数据看板

app.py

@@ -1,3 +1,4 @@
+import io
 import secrets
 from pathlib import Path
 from urllib.parse import quote
@@ -15,6 +16,15 @@ from auth_service import (
 )
 from config import Config
 from database import init_database
+from ocr_dashboard_service import (
+    get_result_matches as get_ocr_result_matches,
+    get_summary as get_ocr_dashboard_summary,
+    list_batches as list_ocr_batches,
+    list_collections as list_ocr_collections,
+    list_evidences as list_ocr_evidences,
+    export_evidences as export_ocr_evidences,
+    list_results as list_ocr_results,
+)
 from acceptance_service import (
     create_acceptance_task,
     get_acceptance_resources,
@@ -298,6 +308,40 @@ def create_app() -> Flask:
     def resume_project_acceptance_task(task_id: str):
         return jsonify({"data": resume_acceptance_task(task_id), "request_id": g.request_id})
 
+    @app.route("/api/v1/data-dashboard/ocr/batches", methods=["GET"])
+    def query_ocr_dashboard_batches():
+        return jsonify({"data": list_ocr_batches(), "request_id": g.request_id})
+
+    @app.route("/api/v1/data-dashboard/ocr/batches/<batch_key>/collections", methods=["GET"])
+    def query_ocr_dashboard_collections(batch_key: str):
+        return jsonify({"data": list_ocr_collections(batch_key), "request_id": g.request_id})
+
+    @app.route("/api/v1/data-dashboard/ocr/summary", methods=["GET"])
+    def query_ocr_dashboard_summary():
+        return jsonify({"data": get_ocr_dashboard_summary(
+            request.args.get("batch_key", "").strip(),
+            request.args.get("collection_key", "").strip(),
+        ), "request_id": g.request_id})
+
+    @app.route("/api/v1/data-dashboard/ocr/results", methods=["GET"])
+    def query_ocr_dashboard_results():
+        return jsonify({"data": list_ocr_results(request.args.to_dict()), "request_id": g.request_id})
+
+    @app.route("/api/v1/data-dashboard/ocr/results/<int:result_id>/matches", methods=["GET"])
+    def query_ocr_dashboard_result_matches(result_id: int):
+        return jsonify({"data": get_ocr_result_matches(result_id), "request_id": g.request_id})
+
+    @app.route("/api/v1/data-dashboard/ocr/evidences", methods=["GET"])
+    def query_ocr_dashboard_evidences():
+        return jsonify({"data": list_ocr_evidences(request.args.to_dict()), "request_id": g.request_id})
+
+    @app.route("/api/v1/data-dashboard/ocr/evidences/export", methods=["GET"])
+    def export_ocr_dashboard_evidences():
+        payload, filename, count = export_ocr_evidences(request.args.to_dict())
+        response = send_file(payload, mimetype="application/zip", as_attachment=True, download_name=filename)
+        response.headers["X-Evidence-Count"] = str(count)
+        return response
+
     @app.route("/api/v1/tasks", methods=["GET"])
     def query_research_tasks():
         result = list_tasks(

auth_service.py

@@ -161,6 +161,57 @@ def get_user_by_session_token(token: str | None) -> dict[str, Any] | None:
         }
 
 
+def create_agent_access_token(session_token: str | None) -> tuple[str, datetime] | None:
+    """为 Link AI 生成独立的短期 Bearer Token。"""
+    if not session_token:
+        return None
+    session_hash = _sha256(session_token)
+    current = now_iso()
+    token = secrets.token_urlsafe(36)
+    expires_at_dt = datetime.now().astimezone() + timedelta(days=Config.SESSION_EXPIRE_DAYS)
+    expires_at = expires_at_dt.isoformat(timespec="seconds")
+    with connection() as conn:
+        row = conn.execute(
+            """SELECT user_id FROM auth_sessions
+               WHERE session_hash=? AND revoked_at IS NULL AND expires_at>?""",
+            (session_hash, current),
+        ).fetchone()
+        if not row:
+            return None
+        conn.execute(
+            """INSERT INTO ai_access_tokens
+               (token_hash,user_id,expires_at,created_at,last_seen_at)
+               VALUES (?,?,?,?,?)""",
+            (_sha256(token), row["user_id"], expires_at, current, current),
+        )
+    return token, expires_at_dt
+
+
+def get_user_by_agent_token(token: str | None) -> dict[str, Any] | None:
+    if not token:
+        return None
+    current = now_iso()
+    with connection() as conn:
+        row = conn.execute(
+            """SELECT t.id AS token_id,u.* FROM ai_access_tokens t
+               JOIN auth_users u ON u.id=t.user_id
+               WHERE t.token_hash=? AND t.revoked_at IS NULL AND t.expires_at>?""",
+            (_sha256(token), current),
+        ).fetchone()
+        if not row:
+            return None
+        conn.execute("UPDATE ai_access_tokens SET last_seen_at=? WHERE id=?", (current, row["token_id"]))
+        return {
+            "id": str(row["ding_user_id"]),
+            "name": row["name"],
+            "account": row["mobile"] or row["email"] or row["ding_user_id"],
+            "avatar": row["avatar"],
+            "union_id": row["union_id"],
+            "open_id": row["open_id"],
+            "raw": json_loads(row["raw_json"]) if row["raw_json"] else {},
+        }
+
+
 def revoke_session(token: str | None) -> None:
     if not token:
         return

config.py

@@ -25,6 +25,11 @@ class Config:
     CRAWLER_POLL_INTERVAL_SECONDS = float(os.getenv("CRAWLER_POLL_INTERVAL_SECONDS", "5"))
     ACCEPTANCE_BROWSER_TIMEOUT_SECONDS = int(os.getenv("ACCEPTANCE_BROWSER_TIMEOUT_SECONDS", "60"))
     ACCEPTANCE_NOTE_WORKERS = int(os.getenv("ACCEPTANCE_NOTE_WORKERS", "4"))
+    SEARCH_QA_API_BASE_URL = os.getenv(
+        "SEARCH_QA_API_BASE_URL", "https://yunshouji.xiaotitv.com"
+    ).rstrip("/")
+    SEARCH_QA_API_KEY = os.getenv("SEARCH_QA_API_KEY", "")
+    SEARCH_QA_TIMEOUT_SECONDS = int(os.getenv("SEARCH_QA_TIMEOUT_SECONDS", "60"))
     CORS_ORIGIN = os.getenv("CORS_ORIGIN", "http://localhost:5173")
     FRONTEND_BASE_URL = os.getenv("FRONTEND_BASE_URL", CORS_ORIGIN).rstrip("/")
     SESSION_COOKIE_NAME = os.getenv("SESSION_COOKIE_NAME", "hc_session")

data_dashboard_service.py

@@ -0,0 +1,59 @@
+from typing import Any
+
+import requests
+
+from config import Config
+from task_service import TaskError
+
+
+def _headers() -> dict[str, str]:
+    if not Config.SEARCH_QA_API_KEY:
+        raise TaskError(
+            "SEARCH_QA_NOT_CONFIGURED",
+            "搜索问一问开放 API 尚未配置",
+            503,
+        )
+    return {"X-API-Key": Config.SEARCH_QA_API_KEY}
+
+
+def _request(method: str, path: str, **kwargs) -> requests.Response:
+    try:
+        session = requests.Session()
+        # 搜索问一问开放 API 必须直连；本机或服务环境的全局代理可能
+        # 指向不可用端口，且不应影响这一条受控的服务端对服务端链路。
+        session.trust_env = False
+        response = session.request(
+            method,
+            f"{Config.SEARCH_QA_API_BASE_URL}{path}",
+            headers={**_headers(), **kwargs.pop("headers", {})},
+            timeout=Config.SEARCH_QA_TIMEOUT_SECONDS,
+            **kwargs,
+        )
+    except requests.RequestException as exc:
+        raise TaskError("SEARCH_QA_UNAVAILABLE", "搜索问一问服务暂时不可用", 502) from exc
+    if response.status_code == 429:
+        raise TaskError("SEARCH_QA_RATE_LIMITED", "搜索问一问接口请求过于频繁，请稍后重试", 429)
+    if response.status_code >= 400:
+        message = "搜索问一问服务请求失败"
+        try:
+            payload = response.json()
+            message = str(payload.get("message") or payload.get("error") or message)
+        except ValueError:
+            pass
+        raise TaskError("SEARCH_QA_UPSTREAM_ERROR", message, 502)
+    return response
+
+
+def list_runs(task_id: str = "", xhs_account_id: str = "", limit: int = 20) -> dict[str, Any]:
+    limit = max(1, min(int(limit), 100))
+    params: dict[str, Any] = {"limit": limit}
+    if task_id:
+        params = {"task_id": task_id}
+    elif xhs_account_id:
+        params["xhs_account_id"] = xhs_account_id
+    payload = _request("GET", "/api/open/v1/search-qa/runs", params=params).json()
+    if not payload.get("ok"):
+        raise TaskError("SEARCH_QA_UPSTREAM_ERROR", "搜索问一问服务返回失败", 502)
+    runs = payload.get("runs") or ([payload["run"]] if payload.get("run") else [])
+    return {"items": runs, "total": len(runs), "query": params}
+

ocr_dashboard_service.py

@@ -0,0 +1,128 @@
+import io
+import re
+import zipfile
+from pathlib import Path
+from typing import Any
+from urllib.parse import urlparse
+
+import requests
+
+from config import Config
+from task_service import TaskError
+
+
+SESSION = requests.Session()
+SESSION.trust_env = False
+
+
+def _get(path: str, params: dict[str, Any] | None = None) -> Any:
+    try:
+        response = SESSION.get(
+            f"{Config.CRAWLER_BASE_URL}/api/v1{path}",
+            params={key: value for key, value in (params or {}).items() if value not in (None, "")},
+            timeout=Config.CRAWLER_TIMEOUT_SECONDS,
+        )
+        response.raise_for_status()
+        payload = response.json()
+    except (requests.RequestException, ValueError) as exc:
+        raise TaskError("OCR_DASHBOARD_UNAVAILABLE", "OCR 看板数据服务暂时不可用", 502) from exc
+    if not payload.get("success") or int(payload.get("code") or 0) != 200:
+        code = int(payload.get("code") or 500)
+        raise TaskError(
+            "OCR_DASHBOARD_UPSTREAM_ERROR",
+            str(payload.get("msg") or "OCR 看板数据查询失败"),
+            404 if code == 404 else 422 if code == 400 else 502,
+        )
+    return payload.get("data")
+
+
+def list_batches() -> list[dict[str, Any]]:
+    return _get("/xhs/ocr/batches") or []
+
+
+def list_collections(batch_key: str) -> list[dict[str, Any]]:
+    if not batch_key.strip():
+        raise TaskError("OCR_BATCH_REQUIRED", "batch_key 不能为空", 422)
+    return _get(f"/xhs/ocr/batches/{batch_key.strip()}/collections") or []
+
+
+def get_summary(batch_key: str = "", collection_key: str = "") -> dict[str, Any]:
+    return _get("/xhs/ocr/dashboard/summary", {
+        "batch_key": batch_key, "collection_key": collection_key,
+    }) or {}
+
+
+def list_results(params: dict[str, Any]) -> dict[str, Any]:
+    allowed = {
+        "batch_key", "collection_key", "question_code", "search_keyword", "city",
+        "department_name", "minsheng_matched", "cobranded_matched", "is_valid",
+        "page", "page_size",
+    }
+    query = {key: value for key, value in params.items() if key in allowed}
+    page = int(query.get("page") or 1)
+    page_size = int(query.get("page_size") or 50)
+    if page < 1 or not 1 <= page_size <= 200:
+        raise TaskError("OCR_INVALID_PAGINATION", "分页参数无效", 422)
+    query.update({"page": page, "page_size": page_size})
+    return _get("/xhs/ocr/dashboard/results", query) or {
+        "items": [], "total": 0, "page": page, "page_size": page_size,
+    }
+
+
+def get_result_matches(result_id: int) -> dict[str, Any]:
+    if result_id < 1:
+        raise TaskError("OCR_RESULT_ID_INVALID", "result_id 必须大于 0", 422)
+    return _get(f"/xhs/ocr/results/{result_id}/matches") or {}
+
+
+def list_evidences(params: dict[str, Any]) -> dict[str, Any]:
+    allowed = {"batch_key", "collection_key", "question_code", "evidence_type", "limit"}
+    query = {key: value for key, value in params.items() if key in allowed}
+    if not str(query.get("batch_key") or "").strip() or not str(query.get("collection_key") or "").strip():
+        raise TaskError("OCR_EVIDENCE_SCOPE_REQUIRED", "batch_key 和 collection_key 不能为空", 422)
+    limit = int(query.get("limit") or 10)
+    if not 1 <= limit <= 100:
+        raise TaskError("OCR_EVIDENCE_LIMIT_INVALID", "limit 必须在 1 到 100 之间", 422)
+    query["limit"] = limit
+    return _get("/xhs/ocr/dashboard/evidences", query) or {
+        "items": [], "total": 0, "limit": limit, "is_complete": False,
+    }
+
+
+def export_evidences(params: dict[str, Any]) -> tuple[io.BytesIO, str, int]:
+    query = dict(params)
+    query["limit"] = 100
+    data = list_evidences(query)
+    items = data.get("items") or []
+    if not items:
+        raise TaskError("OCR_EVIDENCE_EMPTY", "当前筛选范围没有可导出的截图", 404)
+    output = io.BytesIO()
+    used_names: set[str] = set()
+    with zipfile.ZipFile(output, "w", zipfile.ZIP_DEFLATED) as archive:
+        for index, item in enumerate(items, 1):
+            url = str(item.get("cos_url") or "").strip()
+            host = (urlparse(url).hostname or "").lower()
+            if not url.startswith("https://") or not host.endswith(".myqcloud.com"):
+                continue
+            try:
+                response = SESSION.get(url, timeout=Config.CRAWLER_TIMEOUT_SECONDS)
+                response.raise_for_status()
+            except requests.RequestException:
+                continue
+            suffix = Path(str(item.get("source_file_name") or urlparse(url).path)).suffix.lower()
+            if suffix not in {".jpg", ".jpeg", ".png", ".webp"}:
+                suffix = ".jpg"
+            parts = [item.get("question_code"), item.get("search_keyword"),
+                     item.get("device_code"), item.get("city"), str(item.get("sort_order") or index)]
+            stem = "_".join(re.sub(r"[^0-9A-Za-z\u4e00-\u9fff_-]+", "-", str(part))
+                            for part in parts if part not in (None, ""))
+            name = f"{stem or f'evidence-{index}'}{suffix}"
+            if name in used_names:
+                name = f"{Path(name).stem}-{index}{suffix}"
+            used_names.add(name)
+            archive.writestr(name, response.content)
+    if not used_names:
+        raise TaskError("OCR_EVIDENCE_EXPORT_FAILED", "截图下载失败，未生成导出文件", 502)
+    output.seek(0)
+    scope = str(query.get("question_code") or "all")
+    return output, f"holy-crab-evidences-{scope}.zip", len(used_names)

scripts/import_ocr_evidence_to_cos.py

@@ -0,0 +1,145 @@
+#!/usr/bin/env python3
+"""Select representative OCR screenshots, upload them to COS, and emit a DB manifest.
+
+Run this script with a Python environment that provides qcloud_cos. Credentials are
+read from the user-supplied text file and are never written to the manifest or logs.
+"""
+import argparse
+import hashlib
+import json
+import subprocess
+import tempfile
+from pathlib import Path
+from urllib.request import urlopen
+
+from qcloud_cos import CosConfig, CosS3Client
+
+
+BASE_TOKEN = "NrZAbJJG5aGNuwsgSQWcGhl8nKH"
+TABLE_ID = "tbly0jJtr3fMrcHp"
+
+
+def read_cos_config(path: Path) -> dict[str, str]:
+    lines = [line.strip() for line in path.read_text(encoding="utf-8").splitlines() if line.strip()]
+    # Delivery file format: label line followed by value line.
+    labels = {}
+    for index, line in enumerate(lines[:-1]):
+        lower = line.lower().rstrip("：:")
+        if any(name in lower for name in ("secretid", "secretkey", "bucket", "region")):
+            labels[lower] = lines[index + 1]
+    def value(fragment: str) -> str:
+        for label, val in labels.items():
+            if fragment in label:
+                return val
+        raise ValueError(f"COS configuration is missing {fragment}")
+    return {"secret_id": value("secretid"), "secret_key": value("secretkey"),
+            "bucket": value("bucket"), "region": value("region")}
+
+
+def fetch_results(api_base: str, batch_key: str, collection_key: str) -> list[dict]:
+    items = []
+    for page in range(1, 100):
+        url = (f"{api_base.rstrip('/')}/api/v1/data-dashboard/ocr/results?batch_key={batch_key}"
+               f"&collection_key={collection_key}&page={page}&page_size=200")
+        payload = json.loads(urlopen(url, timeout=60).read())
+        data = payload["data"]
+        items.extend(data.get("items") or [])
+        if len(items) >= int(data.get("total") or 0):
+            break
+    return items
+
+
+def select(items: list[dict]) -> list[dict]:
+    selected = []
+    for question in sorted({str(item.get("question_code")) for item in items}):
+        pool = [item for item in items if str(item.get("question_code")) == question
+                and item.get("is_valid") and item.get("screenshot_json")
+                and str(item.get("search_keyword") or "") in str(item.get("ocr_text") or "")]
+        chosen: list[tuple[dict, str]] = []
+        seen_tokens: set[str] = set()
+        used_cities: set[str] = set()
+        buckets = [
+            (lambda x: x.get("cobranded_matched"), "联名信用卡正文露出"),
+            (lambda x: x.get("minsheng_matched") and not x.get("cobranded_matched"), "民生正文露出"),
+            (lambda x: x.get("ranking_triggered"), "榜单露出"),
+            (lambda x: not x.get("minsheng_matched"), "未露出对照"),
+            (lambda x: not x.get("minsheng_matched"), "未露出对照"),
+        ]
+        for predicate, reason in buckets:
+            candidates = sorted((item for item in pool if predicate(item)),
+                                key=lambda item: (item.get("city") in used_cities,
+                                                  item.get("rank_position") or 999,
+                                                  item.get("result_id") or 0))
+            for item in candidates:
+                attachment = item["screenshot_json"][0]
+                token = attachment.get("file_token")
+                if token and token not in seen_tokens:
+                    chosen.append((item, reason)); seen_tokens.add(token); used_cities.add(item.get("city") or "")
+                    break
+        if len(chosen) < 5:
+            remaining = sorted(pool, key=lambda item: (item.get("city") in used_cities,
+                                                       not bool(item.get("minsheng_matched")),
+                                                       item.get("result_id") or 0))
+            for item in remaining:
+                attachment = item["screenshot_json"][0]
+                token = attachment.get("file_token")
+                if not token or token in seen_tokens:
+                    continue
+                reason = "正文露出补充样本" if item.get("minsheng_matched") else "未露出补充对照"
+                chosen.append((item, reason)); seen_tokens.add(token); used_cities.add(item.get("city") or "")
+                if len(chosen) >= 5:
+                    break
+        for order, (item, reason) in enumerate(chosen, 1):
+            attachment = item["screenshot_json"][0]
+            selected.append({**item, "attachment": attachment, "evidence_type": "body_ranking",
+                             "selection_reason": reason, "sort_order": order})
+    return selected
+
+
+def download_attachment(item: dict, output: Path) -> None:
+    subprocess.run(["lark-cli", "base", "+record-download-attachment", "--base-token", BASE_TOKEN,
+                    "--table-id", TABLE_ID, "--record-id", item["source_record_id"],
+                    "--file-token", item["attachment"]["file_token"], "--output", output.name, "--as", "user"],
+                   check=True, cwd=output.parent, stdout=subprocess.DEVNULL)
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--cos-config", required=True, type=Path)
+    parser.add_argument("--batch-key", required=True)
+    parser.add_argument("--collection-key", required=True)
+    parser.add_argument("--api-base", default="http://127.0.0.1:8000")
+    parser.add_argument("--manifest", required=True, type=Path)
+    args = parser.parse_args()
+    cfg = read_cos_config(args.cos_config)
+    client = CosS3Client(CosConfig(Region=cfg["region"], SecretId=cfg["secret_id"], SecretKey=cfg["secret_key"]))
+    chosen = select(fetch_results(args.api_base, args.batch_key, args.collection_key))
+    manifest = []
+    with tempfile.TemporaryDirectory(prefix="holy-crab-evidence-") as temp_dir:
+        for index, item in enumerate(chosen, 1):
+            attachment = item["attachment"]
+            suffix = Path(attachment.get("name") or "evidence.jpg").suffix.lower() or ".jpg"
+            local = Path(temp_dir) / f"{item['question_code']}-{item['result_id']}{suffix}"
+            download_attachment(item, local)
+            digest = hashlib.sha256(local.read_bytes()).hexdigest()
+            key = (f"holy-crab/ocr-evidences/{args.batch_key}/{args.collection_key}/"
+                   f"{item['question_code']}/{item['result_id']}-{digest[:12]}{suffix}")
+            client.upload_file(Bucket=cfg["bucket"], LocalFilePath=str(local), Key=key, PartSize=10, MAXThread=4)
+            url = f"https://{cfg['bucket']}.cos.{cfg['region']}.myqcloud.com/{key}"
+            manifest.append({"batch_key": args.batch_key, "collection_key": args.collection_key,
+                "question_code": item["question_code"], "result_id": item["result_id"],
+                "evidence_type": item["evidence_type"], "sort_order": item["sort_order"],
+                "city": item.get("city"), "device_code": item.get("device_code"),
+                "source_record_id": item["source_record_id"], "source_file_token": attachment["file_token"],
+                "source_file_name": attachment.get("name"), "cos_object_key": key, "cos_url": url,
+                "selection_reason": item["selection_reason"], "metadata": {
+                    "minsheng_matched": bool(item.get("minsheng_matched")),
+                    "cobranded_matched": bool(item.get("cobranded_matched")),
+                    "ranking_triggered": bool(item.get("ranking_triggered")), "rank_position": item.get("rank_position")}})
+            print(f"uploaded {index}/{len(chosen)} {item['question_code']} result={item['result_id']}")
+    args.manifest.write_text(json.dumps(manifest, ensure_ascii=False, indent=2), encoding="utf-8")
+    print(f"manifest={args.manifest} count={len(manifest)}")
+
+
+if __name__ == "__main__":
+    main()

task_service.py

@@ -1,6 +1,7 @@
 import hashlib
 import json
 import secrets
+import shutil
 import string
 import threading
 import time
@@ -401,6 +402,33 @@ def list_tasks(q: str = "", status: str = "", page: int = 1, page_size: int = 10
         }
 
 
+def merge_tasks(public_ids: list[str]) -> dict[str, Any]:
+    ids = list(dict.fromkeys(str(item).strip() for item in public_ids if str(item).strip()))
+    if not ids:
+        raise TaskError("MERGE_TASKS_REQUIRED", "至少需要一个任务 ID", 422)
+    if len(ids) > 50:
+        raise TaskError("MERGE_TASKS_LIMIT", "单次最多合并 50 个任务", 422)
+    tasks = [get_task(public_id) for public_id in ids]
+    return {"task_ids": ids, "tasks": tasks, "total": len(tasks)}
+
+
+def delete_task(public_id: str) -> dict[str, Any]:
+    with connection() as conn:
+        row = conn.execute(
+            "SELECT id,status FROM research_tasks WHERE public_task_id=?", (public_id,)
+        ).fetchone()
+        if not row:
+            raise TaskError("TASK_NOT_FOUND", "任务不存在", 404)
+        if row["status"] not in TERMINAL_STATUSES:
+            raise TaskError("TASK_STATUS_CONFLICT", "只能删除已结束任务", 409)
+        conn.execute("UPDATE research_tasks SET retry_of_task_id=NULL WHERE retry_of_task_id=?", (row["id"],))
+        conn.execute("DELETE FROM research_tasks WHERE id=?", (row["id"],))
+    folder = Config.FILE_STORAGE_PATH / public_id
+    if folder.exists() and folder.is_dir() and not folder.is_symlink():
+        shutil.rmtree(folder)
+    return {"task_id": public_id, "deleted": True}
+
+
 def cancel_task(public_id: str, reason: str = "") -> dict[str, Any]:
     crawler_task_id = None
     with connection() as conn:

tests/test_data_dashboard_service.py

@@ -0,0 +1,19 @@
+import unittest
+from unittest.mock import Mock, patch
+
+from data_dashboard_service import list_runs
+
+
+class DataDashboardServiceTest(unittest.TestCase):
+    @patch("data_dashboard_service._request")
+    def test_list_runs_normalizes_single_run(self, request):
+        response = Mock()
+        response.json.return_value = {"ok": True, "run": {"id": "sqa_1", "status": "completed"}}
+        request.return_value = response
+        result = list_runs(task_id="sqa_1")
+        self.assertEqual(result["total"], 1)
+        self.assertEqual(result["items"][0]["id"], "sqa_1")
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_ocr_dashboard_service.py

@@ -0,0 +1,47 @@
+import unittest
+from unittest.mock import Mock, patch
+
+from ocr_dashboard_service import export_evidences, get_summary, list_batches, list_results
+from task_service import TaskError
+
+
+class OcrDashboardServiceTest(unittest.TestCase):
+    @patch("ocr_dashboard_service.SESSION.get")
+    def test_list_batches_unwraps_framework_response(self, get):
+        response = Mock()
+        response.json.return_value = {"success": True, "code": 200, "data": [{"batch_key": "b1"}]}
+        get.return_value = response
+        self.assertEqual(list_batches()[0]["batch_key"], "b1")
+
+    @patch("ocr_dashboard_service.SESSION.get")
+    def test_summary_forwards_scope(self, get):
+        response = Mock()
+        response.json.return_value = {"success": True, "code": 200, "data": {"questions": []}}
+        get.return_value = response
+        get_summary("b1", "c1")
+        self.assertEqual(get.call_args.kwargs["params"], {"batch_key": "b1", "collection_key": "c1"})
+
+    def test_results_rejects_oversized_page(self):
+        with self.assertRaises(TaskError):
+            list_results({"page_size": "201"})
+
+    @patch("ocr_dashboard_service.SESSION.get")
+    @patch("ocr_dashboard_service.list_evidences")
+    def test_export_evidences_builds_zip(self, list_evidences, get):
+        list_evidences.return_value = {"items": [{
+            "question_code": "Q1", "search_keyword": "山姆会员卡", "device_code": "DEV-1",
+            "city": "厦门", "sort_order": 1, "source_file_name": "shot.jpg",
+            "cos_url": "https://bucket.cos.ap-guangzhou.myqcloud.com/shot.jpg",
+        }]}
+        response = Mock(content=b"image-bytes")
+        get.return_value = response
+
+        payload, filename, count = export_evidences({"batch_key": "b1", "collection_key": "c1"})
+
+        self.assertEqual(filename, "holy-crab-evidences-all.zip")
+        self.assertEqual(count, 1)
+        self.assertTrue(payload.read(2) == b"PK")
+
+
+if __name__ == "__main__":
+    unittest.main()