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

# -*- coding: utf-8 -*-
"""
同城对标检测 - 分析报告生成器
输入: 采集快照 CSV (每次采集累加, 一个商品一行)
输出: 04_输出/对标检测报告_<日期>.html

用法:
    python build_report.py                      # 用最新快照
    python build_report.py --demo               # 用内置模拟数据跑演示
    python build_report.py --file xxx.csv       # 指定快照文件
"""
import csv
import os
import sys
import random
import statistics
from datetime import date

BASE = os.path.dirname(os.path.abspath(__file__))
OUT_DIR = os.path.join(os.path.dirname(BASE), '04_输出')
SNAP_DIR = os.path.join(os.path.dirname(BASE), '03_分析')

SNAP_FIELDS = ['采集日期', '门店编码', '门店名', '门店类型', 'POI链接',
               '商品名', '原价', '团购价', '已售', '评分', '评价数']

PRODUCTS = ['面部深层清洁', '超微小气泡', '光子嫩肤单次', '水光补水',
            '肩颈舒缓60min', '头皮深层清洁', '果酸焕肤', '淡斑管理',
            '背部疏通', '眼部护理', '小气泡3次卡', '面部抗衰套餐']


def med(xs):
    xs = [x for x in xs if x is not None]
    return statistics.median(xs) if xs else 0.0


def pct(x):
    return f'{x * 100:.0f}%'


