yhm-shop-data/store-ops/backend/db.py

# -*- coding: utf-8 -*-
"""
SQLite 存储层
- stores          门店主数据
- metrics         指标明细（按 门店 × 周期 × 日期 × 粒度 唯一）
- batches         导入批次
- thresholds      考核指标阈值（单行 JSON）
- settings        杂项键值

日粒度数据按月聚合，聚合口径统一在这里，前端不重复算。
"""
import json
import os
import re
import sqlite3
from datetime import datetime

import rules

BASE = os.path.dirname(os.path.abspath(__file__))
DATA_DIR = os.path.join(os.path.dirname(BASE), "data")
DB_PATH = os.path.join(DATA_DIR, "app.db")
INBOX = os.path.join(DATA_DIR, "inbox")
PROCESSED = os.path.join(DATA_DIR, "processed")

SCHEMA = """
CREATE TABLE IF NOT EXISTS stores (
  id          TEXT PRIMARY KEY,
  name        TEXT NOT NULL,
  city        TEXT,
  status      TEXT DEFAULT 'active',
  first_seen  TEXT,
  last_seen   TEXT
);
CREATE TABLE IF NOT EXISTS metrics (
  store_id      TEXT NOT NULL,
  period        TEXT NOT NULL,          -- YYYY-MM（月）或 截至日 YYYY-MM-DD（快照）
  date          TEXT NOT NULL DEFAULT '',-- YYYY-MM-DD；月汇总行留空
  granularity   TEXT NOT NULL,          -- daily | period | snapshot
  window_start  TEXT DEFAULT '',        -- 快照：统计窗口开始日
  window_end    TEXT DEFAULT '',        -- 快照：统计窗口结束日（= 数据截至日）
  visit         INTEGER DEFAULT 0,
  deal          INTEGER DEFAULT 0,
  deal_refund   INTEGER DEFAULT 0,
  intend_deal   INTEGER DEFAULT 0,
  used_amount   REAL    DEFAULT 0,
  used_coupons  INTEGER DEFAULT 0,
  intend_refund INTEGER DEFAULT 0,
  videos        INTEGER DEFAULT 0,
  video_views   INTEGER DEFAULT 0,
  score         REAL,
  reviews_new   INTEGER DEFAULT 0,
  good_new      INTEGER DEFAULT 0,
  bad_new       INTEGER DEFAULT 0,
  reviews_total INTEGER DEFAULT 0,
  good_total    INTEGER DEFAULT 0,
  bad_total     INTEGER DEFAULT 0,
  reply_rate    REAL,
  complaint_rate REAL,
  return_rate   REAL,
  lit           INTEGER DEFAULT 0,
  lit_posts     INTEGER DEFAULT 0,
  favs          INTEGER DEFAULT 0,
  rank_note     TEXT,
  batch_id      INTEGER,
  PRIMARY KEY (store_id, period, date, granularity)
);
CREATE INDEX IF NOT EXISTS idx_metrics_period ON metrics(period, granularity);
CREATE TABLE IF NOT EXISTS batches (
  id          INTEGER PRIMARY KEY AUTOINCREMENT,
  filename    TEXT,
  rows_read   INTEGER,
  rows_written INTEGER,
  new_stores  INTEGER,
  period      TEXT,
  date_from   TEXT,
  date_to     TEXT,
  created_at  TEXT,
  note        TEXT,
  file_path   TEXT          -- 原始文件在 data/processed 下的文件名
);

CREATE TABLE IF NOT EXISTS users (
  id            INTEGER PRIMARY KEY AUTOINCREMENT,
  username      TEXT UNIQUE NOT NULL,
  pass_hash     TEXT NOT NULL,
  salt          TEXT NOT NULL,
  role          TEXT NOT NULL DEFAULT 'viewer',
  display_name  TEXT DEFAULT '',
  active        INTEGER DEFAULT 1,
  must_change   INTEGER DEFAULT 0,
  created_at    TEXT,
  last_login    TEXT
);

CREATE TABLE IF NOT EXISTS sessions (
  token_hash  TEXT PRIMARY KEY,
  user_id     INTEGER NOT NULL,
  created_at  TEXT,
  expires_at  TEXT,
  ip          TEXT DEFAULT ''
);
CREATE TABLE IF NOT EXISTS thresholds (
  id         INTEGER PRIMARY KEY CHECK (id = 1),
  data       TEXT NOT NULL,
  updated_at TEXT
);
CREATE TABLE IF NOT EXISTS settings (
  key   TEXT PRIMARY KEY,
  value TEXT
);
"""

