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 合同平台
- 新用户可直接注册并自动进入工作台；正式环境不展示本地初始化入口。
+ 新用户可直接注册并自动进入工作台；登录后会按当前角色进入对应工作区。