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

# -*- coding: utf-8 -*-
"""
数据导入（CSV / Excel xlsx）
- 编码：utf-8-sig → gbk → utf-8(容错)
- 表头：走别名映射，抖音来客导出的不同列名都能认
- 粒度：有日期列/传入日期 → 按日入库(daily)；否则整表当周期汇总(period)
- 幂等：同 门店+周期+日期+粒度 覆盖写入，重复导入不会翻倍
"""
import csv
import io
import os
import re
from datetime import datetime

import db
import xlsx

# canonical -> 可能的表头写法（已归一化：去空格/全角括号转半角）
ALIASES = {
    "store_id": ["门店id", "门店编号", "门店编码", "poiid", "shopid", "storeid", "id"],
    "name": ["门店名称", "门店名", "店铺名称", "shopname", "storename", "name"],
    "city": ["所在城市", "城市", "city"],
    "date": ["日期", "统计日期", "数据日期", "时间", "date"],
    "visit": ["门店页访问次数", "访问次数", "门店访问", "访问量", "visit"],
    "deal": ["门店页成交券数", "成交券数", "成交单量"],
    "deal_refund": ["门店页成交后退款券数", "成交后退款券数"],
    "intend_deal": ["门店意向成交券数", "意向成交券数", "下单券数", "成交券数(意向)"],
    "used_amount": ["门店核销金额", "核销金额", "核销gmv", "核销交易额"],
    "used_coupons": ["门店核销券数", "核销券数", "核销量", "核销单量"],
    "intend_refund": ["门店意向退款券数", "意向退款券数"],
    "videos": ["新发布门店关联视频数", "新发布视频数", "关联视频数", "发布视频数"],
    "video_views": ["门店视频播放次数", "视频播放次数", "播放量"],
    "score": ["门店评分", "评分", "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": ["上榜信息及排名", "榜单排名", "上榜信息"],
}

RATE_FIELDS = {"reply_rate", "complaint_rate", "return_rate"}
INT_FIELDS = {"visit", "deal", "deal_refund", "intend_deal", "used_coupons",
              "intend_refund", "videos", "video_views", "reviews_new", "good_new",
              "bad_new", "reviews_total", "good_total", "bad_total", "lit", "lit_posts", "favs"}
FLOAT_FIELDS = {"used_amount", "score"}
TEXT_FIELDS = {"rank_note"}


def norm_header(h):
    h = (h or "").strip().replace("\ufeff", "")
    h = h.replace("（", "(").replace("）", ")")
    h = re.sub(r"\s+", "", h)
    return h.lower()


def build_map(header):
    """返回 canonical -> 列下标"""
    norm = [norm_header(h) for h in header]
    m = {}
    for canon, names in ALIASES.items():
        for i, h in enumerate(norm):
            if h in [n.lower() for n in names]:
                m[canon] = i
                break
    return m


def decode(raw):
    for enc in ("utf-8-sig", "gbk", "utf-8"):
        try:
            return raw.decode(enc), enc
        except UnicodeDecodeError:
            continue
    return raw.decode("utf-8", errors="replace"), "utf-8(replace)"


def to_num(v, as_rate=False):
    if v is None:
        return None
    s = str(v).strip().replace(",", "").replace(" ", "")
    if s in ("", "-", "--", "None", "null", "nan"):
        return None
    has_pct = s.endswith("%")
    if has_pct:
        s = s[:-1]
    try:
        x = float(s)
    except ValueError:
        return None
    if as_rate:
        # 带 % 或数值大于 1，都视为百分数
        if has_pct or x > 1:
            x = x / 100.0
    return x


def _valid_ymd(y, mo, d):
    """合法性校验：月份 1-12、日期 1-31，不合法返回 None（避免出现 2026-01-00 这种）"""
    try:
        y, mo, d = int(y), int(mo), int(d)
    except (TypeError, ValueError):
        return None
    if not (1 <= mo <= 12) or not (1 <= d <= 31):
        return None
    return y, mo, d


# 生意经「按统计周期」导出的统计日期是区间，如 2026-10-03-2026-11-01 / 2026-10-03~2026-11-01
RANGE_RE = re.compile(
    r"^\s*(20\d{2})[-/.]?(\d{1,2})[-/.](\d{1,2})\s*(?:-|~|～|至|到)\s*"
    r"(20\d{2})[-/.]?(\d{1,2})[-/.](\d{1,2})\s*$")


def _norm_date(y, m, d):
    try:
        return "%04d-%02d-%02d" % (int(y), int(m), int(d))
    except (TypeError, ValueError):
        return ""


def parse_range(v):
    """把统计日期区间解析成 (start, end)；不是区间返回 None"""
    m = RANGE_RE.match(str(v or "").replace(" ", ""))
    if not m:
        return None
    a = _norm_date(m.group(1), m.group(2), m.group(3))
    b = _norm_date(m.group(4), m.group(5), m.group(6))
    if not a or not b or a > b:
        return None
    return a, b


def range_in_filename(filename):
    """文件名里带日期区间（如 …_2026-10-03_2026-11-01.xlsx）→ 返回 (start, end)"""
    m = re.search(r"(20\d{2})[-/.](\d{1,2})[-/.](\d{1,2})\s*[_~-]+\s*"
                  r"(20\d{2})[-/.](\d{1,2})[-/.](\d{1,2})", str(filename or ""))
    if not m:
        return None
    a = _norm_date(m.group(1), m.group(2), m.group(3))
    b = _norm_date(m.group(4), m.group(5), m.group(6))
    if a and b and a <= b:
        return a, b
    return None


def guess_period_date(filename, period, date):
    """
    没给日期/月份时，从文件名猜。顺序：先认「年月日」，再认「年月」，最后认「月」。
      日粒度：2026-10-30 / 2026/10/30 / 2026.10.30 / 2026年10月30日 / 20261030 / 10.28
      月粒度：2026-10 / 2026年10月 / 2026.10 / 202610 / 10月
    关键点：2026-10 这类「年月」不能被误读成「2026 年 1 月 0 日」——
    所以完整日期必须分隔符齐全、且年月日都合法，才按日粒度处理。
    """
    if date and not period:
        period = date[:7]
    if date or period:
        return period, date
    fn = filename or ""

    # 1) 带分隔符的完整日期：2026-10-30、2026/10/30、2026.10.30、2026年10月30日
    m = re.search(r"(20\d{2})\s*[-_/.年]\s*(\d{1,2})\s*[-_/.月]\s*(\d{1,2})(?!\d)", fn)
    if m:
        ymd = _valid_ymd(*m.groups())
        if ymd:
            return "%04d-%02d" % ymd[:2], "%04d-%02d-%02d" % ymd

    # 2) 无分隔符的完整日期：20261030
    m = re.search(r"(?<!\d)(20\d{2})(\d{2})(\d{2})(?!\d)", fn)
    if m:
        ymd = _valid_ymd(*m.groups())
        if ymd:
            return "%04d-%02d" % ymd[:2], "%04d-%02d-%02d" % ymd

    # 3) 带分隔符的年月：2026-10、2026年10月、2026.10
    m = re.search(r"(20\d{2})\s*[-_/.年]\s*(\d{1,2})(?!\d)", fn)
    if m:
        y, mo = int(m.group(1)), int(m.group(2))
        if 1 <= mo <= 12:
            return "%04d-%02d" % (y, mo), ""

    # 4) 无分隔符的年月：202610
    m = re.search(r"(?<!\d)(20\d{2})(\d{2})(?!\d)", fn)
    if m:
        y, mo = int(m.group(1)), int(m.group(2))
        if 1 <= mo <= 12:
            return "%04d-%02d" % (y, mo), ""

    # 5) 只有月份：10月 / 10.28（用当前年份）
    y = datetime.now().year
    m = re.search(r"(\d{1,2})\s*月", fn)
    if m and 1 <= int(m.group(1)) <= 12:
        return "%04d-%02d" % (y, int(m.group(1))), ""
    m = re.search(r"(?<!\d)(\d{1,2})\.(\d{1,2})(?!\d)", fn)
    if m:
        ymd = _valid_ymd(y, m.group(1), m.group(2))
        if ymd:
            return "%04d-%02d" % ymd[:2], "%04d-%02d-%02d" % ymd

    return datetime.now().strftime("%Y-%m"), ""


def import_rows(rows, filename, period=None, date=None, source="csv", window=None, file_path=None):
    """rows: [[str,...], ...]，第一行为表头。CSV 与 Excel 共用此入库逻辑。"""
    rows = [r for r in rows]
    # 去掉全空行/表头前的杂讯行：找到含「门店名称或门店ID」表头的行作为表头
    start = 0
    for i, r in enumerate(rows[:20]):
        normed = [norm_header(c) for c in r]
        if any(h in ("门店id", "门店编号", "门店编码", "poiid") for h in normed) \
           and any("门店名称" in h or h == "店铺名称" for h in normed):
            start = i
            break
    rows = rows[start:]
    if not rows:
        return {"ok": False, "error": "文件为空"}

    reader = rows
    header = [str(c) if c is not None else "" for c in reader[0]]
    cmap = build_map(header)
    missing = [k for k in ("store_id", "name") if k not in cmap]
    if missing:
        return {"ok": False, "error": "表头缺少必要列：门店ID / 门店名称",
                "header": header}

    # window=(start, end) 时强制按「近 N 天快照」入库（生意经按统计周期导出没有日期列）
    win_force = None
    if not window:
        window = range_in_filename(filename)
    if window:
        a, b = (window[0] or "").strip(), (window[1] or "").strip()
        if a and b and a <= b:
            win_force = (a, b)
            if not period:
                period = b
            if not date:
                date = b
    period, date = guess_period_date(filename, period, date)

    conn = db.connect()
    today = datetime.now().date().isoformat()
    rows_read = rows_written = new_stores = duplicates = 0
    seen_keys = set()
    errors = []
    dates = set()
    grams = set()          # 实际入库用到的粒度（daily / period / snapshot）
    win_seen = set()       # 快照的统计窗口
    period_cnt = {}        # 实际入库的周期分布，用于回报「导到哪个月了」
    period_val = period

    with conn:
        batch_id = db.record_batch(conn, {"filename": filename, "rows_read": 0, "rows_written": 0,
                                          "file_path": file_path,
                                          "new_stores": 0, "period": period, "note": "importing"})
        for line_no, r in enumerate(reader[1:], start=2):
            r = [("" if c is None else str(c)) for c in r]
            if not r or not any(c.strip() for c in r):
                continue
            rows_read += 1

            def cell(key):
                i = cmap.get(key)
                return r[i] if i is not None and i < len(r) else None

            sid = (cell("store_id") or "").strip()
            name = (cell("name") or "").strip()
            if not sid or not name:
                if len(errors) < 5:
                    errors.append("第%d行缺少门店ID或名称" % line_no)
                continue

            row_date = (cell("date") or "").strip()
            win_start = win_end = ""
            if row_date:
                rng = parse_range(row_date)
                if rng:                                  # 区间 → 近 N 天快照
                    win_start, win_end = rng
                else:
                    row_date = row_date.replace("/", "-").replace(".", "-")
                    m = re.match(r"(20\d{2})-(\d{1,2})-(\d{1,2})", row_date)
                    if m:
                        row_date = "%s-%02d-%02d" % (m.group(1), int(m.group(2)), int(m.group(3)))
                    else:
                        row_date = date or ""

            if win_force:
                win_start, win_end = win_force
            if win_start and win_end and win_start != win_end:
                # 滚动窗口：period 记「截至日」，每店一行，window_start/end 保留区间
                p, g, row_date = win_end, "snapshot", win_end
            elif win_start and win_end:
                p, g, row_date = win_start[:7], "daily", win_start   # 首尾同一天 = 单日
                win_start = win_end = ""
            else:
                p = row_date[:7] if row_date else period_val
                g = "daily" if row_date else "period"

            key = (sid, p, row_date, g)
            if key in seen_keys:
                duplicates += 1
                if len(errors) < 5:
                    errors.append("第%d行与其他行门店/日期重复，已跳过" % line_no)
                continue
            seen_keys.add(key)
            if row_date:
                dates.add(row_date)

            rec = {"store_id": sid, "period": p, "date": row_date, "granularity": g,
                   "window_start": win_start, "window_end": win_end}
            for key in INT_FIELDS:
                v = to_num(cell(key))
                rec[key] = int(v) if v is not None else 0
            for key in FLOAT_FIELDS:
                rec[key] = to_num(cell(key))
            for key in RATE_FIELDS:
                rec[key] = to_num(cell(key), as_rate=True)
            for key in TEXT_FIELDS:
                rec[key] = (cell(key) or "").strip() or None

            city = (cell("city") or "").strip()
            if db.upsert_store(conn, sid, name, city, today):
                new_stores += 1
            db.insert_metric(conn, rec, batch_id)
            rows_written += 1
            grams.add(g)
            period_cnt[p] = period_cnt.get(p, 0) + 1
            if g == "snapshot" and win_start and win_end:
                win_seen.add((win_start, win_end))

        conn.execute("""UPDATE batches SET rows_read=?, rows_written=?, new_stores=?,
                        period=?, date_from=?, date_to=?, note=? WHERE id=?""",
                     (rows_read, rows_written, new_stores,
                      (max(period_cnt.items(), key=lambda kv: kv[1])[0] if period_cnt
                       else (sorted(dates)[0][:7] if dates else period_val)),
                      min(dates) if dates else "", max(dates) if dates else "",
                      "ok" if rows_written else "无有效数据", batch_id))
    conn.close()

    # 回包口径以「实际写进库的行」为准：跨月文件按行数最多的那个月回报
    out_period = max(period_cnt.items(), key=lambda kv: kv[1])[0] if period_cnt else period_val
    granularity = ("snapshot" if "snapshot" in grams
                   else "daily" if "daily" in grams else "period")
    return {"ok": rows_written > 0, "source": source, "filename": filename,
            "rows_read": rows_read, "rows_written": rows_written,
            "new_stores": new_stores, "granularity": granularity,
            "period": out_period, "date": date,
            "periods": sorted(period_cnt.items(), key=lambda kv: -kv[1]),
            "windows": [{"start": a, "end": b} for a, b in sorted(win_seen)],
            "date_range": [min(dates), max(dates)] if dates else [],
            "skipped": rows_read - rows_written, "duplicates": duplicates, "errors": errors,
            "columns_matched": len(cmap)}


def import_csv(raw, filename, period=None, date=None, window=None, file_path=None):
    text, enc = decode(raw)
    sample = text[:4096]
    try:
        dialect = csv.Sniffer().sniff(sample, delimiters=",\t;")
        delim = dialect.delimiter
    except csv.Error:
        delim = ","
    rows = list(csv.reader(io.StringIO(text), delimiter=delim))
    r = import_rows(rows, filename, period=period, date=date, source="csv", window=window, file_path=file_path)
    r["encoding"] = enc
    return r


def import_xlsx(raw, filename, period=None, date=None, window=None, file_path=None):
    rows, sheets = xlsx.read_bytes(raw)
    r = import_rows(rows, filename, period=period, date=date, source="xlsx", window=window, file_path=file_path)
    r["sheets"] = sheets
    return r


def import_bytes(raw, filename, period=None, date=None, window=None, file_path=None):
    """按扩展名分发：.xlsx 走 Excel 解析，其余按 CSV。window=(start,end) 强制按快照入库。"""
    if filename.lower().endswith(".xlsx"):
        return import_xlsx(raw, filename, period=period, date=date, window=window, file_path=file_path)
    return import_csv(raw, filename, period=period, date=date, window=window, file_path=file_path)


def import_file(path, period=None, date=None, file_path=None):
    with open(path, "rb") as fp:
        raw = fp.read()
    return import_bytes(raw, os.path.basename(path), period=period, date=date, file_path=file_path)