METRIC_FIELDS = ["visit", "deal", "deal_refund", "intend_deal", "used_amount",
                 "used_coupons", "intend_refund", "videos", "video_views", "score",
                 "reviews_new", "good_new", "bad_new", "reviews_total", "good_total",
                 "bad_total", "reply_rate", "complaint_rate", "return_rate",
                 "lit", "lit_posts", "favs", "rank_note"]


def connect():
    os.makedirs(DATA_DIR, exist_ok=True)
    os.makedirs(INBOX, exist_ok=True)
    os.makedirs(PROCESSED, exist_ok=True)
    conn = sqlite3.connect(DB_PATH, timeout=15)
    conn.row_factory = sqlite3.Row
    conn.execute("PRAGMA journal_mode=WAL")
    return conn


def _migrate(conn):
    """老库补齐新增列（SQLite 的 ADD COLUMN 不会报错重复）"""
    cols = {r[1] for r in conn.execute("PRAGMA table_info(metrics)")}
    for c in ("window_start", "window_end"):
        if c not in cols:
            conn.execute("ALTER TABLE metrics ADD COLUMN %s TEXT DEFAULT ''" % c)
    bcols = {r[1] for r in conn.execute("PRAGMA table_info(batches)")}
    if "file_path" not in bcols:
        conn.execute("ALTER TABLE batches ADD COLUMN file_path TEXT")


def backfill_batch_files(conn):
    """老批次补上「原始文件」路径：用 created_at 的时间戳去匹配 processed 里的文件名前缀"""
    try:
        rows = conn.execute("SELECT id, created_at FROM batches "
                            "WHERE file_path IS NULL OR file_path=''").fetchall()
        if not rows or not os.path.isdir(PROCESSED):
            return
        idx = {}
        for f in os.listdir(PROCESSED):
            m = re.match(r"^(\d{8}_\d{6})_", f)
            if m:
                idx[m.group(1)] = f
        for r in rows:
            try:
                stamp = datetime.fromisoformat(r["created_at"]).strftime("%Y%m%d_%H%M%S")
            except (TypeError, ValueError):
                continue
            f = idx.get(stamp)
            if f:
                conn.execute("UPDATE batches SET file_path=? WHERE id=?", (f, r["id"]))
    except Exception:
        pass


def init_db():
    conn = connect()
    with conn:
        conn.executescript(SCHEMA)
        _migrate(conn)
        backfill_batch_files(conn)
        cur = conn.execute("SELECT id FROM thresholds WHERE id=1")
        if not cur.fetchone():
            conn.execute("INSERT INTO thresholds(id, data, updated_at) VALUES (1, ?, ?)",
                         (json.dumps(rules.DEFAULT_THRESHOLDS, ensure_ascii=False),
                          datetime.now().isoformat(timespec="seconds")))
    conn.close()


def get_thresholds():
    """读取阈值；库里没有的新键（v2 新增门槛）用默认值补全，老库不用迁移"""
    conn = connect()
    row = conn.execute("SELECT data FROM thresholds WHERE id=1").fetchone()
    conn.close()
    T = dict(rules.DEFAULT_THRESHOLDS)
    if row:
        try:
            T.update(json.loads(row["data"]))
        except Exception:
            pass
    return T


def set_thresholds(T):
    conn = connect()
    with conn:
        conn.execute("UPDATE thresholds SET data=?, updated_at=? WHERE id=1",
                     (json.dumps(T, ensure_ascii=False),
                      datetime.now().isoformat(timespec="seconds")))
    conn.close()


def reset_thresholds():
    set_thresholds(dict(rules.DEFAULT_THRESHOLDS))
    return dict(rules.DEFAULT_THRESHOLDS)


