#!/usr/bin/env python3 # -*- coding: utf-8 -*- """报销助手 · 邮箱发票自动抓取（IMAP） 用途：定时把报销专用邮箱（407584398@qq.com）里的发票附件抓进 数据/收件箱/， 随后自动跑 解析 → 校验 → 页面生成，一步到位。 用法： /usr/bin/python3 脚本/fetch_invoices_mail.py # 抓取 → 解析 → 校验 → 生成页面 /usr/bin/python3 脚本/fetch_invoices_mail.py --only-fetch # 只抓邮件，不处理 /usr/bin/python3 脚本/fetch_invoices_mail.py --test # 只测登录与连通性 /usr/bin/python3 脚本/fetch_invoices_mail.py --days 30 # 只抓最近 30 天的邮件 配置：数据/邮箱配置.json（授权码请在本机文件里填，勿贴在聊天里） 安全：只读取与下载，不删邮件；已抓取过的邮件按 Message-ID + UID 去重，绝不重复入库。 """ import os import re import sys import json import email import hashlib import imaplib import datetime import subprocess from email.header import decode_header, make_header sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) import invoice_core as core # noqa: E402 CONF = os.path.join(core.DATA, '邮箱配置.json') SEEN = os.path.join(core.DATA, '邮箱已抓取.json') DEFAULT_CONF = { "imap_host": "imap.qq.com", "imap_port": 993, "use_ssl": True, "账号": "407584398@qq.com", "授权码": "", "文件夹": ["INBOX"], "只抓未读": False, "抓取后标记已读": True, "附件后缀": [".pdf", ".xml", ".ofd", ".jpg", ".jpeg", ".png", ".bmp", ".webp", ".zip"], "主题关键词": [], "发件人白名单": [], "抓内嵌图片": True, "_说明": "QQ 邮箱需在「设置 → 账户 → POP3/IMAP/SMTP服务」开启 IMAP 并生成【授权码】，把授权码填到上面『授权码』字段。" } EXT_RANK = {'.xml': 0, '.pdf': 1, '.ofd': 2, '.jpg': 5, '.jpeg': 5, '.png': 5, '.bmp': 6, '.webp': 6, '.tif': 6, '.tiff': 6, '.zip': 9} # 附件伴生文档（不是发票本身，是配套材料）→ 会自动回填到对应发票的补录里 COMPANION = { '住宿清单': ['结账单', '住宿清单', '入住清单', '消费明细', '水单'], '行程单': ['行程单', 'itinerary'], } def companion_kind(name): low = name.lower() for kind, kws in COMPANION.items(): if any(k in low for k in kws): return kind return '' 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 def prefer_best_formats(atts, allowed=None): """同一封邮件里同名不同格式（如 2631...115.ofd/.pdf/.png）只保留最佳格式， 避免同一张发票因多格式重复入库。不同文件名（发票+结账单）都保留。 allowed：当前配置允许落盘的后缀——**只允许的格式参与“最佳格式”评选**。 否则 xml 会赢过 pdf，却因配置不收 xml 被整封跳过 → 真发票永久漏抓。""" best = {} for name, payload in atts: base, ext = os.path.splitext(name) if allowed and ext.lower() not in allowed and ext.lower() != '.zip': continue key = base.lower().strip() rank = EXT_RANK.get(ext.lower(), 8) if key not in best or rank < best[key][0]: best[key] = (rank, name, payload) return [(n, p) for _, n, p in best.values()] def load_conf(): if not os.path.exists(CONF): json.dump(DEFAULT_CONF, open(CONF, 'w', encoding='utf-8'), ensure_ascii=False, indent=2) print('📝 已生成配置模板：%s' % CONF) c = dict(DEFAULT_CONF) c.update(json.load(open(CONF, encoding='utf-8'))) return c SRCFILE = os.path.join(core.DATA, '邮件来源.json') def load_src(): if os.path.exists(SRCFILE): try: return json.load(open(SRCFILE, encoding='utf-8')) except Exception: pass return {'文件': {}} def save_src(s): s['updated'] = core.now() json.dump(s, open(SRCFILE, 'w', encoding='utf-8'), ensure_ascii=False, indent=1) def load_seen(): if os.path.exists(SEEN): try: return json.load(open(SEEN, encoding='utf-8')) except Exception: pass return {'message_ids': [], 'uids': {}, '已抓取文件': []} def save_seen(s): s['updated'] = core.now() json.dump(s, open(SEEN, 'w', encoding='utf-8'), ensure_ascii=False, indent=1) def dec(s): if not s: return '' try: return str(make_header(decode_header(s))) except Exception: return str(s) def safe_name(n): n = re.sub(r'[\\/:*?"<>|\r\n\t]', '_', n or 'attachment') return n.strip()[:120] or 'attachment' def connect(c): host, port = c['imap_host'], int(c['imap_port']) if c.get('use_ssl', True): box = imaplib.IMAP4_SSL(host, port, timeout=45) else: box = imaplib.IMAP4(host, port, timeout=45) box.login(c['账号'], c['授权码']) return box def iter_attachments(msg, conf): """产出 (文件名, 字节) —— 附件 + 可选内嵌图片""" for part in msg.walk(): ctype = part.get_content_type() disp = str(part.get('Content-Disposition') or '') fname = part.get_filename() if fname: fname = dec(fname) is_attach = ('attachment' in disp.lower()) or bool(fname) is_inline_img = conf.get('抓内嵌图片') and ctype.startswith('image/') and 'inline' in disp.lower() if not (is_attach or is_inline_img): continue if not fname: ext = mimetypes_ext(ctype) fname = 'mail_%s%s' % (datetime.datetime.now().strftime('%H%M%S%f'), ext) payload = part.get_payload(decode=True) if not payload: continue yield safe_name(fname), payload def mimetypes_ext(ctype): return {'.pdf': '.pdf'}.get(ctype, {'image/jpeg': '.jpg', 'image/png': '.png', 'application/pdf': '.pdf', 'text/xml': '.xml', 'application/xml': '.xml'}.get(ctype, '.bin')) def fetch(conf, days=None, only_fetch=False, test_only=False): if not conf.get('授权码') or '请填' in str(conf.get('授权码')): print('⚠️ 还没填授权码。请在 %s 的『授权码』字段填入 QQ 邮箱授权码（IMAP 授权码，非登录密码）。' % CONF) print(' 获取路径：QQ邮箱网页版 → 设置 → 账户 → POP3/IMAP/SMTP服务 → 开启 IMAP → 生成授权码') return 0, 0 seen = load_seen() src = load_src() ext_ok = tuple(conf.get('附件后缀') or ['.pdf', '.xml']) since = None if days: since = (datetime.date.today() - datetime.timedelta(days=int(days))).strftime('%d-%b-%Y') try: box = connect(conf) except Exception as e: print('❌ 登录失败：%s' % str(e)[:160]) print(' 常见原因：① 授权码错误或已重置 ② IMAP 服务未开启 ③ 网络/端口被拦') return 0, 0 print('✅ 已登录 %s（%s）' % (conf['账号'], conf['imap_host'])) if test_only: st, folders = box.list() print(' 文件夹数:', len(folders) if folders else 0) box.logout() return 0, 0 got, saved = 0, 0 skipped = [] for folder in (conf.get('文件夹') or ['INBOX']): _ok = False for _ro in (False, True): # 写模式打不开就退只读（部分服务器拒绝对子文件夹 SELECT） try: _t, _d = box.select(folder, readonly=_ro) if _t == 'OK': _ok = True break except Exception: continue if not _ok: print(' ⚠️ 无法打开文件夹 %s，跳过' % folder) continue crit = ['SINCE', since] if since else (['UNSEEN'] if conf.get('只抓未读') else ['ALL']) # 🔴 必须用 UID 检索：序号（seq num）会因删除/归档而整体前移， # 用序号去重会把「新邮件」误判成「老邮件已抓过」→ 静默漏票（2026-10-24 实测漏掉整趟出差）。 st, data = box.uid('SEARCH', None, *crit) ids = (data[0] or b'').split() print(' %s：命中 %d 封（%s）' % (folder, len(ids), ' '.join(crit))) for num in ids[-300:]: # 单次上限，避免误开巨量 try: st, mdata = box.uid('FETCH', num, '(RFC822)') msg = email.message_from_bytes(mdata[0][1]) except Exception: continue mid = (msg.get('Message-ID') or '').strip() uid_key = '%s:uid:%s' % (folder, num.decode()) if mid and mid in seen['message_ids']: continue if uid_key in seen['uids']: continue subj = dec(msg.get('Subject')) frm = dec(msg.get('From')) date = msg.get('Date') or '' if conf.get('主题关键词') and not any(k in subj for k in conf['主题关键词']): continue if conf.get('发件人白名单') and not any(k in frm for k in conf['发件人白名单']): continue got += 1 mail_atts = prefer_best_formats(list(iter_attachments(msg, conf)), allowed=ext_ok) saved_this = 0 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 h = 'sha256:' + hashlib.sha256(payload).hexdigest()[:16] if h in seen.get('已抓取文件', []): continue dst = os.path.join(core.INBOX, fname) if os.path.exists(dst): base, ext2 = os.path.splitext(fname) dst = os.path.join(core.INBOX, '%s_%s%s' % (base, h[-6:], ext2)) open(dst, 'wb').write(payload) seen.setdefault('已抓取文件', []).append(h) saved += 1 saved_this += 1 src.setdefault('文件', {})[os.path.basename(dst)] = { 'message_id': mid, 'subject': subj, 'from': frm, 'date': date, 'kind': companion_kind(fname), 'siblings': [n for n, _ in mail_atts], } print(' 📎 %s%s ←「%s」' % (fname, ' [配套材料]' if companion_kind(fname) else '', subj[:30])) # 🔴 只有「确实落盘」或「本来就没有可用附件」才记已抓取； # 有可用附件却没落盘（下载失败/被跳过）必须留待下次重试，否则该票永久丢失 _eligible = [f for f, _ in expanded if f.lower().endswith(ext_ok)] if saved_this or not _eligible: if mid: seen['message_ids'].append(mid) seen['uids'][uid_key] = core.now() else: skipped.append('%s（附件未落盘，下次重试）' % subj[:20]) print(' ⚠️ 附件未落盘，本次不记录，下次重试：%s' % subj[:36]) if conf.get('抓取后标记已读') and saved_this: try: box.uid('STORE', num, '+FLAGS', '\\Seen') except Exception: pass try: box.logout() except Exception: pass save_seen(seen) save_src(src) print('\n抓取完成：新邮件 %d 封 ｜ 新增附件 %d 个 → %s' % (got, saved, core.INBOX)) if skipped: print('\n⏭ 按口径跳过 %d 个非 PDF 附件（如需一并收取，改 数据/邮箱配置.json 的『附件后缀』）：' % len(skipped)) for x in skipped[:15]: print(' • %s' % x) if len(skipped) > 15: print(' … 等共 %d 个' % len(skipped)) return got, saved def main(argv): conf = load_conf() days = None if '--days' in argv: try: days = argv[argv.index('--days') + 1] except IndexError: pass if '--test' in argv: return fetch(conf, days, test_only=True) got, saved = fetch(conf, days, only_fetch='--only-fetch' in argv) if '--only-fetch' in argv or (got == 0 and saved == 0): return 0 if saved: print('\n▶ 自动处理新票据…') for step in [['/usr/bin/python3', os.path.join(core.BASE, '脚本', 'invoice_parse.py')], ['/usr/bin/python3', os.path.join(core.BASE, '脚本', 'invoice_check.py')], ['/usr/bin/python3', os.path.join(core.BASE, '脚本', 'build_page.py')]]: r = subprocess.run(step, capture_output=True, text=True) tail = [l for l in (r.stdout or '').strip().split('\n') if l.strip()][-3:] print(' ' + ' / '.join(tail)) return 0 if __name__ == '__main__': sys.exit(main(sys.argv[1:]))