diff --git a/backend/Dockerfile b/backend/Dockerfile index 2951ae53..bd88185c 100644 --- a/backend/Dockerfile +++ b/backend/Dockerfile @@ -8,6 +8,9 @@ WORKDIR /app RUN apt-get update && apt-get install -y --no-install-recommends \ build-essential \ curl \ + tesseract-ocr \ + tesseract-ocr-chi-sim \ + tesseract-ocr-eng \ && rm -rf /var/lib/apt/lists/* COPY requirements.txt ./ diff --git a/backend/app/api/reviews.py b/backend/app/api/reviews.py index a9448f08..d54a3524 100644 --- a/backend/app/api/reviews.py +++ b/backend/app/api/reviews.py @@ -52,6 +52,7 @@ async def create_review( normalized_text = text.strip() if text else None file_name = file.filename if file else None file_text: Optional[str] = None + extraction: Optional[dict] = None if file is not None: file_bytes = await file.read() @@ -78,6 +79,8 @@ async def create_review( contract_type=contract_type, text=combined_text, file_name=file_name, + extracted_text=file_text, + extraction_metadata=extraction["metadata"] if extraction else None, ) audit_log_service.record_event( tenant_id=current_user.get("tenant_id"), diff --git a/backend/app/core/config.py b/backend/app/core/config.py index f8b01225..b542a606 100644 --- a/backend/app/core/config.py +++ b/backend/app/core/config.py @@ -7,6 +7,16 @@ class Settings(BaseSettings): app_name: str = "AI Contract Platform" app_env: str = "development" api_v1_prefix: str = "/api/v1" + review_pdf_ocr_enabled: bool = True + review_pdf_ocr_languages: str = "chi_sim+eng" + review_pdf_ocr_max_pages: int = 6 + review_pdf_ocr_render_scale: float = 2.0 + review_pdf_ocr_block_min_chars: int = 40 + review_pdf_ocr_warn_min_chars: int = 120 + review_pdf_ocr_block_min_confidence: float = 45.0 + review_pdf_ocr_warn_min_confidence: float = 70.0 + review_pdf_ocr_warn_max_noise_ratio: float = 0.18 + review_pdf_ocr_block_max_noise_ratio: float = 0.35 shared_ai_platform_base_url: str = "" shared_ai_platform_api_prefix: str = "/api/v0.1" shared_ai_platform_app_id: str = "ai-contract-platform" diff --git a/backend/app/services/review_service.py b/backend/app/services/review_service.py index 394695af..2d75f66e 100644 --- a/backend/app/services/review_service.py +++ b/backend/app/services/review_service.py @@ -10,6 +10,15 @@ import zipfile from xml.etree import ElementTree import pdfplumber +try: + import pypdfium2 as pdfium +except ImportError: # pragma: no cover - optional local dependency path + pdfium = None +try: + import pytesseract + from pytesseract import Output as TesseractOutput +except ImportError: # pragma: no cover - optional local dependency path + pytesseract = None from fastapi import HTTPException, status from reportlab.lib.pagesizes import A4 from reportlab.pdfbase import pdfmetrics @@ -18,6 +27,7 @@ from reportlab.pdfgen import canvas from sqlalchemy.orm import Session +from app.core.config import settings from app.models.review_result import ReviewResult from app.models.review_task import ReviewTask from app.repositories.review_result_repository import ReviewResultRepository @@ -120,6 +130,13 @@ REPORT_PDF_MARKERS = ( "风险项:", "原始文本摘要:", ) +OCR_GATE_PASS = "PASS" +OCR_GATE_WARN = "WARN" +OCR_GATE_BLOCK = "BLOCK" +PDF_SOURCE_KIND_TEXT = "text_pdf" +PDF_SOURCE_KIND_SCANNED = "scanned_pdf" +PDF_SOURCE_KIND_MIXED = "mixed_pdf" +PDF_SOURCE_KIND_UNKNOWN = "unknown_pdf" SUPPORTED_TEXT_SUFFIXES = (".txt", ".md", ".csv", ".json") DOCX_SUFFIX = ".docx" PDF_SUFFIX = ".pdf" @@ -153,6 +170,8 @@ class ReviewService: contract_type: Optional[str], text: Optional[str], file_name: Optional[str], + extracted_text: Optional[str] = None, + extraction_metadata: Optional[Dict[str, Any]] = None, ) -> dict: review_id = f"review-{uuid4().hex[:12]}" title = file_name or "文本输入合同" @@ -168,7 +187,9 @@ class ReviewService: status="pending", source_type=source_type, input_text=text, + extracted_text=extracted_text, ) + review_task._extraction_metadata = extraction_metadata or {} review_task = self.repository.create(review_task) review_task = self._finalize_review_task(review_task) @@ -202,17 +223,50 @@ class ReviewService: def extract_review_upload_text(self, file_bytes: bytes, file_name: Optional[str]) -> Dict[str, Optional[str]]: text = self.extract_text_from_file(file_bytes=file_bytes, file_name=file_name) normalized_name = (file_name or "").lower() + extraction_metadata: Dict[str, Any] = { + "document_kind": None, + "extraction_mode": "text", + "ocr_gate_status": None, + } if normalized_name.endswith(PDF_SUFFIX): + pdf_source = self._classify_pdf_source(file_bytes) + extraction_metadata["document_kind"] = pdf_source["kind"] if self._looks_like_platform_review_report_file_name(normalized_name) or (text and self._looks_like_platform_review_report(text)): return { "text": None, "failure_message": "检测到疑似平台审查报告 PDF,请上传合同原文 PDF / DOCX / TXT,或直接粘贴合同正文后再审查。", + "metadata": { + **extraction_metadata, + "document_kind": "platform_review_report_pdf", + "extraction_mode": "reject_report_pdf", + "ocr_gate_status": OCR_GATE_BLOCK, + }, } + if text: + return { + "text": text, + "failure_message": "无法从上传文件中提取可审查文本,请改为可复制文本 PDF / DOCX / TXT,或直接粘贴合同正文。", + "metadata": { + **extraction_metadata, + "extraction_mode": "text", + "ocr_gate_status": OCR_GATE_PASS, + }, + } + ocr_result = self._extract_text_from_pdf_with_ocr(file_bytes, pdf_source) + return { + "text": ocr_result["text"], + "failure_message": ocr_result["failure_message"], + "metadata": { + **extraction_metadata, + **ocr_result["metadata"], + }, + } return { "text": text, "failure_message": "无法从上传文件中提取可审查文本,请改为可复制文本 PDF / DOCX / TXT,或直接粘贴合同正文。", + "metadata": extraction_metadata, } def validate_review_upload(self, file_name: Optional[str], mime_type: Optional[str], file_bytes: bytes) -> None: @@ -536,6 +590,7 @@ class ReviewService: result = self._get_legal_review_execution_gateway().run_review(review) risk_items = result["risk_items"] risk_summary = result["risk_summary"] + confidence_summary = self._merge_confidence_summary(result.get("confidence_summary"), getattr(review, "_extraction_metadata", None)) review.risk_level = risk_summary["overall"] review.status = "completed" review.processing_time_ms = 1000 + len(risk_items) * 200 @@ -552,13 +607,21 @@ class ReviewService: risk_summary_json=json.dumps(risk_summary, ensure_ascii=False), review_opinion=result["review_opinion"], recommendation=result.get("recommendation") or DEFAULT_RECOMMENDATION, - confidence_summary=json.dumps(result["confidence_summary"], ensure_ascii=False) if result.get("confidence_summary") else None, + confidence_summary=json.dumps(confidence_summary, ensure_ascii=False) if confidence_summary else None, report_file_id=None, ) self.review_result_repository.create(review_result) return self.repository.update(review) + def _merge_confidence_summary(self, provider_confidence: Optional[Dict[str, Any]], extraction_metadata: Optional[Dict[str, Any]]) -> Optional[Dict[str, Any]]: + if not provider_confidence and not extraction_metadata: + return None + merged = dict(provider_confidence or {}) + if extraction_metadata: + merged["document_analysis"] = extraction_metadata + return merged + def _extract_text_from_docx(self, file_bytes: bytes) -> Optional[str]: try: with zipfile.ZipFile(io.BytesIO(file_bytes)) as archive: @@ -603,6 +666,192 @@ class ReviewService: return cleaned return None + def _classify_pdf_source(self, file_bytes: bytes) -> Dict[str, Any]: + try: + with pdfplumber.open(io.BytesIO(file_bytes)) as pdf: + page_count = len(pdf.pages) + text_pages = 0 + image_pages = 0 + for page in pdf.pages[: settings.review_pdf_ocr_max_pages]: + page_text = page.extract_text() + if page_text and page_text.strip(): + text_pages += 1 + if getattr(page, "images", None) and len(page.images) > 0: + image_pages += 1 + except Exception: + return { + "kind": PDF_SOURCE_KIND_UNKNOWN, + "page_count": 0, + "text_pages": 0, + "image_pages": 0, + } + + if text_pages > 0 and image_pages > 0: + kind = PDF_SOURCE_KIND_MIXED + elif text_pages > 0: + kind = PDF_SOURCE_KIND_TEXT + elif image_pages > 0: + kind = PDF_SOURCE_KIND_SCANNED + else: + kind = PDF_SOURCE_KIND_UNKNOWN + + return { + "kind": kind, + "page_count": page_count, + "text_pages": text_pages, + "image_pages": image_pages, + } + + def _extract_text_from_pdf_with_ocr(self, file_bytes: bytes, pdf_source: Dict[str, Any]) -> Dict[str, Any]: + if not settings.review_pdf_ocr_enabled: + return { + "text": None, + "failure_message": "检测到扫描件/图片型 PDF,但当前 OCR 能力未启用,请上传可复制文本 PDF / DOCX / TXT,或直接粘贴合同正文。", + "metadata": { + "document_kind": pdf_source["kind"], + "extraction_mode": "ocr_disabled", + "ocr_gate_status": OCR_GATE_BLOCK, + "ocr_quality": None, + }, + } + + if pytesseract is None or pdfium is None: + return { + "text": None, + "failure_message": "检测到扫描件/图片型 PDF,但当前环境未加载 OCR 运行时,请上传可复制文本 PDF / DOCX / TXT,或直接粘贴合同正文。", + "metadata": { + "document_kind": pdf_source["kind"], + "extraction_mode": "ocr_runtime_missing", + "ocr_gate_status": OCR_GATE_BLOCK, + "ocr_quality": None, + }, + } + + try: + document = pdfium.PdfDocument(io.BytesIO(file_bytes)) + except Exception: + return { + "text": None, + "failure_message": "扫描件/图片型 PDF 解析失败,请上传更清晰的扫描件、可复制文本 PDF / DOCX / TXT,或直接粘贴合同正文。", + "metadata": { + "document_kind": pdf_source["kind"], + "extraction_mode": "ocr_parse_failed", + "ocr_gate_status": OCR_GATE_BLOCK, + "ocr_quality": None, + }, + } + + page_texts: List[List[str]] = [] + confidences: List[float] = [] + pages_to_ocr = min(len(document), settings.review_pdf_ocr_max_pages) + for page_index in range(pages_to_ocr): + page = document[page_index] + bitmap = page.render(scale=settings.review_pdf_ocr_render_scale) + pil_image = bitmap.to_pil() + page_text = pytesseract.image_to_string( + pil_image, + lang=settings.review_pdf_ocr_languages, + config="--psm 6", + ) + data = pytesseract.image_to_data( + pil_image, + lang=settings.review_pdf_ocr_languages, + config="--psm 6", + output_type=TesseractOutput.DICT, + ) + lines = [line.strip() for line in page_text.splitlines() if line.strip()] + if lines: + page_texts.append(lines) + confidences.extend(self._extract_ocr_confidences(data)) + pil_image.close() + bitmap.close() + page.close() + document.close() + + stripped_pages = self._strip_repeated_page_decorations(page_texts) + text = "\n".join("\n".join(lines) for lines in stripped_pages if lines) + cleaned = self._clean_pdf_extracted_text(text) + quality = self._evaluate_ocr_quality(cleaned or "", confidences, pdf_source) + + if quality["status"] == OCR_GATE_BLOCK: + return { + "text": None, + "failure_message": quality["failure_message"], + "metadata": { + "document_kind": pdf_source["kind"], + "extraction_mode": "ocr", + "ocr_gate_status": OCR_GATE_BLOCK, + "ocr_quality": quality, + }, + } + + return { + "text": cleaned, + "failure_message": quality["failure_message"], + "metadata": { + "document_kind": pdf_source["kind"], + "extraction_mode": "ocr", + "ocr_gate_status": quality["status"], + "ocr_quality": quality, + }, + } + + def _extract_ocr_confidences(self, data: Dict[str, List[Any]]) -> List[float]: + values: List[float] = [] + for text, confidence in zip(data.get("text", []), data.get("conf", [])): + if not str(text).strip(): + continue + try: + score = float(confidence) + except (TypeError, ValueError): + continue + if score >= 0: + values.append(score) + return values + + def _evaluate_ocr_quality(self, text: str, confidences: List[float], pdf_source: Dict[str, Any]) -> Dict[str, Any]: + lines = [line.strip() for line in text.splitlines() if line.strip()] + char_count = len(re.sub(r"\s+", "", text)) + token_count = len(re.findall(r"[A-Za-z\u4e00-\u9fff]{2,}", text)) + printable_chars = sum(1 for char in text if char.isprintable() or char in "\r\n\t") + printable_ratio = printable_chars / max(len(text), 1) if text else 0.0 + page_noise_hits = sum(1 for line in lines if PDF_PAGE_NOISE_PATTERN.fullmatch(line) or self._looks_like_page_decoration(line)) + noise_ratio = page_noise_hits / max(len(lines), 1) if lines else 0.0 + mean_confidence = round(sum(confidences) / len(confidences), 2) if confidences else None + + detail = { + "status": OCR_GATE_PASS, + "char_count": char_count, + "token_count": token_count, + "line_count": len(lines), + "printable_ratio": round(printable_ratio, 3), + "noise_ratio": round(noise_ratio, 3), + "mean_confidence": mean_confidence, + "page_count": pdf_source.get("page_count"), + "text_pages": pdf_source.get("text_pages"), + "image_pages": pdf_source.get("image_pages"), + "failure_message": None, + } + + if char_count < settings.review_pdf_ocr_block_min_chars or token_count < 2 or printable_ratio < 0.45: + detail["status"] = OCR_GATE_BLOCK + elif mean_confidence is not None and mean_confidence < settings.review_pdf_ocr_block_min_confidence: + detail["status"] = OCR_GATE_BLOCK + elif noise_ratio > settings.review_pdf_ocr_block_max_noise_ratio: + detail["status"] = OCR_GATE_BLOCK + elif char_count < settings.review_pdf_ocr_warn_min_chars or printable_ratio < 0.72: + detail["status"] = OCR_GATE_WARN + elif mean_confidence is not None and mean_confidence < settings.review_pdf_ocr_warn_min_confidence: + detail["status"] = OCR_GATE_WARN + elif noise_ratio > settings.review_pdf_ocr_warn_max_noise_ratio: + detail["status"] = OCR_GATE_WARN + + if detail["status"] == OCR_GATE_BLOCK: + detail["failure_message"] = "检测到扫描件/图片型 PDF,但 OCR 识别质量不足,请上传更清晰的扫描件、可复制文本 PDF / DOCX / TXT,或直接粘贴合同正文。" + elif detail["status"] == OCR_GATE_WARN: + detail["failure_message"] = "已通过 OCR 提取文本,但识别质量一般,建议人工复核原文后再使用审查结果。" + return detail + def _extract_with_pdfplumber(self, file_bytes: bytes, max_retries: int = 2) -> Optional[str]: """Extract text from PDF using pdfplumber library with retry mechanism. diff --git a/backend/requirements.txt b/backend/requirements.txt index 6c21d558..58e12eaf 100644 --- a/backend/requirements.txt +++ b/backend/requirements.txt @@ -16,4 +16,5 @@ email-validator==2.2.0 reportlab==4.4.1 pdfplumber==0.11.4 python-docx==1.1.2 +pytesseract==0.3.13 pytest==8.3.5 diff --git a/backend/tests/test_review_service.py b/backend/tests/test_review_service.py index 0963be39..a37724a7 100644 --- a/backend/tests/test_review_service.py +++ b/backend/tests/test_review_service.py @@ -268,6 +268,57 @@ class ReviewServiceTestCase(unittest.TestCase): self.assertIsNone(result["text"]) self.assertIn("疑似平台审查报告", result["failure_message"]) + self.assertEqual(result["metadata"]["ocr_gate_status"], "BLOCK") + + def test_extract_review_upload_text_uses_ocr_for_scanned_pdf(self): + self.service._extract_text_from_file = lambda file_bytes, file_name: None + self.service._classify_pdf_source = lambda file_bytes: { + "kind": "scanned_pdf", + "page_count": 1, + "text_pages": 0, + "image_pages": 1, + } + self.service._extract_text_from_pdf_with_ocr = lambda file_bytes, pdf_source: { + "text": "扫描合同正文\n付款安排", + "failure_message": "已通过 OCR 提取文本,但识别质量一般,建议人工复核原文后再使用审查结果。", + "metadata": { + "document_kind": "scanned_pdf", + "extraction_mode": "ocr", + "ocr_gate_status": "WARN", + "ocr_quality": {"status": "WARN", "char_count": 12}, + }, + } + + result = self.service.extract_review_upload_text(b"%PDF-1.4", "scan.pdf") + + self.assertEqual(result["text"], "扫描合同正文\n付款安排") + self.assertEqual(result["metadata"]["extraction_mode"], "ocr") + self.assertEqual(result["metadata"]["ocr_gate_status"], "WARN") + + def test_extract_review_upload_text_blocks_low_quality_ocr(self): + self.service._extract_text_from_file = lambda file_bytes, file_name: None + self.service._classify_pdf_source = lambda file_bytes: { + "kind": "scanned_pdf", + "page_count": 1, + "text_pages": 0, + "image_pages": 1, + } + self.service._extract_text_from_pdf_with_ocr = lambda file_bytes, pdf_source: { + "text": None, + "failure_message": "检测到扫描件/图片型 PDF,但 OCR 识别质量不足,请上传更清晰的扫描件、可复制文本 PDF / DOCX / TXT,或直接粘贴合同正文。", + "metadata": { + "document_kind": "scanned_pdf", + "extraction_mode": "ocr", + "ocr_gate_status": "BLOCK", + "ocr_quality": {"status": "BLOCK", "char_count": 8}, + }, + } + + result = self.service.extract_review_upload_text(b"%PDF-1.4", "scan-low.pdf") + + self.assertIsNone(result["text"]) + self.assertEqual(result["metadata"]["ocr_gate_status"], "BLOCK") + self.assertIn("OCR 识别质量不足", result["failure_message"]) def test_validate_review_upload_accepts_supported_suffix(self): self.service.validate_review_upload("contract.pdf", "application/pdf", b"%PDF-1.7") @@ -306,6 +357,55 @@ class ReviewServiceTestCase(unittest.TestCase): self.assertEqual(context.exception.status_code, 400) self.assertEqual(context.exception.detail, "仅支持 PDF / DOCX / TXT 文件") + def test_evaluate_ocr_quality_returns_pass_for_clean_text(self): + quality = self.service._evaluate_ocr_quality( + "合同正文第一条 付款安排与开票条件明确,甲方应在验收并收到合规发票后十五个工作日内付款。\n合同正文第二条 保密义务、期限与返还责任明确,双方应在合同终止后三年内持续履行保密义务。\n合同正文第三条 违约责任、责任上限与通知流程明确,任何一方违约时应先书面通知并给予整改期限。", + [88.0, 92.0, 84.0], + {"kind": "scanned_pdf", "page_count": 1, "text_pages": 0, "image_pages": 1}, + ) + + self.assertEqual(quality["status"], "PASS") + + def test_evaluate_ocr_quality_returns_warn_for_medium_quality_text(self): + quality = self.service._evaluate_ocr_quality( + "合同正文第一条 付款安排待补充。\n合同正文第二条 保密义务待补充。\n合同正文第三条 发票条件待补充。", + [62.0, 64.0], + {"kind": "scanned_pdf", "page_count": 1, "text_pages": 0, "image_pages": 1}, + ) + + self.assertEqual(quality["status"], "WARN") + + def test_evaluate_ocr_quality_returns_block_for_noise(self): + quality = self.service._evaluate_ocr_quality( + "Page 1 / 2\n1 / 2\nPDF\n", + [20.0, 15.0], + {"kind": "scanned_pdf", "page_count": 1, "text_pages": 0, "image_pages": 1}, + ) + + self.assertEqual(quality["status"], "BLOCK") + + def test_finalize_review_task_merges_extraction_metadata_into_confidence_summary(self): + created = self.service.create_review_task( + tenant_id=self.current_user["tenant_id"], + user_id=self.current_user["id"], + review_as="party_a", + contract_type="service", + text="付款\n保密", + file_name="scan.pdf", + extracted_text="付款\n保密", + extraction_metadata={ + "document_kind": "scanned_pdf", + "extraction_mode": "ocr", + "ocr_gate_status": "WARN", + "ocr_quality": {"status": "WARN", "char_count": 12}, + }, + ) + + result = self.service.get_review_result(self.current_user, created["id"]) + assert result is not None + self.assertEqual(result["confidence_summary"]["document_analysis"]["document_kind"], "scanned_pdf") + self.assertEqual(result["confidence_summary"]["document_analysis"]["ocr_gate_status"], "WARN") + def test_merge_review_text_prefers_both_sources(self): merged = self.service.merge_review_text("手工补充", "自动续约") diff --git a/backend/tests/test_reviews_api.py b/backend/tests/test_reviews_api.py index ae381e1a..bbd29949 100644 --- a/backend/tests/test_reviews_api.py +++ b/backend/tests/test_reviews_api.py @@ -27,17 +27,33 @@ class StubReviewService: return file_bytes.decode("utf-8", errors="ignore") def extract_review_upload_text(self, file_bytes, file_name): + if file_name.endswith(".scan.pdf"): + return { + "text": "扫描合同正文\n付款安排", + "failure_message": "已通过 OCR 提取文本,但识别质量一般,建议人工复核原文后再使用审查结果。", + "metadata": { + "document_kind": "scanned_pdf", + "extraction_mode": "ocr", + "ocr_gate_status": "WARN", + "ocr_quality": {"status": "WARN", "char_count": 12}, + }, + } text = self.extract_text_from_file(file_bytes, file_name) return { "text": text, "failure_message": "无法从上传文件中提取可审查文本,请改为可复制文本 PDF / DOCX / TXT,或直接粘贴合同正文。", + "metadata": { + "document_kind": "text_pdf", + "extraction_mode": "text", + "ocr_gate_status": "PASS", + }, } def validate_review_upload(self, file_name, mime_type, file_bytes): if file_name.endswith(".png") or file_name.endswith(".exe") or file_bytes.startswith(b"\x89PNG") or file_bytes.startswith(b"MZ"): raise HTTPException(status_code=400, detail="仅支持 PDF / DOCX / TXT 文件") - def create_review_task(self, tenant_id, user_id, review_as, contract_type, text, file_name): + def create_review_task(self, tenant_id, user_id, review_as, contract_type, text, file_name, extracted_text=None, extraction_metadata=None): return { "id": "review-123", "status": "completed", @@ -47,6 +63,8 @@ class StubReviewService: "file_name": file_name, "tenant_id": tenant_id, "user_id": user_id, + "extracted_text": extracted_text, + "extraction_metadata": extraction_metadata, } def list_reviews(self, current_user): @@ -297,6 +315,18 @@ class ReviewsApiTestCase(unittest.TestCase): self.assertEqual(body["code"], 40002) self.assertIn("疑似平台审查报告", body["message"]) + def test_create_review_accepts_scanned_pdf_after_ocr(self): + response = self.client.post( + "/api/v1/reviews", + data={"review_as": "party_a"}, + files={"file": ("contract.scan.pdf", b"%PDF", "application/pdf")}, + ) + + self.assertEqual(response.status_code, 200) + body = response.json() + self.assertEqual(body["code"], 0) + self.assertEqual(body["data"]["merged_text"], "扫描合同正文\n付款安排") + def test_list_reviews_uses_current_user_scope(self): response = self.client.get("/api/v1/reviews") diff --git a/backend/tools/pdf_acceptance_gate.py b/backend/tools/pdf_acceptance_gate.py new file mode 100644 index 00000000..69e6d6e0 --- /dev/null +++ b/backend/tools/pdf_acceptance_gate.py @@ -0,0 +1,443 @@ +import argparse +import io +import json +import sys +import time +from dataclasses import dataclass +from datetime import datetime, timezone +from pathlib import Path +from typing import Callable, Dict, List, Optional + +import httpx +from PIL import Image, ImageDraw +from reportlab.lib.utils import ImageReader +from reportlab.pdfbase import pdfmetrics +from reportlab.pdfbase.cidfonts import UnicodeCIDFont +from reportlab.pdfgen import canvas + + +pdfmetrics.registerFont(UnicodeCIDFont("STSong-Light")) + + +@dataclass +class PdfCase: + case_id: str + category: str + label: str + expectation: str + file_name: str + build_bytes: Optional[Callable[[], bytes]] = None + file_path: Optional[str] = None + require_preview: Optional[List[str]] = None + forbid_preview: Optional[List[str]] = None + require_clause_absent: Optional[List[str]] = None + require_message_contains: Optional[str] = None + allow_ocr_statuses: Optional[List[str]] = None + notes: Optional[str] = None + + +def make_text_pdf(lines: List[str], font_name: str = "Helvetica", font_size: int = 12) -> bytes: + buffer = io.BytesIO() + pdf = canvas.Canvas(buffer) + pdf.setFont(font_name, font_size) + y = 780 + for line in lines: + pdf.drawString(72, y, line) + y -= 20 + if y < 72: + pdf.showPage() + pdf.setFont(font_name, font_size) + y = 780 + pdf.save() + return buffer.getvalue() + + +def make_report_like_pdf() -> bytes: + return make_text_pdf( + [ + "审查任务:文本输入合同", + "任务编号:review-123", + "风险汇总:", + "审查意见:建议复核", + "原始文本摘要:合同文本", + ], + font_name="STSong-Light", + ) + + +def make_clean_short_pdf() -> bytes: + return make_text_pdf( + [ + "Service Agreement", + "Party A shall pay within 30 days after acceptance.", + "If Party B delays delivery, liquidated damages apply.", + ] + ) + + +def make_clean_long_pdf() -> bytes: + return make_text_pdf( + [ + "Technology Development Agreement", + "1. Scope: Party B shall develop and deploy the requested system and deliver source code.", + "2. Payment: Party A shall pay milestone one within 15 days after written acceptance.", + "3. The balance shall be paid within 45 days after stable production operation.", + "4. Confidentiality obligations shall survive termination for three years.", + "5. Liability shall be capped except for wilful misconduct and confidentiality breaches.", + ] + ) + + +def make_header_footer_pdf() -> bytes: + buffer = io.BytesIO() + pdf = canvas.Canvas(buffer) + pdf.setFont("Helvetica", 12) + pdf.drawString(72, 780, "Page 1 / 2") + pdf.drawString(72, 760, "PDF extraction boundary sample") + pdf.drawString(72, 720, "Procurement Agreement") + pdf.drawString(72, 700, "Party A shall pay within 15 business days after invoice receipt and acceptance.") + pdf.showPage() + pdf.setFont("Helvetica", 12) + pdf.drawString(72, 780, "Page 2 / 2") + pdf.drawString(72, 760, "PDF extraction boundary sample") + pdf.drawString(72, 720, "Both parties shall define confidentiality obligations clearly.") + pdf.save() + return buffer.getvalue() + + +def make_broken_minimal_pdf() -> bytes: + return b"%PDF-1.4\n1 0 obj\n<<>>\nendobj\ntrailer\n<<>>\n%%EOF" + + +def _make_image_contract(low_res: bool) -> bytes: + size = (420, 240) if low_res else (1200, 800) + image = Image.new("RGB", size, "white") + draw = ImageDraw.Draw(image) + lines = [ + "SCANNED CONTRACT SAMPLE", + "Party A shall pay within 15 days after acceptance.", + "Confidential information shall remain protected.", + ] + y = 20 + for line in lines: + draw.text((20, y), line, fill="black") + y += 48 if not low_res else 28 + + pdf_buffer = io.BytesIO() + pdf = canvas.Canvas(pdf_buffer) + pdf.drawImage(ImageReader(image), 36, 400, width=520, height=300) + pdf.save() + return pdf_buffer.getvalue() + + +def make_image_only_pdf() -> bytes: + return _make_image_contract(low_res=False) + + +def make_low_res_image_only_pdf() -> bytes: + return _make_image_contract(low_res=True) + + +def build_cases(sample_contract_path: Optional[str]) -> List[PdfCase]: + cases: List[PdfCase] = [ + PdfCase( + case_id="A-001", + category="A", + label="干净短文本 PDF", + expectation="must_succeed", + file_name="gate-clean-short.pdf", + build_bytes=make_clean_short_pdf, + require_preview=["Service Agreement", "Party A shall pay within 30 days after acceptance."], + ), + PdfCase( + case_id="A-002", + category="A", + label="干净长文本 PDF", + expectation="must_succeed", + file_name="gate-clean-long.pdf", + build_bytes=make_clean_long_pdf, + require_preview=["Technology Development Agreement", "Party A shall pay milestone one within 15 days"], + ), + PdfCase( + case_id="A-003", + category="A", + label="页眉页脚边界 PDF", + expectation="must_succeed", + file_name="gate-header-footer.pdf", + build_bytes=make_header_footer_pdf, + require_preview=["Procurement Agreement", "Party A shall pay within 15 business days"], + forbid_preview=["Page 1 / 2", "Page 2 / 2", "PDF extraction boundary sample"], + require_clause_absent=["Page 1 / 2", "Page 2 / 2", "PDF extraction boundary sample"], + ), + PdfCase( + case_id="B-001", + category="B", + label="最小坏样本 PDF", + expectation="must_fail", + file_name="gate-broken-minimal.pdf", + build_bytes=make_broken_minimal_pdf, + require_message_contains="无法从上传文件中提取可审查文本", + ), + PdfCase( + case_id="B-002", + category="B", + label="平台审查报告 PDF", + expectation="must_fail", + file_name="review-report-gate.pdf", + build_bytes=make_report_like_pdf, + require_message_contains="疑似平台审查报告 PDF", + ), + PdfCase( + case_id="C-001", + category="C", + label="图片型 PDF(OCR第一批)", + expectation="ocr_supported", + file_name="gate-image-only.pdf", + build_bytes=make_image_only_pdf, + allow_ocr_statuses=["PASS", "WARN"], + notes="OCR 第一批实现后,这类清晰图片型 PDF 应能进入正式审查。", + ), + PdfCase( + case_id="C-002", + category="C", + label="低清晰度图片型 PDF(OCR第一批)", + expectation="ocr_quality_gate", + file_name="gate-image-lowres.pdf", + build_bytes=make_low_res_image_only_pdf, + allow_ocr_statuses=["WARN", "BLOCK"], + notes="OCR 第一批实现后,这类样本允许 WARN 或 BLOCK,但不能伪装成高质量 PASS。", + ), + ] + + if sample_contract_path: + sample_path = Path(sample_contract_path) + if sample_path.exists(): + cases.append( + PdfCase( + case_id="B-003", + category="B", + label="固定回归样本 sample-contract.pdf", + expectation="must_fail", + file_name=sample_path.name, + file_path=str(sample_path), + require_message_contains="无法从上传文件中提取可审查文本", + notes="按当前验收分级清单,这个样本被冻结为必须失败。", + ) + ) + return cases + + +def load_case_bytes(case: PdfCase) -> bytes: + if case.build_bytes is not None: + return case.build_bytes() + if case.file_path: + return Path(case.file_path).read_bytes() + raise RuntimeError(f"Case {case.case_id} has no content source") + + +def build_md(summary: Dict) -> str: + lines = [ + "# PDF Acceptance Gate", + "", + f"- Status: {summary['status']}", + f"- Generated at: {summary['generated_at']}", + f"- API Base: `{summary['api_base']}`", + f"- Artifact dir: `{summary['artifact_dir']}`", + "", + "## Category Totals", + "", + "| Category | Total | Passed | Failed |", + "|---|---:|---:|---:|", + ] + for category in ["A", "B", "C"]: + bucket = summary["category_totals"].get(category, {"total": 0, "passed": 0, "failed": 0}) + lines.append(f"| {category} | {bucket['total']} | {bucket['passed']} | {bucket['failed']} |") + + lines.extend( + [ + "", + "## Case Results", + "", + "| Case | Category | Expectation | Status | Detail |", + "|---|---|---|---|---|", + ] + ) + for case in summary["cases"]: + detail = case.get("detail", "").replace("|", "/") + lines.append(f"| {case['case_id']} | {case['category']} | {case['expectation']} | {case['status']} | {detail} |") + return "\n".join(lines) + + +def register_and_login(client: httpx.Client, api_base: str, password: str) -> Dict[str, str]: + seed = int(time.time() * 1000) + username = f"pdf.gate.{seed}@example.com" + display_name = f"PDF Gate {seed}" + register = client.post( + f"{api_base}/auth/register", + json={"username": username, "password": password, "display_name": display_name}, + ) + register.raise_for_status() + login = client.post( + f"{api_base}/auth/login", + json={"account": username, "password": password}, + ) + login.raise_for_status() + login_data = login.json()["data"] + return {"username": username, "token": login_data["access_token"]} + + +def validate_success_case(client: httpx.Client, api_base: str, headers: Dict[str, str], case: PdfCase, submit_json: Dict) -> Dict: + review_id = submit_json["data"]["id"] + detail = client.get(f"{api_base}/reviews/{review_id}", headers=headers) + detail.raise_for_status() + detail_json = detail.json()["data"] + result = client.get(f"{api_base}/reviews/{review_id}/result", headers=headers) + result.raise_for_status() + result_json = result.json()["data"] + download = client.get(f"{api_base}/reviews/{review_id}/report/download", headers=headers) + download.raise_for_status() + + preview_text = detail_json.get("source_text_preview") or "" + clause_text = "\n".join(item.get("content", "") for item in result_json.get("clauses", [])) + confidence_summary = result_json.get("confidence_summary") or {} + document_analysis = confidence_summary.get("document_analysis") or {} + + problems: List[str] = [] + if detail_json.get("status") != "completed": + problems.append(f"detail.status={detail_json.get('status')}") + if not review_id: + problems.append("missing review_id") + if not download.content: + problems.append("empty report download") + for needle in case.require_preview or []: + if needle not in preview_text: + problems.append(f"preview missing: {needle}") + for needle in case.forbid_preview or []: + if needle in preview_text: + problems.append(f"preview contains noise: {needle}") + for needle in case.require_clause_absent or []: + if needle in clause_text: + problems.append(f"clauses contain noise: {needle}") + if case.allow_ocr_statuses: + ocr_status = document_analysis.get("ocr_gate_status") + if ocr_status not in case.allow_ocr_statuses: + problems.append(f"ocr gate status {ocr_status} not in {case.allow_ocr_statuses}") + + return { + "review_id": review_id, + "detail": detail_json, + "result": result_json, + "download_bytes": len(download.content), + "document_analysis": document_analysis, + "problems": problems, + } + + +def validate_failure_case(case: PdfCase, submit_json: Dict) -> List[str]: + problems: List[str] = [] + if submit_json.get("code") != 40002: + problems.append(f"expected code=40002 got {submit_json.get('code')}") + message = submit_json.get("message", "") + if case.require_message_contains and case.require_message_contains not in message: + problems.append(f"message missing: {case.require_message_contains}") + return problems + + +def run_gate(api_base: str, artifact_dir: Path, sample_contract_path: Optional[str], password: str) -> Dict: + artifact_dir.mkdir(parents=True, exist_ok=True) + timestamp = datetime.now(timezone.utc).isoformat() + client = httpx.Client(timeout=90.0) + auth = register_and_login(client, api_base, password) + headers = {"Authorization": f"Bearer {auth['token']}"} + + cases_summary = [] + category_totals: Dict[str, Dict[str, int]] = {} + + for case in build_cases(sample_contract_path): + payload = load_case_bytes(case) + submit = client.post( + f"{api_base}/reviews", + headers=headers, + files={"file": (case.file_name, payload, "application/pdf")}, + data={"review_as": "party_a"}, + ) + submit.raise_for_status() + submit_json = submit.json() + + case_record = { + "case_id": case.case_id, + "category": case.category, + "label": case.label, + "expectation": case.expectation, + "submit": submit_json, + "status": "PASS", + "detail": "", + "notes": case.notes, + } + + problems: List[str] = [] + if case.expectation in {"must_succeed", "ocr_supported", "ocr_quality_gate"}: + success_detail = validate_success_case(client, api_base, headers, case, submit_json) + case_record["success_detail"] = success_detail + problems.extend(success_detail["problems"]) + else: + problems.extend(validate_failure_case(case, submit_json)) + + if problems: + case_record["status"] = "FAIL" + case_record["detail"] = "; ".join(problems) + else: + if case.expectation == "must_succeed": + case_record["detail"] = f"review_id={case_record['success_detail']['review_id']}" + else: + case_record["detail"] = submit_json.get("message", "") + + bucket = category_totals.setdefault(case.category, {"total": 0, "passed": 0, "failed": 0}) + bucket["total"] += 1 + if case_record["status"] == "PASS": + bucket["passed"] += 1 + else: + bucket["failed"] += 1 + + cases_summary.append(case_record) + + status = "PASS" if all(item["status"] == "PASS" for item in cases_summary) else "FAIL" + summary = { + "name": "pdf-acceptance-gate", + "status": status, + "generated_at": timestamp, + "api_base": api_base, + "artifact_dir": str(artifact_dir), + "registered_user": auth["username"], + "category_totals": category_totals, + "cases": cases_summary, + } + + (artifact_dir / "pdf-acceptance-gate-summary.json").write_text( + json.dumps(summary, ensure_ascii=False, indent=2), + encoding="utf-8", + ) + (artifact_dir / "pdf-acceptance-gate-summary.md").write_text(build_md(summary), encoding="utf-8") + print(json.dumps(summary, ensure_ascii=False, indent=2)) + return summary + + +def main() -> int: + parser = argparse.ArgumentParser(description="Run the PDF acceptance gate against the live API.") + parser.add_argument("--api-base", required=True) + parser.add_argument("--artifact-dir", required=True) + parser.add_argument("--sample-contract-path", default="") + parser.add_argument("--password", default="123456") + args = parser.parse_args() + + summary = run_gate( + api_base=args.api_base.rstrip("/"), + artifact_dir=Path(args.artifact_dir), + sample_contract_path=args.sample_contract_path or None, + password=args.password, + ) + return 0 if summary["status"] == "PASS" else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/project-docs/PDF样本验收分级清单-2026-05-04.md b/project-docs/PDF样本验收分级清单-2026-05-04.md new file mode 100644 index 00000000..76f378a5 --- /dev/null +++ b/project-docs/PDF样本验收分级清单-2026-05-04.md @@ -0,0 +1,237 @@ +# PDF样本验收分级清单(2026-05-04) + +## 1. 目的 + +本清单用于把 PDF 审查样本一次性分成三类,并固化后续验收口径: + +1. `必须成功` +2. `必须失败` +3. `OCR后再支持` + +以后任何 PDF 回归、发布门禁、验收说明,都以本清单为准,不再混用“样本坏了”“能力不支持”“功能失效”这三种不同问题。 + +## 2. 分级总规则 + +### A类:必须成功 + +定义: + +- 文件属于真实合同正文或正文节选 +- PDF 内存在可复制文本层 +- 当前系统不依赖 OCR 也应能完成抽取与审查 + +验收标准: + +1. 上传返回 `code=0` +2. 必须生成 `review_id` +3. 详情页 `status=completed` +4. `source_text_preview` 必须是可读正文,不得主要由页码、页眉、任务元信息组成 +5. `result.clauses` 前几条必须是正文条款,不得主要为噪声 +6. 报告下载必须成功 + +判定原则: + +- A类失败 = 真缺陷 + +### B类:必须失败 + +定义: + +- 文件不是有效合同正文输入 +- 或当前产品明确不应把它当作合同原文继续审查 + +验收标准: + +1. 不得生成伪成功审查结果 +2. 必须返回明确失败提示 +3. 不得生成“看起来 completed、实际上分析对象错误”的审查任务 + +判定原则: + +- B类如果“成功进入审查” = 真缺陷 +- B类如果被明确拦截 = 正常 + +### C类:OCR后再支持 + +定义: + +- 文件可能是有效合同 +- 但当前主要依赖扫描图像 / 图片层 / 无文本层 PDF +- 当前版本不承诺成功,待 OCR 能力上线后再纳入正式成功验收 + +当前版本验收标准: + +1. 可以失败 +2. 失败时应给出明确边界提示 +3. 不得伪成功 + +OCR上线后验收标准: + +1. 先经过“扫描件识别 / 图片型 PDF 识别” +2. 再进入 OCR +3. 再经过 OCR 质量门禁 +4. 质量门禁 `PASS` 才允许进入正式审查 + +判定原则: + +- OCR 上线前:C类失败不记为当前缺陷 +- OCR 上线后:C类是否转入 A类,由样本逐个升级确认 + +## 3. 当前样本分级冻结 + +### A类:必须成功 + +| 样本ID | 样本名称 | 当前结论 | 当前理由 | +|---|---|---|---| +| A-01 | `en_short.pdf` | 必须成功 | 干净短文本 PDF,已验证可稳定完成上传、抽取、审查、报告下载 | +| A-02 | `en_long.pdf` | 必须成功 | 干净长文本 PDF,已验证可稳定完成正式审查链路 | +| A-03 | `en_header_footer.pdf` | 必须成功 | 属于文本型 PDF,只是含页眉页脚边界;系统应通过清洗后成功审查 | + +### B类:必须失败 + +| 样本ID | 样本名称 | 当前结论 | 当前理由 | 正确系统表现 | +|---|---|---|---|---| +| B-01 | `sample-contract.pdf` | 必须失败 | 当前证据显示该文件未形成可抽取正文,且体积仅约 `612 bytes`,不适合作为成功审查样本 | 返回 `40002`,不生成 `review_id` | +| B-02 | `review-report-*.pdf` / 平台审查报告 PDF | 必须失败 | 它是平台输出物,不是合同原文;再次上传会造成“报告审报告”的伪成功 | 返回“疑似平台审查报告 PDF”提示,不生成审查任务 | + +### C类:OCR后再支持 + +| 样本ID | 样本名称 | 当前结论 | 当前阶段处理方式 | OCR上线后目标 | +|---|---|---|---|---| +| C-01 | 中文扫描合同 PDF(待补样本) | OCR后再支持 | 当前可失败,但不能伪成功 | OCR质量达标后转入必须成功 | +| C-02 | 图片型 PDF(待补样本) | OCR后再支持 | 当前可失败,但要提示边界 | OCR质量达标后转入必须成功 | +| C-03 | 拍照合成 PDF(待补样本) | OCR后再支持 | 当前可失败,但要提示边界 | OCR质量达标后按质量分级 | +| C-04 | 低清晰度扫描 PDF(待补样本) | OCR后再支持 | 当前可失败 | OCR后可能仍只允许 `WARN/BLOCK` | +| C-05 | 多页表格扫描 PDF(待补样本) | OCR后再支持 | 当前可失败 | OCR+结构清洗后再评估是否转入 A类 | + +## 4. 关于 `sample-contract.pdf` 的最终口径 + +当前冻结结论: + +- `sample-contract.pdf` **不是“必须成功样本”** +- 它当前被归类为 **B类:必须失败** + +原因: + +1. 现有证据没有证明它是“真实可读扫描合同” +2. 现有证据明确表明它无法形成可抽取正文 +3. 如果强行把它列为成功样本,会倒逼系统接受无正文或垃圾正文,反而降低结果可信度 + +只有在满足以下条件时,才允许它从 B类改判为 C类: + +1. 独立核实该文件确实是“真实合同扫描件” +2. 能确认文件内容肉眼可读、只是不含文本层 +3. OCR 能力正式上线 + +只有在满足以下条件时,才允许它从 C类升级为 A类: + +1. OCR 上线 +2. 该样本 OCR 结果稳定 +3. `source_text_preview` 与 `clauses` 可读且不被噪声污染 +4. 连续回归通过 + +## 5. OCR上线后的质量门禁预定义 + +为避免“扫描件 OCR 出垃圾文本也算成功”,OCR 上线后统一使用三档门禁: + +### PASS + +满足以下特征中的大部分: + +1. 抽取文本长度达到最小正文门槛 +2. 正文有效字符比例高 +3. 没有大段乱码 +4. 没有大段重复页眉页脚 +5. `source_text_preview` 明显是合同正文 +6. `clauses` 前几条不是页码/水印/模板噪声 + +结果: + +- 允许进入正式审查 + +### WARN + +文本可读,但质量一般: + +1. 可抽取正文,但局部有噪声 +2. 版式混乱但仍可人工复核 +3. 条款摘要可用性一般 + +结果: + +- 允许审查 +- 但必须标记“低置信度,建议人工复核” + +### BLOCK + +出现以下任一情况: + +1. 正文长度过短 +2. 噪声比例过高 +3. 主要内容为页码/页眉/水印/图片说明 +4. OCR 乱码严重 +5. 文档类型明显不是合同原文 + +结果: + +- 不允许进入正式审查 +- 返回明确失败原因 + +## 6. 当前发布与回归门禁口径 + +在 OCR 正式上线前,发布门禁按以下规则执行: + +1. A类必须全部通过 +2. B类必须全部被正确拦截 +3. C类允许失败,但必须失败得明确,不能伪成功 + +也就是说: + +- `sample-contract.pdf` 当前失败 **不是剩余产品缺陷** +- `review-report-*.pdf` 如果再被系统成功审查,才是缺陷 +- `en_header_footer.pdf` 如果再次混入页码页眉噪声,则是缺陷 + +## 7. 当前冻结结论 + +从今天开始,PDF 样本验收统一按以下口径执行: + +1. `en_short.pdf / en_long.pdf / en_header_footer.pdf`:`必须成功` +2. `sample-contract.pdf / 平台审查报告 PDF`:`必须失败` +3. `扫描件 / 图片型 / 无文本层合同 PDF`:`OCR后再支持` + +这份分级清单在 OCR 功能正式上线前,视为当前 PDF 能力的正式验收标准。 + +## 8. 自动化门禁落地 + +已新增自动化门禁: + +- `backend/tools/pdf_acceptance_gate.py` +- `scripts/run-pdf-acceptance-gate.ps1` + +当前脚本已固化以下样本规则: + +### A类自动化样本 + +- `A-001` 干净短文本 PDF +- `A-002` 干净长文本 PDF +- `A-003` 页眉页脚边界 PDF + +### B类自动化样本 + +- `B-001` 最小坏样本 PDF +- `B-002` 平台审查报告 PDF +- `B-003` 可选真实 `sample-contract.pdf`(传入路径时启用) + +### C类自动化样本 + +- `C-001` 图片型 PDF(OCR第一批) +- `C-002` 低清晰度图片型 PDF(OCR第一批) + +当前门禁判定规则: + +1. A类全部必须通过 +2. B类全部必须被正确拦截 +3. `C-001` 这类清晰图片型 PDF 在 OCR 第一批中应至少达到 `PASS/WARN` +4. `C-002` 这类低清晰度图片型 PDF 在 OCR 第一批中允许 `WARN/BLOCK`,但不能伪装成高质量 `PASS` + +如果未来 OCR 继续增强,应继续在这支门禁里把更多 C类样本逐个升级为 `PASS / WARN / BLOCK` 判定。 diff --git a/scripts/run-pdf-acceptance-gate.ps1 b/scripts/run-pdf-acceptance-gate.ps1 new file mode 100644 index 00000000..f7ee3af5 --- /dev/null +++ b/scripts/run-pdf-acceptance-gate.ps1 @@ -0,0 +1,167 @@ +param( + [string]$ApiUrl = "http://192.168.3.90:8001/api/v1", + [string]$ArtifactRoot = "", + [string]$SampleContractPath = "", + [string]$MemberPassword = "123456" +) + +$ErrorActionPreference = "Stop" + +$repoRoot = (Resolve-Path (Join-Path $PSScriptRoot "..")).Path +$backendDir = Join-Path $repoRoot "backend" +$timestamp = Get-Date -Format "yyyy-MM-ddTHH-mm-ss-fff" + +if ([string]::IsNullOrWhiteSpace($ArtifactRoot)) { + $ArtifactRoot = Join-Path $repoRoot "project-docs\qa-artifacts\pdf-acceptance-gate" +} +if (-not [System.IO.Path]::IsPathRooted($ArtifactRoot)) { + $ArtifactRoot = Join-Path $repoRoot $ArtifactRoot +} +if (-not [string]::IsNullOrWhiteSpace($SampleContractPath) -and -not [System.IO.Path]::IsPathRooted($SampleContractPath)) { + $SampleContractPath = Join-Path $repoRoot $SampleContractPath +} + +$artifactDir = Join-Path $ArtifactRoot $timestamp +New-Item -ItemType Directory -Force -Path $artifactDir | Out-Null + +$summaryJson = Join-Path $artifactDir "pdf-acceptance-gate-wrapper-summary.json" +$summaryMd = Join-Path $artifactDir "pdf-acceptance-gate-wrapper-summary.md" +$logPath = Join-Path $artifactDir "pdf-acceptance-gate.log" +$gitStatusPath = Join-Path $artifactDir "git-status.txt" + +$checks = New-Object System.Collections.Generic.List[object] + +function Add-Check { + param( + [string]$Key, + [string]$Label, + [ValidateSet("pass", "fail", "warn", "skip")] + [string]$Status, + [string]$Detail, + [bool]$Required = $true, + [hashtable]$Meta = @{} + ) + + $checks.Add([ordered]@{ + key = $Key + label = $Label + status = $Status + required = $Required + detail = $Detail + meta = $Meta + }) | Out-Null +} + +function Save-Json { + param([string]$Path, [object]$Value) + $Value | ConvertTo-Json -Depth 18 | Set-Content -Encoding UTF8 -Path $Path +} + +if (-not (Test-Path "$HOME\.Codex\skills\gstack\bin")) { + Add-Check -Key "gstack" -Label "gstack preflight" -Status "fail" -Detail "gstack is missing" +} else { + Add-Check -Key "gstack" -Label "gstack preflight" -Status "pass" -Detail "GSTACK_OK" +} + +$pythonPath = Join-Path $backendDir ".venv\Scripts\python.exe" +if (-not (Test-Path $pythonPath)) { + Add-Check -Key "python" -Label "backend venv python" -Status "fail" -Detail "Backend venv python missing" -Meta @{ path = $pythonPath } +} else { + Add-Check -Key "python" -Label "backend venv python" -Status "pass" -Detail "Backend venv python found" -Meta @{ path = $pythonPath } +} + +$toolPath = Join-Path $backendDir "tools\pdf_acceptance_gate.py" +if (-not (Test-Path $toolPath)) { + Add-Check -Key "tool" -Label "pdf acceptance tool" -Status "fail" -Detail "Tool missing" -Meta @{ path = $toolPath } +} else { + Add-Check -Key "tool" -Label "pdf acceptance tool" -Status "pass" -Detail "Tool found" -Meta @{ path = $toolPath } +} + +$gitStatus = git -C $repoRoot status --short +($gitStatus -join [Environment]::NewLine) | Set-Content -Encoding UTF8 -Path $gitStatusPath +if ($gitStatus) { + Add-Check -Key "working_tree" -Label "working tree" -Status "warn" -Required $false -Detail "Working tree has uncommitted changes; gate is advisory" -Meta @{ artifact = $gitStatusPath } +} else { + Add-Check -Key "working_tree" -Label "working tree" -Status "pass" -Detail "Working tree is clean" -Meta @{ artifact = $gitStatusPath } +} + +$failedPreflight = @($checks | Where-Object { $_.required -and $_.status -eq "fail" }) +if ($failedPreflight.Count -eq 0) { + $args = @( + $toolPath, + "--api-base", $ApiUrl, + "--artifact-dir", $artifactDir, + "--password", $MemberPassword + ) + if (-not [string]::IsNullOrWhiteSpace($SampleContractPath)) { + $args += @("--sample-contract-path", $SampleContractPath) + } + + Push-Location $backendDir + $previousErrorActionPreference = $ErrorActionPreference + $ErrorActionPreference = "Continue" + try { + $output = & $pythonPath @args 2>&1 + $exitCode = $LASTEXITCODE + } finally { + $ErrorActionPreference = $previousErrorActionPreference + Pop-Location + } + ($output -join [Environment]::NewLine) | Set-Content -Encoding UTF8 -Path $logPath + + if ($exitCode -eq 0) { + Add-Check -Key "pdf_acceptance_gate" -Label "pdf acceptance gate" -Status "pass" -Detail "PDF acceptance gate passed" -Meta @{ artifact_dir = $artifactDir; log = $logPath } + } else { + Add-Check -Key "pdf_acceptance_gate" -Label "pdf acceptance gate" -Status "fail" -Detail "PDF acceptance gate failed" -Meta @{ artifact_dir = $artifactDir; log = $logPath } + } +} + +$failedRequired = @($checks | Where-Object { $_.required -and $_.status -eq "fail" }) +$warningsList = @($checks | Where-Object { $_.status -eq "warn" }) +$overallStatus = if ($failedRequired.Count -gt 0) { "FAIL" } elseif ($warningsList.Count -gt 0) { "WARN" } else { "PASS" } + +$summary = [ordered]@{ + name = "pdf-acceptance-gate-wrapper" + status = $overallStatus + generated_at = (Get-Date).ToUniversalTime().ToString("o") + artifact_dir = $artifactDir + targets = [ordered]@{ + api_url = $ApiUrl + sample_contract_path = $SampleContractPath + } + checks = $checks + totals = [ordered]@{ + total = $checks.Count + failed_required = $failedRequired.Count + warnings = $warningsList.Count + } +} +Save-Json -Path $summaryJson -Value $summary + +$markdown = New-Object System.Collections.Generic.List[string] +$markdown.Add("# PDF Acceptance Gate Wrapper") | Out-Null +$markdown.Add("") | Out-Null +$markdown.Add("- Status: $overallStatus") | Out-Null +$markdown.Add("- Generated at: $($summary.generated_at)") | Out-Null +$markdown.Add("- Artifact dir: $artifactDir") | Out-Null +$markdown.Add("- API URL: $ApiUrl") | Out-Null +$markdown.Add("") | Out-Null +$markdown.Add("| Check | Required | Status | Detail |") | Out-Null +$markdown.Add("|---|---:|---|---|") | Out-Null +foreach ($check in $checks) { + $markdown.Add("| $($check.label) | $($check.required) | $($check.status.ToUpperInvariant()) | $($check.detail) |") | Out-Null +} +$markdown.Add("") | Out-Null +$markdown.Add("Artifacts:") | Out-Null +$markdown.Add("") | Out-Null +$markdown.Add('- `pdf-acceptance-gate-wrapper-summary.json`') | Out-Null +$markdown.Add('- `pdf-acceptance-gate-wrapper-summary.md`') | Out-Null +$markdown.Add('- `pdf-acceptance-gate.log`') | Out-Null +$markdown.Add('- `pdf-acceptance-gate-summary.json`') | Out-Null +$markdown.Add('- `pdf-acceptance-gate-summary.md`') | Out-Null +$markdown | Set-Content -Encoding UTF8 -Path $summaryMd + +Write-Host "PDF_ACCEPTANCE_GATE_$overallStatus $artifactDir" +if ($overallStatus -eq "FAIL") { + throw "PDF acceptance gate failed. See $summaryMd" +}