Create-knowledge-find-decode/acquisition/web_api.py

"""前端取数接口：从 app.db 分页/筛选返回 query 与真实搜索结果。

挂到现有 FastAPI（creation_knowledge/api.py）。前端从「读死 json」改成调这些接口：
  GET /api/runs                              有哪些批次
  GET /api/methods?table=queries|search      某表的打法清单（筛选下拉）
  GET /api/queries?method=&run_id=&page=&size=
  GET /api/search?platform=&method=&ok=&run_id=&page=&size=
每个请求开一个 sqlite 连接（轻、够用）；连接出错时接口照常返回空集，不 500。
"""
from __future__ import annotations

from pathlib import Path
from typing import Optional

from fastapi import APIRouter, Query

from acquisition import store
from acquisition.query import (
    ACTIONS, CARRIER_POS_GROUPED, DEMAND, KTYPE_SUFFIX, MODALITIES, STAGES,
)

router = APIRouter(prefix="/api")
_ROOT = Path(__file__).resolve().parent.parent

# 分类「创作知识 / 非创作知识」用的真实判断提示词（图文 / 视频），前端弹窗展示。读 is_empty 为判据。
_JUDGE_PROMPTS = [("图文（小红书 / 微信公众号）", "extract.txt"), ("视频（抖音）", "extract_video.txt")]

# 各方法里【人工定义】的轴及其全部取值（实质/形式/作用来自分类树，不在此）。单一来源 = query.py。
MANUAL_AXES = {
    "实质 × 创作阶段 × 需求": {"创作阶段": STAGES, "需求点": DEMAND},
    "形式 + 载体位置": {"载体位置": CARRIER_POS_GROUPED},
    "多轴正交组合": {"阶段": STAGES, "动作": ACTIONS, "知识类型": KTYPE_SUFFIX, "模态": MODALITIES},
}


@router.get("/judge-prompts")
def judge_prompts() -> dict:
    """分类创作知识用的真实提示词（图文 extract.txt / 视频 extract_video.txt），原样返回供前端展示。"""
    out = []
    for label, fn in _JUDGE_PROMPTS:
        try:
            text = (_ROOT / "prompts" / fn).read_text("utf-8")
        except Exception as exc:
            text = f"(读取失败: {exc})"
        out.append({"name": label, "file": fn, "text": text})
    return {"prompts": out}


@router.get("/runs")
def runs() -> dict:
    conn = store.connect()
    try:
        return {"runs": store.list_runs(conn)}
    finally:
        conn.close()


@router.get("/manual-axes")
def manual_axes() -> dict:
    """各方法人工定义轴的全部取值（前端「查看人工定义的轴」弹窗用）。"""
    return {"axes": MANUAL_AXES}


@router.get("/methods")
def methods(table: str = Query("queries")) -> dict:
    conn = store.connect()
    try:
        tbl = "search_results" if table == "search" else "queries"
        return {"methods": store.list_methods(conn, tbl)}
    finally:
        conn.close()


@router.get("/queries")
def queries(method: Optional[str] = None, run_id: Optional[str] = None,
            page: int = 1, size: int = 30) -> dict:
    conn = store.connect()
    try:
        return store.list_queries(conn, method=method, run_id=run_id, page=page, size=size)
    finally:
        conn.close()


@router.get("/search")
def search(platform: Optional[str] = None, method: Optional[str] = None,
           ok: Optional[bool] = None, run_id: Optional[str] = None,
           query: Optional[str] = None, page: int = 1, size: int = 30) -> dict:
    conn = store.connect()
    try:
        return store.list_search(conn, platform=platform, method=method, ok=ok,
                                 run_id=run_id, query=query, page=page, size=size)
    finally:
        conn.close()


@router.get("/search/summary")
def search_summary() -> dict:
    """{query文本: {total, ok}}——列表页据此给每条 query 显示「搜索结果」按钮与计数。"""
    conn = store.connect()
    try:
        return {"summary": store.search_summary(conn)}
    finally:
        conn.close()