From 94969f7f809fbf88c357b7d5d9266aaacde99233 Mon Sep 17 00:00:00 2001 From: Codex Date: Sun, 3 May 2026 23:22:27 +0800 Subject: [PATCH] fix: harden pdf review extraction --- backend/app/api/reviews.py | 7 +- .../legal_review_execution_gateway.py | 47 +++++ .../app/services/legal_review_local_engine.py | 77 ++++---- backend/app/services/review_service.py | 164 ++++++++++++++++-- backend/tests/test_review_service.py | 80 ++++++++- backend/tests/test_reviews_api.py | 39 ++++- frontend/src/app/login/page.tsx | 4 +- 7 files changed, 357 insertions(+), 61 deletions(-) create mode 100644 backend/app/services/legal_review_execution_gateway.py diff --git a/backend/app/api/reviews.py b/backend/app/api/reviews.py index 20854065..a9448f08 100644 --- a/backend/app/api/reviews.py +++ b/backend/app/api/reviews.py @@ -54,9 +54,10 @@ async def create_review( file_text: Optional[str] = None if file is not None: - review_service.validate_review_upload(file_name=file_name, mime_type=file.content_type) file_bytes = await file.read() - file_text = review_service.extract_text_from_file(file_bytes=file_bytes, file_name=file_name) + review_service.validate_review_upload(file_name=file_name, mime_type=file.content_type, file_bytes=file_bytes) + extraction = review_service.extract_review_upload_text(file_bytes=file_bytes, file_name=file_name) + file_text = extraction["text"] combined_text = review_service.merge_review_text(text=normalized_text, file_text=file_text) @@ -66,7 +67,7 @@ async def create_review( if file is not None and not file_text and not normalized_text: return { "code": 40002, - "message": "无法从上传文件中提取可审查文本，请改为可复制文本 PDF / DOCX / TXT，或直接粘贴合同正文。", + "message": extraction["failure_message"], "data": None, } diff --git a/backend/app/services/legal_review_execution_gateway.py b/backend/app/services/legal_review_execution_gateway.py new file mode 100644 index 00000000..947760a9 --- /dev/null +++ b/backend/app/services/legal_review_execution_gateway.py @@ -0,0 +1,47 @@ +from typing import Any, Dict + +from app.clients.shared_ai_platform import SharedAIPlatformClient, build_shared_ai_platform_client +from app.core.config import settings +from app.models.review_task import ReviewTask +from app.services.legal_review_local_engine import LocalLegalReviewEngine + + +class LegalReviewExecutionGateway: + def __init__(self, shared_ai_platform_client: SharedAIPlatformClient, local_engine: LocalLegalReviewEngine): + self.shared_ai_platform_client = shared_ai_platform_client + self.local_engine = local_engine + + def run_review(self, review: ReviewTask) -> Dict[str, Any]: + if settings.shared_ai_platform_legal_review_enabled and settings.shared_ai_platform_base_url.strip(): + try: + return self.shared_ai_platform_client.run_legal_review( + { + "id": review.user_id, + "tenant_id": review.tenant_id, + "role": "system", + }, + { + "title": review.title, + "contract_type": review.contract_type, + "review_as": review.review_as, + "text": review.input_text or "", + }, + ) + except Exception: + if not settings.shared_ai_platform_legal_review_fallback_to_local: + raise + + return self.local_engine.run_review( + tenant_id=review.tenant_id, + title=review.title, + contract_type=review.contract_type, + review_as=review.review_as, + text=review.input_text or "", + ) + + +def build_legal_review_execution_gateway(db, review_rule_repository) -> LegalReviewExecutionGateway: + return LegalReviewExecutionGateway( + shared_ai_platform_client=build_shared_ai_platform_client(db), + local_engine=LocalLegalReviewEngine(review_rule_repository), + ) diff --git a/backend/app/services/legal_review_local_engine.py b/backend/app/services/legal_review_local_engine.py index d8b328d8..d7b6c487 100644 --- a/backend/app/services/legal_review_local_engine.py +++ b/backend/app/services/legal_review_local_engine.py @@ -2,49 +2,49 @@ from typing import Any, Dict, List, Optional DEFAULT_RISK_RULES = [ { - "keyword": "违约金", + "keywords": ["违约金", "liquidated damages", "default penalty", "penalty", "default interest"], "name": "违约责任条款需重点核查", "contract_type": "all", "risk_level": "medium", - "description": "合同中出现违约金约定，建议核对触发条件、计算口径及上限是否明确。", - "suggestion": "建议核对触发条件、计算口径及责任上限。", + "description": "合同中涉及违约金、违约责任或罚则安排，建议核对触发条件、计算口径、责任上限和是否属于惩罚性约定。", + "suggestion": "建议核对违约触发条件、计算方式、责任上限以及是否与主合同义务匹配。", "legal_basis": None, }, { - "keyword": "自动续约", + "keywords": ["自动续约", "自动续期", "auto-renewal", "automatic renewal", "evergreen", "renewal term"], "name": "存在自动续约安排", "contract_type": "all", "risk_level": "high", - "description": "合同包含自动续约表述，建议补充通知期限和终止条件，避免被动续约。", - "suggestion": "建议补充续约通知期限和明确终止条件。", - "legal_basis": "《民法典》关于合同解除与续期约定的规则。", + "description": "合同包含自动续约或默认延续表述，建议补充通知期限、退出窗口、续约周期和价格调整机制，避免被动续约。", + "suggestion": "建议补充明确的续约通知期限、退出方式、续约周期与价格调整规则。", + "legal_basis": "《民法典》关于合同解除、期限与续期约定的一般规则。", }, { - "keyword": "免责", - "name": "免责条款需要限制范围", + "keywords": ["免责", "责任限制", "limitation of liability", "liability cap", "exclusion of liability", "disclaimer"], + "name": "免责或责任限制条款需要限制范围", "contract_type": "all", "risk_level": "high", - "description": "检测到免责相关表述，建议明确免责边界，避免责任明显失衡。", - "suggestion": "建议明确免责边界并校验条款效力。", - "legal_basis": "《民法典》关于格式条款、免责条款效力限制的规则。", + "description": "合同涉及免责、责任限制或赔偿上限安排，建议核查是否过度排除关键责任，例如故意、重大过失、保密义务或赔偿义务。", + "suggestion": "建议明确免责边界、责任上限和必要 carve-out，避免责任明显失衡。", + "legal_basis": "《民法典》关于格式条款、免责条款效力限制的一般规则。", }, { - "keyword": "保密", + "keywords": ["保密", "confidentiality", "confidential information", "non-disclosure", "nda"], "name": "保密义务建议细化", "contract_type": "all", "risk_level": "low", - "description": "合同涉及保密义务，建议明确保密信息范围、期限及违约责任。", - "suggestion": "建议明确保密信息范围、保密期限及违约责任。", + "description": "合同涉及保密义务，建议明确保密信息范围、例外情形、期限、披露边界以及返还/销毁义务。", + "suggestion": "建议明确保密信息范围、例外情形、保密期限及违约责任。", "legal_basis": "基于商业秘密保护与违约责任的一般合规要求。", }, { - "keyword": "付款", - "name": "付款条件建议明确", + "keywords": ["付款", "payment", "payment milestone", "milestone", "invoice", "billing", "fees", "acceptance"], + "name": "付款与验收条件建议明确", "contract_type": "all", "risk_level": "medium", - "description": "检测到付款相关条款，建议明确付款节点、验收标准与发票条件。", - "suggestion": "建议明确付款节点、验收标准与发票条件。", - "legal_basis": "《民法典》关于价款支付与履行顺序的规则。", + "description": "合同涉及付款、验收、开票或费用条款，建议明确付款节点、验收标准、开票前置条件和逾期责任。", + "suggestion": "建议明确付款节点、验收标准、发票条件、逾期责任及争议处理口径。", + "legal_basis": "《民法典》关于价款支付与履行顺序的一般规则。", }, ] @@ -83,8 +83,8 @@ class LocalLegalReviewEngine: "risk_summary": risk_summary, "clauses": self._build_clauses(text), "risk_items": risk_items, - "review_opinion": self._build_review_opinion(risk_summary), - "recommendation": DEFAULT_RECOMMENDATION, + "review_opinion": self._build_review_opinion(risk_summary, bool(text.strip()), bool(risk_items)), + "recommendation": self._build_recommendation(risk_summary, bool(text.strip()), bool(risk_items)), "confidence_summary": { "analysis_mode": "local-rule-engine", "review_as": review_as, @@ -92,10 +92,11 @@ class LocalLegalReviewEngine: } def _build_risk_items(self, tenant_id: str, title: str, contract_type: str, text: str) -> List[Dict[str, Any]]: - search_text = f"{title}\n{contract_type}\n{text}" + search_text = f"{title}\n{contract_type}\n{text}".lower() items: List[Dict[str, Any]] = [] for rule in self._load_rules_for_review(tenant_id, contract_type): - if rule["keyword"] in search_text: + keywords = rule.get("keywords") or [] + if any(keyword.lower() in search_text for keyword in keywords if keyword): items.append( { "title": rule["name"], @@ -112,7 +113,7 @@ class LocalLegalReviewEngine: custom_rules = [ { "name": rule.name, - "keyword": rule.keyword, + "keywords": self._expand_keywords(rule.keyword), "risk_level": rule.risk_level, "description": rule.description, "suggestion": rule.suggestion, @@ -120,10 +121,15 @@ class LocalLegalReviewEngine: } for rule in rules ] - custom_keywords = {rule["keyword"] for rule in custom_rules} - fallback_rules = [rule for rule in DEFAULT_RISK_RULES if rule["keyword"] not in custom_keywords] + custom_signatures = {tuple(rule["keywords"]) for rule in custom_rules} + fallback_rules = [rule for rule in DEFAULT_RISK_RULES if tuple(rule["keywords"]) not in custom_signatures] return [*custom_rules, *fallback_rules] + def _expand_keywords(self, keyword: str) -> List[str]: + if "|" in keyword: + return [item.strip() for item in keyword.split("|") if item.strip()] + return [keyword.strip()] + def _build_risk_summary(self, risk_items: List[Dict[str, Any]]) -> Dict[str, Any]: high = sum(1 for item in risk_items if item["level"] == "high") medium = sum(1 for item in risk_items if item["level"] == "medium") @@ -145,10 +151,21 @@ class LocalLegalReviewEngine: segments = [segment.strip() for segment in text.splitlines() if segment.strip()] return [{"title": f"条款{i}", "content": segment} for i, segment in enumerate(segments[:5], start=1)] - def _build_review_opinion(self, risk_summary: Dict[str, Any]) -> str: + def _build_review_opinion(self, risk_summary: Dict[str, Any], has_text: bool, has_risk_items: bool) -> str: overall = risk_summary["overall"] if overall == "high": - return "检测到高风险条款，建议暂停签署并由法务重点复核责任、续约或免责安排。" + return "检测到高风险条款，建议暂停签署并由法务重点复核自动续约、免责或责任限制安排。" if overall == "medium": - return "检测到中风险条款，建议补充付款、违约等关键条件后再推进签署。" + return "检测到中风险条款，建议先补充付款、违约或验收条件后再推进签署。" + if has_text and not has_risk_items: + return "当前文本未命中显著风险关键词，规则层暂判低风险；但如果原文仅概述条款类型而缺少具体约定，仍建议结合完整条款做人工复核。" return DEFAULT_OPINION + + def _build_recommendation(self, risk_summary: Dict[str, Any], has_text: bool, has_risk_items: bool) -> str: + if risk_summary["overall"] == "high": + return "manual_review" + if risk_summary["overall"] == "medium": + return "proceed_with_revision" + if has_text and not has_risk_items: + return "proceed_with_revision" + return DEFAULT_RECOMMENDATION diff --git a/backend/app/services/review_service.py b/backend/app/services/review_service.py index 5e3f7836..5e5848ee 100644 --- a/backend/app/services/review_service.py +++ b/backend/app/services/review_service.py @@ -22,6 +22,7 @@ from app.models.review_result import ReviewResult from app.models.review_task import ReviewTask from app.repositories.review_result_repository import ReviewResultRepository from app.services.file_service import FileService +from app.services.legal_review_execution_gateway import build_legal_review_execution_gateway from app.repositories.review_rule_repository import ReviewRuleRepository from app.repositories.review_task_repository import ReviewTaskRepository @@ -89,6 +90,7 @@ PDF_TEXT_PATTERN = re.compile(rb"\(([^\)]{2,200})\)Tj") PDF_HEX_TEXT_PATTERN = re.compile(rb"<([0-9A-Fa-f]{4,400})>Tj") PDF_ASCII_PATTERN = re.compile(rb"[\x20-\x7e]{4,120}") PDF_TEXT_MARKERS = ("合同", "违约", "免责", "付款", "保密", "续约") +PDF_PAGE_NOISE_PATTERN = re.compile(r"^(page\s+\d+(\s*/\s*\d+|\s+of\s+\d+)?|\d+\s*/\s*\d+)$", re.IGNORECASE) PDF_NOISE_LINE_PATTERN = re.compile( r"^(%PDF|xref|trailer|startxref|\d+\s+\d+\s+obj|endobj|stream|endstream|<<|/Type|/Catalog|/Pages|/Kids|/Parent|/Resources|/Font|/Length)", re.IGNORECASE, @@ -110,6 +112,14 @@ PDF_NOISE_TOKENS = { "pages", "contents", } +REPORT_PDF_MARKERS = ( + "审查任务：", + "任务编号：", + "风险汇总：", + "审查意见：", + "风险项：", + "原始文本摘要：", +) SUPPORTED_TEXT_SUFFIXES = (".txt", ".md", ".csv", ".json") DOCX_SUFFIX = ".docx" PDF_SUFFIX = ".pdf" @@ -132,6 +142,9 @@ class ReviewService: self.review_result_repository = ReviewResultRepository(db) self.file_service = FileService(db) + def _get_legal_review_execution_gateway(self): + return build_legal_review_execution_gateway(self.db, self.review_rule_repository) + def create_review_task( self, tenant_id: str, @@ -186,7 +199,23 @@ class ReviewService: return self._extract_text_from_pdf(file_bytes) return self._extract_text_from_binary(file_bytes) - def validate_review_upload(self, file_name: Optional[str], mime_type: Optional[str]) -> None: + 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() + + if normalized_name.endswith(PDF_SUFFIX): + if text and self._looks_like_platform_review_report(text): + return { + "text": None, + "failure_message": "检测到疑似平台审查报告 PDF，请上传合同原文 PDF / DOCX / TXT，或直接粘贴合同正文后再审查。", + } + + return { + "text": text, + "failure_message": "无法从上传文件中提取可审查文本，请改为可复制文本 PDF / DOCX / TXT，或直接粘贴合同正文。", + } + + def validate_review_upload(self, file_name: Optional[str], mime_type: Optional[str], file_bytes: bytes) -> None: if not file_name: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="上传文件缺少文件名") @@ -198,6 +227,49 @@ class ReviewService: if normalized_mime and not any(normalized_mime.startswith(allowed) for allowed in SUPPORTED_REVIEW_UPLOAD_MIME_TYPES): raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="仅支持 PDF / DOCX / TXT 文件") + detected_type = self._detect_review_upload_type(file_bytes) + expected_type = { + PDF_SUFFIX: "pdf", + DOCX_SUFFIX: "docx", + ".txt": "txt", + }[suffix] + if detected_type != expected_type: + raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="仅支持 PDF / DOCX / TXT 文件") + + def _detect_review_upload_type(self, file_bytes: bytes) -> str: + if file_bytes.startswith(b"%PDF"): + return "pdf" + if file_bytes.startswith(b"PK"): + try: + with zipfile.ZipFile(io.BytesIO(file_bytes)) as archive: + names = set(archive.namelist()) + if "word/document.xml" in names: + return "docx" + except zipfile.BadZipFile: + return "unknown" + if file_bytes.startswith(b"\x89PNG\r\n\x1a\n"): + return "png" + if file_bytes.startswith(b"MZ"): + return "exe" + if not file_bytes: + return "txt" + + sample = file_bytes[:4096] + if b"\x00" in sample: + return "binary" + + try: + decoded = sample.decode("utf-8") + except UnicodeDecodeError: + try: + decoded = sample.decode("gb18030") + except UnicodeDecodeError: + return "binary" + + printable_chars = sum(1 for char in decoded if char.isprintable() or char in "\r\n\t") + ratio = printable_chars / max(len(decoded), 1) + return "txt" if ratio >= 0.85 else "binary" + def list_reviews(self, current_user: Dict[str, Any]) -> dict: reviews = self.repository.list_all(current_user["tenant_id"], current_user["id"]) @@ -403,6 +475,7 @@ class ReviewService: "legal_basis": self._build_legal_basis(risk_items), "review_opinion": result["review_opinion"], "recommendation": result["recommendation"], + "confidence_summary": result.get("confidence_summary"), "source_excerpt": (review.input_text or "未提供合同文本，仅记录了任务元数据。")[:500], } @@ -425,6 +498,7 @@ class ReviewService: "risk_items": review_result.risk_items, "review_opinion": review_result.review_opinion, "recommendation": review_result.recommendation, + "confidence_summary": review_result.confidence, }, ) else: @@ -459,8 +533,9 @@ class ReviewService: review.completed_at = datetime.now(timezone.utc) return self.repository.update(review) - risk_items = self._build_risk_items(review) - risk_summary = self._build_risk_summary(risk_items) + result = self._get_legal_review_execution_gateway().run_review(review) + risk_items = result["risk_items"] + risk_summary = result["risk_summary"] review.risk_level = risk_summary["overall"] review.status = "completed" review.processing_time_ms = 1000 + len(risk_items) * 200 @@ -471,19 +546,13 @@ class ReviewService: review_result = ReviewResult( id=f"result-{uuid4().hex[:12]}", review_task_id=review.id, - basic_info_json=json.dumps({ - "contract_type": review.contract_type, - "party_a": DEFAULT_PARTY_A, - "party_b": DEFAULT_PARTY_B, - "amount": DEFAULT_AMOUNT, - "term": DEFAULT_TERM, - }, ensure_ascii=False), - clauses_json=json.dumps(self._build_clauses(review), ensure_ascii=False), + basic_info_json=json.dumps(result["basic_info"], ensure_ascii=False), + clauses_json=json.dumps(result["clauses"], ensure_ascii=False), risk_items_json=json.dumps(risk_items, ensure_ascii=False), risk_summary_json=json.dumps(risk_summary, ensure_ascii=False), - review_opinion=self._build_review_opinion(risk_summary), - recommendation=DEFAULT_RECOMMENDATION, - confidence_summary=None, + 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, report_file_id=None, ) self.review_result_repository.create(review_result) @@ -547,16 +616,20 @@ class ReviewService: for attempt in range(max_retries + 1): try: with pdfplumber.open(io.BytesIO(file_bytes)) as pdf: - text_parts = [] + page_lines: List[List[str]] = [] # Extract text from each page for page in pdf.pages: page_text = page.extract_text() if page_text: - text_parts.append(page_text) + lines = [line.strip() for line in page_text.splitlines() if line.strip()] + if lines: + page_lines.append(lines) - if text_parts: - extracted_text = "\n".join(text_parts) + if page_lines: + extracted_text = "\n".join( + "\n".join(lines) for lines in self._strip_repeated_page_decorations(page_lines) if lines + ) normalized = self._clean_pdf_extracted_text(extracted_text) # Validate that we extracted meaningful text @@ -656,6 +729,8 @@ class ReviewService: line = raw_line.strip() if not line: continue + if PDF_PAGE_NOISE_PATTERN.fullmatch(line): + continue if PDF_NOISE_LINE_PATTERN.search(line): continue if line.lower().count(" obj") or line.lower().count("endobj"): @@ -669,6 +744,59 @@ class ReviewService: return None return normalized + def _strip_repeated_page_decorations(self, page_lines: List[List[str]]) -> List[List[str]]: + candidate_counts: Dict[str, int] = {} + for lines in page_lines: + edge_candidates = [line for line in [*lines[:2], *lines[-2:]] if line] + for line in edge_candidates: + candidate_counts[line] = candidate_counts.get(line, 0) + 1 + + repeated_edge_lines = { + line + for line, count in candidate_counts.items() + if count >= 2 and self._looks_like_page_decoration(line) + } + + stripped_pages: List[List[str]] = [] + for lines in page_lines: + stripped_pages.append( + [ + line + for line in lines + if line not in repeated_edge_lines and not PDF_PAGE_NOISE_PATTERN.fullmatch(line) + ] + ) + return stripped_pages + + def _looks_like_page_decoration(self, line: str) -> bool: + normalized = line.strip() + if not normalized: + return False + if PDF_PAGE_NOISE_PATTERN.fullmatch(normalized): + return True + if len(normalized) > 80: + return False + word_count = len(re.findall(r"[A-Za-z0-9\u4e00-\u9fff]+", normalized)) + if word_count > 8: + return False + if any(marker in normalized for marker in PDF_TEXT_MARKERS): + return False + return True + + def _looks_like_platform_review_report(self, value: str) -> bool: + marker_hits = sum(1 for marker in REPORT_PDF_MARKERS if marker in value) + if marker_hits >= 3: + return True + normalized_lines = [line.strip() for line in value.splitlines() if line.strip()] + if len(normalized_lines) < 4: + return False + first_window = "\n".join(normalized_lines[:8]) + return ( + "审查任务：" in first_window + and "任务编号：" in first_window + and ("风险汇总：" in value or "审查意见：" in value) + ) + def _looks_like_natural_pdf_text(self, value: str) -> bool: lines = [line.strip() for line in value.splitlines() if line.strip()] if not lines: diff --git a/backend/tests/test_review_service.py b/backend/tests/test_review_service.py index 03d7cd91..0963be39 100644 --- a/backend/tests/test_review_service.py +++ b/backend/tests/test_review_service.py @@ -222,14 +222,86 @@ class ReviewServiceTestCase(unittest.TestCase): self.assertIn("保密义务", extracted) self.assertNotIn("endobj", extracted.lower()) + def test_extract_text_from_pdf_strips_repeated_headers_and_page_numbers(self): + buffer = io.BytesIO() + pdf = canvas.Canvas(buffer) + pdfmetrics.registerFont(UnicodeCIDFont("STSong-Light")) + pdf.setFont("STSong-Light", 12) + pdf.drawString(72, 780, "Page 1 / 2") + pdf.drawString(72, 760, "PDF extraction boundary sample") + pdf.drawString(72, 720, "合同正文第一条 付款安排") + pdf.showPage() + pdf.setFont("STSong-Light", 12) + pdf.drawString(72, 780, "Page 2 / 2") + pdf.drawString(72, 760, "PDF extraction boundary sample") + pdf.drawString(72, 720, "合同正文第二条 保密义务") + pdf.save() + + extracted = self.service.extract_text_from_file(buffer.getvalue(), "boundary.pdf") + + self.assertIsNotNone(extracted) + assert extracted is not None + self.assertIn("付款安排", extracted) + self.assertIn("保密义务", extracted) + self.assertNotIn("Page 1 / 2", extracted) + self.assertNotIn("Page 2 / 2", extracted) + self.assertNotIn("PDF extraction boundary sample", extracted) + + def test_extract_review_upload_text_rejects_platform_report_pdf(self): + buffer = io.BytesIO() + pdf = canvas.Canvas(buffer) + pdfmetrics.registerFont(UnicodeCIDFont("STSong-Light")) + pdf.setFont("STSong-Light", 12) + y = 780 + for line in [ + "审查任务：文本输入合同", + "任务编号：review-123", + "风险汇总：", + "审查意见：建议复核", + "原始文本摘要：合同文本", + ]: + pdf.drawString(72, y, line) + y -= 20 + pdf.save() + + result = self.service.extract_review_upload_text(buffer.getvalue(), "review-report.pdf") + + self.assertIsNone(result["text"]) + self.assertIn("疑似平台审查报告", result["failure_message"]) + def test_validate_review_upload_accepts_supported_suffix(self): - self.service.validate_review_upload("contract.pdf", "application/pdf") - self.service.validate_review_upload("contract.docx", "application/vnd.openxmlformats-officedocument.wordprocessingml.document") - self.service.validate_review_upload("contract.txt", "text/plain") + self.service.validate_review_upload("contract.pdf", "application/pdf", b"%PDF-1.7") + docx_buffer = io.BytesIO() + with zipfile.ZipFile(docx_buffer, "w") as archive: + archive.writestr("word/document.xml", "") + self.service.validate_review_upload( + "contract.docx", + "application/vnd.openxmlformats-officedocument.wordprocessingml.document", + docx_buffer.getvalue(), + ) + self.service.validate_review_upload("contract.txt", "text/plain", "合同正文".encode("utf-8")) def test_validate_review_upload_rejects_unsupported_suffix(self): with self.assertRaises(HTTPException) as context: - self.service.validate_review_upload("contract.png", "image/png") + self.service.validate_review_upload("contract.png", "image/png", b"\x89PNG\r\n\x1a\n") + + self.assertEqual(context.exception.status_code, 400) + self.assertEqual(context.exception.detail, "仅支持 PDF / DOCX / TXT 文件") + + def test_validate_review_upload_rejects_png_renamed_as_txt(self): + with self.assertRaises(HTTPException) as context: + self.service.validate_review_upload("contract.txt", "text/plain", b"\x89PNG\r\n\x1a\n") + + self.assertEqual(context.exception.status_code, 400) + self.assertEqual(context.exception.detail, "仅支持 PDF / DOCX / TXT 文件") + + def test_validate_review_upload_rejects_exe_renamed_as_docx(self): + with self.assertRaises(HTTPException) as context: + self.service.validate_review_upload( + "contract.docx", + "application/vnd.openxmlformats-officedocument.wordprocessingml.document", + b"MZfake-executable", + ) self.assertEqual(context.exception.status_code, 400) self.assertEqual(context.exception.detail, "仅支持 PDF / DOCX / TXT 文件") diff --git a/backend/tests/test_reviews_api.py b/backend/tests/test_reviews_api.py index 4353f839..ae381e1a 100644 --- a/backend/tests/test_reviews_api.py +++ b/backend/tests/test_reviews_api.py @@ -26,8 +26,15 @@ class StubReviewService: return None return file_bytes.decode("utf-8", errors="ignore") - def validate_review_upload(self, file_name, mime_type): - if file_name.endswith(".png") or file_name.endswith(".exe"): + def extract_review_upload_text(self, file_bytes, file_name): + text = self.extract_text_from_file(file_bytes, file_name) + return { + "text": text, + "failure_message": "无法从上传文件中提取可审查文本，请改为可复制文本 PDF / DOCX / TXT，或直接粘贴合同正文。", + } + + 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): @@ -250,8 +257,11 @@ class ReviewsApiTestCase(unittest.TestCase): def test_create_review_rejects_unextractable_supported_file_without_text_fallback(self): class UnextractableReviewService(StubReviewService): - def extract_text_from_file(self, file_bytes, file_name): - return None + def extract_review_upload_text(self, file_bytes, file_name): + return { + "text": None, + "failure_message": "无法从上传文件中提取可审查文本，请改为可复制文本 PDF / DOCX / TXT，或直接粘贴合同正文。", + } app.dependency_overrides[get_review_service] = lambda: UnextractableReviewService() @@ -266,6 +276,27 @@ class ReviewsApiTestCase(unittest.TestCase): self.assertEqual(body["code"], 40002) self.assertIn("无法从上传文件中提取可审查文本", body["message"]) + def test_create_review_rejects_platform_report_pdf_with_specific_message(self): + class ReportLikeReviewService(StubReviewService): + def extract_review_upload_text(self, file_bytes, file_name): + return { + "text": None, + "failure_message": "检测到疑似平台审查报告 PDF，请上传合同原文 PDF / DOCX / TXT，或直接粘贴合同正文后再审查。", + } + + app.dependency_overrides[get_review_service] = lambda: ReportLikeReviewService() + + response = self.client.post( + "/api/v1/reviews", + data={"review_as": "party_a"}, + files={"file": ("review-report.pdf", b"%PDF", "application/pdf")}, + ) + + self.assertEqual(response.status_code, 200) + body = response.json() + self.assertEqual(body["code"], 40002) + self.assertIn("疑似平台审查报告", body["message"]) + def test_list_reviews_uses_current_user_scope(self): response = self.client.get("/api/v1/reviews") diff --git a/frontend/src/app/login/page.tsx b/frontend/src/app/login/page.tsx index d1291a9c..0b60ee42 100644 --- a/frontend/src/app/login/page.tsx +++ b/frontend/src/app/login/page.tsx @@ -14,7 +14,7 @@ const loginHighlights = ["账号注册与密码登录", "登录后直达审查 const loginTips = [ "新用户建议直接注册账号，注册后会自动登录并进入工作台。", - "管理员初始化入口仅在受控本地环境展示，用于联调和验收，不是常规使用入口。", + "管理员与普通成员共用同一登录入口，登录后按角色进入对应工作台。", "后续还可以继续扩展更细的组织成员、权限和找回密码能力。", ]; @@ -238,7 +238,7 @@ function LoginPageContent() {
Account Access 登录 AI 合同平台 - 新用户可直接注册并自动进入工作台；正式环境不展示本地初始化入口。 + 新用户可直接注册并自动进入工作台；登录后会按当前角色进入对应工作区。