Neo-ZQYY/scripts/ops/monthly_revenue_report.py

"""月度经营数据统计报告

按月汇总 2026年1月~3月 的实际收入进项，
日切 08:00 ~ 次日 08:00，输出 markdown 表格到 docs/reports/monthly-revenue-q1.md
"""
import os, sys
from pathlib import Path
from datetime import date, timedelta
from dotenv import load_dotenv

load_dotenv(Path(__file__).resolve().parents[2] / ".env")

dsn = os.environ.get("TEST_DB_DSN")
if not dsn:
    dsn = os.environ.get("PG_DSN", "")
    if "etl_feiqiu" in dsn:
        dsn = dsn.replace("etl_feiqiu", "test_etl_feiqiu")
if not dsn:
    sys.exit("ERROR: TEST_DB_DSN / PG_DSN not set")

os.environ.setdefault("PGCLIENTENCODING", "UTF8")
import psycopg2

today = date.today()

# 三个月的起止日期
months = [
    ("1月", date(2026, 1, 1), date(2026, 1, 31)),
    ("2月", date(2026, 2, 1), date(2026, 2, 28)),
    ("3月", date(2026, 3, 1), min(today, date(2026, 3, 31))),
]

sql = """
WITH params AS (
  SELECT
    %(label)s AS month_label,
    (%(start)s::text || ' 08:00:00+08')::timestamptz AS period_start,
    ((%(end)s::date + interval '1 day')::date::text || ' 08:00:00+08')::timestamptz AS period_end
),
cash_online AS (
  SELECT COALESCE(SUM(h.pay_amount), 0) AS v
  FROM params p, dwd.dwd_settlement_head h
  WHERE h.create_time >= p.period_start AND h.create_time < p.period_end
    AND h.settle_type IN (1, 3)
),
recharge AS (
  SELECT COALESCE(SUM(ro.pay_amount), 0) AS v
  FROM params p, dwd.dwd_recharge_order ro
  WHERE ro.create_time >= p.period_start AND ro.create_time < p.period_end
    AND ro.settle_type = 5
),
groupbuy AS (
  SELECT COALESCE(SUM(pcr.sale_price * 0.75), 0) AS v
  FROM params p, dwd.dwd_platform_coupon_redemption pcr
  WHERE pcr.create_time >= p.period_start AND pcr.create_time < p.period_end
    AND pcr.is_delete = 0
),
member_guest AS (
  SELECT
    COUNT(*) FILTER (WHERE h.member_id > 0) AS member_v,
    COUNT(*) FILTER (WHERE h.member_id = 0) AS guest_v
  FROM params p, dwd.dwd_settlement_head h
  WHERE h.create_time >= p.period_start AND h.create_time < p.period_end
    AND h.settle_type IN (1, 3)
),
new_member AS (
  SELECT COUNT(DISTINCT m.member_id) AS v
  FROM params p, dwd.dim_member m
  WHERE m.create_time >= p.period_start AND m.create_time < p.period_end
    AND m.scd2_is_current = 1
),
old_recharge AS (
  SELECT
    COUNT(DISTINCT ro.member_id) FILTER (WHERE ro.member_id > 0) AS persons,
    COALESCE(SUM(ro.pay_amount) FILTER (WHERE ro.member_id > 0), 0) AS amount
  FROM params p, dwd.dwd_recharge_order ro
  WHERE ro.create_time >= p.period_start AND ro.create_time < p.period_end
    AND ro.settle_type = 5
)
SELECT
  (SELECT month_label FROM params),
  ROUND(co.v + r.v + gb.v, 2) AS total,
  co.v AS cash_online,
  r.v AS recharge,
  ROUND(gb.v, 2) AS groupbuy,
  mg.member_v, mg.guest_v,
  nm.v AS new_members,
  omr.persons AS old_recharge_persons,
  omr.amount AS old_recharge_amount
FROM cash_online co, recharge r, groupbuy gb, member_guest mg, new_member nm, old_recharge omr;
"""

conn = psycopg2.connect(dsn, client_encoding="UTF8")
cur = conn.cursor()

results = []
for label, start, end in months:
    cur.execute(sql, {"label": label, "start": start.isoformat(), "end": end.isoformat()})
    results.append(cur.fetchone())

conn.close()

def fmt_money(v):
    return f"{v:,.2f}"

def fmt_int(v):
    return str(int(v))

# 生成 markdown
lines = []
lines.append(f"# 月度经营数据统计（2026年1月~3月）")
lines.append("")
lines.append("> 数据来源：test_etl_feiqiu (DWD schema)")
lines.append(f"> 统计时间：{today.isoformat()}")
lines.append("> 日切规则：每天 08:00 ~ 次日 08:00（Asia/Shanghai）")
lines.append(f"> 3月数据截至：{months[2][2].isoformat()}")
lines.append("")
lines.append("## 月度汇总")
lines.append("")
lines.append("| 月份 | 实收合计 | 实收（纸币+线上） | 充值 | 团购结算 | 会员到店人次 | 散客到店人次 | 新会员注册数 | 老会员充值人数 | 老会员充值总额 |")
lines.append("|------|--------:|-----------------:|-----:|---------:|------------:|------------:|------------:|--------------:|--------------:|")

totals = [0] * 9  # 9 个数值列的合计

for row in results:
    label, total, cash_online, recharge, groupbuy, member_v, guest_v, new_members, old_persons, old_amount = row
    vals = [float(total), float(cash_online), float(recharge), float(groupbuy),
            int(member_v), int(guest_v), int(new_members), int(old_persons), float(old_amount)]
    for i in range(9):
        totals[i] += vals[i]
    lines.append(
        f"| {label} | {fmt_money(vals[0])} | {fmt_money(vals[1])} | {fmt_money(vals[2])} | {fmt_money(vals[3])} "
        f"| {fmt_int(vals[4])} | {fmt_int(vals[5])} | {fmt_int(vals[6])} | {fmt_int(vals[7])} | {fmt_money(vals[8])} |"
    )

# 合计行
lines.append(
    f"| **Q1合计** | {fmt_money(totals[0])} | {fmt_money(totals[1])} | {fmt_money(totals[2])} | {fmt_money(totals[3])} "
    f"| {fmt_int(totals[4])} | {fmt_int(totals[5])} | {fmt_int(totals[6])} | {fmt_int(totals[7])} | {fmt_money(totals[8])} |"
)

lines.append("")
lines.append('> 金额单位：元。实收合计 = 实收（纸币+线上） + 充值 + 团购结算')
lines.append('> 口径定义与 daily-revenue-latest.md 一致')

repo_root = Path(__file__).resolve().parents[2]
out_path = repo_root / "docs" / "reports" / "monthly-revenue-q1.md"
out_path.parent.mkdir(parents=True, exist_ok=True)
out_path.write_text("\n".join(lines), encoding="utf-8")
print(f"报告已生成: {out_path}")