yhm-shop-data/store-ops/tools/gen_rules_doc.py

# -*- coding: utf-8 -*-
"""把当前生效的检测规则导出成本地文档（DOCX + Markdown）
用法：python3 tools/gen_rules_doc.py
"""
import os, sys, json
from datetime import datetime

BASE = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, os.path.join(BASE, "backend"))
os.chdir(BASE)

import re
import rules, db, codesync


def clean(a):
    """去掉动作行首的 ✅/✓，与工作台展示保持一致"""
    return re.sub(r"^[\s\u2705\u2714\u2611\ufe0f]+", "", str(a or "")).strip()

LAYER = [("traffic", "① 流量层 · 能不能被看到"),
         ("convert", "② 转化层 · 看到了有没有下单"),
         ("fulfill", "③ 履约层 · 下单了有没有核销"),
         ("repute", "④ 口碑层 · 体验好不好")]
TIER = {"R": "违规红线", "B": "考核未达标", "T": "运营提升", "N": "样本不足"}
COND = {
    "TD":   "商责退单率 > 1.27% / 3.18% / 6.35%（预警 / 一般严重 / 非常严重）",
    "TS":   "投诉率 > 0.11% / 0.28% / 0.56%（三级）",
    "CP":   "新增评价 ≥10 → 新增口径；否则累计评价 ≥20 → 累计口径；都不足仍判定但最高只到预警线；超 0.66% / 1.66% / 3.31%",
    "SCORE":"评分 < 4.3（0 分按「未开分」处理，文案与动作单独走）",
    "REPLY":"新增评价 ≥1 且 评价回复率 < 50%",
    "RVOL": "新增评价数 < 10",
    "DEAL0":"访问 > 0 且 门店页成交券数 = 0",
    "ZERO": "门店页成交券数 > 0 且 核销券数 = 0（履约层内命中此项则不再判「核销率低」）",
    "HX":   "门店页成交券数 ≥ 5 且 核销率 < 40%",
    "AMT":  "核销金额 < 10,000 元（独立判定，不受核销率约束）",
    "VISIT":"访问次数 < 100（含 0 访问）",
    "VIDEO":"新发布门店关联视频数 < 10",
    "LIT":  "门店点亮数 < 10 且 访问 > 0",
    "CVR":  "访问 ≥100 且 成交券数 > 0 且 成交券数 ÷ 访问 < 10%",
    "PEND": "估算未核销券（门店页成交券数 − 核销券数）≥ 10 张",
    "RANK0": "上榜信息为空 / 「-」，即好评榜、人气榜均未进入",
    "RANK_IN": "上榜信息只有「入围XX榜」，没有「第 N 名」",
}
TREE = [
    ("① 有 R 级违规（退单 / 投诉 / 差评）", "平台已在处置或限流", "先止血：处理违规项，再谈增长"),
    ("② 零访问 / 访问过低", "没有流量进来，后面几步都无从谈起", "先修曝光（点亮 + 素材 + 榜单）"),
    ("③ 零成交 / 转化率低", "有流量但没下单", "修转化（品项、价格、主图）"),
    ("④ 成交未核销 / 核销率低 / 核销金额低", "客户买了不到店，或到店不核销", "修履约（集中邀约 + 核销激励）"),
    ("⑤ 评分不达标 / 回复率低", "口碑拖累自然流量与转化", "修口碑（处理差评 + 专人回复）"),
    ("⑥ 以上都不命中", "本期达标", "保持动作，关注冲刺目标"),
]
OUTPUT = [
    ("一句话结论", "按诊断树选出的主因生成，如「卡在「履约」：核销率 11%（基准线 40%）」"),
    ("主因", "诊断树从上往下第一个命中的问题"),
    ("全部问题", "命中几个列几个，不再截断"),
    ("整改动作", "每个问题一条对应动作，去重后输出"),
]


def collect():
    T = db.get_thresholds()
    X = db.get_texts()
    period = db.latest_period()
    rows = db.aggregate_period(period)
    hits, sample = {}, {}
    for m in rows:
        d = rules.diagnose(m, T, X)
        for p in d["problems"]:
            hits[p["key"]] = hits.get(p["key"], 0) + 1
            sample.setdefault(p["key"], p)
    groups = []
    for key, name in LAYER:
        items = []
        for rk, tier, layer in rules.RULES:
            if layer != key:
                continue
            p = sample.get(rk)
            items.append({
                "id": rk, "tier": tier, "name": rules.RULE_CAT.get(rk, rk),
                "cond": COND.get(rk, ""),
                "problem": (p["text"] if p else (X.get("t_" + rk) or "")),
                "action": clean(p["action"] if p else (X.get("a_" + rk) or "")),
                "hits": hits.get(rk, 0),
            })
        groups.append((name, items))
    params = [(k, T.get(k), codesync.get_factory().get(k)) for k in rules.DEFAULT_THRESHOLDS]
    return {"period": period, "stores": len(rows), "groups": groups, "params": params,
            "tree": TREE, "output": OUTPUT}


