chore: isolate legacy demo data

.env

@@ -6,6 +6,9 @@
 # Runtime
 # -----------------------------------------------------------------------------
 TZ=Asia/Shanghai
+CK_SQLITE_PATH=legacy_data/app.db
+CK_DATA_DIR=legacy_data
+CK_DEMO_DATA_DIR=legacy_data/demo
 
 # -----------------------------------------------------------------------------
 # OpenRouter / Gemini via OpenRouter for legacy video smoke only

.gitignore

@@ -18,8 +18,9 @@ dist/
 .vite/
 
 # 媒体落盘 / 运行产物 —— 体积大，本地保留不入库
-data/
-runtime/
+/data
+/legacy_data
+/runtime
 
 # 分类树 dump + embedding —— 本地保留不入库，可由 scripts/dump_trees.py + embed_trees.py 重生
 scope_trees/

acquisition/classify.py

@@ -66,8 +66,19 @@ _PROVIDER_THROTTLES: dict[str, _ProviderThrottle] = {}
 _PROVIDER_THROTTLES_LOCK = threading.Lock()
 
 
-def _data_url(public_path: str):
-    fs = ROOT / public_path.lstrip("/")
+def _data_root(settings: Settings) -> Path:
+    p = Path(settings.data_dir or "data")
+    return p if p.is_absolute() else ROOT / p
+
+
+def _public_path_to_file(public_path: str, settings: Settings) -> Path:
+    if public_path.startswith("/data/"):
+        return _data_root(settings) / public_path.removeprefix("/data/")
+    return ROOT / public_path.lstrip("/")
+
+
+def _data_url(public_path: str, settings: Settings):
+    fs = _public_path_to_file(public_path, settings)
     if not fs.exists():
         return None
     return "data:image/jpeg;base64," + base64.b64encode(fs.read_bytes()).decode()
@@ -292,7 +303,7 @@ def classify_imgtext(p: dict, settings: Settings) -> tuple:
     user = [{"type": "text", "text": f"平台：{p.get('platform')}\n标题：{p.get('title', '')}\n"
              f"正文：{(p.get('body_text') or '')[:1500]}\n（下附帖子图片，请一并看完）"}]
     for im in (p.get("images") or [])[:MAX_CARDS]:
