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

# -*- coding: utf-8 -*-
"""
门店检测工作台 —— 后端服务（Python 标准库，无需安装任何依赖）

启动:
    python backend/server.py            # 默认监听 0.0.0.0:8787（本机 + 局域网可访问）
    python backend/server.py --local    # 只允许本机访问（127.0.0.1）
    python backend/server.py --port 9000 --open

接口:
    GET  /api/meta                 周期列表 / 阈值 / 导入历史 / 统计
    GET  /api/dashboard?period=    工作台全量数据（聚合指标 + 检测结果 + KPI + 分布）
    PUT  /api/thresholds           保存自定义考核指标
    POST /api/thresholds/reset     恢复官方默认
    POST /api/import?filename=&period=&date=   导入 CSV（body 为原始字节）
    GET  /api/export.csv?...       导出检测清单
    GET  /api/history              导入批次历史

目录:
    frontend/   静态页面
    data/       数据库 app.db、inbox/（丢文件进来自动导入）、processed/
"""
import json
import mimetypes
import os
import shutil
import socket
import subprocess
import sys
import threading
import time
import traceback
import urllib.parse
import webbrowser
from datetime import datetime, timedelta
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

import auth
import codesync
import db
import ingest
import rules

BASE = os.path.dirname(os.path.abspath(__file__))
ROOT = os.path.dirname(BASE)
FRONTEND = os.path.join(ROOT, "frontend")
SYNC_SCRIPT = os.path.join(ROOT, "sync", "life_sync.mjs")
SYNC_HOUR = 1          # 每天凌晨 1:00 自动同步近30天数据
SYNC_CATCHUP_HOURS = 20  # 距上次成功超过这么久，启动时补跑一次


# ----------------------------------------------------------------------------
# 业务组装
# ----------------------------------------------------------------------------
def build_dashboard(period=None):
    T = db.get_thresholds()
    X = db.get_texts()
    period = period or db.latest_period()
    if not period:
        return {"empty": True, "thresholds": T, "defaults": rules.DEFAULT_THRESHOLDS,
                "texts": X, "text_defaults": rules.DEFAULT_TEXTS,
                "periods": [], "stores": [], "kpis": [], "dist": {}, "period": None}

    rows = db.aggregate_period(period)
    stores = []
    for m in rows:
        d = rules.diagnose(m, T, X)
        stores.append({
            "id": m["id"], "name": m["name"], "city": m["city"],
            "score": m["score"], "hx_rate": m["hx_rate"], "hx_amt": m["used_amount"],
            "hx_cnt": m["used_coupons"], "td": m["return_rate"] or 0,
            "ts": m["complaint_rate"] or 0, "cp": m["cp_rate"], "cp_src": m["cp_src"],
            "reply": m["reply_rate"], "visit": m["visit"], "videos": m["videos"],
            "lit": m.get("lit", 0),
            "deal": m.get("deal", 0), "deal_refund": m.get("deal_refund", 0),
            "intend_deal": m["intend_deal"], "days": m["days"],
            "rank": m.get("rank_note") or "",
            # v2 诊断结果（瘦身版：只传模板键+变量，文案由前端按当前模板渲染）
            "tier": d["tier"], "headline": d["headline"], "hd_key": d["hd_key"],
            "primary_key": (d["primary"] or {}).get("key"),
            "sev": d["sev"],
            "problems": [{"k": p["key"], "t": p["tier"], "l": p["layer"], "v": p["level"],
                          "c": p["cat"], "p": p["tpl"], "a": p["atpl"], "x": p["vars"]}
                         for p in d["problems"]],
        })
    stores.sort(key=lambda x: (-x["sev"], x["hx_amt"]))

    kpis = [
        {"k": "ALL", "label": "全部门店", "value": len(stores), "hot": 0},
        {"k": "R", "label": "超违规预警线(退单/投诉/差评)", "value": sum(
            1 for s in stores if s["tier"] == "R"), "hot": 1},
        {"k": "SCORE", "label": "评分低于基准%s" % T["score_base"], "value": sum(
            1 for s in stores if s["score"] is not None and s["score"] < T["score_base"]), "hot": 1},
        {"k": "HX", "label": "核销率低于基准%.0f%%" % (T["hx_rate_base"] * 100), "value": sum(
            1 for s in stores if s["hx_rate"] is not None and s["hx_rate"] < T["hx_rate_base"]), "hot": 1},
        {"k": "ZERO", "label": "本期零核销", "value": sum(1 for s in stores if s["hx_cnt"] == 0), "hot": 1},
    ]

    cat_cnt = {}
    for s in stores:
        for c in {p["c"] for p in s["problems"]}:
            cat_cnt[c] = cat_cnt.get(c, 0) + 1
    amount_buckets = {"zero": 0, "lt1k": 0, "ltbase": 0, "ok": 0}
    for s in stores:
        a = s["hx_amt"] or 0
        if a == 0:
            amount_buckets["zero"] += 1
        elif a < 1000:
            amount_buckets["lt1k"] += 1
        elif a < T["hx_amt_base"]:
            amount_buckets["ltbase"] += 1
        else:
            amount_buckets["ok"] += 1
    city_cnt = {}
    for s in stores:
        city_cnt[s["city"] or "未知"] = city_cnt.get(s["city"] or "未知", 0) + 1

    return {
        "empty": False,
        "period": period,
        "periods": db.get_periods(),
        "thresholds": T,
        "defaults": rules.DEFAULT_THRESHOLDS,
        "texts": X,
        "text_defaults": rules.DEFAULT_TEXTS,
        "customized": T != codesync.get_factory(),
        "kpis": kpis,
        "stores": stores,
        "dist": {
            "cats": sorted(cat_cnt.items(), key=lambda x: x[1]),
            "amount": amount_buckets,
            "cities": sorted(city_cnt.items(), key=lambda x: -x[1])[:15],
        },
        "stats": db.stats(),
    }


