#!/usr/bin/env python3 # -*- coding: utf-8 -*- """升级 invoice_parse.py： ① 修复 ZIP 补丁未落地（上次 .replace 多传了参数 → TypeError 静默失败） ② PDF 解析改为「文本层 + macOS Vision OCR 视觉顺序」双通道 + 更强正则 真实数电票 PDF 文本流是「标签在前、值在后」的乱序（实测 48/57 缺销方名称）， OCR 读的是视觉顺序（标签紧邻值）→ 命中率大幅提升。 """ import ast import os P = '脚本/invoice_parse.py' s = open(P, encoding='utf-8').read() # ── ① ZIP 解包（正确写法：replace 只传 2 个参数）── if 'expand_zip' not in s: anchor = 'def prefer_best_formats(atts):' assert anchor in s, '未找到 prefer_best_formats' block = '''def expand_zip(name, payload, keep=('.pdf',)): """ZIP 附件解包：打车/出租车发票常以 ZIP 发来（内含发票+行程单）。 解出的文件以『压缩包名_内部文件名』命名，保留来源可追溯；只取 keep 内格式。""" import io import zipfile out = [] try: zf = zipfile.ZipFile(io.BytesIO(payload)) stem = os.path.splitext(name)[0] for zi in zf.infolist(): if zi.is_dir(): continue base = safe_name(os.path.basename(zi.filename)) if not base.lower().endswith(keep): continue out.append(('%s_%s' % (stem, base), zf.read(zi))) except Exception as e: print(' ⚠️ ZIP 解包失败 %s：%s' % (name, str(e)[:50])) return out ''' s = s.replace(anchor, block + anchor) # ZIP 展开：在抓取循环里（fetch 属于同一模块） loop_old = """ for fname, payload in mail_atts: if not fname.lower().endswith(ext_ok): skipped.append('%s（%s）' % (fname, subj[:22])) continue""" loop_new = """ expanded = [] for _n, _p in mail_atts: if _n.lower().endswith('.zip'): inner = expand_zip(_n, _p) if inner: expanded += inner print(' 📦 %s → 解出 %d 个 PDF' % (_n, len(inner))) else: skipped.append('%s（ZIP 内无 PDF，%s）' % (_n, subj[:18])) else: expanded.append((_n, _p)) for fname, payload in expanded: if not fname.lower().endswith(ext_ok): skipped.append('%s（%s）' % (fname, subj[:22])) continue""" if loop_old in s: s = s.replace(loop_old, loop_new) # ── ② OCR 二进制（编译一次，后续调用快）── if 'OCR_BIN' not in s: ins = '''OCR_BIN = '/tmp/baoyu_ocr_pdf' OCR_SWIFT = os.path.expanduser( '~/.hermes/skills/macos/macos-vision-image-processing/scripts/ocr_pdf.swift') def ensure_ocr_bin(): """编译 macOS Vision 的 PDF OCR（一次编译，后续直接调用）""" if os.path.exists(OCR_BIN): return OCR_BIN if not os.path.exists(OCR_SWIFT): return None try: r = subprocess.run(['swiftc', '-O', '-o', OCR_BIN, OCR_SWIFT], capture_output=True, text=True, timeout=240) if os.path.exists(OCR_BIN): return OCR_BIN print(' ⚠️ OCR 编译失败：%s' % (r.stderr or '')[:80]) except Exception as e: print(' ⚠️ OCR 编译异常：%s' % str(e)[:60]) return None def ocr_pdf_text(path): """用 macOS Vision 逐页 OCR（视觉顺序），失败返回空串""" b = ensure_ocr_bin() if not b: return '' try: r = subprocess.run([b, path], capture_output=True, text=True, timeout=180) t = r.stdout or '' if 'TOTAL_PAGES=' in t: t = t.split('===', 1)[0] + t # 保留正文 return t except Exception: return '' ''' s = s.replace('def _num(s):', ins + 'def _num(s):') # ── ③ 正则升级（视觉顺序友好）── old_pats = s[s.index('PDF_PATTERNS = {'):s.index('def _pdf_text(path):')] new_pats = '''PDF_PATTERNS = { '发票号码': [r'发票号码\\s*[：:]?\\s*([0-9]{15,25})', r'\\b([0-9]{20})\\b'], '发票代码': [r'发票代码\\s*[：:]?\\s*([0-9]{10,12})'], '开票日期': [r'开票日期\\s*[：:]?\\s*(20\\d{2}\\s*年\\s*\\d{1,2}\\s*月\\s*\\d{1,2}\\s*日)', r'开票日期\\s*[：:]?\\s*(20\\d{2}-\\d{2}-\\d{2})', r'(20\\d{2}\\s*年\\s*\\d{1,2}\\s*月\\s*\\d{1,2}\\s*日)'], '买方名称': [r'购\\s*买?\\s*方[\\s\\S]{0,24}?名\\s*称\\s*[：:]\\s*([^\\n，,]{2,40})', r'购\\s*买?\\s*方\\s*名\\s*称\\s*[：:]\\s*([^\\n]{2,40})'], '买方税号': [r'购\\s*买?\\s*方[\\s\\S]{0,90}?([0-9A-Z]{18})(?![0-9A-Z])'], '销方名称': [r'销\\s*售\\s*方[\\s\\S]{0,24}?名\\s*称\\s*[：:]\\s*([^\\n，,]{2,40})', r'销\\s*售\\s*方\\s*名\\s*称\\s*[：:]\\s*([^\\n]{2,40})'], '销方税号': [r'销\\s*售\\s*方[\\s\\S]{0,90}?([0-9A-Z]{18})(?![0-9A-Z])'], '价税合计': [r'价\\s*税\\s*合\\s*计[\\s\\S]{0,40}?[（(]\\s*小写\\s*[)）]\\s*[¥￥]?\\s*([\\d,]+\\.\\d{2})', r'[（(]\\s*小写\\s*[)）]\\s*[¥￥]?\\s*([\\d,]+\\.\\d{2})', r'价\\s*税\\s*合\\s*计[\\s\\S]{0,40}?[¥￥]\\s*([\\d,]+\\.\\d{2})', r'小\\s*写[\\s\\S]{0,12}?[¥￥]\\s*([\\d,]+\\.\\d{2})'], '金额': [r'合\\s*计[\\s\\S]{0,18}?[¥￥]?\\s*([\\d,]+\\.\\d{2})[\\s\\S]{0,24}?[¥￥]?\\s*[\\d,]+\\.\\d{2}', r'金\\s*额[\\s\\S]{0,30}?([\\d,]+\\.\\d{2})'], '税额': [r'合\\s*计[\\s\\S]{0,40}?[¥￥]?\\s*[\\d,]+\\.\\d{2}[^\\d]{0,12}[¥￥]?\\s*([\\d,]+\\.\\d{2})', r'税\\s*额[\\s\\S]{0,30}?([\\d,]+\\.\\d{2})'], '发票类型': [r'(电子发票\\s*[（(][^）)]{2,14}[）)])', r'(电子发票（[^）]+）)', r'(增值税[电子]*[普通专用]*发票)'], } ''' s = s.replace(old_pats, new_pats) # ── ④ parse_pdf 双通道 ── old_fn = s[s.index('def parse_pdf(path):'):s.index('# ── 图片（走 macOS Vision OCR）')] new_fn = '''def parse_pdf(path): """文本层 + Vision OCR（视觉顺序）双通道取字段；数值取不到最大值时标『推定』。""" layer = _pdf_text(path) ocr = ocr_pdf_text(path) full = (ocr + '\\n' + layer) if ocr else layer out, missing = {}, [] for field, pats in PDF_PATTERNS.items(): val = None for p in pats: try: m = re.search(p, full, re.S) except re.error: m = None if m: val = (m.group(1) if m.groups() else m.group(0)).strip() break out[field] = val if not val and field in ('发票号码', '开票日期', '销方名称', '价税合计'): missing.append(field) # 价税合计兜底：文本层乱序时取所有 ¥ 金额的最大值（发票总额必然最大），标『推定』 note = '' if not out.get('价税合计'): cands = [float(x.replace(',', '')) for x in re.findall(r'[¥￥]\\s*([\\d,]+\\.\\d{2})', full)] cands += [float(x.replace(',', '')) for x in re.findall(r'([\\d,]+\\.\\d{2})', full)] cands = [c for c in cands if 0 < c < 1000000] if cands: out['价税合计'] = '%.2f' % max(cands) note = '价税合计为文本推定值（取票面最大金额），请核对' missing = [x for x in missing if x != '价税合计'] out['_文本长度'] = len(layer) out['_OCR长度'] = len(ocr) if note: out['_备注'] = note return out, missing ''' s = s.replace(old_fn, new_fn) # parse_file 里把 _备注 带出来 s = s.replace(" '解析备注': err,", " '解析备注': err or raw.get('_备注'),") open(P, 'w', encoding='utf-8').write(s) ast.parse(s) print('✅ invoice_parse.py 已升级（ZIP 解包 + OCR 双通道 + 正则增强），语法通过') print(' expand_zip:', 'expand_zip' in s, '｜ OCR:', 'ocr_pdf_text' in s, '｜ 推定兜底:', '文本推定值' in s)