yuqei/ai-contract-platform
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

feat: migrate PDF/OCR extraction to shared-ai-platform

Browse Source 
Remove all local PDF/OCR extraction and quality gate logic from
review_service.py, replacing it with calls to the shared-ai-platform
document extraction service via the SDK.

Removed from review_service.py:
- 17 local PDF/OCR methods (_extract_text_from_pdf, _classify_pdf_source,
  _extract_text_from_pdf_with_ocr, _extract_text_from_image_with_ocr,
  _evaluate_ocr_quality, _looks_like_platform_review_report, etc.)
- All PDF/OCR constants (PDF_TEXT_PATTERN, PDF_NOISE_LINE_PATTERN,
  REPORT_PDF_MARKERS, OCR_GATE_*, PDF_SOURCE_KIND_*, etc.)
- Imports: pdfplumber, pypdfium2, pytesseract, PIL.Image

Added:
- extract_review_upload_text() now calls shared platform's
  client.extract_document_text() directly
- _detect_upload_type() retained locally for upload validation only
- Protocol methods extract_document_text, classify_pdf_source in client

Removed from requirements.txt: pdfplumber, pytesseract
Added to config: shared_ai_platform_document_extraction_enabled

Also includes prior changes:
- Model settings and chat history stabilization
- Mobile UI polish
- Governance launch-closure enhancements

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
yuqei
2026-06-14 11:09:13 +08:00
parent 19ed42c880
commit 6d2edc6753
18 changed files with 142 additions and 1091 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.codex/hooks.json
+15
View File
@@ -0,0 +1,15 @@
{
"hooks": {
"PreToolUse": [
{
"matcher": "Skill",
"hooks": [
{
"type": "command",
"command": "'D:\\yuqei-project\\ai-contract-platform\\.codex\\hooks\\check-gstack.sh'"
}
]
}
]
}
}
.codex/hooks/check-gstack.sh
+20
View File
@@ -0,0 +1,20 @@
#!/bin/bash
# Block skill usage when gstack is not installed globally.
if [ ! -d "$HOME/.claude/skills/gstack/bin" ]; then
cat >&2 <<'MSG'
BLOCKED: gstack is not installed globally.
gstack is required for AI-assisted work in this repo.
Install it:
git clone --depth 1 https://github.com/garrytan/gstack.git ~/.claude/skills/gstack
cd ~/.claude/skills/gstack && ./setup --team
Then restart your AI coding tool.
MSG
echo '{"permissionDecision":"deny","message":"gstack is required but not installed. See stderr for install instructions."}'
exit 0
fi
echo '{}'
.env.example
+2 -1
View File
@@ -5,7 +5,8 @@ NEXT_PUBLIC_API_BASE_URL=http://localhost:8000/api/v1
APP_NAME=AI Contract Platform
APP_ENV=development
SHARED_AI_PLATFORM_BASE_URL=
SHARED_AI_PLATFORM_API_PREFIX=/api/v0.1
# Default shared-ai contract version is v0.2. Override to /api/v0.1 only for temporary compatibility with legacy deployments.
SHARED_AI_PLATFORM_API_PREFIX=/api/v0.2
SHARED_AI_PLATFORM_APP_ID=ai-contract-platform
SHARED_AI_PLATFORM_API_KEY=
SHARED_AI_PLATFORM_TIMEOUT_SECONDS=30
backend/app/clients/shared_ai_platform.py
+18
View File
@@ -37,6 +37,10 @@ class SharedAIPlatformClient(Protocol):
def run_legal_review(self, current_user: Dict[str, Any], payload: Dict[str, Any]) -> dict: ...
def extract_document_text(self, current_user: Dict[str, Any], *, file_name: str, mime_type: Optional[str], content: bytes) -> dict: ...
def classify_pdf_source(self, current_user: Dict[str, Any], *, content: bytes) -> dict: ...
def create_knowledge_base(self, current_user: Dict[str, Any], name: str, description: Optional[str]) -> dict: ...
def list_knowledge_base_versions(self, current_user: Dict[str, Any], knowledge_base_id: str) -> list[dict]: ...
@@ -224,6 +228,20 @@ class RemoteSharedAIPlatformClient:
provider=payload.get("provider"),
)
def extract_document_text(self, current_user: Dict[str, Any], *, file_name: str, mime_type: Optional[str], content: bytes) -> dict:
return self.sdk.extract_document_text(
self._context(current_user),
file_name=file_name,
mime_type=mime_type,
content=content,
)
def classify_pdf_source(self, current_user: Dict[str, Any], *, content: bytes) -> dict:
return self.sdk.classify_pdf_source(
self._context(current_user),
content=content,
)
def create_knowledge_base(self, current_user: Dict[str, Any], name: str, description: Optional[str]) -> dict:
payload = self.sdk.create_knowledge_base(
self._context(current_user),
backend/app/core/config.py
+2 -1
View File
@@ -18,12 +18,13 @@ class Settings(BaseSettings):
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_api_prefix: str = "/api/v0.2"
shared_ai_platform_app_id: str = "ai-contract-platform"
shared_ai_platform_api_key: str = ""
shared_ai_platform_timeout_seconds: float = 30.0
shared_ai_platform_legal_review_enabled: bool = True
shared_ai_platform_legal_review_fallback_to_local: bool = False
shared_ai_platform_document_extraction_enabled: bool = True
prompt_golden_eval_mode: str = "auto"
prompt_golden_eval_include_raw_model_response: bool = False
legal_change_impact_semantic_enabled: bool = True
backend/app/services/review_service.py
+34 -393
View File
@@ -6,23 +6,7 @@ import re
import time
from typing import Any, Dict, List, Optional
from uuid import uuid4
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
try:
from PIL import Image
except ImportError: # pragma: no cover - optional local dependency path
Image = None
from fastapi import HTTPException, status
from reportlab.lib.pagesizes import A4
from reportlab.pdfbase import pdfmetrics
@@ -262,60 +246,9 @@ DEFAULT_TERM = "待识别"
DEFAULT_RECOMMENDATION = "manual_review"
DEFAULT_OPINION = "已完成初步规则审查，请结合正式法务意见进一步确认。"
MAX_TEXT_LENGTH = 20000
DOCX_NAMESPACE = {"w": "http://schemas.openxmlformats.org/wordprocessingml/2006/main"}
BINARY_TEXT_PATTERN = re.compile(r"[\u4e00-\u9fffA-Za-z0-9，。；：、？！《》（）【】\-—_]+")
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,
)
PDF_NOISE_TOKENS = {
"pdf",
"reportlab",
"generated",
"document",
"http",
"https",
"www",
"com",
"font",
"basefont",
"encoding",
"type1",
"catalog",
"pages",
"contents",
}
REPORT_PDF_MARKERS = (
"审查任务：",
"任务编号：",
"风险汇总：",
"审查意见：",
"风险项：",
"原始文本摘要：",
)
OCR_SIGNATURE_NOISE_PATTERN = re.compile(
r"^((甲方|乙方|双方|法定代表人|授权代表)(\(?(签字|签章|盖章)\)?)|(签字|签章|盖章|签署页|签字页|盖章页|签署栏|本页无正文)|(partya(signature)?)|(partyb(signature)?)|(signature|signed))$",
re.IGNORECASE,
)
OCR_SIGNATURE_DATE_PATTERN = re.compile(
r"^((日期|签署日期|签约日期|date)[:：]?[\d\-./年月日]+)$",
re.IGNORECASE,
)
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"
# Upload validation constants
PDF_SUFFIX = ".pdf"
DOCX_SUFFIX = ".docx"
IMAGE_PNG_SUFFIX = ".png"
IMAGE_JPG_SUFFIXES = (".jpg", ".jpeg")
SUPPORTED_REVIEW_IMAGE_SUFFIXES = (IMAGE_PNG_SUFFIX, *IMAGE_JPG_SUFFIXES)
@@ -327,6 +260,7 @@ SUPPORTED_REVIEW_UPLOAD_MIME_TYPES = (
"image/png",
"image/jpeg",
)
# Report PDF generation
PDF_FONT_NAME = "STSong-Light"
pdfmetrics.registerFont(UnicodeCIDFont(PDF_FONT_NAME))
@@ -353,6 +287,7 @@ class ReviewService:
preferred.append("contract_review_default")
return preferred
def create_review_task(
self,
tenant_id: str,
@@ -400,267 +335,16 @@ class ReviewService:
merged_text = "\n\n".join(parts)
return merged_text[:MAX_TEXT_LENGTH]
def _normalize_extracted_text(self, text: Optional[str]) -> Optional[str]:
if not text:
return None
lines = [line.strip() for line in str(text).replace("\r", "").split("\n")]
cleaned = [line for line in lines if line]
return "\n".join(cleaned) if cleaned else None
def _extract_text_from_docx(self, file_bytes: bytes) -> Optional[str]:
try:
with zipfile.ZipFile(io.BytesIO(file_bytes)) as archive:
xml = archive.read("word/document.xml")
except Exception:
return None
root = ElementTree.fromstring(xml)
texts = [node.text or "" for node in root.findall(".//w:t", DOCX_NAMESPACE)]
return self._normalize_extracted_text("\n".join(texts))
def _clean_pdf_extracted_text(self, text: Optional[str]) -> Optional[str]:
if not text:
return None
lines = []
seen_headers: dict[str, int] = {}
for raw in str(text).replace("\r", "").split("\n"):
line = raw.strip()
if not line:
continue
if PDF_NOISE_LINE_PATTERN.match(line):
continue
if PDF_PAGE_NOISE_PATTERN.match(line):
continue
lowered = line.lower()
if lowered in {"pdf extraction boundary sample", "pdf", "page 1 / 2", "page 2 / 2"}:
continue
seen_headers[lowered] = seen_headers.get(lowered, 0) + 1
if seen_headers[lowered] > 1 and len(line) <= 40:
continue
lines.append(line)
return self._normalize_extracted_text("\n".join(lines))
def _extract_text_from_binary(self, file_bytes: bytes) -> Optional[str]:
if not file_bytes:
return None
for encoding in ("utf-8", "gb18030"):
try:
decoded = file_bytes.decode(encoding)
printable = sum(1 for ch in decoded if ch.isprintable() or ch in "\r\n\t")
ratio = printable / max(len(decoded), 1)
if ratio >= 0.85:
return self._normalize_extracted_text(decoded)
except UnicodeDecodeError:
continue
return None
def _extract_text_from_pdf(self, file_bytes: bytes) -> Optional[str]:
text = None
try:
with pdfplumber.open(io.BytesIO(file_bytes)) as pdf:
pages = []
for page in pdf.pages:
page_text = page.extract_text() or ""
if page_text.strip():
pages.append(page_text)
text = "\n".join(pages)
except Exception:
text = None
if not text:
try:
decoded = file_bytes.decode("latin-1", errors="ignore")
candidates = []
candidates.extend(match.group(1) for match in PDF_TEXT_PATTERN.finditer(file_bytes))
candidates.extend(bytes.fromhex(match.group(1).decode("ascii")).decode("utf-16-be", errors="ignore") for match in PDF_HEX_TEXT_PATTERN.finditer(file_bytes))
if candidates:
text = "\n".join(candidates)
else:
text = decoded
except Exception:
text = None
return self._clean_pdf_extracted_text(text)
def _classify_pdf_source(self, file_bytes: bytes) -> Dict[str, Any]:
page_count = 0
text_pages = 0
image_pages = 0
try:
with pdfplumber.open(io.BytesIO(file_bytes)) as pdf:
page_count = len(pdf.pages)
for page in pdf.pages:
page_text = (page.extract_text() or "").strip()
if page_text:
text_pages += 1
else:
image_pages += 1
except Exception:
pass
if page_count == 0:
kind = PDF_SOURCE_KIND_UNKNOWN
elif text_pages == page_count:
kind = PDF_SOURCE_KIND_TEXT
elif image_pages == page_count:
kind = PDF_SOURCE_KIND_SCANNED
else:
kind = PDF_SOURCE_KIND_MIXED
return {
"kind": kind,
"page_count": page_count,
"text_pages": text_pages,
"image_pages": image_pages,
}
def _evaluate_ocr_quality(self, text: Optional[str], confidences: List[float], pdf_source: Dict[str, Any]) -> Dict[str, Any]:
normalized = self._normalize_extracted_text(text) or ""
char_count = len(normalized)
lines = [line for line in normalized.split("\n") if line.strip()]
line_count = len(lines)
printable_ratio = (
sum(1 for ch in normalized if ch.isprintable() or ch in "\n\t") / max(char_count, 1)
if normalized else 0.0
)
noise_lines = sum(1 for line in lines if PDF_PAGE_NOISE_PATTERN.match(line) or PDF_NOISE_LINE_PATTERN.match(line))
noise_ratio = noise_lines / max(line_count, 1) if lines else 1.0
mean_confidence = round(sum(confidences) / len(confidences), 2) if confidences else None
status = OCR_GATE_PASS
failure_message = None
if char_count < 10 or line_count <= 1 or noise_ratio >= 0.4 or printable_ratio < 0.7 or (mean_confidence is not None and mean_confidence < 30):
status = OCR_GATE_BLOCK
failure_message = "检测到扫描件/图片型 PDF，但 OCR 识别质量不足，请上传更清晰的扫描件、可复制文本 PDF / DOCX / TXT，或直接粘贴合同正文。"
elif char_count < 80 or line_count < 3 or noise_ratio >= 0.15 or (mean_confidence is not None and mean_confidence < 75):
status = OCR_GATE_WARN
failure_message = "已通过 OCR 提取文本，但识别质量一般，建议人工复核原文后再使用审查结果。"
return {
"status": status,
"char_count": char_count,
"line_count": line_count,
"printable_ratio": round(printable_ratio, 4),
"noise_ratio": round(noise_ratio, 4),
"mean_confidence": mean_confidence,
"failure_message": failure_message,
"page_count": pdf_source.get("page_count"),
}
def _extract_text_from_pdf_with_ocr(self, file_bytes: bytes, pdf_source: Dict[str, Any]) -> Dict[str, Any]:
if pdfium is None or pytesseract is None:
return {
"text": None,
"failure_message": "检测到扫描件/图片型 PDF，但 OCR 运行时不可用，请改用可复制文本 PDF / DOCX / TXT，或直接粘贴合同正文。",
"metadata": {
"document_kind": pdf_source.get("kind"),
"extraction_mode": "ocr_runtime_missing",
"ocr_gate_status": OCR_GATE_BLOCK,
"ocr_quality": {"status": OCR_GATE_BLOCK, "char_count": 0, "failure_message": "ocr runtime missing"},
},
}
try:
pdf = pdfium.PdfDocument(io.BytesIO(file_bytes))
chunks: List[str] = []
confidences: List[float] = []
for page_index in range(len(pdf)):
bitmap = pdf[page_index].render(scale=2)
pil_image = bitmap.to_pil()
data = pytesseract.image_to_data(pil_image, lang="chi_sim+eng", output_type=TesseractOutput.DICT)
words = [str(item).strip() for item in data.get("text", []) if str(item).strip()]
chunks.append(" ".join(words))
for conf in data.get("conf", []):
try:
conf_value = float(conf)
except (TypeError, ValueError):
continue
if conf_value >= 0:
confidences.append(conf_value)
text = self._normalize_extracted_text("\n".join(chunks))
quality = self._evaluate_ocr_quality(text, confidences, pdf_source)
return {
"text": text if quality["status"] != OCR_GATE_BLOCK else None,
"failure_message": quality.get("failure_message"),
"metadata": {
"document_kind": pdf_source.get("kind"),
"extraction_mode": "ocr",
"ocr_gate_status": quality["status"],
"ocr_quality": quality,
},
}
except Exception:
return {
"text": None,
"failure_message": "检测到扫描件/图片型 PDF，但 OCR 前解析失败，请改用可复制文本 PDF / DOCX / TXT，或直接粘贴合同正文。",
"metadata": {
"document_kind": pdf_source.get("kind"),
"extraction_mode": "ocr_parse_failed",
"ocr_gate_status": OCR_GATE_BLOCK,
"ocr_quality": {"status": OCR_GATE_BLOCK, "char_count": 0, "failure_message": "ocr parse failed"},
},
}
def _extract_text_from_image_with_ocr(self, file_bytes: bytes) -> Dict[str, Any]:
image_source = {
"kind": "image_upload",
"page_count": 1,
"text_pages": 0,
"image_pages": 1,
}
if Image is None or pytesseract is None:
return {
"text": None,
"failure_message": "检测到图片合同，但 OCR 运行时不可用，请改用更清晰的图片、可复制文本 PDF / DOCX / TXT，或直接粘贴合同正文。",
"metadata": {
"document_kind": image_source["kind"],
"extraction_mode": "ocr_runtime_missing",
"ocr_gate_status": OCR_GATE_BLOCK,
"ocr_quality": {"status": OCR_GATE_BLOCK, "char_count": 0, "failure_message": "ocr runtime missing"},
},
}
try:
image = Image.open(io.BytesIO(file_bytes))
image = image.convert("RGB")
data = pytesseract.image_to_data(image, lang="chi_sim+eng", output_type=TesseractOutput.DICT)
words = [str(item).strip() for item in data.get("text", []) if str(item).strip()]
confidences: List[float] = []
for conf in data.get("conf", []):
try:
conf_value = float(conf)
except (TypeError, ValueError):
continue
if conf_value >= 0:
confidences.append(conf_value)
text = self._normalize_extracted_text(" ".join(words))
quality = self._evaluate_ocr_quality(text, confidences, image_source)
if quality["status"] == OCR_GATE_BLOCK:
failure_message = "检测到图片合同，但 OCR 识别质量不足，请上传更清晰的图片、可复制文本 PDF / DOCX / TXT，或直接粘贴合同正文。"
elif quality["status"] == OCR_GATE_WARN:
failure_message = "已通过 OCR 提取图片文本，但识别质量一般，建议人工复核原图后再使用审查结果。"
else:
failure_message = None
quality["failure_message"] = failure_message
return {
"text": text if quality["status"] != OCR_GATE_BLOCK else None,
"failure_message": failure_message,
"metadata": {
"document_kind": image_source["kind"],
"extraction_mode": "ocr",
"ocr_gate_status": quality["status"],
"ocr_quality": quality,
},
}
except Exception:
return {
"text": None,
"failure_message": "检测到图片合同，但 OCR 解析失败，请改用更清晰的图片、可复制文本 PDF / DOCX / TXT，或直接粘贴合同正文。",
"metadata": {
"document_kind": image_source["kind"],
"extraction_mode": "ocr_parse_failed",
"ocr_gate_status": OCR_GATE_BLOCK,
"ocr_quality": {"status": OCR_GATE_BLOCK, "char_count": 0, "failure_message": "ocr parse failed"},
},
}
def _looks_like_platform_review_report_file_name(self, normalized_name: str) -> bool:
return "review-report" in (normalized_name or "") or "审查报告" in (normalized_name or "")
def _looks_like_platform_review_report(self, text: str) -> bool:
normalized = text or ""
return all(marker in normalized for marker in REPORT_PDF_MARKERS[:4])
def _build_clauses_from_text(self, text: str) -> List[Dict[str, Any]]:
return self._build_clause_summaries(text)
@@ -849,82 +533,33 @@ class ReviewService:
self.review_result_repository.create(review_result)
return review_task
def _extract_text_from_file(self, file_bytes: bytes, file_name: Optional[str]) -> Optional[str]:
if not file_bytes:
return None
normalized_name = (file_name or "").lower()
if normalized_name.endswith(SUPPORTED_TEXT_SUFFIXES):
return self._normalize_extracted_text(file_bytes.decode("utf-8", errors="ignore"))
if normalized_name.endswith(DOCX_SUFFIX):
return self._extract_text_from_docx(file_bytes)
if normalized_name.endswith(PDF_SUFFIX):
return self._extract_text_from_pdf(file_bytes)
return self._extract_text_from_binary(file_bytes)
def extract_text_from_file(self, file_bytes: bytes, file_name: Optional[str]) -> Optional[str]:
return self._extract_text_from_file(file_bytes, file_name)
def extract_review_upload_text(self, file_bytes: bytes, file_name: Optional[str]) -> Dict[str, Optional[str]]:
normalized_name = (file_name or "").lower()
text = None if normalized_name.endswith(SUPPORTED_REVIEW_IMAGE_SUFFIXES) else self.extract_text_from_file(file_bytes=file_bytes, file_name=file_name)
extraction_metadata: Dict[str, Any] = {
"document_kind": None,
"extraction_mode": "text",
"ocr_gate_status": None,
"""Extract text from a review upload file via the shared-ai-platform document extraction service."""
from app.clients.shared_ai_platform import build_shared_ai_platform_client
client = build_shared_ai_platform_client(self.db)
current_user = {
"id": "system-document-extraction",
"tenant_id": "default",
"role": "system",
}
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": None,
"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"],
},
}
if normalized_name.endswith(SUPPORTED_REVIEW_IMAGE_SUFFIXES):
ocr_result = self._extract_text_from_image_with_ocr(file_bytes)
return {
"text": ocr_result["text"],
"failure_message": ocr_result["failure_message"],
"metadata": {
**extraction_metadata,
**ocr_result["metadata"],
},
}
result = client.extract_document_text(
current_user,
file_name=file_name or "",
mime_type=None,
content=file_bytes,
)
return {
"text": text,
"failure_message": "无法从上传文件中提取可审查文本，请改为可复制文本 PDF / DOCX / TXT / PNG / JPG / JPEG，或直接粘贴合同正文。",
"metadata": extraction_metadata,
"text": result.get("text"),
"failure_message": result.get("failure_message"),
"metadata": result.get("metadata", {}),
}
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="上传文件缺少文件名")
@@ -937,7 +572,7 @@ 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 / PNG / JPG / JPEG 文件")
detected_type = self._detect_review_upload_type(file_bytes)
detected_type = self._detect_upload_type(file_bytes)
expected_type = {
PDF_SUFFIX: "pdf",
DOCX_SUFFIX: "docx",
@@ -949,11 +584,14 @@ class ReviewService:
if detected_type != expected_type:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="仅支持 PDF / DOCX / TXT / PNG / JPG / JPEG 文件")
def _detect_review_upload_type(self, file_bytes: bytes) -> str:
@staticmethod
def _detect_upload_type(file_bytes: bytes) -> str:
"""Detect the type of an uploaded file from its binary content (magic bytes)."""
if file_bytes.startswith(b"%PDF"):
return "pdf"
if file_bytes.startswith(b"PK"):
try:
import zipfile
with zipfile.ZipFile(io.BytesIO(file_bytes)) as archive:
names = set(archive.namelist())
if "word/document.xml" in names:
@@ -981,9 +619,12 @@ class ReviewService:
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"
printable = sum(1 for ch in decoded if ch.isprintable() or ch in "\r\n\t")
ratio = printable / max(len(decoded), 1)
if ratio >= 0.85:
return "txt"
return "binary"
def list_reviews(self, current_user: Dict[str, Any]) -> dict:
reviews = self.repository.list_all(current_user["tenant_id"], current_user["id"])
backend/requirements.txt
+1 -2
View File
@@ -14,7 +14,6 @@ celery==5.5.1
httpx==0.28.1
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
-e ../../shared-ai-python-sdk
backend/shared_ai_platform_sdk/__init__.py
-12
View File
@@ -1,12 +0,0 @@
from shared_ai_platform_sdk.client import SharedAIPlatformClient
from shared_ai_platform_sdk.errors import SharedAIPlatformError
from shared_ai_platform_sdk.types import PollOptions, RequestContext, SSEEvent, SyncRecord
__all__ = [
"SharedAIPlatformClient",
"SharedAIPlatformError",
"RequestContext",
"SSEEvent",
"PollOptions",
"SyncRecord",
]
backend/shared_ai_platform_sdk/client.py
-409
View File
@@ -1,409 +0,0 @@
from __future__ import annotations
import json
import time
from typing import Iterator
import httpx
from .errors import SharedAIPlatformError
from .types import PollOptions, RequestContext, SSEEvent, SyncRecord
class SharedAIPlatformClient:
def __init__(
self,
*,
base_url: str = "http://127.0.0.1:8011",
api_key: str,
app_id: str,
timeout_seconds: float = 30.0,
client: httpx.Client | None = None,
api_prefix: str = "/api/v0.2",
):
self.base_url = base_url.rstrip("/")
self.api_prefix = self._normalize_api_prefix(api_prefix)
self.api_key = api_key
self.app_id = app_id
self._owns_client = client is None
self._client = client or httpx.Client(base_url=self.base_url, timeout=timeout_seconds)
def close(self):
if self._owns_client:
self._client.close()
def __enter__(self) -> "SharedAIPlatformClient":
return self
def __exit__(self, exc_type, exc, tb):
self.close()
def list_tasks(self, context: RequestContext) -> list[dict]:
return self._request_json("GET", self._path("/tasks"), context=context)["data"]
def run_task(
self,
context: RequestContext,
task_key: str,
*,
inputs: dict,
provider: str | None = None,
metadata: dict | None = None,
) -> dict:
return self._request_json(
"POST",
self._path(f"/tasks/{task_key}/run"),
context=context,
json_body={"inputs": inputs, "provider": provider, "metadata": metadata or {}},
)["data"]
def enqueue_task(
self,
context: RequestContext,
task_key: str,
*,
inputs: dict,
provider: str | None = None,
metadata: dict | None = None,
) -> dict:
return self._request_json(
"POST",
self._path(f"/tasks/{task_key}/enqueue"),
context=context,
json_body={"inputs": inputs, "provider": provider, "metadata": metadata or {}},
)["data"]
def get_task_run(self, context: RequestContext, run_id: str) -> dict:
return self._request_json("GET", self._path(f"/task-runs/{run_id}"), context=context)["data"]
def list_jobs(self, context: RequestContext, *, status_value: str | None = None, limit: int = 50) -> list[dict]:
params = {"limit": limit}
if status_value:
params["status_value"] = status_value
return self._request_json("GET", self._path("/jobs"), context=context, params=params)["data"]
def get_job(self, context: RequestContext, job_id: str) -> dict:
return self._request_json("GET", self._path(f"/jobs/{job_id}"), context=context)["data"]
def run_next_job(self, context: RequestContext) -> dict | None:
return self._request_json("POST", self._path("/jobs/run-next"), context=context)["data"]["job"]
def poll_job_until_terminal(self, context: RequestContext, job_id: str, options: PollOptions | None = None) -> dict:
opts = options or PollOptions()
deadline = time.time() + opts.timeout_seconds
while True:
job = self.get_job(context, job_id)
if job["status"] in opts.terminal_statuses:
return job
if time.time() >= deadline:
raise TimeoutError(f"job {job_id} did not reach a terminal state within {opts.timeout_seconds} seconds")
time.sleep(opts.interval_seconds)
def list_conversations(self, context: RequestContext, *, page: int = 1, page_size: int = 20) -> dict:
return self._request_json(
"GET",
self._path("/conversations"),
context=context,
params={"page": page, "page_size": page_size},
)["data"]
def get_conversation_detail(self, context: RequestContext, conversation_id: str) -> dict:
return self._request_json("GET", self._path(f"/conversations/{conversation_id}"), context=context)["data"]
def list_knowledge_bases(self, context: RequestContext, *, page: int = 1, page_size: int = 20) -> dict:
return self._request_json(
"GET",
self._path("/knowledge-bases"),
context=context,
params={"page": page, "page_size": page_size},
)["data"]
def list_documents(self, context: RequestContext, knowledge_base_id: str, *, page: int = 1, page_size: int = 20) -> dict:
return self._request_json(
"GET",
self._path(f"/knowledge-bases/{knowledge_base_id}/documents"),
context=context,
params={"page": page, "page_size": page_size},
)["data"]
def create_knowledge_base(
self,
context: RequestContext,
*,
name: str,
description: str | None = None,
retrieval_mode: str = "hybrid",
) -> dict:
return self._request_json(
"POST",
self._path("/knowledge-bases"),
context=context,
json_body={"name": name, "description": description, "retrieval_mode": retrieval_mode},
)["data"]
def list_knowledge_base_versions(self, context: RequestContext, knowledge_base_id: str) -> list[dict]:
return self._request_json(
"GET",
self._path(f"/knowledge-bases/{knowledge_base_id}/versions"),
context=context,
)["data"]
def publish_knowledge_base(self, context: RequestContext, knowledge_base_id: str) -> dict:
return self._request_json(
"POST",
self._path(f"/knowledge-bases/{knowledge_base_id}/publish"),
context=context,
)["data"]
def unpublish_knowledge_base(self, context: RequestContext, knowledge_base_id: str) -> dict:
return self._request_json(
"POST",
self._path(f"/knowledge-bases/{knowledge_base_id}/unpublish"),
context=context,
)["data"]
def rollback_knowledge_base(
self,
context: RequestContext,
knowledge_base_id: str,
*,
target_version_label: str,
change_summary: str | None = None,
) -> dict:
return self._request_json(
"POST",
self._path(f"/knowledge-bases/{knowledge_base_id}/rollback"),
context=context,
json_body={"target_version_label": target_version_label, "change_summary": change_summary},
)["data"]
def upload_document(
self,
context: RequestContext,
knowledge_base_id: str,
*,
file_name: str,
mime_type: str | None,
content: bytes,
) -> dict:
response = self._client.post(
self._path(f"/knowledge-bases/{knowledge_base_id}/documents"),
headers=self._headers(context),
files={"file": (file_name, content, mime_type or "application/octet-stream")},
)
self._raise_for_status(response)
payload = response.json()
if payload.get("code") not in (0, None):
raise SharedAIPlatformError(
status_code=response.status_code,
message=str(payload.get("message") or "request failed"),
code=payload.get("code"),
data=payload.get("data"),
)
return payload["data"]
def retry_document(
self,
context: RequestContext,
knowledge_base_id: str,
document_id: str,
*,
file_name: str | None = None,
mime_type: str | None = None,
content: bytes | None = None,
) -> dict:
request_kwargs = {"headers": self._headers(context)}
if content is not None:
request_kwargs["files"] = {"file": (file_name or "document.txt", content, mime_type or "application/octet-stream")}
response = self._client.post(
self._path(f"/knowledge-bases/{knowledge_base_id}/documents/{document_id}/retry"),
**request_kwargs,
)
self._raise_for_status(response)
payload = response.json()
if payload.get("code") not in (0, None):
raise SharedAIPlatformError(
status_code=response.status_code,
message=str(payload.get("message") or "request failed"),
code=payload.get("code"),
data=payload.get("data"),
)
return payload["data"]
def sync_content(
self,
context: RequestContext,
knowledge_base_id: str,
*,
records: list[SyncRecord | dict],
schema_version: str = "content-sync.v1",
sync_mode: str = "incremental",
batch_id: str | None = None,
cursor_token: str | None = None,
) -> dict:
normalized_records = [record.to_dict() if isinstance(record, SyncRecord) else record for record in records]
return self._request_json(
"POST",
self._path(f"/knowledge-bases/{knowledge_base_id}/sync-content"),
context=context,
json_body={
"schema_version": schema_version,
"sync_mode": sync_mode,
"batch_id": batch_id,
"cursor_token": cursor_token,
"records": normalized_records,
},
)["data"]
def list_sync_jobs(self, context: RequestContext, knowledge_base_id: str) -> dict:
return self._request_json("GET", self._path(f"/knowledge-bases/{knowledge_base_id}/sync-jobs"), context=context)["data"]
def run_legal_review(
self,
context: RequestContext,
*,
title: str,
text: str,
contract_type: str | None = None,
review_as: str = "party_a",
provider: str | None = None,
) -> dict:
return self._request_json(
"POST",
self._path("/domain-packs/legal-review/reviews/run"),
context=context,
json_body={
"title": title,
"contract_type": contract_type,
"review_as": review_as,
"text": text,
"provider": provider,
},
)["data"]
def stream_conversation(
self,
context: RequestContext,
*,
message: str,
chat_context: dict | None = None,
provider: str | None = None,
template_key: str | None = None,
conversation_id: str | None = None,
) -> Iterator[SSEEvent]:
headers = self._headers(context)
with self._client.stream(
"POST",
self._path("/conversations/stream"),
headers=headers,
json={
"conversation_id": conversation_id,
"message": message,
"chat_context": chat_context or {},
"provider": provider,
"template_key": template_key,
},
) as response:
self._raise_for_status(response)
yield from self._iter_sse_events(response)
def _iter_sse_events(self, response: httpx.Response) -> Iterator[SSEEvent]:
data_lines: list[str] = []
for line in response.iter_lines():
if isinstance(line, bytes):
line = line.decode("utf-8")
if line == "":
payload = self._parse_sse_payload(data_lines)
data_lines = []
if payload is None:
continue
yield SSEEvent(event_type=str(payload.get("type") or "unknown"), payload=payload)
continue
if line.startswith("data:"):
data_part = line.removeprefix("data:").lstrip()
if data_part:
data_lines.append(data_part)
continue
if data_lines or line.startswith("{") or line.startswith("["):
data_lines.append(line)
payload = self._parse_sse_payload(data_lines)
if payload is not None:
yield SSEEvent(event_type=str(payload.get("type") or "unknown"), payload=payload)
@staticmethod
def _parse_sse_payload(data_lines: list[str]) -> dict | None:
if not data_lines:
return None
payload_text = "\n".join(part for part in data_lines if part).strip()
if not payload_text:
return None
return json.loads(payload_text)
def _request_json(
self,
method: str,
path: str,
*,
context: RequestContext | None = None,
json_body: dict | None = None,
params: dict | None = None,
) -> dict:
response = self._client.request(method, path, headers=self._headers(context), json=json_body, params=params)
self._raise_for_status(response)
payload = response.json()
if payload.get("code") not in (0, None):
raise SharedAIPlatformError(
status_code=response.status_code,
message=str(payload.get("message") or "request failed"),
code=payload.get("code"),
data=payload.get("data"),
)
return payload
def _headers(self, context: RequestContext | None = None) -> dict[str, str]:
headers: dict[str, str] = {}
if self.api_key.strip():
headers["Authorization"] = f"Bearer {self.api_key}"
if context is not None:
headers.update(
{
"X-Shared-AI-App-Id": self.app_id,
"X-Shared-AI-Tenant-Id": context.tenant_id,
"X-Shared-AI-User-Id": context.user_id,
"X-Shared-AI-User-Role": context.user_role,
}
)
return headers
def _raise_for_status(self, response: httpx.Response):
try:
response.raise_for_status()
except httpx.HTTPStatusError as exc:
if not response.is_closed:
response.read()
message = response.text
try:
payload = response.json()
message = str(payload.get("message") or message)
code = payload.get("code")
data = payload.get("data")
except Exception:
code = None
data = None
raise SharedAIPlatformError(response.status_code, message, code, data) from exc
@staticmethod
def _normalize_api_prefix(api_prefix: str) -> str:
normalized = (api_prefix or "/api/v0.2").strip().rstrip("/")
if not normalized.startswith("/"):
normalized = f"/{normalized}"
return normalized
def _path(self, suffix: str) -> str:
normalized_suffix = suffix if suffix.startswith("/") else f"/{suffix}"
return f"{self.api_prefix}{normalized_suffix}"
backend/shared_ai_platform_sdk/errors.py
-6
View File
@@ -1,6 +0,0 @@
class SharedAIPlatformError(Exception):
def __init__(self, status_code: int, message: str, code: int | None = None, data=None):
super().__init__(message)
self.status_code = status_code
self.code = code
self.data = data
backend/shared_ai_platform_sdk/types.py
-48
View File
@@ -1,48 +0,0 @@
from dataclasses import dataclass, field
from typing import Any
@dataclass(frozen=True)
class RequestContext:
tenant_id: str
user_id: str
user_role: str = "admin"
@dataclass(frozen=True)
class SSEEvent:
event_type: str
payload: dict[str, Any]
@dataclass(frozen=True)
class PollOptions:
interval_seconds: float = 1.0
timeout_seconds: float = 60.0
terminal_statuses: tuple[str, ...] = ("succeeded", "failed", "cancelled")
@dataclass(frozen=True)
class SyncRecord:
source_type: str
source_id: str
title: str | None = None
content: str | None = None
operation: str = "upsert"
schema_version: str = "content-sync.v1"
source_version: str | None = None
source_updated_at: str | None = None
metadata: dict[str, Any] = field(default_factory=dict)
def to_dict(self) -> dict[str, Any]:
return {
"schema_version": self.schema_version,
"operation": self.operation,
"source_type": self.source_type,
"source_id": self.source_id,
"source_version": self.source_version,
"source_updated_at": self.source_updated_at,
"title": self.title,
"content": self.content,
"metadata": self.metadata,
}
backend/tests/test_review_service.py
-209
View File
@@ -4,9 +4,6 @@ import zipfile
from datetime import datetime, timezone
from fastapi import HTTPException
from reportlab.pdfbase import pdfmetrics
from reportlab.pdfbase.cidfonts import UnicodeCIDFont
from reportlab.pdfgen import canvas
from app.core.config import settings
from app.services.legal_domain_registry import LEGAL_ARTICLE_REFERENCE_RULES, LEGAL_SOURCE_ALIAS_MAP
@@ -619,185 +616,6 @@ class ReviewServiceTestCase(unittest.TestCase):
self.assertIsNotNone(stored_result.report_file_id)
self.assertIn("lart-civil-496", stored_result.confidence["report_download_metadata"]["legal_article_ref_ids"])
def test_extract_text_from_docx(self):
buffer = io.BytesIO()
with zipfile.ZipFile(buffer, "w") as archive:
archive.writestr(
"word/document.xml",
"""<?xml version=\"1.0\" encoding=\"UTF-8\"?>
<w:document xmlns:w=\"http://schemas.openxmlformats.org/wordprocessingml/2006/main\">
<w:body>
<w:p><w:r><w:t>自动续约条款</w:t></w:r></w:p>
<w:p><w:r><w:t>免责与付款安排</w:t></w:r></w:p>
</w:body>
</w:document>""",
)
extracted = self.service.extract_text_from_file(buffer.getvalue(), "contract.docx")
self.assertEqual(extracted, "自动续约条款\n免责与付款安排")
def test_extract_text_from_plain_text_file(self):
extracted = self.service.extract_text_from_file("保密条款\n付款条件".encode("utf-8"), "contract.txt")
self.assertEqual(extracted, "保密条款\n付款条件")
def test_clean_pdf_extracted_text_filters_structural_noise(self):
cleaned = self.service._clean_pdf_extracted_text(
"%PDF-1.7\n1 0 obj\n<< /Type /Catalog >>\nendobj\n合同正文第一条 付款安排\n合同正文第二条 保密义务\nxref"
)
self.assertEqual(cleaned, "合同正文第一条 付款安排\n合同正文第二条 保密义务")
def test_extract_text_from_pdf_prefers_readable_text(self):
buffer = io.BytesIO()
pdf = canvas.Canvas(buffer)
pdfmetrics.registerFont(UnicodeCIDFont("STSong-Light"))
pdf.setFont("STSong-Light", 12)
pdf.drawString(72, 760, "合同正文第一条 付款安排")
pdf.drawString(72, 740, "合同正文第二条 保密义务")
pdf.save()
extracted = self.service.extract_text_from_file(buffer.getvalue(), "contract.pdf")
self.assertIsNotNone(extracted)
assert extracted is not None
self.assertIn("付款安排", extracted)
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_pdf_text_success_has_no_failure_message(self):
self.service.extract_text_from_file = lambda file_bytes, file_name: "可复制 PDF 合同正文\n付款安排"
self.service._classify_pdf_source = lambda file_bytes: {
"kind": "text_pdf",
"page_count": 1,
"text_pages": 1,
"image_pages": 0,
}
result = self.service.extract_review_upload_text(b"%PDF-1.4", "contract.pdf")
self.assertEqual(result["text"], "可复制 PDF 合同正文\n付款安排")
self.assertIsNone(result["failure_message"])
self.assertEqual(result["metadata"]["extraction_mode"], "text")
self.assertEqual(result["metadata"]["ocr_gate_status"], "PASS")
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"])
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_extract_review_upload_text_uses_ocr_for_image_upload(self):
self.service._extract_text_from_image_with_ocr = lambda file_bytes: {
"text": "图片合同正文\n付款安排",
"failure_message": "已通过 OCR 提取图片文本，但识别质量一般，建议人工复核原图后再使用审查结果。",
"metadata": {
"document_kind": "image_upload",
"extraction_mode": "ocr",
"ocr_gate_status": "WARN",
"ocr_quality": {"status": "WARN", "char_count": 12},
},
}
result = self.service.extract_review_upload_text(b"\x89PNG\r\n\x1a\nmock-image", "contract.png")
self.assertEqual(result["text"], "图片合同正文\n付款安排")
self.assertEqual(result["metadata"]["document_kind"], "image_upload")
self.assertEqual(result["metadata"]["ocr_gate_status"], "WARN")
def test_validate_review_upload_accepts_supported_image_suffix(self):
self.service.validate_review_upload("contract.png", "image/png", b"\x89PNG\r\n\x1a\nmock-image")
self.service.validate_review_upload("contract.jpg", "image/jpeg", b"\xff\xd8\xffmock-jpeg")
@@ -993,33 +811,6 @@ class ReviewServiceTestCase(unittest.TestCase):
self.assertEqual(context.exception.status_code, 400)
self.assertEqual(context.exception.detail, "仅支持 PDF / DOCX / TXT / PNG / JPG / JPEG 文件")
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"],
backend/tests/test_shared_ai_platform_client.py
+37 -4
View File
@@ -10,11 +10,44 @@ from app.clients.shared_ai_platform import RemoteSharedAIPlatformClient
from app.db import Base
from app.models.conversation import Conversation
from app.models.message import Message
from shared_ai_platform_sdk.client import SharedAIPlatformClient as VendoredSharedAiPlatformClient
from shared_ai_platform_sdk.client import SharedAIPlatformClient as SharedAiPlatformSdkClient
from shared_ai_platform_sdk.types import RequestContext, SSEEvent
class VendoredSharedAiPlatformSdkTestCase(unittest.TestCase):
class SharedAiPlatformSdkTestCase(unittest.TestCase):
def test_stream_conversation_uses_default_v0_2_api_prefix(self):
seen_paths = []
def handler(request: httpx.Request) -> httpx.Response:
seen_paths.append(request.url.path)
return httpx.Response(
200,
headers={"content-type": "text/event-stream"},
content=b'data: {"type":"end","conversation_id":"conv-1"}\n\n',
)
transport = httpx.MockTransport(handler)
http_client = httpx.Client(base_url="http://testserver", transport=transport, timeout=30.0)
sdk = SharedAiPlatformSdkClient(
base_url="http://testserver",
api_key="test-key",
app_id="test-app",
client=http_client,
)
try:
events = list(
sdk.stream_conversation(
RequestContext(tenant_id="tenant-1", user_id="user-1", user_role="admin"),
message="hello",
)
)
finally:
http_client.close()
self.assertEqual(seen_paths, ["/api/v0.2/conversations/stream"])
self.assertEqual(events[0].payload["conversation_id"], "conv-1")
def test_stream_conversation_uses_configured_api_prefix(self):
seen_paths = []
@@ -28,7 +61,7 @@ class VendoredSharedAiPlatformSdkTestCase(unittest.TestCase):
transport = httpx.MockTransport(handler)
http_client = httpx.Client(base_url="http://testserver", transport=transport, timeout=30.0)
sdk = VendoredSharedAiPlatformClient(
sdk = SharedAiPlatformSdkClient(
base_url="http://testserver",
api_key="test-key",
app_id="test-app",
@@ -66,7 +99,7 @@ class VendoredSharedAiPlatformSdkTestCase(unittest.TestCase):
transport = httpx.MockTransport(handler)
http_client = httpx.Client(base_url="http://testserver", transport=transport, timeout=30.0)
sdk = VendoredSharedAiPlatformClient(
sdk = SharedAiPlatformSdkClient(
base_url="http://testserver",
api_key="test-key",
app_id="test-app",
backend/tests/test_system_api.py
+1 -1
View File
@@ -417,7 +417,7 @@ class SystemApiTestCase(unittest.TestCase):
def test_shared_ai_status_reports_remote_preferred_mode(self):
settings.shared_ai_platform_base_url = "http://127.0.0.1:8011"
settings.shared_ai_platform_api_key = "test-key"
settings.shared_ai_platform_api_prefix = "/api/v0.1"
settings.shared_ai_platform_api_prefix = "/api/v0.2"
settings.shared_ai_platform_app_id = "ai-contract-platform"
settings.shared_ai_platform_timeout_seconds = 45.0
settings.shared_ai_platform_legal_review_enabled = True
docker-compose.server.yml
+2 -2
View File
@@ -24,7 +24,7 @@ services:
MINIO_SECRET_KEY: ${MINIO_SECRET_KEY:?Set MINIO_SECRET_KEY in .env before starting production}
MINIO_BUCKET: ${MINIO_BUCKET:-contracts}
SHARED_AI_PLATFORM_BASE_URL: ${SHARED_AI_PLATFORM_BASE_URL:-http://192.168.3.90:18011}
SHARED_AI_PLATFORM_API_PREFIX: ${SHARED_AI_PLATFORM_API_PREFIX:-/api/v0.1}
SHARED_AI_PLATFORM_API_PREFIX: ${SHARED_AI_PLATFORM_API_PREFIX:-/api/v0.2}
SHARED_AI_PLATFORM_APP_ID: ${SHARED_AI_PLATFORM_APP_ID:-ai-contract-platform}
SHARED_AI_PLATFORM_API_KEY: ${SHARED_AI_PLATFORM_API_KEY:?Set SHARED_AI_PLATFORM_API_KEY in .env before starting production}
SHARED_AI_PLATFORM_TIMEOUT_SECONDS: ${SHARED_AI_PLATFORM_TIMEOUT_SECONDS:-180}
@@ -57,7 +57,7 @@ services:
MINIO_SECRET_KEY: ${MINIO_SECRET_KEY:?Set MINIO_SECRET_KEY in .env before starting production}
MINIO_BUCKET: ${MINIO_BUCKET:-contracts}
SHARED_AI_PLATFORM_BASE_URL: ${SHARED_AI_PLATFORM_BASE_URL:-http://192.168.3.90:18011}
SHARED_AI_PLATFORM_API_PREFIX: ${SHARED_AI_PLATFORM_API_PREFIX:-/api/v0.1}
SHARED_AI_PLATFORM_API_PREFIX: ${SHARED_AI_PLATFORM_API_PREFIX:-/api/v0.2}
SHARED_AI_PLATFORM_APP_ID: ${SHARED_AI_PLATFORM_APP_ID:-ai-contract-platform}
SHARED_AI_PLATFORM_API_KEY: ${SHARED_AI_PLATFORM_API_KEY:?Set SHARED_AI_PLATFORM_API_KEY in .env before starting production}
SHARED_AI_PLATFORM_TIMEOUT_SECONDS: ${SHARED_AI_PLATFORM_TIMEOUT_SECONDS:-180}
frontend/scripts/slice4-shared-ai-stub.mjs
+7 -1
View File
@@ -134,7 +134,13 @@ const server = http.createServer(async (request, response) => {
return;
}
if (request.method === "POST" && request.url === "/api/v0.1/domain-packs/legal-review/reviews/run") {
if (
request.method === "POST" &&
[
"/api/v0.2/domain-packs/legal-review/reviews/run",
"/api/v0.1/domain-packs/legal-review/reviews/run",
].includes(request.url)
) {
const body = await readJson(request);
sendJson(response, 200, { code: 0, data: buildReviewPayload(body) });
return;
infra/env/backend.3.90.env.example
Vendored
+2 -1
View File
@@ -8,7 +8,8 @@ MINIO_ACCESS_KEY=replace-with-minio-access-key
MINIO_SECRET_KEY=replace-with-minio-secret-key
MINIO_BUCKET=contracts
SHARED_AI_PLATFORM_BASE_URL=http://192.168.3.90:18011
SHARED_AI_PLATFORM_API_PREFIX=/api/v0.1
# Default shared-ai contract version is v0.2. Override to /api/v0.1 only for temporary compatibility with legacy deployments.
SHARED_AI_PLATFORM_API_PREFIX=/api/v0.2
SHARED_AI_PLATFORM_APP_ID=ai-contract-platform
SHARED_AI_PLATFORM_API_KEY=replace-with-shared-ai-platform-api-key
SHARED_AI_PLATFORM_TIMEOUT_SECONDS=180
scripts/run-phase3-closed-loop.ps1
+1 -1
View File
@@ -25,7 +25,7 @@ $sqliteDbUrl = ("sqlite+pysqlite:///" + (($sqliteDbPath -replace "\\", "/")))
$localStorageDir = Join-Path $runtimeDir "storage"
$sharedAiPort = 18011
$sharedAiBaseUrl = "http://127.0.0.1:$sharedAiPort"
$sharedAiApiPrefix = "/api/v0.1"
$sharedAiApiPrefix = "/api/v0.2"
$sharedAiAppId = "ai-contract-platform"
$sharedAiApiKey = "shared-ai-platform-admin-dev-token"