diff --git a/backend/api/prescreen.py b/backend/api/prescreen.py index 59638de..7ba7f48 100644 --- a/backend/api/prescreen.py +++ b/backend/api/prescreen.py @@ -22,6 +22,8 @@ from __future__ import annotations +import hashlib + from fastapi import APIRouter, Depends, HTTPException, Request from fastapi.exception_handlers import request_validation_exception_handler from fastapi.exceptions import RequestValidationError @@ -163,6 +165,67 @@ def _limit(value: int) -> int: return max(1, min(value, 100)) +# ------------------------------------------------- 既有上传入口的批量摄取 +def _pick_item(batch: dict, filename: str, item_id: str | None) -> dict: + items = batch["items"] + if item_id: + item = next((i for i in items if i["id"] == item_id), None) + if item is None: + raise KeyError("导入条目不存在") + return item + base = filename.replace("\\", "/").rsplit("/", 1)[-1] + matches = [i for i in items + if i["relative_path"].replace("\\", "/").rsplit("/", 1)[-1] == base] + if not matches: + raise KeyError("该文件不在导入批次清单中") + if len(matches) > 1: + raise ValueError("批次中有多个同名文件，请指定 item_id") + return matches[0] + + +def ingest_into_batch(project_id: str, filename: str, content: bytes, actor: str, + batch_id: str, item_id: str | None = None): + """把文件字节摄入**既有**上传链，并回写批次条目（T00/04 §4.2）。 + + 不建立平行上传 API：仍走 `ProjectAssistantService.upload()`，因此解析、 + 去重（同哈希复用原件）、pending_candidates、data_revision、mark_stale 等 + 既有行为完全不变。本函数只做"清单核对 + 状态回写"。 + + 服务端**重算**哈希与大小，不信任客户端清单（B-22 的来源可信度）。 + """ + with _db() as db: + batch = material_import.load(db, batch_id, project_id) + if batch["status"] == "cancelled": + raise store.Conflict("批次已取消，不能继续摄入文件") + + item = _pick_item(batch, filename, item_id) + + def fail(code: ErrorCode, message: str): + with _db() as db: + db.execute("BEGIN IMMEDIATE") + material_import.mark_item(db, batch_id, item["id"], "failed", error_code=code) + raise ValueError(message) + + if item["status"] == "parsed": + raise store.Conflict("该条目已解析完成，不能重复摄入") + digest = hashlib.sha256(content).hexdigest() + if digest != item["sha256"]: + fail(ErrorCode.INVALID_EVIDENCE, "文件内容与清单声明的 sha256 不一致，请重新选择该文件") + if len(content) != item["size_bytes"]: + fail(ErrorCode.INVALID_EVIDENCE, "文件大小与清单声明不一致，请重新选择该文件") + + # 走既有上传链（可能因同哈希而复用既有 document，这是 14 §3 的期望行为） + document = _service().upload(project_id, filename, content, actor) + + with _db() as db: + db.execute("BEGIN IMMEDIATE") + batch = material_import.mark_item(db, batch_id, item["id"], "parsed", + document_id=document["id"]) + # 返回体保持**document 形状**（既有调用方读 id 与 candidates）， + # 批次状态作为附加字段给出，不替换既有契约。 + return dict(document, import_batch=batch) + + async def validation_handler(request: Request, exc: RequestValidationError): """请求体/查询参数校验失败也走统一信封。 diff --git a/backend/api/project_assistant.py b/backend/api/project_assistant.py index f4ecf8d..f9be7f4 100644 --- a/backend/api/project_assistant.py +++ b/backend/api/project_assistant.py @@ -1,7 +1,7 @@ """Unified project API. Every resource read/write derives scope from its project.""" import base64 from urllib.parse import quote -from fastapi import APIRouter, Depends, HTTPException, Request, UploadFile, File +from fastapi import APIRouter, Depends, Form, HTTPException, Request, UploadFile, File from fastapi.responses import Response from pydantic import BaseModel, Field, ConfigDict from api.risk_agent_access import authorize, check_project @@ -221,7 +221,9 @@ def legacy(project_id: str, kind: str, legacy_id: str, request: Request): @router.post("/projects/{project_id}/documents", status_code=201) -async def upload(project_id: str, request: Request, file: UploadFile = File(...)): +async def upload(project_id: str, request: Request, file: UploadFile = File(...), + batch_id: str | None = Form(default=None), + item_id: str | None = Form(default=None)): guard(request, project_id) from services.assistant_materials import LIMIT try: @@ -230,6 +232,13 @@ async def upload(project_id: str, request: Request, file: UploadFile = File(...) await file.close() # Parsing is synchronous CPU work; do not block the ASGI event loop. from starlette.concurrency import run_in_threadpool + # S1：传入 batch_id 时走批量清单核对与状态回写（api/prescreen.ingest_into_batch）。 + # 不传时行为与既有完全一致——前端与既有调用方都不受影响。 + if batch_id: + from api.prescreen import ingest_into_batch + return await run_in_threadpool(lambda: translate(lambda: ingest_into_batch( + project_id, file.filename or "material", content, + request.state.identity["id"], batch_id, item_id))) return await run_in_threadpool(lambda: translate(lambda: service().upload(project_id, file.filename or "material", content, request.state.identity["id"]))) diff --git a/backend/test_prescreen_api.py b/backend/test_prescreen_api.py index c70fd59..53c5cb8 100644 --- a/backend/test_prescreen_api.py +++ b/backend/test_prescreen_api.py @@ -3,6 +3,7 @@ 只挂 prescreen 路由到独立 app，数据库指向临时文件——不触碰真实库。 """ +import hashlib import os import tempfile import unittest @@ -320,5 +321,156 @@ class PrescreenApiTests(unittest.TestCase): self.assertNotIn(banned, source) +class BatchIngestTests(unittest.TestCase): + """S1c′：既有上传入口的批量摄取（T00/04 §4.2）。 + + 文档端点属于 project_assistant 路由，故这里同时挂两个路由，与 app.py 一致。 + """ + + def setUp(self): + self.temp = tempfile.TemporaryDirectory(ignore_cleanup_errors=True) + self.path = Path(self.temp.name) + self.service = ProjectAssistantService(self.path / "assistant.sqlite3") + self.patches = [ + patch.object(api, "_service", lambda: self.service), + patch.dict(os.environ, {"XINGBANG_AUTH_USERS": "", "XINGBANG_REQUIRE_AUTH": "0", + "XINGBANG_LOCAL_MODEL_URL": ""}), + ] + for p in self.patches: + p.start() + from fastapi.exceptions import RequestValidationError as RVE + from api import project_assistant as legacy + # 两个路由各自的 service 工厂都要指向临时库——否则 POST /documents + # 会落到真实的 backend/xingbang_data 库上（测试隔离泄漏）。 + self.patches.append(patch.object(legacy, "service", lambda: self.service)) + for p in self.patches: + p.start() + app = FastAPI() + app.add_exception_handler(RVE, api.validation_handler) + app.include_router(legacy.router) + app.include_router(api.router) + self.client = TestClient(app) + self.base = "/api/xingbang/risk-agent/project-assistant" + self.pre = self.base + "/prescreen" + + def tearDown(self): + self.client.close() + for p in reversed(self.patches): + p.stop() + self.temp.cleanup() + + def make_batch(self, payload: bytes, name="流水.csv"): + entry = {"relative_path": name, "sha256": hashlib.sha256(payload).hexdigest(), + "size_bytes": len(payload), "mime_type": "text/csv"} + response = self.client.post(f"{self.pre}/projects/{PROJECT}/import-batches", + json={"manifest": [entry]}, headers=HEADERS) + self.assertEqual(response.status_code, 202, response.text) + return response.json()["batch"] + + def send(self, batch, payload: bytes, name="流水.csv", item_id=None, batch_id=None): + data = {"batch_id": batch["id"] if batch_id is None else batch_id} + data["item_id"] = item_id if item_id is not None else batch["items"][0]["id"] + return self.client.post(f"{self.base}/projects/{PROJECT}/documents", + files={"file": (name, payload, "text/csv")}, + data=data, headers=HEADERS) + + def test_ingest_marks_item_parsed_and_links_document(self): + payload = "field,value\nfacility.amount,100\n".encode("utf-8-sig") + batch = self.make_batch(payload) + response = self.send(batch, payload) + self.assertEqual(response.status_code, 201, response.text) + document_id = response.json()["id"] + after = self.client.get(f"{self.pre}/import-batches/{batch['id']}", headers=HEADERS).json() + self.assertEqual(after["status"], "ready") + self.assertEqual(after["items"][0]["status"], "parsed") + self.assertEqual(after["items"][0]["document_id"], document_id) + + def test_ingest_by_filename_when_item_id_absent(self): + payload = b"field,value\nfacility.amount,100\n" + batch = self.make_batch(payload) + response = self.client.post(f"{self.base}/projects/{PROJECT}/documents", + files={"file": ("流水.csv", payload, "text/csv")}, + data={"batch_id": batch["id"]}, headers=HEADERS) + self.assertEqual(response.status_code, 201, response.text) + after = self.client.get(f"{self.pre}/import-batches/{batch['id']}", headers=HEADERS).json() + self.assertEqual(after["items"][0]["status"], "parsed") + + def test_hash_mismatch_marks_failed_and_rejects(self): + declared = b"field,value\nfacility.amount,100\n" + batch = self.make_batch(declared) + tampered = b"field,value\nfacility.amount,999\n" + response = self.send(batch, tampered) + self.assertEqual(response.status_code, 422, response.text) + after = self.client.get(f"{self.pre}/import-batches/{batch['id']}", headers=HEADERS).json() + self.assertEqual(after["items"][0]["status"], "failed") + self.assertEqual(after["items"][0]["error_code"], "INVALID_EVIDENCE") + self.assertIsNone(after["items"][0]["document_id"]) + + def test_size_mismatch_marks_failed(self): + payload = b"field,value\nfacility.amount,100\n" + entry = {"relative_path": "x.csv", "sha256": hashlib.sha256(payload).hexdigest(), + "size_bytes": len(payload) + 5, "mime_type": "text/csv"} + batch = self.client.post(f"{self.pre}/projects/{PROJECT}/import-batches", + json={"manifest": [entry]}, headers=HEADERS).json()["batch"] + response = self.send(batch, payload, name="x.csv") + self.assertEqual(response.status_code, 422, response.text) + after = self.client.get(f"{self.pre}/import-batches/{batch['id']}", headers=HEADERS).json() + self.assertEqual(after["items"][0]["status"], "failed") + + def test_item_not_in_batch_rejected(self): + payload = b"field,value\nfacility.amount,100\n" + batch = self.make_batch(payload) + response = self.send(batch, payload, item_id="item-not-mine") + self.assertEqual(response.status_code, 404, response.text) + + def test_filename_not_in_manifest_rejected(self): + payload = b"field,value\nfacility.amount,100\n" + batch = self.make_batch(payload, name="declared.csv") + response = self.client.post(f"{self.base}/projects/{PROJECT}/documents", + files={"file": ("other.csv", payload, "text/csv")}, + data={"batch_id": batch["id"]}, headers=HEADERS) + self.assertEqual(response.status_code, 404, response.text) + + def test_double_ingest_rejected(self): + payload = b"field,value\nfacility.amount,100\n" + batch = self.make_batch(payload) + self.assertEqual(self.send(batch, payload).status_code, 201) + again = self.send(batch, payload) + self.assertEqual(again.status_code, 409, again.text) + + def test_cancelled_batch_refuses_ingest(self): + payload = b"field,value\nfacility.amount,100\n" + batch = self.make_batch(payload) + self.client.post(f"{self.pre}/import-batches/{batch['id']}/cancel", + json={"expected_revision": batch["revision"]}, headers=HEADERS) + response = self.send(batch, payload) + self.assertEqual(response.status_code, 409, response.text) + + def test_plain_upload_without_batch_id_is_unchanged(self): + """既有行为回归：不传 batch_id 时仍走原路径，返回解析候选。""" + payload = b"field,value\nfacility.amount,100\n" + response = self.client.post(f"{self.base}/projects/{PROJECT}/documents", + files={"file": ("plain.csv", payload, "text/csv")}, + headers=HEADERS) + self.assertEqual(response.status_code, 201, response.text) + body = response.json() + self.assertIn("candidates", body) + self.assertNotIn("raw_base64", body) + self.assertTrue(body["id"].startswith("doc-")) + + def test_end_to_end_batch_closes_the_loop(self): + """S1 首次闭环：登记批次 → 上传 → 解析定位 → 可确认。""" + payload = "field,value,currency,unit,period,scope\nfacility.amount,100,CNY,万元,2025,合并\n".encode("utf-8-sig") + batch = self.make_batch(payload) + self.assertEqual(self.send(batch, payload).status_code, 201) + final = self.client.get(f"{self.pre}/import-batches/{batch['id']}", headers=HEADERS).json() + self.assertEqual(final["status"], "ready") + document_id = final["items"][0]["document_id"] + detail = self.client.get(f"{self.base}/projects/{PROJECT}/documents/{document_id}", + headers=HEADERS) + self.assertEqual(detail.status_code, 200, detail.text) + self.assertTrue(detail.json()["candidates"], "解析应产出候选字段") + + if __name__ == "__main__": unittest.main()