diff --git a/app.py b/app.py index e95448a..89e1b5a 100644 --- a/app.py +++ b/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//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//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( diff --git a/auth_service.py b/auth_service.py index 7ee6473..4ffb0e1 100644 --- a/auth_service.py +++ b/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 diff --git a/config.py b/config.py index 7d3fae8..c897457 100644 --- a/config.py +++ b/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") diff --git a/data_dashboard_service.py b/data_dashboard_service.py new file mode 100644 index 0000000..03e0cc8 --- /dev/null +++ b/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} + diff --git a/ocr_dashboard_service.py b/ocr_dashboard_service.py new file mode 100644 index 0000000..3814a22 --- /dev/null +++ b/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) diff --git a/scripts/import_ocr_evidence_to_cos.py b/scripts/import_ocr_evidence_to_cos.py new file mode 100644 index 0000000..1b90aa0 --- /dev/null +++ b/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() diff --git a/task_service.py b/task_service.py index de50d3d..488925f 100644 --- a/task_service.py +++ b/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: diff --git a/tests/test_data_dashboard_service.py b/tests/test_data_dashboard_service.py new file mode 100644 index 0000000..e9bf629 --- /dev/null +++ b/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() diff --git a/tests/test_ocr_dashboard_service.py b/tests/test_ocr_dashboard_service.py new file mode 100644 index 0000000..f8e2653 --- /dev/null +++ b/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()