diff --git a/.claude/agents/test-runner.md b/.claude/agents/test-runner.md new file mode 100644 index 0000000..7e60cac --- /dev/null +++ b/.claude/agents/test-runner.md @@ -0,0 +1,27 @@ +--- +name: test-runner +description: Runs the project's pytest suite + ruff in a fresh context and reports PASS/FAIL with evidence. Bash + Read only; never edits code. Use after any change to confirm nothing regressed. +tools: Bash, Read +--- + +你是獨立的測試執行稽核員，運作在**全新、未被開發者汙染的脈絡**裡。 +你的唯一工作：**實際把整套測試與 lint 跑起來**，如實回報 PASS / FAIL，不修改任何檔案。 + +步驟： +1. 跑測試 + 覆蓋率： + `cd /Users/ron/Desktop/agent-notes && .venv/bin/python -m pytest --cov=agentnotes --cov-report=term-missing` +2. 跑 lint： + `.venv/bin/ruff check src tests` +3. 這些都是離線測試（不需 NotebookLM / Notion / Telegram），也不該動到真實 `.state/`。 + 跑完用 `ls .state/processed .state/downloads 2>&1` 確認這兩個資料夾沒被測試建立。 +4. 若 pytest 或 ruff 有任何失敗： + - 指出**哪個 test 失敗、哪一行 assert**（pytest 會印出 traceback 與 `assert` 左右值）。 + - 用 Read 打開該測試與對應原始碼（`src/agentnotes/...`），判斷是測試過時還是真的 bug，給檔案:行定位與可能原因。 + - **不要嘗試修復**，只回報。 + +輸出格式：**第一行** `PASS` 或 `FAIL`。其後附： +- pytest 結果（通過數 / 失敗數）與 ruff 結果（All checks passed 或錯誤列表）。 +- 覆蓋率總結那一行（TOTAL ...）。 +- 若 FAIL：失敗測試名稱、assert 訊息、可能原因（檔案:行）。 + +鐵則：**沒有親眼看到 pytest 印出全部通過、且 ruff 全綠之前不能回報 PASS**。寧可 FAIL 也不放過。 diff --git a/.env.example b/.env.example index 5243201..eb55802 100644 --- a/.env.example +++ b/.env.example @@ -6,10 +6,13 @@ TELEGRAM_BOT_TOKEN= TELEGRAM_ALLOWED_CHAT_IDS= # --- Notion (via Notion MCP in the Agent SDK) --- -# A Notion page the Notion MCP integration can access; the bot creates the report DB under it on first run. +# A Notion page the Notion MCP integration can access. Each topic (see topics.json) gets +# its own "NotebookLM 報告 - " database, auto-created under THIS page the first time +# that topic is used. Required for any topic whose DB isn't pre-created/cached. NOTION_PARENT_PAGE_ID= -# Leave empty on first run; the bot fills this after auto-creating the "NotebookLM 報告" database. -NOTION_DATABASE_ID= +# (Per-topic DB ids are NOT set here. They live in .state/topic_dbs.json, which the bot +# fills the first time it auto-creates a topic's database — seed it manually to reuse an +# existing DB, e.g. {"ai-agent": ""}.) # --- Auth --- # Leave ANTHROPIC_API_KEY UNSET to use the Claude subscription pool (2026-06-15 billing split). diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..3ae2b9c --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,21 @@ +name: CI + +on: + push: + pull_request: + +jobs: + test: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: "3.11" + cache: pip + - name: Install (package + dev tools) + run: pip install -e ".[dev]" + - name: Lint (ruff) + run: ruff check src tests + - name: Test (pytest + coverage) + run: pytest --cov=agentnotes --cov-report=term-missing diff --git a/.gitignore b/.gitignore index 481d058..40bf161 100644 --- a/.gitignore +++ b/.gitignore @@ -23,3 +23,9 @@ logs/ # vendored notebooklm-skill clone .vendor/ + +# test / lint artifacts +.coverage +.pytest_cache/ +.ruff_cache/ +htmlcov/ diff --git a/PROGRESS.md b/PROGRESS.md index 39534e6..81130e7 100644 --- a/PROGRESS.md +++ b/PROGRESS.md @@ -17,3 +17,12 @@ cwc-style handoff log. The bot appends one block per job so a restart can see hi ## Job log - 2026-06-28 23:32 | 1782660251-5117498666-3 | https://www.youtube.com/watch?v=tTcxVv8HHNw | OK https://www.notion.so/38d5fc56c4398131827cc231a593c709 +- 2026-06-29 15:35 | 1782717988-5117498666-6 | https://www.youtube.com/watch?v=mR-WAvEPRwE | OK https://www.notion.so/38e5fc56c439819e8542e4a612b7516a +- 2026-06-30 15:00 | 1782802268-5117498666-17 | https://www.youtube.com/watch?v=GrNbuWWJYiI | OK https://www.notion.so/38f5fc56c4398180abcad3add5b122c8 +- 2026-06-30 16:10 | 1782806407-5117498666-20 | https://www.youtube.com/watch?v=Xu5gz2qsaz8 | OK https://www.notion.so/38f5fc56c4398123aa88edd9eed1b278 +- 2026-06-30 22:46 | 1782830779-5117498666-25 | ai-agent | https://www.youtube.com/watch?v=mR-WAvEPRwE | EXCEPTION Exception('Claude Code returned an error result: success') +- 2026-06-30 22:48 | 1782830918-5117498666-28 | ai-agent | https://www.youtube.com/watch?v=mR-WAvEPRwE | EXCEPTION Exception('Claude Code returned an error result: success') +- 2026-07-01 00:01 | 1782835304-5117498666-33 | ai-agent | https://www.youtube.com/watch?v=mR-WAvEPRwE | EXCEPTION Exception('Claude Code returned an error result: success') +- 2026-07-01 00:32 | 1782836492-5117498666-38 | ai-agent | https://www.youtube.com/watch?v=mR-WAvEPRwE | OK https://www.notion.so/38f5fc56c4398149a61bf5f78cd1c2e5 +- 2026-07-01 00:39 | 1782837592-5117498666-41 | ai-agent | https://www.youtube.com/watch?v=mR-WAvEPRwE | DUP https://www.notion.so/38f5fc56c4398149a61bf5f78cd1c2e5 +- 2026-07-01 00:48 | 1782837603-5117498666-44 | ai-agent | https://www.youtube.com/watch?v=mR-WAvEPRwE | OK https://www.notion.so/38f5fc56c43981cbbccec8e802505b8d diff --git a/README.md b/README.md index 1bd2991..2cf3f82 100644 --- a/README.md +++ b/README.md @@ -24,9 +24,25 @@ builderA(產報告) → 閘A(報告) → builderB(寫Notion) → 閘B(Notion) ## 行為細節 -- **共用一個 NotebookLM 筆記本**：所有來源都加進名為 **「給 Agent Note」** 的同一個筆記本（不存在才建一次）， - 每次提問用 `ask -s ` 把範圍鎖在這次新增的來源，避免被筆記本內其他來源汙染。筆記本名稱在 - `pipeline.py` 的 `NOTEBOOK_NAME`。 +- **整理既有筆記本（sweep）**：傳 `#topic 整理現有`（或 `sweep`/`sync`），bot 會把該 topic 筆記本裡 + **每個還沒匯出的來源**各自產一份報告、各自一頁 Notion；已匯出的自動跳過（可反覆執行、不重做）。 + URL 來源用正規化網址比對，所以跟 `#topic <連結>` 走的是同一套防重複——已用連結匯過的不會再做一次。 +- **連結或檔案都收**：傳 http/https 連結，或直接上傳附件檔（**PDF / DOCX / TXT / MD**，NotebookLM 能吃的）。 + 附件的 `#topic` 標籤放在 caption。bot 用 Telegram getFile 下載到 `.state/downloads/`，再 `source add --type file` + 進對應筆記本。限制：Telegram bot API 只能下載 ≤20MB 的檔；超過會如實回失敗（不假裝成功）。 +- **依 topic 分流（每個 topic 一本筆記本、一個 Notion DB）**：傳連結時用 `#標籤` 指定 topic，例如 + `#ai 你的連結` 或 `你的連結 #phd`。bot 依 `topics.json` 把來源加進對應的 NotebookLM 筆記本 + （如 `(AgentNote) AI agent`），整理後寫進對應的 Notion 資料庫（如 `NotebookLM 報告 - AI agent`）。 + 好處是順手在 NotebookLM 建立分主題的 RAG 語料庫。**沒標、標到不存在、或一次標多個 topic → bot 不處理， + 直接回你目前可用的 topic 清單**（fail-closed，避免 typo 汙染語料庫）。 +- **同一筆記本內仍鎖來源**：每次提問用 `ask -s ` 把範圍鎖在這次新增的來源，避免被同一本裡其他來源汙染。 +- **防重複**：每個 topic 一份「已處理索引」（`.state/processed/.json`，gitignored）。同一個 (topic, 連結) + 成功處理過後再傳一次 → bot **不重跑**，直接回你上次的 Notion 連結。比對用的是**正規化後的 URL** + （`youtu.be`↔`watch?v=`、`utm_*`/`si`/`fbclid` 等追蹤參數、結尾斜線、`www.`、`http/https`、`#fragment` 都會被忽略）。 + 索引**只在整條成功後才寫**，失敗的 job 仍可重試。要強制重做就在訊息加 `#force`。 +- **topic 設定兩層**：定義在 `topics.json`（進 repo：別名 + 筆記本名 + Notion DB 標題）； + 自動建出來的 `topic -> notion_db_id` 快取在 `.state/topic_dbs.json`（gitignored）。新增 topic = 在 + `topics.json` 加一筆，之後該 topic 的筆記本與 Notion DB 會在第一次用到時自動建立。 - **Notion 頁開頭自動加目錄**：每頁第一個區塊是 Notion 原生 Table of Contents block，依 `##` 標題自動產生， 方便點擊跳轉。 @@ -37,13 +53,34 @@ src/agentnotes/ contract.py Default-FAIL 證據契約（核心防線） evaluators.py 確定性檢查 + 稽核員 prompt（核心防線） agent.py Agent SDK 封裝：builder（可寫/可 Notion）vs evaluator（read-only、新 context） - pipeline.py 固定單趟 pipeline，把 evaluator 當閘 + pipeline.py 單一來源的證據閘 pipeline（_run_source）；run_pipeline=新來源、run_sweep=整理既有 + notebooklm_cli.py Python 直呼 notebooklm CLI（找筆記本 id、列來源）—— sweep 用來逐來源防重複 + topics.py topic registry：#標籤 → (NotebookLM 筆記本, Notion DB) 路由 + db_id 快取 + dedup.py 防重複：URL 正規化 + 每 topic 的「已處理索引」（#force 可覆寫） ingest/base.py InputAdapter 介面（Line 之後接同介面） - ingest/telegram.py getUpdates 長輪詢 + sendMessage + ingest/telegram.py getUpdates 長輪詢 + sendMessage + 解析 #topic + 下載附件檔 main.py 派工迴圈 -.claude/agents/evaluator.md 稽核員子代理人定義 +topics.json topic 定義（別名 / 筆記本名 / Notion DB 標題）—— 進 repo +.state/topic_dbs.json 自動建立後的 topic → notion_db_id 快取（gitignored） +.claude/agents/evaluator.md 稽核員子代理人定義（產出 PASS/FAIL） +.claude/agents/test-runner.md 測試執行子代理人：全新脈絡跑 pytest+ruff 回報 deploy/com.ron.agentnotes.plist launchd 常駐 -tests/test_evidence.py 證據防線的離線測試（不需任何外部服務） +.github/workflows/ci.yml CI：push/PR 自動跑 ruff + pytest（含覆蓋率） + +tests/ pytest 套件（全離線，不需外部服務） + conftest.py 共用 fixtures：cfg / registry / index / good_report / + fake_agents（可調 gate 結果的假 builder+evaluator）/ poll_once（驅動 poll） + test_evidence.py 證據防線：確定性檢查 + Default-FAIL 契約 + test_evaluators.py 確定性檢查的邊界（marker、門檻、失敗標題、計數） + test_contract.py 契約不變量（無證據不過、未知 gate 報錯、磁碟往返） + test_config.py .env 解析 + chat-id allowlist + test_topics.py #標籤 解析、別名正規化、db_id 快取 + test_dedup.py URL 正規化、檔案雜湊、索引、跳過/入帳/#force + test_gate.py verdict 解析 + Gate B 對真實頁面正文做檢查 + test_attachments.py 附件型別 gate、報告 prompt 檔案分支、poll 文件路由 + test_sweep.py 整理既有：指令偵測、逐來源防重複/失敗/#force + test_telegram.py poll() URL 路由（topic 缺/多、多連結、白名單、offset） + test_main.py _title_hint / _job_id / handle_job 派工 ``` 認證：靠本機已登入的 **Claude 訂閱**（2026-06-15 分池計費），**不要**設 `ANTHROPIC_API_KEY`， @@ -73,9 +110,17 @@ cp .env.example .env ## 驗證（由內而外，全部證據導向） ```bash -# A) 證據防線（不需任何外部服務）—— 先確認「防謊報」本身是對的 -python tests/test_evidence.py -# 驗：正常報告 PASS；空白/登入頁/過短 → FAIL；契約沒證據不給過 +# 0) 一次性：裝開發/測試工具（pytest、pytest-cov、ruff） +.venv/bin/python -m pip install -e ".[dev]" + +# A) 整個離線測試套件（不需任何外部服務）—— pytest 自動 discovery +.venv/bin/python -m pytest # 跑全部 +.venv/bin/python -m pytest --cov=agentnotes --cov-report=term-missing # 帶覆蓋率 +.venv/bin/ruff check src tests # lint +# 涵蓋：證據防線（防謊報）、contract、config、topic 路由、防重複、附件、sweep、 +# telegram poll 路由、main 派工。共用 fixtures 在 tests/conftest.py。 +# 也可用「test-runner」子代理人在全新脈絡自動跑 pytest+ruff 並回報 PASS/FAIL +# （.claude/agents/test-runner.md）。CI 在 push/PR 會自動跑同一套（.github/workflows/ci.yml）。 # B) NotebookLM 段：對一支已知影片實跑，人工讀 jobs//report.md 確認繁中、逐段、## header # C) 整鏈：把一個連結傳給 bot → 收到 ✅ + 可開的 Notion 連結 diff --git a/pyproject.toml b/pyproject.toml index ffe1408..59e0a34 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -8,6 +8,9 @@ dependencies = [ "httpx", ] +[project.optional-dependencies] +dev = ["pytest>=8", "pytest-cov", "ruff"] + [project.scripts] agent-notes = "agentnotes.main:run" @@ -17,3 +20,23 @@ where = ["src"] [build-system] requires = ["setuptools>=68"] build-backend = "setuptools.build_meta" + +[tool.pytest.ini_options] +testpaths = ["tests"] +pythonpath = ["src"] +addopts = "-q --strict-markers" + +[tool.ruff] +target-version = "py311" +# 120 not 88/100: the bot's prompts/replies are full of CJK, and ruff counts code +# points — 100 CJK chars already render ~200 columns wide, so a tighter cap would +# force ugly mid-sentence breaks in Chinese string literals. +line-length = 120 +src = ["src", "tests"] + +[tool.ruff.lint] +select = ["E", "F", "W", "I", "UP", "B"] + +[tool.coverage.run] +source = ["agentnotes"] +branch = true diff --git a/src/agentnotes/agent.py b/src/agentnotes/agent.py index 938deb8..1d20661 100644 --- a/src/agentnotes/agent.py +++ b/src/agentnotes/agent.py @@ -37,6 +37,23 @@ class StepTimeout(Exception): """An agent step exceeded its wall-clock budget.""" + +class AgentError(Exception): + """A builder/evaluator step failed for an actionable reason (turned into a clear + user-facing message). Most common: the spawned ``claude`` CLI is not logged in, + which the SDK surfaces as the cryptic "returned an error result: success".""" + + +# Substrings in an SDK error that mean "the spawned claude couldn't even start a turn" +# — overwhelmingly an auth/login problem (e.g. after a CLI upgrade wiped the session). +_AUTH_HINT_MARKERS = ("returned an error result", "not logged in", "/login", "please run") + +_AUTH_HELP = ( + "Claude CLI 可能未登入（常見於 CLI 升級後 session 失效）。" + "請在終端機跑 `claude auth status` 確認，必要時 `claude auth login`" + "（背景常駐可改用 `claude setup-token` 取得長效權杖）。" +) + # Tools the builder may use. "Skill" lets it invoke the notebooklm skill, which # shells out to python via "Bash". BUILDER_TOOLS = ["Skill", "Bash", "Read", "Write", "Edit", "Glob", "Grep"] @@ -142,10 +159,16 @@ async def _consume() -> AgentRun: timeout = float(os.environ.get("JOB_STEP_TIMEOUT_S", DEFAULT_STEP_TIMEOUT_S)) try: return await asyncio.wait_for(_consume(), timeout=timeout) - except asyncio.TimeoutError as e: + except TimeoutError as e: raise StepTimeout( f"agent 步驟超過 {int(timeout)} 秒未完成（可能 NotebookLM/Playwright/MCP 卡住）" ) from e + except Exception as e: + # Map the SDK's cryptic auth failure into an actionable message; re-raise the rest. + text = str(e).lower() + if any(m in text for m in _AUTH_HINT_MARKERS): + raise AgentError(f"{_AUTH_HELP}\n原始錯誤：{e}") from e + raise async def run_builder(prompt: str, *, cwd: Path = ROOT, with_notion: bool = False, diff --git a/src/agentnotes/config.py b/src/agentnotes/config.py index d09af06..8f0339b 100644 --- a/src/agentnotes/config.py +++ b/src/agentnotes/config.py @@ -30,13 +30,12 @@ def _load_dotenv(path: Path) -> None: class Config: telegram_bot_token: str telegram_allowed_chat_ids: frozenset[int] - notion_parent_page_id: str - notion_database_id: str + notion_parent_page_id: str # where a topic's Notion DB is auto-created on first use report_min_cjk_chars: int report_min_headers: int @classmethod - def load(cls) -> "Config": + def load(cls) -> Config: _load_dotenv(ROOT / ".env") raw_ids = os.environ.get("TELEGRAM_ALLOWED_CHAT_IDS", "").strip() ids = frozenset( @@ -46,7 +45,6 @@ def load(cls) -> "Config": telegram_bot_token=os.environ.get("TELEGRAM_BOT_TOKEN", "").strip(), telegram_allowed_chat_ids=ids, notion_parent_page_id=os.environ.get("NOTION_PARENT_PAGE_ID", "").strip(), - notion_database_id=os.environ.get("NOTION_DATABASE_ID", "").strip(), report_min_cjk_chars=int(os.environ.get("REPORT_MIN_CJK_CHARS", "400")), report_min_headers=int(os.environ.get("REPORT_MIN_HEADERS", "3")), ) diff --git a/src/agentnotes/contract.py b/src/agentnotes/contract.py index e464282..e47cf94 100644 --- a/src/agentnotes/contract.py +++ b/src/agentnotes/contract.py @@ -21,7 +21,7 @@ class Contract: state: dict = field(default_factory=dict) @classmethod - def create(cls, job_dir: Path) -> "Contract": + def create(cls, job_dir: Path) -> Contract: """Create a fresh Default-FAIL contract for a job.""" job_dir.mkdir(parents=True, exist_ok=True) state = {g: {"passes": False, "evidence": None, "note": None} for g in GATES} @@ -30,7 +30,7 @@ def create(cls, job_dir: Path) -> "Contract": return c @classmethod - def load(cls, job_dir: Path) -> "Contract": + def load(cls, job_dir: Path) -> Contract: path = job_dir / "contract.json" return cls(path=path, state=json.loads(path.read_text(encoding="utf-8"))) diff --git a/src/agentnotes/dedup.py b/src/agentnotes/dedup.py new file mode 100644 index 0000000..131b5e3 --- /dev/null +++ b/src/agentnotes/dedup.py @@ -0,0 +1,156 @@ +"""Duplicate prevention: per-topic index of already-processed sources. + +The pipeline records every (topic, url) it finishes successfully here. Before a +new job runs, it checks this index: a hit (without #force) means "we already made +a Notion page for this" — reply with the previous link, touch nothing. This keeps +the NotebookLM RAG corpus clean and avoids duplicate Notion pages. + +Two layers, mirroring the topic-db cache: +* on disk under ``.state/processed/.json`` (gitignored runtime state); +* matched on a NORMALIZED url so the same source written two slightly different + ways (youtu.be vs watch?v=, utm_* params, trailing slash) counts as one. + +``base_dir=None`` keeps the index purely in memory (used by offline tests so they +never touch real .state). +""" +from __future__ import annotations + +import hashlib +import json +import logging +from dataclasses import dataclass, field +from datetime import UTC, datetime +from pathlib import Path +from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit + +from .config import STATE_DIR + +log = logging.getLogger("agentnotes.dedup") + +PROCESSED_DIR = STATE_DIR / "processed" + +# Query params that never change which resource a URL points at — drop them so a +# shared/tracked copy of a link dedups against the clean one. +_TRACKING_EXACT = { + "fbclid", "gclid", "gclsrc", "dclid", "si", "igshid", + "mc_cid", "mc_eid", "ref", "ref_src", "spm", "feature", +} + + +def _is_tracking(key: str) -> bool: + k = key.lower() + return k.startswith("utm_") or k in _TRACKING_EXACT + + +def normalize_url(raw: str) -> str: + """Canonicalize a URL into a stable dedup key. + + - scheme coerced to https, host lowercased, leading ``www.`` / ``m.`` dropped; + - YouTube ``youtu.be/ID`` and ``youtube.com/watch?v=ID`` collapse to one key; + - tracking params removed, remaining params sorted, fragment dropped, trailing + slash trimmed. + This is an identity key only (never fetched), so coercing the scheme is safe. + """ + parts = urlsplit((raw or "").strip()) + host = parts.netloc.lower() + if host.startswith("www."): + host = host[4:] + if host == "m.youtube.com": + host = "youtube.com" + path = parts.path + query = [(k, v) for k, v in parse_qsl(parts.query) if not _is_tracking(k)] + + vid = None + if host == "youtu.be": + vid = path.lstrip("/").split("/")[0] + elif host == "youtube.com" and path == "/watch": + vid = dict(query).get("v") + if vid: + host, path, query = "youtube.com", "/watch", [("v", vid)] + + path = path.rstrip("/") + query.sort() + return urlunsplit(("https", host, path, urlencode(query), "")) + + +def file_digest(path: str) -> str: + """Content hash of a local file — the dedup key for uploaded sources (PDF/DOCX/...). + + Keyed on bytes, not filename, so the same paper re-sent under a different name still + dedups. + """ + h = hashlib.sha256() + with open(path, "rb") as f: + for chunk in iter(lambda: f.read(65536), b""): + h.update(chunk) + return "sha256:" + h.hexdigest() + + +@dataclass +class ProcessedIndex: + base_dir: Path | None = PROCESSED_DIR # None == in-memory only (tests) + _cache: dict = field(default_factory=dict) # topic -> {norm_url: {...}} + + @classmethod + def load(cls, base_dir: Path | None = PROCESSED_DIR) -> ProcessedIndex: + return cls(base_dir=base_dir) + + def _path(self, topic: str) -> Path: + return self.base_dir / f"{topic}.json" + + def _topic_map(self, topic: str) -> dict: + if topic in self._cache: + return self._cache[topic] + data: dict = {} + if self.base_dir is not None and self._path(topic).exists(): + try: + data = json.loads(self._path(topic).read_text(encoding="utf-8")) + except (json.JSONDecodeError, OSError) as e: + log.warning("could not read processed index %s: %s", self._path(topic), e) + self._cache[topic] = data + return data + + def _lookup(self, topic: str, key: str) -> dict | None: + return self._topic_map(topic).get(key) + + # ---- explicit-key access (caller computed the key; used by the source layer) ---- + def lookup_key(self, topic: str, key: str) -> dict | None: + return self._lookup(topic, key) + + def record_key(self, topic: str, key: str, notion_url: str, ref: str) -> None: + self._record(topic, key, notion_url, ref) + + def _record(self, topic: str, key: str, notion_url: str, ref: str) -> None: + m = self._topic_map(topic) + m[key] = { + "notion_url": notion_url, + "ref": ref, # original url / filename, for human-readable inspection + "ts": datetime.now(UTC).astimezone().isoformat(timespec="seconds"), + } + if self.base_dir is None: + return + try: + self.base_dir.mkdir(parents=True, exist_ok=True) + self._path(topic).write_text( + json.dumps(m, ensure_ascii=False, indent=2), encoding="utf-8" + ) + except OSError as e: + log.warning("could not persist processed index for %s: %s", topic, e) + + # ---- URL sources (keyed on the normalized URL) ---- + def lookup(self, topic: str, url: str) -> dict | None: + """Return the stored record for an already-processed (topic, url), or None.""" + return self._lookup(topic, normalize_url(url)) + + def record(self, topic: str, url: str, notion_url: str) -> None: + """Mark (topic, url) done. Called only after a job fully succeeds.""" + self._record(topic, normalize_url(url), notion_url, url) + + # ---- File sources (keyed on the file's content hash) ---- + def lookup_file(self, topic: str, file_path: str) -> dict | None: + """Return the stored record for an already-processed (topic, file), or None.""" + return self._lookup(topic, file_digest(file_path)) + + def record_file(self, topic: str, file_path: str, notion_url: str) -> None: + """Mark (topic, file-content) done. Called only after a job fully succeeds.""" + self._record(topic, file_digest(file_path), notion_url, Path(file_path).name) diff --git a/src/agentnotes/ingest/base.py b/src/agentnotes/ingest/base.py index ec86ce0..0bf5045 100644 --- a/src/agentnotes/ingest/base.py +++ b/src/agentnotes/ingest/base.py @@ -22,8 +22,13 @@ def extract_urls(text: str) -> list[str]: class IncomingJob: chat_id: str message_id: str - url: str + url: str # the source URL; "" for an uploaded-file job source_text: str # full original message text (for context / title hints) + topic: str # canonical topic key (which NotebookLM notebook + Notion DB to use) + force: bool = False # #force tag: bypass the duplicate-skip and re-process + file_path: str | None = None # local path of a downloaded attachment (PDF/DOCX/...) + file_name: str | None = None # original attachment filename (for title/source ref) + command: str | None = None # a non-source command, e.g. "sweep" (UC3), else None class InputAdapter(ABC): diff --git a/src/agentnotes/ingest/telegram.py b/src/agentnotes/ingest/telegram.py index 44a59a2..e5d453e 100644 --- a/src/agentnotes/ingest/telegram.py +++ b/src/agentnotes/ingest/telegram.py @@ -7,15 +7,53 @@ from __future__ import annotations import logging +import os from collections.abc import AsyncIterator import httpx from ..config import STATE_DIR +from ..topics import TopicRegistry, extract_tags, normalize from .base import IncomingJob, InputAdapter, extract_urls log = logging.getLogger("agentnotes.telegram") +# The Telegram Bot API can only download files up to 20 MB via getFile. +TELEGRAM_MAX_DOWNLOAD = 20 * 1024 * 1024 + +# Attachment types NotebookLM can ingest as file sources. Accept by MIME or extension. +_ACCEPTED_MIME = { + "application/pdf", + "application/vnd.openxmlformats-officedocument.wordprocessingml.document", # .docx + "text/plain", + "text/markdown", +} +_ACCEPTED_EXT = (".pdf", ".docx", ".txt", ".md") + + +def _accepted_doc(document: dict) -> bool: + mime = (document.get("mime_type") or "").lower() + name = (document.get("file_name") or "").lower() + return mime in _ACCEPTED_MIME or name.endswith(_ACCEPTED_EXT) + + +# "Sweep existing notebook" (UC3) command keywords, compared after normalize(). +_SWEEP_KEYWORDS = {"整理現有", "整理既有", "補齊", "sweep", "sync"} + + +def _is_sweep_command(text: str) -> bool: + """True if the message (minus its #tags) IS exactly a sweep command. + + Exact match, not substring: otherwise ASCII keywords like 'sync'/'sweep' would fire + inside ordinary words ('asynchronous', 'sweep the floor') and — since callers check + this before the URL branch — could hijack a real URL message. A URL-bearing message + can never equal a keyword, so it can never be mistaken for a sweep. + """ + body = text or "" + for tag in extract_tags(body): + body = body.replace(f"#{tag}", "") + return normalize(body) in {normalize(k) for k in _SWEEP_KEYWORDS} + class TelegramAdapter(InputAdapter): name = "telegram" @@ -25,12 +63,15 @@ def __init__( token: str, allowed_chat_ids: frozenset[int] = frozenset(), *, + registry: TopicRegistry, poll_timeout: int = 50, ) -> None: if not token: raise ValueError("TELEGRAM_BOT_TOKEN is empty") self._base = f"https://api.telegram.org/bot{token}" + self._file_base = f"https://api.telegram.org/file/bot{token}" self._allowed = allowed_chat_ids + self._registry = registry self._poll_timeout = poll_timeout self._offset_path = STATE_DIR / "telegram_offset" self._offset: int | None = self._load_offset() @@ -68,8 +109,44 @@ async def _get_updates(self) -> list[dict]: log.warning("getUpdates failed: %s", e) return [] + async def _resolve_topic_or_prompt(self, chat_id, msg, text: str, *, kind: str + ) -> str | None: + """Resolve the #topic tag; if missing/ambiguous, reply with the menu and None. + + A source needs an unambiguous topic, else we'd pollute a RAG corpus. + """ + topic, err = self._registry.resolve_message(text) + if topic is not None: + return topic + hint = ( + "你標了不只一個 topic，請只標一個。\n" + if err == "ambiguous" + else f"{kind}要放哪個 topic？\n" + ) + await self.send_reply( + str(chat_id), hint + self._registry.menu_text(), + reply_to=str(msg.get("message_id")), + ) + return None + + async def _download_document(self, document: dict) -> tuple[str, str]: + """getFile + download an attachment to .state/downloads. Returns (path, name).""" + file_id = document["file_id"] + r = await self._client.get(f"{self._base}/getFile", params={"file_id": file_id}) + r.raise_for_status() + tg_path = r.json()["result"]["file_path"] + dr = await self._client.get(f"{self._file_base}/{tg_path}") + dr.raise_for_status() + name = (document.get("file_name") or os.path.basename(tg_path) or "file").replace("/", "_") + dest_dir = STATE_DIR / "downloads" + dest_dir.mkdir(parents=True, exist_ok=True) + uid = document.get("file_unique_id") or file_id + dest = dest_dir / f"{uid}__{name}" + dest.write_bytes(dr.content) + return str(dest), name + async def poll(self) -> AsyncIterator[IncomingJob]: - """Yield one IncomingJob per URL found in allowed incoming messages. + """Yield one IncomingJob per URL or accepted attachment in allowed messages. The offset is committed (persisted) only AFTER an update's jobs have been consumed — the generator resumes past each `yield` once the caller finishes @@ -90,25 +167,103 @@ async def poll(self) -> AsyncIterator[IncomingJob]: self._commit_offset(next_offset) continue text = msg.get("text", "") or msg.get("caption", "") + document = msg.get("document") + + # ---- attachment (PDF/DOCX/TXT/MD) branch ---- + if document: + async for job in self._handle_document(chat_id, msg, text, document): + yield job + self._commit_offset(next_offset) + continue + + # ---- URL branch (a URL always wins over a command keyword) ---- urls = extract_urls(text) - if not urls: - if text: - await self.send_reply( - str(chat_id), - "請傳一個影片或論文的連結（http/https）給我，我會整理成中文分段報告寫進 Notion。", + if urls: + topic = await self._resolve_topic_or_prompt(chat_id, msg, text, kind="這個連結") + if topic is not None: + force = any(normalize(t) == "force" for t in extract_tags(text)) + for url in urls: + yield IncomingJob( + chat_id=str(chat_id), + message_id=str(msg.get("message_id")), + url=url, + source_text=text, + topic=topic, + force=force, + ) + self._commit_offset(next_offset) + continue + + # ---- "整理現有" command branch (UC3: sweep existing notebook) ---- + if _is_sweep_command(text): + topic = await self._resolve_topic_or_prompt( + chat_id, msg, text, kind="這個指令") + if topic is not None: + force = any(normalize(t) == "force" for t in extract_tags(text)) + yield IncomingJob( + chat_id=str(chat_id), + message_id=str(msg.get("message_id")), + url="", + source_text=text, + topic=topic, + force=force, + command="sweep", ) self._commit_offset(next_offset) continue - for url in urls: - yield IncomingJob( - chat_id=str(chat_id), - message_id=str(msg.get("message_id")), - url=url, - source_text=text, + + # ---- nothing actionable ---- + if text: + await self.send_reply( + str(chat_id), + "請傳一個影片或論文的連結（http/https）給我，我會整理成中文分段報告寫進 Notion。\n\n" + + self._registry.menu_text(), ) - # all jobs for this update have been handled by the caller self._commit_offset(next_offset) + async def _handle_document(self, chat_id, msg, text: str, document: dict + ) -> AsyncIterator[IncomingJob]: + """Validate, download, and yield a job for an attachment (or reply on failure).""" + if not _accepted_doc(document): + await self.send_reply( + str(chat_id), + "這個檔案類型我看不懂，目前支援 PDF / DOCX / TXT / MD。", + reply_to=str(msg.get("message_id")), + ) + return + topic = await self._resolve_topic_or_prompt(chat_id, msg, text, kind="這個檔案") + if topic is None: + return + size = document.get("file_size") or 0 + if size > TELEGRAM_MAX_DOWNLOAD: + await self.send_reply( + str(chat_id), + "這個檔案太大了（超過 20MB，Telegram 機器人無法下載）。請改傳連結，或先壓縮/拆分。", + reply_to=str(msg.get("message_id")), + ) + return + try: + file_path, file_name = await self._download_document(document) + except (httpx.HTTPError, OSError, KeyError, ValueError) as e: + log.warning("download failed for document in chat %s: %s", chat_id, e) + await self.send_reply( + str(chat_id), + "下載這個檔案時失敗，已停下（不會回你假結果）。請再傳一次或改傳連結。", + reply_to=str(msg.get("message_id")), + ) + return + force = any(normalize(t) == "force" for t in extract_tags(text)) + yield IncomingJob( + chat_id=str(chat_id), + message_id=str(msg.get("message_id")), + url="", + source_text=text, + topic=topic, + force=force, + file_path=file_path, + file_name=file_name, + ) + async def _send(self, chat_id: str, text: str, reply_to: str | None = None) -> None: payload: dict = { "chat_id": chat_id, diff --git a/src/agentnotes/main.py b/src/agentnotes/main.py index 5172f38..182f694 100644 --- a/src/agentnotes/main.py +++ b/src/agentnotes/main.py @@ -8,12 +8,15 @@ import asyncio import logging import time -from datetime import datetime, timezone +from datetime import UTC, datetime +from pathlib import Path -from .config import Config, ROOT +from .config import ROOT, Config +from .dedup import ProcessedIndex from .ingest.base import IncomingJob from .ingest.telegram import TelegramAdapter -from .pipeline import run_pipeline +from .pipeline import run_pipeline, run_sweep +from .topics import TopicRegistry, extract_tags log = logging.getLogger("agentnotes") @@ -23,27 +26,68 @@ def _job_id(job: IncomingJob) -> str: def _title_hint(job: IncomingJob) -> str: - text = (job.source_text or "").replace(job.url, "").strip() - return text[:60] if text else "未命名報告" + text = job.source_text or "" + if job.url: + text = text.replace(job.url, "") + for tag in extract_tags(text): # drop the "#topic" / "#force" tokens from the title + text = text.replace(f"#{tag}", "") + text = text.strip() + if text: + return text[:60] + if job.file_name: # attachment with only a tag as caption: fall back to filename + return Path(job.file_name).stem[:60] or "未命名報告" + return "未命名報告" def _log_progress(job_id: str, line: str) -> None: - stamp = datetime.now(timezone.utc).astimezone().strftime("%Y-%m-%d %H:%M") + stamp = datetime.now(UTC).astimezone().strftime("%Y-%m-%d %H:%M") with (ROOT / "PROGRESS.md").open("a", encoding="utf-8") as f: f.write(f"- {stamp} | {job_id} | {line}\n") -async def handle_job(adapter: TelegramAdapter, cfg: Config, job: IncomingJob) -> None: +async def handle_job(adapter: TelegramAdapter, cfg: Config, registry: TopicRegistry, + index: ProcessedIndex, job: IncomingJob) -> None: job_id = _job_id(job) - log.info("dispatch job %s url=%s", job_id, job.url) + + # ---- UC3: "整理現有" — sweep the topic's existing notebook sources ---- + if job.command == "sweep": + log.info("dispatch sweep %s topic=%s force=%s", job_id, job.topic, job.force) + await adapter.send_progress( + job.chat_id, + f"開始整理「{job.topic}」筆記本裡既有的來源…（數量多會花一點時間）", + ) + try: + result = await run_sweep(job_id, cfg, registry, index, job.topic, force=job.force) + except Exception as e: # never crash the loop on one bad job + log.exception("sweep %s crashed", job_id) + _log_progress(job_id, f"{job.topic} | (sweep) | EXCEPTION {e!r}") + await adapter.send_reply( + job.chat_id, + f"⚠️ 整理時發生例外，已停下（不會回你假結果）。\n{type(e).__name__}: {e}", + reply_to=job.message_id, + ) + return + _log_progress(job_id, f"{job.topic} | (sweep) | {'OK' if result.ok else 'FAIL'}") + await adapter.send_reply(job.chat_id, result.message, reply_to=job.message_id) + return + + source_ref = job.url or job.file_name or "(檔案)" # url for links, filename for uploads + log.info("dispatch job %s topic=%s force=%s source=%s", + job_id, job.topic, job.force, source_ref) + kind = "檔案" if job.file_path else "連結" await adapter.send_progress( - job.chat_id, f"收到連結，開始用 NotebookLM 整理中文分段報告…\n{job.url}" + job.chat_id, + f"收到{kind}（topic：{job.topic}），開始用 NotebookLM 整理中文分段報告…\n{source_ref}", ) try: - result = await run_pipeline(job_id, job.url, cfg, title_hint=_title_hint(job)) + result = await run_pipeline( + job_id, job.url, cfg, registry, job.topic, title_hint=_title_hint(job), + force=job.force, index=index, + file_path=job.file_path, file_name=job.file_name, + ) except Exception as e: # never crash the loop on one bad job log.exception("job %s crashed", job_id) - _log_progress(job_id, f"{job.url} | EXCEPTION {e!r}") + _log_progress(job_id, f"{job.topic} | {source_ref} | EXCEPTION {e!r}") await adapter.send_reply( job.chat_id, f"⚠️ 處理時發生例外，已停下（不會回你假結果）。\n{type(e).__name__}: {e}", @@ -51,10 +95,13 @@ async def handle_job(adapter: TelegramAdapter, cfg: Config, job: IncomingJob) -> ) return - _log_progress( - job_id, - f"{job.url} | {'OK ' + (result.notion_url or '') if result.ok else 'FAIL ' + result.message.replace(chr(10), ' ')}", - ) + if result.duplicate: + status = f"DUP {result.notion_url or ''}" + elif result.ok: + status = f"OK {result.notion_url or ''}" + else: + status = "FAIL " + result.message.replace(chr(10), " ") + _log_progress(job_id, f"{job.topic} | {source_ref} | {status}") await adapter.send_reply(job.chat_id, result.message, reply_to=job.message_id) @@ -68,14 +115,19 @@ async def amain() -> None: raise SystemExit( "TELEGRAM_BOT_TOKEN 未設定。請複製 .env.example 成 .env 並填入新 bot 的 token。" ) - adapter = TelegramAdapter(cfg.telegram_bot_token, cfg.telegram_allowed_chat_ids) - log.info("agent-notes up; polling Telegram. allowlist=%s", - sorted(cfg.telegram_allowed_chat_ids) or "(open)") + registry = TopicRegistry.load() + index = ProcessedIndex.load() + adapter = TelegramAdapter( + cfg.telegram_bot_token, cfg.telegram_allowed_chat_ids, registry=registry + ) + log.info("agent-notes up; polling Telegram. allowlist=%s topics=%s", + sorted(cfg.telegram_allowed_chat_ids) or "(open)", + list(registry.defs)) # Jobs run one at a time to avoid two NotebookLM/Playwright sessions colliding. try: async for job in adapter.poll(): - await handle_job(adapter, cfg, job) + await handle_job(adapter, cfg, registry, index, job) finally: await adapter.aclose() diff --git a/src/agentnotes/notebooklm_cli.py b/src/agentnotes/notebooklm_cli.py new file mode 100644 index 0000000..345f610 --- /dev/null +++ b/src/agentnotes/notebooklm_cli.py @@ -0,0 +1,55 @@ +"""Thin, deterministic wrapper over the notebooklm CLI for the bits Python drives +itself (no LLM needed): finding a notebook by title and listing its sources. + +Used by the "sweep existing notebook" flow (UC3): Python lists the topic notebook's +sources so it can dedup per-source and decide which still need a Notion page, then the +agent generates each report. Source-adding and report-writing still go through the +agent/skill — only the read-only listing is done here. +""" +from __future__ import annotations + +import json +import logging +import subprocess +import sys + +log = logging.getLogger("agentnotes.notebooklm_cli") + +DEFAULT_TIMEOUT_S = 120.0 + + +class NotebookLMError(Exception): + """A notebooklm CLI call failed, timed out, or returned unparseable output.""" + + +def _run_json(args: list[str], timeout: float) -> dict: + cmd = [sys.executable, "-m", "notebooklm", *args, "--json"] + try: + r = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout) + except subprocess.TimeoutExpired as e: + raise NotebookLMError(f"notebooklm {' '.join(args)} 逾時（{int(timeout)}s）") from e + except OSError as e: + raise NotebookLMError(f"無法執行 notebooklm CLI：{e}") from e + if r.returncode != 0: + raise NotebookLMError( + f"notebooklm {' '.join(args)} 失敗（rc={r.returncode}）：{r.stderr.strip()[-300:]}" + ) + try: + return json.loads(r.stdout) + except json.JSONDecodeError as e: + raise NotebookLMError(f"notebooklm {' '.join(args)} 回傳非 JSON") from e + + +def find_notebook_id(title: str, *, timeout: float = DEFAULT_TIMEOUT_S) -> str | None: + """Return the id of the notebook whose title EXACTLY equals `title`, or None.""" + data = _run_json(["list"], timeout) + for nb in data.get("notebooks", []): + if nb.get("title") == title: + return nb.get("id") + return None + + +def list_sources(notebook_id: str, *, timeout: float = DEFAULT_TIMEOUT_S) -> list[dict]: + """Return the notebook's sources (each: id, title, type, url, status, ...).""" + data = _run_json(["source", "list", "-n", notebook_id], timeout) + return data.get("sources", []) diff --git a/src/agentnotes/pipeline.py b/src/agentnotes/pipeline.py index 073df28..b9e3f2c 100644 --- a/src/agentnotes/pipeline.py +++ b/src/agentnotes/pipeline.py @@ -14,12 +14,15 @@ import logging import re +from collections.abc import Callable from dataclasses import dataclass from pathlib import Path +from . import notebooklm_cli as nlm from .agent import run_builder, run_evaluator -from .config import Config, JOBS_DIR +from .config import JOBS_DIR, Config from .contract import Contract +from .dedup import ProcessedIndex, file_digest, normalize_url from .evaluators import ( NOTION_EVALUATOR_PROMPT, PAGE_BODY_BEGIN, @@ -29,16 +32,24 @@ check_report_text, parse_verdict, ) +from .topics import TopicRegistry log = logging.getLogger("agentnotes.pipeline") -# All sources go into one shared NotebookLM notebook; each request adds a source to it -# and asks questions scoped to just that source (-s) to avoid cross-source contamination. -NOTEBOOK_NAME = "給 Agent Note" +# Each topic has its own shared NotebookLM notebook (see topics.json). Within a +# notebook, every request adds its source and asks questions scoped to just that +# source (-s) to avoid cross-source contamination. # Match notion.so / www.notion.so / app.notion.com page links. Stop at whitespace, # closing paren/bracket, or markdown emphasis chars so we don't swallow a trailing ** . _NOTION_URL_RE = re.compile(r"https://(?:www\.|app\.)?notion\.(?:so|com)/[^\s)\]*]+") +# A Notion database id echoed back by the builder after auto-creating a topic's DB: +# "database_id = 32-hex" or a dashed UUID. Anchored on the word "database" so a page +# id in the same reply can't be mistaken for the DB id. +_DB_ID_RE = re.compile( + r"database[_ ]?id[^0-9a-fA-F]{0,12}([0-9a-fA-F]{32}|[0-9a-fA-F]{8}-[0-9a-fA-F-]{27})", + re.IGNORECASE, +) def _extract_notion_url(text: str) -> str | None: @@ -49,6 +60,11 @@ def _extract_notion_url(text: str) -> str | None: return m.group(0).rstrip("*_.,;:)]>\"'") +def _extract_db_id(text: str) -> str | None: + m = _DB_ID_RE.search(text or "") + return m.group(1) if m else None + + def _split_verdict_body(text: str) -> tuple[str, str]: """Split a Gate-B evaluator reply into (verdict_part, page_body). @@ -69,22 +85,51 @@ class PipelineResult: ok: bool message: str # user-facing reply text notion_url: str | None = None + duplicate: bool = False # True == skipped because this (topic, url) was done before def _job_dir(job_id: str) -> Path: return JOBS_DIR / job_id -def _report_prompt(report_path: Path, url: str) -> str: +def _report_prompt(report_path: Path, notebook_name: str, *, url: str | None = None, + file_path: str | None = None, file_title: str | None = None, + source_id: str | None = None) -> str: + if source_id: + # The source is ALREADY in the notebook (UC3 sweep): never add it, just scope to it. + source_line = f"來源（已在「{notebook_name}」筆記本內）：{file_title or source_id}" + add_step = ( + f"2. 這個來源**已經在筆記本裡了，絕對不要再 `source add`**。它的 source id = {source_id}" + "（以下記為 SRC）。" + ) + ident = "source id" + elif file_path: + title_opt = f' --title "{file_title}"' if file_title else "" + source_line = f"來源檔案（本地路徑，已由機器人下載）：{file_path}" + add_step = ( + "2. **先**用 `notebooklm source list -n NB --json` 看這個檔案是不是已經是來源（比對檔名/標題）。\n" + f' - 若**尚未加入**：用 `notebooklm source add "{file_path}" --type file -n NB{title_opt}` ' + "把這個檔案加進筆記本。\n" + " - 若**已存在**同一個檔案：不要再 add，直接重用那筆既有來源，避免疊出重複來源。" + ) + ident = "檔名/標題" + else: + source_line = f"來源連結：{url}" + add_step = ( + "2. **先**用 `notebooklm source list -n NB --json` 看這個 URL 是不是已經是筆記本裡的來源。\n" + " - 若**已存在**（同一個 URL）：不要再 add，直接重用那筆既有來源，避免在筆記本裡疊出重複來源。\n" + f' - 若**不存在**：才用 `notebooklm source add "{url}" -n NB` 把這次的來源加進「{notebook_name}」。' + ) + ident = "URL/標題" return f"""請用 notebooklm skill 處理這個來源，產出一份「網誌式」繁體中文報告。 -來源連結：{url} +{source_line} -### 重要：所有來源集中在同一個 NotebookLM 筆記本「{NOTEBOOK_NAME}」（不要每次開新筆記本） -1. 先 `notebooklm list --json`，找出標題剛好等於「{NOTEBOOK_NAME}」的筆記本，取得其 id（記為 NB）。 - - 只有在**完全找不到**時，才用 `notebooklm create "{NOTEBOOK_NAME}"` 建立它（只建這一次，之後一律重用）。 -2. 用 `notebooklm source add "{url}" -n NB` 把這次的來源加進「{NOTEBOOK_NAME}」。 -3. 用 `notebooklm source list -n NB --json` 找出剛剛新增那筆來源的 source id（記為 SRC，對 URL/標題確認是這次這筆）。 +### 重要：這個 topic 的來源集中在同一個 NotebookLM 筆記本「{notebook_name}」（不要每次開新筆記本） +1. 先 `notebooklm list --json`，找出標題剛好等於「{notebook_name}」的筆記本，取得其 id（記為 NB）。 + - 只有在**完全找不到**時，才用 `notebooklm create "{notebook_name}"` 建立它（只建這一次，之後一律重用）。 +{add_step} +3. 用 `notebooklm source list -n NB --json` 找出這個來源對應的 source id（記為 SRC，對{ident}確認是這次這筆）。 若來源還在處理中，先 `notebooklm source wait -n NB` 等就緒。 4. **NotebookLM 預設會「勾選筆記本裡所有來源」，所以每一次提問都一定要加 `-s SRC` 把範圍鎖死在這次的來源** （例：`notebooklm ask -n NB -s SRC --json "..."`）。**只要有任何一次 ask 漏了 `-s`，就會把其他來源也讀進去、 @@ -107,19 +152,27 @@ def _report_prompt(report_path: Path, url: str) -> str: 完成後回我一行：報告字數與 ## 段落數。""" -def _notion_prompt(report_path: Path, db_id: str, url: str, title_hint: str) -> str: - db_clause = ( - f"把報告新增為這個資料庫的一頁：database_id = {db_id}。" - if db_id - else "若還沒有「NotebookLM 報告」資料庫，請在我授權的 parent page 下用 notion-create-database 建一個" - "（屬性：標題 Title、來源 URL、建立時間 Date），然後把報告新增為其中一頁。" - "建完後請在回覆中明確寫出 database_id。" - ) +def _notion_prompt(report_path: Path, db_id: str, db_title: str, parent_page_id: str, + source_ref: str, title_hint: str) -> str: + if db_id: + db_clause = f"把報告新增為這個資料庫的一頁：database_id = {db_id}。" + else: + where = ( + f"在這個 parent page（id = {parent_page_id}）底下" + if parent_page_id + else "在我授權的 parent page 底下" + ) + db_clause = ( + f"若還沒有標題為「{db_title}」的資料庫，請{where}用 notion-create-database 建一個" + f"，標題就用「{db_title}」（屬性：標題 Title、來源 URL、建立時間 Date），" + "然後把報告新增為其中一頁。" + "建完後請在回覆中明確寫出 database_id（格式：`database_id = `）。" + ) return f"""請先 Read 這個檔案的完整內容：{report_path} 然後用 Notion MCP 把它建成一頁 Notion 文章。{db_clause} - 頁面標題用報告的 # 標題（若無則用：{title_hint}）。 -- 「來源」屬性填：{url} +- 「來源」屬性填：{source_ref} - **頁面最前面的第一個區塊，要放一個 Notion 原生的「Table of contents（目錄）」block**， 讓讀者可點擊跳到各 `##` 段落、提升可讀性。這要用 Notion 真正的目錄區塊（會自動依標題產生）， 不是手打的連結清單。若不確定 Notion-flavored markdown 的目錄區塊語法， @@ -132,15 +185,71 @@ def _notion_prompt(report_path: Path, db_id: str, url: str, title_hint: str) -> 完成後**務必**在回覆最後附上該頁的完整 Notion 連結（https://www.notion.so/...）。""" -async def run_pipeline(job_id: str, url: str, cfg: Config, title_hint: str = "未命名報告" - ) -> PipelineResult: +@dataclass +class Source: + """One thing to turn into a report: a URL, an uploaded file, or a source already in + the notebook. Captures the three things that differ between them — the dedup key, the + human ref (Notion 來源 + logs), and the builder-A prompt — so the evidence-gated core + (_run_source) stays single-sourced.""" + key: str # dedup key in the per-topic ProcessedIndex + ref: str # shown as the Notion 來源 property and in logs + eval_ref: str # what the report evaluator is told the source is + _prompt: Callable[[Path, str], str] + + def report_prompt(self, report_path: Path, notebook_name: str) -> str: + return self._prompt(report_path, notebook_name) + + +def url_source(url: str) -> Source: + return Source(normalize_url(url), url, url, + lambda rp, nb: _report_prompt(rp, nb, url=url)) + + +def file_source(file_path: str, file_name: str | None) -> Source: + ref = file_name or file_path + return Source(file_digest(file_path), ref, ref, + lambda rp, nb: _report_prompt(rp, nb, file_path=file_path, file_title=file_name)) + + +def existing_source(src: dict) -> Source: + """A source already in the notebook (UC3). Keyed by its URL when it has one (so it + dedups against a URL job for the same link), else by its NotebookLM source id.""" + sid = src["id"] + title = src.get("title") or sid + url = src.get("url") + key = normalize_url(url) if url else f"nbsrc:{sid}" + return Source(key, title, title, + lambda rp, nb: _report_prompt(rp, nb, source_id=sid, file_title=title)) + + +async def _run_source(job_id: str, source: Source, cfg: Config, registry: TopicRegistry, + topic: str, title_hint: str, *, force: bool, + index: ProcessedIndex) -> PipelineResult: + """The evidence-gated core: build a report for ONE source, write it to Notion, and + report success only once BOTH gates pass. Shared by URL/file jobs and the sweep.""" + # ---------- Dedup: already done for this topic? (skip unless #force) ---------- + if not force: + prev = index.lookup_key(topic, source.key) + if prev: + log.info("[%s] duplicate; skipping (topic=%s)", job_id, topic) + return PipelineResult( + True, + "這篇在「{}」之前已經處理過了，直接給你上次的結果：\n{}\n（要重新產生請在訊息加 #force）".format( + topic, prev.get("notion_url", "") + ), + notion_url=prev.get("notion_url"), + duplicate=True, + ) + job_dir = _job_dir(job_id) contract = Contract.create(job_dir) report_path = job_dir / "report.md" + notebook_name = registry.notebook(topic) # ---------- Step A: build the report ---------- - log.info("[%s] builder A: notebooklm report", job_id) - await run_builder(_report_prompt(report_path, url)) + log.info("[%s] builder A: notebooklm report (notebook=%s, source=%s)", + job_id, notebook_name, source.ref) + await run_builder(source.report_prompt(report_path, notebook_name)) # ---------- Gate A: did a real report actually get written? ---------- if not report_path.exists(): @@ -159,7 +268,7 @@ async def run_pipeline(job_id: str, url: str, cfg: Config, title_hint: str = " verdict = parse_verdict( (await run_evaluator( - REPORT_EVALUATOR_PROMPT.format(report_path=report_path, source_url=url) + REPORT_EVALUATOR_PROMPT.format(report_path=report_path, source_url=source.eval_ref) )).text ) if not verdict.ok: @@ -172,11 +281,22 @@ async def run_pipeline(job_id: str, url: str, cfg: Config, title_hint: str = " ) # ---------- Step B: write to Notion ---------- - log.info("[%s] builder B: write to Notion", job_id) + db_id = registry.notion_db_id(topic) + log.info("[%s] builder B: write to Notion (topic=%s, db=%s)", + job_id, topic, db_id or "") b = await run_builder( - _notion_prompt(report_path, cfg.notion_database_id, url, title_hint), + _notion_prompt(report_path, db_id, registry.notion_db_title(topic), + cfg.notion_parent_page_id, source.ref, title_hint), with_notion=True, ) + # If this topic's DB was just auto-created, cache its id so future jobs reuse it + # instead of building a duplicate DB. + if not db_id: + new_db_id = _extract_db_id(b.text) + if new_db_id: + registry.set_notion_db_id(topic, new_db_id) + else: + log.warning("[%s] could not parse a new database_id from builder reply", job_id) notion_url = _extract_notion_url(b.text) if not notion_url: contract.fail_gate("notion_page", "builder 沒有回傳 Notion 頁面連結（可能沒建成功）") @@ -206,6 +326,9 @@ async def run_pipeline(job_id: str, url: str, cfg: Config, title_hint: str = " # ---------- Only now: success ---------- if not contract.all_pass(): return PipelineResult(False, _fail_msg(contract)) + # Record only after a fully-verified success, so a failed/retried job is never + # mistaken for "already done". + index.record_key(topic, source.key, notion_url, source.ref) return PipelineResult( True, f"✅ 完成！中文分段報告已寫進 Notion：\n{notion_url}", @@ -213,6 +336,72 @@ async def run_pipeline(job_id: str, url: str, cfg: Config, title_hint: str = " ) +async def run_pipeline(job_id: str, url: str, cfg: Config, registry: TopicRegistry, + topic: str, title_hint: str = "未命名報告", *, + force: bool = False, index: ProcessedIndex | None = None, + file_path: str | None = None, file_name: str | None = None + ) -> PipelineResult: + """Process ONE new source — a URL, or an uploaded file — end-to-end.""" + index = index if index is not None else ProcessedIndex.load() + source = file_source(file_path, file_name) if file_path else url_source(url) + return await _run_source(job_id, source, cfg, registry, topic, title_hint, + force=force, index=index) + + +async def run_sweep(job_id: str, cfg: Config, registry: TopicRegistry, + index: ProcessedIndex, topic: str, *, force: bool = False + ) -> PipelineResult: + """UC3: process every not-yet-exported source already in this topic's notebook. + + Python lists the sources deterministically (no LLM), then runs the same evidence-gated + core per source; each source dedups on its own key so the sweep is re-runnable. + """ + notebook_name = registry.notebook(topic) + try: + nb_id = nlm.find_notebook_id(notebook_name) + except nlm.NotebookLMError as e: + return PipelineResult(False, f"⚠️ 無法讀取 NotebookLM：{e}") + if not nb_id: + return PipelineResult( + False, f"⚠️ 找不到筆記本「{notebook_name}」——這個 topic 目前還沒有任何來源。") + try: + sources = nlm.list_sources(nb_id) + except nlm.NotebookLMError as e: + return PipelineResult(False, f"⚠️ 無法列出「{notebook_name}」的來源：{e}") + + # Only sources that are ready AND have an id can be processed; anything else + # (still processing, errored, or malformed) is skipped, not silently mis-handled. + ready = [s for s in sources if s.get("status") == "ready" and s.get("id")] + not_ready = len(sources) - len(ready) + new: list[tuple[str, str]] = [] + skipped = 0 + failed: list[tuple[str, str]] = [] + + for i, s in enumerate(ready, start=1): + source = existing_source(s) + title = s.get("title") or source.key + res = await _run_source(f"{job_id}-{i}", source, cfg, registry, topic, + title_hint=title, force=force, index=index) + if res.duplicate: + skipped += 1 + elif res.ok: + new.append((title, res.notion_url or "")) + else: + failed.append((title, res.message)) + + lines = [f"📚 整理「{topic}」既有來源：共 {len(sources)} 筆"] + if not_ready: + lines.append(f"（{not_ready} 筆尚未就緒（處理中/錯誤），已略過）") + lines.append(f"新產 {len(new)}、跳過 {skipped}、失敗 {len(failed)}") + for t, u in new: + lines.append(f"✅ {t}\n{u}") + for t, m in failed: + lines.append(f"⚠️ {t}：{(m.splitlines() or [''])[0]}") + if not new and not failed: + lines.append("（沒有新的要做——可能都匯出過了，或筆記本是空的）") + return PipelineResult(ok=(not failed), message="\n".join(lines)) + + def _fail_msg(contract: Contract) -> str: reason = contract.first_failure_note() or "未知原因" return f"⚠️ 沒有成功（已停在未完成狀態，不會回你假連結）。\n原因：{reason}" diff --git a/src/agentnotes/topics.py b/src/agentnotes/topics.py new file mode 100644 index 0000000..92845ad --- /dev/null +++ b/src/agentnotes/topics.py @@ -0,0 +1,126 @@ +"""Topic registry: routes each job to a NotebookLM notebook + a Notion database. + +Two layers, deliberately split so the committed definitions never churn with the +per-deployment cache: + +* ``topics.json`` (committed) — definitions: aliases + notebook name + Notion DB + title for each topic. Editing this is the ONLY manual step to add a topic. +* ``.state/topic_dbs.json`` (gitignored) — cache of ``topic -> notion_db_id``, + filled in the first time a topic's Notion DB is auto-created so we never build a + duplicate DB on later jobs. + +A topic is chosen by an inline ``#tag`` in the incoming message; the tag is matched +against the topic key and its aliases after normalization (lowercase, spaces / +hyphens / underscores stripped), so ``#ai-agent`` / ``#aiagent`` / ``#ai`` all +resolve. Unknown / missing / conflicting tags are rejected upstream (fail-closed), +which keeps the RAG corpora clean — a typo can't silently spawn a junk notebook. +""" +from __future__ import annotations + +import json +import logging +import re +from dataclasses import dataclass, field +from pathlib import Path + +from .config import ROOT, STATE_DIR + +log = logging.getLogger("agentnotes.topics") + +TOPICS_FILE = ROOT / "topics.json" +TOPIC_DB_CACHE = STATE_DIR / "topic_dbs.json" + +_NORM_RE = re.compile(r"[\s_\-]+") +# Inline topic tags: "#ai", "#ai-agent", "#博士". Stop at whitespace or another '#'. +_TAG_RE = re.compile(r"#([^\s#]+)") + + +def normalize(s: str) -> str: + return _NORM_RE.sub("", (s or "").strip().lower()) + + +def extract_tags(text: str) -> list[str]: + """Return the raw tag tokens (without '#') found in a message, in order.""" + return _TAG_RE.findall(text or "") + + +@dataclass +class TopicRegistry: + defs: dict # topic_key -> {"aliases": [...], "notebook": str, "notion_db_title": str} + cache: dict # topic_key -> notion_db_id + cache_path: Path | None = None # None == in-memory only (tests) + _alias_index: dict = field(default_factory=dict) # normalized alias/key -> topic_key + + def __post_init__(self) -> None: + index: dict[str, str] = {} + for key, spec in self.defs.items(): + for token in [key, *(spec.get("aliases") or [])]: + norm = normalize(token) + if norm: + index[norm] = key + self._alias_index = index + + @classmethod + def load(cls, *, defs_path: Path = TOPICS_FILE, cache_path: Path = TOPIC_DB_CACHE + ) -> TopicRegistry: + defs = json.loads(defs_path.read_text(encoding="utf-8")) + cache: dict = {} + if cache_path.exists(): + try: + cache = json.loads(cache_path.read_text(encoding="utf-8")) + except (json.JSONDecodeError, OSError) as e: + log.warning("could not read topic db cache %s: %s", cache_path, e) + return cls(defs=defs, cache=cache, cache_path=cache_path) + + # ---- resolution ---- + def resolve(self, tag: str) -> str | None: + """Resolve one tag (alias or key) to a canonical topic key, or None.""" + return self._alias_index.get(normalize(tag)) + + def resolve_message(self, text: str) -> tuple[str | None, str | None]: + """Resolve the topic from a message's inline tags. + + Returns ``(topic, error)``: exactly one of the two is set. ``error`` is a + short reason ("missing" / "ambiguous") so the caller can pick the reply. + """ + topics = {t for t in (self.resolve(tag) for tag in extract_tags(text)) if t} + if len(topics) == 1: + return next(iter(topics)), None + if not topics: + return None, "missing" + return None, "ambiguous" + + # ---- per-topic accessors ---- + def notebook(self, topic: str) -> str: + return self.defs[topic]["notebook"] + + def notion_db_title(self, topic: str) -> str: + return self.defs[topic]["notion_db_title"] + + def notion_db_id(self, topic: str) -> str: + return (self.cache.get(topic) or "").strip() + + def set_notion_db_id(self, topic: str, db_id: str) -> None: + """Cache a newly-created DB id so future jobs reuse it (no duplicate DBs).""" + db_id = (db_id or "").strip() + if not db_id or self.cache.get(topic) == db_id: + return + self.cache[topic] = db_id + if self.cache_path is None: + return + try: + self.cache_path.parent.mkdir(parents=True, exist_ok=True) + self.cache_path.write_text( + json.dumps(self.cache, ensure_ascii=False, indent=2), encoding="utf-8" + ) + except OSError as e: + log.warning("could not persist topic db cache: %s", e) + + # ---- user-facing help ---- + def menu_text(self) -> str: + """A short '請標一個 topic' menu listing each topic's primary tag.""" + lines = [f"- #{key}（{self.notion_db_title(key)}）" for key in self.defs] + return ( + "請在訊息裡用 #標籤 指定要放哪個 topic，例如 `#ai 你的連結`。\n" + "目前可用的 topic：\n" + "\n".join(lines) + ) diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..1ced4ec --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,200 @@ +"""Shared pytest fixtures — the reusable backbone for the test suite. + +Everything offline lives here so individual test files stay short and a new test just +asks for the fixtures it needs: + +* ``cfg`` / ``registry`` / ``index`` — in-memory config, topic registry, dedup index. +* ``good_report`` — a report that passes the deterministic gate (>=400 CJK, >=3 ``##``). +* ``fake_agents`` — installs fake builder/evaluator into the pipeline and returns a + controller (``AgentControl``) to steer gate outcomes and read the builder call count. +* ``poll_once`` — drives ``TelegramAdapter.poll()`` over a single crafted update with no + network, returning ``(jobs, calls)``. +* ``make_update`` / ``make_message`` — tiny Telegram update/message builders. +""" +from __future__ import annotations + +import asyncio +import re + +import pytest + +from agentnotes import pipeline as P +from agentnotes.agent import AgentRun +from agentnotes.config import Config +from agentnotes.dedup import ProcessedIndex +from agentnotes.evaluators import PAGE_BODY_BEGIN, PAGE_BODY_END +from agentnotes.ingest.telegram import TelegramAdapter +from agentnotes.topics import TopicRegistry + +# A report long enough (>=400 CJK) with >=3 "##" headers to pass check_report_text. +GOOD_REPORT = ( + "# 一份合格的測試報告\n\n" + + ("這是一段足夠長的繁體中文內容，用於通過確定性結構檢查。" * 15) + + "\n\n## 第一段\n" + ("內容說明。" * 30) + + "\n\n## 第二段\n" + ("內容說明。" * 30) + + "\n\n## 第三段\n" + ("內容說明。" * 30) + "\n" +) + + +@pytest.fixture +def good_report() -> str: + return GOOD_REPORT + + +@pytest.fixture +def cfg() -> Config: + return Config( + telegram_bot_token="x", + telegram_allowed_chat_ids=frozenset(), + notion_parent_page_id="", + report_min_cjk_chars=400, + report_min_headers=3, + ) + + +def make_registry(*, cache: dict | None = None) -> TopicRegistry: + """Two-topic registry (phd + ai-agent). Pass cache to pre-seed Notion db ids.""" + return TopicRegistry( + defs={ + "phd": {"aliases": ["phd"], "notebook": "(AgentNote) PhD study", + "notion_db_title": "NotebookLM 報告 - PhD Study"}, + "ai-agent": {"aliases": ["ai"], "notebook": "(AgentNote) AI agent", + "notion_db_title": "NotebookLM 報告 - AI agent"}, + }, + cache=dict(cache or {"phd": "db_phd", "ai-agent": "db_ai"}), + cache_path=None, + ) + + +@pytest.fixture +def registry() -> TopicRegistry: + return make_registry() + + +@pytest.fixture +def index() -> ProcessedIndex: + return ProcessedIndex(base_dir=None) # in-memory, never touches real .state + + +class AgentControl: + """Knobs for the fake builder/evaluator installed by the ``fake_agents`` fixture. + + Defaults make a clean run succeed. Tweak before calling the pipeline: + * ``gate_a_ok`` / ``gate_b_ok`` — flip a gate's verdict. + * ``gate_a_text`` / ``gate_b_text`` — return this exact evaluator reply (full + control, e.g. to test verdict parsing or page-body markers). + * ``fail_marker`` — if this string appears in the gate-A evaluator prompt, that + one source fails gate A (used to fail a single source mid-sweep). + * ``write_report`` — whether builder A actually writes report.md. + * ``builder_calls`` — read-only count of builder invocations. + """ + + def __init__(self) -> None: + self.gate_a_ok = True + self.gate_b_ok = True + self.gate_a_text: str | None = None + self.gate_b_text: str | None = None + self.fail_marker: str | None = None + self.write_report = True + self.builder_calls = 0 + + +@pytest.fixture +def fake_agents(monkeypatch) -> AgentControl: + ctl = AgentControl() + + async def builder(prompt, **kw): + ctl.builder_calls += 1 + if "Notion MCP" in prompt: # builder B: pretend a Notion page was created + return AgentRun("已建立頁面：https://www.notion.so/deadbeefcafe", None, False) + if ctl.write_report: # builder A: write the report to the path named in the prompt + m = re.search(r"(/\S+report\.md)", prompt) + if m: + from pathlib import Path + p = Path(m.group(1)) + p.parent.mkdir(parents=True, exist_ok=True) + p.write_text(GOOD_REPORT, encoding="utf-8") + return AgentRun("報告 OK", None, False) + + async def evaluator(prompt, **kw): + if "notion-fetch" in prompt or PAGE_BODY_BEGIN in prompt: # gate B + if ctl.gate_b_text is not None: + return AgentRun(ctl.gate_b_text, None, False) + if ctl.gate_b_ok: + return AgentRun(f"PASS：完整\n{PAGE_BODY_BEGIN}\n{GOOD_REPORT}\n{PAGE_BODY_END}", + None, False) + return AgentRun(f"FAIL：頁面空白\n{PAGE_BODY_BEGIN}\n{PAGE_BODY_END}", None, False) + # gate A + if ctl.gate_a_text is not None: + return AgentRun(ctl.gate_a_text, None, False) + if ctl.fail_marker and ctl.fail_marker in prompt: + return AgentRun("FAIL：來源讀不到", None, False) + return AgentRun("PASS：報告 OK" if ctl.gate_a_ok else "FAIL：不合格", None, False) + + monkeypatch.setattr(P, "run_builder", builder) + monkeypatch.setattr(P, "run_evaluator", evaluator) + return ctl + + +# --------- Telegram poll() driver (no network) --------- + +class _StopDrive(Exception): + """poll()'s second _get_updates raises this to end a single-batch drive.""" + + +def make_message(**fields) -> dict: + base = {"message_id": 10, "chat": {"id": 123}} + base.update(fields) + return base + + +def make_update(update_id: int = 7, **message_fields) -> dict: + return {"update_id": update_id, "message": make_message(**message_fields)} + + +@pytest.fixture +def poll_once(): + """Drive poll() over ONE crafted update. Returns (jobs, calls). + + calls = {"replies": [...], "commits": [...], "downloads": int}. Pass allowed= to set + the chat allowlist and download= to override the attachment downloader. + """ + def _run(registry, update, *, allowed=frozenset(), download=None): + adapter = TelegramAdapter("x:y", allowed, registry=registry) + calls = {"replies": [], "commits": [], "downloads": 0} + batches = [[update]] + + async def fake_get_updates(): + if batches: + return batches.pop(0) + raise _StopDrive() + + async def fake_send_reply(chat_id, text, reply_to=None): + calls["replies"].append(text) + + async def fake_download(document): + calls["downloads"] += 1 + name = document.get("file_name", "f") + return (f"/tmp/dl/{name}", name) + + adapter._get_updates = fake_get_updates + adapter.send_reply = fake_send_reply + adapter._commit_offset = lambda off: calls["commits"].append(off) + adapter._download_document = download or fake_download + + async def drive(): + jobs = [] + gen = adapter.poll() + try: + while True: + jobs.append(await gen.__anext__()) + except (_StopDrive, StopAsyncIteration): + pass + finally: + await gen.aclose() + await adapter.aclose() + return jobs + + return asyncio.run(drive()), calls + + return _run diff --git a/tests/test_agent.py b/tests/test_agent.py new file mode 100644 index 0000000..9baf1d3 --- /dev/null +++ b/tests/test_agent.py @@ -0,0 +1,56 @@ +"""agent.py: SDK wrapper behavior — result aggregation, the auth-failure mapping, and +the wall-clock timeout. `query` is monkeypatched so nothing spawns a real claude CLI.""" +import asyncio +from types import SimpleNamespace + +import pytest + +from agentnotes import agent as A + + +def test_not_logged_in_is_mapped_to_actionable_error(monkeypatch): + async def boom_query(**kw): + # the exact cryptic exception the SDK raises when the spawned claude isn't logged in + raise Exception("Claude Code returned an error result: success") + yield # unreachable, but makes this function an async generator + + monkeypatch.setattr(A, "query", boom_query) + with pytest.raises(A.AgentError) as ei: + asyncio.run(A.run_builder("hi")) + msg = str(ei.value) + assert "claude auth login" in msg and "未登入" in msg + assert "returned an error result" in msg # original cause preserved + + +def test_unrelated_error_is_not_swallowed(monkeypatch): + async def boom_query(**kw): + raise RuntimeError("disk on fire") + yield + + monkeypatch.setattr(A, "query", boom_query) + with pytest.raises(RuntimeError): # not reclassified as AgentError + asyncio.run(A.run_builder("hi")) + + +def test_aggregates_result_text_and_cost(monkeypatch): + async def ok_query(**kw): + yield SimpleNamespace(content=[SimpleNamespace(text="部分文字")], + result=None, total_cost_usd=None, is_error=False) + yield SimpleNamespace(content=None, result="最終答案", + total_cost_usd=0.02, is_error=False) + + monkeypatch.setattr(A, "query", ok_query) + run = asyncio.run(A.run_builder("hi")) + assert run.text == "最終答案" # prefers the ResultMessage .result + assert run.cost_usd == 0.02 and run.is_error is False + + +def test_timeout_raises_step_timeout(monkeypatch): + async def slow_query(**kw): + await asyncio.sleep(5) + yield SimpleNamespace(content=None, result="x", total_cost_usd=None, is_error=False) + + monkeypatch.setattr(A, "query", slow_query) + monkeypatch.setenv("JOB_STEP_TIMEOUT_S", "0.1") + with pytest.raises(A.StepTimeout): + asyncio.run(A.run_builder("hi")) diff --git a/tests/test_attachments.py b/tests/test_attachments.py new file mode 100644 index 0000000..d86ce20 --- /dev/null +++ b/tests/test_attachments.py @@ -0,0 +1,76 @@ +"""UC2 attachments: the type gate, the report-prompt file branch, and poll() routing of +document messages (no network).""" +import pytest + +from agentnotes.ingest.telegram import _accepted_doc +from agentnotes.pipeline import _report_prompt +from conftest import make_update + + +@pytest.mark.parametrize("document, accepted", [ + ({"mime_type": "application/pdf"}, True), + ({"mime_type": "application/vnd.openxmlformats-officedocument.wordprocessingml.document"}, True), + ({"mime_type": "application/octet-stream", "file_name": "paper.PDF"}, True), + ({"file_name": "notes.txt"}, True), + ({"file_name": "readme.md"}, True), + ({"mime_type": "image/png", "file_name": "x.png"}, False), + ({"mime_type": "application/zip", "file_name": "a.zip"}, False), + ({}, False), +]) +def test_accepted_doc(document, accepted): + assert _accepted_doc(document) is accepted + + +def test_report_prompt_file_branch(): + fp = _report_prompt(__import__("pathlib").Path("/tmp/report.md"), "(AgentNote) PhD study", + file_path="/tmp/dl/thesis.pdf", file_title="thesis.pdf") + assert "--type file" in fp + assert "/tmp/dl/thesis.pdf" in fp + assert '--title "thesis.pdf"' in fp + assert "-s SRC" in fp # scoping rule preserved + assert "None" not in fp + + +def test_report_prompt_url_branch(): + up = _report_prompt(__import__("pathlib").Path("/tmp/report.md"), "(AgentNote) AI agent", + url="https://youtu.be/abc") + assert "https://youtu.be/abc" in up + assert "--type file" not in up + + +# ---- poll() routing of document messages ---- + +def _doc(**over): + d = {"file_id": "F", "file_unique_id": "U", "file_name": "thesis.pdf", + "mime_type": "application/pdf", "file_size": 1000} + d.update(over) + return d + + +def test_accepted_pdf_yields_one_file_job(registry, poll_once): + jobs, calls = poll_once(registry, make_update(caption="#phd", document=_doc())) + assert len(jobs) == 1 + j = jobs[0] + assert j.url == "" and j.file_path and j.file_name == "thesis.pdf" and j.topic == "phd" + assert calls["downloads"] == 1 + + +def test_captionless_pdf_asks_for_topic_not_a_link(registry, poll_once): + jobs, calls = poll_once(registry, make_update(document=_doc(file_size=None))) + assert jobs == [] + assert calls["downloads"] == 0 # topic checked before download + assert calls["replies"] and "topic" in calls["replies"][0] and "http/https" not in calls["replies"][0] + + +def test_unsupported_type_rejected(registry, poll_once): + jobs, calls = poll_once(registry, make_update( + caption="#phd", document=_doc(file_name="pic.png", mime_type="image/png"))) + assert jobs == [] and calls["downloads"] == 0 + assert calls["replies"] and "看不懂" in calls["replies"][0] + + +def test_oversized_file_rejected_without_download(registry, poll_once): + jobs, calls = poll_once(registry, make_update( + caption="#phd", document=_doc(file_size=25 * 1024 * 1024))) + assert jobs == [] and calls["downloads"] == 0 + assert calls["replies"] and "太大" in calls["replies"][0] diff --git a/tests/test_config.py b/tests/test_config.py new file mode 100644 index 0000000..438d0c3 --- /dev/null +++ b/tests/test_config.py @@ -0,0 +1,51 @@ +"""Config + .env parsing.""" +import os + +from agentnotes.config import Config, _load_dotenv + + +def test_load_dotenv_parses_and_never_overrides(tmp_path, monkeypatch): + for k in ("AN_A", "AN_B", "AN_C", "AN_KEEP"): + monkeypatch.delenv(k, raising=False) + monkeypatch.setenv("AN_KEEP", "keep-me") # pre-existing var must NOT be overridden + p = tmp_path / ".env" + p.write_text( + "# a comment line\n" + "\n" + "AN_A=hello\n" + 'AN_B="quoted value"\n' + "AN_C='single quoted'\n" + "JUST_A_LINE_NO_EQUALS\n" + "AN_KEEP=should-not-win\n", + encoding="utf-8", + ) + _load_dotenv(p) + assert os.environ["AN_A"] == "hello" + assert os.environ["AN_B"] == "quoted value" + assert os.environ["AN_C"] == "single quoted" + assert "JUST_A_LINE_NO_EQUALS" not in os.environ + assert os.environ["AN_KEEP"] == "keep-me" + + +def test_load_dotenv_missing_file_is_noop(tmp_path): + _load_dotenv(tmp_path / "nope.env") # must not raise + + +def test_config_allowlist_parsing(monkeypatch): + monkeypatch.setenv("TELEGRAM_BOT_TOKEN", "tok") + monkeypatch.setenv("TELEGRAM_ALLOWED_CHAT_IDS", "123, -456 , notanint, 78.9, 789, ") + monkeypatch.setenv("NOTION_PARENT_PAGE_ID", "pp") + monkeypatch.setenv("REPORT_MIN_CJK_CHARS", "250") + monkeypatch.setenv("REPORT_MIN_HEADERS", "5") + cfg = Config.load() + assert cfg.telegram_bot_token == "tok" + assert cfg.telegram_allowed_chat_ids == frozenset({123, -456, 789}) + assert cfg.notion_parent_page_id == "pp" + assert cfg.report_min_cjk_chars == 250 + assert cfg.report_min_headers == 5 + + +def test_config_empty_allowlist_is_open(monkeypatch): + monkeypatch.setenv("TELEGRAM_BOT_TOKEN", "tok") + monkeypatch.setenv("TELEGRAM_ALLOWED_CHAT_IDS", "") + assert Config.load().telegram_allowed_chat_ids == frozenset() diff --git a/tests/test_contract.py b/tests/test_contract.py new file mode 100644 index 0000000..f262f4f --- /dev/null +++ b/tests/test_contract.py @@ -0,0 +1,47 @@ +"""Default-FAIL contract invariants — the structural anti-silent-failure guard.""" +import pytest + +from agentnotes.contract import GATES, Contract + + +def test_fresh_contract_defaults_to_fail(tmp_path): + c = Contract.create(tmp_path / "job") + assert (tmp_path / "job" / "contract.json").exists() + assert not c.all_pass() + assert not any(c.passed(g) for g in GATES) + assert c.first_failure_note() is not None + + +def test_unknown_gate_raises(tmp_path): + c = Contract.create(tmp_path / "job") + with pytest.raises(KeyError): + c.pass_gate("does_not_exist", evidence="x") + with pytest.raises(KeyError): + c.fail_gate("does_not_exist", "x") + + +def test_pass_gate_requires_evidence(tmp_path): + c = Contract.create(tmp_path / "job") + with pytest.raises(ValueError): + c.pass_gate(GATES[0], evidence="") + + +def test_fail_then_pass_and_persist(tmp_path): + job = tmp_path / "job" + c = Contract.create(job) + + c.fail_gate(GATES[0], "boom reason") + assert not c.passed(GATES[0]) + assert c.first_failure_note() == "boom reason" + + c.pass_gate(GATES[0], evidence="ev0", note="note0") + assert c.passed(GATES[0]) + assert not c.all_pass() # second gate still open + + c.pass_gate(GATES[1], evidence="ev1") + assert c.all_pass() + assert c.first_failure_note() is None + + reloaded = Contract.load(job) + assert reloaded.all_pass() + assert reloaded.state[GATES[0]]["note"] == "note0" diff --git a/tests/test_dedup.py b/tests/test_dedup.py new file mode 100644 index 0000000..e424917 --- /dev/null +++ b/tests/test_dedup.py @@ -0,0 +1,116 @@ +"""Duplicate prevention: URL normalization, content hashing, the index, and the +pipeline-level skip/record-on-success/#force behavior.""" +import asyncio + +import pytest + +from agentnotes import pipeline as P +from agentnotes.dedup import ProcessedIndex, file_digest, normalize_url + + +@pytest.mark.parametrize("a, b, same", [ + ("https://youtu.be/abc123", "https://www.youtube.com/watch?v=abc123", True), + ("https://ex.com/a?utm_source=x&id=7", "https://ex.com/a?id=7", True), + ("https://youtu.be/abc123?si=track", "https://youtu.be/abc123", True), + ("https://www.ex.com/a/#sec", "https://ex.com/a", True), + ("http://ex.com/a", "https://ex.com/a", True), + ("https://youtu.be/aaa", "https://youtu.be/bbb", False), + ("https://ex.com/a?id=7", "https://ex.com/a?id=8", False), +]) +def test_normalize_url(a, b, same): + assert (normalize_url(a) == normalize_url(b)) is same + + +def test_file_digest_is_content_keyed(tmp_path): + p1 = tmp_path / "paper.pdf" + p1.write_bytes(b"%PDF-1.4 fake") + p2 = tmp_path / "renamed.pdf" + p2.write_bytes(b"%PDF-1.4 fake") # same bytes, different name + p3 = tmp_path / "other.pdf" + p3.write_bytes(b"%PDF-1.4 different") + assert file_digest(str(p1)) == file_digest(str(p2)) + assert file_digest(str(p1)) != file_digest(str(p3)) + + +def test_index_url_roundtrip_with_normalization(): + idx = ProcessedIndex(base_dir=None) + assert idx.lookup("phd", "https://youtu.be/x") is None + idx.record("phd", "https://youtu.be/x", "https://notion.so/p1") + assert idx.lookup("phd", "https://www.youtube.com/watch?v=x")["notion_url"] == "https://notion.so/p1" + assert idx.lookup("ai-agent", "https://youtu.be/x") is None # per-topic isolation + + +def test_index_persists_to_disk(tmp_path): + i1 = ProcessedIndex(base_dir=tmp_path) + i1.record("phd", "https://ex.com/paper?utm_source=z", "https://notion.so/p2") + assert (tmp_path / "phd.json").exists() + i2 = ProcessedIndex(base_dir=tmp_path) # fresh instance reads disk + assert i2.lookup("phd", "https://ex.com/paper")["notion_url"] == "https://notion.so/p2" + + +def test_file_index_roundtrip(tmp_path): + p = tmp_path / "a.pdf" + p.write_bytes(b"bytes") + idx = ProcessedIndex(base_dir=None) + assert idx.lookup_file("phd", str(p)) is None + idx.record_file("phd", str(p), "https://notion.so/fp") + assert idx.lookup_file("phd", str(p))["notion_url"] == "https://notion.so/fp" + + +def test_set_db_id_empty_noop(): + idx = ProcessedIndex(base_dir=None) + idx.record("phd", "https://x", "u") + before = idx.lookup("phd", "https://x") + idx.record("phd", "https://x", "u") # idempotent + assert idx.lookup("phd", "https://x") == before + + +# ---- pipeline-level dedup ---- + +def test_url_job_records_then_skips(cfg, registry, fake_agents): + idx = ProcessedIndex(base_dir=None) + url = "https://youtu.be/dedup1" + r1 = asyncio.run(P.run_pipeline("t-d-1", url, cfg, registry, "phd", index=idx)) + assert r1.ok and not r1.duplicate and r1.notion_url + assert fake_agents.builder_calls >= 2 + assert idx.lookup("phd", url) is not None + + fake_agents.builder_calls = 0 + r2 = asyncio.run(P.run_pipeline("t-d-2", url, cfg, registry, "phd", index=idx)) + assert r2.duplicate and r2.ok and r2.notion_url == r1.notion_url + assert fake_agents.builder_calls == 0 # builder never invoked on a duplicate + + +def test_force_bypasses_skip(cfg, registry, fake_agents): + idx = ProcessedIndex(base_dir=None) + url = "https://youtu.be/dedup2" + asyncio.run(P.run_pipeline("t-d-3", url, cfg, registry, "phd", index=idx)) + fake_agents.builder_calls = 0 + r = asyncio.run(P.run_pipeline("t-d-4", url, cfg, registry, "phd", force=True, index=idx)) + assert not r.duplicate and r.ok and fake_agents.builder_calls >= 2 + + +def test_file_job_dedups_on_content(cfg, registry, fake_agents, tmp_path): + fp = tmp_path / "thesis.pdf" + fp.write_bytes(b"%PDF thesis") + idx = ProcessedIndex(base_dir=None) + r1 = asyncio.run(P.run_pipeline("t-f-1", "", cfg, registry, "phd", + index=idx, file_path=str(fp), file_name="thesis.pdf")) + assert r1.ok and idx.lookup_file("phd", str(fp)) is not None + fake_agents.builder_calls = 0 + r2 = asyncio.run(P.run_pipeline("t-f-2", "", cfg, registry, "phd", + index=idx, file_path=str(fp), file_name="thesis.pdf")) + assert r2.duplicate and fake_agents.builder_calls == 0 + + +def test_failed_job_is_not_recorded_and_retries(cfg, registry, fake_agents): + idx = ProcessedIndex(base_dir=None) + url = "https://youtu.be/willfail" + fake_agents.gate_b_ok = False # Notion page comes back blank -> gate B fails + r = asyncio.run(P.run_pipeline("t-fail-1", url, cfg, registry, "phd", index=idx)) + assert not r.ok and not r.duplicate + assert idx.lookup("phd", url) is None # a failure must stay retriable + + fake_agents.gate_b_ok = True + r2 = asyncio.run(P.run_pipeline("t-fail-2", url, cfg, registry, "phd", index=idx)) + assert r2.ok and not r2.duplicate diff --git a/tests/test_evaluators.py b/tests/test_evaluators.py new file mode 100644 index 0000000..8bee1ba --- /dev/null +++ b/tests/test_evaluators.py @@ -0,0 +1,78 @@ +"""Deterministic evaluator edge cases — the boundaries test_evidence doesn't pin.""" +import pytest + +from agentnotes.evaluators import ( + check_notion_fetch, + check_report_text, + count_cjk, + count_headers, + find_failure_marker, + parse_verdict, +) + + +def _long_report(extra: str = "") -> str: + body = "這是一段足夠長的繁體中文內容，用來確保字數遠超過門檻。" * 15 + return (f"# 一份合格的報告{extra}\n\n{body}\n\n" + f"## 第一段\n{'內容說明。' * 30}\n\n" + f"## 第二段\n{'內容說明。' * 30}\n\n" + f"## 第三段\n{'內容說明。' * 30}\n") + + +def test_long_report_mentioning_error_word_still_passes(): + assert check_report_text(_long_report() + "\n影片中提到一個常見的錯誤做法。").ok + + +def test_short_text_with_login_marker_fails(): + assert not check_report_text("請登入 Google 帳號後再試一次。").ok + + +@pytest.mark.parametrize("kwargs", [{"min_cjk": 100000}, {"min_headers": 99}]) +def test_thresholds_fail_a_good_report(kwargs): + assert not check_report_text(_long_report(), **kwargs).ok + + +def test_good_report_passes_defaults(): + assert check_report_text(_long_report()).ok + + +@pytest.mark.parametrize("marker", ["處理失敗", "無法產出", "未登入", "錯誤報告"]) +def test_failure_title_heuristic(marker): + t = f"# {marker}：說明\n\n{'內容。' * 300}\n## a\n內文\n## b\n內文\n## c\n內文\n" + assert not check_report_text(t).ok + + +def test_count_cjk_counts_only_han(): + assert count_cjk("abc 中文 x！") == 2 + + +def test_count_headers(): + assert count_headers("# a\n## b\nplain\n### c") == 3 + + +def test_find_failure_marker_is_case_insensitive(): + assert find_failure_marker("Please Sign In") == "sign in" + + +def test_find_failure_marker_none_on_clean_text(): + assert find_failure_marker("乾淨的內容") is None + + +def test_notion_good_body_passes(): + assert check_notion_fetch(_long_report()).ok + + +def test_notion_high_threshold_fails(): + assert not check_notion_fetch(_long_report(), min_cjk=100000).ok + + +def test_notion_blank_fails(): + assert not check_notion_fetch(" \n ").ok + + +def test_parse_verdict_leading_blank_lines(): + assert parse_verdict("\n\n PASS：可以").ok + + +def test_parse_verdict_none_fails_closed(): + assert not parse_verdict(None).ok diff --git a/tests/test_evidence.py b/tests/test_evidence.py index 3f9d06c..24f96c0 100644 --- a/tests/test_evidence.py +++ b/tests/test_evidence.py @@ -1,125 +1,71 @@ -"""Offline verification of the anti-silent-failure spine. +"""The anti-silent-failure spine: deterministic gate + Default-FAIL contract.""" +import pytest -No SDK, no network. Proves the deterministic gate PASSES a real report and FAILS -the real-world silent-failure modes, and that the contract refuses to claim -success unless a gate was actually flipped with evidence. +from agentnotes.contract import Contract +from agentnotes.evaluators import check_notion_fetch, check_report_text, parse_verdict -Run: python tests/test_evidence.py -""" -import sys -import tempfile -from pathlib import Path - -sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) +LOGIN_GATE = "Sign in with Google to continue.\n請登入 Google 帳號以使用 NotebookLM。" +EMPTY = " \n \n" +THIN = "# 標題\n太短了。" -from agentnotes.contract import Contract # noqa: E402 -from agentnotes.evaluators import ( # noqa: E402 - check_notion_fetch, - check_report_text, - parse_verdict, +# Real production failure: a long, well-formed Chinese failure report with ## headers — +# passes length/header checks; the TITLE is what gives it away. +FAILURE_REPORT = ( + "# NotebookLM 處理失敗報告：無法產出來源的網誌式內容\n\n" + + ("本報告如實記錄處理來源失敗的原因，依任務要求不編造內容，刻意寫得夠長以證明" + "光靠長度檢查仍可能放行一份其實是失敗的報告。" * 6) + + "\n\n## 執行過程\n環境中找不到有效登入工作階段，notebooklm doctor 顯示 Auth 為 fail。\n" + + "\n## 失敗原因\n根本原因是 NotebookLM 尚未登入，需互動式瀏覽器完成 Google 驗證。\n" ) -GOOD_REPORT = """# 影片重點整理：分散式系統的共識演算法 -## 背景與問題 -這部影片從分散式系統為什麼難講起，說明在多節點環境下，網路延遲與節點故障 -讓「大家對同一件事達成一致」變得困難。作者用銀行轉帳的例子帶出一致性的重要。 +def test_good_report_passes(good_report): + assert check_report_text(good_report).ok -## Paxos 的核心概念 -接著介紹 Paxos，分成提案者與接受者兩種角色。重點在於多數決：只要超過半數 -節點同意，提案就成立，因此可以容忍少數節點失效仍維持一致。 -## Raft 為何更易理解 -影片比較 Raft 與 Paxos，指出 Raft 把問題拆成領導者選舉、日誌複製與安全性 -三塊，讓工程師更容易實作與除錯，這也是它被廣泛採用的原因。 +def test_empty_report_fails(): + assert not check_report_text(EMPTY).ok -## 領導者選舉的細節 -影片進一步拆解 Raft 的領導者選舉：每個節點維護一個任期編號，當追隨者在逾時時間內 -沒有收到領導者的心跳，就會自己變成候選者並發起投票。候選者向其他節點請求選票， -拿到多數票就晉升為新的領導者。任期編號確保舊領導者復活後不會造成混亂，因為它的 -任期較小會被拒絕。這段用動畫示範了腦裂情況下系統如何靠多數決收斂回單一領導者。 -## 日誌複製與一致性保證 -領導者把每個客戶端請求當成一筆日誌條目，先寫進自己的日誌再廣播給追隨者。只有當 -多數節點都確認寫入，這筆條目才會被標記為已提交並真正套用到狀態機。影片強調這個 -「先複製、多數確認、才提交」的順序，是 Raft 能在故障下仍保證一致性的關鍵機制。 +def test_login_gate_fails(): + assert not check_report_text(LOGIN_GATE).ok -## 實務取捨與總結 -最後談到延遲與可用性的取捨：節點越多容錯越強但寫入延遲也越高，作者建議多數應用 -直接採用成熟的 Raft 實作（例如 etcd 背後的函式庫）而不要自己造輪子，並提醒一致性 -不是免費的，務必依業務需求衡量強一致與高可用之間的平衡點。 -""" -LOGIN_GATE = "Sign in with Google to continue.\n請登入 Google 帳號以使用 NotebookLM。" -EMPTY = " \n \n" -THIN = "# 標題\n太短了。" +def test_thin_report_fails(): + assert not check_report_text(THIN).ok -# Real production failure: builder honestly wrote a long, well-formed Chinese failure -# report with ## headers (passes length/header checks) — the TITLE gives it away. -FAILURE_REPORT = """# NotebookLM 處理失敗報告：無法產出來源的網誌式內容 -本報告如實記錄使用 NotebookLM 處理來源的嘗試過程與失敗原因，依任務要求不編造內容。 -這段文字刻意寫得夠長、包含足夠的中文字數，並且有多個段落標題，用來證明就算長度與 -header 數量都達標，光靠確定性長度檢查仍可能放行一份「其實是失敗」的報告，因此需要 -標題啟發式與語意稽核員一起把關，才能真正擋住 silent failure 這種看似成功的情況。 +def test_long_failure_titled_report_fails(): + assert not check_report_text(FAILURE_REPORT).ok -## 執行過程 -環境中找不到有效的登入工作階段，notebooklm doctor 顯示 Auth 為 fail，因此無法把連結 -加入筆記本，也無法取得任何帶引用的研究答案，整個流程在驗證關卡就被擋下來了。 -## 失敗原因 -根本原因是 NotebookLM 尚未登入，且登入需要互動式瀏覽器完成 Google 驗證，無法在無頭 -環境自動完成，為避免違反不得編造內容的要求，本報告不以記憶或猜測填補論文細節。 -""" +def test_good_report_has_enough_headers(good_report): + assert check_report_text(good_report).metrics["headers"] >= 3 -failures = [] +def test_notion_real_content_passes(good_report): + assert check_notion_fetch(good_report).ok -def expect(name, cond): - status = "PASS" if cond else "FAIL" - print(f" [{status}] {name}") - if not cond: - failures.append(name) +def test_notion_blank_fails(): + assert not check_notion_fetch(EMPTY).ok -print("== deterministic report checks ==") -expect("real report passes", check_report_text(GOOD_REPORT).ok) -expect("empty report fails", not check_report_text(EMPTY).ok) -expect("login gate fails", not check_report_text(LOGIN_GATE).ok) -expect("thin report fails", not check_report_text(THIN).ok) -expect("long failure-titled report fails (real prod case)", not check_report_text(FAILURE_REPORT).ok) -r = check_report_text(GOOD_REPORT) -expect("good report counts >=3 headers", r.metrics["headers"] >= 3) -print("== deterministic notion checks ==") -expect("real notion content passes", check_notion_fetch(GOOD_REPORT).ok) -expect("blank notion page fails", not check_notion_fetch(EMPTY).ok) +@pytest.mark.parametrize("text, ok", [ + ("PASS：很好", True), ("FAIL：登入頁", False), ("???", False), +]) +def test_parse_verdict_basics(text, ok): + assert parse_verdict(text).ok is ok -print("== verdict parsing ==") -expect("PASS line parsed true", parse_verdict("PASS：很好").ok) -expect("FAIL line parsed false", not parse_verdict("FAIL：登入頁").ok) -expect("garbage parsed false", not parse_verdict("???").ok) -print("== contract: refuses success without evidence ==") -with tempfile.TemporaryDirectory() as d: - job = Path(d) / "job1" +def test_contract_refuses_success_without_evidence(tmp_path): + job = tmp_path / "job1" c = Contract.create(job) - expect("fresh contract not all_pass", not c.all_pass()) - try: + assert not c.all_pass() + with pytest.raises(ValueError): c.pass_gate("notebooklm_report", evidence="") - expect("empty-evidence pass rejected", False) - except ValueError: - expect("empty-evidence pass rejected", True) c.pass_gate("notebooklm_report", evidence="讀了 report.md：1200 中文字、4 header") - expect("one gate passed != all_pass", not c.all_pass()) + assert not c.all_pass() # one gate is not enough c.pass_gate("notion_page", evidence="notion-fetch 回讀：非空、header 在") - expect("both gates -> all_pass", c.all_pass()) - # reload from disk to prove persistence - c2 = Contract.load(job) - expect("contract persists across reload", c2.all_pass()) - -print() -if failures: - print(f"❌ {len(failures)} check(s) failed: {failures}") - sys.exit(1) -print("✅ all evidence-spine checks passed") + assert c.all_pass() + assert Contract.load(job).all_pass() # persists across reload diff --git a/tests/test_gate.py b/tests/test_gate.py index 26938cd..0dec8a9 100644 --- a/tests/test_gate.py +++ b/tests/test_gate.py @@ -1,128 +1,54 @@ -"""Gate logic regression tests (no SDK, no network) — would have caught C1/C2. - -Injects fake builder/evaluator agents into the pipeline so the gate ordering and the -success decision can be tested offline. - -Run: python tests/test_gate.py -""" +"""Gate logic: tolerant verdict parsing + Gate B running the deterministic check on the +real fetched page body (no substring escape hatch).""" import asyncio -import sys -from pathlib import Path - -sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) - -from agentnotes import pipeline as P # noqa: E402 -from agentnotes.agent import AgentRun # noqa: E402 -from agentnotes.config import JOBS_DIR, Config # noqa: E402 -from agentnotes.evaluators import ( # noqa: E402 - PAGE_BODY_BEGIN, - PAGE_BODY_END, - parse_verdict, -) - -# A report that passes the deterministic check (>=400 CJK, >=3 "##" headers). -GOOD_REPORT = """# 殘差學習如何讓深層網路可訓練 - -這份報告根據來源的帶引用整理而成，逐主題說明它要解決的問題、提出的核心方法，以及在多個 -競賽上的實際表現。內容刻意寫得夠長、包含足夠的中文字數與多個分段標題，用來確認當報告本身 -是合格的時候，確定性檢查與稽核員都會放行，而結構也完整到足以被獨立驗證。 - -## 一、它要解決的問題：越深的網路越難訓練 - -在這個方法出現之前，研究者已經知道增加深度能提升表現，但實務上一旦把層數疊得太高，最佳化 -就會變得非常棘手，深度反而不再是助力。這篇來源正是針對這個訓練障礙，提出一套全新的框架， -讓比以往更深的網路也能被穩定地訓練出來，而不會在加深之後表現不升反降。 - -## 二、核心方法：把學整個函數改成學殘差 - -方法的精髓在於對網路層的重新表述：與其讓每一組網路層從零開始學習理想的輸出，不如讓它只學習 -相對於該層輸入的殘差，也就是輸出與輸入之間的差值，再把這個差值加回原本的輸入。這個轉換看似 -只是換個角度，卻從根本上降低了最佳化的難度，並讓網路能真正從增加深度當中換到準確率的提升。 - -## 三、實際成績：跨任務的全面領先 - -理論最終仍要靠成績說話，而它的表現相當有說服力：在大型影像辨識競賽上取得很低的辨識錯誤並 -奪得名次，同時把這種極深的特徵表示法遷移到物件偵測、定位與影像分割等多個下游任務，全部都 -拿到領先的結果，證明這套方法帶來的並非單一任務的調參優勢，而是一種能廣泛遷移的通用能力。 -""" - -cfg = Config( - telegram_bot_token="x", - telegram_allowed_chat_ids=frozenset(), - notion_parent_page_id="", - notion_database_id="db123", - report_min_cjk_chars=400, - report_min_headers=3, -) - -failures = [] - - -def expect(name, cond): - print(f" [{'PASS' if cond else 'FAIL'}] {name}") - if not cond: - failures.append(name) +import pytest -def _install_fakes(job_id, *, notion_eval_text, gate_a="PASS：報告 OK"): - """Patch pipeline.run_builder/run_evaluator. Builder#1 writes report.md; builder#2 - returns a Notion URL. Evaluator#1 is gate A; evaluator#2 is gate B.""" - report_path = JOBS_DIR / job_id / "report.md" - state = {"b": 0, "e": 0} +from agentnotes import pipeline as P +from agentnotes.dedup import ProcessedIndex +from agentnotes.evaluators import PAGE_BODY_BEGIN, PAGE_BODY_END, parse_verdict - async def fake_builder(prompt, **kw): - state["b"] += 1 - if state["b"] == 1: - report_path.parent.mkdir(parents=True, exist_ok=True) - report_path.write_text(GOOD_REPORT, encoding="utf-8") - return AgentRun("1500 字、3 段", None, False) - return AgentRun("已建立頁面：https://www.notion.so/deadbeefcafe", None, False) - async def fake_eval(prompt, **kw): - state["e"] += 1 - return AgentRun(gate_a if state["e"] == 1 else notion_eval_text, None, False) +@pytest.mark.parametrize("text, ok", [ + ("PASS：好", True), + ("FAIL：登入頁", False), + ("好的，我看過了\nPASS：完整", True), # preamble then PASS line + ("**PASS**：完整", True), # bold + ("我判定 PASS：好", True), # inline + ("FAIL：頁面空白，無法 PASS", False), # FAIL line mentioning the word PASS + ("this is not a PASS, it's a FAIL", False), + ("???", False), + ("", False), +]) +def test_parse_verdict(text, ok): + assert parse_verdict(text).ok is ok - P.run_builder, P.run_evaluator = fake_builder, fake_eval +def _run_gate_b(cfg, registry, fake_agents, job_id, gate_b_text): + fake_agents.gate_b_text = gate_b_text + return asyncio.run(P.run_pipeline( + job_id, "https://x.org/p", cfg, registry, "phd", index=ProcessedIndex(base_dir=None))) -def run_gate_b(job_id, notion_eval_text): - _install_fakes(job_id, notion_eval_text=notion_eval_text) - return asyncio.run(P.run_pipeline(job_id, "https://x.org/p", cfg)) +def test_gateb_rejects_fail_verdict_with_pass_substring(cfg, registry, fake_agents): + # The exact attack the naive version let through: verdict FAIL but text contains 'PASS'. + r = _run_gate_b(cfg, registry, fake_agents, "t-gb-1", + f"FAIL：頁面空白，無法 PASS\n{PAGE_BODY_BEGIN}\n{PAGE_BODY_END}") + assert not r.ok -print("== parse_verdict (C2: tolerate preamble/markdown, fail-closed) ==") -expect("plain PASS", parse_verdict("PASS：好").ok) -expect("plain FAIL", not parse_verdict("FAIL：登入頁").ok) -expect("preamble then PASS line", parse_verdict("好的，我看過了\nPASS：完整").ok) -expect("bold **PASS**", parse_verdict("**PASS**：完整").ok) -expect("inline '我判定 PASS：好'", parse_verdict("我判定 PASS：好").ok) -expect("FAIL mentioning the word PASS -> FAIL", not parse_verdict("FAIL：頁面空白，無法 PASS").ok) -expect("'not a PASS, it is a FAIL' -> FAIL", not parse_verdict("this is not a PASS, it's a FAIL").ok) -expect("garbage -> FAIL", not parse_verdict("???").ok) -expect("empty -> FAIL", not parse_verdict("").ok) -print("== Gate B (C1: deterministic check runs on real page body, no substring hatch) ==") -# The exact attack the old code let through: verdict FAIL but text contains 'PASS'. -r1 = run_gate_b("test-gateb-1", f"FAIL：頁面空白，無法 PASS\n{PAGE_BODY_BEGIN}\n{PAGE_BODY_END}") -expect("malicious FAIL-with-PASS-substring does NOT pass", not r1.ok) +def test_gateb_rejects_pass_verdict_with_empty_body(cfg, registry, fake_agents): + r = _run_gate_b(cfg, registry, fake_agents, "t-gb-2", + f"PASS：看起來有寫\n{PAGE_BODY_BEGIN}\n{PAGE_BODY_END}") + assert not r.ok -# Verdict says PASS but the fetched body is empty -> deterministic check must fail it. -r2 = run_gate_b("test-gateb-2", f"PASS：看起來有寫\n{PAGE_BODY_BEGIN}\n{PAGE_BODY_END}") -expect("PASS verdict + empty body does NOT pass", not r2.ok) -# Genuine success: PASS verdict + real body echoed back. -r3 = run_gate_b( - "test-gateb-3", - f"PASS：分段完整保留\n{PAGE_BODY_BEGIN}\n{GOOD_REPORT}\n{PAGE_BODY_END}", -) -expect("PASS verdict + real body passes, returns notion url", r3.ok and bool(r3.notion_url)) +def test_gateb_accepts_pass_with_real_body(cfg, registry, fake_agents, good_report): + r = _run_gate_b(cfg, registry, fake_agents, "t-gb-3", + f"PASS：分段完整\n{PAGE_BODY_BEGIN}\n{good_report}\n{PAGE_BODY_END}") + assert r.ok and r.notion_url -# Missing markers entirely -> can't verify body -> fail closed even on a PASS verdict. -r4 = run_gate_b("test-gateb-4", "PASS：我覺得有寫進去（但沒有附正文）") -expect("PASS verdict but no body markers -> fail closed", not r4.ok) -print() -if failures: - print(f"❌ {len(failures)} failed: {failures}") - sys.exit(1) -print("✅ all gate tests passed") +def test_gateb_fails_closed_without_body_markers(cfg, registry, fake_agents): + r = _run_gate_b(cfg, registry, fake_agents, "t-gb-4", "PASS：我覺得有寫進去（但沒附正文）") + assert not r.ok diff --git a/tests/test_main.py b/tests/test_main.py new file mode 100644 index 0000000..4ece6fc --- /dev/null +++ b/tests/test_main.py @@ -0,0 +1,127 @@ +"""main.py helpers (_title_hint, _job_id) and handle_job dispatch.""" +import asyncio + +import pytest + +from agentnotes import main as M +from agentnotes.ingest.base import IncomingJob +from agentnotes.pipeline import PipelineResult + + +@pytest.mark.parametrize("job, expected", [ + (IncomingJob("c", "m", "https://youtu.be/x", "看這個影片 #phd https://youtu.be/x", "phd"), + "看這個影片"), + (IncomingJob("c", "m", "", "#phd", "phd", file_path="/d/My Thesis.pdf", + file_name="My Thesis.pdf"), "My Thesis"), + (IncomingJob("c", "m", "https://x", "https://x", "phd"), "未命名報告"), +]) +def test_title_hint(job, expected): + assert M._title_hint(job) == expected + + +def test_job_id_shape(): + j = IncomingJob("123", "9", "https://x", "https://x", "phd") + assert M._job_id(j).endswith("-123-9") + + +class FakeAdapter: + def __init__(self): + self.progress = [] + self.replies = [] + + async def send_progress(self, chat, text): + self.progress.append((chat, text)) + + async def send_reply(self, chat, text, reply_to=None): + self.replies.append((chat, text, reply_to)) + + +@pytest.fixture +def dispatch(monkeypatch, cfg, registry, index): + """Returns (run, calls): run(job, *, pipeline_result|sweep_result|raises) dispatches + handle_job with fakes and returns the FakeAdapter.""" + calls = {} + monkeypatch.setattr(M, "_log_progress", lambda *a, **k: None) # don't touch PROGRESS.md + + state = {"pipeline_result": PipelineResult(True, "done", notion_url="u"), + "sweep_result": PipelineResult(True, "swept-summary"), + "raises": False} + + async def fake_pipeline(job_id, url, c, r, topic, title_hint="未命名報告", *, + force=False, index=None, file_path=None, file_name=None): + if state["raises"]: + raise RuntimeError("kaboom") + calls["pipeline"] = {"url": url, "topic": topic, "force": force, + "file_path": file_path, "file_name": file_name} + return state["pipeline_result"] + + async def fake_sweep(job_id, c, r, idx, topic, *, force=False): + calls["sweep"] = {"topic": topic, "force": force} + return state["sweep_result"] + + monkeypatch.setattr(M, "run_pipeline", fake_pipeline) + monkeypatch.setattr(M, "run_sweep", fake_sweep) + + def run(job): + calls.clear() + a = FakeAdapter() + asyncio.run(M.handle_job(a, cfg, registry, index, job)) + return a + + return run, calls, state + + +def test_normal_url_job_calls_pipeline(dispatch): + run, calls, _ = dispatch + a = run(IncomingJob("123", "9", "https://youtu.be/x", "https://youtu.be/x", "phd")) + assert "pipeline" in calls and "sweep" not in calls + assert calls["pipeline"]["url"] == "https://youtu.be/x" and calls["pipeline"]["file_path"] is None + assert any("done" in t for _, t, _ in a.replies) + + +def test_file_job_threads_file_args(dispatch): + run, calls, _ = dispatch + run(IncomingJob("123", "9", "", "#phd", "phd", file_path="/d/t.pdf", file_name="t.pdf")) + assert calls["pipeline"]["file_path"] == "/d/t.pdf" and calls["pipeline"]["file_name"] == "t.pdf" + + +def test_force_flag_forwarded(dispatch): + run, calls, _ = dispatch + run(IncomingJob("123", "9", "https://x", "https://x #force", "phd", force=True)) + assert calls["pipeline"]["force"] is True + + +def test_sweep_command_routes_to_run_sweep(dispatch): + run, calls, _ = dispatch + a = run(IncomingJob("123", "9", "", "#phd 整理現有", "phd", command="sweep")) + assert "sweep" in calls and "pipeline" not in calls + assert any("swept-summary" in t for _, t, _ in a.replies) + + +def test_duplicate_result_relayed(dispatch): + run, calls, state = dispatch + state["pipeline_result"] = PipelineResult(True, "已經處理過了", notion_url="u", duplicate=True) + a = run(IncomingJob("123", "9", "https://x", "https://x", "phd")) + assert any("已經處理過了" in t for _, t, _ in a.replies) + + +def test_exception_is_caught(dispatch): + run, calls, state = dispatch + state["raises"] = True + a = run(IncomingJob("123", "9", "https://x", "https://x", "phd")) # must not raise + assert any("例外" in t for _, t, _ in a.replies) + + +def test_auth_error_message_reaches_user(monkeypatch, cfg, registry, index): + """An AgentError (e.g. CLI not logged in) should surface a clear, actionable reply.""" + from agentnotes.agent import AgentError + monkeypatch.setattr(M, "_log_progress", lambda *a, **k: None) + + async def raise_auth(*a, **k): + raise AgentError("Claude CLI 可能未登入。請在終端機跑 `claude auth login`。") + + monkeypatch.setattr(M, "run_pipeline", raise_auth) + a = FakeAdapter() + asyncio.run(M.handle_job(a, cfg, registry, index, + IncomingJob("1", "2", "https://x", "https://x", "phd"))) + assert any("登入" in t for _, t, _ in a.replies) diff --git a/tests/test_sweep.py b/tests/test_sweep.py new file mode 100644 index 0000000..35cabc4 --- /dev/null +++ b/tests/test_sweep.py @@ -0,0 +1,125 @@ +"""UC3 sweep: command detection, deterministic source listing, per-source dedup/skip/ +failure handling, and #force.""" +import asyncio + +import pytest + +from agentnotes import pipeline as P +from agentnotes.dedup import ProcessedIndex, normalize_url +from agentnotes.ingest.telegram import _is_sweep_command + + +@pytest.mark.parametrize("text, is_sweep", [ + ("#phd 整理現有", True), + ("#phd sync", True), + ("整理現有 #phd", True), + ("整理現有", True), # works even before topic resolution + ("#phd", False), + ("#phd asynchronous", False), # R1: no ASCII-substring hijack + ("#phd let's sync up", False), + ("#phd sweep the floor", False), + ("#phd 整理現有的論文", False), # extra words -> not exact + ("#phd https://youtu.be/x", False), + ("#phd https://youtu.be/x please sync this", False), # URL message never a sweep +]) +def test_is_sweep_command(text, is_sweep): + assert _is_sweep_command(text) is is_sweep + + +def _patch_cli(monkeypatch, sources, notebook_id="NB1"): + monkeypatch.setattr(P.nlm, "find_notebook_id", lambda title, **kw: notebook_id) + monkeypatch.setattr(P.nlm, "list_sources", lambda nb, **kw: list(sources)) + + +SOURCES = [ + {"id": "s1", "title": "Paper One", "type": "youtube", + "url": "https://youtu.be/aaa", "status": "ready"}, + {"id": "s2", "title": "Paper Two (file)", "type": "file", "status": "ready"}, + {"id": "s3", "title": "Still Processing", "type": "youtube", + "url": "https://youtu.be/ccc", "status": "processing"}, +] + + +def test_first_pass_processes_ready_skips_not_ready(cfg, registry, fake_agents, monkeypatch): + _patch_cli(monkeypatch, SOURCES) + idx = ProcessedIndex(base_dir=None) + r = asyncio.run(P.run_sweep("sw-a", cfg, registry, idx, "phd")) + assert r.ok + assert "共 3 筆" in r.message and "尚未就緒" in r.message and "新產 2" in r.message + assert idx.lookup_key("phd", normalize_url("https://youtu.be/aaa")) is not None + assert idx.lookup_key("phd", "nbsrc:s2") is not None + + +def test_rerun_is_idempotent(cfg, registry, fake_agents, monkeypatch): + _patch_cli(monkeypatch, SOURCES) + idx = ProcessedIndex(base_dir=None) + asyncio.run(P.run_sweep("sw-b1", cfg, registry, idx, "phd")) + r = asyncio.run(P.run_sweep("sw-b2", cfg, registry, idx, "phd")) + assert "新產 0" in r.message and "跳過 2" in r.message + + +def test_source_already_done_via_url_job_is_skipped(cfg, registry, fake_agents, monkeypatch): + _patch_cli(monkeypatch, SOURCES) + idx = ProcessedIndex(base_dir=None) + idx.record("phd", "https://www.youtube.com/watch?v=aaa", "https://notion.so/prev") + r = asyncio.run(P.run_sweep("sw-c", cfg, registry, idx, "phd")) + assert "新產 1" in r.message and "跳過 1" in r.message + + +def test_missing_notebook_fails_honestly(cfg, registry, fake_agents, monkeypatch): + monkeypatch.setattr(P.nlm, "find_notebook_id", lambda title, **kw: None) + r = asyncio.run(P.run_sweep("sw-d", cfg, registry, ProcessedIndex(base_dir=None), "phd")) + assert not r.ok and "找不到筆記本" in r.message + + +def test_empty_notebook(cfg, registry, fake_agents, monkeypatch): + _patch_cli(monkeypatch, []) + r = asyncio.run(P.run_sweep("sw-e", cfg, registry, ProcessedIndex(base_dir=None), "phd")) + assert "新產 0" in r.message and "沒有新的要做" in r.message + + +@pytest.mark.parametrize("src, key", [ + ({"id": "z9", "url": "", "title": "t"}, "nbsrc:z9"), + ({"id": "z9", "url": "https://youtu.be/aaa", "title": "t"}, normalize_url("https://youtu.be/aaa")), +]) +def test_existing_source_key(src, key): + assert P.existing_source(src).key == key + + +def test_malformed_source_without_id_is_skipped(cfg, registry, fake_agents, monkeypatch): + _patch_cli(monkeypatch, [ + {"title": "no id", "status": "ready"}, + {"id": "ok1", "title": "Has id", "type": "file", "status": "ready"}, + ]) + r = asyncio.run(P.run_sweep("sw-f", cfg, registry, ProcessedIndex(base_dir=None), "phd")) + assert "新產 1" in r.message and r.ok + + +def test_per_source_failure_counted_not_recorded_then_retried(cfg, registry, fake_agents, monkeypatch): + two = [ + {"id": "g1", "title": "GOODSRC", "type": "youtube", + "url": "https://youtu.be/good", "status": "ready"}, + {"id": "b1", "title": "BADSRC", "type": "youtube", + "url": "https://youtu.be/bad", "status": "ready"}, + ] + _patch_cli(monkeypatch, two) + fake_agents.fail_marker = "BADSRC" # gate A fails for the source whose title is BADSRC + idx = ProcessedIndex(base_dir=None) + r = asyncio.run(P.run_sweep("sw-g", cfg, registry, idx, "phd")) + assert "新產 1" in r.message and "失敗 1" in r.message and not r.ok + assert idx.lookup_key("phd", normalize_url("https://youtu.be/good")) is not None + assert idx.lookup_key("phd", normalize_url("https://youtu.be/bad")) is None + + fake_agents.fail_marker = None # retry: now everything succeeds + r2 = asyncio.run(P.run_sweep("sw-g2", cfg, registry, idx, "phd")) + assert "新產 1" in r2.message and "跳過 1" in r2.message + + +def test_force_redoes_recorded_source(cfg, registry, fake_agents, monkeypatch): + one = [{"id": "f1", "title": "Forced", "type": "youtube", + "url": "https://youtu.be/fff", "status": "ready"}] + _patch_cli(monkeypatch, one) + idx = ProcessedIndex(base_dir=None) + asyncio.run(P.run_sweep("sw-h", cfg, registry, idx, "phd")) + r = asyncio.run(P.run_sweep("sw-h2", cfg, registry, idx, "phd", force=True)) + assert "新產 1" in r.message and "跳過 0" in r.message diff --git a/tests/test_telegram.py b/tests/test_telegram.py new file mode 100644 index 0000000..937834a --- /dev/null +++ b/tests/test_telegram.py @@ -0,0 +1,51 @@ +"""Telegram poll() URL routing (no network), via the poll_once driver.""" +from conftest import make_update + + +def test_url_with_topic_yields_one_job(registry, poll_once): + jobs, calls = poll_once(registry, make_update(text="#phd https://youtu.be/x")) + assert len(jobs) == 1 + j = jobs[0] + assert (j.url == "https://youtu.be/x" and j.topic == "phd" + and j.force is False and j.file_path is None and j.command is None) + assert calls["commits"] == [8] # update_id 7 -> offset 8 + + +def test_force_flag(registry, poll_once): + jobs, _ = poll_once(registry, make_update(text="#phd #force https://youtu.be/x")) + assert len(jobs) == 1 and jobs[0].force is True + + +def test_multiple_urls_yield_multiple_jobs(registry, poll_once): + jobs, _ = poll_once(registry, make_update(text="#phd https://youtu.be/a https://youtu.be/b")) + assert {j.url for j in jobs} == {"https://youtu.be/a", "https://youtu.be/b"} + + +def test_missing_topic_replies_menu_no_job(registry, poll_once): + jobs, calls = poll_once(registry, make_update(text="https://youtu.be/x")) + assert jobs == [] + assert calls["replies"] and "topic" in calls["replies"][0] + assert calls["commits"] == [8] + + +def test_ambiguous_topic_replies_no_job(registry, poll_once): + jobs, calls = poll_once(registry, make_update(text="#ai #phd https://youtu.be/x")) + assert jobs == [] + assert calls["replies"] and "只標一個" in calls["replies"][0] + + +def test_non_allowed_chat_is_ignored(registry, poll_once): + jobs, calls = poll_once(registry, make_update(text="#phd https://youtu.be/x"), + allowed=frozenset({999})) + assert jobs == [] and calls["replies"] == [] + assert calls["commits"] == [8] + + +def test_plain_text_replies_menu(registry, poll_once): + jobs, calls = poll_once(registry, make_update(text="#phd hello there")) + assert jobs == [] and calls["replies"] + + +def test_update_without_message_is_skipped(registry, poll_once): + jobs, calls = poll_once(registry, {"update_id": 7}) + assert jobs == [] and calls["replies"] == [] and calls["commits"] == [8] diff --git a/tests/test_topics.py b/tests/test_topics.py new file mode 100644 index 0000000..c8f1e52 --- /dev/null +++ b/tests/test_topics.py @@ -0,0 +1,86 @@ +"""Topic routing: tag extraction, alias-normalized resolution, db-id cache.""" +import json + +import pytest + +from agentnotes.pipeline import _extract_db_id +from agentnotes.topics import TOPICS_FILE, TopicRegistry, extract_tags + +DEFS = { + "ai-agent": {"aliases": ["ai", "agent", "ai agent", "ai-agent"], + "notebook": "(AgentNote) AI agent", + "notion_db_title": "NotebookLM 報告 - AI agent"}, + "phd": {"aliases": ["phd", "phd study", "博士", "論文"], + "notebook": "(AgentNote) PhD study", + "notion_db_title": "NotebookLM 報告 - PhD Study"}, +} + + +def reg(cache=None, cache_path=None): + return TopicRegistry(defs=DEFS, cache=dict(cache or {}), cache_path=cache_path) + + +@pytest.mark.parametrize("text, tags", [ + ("#ai https://x.org", ["ai"]), + ("https://x.org #phd", ["phd"]), + ("#ai #phd x", ["ai", "phd"]), + ("https://x.org", []), +]) +def test_extract_tags(text, tags): + assert extract_tags(text) == tags + + +@pytest.mark.parametrize("tag, topic", [ + ("ai-agent", "ai-agent"), ("ai", "ai-agent"), ("AI_Agent", "ai-agent"), + ("aiagent", "ai-agent"), ("博士", "phd"), ("crypto", None), +]) +def test_resolve_alias(tag, topic): + assert reg().resolve(tag) == topic + + +@pytest.mark.parametrize("text, expected", [ + ("#phd 看這個 https://x", ("phd", None)), + ("https://x", (None, "missing")), + ("#crypto https://x", (None, "missing")), + ("#ai #phd https://x", (None, "ambiguous")), + ("#ai #agent https://x", ("ai-agent", None)), # same topic twice is fine +]) +def test_resolve_message(text, expected): + assert reg().resolve_message(text) == expected + + +def test_db_id_cache_roundtrip_on_disk(tmp_path): + cache_path = tmp_path / "topic_dbs.json" + r1 = reg(cache_path=cache_path) + assert r1.notion_db_id("phd") == "" + r1.set_notion_db_id("phd", "abc123") + assert r1.notion_db_id("phd") == "abc123" + assert json.loads(cache_path.read_text())["phd"] == "abc123" + # a fresh registry reading the committed defs + this cache sees it + r2 = TopicRegistry.load(defs_path=TOPICS_FILE, cache_path=cache_path) + assert r2.notion_db_id("phd") == "abc123" + + +def test_set_db_id_empty_does_not_clobber(): + r = reg(cache={"phd": "x"}) + r.set_notion_db_id("phd", "") + assert r.notion_db_id("phd") == "x" + + +@pytest.mark.parametrize("text, expected", [ + ("已建立。database_id = 0123456789abcdef0123456789abcdef", + "0123456789abcdef0123456789abcdef"), + ("database_id: 5f4a7cfd-defa-41e7-8fc3-b66a52985b4c", + "5f4a7cfd-defa-41e7-8fc3-b66a52985b4c"), + ("頁面連結 https://www.notion.so/abcdef0123456789abcdef0123456789", None), +]) +def test_extract_db_id(text, expected): + assert _extract_db_id(text) == expected + + +def test_accessors_and_menu(): + r = reg() + assert r.notebook("phd") == "(AgentNote) PhD study" + assert r.notion_db_title("ai-agent") == "NotebookLM 報告 - AI agent" + menu = r.menu_text() + assert "#ai-agent" in menu and "#phd" in menu diff --git a/tests/test_url.py b/tests/test_url.py index 722487a..bd5770b 100644 --- a/tests/test_url.py +++ b/tests/test_url.py @@ -1,33 +1,16 @@ -"""Regression: Notion URL extraction must not swallow trailing markdown. +"""Notion URL extraction must not swallow trailing markdown/punctuation.""" +import pytest -A builder reply like "****" once leaked the trailing ** into the link sent -to the user. Run: python tests/test_url.py -""" -import sys -from pathlib import Path +from agentnotes.pipeline import _extract_notion_url -sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) -from agentnotes.pipeline import _extract_notion_url # noqa: E402 - -CASES = { - "看這裡 **https://www.notion.so/38d5fc56c439815cb9d5ebb5a4ea43d1**": - "https://www.notion.so/38d5fc56c439815cb9d5ebb5a4ea43d1", - "連結：https://www.notion.so/abc123).": "https://www.notion.so/abc123", - "done https://app.notion.com/p/xyz789": "https://app.notion.com/p/xyz789", - "[頁面](https://notion.so/page-42)": "https://notion.so/page-42", - "no url here": None, -} - -failures = [] -for inp, exp in CASES.items(): - got = _extract_notion_url(inp) - ok = got == exp - print(f" [{'PASS' if ok else 'FAIL'}] {got!r}") - if not ok: - failures.append((inp, exp, got)) - -if failures: - print(f"❌ {len(failures)} failed: {failures}") - sys.exit(1) -print("✅ notion url extraction OK") +@pytest.mark.parametrize("text, expected", [ + ("看這裡 **https://www.notion.so/38d5fc56c439815cb9d5ebb5a4ea43d1**", + "https://www.notion.so/38d5fc56c439815cb9d5ebb5a4ea43d1"), + ("連結：https://www.notion.so/abc123).", "https://www.notion.so/abc123"), + ("done https://app.notion.com/p/xyz789", "https://app.notion.com/p/xyz789"), + ("[頁面](https://notion.so/page-42)", "https://notion.so/page-42"), + ("no url here", None), +]) +def test_extract_notion_url(text, expected): + assert _extract_notion_url(text) == expected diff --git a/topics.json b/topics.json new file mode 100644 index 0000000..b788f49 --- /dev/null +++ b/topics.json @@ -0,0 +1,12 @@ +{ + "ai-agent": { + "aliases": ["ai", "agent", "ai agent", "ai-agent", "aiagent"], + "notebook": "(AgentNote) AI agent", + "notion_db_title": "NotebookLM 報告 - AI agent" + }, + "phd": { + "aliases": ["phd", "phd study", "phd-study", "博士", "博士論文", "論文"], + "notebook": "(AgentNote) PhD study", + "notion_db_title": "NotebookLM 報告 - PhD Study" + } +}