#!/usr/bin/env python3 # -*- coding: utf-8 -*- """报销助手 · 校验引擎（依据 CA-01-03-A3《付款、费用报销管理制度》） 状态优先级：异常 > 重复 > 待补 > 待核 > 通过 判定依据全部来自 数据/规则库.json（制度落地副本）。规则缺失/口径待确认的项**不判定**，只输出提示。 新增字段（写在这些发票记录的 `补录` 里，页面可编辑）： {城市档位, 间夜数, 就餐人数, 公里数, 能源类型, 事前申请, 住宿清单, 行程单} 用法： /usr/bin/python3 脚本/invoice_check.py # 重算全台账并保存 /usr/bin/python3 脚本/invoice_check.py --list # 只看问题清单 """ import os import re import sys from collections import defaultdict sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) import invoice_core as core # noqa: E402 ORDER = ['异常', '重复', '待补', '待核', '通过'] REQUIRED = ['发票号码', '开票日期', '价税合计'] # 销方名称是元数据，缺失只提示不判待补 TIERS = ['超一类城市', '一类城市', '二类城市'] def _amt(x): try: return float(x) except (TypeError, ValueError): return None def _f(x): try: return float(str(x).strip()) except (TypeError, ValueError): return None def _lim(rules, *path, default=None): cur = rules for k in path: if not isinstance(cur, dict) or k not in cur: return default cur = cur[k] if isinstance(cur, dict): return cur return cur if cur is not None else default def _bad_name(v): """名称字段是否无效（纯号码/标签残留/非中文）→ 视为『未从票面提取到』，不参与判定""" s = str(v or '').strip() if len(s) < 3: return True if re.fullmatch(r'[0-9A-Z]{8,}', s): return True if any(t in s for t in ('纳税人识别号', '统一社会信用代码', '名称：', '号码', '下载次数', '开票人', '备注', '价税合计')): return True return not re.search(r'[\u4e00-\u9fa5]', s) def check_record(rec, rules, led=None, peers=None): if rec.get('类型') == '附件': return {'状态': '附件', '问题': [], '提示': ['配套材料，随发票一并提交']} issues, notes, status = [], [], [] # 票面人工覆盖：rec['票面'] 优先（页面补录后回灌） if rec.get('票面'): for k, v in (rec['票面'] or {}).items(): if v not in (None, ''): rec[k] = v rec['_票面已人工补录'] = True sup = rec.get('补录') or {} cat = (rec.get('分类') or {}).get('费用类别') or '' total = _amt(rec.get('价税合计')) amt, tax = _amt(rec.get('金额')), _amt(rec.get('税额')) # ① 结构：必填缺失 miss = [f for f in REQUIRED if not rec.get(f)] if miss: status.append('待补') issues.append('缺字段：%s' % '、'.join(miss)) if not rec.get('销方名称'): notes.append('票面销方名称未提取 → 可在页面补录（不影响判定）') elif rec.get('_销方推定') or not rec.get('销方税号'): notes.append('销方名称为推定值（票面无『销售方』字样或未取到销方税号）→ 请核对') # ② 金额自洽 if total is not None: if total <= 0: status.append('异常') issues.append('价税合计 ≤ 0') if amt is not None and tax is not None and abs((amt + tax) - total) > 0.02: # 发票「金额+税额」必然等于价税合计；不等几乎都是提取误差 → 待核，不判异常 status.append('待核') notes.append('金额/税额与合计不一致（%.2f + %.2f ≠ %.2f），疑为提取误差，请核对票面' % (amt, tax, total)) # ③ 抬头 / 税号 ci = rules.get('公司开票信息') or {} if core.is_configured(ci.get('名称')): if _bad_name(rec.get('买方名称')): if rec.get('_票面含公司名称') is True: notes.append('抬头：字段未结构化提取，但票面文本含公司名称 → 文本核验通过') elif rec.get('_票面含公司名称') is False: status.append('待核') issues.append('票面文本未出现公司名称（%s）→ 很可能抬头开错/个人抬头，请核对' % ci['名称']) else: status.append('待补') issues.append('票面抬头未能可靠提取 → 请在页面补录「买方名称」后重判') elif ci['名称'].strip() not in str(rec['买方名称']): status.append('异常') issues.append('抬头不符（票面：%s）→ 须重开' % rec['买方名称']) else: notes.append('规则未配置：公司抬头') bt = str(rec.get('买方税号') or '').strip() if core.is_configured(ci.get('税号')): # 判定顺序（重要）：字段可用就直接比对 → 字段不可用才退回文本核验 → 都不行才待补 _exp = str(ci['税号']).strip().upper() if re.fullmatch(r'[0-9A-Z]{18}', bt.upper()): if bt.upper() != _exp: if rec.get('_票面含公司税号') is True: notes.append('税号字段（%s）与公司税号不一致，但票面文本含公司税号 → 疑为提取误差，请核对' % bt) else: status.append('异常') issues.append('税号不符（票面：%s）→ 须重开' % bt) elif rec.get('_票面含公司税号') is True: notes.append('税号：字段未结构化提取，但票面文本含公司税号 → 文本核验通过') elif _bad_name(rec.get('买方名称')): status.append('待补') issues.append('票面税号未能可靠提取（与抬头同源）→ 请在页面补录后重判') else: status.append('待补') issues.append('票面税号未能可靠提取 → 请在页面补录「买方税号」后重判') else: notes.append('规则未配置：公司税号') # ④ 发票类型 cr = rules.get('发票合规规则') or {} allow, deny = cr.get('允许的发票类型') or [], cr.get('禁止的发票类型') or [] t = re.sub(r'\s+', '', str(rec.get('发票类型') or '')) # OCR 常把换行/空格插进类型名 if t: if any(d and re.sub(r'\s+', '', d) in t for d in deny): status.append('异常') issues.append('发票类型不可报：%s' % t) else: # 宽松白名单：命中任一关键类别即视为允许（OCR 误字如「普→晋」不应误报） keys = ['电子发票', '增值税', '数电票', '铁路电子客票', '航空运输电子客票', '行程单'] if not any(k in t for k in keys): status.append('待核') issues.append('发票类型不在允许清单：%s' % t) # ⑤ 开票日期：跨年票失效（制度 §5.3.1 年末关账后票据失效） d = rec.get('开票日期') if d and _amt(d[:4]): rng = cr.get('开票日期允许范围') or {} earliest = str(rng.get('最早') or '') m = re.search(r'(20\d{2})', earliest) if m and d[:4] != m.group(1): status.append('异常') issues.append('跨年度票据（%s）：%s 年度票据须在当年年末关账前报销' % (d, m.group(1))) # ⑥ 重复 / 连号 # 比对池只含发票（配套材料/清单的号码与发票可能重叠，不能算重复报销） pool = [p for p in (led or {}).get('发票', []) if p.get('类型') != '附件'] if peers: pool += [p for p in peers if p is not rec] num = rec.get('发票号码') if num: same = [p for p in pool if p.get('发票号码') == num and p is not rec] if same: status.append('重复') issues.append('发票号码已存在（%s）→ 制度：电子发票只能使用一次' % same[0].get('id')) if rec.get('销方名称') and re.search(r'\d{6,}$', str(num)): tail = int(re.search(r'(\d{6,})$', str(num)).group(1)) near = [p for p in pool if p.get('销方名称') == rec.get('销方名称') and re.search(r'(\d{6,})$', str(p.get('发票号码') or '')) and abs(int(re.search(r'(\d{6,})$', str(p['发票号码'])).group(1)) - tail) <= 2] if len(near) >= 2: status.append('待核') issues.append('疑似连号发票（同销方相邻号 %d 张）' % (len(near) + 1)) # ⑦ 分项标准校验 lim = rules.get('标准与限额') or {} if cat == '差旅-住宿': tier = sup.get('城市档位') nights = _f(sup.get('间夜数')) h = lim.get('差旅住宿') or {} if not tier: status.append('待补') issues.append('缺信息：出差城市档位（用于判定住宿标准）') if not nights: status.append('待补') issues.append('缺信息：间夜数（用于计算日均房价）') if tier and nights and total: cap = _amt((h.get(tier) or {}).get('上限')) if cap: daily = total / nights if daily > cap + 0.01: status.append('异常') issues.append('住宿超标：日均 %.0f 元 > %s上限 %s 元/晚（需《费用超标申请》，按上限报销）' % (daily, tier, cap)) else: notes.append('住宿日均 %.0f 元 ≤ %s上限 %s 元（含税）' % (daily, tier, cap)) else: notes.append('规则未配置：%s 住宿上限' % tier) if not sup.get('住宿清单'): status.append('待补') issues.append('缺附件：酒店住宿清单（制度必备）') if cat in ('业务招待-餐厅', '业务招待-外卖'): people = _f(sup.get('就餐人数')) b = lim.get('业务招待') or {} cap = _amt(b.get('餐厅招待_人均上限') if cat == '业务招待-餐厅' else b.get('外卖点餐_人均上限')) if not people: status.append('待补') issues.append('缺信息：就餐人数（用于计算人均）') elif total and cap: per = total / people if per > cap + 0.01: status.append('异常') issues.append('招待超标：人均 %.0f 元 > 上限 %s 元/人/餐（需《费用超标申请》）' % (per, cap)) else: notes.append('招待人均 %.0f 元 ≤ 上限 %s 元' % (per, cap)) if not (rec.get('分类') or {}).get('关联客户'): status.append('待补') issues.append('缺信息：关联客户') if cat == '差旅-目的地交通费' or cat == '差旅-目的地交通': cap = _amt((lim.get('目的地交通费') or {}).get('上限')) days = _f(sup.get('天数')) if cap and total and days and total / days > cap + 0.01: status.append('异常') issues.append('目的地交通费超标：日均 %.0f 元 > 上限 %s 元/天' % (total / days, cap)) elif not days: notes.append('未填出差天数，未核目的地交通费日均（上限 %s 元/天）' % cap) if cat == '差旅-出差补助': s = _amt((lim.get('出差补助') or {}).get('标准')) days = _f(sup.get('天数')) if s and days and total: expect = s * days if abs(total - expect) > max(40, expect * 0.3): status.append('待核') issues.append('补助金额与标准不符：票面 %.0f 元 vs %s 元/天 × %s 天 = %.0f 元（不满一天减半 40 元）' % (total, s, days, expect)) if cat == '私车公用': km = _f(sup.get('公里数')) p = lim.get('私车公用') or {} rate = _amt((p.get('燃油费_电车') or {}).get('标准')) if sup.get('能源类型') == '电车' \ else _amt((p.get('燃油费_油车') or {}).get('标准')) if not km: status.append('待补') issues.append('缺信息：行驶公里数') if not sup.get('能源类型'): status.append('待补') issues.append('缺信息：能源类型（油车 1.5 元/公里 / 电车 0.8 元/公里）') if km and rate and total: cap_amt = km * rate if total > cap_amt + 0.01: status.append('待核') issues.append('金额 %.0f 元 高于燃油费上限（%s 公里 × %s 元 = %.0f 元）；停车费/路桥费凭票实报实销，请核对拆分' % (total, km, rate, cap_amt)) if not sup.get('行驶记录明细表'): status.append('待补') issues.append('缺附件：《行驶记录明细表》（制度必备）') # ⑧ 事前申请（制度：各类申请需事前申请，报销时须关联） pre = sup.get('事前申请') need = '' if cat.startswith('业务招待'): need = '《招待申请》' elif cat.startswith('差旅') or cat == '私车公用': need = '外出申请 或 出差申请' if cat == '差旅-市内交通(加班)' or cat == '差旅-市内交通': need = '加班申请' if need: if pre == '已关联': notes.append('已关联事前申请：%s' % need) elif pre == '未关联': status.append('待补') issues.append('未关联事前申请（%s）→ 制度要求事前申请' % need) else: status.append('待补') issues.append('待确认是否已关联事前申请（%s）' % need) # ⑨ 行程单（飞机票、网约车必须附） src = '%s %s' % (rec.get('销方名称') or '', rec.get('项目名称') or '') # 数电票版《航空运输电子客票行程单》本身即报销凭证 → 自证，不再要求另附行程单 _self = '航空运输电子客票行程单' in str(rec.get('发票类型') or '') if any(k in src for k in ['航空', '机票', '航空运输', '网约车', '滴滴', '出行']) and not _self: if not sup.get('行程单'): status.append('待补') issues.append('缺附件：行程单（飞机票、网约车必须附）') elif _self and not sup.get('行程单'): notes.append('数电票版行程单，本身即报销凭证（无需另附）') for s in ORDER: if s in status: final = s break else: final = '通过' return {'状态': final, '问题': issues, '提示': notes} def main(argv): led = core.load_ledger() rules = core.load_rules() only = '--list' in argv stats = defaultdict(int) problem = [] n_att = 0 for r in led['发票']: res = check_record(r, rules, led) r['校验'] = res if r.get('类型') == '附件': n_att += 1 continue stats[res['状态']] += 1 if res['状态'] != '通过': problem.append((res['状态'], r)) core.save_ledger(led) print('校验完成：发票 %d 张 + 配套材料 %d 份（依据 规则库 v%s）' % (len(led['发票']) - n_att, n_att, rules.get('version', '?'))) for k in ORDER: if stats.get(k): print(' %-4s %d 张' % (k, stats[k])) print() if only or problem: print('问题清单：') for st, r in sorted(problem, key=lambda x: ORDER.index(x[0])): amt = r.get('价税合计') print(' [%s] %-20s %-11s %9s %s' % (st, r.get('id'), (r.get('开票日期') or '日期?'), ('%.2f' % amt) if isinstance(amt, (int, float)) else '金额?', '；'.join((r['校验']['问题'] or [])[:2]))) return 0 if __name__ == '__main__': sys.exit(main(sys.argv[1:]))