antissoft / lvqianmeiye_ERP

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""按门店统计会员分层（截止 2025-12 ~ 2026-05 各月末，全部为累计口径）。MySQL 5.7 兼容。

累计口径：
- 总人数：截止日前已注册的有效会员(khlx=3)
- 新客：截止日前在该店有过「首次有效成交」的去重人头（历史累计，非当月）
- 常到店/预警/灰/黑客：截止日当天的分层快照（互斥）
- 退款：截止日前在该店有过退卡的去重人头（历史累计，非当月）
"""
import csv
import os
from collections import defaultdict
from datetime import date

import pymysql

DB = dict(
    host="rm-2vccze142rc9a8f58bo.mysql.cn-chengdu.rds.aliyuncs.com",
    user="nettest",
    password="nettest",
    database="lqerp",
    port=3306,
    charset="utf8mb4",
    connect_timeout=30,
    read_timeout=600,
)

# (年, 月) 列表
PERIODS = [(2025, 12), (2026, 1), (2026, 2), (2026, 3), (2026, 4), (2026, 5)]
DAYS_ACTIVE = 90
DAYS_WARN_MAX = 150

ROOT = os.path.dirname(
    os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
)
OUT_CSV = os.path.join(
    ROOT, "ExportFiles", "门店会员分层统计_2025年12月-2026年5月截止_累计.csv"
)


def month_end(y: int, m: int) -> date:
    if m == 12:
        return date(y, 12, 31)
    return date(y, m + 1, 1).fromordinal(date(y, m + 1, 1).toordinal() - 1)


def classify(days_since: int, remain: float) -> str:
    if days_since <= DAYS_ACTIVE:
        return "active"
    if days_since <= DAYS_WARN_MAX:
        return "warn"
    if remain > 0:
        return "gray"
    return "black"


def run_month(cur, cutoff: date, label: str) -> list:
    cutoff_s = f"{cutoff} 23:59:59"

    cur.execute(
        """
        SELECT F_Id AS store_id, dm AS store_name FROM lq_mdxx
        """
    )
    store_names = {r["store_id"]: r["store_name"] for r in cur.fetchall()}

    cur.execute(
        """
        SELECT kh.F_Id AS member_id, kh.gsmd AS store_id, kh.F_CreateTime AS create_time
        FROM lq_khxx kh
        WHERE kh.F_IsEffective = 1 AND kh.khlx = '3'
          AND kh.gsmd IS NOT NULL AND kh.gsmd <> ''
          AND kh.F_CreateTime <= %s
        """,
        (cutoff_s,),
    )
    members = cur.fetchall()
    if not members:
        return []

    member_ids = [m["member_id"] for m in members]

    cur.execute(
        """
        SELECT xh.hy AS member_id, MAX(xh.hksj) AS last_hksj
        FROM lq_xh_hyhk xh
        WHERE xh.F_IsEffective = 1 AND xh.hksj <= %s
        GROUP BY xh.hy
        """,
        (cutoff_s,),
    )
    last_visit = {r["member_id"]: r["last_hksj"] for r in cur.fetchall()}

    # 剩余项目次数：开单-消耗-退卡（按会员+门店）
    remain_sql = """
        SELECT t.member_id, t.store_id,
               GREATEST(COALESCE(b.cnt,0)-COALESCE(c.cnt,0)-COALESCE(r.cnt,0), 0) AS remain_cnt
        FROM (
            SELECT DISTINCT m.F_Id AS member_id, m.gsmd AS store_id
            FROM lq_khxx m
            WHERE m.F_Id IN ({ids})
        ) t
        LEFT JOIN (
            SELECT px.F_MemberId AS member_id, kd.djmd AS store_id,
                   SUM(CAST(px.F_ProjectNumber AS DECIMAL(18,2))) AS cnt
            FROM lq_kd_pxmx px
            INNER JOIN lq_kd_kdjlb kd ON px.glkdbh = kd.F_Id
            WHERE px.F_IsEffective = 1 AND kd.F_IsEffective = 1
              AND kd.kdrq <= %s AND px.F_MemberId IN ({ids})
            GROUP BY px.F_MemberId, kd.djmd
        ) b ON b.member_id = t.member_id AND b.store_id = t.store_id
        LEFT JOIN (
            SELECT px.F_MemberId AS member_id, xh.md AS store_id,
                   SUM(CAST(px.F_ProjectNumber AS DECIMAL(18,2))) AS cnt
            FROM lq_xh_pxmx px
            INNER JOIN lq_xh_hyhk xh ON px.F_ConsumeInfoId = xh.F_Id
            WHERE px.F_IsEffective = 1 AND xh.F_IsEffective = 1
              AND xh.hksj <= %s AND px.F_MemberId IN ({ids})
            GROUP BY px.F_MemberId, xh.md
        ) c ON c.member_id = t.member_id AND c.store_id = t.store_id
        LEFT JOIN (
            SELECT px.F_MemberId AS member_id, hytk.md AS store_id,
                   SUM(CAST(mx.F_ProjectNumber AS DECIMAL(18,2))) AS cnt
            FROM lq_hytk_mx mx
            INNER JOIN lq_hytk_hytk hytk ON mx.F_RefundInfoId = hytk.F_Id
            INNER JOIN lq_kd_pxmx px ON mx.F_BillingItemId = px.F_Id AND px.F_IsEffective = 1
            WHERE mx.F_IsEffective = 1 AND hytk.F_IsEffective = 1
              AND hytk.tksj <= %s AND px.F_MemberId IN ({ids})
            GROUP BY px.F_MemberId, hytk.md
        ) r ON r.member_id = t.member_id AND r.store_id = t.store_id
    """

    remain_map = {}
    batch = 800
    for i in range(0, len(member_ids), batch):
        chunk = member_ids[i : i + batch]
        ph = ",".join(["%s"] * len(chunk))
        sql = remain_sql.format(ids=ph)
        params = chunk + [cutoff_s] + chunk + [cutoff_s] + chunk + [cutoff_s] + chunk
        cur.execute(sql, params)
        for r in cur.fetchall():
            remain_map[(r["member_id"], r["store_id"])] = float(r["remain_cnt"] or 0)

    tier = defaultdict(
        lambda: dict(
            total=0, active=0, warn=0, gray=0, black=0, new=0, refund=0
        )
    )

    for m in members:
        sid = m["store_id"]
        mid = m["member_id"]
        lv = last_visit.get(mid)
        anchor = lv if lv else m["create_time"]
        if anchor is None:
            days_since = 9999
        else:
            if hasattr(anchor, "date"):
                days_since = (cutoff - anchor.date()).days
            else:
                days_since = (cutoff - anchor).days

        rem = remain_map.get((mid, sid), 0.0)
        tier[sid]["total"] += 1
        cat = classify(days_since, rem)
        tier[sid][cat] += 1

    # 累计新客：截至 cutoff，在该店首次有效成交的去重人头
    cur.execute(
        """
        SELECT first_bill.store_id, COUNT(*) AS cnt
        FROM (
            SELECT kd.kdhy AS member_id, kd.djmd AS store_id, MIN(kd.kdrq) AS first_kdrq
            FROM lq_kd_kdjlb kd
            WHERE kd.F_IsEffective = 1 AND kd.sfyj > 0
              AND kd.djmd IS NOT NULL AND kd.djmd <> ''
            GROUP BY kd.kdhy, kd.djmd
        ) first_bill
        WHERE first_bill.first_kdrq <= %s
        GROUP BY first_bill.store_id
        """,
        (cutoff_s,),
    )
    for r in cur.fetchall():
        tier[r["store_id"]]["new"] = int(r["cnt"])

    # 累计退款：截至 cutoff，在该店有过退卡的去重人头
    cur.execute(
        """
        SELECT hytk.md AS store_id, COUNT(DISTINCT hytk.hy) AS cnt
        FROM lq_hytk_hytk hytk
        WHERE hytk.F_IsEffective = 1
          AND hytk.tksj <= %s
          AND hytk.md IS NOT NULL AND hytk.md <> ''
        GROUP BY hytk.md
        """,
        (cutoff_s,),
    )
    for r in cur.fetchall():
        tier[r["store_id"]]["refund"] = int(r["cnt"])

    rows = []
    all_store_ids = set(tier.keys()) | set(store_names.keys())
    for sid in all_store_ids:
        t = tier[sid]
        if t["total"] == 0 and t["new"] == 0 and t["refund"] == 0:
            continue
        rows.append(
            {
                "截止月份": label,
                "门店ID": sid,
                "门店名称": store_names.get(sid, sid),
                "总人数": t["total"],
                "新客": t["new"],
                "常到店": t["active"],
                "预警客": t["warn"],
                "灰客": t["gray"],
                "黑客": t["black"],
                "退款": t["refund"],
            }
        )
    rows.sort(key=lambda x: (-x["总人数"], x["门店名称"]))
    return rows


def main():
    os.makedirs(os.path.join(ROOT, "ExportFiles"), exist_ok=True)
    conn = pymysql.connect(**DB, cursorclass=pymysql.cursors.DictCursor)
    all_rows = []
    try:
        with conn.cursor() as cur:
            for y, m in PERIODS:
                cutoff = month_end(y, m)
                label = f"{y}年{m}月截止(累计)"
                print(f"统计 {label} ...", flush=True)
                all_rows.extend(run_month(cur, cutoff, label))
    finally:
        conn.close()

    fields = [
        "截止月份",
        "门店ID",
        "门店名称",
        "总人数",
        "新客",
        "常到店",
        "预警客",
        "灰客",
        "黑客",
        "退款",
    ]
    with open(OUT_CSV, "w", newline="", encoding="utf-8-sig") as f:
        w = csv.DictWriter(f, fieldnames=fields)
        w.writeheader()
        w.writerows(all_rows)

    print(f"已写入 {OUT_CSV}，共 {len(all_rows)} 行\n全公司汇总：")
    agg = defaultdict(lambda: defaultdict(int))
    for row in all_rows:
        k = row["截止月份"]
        for col in fields[3:]:
            agg[k][col] += row[col]
    for label in sorted(agg.keys()):
        a = agg[label]
        print(
            f"  {label}: 总{a['总人数']} 新{a['新客']} 常{a['常到店']} "
            f"预警{a['预警客']} 灰{a['灰客']} 黑{a['黑客']} 退{a['退款']}"
        )


if __name__ == "__main__":
    main()