# ---------------------------------------------------------------- demo data
def build_demo():
    """构造 10 家本店 x 4~5 家竞对 的模拟快照, 用于验证报告长什么样"""
    rnd = random.Random(20261029)
    today = date.today().isoformat()
    rows = []
    profiles = [
        ('C001', 168, 1.28, 4.62, 0.55),
        ('C002', 198, 0.92, 4.81, 1.35),
        ('C003', 128, 1.05, 4.75, 1.10),
        ('C004', 158, 1.42, 4.51, 0.42),
        ('C005', 178, 0.88, 4.86, 1.62),
        ('C006', 148, 1.12, 4.68, 0.86),
        ('C007', 188, 1.35, 4.58, 0.51),
        ('C008', 138, 0.98, 4.79, 1.24),
        ('C009', 168, 1.08, 4.70, 0.95),
        ('C010', 208, 1.25, 4.55, 0.60),
    ]
    for code, base_price, price_k, rating, strength in profiles:
        n_comp = rnd.choice([4, 4, 5])
        rows.append(dict(采集日期=today, 门店编码=code, 门店名=f'{code}门店',
                         门店类型='本店', POI链接=f'https://v.douyin.com/{code}/',
                         商品名='', 原价=0, 团购价=0, 已售=0,
                         评分=rating, 评价数=rnd.randint(300, 2400)))
        own_n = min(rnd.randint(6, 11), len(PRODUCTS))
        own_sold_total = int(rnd.randint(600, 3000) * strength)
        for nm in rnd.sample(PRODUCTS, own_n):
            price = round(base_price * rnd.uniform(0.6, 1.6))
            rows.append(dict(采集日期=today, 门店编码=code, 门店名=f'{code}门店',
                             门店类型='本店', POI链接=f'https://v.douyin.com/{code}/',
                             商品名=nm, 原价=round(price * rnd.uniform(1.6, 2.8)),
                             团购价=price, 已售=max(3, own_sold_total // own_n + rnd.randint(-40, 40)),
                             评分=rating, 评价数=0))
        for i in range(n_comp):
            cname = f'{code}商圈·竞对{i + 1}'
            c_rating = round(rating + rnd.uniform(-0.35, 0.35), 1)
            c_n = min(rnd.randint(4, 14), len(PRODUCTS))
            comp_pool = int(rnd.randint(500, 3200) / max(strength, 0.3))
            for nm in rnd.sample(PRODUCTS, c_n):
                price = round(base_price * rnd.uniform(0.7, 1.5) / price_k)
                rows.append(dict(采集日期=today, 门店编码=code, 门店名=cname,
                                 门店类型='竞对', POI链接=f'https://v.douyin.com/x{code}{i}/',
                                 商品名=nm, 原价=round(price * rnd.uniform(1.5, 2.6)),
                                 团购价=price, 已售=max(2, comp_pool // c_n + rnd.randint(-30, 30)),
                                 评分=c_rating, 评价数=rnd.randint(100, 2000)))
    return rows


# ---------------------------------------------------------------- analysis
def analyse(rows):
    groups = {}
    for r in rows:
        groups.setdefault(r['门店编码'], []).append(r)

    result = []
    for code, rs in groups.items():
        own = [r for r in rs if r['门店类型'] == '本店' and r['商品名']]
        comp = [r for r in rs if r['门店类型'] == '竞对' and r['商品名']]
        if not own:
            continue
        comp_names = {r['门店名'] for r in comp}
        n = len(comp_names) or 1

        own_price = med([r['团购价'] for r in own])
        comp_price = med([r['团购价'] for r in comp])
        own_cnt = len({r['商品名'] for r in own})
        cnt_by_store = {}
        for r in comp:
            cnt_by_store.setdefault(r['门店名'], set()).add(r['商品名'])
        comp_cnt = med([len(v) for v in cnt_by_store.values()]) if cnt_by_store else own_cnt

        own_sold = sum(r['已售'] for r in own)
        comp_sold = sum(r['已售'] for r in comp)
        total = own_sold + comp_sold or 1
        share = own_sold / total
        share_idx = share * (n + 1)

        price_idx = own_price / comp_price if comp_price else 1.0
        own_rating = next((r['评分'] for r in own), 0)
        comp_rating = med([r['评分'] for r in rs if r['门店类型'] == '竞对'])
        gap = own_rating - (comp_rating or own_rating)

        composite = (0.5 * share_idx
                     + 0.3 * (1 + gap / 0.5 * 0.2)
                     + 0.2 * (1 + (1 - price_idx) * 0.5))

        result.append(dict(
            编码=code, 门店名=f'{code}门店', 竞对数=n,
            本店商品数=own_cnt, 竞对商品数=comp_cnt,
            本店价=own_price, 同城价=comp_price, 价格指数=price_idx,
            本店销量=own_sold, 同城销量=comp_sold, 份额=share, 份额指数=share_idx,
            本店评分=own_rating, 同城评分=comp_rating, 评分差=gap,
            竞争力=composite,
        ))
    result.sort(key=lambda x: x['份额指数'])
    return result


def judge(m):
    """检测规则 -> (标签, 等级, 归因建议)"""
    rules = []
    if m['价格指数'] > 1.2 and m['份额指数'] < 0.8:
        rules.append(('定价偏高', 'danger',
                      f"价格高于同城中位 {pct(m['价格指数'] - 1)}，销量份额仅为平均水平的 {m['份额指数']:.2f} 倍。"
                      f"优先检查主图与团购标题，其次考虑下调入门款价格。"))
    if m['价格指数'] < 0.85 and m['份额指数'] < 0.8:
        rules.append(('性价比未被感知', 'warn',
                      f"价格低于同城 {pct(1 - m['价格指数'])}，但份额只有平均的 {m['份额指数']:.2f} 倍。"
                      f"不是价格问题，是曝光或评价数量不够，检查达人铺设与评价增速。"))
    if m['竞对商品数'] and m['本店商品数'] < m['竞对商品数'] * 0.6:
        rules.append(('品项不足', 'warn',
                      f"在售 {m['本店商品数']} 个，同城中位 {m['竞对商品数']:.0f} 个。"
                      f"补齐引流款+利润款的组合，至少到中位水平。"))
    if m['评分差'] < -0.15:
        rules.append(('口碑落后', 'danger',
                      f"评分低于同城中位 {abs(m['评分差']):.2f} 分。差评未处理会直接压转化，先处理近 30 天差评。"))
    if m['份额指数'] > 1.3 and m['价格指数'] < 1.1:
        rules.append(('优势店', 'good', '价格不高于同城且份额领先，把这家店的商品结构和内容打法复制到同区域其他店。'))
    if not rules:
        rules.append(('与商圈持平', 'flat', '各项指标接近同城水平，暂无明显短板。'))
    return rules


# ---------------------------------------------------------------- render
CSS = """
*{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:1080px;margin:0 auto}
h1{font-size:22px;font-weight:600;letter-spacing:.2px}
.sub{color:#8b93a1;font-size:13px;margin-top:8px}
.kpis{display:grid;grid-template-columns:repeat(4,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}
.kpi .l{color:#8b93a1;font-size:12px;margin-top:2px}
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:right;color:#8b93a1;font-weight:500;font-size:12px;padding:9px 10px;
   border-bottom:1px solid #2a2f3a;white-space:nowrap}
th:first-child,th:nth-child(2){text-align:left}
td{padding:10px;border-bottom:1px solid #1d212a;text-align:right;white-space:nowrap}
td:first-child,td:nth-child(2){text-align:left}
tr:hover td{background:#151920}
.tag{display:inline-block;padding:2px 8px;border-radius:5px;font-size:11px;font-weight:500}
.t-danger{background:rgba(255,107,107,.14);color:#ff8585}
.t-warn{background:rgba(240,168,60,.14);color:#f0b45c}
.t-good{background:rgba(78,201,160,.14);color:#5fd6ae}
.t-flat{background:rgba(140,148,163,.14);color:#9aa3b2}
.cards{display:grid;grid-template-columns:1fr;gap:10px}
.card{background:#171a21;border:1px solid #252a34;border-left:3px solid #2f3745;
      border-radius:10px;padding:14px 18px}
.card.danger{border-left-color:#ff6b6b}
.card.warn{border-left-color:#f0a83c}
.card.good{border-left-color:#4ec9a0}
.card .h{display:flex;align-items:center;gap:10px;flex-wrap:wrap}
.card .nm{font-weight:600;font-size:14px}
.card .msg{color:#a8b0be;font-size:13px;margin-top:6px}
.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}
.up{color:#ff8585}.down{color:#5fd6ae}
"""


def render(metrics, meta):
    total_comp = sum(m['竞对数'] for m in metrics)
    share_avg = statistics.mean([m['份额指数'] for m in metrics]) if metrics else 0
    below = [m for m in metrics if m['份额指数'] < 0.8]
    issues = [(m, r) for m in metrics for r in judge(m) if r[1] in ('danger', 'warn')]

    kpis = [('试点门店', f'{len(metrics)} 家'),
            ('对标门店', f'{total_comp} 家'),
            ('份额落后门店', f'{len(below)} 家'),
            ('平均份额指数', f'{share_avg:.2f}')]

    rows_html = []
    for m in metrics:
        j = judge(m)[0]
        rows_html.append(f"""<tr>
<td>{m['门店名']}</td><td>{m['竞对数']}</td>
<td>{m['本店商品数']} / {m['竞对商品数']:.0f}</td>
<td>{m['本店价']:.0f} / {m['同城价']:.0f}</td>
<td class="{'up' if m['价格指数'] > 1.12 else 'down' if m['价格指数'] < 0.9 else ''}">{m['价格指数']:.2f}</td>
<td>{pct(m['份额'])}</td>
<td class="{'up' if m['份额指数'] < 0.8 else 'down' if m['份额指数'] > 1.2 else ''}">{m['份额指数']:.2f}</td>
<td>{m['本店评分']:.1f} / {m['同城评分']:.1f}</td>
<td>{m['竞争力']:.2f}</td>
<td><span class="tag t-{j[1]}">{j[0]}</span></td>
</tr>""")

    cards_html = []
    for m, (label, level, msg) in issues:
        cards_html.append(f"""<div class="card {level}">
<div class="h"><span class="nm">{m['门店名']}</span><span class="tag t-{level}">{label}</span></div>
<div class="msg">{msg}</div>
<div class="nums">在售 {m['本店商品数']} 品（同城中位 {m['竞对商品数']:.0f}） · 价格指数 {m['价格指数']:.2f} · 份额指数 {m['份额指数']:.2f} · 评分 {m['本店评分']:.1f}</div>
</div>""")
    if not cards_html:
        cards_html.append('<div class="card good"><div class="h"><span class="nm">本期未命中异常规则</span></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>同城对标检测报告</title><style>{CSS}</style></head><body><div class="wrap">
<h1>同城对标检测报告</h1>
<div class="sub">采集日期 {meta['date']} · 口径为抖音前台公开可观测指标 · 仅用于试点验证</div>
<div class="kpis">{''.join(f'<div class="kpi"><div class="v">{v}</div><div class="l">{k}</div></div>' for k, v in kpis)}</div>
<h2><span class="bar"></span>命中异常的 {len(issues)} 项，按优先级处理</h2>
<div class="cards">{''.join(cards_html)}</div>
<h2><span class="bar"></span>对标明细（按份额落后排序）</h2>
<table><thead><tr>
<th>门店</th><th>对标店</th><th>在售品(本/中位)</th><th>价格中位(本/同城)</th><th>价格指数</th>
<th>销量份额</th><th>份额指数</th><th>评分(本/中位)</th><th>竞争力</th><th>判定</th>
</tr></thead><tbody>{''.join(rows_html)}</tbody></table>
<div class="note">
<b>指数说明</b>　份额指数 1.00 = 与同组门店平均份额持平；价格指数 1.00 = 与同城价格中位持平；竞争力指数 = 份额 50% + 口碑 30% + 价格 20%。<br>
<b>口径局限</b>　销量来自前台「已售」累计值，含历史，单次采集不能代表当日销量；要算真实日增需连续两次采集取差值。曝光、转化率、核销率、退款率属商家私有数据，外部采集拿不到，本报告不涉及。<br>
<b>试点性质</b>　{'本页数据为模拟样例，用于确认字段与判定规则是否可用，不代表真实经营结论。' if meta['demo'] else '数据由实际采集生成。'}
</div>
</div></body></html>"""


def main():
    args = sys.argv[1:]
    demo = '--demo' in args
    rows, src = [], ''

    if demo:
        rows = build_demo()
    else:
        path = None
        if '--file' in args:
            path = args[args.index('--file') + 1]
        else:
            files = sorted(f for f in os.listdir(SNAP_DIR) if f.endswith('.csv')) if os.path.isdir(SNAP_DIR) else []
            if files:
                path = os.path.join(SNAP_DIR, files[-1])
        if not path or not os.path.exists(path):
            print('未找到采集快照 CSV。可先用 --demo 查看报告形态。')
            sys.exit(1)
        src = path
        with open(path, encoding='utf-8-sig') as f:
            for r in csv.DictReader(f):
                r['团购价'] = float(r.get('团购价') or 0)
                r['原价'] = float(r.get('原价') or 0)
                r['已售'] = float(r.get('已售') or 0)
                r['评分'] = float(r.get('评分') or 0)
                rows.append(r)

    metrics = analyse(rows)
    if not metrics:
        print('快照里没有可分析的数据，检查「门店类型」是否填了「本店 / 竞对」。')
        sys.exit(1)

    meta = dict(date=rows[0]['采集日期'] if rows else date.today().isoformat(), demo=demo)
    os.makedirs(OUT_DIR, exist_ok=True)
    suffix = '示例' if demo else meta['date'].replace('-', '')
    out = os.path.join(OUT_DIR, f'对标检测报告_{suffix}.html')
    with open(out, 'w', encoding='utf-8') as f:
        f.write(render(metrics, meta))

    print(f'报告已生成: {out}')
    print(f'分析门店 {len(metrics)} 家, 竞对 {sum(m["竞对数"] for m in metrics)} 家')
    for m in metrics[:5]:
        print(f"  {m['门店名']}  份额指数 {m['份额指数']:.2f}  价格指数 {m['价格指数']:.2f}  {judge(m)[0][0]}")
    if src:
        print(f'数据源: {src}')


if __name__ == '__main__':
    main()