def build_export(query):
    """导出门店实际数据（纯指标，不含诊断结论/问题/整改动作）"""
    d = build_dashboard(query.get("period", [None])[0])
    rows = d["stores"]
    kpi = query.get("kpi", ["ALL"])[0]
    city = query.get("city", [""])[0]
    cat = query.get("cat", [""])[0]
    score_band = query.get("score", [""])[0]
    q = query.get("q", [""])[0].lower()

    if kpi == "R":
        rows = [s for s in rows if s["tier"] == "R"]
    elif kpi == "SCORE":
        rows = [s for s in rows if s["score"] is not None and s["score"] < d["thresholds"]["score_base"]]
    elif kpi == "HX":
        rows = [s for s in rows if s["hx_rate"] is not None and s["hx_rate"] < d["thresholds"]["hx_rate_base"]]
    elif kpi == "ZERO":
        rows = [s for s in rows if s["hx_cnt"] == 0]
    if city:
        rows = [s for s in rows if s["city"] == city]
    if cat:
        rows = [s for s in rows if cat in {p["c"] for p in (s.get("problems") or [])}]
    if score_band:
        # 评分档位：闭区间按【】语义含端点（4.3 同时落在 4.0-4.3 与 4.3-4.5）
        def in_band(v):
            if v is None:
                return False
            x = round(v, 1)
            if score_band == "lt40":
                return x < 4.0
            if score_band == "40_43":
                return 4.0 <= x <= 4.3
            if score_band == "43_45":
                return 4.3 <= x <= 4.5
            if score_band == "gte45":
                return x >= 4.5
            return True
        rows = [s for s in rows if in_band(s["score"])]
    if q:
        rows = [s for s in rows if q in s["name"].lower() or q in s["id"]]

    def pct(v, n=1):
        return "" if v is None else ("%." + str(n) + "f%%") % (v * 100)

    head = ["门店ID", "门店名称", "城市", "评分", "访问次数", "成交券数", "核销券数",
            "核销率", "核销金额", "待核销券(估算)", "商责退单率", "投诉率", "差评率",
            "评价回复率", "发布视频数", "门店点亮数", "上榜信息及排名"]
    lines = [",".join(head)]
    for s in rows:
        pend = max((s["deal"] or 0) - (s["hx_cnt"] or 0), 0)
        cells = [s["id"], s["name"], s["city"],
                 "" if s["score"] is None else "%.1f" % s["score"],
                 str(s["visit"]), str(s["deal"]), str(s["hx_cnt"]),
                 pct(s["hx_rate"]), "%.0f" % (s["hx_amt"] or 0), str(pend),
                 "%.2f%%" % (s["td"] * 100), "%.2f%%" % (s["ts"] * 100),
                 "" if s["cp"] is None else "%.2f%%" % (s["cp"] * 100),
                 "" if s["reply"] is None else "%.0f%%" % (s["reply"] * 100),
                 str(s["videos"]), str(s.get("lit") or 0),
                 s.get("rank") or ""]
        # 单元格内的换行压成空格，避免 Excel 里出现多行单元格
        lines.append(",".join(
            '"%s"' % str(c).replace("\r", " ").replace("\n", " ").replace('"', '""')
            for c in cells))
    return ("\ufeff" + "\r\n".join(lines)).encode("utf-8"), d["period"]


