新增日志+针对请求体连接进行处理

README.md

@@ -116,6 +116,25 @@ sh deploy.sh restart    # 停止后重新后台启动
 
 日志位于 `output/api.log`，PID 位于 `output/api.pid`。
 
+请求审计日志位于 `output/requests.log`，UTF-8 JSONL 格式，每行一条事件，追加写入。
+记录原始 JSON 请求体、接口响应状态码及 JSON 正文，以及每次回调的地址、载荷、
+HTTP 状态码、响应正文或异常；通过 `request_id` 关联，成功创建任务时它也是任务 ID。
+校验失败的请求也记录。multipart 记录表单字段和文件名/大小/类型，不记录文件二进制。
+
+`processing_request` 在校验完成、提交后台任务之前记录最终处理参数：`body` 中为复制到
+任务目录后的实际文件绝对路径、规范化回调地址、txbId 和生效的清理开关；两个路径模式
+标记均为 true，表示这些最终路径已经是绝对路径。JSON 模式额外通过 `source_paths`
+记录删除 static、目录拼接后的源文件绝对路径，便于对照原始请求。CLA_FILE 仅保存，
+不参与生成，日志通过 `clarification_used_in_pipeline=false` 明确标注。
+
+下载响应仅记录状态码，不记录 DOCX 内容。日志不随任务中间文件清理，重启后仍保留。
+回调 HTTP 200 的业务错误正文会原样记录，原有按 HTTP 状态判断投递成功的规则不变。
+日志包含请求中的路径和回调信息，仅保存在服务器，按运维要求限制访问并定期归档。
+
+```bash
+tail -f output/requests.log
+```
+
 可用环境变量：`MODEL_REPO`（默认
 `Maiteka/gpt2-chinese-cluecorpussmall-onnx`）、`MODEL_DIR`（默认
 `models/gpt2-chinese-cluecorpussmall-onnx`）、`UV_CACHE_DIR`（默认
