Files
guoju0808/ry-ocr/app/services/invoice_extractor.py
T
郭庆泰 c3eb8ed9c3 feat: OCR 服务 + 会议材料模块
ry-ocr/ (新)
  本地发票识别微服务 (PaddleOCR 3.x + FastAPI, 8801)
  - QR 优先: 扫到二维码即取开票时间/发票号/金额; 没扫到/格式不合法直接判非发票, 不跑 OCR
  - 配置 QR_FULL_OCR 控制快路径(false, 0.2s)还是全字段(true, 4.5s)
  - /recognize/invoice (multipart) + /recognize/invoice/by-path (本地路径, 白名单) + /recognize/text
  - is_invoice / from_qr / qr_raw / qr_error / error_code 字段
  - 12 字段发票抽取 (regex + 启发式, 左右主体识别)
  - 超时保护 (15s 单页 / 60s 总流程) + PaddleOCR 单例 + ThreadPoolExecutor

ry-api/ruoyi-business/
  - pom.xml: 加 hutool-http/json/core 5.8.27, lombok 1.18.30 (OcrClient @Slf4j 所需)
  - ocr/: OcrClient + InvoiceResult/Fields/Line + ZipExtractor + InvoiceOcrScheduler
  - oss/: OssUploader + OssConfMeta (OCR 识别后重传 OSS)
  - config/: OcrConfig + OcrExecutorConfig (后台线程池)
  - service/impl/InvoiceOcrService: 后台提交 OCR, ZIP 路径解压识别, 替换场景先清旧
  - 会议材料 CRUD 全套 (BizMeetingAuditLog/Executor/Invoice/Material/Supervisor):
    controller + service + mapper + domain + xml

ry-vue3/
  - MeetingDetail.vue (新建): 会议详情页 (含评分维度章节, 改只读)
  - Meetings.vue / OssFileUploader.vue / router / Login.vue: 适配新字段

ry-api/ruoyi-admin/
  - RuoYiApplication.java + application.yml: 启用 @Async 异步支持

_self/
  - manager_meetings.md / manager_meeting_detail.md: 文档
2026-08-22 00:23:22 +08:00