# ----------------------------------------------------------------------------
# 近30天数据每日自动同步（生意经）
# 放在服务进程里跑：launchd 会被 macOS 的 ~/Documents 保护挡住，服务本身没有这个问题
# ----------------------------------------------------------------------------
SYNC_LOCK = threading.Lock()
SYNC_STATE = {"running": False, "last_start": None, "last_end": None,
              "last_ok": None, "last_msg": "", "trigger": ""}


def _ego_bin():
    """定位 ego-browser：PATH 里没有时按常见安装位置找"""
    import shutil
    for c in (os.environ.get("EGO_BIN"),
              os.path.expanduser("~/.local/bin/ego-browser"),
              "/usr/local/bin/ego-browser", "/opt/homebrew/bin/ego-browser"):
        if c and os.path.isfile(c) and os.access(c, os.X_OK):
            return c
    return shutil.which("ego-browser")


def run_sync(trigger="manual"):
    """跑一次同步；同一时间只允许一个"""
    if not SYNC_LOCK.acquire(blocking=False):
        return {"ok": False, "error": "已有同步任务在进行中"}
    try:
        if not os.path.isfile(SYNC_SCRIPT):
            return {"ok": False, "error": "找不到同步脚本 %s" % SYNC_SCRIPT}
        ego = _ego_bin()
        if not ego:
            return {"ok": False, "error": "找不到 ego-browser"}

        SYNC_STATE.update({"running": True, "trigger": trigger,
                           "last_start": datetime.now().isoformat(timespec="seconds"),
                           "last_msg": ""})
        print("[sync] 开始（%s）" % trigger)
        code = ("globalThis.SYNC_ARGS = [];"
                "await import('file://' + encodeURI(%r));" % SYNC_SCRIPT)
        pr = subprocess.run([ego, "nodejs", "-e", code], cwd=ROOT,
                            capture_output=True, text=True, timeout=1800)
        out = (pr.stdout or "") + (pr.stderr or "")
        ok = pr.returncode == 0
        tail = [l for l in out.strip().splitlines() if l.strip()][-3:]
        SYNC_STATE.update({"running": False, "last_ok": ok,
                           "last_end": datetime.now().isoformat(timespec="seconds"),
                           "last_msg": " | ".join(tail)[:500]})
        db.set_setting("last_sync", datetime.now().isoformat(timespec="seconds"))
        print("[sync] 结束 ok=%s" % ok)
        return {"ok": ok, "output": tail, "returncode": pr.returncode}
    except Exception as e:
        traceback.print_exc()
        SYNC_STATE.update({"running": False, "last_ok": False,
                           "last_end": datetime.now().isoformat(timespec="seconds"),
                           "last_msg": str(e)[:500]})
        return {"ok": False, "error": str(e)}
    finally:
        SYNC_LOCK.release()


def _sync_thread(trigger):
    def _run():
        try:
            run_sync(trigger)
        except Exception:
            traceback.print_exc()
    threading.Thread(target=_run, daemon=True).start()


def daily_sync_loop(log=print):
    """每天 SYNC_HOUR:00 触发；启动时若距上次成功太久则补跑"""
    # 启动补跑
    try:
        last = db.get_setting("last_sync")
        stale = True
        if last:
            try:
                stale = (datetime.now() - datetime.fromisoformat(last)).total_seconds() > SYNC_CATCHUP_HOURS * 3600
            except ValueError:
                stale = True
        if stale:
            time.sleep(90)          # 等服务和收件箱 watcher 起来
            log("[sync] 距上次同步超过 %d 小时，启动补跑一次" % SYNC_CATCHUP_HOURS)
            _sync_thread("catchup")
    except Exception:
        traceback.print_exc()

    while True:
        try:
            now = datetime.now()
            nxt = now.replace(hour=SYNC_HOUR, minute=0, second=0, microsecond=0)
            if nxt <= now:
                nxt = nxt + timedelta(days=1)
            wait = (nxt - now).total_seconds()
            log("[sync] 下次自动同步：%s" % nxt.strftime("%Y-%m-%d %H:%M"))
            time.sleep(max(60, wait))
            _sync_thread("schedule")
            time.sleep(120)         # 留出执行时间，避免同分钟重复触发
        except Exception:
            traceback.print_exc()
            time.sleep(300)


