feat(acquisition): migrate query and platform adapters

acquisition/classification/__init__.py

@@ -0,0 +1,17 @@
+"""Formal coarse-classification package."""
+
+from acquisition.classification.coarse import (
+    ClassificationResult,
+    classify_imgtext,
+    classify_video,
+    coarse_classify_item,
+    prompt_version,
+)
+
+__all__ = [
+    "ClassificationResult",
+    "classify_imgtext",
+    "classify_video",
+    "coarse_classify_item",
+    "prompt_version",
+]

acquisition/classification/coarse.py

@@ -0,0 +1,130 @@
+"""Formal coarse classifier for creation-knowledge candidates."""
+from __future__ import annotations
+
+import hashlib
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any
+
+from acquisition.classify import (
+    MAX_CARDS,
+    _data_url,
+    _is_http_url,
+    _judge,
+    classify_video as _legacy_classify_video,
+)
+from core.config import Settings
+from core.prompts import load_prompt
+
+ROOT = Path(__file__).resolve().parents[2]
+
+
+@dataclass(frozen=True)
+class ClassificationResult:
+    is_creation_knowledge: bool | None
+    label: str | None
+    confidence: float | None
+    reason: str
+    knowledge: str = ""
+    prompt_version: str | None = None
+    result_payload: dict[str, Any] = field(default_factory=dict)
+    status: str = "classified"
+    error_message: str | None = None
+
+
+def prompt_version(*names: str) -> str:
+    h = hashlib.sha256()
+    for name in names:
+        path = ROOT / "prompts" / f"{name}.txt"
+        h.update(name.encode("utf-8"))
+        if path.exists():
+            h.update(path.read_bytes())
+    return h.hexdigest()[:16]
+
+
+def classify_imgtext(payload: dict[str, Any], settings: Settings) -> tuple:
+    """Classify image-text content, accepting both HTTP image URLs and /data paths."""
+    user = [
+        {
+            "type": "text",
+            "text": (
+                f"平台：{payload.get('platform')}\n"
+                f"标题：{payload.get('title', '')}\n"
+                f"正文：{(payload.get('body_text') or '')[:1500]}\n"
+                "（下附帖子图片，请一并看完）"
+            ),
+        }
+    ]
+    for image in (payload.get("images") or [])[:MAX_CARDS]:
+        if _is_http_url(image):
+            user.append({"type": "image_url", "image_url": {"url": image}})
+            continue
+        data_url = _data_url(image, settings)
+        if data_url:
+            user.append({"type": "image_url", "image_url": {"url": data_url}})
+    messages = [
+        {"role": "system", "content": load_prompt("classify_imgtext")},
+        {"role": "user", "content": user},
+    ]
+    return _judge(messages, settings, timeout=120)
+
+
+def classify_video(payload: dict[str, Any], settings: Settings) -> tuple:
+    return _legacy_classify_video(payload, settings)
+
+
+def coarse_classify_item(
+    *,
+    platform: str,
+    title: str = "",
+    body_text: str = "",
+    image_urls: list[str] | None = None,
+    video_url: str = "",
+    settings: Settings,
+) -> ClassificationResult:
+    if platform == "douyin" or video_url:
+        version = prompt_version("classify_video")
+        is_creation, reason, knowledge, points = classify_video(
+            {
+                "platform": platform,
+                "title": title,
+                "body_text": body_text,
+                "video": video_url,
+            },
+            settings,
+        )
+    else:
+        version = prompt_version("classify_imgtext")
+        is_creation, reason, knowledge, points = classify_imgtext(
+            {
+                "platform": platform,
+                "title": title,
+                "body_text": body_text,
+                "images": image_urls or [],
+            },
+            settings,
+        )
+
+    if is_creation is None:
+        return ClassificationResult(
+            is_creation_knowledge=None,
+            label=None,
+            confidence=None,
+            reason=reason,
+            knowledge=knowledge,
+            prompt_version=version,
+            result_payload={"knowledge": knowledge, "points": points},
+            status="failed",
+            error_message=reason,
+        )
+    is_hit = bool(is_creation)
+    return ClassificationResult(
+        is_creation_knowledge=is_hit,
+        label="creation" if is_hit else "not_creation",
+        confidence=1.0,
+        reason=reason,
+        knowledge=knowledge,
+        prompt_version=version,
+        result_payload={"knowledge": knowledge, "points": points},
+        status="classified",
+    )

acquisition/classify.py

@@ -6,18 +6,15 @@
   图文（小红书/微信）：prompts/classify_imgtext.txt（标题+正文+全部图喂多模态模型）
   视频（抖音）：prompts/classify_video.txt（OSS/CDN URL 直喂；本地大视频 fallback 时先压 480p 保音轨）
 都输出 {is_empty, reason, knowledge}：is_empty 即创作闸；is_empty=false 时连带提炼出具体创作知识点。