299 lines
10 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""从 OCR 文本/行里抽取发票字段
适配中国大陆 增值税发票(电子普票 / 专票 / 电子专票 / 数电票)
关键策略:
- 主体(销售方/购买方)按 OCR box **坐标**判断归属(左右两栏)
- 名称提取加 stop word,避免 OCR 单行文本混淆
"""
from __future__ import annotations
import re
from typing import List, Optional, Tuple
from app.models import InvoiceFields, OCRLine
from app.utils import (
amount_consistent,
extract_cn_amount,
extract_num_amount,
extract_pretax_amount,
extract_tax_amount,
extract_total_amount,
)
# ---------- 发票类型 ----------
_INVOICE_TYPES = [
"增值税电子专用发票",
"增值税电子普通发票",
"增值税专用发票",
"增值税普通发票",
"通用机打发票",
"数电票",
"电子发票",
]
# ---------- 发票号码 ----------
_NO_PATTERN = re.compile(
r"(?:发\s*票\s*号\s*码|号\s*码|No\.?|号)\s*[:]?\s*(\d{8,20})",
re.IGNORECASE,
)
# ---------- 发票代码 ----------
_CODE_PATTERN = re.compile(
r"(?:发\s*票\s*代\s*码|代\s*码)\s*[:]?\s*(\d{10,12}|\d{8,12})",
)
# ---------- 开票日期 ----------
_DATE_PATTERN = re.compile(
r"(?:开\s*票\s*日\s*期|日\s*期)\s*[:]?\s*"
r"(\d{4})\s*[年/\.]\s*(\d{1,2})\s*[月/\.]\s*(\d{1,2})",
)
# ---------- 纳税人识别号(必须至少含 1 个字母,排除纯数字发票号)----------
_TAX_NO_PATTERN = re.compile(r"((?=[0-9A-Z]*[A-Z])[0-9A-Z]{18})")
# ---------- 主体标签 ----------
_BUYER_LABEL = re.compile(r"购\s*买\s*方\s*(?:信\s*息|名\s*称|)")
_SELLER_LABEL = re.compile(r"销\s*售\s*方\s*(?:信\s*息|名\s*称|)")
# ---------- 名称(带 stop word 截断)----------
_NAME_STOP = r"(?:销售方|购买方|统一社会信用|纳税人|项目名称|规格型号|^单位$|^数量$|^单价$|^金额|^税率|^税额|备注|收款人|复核|开票人|价税合计|小写|大写)"
_NAME_PATTERN = re.compile(
rf"名\s*称\s*[:]\s*"
rf"((?:(?!{_NAME_STOP})[^\n\r]){{2,60}}?(?:公司|商店|厂|店|部|中心|工作室))"
)
# ---------- 工具函数 ----------
def _norm(text: str) -> str:
return re.sub(r"\s+", " ", text or "").strip()
def _box_center(box) -> Tuple[float, float]:
"""box: [[x1,y1], [x2,y2], [x3,y3], [x4,y4]] → (cx, cy)"""
if not box or len(box) < 4:
return (0.0, 0.0)
xs = [p[0] for p in box]
ys = [p[1] for p in box]
return ((min(xs) + max(xs)) / 2, (min(ys) + max(ys)) / 2)
def _detect_invoice_type(text: str) -> Optional[str]:
for t in _INVOICE_TYPES:
if t in text:
return t
return None
def _extract_invoice_no(text: str) -> Optional[str]:
m = _NO_PATTERN.search(text)
return m.group(1) if m else None
def _extract_invoice_code(text: str) -> Optional[str]:
m = _CODE_PATTERN.search(text)
return m.group(1) if m else None
def _extract_date(text: str) -> Optional[str]:
m = _DATE_PATTERN.search(text)
if not m:
return None
y, mo, d = m.groups()
return f"{int(y):04d}-{int(mo):02d}-{int(d):02d}"
def _clean_name(name: str) -> str:
name = re.sub(r"^[\s:,,。、]+", "", name)
name = name.split("纳税人")[0].split("统一社会")[0]
name = re.sub(r"[^一-龥A-Za-z0-9()()·\-]", "", name)
return name.rstrip(":;,,。、 ").strip()
def _extract_parties_from_lines(lines: List[OCRLine]) -> tuple:
"""基于 box 坐标的主体识别
1) 找到 "购买方信息"/"销售方信息" 标签所在位置(box 中心)
2) 把每个 "名称: xxx" 行按 x 坐标归属到对应栏
3) 兜底:左右栏无法区分时按 y 坐标(top/bottom
"""
# 找标签坐标
buyer_label_box = None
seller_label_box = None
for line in lines:
if _BUYER_LABEL.search(line.text) and not buyer_label_box:
buyer_label_box = line.box
if _SELLER_LABEL.search(line.text) and not seller_label_box:
seller_label_box = line.box
by_x = buyer_label_box is not None and seller_label_box is not None and \
abs(_box_center(buyer_label_box)[0] - _box_center(seller_label_box)[0]) > 50
# 找所有 "名称: ..." 行的 box
name_lines = []
for line in lines:
m = _NAME_PATTERN.search(line.text)
if m:
name_lines.append((line, _clean_name(m.group(1))))
# 找所有税号
tax_lines = []
for line in lines:
m = _TAX_NO_PATTERN.search(line.text)
if m:
tax_lines.append((line, m.group(1)))
seller_name = buyer_name = None
seller_tax = buyer_tax = None
if by_x and buyer_label_box and seller_label_box:
# 左右栏布局:用 x 坐标归属
bx, _ = _box_center(buyer_label_box)
sx, _ = _box_center(seller_label_box)
mid = (bx + sx) / 2
for line, name in name_lines:
cx, _ = _box_center(line.box)
if cx < mid and not buyer_name:
buyer_name = name
elif cx >= mid and not seller_name:
seller_name = name
for line, tax in tax_lines:
cx, _ = _box_center(line.box)
if cx < mid and not buyer_tax:
buyer_tax = tax
elif cx >= mid and not seller_tax:
seller_tax = tax
else:
# 上下栏 / 单栏:按 y 坐标(第一个 = 购买方)
name_lines.sort(key=lambda x: (_box_center(x[0].box)[1], _box_center(x[0].box)[0]))
tax_lines.sort(key=lambda x: (_box_center(x[0].box)[1], _box_center(x[0].box)[0]))
if len(name_lines) >= 1:
buyer_name = name_lines[0][1]
if len(name_lines) >= 2:
seller_name = name_lines[1][1]
if len(tax_lines) >= 1:
buyer_tax = tax_lines[0][1]
if len(tax_lines) >= 2:
seller_tax = tax_lines[1][1]
return seller_name, seller_tax, buyer_name, buyer_tax
def _extract_parties_from_text(text: str) -> tuple:
"""无 box 信息时的 fallback:按显式标签分段"""
seller_name = buyer_name = None
seller_tax = buyer_tax = None
for m in _BUYER_LABEL.finditer(text):
chunk = text[m.end():m.end() + 200]
nm = _NAME_PATTERN.search(chunk)
tm = _TAX_NO_PATTERN.search(chunk[:100])
if nm and not buyer_name:
buyer_name = _clean_name(nm.group(1))
if tm and not buyer_tax:
buyer_tax = tm.group(1)
if buyer_name and buyer_tax:
break
for m in _SELLER_LABEL.finditer(text):
chunk = text[m.end():m.end() + 200]
nm = _NAME_PATTERN.search(chunk)
tm = _TAX_NO_PATTERN.search(chunk[:100])
if nm and not seller_name:
seller_name = _clean_name(nm.group(1))
if tm and not seller_tax:
seller_tax = tm.group(1)
if seller_name and seller_tax:
break
if not buyer_name:
name_positions = [(m.start(), m.group(1)) for m in _NAME_PATTERN.finditer(text)]
if len(name_positions) >= 1:
buyer_name = _clean_name(name_positions[0][1])
if len(name_positions) >= 2:
seller_name = _clean_name(name_positions[1][1])
tax_positions = _TAX_NO_PATTERN.findall(text)
if not buyer_tax and len(tax_positions) >= 1:
buyer_tax = tax_positions[0]
if not seller_tax and len(tax_positions) >= 2:
seller_tax = tax_positions[1]
return seller_name, seller_tax, buyer_name, buyer_tax
def extract(text: str, lines: Optional[List[OCRLine]] = None) -> InvoiceFields:
raw = _norm(text)
fields = InvoiceFields()
fields.invoice_type = _detect_invoice_type(raw)
fields.invoice_code = _extract_invoice_code(raw)
fields.invoice_no = _extract_invoice_no(raw)
fields.invoice_date = _extract_date(raw)
# 金额
total = extract_total_amount(raw)
tax = extract_tax_amount(raw)
pretax = extract_pretax_amount(raw)
# 兜底:从所有 .xx 数字中找满足 tax + pretax = total 的组合
if (tax is None or pretax is None) and total is not None:
candidates = sorted(
{float(x) for x in re.findall(r"(\d+\.\d{2})", raw) if float(x) < total},
reverse=True,
)
for i, a in enumerate(candidates):
for b in candidates[i + 1:]:
if abs(a + b - total) < 0.011:
if pretax is None:
pretax = round(a, 2)
if tax is None:
tax = round(b, 2)
break
if tax is not None and pretax is not None:
break
# 单数字兜底:如果 total 不在候选里,只剩一个候选时
if (tax is None or pretax is None) and len(candidates) == 1:
only = round(candidates[0], 2)
if pretax is None and tax is None:
pretax = only
tax = round(total - only, 2)
elif tax is None:
tax = only
elif pretax is None:
pretax = only
# 二次兜底:合计 - 任一 = 另一
if tax is None and total is not None and pretax is not None:
tax = round(total - pretax, 2)
if pretax is None and total is not None and tax is not None:
pretax = round(total - tax, 2)
fields.amount = total
fields.tax_amount = tax
fields.amount_pretax = pretax
fields.amount_cn = extract_cn_amount(raw)
fields.amount_match = amount_consistent(fields.amount_cn, fields.amount)
# 主体
if lines:
seller_name, seller_tax, buyer_name, buyer_tax = _extract_parties_from_lines(lines)
if not (seller_name and buyer_name):
sn, st, bn, bt = _extract_parties_from_text(raw)
seller_name = seller_name or sn
buyer_name = buyer_name or bn
seller_tax = seller_tax or st
buyer_tax = buyer_tax or bt
else:
seller_name, seller_tax, buyer_name, buyer_tax = _extract_parties_from_text(raw)
fields.seller_name = seller_name
fields.seller_tax_no = seller_tax
fields.buyer_name = buyer_name
fields.buyer_tax_no = buyer_tax
return fields