# ----------------------------------------------------------------------------
# 收件箱自动导入
# ----------------------------------------------------------------------------
def watch_inbox(interval=20, log=print):
    """data/inbox 里丢 CSV / Excel 就自动导入，处理完移到 data/processed"""
    while True:
        try:
            files = [f for f in os.listdir(db.INBOX)
                     if f.lower().endswith((".csv", ".xlsx")) and not f.startswith("~$")]
            for f in files:
                path = os.path.join(db.INBOX, f)
                try:
                    stamp = datetime.now().strftime("%Y%m%d_%H%M%S")
                    dest = "%s_%s" % (stamp, f)
                    r = ingest.import_file(path, file_path=dest)
                    shutil.move(path, os.path.join(db.PROCESSED, dest))
                    log("[inbox] %s → 导入 %d 行（%s）" % (f, r.get("rows_written", 0), r.get("period")))
                except Exception as e:
                    log("[inbox] %s 导入失败: %s" % (f, e))
                    shutil.move(path, os.path.join(db.PROCESSED, "失败_" + f))
        except Exception:
            traceback.print_exc()
        time.sleep(interval)


# ----------------------------------------------------------------------------
# HTTP
# ----------------------------------------------------------------------------
# ---------------------------------------------------------------- 权限
# 路由 → 需要的权限点。没列出的 /api/* 一律只要登录（view）
PERM_RULES = [
    ("POST",   "/api/login",            None),      # 公开
    ("POST",   "/api/logout",           None),
    ("GET",    "/api/auth/state",       None),
    ("GET",    "/api/me",               "view"),
    ("GET",    "/api/users",            "users"),
    ("POST",   "/api/users",            "users"),
    ("PUT",    "/api/users",            "users"),
    ("DELETE", "/api/users",            "users"),
    ("GET",    "/api/export.file",      "export"),
    ("GET",    "/api/export.csv",       "export"),
    ("GET",    "/api/batch/",           "export"),
    ("POST",   "/api/import",           "import"),
    ("POST",   "/api/sync",             "sync"),
    ("PUT",    "/api/thresholds",       "config"),
    ("POST",   "/api/thresholds/reset", "config"),
    ("PUT",    "/api/copytext",         "config"),
    ("POST",   "/api/copytext/reset",   "config"),
]
SESSION_DAYS = 30


def required_perm(method, path):
    for m, pref, perm in PERM_RULES:
        if m == method and (path == pref or path.startswith(pref)):
            return perm, True
    return "view", False


def ensure_admin():
    """首次运行创建初始管理员"""
    if db.user_count() > 0:
        return None
    db.create_user("admin", "admin123", "admin", "管理员")
    u = db.get_user_by_name("admin")
    db.set_password(u["id"], "admin123", must_change=1)
    return "admin123"


