"""OCR upload orchestration, field extraction, and standards evaluation.""" import re from datetime import date from pathlib import Path from typing import Any from uuid import uuid4 from fastapi import UploadFile from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.core.config import get_settings from app.core.exceptions import AppException from app.models.entities import StandardRule from app.schemas.ocr import ( OcrBatchResult, OcrExtractedFields, OcrField, OcrFileResult, OcrMeasurement, OcrStandardJudgment, ) from app.services.baidu_ocr import BaiduOcrClient from app.services.llm_client import LLMClient _FIELD_ALIASES = { "product_name": ("产品名称", "产品名", "品名"), "product_category": ("产品品类", "品类"), "supplier": ("供应商", "供方"), "factory": ("加工厂", "生产工厂", "工厂"), "material_type": ("物料名称", "物料类型", "物料"), "stage_new": ("发生阶段", "发现阶段", "产品阶段", "阶段"), "description": ("异常描述", "问题描述", "不良描述", "异常现象"), "defect_level": ("缺陷等级", "异常等级", "等级"), } _MEASUREMENT_WORDS = ( "pH", "PH", "色差", "铅", "汞", "砷", "镉", "力矩", "喷出量", "附着力", "含量", "粘度", "黏度", "透光率", "菌落", "温度", "湿度", "不合格率", ) _NON_MEASUREMENT_NAMES = { "数量", "抽检数量", "产品规格", "规格", "容量", "日期", "生产日期", "发生日期", "单号", "订单号", "批次", "批号", "页码", } _UNIT_PATTERN = r"(?:mg/kg|μg/g|ug/g|g/L|mg/L|N[·.]?m|mPa[·.]?s|Pa[·.]?s|mm|cm|nm|%|级|℃|°C)?" class OcrService: def __init__( self, session: AsyncSession, *, client: BaiduOcrClient | None = None, llm: LLMClient | None = None, ) -> None: self.session = session self.client = client or BaiduOcrClient() self.llm = llm or LLMClient() self.settings = get_settings() async def recognize(self, files: list[UploadFile]) -> OcrBatchResult: if not files: raise AppException("请至少上传一个图片或 PDF", code=422, status_code=422) if len(files) > self.settings.ocr_max_batch_files: raise AppException( f"一次最多上传 {self.settings.ocr_max_batch_files} 个文件", code=422, status_code=422, ) rules = list( ( await self.session.scalars( select(StandardRule).where(StandardRule.status == "启用").order_by(StandardRule.id) ) ).all() ) results: list[OcrFileResult] = [] try: for upload in files: results.append(await self._recognize_one(upload, rules)) except Exception: for result in results: self.upload_path(result.file_id).unlink(missing_ok=True) raise fields = self._merge_fields([item.fields for item in results]) measurements = [measurement for item in results for measurement in item.measurements] judgments = [judgment for item in results for judgment in item.judgments] return OcrBatchResult( files=results, fields=fields, measurements=measurements, judgments=judgments, requires_review=any(item.requires_review for item in results), ) async def judge(self, measurements: list[OcrMeasurement]) -> list[OcrStandardJudgment]: rules = list( ( await self.session.scalars( select(StandardRule).where(StandardRule.status == "启用").order_by(StandardRule.id) ) ).all() ) return self._judge_measurements(measurements, rules) async def _recognize_one(self, upload: UploadFile, rules: list[StandardRule]) -> OcrFileResult: content = await self._read_limited(upload) content_type, suffix = self._detect_type(content) original_name = Path(upload.filename or f"upload{suffix}").name file_id = f"{uuid4().hex}{suffix}" target = self.upload_path(file_id) target.parent.mkdir(parents=True, exist_ok=True) target.write_bytes(content) try: if content_type == "application/pdf": raw_text, provider_confidence, page_count = await self.client.recognize_document(content, original_name) engine = "paddle-vl-parser" else: raw_text, provider_confidence, page_count = await self.client.recognize_image(content) engine = "webimage" extracted = await self._extract(raw_text, provider_confidence) fields, measurements, extraction_confidence = extracted judgments = self._judge_measurements(measurements, rules) warnings = self._warnings(fields, measurements, judgments, provider_confidence) requires_review = bool(warnings) or any(field.needs_review for field in self._field_values(fields)) return OcrFileResult( file_id=file_id, file_name=original_name, content_type=content_type, engine=engine, raw_text=raw_text, confidence=provider_confidence if provider_confidence is not None else extraction_confidence, page_count=page_count, fields=fields, measurements=measurements, judgments=judgments, warnings=warnings, requires_review=requires_review, ) except Exception: target.unlink(missing_ok=True) raise finally: await upload.close() async def _extract( self, text: str, provider_confidence: float | None, ) -> tuple[OcrExtractedFields, list[OcrMeasurement], float | None]: fallback_fields = self._extract_fields_by_labels(text, provider_confidence) fallback_measurements = self._extract_measurements(text, provider_confidence) if not self.llm.configured: return fallback_fields, fallback_measurements, provider_confidence try: data = await self.llm.extract_ocr_fields(text) except Exception: # noqa: BLE001 - OCR remains usable when the optional entity extractor fails return fallback_fields, fallback_measurements, provider_confidence confidence = self._as_confidence(data.get("confidence")) fields = self._fields_from_llm(data, confidence, fallback_fields) measurements = self._measurements_from_llm(data.get("measurements"), text, confidence) return fields, measurements or fallback_measurements, confidence def _fields_from_llm( self, data: dict[str, Any], confidence: float | None, fallback: OcrExtractedFields, ) -> OcrExtractedFields: values: dict[str, Any] = {} for key in _FIELD_ALIASES: value = data.get(key) if key == "defect_level" and value: value = str(value).upper().replace("级", "") if value not in {"A", "B", "C"}: value = None if value is None or not str(value).strip(): values[key] = getattr(fallback, key) else: values[key] = self._field(str(value).strip(), confidence) occurred_on = self._as_date(data.get("occurred_on")) or fallback.occurred_on return OcrExtractedFields(**values, occurred_on=occurred_on) def _measurements_from_llm( self, raw: Any, source_text: str, default_confidence: float | None, ) -> list[OcrMeasurement]: if not isinstance(raw, list): return [] results: list[OcrMeasurement] = [] for item in raw[:50]: if not isinstance(item, dict): continue name = str(item.get("item") or "").strip() raw_value = str(item.get("raw_value") or item.get("value") or "").strip() if ( not name or self._normalize(name) in {self._normalize(value) for value in _NON_MEASUREMENT_NAMES} or not raw_value or not self._numeric_appears(raw_value, source_text) ): continue value = self._as_float(item.get("value")) if value is None: value = self._as_float(raw_value) confidence = self._as_confidence(item.get("confidence")) or default_confidence results.append( OcrMeasurement( item=name[:100], raw_value=raw_value[:100], value=value, unit=str(item.get("unit") or "").strip()[:30] or self._unit_from(raw_value), confidence=confidence, needs_review=confidence is None or confidence < self.settings.ocr_judgment_confidence, ) ) return results def _extract_fields_by_labels(self, text: str, confidence: float | None) -> OcrExtractedFields: values: dict[str, OcrField] = {} for field_name, aliases in _FIELD_ALIASES.items(): found: str | None = None for alias in aliases: match = re.search(rf"{re.escape(alias)}\s*[::]?\s*([^\n||]{{1,120}})", text, re.IGNORECASE) if match: found = match.group(1).strip(" ::;;") break if field_name == "defect_level" and found: level = re.search(r"\b([ABC])\s*级?\b", found, re.IGNORECASE) found = level.group(1).upper() if level else None values[field_name] = self._field(found, confidence) occurred = None date_match = re.search(r"(?:发生日期|日期)\s*[::]?\s*(20\d{2})[-/.年](\d{1,2})[-/.月](\d{1,2})", text) if date_match: occurred = self._as_date("-".join(date_match.groups())) return OcrExtractedFields(**values, occurred_on=occurred) def _extract_measurements(self, text: str, confidence: float | None) -> list[OcrMeasurement]: results: list[OcrMeasurement] = [] seen: set[tuple[str, str]] = set() pattern = re.compile( rf"(?P[A-Za-z\u4e00-\u9fffΔδ()()·/\-]{{1,32}}?)\s*[::]?\s*" rf"(?P[+\-]?\d+(?:\.\d+)?)\s*(?P{_UNIT_PATTERN})", re.IGNORECASE, ) for line in text.splitlines(): if not any(word.lower() in line.lower() for word in _MEASUREMENT_WORDS): continue for match in pattern.finditer(line): name = match.group("item").strip(" ::;;-||") raw_value = f"{match.group('value')}{match.group('unit') or ''}" key = (name, raw_value) if not name or key in seen: continue seen.add(key) results.append( OcrMeasurement( item=name[:100], raw_value=raw_value, value=float(match.group("value")), unit=match.group("unit") or None, confidence=confidence, needs_review=confidence is None or confidence < self.settings.ocr_judgment_confidence, ) ) return results[:50] def _judge_measurements( self, measurements: list[OcrMeasurement], rules: list[StandardRule], ) -> list[OcrStandardJudgment]: judgments: list[OcrStandardJudgment] = [] for measurement in measurements: rule = self._best_rule(measurement.item, rules) if rule is None: judgments.append( OcrStandardJudgment( measurement_item=measurement.item, measured_value=measurement.raw_value, result="未匹配标准", reason="标准库中没有找到对应检测项", ) ) continue result, reason = self.evaluate_standard(measurement, rule.judge_standard) if measurement.needs_review and result in {"合格", "不合格"}: result = "待人工确认" reason = ( f"识别置信度低于 {self.settings.ocr_judgment_confidence:.2f}," "需确认检测值后再采用自动判定" ) judgments.append( OcrStandardJudgment( measurement_item=measurement.item, measured_value=measurement.raw_value, standard_rule_id=rule.id, risk_point=rule.risk_point, standard_value=rule.judge_standard, result=result, basis=rule.regulation, reason=reason, ) ) return judgments @staticmethod def _best_rule(item: str, rules: list[StandardRule]) -> StandardRule | None: needle = OcrService._normalize(item) if not needle: return None scored: list[tuple[float, StandardRule]] = [] for rule in rules: haystack = OcrService._normalize(f"{rule.risk_point} {rule.judge_standard}") if needle in haystack or haystack in needle: score = 2.0 + min(len(needle), len(haystack)) / 100 else: left = set(OcrService._bigrams(needle)) right = set(OcrService._bigrams(haystack)) score = len(left & right) / max(1, len(left | right)) threshold = 0.45 if len(needle) <= 4 else 0.3 if score >= threshold: scored.append((score, rule)) return max(scored, key=lambda pair: pair[0])[1] if scored else None @staticmethod def evaluate_standard(measurement: OcrMeasurement, standard: str) -> tuple[str, str | None]: if measurement.value is None: return "待人工确认", "检测值不是可比较的数字" normalized = standard.replace("~", "~").replace("—", "-").replace("至", "~") range_match = re.search(r"(-?\d+(?:\.\d+)?)\s*(?:~|-)\s*(-?\d+(?:\.\d+)?)", normalized) if range_match: mismatch = OcrService._unit_mismatch(measurement.unit, normalized[range_match.end() :]) if mismatch: return "待人工确认", mismatch lower, upper = map(float, range_match.groups()) passed = lower <= measurement.value <= upper return ("合格" if passed else "不合格"), None comparator = re.search(r"(<=|>=|≤|≥|<|>)\s*(?:±\s*)?(-?\d+(?:\.\d+)?)", normalized) if comparator: mismatch = OcrService._unit_mismatch(measurement.unit, normalized[comparator.end() :]) if mismatch: return "待人工确认", mismatch operator, raw_limit = comparator.groups() limit = float(raw_limit) value = abs(measurement.value) if "±" in comparator.group(0) else measurement.value passed = { "<=": value <= limit, "≤": value <= limit, ">=": value >= limit, "≥": value >= limit, "<": value < limit, ">": value > limit, }[operator] return ("合格" if passed else "不合格"), None return "待人工确认", "该标准不是系统可自动计算的数值范围" def _warnings( self, fields: OcrExtractedFields, measurements: list[OcrMeasurement], judgments: list[OcrStandardJudgment], confidence: float | None, ) -> list[str]: warnings: list[str] = [] if confidence is None: warnings.append("服务未返回文字级置信度,请人工核对识别结果") elif confidence < self.settings.ocr_review_confidence: warnings.append("OCR 置信度低于 0.8,请重点核对标黄字段") if not fields.product_name.value: warnings.append("未识别到产品名称") if not fields.description.value: warnings.append("未识别到异常描述") if measurements and any(item.result == "不合格" for item in judgments): warnings.append("检测数值存在不合格项,请确认后提交审核") return warnings async def _read_limited(self, upload: UploadFile) -> bytes: limit = self.settings.ocr_max_file_mb * 1024 * 1024 content = await upload.read(limit + 1) if len(content) > limit: raise AppException(f"单个文件不能超过 {self.settings.ocr_max_file_mb}MB", code=413, status_code=413) if not content: raise AppException("不能上传空文件", code=422, status_code=422) return content @staticmethod def _detect_type(content: bytes) -> tuple[str, str]: if content.startswith(b"%PDF-"): return "application/pdf", ".pdf" if content.startswith(b"\x89PNG\r\n\x1a\n"): return "image/png", ".png" if content.startswith(b"\xff\xd8\xff"): return "image/jpeg", ".jpg" raise AppException("仅支持 JPG、PNG 或 PDF 文件", code=422, status_code=422) def upload_path(self, file_id: str) -> Path: if not re.fullmatch(r"[0-9a-f]{32}\.(?:jpg|png|pdf)", file_id): raise AppException("附件不存在", code=404, status_code=404) root = self.settings.ocr_upload_dir.resolve() return root / file_id @staticmethod def _field(value: str | None, confidence: float | None) -> OcrField: threshold = get_settings().ocr_review_confidence return OcrField( value=value, confidence=confidence, needs_review=not value or confidence is None or confidence < threshold, ) @staticmethod def _field_values(fields: OcrExtractedFields) -> list[OcrField]: return [getattr(fields, key) for key in _FIELD_ALIASES] @staticmethod def _merge_fields(items: list[OcrExtractedFields]) -> OcrExtractedFields: merged: dict[str, Any] = {} for key in _FIELD_ALIASES: candidates = [getattr(item, key) for item in items if getattr(item, key).value] merged[key] = max(candidates, key=lambda field: field.confidence or -1) if candidates else OcrField() merged["occurred_on"] = next((item.occurred_on for item in items if item.occurred_on), None) return OcrExtractedFields(**merged) @staticmethod def _as_confidence(value: Any) -> float | None: try: number = float(value) except (TypeError, ValueError): return None return min(1.0, max(0.0, number)) @staticmethod def _as_float(value: Any) -> float | None: match = re.search(r"[+\-]?\d+(?:\.\d+)?", str(value or "")) return float(match.group()) if match else None @staticmethod def _as_date(value: Any) -> date | None: if not value: return None try: return date.fromisoformat(str(value).strip()[:10]) except ValueError: return None @staticmethod def _unit_from(value: str) -> str | None: match = re.search(_UNIT_PATTERN + r"$", value, re.IGNORECASE) return match.group() if match and match.group() else None @staticmethod def _unit_mismatch(measured_unit: str | None, standard_tail: str) -> str | None: if not measured_unit: return None standard_unit = re.match(rf"\s*({_UNIT_PATTERN})", standard_tail, re.IGNORECASE) expected = standard_unit.group(1) if standard_unit and standard_unit.group(1) else None if not expected: return None def normalize(unit: str) -> str: return unit.lower().replace(".", "·").replace("°c", "℃") if normalize(measured_unit) != normalize(expected): return f"检测值单位 {measured_unit} 与标准单位 {expected} 不一致" return None @staticmethod def _numeric_appears(value: str, text: str) -> bool: number = re.search(r"[+\-]?\d+(?:\.\d+)?", value) return bool(number and number.group() in text) @staticmethod def _normalize(value: str) -> str: return re.sub(r"[^a-z0-9\u4e00-\u9fffΔδ]", "", value.lower()) @staticmethod def _bigrams(value: str) -> list[str]: return [value[index : index + 2] for index in range(max(1, len(value) - 1))]