# -*- coding: utf-8 -*- """ 门店月度检测引擎 v0.1 —— 基于飞书「门店考核指标」官方基准线 输入: 门店数据_10月.csv (每月从飞书表导出/更新) 输出: 04_输出/门店检测报告_<月份>.html + 门店检测清单_<月份>.csv 阈值集中在 THRESH, 与飞书《门店考核指标》表一致; 口径如有调整, 只改这里。 """ import csv import os import sys BASE = os.path.dirname(os.path.abspath(__file__)) OUT = os.path.join(os.path.dirname(BASE), '04_输出') # ---- 阈值(来自飞书「门店考核指标」表) ---- THRESH = dict( score_base=4.3, score_target=4.5, # 门店评分: 基准4.3, 目标高于4.5 hx_rate_base=0.40, hx_rate_target=0.60, # 核销率: 基准40%, 目标60% hx_amt_base=12000, # 核销金额: 基准12000 tuDan=[0.0127, 0.0318, 0.0635], # 商责退单率: 预警/一般/非常严重(小数) touSu=[0.0011, 0.0028, 0.0056], # 投诉率(小数) chaPing=[0.0066, 0.0166, 0.0331], # 差评率(小数) reply_base=50, # 评价回复率(%) visit_low=100, # 月访问过低线 ) LV = {'R3': 3, 'R2': 2, 'R1': 1, 'B': 0.6, 'T': 0.3} def f(v, pct=False): try: x = float(str(v).replace('%', '').replace(',', '')) return x / 100 if pct else x except (ValueError, TypeError): return None def band(v, lines): """返回违规等级 R1/R2/R3 或 None""" if v is None: return None if v > lines[2]: return 'R3' if v > lines[1]: return 'R2' if v > lines[0]: return 'R1' return None def detect(rows): out, seen, id_conflict = [], {}, [] for r in rows: sid = r['门店ID'].strip() if sid in seen: if seen[sid] != r['门店名称']: id_conflict.append((sid, seen[sid], r['门店名称'])) continue # 重复行去重(保留首行) seen[sid] = r['门店名称'] g = {k: f(r.get(k, ''), pct='率' in k or k == '评价回复率') for k in ( '门店页访问次数', '门店页成交券数', '门店核销金额', '门店核销券数', '门店意向成交券数', '门店评分', '新增评价数', '新增中差评数', '累计评价数', '累计中差评数', '评价回复率', '经营风险投诉率', '经营风险商责退单率', '新发布门店关联视频数', '门店收藏数')} if g['门店意向成交券数'] and g['门店意向成交券数'] > 0: hx_rate = (g['门店核销券数'] or 0) / g['门店意向成交券数'] else: hx_rate = None if (g['新增评价数'] or 0) >= 10: cp_rate = (g['新增中差评数'] or 0) / g['新增评价数'] cp_src = '新增' elif (g['累计评价数'] or 0) >= 20: cp_rate = (g['累计中差评数'] or 0) / g['累计评价数'] cp_src = '累计' else: cp_rate, cp_src = None, '' # 样本太少, 不判差评率, 避免小样本误报 flags = [] lv = band(g['经营风险商责退单率'], THRESH['tuDan']) if lv: lv_name = {'R1': '预警线', 'R2': '一般严重线', 'R3': '非常严重线'}[lv] flags.append((lv, f"商责退单率{g['经营风险商责退单率'] * 100:.2f}%超{lv_name}", '催办到店核销、主动处理退单商责, 避免平台处置')) lv = band(g['经营风险投诉率'], THRESH['touSu']) if lv: flags.append((lv, f"投诉率{g['经营风险投诉率'] * 100:.2f}%超线", '24小时内响应投诉工单, 复盘服务环节')) lv = band(cp_rate, THRESH['chaPing']) if lv: flags.append((lv, f"差评率({cp_src}口径){cp_rate * 100:.2f}%超线", '处理近30天差评, 主攻优质深度好评')) if g['门店评分'] is not None and g['门店评分'] < THRESH['score_base']: flags.append(('B', f"评分{g['门店评分']:.1f}低于考核基准{THRESH['score_base']}", '引导好评+回复差评, 目标先回4.3再冲4.5')) if hx_rate is not None and hx_rate < THRESH['hx_rate_base']: flags.append(('B', f"核销率{hx_rate * 100:.0f}%低于基准{THRESH['hx_rate_base'] * 100:.0f}%", '电话/私信触达待核销客户, 加速到店')) if (g['门店核销金额'] or 0) < THRESH['hx_amt_base']: if (g['门店核销券数'] or 0) == 0: flags.append(('B', '本月零核销', '检查团购在售状态与价格, 重新起量')) else: flags.append(('B', f"核销金额{g['门店核销金额']:.0f}元低于基准{THRESH['hx_amt_base']}", '补齐在售品结构, 冲单量')) if (g['评价回复率'] is not None and g['评价回复率'] < THRESH['reply_base'] / 100): flags.append(('T', f"评价回复率{g['评价回复率'] * 100:.0f}%偏低", '安排专人回复全部评价')) if (g['门店页访问次数'] or 0) < THRESH['visit_low']: flags.append(('T', f"月访问仅{g['门店页访问次数']:.0f}次", '检查POI完整度, 点亮门店+更新AIGC素材')) if (g['新发布门店关联视频数'] or 0) == 0: flags.append(('T', '本月零视频', '后台AIGC工具定期产出图片/视频')) sev = max((LV[lv] for lv, _, _ in flags), default=0) out.append(dict(门店ID=sid, 门店名称=r['门店名称'], 城市=r['所在城市'], 评分=g['门店评分'], 核销率=hx_rate, 核销金额=g['门店核销金额'] or 0, 核销券数=g['门店核销券数'] or 0, 商责退单率=g['经营风险商责退单率'] or 0, 投诉率=g['经营风险投诉率'] or 0, 差评率=cp_rate, 差评口径=cp_src, 回复率=g['评价回复率'], 月访问=g['门店页访问次数'] or 0, 新视频=g['新发布门店关联视频数'] or 0, flags=flags, sev=sev)) out.sort(key=lambda x: (-x['sev'], x['核销金额'])) return out, id_conflict def actions(m): acts, seen = [], set() for _, _, a in m['flags']: if a not in seen: seen.add(a) acts.append(a) return acts CSS = open(os.path.join(BASE, 'report_style.css'), encoding='utf-8').read() \ if os.path.exists(os.path.join(BASE, 'report_style.css')) else """ *{box-sizing:border-box;margin:0;padding:0} body{background:#0f1115;color:#e6e8ec;font-family:-apple-system,"Segoe UI","Microsoft YaHei",sans-serif; font-size:14px;line-height:1.65;padding:36px 40px 64px} .wrap{max-width:1180px;margin:0 auto} h1{font-size:22px;font-weight:600}.sub{color:#8b93a1;font-size:13px;margin-top:8px} .kpis{display:grid;grid-template-columns:repeat(5,1fr);gap:12px;margin:26px 0 30px} .kpi{background:#171a21;border:1px solid #252a34;border-radius:12px;padding:16px 18px} .kpi .v{font-size:24px;font-weight:600;font-variant-numeric:tabular-nums;white-space:nowrap} .kpi .l{color:#8b93a1;font-size:12px;margin-top:2px} .kpi.warn .v{color:#ff8585} h2{font-size:15px;font-weight:600;margin:34px 0 14px;display:flex;align-items:center;gap:8px} h2 .bar{width:3px;height:14px;background:#6aa9ff;border-radius:2px} table{width:100%;border-collapse:collapse;font-size:13px;font-variant-numeric:tabular-nums} th{text-align:left;color:#8b93a1;font-weight:500;font-size:12px;padding:9px 10px;border-bottom:1px solid #2a2f3a;white-space:nowrap} td{padding:10px;border-bottom:1px solid #1d212a;vertical-align:top} tr:hover td{background:#151920} .tag{display:inline-block;padding:2px 8px;border-radius:5px;font-size:11px;font-weight:500;margin:2px 4px 2px 0;white-space:nowrap} .t-R3{background:rgba(255,60,60,.2);color:#ff6b6b} .t-R2{background:rgba(255,107,107,.14);color:#ff8585} .t-R1{background:rgba(240,168,60,.16);color:#f0b45c} .t-B{background:rgba(109,140,255,.14);color:#8fa8ff} .t-T{background:rgba(140,148,163,.14);color:#9aa3b2} .cards{display:grid;gap:10px} .card{background:#171a21;border:1px solid #252a34;border-left:3px solid #2f3745;border-radius:10px;padding:14px 18px} .card.R3{border-left-color:#ff3c3c}.card.R2{border-left-color:#ff6b6b}.card.R1{border-left-color:#f0a83c} .card .h{display:flex;align-items:center;gap:10px;flex-wrap:wrap} .card .nm{font-weight:600;font-size:14px} .card .city{color:#6f7787;font-size:12px} .card .msg{color:#a8b0be;font-size:13px;margin-top:6px} .card .act{color:#5fd6ae;font-size:12px;margin-top:4px} .card .nums{color:#6f7787;font-size:12px;margin-top:6px;font-variant-numeric:tabular-nums} .note{color:#767e8c;font-size:12px;line-height:1.9;margin-top:14px} .note b{color:#9aa3b2;font-weight:500} """ def render(metrics, conflicts, month): def cnt(p): return sum(1 for m in metrics if any(lv == p or (p == 'R' and lv in ('R1', 'R2', 'R3')) for lv, _, _ in m['flags'])) kpis = [('检测门店', f'{len(metrics)} 家'), ('超违规预警线(退单/投诉/差评)', f'{cnt("R")} 家'), ('评分低于基准4.3', f'{sum(1 for m in metrics if m["评分"] is not None and m["评分"] < THRESH["score_base"])} 家'), ('核销率低于基准40%', f'{sum(1 for m in metrics if m["核销率"] is not None and m["核销率"] < THRESH["hx_rate_base"])} 家'), ('本月零核销', f'{sum(1 for m in metrics if m["核销金额"] == 0)} 家')] worst = [m for m in metrics if m['sev'] >= LV['R1']][:20] cards = [] for m in worst: top = max(m['flags'], key=lambda x: LV[x[0]]) cards.append(f"""
{m['门店名称']}{m['城市']} {''.join(f'{lb}' for lv, lb, _ in m['flags'][:4])}
{actions(m)[0]}
评分 {m['评分']} · 核销率 {f'{m["核销率"]*100:.0f}%' if m['核销率'] is not None else '-'} · 核销金额 {m['核销金额']:.0f} 元 · 退单率 {m['商责退单率'] * 100:.2f}% · 月访问 {m['月访问']:.0f}
""") trs = [] for m in metrics[:150]: tags = ''.join(f'{lb}' for lv, lb, _ in m['flags']) trs.append(f"""{m['门店名称']}{m['城市']} {m['评分'] if m['评分'] is not None else '-'} {f'{m["核销率"]*100:.0f}%' if m['核销率'] is not None else '-'} {m['核销金额']:.0f}{m['商责退单率'] * 100:.2f}%{tags}""") conf_html = '' if conflicts: rows = ''.join(f'
· {sid}: {n1} / {n2} —— 两个门店共用一个ID, 需在主数据里核实
' for sid, n1, n2 in conflicts) conf_html = f'

主数据风险：同 ID 不同名

{rows}
' return f""" 门店月度检测报告 {month}