def get_setting(key, default=None):
    conn = connect()
    row = conn.execute("SELECT value FROM settings WHERE key=?", (key,)).fetchone()
    conn.close()
    return row["value"] if row else default


def set_setting(key, value):
    conn = connect()
    with conn:
        conn.execute("INSERT INTO settings(key, value) VALUES(?,?) "
                     "ON CONFLICT(key) DO UPDATE SET value=excluded.value", (key, str(value)))
    conn.close()


# ---- 文案模板（settings 表单行 JSON，key='texts'）----
def get_texts():
    """
    读取文案。库里存的是整份模板，升级后旧存档会盖住新默认值，所以按规则迁移：
      存档值 == v1 默认值  → 用户没改过，用新默认值
      存档值 != v1 默认值  → 用户自定义过，保留
    """
    conn = connect()
    row = conn.execute("SELECT value FROM settings WHERE key='texts'").fetchone()
    conn.close()
    merged = dict(rules.DEFAULT_TEXTS)
    if row:
        try:
            stored = json.loads(row["value"])
        except Exception:
            stored = {}
        for k, v in stored.items():
            if k not in rules.DEFAULT_TEXTS:
                continue                      # 已废弃的旧键直接丢弃
            if rules.V1_TEXTS.get(k) == v:
                continue                      # 与 v1 默认一致 = 没自定义过
            merged[k] = v
    return merged


def set_texts(X):
    """只存「与默认值不同」的键，默认值以后再改也能自动生效"""
    diff = {k: v for k, v in X.items()
            if k in rules.DEFAULT_TEXTS and v != rules.DEFAULT_TEXTS[k]}
    conn = connect()
    with conn:
        conn.execute(
            "INSERT INTO settings(key, value) VALUES('texts', ?) "
            "ON CONFLICT(key) DO UPDATE SET value=excluded.value",
            (json.dumps(diff, ensure_ascii=False),))
    conn.close()


def reset_texts():
    set_texts(dict(rules.DEFAULT_TEXTS))
    return dict(rules.DEFAULT_TEXTS)


def _pick_granularity(conn, period):
    """
    快照（近30天滚动窗口）优先级最高：一个 period 只有一份快照行/店。
    同一周期两种粒度都存在时的取舍：
    日数据覆盖 >= 15 天才用日粒度（避免刚导 1 天日报就把整月视图切成残缺的日聚合）；
    否则用月汇总行。
    """
    has_snap = conn.execute("SELECT 1 FROM metrics WHERE period=? AND granularity='snapshot' LIMIT 1",
                            (period,)).fetchone()
    if has_snap:
        return "snapshot"
    has_daily = conn.execute("SELECT 1 FROM metrics WHERE period=? AND granularity='daily' LIMIT 1",
                             (period,)).fetchone()
    has_month = conn.execute("SELECT 1 FROM metrics WHERE period=? AND granularity='period' LIMIT 1",
                             (period,)).fetchone()
    if has_daily and has_month:
        n = conn.execute("SELECT COUNT(DISTINCT date) c FROM metrics WHERE period=? "
                         "AND granularity='daily' AND date<>''", (period,)).fetchone()["c"]
        return "daily" if n >= 15 else "period"
    return "daily" if has_daily else "period"


def get_periods():
    """
    返回周期列表。三种粒度：
      快照 snapshot —— 近 N 天滚动窗口，period 记「截至日」，带 window_start/end
      daily         —— 当月按日导入
      period        —— 月汇总
    """
    conn = connect()
    rows = conn.execute("""
        SELECT period, granularity, COUNT(DISTINCT store_id) AS stores, COUNT(DISTINCT date) AS days
        FROM metrics GROUP BY period, granularity
    """).fetchall()
    plans = {}
    for r in rows:
        plans.setdefault(r["period"], {})[r["granularity"]] = r
    out = []
    for p, gs in plans.items():
        g = _pick_granularity(conn, p)
        r = gs.get(g) or next(iter(gs.values()))
        item = {"period": p, "stores": r["stores"], "granularity": g,
                "days": r["days"] if g == "daily" else 0}
        if g == "snapshot":
            w = conn.execute("SELECT window_start, window_end FROM metrics "
                             "WHERE period=? AND granularity='snapshot' LIMIT 1", (p,)).fetchone()
            ws, we = (w["window_start"] or "", w["window_end"] or "") if w else ("", "")
            d = _days_between(ws, we)
            item.update({"window_start": ws, "window_end": we, "as_of": we or p,
                         "days": d or 30,
                         "label": "近%d天 · 截至 %s" % (d or 30, (we or p)[5:])})
        item["sort_key"] = (item.get("as_of") or (p + "-01"))
        out.append(item)
    conn.close()
    return sorted(out, key=lambda x: x["sort_key"], reverse=True)