def build_md(d):
    L = ["# 门店检测工作台 · 检测规则说明", "",
         "> 数据周期 %s · %d 家门店 · 导出时间 %s" %
         (d["period"], d["stores"], datetime.now().strftime("%Y-%m-%d %H:%M")), ""]
    L += ["## 一、核心口径", "",
          "- 核销率 = 核销券数 ÷ **门店页成交券数**", "- 待核销 = 门店页成交券数 − 核销券数（估算，未扣退款）",
          "- 差评率：新增评价 ≥10 用新增口径；否则累计 ≥20 用累计口径；都不足仍判定但最高只到预警线",
          "- 评分 0 分 = 未开分，仍计入「评分不达标」", "- 评价回复率：只在有新增评价时判定",
          "- **四层并行**：流量 / 转化 / 履约 / 口碑各自独立判定、互不压制；只有履约层内部「成交未核销 → 核销率低」按根因去重",
          ""]
    L += ["## 二、规则明细", ""]
    for name, items in d["groups"]:
        L += ["### " + name, "", "| ID | 档位 | 指标 | 判定条件 | 问题文案 | 整改动作 | 命中 |",
              "|---|---|---|---|---|---|---|"]
        for it in items:
            L.append("| %s | %s | %s | %s | %s | %s | %d 家 |" %
                     (it["id"], TIER.get(it["tier"], it["tier"]), it["name"], it["cond"],
                      it["problem"], it["action"].replace("\n", "<br>"), it["hits"]))
        L.append("")
    L += ["## 三、可调阈值", "", "| 参数 | 当前值 | 出厂值 |", "|---|---|---|"]
    for k, cur, fac in d["params"]:
        L.append("| %s | %s | %s |" % (k, cur, fac))
    L += ["", "## 四、诊断树（主因判定顺序）", "", "| 顺序 | 命中条件 | 含义 | 处理方向 |", "|---|---|---|---|"]
    for i, (a, b, c) in enumerate(d["tree"], 1):
        L.append("| %d | %s | %s | %s |" % (i, a, b, c))
    L += ["", "## 五、输出结构", "", "| 字段 | 说明 |", "|---|---|"]
    for a, b in d["output"]:
        L.append("| %s | %s |" % (a, b))
    return "\n".join(L)


