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 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")

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


@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()