def latest_period():
    ps = get_periods()
    return ps[0]["period"] if ps else None


def _days_between(a, b):
    """两个 YYYY-MM-DD 之间的天数（含首尾）"""
    try:
        from datetime import date
        d1 = date.fromisoformat(a)
        d2 = date.fromisoformat(b)
        return (d2 - d1).days + 1
    except Exception:
        return 0


def _weighted(items, weight_key, value_key):
    """加权平均：权重为 None/0 时退化为简单平均"""
    num = den = 0.0
    simple = []
    for it in items:
        v = it.get(value_key)
        if v is None:
            continue
        w = it.get(weight_key) or 0
        num += v * w
        den += w
        simple.append(v)
    if den > 0:
        return num / den
    return (sum(simple) / len(simple)) if simple else None


def aggregate_period(period):
    """
    按月聚合。日粒度优先；没有日粒度就用月汇总行。
    返回每店一条，含派生指标 hx_rate / cp_rate / cp_src。
    """
    conn = connect()
    g = _pick_granularity(conn, period)
    rows = conn.execute("""
        SELECT m.*, s.name, s.city FROM metrics m JOIN stores s ON s.id = m.store_id
        WHERE m.period=? AND m.granularity=?
        ORDER BY m.store_id, m.date
    """, (period, g)).fetchall()
    conn.close()

    grouped = {}
    for r in rows:
        grouped.setdefault(r["store_id"], []).append(dict(r))

    out = []
    for sid, items in grouped.items():
        latest = max(items, key=lambda x: (x["date"] or ""))
        total = {k: 0 for k in ("visit", "intend_deal", "used_coupons", "used_amount",
                                "deal", "deal_refund", "videos", "video_views", "reviews_new",
                                "bad_new", "reviews_total", "bad_total", "favs", "lit")}
        for it in items:
            for k in total:
                total[k] += (it.get(k) or 0)

        score = None
        rank_note = ""
        for it in sorted(items, key=lambda x: x["date"] or ""):
            if it.get("score") is not None:
                score = it["score"]          # 取最近一次有值（含 0 分未开分）的评分
            v = (it.get("rank_note") or "").strip()
            if v and v not in ("-", "无"):
                rank_note = v                # 取最近一次有内容的上榜信息

        # 核销率分母 = 门店页成交券数（2026-11-01 用户确定口径）
        hx_rate = rules.diff_rate(total["used_coupons"], total["deal"])
        cp_rate, cp_src, _cp_small = rules.bad_rate(
            total["reviews_new"], total["bad_new"], total["reviews_total"], total["bad_total"])
        out.append({
            "id": sid,
            "name": latest["name"],
            "city": latest["city"] or "",
            "visit": total["visit"],
            "deal": total["deal"],
            "deal_refund": total["deal_refund"],
            "intend_deal": total["intend_deal"],
            "used_coupons": total["used_coupons"],
            "used_amount": round(total["used_amount"], 2),
            "score": score,
            "hx_rate": hx_rate,
            "cp_rate": cp_rate,
            "cp_src": cp_src,
            # 评价明细：差评率口径与样本量判定需要原始计数
            "reviews_new": total["reviews_new"],
            "bad_new": total["bad_new"],
            "reviews_total": total["reviews_total"],
            "bad_total": total["bad_total"],
            "lit": total["lit"],
            "rank_note": rank_note,
            "return_rate": _weighted(items, "deal", "return_rate"),
            "complaint_rate": _weighted(items, "deal", "complaint_rate"),
            "reply_rate": _weighted(items, "reviews_new", "reply_rate"),
            "videos": total["videos"],
            "video_views": total["video_views"],
            "favs": total["favs"],
            "days": (_days_between(items[0].get("window_start") or "",
                                   items[0].get("window_end") or "")
                     if g == "snapshot"
                     else len({it["date"] for it in items if it["date"]})),
            "window_start": items[0].get("window_start") or "",
            "window_end": items[0].get("window_end") or "",
        })
    return out