-        u = _data_url(im)
+        u = _data_url(im, settings)
         if u:
             user.append({"type": "image_url", "image_url": {"url": u}})
     messages = [{"role": "system", "content": load_prompt("classify_imgtext")},
@@ -309,7 +320,7 @@ def classify_video(p: dict, settings: Settings) -> tuple:
     if _is_http_url(rel):
         media = rel
     else:
-        mp4 = ROOT / rel.lstrip("/")
+        mp4 = _public_path_to_file(rel, settings)
         if not rel or not mp4.exists():
             return None, "无视频", "", ""
         use = _compress(mp4) if mp4.stat().st_size > COMPRESS_OVER_MB * 1048576 else mp4

acquisition/creation_search.py

@@ -98,6 +98,11 @@ def _dedup(values: Iterable[str]) -> list[str]:
     return out
 
 
+def _data_root(settings: Settings) -> Path:
+    p = Path(settings.data_dir or "data")
+    return p if p.is_absolute() else ROOT / p
+
+
 def _save_images(urls: list[str], *, platform: str, base_dir: Path, public_base: str,
                  downloader: Downloader = _default_download, max_images: int = 12) -> list[str]:
     out: list[str] = []
@@ -180,7 +185,7 @@ def search_candidates(platform: str, query: str, *, settings: Settings, limit: i
 def _process_xhs(candidate: Candidate, *, query_hash: str, settings: Settings,
                  rate_limiter: PlatformRateLimiter, downloader: Downloader) -> dict:
     post = fetch_post_detail(candidate.source_id or candidate.url, settings=settings, rate_limiter=rate_limiter)
-    base = ROOT / "data" / "media" / "xiaohongshu" / query_hash / (post.content_id or candidate.source_id)
+    base = _data_root(settings) / "media" / "xiaohongshu" / query_hash / (post.content_id or candidate.source_id)
     pub = f"/data/media/xiaohongshu/{query_hash}/{post.content_id or candidate.source_id}"
     images = _save_images(post.image_urls or [candidate.cover_url], platform="xiaohongshu",
                           base_dir=base, public_base=pub, downloader=downloader)
@@ -205,7 +210,7 @@ def _process_weixin(candidate: Candidate, *, query_hash: str, settings: Settings
         candidate.url, settings=settings, rate_limiter=rate_limiter
     )
     h = _hash(candidate.url)
-    base = ROOT / "data" / "media" / "weixin" / query_hash / h
+    base = _data_root(settings) / "media" / "weixin" / query_hash / h
     pub = f"/data/media/weixin/{query_hash}/{h}"
     images = _save_images(detail_images or [candidate.cover_url], platform="weixin",
                           base_dir=base, public_base=pub, downloader=downloader)

acquisition/store.py

@@ -10,10 +10,13 @@
 from __future__ import annotations
 
 import json
+import os
 import sqlite3
 from pathlib import Path
 from typing import Any, Iterable, Optional
 
+from core.config import load_env_file
+
 ROOT = Path(__file__).resolve().parent.parent
 DB_PATH = ROOT / "data" / "app.db"
 
@@ -134,9 +137,24 @@ _MIGRATIONS = [
 ]
 
 
-def connect(db_path: Path | str = DB_PATH) -> sqlite3.Connection:
+def resolve_db_path(db_path: Path | str | None = None, *, env_file: str | Path | None = None) -> Path:
+    """Resolve the SQLite path at call time.
+
+    `DB_PATH` stays as the default new-data location for compatibility, while
+    CK_SQLITE_PATH can point old demo/API runs at legacy_data/app.db.
+    """
+    if db_path:
+        p = Path(db_path)
+    else:
+        source = load_env_file(env_file or os.getenv("CK_ENV_FILE", ".env"))
+        raw = os.getenv("CK_SQLITE_PATH") or source.get("CK_SQLITE_PATH") or str(DB_PATH)
+        p = Path(raw)
+    return p if p.is_absolute() else ROOT / p
+
+
+def connect(db_path: Path | str | None = None) -> sqlite3.Connection:
     """打开（必要时建）库，建表，行按 dict 取。busy_timeout 让多进程并发写不报锁。"""
-    p = Path(db_path)
+    p = resolve_db_path(db_path)
     p.parent.mkdir(parents=True, exist_ok=True)
     conn = sqlite3.connect(str(p), timeout=30)
     conn.row_factory = sqlite3.Row

core/config.py

@@ -104,6 +104,7 @@ class Settings:
     bailian_vl_model: str = "qwen-vl-plus"
     bailian_text_model: str = "qwen-plus"
     bailian_timeout_seconds: float = 120.0
+    demo_data_dir: str = ""
 
     @classmethod
     def from_env(cls, env_file: str | Path = ".env") -> "Settings":
@@ -157,6 +158,10 @@ class Settings:
             frames_dir=env_value("CK_FRAMES_DIR", file_env, "runtime/frames"),
             douyin_ratio=env_value("CK_DOUYIN_RATIO", file_env, "540p"),
             data_dir=env_value("CK_DATA_DIR", file_env, "data"),
+            demo_data_dir=env_value(
+                "CK_DEMO_DATA_DIR", file_env,
+                str(Path(env_value("CK_DATA_DIR", file_env, "data")) / "demo"),
+            ),
             oss_upload_url=env_value(
                 "CRAWLER_OSS_UPLOAD_URL", file_env,
                 "http://crawler-upload-v2.aiddit.com/crawler/oss/upload_stream",

scripts/decompose.py

@@ -31,7 +31,6 @@ from scripts.scope_link import ScopeLinker
 
 ROOT = Path(__file__).resolve().parent.parent
 FIX = ROOT / "tests" / "fixtures"
-DATA = ROOT / "data" / "demo"
 SKILL = ROOT / "创作知识提取-skill"
 PHASE1 = (SKILL / "extraction" / "phase1-frame.md").read_text(encoding="utf-8")
 PHASE2 = (SKILL / "extraction" / "phase2-scope.md").read_text(encoding="utf-8")
@@ -95,6 +94,11 @@ def _image_data_url(path: Path) -> str:
     return f"data:{mime};base64,{data}"
 
 
+def _demo_data_root(settings: Settings) -> Path:
+    p = Path(settings.demo_data_dir or (Path(settings.data_dir or "data") / "demo"))
+    return p if p.is_absolute() else ROOT / p
+
+
 def read_one(src: dict, post, settings: Settings, extractor: BailianExtractor):
     live = src["from"] == "live"
     if post.video_urls:  # 视频帖
@@ -108,7 +112,7 @@ def read_one(src: dict, post, settings: Settings, extractor: BailianExtractor):
             ec = video_extract.extract_video(post, settings=settings, oss_video_url=oss_url, public_url=oss_url)
             pub = oss_url
         else:        # fixture / 转存失败：下载 mp4 + base64 兜底（落盘按平台分目录，post.id 已含平台前缀）
-            save = DATA / post.platform / post.id / "video.mp4"
+            save = _demo_data_root(settings) / post.platform / post.id / "video.mp4"
             pub = f"/data/demo/{post.platform}/{post.id}/video.mp4"
             ec = video_extract.extract_video(post, settings=settings, save_path=save, public_url=pub)
         media = {"type": "video", "video_url": pub, "images": []}
@@ -124,7 +128,7 @@ def read_one(src: dict, post, settings: Settings, extractor: BailianExtractor):
                     c.url = imgs[c.index - 1]
             ec = extractor.extract(post)
         else:     # fixture：把预下载图片用 data URL 喂模型，展示仍用 /data 路径
-            local_dir = DATA / "xiaohongshu" / post.id
+            local_dir = _demo_data_root(settings) / "xiaohongshu" / post.id
             display_imgs = [
                 f"/data/demo/xiaohongshu/{post.id}/image_{n}.webp"
                 for n in range(1, len(post.image_urls) + 1)

scripts/run_creation_search.py

@@ -41,7 +41,7 @@ def _parse_platforms(value: str | None) -> list[str]:
 def run_platform_workers(*, run_id: str, queries: list[str], platforms: list[str],
                          settings: Settings, search_limit: int, display_limit: int,
                          classify: bool, skip_done: bool,
-                         db_path=store.DB_PATH) -> int:
+                         db_path: Path | str | None = None) -> int:
     """Run long-lived platform workers. Return failed platform-job count."""
 
     def _platform_worker(platform: str) -> tuple[str, int, int]:
@@ -109,12 +109,14 @@ def main() -> None:
     ap.add_argument("--search-limit", type=int, default=10)
     ap.add_argument("--display-limit", type=int, default=5)
     ap.add_argument("--query-file", default=str(Path("data/queries/creation_demo.json")))
+    ap.add_argument("--db-path", help="SQLite 结果库路径；默认读 CK_SQLITE_PATH，否则 data/app.db")
     ap.add_argument("--no-classify", action="store_true", help="只采集不调用 AI 判断")
     ap.add_argument("--resume", action="store_true", help="保留同 run_id 已有结果，继续未完成 job")
     ap.add_argument("--skip-done", action="store_true", help="跳过已 done 且 display_count>=display_limit 的 job")
     args = ap.parse_args()
 
     settings = Settings.from_env(os.getenv("CK_ENV_FILE", ".env"))
+    db_path = store.resolve_db_path(args.db_path)
     queries = [args.query.strip()] if args.query else load_creation_queries(args.query_file)
     if args.limit_queries and args.limit_queries > 0:
         queries = queries[:args.limit_queries]
@@ -122,7 +124,7 @@ def main() -> None:
     if not queries:
         raise SystemExit("no queries to run")
 
-    conn = store.connect()
+    conn = store.connect(db_path)
     if args.resume and store.creation_run_exists(conn, args.run_id):
         print(f"resume run_id={args.run_id}")
     else:
@@ -150,9 +152,10 @@ def main() -> None:
         display_limit=args.display_limit,
         classify=not args.no_classify,
         skip_done=args.skip_done or args.resume,
+        db_path=db_path,
     )
 
-    c = store.connect()
+    c = store.connect(db_path)
     store.finish_creation_run(c, args.run_id, status="finished" if failed == 0 else "partial", ts=int(time.time()))
     c.close()
     print(f"done run_id={args.run_id} failed_platform_jobs={failed}")

scripts/serve_old_web.py

@@ -6,6 +6,7 @@ This tiny server preserves those paths without changing the old frontend.
 """
 from __future__ import annotations
 
+import os
 from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer
 from pathlib import Path
 from urllib.parse import unquote, urlparse
@@ -17,6 +18,31 @@ PROMPTS = ROOT / "prompts"
 SKILL_EXTRACTION = ROOT / "创作知识提取-skill" / "extraction"
 
 
+def _load_env_file(path: str | Path) -> dict[str, str]:
+    p = Path(path)
+    if not p.is_absolute():
+        p = ROOT / p
+    if not p.exists():
+        return {}
+    out: dict[str, str] = {}
+    for line in p.read_text(encoding="utf-8").splitlines():
+        stripped = line.strip()
+        if not stripped or stripped.startswith("#") or "=" not in stripped:
+            continue
+        key, value = stripped.split("=", 1)
+        out[key.strip()] = value.strip().strip('"').strip("'")
+    return out
+
+
+def _data_root() -> Path:
+    env = _load_env_file(os.getenv("CK_ENV_FILE", ".env"))
+    raw = os.getenv("CK_DATA_DIR") or env.get("CK_DATA_DIR")
+    if not raw:
+        raw = "legacy_data" if (ROOT / "legacy_data").exists() else "data"
+    p = Path(raw)
+    return p if p.is_absolute() else ROOT / p
+
+
 class LegacyDemoHandler(SimpleHTTPRequestHandler):
     def translate_path(self, path: str) -> str:
         parsed = urlparse(path)
@@ -24,7 +50,7 @@ class LegacyDemoHandler(SimpleHTTPRequestHandler):
         if clean in {"", "/"}:
             return str(WEB / "index.html")
         if clean.startswith("/data/"):
-            return str(ROOT / clean.lstrip("/"))
+            return str(_data_root() / clean.removeprefix("/data/"))
         if clean.startswith("/runs/"):
             return str(WEB / clean.lstrip("/"))
         if clean.startswith("/prompt-read/"):

tests/test_creation_search.py

@@ -3,10 +3,12 @@ from __future__ import annotations
 import httpx
 
 import acquisition.classify as classify_module
+import acquisition.creation_search as creation_search_module
 from acquisition import store
-from acquisition.classify import _providers, classify_video
+from acquisition.classify import _providers, classify_imgtext, classify_video
 from acquisition.creation_search import Candidate, PlatformRateLimiter, run_platform_query
 from acquisition.crawler import RateLimiter
+from core.models import Post
 from core.config import PgConfig, Settings
 from scripts import run_creation_search as run_cli
 
@@ -129,6 +131,27 @@ def test_classify_video_accepts_cdn_url(monkeypatch):
     assert seen["url"] == "https://cdn.test/video.mp4"
 
 
+def test_classify_imgtext_maps_public_data_to_settings_data_dir(tmp_path, monkeypatch):
+    legacy = tmp_path / "legacy_data"
+    img = legacy / "media" / "x.jpg"
+    img.parent.mkdir(parents=True)
+    img.write_bytes(b"\xff\xd8fake-jpeg")
+    settings = _settings()
+    settings.data_dir = str(legacy)
+    seen = {}
+
+    def fake_judge(messages, settings, timeout):
+        seen["parts"] = messages[1]["content"]
+        return 1, "ok", "knowledge", ""
+
+    monkeypatch.setattr("acquisition.classify._judge", fake_judge)
+    res = classify_imgtext({"images": ["/data/media/x.jpg"]}, settings)
+
+    assert res == (1, "ok", "knowledge", "")
+    image_parts = [p for p in seen["parts"] if p["type"] == "image_url"]
+    assert image_parts and image_parts[0]["image_url"]["url"].startswith("data:image/jpeg;base64,")
+
+
 def test_qwen_provider_uses_dashscope_credentials(monkeypatch):
     monkeypatch.setenv("CLASSIFY_PROVIDER", "qwen")
     monkeypatch.setenv("CLASSIFY_MODEL", "qwen3.7-plus")
@@ -228,6 +251,30 @@ def test_run_platform_query_skips_bad_items_and_fills_display_limit(tmp_path, mo
     assert len(detail["platforms"]["weixin"]["items"]) == 5
 
 
+def test_xhs_media_saved_under_settings_data_dir(tmp_path, monkeypatch):
+    settings = _settings()
+    settings.data_dir = str(tmp_path / "legacy_data")
+    post = Post(
+        id="xhs_1", platform="xiaohongshu", url="https://xhs.test/1",
+        content_id="cid1", title="t", body_text="b", image_urls=["https://img.test/a.jpg"],
+    )
+
+    monkeypatch.setattr(creation_search_module, "fetch_post_detail", lambda *a, **k: post)
+    out = creation_search_module._process_xhs(
+        Candidate(rank=1, source_id="cid1"),
+        query_hash="qh",
+        settings=settings,
+        rate_limiter=PlatformRateLimiter("xiaohongshu", delegate=RateLimiter(
+            min_interval_seconds=0, max_interval_seconds=0, sleep_fn=lambda _: None,
+        )),
+        downloader=lambda url, platform: b"\xff\xd8fake-jpeg",
+    )
+
+    expected = tmp_path / "legacy_data" / "media" / "xiaohongshu" / "qh" / "cid1" / "image_1.jpg"
+    assert expected.exists()
+    assert out["image_urls"] == ["/data/media/xiaohongshu/qh/cid1/image_1.jpg"]
+
+
 def test_creation_job_done_and_classification_queue(tmp_path):
     c = store.connect(tmp_path / "app.db")
     store.create_creation_run(c, "r1", total_queries=1, ts=1)
@@ -311,3 +358,31 @@ def test_run_workers_skip_done_does_not_rerun(tmp_path, monkeypatch):
     )
     assert failed == 0
     assert calls == []
+
+
+def test_run_workers_resolves_env_db_path(tmp_path, monkeypatch):
+    db = tmp_path / "legacy_data" / "app.db"
+    c = store.connect(db)
+    store.create_creation_run(c, "r1", total_queries=1, ts=1)
+    store.ensure_creation_job(c, "r1", "q", "weixin", ts=1)
+    c.close()
+    monkeypatch.setenv("CK_SQLITE_PATH", str(db))
+
+    def fake_run(conn, *, platform, run_id, query, display_limit, **kwargs):
+        store.update_creation_job(
+            conn, run_id, query, platform, status="done",
+            searched_count=10, display_count=display_limit, ts=2,
+        )
+        return {"platform": platform, "status": "done", "display_count": display_limit, "error": ""}
+
+    monkeypatch.setattr(run_cli, "run_platform_query", fake_run)
+    failed = run_cli.run_platform_workers(
+        run_id="r1", queries=["q"], platforms=["weixin"],
+        settings=_settings(), search_limit=10, display_limit=5,
+        classify=False, skip_done=False,
+    )
+
+    c = store.connect(db)
+    summary = store.creation_search_summary(c, run_id="r1")["queries"]["q"]["platforms"]
+    assert failed == 0
+    assert summary["weixin"]["status"] == "done"

tests/test_legacy_paths.py

@@ -0,0 +1,19 @@
+from __future__ import annotations
+
+from scripts.serve_old_web import LegacyDemoHandler, _data_root
+
+
+def test_serve_old_web_data_root_uses_ck_data_dir(tmp_path, monkeypatch):
+    legacy = tmp_path / "legacy_data"
+    monkeypatch.setenv("CK_DATA_DIR", str(legacy))
+
+    assert _data_root() == legacy
+
+
+def test_serve_old_web_maps_data_url_to_legacy_root(tmp_path, monkeypatch):
+    legacy = tmp_path / "legacy_data"
+    monkeypatch.setenv("CK_DATA_DIR", str(legacy))
+
+    mapped = LegacyDemoHandler.translate_path(object(), "/data/demo/x/image_1.webp")
+
+    assert mapped == str(legacy / "demo" / "x" / "image_1.webp")

tests/test_store.py

@@ -8,6 +8,18 @@ def _db(tmp_path):
     return store.connect(tmp_path / "t.db")
 
 
+def test_connect_uses_ck_sqlite_path(tmp_path, monkeypatch):
+    db = tmp_path / "legacy_data" / "app.db"
+    monkeypatch.setenv("CK_SQLITE_PATH", str(db))
+
+    c = store.connect()
+    try:
+        assert db.exists()
+        assert store.list_runs(c) == []
+    finally:
+        c.close()
+
+
 def test_insert_queries_idempotent_and_axes(tmp_path):
     c = _db(tmp_path)
     items = [{"query": "q1", "实质": "军人", "阶段": "灵感"}, {"query": "q2", "形式": "排比"}]