class Handler(BaseHTTPRequestHandler):
    server_version = "StoreOps/1.0"

    def log_message(self, fmt, *args):
        sys.stdout.write("  %s %s\n" % (self.command, self.path))

    # ---- helpers ----
    def _send(self, code, body, ctype="application/json; charset=utf-8", extra=None):
        self.send_response(code)
        self.send_header("Content-Type", ctype)
        self.send_header("Content-Length", str(len(body)))
        self.send_header("Cache-Control", "no-store")
        for k, v in (extra or {}).items():
            self.send_header(k, v)
        self.end_headers()
        self.wfile.write(body)

    def _json(self, obj, code=200):
        self._send(code, json.dumps(obj, ensure_ascii=False, default=str).encode("utf-8"))

    def _err(self, msg, code=400):
        self._json({"ok": False, "error": msg}, code)

    def _cookies(self):
        out = {}
        for part in (self.headers.get("Cookie") or "").split(";"):
            if "=" in part:
                k, v = part.split("=", 1)
                out[k.strip()] = v.strip()
        return out

    def _auth(self):
        self.user = None
        self.sid = None
        sid = self._cookies().get("sid")
        if sid:
            h = auth.hash_token(sid)
            u, _ = db.get_session(h)
            if u:
                self.user, self.sid = u, h
        return self.user

    def _guard(self, method, path):
        """返回 True 表示已放行；否则已经写好响应，调用方直接 return"""
        need, matched = required_perm(method, path)
        if need is None:
            return True
        if not self.user:
            self._json({"ok": False, "error": "请先登录", "need_login": True}, 401)
            return False
        if not auth.can(self.user["role"], need):
            self._json({"ok": False, "error": "当前账号（%s）没有这个权限" %
                                          auth.role_label(self.user["role"]), "forbidden": need}, 403)
            return False
        return True

    def _body(self):
        n = int(self.headers.get("Content-Length") or 0)
        return self.rfile.read(n) if n else b""

    def do_DELETE(self):
        try:
            u = urllib.parse.urlparse(self.path)
            self._auth()
            if u.path.startswith("/api/") and not self._guard("DELETE", u.path):
                return
            if u.path.startswith("/api/users/"):
                uid = int(u.path.rsplit("/", 1)[1])
                return self._user_delete(uid)
            return self._err("没有这个接口", 404)
        except Exception as e:
            traceback.print_exc()
            self._err("服务端异常: %s" % e, 500)

    # ---- 账号相关 ----
    def _login(self):
        try:
            body = json.loads(self._body().decode("utf-8") or "{}")
        except (ValueError, UnicodeDecodeError):
            return self._err("请求格式不对")
        name = (body.get("username") or "").strip()
        pw = body.get("password") or ""
        u = db.get_user_by_name(name)
        if not u or not auth.verify_password(pw, u["pass_hash"], u["salt"]):
            time.sleep(0.4)          # 拖慢暴力破解
            return self._err("账号或密码不对", 401)
        if not u["active"]:
            return self._err("该账号已被停用，请联系管理员", 403)
        token = auth.new_token()
        exp = (datetime.now() + timedelta(days=SESSION_DAYS)).isoformat(timespec="seconds")
        db.create_session(auth.hash_token(token), u["id"], exp,
                          self.headers.get("X-Forwarded-For", "") or self.client_address[0])
        db.touch_login(u["id"])
        payload = json.dumps({"ok": True, "must_change": bool(u["must_change"]),
                              "role": u["role"],
                              "display_name": u["display_name"]}).encode()
        return self._send(200, payload, "application/json; charset=utf-8", {
            "Set-Cookie": "sid=%s; Path=/; Max-Age=%d; HttpOnly; SameSite=Lax"
                          % (token, SESSION_DAYS * 86400)})

    def _user_create(self):
        try:
            b = json.loads(self._body().decode("utf-8") or "{}")
        except (ValueError, UnicodeDecodeError):
            return self._err("请求格式不对")
        name = (b.get("username") or "").strip()
        pw = b.get("password") or ""
        role = b.get("role") or "viewer"
        if not name:
            return self._err("用户名不能为空")
        if db.get_user_by_name(name):
            return self._err("这个用户名已存在", 409)
        if role not in auth.ROLES:
            return self._err("角色不对")
        ok, msg = auth.check_strength(pw)
        if not ok:
            return self._err(msg)
        uid = db.create_user(name, pw, role, (b.get("display_name") or name).strip())
        return self._json({"ok": True, "id": uid})

    def _me_update(self):
        """改自己的密码"""
        try:
            b = json.loads(self._body().decode("utf-8") or "{}")
        except (ValueError, UnicodeDecodeError):
            return self._err("请求格式不对")
        u = self.user
        new = b.get("password") or ""
        ok, msg = auth.check_strength(new)
        if not ok:
            return self._err(msg)
        if not u["must_change"]:
            old = b.get("old_password") or ""
            if not auth.verify_password(old, u["pass_hash"], u["salt"]):
                return self._err("原密码不对", 403)
        db.set_password(u["id"], new, must_change=0)
        return self._json({"ok": True})

    def _user_update(self, uid):
        try:
            b = json.loads(self._body().decode("utf-8") or "{}")
        except (ValueError, UnicodeDecodeError):
            return self._err("请求格式不对")
        u = db.get_user(uid)
        if not u:
            return self._err("账号不存在", 404)

        role = b.get("role")
        if role is not None and role not in auth.ROLES:
            return self._err("角色不对")
        # 不能把最后一个管理员降级
        if role and role != "admin" and u["role"] == "admin" and db.admin_count() <= 1:
            return self._err("至少要保留一个管理员")

        active = b.get("active")
        if active is not None and not active and u["username"] == "admin":
            return self._err("admin 是内置账号，不能停用")
        if active is not None and not active and u["role"] == "admin" and db.admin_count() <= 1:
            return self._err("至少要保留一个启用的管理员")
        if active is not None and not active and uid == self.user["id"]:
            return self._err("不能停用自己正在用的账号")

        db.update_user(uid, role=role, display_name=b.get("display_name"), active=active)

        pw = b.get("password")
        if pw:
            ok, msg = auth.check_strength(pw)
            if not ok:
                return self._err(msg)
            db.set_password(uid, pw, must_change=1 if b.get("force_change") else 0)
            db.delete_user_sessions(uid)      # 改密码后强制重新登录
        return self._json({"ok": True})

    def _user_delete(self, uid):
        if self.user["id"] == uid:
            return self._err("不能删除自己正在用的账号")
        u = db.get_user(uid)
        if not u:
            return self._err("账号不存在", 404)
        if u["username"] == "admin":
            return self._err("admin 是内置账号，不能删除")
        if u["role"] == "admin" and db.admin_count() <= 1:
            return self._err("至少要保留一个管理员")
        db.delete_user(uid)
        return self._json({"ok": True})

    def _static(self, path):
        if path in ("/", ""):
            if not getattr(self, "user", None):
                path = "/login.html"       # 没登录先看登录页
            else:
                path = "/index.html"
        fp = os.path.normpath(os.path.join(FRONTEND, path.lstrip("/")))
        if not fp.startswith(FRONTEND) or not os.path.isfile(fp):
            self._send(404, b"Not Found", "text/plain; charset=utf-8")
            return
        ctype = mimetypes.guess_type(fp)[0] or "application/octet-stream"
        with open(fp, "rb") as f:
            data = f.read()
        self._send(200, data, ctype)

    # ---- routes ----
    def do_GET(self):
        try:
            u = urllib.parse.urlparse(self.path)
            path, q = u.path, urllib.parse.parse_qs(u.query)
            self._auth()
            if path.startswith("/api/"):
                if not self._guard("GET", path):
                    return
            if path == "/api/auth/state":
                n = db.user_count()
                admin = db.get_user_by_name("admin")
                return self._json({
                    "ok": True, "needs_setup": n == 0,
                    "default_admin": bool(admin and admin["must_change"]),
                    "logged_in": bool(self.user),
                })
            if path == "/api/me":
                u2 = self.user
                return self._json({"ok": True, "user": {
                    "id": u2["id"], "username": u2["username"], "role": u2["role"],
                    "role_label": auth.role_label(u2["role"]),
                    "display_name": u2["display_name"],
                    "must_change": bool(u2["must_change"]),
                    "perms": sorted(auth.GRANTS.get(u2["role"], set())),
                }})
            if path == "/api/users":
                return self._json({"ok": True, "users": db.list_users(),
                                   "roles": [{"key": k, **auth.ROLES[k]} for k in auth.ROLE_ORDER],
                                   "perms": auth.PERMS,
                                   "grants": {k: sorted(v) for k, v in auth.GRANTS.items()}})
            if path == "/api/meta":
                return self._json({
                    "ok": True,
                    "periods": db.get_periods(),
                    "thresholds": db.get_thresholds(),
                    "defaults": rules.DEFAULT_THRESHOLDS,
                    "factory": codesync.get_factory(),
                    "customized": db.get_thresholds() != codesync.get_factory(),
                    "history": db.import_history(),
                    "stats": db.stats(),
                    "inbox": {"path": db.INBOX, "processed": db.PROCESSED},
                })
            if path == "/api/dashboard":
                return self._json(build_dashboard(q.get("period", [None])[0]))
            if path == "/api/sync/status":
                return self._json({"ok": True, "state": SYNC_STATE,
                                   "last_sync": db.get_setting("last_sync"),
                                   "schedule": "每天 %02d:00" % SYNC_HOUR,
                                   "script": SYNC_SCRIPT})
            if path == "/api/history":
                return self._json({"ok": True, "history": db.import_history(50)})
            if path == "/api/copytext":
                return self._json({"ok": True, "texts": db.get_texts(),
                                   "defaults": rules.DEFAULT_TEXTS})
            if path.startswith("/api/batch/") and path.endswith("/file"):
                try:
                    bid = int(path.split("/")[3])
                except (ValueError, IndexError):
                    return self._err("批次号不正确", 400)
                return self._send_batch_file(db.get_batch(bid))
            if path == "/api/export.file":
                return self._send_batch_file(db.latest_batch_with_file(q.get("period", [None])[0]))
            if path == "/api/export.csv":
                body, period = build_export(q)
                name = "门店数据_%s.csv" % (period or "全部")
                return self._send(200, body, "text/csv; charset=utf-8",
                                  {"Content-Disposition":
                                   "attachment; filename*=UTF-8''%s" % urllib.parse.quote(name)})
            return self._static(path)
        except Exception as e:
            traceback.print_exc()
            self._err("服务端异常: %s" % e, 500)

    def _send_batch_file(self, batch):
        """原样返回某次导入的原始文件"""
        if not batch or not batch.get("file_path"):
            return self._err("这个批次没有留存原始文件", 404)
        p = os.path.join(db.PROCESSED, batch["file_path"])
        if not os.path.isfile(p):
            return self._err("原始文件已不存在（可能被清理）", 404)
        import re as _re
        name = _re.sub(r"^\d{8}_\d{6}_", "", batch["file_path"])
        ctype = ("application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
                 if name.lower().endswith(".xlsx") else "text/csv; charset=utf-8")
        with open(p, "rb") as fp:
            body = fp.read()
        return self._send(200, body, ctype,
                          {"Content-Disposition":
                           "attachment; filename*=UTF-8''%s" % urllib.parse.quote(name)})

    def do_PUT(self):
        try:
            u = urllib.parse.urlparse(self.path)
            self._auth()
            if u.path.startswith("/api/") and not self._guard("PUT", u.path):
                return
            if u.path == "/api/me":
                return self._me_update()
            if u.path.startswith("/api/users/"):
                return self._user_update(int(u.path.rsplit("/", 1)[1]))
            if u.path == "/api/thresholds":
                body = json.loads(self._body().decode("utf-8"))
                merged = db.get_thresholds()
                merged.update(body)
                errs = rules.validate(merged)
                if errs:
                    return self._err("；".join(errs), 422)
                db.set_thresholds(merged)
                ok, msg = codesync.write_defaults(merged)
                return self._json({"ok": True, "thresholds": db.get_thresholds(),
                                   "code_synced": ok, "code_msg": msg,
                                   "factory": codesync.get_factory()})
            if u.path == "/api/copytext":
                body = json.loads(self._body().decode("utf-8"))
                if not isinstance(body, dict):
                    return self._err("文案格式不正确", 422)
                merged = db.get_texts()
                for k, v in body.items():
                    if k in rules.DEFAULT_TEXTS:
                        merged[k] = str(v if v is not None else "")
                db.set_texts(merged)
                return self._json({"ok": True, "texts": db.get_texts()})
            self._err("未知接口", 404)
        except Exception as e:
            traceback.print_exc()
            self._err("保存失败: %s" % e, 500)

    def do_POST(self):
        try:
            u = urllib.parse.urlparse(self.path)
            q = urllib.parse.parse_qs(u.query)
            self._auth()
            if u.path.startswith("/api/") and not self._guard("POST", u.path):
                return
            if u.path == "/api/login":
                return self._login()
            if u.path == "/api/logout":
                if self.sid:
                    db.delete_session(self.sid)
                return self._send(200, json.dumps({"ok": True}).encode(), 
                                  "application/json; charset=utf-8",
                                  {"Set-Cookie": "sid=; Path=/; Max-Age=0; HttpOnly; SameSite=Lax"})
            if u.path == "/api/users":
                return self._user_create()
            if u.path == "/api/import":
                raw = self._body()
                if not raw:
                    return self._err("文件内容为空")
                filename = q.get("filename", ["upload.csv"])[0]
                period = q.get("period", [None])[0] or None
                date = q.get("date", [None])[0] or None
                stamp = datetime.now().strftime("%Y%m%d_%H%M%S")
                dest = "%s_%s" % (stamp, filename)
                try:
                    os.makedirs(db.PROCESSED, exist_ok=True)
                    with open(os.path.join(db.PROCESSED, dest), "wb") as fp:
                        fp.write(raw)
                except OSError:
                    dest = None
                result = ingest.import_bytes(raw, filename, period=period, date=date, file_path=dest)
                result["ok"] = result.get("ok", False)
                return self._json(result, 200 if result["ok"] else 422)
            if u.path == "/api/sync":
                if SYNC_STATE.get("running"):
                    return self._json({"ok": False, "error": "已有同步任务在进行中"}, 409)
                threading.Thread(target=run_sync, args=("manual",), daemon=True).start()
                return self._json({"ok": True, "msg": "已开始同步，约 3-5 分钟"})
            if u.path == "/api/thresholds/reset":
                fac = codesync.get_factory()
                db.set_thresholds(dict(fac))
                ok, msg = codesync.write_defaults(fac)
                return self._json({"ok": True, "thresholds": db.get_thresholds(),
                                   "code_synced": ok, "code_msg": msg,
                                   "factory": fac})
            if u.path == "/api/copytext/reset":
                return self._json({"ok": True, "texts": db.reset_texts()})
            self._err("未知接口", 404)
        except Exception as e:
            traceback.print_exc()
            self._err("导入失败: %s" % e, 500)