def upsert_store(conn, sid, name, city, today):
    cur = conn.execute("SELECT id FROM stores WHERE id=?", (sid,))
    if cur.fetchone():
        conn.execute("UPDATE stores SET name=?, city=?, last_seen=? WHERE id=?",
                     (name, city, today, sid))
        return False
    conn.execute("INSERT INTO stores(id, name, city, first_seen, last_seen) VALUES (?,?,?,?,?)",
                 (sid, name, city, today, today))
    return True


def insert_metric(conn, row, batch_id):
    cols = ", ".join(METRIC_FIELDS)
    qs = ", ".join(["?"] * len(METRIC_FIELDS))
    vals = [row.get(k) for k in METRIC_FIELDS]
    conn.execute(
        "INSERT OR REPLACE INTO metrics(store_id, period, date, granularity, window_start, window_end, "
        "%s, batch_id) VALUES (?,?,?,?,?,?,%s,?)" % (cols, qs),
        [row["store_id"], row["period"], row.get("date", ""), row["granularity"],
         row.get("window_start", ""), row.get("window_end", "")] + vals + [batch_id])


def record_batch(conn, info):
    cur = conn.execute(
        "INSERT INTO batches(filename, rows_read, rows_written, new_stores, period, date_from, date_to, "
        "created_at, note, file_path) VALUES (?,?,?,?,?,?,?,?,?,?)",
        (info.get("filename"), info.get("rows_read"), info.get("rows_written"),
         info.get("new_stores"), info.get("period"), info.get("date_from"),
         info.get("date_to"), datetime.now().isoformat(timespec="seconds"), info.get("note"),
         info.get("file_path")))
    return cur.lastrowid


# ---------------------------------------------------------------- 账号
def user_count():
    conn = connect()
    n = conn.execute("SELECT COUNT(*) FROM users").fetchone()[0]
    conn.close()
    return n


def get_user_by_name(username):
    conn = connect()
    r = conn.execute("SELECT * FROM users WHERE username=?", (username,)).fetchone()
    conn.close()
    return dict(r) if r else None


def get_user(uid):
    conn = connect()
    r = conn.execute("SELECT * FROM users WHERE id=?", (uid,)).fetchone()
    conn.close()
    return dict(r) if r else None


def list_users():
    conn = connect()
    rows = [dict(r) for r in conn.execute(
        "SELECT id, username, role, display_name, active, must_change, created_at, last_login "
        "FROM users ORDER BY CASE role WHEN 'admin' THEN 0 WHEN 'editor' THEN 1 ELSE 2 END, id")]
    conn.close()
    return rows


def create_user(username, password, role, display_name=""):
    import auth
    h, salt = auth.hash_password(password)
    conn = connect()
    try:
        cur = conn.execute(
            "INSERT INTO users(username, pass_hash, salt, role, display_name, active, must_change, created_at) "
            "VALUES (?,?,?,?,?,1,0,?)",
            (username, h, salt, role, display_name or username,
             datetime.now().isoformat(timespec="seconds")))
        conn.commit()
        return cur.lastrowid
    finally:
        conn.close()


def update_user(uid, role=None, display_name=None, active=None):
    sets, vals = [], []
    if role is not None:
        sets.append("role=?"); vals.append(role)
    if display_name is not None:
        sets.append("display_name=?"); vals.append(display_name)
    if active is not None:
        sets.append("active=?"); vals.append(1 if active else 0)
    if not sets:
        return
    vals.append(uid)
    conn = connect()
    conn.execute("UPDATE users SET %s WHERE id=?" % ",".join(sets), vals)
    conn.commit()
    conn.close()