-结果按 url 写 app.db 的 post_class（upsert 覆盖，可重判）。并发；与微信补下载并行安全（busy_timeout）。
-只读 prompts、走 OpenRouter / Ark / Qwen——不改 skill/creation_knowledge，不解构、不入 ingest。
-用法：PYTHONPATH=. CK_ENV_FILE=.env python -m acquisition.classify [平台名...]
+正式链路只复用本模块的图文/视频判定函数；旧 SQLite 批量补判 CLI 已归档。
+只读 prompts、走 OpenRouter / Ark / Qwen，不解构、不入 ingest。
 """
 from __future__ import annotations
 
 import base64
-import concurrent.futures as cf
 import json
 import os
 import subprocess
-import sys
 import tempfile
 import threading
 import time
@@ -26,7 +23,6 @@ from urllib.parse import urlparse
 
 import httpx
 
-from acquisition import store
 from core.config import Settings, load_env_file
 from core.prompts import load_prompt
 
@@ -303,6 +299,9 @@ 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]:
+        if _is_http_url(im):
+            user.append({"type": "image_url", "image_url": {"url": im}})
+            continue
         u = _data_url(im, settings)
         if u:
             user.append({"type": "image_url", "image_url": {"url": u}})
@@ -338,48 +337,8 @@ def classify_video(p: dict, settings: Settings) -> tuple:
     return _judge(messages, settings, timeout=300)
 
 
-def _safe(fn, *a) -> tuple:
-    try:
-        return fn(*a)
-    except Exception as exc:
-        return None, f"判定失败: {str(exc)[:60]}", "", ""
-
-
-def main() -> None:
-    settings = Settings.from_env()
-    conn = store.connect()
-    platforms = sys.argv[1:] or PLATFORMS
-    posts = store.posts_to_classify(conn, platforms)
-    imgs = [p for p in posts if p["platform"] != "douyin"]
-    vids = [p for p in posts if p["platform"] == "douyin"]
-    total = len(posts)
-    print(f"收紧重判：图文 {len(imgs)}（并发{IMG_WORKERS}）+ 抖音视频 {len(vids)}（并发{VID_WORKERS}，大视频先压）")
-    ts = int(time.time())
-    done = {"n": 0, "fail": 0}
-
-    def _write(p, res):
-        ic, reason, knowledge, points = res
-        if ic is None:
-            done["fail"] += 1
-        else:
-            store.upsert_class(conn, p["url"], ic, reason, ts, knowledge, points)
-        done["n"] += 1
-        if done["n"] % 20 == 0:
-            print(f"  {done['n']}/{total}（失败 {done['fail']}）")
-
-    with cf.ThreadPoolExecutor(IMG_WORKERS) as ex:
-        futs = {ex.submit(_safe, classify_imgtext, p, settings): p for p in imgs}
-        for fut in cf.as_completed(futs):
-            _write(futs[fut], fut.result())
-    with cf.ThreadPoolExecutor(VID_WORKERS) as ex:
-        futs = {ex.submit(_safe, classify_video, p, settings): p for p in vids}
-        for fut in cf.as_completed(futs):
-            _write(futs[fut], fut.result())
-
-    c = store.class_counts(conn)
-    conn.close()
-    print(f"完成：创作知识 {c['creation']} / 非创作知识 {c['non_creation']}（本轮失败 {done['fail']}，可重跑补判）")
-
-
 if __name__ == "__main__":
-    main()
+    raise SystemExit(
+        "acquisition.classify now exposes classifier functions only; "
+        "use acquisition/classification/coarse.py or the formal acquisition runner."
+    )

acquisition/media/__init__.py

@@ -0,0 +1,5 @@
+"""Formal media stabilization package."""
+
+from acquisition.media.service import StabilizedMedia, stabilize_media_urls
+
+__all__ = ["StabilizedMedia", "stabilize_media_urls"]

acquisition/media/oss_client.py

@@ -0,0 +1,6 @@
+"""Formal OSS client wrapper for media stabilization."""
+from __future__ import annotations
+
+from acquisition.oss import OssError, parse_oss_response, to_oss, upload_stream
+
+__all__ = ["OssError", "parse_oss_response", "to_oss", "upload_stream"]

acquisition/media/service.py

@@ -0,0 +1,75 @@
+"""Media stabilization service for the formal acquisition pipeline."""
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Callable
+
+from acquisition.media.oss_client import to_oss
+from core.config import Settings
+
+
+@dataclass(frozen=True)
+class StabilizedMedia:
+    media_type: str
+    source_url: str
+    cdn_url: str
+    position: int
+    status: str = "ready"
+
+
+Uploader = Callable[[str, str], str]
+
+
+def stabilize_media_urls(
+    *,
+    image_urls: list[str] | None = None,
+    video_urls: list[str] | None = None,
+    settings: Settings | None = None,
+    uploader: Uploader | None = None,
+) -> list[StabilizedMedia]:
+    """Transfer image/video URLs to OSS/CDN and return DB-ready media rows.
+
+    The uploader falls back to the original source URL when OSS transfer fails,
+    so acquisition can still store the candidate and make the failure visible in
+    later review instead of dropping the item.
+    """
+    upload = uploader or (
+        lambda url, media_type: to_oss(url, media_type, settings=settings)
+    )
+    rows: list[StabilizedMedia] = []
+    position = 0
+    for url in image_urls or []:
+        if not url:
+            continue
+        position += 1
+        try:
+            cdn = upload(url, "image")
+        except Exception:
+            cdn = url
+        rows.append(
+            StabilizedMedia(
+                media_type="image",
+                source_url=url,
+                cdn_url=cdn,
+                position=position,
+                status="ready" if cdn else "failed",
+            )
+        )
+    for url in video_urls or []:
+        if not url:
+            continue
+        position += 1
+        try:
+            cdn = upload(url, "video")
+        except Exception:
+            cdn = url
+        rows.append(
+            StabilizedMedia(
+                media_type="video",
+                source_url=url,
+                cdn_url=cdn,
+                position=position,
+                status="ready" if cdn else "failed",
+            )
+        )
+    return rows

acquisition/platforms/__init__.py

@@ -0,0 +1,27 @@
+"""Formal platform adapters package."""
+
+from acquisition.platforms.base import PlatformAdapter, PlatformCandidate, PlatformItem
+from acquisition.platforms.douyin import DouyinAdapter
+from acquisition.platforms.weixin import WeixinAdapter
+from acquisition.platforms.xiaohongshu import XiaohongshuAdapter
+
+
+def get_platform_adapter(platform: str) -> PlatformAdapter:
+    if platform == "xiaohongshu":
+        return XiaohongshuAdapter()
+    if platform == "weixin":
+        return WeixinAdapter()
+    if platform == "douyin":
+        return DouyinAdapter()
+    raise ValueError(f"unsupported platform: {platform}")
+
+
+__all__ = [
+    "PlatformAdapter",
+    "PlatformCandidate",
+    "PlatformItem",
+    "DouyinAdapter",
+    "WeixinAdapter",
+    "XiaohongshuAdapter",
+    "get_platform_adapter",
+]

acquisition/platforms/base.py

@@ -0,0 +1,59 @@
+"""Formal platform adapter contracts for acquisition."""
+from __future__ import annotations
+
+from typing import Any, Protocol
+
+from pydantic import BaseModel, ConfigDict, Field
+
+from core.config import Settings
+
+
+class PlatformCandidate(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    rank: int
+    platform: str
+    source_id: str = ""
+    url: str = ""
+    title: str = ""
+    author: str = ""
+    cover_url: str = ""
+    raw: dict[str, Any] = Field(default_factory=dict)
+
+
+class PlatformItem(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    platform: str
+    source_id: str = ""
+    url: str = ""
+    content_type: str = ""
+    title: str = ""
+    author: str = ""
+    body_text: str = ""
+    image_urls: list[str] = Field(default_factory=list)
+    video_urls: list[str] = Field(default_factory=list)
+    raw: dict[str, Any] = Field(default_factory=dict)
+
+
+class PlatformAdapter(Protocol):
+    platform: str
+
+    def search(
+        self,
+        query: str,
+        *,
+        settings: Settings,
+        limit: int,
+        rate_limiter: Any,
+    ) -> list[PlatformCandidate]:
+        ...
+
+    def fetch_detail(
+        self,
+        candidate: PlatformCandidate,
+        *,
+        settings: Settings,
+        rate_limiter: Any,
+    ) -> PlatformItem:
+        ...

acquisition/platforms/douyin.py

@@ -0,0 +1,64 @@
+"""Douyin formal acquisition adapter."""
+from __future__ import annotations
+
+from typing import Any
+
+from acquisition.crawler import fetch_post_detail
+from acquisition.platforms.base import PlatformCandidate, PlatformItem
+from acquisition.search import search_keyword
+from core.config import Settings
+
+
+class DouyinAdapter:
+    platform = "douyin"
+
+    def search(
+        self,
+        query: str,
+        *,
+        settings: Settings,
+        limit: int,
+        rate_limiter: Any,
+    ) -> list[PlatformCandidate]:
+        ids = search_keyword(
+            query,
+            platform=self.platform,
+            content_type="视频",
+            limit=limit,
+            settings=settings,
+            rate_limiter=rate_limiter,
+        )
+        return [
+            PlatformCandidate(
+                rank=i,
+                platform=self.platform,
+                source_id=content_id,
+                raw={"id": content_id},
+            )
+            for i, content_id in enumerate(ids, start=1)
+        ]
+
+    def fetch_detail(
+        self,
+        candidate: PlatformCandidate,
+        *,
+        settings: Settings,
+        rate_limiter: Any,
+    ) -> PlatformItem:
+        post = fetch_post_detail(
+            candidate.source_id,
+            settings=settings,
+            rate_limiter=rate_limiter,
+        )
+        return PlatformItem(
+            platform=self.platform,
+            source_id=post.content_id or candidate.source_id,
+            url=post.url or candidate.url,
+            content_type=post.content_type or "video",
+            title=post.title or candidate.title,
+            author=post.author_name or candidate.author,
+            body_text=post.body_text or "",
+            image_urls=post.image_urls,
+            video_urls=post.video_urls,
+            raw=post.raw if isinstance(post.raw, dict) else {},
+        )

acquisition/platforms/weixin.py

@@ -0,0 +1,69 @@
+"""Weixin formal acquisition adapter."""
+from __future__ import annotations
+
+import hashlib
+from typing import Any
+
+from acquisition.platforms.base import PlatformCandidate, PlatformItem
+from acquisition.search import fetch_weixin_detail, search_weixin
+from core.config import Settings
+
+
+def _hash(value: str, n: int = 16) -> str:
+    return hashlib.md5(value.encode("utf-8")).hexdigest()[:n]
+
+
+class WeixinAdapter:
+    platform = "weixin"
+
+    def search(
+        self,
+        query: str,
+        *,
+        settings: Settings,
+        limit: int,
+        rate_limiter: Any,
+    ) -> list[PlatformCandidate]:
+        rows = search_weixin(
+            query,
+            limit=limit,
+            settings=settings,
+            rate_limiter=rate_limiter,
+        )
+        return [
+            PlatformCandidate(
+                rank=i,
+                platform=self.platform,
+                source_id=_hash(row.get("url") or ""),
+                url=row.get("url") or "",
+                title=row.get("title") or "",
+                author=row.get("nick_name") or "",
+                cover_url=row.get("cover_url") or "",
+                raw=row,
+            )
+            for i, row in enumerate(rows, start=1)
+        ]
+
+    def fetch_detail(
+        self,
+        candidate: PlatformCandidate,
+        *,
+        settings: Settings,
+        rate_limiter: Any,
+    ) -> PlatformItem:
+        body_text, images = fetch_weixin_detail(
+            candidate.url,
+            settings=settings,
+            rate_limiter=rate_limiter,
+        )
+        return PlatformItem(
+            platform=self.platform,
+            source_id=candidate.source_id or _hash(candidate.url),
+            url=candidate.url,
+            content_type="图文",
+            title=candidate.title,
+            author=candidate.author,
+            body_text=body_text,
+            image_urls=images or [candidate.cover_url],
+            raw=candidate.raw,
+        )

acquisition/platforms/xiaohongshu.py

@@ -0,0 +1,67 @@
+"""Xiaohongshu formal acquisition adapter."""
+from __future__ import annotations
+
+from typing import Any
+
+from acquisition.crawler import fetch_post_detail
+from acquisition.platforms.base import PlatformCandidate, PlatformItem
+from acquisition.search import search_xiaohongshu
+from core.config import Settings
+
+
+class XiaohongshuAdapter:
+    platform = "xiaohongshu"
+
+    def search(
+        self,
+        query: str,
+        *,
+        settings: Settings,
+        limit: int,
+        rate_limiter: Any,
+    ) -> list[PlatformCandidate]:
+        rows = search_xiaohongshu(
+            query,
+            content_type=settings.search_content_type,
+            limit=limit,
+            settings=settings,
+            rate_limiter=rate_limiter,
+        )
+        return [
+            PlatformCandidate(
+                rank=i,
+                platform=self.platform,
+                source_id=row.get("id") or "",
+                url=row.get("url") or "",
+                title=row.get("title") or "",
+                author=row.get("nick_name") or "",
+                cover_url=row.get("cover_url") or "",
+                raw=row,
+            )
+            for i, row in enumerate(rows, start=1)
+        ]
+
+    def fetch_detail(
+        self,
+        candidate: PlatformCandidate,
+        *,
+        settings: Settings,
+        rate_limiter: Any,
+    ) -> PlatformItem:
+        post = fetch_post_detail(
+            candidate.source_id or candidate.url,
+            settings=settings,
+            rate_limiter=rate_limiter,
+        )
+        return PlatformItem(
+            platform=self.platform,
+            source_id=post.content_id or candidate.source_id,
+            url=post.url or candidate.url,
+            content_type=post.content_type or "图文",
+            title=post.title or candidate.title,
+            author=post.author_name or candidate.author,
+            body_text=post.body_text or "",
+            image_urls=post.image_urls or [candidate.cover_url],
+            video_urls=post.video_urls,
+            raw=post.raw if isinstance(post.raw, dict) else {},
+        )

acquisition/queries/__init__.py

@@ -0,0 +1,17 @@
+"""Formal query generation and filtering package."""
+
+from acquisition.queries.builder import (
+    QueryBuildOptions,
+    build_creation_query_batch,
+    persist_query_batch,
+)
+from acquisition.queries.filter import VALID_MIN, filter_queries, prompt_version
+
+__all__ = [
+    "QueryBuildOptions",
+    "build_creation_query_batch",
+    "persist_query_batch",
+    "VALID_MIN",
+    "filter_queries",
+    "prompt_version",
+]

acquisition/queries/builder.py

@@ -0,0 +1,263 @@
+"""Formal creation-query batch builder."""
+from __future__ import annotations
+
+import json
+import random
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+
+from acquisition.query import ACTIONS, STAGES as OLD_STAGES, _nonleaf_d4
+from acquisition.queries.filter import filter_queries, prompt_version
+from acquisition.repositories.base import AcquisitionRepository
+from core.config import Settings
+
+ROOT = Path(__file__).resolve().parents[2]
+TREES = ROOT / "scope_trees" / "trees_index.json"
+KTYPE = ["怎么做", "有哪些", "为什么"]
+MODALITY = ["视频", "图文"]
+INTENT = ["灵感", "选题", "脚本"]
+
+
+@dataclass(frozen=True)
+class QueryBuildOptions:
+    per: int = 30
+    batch_n: int = 30
+    seed: int = 7
+    dry: bool = False
+
+
+def _segs(path: str | None) -> list[str]:
+    return [x for x in (path or "").split("/") if x]
+
+
+def _leaves(idx: list[dict[str, Any]], source_type: str, under: str | None = None) -> list[str]:
+    paths = [(_segs(n.get("path")), n.get("name")) for n in idx if n.get("source_type") == source_type]
+    if under:
+        paths = [(s, nm) for s, nm in paths if under in s]
+    all_paths = {"/".join(s) for s, _ in paths}
+    out: list[str] = []
+    seen: set[str] = set()
+    for segs, name in paths:
+        if len(segs) < 2:
+            continue
+        full = "/".join(segs)
+        is_leaf = not any(other != full and other.startswith(full + "/") for other in all_paths)
+        value = name or segs[-1]
+        if is_leaf and value and value not in seen:
+            seen.add(value)
+            out.append(value)
+    return out
+
+
+def _nonleaf(
+    idx: list[dict[str, Any]],
+    source_type: str,
+    depths: tuple[int, ...] = (3, 4),
+    under: str | None = None,
+) -> list[str]:
+    paths = [(_segs(n.get("path")), n.get("name")) for n in idx if n.get("source_type") == source_type]
+    if under:
+        paths = [(s, nm) for s, nm in paths if under in s]
+    all_paths = {"/".join(s) for s, _ in paths}
+    out: list[str] = []
+    seen: set[str] = set()
+    for segs, name in paths:
+        if len(segs) not in depths:
+            continue
+        full = "/".join(segs)
+        is_nonleaf = any(other != full and other.startswith(full + "/") for other in all_paths)
+        value = name or (segs[-1] if segs else "")
+        if is_nonleaf and value and value not in seen:
+            seen.add(value)
+            out.append(value)
+    return out
+
+
+def build_creation_query_batch(
+    settings: Settings,
+    *,
+    tree_path: Path = TREES,
+    options: QueryBuildOptions | None = None,
+) -> dict[str, Any]:
+    """Build a creation-query batch without writing files or database rows."""
+    opts = options or QueryBuildOptions()
+    rng = random.Random(opts.seed)
+    idx = json.loads(tree_path.read_text("utf-8"))
+    shi = _nonleaf(idx, "实质", depths=(3, 4), under="理念")
+    xing = _nonleaf(idx, "形式", depths=(3, 4), under="架构")
+    purpose_pool = _leaves(idx, "作用") + _leaves(idx, "感受") + _leaves(idx, "意图")
+    shi_batch = rng.sample(shi, min(opts.batch_n, len(shi)))
+    xing_batch = rng.sample(xing, min(opts.batch_n, len(xing)))
+    purpose_batch = rng.sample(purpose_pool, min(opts.batch_n, len(purpose_pool)))
+    f6_zy = _nonleaf_d4("作用", 10)
+    f6_stage_act = [(s, a) for s in OLD_STAGES for a in ACTIONS] + [("", "")]
+
+    if not shi_batch or not xing_batch or not purpose_batch or not f6_zy:
+        raise RuntimeError("scope tree does not contain enough creation query axes")
+
+    master: list[dict[str, str]] = []
+    for i in range(opts.per):
+        stage, action = f6_stage_act[i % len(f6_stage_act)]
+        master.append(
+            {
+                "实质": shi_batch[i % len(shi_batch)],
+                "形式": xing_batch[i % len(xing_batch)],
+                "目的": purpose_batch[i % len(purpose_batch)],
+                "模态": MODALITY[i % len(MODALITY)],
+                "业务阶段": INTENT[i % len(INTENT)],
+                "知识类型": KTYPE[(i // 3) % len(KTYPE)],
+                "阶段": stage or "/",
+                "动作": action or "/",
+                "_st": stage,
+                "_ac": action,
+                "作用": f6_zy[i % len(f6_zy)],
+            }
+        )
+
+    def project(i: int, keys: list[str]) -> dict[str, Any]:
+        row = master[i]
+        return {"parts": {key: row[key] for key in keys}}
+
+    def gen_old(
+        i: int,
+        *,
+        shi_axis: bool = False,
+        xing_axis: bool = False,
+        purpose_axis: bool = False,
+        effect_axis: bool = True,
+    ) -> dict[str, Any]:
+        row = master[i]
+        segment = (row["_st"] + row["_ac"]) if row["_ac"] else ""
+        head = (
+            ([row["实质"]] if shi_axis else [])
+            + ([row["形式"]] if xing_axis else [])
+            + ([row["目的"]] if purpose_axis else [])
+        )
+        tail = ([row["作用"]] if effect_axis else []) + [row["知识类型"]]
+        query = " ".join(head + ([segment] if segment else []) + tail)
+        parts: dict[str, str] = {}
+        if shi_axis:
+            parts["实质"] = row["实质"]
+        if xing_axis:
+            parts["形式"] = row["形式"]
+        if purpose_axis:
+            parts["目的"] = row["目的"]
+        parts["阶段"], parts["动作"] = row["阶段"], row["动作"]
+        if effect_axis:
+            parts["作用"] = row["作用"]
+        parts["知识类型"] = row["知识类型"]
+        return {"parts": parts, "query": query}
+
+    families = [
+        {"key": "f1", "axes": ["实质", "模态", "业务阶段", "知识类型"], "gen": lambda i: project(i, ["实质", "模态", "业务阶段", "知识类型"])},
+        {"key": "f2", "axes": ["形式", "模态", "业务阶段", "知识类型"], "gen": lambda i: project(i, ["形式", "模态", "业务阶段", "知识类型"])},
+        {"key": "f4", "axes": ["作用/感受/意图", "模态", "业务阶段", "知识类型"], "gen": lambda i: project(i, ["目的", "模态", "业务阶段", "知识类型"])},
+        {"key": "f3", "axes": ["实质", "形式", "模态", "业务阶段", "知识类型"], "gen": lambda i: project(i, ["实质", "形式", "模态", "业务阶段", "知识类型"])},
+        {"key": "f5", "axes": ["模态", "业务阶段", "知识类型"], "gen": lambda i: project(i, ["模态", "业务阶段", "知识类型"])},
+        {"key": "a_shi", "axes": ["实质", "阶段", "动作", "作用", "知识类型"], "gen": lambda i: gen_old(i, shi_axis=True)},
+        {"key": "a_xing", "axes": ["形式", "阶段", "动作", "作用", "知识类型"], "gen": lambda i: gen_old(i, xing_axis=True)},
+        {"key": "a_both", "axes": ["实质", "形式", "阶段", "动作", "作用", "知识类型"], "gen": lambda i: gen_old(i, shi_axis=True, xing_axis=True)},
+        {"key": "a_purpose", "axes": ["作用/感受/意图", "阶段", "动作", "作用", "知识类型"], "gen": lambda i: gen_old(i, purpose_axis=True)},
+        {"key": "a_tail", "axes": ["阶段", "动作", "作用", "知识类型"], "gen": lambda i: gen_old(i)},
+        {"key": "b_shi", "axes": ["实质", "阶段", "动作", "知识类型"], "gen": lambda i: gen_old(i, shi_axis=True, effect_axis=False)},
+        {"key": "b_xing", "axes": ["形式", "阶段", "动作", "知识类型"], "gen": lambda i: gen_old(i, xing_axis=True, effect_axis=False)},
+        {"key": "b_both", "axes": ["实质", "形式", "阶段", "动作", "知识类型"], "gen": lambda i: gen_old(i, shi_axis=True, xing_axis=True, effect_axis=False)},
+        {"key": "b_purpose", "axes": ["作用/感受/意图", "阶段", "动作", "知识类型"], "gen": lambda i: gen_old(i, purpose_axis=True, effect_axis=False)},
+        {"key": "b_tail", "axes": ["阶段", "动作", "知识类型"], "gen": lambda i: gen_old(i, effect_axis=False)},
+    ]
+    order = ["实质", "形式", "目的", "模态", "业务阶段", "知识类型"]
+
+    out: dict[str, Any] = {
+        "axis_values": {
+            "实质": shi,
+            "形式": xing,
+            "目的池": purpose_pool,
+            "业务阶段": INTENT,
+            "模态": MODALITY,
+            "知识类型": KTYPE,
+            "阶段": OLD_STAGES,
+            "动作": ACTIONS,
+            "作用": f6_zy,
+        },
+        "metadata": {
+            "seed": opts.seed,
+            "per": opts.per,
+            "batch_n": opts.batch_n,
+            "dry": opts.dry,
+            "query_filter_prompt_version": prompt_version(),
+        },
+        "families": [],
+    }
+    for family in families:
+        name = " × ".join(family["axes"])
+        seen: set[str] = set()
+        items: list[dict[str, Any]] = []
+        for i in range(opts.per):
+            generated = family["gen"](i)
+            parts = generated["parts"]
+            query = generated.get("query") or " ".join(parts[k] for k in order if k in parts)
+            if query in seen:
+                continue
+            seen.add(query)
+            items.append({"query": query, "parts": parts})
+        verdicts = (
+            [
+                {"keep": True, "valid": None, "relevant": True, "reason": "(dry:未过筛)"}
+                for _ in items
+            ]
+            if opts.dry
+            else filter_queries([it["query"] for it in items], settings)
+        )
+        for item, verdict in zip(items, verdicts):
+            item.update(verdict)
+        out["families"].append(
+            {"key": family["key"], "name": name, "axes": family["axes"], "items": items}
+        )
+    return out
+
+
+def persist_query_batch(
+    repo: AcquisitionRepository,
+    generated: dict[str, Any],
+    *,
+    name: str,
+    source_type: str = "generated",
+    generation_method: str = "creation_demo_v1",
+    target_platforms: list[str] | None = None,
+) -> tuple[Any, int]:
+    """Persist generated families into formal query batch/query rows."""
+    batch = repo.create_query_batch(
+        name=name,
+        source_type=source_type,
+        generation_method=generation_method,
+        target_platforms=target_platforms or ["xiaohongshu", "weixin", "douyin"],
+        status="ready",
+        metadata=generated.get("metadata") or {},
+    )
+    count = 0
+    sort_order = 0
+    for family in generated.get("families") or []:
+        for item in family.get("items") or []:
+            sort_order += 1
+            repo.add_query(
+                batch_id=batch.id,
+                query_text=item["query"],
+                axes=item.get("parts") or {},
+                keep=bool(item.get("keep", True)),
+                filter_reason=item.get("reason") or "",
+                status="ready",
+                sort_order=sort_order,
+                metadata={
+                    "family_key": family.get("key"),
+                    "family_name": family.get("name"),
+                    "family_axes": family.get("axes") or [],
+                    "valid": item.get("valid"),
+                    "relevant": item.get("relevant"),
+                    "query_filter_prompt_version": (generated.get("metadata") or {}).get(
+                        "query_filter_prompt_version"
+                    ),
+                },
+            )
+            count += 1
+    return batch, count

acquisition/queries/filter.py

@@ -0,0 +1,137 @@
+"""Formal query filter used by generated query batches."""
+from __future__ import annotations
+
+import hashlib
+import json
+import os
+from pathlib import Path
+
+import httpx
+
+from core.config import Settings, load_env_file
+
+PROMPT = Path(__file__).resolve().parent.parent / "query_filter.txt"
+VALID_MIN = 6
+
+
+def prompt_version() -> str:
+    h = hashlib.sha256()
+    h.update(PROMPT.read_bytes())
+    return h.hexdigest()[:16]
+
+
+def filter_queries(
+    queries: list[str],
+    settings: Settings,
+    *,
+    batch: int = 40,
+) -> list[dict]:
+    """Batch-filter query strings and keep output aligned with input order."""
+    out: list[dict] = []
+    for i in range(0, len(queries), batch):
+        out.extend(_filter_batch(queries[i:i + batch], settings))
+    return out
+
+
+def _filter_batch(queries: list[str], settings: Settings) -> list[dict]:
+    if not queries:
+        return []
+    user = json.dumps(
+        [{"idx": i, "query": q} for i, q in enumerate(queries)],
+        ensure_ascii=False,
+    )
+    messages = [
+        {"role": "system", "content": PROMPT.read_text("utf-8")},
+        {"role": "user", "content": user},
+    ]
+    try:
+        txt = _chat_content(settings, messages)
+        data = json.loads(txt)
+        arr = data if isinstance(data, list) else next(
+            (v for v in data.values() if isinstance(v, list)),
+            [],
+        )
+        by = {d.get("idx"): d for d in arr if isinstance(d, dict)}
+        out = []
+        for i in range(len(queries)):
+            d = by.get(i, {})
+            try:
+                valid = int(d.get("valid", 10))
+            except (TypeError, ValueError):
+                valid = 10
+            relevant = bool(d.get("relevant", True))
+            out.append(
+                {
+                    "keep": valid >= VALID_MIN and relevant,
+                    "valid": valid,
+                    "relevant": relevant,
+                    "reason": str(d.get("reason", ""))[:50],
+                }
+            )
+        return out
+    except Exception as exc:
+        reason = f"筛选失败:{str(exc)[:30]}"
+        return [
+            {"keep": True, "valid": 10, "relevant": True, "reason": reason}
+            for _ in queries
+        ]
+
+
+def _chat_content(settings: Settings, messages: list[dict]) -> str:
+    body = {
+        "model": settings.llm_model,
+        "messages": messages,
+        "response_format": {"type": "json_object"},
+    }
+    env = load_env_file(os.getenv("CK_ENV_FILE", ".env"))
+    prefer_ark = os.getenv("QUERY_FILTER_PROVIDER") == "ark" or bool(
+        os.getenv("ARK_CHAT_MODEL")
+    )
+    openrouter_exc: Exception | None = None
+    if settings.openrouter_api_key and not prefer_ark:
+        try:
+            resp = httpx.post(
+                settings.openrouter_base_url.rstrip("/") + "/chat/completions",
+                headers={
+                    "Authorization": f"Bearer {settings.openrouter_api_key}",
+                    "Content-Type": "application/json",
+                },
+                json=body,
+                timeout=120,
+            )
+            resp.raise_for_status()
+            return resp.json()["choices"][0]["message"]["content"]
+        except Exception as exc:
+            openrouter_exc = exc
+
+    ark_key = os.getenv("ARK_API_KEY") or env.get("ARK_API_KEY")
+    if ark_key:
+        ark_model = (
+            os.getenv("ARK_CHAT_MODEL")
+            or env.get("ARK_CHAT_MODEL")
+            or "doubao-seed-1-6-flash-250615"
+        )
+        ark_url = (
+            os.getenv("ARK_CHAT_URL")
+            or env.get("ARK_CHAT_URL")
+            or "https://ark.cn-beijing.volces.com/api/v3/chat/completions"
+        )
+        resp = httpx.post(
+            ark_url,
+            headers={
+                "Authorization": f"Bearer {ark_key}",
+                "Content-Type": "application/json",
+            },
+            json={
+                "model": ark_model,
+                "messages": messages,
+                "response_format": {"type": "json_object"},
+            },
+            timeout=120,
+        )
+        resp.raise_for_status()
+        return resp.json()["choices"][0]["message"]["content"]
+
+    if openrouter_exc:
+        raise openrouter_exc
+    raise RuntimeError("missing OpenRouter/Ark chat credentials")

acquisition/query_filter.py

@@ -1,95 +1,6 @@
-"""创作 query 筛选器：调用 acquisition/query_filter.txt，批量判一组 query。
-
-模型对每条 query 打两项：语义合法性 valid(0-10) + 创作相关性 relevant(bool)。
-最终保留 keep = valid ≥ VALID_MIN 且 relevant —— 阈值放代码里(好调，不改提示词)。
-  · valid：词组合本身说不说得通(机械正交易产生的废话短语在此淘汰)。
-  · relevant：搜回来是否和创作知识相关(三把尺=设计决策vs工艺/可迁移/业务阶段；再排 制作/题材本身/应试政企/作品)。
-返回与输入等长的 [{keep, valid, relevant, reason}]。供 build_creation_demo、filter_multiaxis 等共用。
-"""
+"""Legacy import shim for the formal query filter."""
 from __future__ import annotations
 
-import json
-import os
-from pathlib import Path
-
-import httpx
-
-from core.config import Settings, load_env_file
-
-PROMPT = Path(__file__).resolve().parent / "query_filter.txt"
-VALID_MIN = 6   # 语义合法性阈值：valid ≥ 此值才算说得通；调这里即可松紧
-
-
-def filter_queries(queries: list[str], settings: Settings, *, batch: int = 40) -> list[dict]:
-    """批量过滤（分批调用，避免一次喂太多）。返回 [{keep, reason}]，与 queries 对齐。"""
-    out: list[dict] = []
-    for i in range(0, len(queries), batch):
-        out.extend(_filter_batch(queries[i:i + batch], settings))
-    return out
-
-
-def _filter_batch(queries: list[str], settings: Settings) -> list[dict]:
-    if not queries:
-        return []
-    user = json.dumps([{"idx": i, "query": q} for i, q in enumerate(queries)], ensure_ascii=False)
-    messages = [
-        {"role": "system", "content": PROMPT.read_text("utf-8")},
-        {"role": "user", "content": user},
-    ]
-    try:
-        txt = _chat_content(settings, messages)
-        # query_filter 要求输出数组；有的模型会包一层 {"result":[...]}，都兜住
-        data = json.loads(txt)
-        arr = data if isinstance(data, list) else next((v for v in data.values() if isinstance(v, list)), [])
-        by = {d.get("idx"): d for d in arr if isinstance(d, dict)}
-        out = []
-        for i in range(len(queries)):
-            d = by.get(i, {})
-            try:
-                valid = int(d.get("valid", 10))
-            except (TypeError, ValueError):
-                valid = 10
-            relevant = bool(d.get("relevant", True))
-            out.append({"keep": valid >= VALID_MIN and relevant,   # 阈值在代码里判
-                        "valid": valid, "relevant": relevant,
-                        "reason": str(d.get("reason", ""))[:50]})
-        return out
-    except Exception as exc:
-        return [{"keep": True, "valid": 10, "relevant": True, "reason": f"筛选失败:{str(exc)[:30]}"} for _ in queries]
-
-
-def _chat_content(settings: Settings, messages: list[dict]) -> str:
-    """Return chat content. Prefer OpenRouter; fall back to Ark when OpenRouter is region-blocked locally."""
-    body = {"model": settings.llm_model, "messages": messages, "response_format": {"type": "json_object"}}
-    env = load_env_file(os.getenv("CK_ENV_FILE", ".env"))
-    prefer_ark = os.getenv("QUERY_FILTER_PROVIDER") == "ark" or bool(os.getenv("ARK_CHAT_MODEL"))
-    openrouter_exc: Exception | None = None
-    if settings.openrouter_api_key and not prefer_ark:
-        try:
-            resp = httpx.post(
-                settings.openrouter_base_url.rstrip("/") + "/chat/completions",
-                headers={"Authorization": f"Bearer {settings.openrouter_api_key}", "Content-Type": "application/json"},
-                json=body,
-                timeout=120,
-            )
-            resp.raise_for_status()
-            return resp.json()["choices"][0]["message"]["content"]
-        except Exception as exc:
-            openrouter_exc = exc
-
-    ark_key = os.getenv("ARK_API_KEY") or env.get("ARK_API_KEY")
-    if ark_key:
-        ark_model = os.getenv("ARK_CHAT_MODEL") or env.get("ARK_CHAT_MODEL") or "doubao-seed-1-6-flash-250615"
-        ark_url = os.getenv("ARK_CHAT_URL") or env.get("ARK_CHAT_URL") or "https://ark.cn-beijing.volces.com/api/v3/chat/completions"
-        resp = httpx.post(
-            ark_url,
-            headers={"Authorization": f"Bearer {ark_key}", "Content-Type": "application/json"},
-            json={"model": ark_model, "messages": messages, "response_format": {"type": "json_object"}},
-            timeout=120,
-        )
-        resp.raise_for_status()
-        return resp.json()["choices"][0]["message"]["content"]
+from acquisition.queries.filter import VALID_MIN, filter_queries, prompt_version
 
-    if openrouter_exc:
-        raise openrouter_exc
-    raise RuntimeError("missing OpenRouter/Ark chat credentials")
+__all__ = ["VALID_MIN", "filter_queries", "prompt_version"]

acquisition/runner.py

@@ -0,0 +1,300 @@
+"""Formal acquisition runner backed by the repository boundary."""
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any, Callable
+from uuid import UUID
+
+from acquisition.classification.coarse import ClassificationResult, coarse_classify_item
+from acquisition.crawler import RateLimiter
+from acquisition.domain import AcquisitionJob, Query
+from acquisition.media.service import StabilizedMedia, stabilize_media_urls
+from acquisition.platforms import PlatformAdapter, get_platform_adapter
+from acquisition.repositories.base import AcquisitionRepository
+from core.config import Settings
+
+DEFAULT_PLATFORMS = ("xiaohongshu", "weixin", "douyin")
+
+AdapterFactory = Callable[[str], PlatformAdapter]
+Classifier = Callable[..., ClassificationResult]
+MediaStabilizer = Callable[..., list[StabilizedMedia]]
+RateLimiterFactory = Callable[[str], Any]
+
+
+@dataclass(frozen=True)
+class RunBatchResult:
+    run_id: UUID
+    total_jobs: int
+    done: int
+    partial: int
+    failed: int
+    skipped: int
+
+
+class PlatformRateLimiter:
+    """Share one rate bucket per platform across search and detail calls."""
+
+    def __init__(self, platform: str, delegate: RateLimiter | None = None) -> None:
+        self.platform = platform
+        self.delegate = delegate or RateLimiter(
+            min_interval_seconds=10.0,
+            max_interval_seconds=12.0,
+        )
+
+    def wait(self, bucket: str) -> None:
+        self.delegate.wait(self.platform)
+
+
+def _query_id(query: Query) -> UUID:
+    if query.id is None:
+        raise RuntimeError("formal acquisition queries must have id before running")
+    return query.id
+
+
+def _job_id(job: AcquisitionJob) -> UUID:
+    if job.id is None:
+        raise RuntimeError("formal acquisition jobs must have id before running")
+    return job.id
+
+
+def _best_video_url(media: list[StabilizedMedia]) -> str:
+    for row in media:
+        if row.media_type == "video" and row.cdn_url:
+            return row.cdn_url
+    return ""
+
+
+def _image_urls(media: list[StabilizedMedia]) -> list[str]:
+    return [row.cdn_url for row in media if row.media_type == "image" and row.cdn_url]
+
+
+def _source_payload(candidate: Any, detail: Any) -> dict[str, Any]:
+    return {
+        "candidate": candidate.model_dump() if hasattr(candidate, "model_dump") else {},
+        "detail": detail.raw if isinstance(getattr(detail, "raw", None), dict) else {},
+    }
+
+
+def _record_candidate(
+    repo: AcquisitionRepository,
+    *,
+    job: AcquisitionJob,
+    query: Query,
+    platform: str,
+    candidate: Any,
+    detail: Any,
+    settings: Settings,
+    media_stabilizer: MediaStabilizer,
+    classifier: Classifier,
+    classify: bool,
+) -> bool:
+    media_rows = media_stabilizer(
+        image_urls=detail.image_urls,
+        video_urls=detail.video_urls,
+        settings=settings,
+    )
+    item = repo.upsert_candidate_item(
+        platform=platform,
+        job_id=_job_id(job),
+        query_id=_query_id(query),
+        platform_item_id=detail.source_id or candidate.source_id or None,
+        canonical_url=detail.url or candidate.url or None,
+        content_type=detail.content_type or None,
+        title=detail.title or None,
+        author_name=detail.author or None,
+        raw_summary=(detail.body_text or "")[:1000],
+        status="candidate",
+        source_payload=_source_payload(candidate, detail),
+        metadata={"candidate_rank": candidate.rank},
+    )
+    if item.id is None:
+        raise RuntimeError("repository returned candidate without id")
+
+    for row in media_rows:
+        repo.add_media_asset(
+            item_id=item.id,
+            media_type=row.media_type,
+            source_url=row.source_url,
+            oss_url=row.cdn_url,
+            cdn_url=row.cdn_url,
+            position=row.position,
+            status=row.status,
+            source_payload={},
+            metadata={},
+        )
+
+    if classify:
+        result = classifier(
+            platform=platform,
+            title=detail.title,
+            body_text=detail.body_text,
+            image_urls=_image_urls(media_rows),
+            video_url=_best_video_url(media_rows),
+            settings=settings,
+        )
+        repo.add_item_classification(
+            item_id=item.id,
+            is_creation_knowledge=result.is_creation_knowledge,
+            label=result.label,
+            confidence=result.confidence,
+            reason=result.reason,
+            model_name=settings.video_model,
+            prompt_version=result.prompt_version,
+            result_payload=result.result_payload,
+            status=result.status,
+            error_message=result.error_message,
+        )
+    return True
+
+
+def run_batch(
+    repo: AcquisitionRepository,
+    *,
+    batch_id: UUID,
+    settings: Settings,
+    platforms: tuple[str, ...] | list[str] = DEFAULT_PLATFORMS,
+    search_limit: int = 10,
+    display_limit: int = 5,
+    classify: bool = True,
+    resume: bool = True,
+    skip_done: bool = True,
+    run_key: str | None = None,
+    adapter_factory: AdapterFactory = get_platform_adapter,
+    media_stabilizer: MediaStabilizer = stabilize_media_urls,
+    classifier: Classifier = coarse_classify_item,
+    rate_limiter_factory: RateLimiterFactory | None = None,
+) -> RunBatchResult:
+    """Run query x platform acquisition and write formal cloud-state rows."""
+    queries = repo.list_queries_for_batch(batch_id, keep=True)
+    run = repo.create_acquisition_run(
+        batch_id=batch_id,
+        run_key=run_key or f"acquisition:{batch_id}",
+        status="running",
+        metadata={
+            "platforms": list(platforms),
+            "search_limit": search_limit,
+            "display_limit": display_limit,
+            "classify": classify,
+            "resume": resume,
+        },
+    )
+    if run.id is None:
+        raise RuntimeError("repository returned acquisition run without id")
+
+    total_jobs = len(queries) * len(platforms)
+    done = partial = failed = skipped = 0
+    gates: dict[str, Any] = {}
+
+    for query in queries:
+        query_id = _query_id(query)
+        for platform in platforms:
+            job = repo.ensure_acquisition_job(
+                run_id=run.id,
+                query_id=query_id,
+                platform=platform,
+                search_limit=search_limit,
+                display_limit=display_limit,
+                status="pending",
+                metadata={"query_text": query.query_text},
+            )
+            if skip_done and job.status == "done":
+                skipped += 1
+                continue
+
+            attempts = job.attempt_count + 1
+            job = repo.update_acquisition_job(
+                _job_id(job),
+                status="running",
+                attempt_count=attempts,
+                error_message=None,
+            )
+            errors: list[str] = []
+            display_count = 0
+            searched_count = 0
+            try:
+                adapter = adapter_factory(platform)
+                gate = gates.get(platform)
+                if gate is None:
+                    gate = (
+                        rate_limiter_factory(platform)
+                        if rate_limiter_factory
+                        else PlatformRateLimiter(platform)
+                    )
+                    gates[platform] = gate
+                candidates = adapter.search(
+                    query.query_text,
+                    settings=settings,
+                    limit=search_limit,
+                    rate_limiter=gate,
+                )
+                searched_count = len(candidates)
+                for candidate in candidates:
+                    if display_count >= display_limit:
+                        break
+                    try:
+                        detail = adapter.fetch_detail(
+                            candidate,
+                            settings=settings,
+                            rate_limiter=gate,
+                        )
+                        if _record_candidate(
+                            repo,
+                            job=job,
+                            query=query,
+                            platform=platform,
+                            candidate=candidate,
+                            detail=detail,
+                            settings=settings,
+                            media_stabilizer=media_stabilizer,
+                            classifier=classifier,
+                            classify=classify,
+                        ):
+                            display_count += 1
+                    except Exception as exc:
+                        errors.append(str(exc)[:160])
+                        continue
+
+                status = "done" if display_count >= display_limit else (
+                    "partial" if display_count else "failed"
+                )
+                if status == "done":
+                    done += 1
+                elif status == "partial":
+                    partial += 1
+                else:
+                    failed += 1
+                repo.update_acquisition_job(
+                    _job_id(job),
+                    status=status,
+                    attempt_count=attempts,
+                    error_message=None if status != "failed" else "; ".join(errors[-3:]),
+                    metadata={
+                        "query_text": query.query_text,
+                        "searched_count": searched_count,
+                        "display_count": display_count,
+                        "errors": errors[-3:],
+                    },
+                )
+            except Exception as exc:
+                failed += 1
+                repo.update_acquisition_job(
+                    _job_id(job),
+                    status="failed",
+                    attempt_count=attempts,
+                    error_message=str(exc)[:300],
+                    metadata={
+                        "query_text": query.query_text,
+                        "searched_count": searched_count,
+                        "display_count": display_count,
+                        "errors": errors[-3:],
+                    },
+                )
+
+    return RunBatchResult(
+        run_id=run.id,
+        total_jobs=total_jobs,
+        done=done,
+        partial=partial,
+        failed=failed,
+        skipped=skipped,
+    )

scripts/build_creation_demo.py

@@ -1,205 +1,95 @@
-"""创作知识 query 正交 demo：多种正交组合 → LLM 评 valid/relevant(query_filter.txt) → 存 JSON。
+#!/usr/bin/env python3
+"""Build formal creation-query batches.
 
-不真实搜，只产 query 供前端看。轴严格取分类树的"创作支"：
-  实质 = 实质树·理念支(排除表象)   形式 = 形式树·架构支(排除呈现)
-  目的池 = 作用树 + 感受树 + 意图树 全部合一(随机取)
-  业务阶段 = 灵感/选题/脚本(只这三个裸阶段，不展开)——脊柱，每族必带
-  模态 = 视频/图文   知识类型 = 怎么做/有哪些/为什么
-新尾缀：模态 × 业务阶段 × 知识类型；老尾缀对照：阶段 × 动作 × [作用] × 知识类型。
-实质 / 形式 / 目的等轴从同一张 MASTER 主表投影，方便控制变量横向对比。
-每条原串过 query_filter.txt(valid + relevant)，存 keep+valid+reason 供前端展示。
-用法：PYTHONPATH=. CK_ENV_FILE=.env python scripts/build_creation_demo.py
+Default mode only prints a summary. Use --export-json for a local review file
+or --persist to write query_batches/queries into the formal PostgreSQL store.
 """
 from __future__ import annotations
 
