from __future__ import annotations import json import re from concurrent.futures import ThreadPoolExecutor from datetime import datetime, timezone from sqlalchemy.orm import Session from database import SessionLocal from models import ( QuizRecord, User, UserSettings, WikiLink, WikiLog, WikiPage, WikiSource, Word, ) from services.llm_service import LlmServiceError, llm_service from services.wiki_llm_quota_service import WikiLlmQuotaExceeded, wiki_llm_quota_service def utc_now_iso() -> str: return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") def markdown_escape(value: str | None) -> str: return (value or "").replace("\r", " ").replace("\n", " ").strip() def word_title(word: Word) -> str: if word.source_lang == "en": return word.source_text if word.target_lang == "en": return word.target_text return word.source_text def word_meaning(word: Word) -> str: if word.source_lang == "zh": return word.source_text if word.target_lang == "zh": return word.target_text return word.target_text def slug_part(value: str) -> str: normalized = re.sub(r"[^a-z0-9]+", "-", value.lower()).strip("-") return normalized[:72] or "word" class WikiService: GRAPH_NODE_LIMIT = 120 def __init__(self) -> None: # Serialize Wiki compilation so repeated answers for the same word cannot # race on page/link upserts. API requests never wait for this queue. self._executor = ThreadPoolExecutor(max_workers=1, thread_name_prefix="wordloop-wiki") def enqueue_word(self, user_id: int, word_id: int) -> None: self._executor.submit(self._ingest_in_background, user_id, word_id, None) def enqueue_quiz(self, user_id: int, word_id: int, record_id: int) -> None: self._executor.submit(self._ingest_in_background, user_id, word_id, record_id) def _ingest_in_background( self, user_id: int, word_id: int, record_id: int | None, ) -> None: db = SessionLocal() try: user = db.query(User).filter(User.id == user_id).first() word = ( db.query(Word) .filter(Word.id == word_id, Word.user_id == user_id) .first() ) if not user or not word: return if record_id is None: self.ingest_word(db, user, word) return record = ( db.query(QuizRecord) .filter( QuizRecord.id == record_id, QuizRecord.user_id == user_id, QuizRecord.word_id == word_id, ) .first() ) if record: self.ingest_quiz(db, user, word, record) except Exception: db.rollback() finally: db.close() def get_settings(self, db: Session, user: User) -> UserSettings: settings = db.query(UserSettings).filter(UserSettings.user_id == user.id).first() if not settings: settings = UserSettings(user_id=user.id) db.add(settings) db.flush() return settings def status(self, db: Session, user: User) -> dict: settings = self.get_settings(db, user) page_count = db.query(WikiPage).filter(WikiPage.user_id == user.id).count() link_count = db.query(WikiLink).filter(WikiLink.user_id == user.id).count() latest = ( db.query(WikiPage) .filter(WikiPage.user_id == user.id, WikiPage.category == "word") .order_by(WikiPage.updated_at.desc()) .first() ) latest_log = ( db.query(WikiLog) .filter(WikiLog.user_id == user.id) .order_by(WikiLog.created_at.desc()) .first() ) quota = wiki_llm_quota_service.quota_dict(db, user) llm_ready = llm_service.is_configured() return { "enabled": bool(settings.ai_wiki_enabled), "auto_organize": bool(settings.ai_wiki_auto_organize), "page_count": page_count, "link_count": link_count, "last_organized_at": latest_log.created_at if latest_log else None, "latest_page": self.page_dict(latest) if latest else None, "llm_configured": llm_ready, "llm_model": llm_service.model_name() if llm_ready else None, "compiler_mode": "deepseek" if llm_ready else "deterministic", "daily_llm_limit": quota["daily_limit"], "llm_used_today": quota["used_today"], "llm_remaining_today": quota["remaining_today"], "quota_exceeded": quota["quota_exceeded"], "recharge_message": quota["recharge_message"], } def update_settings( self, db: Session, user: User, *, enabled: bool | None, auto_organize: bool | None, ) -> dict: settings = self.get_settings(db, user) was_enabled = bool(settings.ai_wiki_enabled) if enabled is not None: settings.ai_wiki_enabled = 1 if enabled else 0 if not enabled: settings.ai_wiki_auto_organize = 0 if auto_organize is not None: settings.ai_wiki_auto_organize = ( 1 if auto_organize and bool(settings.ai_wiki_enabled) else 0 ) db.commit() if bool(settings.ai_wiki_enabled) and not was_enabled: self.rebuild(db, user) return self.status(db, user) def graph(self, db: Session, user: User) -> dict: pages = ( db.query(WikiPage) .filter(WikiPage.user_id == user.id) .order_by(WikiPage.updated_at.desc(), WikiPage.id.desc()) .limit(self.GRAPH_NODE_LIMIT) .all() ) if not pages: return {"nodes": [], "links": []} page_ids = {page.id for page in pages} links = ( db.query(WikiLink) .filter( WikiLink.user_id == user.id, WikiLink.source_page_id.in_(page_ids), WikiLink.target_page_id.in_(page_ids), ) .all() ) degree: dict[int, int] = {page.id: 0 for page in pages} for link in links: degree[link.source_page_id] = degree.get(link.source_page_id, 0) + 1 degree[link.target_page_id] = degree.get(link.target_page_id, 0) + 1 return { "nodes": [ { "id": str(page.id), "slug": page.slug, "title": page.title, "category": page.category, "summary": page.summary, "source_count": page.source_count, "updated_at": page.updated_at, "degree": degree.get(page.id, 0), } for page in pages ], "links": [ { "source": str(link.source_page_id), "target": str(link.target_page_id), "relation": link.relation, } for link in links ], } def ingest_word(self, db: Session, user: User, word: Word) -> None: settings = self.get_settings(db, user) if not bool(settings.ai_wiki_enabled): return self._capture_word_source(db, user, word) if bool(settings.ai_wiki_auto_organize): db.flush() self._compile_word(db, user, word) self._compile_index(db, user) db.commit() def ingest_quiz( self, db: Session, user: User, word: Word, record: QuizRecord, ) -> None: settings = self.get_settings(db, user) if not bool(settings.ai_wiki_enabled): return self._capture_quiz_source(db, user, word, record) if bool(settings.ai_wiki_auto_organize): db.flush() self._compile_word(db, user, word) self._compile_index(db, user) db.commit() def rebuild(self, db: Session, user: User) -> dict: settings = self.get_settings(db, user) if not bool(settings.ai_wiki_enabled): return self.status(db, user) if llm_service.is_configured(): wiki_llm_quota_service.require_available(db, user) words = ( db.query(Word) .filter(Word.user_id == user.id) .order_by(Word.created_at.asc()) .all() ) for word in words: self._capture_word_source(db, user, word) records = ( db.query(QuizRecord) .filter(QuizRecord.user_id == user.id, QuizRecord.word_id == word.id) .order_by(QuizRecord.created_at.asc()) .all() ) for record in records: self._capture_quiz_source(db, user, word, record) db.flush() self._compile_word(db, user, word, use_llm=False) self._compile_index(db, user) self._log(db, user, "rebuild", "index", f"重建 {len(words)} 个单词知识页") db.commit() return self.status(db, user) def remove_word(self, db: Session, user: User, word: Word) -> None: pages = ( db.query(WikiPage) .filter(WikiPage.user_id == user.id, WikiPage.word_id == word.id) .all() ) page_ids = [page.id for page in pages] if page_ids: db.query(WikiLink).filter( WikiLink.user_id == user.id, ( WikiLink.source_page_id.in_(page_ids) | WikiLink.target_page_id.in_(page_ids) ), ).delete(synchronize_session=False) db.query(WikiPage).filter( WikiPage.user_id == user.id, WikiPage.id.in_(page_ids) ).delete(synchronize_session=False) db.query(WikiSource).filter( WikiSource.user_id == user.id, WikiSource.word_id == word.id ).delete(synchronize_session=False) def refresh_index(self, db: Session, user: User) -> None: settings = self.get_settings(db, user) if bool(settings.ai_wiki_enabled): self._compile_index(db, user) def page_dict(self, page: WikiPage) -> dict: return { "slug": page.slug, "title": page.title, "category": page.category, "summary": page.summary, "content": page.content, "source_count": page.source_count, "updated_at": page.updated_at, } def _capture_word_source(self, db: Session, user: User, word: Word) -> None: key = f"word:{word.id}" if ( db.query(WikiSource) .filter(WikiSource.user_id == user.id, WikiSource.source_key == key) .first() ): return payload = { "word_id": word.id, "source_text": word.source_text, "target_text": word.target_text, "phonetic": word.phonetic, "example_en": word.example_en, "example_cn": word.example_cn, "book_id": word.book_id, "created_at": word.created_at, } db.add( WikiSource( user_id=user.id, word_id=word.id, source_key=key, source_type="word", payload=json.dumps(payload, ensure_ascii=False), created_at=word.created_at, ) ) def _capture_quiz_source( self, db: Session, user: User, word: Word, record: QuizRecord, ) -> None: key = f"quiz:{record.id}" if ( db.query(WikiSource) .filter(WikiSource.user_id == user.id, WikiSource.source_key == key) .first() ): return payload = { "quiz_record_id": record.id, "word_id": word.id, "question_type": record.question_type, "user_answer": record.user_answer, "correct_answer": record.correct_answer, "is_correct": bool(record.is_correct), "duration_seconds": record.duration_seconds, "created_at": record.created_at, } db.add( WikiSource( user_id=user.id, word_id=word.id, source_key=key, source_type="quiz", payload=json.dumps(payload, ensure_ascii=False), created_at=record.created_at, ) ) def _compile_word( self, db: Session, user: User, word: Word, *, require_llm_quota: bool = False, use_llm: bool = True, ) -> WikiPage: now = utc_now_iso() title = markdown_escape(word_title(word)) meaning = markdown_escape(word_meaning(word)) slug = f"vocabulary/{word.id}-{slug_part(title)}" sources = ( db.query(WikiSource) .filter(WikiSource.user_id == user.id, WikiSource.word_id == word.id) .order_by(WikiSource.created_at.desc()) .all() ) records = ( db.query(QuizRecord) .filter(QuizRecord.user_id == user.id, QuizRecord.word_id == word.id) .order_by(QuizRecord.created_at.desc()) .limit(8) .all() ) status_page = self._compile_collection( db, user, f"learning/{word.status}", self._status_title(word.status), f"当前处于“{self._status_title(word.status)}”状态的单词集合。", ) book_slug = f"books/{word.book_id if word.book_id > 0 else 'accumulation'}" book_page = self._compile_collection( db, user, book_slug, "词书练习" if word.book_id > 0 else "日常积累", "按学习来源整理的词汇集合。", ) attempts = [ ( f"- {record.created_at} · {record.question_type} · " f"{'答对' if record.is_correct else '答错'} · " f"回答：{markdown_escape(record.user_answer)}" ) for record in records ] content = "\n".join( [ "---", f"title: {title}", f"word_id: {word.id}", f"status: {word.status}", f"mastery: {word.mastery_score}", f"updated_at: {now}", "tags: [wordloop, vocabulary]", "---", "", f"# {title}", "", f"**释义：** {meaning}", f"**音标：** {markdown_escape(word.phonetic) or '暂无'}", f"**掌握度：** {word.mastery_score}% · 正确 {word.correct_count} 次 · " f"错误 {word.wrong_count} 次", "", "## 例句", "", markdown_escape(word.example_en) or "暂无英文例句。", "", markdown_escape(word.example_cn) or "暂无中文例句。", "", "## 学习轨迹", "", *(attempts or ["- 暂无练习记录。"]), "", "## 关联", "", f"- [[{status_page.slug}|{status_page.title}]]", f"- [[{book_page.slug}|{book_page.title}]]", "", "## 来源", "", f"- WordLoop 原始学习事件 {len(sources)} 条", ] ) summary = f"{meaning}；掌握度 {word.mastery_score}%，累计练习 {word.train_count} 次。" if use_llm: content, summary, llm_used = self._maybe_enhance_with_llm( db, user, title=title, draft_content=content, draft_summary=summary, require_quota=require_llm_quota, ) else: llm_used = False page = self._upsert_page( db, user, slug=slug, title=title, category="word", summary=summary, content=content, source_count=len(sources), word_id=word.id, now=now, ) db.flush() db.query(WikiLink).filter( WikiLink.user_id == user.id, WikiLink.source_page_id == page.id ).delete(synchronize_session=False) for target, relation in ((status_page, "learning_status"), (book_page, "collection")): db.add( WikiLink( user_id=user.id, source_page_id=page.id, target_page_id=target.id, relation=relation, created_at=now, ) ) detail = f"更新词汇页 {title}" if llm_used: detail += "（DeepSeek 整理）" self._log(db, user, "ingest", page.slug, detail) return page def _maybe_enhance_with_llm( self, db: Session, user: User, *, title: str, draft_content: str, draft_summary: str, require_quota: bool, ) -> tuple[str, str, bool]: if not llm_service.is_configured(): return draft_content, draft_summary, False try: if require_quota: wiki_llm_quota_service.require_available(db, user) elif wiki_llm_quota_service.remaining_today(db, user) <= 0: return draft_content, draft_summary, False enhanced = llm_service.enhance_wiki_page(title=title, draft_markdown=draft_content) wiki_llm_quota_service.consume(db, user) enhanced_summary = self._extract_summary(enhanced, draft_summary) return enhanced, enhanced_summary, True except WikiLlmQuotaExceeded: if require_quota: raise return draft_content, draft_summary, False except LlmServiceError: return draft_content, draft_summary, False def _extract_summary(self, content: str, fallback: str) -> str: for line in content.splitlines(): stripped = line.strip() if not stripped or stripped.startswith("#") or stripped.startswith("---"): continue if stripped.startswith("**") and stripped.endswith("**"): return stripped.strip("*").strip() if len(stripped) <= 120: return stripped return fallback def _compile_collection( self, db: Session, user: User, slug: str, title: str, summary: str, ) -> WikiPage: now = utc_now_iso() content = f"# {title}\n\n{summary}\n\n此页面由 WordLoop Wiki 自动维护。" return self._upsert_page( db, user, slug=slug, title=title, category="collection", summary=summary, content=content, source_count=0, word_id=None, now=now, ) def _compile_index(self, db: Session, user: User) -> WikiPage: now = utc_now_iso() self._prune_orphan_collections(db, user) pages = ( db.query(WikiPage) .filter(WikiPage.user_id == user.id, WikiPage.category == "word") .order_by(WikiPage.updated_at.desc()) .all() ) entries = [f"- [[{page.slug}|{page.title}]] — {page.summary}" for page in pages] content = "\n".join( [ "# WordLoop Wiki", "", "由学习记录持续编译的个人知识库。", "", "## 词汇页", "", *(entries or ["- 暂无知识页。"]), ] ) return self._upsert_page( db, user, slug="index", title="WordLoop Wiki", category="index", summary=f"共 {len(pages)} 个词汇知识页。", content=content, source_count=sum(page.source_count for page in pages), word_id=None, now=now, ) def _prune_orphan_collections(self, db: Session, user: User) -> None: db.flush() collections = ( db.query(WikiPage) .filter(WikiPage.user_id == user.id, WikiPage.category == "collection") .all() ) for page in collections: inbound = ( db.query(WikiLink) .filter( WikiLink.user_id == user.id, WikiLink.target_page_id == page.id, ) .count() ) if inbound == 0: db.delete(page) db.flush() def _upsert_page( self, db: Session, user: User, *, slug: str, title: str, category: str, summary: str, content: str, source_count: int, word_id: int | None, now: str, ) -> WikiPage: page = ( db.query(WikiPage) .filter(WikiPage.user_id == user.id, WikiPage.slug == slug) .first() ) if not page: page = WikiPage( user_id=user.id, slug=slug, created_at=now, ) db.add(page) page.word_id = word_id page.title = title page.category = category page.summary = summary page.content = content page.source_count = source_count page.updated_at = now db.flush() return page def _log( self, db: Session, user: User, action: str, page_slug: str | None, detail: str, ) -> None: db.add( WikiLog( user_id=user.id, action=action, page_slug=page_slug, detail=detail, created_at=utc_now_iso(), ) ) def _status_title(self, status: str) -> str: return { "new": "新词", "learning": "学习中", "mastered": "已掌握", "weak": "易错词", }.get(status, status) wiki_service = WikiService()