#!/usr/bin/env python3 # -*- coding: utf-8 -*- """正确落地两个补丁（上次把两个文件的锚点搞混了）： A. fetch_invoices_mail.py ← ZIP 解包（打车/高速通行费发票以车牌命名 ZIP 发来） B. invoice_parse.py ← OCR 双通道 + 正则增强（真实数电票文本流乱序） 每个补丁独立校验锚点，落地后逐项断言。 """ import ast import sys # ══════════ A. 抓取脚本：ZIP 解包 ══════════ FA = '脚本/fetch_invoices_mail.py' a = open(FA, encoding='utf-8').read() if 'expand_zip' not in a: anchor = 'def prefer_best_formats(atts):' assert anchor in a, 'A-锚点缺失：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 ''' a = a.replace(anchor, block + anchor, 1) old_loop = """ for fname, payload in mail_atts: if not fname.lower().endswith(ext_ok): skipped.append('%s（%s）' % (fname, subj[:22])) continue""" new_loop = """ 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 old_loop in a: a = a.replace(old_loop, new_loop, 1) elif 'expanded.append((_n, _p))' not in a: print('⚠️ A-循环锚点未命中，需人工检查') open(FA, 'w', encoding='utf-8').write(a) ast.parse(a) print('A ✅ fetch_invoices_mail.py: expand_zip=%d 循环改造=%d' % (a.count('expand_zip'), a.count('expanded.append'))) # ══════════ B. 解析脚本：OCR 双通道 + 正则增强 ══════════ PB = '脚本/invoice_parse.py' b = open(PB, encoding='utf-8').read() if 'OCR_BIN' not in b: 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（一次编译，后续直接调用，约 1s/页）""" if os.path.exists(OCR_BIN): return OCR_BIN if not os.path.exists(OCR_SWIFT): return None try: subprocess.run(['swiftc', '-O', '-o', OCR_BIN, OCR_SWIFT], capture_output=True, text=True, timeout=240) except Exception: pass return OCR_BIN if os.path.exists(OCR_BIN) else None def ocr_pdf_text(path): """逐页 OCR（读到的是视觉顺序：标签紧邻值）——数电票文本层常是『标签在前值在后』的乱序， OCR 用于补齐 买方/销方名称、价税合计等字段。失败返回空串。""" exe = ensure_ocr_bin() if not exe: return '' try: r = subprocess.run([exe, path], capture_output=True, text=True, timeout=180) return (r.stdout or '').strip() except Exception: return '' ''' assert 'def _num(s):' in b, 'B-锚点缺失：_num' b = b.replace('def _num(s):', ins + 'def _num(s):', 1) # 正则增强 i0 = b.index('PDF_PATTERNS = {') i1 = b.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{1,2}[-/]\\d{1,2})', r'(20\\d{2}\\s*年\\s*\\d{1,2}\\s*月\\s*\\d{1,2}\\s*日)', r'开票日期\\s*[：:]?\\s*((?:20\\d{2}\\s*年[^\\n]{0,14}日))'], '买方名称': [r'购\\s*买?\\s*方[\\s\\S]{0,26}?名\\s*称\\s*[：:]\\s*([^\\n，,]{2,44})', r'购\\s*买?\\s*方\\s*名\\s*称\\s*[：:]\\s*([^\\n]{2,44})'], '买方税号': [r'购\\s*买?\\s*方[\\s\\S]{0,100}?([0-9A-Z]{18})(?![0-9A-Z])'], '销方名称': [r'销\\s*售\\s*方[\\s\\S]{0,26}?名\\s*称\\s*[：:]\\s*([^\\n，,]{2,44})', r'销\\s*售\\s*方\\s*名\\s*称\\s*[：:]\\s*([^\\n]{2,44})'], '销方税号': [r'销\\s*售\\s*方[\\s\\S]{0,100}?([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,18}?[¥￥]?\\s*([\\d,]+\\.\\d{2})'], '税额': [r'合\\s*计[\\s\\S]{0,44}?[¥￥]?\\s*[\\d,]+\\.\\d{2}[^\\d]{0,14}[¥￥]?\\s*([\\d,]+\\.\\d{2})'], '发票类型': [r'(电子发票\\s*[（(][^）)]{2,16}[）)])', r'(增值税[电子]*[普通专用]*发票)'], } ''' b = b[:i0] + new_pats + b[i1:] # parse_pdf 双通道 j0 = b.index('def parse_pdf(path):') j1 = b.index('# ── 图片（走 macOS Vision OCR）') new_fn = '''def parse_pdf(path): """文本层 + 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)] if not cands: 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) out['_备注'] = note return out, missing ''' b = b[:j0] + new_fn + b[j1:] if "'解析备注': err," in b: b = b.replace("'解析备注': err,", "'解析备注': err or raw.get('_备注'),", 1) open(PB, 'w', encoding='utf-8').write(b) ast.parse(b) print('B ✅ invoice_parse.py: OCR=%d 推定兜底=%d 文本层仍用=%d' % (b.count('ocr_pdf_text'), b.count('推定值'), b.count('_pdf_text(path)'))) print('\n两个补丁均已落地，语法通过')