def set_password(uid, password, must_change=0):
    import auth
    h, salt = auth.hash_password(password)
    conn = connect()
    conn.execute("UPDATE users SET pass_hash=?, salt=?, must_change=? WHERE id=?",
                 (h, salt, 1 if must_change else 0, uid))
    conn.commit()
    conn.close()


def delete_user(uid):
    conn = connect()
    conn.execute("DELETE FROM sessions WHERE user_id=?", (uid,))
    conn.execute("DELETE FROM users WHERE id=?", (uid,))
    conn.commit()
    conn.close()


def admin_count():
    conn = connect()
    n = conn.execute("SELECT COUNT(*) FROM users WHERE role='admin' AND active=1").fetchone()[0]
    conn.close()
    return n


def touch_login(uid):
    conn = connect()
    conn.execute("UPDATE users SET last_login=? WHERE id=?",
                 (datetime.now().isoformat(timespec="seconds"), uid))
    conn.commit()
    conn.close()


# ---------------------------------------------------------------- 会话
def create_session(token_hash, user_id, expires_at, ip=""):
    conn = connect()
    conn.execute("DELETE FROM sessions WHERE expires_at < ?",
                 (datetime.now().isoformat(timespec="seconds"),))
    conn.execute("INSERT INTO sessions(token_hash, user_id, created_at, expires_at, ip) VALUES (?,?,?,?,?)",
                 (token_hash, user_id, datetime.now().isoformat(timespec="seconds"), expires_at, ip))
    conn.commit()
    conn.close()


def get_session(token_hash):
    """返回 (user_dict, expires_at) 或 (None, None)"""
    conn = connect()
    r = conn.execute("SELECT * FROM sessions WHERE token_hash=?", (token_hash,)).fetchone()
    if not r:
        conn.close()
        return None, None
    if (r["expires_at"] or "") < datetime.now().isoformat(timespec="seconds"):
        conn.execute("DELETE FROM sessions WHERE token_hash=?", (token_hash,))
        conn.commit()
        conn.close()
        return None, None
    u = conn.execute("SELECT * FROM users WHERE id=?", (r["user_id"],)).fetchone()
    conn.close()
    if not u or not u["active"]:
        return None, None
    return dict(u), r["expires_at"]


def delete_session(token_hash):
    conn = connect()
    conn.execute("DELETE FROM sessions WHERE token_hash=?", (token_hash,))
    conn.commit()
    conn.close()


def delete_user_sessions(uid):
    conn = connect()
    conn.execute("DELETE FROM sessions WHERE user_id=?", (uid,))
    conn.commit()
    conn.close()


def get_batch(batch_id):
    conn = connect()
    r = conn.execute("SELECT * FROM batches WHERE id=?", (batch_id,)).fetchone()
    conn.close()
    return dict(r) if r else None


def latest_batch_with_file(period=None):
    """找某个周期（或全局最新）里带原始文件的批次"""
    conn = connect()
    sql = ("SELECT * FROM batches WHERE file_path IS NOT NULL AND file_path<>''")
    args = []
    if period:
        sql += " AND period=?"
        args.append(period)
    sql += " ORDER BY id DESC LIMIT 1"
    r = conn.execute(sql, args).fetchone()
    if not r and period:
        r = conn.execute("SELECT * FROM batches WHERE file_path IS NOT NULL AND file_path<>'' "
                         "ORDER BY id DESC LIMIT 1").fetchone()
    conn.close()
    return dict(r) if r else None


def import_history(limit=20):
    conn = connect()
    rows = conn.execute("SELECT * FROM batches ORDER BY id DESC LIMIT ?", (limit,)).fetchall()
    conn.close()
    return [dict(r) for r in rows]


def stats():
    conn = connect()
    n_stores = conn.execute("SELECT COUNT(*) c FROM stores").fetchone()["c"]
    n_metrics = conn.execute("SELECT COUNT(*) c FROM metrics").fetchone()["c"]
    n_batches = conn.execute("SELECT COUNT(*) c FROM batches").fetchone()["c"]
    conn.close()
    return {"stores": n_stores, "metrics": n_metrics, "batches": n_batches}