#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ ERP 特殊要求库后处理（2026-09-20 用户四点反馈中的 1、3） ========================================================= ① 清掉「一次性处理」类条目（加急/本单/合同后补/等销售通知…）—— 它们不是客户特殊要求 ③ 把「出现次数」从近似值改成准确值：按 **单据编号去重** 统计 （原实现按行累加，同一份导出文件被上传到多个目录时会重复计数） 用法：python3 fix_erp_after.py # 只预览 python3 fix_erp_after.py --apply # 写入 """ import os import re import sys import json import glob import shutil from collections import defaultdict, Counter ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) LIB = os.path.join(ROOT, '数据', 'customer_special_requirements.json') REVIEW = os.path.join(ROOT, '数据', 'sr_review.json') # 明确的一次性处理短语（剥离后若再无实质内容，才判为一次性） ONEOFF_STRIP = re.compile( r'加急|合同\s*(待补|后补|下周补|补到|原件随货)?|等销售通知|发货时间待.{0,6}通知|先发货|今天发货|明天发货' r'|请于.{0,14}(前)?(寄出|发出|备好|发货)|优先(申请)?发出|有多少先发多少|货款预计[^，。；;]*' r'|预测单[：:\s]*FO\d+|关联(订单|ZHBR)[^，。；;]*|^待确认|^待补$|^无$|再发货|先做合同|做合同|走合同') # 「要求」特征词：出现即说明有实质内容，保留 REQ_KW = re.compile(r'须|需|要求|必须|不得|盖章|盖.{0,4}章|分装|标签|logo|LOGO|白标|COA|MSDS|批次|批号|效期|有效期' r'|温度|冰袋|干冰|体系|浓度|规格|版本|协议|随货|打包|箱|管|ml|T/包|套|稀释|对照|阴性|阳性|质检' r'|账期|工作日|重新制备|备货|预留|检测|报告|说明|附|贴|打印|寄|发') def _s(v): if v is None: return '' t = str(v).strip() return '' if t.lower() in ('nan', 'none', 'null', 'nat') else t def is_oneoff(content): """一次性处理判据：剥离一次性短语后没有实质内容残留。 关键：像「加急，送货单盖仓库章」「先发货，账期按10个工作日」这类，剥离后仍留有 盖章/账期等实质要求 → 必须保留（第一版规则按长度+关键词误删过这类，见 git 记录）。""" c = _s(content) if not c: return False if not ONEOFF_STRIP.search(c): return False residue = re.sub(r'[，,。；;、！!？?\s]+', '', ONEOFF_STRIP.sub('', c)) if len(residue) >= 6: return False # 还有实质内容 → 保留 if REQ_KW.search(residue): return False return True def build_hits(rules=None): """读全部 ERP 销售管理导出 → (客户, 备注文本) → 单据编号集合（行级去重，避免同一文件多次上传重复计数）""" import pandas as pd import warnings warnings.filterwarnings('ignore') paths = sorted(glob.glob(os.path.join(ROOT, '数据', '**', '销售订单导出*.xlsx'), recursive=True)) seen_row, groups = set(), defaultdict(set) idx = defaultdict(list) # (客户, 单据号) → [备注文本] files = 0 for p in paths: try: d = pd.read_excel(p) except Exception: continue col = lambda k: next((c for c in d.columns if k in str(c)), None) c_cust, c_doc, c_sku = col('销售订单关联客户'), col('单据编号'), col('货号#') c_i, c_c = col('备注(内部)'), col('备注(客户)') if not (c_cust and (c_i or c_c)): continue files += 1 for _, r in d.iterrows(): cust = _s(r.get(c_cust)) if not cust or '「' in cust: continue txt = '\n'.join(x for x in (_s(r.get(c_i)), _s(r.get(c_c))) if x) if len(txt) < 4: continue doc = _s(r.get(c_doc)) key = (doc, _s(r.get(c_sku)), re.sub(r'\s+', '', txt)) if key in seen_row: # 行级去重：同一份文件重复上传不再重复计数 continue seen_row.add(key) nt = re.sub(r'\s+', '', txt) groups[(cust, nt)].add(doc) # 索引只按单据号：ERP 里客户名常带全称后缀（如「福建省水产研究所（福建水产病害防治研究所）」）， # 与库里的简称对不上，用（客户+单据号）建索引会大面积匹配失败 idx[doc].append((cust, nt)) return groups, idx, files def match_hits(groups, idx, customer, doc, evidence): """给一条要求找它所属的候选组 → 该要求在多少张不同订单里出现过（按单据编号去重）""" cands = idx.get(doc) if not cands: return 0 ev = re.sub(r'\s+', '', (evidence or '').split('（该要求在')[0]) if not ev: return 0 best, bs = None, 0 for cust, nt in cands: if ev[:24] in nt: return len(groups.get((cust, nt)) or []) n = 0 while n < min(len(ev), len(nt)) and ev[n] == nt[n]: n += 1 if n > bs: bs, best = n, (cust, nt) if best and bs >= 8: return len(groups.get(best) or []) return 0 def main(apply): groups, idx, files = build_hits() print(' 源文件 %d 个 ｜ 候选组 %d ｜ 索引单据 %d' % (files, len(groups), len(idx))) def _clean_ev(s): s = _s(s) s = re.sub(r'^(备注原文|依据原文)[：:]\s*', '', s) s = re.split(r'（该要求在', s)[0] s = re.sub(r'📎', '', s) return re.sub(r'\s+', '', s) # ── ③ 准确出现次数（按单据编号去重）── fixed = 0 for path, is_lib in ((LIB, True), (REVIEW, False)): d = json.load(open(path, encoding='utf-8')) rows = d.get('items') if is_lib else d.get('rows') for it in rows: cust = it.get('customer') or '' if is_lib: m = re.search(r'ERP订单\s*(\S+?)（', it.get('source') or '') doc = m.group(1) if m else '' ev = _clean_ev(it.get('note')) else: doc = (it.get('activity_id') or '').strip() ev = _clean_ev(it.get('evidence')) h = match_hits(groups, idx, cust, doc, ev) old = it.get('hits') or 0 if h: it['hits'] = h if is_lib: note = re.sub(r'（该要求在 \d+ 张订单中出现）', '', it.get('note') or '') it['note'] = note + (('（该要求在 %d 张订单中出现）' % h) if h > 1 else '') if old != h: fixed += 1 elif old: it.pop('hits', None) # 匹配不上就清掉陈旧值，不留错的数 if is_lib: it['note'] = re.sub(r'（该要求在 \d+ 张订单中出现）', '', it.get('note') or '') if is_lib: d['version'] = int(d.get('version') or 1) + 1 json.dump(d, open(path, 'w', encoding='utf-8'), ensure_ascii=False, indent=1) print(' ✅ 出现次数重算：%d 条有变化' % fixed) # ── ① 清掉一次性处理 ── lib = json.load(open(LIB, encoding='utf-8')) keep, drop = [], [] for it in lib['items']: (drop if is_oneoff(it.get('content')) else keep).append(it) print(' 一次性处理条目：%d 条（从 %d 条中筛出）' % (len(drop), len(lib['items']))) for it in drop[:12]: print(' × [%s] %s' % (it['customer'][:14], (it.get('content') or '')[:56])) if apply and drop: lib['items'] = keep lib['version'] = int(lib.get('version') or 1) + 1 json.dump(lib, open(LIB, 'w', encoding='utf-8'), ensure_ascii=False, indent=1) print(' ✅ 已从正式库清除，剩余 %d 条' % len(keep)) # 复核页同步标记 rev = json.load(open(REVIEW, encoding='utf-8')) n = 0 for r in rev['rows']: if is_oneoff(r.get('content')): probs = [p for p in (r.get('problems') or []) if '一次性' not in p] probs.append('已从正式库清除（一次性处理，非客户特殊要求）') r['problems'] = probs r['ok'] = False n += 1 json.dump(rev, open(REVIEW, 'w', encoding='utf-8'), ensure_ascii=False, indent=1) print(' ✅ 复核页标记 %d 行' % n) if apply and drop: shutil.copy2(LIB, LIB + '.bak_before_oneoff_clean') if __name__ == '__main__': main('--apply' in sys.argv[1:])