def bonjour_name():
    """取 Bonjour 名（这台机器 hostname 被设成了 IP，不能直接用）"""
    try:
        out = subprocess.run(["scutil", "--get", "LocalHostName"],
                             capture_output=True, text=True, timeout=3).stdout.strip()
        if out:
            return out
    except Exception:
        pass
    h = socket.gethostname().split(".")[0]
    return None if _re_is_ip(h) else h


def _re_is_ip(v):
    import re as _r
    return bool(_r.match(r"^\d{1,3}(\.\d{1,3}){3}$", v or ""))


def lan_ips():
    """列出本机可用的局域网 IPv4 地址（用于手机/其他电脑访问）"""
    ips = []
    try:
        s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        s.connect(("8.8.8.8", 80))
        ips.append(s.getsockname()[0])
        s.close()
    except Exception:
        pass
    try:
        for info in socket.getaddrinfo(socket.gethostname(), None, socket.AF_INET):
            ip = info[4][0]
            if ip not in ips:
                ips.append(ip)
    except Exception:
        pass
    return [ip for ip in ips if not ip.startswith("127.")]


def main():
    port = 8787
    args = sys.argv[1:]
    if "--port" in args:
        port = int(args[args.index("--port") + 1])
    lan = "--local" not in args          # 默认开放给局域网，--local 只绑本机
    host = "0.0.0.0" if lan else "127.0.0.1"
    db.init_db()
    init_pw = ensure_admin()
    if init_pw:
        print(" [首次启动] 已创建管理员账号 admin / %s —— 登录后请立刻修改密码" % init_pw)
    threading.Thread(target=watch_inbox, args=(20, lambda m: print(m)), daemon=True).start()
    # 自动同步依赖 ego-browser（要能操作已登录生意经的浏览器）。
    # 服务器上没有它就别启这个线程，免得每天 01:00 报一次错。
    can_sync = bool(_ego_bin()) and os.path.isfile(SYNC_SCRIPT)
    if can_sync:
        threading.Thread(target=daily_sync_loop, args=(lambda m: print(m),), daemon=True).start()
    srv = ThreadingHTTPServer((host, port), Handler)
    url = "http://127.0.0.1:%d" % port
    print("=" * 62)
    print(" 门店检测工作台已启动")
    print(" 本机访问: %s" % url)
    if lan:
        ips = lan_ips()
        if ips:
            for ip in ips:
                print(" 局域网:   http://%s:%d" % (ip, port))
        else:
            print(" 局域网:   未取到本机 IP，请在命令行执行 ipconfig 查看 IPv4 地址")
        hn = bonjour_name()
        if hn:
            print(" 更好记的写法: http://%s.local:%d   （同一 WiFi 下的 Mac / iPhone 可用）" % (hn, port))
        print(" 打不开时依次检查：① 两台设备是否同一 WiFi ② 本机是否休眠 ③ 系统设置→网络→防火墙")
    else:
        print(" 已限制为仅本机访问（--local）")
    print(" 数据目录: %s" % db.DATA_DIR)
    print(" 收件箱:   %s  （CSV/Excel 丢进去自动导入）" % db.INBOX)
    if can_sync:
        print(" 数据同步: 每天 %02d:00 自动拉取近30天数据（启动时距上次超过 %d 小时会补跑）"
              % (SYNC_HOUR, SYNC_CATCHUP_HOURS))
        print(" 立即同步: 左侧栏【同步数据】，或 POST /api/sync")
    else:
        print(" 数据同步: 未启用（这台机器上没有 ego-browser）")
        print("           请用【导入数据】上传从生意经导出的 Excel")
    print(" 停止服务: Ctrl + C")
    print("=" * 62)
    if "--open" in args:
        threading.Timer(1.0, lambda: webbrowser.open(url)).start()
    try:
        srv.serve_forever()
    except KeyboardInterrupt:
        print("\n已停止")
        srv.shutdown()


if __name__ == "__main__":
    main()