@@ -195,6 +214,11 @@ curl.exe -X POST "http://127.0.0.1:8000/api/v1/final-review" `
 
 ### REFERENCE_BID 路径处理
 
+JSON 模式的 `TENDER_FILE`、`PROCUREMENT_FILE`、`REFERENCE_BID`、`CLA_FILE` 在路径
+拼接或绝对读取之前，统一删除独立的 `static/`（或 `static\`）目录段。例如
+`static/file/example.docx` 变成 `file/example.docx`。保留其他目录、文件名和绝对路径根；
+不修改配置目录本身，`mystatic/`、`static.docx` 不受影响。multipart 上传不做路径清理。
+
 JSON 模式下：
 
 - `REFERENCE_BID` 为绝对路径时直接使用。

pyproject.toml

@@ -35,6 +35,7 @@ py-modules = [
     "context_search",
     "api_worker",
     "http_api",
+    "request_logging",
     "main",
     "models",
     "workflow",

scripts/test_step4.py

@@ -205,12 +205,7 @@ def _build_report(outline, analysis=None):
             errors.append(f"模板章块顺序错误: {chapter.id} {chapter.title}: {kinds}")
 
     records = [_chapter_record(chapter, entry_counts) for chapter in outline.chapters]
-    for record in records:
-        if record["unresolved_placeholders"]:
-            errors.append(
-                f"章节产物仍有未解析占位符: {record['id']} {record['title']}: "
-                + ", ".join(record["unresolved_placeholders"][:10])
-            )
+    # 残留占位符保留在章节报告中供人工检查，不阻断后续审核和 DOCX 导出。
     project_fields = dict(
         getattr(analysis, "project_fields", {}) or {}
     ) if analysis is not None else {}

scripts/tests/test_api.py

@@ -135,6 +135,11 @@ class ApiWorkerTests(unittest.TestCase):
 
 class FinalReviewApiTests(unittest.TestCase):
     def setUp(self):
+        log_dir = tempfile.TemporaryDirectory()
+        self.addCleanup(log_dir.cleanup)
+        log_patch = patch("request_logging.LOG_PATH", Path(log_dir.name) / "requests.log")
+        log_patch.start()
+        self.addCleanup(log_patch.stop)
         with http_api._JOBS_LOCK:
             http_api._JOBS.clear()
 
@@ -731,6 +736,9 @@ class FinalReviewApiTests(unittest.TestCase):
         class Response:
             status = 200
 
+            def read(self):
+                return b'{"code":200}'
+
             def __enter__(self):
                 return self
 

scripts/tests/test_request_logging.py

@@ -0,0 +1,173 @@
+from io import BytesIO
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+import json
+from pathlib import Path
+import tempfile
+import threading
+import unittest
+from unittest.mock import patch
+from urllib.error import HTTPError, URLError
+
+from fastapi.testclient import TestClient
+import http_api
+import request_logging
+from scripts.tests.test_api import RecordingExecutor, _write_json_input_files, _valid_files, _write_docx
+
+
+class RequestLoggingTests(unittest.TestCase):
+    def setUp(self):
+        self.temp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.temp.cleanup)
+        self.root = Path(self.temp.name)
+        self.log = self.root / "requests.log"
+        self.patch = patch.object(request_logging, "LOG_PATH", self.log)
+        self.patch.start()
+        self.addCleanup(self.patch.stop)
+
+    def events(self):
+        return [json.loads(line) for line in self.log.read_text(encoding="utf-8").splitlines()]
+
+    def test_static_directory_only_preserves_root_and_filename(self):
+        for source, expected in [
+            ("static/file/a.docx", "file/a.docx"),
+            ("/static/file/static/a.docx", "/file/a.docx"),
+            (r"C:\static\file\a.docx", r"C:\file\a.docx"),
+            ("/data/static//file/a.docx", "/data/file/a.docx"),
+            ("/data/mystatic/static.docx", "/data/mystatic/static.docx"),
+            ("//server/share/static/a.docx", "//server/share/a.docx"),
+        ]:
+            with self.subTest(source=source):
+                self.assertEqual(http_api._strip_static_directory(source), expected)
+
+    def test_four_paths_cleaned_and_original_request_logged_for_both_modes(self):
+        tender, procurement, reference = _write_json_input_files(self.root)
+        cla = self.root / "clarification.docx"
+        cla.write_bytes(b"optional")
+        for absolute in (True, False):
+            with self.subTest(absolute=absolute):
+                paths = dict(zip(("TENDER_FILE", "PROCUREMENT_FILE", "REFERENCE_BID", "CLA_FILE"),
+                                 (tender, procurement, reference, cla)))
+                payload = {key: str(path.parent / "static" / path.name) if absolute else "static/" + path.name
+                           for key, path in paths.items()}
+                payload.update(CALLBACK_URL="http://localhost/callback", txbId="TEST",
+                               TPC_IS_ABSOLUTE=absolute, REFERENCE_IS_ABSOLUTE=absolute)
+                executor = RecordingExecutor()
+                with patch.object(http_api, "API_WORK_ROOT", self.root / "jobs"), \
+                     patch.object(http_api, "TPC_DIR", str(self.root)), \
+                     patch.object(http_api, "REFERENCE_DIR", str(self.root)), \
+                     patch.object(http_api, "_JOB_EXECUTOR", executor):
+                    response = TestClient(http_api.app).post("/api/v1/final-review", json=payload)
+                self.assertEqual(response.status_code, 202, response.text)
+                job = executor.submitted[0][1][0]
+                self.assertEqual(set(job["files"]), set(paths))
+                records = [e for e in self.events() if e["request_id"] == response.json()["request_id"]]
+                self.assertEqual(records[0]["body"], payload)
+                effective = next(e for e in records if e["event"] == "processing_request")
+                self.assertEqual(effective["source_paths"], {key: str(path.resolve()) for key, path in paths.items()})
+                self.assertEqual(effective["body"]["TENDER_FILE"], str(job["tender_path"].resolve()))
+                self.assertTrue(effective["body"]["TPC_IS_ABSOLUTE"])
+                self.assertTrue(effective["body"]["CLEAN_INTERMEDIATE"])
+                self.assertEqual(effective["body"]["CALLBACK_URL"], "http://localhost/callback")
+                self.assertEqual(records[-1]["body"], response.json())
+                self.assertEqual(records[-1]["status"], 202)
+
+    def test_rejected_and_invalid_json_requests_are_logged(self):
+        client = TestClient(http_api.app)
+        for body, status in [("{broken", 400), ('{"txbId":"TEST"}', 422)]:
+            response = client.post("/api/v1/final-review", content=body,
+                                   headers={"Content-Type": "application/json"})
+            self.assertEqual(response.status_code, status)
+            records = self.events()[-2:]
+            self.assertEqual(records[0]["request_id"], records[1]["request_id"])
+            self.assertEqual(records[1]["body"], response.json())
+        self.assertFalse(any(e["event"] == "processing_request" for e in self.events()))
+
+    def test_multipart_records_fields_and_file_metadata_without_binary(self):
+        with patch.object(http_api, "API_WORK_ROOT", self.root / "jobs"), \
+             patch.object(http_api, "_JOB_EXECUTOR", RecordingExecutor()):
+            response = TestClient(http_api.app).post("/api/v1/final-review",
+                data={"CALLBACK_URL": "http://localhost/callback", "txbId": "TEST"},
+                files=_valid_files())
+        self.assertEqual(response.status_code, 202)
+        record = next(e for e in self.events() if e["event"] == "request")
+        fields = {entry["name"]: entry["value"] for entry in record["body"]}
+        self.assertEqual(fields["txbId"], "TEST")
+        self.assertGreater(fields["TENDER_FILE"]["size"], 0)
+        self.assertNotIn("%PDF", self.log.read_text(encoding="utf-8"))
+
+    def test_callback_http_error_retry_and_business_response_logged(self):
+        class Response(BytesIO):
+            status = 200
+        responses = [HTTPError("http://localhost", 503, "unavailable", {}, BytesIO(b"busy")),
+                     Response(b'{"code":400,"message":"unknown id"}')]
+        with patch.object(http_api, "urlopen", side_effect=responses), \
+             patch.object(http_api.time, "sleep"), patch.object(http_api, "CALLBACK_MAX_RETRIES", 2):
+            http_api._send_callback("http://localhost/callback", {"request_id": "TEST"})
+        records = self.events()
+        self.assertEqual([e["event"] for e in records],
+                         ["callback_request", "callback_response"] * 2)
+        self.assertEqual(records[1]["status"], 503)
+        self.assertEqual(records[1]["response_body"], "busy")
+        self.assertEqual(records[3]["response_body"]["code"], 400)
+        self.assertEqual(records[3]["attempt"], 2)
+
+    def test_callback_network_failure_logged(self):
+        with patch.object(http_api, "urlopen", side_effect=URLError("offline")), \
+             patch.object(http_api, "CALLBACK_MAX_RETRIES", 1):
+            with self.assertRaises(RuntimeError):
+                http_api._send_callback("http://localhost/callback", {"request_id": "TEST"})
+        self.assertIn("offline", self.events()[-1]["error"])
+
+    def test_log_write_failure_does_not_fail_request(self):
+        with patch.object(request_logging, "LOG_PATH", self.root), \
+             self.assertLogs("request_logging", level="ERROR"):
+            self.assertEqual(TestClient(http_api.app).get("/health").status_code, 200)
+
+    def test_local_http_callback_and_job_cleanup_keep_correlated_log(self):
+        received = []
+
+        class Handler(BaseHTTPRequestHandler):
+            def do_POST(self):
+                received.append(json.loads(self.rfile.read(int(self.headers["Content-Length"]))))
+                self.send_response(200)
+                self.end_headers()
+                self.wfile.write(b'{"code":200}')
+
+            def log_message(self, *_args):
+                pass
+
+        server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
+        thread = threading.Thread(target=server.serve_forever, daemon=True)
+        thread.start()
+        try:
+            executor = RecordingExecutor()
+            with patch.object(http_api, "API_WORK_ROOT", self.root / "jobs"), \
+                 patch.object(http_api, "_JOB_EXECUTOR", executor), \
+                 patch.object(http_api, "_run_pipeline_worker",
+                              side_effect=lambda _a, _b, _c, output, _log: _write_docx(output)):
+                client = TestClient(http_api.app)
+                response = client.post("/api/v1/final-review", files=_valid_files(),
+                    data={"CALLBACK_URL": f"http://127.0.0.1:{server.server_port}/callback", "txbId": "TEST"})
+                self.assertEqual(response.status_code, 202)
+                function, args = executor.submitted[0]
+                function(*args)
+                job_id = response.json()["request_id"]
+                self.assertTrue(client.get(f"/api/v1/jobs/{job_id}").json()["callback_delivered"])
+                self.assertEqual(client.get(f"/api/v1/jobs/{job_id}/file").status_code, 200)
+            self.assertEqual(received[0]["request_id"], job_id)
+            self.assertFalse(args[0]["job_dir"].exists())
+            callback = next(e for e in self.events() if e["event"] == "callback_response")
+            self.assertEqual(callback["request_id"], job_id)
+            self.assertEqual(callback["response_body"], {"code": 200})
+            self.assertEqual(self.events()[-1]["body"], "[non-JSON response omitted]")
+        finally:
+            server.shutdown()
+            server.server_close()
+            thread.join()
+
+    def test_early_server_failure_still_logs_original_json(self):
+        with patch.object(http_api, "TEMPLATE_PATH", self.root / "missing.docx"):
+            response = TestClient(http_api.app).post("/api/v1/final-review", json={"txbId": "TEST"})
+        self.assertEqual(response.status_code, 500)
+        self.assertEqual(self.events()[0]["body"], {"txbId": "TEST"})
+        self.assertEqual(self.events()[-1]["status"], 500)

scripts/tests/test_step4_reporting.py

@@ -140,11 +140,12 @@ class Step4ReportingTests(unittest.TestCase):
         self.assertFalse(report["passed"])
         self.assertIn("直接落点缺少补充正文", report["errors"][0])
 
-    def test_unresolved_placeholder_in_chapter_artifact_fails_gate(self):
+    def test_unresolved_placeholders_are_reported_without_failing_gate(self):
         with tempfile.TemporaryDirectory() as directory:
             artifact_path = os.path.join(directory, "chapter.docx")
             document = Document()
-            document.add_paragraph("项目编号：%%项目编号%%")
+            document.add_paragraph("%%报价汇总表%%")
+            document.add_table(rows=1, cols=1).cell(0, 0).text = "%%服务要求%%"
             document.save(artifact_path)
             chapter = Chapter(
                 id="1",
@@ -156,11 +157,13 @@ class Step4ReportingTests(unittest.TestCase):
                 chapters=[chapter],
             ))
 
-        self.assertFalse(report["passed"])
-        self.assertTrue(any(
-            "章节产物仍有未解析占位符" in error
-            for error in report["errors"]
-        ))
+        self.assertTrue(report["passed"], report["errors"])
+        self.assertEqual(report["errors"], [])
+        self.assertEqual(
+            report["chapters"][0]["unresolved_placeholders"],
+            ["%%报价汇总表%%", "%%服务要求%%"],
+            "残留占位符应保留在报告中，但不得重新加入阻断 errors。",
+        )
 
 
 if __name__ == "__main__":

src/http_api.py

@@ -14,6 +14,7 @@ import sys
 import threading
 import time
 from urllib.parse import urlsplit
+from urllib.error import HTTPError
 from urllib.request import Request as UrlRequest, urlopen
 from uuid import uuid4
 import xml.etree.ElementTree as ET
@@ -22,6 +23,7 @@ import zipfile
 from dotenv import load_dotenv
 from fastapi import FastAPI, HTTPException, Request, UploadFile
 from fastapi.responses import FileResponse
+from request_logging import RequestLogMiddleware, decode_body, write_event
 
 
 PROJECT_ROOT = Path(__file__).resolve().parent.parent
@@ -66,6 +68,9 @@ app = FastAPI(
 )
 
 
+app.add_middleware(RequestLogMiddleware)
+
+
 def _require_suffix(upload: UploadFile, field_name: str, expected: str) -> str:
     suffix = Path(upload.filename or "").suffix.lower()
     if suffix != expected:
@@ -210,6 +215,7 @@ def _parse_path_is_absolute(value, field_name: str) -> bool:
 
 def _resolve_reference_path(reference_value: str, is_absolute: bool) -> str:
     """按配置顺序查找相对参考投书，绝对路径模式原样返回。"""
+    reference_value = _strip_static_directory(reference_value)
     if is_absolute:
         return reference_value
     reference_dirs = [
@@ -234,8 +240,14 @@ def _resolve_reference_path(reference_value: str, is_absolute: bool) -> str:
     )
 
 
+def _strip_static_directory(value: str) -> str:
+    # Match directory components only, preserving roots and file names.
+    return re.sub(r"(?<![^/\\])static[/\\]+", "", value)
+
+
 def _resolve_tpc_path(path_value: str, is_absolute: bool, field_name: str) -> str:
     """TPC 相对路径与统一根目录拼接；绝对路径模式原样返回。"""
+    path_value = _strip_static_directory(path_value)
     if is_absolute:
         return path_value
     if not TPC_DIR:
@@ -382,6 +394,9 @@ def _send_callback(callback_url: str, payload: dict) -> None:
     body = json.dumps(payload, ensure_ascii=False).encode("utf-8")
     last_error: Exception | None = None
     for attempt in range(1, CALLBACK_MAX_RETRIES + 1):
+        fields = {"request_id": payload.get("request_id"), "url": callback_url,
+                  "attempt": attempt, "body": payload}
+        write_event("callback_request", **fields)
         request = UrlRequest(
             callback_url,
             data=body,
@@ -390,10 +405,21 @@ def _send_callback(callback_url: str, payload: dict) -> None:
         )
         try:
             with urlopen(request, timeout=CALLBACK_TIMEOUT_SECONDS) as response:
+                write_event("callback_response", **fields, status=response.status,
+                            response_body=decode_body(response.read()))
                 if 200 <= response.status < 300:
                     return
                 raise RuntimeError(f"回调返回 HTTP {response.status}")
         except Exception as exc:
+            if isinstance(exc, HTTPError):
+                try:
+                    response_body = decode_body(exc.read())
+                finally:
+                    exc.close()
+                write_event("callback_response", **fields, status=exc.code,
+                            response_body=response_body, error=str(exc))
+            else:
+                write_event("callback_error", **fields, error=str(exc))
             last_error = exc
             if attempt < CALLBACK_MAX_RETRIES:
                 time.sleep(min(2 ** (attempt - 1), 5))
@@ -595,6 +621,14 @@ async def generate_final_review(
 
     if is_multipart:
         form = await request.form()
+        request.state.audit_form = [
+            {"name": name, "value": {"filename": value.filename,
+             "size": value.size, "content_type": value.content_type}
+             if hasattr(value, "filename") else value}
+            for name, value in form.multi_items()
+        ]
+        write_event("request_form", request_id=request.state.audit_id,
+                    body=request.state.audit_form)
         callback_url = _normalize_callback_url(
             _required_form_value(form, "CALLBACK_URL")
         )
@@ -660,7 +694,7 @@ async def generate_final_review(
         _require_suffix_from_string(reference_suffix, "REFERENCE_BID", ".docx")
 
     API_WORK_ROOT.mkdir(parents=True, exist_ok=True)
-    request_id = uuid4().hex
+    request_id = request.state.audit_id
     request_root = API_WORK_ROOT / request_id
     job_dir = request_root / "work"
     job_dir.mkdir(parents=True, exist_ok=True)
@@ -756,6 +790,28 @@ async def generate_final_review(
         }
         with _JOBS_LOCK:
             _JOBS[request_id] = dict(job)
+        write_event(
+            "processing_request", request_id=request_id,
+            body={
+                "CALLBACK_URL": callback_url,
+                "txbId": txb_id,
+                "TENDER_FILE": str(tender_path.resolve()),
+                "PROCUREMENT_FILE": str(procurement_path.resolve()),
+                "REFERENCE_BID": str(reference_path.resolve()),
+                "CLA_FILE": str(clarification_path.resolve()) if "CLA_FILE" in files else "",
+                "TPC_IS_ABSOLUTE": True,
+                "REFERENCE_IS_ABSOLUTE": True,
+                "CLEAN_INTERMEDIATE": clean_intermediate,
+            },
+            source_paths=None if is_multipart else {
+                "TENDER_FILE": str(Path(tender_source).resolve()),
+                "PROCUREMENT_FILE": str(Path(procurement_source).resolve()),
+                "REFERENCE_BID": str(Path(reference_source).resolve()),
+                "CLA_FILE": str(Path(cla_source).resolve()) if cla_source else "",
+            },
+            output_path=output_path,
+            clarification_used_in_pipeline=False,
+        )
         _JOB_EXECUTOR.submit(_process_job, job)
     except HTTPException:
         shutil.rmtree(request_root, ignore_errors=True)

src/request_logging.py

@@ -0,0 +1,95 @@
+"""Append-only JSONL audit events, independent of per-job cleanup."""
+
+from datetime import datetime, timezone
+import json
+import logging
+from pathlib import Path
+import threading
+from uuid import uuid4
+
+
+LOG_PATH = Path(__file__).resolve().parent.parent / "output" / "requests.log"
+_LOCK = threading.Lock()
+
+
+def write_event(event: str, **fields) -> None:
+    record = {"time": datetime.now(timezone.utc).isoformat(), "event": event, **fields}
+    try:
+        with _LOCK:
+            LOG_PATH.parent.mkdir(parents=True, exist_ok=True)
+            with LOG_PATH.open("a", encoding="utf-8") as stream:
+                stream.write(json.dumps(record, ensure_ascii=False) + "\n")
+    except OSError:
+        logging.getLogger(__name__).exception("Cannot write request log")
+
+
+def decode_body(body: bytes):
+    text = body.decode("utf-8", errors="replace")
+    try:
+        return json.loads(text)
+    except ValueError:
+        return text
+
+
+class RequestLogMiddleware:
+    def __init__(self, app):
+        self.app = app
+
+    async def __call__(self, scope, receive, send):
+        if scope["type"] != "http":
+            return await self.app(scope, receive, send)
+        state = scope.setdefault("state", {})
+        request_id = state["audit_id"] = uuid4().hex
+        headers = dict(scope.get("headers", []))
+        multipart = b"multipart/form-data" in headers.get(b"content-type", b"").lower()
+        request_body = bytearray()
+        response_body = bytearray()
+        status = 500
+        response_json = False
+        request_logged = False
+
+        def log_request():
+            nonlocal request_logged
+            if not request_logged:
+                write_event("request", request_id=request_id, method=scope["method"],
+                            path=scope["path"], query=scope.get("query_string", b"").decode("utf-8", errors="replace"),
+                            body=state.get("audit_form", {"format": "multipart"}) if multipart else decode_body(bytes(request_body)))
+                request_logged = True
+
+        buffered = []
+        if not multipart:
+            while True:
+                message = await receive()
+                buffered.append(message)
+                if message["type"] != "http.request":
+                    break
+                request_body.extend(message.get("body", b""))
+                if not message.get("more_body", False):
+                    break
+            log_request()
+        pending = iter(buffered)
+
+        async def read():
+            message = next(pending, None)
+            return message if message is not None else await receive()
+
+        async def write(message):
+            nonlocal status, response_json
+            if message["type"] == "http.response.start":
+                log_request()
+                status = message["status"]
+                response_json = b"application/json" in dict(message.get("headers", [])).get(b"content-type", b"")
+            elif message["type"] == "http.response.body" and response_json:
+                response_body.extend(message.get("body", b""))
+            await send(message)
+
+        try:
+            await self.app(scope, read, write)
+        except Exception as exc:
+            log_request()
+            write_event("response", request_id=request_id, status=500,
+                        body="Internal Server Error", error=str(exc))
+            raise
+        else:
+            write_event("response", request_id=request_id, status=status,
+                        body=decode_body(bytes(response_body)) if response_json else "[non-JSON response omitted]")