yhm-shop-data/对标试点/03_分析/detect.py

# -*- 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"""<div class="card {top[0]}">
<div class="h"><span class="nm">{m['门店名称']}</span><span class="city">{m['城市']}</span>
{''.join(f'<span class="tag t-{lv}">{lb}</span>' for lv, lb, _ in m['flags'][:4])}</div>
<div class="msg">{actions(m)[0]}</div>
<div class="nums">评分 {m['评分']} · 核销率 {f'{m["核销率"]*100:.0f}%' if m['核销率'] is not None else '-'} · 核销金额 {m['核销金额']:.0f} 元 · 退单率 {m['商责退单率'] * 100:.2f}% · 月访问 {m['月访问']:.0f}</div>
</div>""")

    trs = []
    for m in metrics[:150]:
        tags = ''.join(f'<span class="tag t-{lv}">{lb}</span>' for lv, lb, _ in m['flags'])
        trs.append(f"""<tr><td>{m['门店名称']}</td><td>{m['城市']}</td>
<td>{m['评分'] if m['评分'] is not None else '-'}</td>
<td>{f'{m["核销率"]*100:.0f}%' if m['核销率'] is not None else '-'}</td>
<td>{m['核销金额']:.0f}</td><td>{m['商责退单率'] * 100:.2f}%</td><td>{tags}</td></tr>""")

    conf_html = ''
    if conflicts:
        rows = ''.join(f'<div>· {sid}: {n1} / {n2} —— 两个门店共用一个ID, 需在主数据里核实</div>'
                       for sid, n1, n2 in conflicts)
        conf_html = f'<h2><span class="bar"></span>主数据风险：同 ID 不同名</h2><div class="cards"><div class="card warn">{rows}</div></div>'

    return f"""<!DOCTYPE html><html lang="zh-CN"><head><meta charset="utf-8">
<meta name="viewport" content="width=device-width,initial-scale=1">
<title>门店月度检测报告 {month}</title><style>{CSS}</style></head><body><div class="wrap">
<h1>门店月度检测报告 · {month}</h1>
<div class="sub">检测规则来自《门店考核指标》官方基准线 · 覆盖 {len(metrics)} 家门店 · 完整清单见同名 CSV</div>
<div class="kpis">{''.join(f'<div class="kpi{" warn" if i in (1,) else ""}"><div class="v">{v}</div><div class="l">{k}</div></div>' for i, (k, v) in enumerate(kpis))}</div>
<h2><span class="bar"></span>优先处理：超违规线门店 TOP{len(cards)}</h2>
<div class="cards">{''.join(cards)}</div>
{conf_html}
<h2><span class="bar"></span>检测明细（前 150 行，按严重度排序）</h2>
<table><thead><tr><th>门店</th><th>城市</th><th>评分</th><th>核销率</th><th>核销金额</th><th>退单率</th><th>触发问题</th></tr></thead>
<tbody>{''.join(trs)}</tbody></table>
<div class="note">
<b>规则口径</b>　核销率 = 核销券数 ÷ 意向成交券数；差评率优先用新增口径(本月新增中差评÷新增评价)，本月无新增评价时回退累计口径并标注。<br>
<b>等级说明</b>　红色=超出违规线(退单/投诉/差评三线)，蓝色=低于考核基准，灰色=提升项。<br>
<b>主数据</b>　本月去重 {sum(1 for _ in metrics)} 家，去重依据门店ID；同ID不同名的冲突单列在报告里，需要人工核实。
</div></div></body></html>"""


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()