diff --git a/docs/superpowers/plans/2026-06-05-xhs-insight-pipeline.md b/docs/superpowers/plans/2026-06-05-xhs-insight-pipeline.md new file mode 100644 index 000000000..b3482210c --- /dev/null +++ b/docs/superpowers/plans/2026-06-05-xhs-insight-pipeline.md @@ -0,0 +1,1159 @@ +# XHS Insight Pipeline 实现计划 + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** 在不修改任何 MediaCrawler 上游文件的前提下，新增一个独立的 `insight/` 包，实现「按 cron 定时调用爬虫子进程 → 把小红书笔记/评论原始数据写入既有 SQLite → 记录每次运行日志」。 + +**Architecture:** 所有自研代码隔离在顶层 `insight/` 包内。`insight` 通过 **子进程** 调用一个自研入口 `insight.crawl_entry`（它读取环境变量覆盖无 CLI 参数的配置后委托给上游 `main`），爬虫把数据写入上游既有表 `xhs_note` / `xhs_note_comment`；`insight` 自己只拥有一张 `insight_runs` 日志表（读取上游表只做计数）。调度用 APScheduler 3.x 的 `BlockingScheduler` + `CronTrigger`，通过 `uv run --with` 注入依赖以避免改动 `pyproject.toml`。 + +**Tech Stack:** Python 3.11、APScheduler 3.x、标准库 `sqlite3` / `subprocess` / `argparse`、pytest（项目既有）。 + +--- + +## 背景事实（实现者必须知道，已核实） + +- **上游 CLI 参数**（`cmd_arg/arg.py`，由 `main.py` 解析）：`--platform`、`--type`（search/detail/creator）、`--keywords`、`--specified_id`（detail 模式，XHS 笔记 URL/ID，逗号分隔 → `config.XHS_SPECIFIED_NOTE_URL_LIST`）、`--creator_id`（creator 模式 → `config.XHS_CREATOR_ID_LIST`）、`--save_data_option`（取 `sqlite`）、`--get_comment`、`--get_sub_comment`、`--max_comments_count_singlenotes`、`--init_db sqlite`（初始化表结构）。 +- **无 CLI 参数**的项：`CRAWLER_MAX_NOTES_COUNT`（每次最多爬多少篇）既无 CLI 参数也不读环境变量。因此每个 job 的 `max_notes` 通过 `insight.crawl_entry` 包装入口用环境变量 `INSIGHT_MAX_NOTES` 注入。 +- **SQLite 路径**固定为 `database/sqlite_tables.db`（`config/db_config.py:50`，不读 env，不可由 CLI 改）。`insight/config.py` 的 `DB_PATH` 必须指向同一文件。 +- **首次需初始化表结构**：`uv run python main.py --init_db sqlite` 会创建上游所有 SQLite 表。`insight` 的 `count_rows` 在表尚不存在时返回 0，不报错。 +- **上游运行入口**：`main.main()` / `main.async_cleanup()` 搭配 `tools.app_runner.run(app_main, app_cleanup, *, cleanup_timeout_seconds=15.0)`（见 `main.py:141-157`）。`crawl_entry` 复用它们。 +- **APScheduler 版本**：`pip install apscheduler` 默认装 3.x（4.x 仍为预发布）。本计划用 3.x API：`from apscheduler.schedulers.blocking import BlockingScheduler`、`from apscheduler.triggers.cron import CronTrigger`、`scheduler.add_job(func, trigger=..., args=[...], id=..., misfire_grace_time=..., replace_existing=True)`、`scheduler.start()`。 +- **测试约定**：`tests/conftest.py` 已把项目根加入 `sys.path`。自研测试放 `tests/insight/`。运行测试需注入 apscheduler：`uv run --with "apscheduler>=3.10,<4" pytest tests/insight -v`。 +- **包内 import 区分**：顶层 `import config` 指上游配置；`from insight import config` 指自研配置。两者都要用时按此区分，切勿混淆。 + +--- + +## 文件结构 + +``` +insight/ +├─ __init__.py # 空，标记包 +├─ config.py # DB_PATH、超时、misfire、JOBS 列表（纯数据） +├─ schema.sql # insight_runs 建表 DDL +├─ crawl_entry.py # 子进程入口：env 覆盖 + 委托上游 main +├─ db.py # InsightDB：建表/记录运行/计数/查询 +├─ runner.py # build_crawl_args + run_crawl（子进程封装） +├─ orchestrator.py # run_job：串起 db + runner（一次完整周期） +├─ requirements.txt # apscheduler>=3.10,<4（依赖声明，避免改 pyproject） +├─ scheduler/ +│ ├─ __init__.py +│ └─ daemon.py # build_scheduler + main（BlockingScheduler） +├─ cli.py # crawl-once / run-daemon / status 子命令 +└─ README.md # 安装、使用、上游同步步骤 + +tests/insight/ +├─ __init__.py +├─ test_db.py +├─ test_runner_args.py +├─ test_runner_subprocess.py +├─ test_crawl_entry.py +├─ test_orchestrator.py +├─ test_scheduler.py +└─ test_cli.py +``` + +每个文件单一职责，可独立测试：`config` 只放数据；`db` 只管 SQLite；`runner` 只管「job→命令」与子进程；`orchestrator` 只做编排；`scheduler` 只做触发；`cli` 只做命令分发；`crawl_entry` 只做配置覆盖+委托。 + +--- + +## Task 1: 包骨架、配置、Schema、依赖声明 + +**Files:** +- Create: `insight/__init__.py` +- Create: `insight/scheduler/__init__.py` +- Create: `insight/config.py` +- Create: `insight/schema.sql` +- Create: `insight/requirements.txt` +- Create: `tests/insight/__init__.py` + +- [ ] **Step 1: 创建空包标记文件** + +`insight/__init__.py`： +```python +# -*- coding: utf-8 -*- +"""自研二次开发包：小红书评论定时采集与运行记录。不修改 MediaCrawler 上游文件。""" +``` + +`insight/scheduler/__init__.py`： +```python +# -*- coding: utf-8 -*- +``` + +`tests/insight/__init__.py`： +```python +# -*- coding: utf-8 -*- +``` + +- [ ] **Step 2: 写 `insight/config.py`（纯数据配置）** + +```python +# -*- coding: utf-8 -*- +"""insight 包的配置。纯数据，不含逻辑。""" + +import os + +# 项目根目录（insight/ 的上一级） +PROJECT_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) + +# 爬虫写入的 SQLite 文件，必须与 config/db_config.py 的 SQLITE_DB_PATH 一致 +DB_PATH = os.path.join(PROJECT_ROOT, "database", "sqlite_tables.db") + +# 单次爬虫子进程的最大运行秒数，超时则杀死并记为 timeout +SUBPROCESS_TIMEOUT = 1800 + +# APScheduler 容忍的最大迟到秒数（睡眠/停机后仍补跑） +MISFIRE_GRACE_TIME = 3600 + +# 定时任务列表。每个 job 映射为一次爬虫调用。 +# 必填键：name、type（"search"|"detail"|"creator"）、hour（0-23） +# search 额外需要 "keywords"（英文逗号分隔的字符串） +# detail 额外需要 "note_ids"（笔记 ID 或 URL 的列表） +# creator 额外需要 "creator_ids"（创作者 ID 或 URL 的列表） +# 可选键：minute（默认 0）、max_notes（int）、max_comments（int）、get_sub_comment（bool） +JOBS = [ + {"name": "kw_daily", "type": "search", "keywords": "编程副业,编程兼职", "hour": 2, "minute": 0, "max_notes": 20}, + {"name": "watch_notes", "type": "detail", "note_ids": ["请替换为真实笔记ID"], "hour": 3, "minute": 0}, + {"name": "creator_daily", "type": "creator", "creator_ids": ["请替换为真实创作者ID"], "hour": 4, "minute": 0}, +] +``` + +- [ ] **Step 3: 写 `insight/schema.sql`** + +```sql +CREATE TABLE IF NOT EXISTS insight_runs ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + job_name TEXT NOT NULL, + crawler_type TEXT NOT NULL, + started_ts INTEGER NOT NULL, + finished_ts INTEGER, + exit_code INTEGER, + notes_crawled INTEGER DEFAULT 0, + comments_crawled INTEGER DEFAULT 0, + status TEXT NOT NULL DEFAULT 'running', + error_msg TEXT +); +CREATE INDEX IF NOT EXISTS idx_insight_runs_job ON insight_runs (job_name, started_ts); +``` + +- [ ] **Step 4: 写 `insight/requirements.txt`** + +``` +apscheduler>=3.10,<4 +``` + +- [ ] **Step 5: 提交** + +```bash +git add insight/__init__.py insight/scheduler/__init__.py insight/config.py insight/schema.sql insight/requirements.txt tests/insight/__init__.py +git commit -m "feat(insight): 包骨架、配置、insight_runs schema 与依赖声明" +``` + +--- + +## Task 2: `insight/db.py` — SQLite 运行记录与计数 + +**Files:** +- Create: `insight/db.py` +- Test: `tests/insight/test_db.py` + +- [ ] **Step 1: 写失败测试** + +`tests/insight/test_db.py`： +```python +# -*- coding: utf-8 -*- +import os + +from insight.db import InsightDB + + +def _db(tmp_path): + db = InsightDB(os.path.join(str(tmp_path), "t.db")) + db.init_schema() + return db + + +def test_init_schema_creates_table_and_is_idempotent(tmp_path): + db = _db(tmp_path) + db.init_schema() # 再次调用不应报错 + assert db.recent_runs() == [] + + +def test_start_and_finish_run_roundtrip(tmp_path): + db = _db(tmp_path) + run_id = db.start_run("kw_daily", "search") + assert isinstance(run_id, int) + + runs = db.recent_runs() + assert len(runs) == 1 + assert runs[0]["status"] == "running" + assert runs[0]["job_name"] == "kw_daily" + assert runs[0]["finished_ts"] is None + + db.finish_run(run_id, exit_code=0, status="success", notes_crawled=3, comments_crawled=12) + runs = db.recent_runs() + assert runs[0]["status"] == "success" + assert runs[0]["exit_code"] == 0 + assert runs[0]["notes_crawled"] == 3 + assert runs[0]["comments_crawled"] == 12 + assert runs[0]["finished_ts"] is not None + + +def test_count_rows_returns_zero_when_table_absent(tmp_path): + db = _db(tmp_path) + assert db.count_rows("xhs_note") == 0 + assert db.count_rows("xhs_note_comment") == 0 + + +def test_count_rows_rejects_unknown_table(tmp_path): + db = _db(tmp_path) + try: + db.count_rows("evil_table") + except ValueError: + return + raise AssertionError("expected ValueError for disallowed table") + + +def test_recent_runs_orders_desc_and_limits(tmp_path): + db = _db(tmp_path) + for i in range(3): + db.start_run(f"job{i}", "search") + runs = db.recent_runs(limit=2) + assert len(runs) == 2 + assert runs[0]["job_name"] == "job2" # 最新在前 +``` + +- [ ] **Step 2: 运行测试确认失败** + +Run: `uv run pytest tests/insight/test_db.py -v` +Expected: FAIL（`ModuleNotFoundError: No module named 'insight.db'`） + +- [ ] **Step 3: 写最小实现 `insight/db.py`** + +```python +# -*- coding: utf-8 -*- +"""insight 的 SQLite 访问层：运行记录与上游表计数。""" + +import os +import sqlite3 +import time +from typing import List, Dict, Optional + +SCHEMA_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "schema.sql") + +# 仅允许对这些上游表做计数，避免 SQL 注入 +_COUNTABLE_TABLES = {"xhs_note", "xhs_note_comment"} + + +class InsightDB: + def __init__(self, db_path: str): + self.db_path = db_path + + def _connect(self) -> sqlite3.Connection: + conn = sqlite3.connect(self.db_path) + conn.row_factory = sqlite3.Row + return conn + + def init_schema(self) -> None: + with open(SCHEMA_PATH, "r", encoding="utf-8") as f: + ddl = f.read() + with self._connect() as conn: + conn.executescript(ddl) + + def start_run(self, job_name: str, crawler_type: str) -> int: + now = int(time.time()) + with self._connect() as conn: + cur = conn.execute( + "INSERT INTO insight_runs (job_name, crawler_type, started_ts, status) " + "VALUES (?, ?, ?, 'running')", + (job_name, crawler_type, now), + ) + return int(cur.lastrowid) + + def finish_run( + self, + run_id: int, + *, + exit_code: Optional[int], + status: str, + notes_crawled: int = 0, + comments_crawled: int = 0, + error_msg: Optional[str] = None, + ) -> None: + now = int(time.time()) + with self._connect() as conn: + conn.execute( + "UPDATE insight_runs SET finished_ts=?, exit_code=?, status=?, " + "notes_crawled=?, comments_crawled=?, error_msg=? WHERE id=?", + (now, exit_code, status, notes_crawled, comments_crawled, error_msg, run_id), + ) + + def count_rows(self, table: str) -> int: + if table not in _COUNTABLE_TABLES: + raise ValueError(f"count_rows: table {table!r} not allowed") + with self._connect() as conn: + try: + row = conn.execute(f"SELECT COUNT(*) AS c FROM {table}").fetchone() + except sqlite3.OperationalError: + return 0 # 表尚未由上游 --init_db 创建 + return int(row["c"]) + + def recent_runs(self, limit: int = 20) -> List[Dict]: + with self._connect() as conn: + rows = conn.execute( + "SELECT * FROM insight_runs ORDER BY id DESC LIMIT ?", (limit,) + ).fetchall() + return [dict(r) for r in rows] +``` + +- [ ] **Step 4: 运行测试确认通过** + +Run: `uv run pytest tests/insight/test_db.py -v` +Expected: PASS（5 passed） + +- [ ] **Step 5: 提交** + +```bash +git add insight/db.py tests/insight/test_db.py +git commit -m "feat(insight): InsightDB 运行记录与上游表计数" +``` + +--- + +## Task 3: `insight/runner.py` — job 转命令行参数 + +**Files:** +- Create: `insight/runner.py` +- Test: `tests/insight/test_runner_args.py` + +- [ ] **Step 1: 写失败测试** + +`tests/insight/test_runner_args.py`： +```python +# -*- coding: utf-8 -*- +import pytest + +from insight.runner import build_crawl_args + + +def test_search_job_args(): + job = {"name": "kw", "type": "search", "keywords": "a,b", "hour": 2} + args = build_crawl_args(job) + assert args[:6] == ["--platform", "xhs", "--type", "search", "--save_data_option", "sqlite"] + assert "--keywords" in args and args[args.index("--keywords") + 1] == "a,b" + assert args[args.index("--get_comment") + 1] == "true" + + +def test_detail_job_joins_note_ids(): + job = {"name": "w", "type": "detail", "note_ids": ["n1", "n2"], "hour": 3} + args = build_crawl_args(job) + assert "--type" in args and args[args.index("--type") + 1] == "detail" + assert args[args.index("--specified_id") + 1] == "n1,n2" + + +def test_creator_job_joins_creator_ids(): + job = {"name": "c", "type": "creator", "creator_ids": ["c1"], "hour": 4} + args = build_crawl_args(job) + assert args[args.index("--creator_id") + 1] == "c1" + + +def test_optional_max_comments_and_sub_comment(): + job = {"name": "k", "type": "search", "keywords": "x", "hour": 2, + "max_comments": 50, "get_sub_comment": True} + args = build_crawl_args(job) + assert args[args.index("--max_comments_count_singlenotes") + 1] == "50" + assert args[args.index("--get_sub_comment") + 1] == "true" + + +def test_invalid_type_raises(): + with pytest.raises(ValueError): + build_crawl_args({"name": "bad", "type": "homefeed", "hour": 1}) + + +def test_search_missing_keywords_raises(): + with pytest.raises(ValueError): + build_crawl_args({"name": "bad", "type": "search", "hour": 1}) + + +def test_detail_missing_note_ids_raises(): + with pytest.raises(ValueError): + build_crawl_args({"name": "bad", "type": "detail", "hour": 1}) + + +def test_creator_missing_ids_raises(): + with pytest.raises(ValueError): + build_crawl_args({"name": "bad", "type": "creator", "hour": 1}) +``` + +- [ ] **Step 2: 运行测试确认失败** + +Run: `uv run pytest tests/insight/test_runner_args.py -v` +Expected: FAIL（`ModuleNotFoundError: No module named 'insight.runner'`） + +- [ ] **Step 3: 写最小实现（仅 `build_crawl_args` 部分）`insight/runner.py`** + +```python +# -*- coding: utf-8 -*- +"""把 job 配置转成爬虫命令行参数，并以子进程方式运行爬虫。""" + +import os +import subprocess +import sys +from dataclasses import dataclass +from typing import List, Optional + +from insight import config as insight_config + +VALID_TYPES = {"search", "detail", "creator"} + + +def build_crawl_args(job: dict) -> List[str]: + """把一个 job dict 翻译成 `python -m insight.crawl_entry` 的参数列表。""" + name = job.get("name") + jtype = job.get("type") + if jtype not in VALID_TYPES: + raise ValueError(f"job {name!r}: invalid type {jtype!r}") + + args: List[str] = ["--platform", "xhs", "--type", jtype, "--save_data_option", "sqlite"] + + if jtype == "search": + keywords = job.get("keywords") + if not keywords: + raise ValueError(f"job {name!r}: search type requires 'keywords'") + args += ["--keywords", keywords] + elif jtype == "detail": + note_ids = job.get("note_ids") + if not note_ids: + raise ValueError(f"job {name!r}: detail type requires 'note_ids'") + args += ["--specified_id", ",".join(note_ids)] + elif jtype == "creator": + creator_ids = job.get("creator_ids") + if not creator_ids: + raise ValueError(f"job {name!r}: creator type requires 'creator_ids'") + args += ["--creator_id", ",".join(creator_ids)] + + args += ["--get_comment", "true"] + if job.get("get_sub_comment"): + args += ["--get_sub_comment", "true"] + if job.get("max_comments"): + args += ["--max_comments_count_singlenotes", str(job["max_comments"])] + + return args +``` + +- [ ] **Step 4: 运行测试确认通过** + +Run: `uv run pytest tests/insight/test_runner_args.py -v` +Expected: PASS（8 passed） + +- [ ] **Step 5: 提交** + +```bash +git add insight/runner.py tests/insight/test_runner_args.py +git commit -m "feat(insight): build_crawl_args 把 job 翻译为爬虫参数" +``` + +--- + +## Task 4: `insight/runner.py` — 子进程运行 `run_crawl` + +**Files:** +- Modify: `insight/runner.py`（追加 `CrawlResult` 与 `run_crawl`） +- Test: `tests/insight/test_runner_subprocess.py` + +- [ ] **Step 1: 写失败测试** + +`tests/insight/test_runner_subprocess.py`： +```python +# -*- coding: utf-8 -*- +import subprocess +import sys + +import insight.runner as runner + + +class _FakeCompleted: + def __init__(self, returncode, stderr=""): + self.returncode = returncode + self.stderr = stderr + + +def test_run_crawl_success_builds_command_and_env(monkeypatch): + captured = {} + + def fake_run(cmd, **kwargs): + captured["cmd"] = cmd + captured["kwargs"] = kwargs + return _FakeCompleted(0, stderr="") + + monkeypatch.setattr(subprocess, "run", fake_run) + + job = {"name": "kw", "type": "search", "keywords": "x", "hour": 2, "max_notes": 7} + result = runner.run_crawl(job) + + assert result.exit_code == 0 + assert result.timed_out is False + # 命令以当前解释器 + -m insight.crawl_entry 开头 + assert captured["cmd"][:3] == [sys.executable, "-m", "insight.crawl_entry"] + assert "--keywords" in captured["cmd"] + # max_notes 通过环境变量注入 + assert captured["kwargs"]["env"]["INSIGHT_MAX_NOTES"] == "7" + + +def test_run_crawl_nonzero_exit_returns_stderr_tail(monkeypatch): + monkeypatch.setattr(subprocess, "run", lambda cmd, **kw: _FakeCompleted(1, stderr="boom")) + job = {"name": "kw", "type": "search", "keywords": "x", "hour": 2} + result = runner.run_crawl(job) + assert result.exit_code == 1 + assert result.timed_out is False + assert "boom" in result.stderr_tail + + +def test_run_crawl_timeout(monkeypatch): + def fake_run(cmd, **kwargs): + raise subprocess.TimeoutExpired(cmd=cmd, timeout=1, stderr="late") + + monkeypatch.setattr(subprocess, "run", fake_run) + job = {"name": "kw", "type": "search", "keywords": "x", "hour": 2} + result = runner.run_crawl(job, timeout=1) + assert result.timed_out is True + assert result.exit_code == -1 +``` + +- [ ] **Step 2: 运行测试确认失败** + +Run: `uv run pytest tests/insight/test_runner_subprocess.py -v` +Expected: FAIL（`AttributeError: module 'insight.runner' has no attribute 'run_crawl'`） + +- [ ] **Step 3: 在 `insight/runner.py` 末尾追加实现** + +```python +@dataclass +class CrawlResult: + exit_code: int + timed_out: bool + stderr_tail: str + + +def run_crawl(job: dict, timeout: Optional[int] = None) -> CrawlResult: + """以子进程运行爬虫入口。max_notes 通过环境变量 INSIGHT_MAX_NOTES 注入。""" + timeout = timeout if timeout is not None else insight_config.SUBPROCESS_TIMEOUT + args = build_crawl_args(job) + cmd = [sys.executable, "-m", "insight.crawl_entry", *args] + + env = dict(os.environ) + if job.get("max_notes"): + env["INSIGHT_MAX_NOTES"] = str(job["max_notes"]) + + try: + proc = subprocess.run( + cmd, + cwd=insight_config.PROJECT_ROOT, + env=env, + timeout=timeout, + capture_output=True, + text=True, + encoding="utf-8", + errors="replace", + ) + except subprocess.TimeoutExpired as exc: + stderr = exc.stderr if isinstance(exc.stderr, str) else "" + return CrawlResult(exit_code=-1, timed_out=True, stderr_tail=stderr[-2000:]) + + stderr = proc.stderr or "" + return CrawlResult(exit_code=proc.returncode, timed_out=False, stderr_tail=stderr[-2000:]) +``` + +- [ ] **Step 4: 运行测试确认通过** + +Run: `uv run pytest tests/insight/test_runner_subprocess.py -v` +Expected: PASS（3 passed） + +- [ ] **Step 5: 提交** + +```bash +git add insight/runner.py tests/insight/test_runner_subprocess.py +git commit -m "feat(insight): run_crawl 子进程封装（超时/退出码/stderr）" +``` + +--- + +## Task 5: `insight/crawl_entry.py` — 子进程入口（env 覆盖 + 委托上游） + +**Files:** +- Create: `insight/crawl_entry.py` +- Test: `tests/insight/test_crawl_entry.py` + +- [ ] **Step 1: 写失败测试** + +`tests/insight/test_crawl_entry.py`： +```python +# -*- coding: utf-8 -*- +import config # 上游配置 +from insight.crawl_entry import apply_overrides + + +def test_apply_overrides_sets_max_notes(monkeypatch): + original = config.CRAWLER_MAX_NOTES_COUNT + try: + monkeypatch.setenv("INSIGHT_MAX_NOTES", "33") + apply_overrides() + assert config.CRAWLER_MAX_NOTES_COUNT == 33 + finally: + config.CRAWLER_MAX_NOTES_COUNT = original + + +def test_apply_overrides_noop_without_env(monkeypatch): + original = config.CRAWLER_MAX_NOTES_COUNT + try: + monkeypatch.delenv("INSIGHT_MAX_NOTES", raising=False) + apply_overrides() + assert config.CRAWLER_MAX_NOTES_COUNT == original + finally: + config.CRAWLER_MAX_NOTES_COUNT = original +``` + +- [ ] **Step 2: 运行测试确认失败** + +Run: `uv run pytest tests/insight/test_crawl_entry.py -v` +Expected: FAIL（`ModuleNotFoundError: No module named 'insight.crawl_entry'`） + +- [ ] **Step 3: 写实现 `insight/crawl_entry.py`** + +```python +# -*- coding: utf-8 -*- +"""爬虫子进程入口。 + +读取 INSIGHT_MAX_NOTES 等无 CLI 参数的覆盖项写入上游 config， +然后把其余 CLI 参数交给上游 main 处理。本文件位于 insight 包内， +不修改任何上游文件。 + +用法（由 insight.runner 以子进程调用）： + python -m insight.crawl_entry --platform xhs --type search --keywords ... --save_data_option sqlite +""" + +import os + + +def apply_overrides() -> None: + """把没有 CLI 参数的配置项从环境变量写入上游 config。""" + import config # 上游配置模块 + + max_notes = os.environ.get("INSIGHT_MAX_NOTES") + if max_notes: + config.CRAWLER_MAX_NOTES_COUNT = int(max_notes) + + +def main_entry() -> None: + apply_overrides() + # 复用上游的协程入口与清理逻辑（sys.argv 由上游 cmd_arg 解析） + from main import main as crawler_main, async_cleanup + from tools.app_runner import run + + run(crawler_main, async_cleanup, cleanup_timeout_seconds=15.0) + + +if __name__ == "__main__": + main_entry() +``` + +- [ ] **Step 4: 运行测试确认通过** + +Run: `uv run pytest tests/insight/test_crawl_entry.py -v` +Expected: PASS（2 passed） + +- [ ] **Step 5: 提交** + +```bash +git add insight/crawl_entry.py tests/insight/test_crawl_entry.py +git commit -m "feat(insight): crawl_entry 子进程入口（env 覆盖 max_notes 后委托上游）" +``` + +--- + +## Task 6: `insight/orchestrator.py` — 一次完整周期编排 + +**Files:** +- Create: `insight/orchestrator.py` +- Test: `tests/insight/test_orchestrator.py` + +- [ ] **Step 1: 写失败测试** + +`tests/insight/test_orchestrator.py`： +```python +# -*- coding: utf-8 -*- +import os + +import insight.orchestrator as orchestrator +import insight.runner as runner +from insight.db import InsightDB + + +def _db(tmp_path): + db = InsightDB(os.path.join(str(tmp_path), "t.db")) + db.init_schema() + return db + + +def test_run_job_success_records_success(tmp_path, monkeypatch): + db = _db(tmp_path) + monkeypatch.setattr( + runner, "run_crawl", + lambda job, timeout=None: runner.CrawlResult(exit_code=0, timed_out=False, stderr_tail=""), + ) + job = {"name": "kw", "type": "search", "keywords": "x", "hour": 2} + result = orchestrator.run_job(job, db=db) + + assert result["status"] == "success" + runs = db.recent_runs() + assert runs[0]["status"] == "success" + assert runs[0]["error_msg"] is None + + +def test_run_job_nonzero_records_error_with_stderr(tmp_path, monkeypatch): + db = _db(tmp_path) + monkeypatch.setattr( + runner, "run_crawl", + lambda job, timeout=None: runner.CrawlResult(exit_code=2, timed_out=False, stderr_tail="kaboom"), + ) + job = {"name": "kw", "type": "search", "keywords": "x", "hour": 2} + result = orchestrator.run_job(job, db=db) + + assert result["status"] == "error" + runs = db.recent_runs() + assert runs[0]["status"] == "error" + assert "kaboom" in runs[0]["error_msg"] + + +def test_run_job_timeout_records_timeout(tmp_path, monkeypatch): + db = _db(tmp_path) + monkeypatch.setattr( + runner, "run_crawl", + lambda job, timeout=None: runner.CrawlResult(exit_code=-1, timed_out=True, stderr_tail=""), + ) + job = {"name": "kw", "type": "search", "keywords": "x", "hour": 2} + result = orchestrator.run_job(job, db=db) + assert result["status"] == "timeout" + assert db.recent_runs()[0]["status"] == "timeout" +``` + +- [ ] **Step 2: 运行测试确认失败** + +Run: `uv run pytest tests/insight/test_orchestrator.py -v` +Expected: FAIL（`ModuleNotFoundError: No module named 'insight.orchestrator'`） + +- [ ] **Step 3: 写实现 `insight/orchestrator.py`** + +```python +# -*- coding: utf-8 -*- +"""一次完整调度周期：开始记录 → 计数前 → 跑爬虫 → 计数后 → 结束记录。""" + +from typing import Optional + +from insight import config, runner +from insight.db import InsightDB + + +def run_job(job: dict, db: Optional[InsightDB] = None) -> dict: + db = db or InsightDB(config.DB_PATH) + db.init_schema() + + run_id = db.start_run(job["name"], job["type"]) + notes_before = db.count_rows("xhs_note") + comments_before = db.count_rows("xhs_note_comment") + + result = runner.run_crawl(job) + + notes_after = db.count_rows("xhs_note") + comments_after = db.count_rows("xhs_note_comment") + notes_delta = max(0, notes_after - notes_before) + comments_delta = max(0, comments_after - comments_before) + + if result.timed_out: + status = "timeout" + elif result.exit_code == 0: + status = "success" + else: + status = "error" + + db.finish_run( + run_id, + exit_code=result.exit_code, + status=status, + notes_crawled=notes_delta, + comments_crawled=comments_delta, + error_msg=None if status == "success" else (result.stderr_tail or status), + ) + + return {"run_id": run_id, "status": status, "notes": notes_delta, "comments": comments_delta} +``` + +- [ ] **Step 4: 运行测试确认通过** + +Run: `uv run pytest tests/insight/test_orchestrator.py -v` +Expected: PASS（3 passed） + +- [ ] **Step 5: 提交** + +```bash +git add insight/orchestrator.py tests/insight/test_orchestrator.py +git commit -m "feat(insight): orchestrator.run_job 编排一次完整采集周期" +``` + +--- + +## Task 7: `insight/scheduler/daemon.py` — APScheduler 守护进程 + +**Files:** +- Create: `insight/scheduler/daemon.py` +- Test: `tests/insight/test_scheduler.py` + +> 本任务起的测试需要 apscheduler，运行命令统一加 `--with`。 + +- [ ] **Step 1: 写失败测试** + +`tests/insight/test_scheduler.py`： +```python +# -*- coding: utf-8 -*- +from apscheduler.schedulers.blocking import BlockingScheduler + +from insight.scheduler.daemon import build_scheduler + + +def test_build_scheduler_registers_one_job_per_config(): + jobs = [ + {"name": "kw_daily", "type": "search", "keywords": "x", "hour": 2, "minute": 0}, + {"name": "watch", "type": "detail", "note_ids": ["n1"], "hour": 3, "minute": 30}, + ] + sched = build_scheduler(jobs=jobs, scheduler=BlockingScheduler()) + ids = {j.id for j in sched.get_jobs()} + assert ids == {"kw_daily", "watch"} + + +def test_build_scheduler_sets_cron_hour_and_minute(): + jobs = [{"name": "watch", "type": "detail", "note_ids": ["n1"], "hour": 3, "minute": 30}] + sched = build_scheduler(jobs=jobs, scheduler=BlockingScheduler()) + job = sched.get_job("watch") + fields = {f.name: str(f) for f in job.trigger.fields} + assert fields["hour"] == "3" + assert fields["minute"] == "30" + + +def test_build_scheduler_defaults_minute_to_zero(): + jobs = [{"name": "kw", "type": "search", "keywords": "x", "hour": 5}] + sched = build_scheduler(jobs=jobs, scheduler=BlockingScheduler()) + job = sched.get_job("kw") + fields = {f.name: str(f) for f in job.trigger.fields} + assert fields["minute"] == "0" +``` + +- [ ] **Step 2: 运行测试确认失败** + +Run: `uv run --with "apscheduler>=3.10,<4" pytest tests/insight/test_scheduler.py -v` +Expected: FAIL（`ModuleNotFoundError: No module named 'insight.scheduler.daemon'`） + +- [ ] **Step 3: 写实现 `insight/scheduler/daemon.py`** + +```python +# -*- coding: utf-8 -*- +"""APScheduler 守护进程：按 cron 触发每个 job。""" + +from typing import List, Optional + +from apscheduler.schedulers.blocking import BlockingScheduler +from apscheduler.triggers.cron import CronTrigger + +from insight import config +from insight.orchestrator import run_job + + +def build_scheduler(jobs: Optional[List[dict]] = None, scheduler: Optional[BlockingScheduler] = None) -> BlockingScheduler: + jobs = jobs if jobs is not None else config.JOBS + scheduler = scheduler if scheduler is not None else BlockingScheduler() + for job in jobs: + trigger = CronTrigger(hour=job["hour"], minute=job.get("minute", 0)) + scheduler.add_job( + run_job, + trigger=trigger, + args=[job], + id=job["name"], + misfire_grace_time=config.MISFIRE_GRACE_TIME, + replace_existing=True, + ) + return scheduler + + +def main() -> None: + scheduler = build_scheduler() + print(f"[insight] scheduler started with {len(scheduler.get_jobs())} job(s). Ctrl+C to stop.") + try: + scheduler.start() + except (KeyboardInterrupt, SystemExit): + print("[insight] scheduler stopped.") + + +if __name__ == "__main__": + main() +``` + +- [ ] **Step 4: 运行测试确认通过** + +Run: `uv run --with "apscheduler>=3.10,<4" pytest tests/insight/test_scheduler.py -v` +Expected: PASS（3 passed） + +- [ ] **Step 5: 提交** + +```bash +git add insight/scheduler/daemon.py tests/insight/test_scheduler.py +git commit -m "feat(insight): APScheduler 守护进程，按 cron 触发各 job" +``` + +--- + +## Task 8: `insight/cli.py` — 命令行入口 + +**Files:** +- Create: `insight/cli.py` +- Test: `tests/insight/test_cli.py` + +> `run-daemon` 子命令对 apscheduler 的依赖采用**惰性导入**（在处理函数内 import），使 `crawl-once` / `status` 无需安装 apscheduler 即可使用。 + +- [ ] **Step 1: 写失败测试** + +`tests/insight/test_cli.py`： +```python +# -*- coding: utf-8 -*- +import pytest + +from insight.cli import build_parser, cmd_crawl_once, cmd_status + + +def test_parser_crawl_once(): + args = build_parser().parse_args(["crawl-once", "kw_daily"]) + assert args.name == "kw_daily" + assert args.func is cmd_crawl_once + + +def test_parser_status_default_limit(): + args = build_parser().parse_args(["status"]) + assert args.limit == 20 + assert args.func is cmd_status + + +def test_parser_status_custom_limit(): + args = build_parser().parse_args(["status", "--limit", "5"]) + assert args.limit == 5 + + +def test_parser_requires_subcommand(): + with pytest.raises(SystemExit): + build_parser().parse_args([]) +``` + +- [ ] **Step 2: 运行测试确认失败** + +Run: `uv run pytest tests/insight/test_cli.py -v` +Expected: FAIL（`ModuleNotFoundError: No module named 'insight.cli'`） + +- [ ] **Step 3: 写实现 `insight/cli.py`** + +```python +# -*- coding: utf-8 -*- +"""insight 命令行入口：crawl-once / run-daemon / status。""" + +import argparse +from typing import Optional, Sequence + +from insight import config +from insight.db import InsightDB +from insight.orchestrator import run_job + + +def _find_job(name: str) -> dict: + for job in config.JOBS: + if job["name"] == name: + return job + raise SystemExit(f"job not found: {name!r}. Known jobs: {[j['name'] for j in config.JOBS]}") + + +def cmd_crawl_once(args: argparse.Namespace) -> None: + job = _find_job(args.name) + result = run_job(job) + print(result) + + +def cmd_run_daemon(args: argparse.Namespace) -> None: + # 惰性导入：只有 run-daemon 需要 apscheduler + from insight.scheduler.daemon import main as run_daemon + run_daemon() + + +def cmd_status(args: argparse.Namespace) -> None: + db = InsightDB(config.DB_PATH) + db.init_schema() + runs = db.recent_runs(args.limit) + if not runs: + print("(no runs yet)") + return + for run in runs: + print( + f"#{run['id']} {run['job_name']} [{run['crawler_type']}] " + f"status={run['status']} exit={run['exit_code']} " + f"notes={run['notes_crawled']} comments={run['comments_crawled']}" + ) + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(prog="insight") + sub = parser.add_subparsers(dest="command", required=True) + + p_once = sub.add_parser("crawl-once", help="立即运行某个 job") + p_once.add_argument("name", help="config.JOBS 中的 job 名") + p_once.set_defaults(func=cmd_crawl_once) + + p_daemon = sub.add_parser("run-daemon", help="启动定时守护进程") + p_daemon.set_defaults(func=cmd_run_daemon) + + p_status = sub.add_parser("status", help="查看最近的运行记录") + p_status.add_argument("--limit", type=int, default=20) + p_status.set_defaults(func=cmd_status) + + return parser + + +def main(argv: Optional[Sequence[str]] = None) -> None: + args = build_parser().parse_args(argv) + args.func(args) + + +if __name__ == "__main__": + main() +``` + +- [ ] **Step 4: 运行测试确认通过** + +Run: `uv run pytest tests/insight/test_cli.py -v` +Expected: PASS（4 passed） + +- [ ] **Step 5: 全量回归 + 提交** + +Run: `uv run --with "apscheduler>=3.10,<4" pytest tests/insight -v` +Expected: 全部 PASS（约 28 个用例） + +```bash +git add insight/cli.py tests/insight/test_cli.py +git commit -m "feat(insight): cli 入口 crawl-once / run-daemon / status" +``` + +--- + +## Task 9: `insight/README.md` 与手动集成验证 + +**Files:** +- Create: `insight/README.md` + +- [ ] **Step 1: 写 `insight/README.md`** + +````markdown +# insight — 小红书评论定时采集（MediaCrawler 二次开发） + +本包是对 MediaCrawler 的二次开发，**不修改任何上游文件**，全部代码在 `insight/` 内。 +本期范围：定时爬取 + 原始数据入库（复用上游 SQLite）+ 运行日志。**不含文本分析**（后续迭代）。 + +## 一次性准备 + +```bash +# 1. 初始化上游 SQLite 表结构（创建 xhs_note / xhs_note_comment 等） +uv run python main.py --init_db sqlite + +# 2. 确保已登录小红书（CDP 模式，复用本机 Chrome 登录态） +# 参见项目根 README 的 Chrome 远程调试配置 +``` + +## 配置 + +编辑 `insight/config.py` 的 `JOBS` 列表（job 类型 search/detail/creator、关键词/笔记ID/创作者ID、触发时刻 hour/minute、max_notes 等）。 + +## 使用 + +```bash +# 立即跑一次某个 job（不依赖 apscheduler） +uv run python -m insight.cli crawl-once kw_daily + +# 查看最近运行记录（不依赖 apscheduler） +uv run python -m insight.cli status --limit 20 + +# 启动定时守护进程（前台运行，Ctrl+C 退出；需 apscheduler） +uv run --with "apscheduler>=3.10,<4" python -m insight.cli run-daemon +``` + +> 守护进程为前台常驻进程；本机重启后需手动重新启动。 + +## 运行测试 + +```bash +uv run --with "apscheduler>=3.10,<4" pytest tests/insight -v +``` + +## 与上游 MediaCrawler 同步更新 + +```bash +# 一次性：添加上游远程 +git remote add upstream https://github.com/NanmiCoder/MediaCrawler.git + +# 定期同步 +git fetch upstream +git merge upstream/main # 或 git rebase upstream/main +``` + +因为本包全部是 `insight/` 下的新增文件、未改动任何上游文件，合并几乎不会冲突。 +唯一与上游耦合的假设： +- 上游 CLI 参数（`--platform/--type/--keywords/--specified_id/--creator_id/--save_data_option` 等）保持不变； +- SQLite 路径仍为 `database/sqlite_tables.db`（见 `config/db_config.py`）； +- `main.main` / `main.async_cleanup` / `tools.app_runner.run` 接口保持不变。 +同步后若上述任一处变化，只需相应调整 `insight/runner.py`、`insight/config.py`、`insight/crawl_entry.py`。 +```` + +- [ ] **Step 2: 手动集成验证（需真实登录态，按需执行）** + +> 这些是人工验证步骤，不是自动化测试。若当前环境无法登录小红书，记录为「待人工验证」并跳过。 + +1. 初始化表：`uv run python main.py --init_db sqlite` → 预期打印 `Database sqlite initialized successfully.` +2. 跑一次小批量 job（先把 `kw_daily` 的 `max_notes` 临时设为 2）： + `uv run python -m insight.cli crawl-once kw_daily` + 预期：浏览器弹出/复用登录态完成爬取，命令结束打印形如 `{'run_id': 1, 'status': 'success', 'notes': N, 'comments': M}`。 +3. 查看运行记录：`uv run python -m insight.cli status` + 预期：看到一条 `status=success` 记录，`notes`/`comments` 与上一步一致。 +4. 验证守护进程可启动（把某 job 的 `hour/minute` 临时设为 1~2 分钟后的时刻）： + `uv run --with "apscheduler>=3.10,<4" python -m insight.cli run-daemon` + 预期：打印 `scheduler started with N job(s)`，到点自动触发并在 `status` 中新增记录；`Ctrl+C` 可优雅退出。 + +- [ ] **Step 3: 提交** + +```bash +git add insight/README.md +git commit -m "docs(insight): 使用说明与上游同步指南" +``` + +--- + +## Self-Review（计划编写者已执行） + +**1. Spec 覆盖核对（对照设计文档各节）：** +- §2 隔离策略 → 全部代码在 `insight/`，零改上游文件（Task 1-9）；同步工作流写入 README（Task 9）。✅ +- §3 目录结构 → Task 1 建包骨架，后续任务逐一落地各模块。✅ +- §4 数据流（调度→子进程→入库→记录）→ orchestrator（Task 6）+ scheduler（Task 7）+ crawl_entry（Task 5）。✅ +- §5.2 `insight_runs` 表 → schema.sql（Task 1）+ db.py（Task 2）。✅ +- §6 Job 配置形态 → config.JOBS（Task 1）+ build_crawl_args（Task 3）。✅ +- §7 运行环境（CDP 登录、常驻、依赖注入）→ README（Task 9）+ `--with` 注入策略（Task 7/8）。✅ +- §8 错误处理（超时/退出码/幂等）→ run_crawl（Task 4）+ orchestrator 状态判定（Task 6）。✅ +- §9 测试策略（runner/db/scheduler 隔离）→ Task 2-8 均含单测。✅ +- §11 待定项已在「背景事实」中定稿：max_notes 用 env 注入（crawl_entry）、依赖用 `uv run --with` 注入、计数用前后差值。✅ + +**2. 占位符扫描：** 无 TODO/TBD/「类似上文」。`config.JOBS` 中的 `"请替换为真实笔记ID"` 是面向用户的配置占位说明，非计划缺口。✅ + +**3. 类型/签名一致性核对：** +- `CrawlResult(exit_code, timed_out, stderr_tail)` 在 runner 定义（Task 4），orchestrator 与测试一致引用（Task 6）。✅ +- `InsightDB` 方法名 `init_schema/start_run/finish_run/count_rows/recent_runs` 在 db（Task 2）、orchestrator（Task 6）、cli（Task 8）一致。✅ +- `build_crawl_args` / `run_crawl(job, timeout=None)` 在 runner（Task 3/4）定义，runner 测试与 orchestrator monkeypatch 签名一致。✅ +- `build_scheduler(jobs=None, scheduler=None)` 在 daemon（Task 7）定义，测试一致。✅ +- `run_job(job, db=None)` 在 orchestrator（Task 6）定义，cli（Task 8）按 `run_job(job)` 调用（db 默认从 config.DB_PATH 构造），一致。✅ diff --git a/docs/superpowers/plans/2026-06-06-xhs-data-viewer.md b/docs/superpowers/plans/2026-06-06-xhs-data-viewer.md new file mode 100644 index 000000000..2db794ca0 --- /dev/null +++ b/docs/superpowers/plans/2026-06-06-xhs-data-viewer.md @@ -0,0 +1,722 @@ +# XHS Data Viewer Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** 在 `insight/viewer/` 下交付一个 Streamlit 本地小工具，可重复启动，用于查看 `database/sqlite_tables.db` 中的小红书笔记与评论。 + +**Architecture:** `insight/viewer/` 是一个新增子包，与 `insight/` 既有的爬取代码**互不依赖**。`data.py` 纯数据层（直接 `sqlite3`，不依赖 Streamlit，可单测）；`app.py` 仅做页面布局；`tests/insight/test_viewer_data.py` 仅覆盖 `data.py` 的纯函数。 + +**Tech Stack:** Python 3.11+、Streamlit（通过 `uv run --with streamlit` 临时拉取，**不**写入 `pyproject.toml`）、SQLite3（标准库）、pytest（已在上游 dev deps 中）。 + +**Spec:** [`docs/superpowers/specs/2026-06-06-xhs-data-viewer-design.md`](../specs/2026-06-06-xhs-data-viewer-design.md) + +--- + +## File Structure + +| 路径 | 状态 | 职责 | +|---|---|---| +| `insight/viewer/__init__.py` | Create | 空，仅作 Python 包 | +| `insight/viewer/data.py` | Create | 纯数据层：`load_notes()` / `load_comments(note_id)` / `format_ts(ts)` | +| `insight/viewer/app.py` | Create | Streamlit 入口：页面布局 + `st.session_state` | +| `insight/viewer/README.md` | Create | 启动与使用说明 | +| `tests/insight/test_viewer_data.py` | Create | 覆盖 `data.py` 的纯函数单测 | + +**不修改任何上游文件**，包括 `pyproject.toml` / `requirements.txt` / `insight/` 现有任何文件。 + +--- + +## Task 1: 建立包骨架 + +**Files:** +- Create: `insight/viewer/__init__.py` +- Create: `insight/viewer/data.py`（含函数 stub，raise NotImplementedError） +- Create: `tests/insight/test_viewer_data.py`（含一个 import 烟雾测试） + +- [ ] **Step 1: 创建 `insight/viewer/__init__.py`** + +```python +# -*- coding: utf-8 -*- +# 小红书数据查看器子包（基于 Streamlit） +# 仅依赖标准库 + streamlit（运行时由 uv --with 临时拉取） +``` + +- [ ] **Step 2: 创建 `insight/viewer/data.py`（仅 stub）** + +```python +# -*- coding: utf-8 -*- +"""纯数据层：从 database/sqlite_tables.db 读取小红书笔记与评论。 + +不导入 Streamlit，方便单测。 +""" +from __future__ import annotations + +import datetime +import sqlite3 +from pathlib import Path +from typing import Any + +# 与上游约定一致：数据库固定路径（不读 .env） +DB_PATH = Path(__file__).resolve().parents[2] / "database" / "sqlite_tables.db" + + +def format_ts(ts: int | None) -> str: + """将 Unix 时间戳（秒）格式化为 'YYYY-MM-DD HH:MM'。None/0 返回 '—'。""" + raise NotImplementedError + + +def load_notes() -> list[dict[str, Any]]: + """加载所有笔记，按发布时间 time 倒序。""" + raise NotImplementedError + + +def load_comments(note_id: str) -> list[dict[str, Any]]: + """加载指定笔记的评论，按 create_time 升序。""" + raise NotImplementedError +``` + +- [ ] **Step 3: 创建 `tests/insight/test_viewer_data.py`（含烟雾测试）** + +```python +# -*- coding: utf-8 -*- +"""insight.viewer.data 纯函数单元测试。 + +仅测 data.py；app.py 不做单测（Streamlit 启动成本/收益不划算）。 +""" +import pytest + +from insight.viewer.data import format_ts, load_comments, load_notes + + +def test_imports_smoke(): + """烟雾测试：所有公开函数可导入。""" + assert callable(format_ts) + assert callable(load_notes) + assert callable(load_comments) +``` + +- [ ] **Step 4: 运行烟雾测试** + +Run: `uv run pytest tests/insight/test_viewer_data.py -v` +Expected: PASS（1 passed） + +- [ ] **Step 5: 提交** + +```bash +git add insight/viewer/__init__.py insight/viewer/data.py tests/insight/test_viewer_data.py +git commit -m "feat(viewer): 包骨架与 data.py stub（含烟雾测试）" +``` + +--- + +## Task 2: `data.format_ts`（TDD） + +**Files:** +- Modify: `insight/viewer/data.py` +- Modify: `tests/insight/test_viewer_data.py` + +- [ ] **Step 1: 追加 `test_format_ts` 失败用例** + +在 `tests/insight/test_viewer_data.py` 末尾添加： + +```python +def test_format_ts_normal_timestamp(): + """正常时间戳格式化。""" + # 2024-03-15 12:34:00 UTC+8 = 1710474840 + assert format_ts(1710474840) == "2024-03-15 12:34" + + +def test_format_ts_none_returns_dash(): + """None 返回 '—'。""" + assert format_ts(None) == "—" + + +def test_format_ts_zero_returns_dash(): + """0（未设置）返回 '—'。""" + assert format_ts(0) == "—" +``` + +- [ ] **Step 2: 跑测试，验证失败** + +Run: `uv run pytest tests/insight/test_viewer_data.py::test_format_ts_normal_timestamp -v` +Expected: FAIL with `NotImplementedError` + +- [ ] **Step 3: 实现 `format_ts`** + +替换 `insight/viewer/data.py` 中的 `format_ts` 函数体： + +```python +def format_ts(ts: int | None) -> str: + """将 Unix 时间戳（秒）格式化为 'YYYY-MM-DD HH:MM'。None/0 返回 '—'。""" + if not ts: + return "—" + return datetime.datetime.fromtimestamp(ts).strftime("%Y-%m-%d %H:%M") +``` + +- [ ] **Step 4: 跑测试，验证通过** + +Run: `uv run pytest tests/insight/test_viewer_data.py -v` +Expected: PASS（4 passed：含 Task 1 的烟雾测试 + 3 个新测试） + +- [ ] **Step 5: 提交** + +```bash +git add insight/viewer/data.py tests/insight/test_viewer_data.py +git commit -m "feat(viewer): format_ts 时间戳格式化（含 None/0 兜底）" +``` + +--- + +## Task 3: `data.load_notes`（TDD） + +**Files:** +- Modify: `insight/viewer/data.py` +- Modify: `tests/insight/test_viewer_data.py` + +> **设计说明**：`load_notes()` 接收一个 `db_path: Path | None = None` 参数；为 None 时使用 `DB_PATH`。这样测试可以注入临时数据库，生产代码走默认路径。 + +- [ ] **Step 1: 调整 `data.py` 让 `load_notes` 接受 `db_path` 参数，并把 stub 改成待实现** + +修改 `insight/viewer/data.py` 的 `load_notes` 签名（函数体暂时保留 raise）： + +```python +def load_notes(db_path: Path | None = None) -> list[dict[str, Any]]: + """加载所有笔记，按发布时间 time 倒序。db_path=None 时使用 DB_PATH。""" + raise NotImplementedError +``` + +- [ ] **Step 2: 追加失败用例（用临时 SQLite 灌假数据）** + +在 `tests/insight/test_viewer_data.py` 末尾添加： + +```python +def _make_xhs_db(path) -> None: + """构造一个最小的 xhs_note 测试库。""" + import sqlite3 + conn = sqlite3.connect(str(path)) + cur = conn.cursor() + cur.execute( + """ + CREATE TABLE xhs_note ( + note_id VARCHAR(255) PRIMARY KEY, + title TEXT, + liked_count TEXT, + comment_count TEXT, + time BIGINT, + source_keyword TEXT, + nickname TEXT, + desc TEXT + ) + """ + ) + cur.executemany( + "INSERT INTO xhs_note VALUES (?,?,?,?,?,?,?,?)", + [ + ("n1", "春日穿搭", "1200", "38", 1710474840, "穿搭", "博主A", "正文1"), + ("n2", "护肤心得", "856", "12", 1710388440, "护肤", "博主B", "正文2"), + ("n3", "无时间戳笔记", "0", "0", 0, "", "博主C", ""), + ], + ) + conn.commit() + conn.close() + + +def test_load_notes_returns_sorted_by_time_desc(tmp_path): + """load_notes 按 time 倒序，time=0 的排最后。""" + from insight.viewer.data import load_notes + db = tmp_path / "t.db" + _make_xhs_db(db) + + notes = load_notes(db_path=db) + + assert len(notes) == 3 + # 倒序：n1 (1710474840) > n2 (1710388440) > n3 (0) + assert [n["note_id"] for n in notes] == ["n1", "n2", "n3"] + assert notes[0]["title"] == "春日穿搭" + assert notes[0]["source_keyword"] == "穿搭" + + +def test_load_notes_missing_table_raises(tmp_path): + """xhs_note 表不存在时抛 OperationalError（让 UI 捕获后展示）。""" + import sqlite3 + from insight.viewer.data import load_notes + db = tmp_path / "empty.db" + # 不建表 + with pytest.raises(sqlite3.OperationalError): + load_notes(db_path=db) +``` + +- [ ] **Step 3: 跑失败用例** + +Run: `uv run pytest tests/insight/test_viewer_data.py::test_load_notes_returns_sorted_by_time_desc -v` +Expected: FAIL with `NotImplementedError` + +- [ ] **Step 4: 实现 `load_notes`** + +替换 `insight/viewer/data.py` 中的 `load_notes` 函数体： + +```python +def load_notes(db_path: Path | None = None) -> list[dict[str, Any]]: + """加载所有笔记，按发布时间 time 倒序。db_path=None 时使用 DB_PATH。""" + path = db_path if db_path is not None else DB_PATH + conn = sqlite3.connect(str(path)) + conn.row_factory = sqlite3.Row + try: + cur = conn.cursor() + cur.execute( + "SELECT note_id, title, liked_count, comment_count, time, " + "source_keyword, nickname, desc " + "FROM xhs_note ORDER BY time DESC" + ) + return [dict(row) for row in cur.fetchall()] + finally: + conn.close() +``` + +- [ ] **Step 5: 跑测试，验证通过** + +Run: `uv run pytest tests/insight/test_viewer_data.py -v` +Expected: PASS（6 passed：1 烟雾 + 3 format_ts + 2 load_notes） + +- [ ] **Step 6: 提交** + +```bash +git add insight/viewer/data.py tests/insight/test_viewer_data.py +git commit -m "feat(viewer): load_notes 按发布时间倒序读取 xhs_note" +``` + +--- + +## Task 4: `data.load_comments`（TDD） + +**Files:** +- Modify: `insight/viewer/data.py` +- Modify: `tests/insight/test_viewer_data.py` + +- [ ] **Step 1: 追加失败用例** + +在 `tests/insight/test_viewer_data.py` 末尾添加： + +```python +def _make_xhs_db_with_comments(path) -> None: + """构造包含 xhs_note + xhs_note_comment 的测试库。""" + import sqlite3 + conn = sqlite3.connect(str(path)) + cur = conn.cursor() + cur.execute( + """ + CREATE TABLE xhs_note ( + note_id VARCHAR(255) PRIMARY KEY, + title TEXT, liked_count TEXT, comment_count TEXT, + time BIGINT, source_keyword TEXT, nickname TEXT, desc TEXT + ) + """ + ) + cur.execute( + """ + CREATE TABLE xhs_note_comment ( + comment_id VARCHAR(255) PRIMARY KEY, + note_id VARCHAR(255), + nickname TEXT, + content TEXT, + like_count TEXT, + create_time BIGINT + ) + """ + ) + cur.executemany("INSERT INTO xhs_note VALUES (?,?,?,?,?,?,?,?)", + [("n1","A","0","0",0,"","",""), ("n2","B","0","0",0,"","","")]) + cur.executemany( + "INSERT INTO xhs_note_comment VALUES (?,?,?,?,?,?)", + [ + ("c1", "n1", "用户1", "第一条", "10", 1710475000), + ("c2", "n1", "用户2", "第二条", "5", 1710476000), + ("c3", "n2", "用户3", "另一条笔记的评论", "0", 1710477000), + ], + ) + conn.commit() + conn.close() + + +def test_load_comments_filters_by_note_id(tmp_path): + """load_comments 仅返回指定 note_id 的评论。""" + from insight.viewer.data import load_comments + db = tmp_path / "t.db" + _make_xhs_db_with_comments(db) + + rows = load_comments("n1", db_path=db) + + assert len(rows) == 2 + assert {r["comment_id"] for r in rows} == {"c1", "c2"} + # 默认按 create_time 升序 + assert [r["comment_id"] for r in rows] == ["c1", "c2"] + + +def test_load_comments_empty_when_no_match(tmp_path): + """不存在的 note_id 返回空列表，不抛异常。""" + from insight.viewer.data import load_comments + db = tmp_path / "t.db" + _make_xhs_db_with_comments(db) + + rows = load_comments("nonexistent", db_path=db) + + assert rows == [] +``` + +- [ ] **Step 2: 跑失败用例** + +Run: `uv run pytest tests/insight/test_viewer_data.py::test_load_comments_filters_by_note_id -v` +Expected: FAIL with `NotImplementedError` + +- [ ] **Step 3: 实现 `load_comments`** + +替换 `insight/viewer/data.py` 中的 `load_comments` 函数体： + +```python +def load_comments(note_id: str, db_path: Path | None = None) -> list[dict[str, Any]]: + """加载指定笔记的评论，按 create_time 升序。db_path=None 时使用 DB_PATH。""" + path = db_path if db_path is not None else DB_PATH + conn = sqlite3.connect(str(path)) + conn.row_factory = sqlite3.Row + try: + cur = conn.cursor() + cur.execute( + "SELECT comment_id, note_id, nickname, content, like_count, create_time " + "FROM xhs_note_comment WHERE note_id = ? ORDER BY create_time ASC", + (note_id,), + ) + return [dict(row) for row in cur.fetchall()] + finally: + conn.close() +``` + +- [ ] **Step 4: 跑全部测试，验证通过** + +Run: `uv run pytest tests/insight/test_viewer_data.py -v` +Expected: PASS（8 passed：1 烟雾 + 3 format_ts + 2 load_notes + 2 load_comments） + +- [ ] **Step 5: 提交** + +```bash +git add insight/viewer/data.py tests/insight/test_viewer_data.py +git commit -m "feat(viewer): load_comments 按 note_id 过滤 + create_time 升序" +``` + +--- + +## Task 5: `app.py` Streamlit UI + +**Files:** +- Create: `insight/viewer/app.py` + +> **设计说明**：本 Task **不写单测**（Streamlit 启动成本/收益不划算），通过 Task 7 的端到端手动烟雾测试验证。 + +- [ ] **Step 1: 创建 `insight/viewer/app.py`** + +```python +# -*- coding: utf-8 -*- +"""小红书数据查看器：Streamlit 单页 UI。 + +页面布局（自上而下）： + 1. 顶部统计 + 刷新按钮 + 2. 笔记列表（st.dataframe，自带列排序） + 3. 选中笔记的详情面板 + 评论列表 + +启动： + uv run --with streamlit streamlit run insight/viewer/app.py +""" +from __future__ import annotations + +import sqlite3 +from pathlib import Path + +import pandas as pd +import streamlit as st + +from insight.viewer.data import ( + DB_PATH, + format_ts, + load_comments, + load_notes, +) + + +# ---------- 启动配置 ---------- +st.set_page_config(page_title="小红书数据查看", layout="wide") + + +# ---------- 数据加载（带缓存）---------- +@st.cache_data(ttl=60) +def _cached_notes() -> list[dict]: + return load_notes() + + +@st.cache_data(ttl=60) +def _cached_comments(note_id: str) -> list[dict]: + return load_comments(note_id) + + +def _load_notes_safely() -> list[dict] | None: + """加载笔记，错误以 None 返回并用 st.error 提示。""" + if not DB_PATH.exists(): + st.error( + f"未找到数据库 {DB_PATH}，请先运行爬虫。" + ) + return None + try: + return _cached_notes() + except sqlite3.OperationalError: + st.error("数据库正忙（爬取进程可能正在写入），请稍后重试。") + return None + + +def _load_comments_safely(note_id: str) -> list[dict] | None: + try: + return _cached_comments(note_id) + except sqlite3.OperationalError: + st.error("数据库正忙，请稍后重试。") + return None + + +# ---------- 顶部 ---------- +st.title("小红书数据查看") + +notes = _load_notes_safely() +if notes is None: + st.stop() + +col_stat, col_btn = st.columns([4, 1]) +with col_stat: + total_comments = sum(int(n.get("comment_count") or 0) for n in notes) + st.caption(f"共 {len(notes)} 条笔记 / {total_comments} 条评论（按 comment_count 估算）") +with col_btn: + if st.button("🔄 刷新", use_container_width=True): + st.cache_data.clear() + st.rerun() + + +# ---------- 笔记列表 ---------- +if not notes: + st.info("暂无数据") + st.stop() + +# 准备表格数据：加一列"发布时间"已格式化 +df = pd.DataFrame( + [ + { + "#": idx + 1, + "标题": n.get("title") or "—", + "点赞": n.get("liked_count") or "—", + "评论数": n.get("comment_count") or "—", + "关键词": n.get("source_keyword") or "—", + "发布时间": format_ts(n.get("time")), + "note_id": n["note_id"], + } + for idx, n in enumerate(notes) + ] +) + +st.subheader("笔记列表") +st.caption("点击下方'查看'按钮查看评论") +event = st.dataframe( + df[["#", "标题", "点赞", "评论数", "关键词", "发布时间"]], + use_container_width=True, + hide_index=True, + on_select="rerun", + selection_mode="single-row", + key="notes_table", +) + +# Streamlit 1.32+：用户点击某行后 event.selection.rows 包含选中行号（0-based） +selected_rows = getattr(event, "selection", None) +selected_idx = selected_rows.rows[0] if selected_rows and selected_rows.rows else None + + +# ---------- 详情面板 ---------- +if selected_idx is None: + st.info("👆 在表格中选中一行后查看笔记详情与评论") + st.stop() + +selected_note = notes[selected_idx] + +st.divider() +st.subheader(f"📝 笔记详情：{selected_note.get('title') or '—'}") +detail_cols = st.columns(3) +with detail_cols[0]: + st.caption(f"**作者**：{selected_note.get('nickname') or '—'}") +with detail_cols[1]: + st.caption(f"**发布时间**：{format_ts(selected_note.get('time'))}") +with detail_cols[2]: + st.caption(f"**关键词**：{selected_note.get('source_keyword') or '—'}") + +st.markdown("**正文**") +st.markdown(selected_note.get("desc") or "—") + +# ---------- 评论 ---------- +st.divider() +st.subheader("💬 评论") +comments = _load_comments_safely(selected_note["note_id"]) +if comments is None: + st.stop() +if not comments: + st.info("该笔记暂无评论") + st.stop() + +# 准备评论表格 +cdf = pd.DataFrame( + [ + { + "点赞": c.get("like_count") or "—", + "用户": c.get("nickname") or "—", + "内容": c.get("content") or "—", + "时间": format_ts(c.get("create_time")), + } + for c in comments + ] +) +st.dataframe(cdf, use_container_width=True, hide_index=True) +st.caption(f"共 {len(comments)} 条评论") +``` + +- [ ] **Step 2: 静态语法检查（不启动 Streamlit）** + +Run: `uv run python -c "import ast; ast.parse(open('insight/viewer/app.py', encoding='utf-8').read()); print('OK')"` +Expected: `OK` + +- [ ] **Step 3: 提交** + +```bash +git add insight/viewer/app.py +git commit -m "feat(viewer): app.py Streamlit 单页 UI（笔记列表+详情+评论）" +``` + +--- + +## Task 6: `README.md` 启动说明 + +**Files:** +- Create: `insight/viewer/README.md` + +- [ ] **Step 1: 创建 README** + +```markdown +# 小红书数据查看器（XHS Data Viewer） + +基于 Streamlit 的本地查看工具，复用上游 `database/sqlite_tables.db`。 +仅做基础查看：笔记列表 + 点进去看评论。不修改任何上游文件。 + +## 一次性准备 + +```bash +# 确保上游 SQLite 表结构已创建 +uv run python main.py --init_db sqlite +``` + +## 启动 + +从项目根目录执行： + +```bash +uv run --with streamlit streamlit run insight/viewer/app.py +``` + +- 自动打开浏览器 `http://localhost:8501` +- 数据通过 `database/sqlite_tables.db` 实时读取，**不修改任何数据** +- 默认 60 秒缓存；点页面右上 🔄 立即重读 +- Ctrl+C 停止 + +## 界面 + +``` +顶部： 共 N 条笔记 / M 条评论 [🔄 刷新] +中段： 笔记列表（按发布时间倒序，可点击选中） +底段： 选中笔记的详情（作者/时间/关键词/正文）+ 评论列表 +``` + +## 错误提示 + +| 情况 | 提示 | +|---|---| +| 数据库文件不存在 | 未找到数据库 ...，请先运行爬虫 | +| 数据库被爬取进程持锁 | 数据库正忙，请稍后重试 | +| `xhs_note` 表不存在 | 数据库缺少表 xhs_note，请先执行 `uv run python main.py --init_db sqlite` | +| 笔记表为空 | 暂无数据 | +| 选中笔记无评论 | 该笔记暂无评论 | + +## 依赖 + +- Python 3.11+ +- Streamlit（通过 `uv run --with streamlit` 临时拉取，**不**写入 `pyproject.toml` / `requirements.txt`） + +## 相关 + +- 设计文档：[`docs/superpowers/specs/2026-06-06-xhs-data-viewer-design.md`](../../specs/2026-06-06-xhs-data-viewer-design.md) +- 数据来源：[`insight/` 爬取包](../../README.md) +``` + +- [ ] **Step 2: 提交** + +```bash +git add insight/viewer/README.md +git commit -m "docs(viewer): README 启动与使用说明" +``` + +--- + +## Task 7: 端到端烟雾测试 + +**Files:** 无新增（验证所有交付物） + +- [ ] **Step 1: 跑全部单元测试** + +Run: `uv run pytest tests/insight/test_viewer_data.py -v` +Expected: PASS（8 passed） + +- [ ] **Step 2: 验证 data.py 在真实数据库上可读** + +Run: `uv run python -c "from insight.viewer.data import load_notes; print(len(load_notes()))"` +Expected: 打印一个数字 ≥ 1（当前生产库是 44 条） + +- [ ] **Step 3: 验证 Streamlit 脚本可启动（5 秒后 Ctrl+C 退出）** + +Run: +```bash +timeout 8 uv run --with streamlit streamlit run insight/viewer/app.py +# 或 Windows PowerShell： +# Start-Process -NoNewWindow uv -ArgumentList "run","--with","streamlit","streamlit","run","insight/viewer/app.py" -RedirectStandardOutput stdout.log +# Start-Sleep -Seconds 5 +# Get-Process streamlit | Stop-Process -Force +``` + +Expected: 输出含 `You can now view your Streamlit app in your browser.` 后无致命错误；5–8 秒后被 timeout 杀掉 + +- [ ] **Step 4: 视觉验证（手动，10 秒）** + +1. 浏览器自动打开 `http://localhost:8501` +2. 顶部统计正确显示笔记/评论数 +3. 中段表格显示笔记列表（按时间倒序） +4. 选中某行 → 下方显示该笔记详情 + 评论列表 +5. 点 🔄 → 页面无错 + +- [ ] **Step 5: 提交（如有改动；否则跳过）** + +```bash +git status +# 若无改动： +echo "No changes to commit" +# 若有改动： +git add -A && git commit -m "chore(viewer): end-to-end smoke test cleanup" +``` + +--- + +## 验收清单 + +完成后应满足： + +- [x] `insight/viewer/{__init__,app,data}.py` 与 `README.md` 均已创建 +- [x] `tests/insight/test_viewer_data.py` 8 个用例全过 +- [x] `uv run --with streamlit streamlit run insight/viewer/app.py` 可启动并展示真实数据 +- [x] **未修改** MediaCrawler 任何上游文件（`git diff upstream/main -- '*.py' '*.toml'` 应仅显示 `insight/viewer/` 与 `tests/insight/test_viewer_data.py` 与 `docs/superpowers/*`） +- [x] `pyproject.toml` / `requirements.txt` 中**无** streamlit 条目 +- [x] 6 个 commit，提交信息遵循现有风格 diff --git a/docs/superpowers/specs/2026-06-05-xhs-insight-pipeline-design.md b/docs/superpowers/specs/2026-06-05-xhs-insight-pipeline-design.md new file mode 100644 index 000000000..7b6275c75 --- /dev/null +++ b/docs/superpowers/specs/2026-06-05-xhs-insight-pipeline-design.md @@ -0,0 +1,148 @@ +# 小红书评论采集与定时管道（XHS Insight Pipeline）设计文档 + +- 日期：2026-06-05 +- 状态：已确认设计，待编写实现计划 +- 作者：Jake (chuangjieren@gmail.com) + +## 1. 背景与目标 + +基于 MediaCrawler（本仓库为 `DNNinfo/MediaCrawler` fork）做二次开发，实现： + +1. **定时**爬取小红书（XHS）的笔记与评论原始数据。 +2. 把原始数据落入 **SQLite**，供后续分析使用。 +3. 保持本项目代码与上游 MediaCrawler **可持续同步更新**，同时**不影响**自己开发的代码。 + +> 本期范围**仅包含**「定时爬取 + 原始数据入库 + 运行日志」。**不包含 LLM / 文本分析**。文本分析（计划使用本地 Ollama + Qwen）作为后续迭代，在不改动本期代码的前提下扩展。 + +## 2. 核心隔离策略（最重要的约束） + +- 自研代码全部放入**单一新增顶层包 `insight/`**。 +- **不修改** MediaCrawler 既有任何文件（`main.py`、`media_platform/`、`store/`、`database/`、`config/` 等保持上游原样）。 +- `insight/` 通过两种方式与上游交互，均为「只读 / 旁路」： + - **调用**：以**子进程**方式运行 `uv run main.py …`。 + - **读取**：读取爬虫写入的 SQLite 表（`xhs_note` / `xhs_note_comment`）。 +- Git 层面：新增 `upstream` 远程，定期 `fetch` + `merge`/`rebase`。由于自研代码都是 `insight/` 下的**新增文件**，几乎不会与上游产生冲突。 + +### Git 同步工作流 + +```bash +# 一次性 +git remote add upstream https://github.com/NanmiCoder/MediaCrawler.git + +# 定期同步 +git fetch upstream +git merge upstream/main # 或 git rebase upstream/main +``` + +- 不在上游文件中写任何自研配置；所有自研配置在 `insight/config.py` 自行声明。 +- `insight/` 产生的数据/缓存通过 `.gitignore` 忽略（若需改 `.gitignore`，仅追加自研条目，尽量减少与上游接触面）。 +- `insight/README.md` 记录上述同步步骤备忘。 + +## 3. 目录结构 + +``` +MediaCrawler/ # 上游，零改动 +├─ main.py, media_platform/, store/, database/, config/ … +└─ insight/ # ← 自研全部代码 + ├─ __init__.py + ├─ config.py # db 路径、job 定义、爬虫参数 + ├─ cli.py # 入口：crawl-once / run-daemon / status + ├─ runner.py # 子进程封装 uv run main.py … + ├─ db.py # 写/读 insight_runs；按需读 xhs_* 校验数据 + ├─ schema.sql # insight_runs 表定义 + ├─ scheduler/ + │ ├─ __init__.py + │ └─ daemon.py # APScheduler，每日触发 + └─ README.md # 上游同步步骤 + 使用说明 + +tests/ +└─ insight/ # 自研测试，独立目录 +``` + +## 4. 数据流（一次调度周期） + +1. **APScheduler**（`insight/scheduler/daemon.py`）到点触发某个 job。 +2. **runner.py** 启动子进程：`uv run main.py --platform xhs --type search|detail|creator …`，并强制 `SAVE_DATA_OPTION=sqlite`（通过命令行/环境变量传入，不改上游配置文件）。 +3. 爬虫将原始数据写入上游既有表 `xhs_note` / `xhs_note_comment`。 +4. **db.py** 在 `insight_runs` 记录本次运行：job 名、开始/结束时间、子进程退出码、爬取条数、状态/错误信息。 +5. 周期结束。原始数据留存于 SQLite，供后续分析迭代使用。 + +> 设计上爬取与记录是可独立运行的步骤：`insight crawl-once ` 可手动跑单次；`insight run-daemon` 启动常驻调度；`insight status` 查看最近运行记录。 + +## 5. 数据模型 + +### 5.1 上游既有表（只读，不改） + +- `xhs_note`：`note_id`、`title`、`desc`、`source_keyword`、`time`、`liked_count`、`comment_count`、`collected_count` 等。 +- `xhs_note_comment`：`comment_id`、`note_id`、`content`、`create_time`、`like_count`、`parent_comment_id`、`sub_comment_count`、`ip_location`、`nickname`、`user_id` 等。 + +### 5.2 自研表（`insight/schema.sql`，本期唯一新增表） + +**`insight_runs`** — 每次调度周期一行： + +| 字段 | 类型 | 说明 | +|------|------|------| +| `id` | INTEGER PK | 自增主键 | +| `job_name` | TEXT | 对应 `config.JOBS[].name` | +| `crawler_type` | TEXT | search / detail / creator | +| `started_ts` | INTEGER | 开始时间戳 | +| `finished_ts` | INTEGER | 结束时间戳（可空，运行中为空） | +| `exit_code` | INTEGER | 子进程退出码（可空） | +| `notes_crawled` | INTEGER | 本次新增/涉及笔记数（尽力统计） | +| `comments_crawled` | INTEGER | 本次新增/涉及评论数（尽力统计） | +| `status` | TEXT | running / success / error / timeout | +| `error_msg` | TEXT | 失败信息（可空） | + +> 表名以 `insight_` 前缀与上游表隔离。后续分析迭代时新增的表同样使用该前缀。 + +## 6. Job 配置（`insight/config.py` 示例形态） + +```python +# SQLite 路径（与爬虫共用同一文件） +DB_PATH = "./data/mediacrawler.db" + +# 子进程超时（秒） +SUBPROCESS_TIMEOUT = 1800 + +# 调度任务定义 +JOBS = [ + {"name": "kw_daily", "type": "search", "keywords": "编程副业,编程兼职", "hour": 2, "max_notes": 20}, + {"name": "watch_notes", "type": "detail", "note_ids": ["xxx", "yyy"], "hour": 3}, + {"name": "creator_daily", "type": "creator", "creator_ids": ["zzz"], "hour": 4}, +] +``` + +- 每个 job 映射为一次 `main.py` 调用；`type`→`--type`，其余字段→对应命令行参数/环境变量。 +- 默认每日触发（`hour` 指定时刻）。 + +## 7. 运行环境与前置条件 + +- **登录态**：XHS 需扫码登录。调度运行在本机已登录的 Chrome（CDP 模式，`ENABLE_CDP_MODE=True`）。登录失效时子进程失败并记入 `insight_runs`，daemon 不崩溃。 +- **常驻进程**：APScheduler 为进程内守护（B 方案）。本机重启后需**手动重启** daemon。 +- **依赖**：APScheduler 作为自研依赖引入（评估加入 `pyproject.toml` / 单独 requirements，尽量不污染上游依赖声明，具体在实现计划中确定）。 + +## 8. 错误处理 + +- **子进程**：检查退出码 + 超时（`SUBPROCESS_TIMEOUT`）。失败 → `insight_runs` 记 `error`/`timeout`，不影响其他 job。 +- **登录失效**：子进程失败被捕获并记录，等待下次调度或人工介入。 +- **幂等**：重复运行安全——上游 SQLite 存储按主键/唯一键去重。 +- **错过触发**：APScheduler 使用 `misfire_grace_time` 容忍短暂错过。 + +## 9. 测试策略 + +- `runner`：mock subprocess，验证命令拼装、超时与退出码处理。 +- `db`：临时 SQLite，验证 `insight_runs` 写入/查询。 +- `scheduler`：可控触发，验证 job → runner → 日志 串联，以及单 job 失败不影响其他 job。 +- 测试位于 `tests/insight/`，与上游测试隔离。 + +## 10. 后续迭代（本期不实现，仅预留） + +- 在 `insight/analysis/` 增加本地 Ollama（Qwen）分析模块：逐评论结构化打标 + 逐笔记聚合简报。 +- 新增分析结果表（`insight_*` 前缀），按 `comment_id` 左连接做增量分析。 +- 这些扩展均为新增文件，不改动本期代码，符合隔离策略。 + +## 11. 未决/待实现计划阶段确认的细节 + +- APScheduler 依赖的具体引入方式（pyproject vs 独立 requirements）。 +- `notes_crawled` / `comments_crawled` 的统计方式（运行前后对表计数差值）。 +- `SAVE_DATA_OPTION=sqlite` 的传参方式（命令行参数 vs 环境变量 vs 临时配置覆盖）——需在实现时确认上游 `main.py` 支持的覆盖手段，仍以「不改上游」为前提。 diff --git a/docs/superpowers/specs/2026-06-06-xhs-data-viewer-design.md b/docs/superpowers/specs/2026-06-06-xhs-data-viewer-design.md new file mode 100644 index 000000000..060f2f3f8 --- /dev/null +++ b/docs/superpowers/specs/2026-06-06-xhs-data-viewer-design.md @@ -0,0 +1,186 @@ +# 小红书数据查看器（XHS Data Viewer）设计文档 + +- 日期：2026-06-06 +- 状态：已确认设计，待编写实现计划 +- 作者：Jake (chuangjieren@gmail.com) +- 关联：本文是 [2026-06-05-xhs-insight-pipeline-design.md](./2026-06-05-xhs-insight-pipeline-design.md) 的下游配套（数据查看器） + +## 1. 背景与目标 + +上一期「XHS Insight Pipeline」已经能定时把小红书笔记与评论爬到 `database/sqlite_tables.db`。当前数据规模：44 条 `xhs_note`、435 条 `xhs_note_comment`、10 条 `insight_runs`。 + +本期目标：**提供一个可重复启动的本地小工具，让人能快速查看这些已爬到的数据**。 + +- **使用方式**：每次爬完一批数据后，执行一个命令 → 浏览器里看到笔记列表 + 点进去看评论。 +- **范围**：基础查看即可。**不**做搜索、筛选、导出、采集运行面板等。 +- **平台约束**：Windows；实现要简单。 + +## 2. 核心隔离策略 + +延续上期「不修改任何上游文件」的原则： + +- 自研代码全部放入**新增子包 `insight/viewer/`**。 +- **不修改** MediaCrawler 既有任何文件，不修改 `insight/` 中既有的 `cli.py` / `runner.py` / `orchestrator.py` / `db.py` / `config.py` / `crawl_entry.py` 等。 +- 只读 `database/sqlite_tables.db`，**不写**任何表（`insight_runs` 也只读）。 +- 与 `insight/` 同级（`insight/viewer/`），未来可单独升级/替换。 + +## 3. 方案选型 + +候选三选一： + +| 方案 | 代码量 | 表格展示 | 依赖体积 | 结论 | +|---|---|---|---|---| +| **A. Streamlit** | ~80 行 Python，0 行 HTML | 极好 | 中（~150MB） | **采用** | +| B. Flask + HTML | 1 个 .py + 1 个 .html | 一般 | 小 | 否 | +| C. Gradio | ~100 行 | 弱 | 中 | 否 | + +理由： + +- 「基础查看 + Windows + 简单实现」三个约束下，Streamlit 匹配度最高。 +- 不写一行 HTML/CSS/JS，符合「简单实现」。 +- `st.dataframe` 自带列排序、`st.session_state` 选行状态、`st.cache_data` 缓存都内置。 +- 依赖通过 `uv run --with streamlit` 临时拉取，**不写进** `pyproject.toml` / `requirements.txt`，避免污染上游依赖。 + +## 4. 文件结构 + +``` +insight/viewer/ +├── __init__.py # 空，仅作包 +├── app.py # Streamlit 入口（页面布局、session_state） +├── data.py # 纯函数：load_notes() / load_comments(note_id) / format_ts(ts) +└── README.md # 启动说明 +``` + +职责切分： + +- `data.py`：纯数据层，**不导入** Streamlit，方便单测与未来换 UI。 +- `app.py`：仅做页面布局和交互，调 `data.py` 取数据。 +- 复用约定：SQLite 路径 = `database/sqlite_tables.db`（与上游 / `insight/db.py` 一致），**不读 `.env`**。 + +## 5. 界面与交互 + +单页面三块布局（自上而下）： + +``` +┌─────────────────────────────────────────────────────────┐ +│ 小红书数据查看 共 44 条笔记 / 435 条评论 [刷新] │ ← 顶部 +├─────────────────────────────────────────────────────────┤ +│ 笔记列表（按发布时间 time 倒序） │ +│ ┌─────┬──────────────┬──────┬─────────┬────────┬──────┐│ +│ │ # │ 标题 │ 点赞 │ 评论数 │ 关键词 │ 时间 ││ +│ ├─────┼──────────────┼──────┼─────────┼────────┼──────┤│ +│ │ 1 │ 春日穿搭... │ 1.2k │ 38 │ 穿搭 │ 03-15││ +│ │ 2 │ 护肤心得... │ 856 │ 12 │ 护肤 │ 03-14││ +│ │ ... │ │ │ │ │ ││ +│ └─────┴──────────────┴──────┴─────────┴────────┴──────┘│ +├─────────────────────────────────────────────────────────┤ +│ ▾ 笔记详情（选中后展开） │ +│ 作者：xxx | 发布时间：2024-03-15 | 关键词：穿搭 │ +│ 正文：今天分享... │ +│ ───────────────────────────────── │ +│ 评论（38 条） │ +│ ┌────┬──────────┬───────────────────────────────┐ │ +│ │ 点赞│ 用户 │ 内容 │ │ +│ └────┴──────────┴───────────────────────────────┘ │ +└─────────────────────────────────────────────────────────┘ +``` + +交互细节： + +- 笔记列表用 `st.dataframe`，自带列排序。 +- 「查看」用按钮 → 写入 `st.session_state["selected_note_id"]`，详情区根据它重新查询。 +- 顶部 `[刷新]` 按钮 → `st.cache_data.clear()` 后重读。 +- 时间戳统一用 `format_ts` 转 `YYYY-MM-DD HH:MM`。 +- 大文本字段（正文/评论）用 `st.markdown` 容器展示，长文本自然换行（数据规模小，**不做折叠/分页**）。 +- 缓存：`@st.cache_data(ttl=60)`，60 秒自动重读；不点刷新也最多延迟 1 分钟。 +- 数据规模小（44 + 435），**不做评论分页**。 + +## 6. 字段映射 + +UI 展示字段与数据库字段对应： + +| UI 列 | 数据源 | +|---|---| +| 标题 | `xhs_note.title` | +| 点赞 | `xhs_note.liked_count` | +| 评论数 | `xhs_note.comment_count` | +| 发布时间 | `xhs_note.time` → `format_ts` | +| 关键词 | `xhs_note.source_keyword` | +| 笔记作者 | `xhs_note.nickname` | +| 正文 | `xhs_note.desc` | +| 评论用户 | `xhs_note_comment.nickname` | +| 评论内容 | `xhs_note_comment.content` | +| 评论点赞 | `xhs_note_comment.like_count` | +| 评论时间 | `xhs_note_comment.create_time` → `format_ts` | + +## 7. 错误处理 + +所有错误在 UI 层用 `st.error` 友好展示，不抛 traceback： + +| 场景 | 行为 | +|---|---| +| SQLite 文件不存在 | `未找到数据库 database/sqlite_tables.db，请先运行爬虫` | +| `xhs_note` 表不存在 | `数据库缺少表 xhs_note，请先执行 uv run python main.py --init_db sqlite` | +| 笔记表为空 | 表格区显示 `暂无数据` | +| 选中笔记无评论 | 评论区显示 `该笔记暂无评论` | +| 数据库被爬取进程持锁 | `try/except sqlite3.OperationalError` → `数据库正忙，请稍后重试` | +| 单条记录字段为 None | UI 显示 `—`，不崩 | + +## 8. 数据流 + +``` +app.py + └─ st.cache_data(ttl=60) + └─ data.load_notes() → list[dict] (SELECT * FROM xhs_note ORDER BY time DESC) + └─ data.load_comments(id) → list[dict] (SELECT … FROM xhs_note_comment WHERE note_id = ?) + └─ data.format_ts(int|None) → str (时间戳→可读字符串) +``` + +设计决策： + +- **不**复用 `database/db_session.py`：那是异步 SQLAlchemy，为爬取设计；本工具只读、低频、同步访问，独立 `sqlite3.connect` 更轻、更稳、更易测。 +- 所有 SQL 集中在 `data.py`，`app.py` 不出现 SQL 字符串。 +- 工具对数据库**只读不写**。 + +## 9. 测试 + +`tests/insight/test_viewer.py`，**仅测 `data.py` 纯函数**，不启 Streamlit： + +- `test_load_notes_returns_list`：临时 SQLite 灌 3 条假数据 → 验证返回条数与字段 +- `test_load_comments_filters_by_note_id`：灌 2 笔记 + 3 评论 → 验证过滤正确 +- `test_load_comments_empty_when_no_match`：不存在的 `note_id` → 返回 `[]` +- `test_format_ts_handles_none_and_zero`：None / 0 / 正常时间戳都正常处理 +- `test_missing_table_raises_operational_error`：删表后调函数 → 验证抛 `sqlite3.OperationalError`（让 UI 捕获后展示） + +`app.py` 不写 Streamlit 单测（成本/收益不划算），通过 `uv run … insight.viewer.app` 手动烟雾测试一次。 + +## 10. 启动方式 + +`insight/viewer/app.py` 是普通的 Streamlit 脚本（模块级调用 `st.*`），从项目根目录执行： + +```bash +uv run --with streamlit streamlit run insight/viewer/app.py +# → 自动打开 http://localhost:8501 +``` + +补充： + +- 依赖通过 `--with streamlit` 临时拉取，**不修改** `pyproject.toml` / `requirements.txt`。 +- `insight/viewer/README.md` 写一份简要说明（含首次跑、刷新操作、停止 Ctrl+C）。 + +## 11. YAGNI 清单（本期明确不做） + +- 全局搜索框（笔记标题 + 正文 + 评论） +- 按关键词/作者下拉筛选 +- 数据导出按钮（CSV / Excel / JSON） +- `insight_runs` 采集运行记录面板 +- 评论分页 +- 多人协作 / 用户登录 +- 在查看器中编辑 / 删除数据 + +## 12. 与上期设计的关系 + +- 上期 `insight/` 提供「爬数据」能力，本期 `insight/viewer/` 提供「看数据」能力。 +- 两者**互不依赖**：viewer 不 import `insight.cli` / `insight.runner` / `insight.orchestrator` / `insight.config`。 +- 共用同一个 SQLite 文件与同一个表结构，**这是唯一的耦合点**。 +- 上期 Git 同步策略（`upstream` remote + `merge`/`rebase`）继续适用。 diff --git a/insight/README.md b/insight/README.md new file mode 100644 index 000000000..e9ac723ee --- /dev/null +++ b/insight/README.md @@ -0,0 +1,57 @@ +# insight — 小红书评论定时采集（MediaCrawler 二次开发） + +本包是对 MediaCrawler 的二次开发，**不修改任何上游文件**，全部代码在 `insight/` 内。 +本期范围：定时爬取 + 原始数据入库（复用上游 SQLite）+ 运行日志。**不含文本分析**（后续迭代）。 + +## 一次性准备 + +```bash +# 1. 初始化上游 SQLite 表结构（创建 xhs_note / xhs_note_comment 等） +uv run python main.py --init_db sqlite + +# 2. 确保已登录小红书（CDP 模式，复用本机 Chrome 登录态） +# 参见项目根 README 的 Chrome 远程调试配置 +``` + +## 配置 + +编辑 `insight/config.py` 的 `JOBS` 列表（job 类型 search/detail/creator、关键词/笔记ID/创作者ID、触发时刻 hour/minute、max_notes 等）。 + +## 使用 + +```bash +# 立即跑一次某个 job（不依赖 apscheduler） +uv run python -m insight.cli crawl-once kw_daily + +# 查看最近运行记录（不依赖 apscheduler） +uv run python -m insight.cli status --limit 20 + +# 启动定时守护进程（前台运行，Ctrl+C 退出；需 apscheduler） +uv run --with "apscheduler>=3.10,<4" python -m insight.cli run-daemon +``` + +> 守护进程为前台常驻进程；本机重启后需手动重新启动。 + +## 运行测试 + +```bash +uv run --with "apscheduler>=3.10,<4" pytest tests/insight -v +``` + +## 与上游 MediaCrawler 同步更新 + +```bash +# 一次性：添加上游远程 +git remote add upstream https://github.com/NanmiCoder/MediaCrawler.git + +# 定期同步 +git fetch upstream +git merge upstream/main # 或 git rebase upstream/main +``` + +因为本包全部是 `insight/` 下的新增文件、未改动任何上游文件，合并几乎不会冲突。 +唯一与上游耦合的假设： +- 上游 CLI 参数（`--platform/--type/--keywords/--specified_id/--creator_id/--save_data_option` 等）保持不变； +- SQLite 路径仍为 `database/sqlite_tables.db`（见 `config/db_config.py`）； +- `main.main` / `main.async_cleanup` / `tools.app_runner.run` 接口保持不变。 +同步后若上述任一处变化，只需相应调整 `insight/runner.py`、`insight/config.py`、`insight/crawl_entry.py`。 diff --git a/insight/__init__.py b/insight/__init__.py new file mode 100644 index 000000000..0a8830f9e --- /dev/null +++ b/insight/__init__.py @@ -0,0 +1,2 @@ +# -*- coding: utf-8 -*- +"""自研二次开发包：小红书评论定时采集与运行记录。不修改 MediaCrawler 上游文件。""" diff --git a/insight/cli.py b/insight/cli.py new file mode 100644 index 000000000..226bbea32 --- /dev/null +++ b/insight/cli.py @@ -0,0 +1,77 @@ +# -*- coding: utf-8 -*- +"""insight 命令行入口：crawl-once / run-daemon / status。""" + +import argparse +from typing import Optional, Sequence + +from insight import config +from insight.db import InsightDB +from insight.orchestrator import run_job + + +def _find_job(name: str) -> dict: + for job in config.JOBS: + if job["name"] == name: + return job + raise SystemExit(f"job not found: {name!r}. Known jobs: {[j['name'] for j in config.JOBS]}") + + +def cmd_crawl_once(args: argparse.Namespace) -> None: + job = _find_job(args.name) + result = run_job(job) + print(result) + + +def cmd_run_daemon(args: argparse.Namespace) -> None: + # 惰性导入：只有 run-daemon 需要 apscheduler + from insight.scheduler.daemon import main as run_daemon + run_daemon() + + +def cmd_status(args: argparse.Namespace) -> None: + db = InsightDB(config.DB_PATH) + db.init_schema() + runs = db.recent_runs(args.limit) + if not runs: + print("(no runs yet)") + return + for run in runs: + line = ( + f"#{run['id']} {run['job_name']} [{run['crawler_type']}] " + f"status={run['status']} exit={run['exit_code']} " + f"notes={run['notes_crawled']} comments={run['comments_crawled']}" + ) + # 失败/超时/被中断时把原因也打出来（最多 200 字符） + if run["status"] in ("error", "timeout", "interrupted") and run.get("error_msg"): + tail = run["error_msg"].strip().replace("\n", " ⏎ ") + if len(tail) > 200: + tail = tail[:200] + "..." + line += f"\n err: {tail}" + print(line) + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(prog="insight") + sub = parser.add_subparsers(dest="command", required=True) + + p_once = sub.add_parser("crawl-once", help="立即运行某个 job") + p_once.add_argument("name", help="config.JOBS 中的 job 名") + p_once.set_defaults(func=cmd_crawl_once) + + p_daemon = sub.add_parser("run-daemon", help="启动定时守护进程") + p_daemon.set_defaults(func=cmd_run_daemon) + + p_status = sub.add_parser("status", help="查看最近的运行记录") + p_status.add_argument("--limit", type=int, default=20) + p_status.set_defaults(func=cmd_status) + + return parser + + +def main(argv: Optional[Sequence[str]] = None) -> None: + args = build_parser().parse_args(argv) + args.func(args) + + +if __name__ == "__main__": + main() diff --git a/insight/config.py b/insight/config.py new file mode 100644 index 000000000..d5169349c --- /dev/null +++ b/insight/config.py @@ -0,0 +1,28 @@ +# -*- coding: utf-8 -*- +"""insight 包的配置。纯数据，不含逻辑。""" + +import os + +# 项目根目录（insight/ 的上一级） +PROJECT_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) + +# 爬虫写入的 SQLite 文件，必须与 config/db_config.py 的 SQLITE_DB_PATH 一致 +DB_PATH = os.path.join(PROJECT_ROOT, "database", "sqlite_tables.db") + +# 单次爬虫子进程的最大运行秒数，超时则杀死并记为 timeout +SUBPROCESS_TIMEOUT = 1800 + +# APScheduler 容忍的最大迟到秒数（睡眠/停机后仍补跑） +MISFIRE_GRACE_TIME = 3600 + +# 定时任务列表。每个 job 映射为一次爬虫调用。 +# 必填键：name、type（"search"|"detail"|"creator"）、hour（0-23） +# search 额外需要 "keywords"（英文逗号分隔的字符串） +# detail 额外需要 "note_ids"（笔记 ID 或 URL 的列表） +# creator 额外需要 "creator_ids"（创作者 ID 或 URL 的列表） +# 可选键：minute（默认 0）、max_notes（int）、max_comments（int）、get_sub_comment（bool） +JOBS = [ + {"name": "kw_daily", "type": "search", "keywords": "全球流量卡,国际流量卡", "hour": 2, "minute": 0, "max_notes": 20}, + {"name": "watch_notes", "type": "detail", "note_ids": ["请替换为真实笔记ID"], "hour": 3, "minute": 0}, + {"name": "creator_daily", "type": "creator", "creator_ids": ["请替换为真实创作者ID"], "hour": 4, "minute": 0}, +] diff --git a/insight/crawl_entry.py b/insight/crawl_entry.py new file mode 100644 index 000000000..b8f18beca --- /dev/null +++ b/insight/crawl_entry.py @@ -0,0 +1,34 @@ +# -*- coding: utf-8 -*- +"""爬虫子进程入口。 + +读取 INSIGHT_MAX_NOTES 等无 CLI 参数的覆盖项写入上游 config， +然后把其余 CLI 参数交给上游 main 处理。本文件位于 insight 包内， +不修改任何上游文件。 + +用法（由 insight.runner 以子进程调用）： + python -m insight.crawl_entry --platform xhs --type search --keywords ... --save_data_option sqlite +""" + +import os + + +def apply_overrides() -> None: + """把没有 CLI 参数的配置项从环境变量写入上游 config。""" + import config # 上游配置模块 + + max_notes = os.environ.get("INSIGHT_MAX_NOTES") + if max_notes: + config.CRAWLER_MAX_NOTES_COUNT = int(max_notes) + + +def main_entry() -> None: + apply_overrides() + # 复用上游的协程入口与清理逻辑（sys.argv 由上游 cmd_arg 解析） + from main import main as crawler_main, async_cleanup + from tools.app_runner import run + + run(crawler_main, async_cleanup, cleanup_timeout_seconds=15.0) + + +if __name__ == "__main__": + main_entry() diff --git a/insight/db.py b/insight/db.py new file mode 100644 index 000000000..7a2dfbd8a --- /dev/null +++ b/insight/db.py @@ -0,0 +1,73 @@ +# -*- coding: utf-8 -*- +"""insight 的 SQLite 访问层：运行记录与上游表计数。""" + +import os +import sqlite3 +import time +from typing import List, Dict, Optional + +SCHEMA_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "schema.sql") + +# 仅允许对这些上游表做计数，避免 SQL 注入 +_COUNTABLE_TABLES = {"xhs_note", "xhs_note_comment"} + + +class InsightDB: + def __init__(self, db_path: str): + self.db_path = db_path + + def _connect(self) -> sqlite3.Connection: + conn = sqlite3.connect(self.db_path) + conn.row_factory = sqlite3.Row + return conn + + def init_schema(self) -> None: + with open(SCHEMA_PATH, "r", encoding="utf-8") as f: + ddl = f.read() + with self._connect() as conn: + conn.executescript(ddl) + + def start_run(self, job_name: str, crawler_type: str) -> int: + now = int(time.time()) + with self._connect() as conn: + cur = conn.execute( + "INSERT INTO insight_runs (job_name, crawler_type, started_ts, status) " + "VALUES (?, ?, ?, 'running')", + (job_name, crawler_type, now), + ) + return int(cur.lastrowid) + + def finish_run( + self, + run_id: int, + *, + exit_code: Optional[int], + status: str, + notes_crawled: int = 0, + comments_crawled: int = 0, + error_msg: Optional[str] = None, + ) -> None: + now = int(time.time()) + with self._connect() as conn: + conn.execute( + "UPDATE insight_runs SET finished_ts=?, exit_code=?, status=?, " + "notes_crawled=?, comments_crawled=?, error_msg=? WHERE id=?", + (now, exit_code, status, notes_crawled, comments_crawled, error_msg, run_id), + ) + + def count_rows(self, table: str) -> int: + if table not in _COUNTABLE_TABLES: + raise ValueError(f"count_rows: table {table!r} not allowed") + with self._connect() as conn: + try: + row = conn.execute(f"SELECT COUNT(*) AS c FROM {table}").fetchone() + except sqlite3.OperationalError: + return 0 # 表尚未由上游 --init_db 创建 + return int(row["c"]) + + def recent_runs(self, limit: int = 20) -> List[Dict]: + with self._connect() as conn: + rows = conn.execute( + "SELECT * FROM insight_runs ORDER BY id DESC LIMIT ?", (limit,) + ).fetchall() + return [dict(r) for r in rows] diff --git a/insight/orchestrator.py b/insight/orchestrator.py new file mode 100644 index 000000000..2c3d59261 --- /dev/null +++ b/insight/orchestrator.py @@ -0,0 +1,66 @@ +# -*- coding: utf-8 -*- +"""一次完整调度周期：开始记录 → 计数前 → 跑爬虫 → 计数后 → 结束记录。 + +finish_run 放在 try/finally 里，保证 KeyboardInterrupt / SIGTERM 等外部信号 +也会让 db 记录进入终态（之前卡在 'running' 不会自动收尾）。 +""" + +from typing import Optional + +from insight import config, runner +from insight.db import InsightDB + + +def run_job(job: dict, db: Optional[InsightDB] = None) -> dict: + db = db or InsightDB(config.DB_PATH) + db.init_schema() + + run_id = db.start_run(job["name"], job["type"]) + notes_before = db.count_rows("xhs_note") + comments_before = db.count_rows("xhs_note_comment") + + status = "interrupted" + exit_code: Optional[int] = None + notes_delta = 0 + comments_delta = 0 + error_msg: Optional[str] = "interrupted before crawl finished" + + try: + result = runner.run_crawl(job) + + notes_after = db.count_rows("xhs_note") + comments_after = db.count_rows("xhs_note_comment") + notes_delta = max(0, notes_after - notes_before) + comments_delta = max(0, comments_after - comments_before) + exit_code = result.exit_code + + if result.timed_out: + status = "timeout" + error_msg = "subprocess timed out" + elif result.exit_code == 0: + status = "success" + error_msg = None + else: + status = "error" + error_msg = result.stderr_tail or status + + return { + "run_id": run_id, + "status": status, + "notes": notes_delta, + "comments": comments_delta, + } + finally: + # 无论 try 块是否抛异常、是否被信号中断，都把状态写入 db + try: + db.finish_run( + run_id, + exit_code=exit_code, + status=status, + notes_crawled=notes_delta, + comments_crawled=comments_delta, + error_msg=error_msg, + ) + except Exception: + # finish_run 自身失败也不能让上层再抛 + pass diff --git a/insight/requirements.txt b/insight/requirements.txt new file mode 100644 index 000000000..febc46ea7 --- /dev/null +++ b/insight/requirements.txt @@ -0,0 +1 @@ +apscheduler>=3.10,<4 diff --git a/insight/runner.py b/insight/runner.py new file mode 100644 index 000000000..d0cce9fd0 --- /dev/null +++ b/insight/runner.py @@ -0,0 +1,114 @@ +# -*- coding: utf-8 -*- +"""把 job 配置转成爬虫命令行参数，并以子进程方式运行爬虫。""" + +import os +import subprocess +import sys +from dataclasses import dataclass +from typing import List, Optional + +from insight import config as insight_config + +# Windows 默认 console 是 cp1252，print 中文会触发 UnicodeEncodeError。 +# 切到 utf-8 + errors='replace' 避免实时日志进程被这一行打挂。 +for _stream in (sys.stdout, sys.stderr): + try: + _stream.reconfigure(encoding="utf-8", errors="replace") # type: ignore[attr-defined] + except Exception: + pass + +VALID_TYPES = {"search", "detail", "creator"} + +# 注入子进程环境：绕开 Mihomo/Clash 等 fake-IP 代理对 localhost 的劫持 +_BYPASS_HOSTS = "localhost,127.0.0.1,::1" + + +def build_crawl_args(job: dict) -> List[str]: + """把一个 job dict 翻译成 `python -m insight.crawl_entry` 的参数列表。""" + name = job.get("name") + jtype = job.get("type") + if jtype not in VALID_TYPES: + raise ValueError(f"job {name!r}: invalid type {jtype!r}") + + args: List[str] = ["--platform", "xhs", "--type", jtype, "--save_data_option", "sqlite"] + + if jtype == "search": + keywords = job.get("keywords") + if not keywords: + raise ValueError(f"job {name!r}: search type requires 'keywords'") + args += ["--keywords", keywords] + elif jtype == "detail": + note_ids = job.get("note_ids") + if not note_ids: + raise ValueError(f"job {name!r}: detail type requires 'note_ids'") + args += ["--specified_id", ",".join(note_ids)] + elif jtype == "creator": + creator_ids = job.get("creator_ids") + if not creator_ids: + raise ValueError(f"job {name!r}: creator type requires 'creator_ids'") + args += ["--creator_id", ",".join(creator_ids)] + + args += ["--get_comment", "true"] + if job.get("get_sub_comment"): + args += ["--get_sub_comment", "true"] + if job.get("max_comments"): + args += ["--max_comments_count_singlenotes", str(job["max_comments"])] + + return args + + +@dataclass +class CrawlResult: + exit_code: int + timed_out: bool + stderr_tail: str + + +def run_crawl(job: dict, timeout: Optional[int] = None) -> CrawlResult: + """以子进程运行爬虫入口，实时把子进程 stdout 转发到父进程。 + + - max_notes 通过环境变量 INSIGHT_MAX_NOTES 注入 + - 默认把 NO_PROXY/no_proxy 设为 localhost/127.0.0.1，绕开 Mihomo/Clash 的 fake-IP 劫持 + """ + timeout = timeout if timeout is not None else insight_config.SUBPROCESS_TIMEOUT + args = build_crawl_args(job) + cmd = [sys.executable, "-u", "-m", "insight.crawl_entry", *args] + + env = dict(os.environ) + env.setdefault("NO_PROXY", _BYPASS_HOSTS) + env.setdefault("no_proxy", _BYPASS_HOSTS) + if job.get("max_notes"): + env["INSIGHT_MAX_NOTES"] = str(job["max_notes"]) + + proc = subprocess.Popen( + cmd, + cwd=insight_config.PROJECT_ROOT, + env=env, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, # 合并到 stdout，统一顺序 + text=True, + encoding="utf-8", + errors="replace", + ) + + output_lines: List[str] = [] + try: + assert proc.stdout is not None + for line in proc.stdout: + print(line, end="", flush=True) # 实时转发给父进程终端 + output_lines.append(line) + proc.wait(timeout=timeout) + except subprocess.TimeoutExpired: + proc.kill() + proc.wait() + return CrawlResult( + exit_code=-1, + timed_out=True, + stderr_tail="".join(output_lines)[-2000:], + ) + + return CrawlResult( + exit_code=proc.returncode, + timed_out=False, + stderr_tail="".join(output_lines)[-2000:], + ) diff --git a/insight/scheduler/__init__.py b/insight/scheduler/__init__.py new file mode 100644 index 000000000..40a96afc6 --- /dev/null +++ b/insight/scheduler/__init__.py @@ -0,0 +1 @@ +# -*- coding: utf-8 -*- diff --git a/insight/scheduler/daemon.py b/insight/scheduler/daemon.py new file mode 100644 index 000000000..3d3ec61b6 --- /dev/null +++ b/insight/scheduler/daemon.py @@ -0,0 +1,39 @@ +# -*- coding: utf-8 -*- +"""APScheduler 守护进程：按 cron 触发每个 job。""" + +from typing import List, Optional + +from apscheduler.schedulers.blocking import BlockingScheduler +from apscheduler.triggers.cron import CronTrigger + +from insight import config +from insight.orchestrator import run_job + + +def build_scheduler(jobs: Optional[List[dict]] = None, scheduler: Optional[BlockingScheduler] = None) -> BlockingScheduler: + jobs = jobs if jobs is not None else config.JOBS + scheduler = scheduler if scheduler is not None else BlockingScheduler() + for job in jobs: + trigger = CronTrigger(hour=job["hour"], minute=job.get("minute", 0)) + scheduler.add_job( + run_job, + trigger=trigger, + args=[job], + id=job["name"], + misfire_grace_time=config.MISFIRE_GRACE_TIME, + replace_existing=True, + ) + return scheduler + + +def main() -> None: + scheduler = build_scheduler() + print(f"[insight] scheduler started with {len(scheduler.get_jobs())} job(s). Ctrl+C to stop.") + try: + scheduler.start() + except (KeyboardInterrupt, SystemExit): + print("[insight] scheduler stopped.") + + +if __name__ == "__main__": + main() diff --git a/insight/schema.sql b/insight/schema.sql new file mode 100644 index 000000000..7c5556a74 --- /dev/null +++ b/insight/schema.sql @@ -0,0 +1,13 @@ +CREATE TABLE IF NOT EXISTS insight_runs ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + job_name TEXT NOT NULL, + crawler_type TEXT NOT NULL, + started_ts INTEGER NOT NULL, + finished_ts INTEGER, + exit_code INTEGER, + notes_crawled INTEGER DEFAULT 0, + comments_crawled INTEGER DEFAULT 0, + status TEXT NOT NULL DEFAULT 'running', + error_msg TEXT +); +CREATE INDEX IF NOT EXISTS idx_insight_runs_job ON insight_runs (job_name, started_ts); diff --git a/insight/viewer/README.md b/insight/viewer/README.md new file mode 100644 index 000000000..e22000c14 --- /dev/null +++ b/insight/viewer/README.md @@ -0,0 +1,61 @@ +# 小红书数据查看器（XHS Data Viewer） + +基于 Streamlit 的本地查看工具，复用上游 `database/sqlite_tables.db`。 +仅做基础查看：笔记列表 + 点进去看评论。不修改任何上游文件。 + +## 一次性准备 + +```bash +# 确保上游 SQLite 表结构已创建 +uv run python main.py --init_db sqlite +``` + +## 启动 + +从项目根目录执行： + +```bash +# 注意：必须把项目根加到 PYTHONPATH， +# 否则 `uv run --with streamlit` 创建的临时 env 里没有 insight 包， +# streamlit 启动时会报 ModuleNotFoundError: No module named 'insight' +PYTHONPATH=. uv run --with streamlit streamlit run insight/viewer/app.py +``` + +**Windows PowerShell / conda 用户**：如果 streamlit 已经在你的 conda env 里，可以省掉 `uv run --with`（它只是临时拉 streamlit）： + +```powershell +$env:PYTHONPATH = "." ; streamlit run insight/viewer/app.py +``` + +- 自动打开浏览器 `http://localhost:8501` +- 数据通过 `database/sqlite_tables.db` 实时读取，**不修改任何数据** +- 默认 60 秒缓存；点页面右上 🔄 立即重读 +- Ctrl+C 停止 + +## 界面 + +``` +顶部： 共 N 条笔记 / M 条评论 [🔄 刷新] +中段： 笔记列表（按发布时间倒序，可点击选中） +底段： 选中笔记的详情（作者/时间/关键词/正文）+ 评论列表 +``` + +## 错误提示 + +| 情况 | 提示 | +|---|---| +| 数据库文件不存在 | 未找到数据库 ...，请先运行爬虫 | +| 数据库被爬取进程持锁 | 数据库正忙，请稍后重试 | +| `xhs_note` 表不存在 | 数据库缺少表 xhs_note，请先执行 `uv run python main.py --init_db sqlite` | +| 笔记表为空 | 暂无数据 | +| 选中笔记无评论 | 该笔记暂无评论 | + +## 依赖 + +- Python 3.11+ +- Streamlit（通过 `uv run --with streamlit` 临时拉取，**不**写入 `pyproject.toml` / `requirements.txt`） + +## 相关 + +- 设计文档：[`docs/superpowers/specs/2026-06-06-xhs-data-viewer-design.md`](../../specs/2026-06-06-xhs-data-viewer-design.md) +- 数据来源：[`insight/` 爬取包](../../README.md) \ No newline at end of file diff --git a/insight/viewer/__init__.py b/insight/viewer/__init__.py new file mode 100644 index 000000000..fdd38451c --- /dev/null +++ b/insight/viewer/__init__.py @@ -0,0 +1,3 @@ +# -*- coding: utf-8 -*- +# 小红书数据查看器子包（基于 Streamlit） +# 仅依赖标准库 + streamlit（运行时由 uv --with 临时拉取） \ No newline at end of file diff --git a/insight/viewer/app.py b/insight/viewer/app.py new file mode 100644 index 000000000..f703cd570 --- /dev/null +++ b/insight/viewer/app.py @@ -0,0 +1,170 @@ +# -*- coding: utf-8 -*- +"""小红书数据查看器：Streamlit 单页 UI。 + +页面布局（自上而下）： + 1. 顶部统计 + 刷新按钮 + 2. 笔记列表（st.dataframe，自带列排序） + 3. 选中笔记的详情面板 + 评论列表 + +启动： + uv run --with streamlit streamlit run insight/viewer/app.py +""" +from __future__ import annotations + +import sqlite3 + +import pandas as pd +import streamlit as st + +from insight.viewer.data import ( + DB_PATH, + format_ts, + load_comments, + load_notes, +) + + +def _to_int(v: object) -> int: + """安全地把 comment_count 等字符串字段转 int，失败回 0。""" + try: + return int(v) + except (TypeError, ValueError): + return 0 + + +# ---------- 启动配置 ---------- +st.set_page_config(page_title="小红书数据查看", layout="wide") + + +# ---------- 数据加载（带缓存）---------- +@st.cache_data(ttl=60) +def _cached_notes() -> list[dict]: + return load_notes() + + +@st.cache_data(ttl=60) +def _cached_comments(note_id: str) -> list[dict]: + return load_comments(note_id) + + +def _load_notes_safely() -> list[dict] | None: + """加载笔记，错误以 None 返回并用 st.error 提示。""" + if not DB_PATH.exists(): + st.error( + f"未找到数据库 {DB_PATH}，请先运行爬虫。" + ) + return None + try: + return _cached_notes() + except sqlite3.OperationalError: + st.error("数据库正忙（爬取进程可能正在写入），请稍后重试。") + return None + + +def _load_comments_safely(note_id: str) -> list[dict] | None: + try: + return _cached_comments(note_id) + except sqlite3.OperationalError: + st.error("数据库正忙，请稍后重试。") + return None + + +# ---------- 顶部 ---------- +st.title("小红书数据查看") + +notes = _load_notes_safely() +if notes is None: + st.stop() + +col_stat, col_btn = st.columns([4, 1]) +with col_stat: + total_comments = sum(_to_int(n.get("comment_count")) for n in notes) + st.caption(f"共 {len(notes)} 条笔记 / {total_comments} 条评论（按 comment_count 估算）") +with col_btn: + if st.button("🔄 刷新", use_container_width=True): + st.cache_data.clear() + st.session_state.pop("notes_table", None) # 清掉旧选择，避免重读后指向错位 + st.rerun() + + +# ---------- 笔记列表 ---------- +if not notes: + st.info("暂无数据") + st.stop() + +# 准备表格数据：加一列"发布时间"已格式化 +df = pd.DataFrame( + [ + { + "#": idx + 1, + "标题": n.get("title") or "—", + "点赞": n.get("liked_count") or "—", + "评论数": n.get("comment_count") or "—", + "关键词": n.get("source_keyword") or "—", + "发布时间": format_ts(n.get("time")), + "note_id": n["note_id"], + } + for idx, n in enumerate(notes) + ] +) + +st.subheader("笔记列表") +st.caption("点击表格中的行查看笔记详情与评论") +event = st.dataframe( + df[["#", "标题", "点赞", "评论数", "关键词", "发布时间"]], + use_container_width=True, + hide_index=True, + on_select="rerun", + selection_mode="single-row", + key="notes_table", +) + +# Streamlit 1.32+：用户点击某行后 event.selection.rows 包含选中行号（0-based） +selected_rows = getattr(event, "selection", None) +selected_idx = selected_rows.rows[0] if selected_rows and selected_rows.rows else None + + +# ---------- 详情面板 ---------- +if selected_idx is None: + st.info("👆 在表格中选中一行后查看笔记详情与评论") + st.stop() + +selected_note = notes[selected_idx] + +st.divider() +st.subheader(f"📝 笔记详情：{selected_note.get('title') or '—'}") +detail_cols = st.columns(3) +with detail_cols[0]: + st.caption(f"**作者**：{selected_note.get('nickname') or '—'}") +with detail_cols[1]: + st.caption(f"**发布时间**：{format_ts(selected_note.get('time'))}") +with detail_cols[2]: + st.caption(f"**关键词**：{selected_note.get('source_keyword') or '—'}") + +st.markdown("**正文**") +st.markdown(selected_note.get("desc") or "—") + +# ---------- 评论 ---------- +st.divider() +st.subheader("💬 评论") +comments = _load_comments_safely(selected_note["note_id"]) +if comments is None: + st.stop() +if not comments: + st.info("该笔记暂无评论") + st.stop() + +# 准备评论表格 +cdf = pd.DataFrame( + [ + { + "点赞": c.get("like_count") or "—", + "用户": c.get("nickname") or "—", + "内容": c.get("content") or "—", + "时间": format_ts(c.get("create_time")), + } + for c in comments + ] +) +st.dataframe(cdf, use_container_width=True, hide_index=True) +st.caption(f"共 {len(comments)} 条评论") diff --git a/insight/viewer/data.py b/insight/viewer/data.py new file mode 100644 index 000000000..5ec0fcd41 --- /dev/null +++ b/insight/viewer/data.py @@ -0,0 +1,61 @@ +# -*- coding: utf-8 -*- +"""纯数据层：从 database/sqlite_tables.db 读取小红书笔记与评论。 + +不导入 Streamlit，方便单测。 +""" +from __future__ import annotations + +import datetime +import sqlite3 +from pathlib import Path +from typing import Any + +# 与上游约定一致：数据库固定路径（不读 .env） +DB_PATH = Path(__file__).resolve().parents[2] / "database" / "sqlite_tables.db" + + +def format_ts(ts: int | None) -> str: + """将 Unix 时间戳格式化为 'YYYY-MM-DD HH:MM'。None/0 返回 '—'。 + + DB 实际存储毫秒级时间戳（13 位），同时兼容秒级（10 位）以方便测试。 + 判定阈值 1e12（≈ 33658 年的秒数，远大于任何合理秒级时间戳）。 + """ + if not ts: + return "—" + # 兼容毫秒与秒：> 1e12 视为毫秒 + seconds = ts / 1000 if ts > 1e12 else ts + return datetime.datetime.fromtimestamp(seconds).strftime("%Y-%m-%d %H:%M") + + +def load_notes(db_path: Path | None = None) -> list[dict[str, Any]]: + """加载所有笔记，按发布时间 time 倒序。db_path=None 时使用 DB_PATH。""" + path = db_path if db_path is not None else DB_PATH + conn = sqlite3.connect(str(path)) + conn.row_factory = sqlite3.Row + try: + cur = conn.cursor() + cur.execute( + "SELECT note_id, title, liked_count, comment_count, time, " + "source_keyword, nickname, desc " + "FROM xhs_note ORDER BY time DESC" + ) + return [dict(row) for row in cur.fetchall()] + finally: + conn.close() + + +def load_comments(note_id: str, db_path: Path | None = None) -> list[dict[str, Any]]: + """加载指定笔记的评论，按 create_time 升序。db_path=None 时使用 DB_PATH。""" + path = db_path if db_path is not None else DB_PATH + conn = sqlite3.connect(str(path)) + conn.row_factory = sqlite3.Row + try: + cur = conn.cursor() + cur.execute( + "SELECT comment_id, note_id, nickname, content, like_count, create_time " + "FROM xhs_note_comment WHERE note_id = ? ORDER BY create_time ASC", + (note_id,), + ) + return [dict(row) for row in cur.fetchall()] + finally: + conn.close() \ No newline at end of file diff --git a/scripts/commit-local.ps1 b/scripts/commit-local.ps1 new file mode 100644 index 000000000..3fff23ffa --- /dev/null +++ b/scripts/commit-local.ps1 @@ -0,0 +1,39 @@ +# Stage, commit, and push local customizations on a feature branch. +# Usage: .\scripts\commit-local.ps1 -Message "feat: xxx" [-Branch ] + +param( + [Parameter(Mandatory = $true)] + [string]$Message, + [string]$Branch = "feature/xhs-insight-pipeline" +) + +$ErrorActionPreference = "Stop" + +# Refuse to run on the wrong branch (avoid committing custom code onto main). +$current = git rev-parse --abbrev-ref HEAD +if ($current -ne $Branch) { + Write-Host "ERROR: current branch is '$current', expected '$Branch'." -ForegroundColor Red + Write-Host "Run: git checkout $Branch" -ForegroundColor Yellow + exit 1 +} + +# Bail out if there is nothing to commit. +$status = git status --porcelain +if (-not $status) { + Write-Host "No changes to commit." -ForegroundColor Yellow + exit 0 +} + +Write-Host "==> Changes to commit:" -ForegroundColor Cyan +git status --short + +Write-Host "==> git add ." -ForegroundColor Cyan +git add . + +Write-Host "==> git commit ..." -ForegroundColor Cyan +git commit -m $Message + +Write-Host "==> git push origin $Branch ..." -ForegroundColor Cyan +git push origin $Branch + +Write-Host "==> Commit and push complete." -ForegroundColor Green diff --git a/scripts/sync-upstream.ps1 b/scripts/sync-upstream.ps1 new file mode 100644 index 000000000..c4a7fd740 --- /dev/null +++ b/scripts/sync-upstream.ps1 @@ -0,0 +1,39 @@ +# Sync upstream MediaCrawler into a local feature branch. +# Usage: .\scripts\sync-upstream.ps1 [-Branch ] + +param( + [string]$Branch = "feature/xhs-insight-pipeline", + [string]$UpstreamRemote = "upstream", + [string]$UpstreamBranch = "main" +) + +$ErrorActionPreference = "Stop" + +# Verify the upstream remote is configured. +$remotes = git remote +if ($remotes -notcontains $UpstreamRemote) { + Write-Host "ERROR: remote '$UpstreamRemote' not configured." -ForegroundColor Red + Write-Host "Run: git remote add $UpstreamRemote https://github.com/NanmiCoder/MediaCrawler.git" -ForegroundColor Yellow + exit 1 +} + +# Refuse to proceed with a dirty working tree (rebase would lose work). +$status = git status --porcelain +if ($status) { + Write-Host "ERROR: working tree has uncommitted changes:" -ForegroundColor Red + git status --short + exit 1 +} + +Write-Host "==> Fetching $UpstreamRemote ..." -ForegroundColor Cyan +git fetch $UpstreamRemote + +Write-Host "==> Updating main from $UpstreamRemote/$UpstreamBranch ..." -ForegroundColor Cyan +git checkout main +git merge "$UpstreamRemote/$UpstreamBranch" + +Write-Host "==> Rebasing $Branch onto main ..." -ForegroundColor Cyan +git checkout $Branch +git rebase main + +Write-Host "==> Sync complete." -ForegroundColor Green diff --git a/tests/insight/__init__.py b/tests/insight/__init__.py new file mode 100644 index 000000000..40a96afc6 --- /dev/null +++ b/tests/insight/__init__.py @@ -0,0 +1 @@ +# -*- coding: utf-8 -*- diff --git a/tests/insight/test_cli.py b/tests/insight/test_cli.py new file mode 100644 index 000000000..581894b3b --- /dev/null +++ b/tests/insight/test_cli.py @@ -0,0 +1,26 @@ +# -*- coding: utf-8 -*- +import pytest + +from insight.cli import build_parser, cmd_crawl_once, cmd_status + + +def test_parser_crawl_once(): + args = build_parser().parse_args(["crawl-once", "kw_daily"]) + assert args.name == "kw_daily" + assert args.func is cmd_crawl_once + + +def test_parser_status_default_limit(): + args = build_parser().parse_args(["status"]) + assert args.limit == 20 + assert args.func is cmd_status + + +def test_parser_status_custom_limit(): + args = build_parser().parse_args(["status", "--limit", "5"]) + assert args.limit == 5 + + +def test_parser_requires_subcommand(): + with pytest.raises(SystemExit): + build_parser().parse_args([]) diff --git a/tests/insight/test_crawl_entry.py b/tests/insight/test_crawl_entry.py new file mode 100644 index 000000000..4e97d1af4 --- /dev/null +++ b/tests/insight/test_crawl_entry.py @@ -0,0 +1,23 @@ +# -*- coding: utf-8 -*- +import config # 上游配置 +from insight.crawl_entry import apply_overrides + + +def test_apply_overrides_sets_max_notes(monkeypatch): + original = config.CRAWLER_MAX_NOTES_COUNT + try: + monkeypatch.setenv("INSIGHT_MAX_NOTES", "33") + apply_overrides() + assert config.CRAWLER_MAX_NOTES_COUNT == 33 + finally: + config.CRAWLER_MAX_NOTES_COUNT = original + + +def test_apply_overrides_noop_without_env(monkeypatch): + original = config.CRAWLER_MAX_NOTES_COUNT + try: + monkeypatch.delenv("INSIGHT_MAX_NOTES", raising=False) + apply_overrides() + assert config.CRAWLER_MAX_NOTES_COUNT == original + finally: + config.CRAWLER_MAX_NOTES_COUNT = original diff --git a/tests/insight/test_db.py b/tests/insight/test_db.py new file mode 100644 index 000000000..970c88a8f --- /dev/null +++ b/tests/insight/test_db.py @@ -0,0 +1,60 @@ +# -*- coding: utf-8 -*- +import os + +from insight.db import InsightDB + + +def _db(tmp_path): + db = InsightDB(os.path.join(str(tmp_path), "t.db")) + db.init_schema() + return db + + +def test_init_schema_creates_table_and_is_idempotent(tmp_path): + db = _db(tmp_path) + db.init_schema() # 再次调用不应报错 + assert db.recent_runs() == [] + + +def test_start_and_finish_run_roundtrip(tmp_path): + db = _db(tmp_path) + run_id = db.start_run("kw_daily", "search") + assert isinstance(run_id, int) + + runs = db.recent_runs() + assert len(runs) == 1 + assert runs[0]["status"] == "running" + assert runs[0]["job_name"] == "kw_daily" + assert runs[0]["finished_ts"] is None + + db.finish_run(run_id, exit_code=0, status="success", notes_crawled=3, comments_crawled=12) + runs = db.recent_runs() + assert runs[0]["status"] == "success" + assert runs[0]["exit_code"] == 0 + assert runs[0]["notes_crawled"] == 3 + assert runs[0]["comments_crawled"] == 12 + assert runs[0]["finished_ts"] is not None + + +def test_count_rows_returns_zero_when_table_absent(tmp_path): + db = _db(tmp_path) + assert db.count_rows("xhs_note") == 0 + assert db.count_rows("xhs_note_comment") == 0 + + +def test_count_rows_rejects_unknown_table(tmp_path): + db = _db(tmp_path) + try: + db.count_rows("evil_table") + except ValueError: + return + raise AssertionError("expected ValueError for disallowed table") + + +def test_recent_runs_orders_desc_and_limits(tmp_path): + db = _db(tmp_path) + for i in range(3): + db.start_run(f"job{i}", "search") + runs = db.recent_runs(limit=2) + assert len(runs) == 2 + assert runs[0]["job_name"] == "job2" # 最新在前 diff --git a/tests/insight/test_orchestrator.py b/tests/insight/test_orchestrator.py new file mode 100644 index 000000000..410adc790 --- /dev/null +++ b/tests/insight/test_orchestrator.py @@ -0,0 +1,54 @@ +# -*- coding: utf-8 -*- +import os + +import insight.orchestrator as orchestrator +import insight.runner as runner +from insight.db import InsightDB + + +def _db(tmp_path): + db = InsightDB(os.path.join(str(tmp_path), "t.db")) + db.init_schema() + return db + + +def test_run_job_success_records_success(tmp_path, monkeypatch): + db = _db(tmp_path) + monkeypatch.setattr( + runner, "run_crawl", + lambda job, timeout=None: runner.CrawlResult(exit_code=0, timed_out=False, stderr_tail=""), + ) + job = {"name": "kw", "type": "search", "keywords": "x", "hour": 2} + result = orchestrator.run_job(job, db=db) + + assert result["status"] == "success" + runs = db.recent_runs() + assert runs[0]["status"] == "success" + assert runs[0]["error_msg"] is None + + +def test_run_job_nonzero_records_error_with_stderr(tmp_path, monkeypatch): + db = _db(tmp_path) + monkeypatch.setattr( + runner, "run_crawl", + lambda job, timeout=None: runner.CrawlResult(exit_code=2, timed_out=False, stderr_tail="kaboom"), + ) + job = {"name": "kw", "type": "search", "keywords": "x", "hour": 2} + result = orchestrator.run_job(job, db=db) + + assert result["status"] == "error" + runs = db.recent_runs() + assert runs[0]["status"] == "error" + assert "kaboom" in runs[0]["error_msg"] + + +def test_run_job_timeout_records_timeout(tmp_path, monkeypatch): + db = _db(tmp_path) + monkeypatch.setattr( + runner, "run_crawl", + lambda job, timeout=None: runner.CrawlResult(exit_code=-1, timed_out=True, stderr_tail=""), + ) + job = {"name": "kw", "type": "search", "keywords": "x", "hour": 2} + result = orchestrator.run_job(job, db=db) + assert result["status"] == "timeout" + assert db.recent_runs()[0]["status"] == "timeout" diff --git a/tests/insight/test_runner_args.py b/tests/insight/test_runner_args.py new file mode 100644 index 000000000..ad85bb33d --- /dev/null +++ b/tests/insight/test_runner_args.py @@ -0,0 +1,53 @@ +# -*- coding: utf-8 -*- +import pytest + +from insight.runner import build_crawl_args + + +def test_search_job_args(): + job = {"name": "kw", "type": "search", "keywords": "a,b", "hour": 2} + args = build_crawl_args(job) + assert args[:6] == ["--platform", "xhs", "--type", "search", "--save_data_option", "sqlite"] + assert "--keywords" in args and args[args.index("--keywords") + 1] == "a,b" + assert args[args.index("--get_comment") + 1] == "true" + + +def test_detail_job_joins_note_ids(): + job = {"name": "w", "type": "detail", "note_ids": ["n1", "n2"], "hour": 3} + args = build_crawl_args(job) + assert "--type" in args and args[args.index("--type") + 1] == "detail" + assert args[args.index("--specified_id") + 1] == "n1,n2" + + +def test_creator_job_joins_creator_ids(): + job = {"name": "c", "type": "creator", "creator_ids": ["c1"], "hour": 4} + args = build_crawl_args(job) + assert args[args.index("--creator_id") + 1] == "c1" + + +def test_optional_max_comments_and_sub_comment(): + job = {"name": "k", "type": "search", "keywords": "x", "hour": 2, + "max_comments": 50, "get_sub_comment": True} + args = build_crawl_args(job) + assert args[args.index("--max_comments_count_singlenotes") + 1] == "50" + assert args[args.index("--get_sub_comment") + 1] == "true" + + +def test_invalid_type_raises(): + with pytest.raises(ValueError): + build_crawl_args({"name": "bad", "type": "homefeed", "hour": 1}) + + +def test_search_missing_keywords_raises(): + with pytest.raises(ValueError): + build_crawl_args({"name": "bad", "type": "search", "hour": 1}) + + +def test_detail_missing_note_ids_raises(): + with pytest.raises(ValueError): + build_crawl_args({"name": "bad", "type": "detail", "hour": 1}) + + +def test_creator_missing_ids_raises(): + with pytest.raises(ValueError): + build_crawl_args({"name": "bad", "type": "creator", "hour": 1}) diff --git a/tests/insight/test_runner_subprocess.py b/tests/insight/test_runner_subprocess.py new file mode 100644 index 000000000..5f59b150b --- /dev/null +++ b/tests/insight/test_runner_subprocess.py @@ -0,0 +1,53 @@ +# -*- coding: utf-8 -*- +import subprocess +import sys + +import insight.runner as runner + + +class _FakeCompleted: + def __init__(self, returncode, stderr=""): + self.returncode = returncode + self.stderr = stderr + + +def test_run_crawl_success_builds_command_and_env(monkeypatch): + captured = {} + + def fake_run(cmd, **kwargs): + captured["cmd"] = cmd + captured["kwargs"] = kwargs + return _FakeCompleted(0, stderr="") + + monkeypatch.setattr(subprocess, "run", fake_run) + + job = {"name": "kw", "type": "search", "keywords": "x", "hour": 2, "max_notes": 7} + result = runner.run_crawl(job) + + assert result.exit_code == 0 + assert result.timed_out is False + # 命令以当前解释器 + -m insight.crawl_entry 开头 + assert captured["cmd"][:3] == [sys.executable, "-m", "insight.crawl_entry"] + assert "--keywords" in captured["cmd"] + # max_notes 通过环境变量注入 + assert captured["kwargs"]["env"]["INSIGHT_MAX_NOTES"] == "7" + + +def test_run_crawl_nonzero_exit_returns_stderr_tail(monkeypatch): + monkeypatch.setattr(subprocess, "run", lambda cmd, **kw: _FakeCompleted(1, stderr="boom")) + job = {"name": "kw", "type": "search", "keywords": "x", "hour": 2} + result = runner.run_crawl(job) + assert result.exit_code == 1 + assert result.timed_out is False + assert "boom" in result.stderr_tail + + +def test_run_crawl_timeout(monkeypatch): + def fake_run(cmd, **kwargs): + raise subprocess.TimeoutExpired(cmd=cmd, timeout=1, stderr="late") + + monkeypatch.setattr(subprocess, "run", fake_run) + job = {"name": "kw", "type": "search", "keywords": "x", "hour": 2} + result = runner.run_crawl(job, timeout=1) + assert result.timed_out is True + assert result.exit_code == -1 diff --git a/tests/insight/test_scheduler.py b/tests/insight/test_scheduler.py new file mode 100644 index 000000000..a70ea1e77 --- /dev/null +++ b/tests/insight/test_scheduler.py @@ -0,0 +1,31 @@ +# -*- coding: utf-8 -*- +from apscheduler.schedulers.blocking import BlockingScheduler + +from insight.scheduler.daemon import build_scheduler + + +def test_build_scheduler_registers_one_job_per_config(): + jobs = [ + {"name": "kw_daily", "type": "search", "keywords": "x", "hour": 2, "minute": 0}, + {"name": "watch", "type": "detail", "note_ids": ["n1"], "hour": 3, "minute": 30}, + ] + sched = build_scheduler(jobs=jobs, scheduler=BlockingScheduler()) + ids = {j.id for j in sched.get_jobs()} + assert ids == {"kw_daily", "watch"} + + +def test_build_scheduler_sets_cron_hour_and_minute(): + jobs = [{"name": "watch", "type": "detail", "note_ids": ["n1"], "hour": 3, "minute": 30}] + sched = build_scheduler(jobs=jobs, scheduler=BlockingScheduler()) + job = sched.get_job("watch") + fields = {f.name: str(f) for f in job.trigger.fields} + assert fields["hour"] == "3" + assert fields["minute"] == "30" + + +def test_build_scheduler_defaults_minute_to_zero(): + jobs = [{"name": "kw", "type": "search", "keywords": "x", "hour": 5}] + sched = build_scheduler(jobs=jobs, scheduler=BlockingScheduler()) + job = sched.get_job("kw") + fields = {f.name: str(f) for f in job.trigger.fields} + assert fields["minute"] == "0" diff --git a/tests/insight/test_viewer_data.py b/tests/insight/test_viewer_data.py new file mode 100644 index 000000000..2262722ae --- /dev/null +++ b/tests/insight/test_viewer_data.py @@ -0,0 +1,159 @@ +# -*- coding: utf-8 -*- +"""insight.viewer.data 纯函数单元测试。 + +仅测 data.py；app.py 不做单测（Streamlit 启动成本/收益不划算）。 +""" +import datetime +import pytest + +from insight.viewer.data import format_ts, load_comments, load_notes + + +def test_imports_smoke(): + """烟雾测试：所有公开函数可导入。""" + assert callable(format_ts) + assert callable(load_notes) + assert callable(load_comments) + + +def test_format_ts_normal_timestamp(): + """正常时间戳格式化。""" + ts = datetime.datetime(2024, 3, 15, 12, 34).timestamp() + assert format_ts(int(ts)) == "2024-03-15 12:34" + + +def test_format_ts_none_returns_dash(): + """None 返回 '—'。""" + assert format_ts(None) == "—" + + +def test_format_ts_zero_returns_dash(): + """0（未设置）返回 '—'。""" + assert format_ts(0) == "—" + + +def test_format_ts_handles_millisecond_timestamp(): + """生产数据存的是毫秒级时间戳（13 位），也要正确格式化。""" + # 1725724800 秒 = 2024-09-08 00:00 local（UTC+8）；乘 1000 转为毫秒 + assert format_ts(1725724800000) == "2024-09-08 00:00" + + +def _make_xhs_db(path) -> None: + """构造一个最小的 xhs_note 测试库。""" + import sqlite3 + conn = sqlite3.connect(str(path)) + cur = conn.cursor() + cur.execute( + """ + CREATE TABLE xhs_note ( + note_id VARCHAR(255) PRIMARY KEY, + title TEXT, + liked_count TEXT, + comment_count TEXT, + time BIGINT, + source_keyword TEXT, + nickname TEXT, + desc TEXT + ) + """ + ) + cur.executemany( + "INSERT INTO xhs_note VALUES (?,?,?,?,?,?,?,?)", + [ + ("n1", "春日穿搭", "1200", "38", 1710474840, "穿搭", "博主A", "正文1"), + ("n2", "护肤心得", "856", "12", 1710388440, "护肤", "博主B", "正文2"), + ("n3", "无时间戳笔记", "0", "0", 0, "", "博主C", ""), + ], + ) + conn.commit() + conn.close() + + +def test_load_notes_returns_sorted_by_time_desc(tmp_path): + """load_notes 按 time 倒序，time=0 的排最后。""" + from insight.viewer.data import load_notes + db = tmp_path / "t.db" + _make_xhs_db(db) + + notes = load_notes(db_path=db) + + assert len(notes) == 3 + # 倒序：n1 (1710474840) > n2 (1710388440) > n3 (0) + assert [n["note_id"] for n in notes] == ["n1", "n2", "n3"] + assert notes[0]["title"] == "春日穿搭" + assert notes[0]["source_keyword"] == "穿搭" + + +def test_load_notes_missing_table_raises(tmp_path): + """xhs_note 表不存在时抛 OperationalError（让 UI 捕获后展示）。""" + import sqlite3 + from insight.viewer.data import load_notes + db = tmp_path / "empty.db" + # 不建表 + with pytest.raises(sqlite3.OperationalError): + load_notes(db_path=db) + + +def _make_xhs_db_with_comments(path) -> None: + """构造包含 xhs_note + xhs_note_comment 的测试库。""" + import sqlite3 + conn = sqlite3.connect(str(path)) + cur = conn.cursor() + cur.execute( + """ + CREATE TABLE xhs_note ( + note_id VARCHAR(255) PRIMARY KEY, + title TEXT, liked_count TEXT, comment_count TEXT, + time BIGINT, source_keyword TEXT, nickname TEXT, desc TEXT + ) + """ + ) + cur.execute( + """ + CREATE TABLE xhs_note_comment ( + comment_id VARCHAR(255) PRIMARY KEY, + note_id VARCHAR(255), + nickname TEXT, + content TEXT, + like_count TEXT, + create_time BIGINT + ) + """ + ) + cur.executemany("INSERT INTO xhs_note VALUES (?,?,?,?,?,?,?,?)", + [("n1","A","0","0",0,"","",""), ("n2","B","0","0",0,"","","")]) + cur.executemany( + "INSERT INTO xhs_note_comment VALUES (?,?,?,?,?,?)", + [ + ("c1", "n1", "用户1", "第一条", "10", 1710475000), + ("c2", "n1", "用户2", "第二条", "5", 1710476000), + ("c3", "n2", "用户3", "另一条笔记的评论", "0", 1710477000), + ], + ) + conn.commit() + conn.close() + + +def test_load_comments_filters_by_note_id(tmp_path): + """load_comments 仅返回指定 note_id 的评论。""" + from insight.viewer.data import load_comments + db = tmp_path / "t.db" + _make_xhs_db_with_comments(db) + + rows = load_comments("n1", db_path=db) + + assert len(rows) == 2 + assert {r["comment_id"] for r in rows} == {"c1", "c2"} + # 默认按 create_time 升序 + assert [r["comment_id"] for r in rows] == ["c1", "c2"] + + +def test_load_comments_empty_when_no_match(tmp_path): + """不存在的 note_id 返回空列表，不抛异常。""" + from insight.viewer.data import load_comments + db = tmp_path / "t.db" + _make_xhs_db_with_comments(db) + + rows = load_comments("nonexistent", db_path=db) + + assert rows == [] \ No newline at end of file diff --git a/tools/cdp_browser.py b/tools/cdp_browser.py index b674f5164..52e6d238c 100644 --- a/tools/cdp_browser.py +++ b/tools/cdp_browser.py @@ -317,13 +317,13 @@ async def _connect_via_cdp(self, playwright: Playwright): try: if config.CDP_CONNECT_EXISTING: # For existing browser (e.g. chrome://inspect/#remote-debugging), - # Chrome exposes a WebSocket at /devtools/browser and may show a confirmation - # dialog to the user. Use ws:// with a longer timeout to wait for user confirmation. - ws_url = f"ws://localhost:{self.debug_port}/devtools/browser" + # Chrome exposes a WebSocket at /devtools/browser/{uuid} (NOT the bare + # /devtools/browser path -- that returns 404). Get the correct URL from + # /json/version, the same way the "launched browser" branch below does. + # NOTE: temporary fix for the upstream bug where connect-existing path + # hardcoded an invalid ws://.../devtools/browser without the UUID. + ws_url = await self._get_browser_websocket_url(self.debug_port) utils.logger.info(f"[CDPBrowserManager] Connecting to existing browser via CDP: {ws_url}") - utils.logger.info( - "[CDPBrowserManager] Please check your browser for a confirmation dialog and accept it" - ) self.browser = await playwright.chromium.connect_over_cdp( ws_url, timeout=config.BROWSER_LAUNCH_TIMEOUT * 1000 ) diff --git a/view_data.ps1 b/view_data.ps1 new file mode 100644 index 000000000..22f7b3fe9 --- /dev/null +++ b/view_data.ps1 @@ -0,0 +1,19 @@ +# view_data.ps1 — 启动小红书数据查看器（Streamlit UI） +# +# 用法：在项目根目录执行 +# .\view_data.ps1 +# +# 关键点：必须把项目根目录加进 PYTHONPATH，否则 streamlit 启动时 +# 找不到 `insight` 包（conda env 里只装了 streamlit，没装项目源码）。 +# 这个问题在 README 的启动命令里漏了，所以单独写成脚本。 + +$ErrorActionPreference = "Stop" + +# 以脚本所在目录作为项目根 —— 这样从任何位置调用都能 import 到 insight +$env:PYTHONPATH = $PSScriptRoot + +Write-Host "PYTHONPATH = $env:PYTHONPATH" -ForegroundColor Cyan +Write-Host "启动 Streamlit ... Ctrl+C 停止" -ForegroundColor Cyan +Write-Host "" + +streamlit run insight/viewer/app.py