from __future__ import annotations import logging import re from dataclasses import dataclass, field from datetime import date from decimal import Decimal, InvalidOperation logger = logging.getLogger(__name__) @dataclass(frozen=True) class ParsedInvoice: invoice_code: str = field(default="") invoice_number: str = field(default="") invoice_date: date | None = field(default=None) amount: Decimal | None = field(default=None) tax_amount: Decimal | None = field(default=None) total_amount: Decimal | None = field(default=None) buyer_name: str = field(default="") seller_name: str = field(default="") project_name: str = field(default="") category: str = field(default="other") class InvoiceParser: """通过正则表达式和关键词匹配从 OCR/PDF 文本中提取结构化发票字段""" _CODE_PATTERN: re.Pattern[str] = re.compile(r"(?:发票代码|No\.)[^\d]*(\d{10,12})") _NUMBER_PATTERN: re.Pattern[str] = re.compile(r"(?:发票号码|No\.)[^\d]*(\d{20}|\d{8})(?!\d)") _PROJECT_PATTERN: re.Pattern[str] = re.compile(r"\*[^*]+\*([^\s\n\r]{2,50})") _DATE_CN_PATTERN: re.Pattern[str] = re.compile(r"(\d{4})年(\d{1,2})月(\d{1,2})日") _DATE_ISO_PATTERN: re.Pattern[str] = re.compile(r"(\d{4})-(\d{2})-(\d{2})") _AMOUNT_PATTERN: re.Pattern[str] = re.compile(r"合\s*计\s*[¥￥]([\d,]+\.\d{2})\s*[¥￥]([\d,]+\.\d{2})") _TOTAL_PATTERN: re.Pattern[str] = re.compile(r"[（(]小写[）)]\s*[¥￥]([\d,]+\.\d{2})") _BUYER_PATTERN: re.Pattern[str] = re.compile( r"(?:购买方名称|购方名称)[：:]\s*([^\n\r]{2,50})" r"|(?:购\s*名称|买\s*名称)[：:]\s*([^销\n\r]{2,50})" r"|名称[：:]\s*([^\n\r]{2,50})" ) _SELLER_PATTERN: re.Pattern[str] = re.compile( r"(?:销售方名称|销方名称)[：:]\s*([^\n\r]{2,50})" r"|销\s*名称[：:]\s*([^\n\r]{2,50})" ) _CATEGORY_KEYWORDS: list[tuple[str, str]] = [ ("增值税专用发票", "vat_special"), ("增值税电子普通发票", "vat_electronic"), ("电子普通发票", "vat_electronic"), ("增值税普通发票", "vat_normal"), ("机动车销售统一发票", "vehicle_sales"), ("铁路电子客票", "train_ticket"), ("火车票", "train_ticket"), ("出租车发票", "taxi_receipt"), ("出租车", "taxi_receipt"), ("定额发票", "quota_invoice"), ("航空运输电子客票", "air_itinerary"), ("飞机行程单", "air_itinerary"), ("过路费", "toll_receipt"), ("通行费", "toll_receipt"), ] def parse(self, raw_text: str) -> ParsedInvoice: """从原始文本提取结构化字段""" invoice_code = self._extract_invoice_code(raw_text) invoice_number = self._extract_invoice_number(raw_text) invoice_date = self._extract_date(raw_text) amount = self._extract_amount(raw_text) tax_amount = self._extract_tax_amount(raw_text) total_amount = self._extract_total_amount(raw_text) buyer_name = self._extract_buyer_name(raw_text) seller_name = self._extract_seller_name(raw_text) project_name = self._extract_project_name(raw_text) category = self.detect_category(raw_text) logger.info( "解析发票完成: code=%s, number=%s, category=%s", invoice_code, invoice_number, category, ) return ParsedInvoice( invoice_code=invoice_code, invoice_number=invoice_number, invoice_date=invoice_date, amount=amount, tax_amount=tax_amount, total_amount=total_amount, buyer_name=buyer_name, seller_name=seller_name, project_name=project_name, category=category, ) def detect_category(self, raw_text: str) -> str: """根据关键词判定发票类目，无法判定时返回 'other'""" for keyword, category in self._CATEGORY_KEYWORDS: if keyword in raw_text: return category return "other" def format_to_text(self, parsed: ParsedInvoice) -> str: """将结构化数据格式化为标准文本（用于 round-trip 验证）""" lines: list[str] = [] lines.append(f"发票代码:{parsed.invoice_code}") lines.append(f"发票号码:{parsed.invoice_number}") if parsed.invoice_date is not None: lines.append( f"开票日期:{parsed.invoice_date.year}年{parsed.invoice_date.month:02d}月{parsed.invoice_date.day:02d}日" ) else: lines.append("开票日期:") lines.append(f"金额:{parsed.amount:.2f}" if parsed.amount is not None else "金额:") lines.append(f"税额:{parsed.tax_amount:.2f}" if parsed.tax_amount is not None else "税额:") lines.append(f"价税合计:{parsed.total_amount:.2f}" if parsed.total_amount is not None else "价税合计:") lines.append(f"购买方名称:{parsed.buyer_name}") lines.append(f"销售方名称:{parsed.seller_name}") lines.append(f"发票类目:{parsed.category}") return "\n".join(lines) def _extract_invoice_code(self, text: str) -> str: m = self._CODE_PATTERN.search(text) return m.group(1) if m else "" def _extract_invoice_number(self, text: str) -> str: m = self._NUMBER_PATTERN.search(text) return m.group(1) if m else "" def _extract_date(self, text: str) -> date | None: m = self._DATE_CN_PATTERN.search(text) if m: try: return date(int(m.group(1)), int(m.group(2)), int(m.group(3))) except ValueError: logger.warning("无效日期（中文格式）: %s", m.group(0)) m = self._DATE_ISO_PATTERN.search(text) if m: try: return date(int(m.group(1)), int(m.group(2)), int(m.group(3))) except ValueError: logger.warning("无效日期（ISO格式）: %s", m.group(0)) return None def _parse_decimal(self, value: str) -> Decimal | None: try: return Decimal(value.replace(",", "")) except InvalidOperation: logger.warning("无法解析金额: %s", value) return None def _extract_amount(self, text: str) -> Decimal | None: m = self._AMOUNT_PATTERN.search(text) return self._parse_decimal(m.group(1)) if m else None def _extract_tax_amount(self, text: str) -> Decimal | None: m = self._AMOUNT_PATTERN.search(text) return self._parse_decimal(m.group(2)) if m else None def _extract_total_amount(self, text: str) -> Decimal | None: m = self._TOTAL_PATTERN.search(text) return self._parse_decimal(m.group(1)) if m else None def _extract_buyer_name(self, text: str) -> str: for pattern in [ r"购买方名称[：:]\s*([^\n\r]{2,50})", r"购方名称[：:]\s*([^\n\r]{2,50})", r"(?:购\s*名称|买\s*名称)[：:]\s*([^\n\r销]{2,40})", r"名称[：:]\s*([^\n\r]{2,50})", ]: m = re.search(pattern, text) if m: return m.group(1).strip() return "" def _extract_seller_name(self, text: str) -> str: for pattern in [ r"销售方名称[：:]\s*([^\n\r]{2,50})", r"销方名称[：:]\s*([^\n\r]{2,50})", r"销\s*名称[：:]\s*([^\n\r]{2,50})", ]: m = re.search(pattern, text) if m: return m.group(1).strip() return "" def _extract_project_name(self, text: str) -> str: m = self._PROJECT_PATTERN.search(text) return m.group(1).strip() if m else ""