+import argparse
 import json
-import random
-import sys
 from pathlib import Path
 
-from acquisition.query import ACTIONS, _nonleaf_d4   # 老正交动作轴 + 4级非叶子取作用节点
-from acquisition.query import STAGES as OLD_STAGES   # 老的裸阶段 灵感/选题/脚本
-from acquisition.query_filter import filter_queries   # 共享筛选器(query_filter.txt)
-from core.config import Settings
-
-ROOT = Path(__file__).resolve().parent.parent
-TREES = ROOT / "scope_trees" / "trees_index.json"
-OUT = ROOT / "data" / "queries" / "creation_demo.json"
-PER = 30
-BATCH_N = 30        # 全 demo 统一抽这么多个「实质 / 形式」，各族共用同一批；PER=BATCH_N 保证整批都用上、左右一一对应
-KTYPE = ["怎么做", "有哪些", "为什么"]
-MODALITY = ["视频", "图文"]   # 被创作内容的形态(与教学帖本身格式无关),正交进所有家族
-# 业务阶段(脊柱)：只留 3 个裸阶段，不再展开成 query构造.md 的衍生词(找素材/开头/钩子/标题…全去掉)
-INTENT = ["灵感", "选题", "脚本"]
-
-
-def _segs(p):
-    return [x for x in (p or "").split("/") if x]
-
-
-def _leaves(idx, source_type, under=None):
-    """某树某支下的叶子节点名(没有更深子节点的=元素层)。under 限定分支。"""
-    paths = [(_segs(n["path"]), n.get("name")) for n in idx if n.get("source_type") == source_type]
-    if under:
-        paths = [(s, nm) for s, nm in paths if under in s]
-    allp = {"/".join(s) for s, _ in paths}
-    out, seen = [], set()
-    for s, nm in paths:
-        if len(s) < 2:
-            continue
-        full = "/".join(s)
-        is_leaf = not any(o != full and o.startswith(full + "/") for o in allp)
-        name = nm or s[-1]
-        if is_leaf and name and name not in seen:
-            seen.add(name)
-            out.append(name)
-    return out
-
-
-def _nonleaf(idx, source_type, depths=(3, 4), under=None):
-    """某树某支下、指定层级的【非叶子"类目"节点】(底下还有元素,不取元素本身)。
-    对齐制作侧取法:实质/形式 取 depth 3-4 的类目层,而非最深的元素层。"""
-    paths = [(_segs(n["path"]), n.get("name")) for n in idx if n.get("source_type") == source_type]
-    if under:
-        paths = [(s, nm) for s, nm in paths if under in s]
-    allp = {"/".join(s) for s, _ in paths}
-    out, seen = [], set()
-    for s, nm in paths:
-        if len(s) not in depths:
-            continue
-        full = "/".join(s)
-        is_nonleaf = any(o != full and o.startswith(full + "/") for o in allp)
-        name = nm or (s[-1] if s else "")
-        if is_nonleaf and name and name not in seen:
-            seen.add(name)
-            out.append(name)
-    return out
-
-
-def main():
-    settings = Settings.from_env()
-    rng = random.Random(7)
-    DRY = "--dry" in sys.argv          # 干跑：跳过 LLM 筛选(全 keep)，只验证生成结构/控制变量
-    idx = json.loads(TREES.read_text("utf-8"))
-    SHI = _nonleaf(idx, "实质", depths=(3, 4), under="理念")   # 类目层,非元素
-    XING = _nonleaf(idx, "形式", depths=(3, 4), under="架构")  # 类目层,非元素
-    POOL = _leaves(idx, "作用") + _leaves(idx, "感受") + _leaves(idx, "意图")
-    print(f"实质 {len(SHI)} / 形式 {len(XING)} / 目的池 {len(POOL)} / 业务阶段 {len(INTENT)}")
-
-    # 统一批：30 实质 / 30 形式 / 30 目的(作用感受意图)，全 demo 共用
-    SHI_BATCH = rng.sample(SHI, min(BATCH_N, len(SHI)))
-    XING_BATCH = rng.sample(XING, min(BATCH_N, len(XING)))
-    POOL_BATCH = rng.sample(POOL, min(BATCH_N, len(POOL)))
-    F6_ZY = _nonleaf_d4("作用", 10)                                       # 老正交的"作用"(4级非叶子)
-    F6_STAGE_ACT = [(s, a) for s in OLD_STAGES for a in ACTIONS] + [("", "")]   # 阶段×动作 + 无动作变体
-    print(f"统一批: 实质×{len(SHI_BATCH)} 形式×{len(XING_BATCH)} 目的×{len(POOL_BATCH)}")
-
-    # 主表：第 i 行把【所有轴】取值一次定死；各组合方式只是从这行挑自己用到的轴(投影)，严格控制变量
-    # ——同一个实质(在第 i 行)无论出现在哪种组合里，配的模态/业务阶段/知识类型都相同。
-    MASTER = []
-    for i in range(PER):
-        st, ac = F6_STAGE_ACT[i % len(F6_STAGE_ACT)]
-        MASTER.append({
-            "实质": SHI_BATCH[i % len(SHI_BATCH)],
-            "形式": XING_BATCH[i % len(XING_BATCH)],
-            "目的": POOL_BATCH[i % len(POOL_BATCH)],
-            "模态": MODALITY[i % len(MODALITY)],
-            "业务阶段": INTENT[i % len(INTENT)],
-            "知识类型": KTYPE[(i // 3) % len(KTYPE)],     # 与业务阶段解耦，纯尾缀族也能多出几种
-            "阶段": st or "/", "动作": ac or "/", "_st": st, "_ac": ac,
-            "作用": F6_ZY[i % len(F6_ZY)],
-        })
-
-    def proj(i, keys):       # 新脊柱族：从主表第 i 行取这些 parts 键(query 由 order 拼)
-        row = MASTER[i]
-        return {"parts": {k: row[k] for k in keys}}
-
-    def gen_old(i, *, shi=False, xing=False, purpose=False, zy=True):    # 老正交族：[实质] [形式] [目的] (阶段+动作) [作用] 知识类型
-        r = MASTER[i]
-        seg = (r["_st"] + r["_ac"]) if r["_ac"] else ""               # 脚本撰写 / 空
-        head = ([r["实质"]] if shi else []) + ([r["形式"]] if xing else []) + ([r["目的"]] if purpose else [])
-        tail = ([r["作用"]] if zy else []) + [r["知识类型"]]
-        q = " ".join(head + ([seg] if seg else []) + tail)
-        parts = {}
-        if shi:
-            parts["实质"] = r["实质"]
-        if xing:
-            parts["形式"] = r["形式"]
-        if purpose:
-            parts["目的"] = r["目的"]
-        parts["阶段"], parts["动作"] = r["阶段"], r["动作"]
-        if zy:
-            parts["作用"] = r["作用"]
-        parts["知识类型"] = r["知识类型"]
-        return {"parts": parts, "query": q}
-
-    # 每种组合方式：生成器 + 用到的轴。name = axes 用「×」连接，直接反映正交结构。
-    # axes 顺序即前端列顺序；前 5 种使用新尾缀，后 6 种用于对照老尾缀。
-    families = [
-        {"key": "f1", "axes": ["实质", "模态", "业务阶段", "知识类型"],
-         "gen": lambda i: proj(i, ["实质", "模态", "业务阶段", "知识类型"])},
-        {"key": "f2", "axes": ["形式", "模态", "业务阶段", "知识类型"],
-         "gen": lambda i: proj(i, ["形式", "模态", "业务阶段", "知识类型"])},
-        {"key": "f4", "axes": ["作用/感受/意图", "模态", "业务阶段", "知识类型"],
-         "gen": lambda i: proj(i, ["目的", "模态", "业务阶段", "知识类型"])},
-        {"key": "f3", "axes": ["实质", "形式", "模态", "业务阶段", "知识类型"],
-         "gen": lambda i: proj(i, ["实质", "形式", "模态", "业务阶段", "知识类型"])},
-        {"key": "f5", "axes": ["模态", "业务阶段", "知识类型"],
-         "gen": lambda i: proj(i, ["模态", "业务阶段", "知识类型"])},
-        # 老正交·尾缀A：阶段 × 动作 × 作用 × 知识类型(五种正交)
-        {"key": "a_shi", "axes": ["实质", "阶段", "动作", "作用", "知识类型"],
-         "gen": lambda i: gen_old(i, shi=True, zy=True)},
-        {"key": "a_xing", "axes": ["形式", "阶段", "动作", "作用", "知识类型"],
-         "gen": lambda i: gen_old(i, xing=True, zy=True)},
-        {"key": "a_both", "axes": ["实质", "形式", "阶段", "动作", "作用", "知识类型"],
-         "gen": lambda i: gen_old(i, shi=True, xing=True, zy=True)},
-        {"key": "a_purpose", "axes": ["作用/感受/意图", "阶段", "动作", "作用", "知识类型"],
-         "gen": lambda i: gen_old(i, purpose=True, zy=True)},
-        {"key": "a_tail", "axes": ["阶段", "动作", "作用", "知识类型"],
-         "gen": lambda i: gen_old(i, zy=True)},
-        # 老正交·尾缀B：阶段 × 动作 × 知识类型(抽掉作用，五种正交)
-        {"key": "b_shi", "axes": ["实质", "阶段", "动作", "知识类型"],
-         "gen": lambda i: gen_old(i, shi=True, zy=False)},
-        {"key": "b_xing", "axes": ["形式", "阶段", "动作", "知识类型"],
-         "gen": lambda i: gen_old(i, xing=True, zy=False)},
-        {"key": "b_both", "axes": ["实质", "形式", "阶段", "动作", "知识类型"],
-         "gen": lambda i: gen_old(i, shi=True, xing=True, zy=False)},
-        {"key": "b_purpose", "axes": ["作用/感受/意图", "阶段", "动作", "知识类型"],
-         "gen": lambda i: gen_old(i, purpose=True, zy=False)},
-        {"key": "b_tail", "axes": ["阶段", "动作", "知识类型"],
-         "gen": lambda i: gen_old(i, zy=False)},
-    ]
-    # 各部件按固定顺序拼成原串（内容维度在前，模态贴题材后，业务阶段+知识类型收尾）
-    order = ["实质", "形式", "目的", "模态", "业务阶段", "知识类型"]
-
-    # 业务阶段只剩 灵感/选题/脚本 三个，扁平数组(前端按池子平铺，不再分组缩进)
-    out = {"axis_values": {"实质": SHI, "形式": XING, "目的池": POOL, "业务阶段": INTENT,
-                           "模态": MODALITY, "知识类型": KTYPE,
-                           "阶段": OLD_STAGES, "动作": ACTIONS, "作用": F6_ZY},   # 老尾缀对照轴
-           "families": []}
-    for fam in families:
-        name = " × ".join(fam["axes"])   # 家族名直接反映正交结构
-        seen, items = set(), []
-        for i in range(PER):                          # 单趟过主表、去重(纯尾缀族会自然少于 PER)
-            g = fam["gen"](i)
-            parts = g["parts"]
-            q = g.get("query") or " ".join(parts[k] for k in order if k in parts)   # f6/f7 自带 query 串
-            if q in seen:
-                continue
-            seen.add(q)
-            items.append({"query": q, "parts": parts})
-        verdicts = ([{"keep": True, "valid": None, "relevant": True, "reason": "(dry:未过筛)"} for _ in items]
-                    if DRY else filter_queries([it["query"] for it in items], settings))
-        for it, v in zip(items, verdicts):
-            it.update(v)
-        kept = sum(1 for it in items if it["keep"])
-        print(f"[{name}] 生成 {len(items)} 条, 筛后保留 {kept}")
-        out["families"].append({"key": fam["key"], "name": name, "axes": fam["axes"], "items": items})
-
-    OUT.parent.mkdir(parents=True, exist_ok=True)
-    OUT.write_text(json.dumps(out, ensure_ascii=False, indent=1), encoding="utf-8")
-    print(f"→ {OUT}")
+from acquisition.queries.builder import (
+    QueryBuildOptions,
+    TREES,
+    build_creation_query_batch,
+    persist_query_batch,
+)
+from acquisition.repositories.postgres import PostgresAcquisitionRepository
+from core.config import CreationDbConfig, Settings
+from core.db_session import transaction
+
+
+def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--env-file", default=".env")
+    parser.add_argument("--tree-path", type=Path)
+    parser.add_argument("--per", type=int, default=30)
+    parser.add_argument("--batch-n", type=int, default=30)
+    parser.add_argument("--seed", type=int, default=7)
+    parser.add_argument("--dry", action="store_true", help="Skip LLM query filtering")
+    parser.add_argument("--export-json", type=Path, help="Optional local review export")
+    parser.add_argument("--persist", action="store_true", help="Persist to formal PG")
+    parser.add_argument("--name", default="creation-demo")
+    return parser.parse_args(argv)
+
+
+def _summary(generated: dict) -> dict:
+    families = generated.get("families") or []
+    total = sum(len(f.get("items") or []) for f in families)
+    kept = sum(
+        1
+        for family in families
+        for item in family.get("items") or []
+        if item.get("keep", True)
+    )
+    return {
+        "family_count": len(families),
+        "query_count": total,
+        "kept_count": kept,
+        "metadata": generated.get("metadata") or {},
+    }
+
+
+def main(argv: list[str] | None = None) -> int:
+    args = parse_args(argv)
+    settings = Settings.from_env(args.env_file)
+    generated = build_creation_query_batch(
+        settings,
+        tree_path=args.tree_path or TREES,
+        options=QueryBuildOptions(
+            per=args.per,
+            batch_n=args.batch_n,
+            seed=args.seed,
+            dry=args.dry,
+        ),
+    )
+    summary = _summary(generated)
+
+    if args.export_json:
+        args.export_json.parent.mkdir(parents=True, exist_ok=True)
+        args.export_json.write_text(
+            json.dumps(generated, ensure_ascii=False, indent=1),
+            encoding="utf-8",
+        )
+        summary["export_json"] = str(args.export_json)
+
+    if args.persist:
+        db_config = CreationDbConfig.from_env(args.env_file)
+        with transaction(db_config) as conn:
+            repo = PostgresAcquisitionRepository(conn)
+            batch, count = persist_query_batch(
+                repo,
+                generated,
+                name=args.name,
+            )
+        summary["batch_id"] = str(batch.id)
+        summary["persisted_queries"] = count
+
+    print(json.dumps(summary, ensure_ascii=False, indent=2))
+    return 0
 
 
 if __name__ == "__main__":
-    main()
+    raise SystemExit(main())

scripts/run_acquisition.py

@@ -0,0 +1,58 @@
+#!/usr/bin/env python3
+"""Run formal acquisition against a cloud query batch."""
+from __future__ import annotations
+
+import argparse
+import json
+from uuid import UUID
+
+from acquisition.repositories.postgres import PostgresAcquisitionRepository
+from acquisition.runner import DEFAULT_PLATFORMS, run_batch
+from core.config import CreationDbConfig, Settings
+from core.db_session import transaction
+
+
+def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--batch-id", required=True, help="Formal query batch UUID")
+    parser.add_argument(
+        "--platform",
+        action="append",
+        choices=DEFAULT_PLATFORMS,
+        help="Platform to run. Repeat to run multiple platforms. Default: all.",
+    )
+    parser.add_argument("--search-limit", type=int, default=10)
+    parser.add_argument("--display-limit", type=int, default=5)
+    parser.add_argument("--no-classify", action="store_true")
+    parser.add_argument("--no-resume", action="store_true")
+    parser.add_argument("--no-skip-done", action="store_true")
+    parser.add_argument("--run-key")
+    parser.add_argument("--env-file", default=".env")
+    return parser.parse_args(argv)
+
+
+def main(argv: list[str] | None = None) -> int:
+    args = parse_args(argv)
+    settings = Settings.from_env(args.env_file)
+    db_config = CreationDbConfig.from_env(args.env_file)
+    platforms = tuple(args.platform or DEFAULT_PLATFORMS)
+    with transaction(db_config) as conn:
+        repo = PostgresAcquisitionRepository(conn)
+        result = run_batch(
+            repo,
+            batch_id=UUID(args.batch_id),
+            settings=settings,
+            platforms=platforms,
+            search_limit=args.search_limit,
+            display_limit=args.display_limit,
+            classify=not args.no_classify,
+            resume=not args.no_resume,
+            skip_done=not args.no_skip_done,
+            run_key=args.run_key,
+        )
+    print(json.dumps(result.__dict__, ensure_ascii=False, default=str, indent=2))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())