门店月度检测报告 · {month}

检测规则来自《门店考核指标》官方基准线 · 覆盖 {len(metrics)} 家门店 · 完整清单见同名 CSV
{''.join(f'
{v}
{k}
' for i, (k, v) in enumerate(kpis))}

优先处理：超违规线门店 TOP{len(cards)}

{''.join(cards)}
{conf_html}

检测明细（前 150 行，按严重度排序）

{''.join(trs)}
门店城市评分核销率核销金额退单率触发问题
规则口径　核销率 = 核销券数 ÷ 意向成交券数；差评率优先用新增口径(本月新增中差评÷新增评价)，本月无新增评价时回退累计口径并标注。
等级说明　红色=超出违规线(退单/投诉/差评三线)，蓝色=低于考核基准，灰色=提升项。
主数据　本月去重 {sum(1 for _ in metrics)} 家，去重依据门店ID；同ID不同名的冲突单列在报告里，需要人工核实。
""" def main(): args = sys.argv[1:] src = args[0] if args and args[0].endswith('.csv') else os.path.join(BASE, '门店数据_10月.csv') month = '2026-10' if '--month' in args: month = args[args.index('--month') + 1] rows = list(csv.DictReader(open(src, encoding='utf-8-sig'))) metrics, conflicts = detect(rows) os.makedirs(OUT, exist_ok=True) tag = month.replace('_', '') html = os.path.join(OUT, f'门店检测报告_{tag}.html') csvout = os.path.join(OUT, f'门店检测清单_{tag}.csv') with open(html, 'w', encoding='utf-8') as fp: fp.write(render(metrics, conflicts, month)) with open(csvout, 'w', encoding='utf-8-sig', newline='') as fp: w = csv.writer(fp) w.writerow(['门店ID', '门店名称', '城市', '评分', '核销率', '核销金额', '商责退单率', '投诉率', '差评率', '差评口径', '评价回复率', '月访问', '本月新视频', '触发问题', '建议动作']) for m in metrics: w.writerow([m['门店ID'], m['门店名称'], m['城市'], m['评分'] if m['评分'] is not None else '', f"{m['核销率']:.4f}" if m['核销率'] is not None else '', f"{m['核销金额']:.2f}", f"{m['商责退单率'] * 100:.2f}%", f"{m['投诉率'] * 100:.2f}%", f"{m['差评率'] * 100:.2f}" if m['差评率'] is not None else '', m['差评口径'], f"{m['回复率'] * 100:.0f}" if m['回复率'] is not None else '', f"{m['月访问']:.0f}", f"{m['新视频']:.0f}", ' | '.join(f'{lb}' for _, lb, _ in m['flags']), ' ；'.join(actions(m))]) print(f'报告: {html}') print(f'清单: {csvout}') print(f'检测 {len(metrics)} 家, 红线 {sum(1 for m in metrics if m["sev"] >= 1)} 家, 同ID不同名冲突 {len(conflicts)} 组') if __name__ == '__main__': main()