def build_docx(d, path):
    from docx import Document
    from docx.shared import Pt, Cm, RGBColor
    from docx.enum.text import WD_ALIGN_PARAGRAPH
    from docx.oxml.ns import qn

    from docx.enum.section import WD_ORIENT
    from docx.oxml import OxmlElement

    doc = Document()
    sec = doc.sections[0]
    # 横向 A4：规则表列多，竖版会把「整改动作」挤成竖条
    sec.orientation = WD_ORIENT.LANDSCAPE
    sec.page_width, sec.page_height = Cm(29.7), Cm(21.0)
    sec.left_margin = sec.right_margin = Cm(1.5)
    sec.top_margin = sec.bottom_margin = Cm(1.5)

    st = doc.styles["Normal"]
    st.font.name = "PingFang SC"
    st.font.size = Pt(10)
    st._element.rPr.rFonts.set(qn("w:eastAsia"), "PingFang SC")
    st.paragraph_format.space_after = Pt(4)
    st.paragraph_format.line_spacing = 1.35

    def cn(run, size=None, bold=None, color=None):
        run.font.name = "PingFang SC"
        run._element.rPr.rFonts.set(qn("w:eastAsia"), "PingFang SC")
        if size: run.font.size = Pt(size)
        if bold is not None: run.font.bold = bold
        if color: run.font.color.rgb = RGBColor(*color)

    t = doc.add_paragraph()
    cn(t.add_run("门店检测工作台 · 检测规则说明"), 18, True, (0x21, 0x1a, 0x10))
    t.alignment = WD_ALIGN_PARAGRAPH.LEFT
    meta = doc.add_paragraph()
    cn(meta.add_run("数据周期 %s · %d 家门店 · 导出时间 %s" %
                    (d["period"], d["stores"], datetime.now().strftime("%Y-%m-%d %H:%M"))), 9, False, (0x93, 0x87, 0x6f))

    def h(text, size=13):
        p = doc.add_paragraph()
        p.paragraph_format.space_before = Pt(14)
        p.paragraph_format.space_after = Pt(6)
        cn(p.add_run(text), size, True, (0x21, 0x1a, 0x10))
        return p

    def table(headers, rows, widths):
        tb = doc.add_table(rows=1, cols=len(headers))
        tb.style = "Table Grid"
        tb.autofit = False
        # 固定布局，否则 Word 会按内容重新分配列宽
        tblPr = tb._tbl.tblPr
        layout = OxmlElement("w:tblLayout")
        layout.set(qn("w:type"), "fixed")
        tblPr.append(layout)
        tblW = OxmlElement("w:tblW")
        tblW.set(qn("w:type"), "dxa")
        tblW.set(qn("w:w"), str(int(sum(widths) * 567)))
        tblPr.append(tblW)
        for i, w in enumerate(widths):
            tb.columns[i].width = Cm(w)
            for cell in tb.columns[i].cells:
                cell.width = Cm(w)
        for i, hd in enumerate(headers):
            c = tb.rows[0].cells[i]
            c.text = ""
            cn(c.paragraphs[0].add_run(hd), 9, True)
        for row in rows:
            cells = tb.add_row().cells
            for i, v in enumerate(row):
                cells[i].width = Cm(widths[i])
                cells[i].text = ""
                p = cells[i].paragraphs[0]
                p.paragraph_format.space_after = Pt(2)
                cn(p.add_run(str(v)), 8.5)
        return tb

    # 一、口径
    h("一、核心口径")
    for line in [
        "核销率 = 核销券数 ÷ 门店页成交券数（分母为 0 时不判定）",
        "待核销 = 门店页成交券数 − 核销券数（估算值，未扣退款，仅作邀约优先级参考）",
        "差评率：新增评价 ≥10 用新增口径；否则累计评价 ≥20 用累计口径；都不足仍判定，但最高只到预警线",
        "评分：取周期内最近一次有值；0 分视为「未开分」，仍计入「评分不达标」",
        "评价回复率：只在有新增评价时判定（没有新评价就没有回复义务）",
        "四层并行：流量 / 转化 / 履约 / 口碑各自独立判定、互不压制；只有履约层内部「成交未核销 → 核销率低」按根因去重",
    ]:
        p = doc.add_paragraph(style="List Bullet")
        cn(p.add_run(line), 9.5)

    # 二、规则明细
    h("二、规则明细（共 %d 条）" % sum(len(i) for _, i in d["groups"]))
    p0 = doc.add_paragraph()
    cn(p0.add_run("说明：「问题文案」列给的是按当前数据渲染出的真实示例，实际输出时会替换成各门店自己的数值。"),
       8.5, False, (0x93, 0x87, 0x6f))
    for name, items in d["groups"]:
        p = doc.add_paragraph()
        p.paragraph_format.space_before = Pt(10)
        cn(p.add_run(name), 10.5, True, (0x8a, 0x6a, 0x2f))
        table(["ID", "档位", "指标", "判定条件", "问题文案", "整改动作", "命中"],
              [[it["id"], TIER.get(it["tier"], it["tier"]), it["name"], it["cond"],
                it["problem"], it["action"].replace("\n", " / "), "%d 家" % it["hits"]]
               for it in items],
              [1.4, 1.9, 2.4, 7.0, 5.4, 6.2, 1.4])

    # 三、阈值
    h("三、可调阈值（%d 项）" % len(d["params"]))
    doc.add_paragraph().add_run()
    p = doc.add_paragraph()
    cn(p.add_run("在【指标配置】面板改数字即可，保存时会同步写回代码默认值。"), 9, False, (0x93, 0x87, 0x6f))
    table(["参数", "当前值", "出厂值"],
          [[k, str(cur), str(fac)] for k, cur, fac in d["params"]],
          [9.0, 4.0, 4.0])

    # 四、诊断树
    h("四、诊断树（主因判定顺序）")
    table(["顺序", "命中条件", "含义", "处理方向"],
          [[str(i), a, b, c] for i, (a, b, c) in enumerate(d["tree"], 1)],
          [1.2, 9.5, 7.0, 8.0])

    # 五、输出
    h("五、每家门店的输出结构")
    table(["字段", "说明"], [[a, b] for a, b in d["output"]], [3.2, 14.2])

    doc.save(path)
    return path


if __name__ == "__main__":
    d = collect()
    out = os.path.join(os.path.dirname(BASE), "检测规则说明_%s" % datetime.now().strftime("%Y%m%d"))
    md = out + ".md"
    open(md, "w", encoding="utf-8").write(build_md(d))
    print(md)
    docx_path = out + ".docx"
    build_docx(d, docx_path)
    print(docx_path)