server-core/app/training/main.py

"""云超服 OPC 培训站 · FastAPI 子应用（已迁入 server-core 总后端）

由 server-core/dispatcher.py 统一对外（opc.pinesound.cn）：`/api/*` 路由到本子应用，
`/auth`、`/opc`、`/admin` 等由平台应用（app.main）服务。独立使用 data/opc.db 业务库。
"""
import asyncio
import json
from datetime import datetime, timedelta
import os
import re
import secrets
import time
from pathlib import Path
from typing import Optional
from fastapi import FastAPI, Header, Request, HTTPException, UploadFile, File, Form
from fastapi.middleware.cors import CORSMiddleware
from fastapi.staticfiles import StaticFiles

from . import db, opc_engine, policy_engine, policy_data, survey_data
from .auth import hash_password, make_token, verify_token, send_sms_code, check_sms_code, PHONE_RE
from .opc_engine import SECTIONS, AXIS_NAMES, ADAPT_LABELS, PROFILES

from contextlib import asynccontextmanager

# 平台域（任务/OPC 身份）访问：复用唯一总库 + 四层 Repository
from ..infrastructure.db import AsyncSessionLocal
from ..infrastructure.repositories import Database
from ..services.task_service import TaskService
from ..jwt import decode_access_token
from ..services import sms as platform_sms
from .. import config as _cfg


@asynccontextmanager
async def lifespan(app):
    # 建表/种子由 alembic + scripts/db/seed.py 非运行态完成
    yield


app = FastAPI(title="云超服 OPC 培训站后端", version="0.1", lifespan=lifespan)

ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))  # server-core 根

# ---- 上传目录 + 静态挂载（历史本地文件；新上传统一走 OSS，见 infrastructure.oss） ----
UPLOAD_DIR = os.path.join(ROOT, "serverdata", "uploads")
os.makedirs(UPLOAD_DIR, exist_ok=True)
app.mount("/uploads", StaticFiles(directory=UPLOAD_DIR), name="uploads")

# ---- OSS 统一下载路由：/oss/<key> 307 跳预签名 URL（桶私有也可读）；未配置 OSS 时回退本地文件 ----
from fastapi.responses import FileResponse, RedirectResponse  # noqa: E402
from ..infrastructure.oss import oss as _oss, resolve_url as _resolve_url  # noqa: E402


@app.get("/oss/{key:path}")
async def oss_download(key: str):
    try:
        key = _oss.clean_key(key)
    except ValueError:
        raise HTTPException(400, "非法对象路径")
    if not _oss.enabled:
        local = Path(UPLOAD_DIR) / key
        if not local.is_file():
            raise HTTPException(404, "文件不存在")
        return FileResponse(local)
    return RedirectResponse(await _oss.download_url(key), status_code=307)


@app.post("/api/oss/presign")
async def oss_presign(request: Request, authorization: str = Header(default="")):
    """生成直传链接：body {filename, contentType?, dir?} → {key, uploadUrl, objectUrl}（前端 PUT 直传 OSS）。"""
    require_auth(authorization)
    body = await request.json()
    name = os.path.basename(str(body.get("filename") or "file"))
    ext = os.path.splitext(name)[1].lower()
    if ext not in (".jpg", ".jpeg", ".png", ".webp", ".gif", ".pdf", ".doc", ".docx",
                   ".mp4", ".mov", ".m4a", ".mp3"):
        raise HTTPException(400, f"不支持的文件类型（{ext or '无扩展名'}）")
    subdir = str(body.get("dir") or "misc").strip("/ ").replace("..", "")
    from ..services.media_upload import build_key
    key = build_key(subdir, ext)
    upload_url = await _oss.presigned_put(key, content_type=body.get("contentType") or "application/octet-stream")
    # objectUrl 为 CDN/OSS 直链，前端直传完成后直接使用，不经服务端代理
    return {"ok": True, "key": key, "uploadUrl": upload_url, "objectUrl": _oss.direct_url(key)}

# ---- 微信小程序配置（真实登录） ----
# 微信 AppID / AppSecret 由 server-core 环境变量或 server-core/.env 提供。
def _load_dotenv():
    """轻量 .env 加载（server-core/.env），不依赖 python-dotenv"""
    env_path = os.path.join(ROOT, ".env")
    if os.path.exists(env_path):
        with open(env_path, encoding="utf-8") as f:
            for line in f:
                line = line.strip()
                if not line or line.startswith("#") or "=" not in line:
                    continue
                k, _, v = line.partition("=")
                os.environ.setdefault(k.strip(), v.strip().strip('"').strip("'"))


_load_dotenv()
# 微信小程序凭据：显式 WX_APPID/WX_SECRET 环境变量优先，否则回退平台 config
# （PINEAGENTS_WECHAT_APPID/SECRET）。此前仅读 WX_APPID/WX_SECRET，与 .env 键名
# 不一致导致 wx-phone 手机号绑定误报「未配置微信 AppID/Secret」。
WX_APPID = os.environ.get("WX_APPID", "") or _cfg.WECHAT_APPID
WX_SECRET = os.environ.get("WX_SECRET", "") or _cfg.WECHAT_SECRET
WX_HAS_CONFIG = bool(WX_APPID and WX_SECRET)

# 站点公开基础地址(用于把 /uploads 相对路径补全为完整 HTTPS 地址，供小程序/网页加载图片)
PUBLIC_BASE = os.environ.get("PUBLIC_BASE", "https://opc.pinesound.cn").rstrip("/")


def abs_url(u):
    """把相对路径(如 /uploads/x.png)补全为绝对地址；已是完整 URL 则原样返回"""
    if not u:
        return u
    u = str(u).strip()
    if u.startswith("http://") or u.startswith("https://"):
        return u
    return PUBLIC_BASE + u

import httpx  # 微信 code2session 请求用
import logging

async def _wx_code2session(code: str):
    """用登录 code 换 openid/session_key（真实微信登录核心）"""
    url = (
        "https://api.weixin.qq.com/sns/jscode2session"
        f"?appid={WX_APPID}&secret={WX_SECRET}&js_code={code}&grant_type=authorization_code"
    )
    async with httpx.AsyncClient(timeout=10) as client:
        r = await client.get(url)
        return r.json()  # {openid, session_key} 或 {errcode, errmsg}

async def _wx_get_access_token():
    """获取小程序 access_token（复用 wechat.py 统一缓存与配置；未配置时抛可读错误）。"""
    from ..services import wechat
    return await wechat._wx_access_token()

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
)



def now_iso():
    return db.now_iso()


def require_auth(authorization: str):
    """校验并返回当前用户（统一账号）。令牌为平台 JWT（/auth/* 登录签发）。"""
    token = (authorization or "").removeprefix("Bearer ").strip()
    payload = decode_access_token(token)
    if not payload:
        raise HTTPException(status_code=401, detail="未授权或登录已过期")
    return {"username": payload.get("username", ""), "sub": payload.get("sub", ""), "user_id": payload.get("sub", "")}


def _current_user(payload: dict) -> dict | None:
    """按平台 JWT( sub/username )取平台 users 行（统一账号源）。

    ⚠️ 同步旁路（经 training 数据层走 DATABASE_URL），仅限确认与平台同库的场景；
    业务端点一律用 `_current_user_async`（平台 Database，与 /auth 同源 DATABASE_URL）。
    """
    u = None
    if payload.get("sub"):
        u = db.fetch_one("users", id=payload["sub"])
    if u is None:
        u = db.fetch_one("users", username=payload.get("username"))
    return u


async def _current_user_async(payload: dict) -> dict | None:
    """按平台 JWT( sub/username )取平台 users 行——走平台 Database（与 /auth 同源）。

    旧实现直连 SQLite 文件，与平台 `DATABASE_URL` 不同源，平台侧创建/合并的账号
    （如运营端绑定账号）在此查不到 → 报名/接单等全部 401「登录状态异常」。
    统一改走平台仓储。
    """
    pdb = Database()
    try:
        u = None
        if payload.get("sub"):
            u = await pdb.users.get_by_id(payload["sub"])
        if u is None and payload.get("username"):
            u = await pdb.users.get_by_username(payload["username"])
        return u
    finally:
        await pdb.close()


# ================= 工具 =================
def read_answers(body: dict, key="answers"):
    a = body.get(key)
    return a if isinstance(a, dict) else {}


def _parse_topics(t):
    """accounts.topics 为 JSON 文本 → list，解析失败兜底 []"""
    if not t:
        return []
    if isinstance(t, list):
        return t
    try:
        v = json.loads(t)
        return v if isinstance(v, list) else []
    except Exception:
        return []


def _count_applied(event_id):
    """报名占用数（容量闸用）= 全部未取消/未驳回的报名（含待审核、待支付），防止超售。"""
    conn = db.get_conn()
    n = conn.execute("SELECT COUNT(*) AS c FROM bookings WHERE event_id=? AND status NOT IN ('cancelled','rejected')", (event_id,)).fetchone()["c"]
    conn.close()
    return n


def _count_enrolled(event_id):
    """已报名（成功）人数 = 审核通过/已确认 且 未取消/未驳回 的报名。

    展示口径：待审核（人工审核模式）、待支付、已取消、已驳回不计入；
    审核通过 +1、取消/删除/驳回自动 -1，与生命周期状态实时一致。
    """
    conn = db.get_conn()
    n = conn.execute("SELECT COUNT(*) AS c FROM bookings WHERE event_id=? AND audit_status IN ('approved','confirmed') AND status NOT IN ('cancelled','rejected')", (event_id,)).fetchone()["c"]
    conn.close()
    return n


def _user_payload(u):
    """统一构建 /me 与 /update-profile 返回的用户资料（读取平台 users，含报名资料）"""
    return {
        "username": u.get("username", ""),
        "name": u.get("nickname") or u.get("username", ""),
        "avatar": _resolve_url(u.get("avatar", "")),
        "phone": u.get("phone", "") or "",
        "phoneBound": bool(u.get("phone", "")),
        "status": u.get("opc_status", "") or "",
        "topics": _parse_topics(u.get("topics", "")),
        "source": u.get("source", "") or "",
    }


def _set_user_phone(username: str, phone: str) -> None:
    """把手机号写入平台 users（统一账号）。"""
    u = db.fetch_one("users", username=username) or db.fetch_one("users", id=username)
    if u:
        db.update_row("users", u["id"], {"phone": phone})


def jsonify(obj):
    # 处理 datetime 等 → dict
    return obj


# ================= 域名验证文件（微信/腾讯域名校验，须在根路径返回） =================
@app.get("/SpXvScDiDT.txt")
async def domain_verify():
    from fastapi.responses import Response
    return Response(content="d813c7bbab566807a833fbc3b0e20a9c", media_type="text/plain")


# ================= 认证 =================
@app.get("/api/health")
async def health():
    return {"ok": True, "service": "opc-fastapi"}


@app.post("/api/auth/register")
async def register():
    # 统一账号：注册改由平台 /auth/register 等，避免双套账号
    raise HTTPException(410, "请使用 /auth（统一账号）")
    entry = {"id": db.gen_id("U"), "username": username, "password": hash_password(password),
             "name": name or username, "contact": contact, "identities": json.dumps(["opc_member|certified"]),
             "created_at": now_iso()}
    await asyncio.to_thread(db.insert, "accounts", entry)
    return {"ok": True, "token": make_token(username), "username": username, "name": name or username}


@app.post("/api/auth/login")
async def login():
    # 统一账号：登录改由平台 /auth/login 等，避免双套账号
    raise HTTPException(410, "登录请使用 /auth（统一账号）")


# ============================= 扫码接单（任务中心）=============================
@app.post("/api/tasks/claim-by-code")
async def tasks_claim_by_code(request: Request):
    """扫码接单：task_code → 领单（published → claimed，记录 claimed_by/流水）。"""
    body = await request.json()
    code = str(body.get("task_code") or body.get("code") or "").strip()
    if not code:
        raise HTTPException(status_code=400, detail="缺少任务码")
    auth = request.headers.get("Authorization", "")
    async with AsyncSessionLocal() as session:
        pdb = Database(session=session)
        payload = require_auth(auth)
        # 统一账号：直接按平台 JWT 定位 users 行（accounts 表已废弃清空，勿再查）。
        user = await _current_user_async(payload)
        if not user:
            raise HTTPException(status_code=401, detail="请先登录")
        # 叠加制身份：OPC 为所有账号基础权限，任何角色均可领单（role=user.role）。
        actor = dict(user)
        actor["id"] = user["id"]
        actor["username"] = user["username"]
        actor["nickname"] = user.get("nickname") or user.get("username")
        actor["port"] = "opc"
        actor["role"] = user.get("role") or "opc_member"
        task = await TaskService(pdb).claim(code, actor, source="scan")
        return {"ok": True, "task": task}


@app.get("/api/tasks/my")
async def tasks_my_tasks(request: Request):
    """我的接单（扫码接单领取的任务）。"""
    auth = request.headers.get("Authorization", "")
    async with AsyncSessionLocal() as session:
        pdb = Database(session=session)
        payload = require_auth(auth)
        # 统一账号：按平台 JWT 定位 users 行（accounts 表已废弃清空，勿再查）。
        user = await _current_user_async(payload)
        if not user:
            raise HTTPException(status_code=401, detail="请先登录")
        items = await pdb.tasks.list(status="claimed") + await pdb.tasks.list(status="doing")
        mine = [t for t in items if t.get("claimed_by") == user["id"]]
        return {"items": mine}


@app.post("/api/auth/send-code")
async def send_code():
    # 统一账号：验证码改由平台 /auth/send-code 下发，避免双套账号
    raise HTTPException(410, "登录请使用 /auth/send-code（统一账号）")


def _same_phone_accounts(phone):
    """找出占用同一手机号的所有账号（username=phone 为主账号，或 phone=phone 的其它账号）"""
    conn = db.get_conn()
    rows = conn.execute("SELECT * FROM accounts WHERE username=? OR phone=?", (phone, phone)).fetchall()
    conn.close()
    return [dict(r) for r in rows]


def _adopt_phone(acct, phone):
    """把同一手机号下其它账号的数据合并进 acct（保留 acct），最后只留一个账号、双向可命中。

    合并规则：若被并入账号带 wxid 而 acct 没有，则把 wxid 迁移到 acct（保证微信登录按 wxid 仍能命中）；
    再把重复账号的报名/测评/日志等归属数据挪到 acct，删除重复账号。最终同一手机号只对应一个账号。
    """
    phone = str(phone or "")
    conn = db.get_conn()
    keep_id = acct["id"]
    acct_username = acct["username"]
    conn.execute("UPDATE accounts SET phone=?, phone_bound=1 WHERE id=?", (phone, keep_id))
    dups = [dict(r) for r in conn.execute(
        "SELECT * FROM accounts WHERE (username=? OR phone=?) AND id != ?", (phone, phone, keep_id)
    ).fetchall()]
    for d in dups:
        # 迁移微信标识，避免 acct 属手机号账号时丢失 wxid 而微信登录失效
        if d.get("wxid") and not conn.execute("SELECT wxid FROM accounts WHERE id=?", (keep_id,)).fetchone()["wxid"]:
            conn.execute("UPDATE accounts SET wxid=? WHERE id=?", (d["wxid"], keep_id))
        # 迁移归属数据（报名 + 各日志）
        conn.execute("UPDATE bookings SET username=?, contact=? WHERE username=? OR contact=?",
                     (acct_username, phone, d["username"], d.get("phone") or ""))
        for t in ("tests", "policy_logs", "plan_logs", "survey_logs"):
            conn.execute(f"UPDATE {t} SET username=? WHERE username=?", (acct_username, d["username"]))
        conn.execute("DELETE FROM accounts WHERE id=?", (d["id"],))
    conn.commit()
    conn.close()
    acct = db.fetch_one("accounts", id=keep_id)
    return acct


def _resolve_phone_account(phone):
    """返回该手机号对应的唯一账号（优先按 username=phone，否则按 phone=phone），并合并重复账号。"""
    acct = db.fetch_one("accounts", username=phone)
    if not acct:
        for a in _same_phone_accounts(phone):
            if a.get("phone") == phone and a["id"]:
                acct = a
                break
    if not acct:
        return None
    return _adopt_phone(acct, phone)


def auth_by_phone(body: dict):
    phone = str(body.get("phone", "")).strip()
    code = str(body.get("code", "")).strip()
    if not PHONE_RE.match(phone):
        raise HTTPException(400, "请输入正确的 11 位手机号")
    if not check_sms_code(phone, code):
        raise HTTPException(401, "验证码错误或已过期，请重新获取")
    # 统一按手机号解析：命中现有账号（含已绑该号的微信账号）则合并复用，避免同一手机号多个账号
    acct = _resolve_phone_account(phone)
    is_new = False
    if not acct:
        acct = {"id": db.gen_id("U"), "username": phone, "phone": phone, "name": str(body.get("name", "")).strip() or phone,
                "phone_bound": 1, "identities": json.dumps(["opc_member|certified"]), "created_at": now_iso()}
        db.insert("accounts", acct)
        is_new = True
    return acct, is_new


@app.post("/api/auth/phone-login")
async def phone_login():
    # 统一账号：手机号登录改由平台 /auth/phone-login，避免双套账号
    raise HTTPException(410, "登录请使用 /auth/phone-login（统一账号）")


@app.post("/api/auth/logout")
async def logout():
    return {"ok": True}


@app.get("/api/auth/verify")
async def verify(authorization: str = Header(default="")):
    payload = require_auth(authorization)
    return {"ok": True, "username": payload["username"]}


# -------- 微信登录 / 我的 / 绑定手机 --------
@app.post("/api/auth/wx-login")
async def wx_login():
    # 统一账号：微信登录改由平台 /auth/wx-login，避免双套账号
    raise HTTPException(410, "登录请使用 /auth/wx-login（统一账号）")


@app.get("/api/auth/me")
async def me(authorization: str = Header(default="")):
    payload = require_auth(authorization)
    u = await _current_user_async(payload)
    if not u:
        raise HTTPException(404, "账号不存在")
    return {"ok": True, "user": _user_payload(u)}


@app.post("/api/auth/update-profile")
async def update_profile(req: Request, authorization: str = Header(default="")):
    """更新当前用户资料：昵称/头像 + 报名资料（status/topics/source）"""
    payload = require_auth(authorization)
    b = await req.json()
    u = await _current_user_async(payload)
    if not u:
        raise HTTPException(404, "账号不存在")
    patch = {}
    if b.get("name") is not None and str(b["name"]).strip():
        patch["nickname"] = str(b["name"]).strip()
    if b.get("avatar") is not None:
        from ..infrastructure.oss import to_object_path
        patch["avatar"] = to_object_path(str(b["avatar"]).strip())
    if b.get("status") is not None:
        patch["opc_status"] = str(b["status"]).strip()
    if b.get("source") is not None:
        patch["source"] = str(b["source"]).strip()
    if b.get("topics") is not None:
        patch["topics"] = json.dumps(b["topics"], ensure_ascii=False) if isinstance(b["topics"], list) else str(b["topics"])
    if patch:
        await asyncio.to_thread(db.update_row, "users", u["id"], patch)
    u = await _current_user_async(payload)
    return {"ok": True, "user": _user_payload(u)}


@app.post("/api/auth/wx-phone")
async def wx_phone(req: Request, authorization: str = Header(default="")):
    """微信官方手机号一键绑定：前端 <Button openType="getPhoneNumber"> 回调的 code → 手机号"""
    if not WX_HAS_CONFIG:
        raise HTTPException(400, "后端未配置微信 AppID/Secret，无法使用微信手机号绑定")
    payload = require_auth(authorization)
    b = await req.json()
    pcode = str(b.get("code", ""))
    if not pcode:
        raise HTTPException(400, "缺少手机号授权凭证")
    try:
        token = await _wx_get_access_token()
    except Exception as exc:  # noqa: BLE001 - 微信凭据/网络异常统一转可读错误
        raise HTTPException(400, f"微信 access_token 获取失败，请检查 AppID/Secret 配置：{exc}") from exc
    if not token:
        raise HTTPException(400, "微信 access_token 获取失败，请检查 AppID/Secret 配置")
    url = f"https://api.weixin.qq.com/wxa/business/getuserphonenumber?access_token={token}"
    async with httpx.AsyncClient(timeout=10) as client:
        r = await client.post(url, json={"code": pcode})
        data = r.json()
    if data.get("errcode"):
        raise HTTPException(401, f"获取手机号失败：{data.get('errmsg', '')}")
    phone = (data.get("phone_info") or {}).get("purePhoneNumber", "")
    if not phone:
        raise HTTPException(400, "未获取到手机号")
    u = await _current_user_async(payload)
    if not u:
        raise HTTPException(404, "账号不存在")
    await asyncio.to_thread(_set_user_phone, u["username"], phone)
    return {"ok": True, "phoneBound": True, "phone": phone}


@app.post("/api/auth/bind-phone")
async def bind_phone(req: Request, authorization: str = Header(default="")):
    payload = require_auth(authorization)
    b = await req.json()
    phone = str(b.get("phone", "")).strip()
    code = str(b.get("code", "")).strip()
    if not PHONE_RE.match(phone):
        raise HTTPException(400, "请输入正确的 11 位手机号")
    if not await platform_sms.verify(phone, code):
        raise HTTPException(401, "验证码错误或已过期")
    u = await _current_user_async(payload)
    if not u:
        raise HTTPException(404, "账号不存在")
    await asyncio.to_thread(_set_user_phone, u["username"], phone)
    return {"ok": True, "phoneBound": True, "phone": phone}


# ================= 排期 / 活动 =================
def _upcoming(all_events):
    now = int(__import__("time").time() * 1000)
    upcoming = []
    for e in all_events:
        from datetime import datetime
        try:
            t = datetime.fromisoformat(e["start_at"]).timestamp() * 1000
        except Exception:
            t = 0
        if t >= now - 1000 * 60 * 30:
            upcoming.append(e)
    upcoming.sort(key=lambda e: e["start_at"])
    return upcoming


# ================= 活动体系常量（区域/类型/模式） =================
EVENT_REGIONS = ("昆明", "曲靖", "大理", "丽江", "版纳", "楚雄", "红河", "其他")
EVENT_CATEGORIES = ("峰会", "竞赛", "沙龙", "培训", "展会", "投融资")
EVENT_MODES = ("online", "offline", "both")  # 线上 / 线下 / 线上+线下


async def _visible_park_ids(payload: dict | None) -> set[str]:
    """当前用户可见的本园区活动范围：所属园区（park_members）集合；运营方=全部（返回 None 语义用 '*'）。

    未登录 → 空集合（只能看公开活动）。
    """
    if not payload or not payload.get("sub"):
        return set()
    try:
        user = await _current_user_async(payload)
        if user and _is_platform_operator(user.get("role") or ""):
            return {"*"}
        from ..infrastructure.models import ParkMember
        from sqlalchemy import select as _select
        pdb = Database()
        try:
            rows = (await pdb.session.execute(
                _select(ParkMember).where(ParkMember.user_id == payload.get("sub", ""),
                                          ParkMember.status == "active"))).scalars().all()
            return {m.park_id for m in rows}
        finally:
            await pdb.close()
    except Exception:
        return set()


def _category_type(category: str) -> str:
    """旧 type 列派生（兼容旧读取方）：沙龙→salon，其余→free。"""
    return "salon" if category == "沙龙" else "free"


def _review_status_of(b: dict, current: str = "") -> str:
    """发布/编辑时的审核状态：submit=draft 存草稿，否则提交审核 pending。"""
    return "draft" if str(b.get("submit") or "") == "draft" else "pending"


def _parse_price_fen(b: dict):
    """活动定价入参归一：priceFen(分) / priceYuan(元) / price_fen(分, admin 桥) → 整数分。

    返回 None 表示请求未携带价格字段（更新时跳过）；create 时由调用方 `or 0` 兜底。
    """
    if "priceFen" in b or "price_fen" in b:
        try:
            return max(0, int(b.get("priceFen", b.get("price_fen")) or 0))
        except (TypeError, ValueError):
            return 0
    if "priceYuan" in b:
        try:
            return max(0, int(round(float(b.get("priceYuan") or 0) * 100)))
        except (TypeError, ValueError):
            return 0
    return None


# ================= 报名体系升级：自定义表单 / 截止 / 管理员 / 隐藏信息 / 短信 =================

FORM_FIELD_TYPES = ("text", "textarea", "number", "phone", "select", "radio", "checkbox", "date")
FORM_FIELDS_MAX = 20
FORM_DATA_MAX_BYTES = 65536
ADMIN_PHONES_MAX = 5
EVENT_REMINDER_TOKEN = os.environ.get("EVENT_REMINDER_TOKEN", "pine-event-reminder-2026")


def _load_json(raw, default):
    """安全 JSON 解析（Text 列统一存储 JSON 字符串）。"""
    try:
        v = json.loads(raw or "")
        return v
    except (TypeError, ValueError):
        return default


def normalize_form_fields(raw) -> tuple[list, str]:
    """校验并规范化报名表单字段定义（events.form_fields_json）。

    支持 8 种类型；必填/选填/hidden 标记；select/radio/checkbox 需 1~20 个选项；
    最多 20 个字段，按 sort 排序返回。返回 (fields, error)，error 为空表示通过。
    """
    if not raw:
        return [], ""
    if not isinstance(raw, list) or len(raw) > FORM_FIELDS_MAX:
        return [], f"报名表单字段最多 {FORM_FIELDS_MAX} 个"
    seen = set()
    out = []
    for i, f in enumerate(raw):
        if not isinstance(f, dict):
            return [], f"字段 {i + 1} 格式不合法"
        fid = str(f.get("id") or "").strip() or f"f_{i + 1}"
        if fid in seen:
            return [], f"字段 ID 重复：{fid}"
        seen.add(fid)
        ftype = f.get("type")
        if ftype not in FORM_FIELD_TYPES:
            return [], f"字段 {fid} 类型不支持：{ftype}"
        label = str(f.get("label") or "").strip()
        if not label or len(label) > 50:
            return [], f"字段 {fid} 缺少标题或标题过长（≤50 字）"
        opts = f.get("options")
        if ftype in ("select", "radio", "checkbox"):
            if not isinstance(opts, list) or not opts or len(opts) > 20 or any(not str(o).strip() for o in opts):
                return [], f"字段 {fid} 的选项须为 1~20 个非空字符串"
            opts = list(dict.fromkeys(str(o).strip() for o in opts))  # 去重，避免重复渲染
        else:
            opts = []
        out.append({
            "id": fid,
            "type": ftype,
            "label": label,
            "required": bool(f.get("required")),
            "hidden": bool(f.get("hidden")),
            "placeholder": str(f.get("placeholder") or "")[:100],
            "options": opts,
            "sort": int(f.get("sort") or 0),
        })
    out.sort(key=lambda x: x["sort"])
    return out, ""


def normalize_form_data(form_fields: list, raw) -> tuple[dict, dict]:
    """校验并归一化报名表单值（bookings.form_data_json）。

    - 值与定义解耦存储：值为 {f_id: value}；读取时按定义渲染。
    - 值类型按字段类型强制：text/textarea/date/phone/number/select/radio → string；
      checkbox → 去重字符串数组（≤10 项）。
    - 返回 (values, errors)；errors 为 {f_id: message}，空表示通过。
    """
    if not form_fields:
        return {}, {}
    raw = raw if isinstance(raw, dict) else {}
    values: dict[str, object] = {}
    errors: dict[str, str] = {}
    for f in form_fields:
        fid = f["id"]
        val = raw.get(fid, "")
        if f["type"] == "checkbox":
            v = val if isinstance(val, list) else ([val] if val not in ("", None) else [])
            v = [str(x).strip() for x in v][:10]
            v = list(dict.fromkeys(v))
            if f["required"] and not v:
                errors[fid] = "请选择至少一项"
                continue
            if any(x not in f["options"] for x in v):
                errors[fid] = "选项不合法"
                continue
            values[fid] = v
            continue
        s = "" if val is None else str(val).strip()
        if f["required"] and not s:
            errors[fid] = "此项必填"
            continue
        if not s:
            values[fid] = ""
            continue
        if f["type"] == "number":
            if not re.fullmatch(r"\d{1,12}(\.\d{1,2})?", s):
                errors[fid] = "请输入有效数字（最多 12 位整数、2 位小数）"
                continue
        elif f["type"] == "phone":
            if not PHONE_RE.match(s):
                errors[fid] = "请输入正确的 11 位手机号"
                continue
        elif f["type"] == "text":
            if len(s) > 200:
                errors[fid] = "单行文本最长 200 字"
                continue
        elif f["type"] == "textarea":
            if len(s) > 2000:
                errors[fid] = "多行文本最长 2000 字"
                continue
        elif f["type"] == "date":
            try:
                datetime.fromisoformat(s.replace("Z", "+00:00"))
            except ValueError:
                errors[fid] = "日期格式不正确"
                continue
        elif f["type"] in ("select", "radio") and f["options"] and s not in f["options"]:
            errors[fid] = "选项不合法"
            continue
        values[fid] = s
    # 总量闸：整体 ≤ 64KB，防滥用
    try:
        if len(json.dumps(values, ensure_ascii=False).encode("utf-8")) > FORM_DATA_MAX_BYTES:
            return {}, {"__global__": "报名信息过长，请精简后提交"}
    except Exception:
        pass
    return values, errors


def normalize_activity_admins(raw) -> tuple[list, str]:
    """校验活动管理员（activity_admins_json）：手机号绑定 1~5 人，支持 {phone,name} 或裸手机号。"""
    if not raw:
        return [], ""
    if not isinstance(raw, list) or not raw or len(raw) > ADMIN_PHONES_MAX:
        return [], f"活动管理员最多 {ADMIN_PHONES_MAX} 人"
    out = []
    seen = set()
    for a in raw:
        phone = str((a or {}).get("phone") if isinstance(a, dict) else a or "").strip()
        if not PHONE_RE.match(phone):
            return [], f"管理员手机号不合法：{phone or '空'}"
        if phone in seen:
            return [], f"管理员手机号重复：{phone}"
        seen.add(phone)
        out.append({
            "phone": phone,
            "name": str((a or {}).get("name") if isinstance(a, dict) else "")[:50],
        })
    return out, ""


def normalize_publishers(raw) -> tuple[list, str]:
    """校验发布方（publishers_json）：手机号绑定 1~5 人，可编辑活动+审核报名。"""
    if not raw:
        return [], ""
    if not isinstance(raw, list) or not raw or len(raw) > ADMIN_PHONES_MAX:
        return [], f"发布方最多 {ADMIN_PHONES_MAX} 人"
    out = []
    seen = set()
    for a in raw:
        phone = str((a or {}).get("phone") if isinstance(a, dict) else a or "").strip()
        if not PHONE_RE.match(phone):
            return [], f"发布方手机号不合法：{phone or '空'}"
        if phone in seen:
            return [], f"发布方手机号重复：{phone}"
        seen.add(phone)
        out.append({
            "phone": phone,
            "name": str((a or {}).get("name") if isinstance(a, dict) else "")[:50],
        })
    return out, ""


def normalize_secret(raw) -> dict:
    """校验活动隐藏信息（secret_json）：{text, image, url, urlLabel} 均可选。"""
    if not isinstance(raw, dict):
        return {}
    return {
        "text": str(raw.get("text") or "")[:2000],
        "image": str(raw.get("image") or "").strip(),
        "url": str(raw.get("url") or "").strip(),
        "urlLabel": str(raw.get("urlLabel") or "")[:50],
    }


def _signup_state(e) -> str:
    """报名状态派生：设置 signup_deadline 且已过 → closed；未设置或未到 → open。"""
    dl = e.get("signup_deadline") or ""
    if not dl or _parse_ms(dl) <= 0:
        return "open"
    return "closed" if int(time.time() * 1000) > _parse_ms(dl) else "open"


async def _event_role_base(payload: dict | None, acct: dict | None, e: dict) -> tuple[bool, str, str]:
    """活动权限公共判断：返回 (是否已命中, 用户uid, 手机号)。创建者/运营方/园区管理员命中时第一个值为 True。"""
    if not payload:
        return False, "", ""
    uid = payload.get("sub") or payload.get("user_id") or ""
    if uid and e.get("publisher_id") == uid:
        return True, uid, ""
    phone = (acct or {}).get("phone") or ""
    role = (acct or {}).get("role") or ""
    if _is_platform_operator(role):
        return True, uid, phone
    if (e.get("scope") or "public") == "park":
        my_park = await _carrier_park_id(payload)
        if my_park and e.get("target_tenant_id") == my_park:
            return True, uid, phone
    return False, uid, phone


async def _can_edit_event(payload: dict | None, acct: dict | None, e: dict) -> bool:
    """活动编辑权限：创建者 / 发布方（手机号绑定）/ 平台运营方 / 本园区管理员。活动管理员（activity_admins）不可编辑。"""
    hit, _uid, phone = await _event_role_base(payload, acct, e)
    if hit:
        return True
    if phone:
        publishers = _load_json(e.get("publishers_json") or "", [])
        if isinstance(publishers, list) and any((p or {}).get("phone") == phone for p in publishers):
            return True
    return False


async def _can_review_booking(payload: dict | None, acct: dict | None, e: dict) -> bool:
    """报名审核权限：创建者 / 发布方 / 活动管理员（手机号绑定）/ 平台运营方 / 本园区管理员。"""
    hit, _uid, phone = await _event_role_base(payload, acct, e)
    if hit:
        return True
    if phone:
        for key in ("publishers_json", "activity_admins_json"):
            arr = _load_json(e.get(key) or "", [])
            if isinstance(arr, list) and any((a or {}).get("phone") == phone for a in arr):
                return True
    return False


async def _can_manage_event(payload: dict | None, acct: dict | None, e: dict) -> bool:
    """兼容别名：活动管理（编辑）权限 = _can_edit_event。审核报名请用 _can_review_booking。"""
    return await _can_edit_event(payload, acct, e)


def _secret_for(e: dict, viewer: dict | None, acct: dict | None, can_manage: bool) -> dict | None:
    """活动隐藏信息可见性：管理者直接可见；报名者仅审核通过（approved/confirmed）后本人可见。"""
    secret = _load_json(e.get("secret_json") or "", {})
    if not isinstance(secret, dict) or not (secret.get("text") or secret.get("image") or secret.get("url")):
        return None
    if can_manage:
        return secret
    if not viewer:
        return None
    # 注意：bookings.username 存的是平台登录名（create_booking 写入 acct.username），
    # 而 JWT 的 sub 是用户 id —— 必须优先用 username 匹配，否则审核通过后也取不到专属信息
    username = viewer.get("username") or viewer.get("sub") or ""
    phone = (acct or {}).get("phone") or ""
    if not username and not phone:
        return None
    conn = db.get_conn()
    try:
        rows = conn.execute(
            "SELECT username, contact, audit_status, status FROM bookings WHERE event_id=?",
            (e.get("id"),)).fetchall()
    finally:
        conn.close()
    for r in rows:
        if r["username"] in (username, phone) or r["contact"] in (username, phone):
            if (r["audit_status"] or "") in ("approved", "confirmed") and r["status"] != "cancelled":
                return secret
            return None
    return None


async def _apply_booking_audit(bid: str, decision: str, comment: str, payload: dict, acct: dict) -> dict:
    """报名审核统一收口：权限校验 + 状态联动 + 审核记录 + 审核短信（幂等）。

    审核动作必须经此入口（PATCH 与 /review 共用），杜绝绕过权限的直改。
    """
    if decision not in ("approved", "rejected"):
        raise HTTPException(400, "审核状态仅支持 approved/rejected")
    bk = await asyncio.to_thread(db.fetch_by_id, "bookings", bid)
    if not bk:
        raise HTTPException(404, "报名记录不存在")
    ev = await asyncio.to_thread(db.fetch_by_id, "events", bk.get("event_id", ""))
    if not ev:
        raise HTTPException(404, "活动不存在")
    if not await _can_review_booking(payload, acct, ev):
        raise HTTPException(403, "仅发布者/发布方/活动管理员/运营方可审核该活动报名")
    if (bk.get("audit_status") or "pending") == decision and bk.get("audit_at"):
        return bk  # 幂等：同结果不重复发短信
    from ..infrastructure.repositories import utcnow_iso
    # 状态联动：通过 → confirmed（付费未支付除外，保持待支付）；驳回 → rejected
    status_new = bk.get("status")
    if decision == "approved":
        pay_required = int(ev.get("price_fen") or 0) > 0
        if not (pay_required and bk.get("pay_status") != "paid"):
            status_new = "confirmed"
    else:
        status_new = "rejected"
    patch = {
        "audit_status": decision,
        "audit_by": (acct or {}).get("nickname") or payload.get("username", ""),
        "audit_at": utcnow_iso(),
        "audit_comment": str(comment or "")[:500],
        "status": status_new,
    }
    await asyncio.to_thread(db.update_row, "bookings", bid, patch)
    # 审核短信（幂等：audit_sms_at 为空才发送；失败不阻断审核）
    if not bk.get("audit_sms_at"):
        sent = False
        try:
            pdb = Database()
            try:
                cfg = await platform_sms.load_sms_config(pdb)
                phone = str(bk.get("contact") or "")
                if phone and PHONE_RE.match(phone):
                    await platform_sms.send_template(cfg, "booking_audit", phone, {
                        "event_title": ev.get("title") or "",
                        "status": "通过" if decision == "approved" else "未通过",
                        "reason": str(comment or ""),
                        "time": ev.get("start_at") or "",
                        "location": ev.get("location") or "",
                    })
                    sent = True
            finally:
                await pdb.close()
        except Exception as exc:  # noqa: BLE001 - 短信失败不阻断审核
            logging.getLogger(__name__).warning("booking_audit sms 失败: %s", exc)
        if sent:
            await asyncio.to_thread(db.update_row, "bookings", bid, {"audit_sms_at": now_iso()})
    return await asyncio.to_thread(db.fetch_by_id, "bookings", bid)


async def scan_event_reminders() -> dict:
    """活动前提醒扫描（定时任务入口）：对「开始时间 − remind_before_min 窗口内」
    且审核通过（approved/confirmed）、未取消、未提醒过的报名，发送 event_reminder 短信。
    """
    stats = {"scanned": 0, "sent": 0, "skipped": 0}
    now_ms = int(time.time() * 1000)
    events = await asyncio.to_thread(db.list_all, "events")
    conn = await asyncio.to_thread(db.get_conn, )
    try:
        bookings = conn.execute("SELECT * FROM bookings").fetchall()
    finally:
        conn.close()
    for ev in events:
        if (ev.get("review_status") or "approved") != "approved":
            continue
        remind_min = int(ev.get("remind_before_min") or 0)
        if remind_min <= 0:
            continue
        start_ms = _parse_ms(ev.get("start_at") or "")
        if start_ms <= 0:
            continue
        # 窗口：开始前 remind_before_min 至开始时刻（提前量负值/越界自然跳过）
        window_open = start_ms - remind_min * 60000
        if not (window_open <= now_ms <= start_ms):
            continue
        for b in bookings:
            if b.get("event_id") != ev["id"]:
                continue
            if (b.get("audit_status") or "") not in ("approved", "confirmed") or b.get("status") == "cancelled":
                continue
            if b.get("reminded_at"):
                continue
            phone = str(b.get("contact") or "")
            if not PHONE_RE.match(phone):
                continue
            sent = False
            try:
                pdb = Database()
                try:
                    cfg = await platform_sms.load_sms_config(pdb)
                    await platform_sms.send_template(cfg, "event_reminder", phone, {
                        "event_title": ev.get("title") or "",
                        "start_at": ev.get("start_at") or "",
                        "location": ev.get("location") or "",
                    })
                    sent = True
                finally:
                    await pdb.close()
            except Exception as exc:  # noqa: BLE001
                logging.getLogger(__name__).warning("event_reminder sms 失败: %s", exc)
            if sent:
                await asyncio.to_thread(db.update_row, "bookings", b["id"], {"reminded_at": now_iso()})
                stats["sent"] += 1
            stats["scanned"] += 1
    return stats


def event_out(e, detail: bool = False, viewer: dict | None = None,
              acct: dict | None = None, can_manage: bool = False, can_review: bool = False):
    """SQLite 行(snake_case) → 前端契约(camelCase) 统一映射。

    detail=True（详情页）才返回 body 富文本大字段，列表不返以免拖垮接口。
    can_manage = 可编辑活动；can_review = 可审核报名。
    """
    if not e:
        return e
    out = {
        "id": e.get("id"),
        "type": e.get("type"),
        "category": e.get("category") or "",
        "region": e.get("region") or "",
        "mode": e.get("mode"),
        "title": e.get("title"),
        "subtitle": e.get("subtitle"),
        "desc": e.get("desc"),
        "location": e.get("location"),
        "host": e.get("host"),
        "image": abs_url(e.get("image")),
        "shareImgTimeline": abs_url(e.get("share_img_timeline")),
        "shareImgMessage": abs_url(e.get("share_img_message")),
        "link": e.get("link"),
        "startAt": e.get("start_at"),
        "endAt": e.get("end_at"),
        "checkinAt": e.get("checkin_at"),
        "durationMin": e.get("duration_min"),
        "capacity": e.get("capacity"),
        "status": e.get("status"),
        "auditMode": e.get("audit_mode") or "auto",
        "showCapacity": bool(e.get("show_capacity")),
        "priceFen": int(e.get("price_fen") or 0),
        "scope": e.get("scope") or "public",
        "signupMode": e.get("signup_mode") or "mini",
        "externalUrl": e.get("external_url") or "",
        "reviewStatus": e.get("review_status") or "approved",
        "reviewComment": e.get("review_comment") or "",
        "publisherId": e.get("publisher_id") or "",
        "publisherName": e.get("publisher_name") or "",
        "enrolled": _count_enrolled(e.get("id")),
        "applied": _count_applied(e.get("id")),
        # ── 报名体系升级 ──
        "signupDeadline": e.get("signup_deadline") or "",
        "signupState": _signup_state(e),
        "remindBeforeMin": int(e.get("remind_before_min") or 1440),
        "showDefaultQuestion": bool(e.get("show_default_question", 1)),
        "canManage": can_manage,
        "canEdit": can_manage,
        "canReview": can_review or can_manage,
    }
    if detail:
        out["body"] = e.get("body") or ""
        out["notice"] = e.get("notice") or ""
        out["formFields"] = _load_json(e.get("form_fields_json") or "", [])
        if not isinstance(out["formFields"], list):
            out["formFields"] = []
        out["secret"] = _secret_for(e, viewer, acct, can_manage)
        out["activityAdmins"] = _load_json(e.get("activity_admins_json") or "", []) if (can_manage or can_review) else []
        out["publishers"] = _load_json(e.get("publishers_json") or "", []) if (can_manage or can_review) else []
    return out


def _park_out(r: dict, detail: bool = False):
    """park_tenants 行(snake_case) → C 端载体资料(camelCase)。列表不含富文本大字段。"""
    try:
        stats = json.loads(r.get("stats_json") or "[]")
    except (TypeError, ValueError):
        stats = []
    intro = []
    try:
        intro = json.loads(r.get("intro_json") or "[]")
    except (TypeError, ValueError):
        intro = []
    out = {
        "id": r.get("id"),
        "name": r.get("name"),
        "city": r.get("city") or "",
        "lng": r.get("lng"),
        "lat": r.get("lat"),
        "cover": abs_url(r.get("cover") or ""),
        "address": r.get("address") or "",
        "contact": r.get("contact") or "",
        "tags": [t for t in (r.get("tags") or "").split(",") if t],
        "stats": stats if isinstance(stats, list) else [],
        "summary": next((x for x in intro if x), ""),
    }
    if detail:
        out.update({
            "introHtml": r.get("intro_html") or "",
            "policyHtml": r.get("policy_html") or "",
            "admissionHtml": r.get("admission_html") or "",
        })
    return out


@app.get("/api/parks")
async def parks_list(region: Optional[str] = None):
    """C 端载体列表（云南载体地图/列表页）：active 园区 + 坐标/摘要，不含富文本大字段。"""
    rows = await asyncio.to_thread(db.list_all, "park_tenants")
    items = [r for r in rows if (r.get("status") or "active") == "active"]
    if region:
        items = [r for r in items if (r.get("city") or "其他") == region]
    items.sort(key=lambda r: (r.get("lng") is None, r.get("city") or "其他"))
    return {"ok": True, "list": [_park_out(r) for r in items]}


@app.get("/api/parks/{pid}")
async def parks_detail(pid: str):
    """C 端载体详情：含园区介绍/政策/入驻流程富文本与在园企业数。"""
    r = await asyncio.to_thread(db.fetch_by_id, "park_tenants", pid)
    if not r or (r.get("status") or "active") != "active":
        raise HTTPException(status_code=404, detail="园区不存在")
    out = _park_out(r, detail=True)
    comps = [c for c in await asyncio.to_thread(db.list_all, "park_companies")
             if c.get("tenant_id") == pid and (c.get("status") or "") == "active"]
    out["companyCount"] = len(comps)
    out["companies"] = [{"id": c.get("id"), "name": c.get("name"), "industry": c.get("industry") or "",
                         "bio": c.get("bio") or ""} for c in comps[:60]]
    return {"ok": True, "park": out}


@app.get("/api/events")
async def events(
    request: Request,
    current: Optional[str] = None,
    bookable: Optional[str] = None,
    category: Optional[str] = None,
    region: Optional[str] = None,
    mode: Optional[str] = None,
):
    all_events = await asyncio.to_thread(db.list_all, "events")
    # 园区活动可见性：本园区活动仅所属园区账号可见（运营方/未登录规则见 _visible_park_ids）
    token = (request.headers.get("authorization") or "").removeprefix("Bearer ").strip()
    vis_parks = await _visible_park_ids(decode_access_token(token) if token else None)
    if "*" not in vis_parks:
        all_events = [e for e in all_events
                      if (e.get("scope") or "public") == "public" or e.get("target_tenant_id") in vis_parks]
    # 服务端三维筛选（类型/区域/模式）
    if category:
        all_events = [e for e in all_events if (e.get("category") or "") == category]
    if region:
        all_events = [e for e in all_events if (e.get("region") or "") == region]
    if mode:
        all_events = [e for e in all_events if e.get("mode") == mode or e.get("mode") == "both"]
    if current is not None or bookable is not None:
        # 首页/可报名列表只展示已通过审核的活动
        approved = [e for e in all_events if (e.get("review_status") or "approved") == "approved"]
        upcoming = _upcoming(approved)
        if current is not None:
            free = next((e for e in upcoming if e["type"] == "free"), None)
            salon = next((e for e in upcoming if e["type"] == "salon"), None)
            return {"ok": True, "next": await asyncio.to_thread(event_out, upcoming[0] if upcoming else None),
                    "free": await asyncio.to_thread(event_out, free), "salon": await asyncio.to_thread(event_out, salon),
                    "upcoming": [await asyncio.to_thread(event_out, e) for e in upcoming[:6]]}
        if bookable is not None:
            # 可公开报名：排除 待开放(pending)/邀请中(invite)/已结束(done)/已满员(full)
            lst = [e for e in upcoming if e["status"] not in ("closed", "done", "pending", "invite", "full")][:30]
            return {"ok": True, "list": [await asyncio.to_thread(event_out, e) for e in lst]}
    return {"ok": True, "list": [await asyncio.to_thread(event_out, e) for e in all_events]}


@app.get("/api/events/mine")
async def my_events(authorization: str = Header(default="")):
    """我的活动（发布者视角，含审核状态）。"""
    payload = require_auth(authorization)
    all_events = await asyncio.to_thread(db.list_all, "events")
    mine = [e for e in all_events if e.get("publisher_id") == payload.get("user_id", "")]
    mine.sort(key=lambda e: e.get("created_at") or "", reverse=True)
    return {"ok": True, "list": [await asyncio.to_thread(event_out, e) for e in mine]}


@app.get("/api/events/managed")
async def managed_events(authorization: str = Header(default="")):
    """我管理的活动：发布方（publishers）+ 活动管理员（activity_admins）手机号命中的活动。"""
    payload = require_auth(authorization)
    acct = await _current_user_async(payload)
    phone = (acct or {}).get("phone") or ""
    if not phone:
        return {"ok": True, "list": []}
    all_events = await asyncio.to_thread(db.list_all, "events")
    mine = []
    for e in all_events:
        hit = False
        for key in ("publishers_json", "activity_admins_json"):
            arr = _load_json(e.get(key) or "", [])
            if isinstance(arr, list) and any((a or {}).get("phone") == phone for a in arr):
                hit = True
                break
        if hit:
            mine.append(e)
    mine.sort(key=lambda e: e.get("created_at") or "", reverse=True)
    return {"ok": True, "list": [await asyncio.to_thread(event_out, e) for e in mine]}


@app.get("/api/events/{eid}/qrcode")
async def event_qrcode(eid: str, authorization: str = Header(default="")):
    """报名小程序码：发布者/活动管理员/运营方生成，微信扫码直达该活动详情页报名。

    未配置 WECHAT_APPID/SECRET 时返回 503 并提示配置；小程序码为无限量 wxacode，
    scene 携带活动 id（扫码后由活动详情页解析）。
    """
    payload = require_auth(authorization)
    acct = await _current_user_async(payload)
    ev = await asyncio.to_thread(db.fetch_by_id, "events", eid)
    if not ev:
        raise HTTPException(404, "活动不存在")
    if not await _can_review_booking(payload, acct, ev):
        raise HTTPException(403, "仅发布者/发布方/活动管理员/运营方可生成报名码")
    scene = f"e={eid}"
    try:
        from ..services import wechat
        png = await wechat.get_wxacode(scene, page="pages-extra/event-detail/index")
    except Exception as exc:  # noqa: BLE001 - 统一转为可读错误（常见：未配置微信凭据）
        msg = str(exc) or "生成失败"
        if "access_token" in msg or "凭据" in msg:
            raise HTTPException(503, "未配置微信小程序凭据（WECHAT_APPID/WECHAT_SECRET），无法生成报名码") from exc
        raise HTTPException(503, f"报名码生成失败：{msg}") from exc
    import base64 as _b64
    from .. import config as _cfg
    return {"ok": True, "scene": scene, "envVersion": _cfg.WECHAT_WXACODE_ENV,
            "image": f"data:image/png;base64,{_b64.b64encode(png).decode('ascii')}"}


@app.get("/api/events/{eid}")
async def event_detail(eid: str, authorization: str = Header(default="")):
    e = await asyncio.to_thread(db.fetch_by_id, "events", eid)
    if not e:
        raise HTTPException(404, "活动不存在")
    # 未过审活动仅发布者可见（详情携带 body 富文本）
    token = (authorization or "").removeprefix("Bearer ").strip()
    payload = decode_access_token(token) if token else None
    acct = await _current_user_async(payload) if payload else None
    is_publisher = bool(payload) and ((payload.get("sub") or payload.get("user_id")) == e.get("publisher_id"))
    # 未过审：仅发布者可见
    if (e.get("review_status") or "approved") != "approved" and not is_publisher:
        raise HTTPException(404, "活动不存在")
    # 园区活动：仅所属园区账号/运营方可看
    if not is_publisher and (e.get("scope") or "public") == "park":
        vis_parks = await _visible_park_ids(payload)
        if "*" not in vis_parks and e.get("target_tenant_id") not in vis_parks:
            raise HTTPException(404, "活动不存在")
    can_manage = bool(payload) and await _can_manage_event(payload, acct, e)
    can_review = bool(payload) and await _can_review_booking(payload, acct, e)
    return {"ok": True, "event": await asyncio.to_thread(
        event_out, e, detail=True, viewer=payload, acct=acct, can_manage=can_manage, can_review=can_review)}


@app.post("/api/events")
async def create_event(req: Request, authorization: str = Header(default="")):
    payload = require_auth(authorization)
    b = await req.json()
    if not b.get("title") or not b.get("startAt"):
        raise HTTPException(400, "请填写活动名称 / 开始时间")
    category = str(b.get("category", "") or "").strip()
    if category and category not in EVENT_CATEGORIES:
        raise HTTPException(400, "活动类型不合法")
    region = str(b.get("region", "") or "").strip() or "其他"
    if region not in EVENT_REGIONS:
        raise HTTPException(400, "活动区域不合法")
    mode = b.get("mode") if b.get("mode") in EVENT_MODES else "offline"
    signup_mode = "external" if b.get("signupMode") == "external" else "mini"
    external_url = str(b.get("externalUrl", "") or "").strip()
    if signup_mode == "external" and not external_url:
        raise HTTPException(400, "外链报名须填写报名链接")
    scope = "park" if b.get("scope") == "park" else "public"
    target_tenant_id = str(b.get("targetTenantId", "") or "").strip()
    if scope == "park" and not target_tenant_id:
        raise HTTPException(400, "本园区活动须选择目标园区")
    if not b.get("endAt"):
        raise HTTPException(400, "请填写结束时间")
    # ── 报名体系升级字段校验 ──
    form_fields, ff_err = normalize_form_fields(b.get("formFields"))
    if ff_err:
        raise HTTPException(400, ff_err)
    admins, adm_err = normalize_activity_admins(b.get("activityAdmins"))
    if adm_err:
        raise HTTPException(400, adm_err)
    publishers, pub_err = normalize_publishers(b.get("publishers"))
    if pub_err:
        raise HTTPException(400, pub_err)
    secret = normalize_secret(b.get("secret"))
    try:
        remind = int(b.get("remindBeforeMin") if b.get("remindBeforeMin") is not None else 1440)
    except (TypeError, ValueError):
        remind = 1440
    if remind < 0 or remind > 10080:
        raise HTTPException(400, "提醒提前量须在 0~10080 分钟之间")
    deadline = str(b.get("signupDeadline", "") or "").strip()
    if deadline and _parse_ms(deadline) <= 0:
        raise HTTPException(400, "报名截止时间格式不正确")
    # 发布者（任意登录账号均可发布）
    user = await _current_user_async(payload)
    entry = {
        "id": db.gen_id("E"),
        "type": _category_type(category),
        "category": category,
        "region": region,
        "mode": mode,
        "title": str(b.get("title", "")).strip(),
        "subtitle": str(b.get("subtitle", "") or ""),
        "desc": str(b.get("desc", "") or ""),
        "body": str(b.get("body", "") or ""),
        "notice": str(b.get("notice", "") or ""),
        "location": str(b.get("location", "") or ""),
        "host": str(b.get("host", "") or "").strip(),
        "image": str(b.get("image", "") or "").strip(),
        "share_img_timeline": str(b.get("shareImgTimeline", "") or "").strip(),
        "share_img_message": str(b.get("shareImgMessage", "") or "").strip(),
        "link": str(b.get("link", "") or ""),
        "start_at": str(b.get("startAt", "") or ""),
        "end_at": str(b.get("endAt", "") or ""),
        "checkin_at": str(b.get("checkinAt", "") or ""),
        "duration_min": int(b.get("durationMin") or 90),
        "capacity": int(b.get("capacity") or 0),
        "status": str(b.get("status", "") or "open"),
        "audit_mode": 'manual' if b.get("auditMode") == "manual" else "auto",
        "show_capacity": 1 if b.get("showCapacity") else 0,
        "price_fen": _parse_price_fen(b) or 0,
        "scope": scope,
        "target_tenant_id": target_tenant_id,
        "signup_mode": signup_mode,
        "external_url": external_url,
        "review_status": _review_status_of(b),
        "publisher_id": payload.get("user_id", ""),
        "publisher_name": (user or {}).get("nickname") or payload.get("username", ""),
        # ── 报名体系升级 ──
        "form_fields_json": json.dumps(form_fields, ensure_ascii=False),
        "activity_admins_json": json.dumps(admins, ensure_ascii=False),
        "publishers_json": json.dumps(publishers, ensure_ascii=False),
        "secret_json": json.dumps(secret, ensure_ascii=False),
        "signup_deadline": deadline,
        "remind_before_min": remind,
        "show_default_question": 0 if b.get("showDefaultQuestion") is False else 1,
    }
    await asyncio.to_thread(db.insert, "events", entry)
    return {"ok": True, "entry": await asyncio.to_thread(event_out, entry, detail=True)}


@app.put("/api/events/{eid}")
async def update_event(eid: str, req: Request, authorization: str = Header(default="")):
    payload = require_auth(authorization)
    b = await req.json()
    existing = await asyncio.to_thread(db.fetch_by_id, "events", eid)
    if not existing:
        raise HTTPException(404, "活动不存在")
    # 仅发布者或平台运营方可编辑
    if existing.get("publisher_id") != payload.get("user_id", ""):
        user = await _current_user_async(payload)
        if (user or {}).get("role") not in ("operator", "operator_internal", "op_admin", "op_super_admin", "admin", "superadmin"):
            raise HTTPException(403, "仅发布者或运营方可编辑该活动")
    patch = {}
    for k in ["title", "subtitle", "desc", "location", "link", "start_at", "end_at", "checkin_at", "status", "host", "image"]:
        # 前端发 startAt/start_at、endAt/end_at、checkinAt/checkin_at 兼容
        alias = {"start_at": "startAt", "end_at": "endAt", "checkin_at": "checkinAt"}.get(k)
        src = k if k in b else (alias if alias else None)
        if src and b.get(src) is not None:
            patch[k] = b[src]
    if patch.get("image"):
        from ..infrastructure.oss import to_object_path
        patch["image"] = to_object_path(str(patch["image"]))
    # 微信分享图（朋友圈 1:1 / 会话 5:4；空字符串=清除，回退封面）
    for src, col in (("shareImgTimeline", "share_img_timeline"), ("shareImgMessage", "share_img_message")):
        if b.get(src) is not None:
            v = str(b.get(src) or "").strip()
            if v:
                from ..infrastructure.oss import to_object_path
                v = to_object_path(v)
            patch[col] = v
    if b.get("category") is not None:
        cat = str(b.get("category") or "").strip()
        if cat and cat not in EVENT_CATEGORIES:
            raise HTTPException(400, "活动类型不合法")
        patch["category"] = cat
        patch["type"] = _category_type(cat)
    if b.get("region") is not None:
        reg = str(b.get("region") or "").strip() or "其他"
        if reg not in EVENT_REGIONS:
            raise HTTPException(400, "活动区域不合法")
        patch["region"] = reg
    if b.get("mode") in EVENT_MODES:
        patch["mode"] = b["mode"]
    if b.get("body") is not None:
        patch["body"] = str(b.get("body") or "")
    if b.get("notice") is not None:
        patch["notice"] = str(b.get("notice") or "")
    if b.get("scope") in ("park", "public"):
        patch["scope"] = b["scope"]
    if b.get("targetTenantId") is not None:
        patch["target_tenant_id"] = str(b.get("targetTenantId") or "").strip()
    if b.get("signupMode") in ("mini", "external"):
        patch["signup_mode"] = b["signupMode"]
    if b.get("externalUrl") is not None:
        patch["external_url"] = str(b.get("externalUrl") or "").strip()
    if b.get("durationMin") is not None:
        patch["duration_min"] = int(b.get("durationMin") or 90)
    if b.get("capacity") is not None:
        patch["capacity"] = int(b.get("capacity") or 0)
    if b.get("auditMode") is not None:
        patch["audit_mode"] = "manual" if b["auditMode"] == "manual" else "auto"
    if b.get("showCapacity") is not None:
        patch["show_capacity"] = 1 if b["showCapacity"] else 0
    # ── 报名体系升级字段 ──
    if b.get("formFields") is not None:
        ff, ff_err = normalize_form_fields(b.get("formFields"))
        if ff_err:
            raise HTTPException(400, ff_err)
        patch["form_fields_json"] = json.dumps(ff, ensure_ascii=False)
    if b.get("activityAdmins") is not None:
        adm, adm_err = normalize_activity_admins(b.get("activityAdmins"))
        if adm_err:
            raise HTTPException(400, adm_err)
        patch["activity_admins_json"] = json.dumps(adm, ensure_ascii=False)
    if b.get("publishers") is not None:
        pub, pub_err = normalize_publishers(b.get("publishers"))
        if pub_err:
            raise HTTPException(400, pub_err)
        patch["publishers_json"] = json.dumps(pub, ensure_ascii=False)
    if b.get("secret") is not None:
        patch["secret_json"] = json.dumps(normalize_secret(b.get("secret")), ensure_ascii=False)
    if b.get("remindBeforeMin") is not None:
        try:
            remind = int(b.get("remindBeforeMin"))
        except (TypeError, ValueError):
            raise HTTPException(400, "提醒提前量须为整数分钟")
        if remind < 0 or remind > 10080:
            raise HTTPException(400, "提醒提前量须在 0~10080 分钟之间")
        patch["remind_before_min"] = remind
    if b.get("signupDeadline") is not None:
        dl = str(b.get("signupDeadline") or "").strip()
        if dl and _parse_ms(dl) <= 0:
            raise HTTPException(400, "报名截止时间格式不正确")
        patch["signup_deadline"] = dl
    if b.get("showDefaultQuestion") is not None:
        patch["show_default_question"] = 0 if b.get("showDefaultQuestion") is False else 1
    pf = _parse_price_fen(b)
    if pf is not None:
        patch["price_fen"] = pf
    if not patch:
        raise HTTPException(400, "无更新字段")
    if not await asyncio.to_thread(db.fetch_by_id, "events", eid):
        raise HTTPException(404, "排期不存在")
    # 编辑已过审活动 → 重新提交审核（存草稿除外）
    if (existing.get("review_status") or "approved") == "approved":
        patch["review_status"] = "pending"
        patch["review_comment"] = ""
    await asyncio.to_thread(db.update_row, "events", eid, patch)
    return {"ok": True, "entry": await asyncio.to_thread(event_out, await asyncio.to_thread(db.fetch_by_id, "events", eid), detail=True)}


@app.delete("/api/events/{eid}")
async def delete_event(eid: str, authorization: str = Header(default="")):
    payload = require_auth(authorization)
    e = await asyncio.to_thread(db.fetch_by_id, "events", eid)
    if e:
        if e.get("publisher_id") != payload.get("user_id", ""):
            user = await _current_user_async(payload)
            if (user or {}).get("role") not in ("operator", "operator_internal", "op_admin", "op_super_admin", "admin", "superadmin"):
                raise HTTPException(403, "仅发布者或运营方可删除该活动")
    await asyncio.to_thread(db.delete_row, "events", eid)
    return {"ok": True}


# ================= 活动审核（运营端 / 园区端统一入口） =================
def _is_platform_operator(role: str) -> bool:
    return role in ("operator", "operator_internal", "op_admin", "op_super_admin", "admin", "superadmin")


async def _carrier_park_id(payload: dict) -> str:
    """carrier 账号绑定的园区 id（叠加制：park_members.admin / park_tenants.operator_user_id）。"""
    from ..infrastructure.models import ParkMember, ParkTenant
    from sqlalchemy import select as _select

    uid = payload.get("sub", "")
    pdb = Database()
    try:
        m = await pdb.session.scalar(_select(ParkMember).where(
            ParkMember.user_id == uid, ParkMember.member_type == "admin",
            ParkMember.status == "active").limit(1))
        if m is not None:
            return m.park_id
        t = await pdb.session.scalar(_select(ParkTenant).where(
            ParkTenant.operator_user_id == uid).limit(1))
        return t.id if t is not None else ""
    finally:
        await pdb.close()


@app.post("/api/events/{eid}/review")
async def review_event(eid: str, req: Request, authorization: str = Header(default="")):
    """活动审核：公开活动=运营端审；本园区活动=目标园区管理员审。"""
    payload = require_auth(authorization)
    b = await req.json()
    decision = b.get("status")
    if decision not in ("approved", "rejected"):
        raise HTTPException(400, "审核状态仅支持 approved/rejected")
    e = await asyncio.to_thread(db.fetch_by_id, "events", eid)
    if not e:
        raise HTTPException(404, "活动不存在")

    user = await _current_user_async(payload)
    role = (user or {}).get("role") or ""
    allowed = False
    if _is_platform_operator(role):
        allowed = True
    else:
        # 园区管理员：仅可审核「本园区」范围内活动
        my_park = await _carrier_park_id(payload)
        if my_park and (e.get("scope") or "public") == "park" and e.get("target_tenant_id") == my_park:
            allowed = True
    if not allowed:
        raise HTTPException(403, "无权审核该活动")

    from ..infrastructure.repositories import utcnow_iso
    patch = {
        "review_status": decision,
        "review_comment": str(b.get("comment", "") or ""),
        "review_by": (user or {}).get("nickname") or payload.get("username", ""),
        "review_at": utcnow_iso(),
    }
    await asyncio.to_thread(db.update_row, "events", eid, patch)
    await write_event_audit(payload, e, decision, patch["review_comment"])
    return {"ok": True, "entry": await asyncio.to_thread(event_out, await asyncio.to_thread(db.fetch_by_id, "events", eid), detail=True)}


async def write_event_audit(payload: dict, e: dict, decision: str, comment: str) -> None:
    """审核动作落审计日志（best-effort）。"""
    try:
        pdb = Database()
        try:
            await pdb.audit.add(action="event.review", resource="event", resource_id=e.get("id", ""),
                          detail=f"decision={decision} comment={comment}",
                          user_id=payload.get("user_id", ""))
        finally:
            await pdb.close()
    except Exception:
        pass


@app.get("/api/events/pending/list")
async def pending_events(authorization: str = Header(default=""), scope: str = "", reviewStatus: str = "pending"):
    """待审核列表：运营端=全部；园区管理员=本园区范围内。支持 scope/reviewStatus 过滤。"""
    payload = require_auth(authorization)
    user = await _current_user_async(payload)
    role = (user or {}).get("role") or ""
    all_events = await asyncio.to_thread(db.list_all, "events")
    if _is_platform_operator(role):
        pass
    else:
        my_park = await _carrier_park_id(payload)
        if not my_park:
            raise HTTPException(403, "仅运营方或园区管理员可查看审核列表")
        all_events = [e for e in all_events
                      if (e.get("scope") or "public") == "park" and e.get("target_tenant_id") == my_park]
    if scope:
        all_events = [e for e in all_events if (e.get("scope") or "public") == scope]
    if reviewStatus and reviewStatus != "all":
        all_events = [e for e in all_events if (e.get("review_status") or "approved") == reviewStatus]
    all_events.sort(key=lambda e: e.get("created_at") or "", reverse=True)
    return {"ok": True, "list": [await asyncio.to_thread(event_out, e) for e in all_events]}


# ================= 报名 =================
@app.post("/api/bookings")
async def create_booking(req: Request, authorization: str = Header(default="")):
    b = await req.json()
    event = await asyncio.to_thread(db.fetch_by_id, "events", str(b.get("eventId", "")).strip())
    # 校验活动可报名：公开报名仅 open/full；待开放/邀请中/已结束不可报名
    if not event:
        raise HTTPException(404, "活动不存在")
    if event["status"] not in ("open", "full"):
        label = {"pending": "活动待开放", "invite": "本场为定向邀请", "done": "本场已结束"}.get(event["status"], "暂不可报名")
        raise HTTPException(400, label)
    # 审核闸：未通过审核的活动不可报名
    if (event.get("review_status") or "approved") != "approved":
        raise HTTPException(400, "活动未通过审核，暂不可报名")
    # 报名截止闸：设置了截止时间且已过 → 拒绝报名
    if _signup_state(event) == "closed":
        raise HTTPException(400, "本场报名已截止")
    # 名额闸：限制人数（capacity>0）时，占用数（含待审核/待支付，防超售）达到上限即拒绝
    _capacity = int(event.get("capacity") or 0)
    if _capacity > 0 and _count_applied(event["id"]) >= _capacity:
        raise HTTPException(400, "本场名额已满")
    # 报名方式闸：外链报名活动不出站内报名
    if (event.get("signup_mode") or "mini") == "external":
        raise HTTPException(400, "该活动为外部报名，请前往报名链接")

    # 强制登录：必须携带有效 token（无登录 → 401）
    payload = require_auth(authorization)
    acct = await _current_user_async(payload)
    if not acct:
        raise HTTPException(401, "登录状态异常，请重新登录")
    # 报名必须已绑定手机号（登录后自动/引导绑定），保证报名有联系方式
    if not acct.get("phone"):
        raise HTTPException(400, "请先绑定手机号后再报名")

    # 关联键取平台 users 行的 username（JWT 断言可能缺失，且保证与 /api/bookings/mine 等一致）
    username = acct.get("username") or payload.get("username") or ""
    # 报名资料（姓名/状态/主题/来源）取账号中已保存的个人中心设置，报名区无需再填；question 按场次从请求体取
    name = (acct.get("nickname") or "") or (str(b.get("name", "")).strip() or username)
    status_label = (acct.get("opc_status") or "") or str(b.get("status", "") or "").strip()
    topics_val = _parse_topics(acct.get("topics"))
    if not topics_val:
        topics_val = b.get("topics", []) if isinstance(b.get("topics"), list) else []
    source = (acct.get("source") or "") or str(b.get("source", "") or "").strip()
    # 免费且非人工审核 → 报名即通过：status=confirmed 与 audit_status=approved 保持一致；
    # 人工审核 / 付费活动 → 先挂起（pending），审核通过或支付成功后再由相应流程置 confirmed
    pay_required = int(event.get("price_fen") or 0) > 0
    audit_auto = event.get("audit_mode") != "manual"
    # 自定义表单：按活动定义逐字段校验（必填/类型/枚举/长度），错误返回字段级提示
    form_fields = _load_json(event.get("form_fields_json") or "", [])
    if not isinstance(form_fields, list):
        form_fields = []
    form_values, form_errors = normalize_form_data(form_fields, b.get("formData"))
    if form_errors:
        fkey = next(iter(form_errors))
        raise HTTPException(400, {"code": "FIELD_INVALID", "field": fkey, "message": form_errors[fkey]})
    entry = {
        "id": db.gen_id("B"), "created_at": now_iso(),
        "status": "confirmed" if (not pay_required and audit_auto) else "pending",
        "username": username, "name": name,
        "contact": (acct.get("phone") or ""), "status_label": status_label,
        "want": b.get("want", ""),
        "event_id": event["id"], "event_title": event["title"], "event_start": event["start_at"],
        "topics": json.dumps(topics_val, ensure_ascii=False),
        "question": b.get("question", ""), "source": source,
        "audit_status": "approved" if (not pay_required and audit_auto) else "pending",
        # ── 报名体系升级：自定义表单值（与定义解耦，定义在 events.form_fields_json） ──
        "form_data_json": json.dumps(form_values, ensure_ascii=False),
    }
    # 付费活动：报名先挂起（unpaid/不自动通过），支付成功后由回调置 paid/confirmed/approved
    if pay_required:
        entry["pay_status"] = "unpaid"
        entry["audit_status"] = "pending"
    await asyncio.to_thread(db.insert, "bookings", entry)
    resp = {"ok": True, "id": entry["id"], "createdAt": entry["created_at"],
            "username": username, "name": name,
            "auditStatus": entry["audit_status"],
            "payRequired": pay_required,
            "priceFen": int(event.get("price_fen") or 0)}
    if pay_required:
        # Web 端扫码支付：生成「支付小程序码」，微信扫码直达小程序支付页（scene=报名单 id）
        try:
            from ..services import wechat
            png = await wechat.get_wxacode(entry["id"], page="pages-extra/pay-qr/index")
            import base64 as _b64
            resp["payQrImage"] = f"data:image/png;base64,{_b64.b64encode(png).decode('ascii')}"
        except Exception as exc:  # noqa: BLE001 - 未配置微信凭据时不阻断报名
            logging.getLogger(__name__).warning("pay-qr wxacode 生成失败: %s", exc)
            resp["payQrImage"] = ""
    return resp


# ---------------- Web 扫码支付：小程序支付页取单 ----------------
@app.get("/api/bookings/pay-qr/{scene}")
async def booking_pay_qr(scene: str):
    """小程序扫码进入支付页：按 scene（=报名单 id）取报名+活动摘要。未支付才可取。"""
    booking = await asyncio.to_thread(db.fetch_by_id, "bookings", scene)
    if not booking:
        raise HTTPException(404, "支付单不存在或已过期")
    event = await asyncio.to_thread(db.fetch_by_id, "events", booking.get("event_id", ""))
    if (booking.get("pay_status") or "") == "paid":
        return {"ok": True, "paid": True, "booking": {"id": booking["id"], "eventTitle": booking.get("event_title", "")}}
    return {
        "ok": True, "paid": False,
        "booking": {
            "id": booking["id"], "eventTitle": booking.get("event_title", ""),
            "eventStart": booking.get("event_start", ""), "name": booking.get("name", ""),
            "payStatus": booking.get("pay_status", "unpaid"),
            "auditStatus": booking.get("audit_status", "pending"),
            "priceFen": int(event.get("price_fen") or 0) if event else 0,
        },
    }


# ---------------- 报名支付（付费活动：课程/沙龙定价，复用 app/pay 支付链路） ----------------
async def _owned_booking(bid: str, payload: dict, acct: dict | None) -> dict:
    """按 id 取报名单并校验归属（username / 绑定手机号），通过返回行，否则 404/403。"""
    bk = await asyncio.to_thread(db.fetch_by_id, "bookings", bid)
    if not bk:
        raise HTTPException(404, "报名记录不存在")
    ids = {payload.get("username", "")}
    if acct and acct.get("phone"):
        ids.add(acct["phone"])
    if bk.get("username") not in ids and (bk.get("contact") or "") not in ids:
        raise HTTPException(403, "无权操作该记录")
    return bk


@app.post("/api/bookings/{bid}/pay")
async def pay_booking(bid: str, authorization: str = Header(default="")):
    """发起/继续支付：创建或复用未过期活动订单，返回 wx.requestPayment 参数。"""
    payload = require_auth(authorization)
    acct = await _current_user_async(payload)
    if not acct:
        raise HTTPException(401, "登录状态异常，请重新登录")
    bk = await _owned_booking(bid, payload, acct)
    if bk.get("pay_status") == "paid":
        return {"ok": True, "paid": True}
    ev = await asyncio.to_thread(db.fetch_by_id, "events", bk.get("event_id", ""))
    if not ev:
        raise HTTPException(404, "活动不存在")
    if int(ev.get("price_fen") or 0) <= 0:
        raise HTTPException(400, "本场活动免费，无需支付")
    from ..pay.event_service import create_order_for_booking
    pdb = Database()
    try:
        r = await create_order_for_booking(pdb, acct, dict(bk), dict(ev))
    finally:
        await pdb.close()
    return {"ok": True, "paid": False, **r}


@app.post("/api/bookings/{bid}/pay-status")
async def booking_pay_status(bid: str, authorization: str = Header(default="")):
    """查支付状态（微信侧对账兜底）：前端 wx.requestPayment 成功后轮询。"""
    payload = require_auth(authorization)
    acct = await _current_user_async(payload)
    bk = await _owned_booking(bid, payload, acct)
    from ..pay.event_service import booking_pay_status as _status
    pdb = Database()
    try:
        r = await _status(pdb, dict(bk))
    finally:
        await pdb.close()
    return {"ok": True, **r}


BOOKING_UNPAID_TTL_MIN = 15   # 未支付报名保留时长，超时自动关闭并删除


def _expire_unpaid_bookings() -> int:
    """未支付报名 15 分钟自动关闭：删除报名行 + 关闭关联支付订单（幂等，读取时懒执行）。"""
    conn = db.get_conn()
    rows = conn.execute(
        "SELECT id, event_id FROM bookings WHERE pay_status='unpaid' AND created_at <= ?",
        ((datetime.utcnow() - timedelta(minutes=BOOKING_UNPAID_TTL_MIN)).strftime('%Y-%m-%dT%H:%M:%SZ'),)
    ).fetchall()
    conn.close()
    if not rows:
        return 0
    ids = [r["id"] for r in rows]
    conn = db.get_conn()
    conn.executemany("DELETE FROM bookings WHERE id=?", [(i,) for i in ids])
    conn.commit()
    conn.close()
    # 关联支付订单一并关闭（异步引擎订单由对账兜底，这里同步改单）
    try:
        c2 = db.get_conn()
        c2.executemany(
            "UPDATE event_orders SET status='closed' WHERE booking_id=? AND status='pending'",
            [(i,) for i in ids])
        c2.commit()
        c2.close()
    except Exception:
        pass
    return len(ids)


def _decode_bk(row):
    d = dict(row)
    try:
        d["topics"] = json.loads(d.get("topics") or "[]")
    except Exception:
        d["topics"] = []
    d["payStatus"] = d.get("pay_status") or ""
    d["orderNo"] = d.get("order_no") or ""
    d["appealStatus"] = d.get("appeal_status") or ""
    d["appealReason"] = d.get("appeal_reason") or ""
    # ── 报名体系升级：自定义表单值 / 审核记录 / 隐藏信息可见性 ──
    try:
        d["formData"] = json.loads(d.get("form_data_json") or "{}")
    except Exception:
        d["formData"] = {}
    d["auditBy"] = d.get("audit_by") or ""
    d["auditAt"] = d.get("audit_at") or ""
    d["auditComment"] = d.get("audit_comment") or ""
    d["auditSmsAt"] = d.get("audit_sms_at") or ""
    d["canViewSecret"] = (d.get("audit_status") or "") in ("approved", "confirmed") and d.get("status") != "cancelled"
    return d


@app.post("/api/bookings/{bid}/cancel")
async def cancel_booking(bid: str, authorization: str = Header(default="")):
    """取消报名（仅报名人）—— 仅限「待参加」（审核通过且未开始/未结束）。

    规则：
    - 未支付 / 待审核 → 应删除报名（走 delete 端点），不在此处理；
    - 已支付 → 不可取消（退款走线下，联系运营方）；
    - 已签到 / 活动已结束 → 不可取消。
    """
    payload = require_auth(authorization)
    bk = await asyncio.to_thread(db.fetch_by_id, "bookings", bid)
    if not bk:
        raise HTTPException(404, "报名记录不存在")
    # 归属校验：报名单 username 或联系手机号 = 当前账号
    acct = await _current_user_async(payload)
    ids = {payload.get("username", "")}
    if acct and acct.get("phone"):
        ids.add(acct["phone"])
    if bk.get("username") not in ids and (bk.get("contact") or "") not in ids:
        raise HTTPException(403, "仅报名人可取消报名")
    # 状态闸
    if (bk.get("pay_status") or "") == "unpaid" or bk.get("status") == "pending":
        raise HTTPException(400, "未支付/待审核的报名请直接删除报名")
    if (bk.get("pay_status") or "") == "paid":
        raise HTTPException(400, "已支付的报名不可取消，如需退款请联系运营方")
    if bk.get("checkin_at"):
        raise HTTPException(400, "该场次已签到，不可取消")
    if bk.get("status") in ("converted", "cancelled"):
        raise HTTPException(400, "该报名已完结或已取消")
    if bk.get("audit_status") not in ("approved", "confirmed"):
        raise HTTPException(400, "报名尚未审核通过，请删除报名后重新报名")
    event = await asyncio.to_thread(db.fetch_by_id, "events", bk.get("event_id", ""))
    if event:
        end_ms = _parse_ms(event.get("end_at")) or 0
        if end_ms and end_ms < int(time.time() * 1000):
            raise HTTPException(400, "活动已结束，无需取消")
    await asyncio.to_thread(db.update_row, "bookings", bid, {"status": "cancelled", "audit_status": "rejected"})
    return {"ok": True, "id": bid, "status": "cancelled"}


@app.post("/api/bookings/{bid}/delete")
async def delete_booking(bid: str, authorization: str = Header(default="")):
    """删除报名（仅报名人）—— 适用未支付 / 待审核状态（物理删除，名额即刻释放）。"""
    payload = require_auth(authorization)
    bk = await asyncio.to_thread(db.fetch_by_id, "bookings", bid)
    if not bk:
        raise HTTPException(404, "报名记录不存在")
    acct = await _current_user_async(payload)
    ids = {payload.get("username", "")}
    if acct and acct.get("phone"):
        ids.add(acct["phone"])
    if bk.get("username") not in ids and (bk.get("contact") or "") not in ids:
        raise HTTPException(403, "仅报名人可删除报名")
    if (bk.get("pay_status") or "") == "paid":
        raise HTTPException(400, "已支付的报名不可删除，如需退款请联系运营方")
    if bk.get("status") == "cancelled":
        raise HTTPException(400, "该报名已取消，无需删除")
    if bk.get("audit_status") == "approved" and (bk.get("pay_status") or "") != "unpaid":
        raise HTTPException(400, "报名已生效（待参加），请使用取消报名")
    conn = db.get_conn()
    conn.execute("DELETE FROM bookings WHERE id=?", (bid,))
    conn.commit()
    conn.close()
    try:
        c2 = db.get_conn()
        c2.execute("UPDATE event_orders SET status='closed' WHERE booking_id=? AND status='pending'", (bid,))
        c2.commit()
        c2.close()
    except Exception:
        pass
    return {"ok": True, "id": bid, "deleted": True}


# ── 爽约申诉 ──
@app.post("/api/bookings/{bid}/appeal")
async def appeal_booking(bid: str, req: Request, authorization: str = Header(default="")):
    """爽约申诉（仅报名人）：活动结束后被标记爽约 → 提交申诉，由运营端复核。"""
    payload = require_auth(authorization)
    b = await req.json()
    reason = str(b.get("reason", "") or "").strip()
    if not reason:
        raise HTTPException(400, "请填写申诉理由")
    bk = await asyncio.to_thread(db.fetch_by_id, "bookings", bid)
    if not bk:
        raise HTTPException(404, "报名记录不存在")
    acct = await _current_user_async(payload)
    ids = {payload.get("username", "")}
    if acct and acct.get("phone"):
        ids.add(acct["phone"])
    if bk.get("username") not in ids and (bk.get("contact") or "") not in ids:
        raise HTTPException(403, "仅报名人可申诉")
    if bk.get("checkin_at"):
        raise HTTPException(400, "该报名已签到，无需申诉")
    if (bk.get("appeal_status") or "") == "pending":
        raise HTTPException(400, "申诉审核中，请勿重复提交")
    from ..infrastructure.repositories import utcnow_iso
    await asyncio.to_thread(db.update_row, "bookings", bid, {
        "appeal_status": "pending", "appeal_reason": reason, "appeal_at": utcnow_iso(),
    })
    return {"ok": True, "id": bid, "appealStatus": "pending"}


@app.get("/api/bookings/mine")
async def my_bookings(authorization: str = Header(default="")):
    payload = require_auth(authorization)
    acct = await _current_user_async(payload)
    ids = {payload["username"]}
    if acct and acct.get("phone"):
        ids.add(acct["phone"])
    await asyncio.to_thread(_expire_unpaid_bookings)   # 未支付 15 分钟自动关闭并删除
    conn = await asyncio.to_thread(db.get_conn, )
    rows = conn.execute("SELECT * FROM bookings").fetchall()
    events = {r["id"]: dict(r) for r in conn.execute("SELECT * FROM events").fetchall()}
    conn.close()
    lst = [_decode_bk(r) for r in rows if r["username"] in ids or (r["contact"] or "") in ids]
    now_ms = int(time.time() * 1000)
    for x in lst:
        # 前端（我的报名）用 camelCase，补别名保证一致的契约（snake 亦保留）
        x["eventTitle"] = x.get("event_title")
        x["eventStart"] = x.get("event_start")
        x["createdAt"] = x.get("created_at")
        x["checkinAt"] = x.get("checkin_at")
        x["auditStatus"] = x.get("audit_status") or "pending"
        ev = events.get(x.get("event_id"))
        if ev:
            open_ms = _parse_ms(ev.get("checkin_at") or ev.get("start_at"))
            end_ms = _parse_ms(ev.get("end_at")) or (open_ms + (ev.get("duration_min") or 90) * 60000)
            x["event_status"] = ev.get("status")
            # 活动审核状态（供前端过滤：审核中的活动不进入倒计时/待参加列表）
            x["eventReviewStatus"] = ev.get("review_status") or "approved"
            x["event_end"] = ev.get("end_at")
            x["event_checkin_at"] = ev.get("checkin_at")
            x["priceFen"] = int(ev.get("price_fen") or 0)
            x["can_checkin"] = bool(open_ms <= now_ms <= end_ms and ev.get("status") != "done")
            # 报名体系升级：附表单定义（供按定义渲染 formData 与隐藏字段）
            ff = _load_json(ev.get("form_fields_json") or "", [])
            x["formFields"] = ff if isinstance(ff, list) else []
            x["eventSignupDeadline"] = ev.get("signup_deadline") or ""
            # 报名体系升级：审核通过后可查看活动的隐藏信息（专属信息）
            if x.get("canViewSecret"):
                sv = _load_json(ev.get("secret_json") or "", {})
                x["secret"] = sv if isinstance(sv, dict) and (sv.get("text") or sv.get("image") or sv.get("url")) else None
            else:
                x["secret"] = None
        else:
            x["event_status"] = None
            x["eventReviewStatus"] = "approved"
            x["event_end"] = None
            x["event_checkin_at"] = None
            x["can_checkin"] = False
            x["formFields"] = []
        # ── 生命周期派生 ──
        ended_ev = x.get("event_status") == "done" or (x.get("event_end") and _parse_ms(x.get("event_end")) and _parse_ms(x.get("event_end")) < now_ms)
        x["no_show"] = bool(ended_ev and not x.get("checkin_at")
                            and (x.get("audit_status") or "") in ("approved", "confirmed")
                            and x.get("status") != "cancelled")
        x["cancelled"] = x.get("status") == "cancelled"
        # 删除报名（未支付/待审核）
        x["can_delete"] = (not x["cancelled"]) and (x.get("payStatus") or "") != "paid" \
            and ((x.get("payStatus") or "") == "unpaid" or (x.get("audit_status") or "pending") == "pending")
        # 取消报名（待参加：审核通过且未签到、未结束）
        x["can_cancel"] = (not x["cancelled"]) and not x["no_show"] and (x.get("payStatus") or "") != "paid" \
            and not x.get("checkin_at") and (x.get("audit_status") or "") in ("approved", "confirmed") \
            and x.get("event_status") not in ("done",)
        # 展示状态优先级：已取消 > 爽约(申诉中/未申诉) > 未支付 > 审核中 > 未通过 > 已结束 > 待参加
        if x["cancelled"]:
            x["displayStatus"] = "cancelled"
        elif x["no_show"]:
            x["displayStatus"] = "appealing" if (x.get("appeal_status") or "") == "pending" else "no_show"
        elif (x.get("payStatus") or "") == "unpaid":
            x["displayStatus"] = "unpaid"
        elif (x.get("audit_status") or "pending") == "pending":
            x["displayStatus"] = "pending"
        elif (x.get("audit_status") or "") == "rejected":
            x["displayStatus"] = "rejected"
        elif ended_ev:
            x["displayStatus"] = "done"
        else:
            x["displayStatus"] = "upcoming"
    lst.sort(key=lambda x: x["created_at"], reverse=True)
    return {"ok": True, "list": lst}


@app.get("/api/bookings/by-event/{event_id}")
async def my_booking_for_event(event_id: str, authorization: str = Header(default="")):
    """查当前登录用户是否已报名该活动及其审核状态（详情页判断用）"""
    payload = require_auth(authorization)
    acct = await _current_user_async(payload)
    ids = {payload["username"]}
    if acct and acct.get("phone"):
        ids.add(acct["phone"])
    conn = await asyncio.to_thread(db.get_conn, )
    rows = conn.execute("SELECT * FROM bookings WHERE event_id=?", (event_id,)).fetchall()
    conn.close()
    hit = None
    for r in rows:
        if r["username"] in ids or (r["contact"] or "") in ids:
            hit = _decode_bk(r)
            break
    if not hit:
        return {"ok": True, "booked": False, "auditStatus": None}
    return {"ok": True, "booked": True, "id": hit["id"], "auditStatus": hit.get("audit_status") or "pending",
            "status": hit.get("status"), "checkinAt": hit.get("checkin_at"),
            "payStatus": hit.get("pay_status") or "", "orderNo": hit.get("order_no") or "",
            "auditComment": hit.get("auditComment") or "",
            "canViewSecret": bool(hit.get("canViewSecret"))}


@app.post("/api/checkins")
async def checkin(req: Request, authorization: str = Header(default="")):
    payload = require_auth(authorization)
    b = await req.json()
    acct = await _current_user_async(payload)
    ids = {payload["username"]}
    if acct and acct.get("phone"):
        ids.add(acct["phone"])
    bk = await asyncio.to_thread(db.fetch_by_id, "bookings", b.get("bookingId", ""))
    if not bk:
        raise HTTPException(404, "报名记录不存在")
    if bk["username"] not in ids and (bk.get("contact") or "") not in ids:
        raise HTTPException(403, "无权操作该记录")
    if bk.get("checkin_at"):
        raise HTTPException(400, "该场次已签到")
    # 付费闸：定价活动必须先完成支付才能签到
    if bk.get("pay_status") == "unpaid":
        raise HTTPException(400, "请先完成支付后再签到")
    # 时间闸：只能在「签到开放时间 ≤ now ≤ 结束时间」内签到
    ev = await asyncio.to_thread(db.fetch_by_id, "events", bk.get("event_id", ""))
    open_ms = _parse_ms(ev.get("checkin_at") or ev.get("start_at")) if ev else 0
    end_ms = (_parse_ms(ev.get("end_at")) if ev and ev.get("end_at") else 0) or (_parse_ms(ev.get("start_at")) + (ev.get("duration_min") or 90) * 60000 if ev else 0)
    now_ms = int(time.time() * 1000)
    if now_ms < open_ms:
        raise HTTPException(400, "未到签到时间，活动开始后再签到")
    if now_ms > end_ms or (ev and ev.get("status") == "done"):
        raise HTTPException(400, "本场已结束，无法签到")
    await asyncio.to_thread(db.update_row, "bookings", bk["id"], {"checkin_at": now_iso()})
    return {"ok": True, "entry": _decode_bk(await asyncio.to_thread(db.fetch_by_id, "bookings", bk["id"]))}


@app.get("/api/bookings")
async def admin_bookings(authorization: str = Header(default=""), status: Optional[str] = None, audit: Optional[str] = None):
    require_auth(authorization)
    await asyncio.to_thread(_expire_unpaid_bookings)   # 未支付超时清理
    lst = [_decode_bk(r) for r in await asyncio.to_thread(db.list_all, "bookings")]
    lst.sort(key=lambda x: x["created_at"], reverse=True)
    if status:
        lst = [x for x in lst if x["status"] == status]
    if audit:
        lst = [x for x in lst if x.get("audit_status") == audit]
    return {"ok": True, "list": lst}


@app.patch("/api/bookings/{bid}")
async def update_booking(bid: str, req: Request, authorization: str = Header(default="")):
    """更新报名单（管理视角）：发布者/活动管理员/运营方可编辑资料或执行审核（触发短信）。

    原实现仅 require_auth 即可任意改审核状态（越权风险）；现收口到
    _apply_booking_audit 统一权限校验 + 状态联动 + 审计 + 短信幂等。
    """
    payload = require_auth(authorization)
    acct = await _current_user_async(payload)
    if not acct:
        raise HTTPException(401, "登录状态异常，请重新登录")
    b = await req.json()
    bk = await asyncio.to_thread(db.fetch_by_id, "bookings", bid)
    if not bk:
        raise HTTPException(404, "预约不存在")
    ev = await asyncio.to_thread(db.fetch_by_id, "events", bk.get("event_id", ""))
    if not ev:
        raise HTTPException(404, "活动不存在")
    if not await _can_review_booking(payload, acct, ev):
        raise HTTPException(403, "仅发布者/发布方/活动管理员/运营方可操作该活动报名")
    patch = {}
    if b.get("status") in ("pending", "confirmed", "arrived", "converted", "rejected", "cancelled"):
        patch["status"] = b["status"]
    if b.get("name") is not None:
        patch["name"] = str(b.get("name") or "").strip()[:100]
    if b.get("contact") is not None:
        patch["contact"] = str(b.get("contact") or "").strip()[:30]
    if patch:
        await asyncio.to_thread(db.update_row, "bookings", bid, patch)
    # 审核动作：auditStatus 变化 → 统一审核入口（状态联动 + 审计 + 短信）
    if b.get("auditStatus") in ("approved", "rejected") and b.get("auditStatus") != (bk.get("audit_status") or "pending"):
        await _apply_booking_audit(bid, b["auditStatus"], b.get("auditComment") or "", payload, acct)
    return {"ok": True, "entry": _decode_bk(await asyncio.to_thread(db.fetch_by_id, "bookings", bid))}


@app.post("/api/bookings/{bid}/review")
async def review_booking(bid: str, req: Request, authorization: str = Header(default="")):
    """报名审核统一入口（小程序/管理端共用）：approved/rejected + 原因，触发审核短信。

    权限：发布者 / 活动管理员（手机号绑定）/ 平台运营方 / 本园区管理员。
    """
    payload = require_auth(authorization)
    acct = await _current_user_async(payload)
    if not acct:
        raise HTTPException(401, "登录状态异常，请重新登录")
    b = await req.json()
    bk = await _apply_booking_audit(bid, b.get("status"), b.get("comment") or "", payload, acct)
    return {"ok": True, "entry": _decode_bk(bk)}


@app.get("/api/events/{eid}/bookings")
async def event_bookings(eid: str, authorization: str = Header(default=""),
                         audit: Optional[str] = None, status: Optional[str] = None):
    """活动报名列表（管理视角）：发布者/活动管理员/运营方/园区管理员。

    返回自定义表单值（formData）与审核记录（auditBy/auditAt/auditComment），
    附活动表单定义（formFields）供前端按定义渲染。
    """
    payload = require_auth(authorization)
    acct = await _current_user_async(payload)
    if not acct:
        raise HTTPException(401, "登录状态异常，请重新登录")
    e = await asyncio.to_thread(db.fetch_by_id, "events", eid)
    if not e:
        raise HTTPException(404, "活动不存在")
    if not await _can_review_booking(payload, acct, e):
        raise HTTPException(403, "仅发布者/发布方/活动管理员/运营方可查看该活动报名")
    conn = await asyncio.to_thread(db.get_conn, )
    try:
        rows = conn.execute("SELECT * FROM bookings WHERE event_id=?", (eid,)).fetchall()
    finally:
        conn.close()
    lst = [_decode_bk(r) for r in rows]
    lst.sort(key=lambda x: x["created_at"] or "", reverse=True)
    if audit:
        lst = [x for x in lst if x.get("audit_status") == audit]
    if status:
        lst = [x for x in lst if x["status"] == status]
    form_fields = _load_json(e.get("form_fields_json") or "", [])
    if not isinstance(form_fields, list):
        form_fields = []
    return {"ok": True, "list": lst, "formFields": form_fields,
            "event": {"id": e["id"], "title": e.get("title", ""), "signupDeadline": e.get("signup_deadline") or ""}}


@app.post("/api/events/reminders/scan")
async def event_reminders_scan(request: Request, x_internal_token: str = Header(default="")):
    """活动前提醒短信扫描（定时任务入口，幂等）。

    调用方须携带 X-Internal-Token（ENV EVENT_REMINDER_TOKEN，默认开发值），
    由外部 cron 或部署脚本周期性调用；窗口内已提醒的报名不会重复发送。
    """
    if x_internal_token != EVENT_REMINDER_TOKEN:
        raise HTTPException(403, "内部令牌校验失败")
    stats = await scan_event_reminders()
    return {"ok": True, **stats}


@app.delete("/api/bookings/{bid}")
async def delete_booking(bid: str, authorization: str = Header(default="")):
    require_auth(authorization)
    await asyncio.to_thread(db.delete_row, "bookings", bid)
    return {"ok": True}


# ================= OPC 测评 =================
@app.get("/api/tests/opc/questions")
async def opc_questions(version: Optional[str] = "full"):
    v = "quick" if version == "quick" else "full"
    return {
        "ok": True, "version": v,
        "questions": [{"id": q["id"], "part": q["part"], "question": q["question"], "A": q["A"], "B": q["B"]}
                      for q in opc_engine.current_questions(v)],
        "sections": SECTIONS, "axisNames": AXIS_NAMES, "dimLabels": ADAPT_LABELS,
        "disclaimer": PROFILES["disclaimer"], "quickNote": PROFILES["quickNote"],
    }


@app.post("/api/tests/opc/calculate")
async def opc_calculate(req: Request):
    b = await req.json()
    version = "quick" if b.get("version") == "quick" else "full"
    answers = read_answers(b)
    try:
        result = opc_engine.expand_result(opc_engine.calculate(answers, version))
        return {"ok": True, "result": result}
    except Exception:
        raise HTTPException(400, "测评计算失败，请稍后重试")


@app.post("/api/tests")
async def report_test(req: Request):
    b = await req.json()
    if not b.get("typeCode"):
        raise HTTPException(400, "缺少测评结果")
    # 逐题答案（键=题目id，值=A/B），原样存储为 JSON，供 admin 详情页回溯
    answers_raw = b.get("answers")
    answers_json = json.dumps(answers_raw, ensure_ascii=False) if answers_raw is not None else None
    entry = {"id": db.gen_id("T"), "created_at": now_iso(), "username": str(b.get("username", "")).strip(),
             "type_code": b.get("typeCode"), "persona": b.get("persona", ""),
             "adapt_index": int(b.get("adaptIndex") or 0), "adapt_level": b.get("adaptLevel", ""),
             "tracks": json.dumps(b.get("tracks", []), ensure_ascii=False), "version": b.get("version", ""),
             "answers": answers_json}
    await asyncio.to_thread(db.insert, "tests", entry)
    return {"ok": True, "id": entry["id"]}


@app.get("/api/tests")
async def admin_tests(authorization: str = Header(default="")):
    require_auth(authorization)
    return {"ok": True, "list": await asyncio.to_thread(db.list_all, "tests")}


# ================= 政策 / 流程 / 调研：题目下发 + 结果生成（逻辑后置于后端） =================
@app.get("/api/policy/questions")
async def policy_questions():
    return {"ok": True, "questions": policy_data.PT_QUESTIONS}


@app.post("/api/policy/calculate")
async def policy_calculate(req: Request):
    b = await req.json()
    ans = b.get("answers") if isinstance(b.get("answers"), dict) else {}
    result = policy_engine.compute_policy(ans)
    return {"ok": True, "result": result}


@app.get("/api/plan/config")
async def plan_config():
    return {"ok": True, "regions": policy_data.SP_REGIONS, "status": policy_data.SP_STATUS}


@app.post("/api/plan/generate")
async def plan_generate(req: Request):
    b = await req.json()
    plan = policy_engine.build_plan({
        "needPark": bool(b.get("needPark")),
        "needRegister": bool(b.get("needRegister")),
        "hasStaff": bool(b.get("hasStaff")),
    })
    return {"ok": True, "plan": plan}


@app.get("/api/survey/questions")
async def survey_questions():
    return {"ok": True, "likert": survey_data.LIKERT_OPTIONS,
            "sections": survey_data.SURVEY_SECTIONS, "questions": survey_data.SURVEY_QUESTIONS}


@app.post("/api/survey/submit")
async def survey_submit(req: Request):
    b = await req.json()
    if not isinstance(b.get("answers"), dict):
        raise HTTPException(400, "缺少调研答案")
    entry = {"id": db.gen_id("S"), "created_at": now_iso(), "username": str(b.get("username", "")).strip(),
             "source": str(b.get("source", "")).strip() or "web", "answers": json.dumps(b.get("answers"), ensure_ascii=False)}
    await asyncio.to_thread(db.insert, "survey_logs", entry)
    return {"ok": True, "id": entry["id"], "createdAt": entry["created_at"]}


# ================= 日志上报（政策 / 流程 / 调研） =================
@app.post("/api/policy-logs")
async def report_policy(req: Request):
    b = await req.json()
    if not b.get("answers"):
        raise HTTPException(400, "缺少测评结果")
    entry = {"id": db.gen_id("P"), "created_at": now_iso(), "username": str(b.get("username", "")).strip(),
             "answers": json.dumps(b.get("answers"), ensure_ascii=False),
             "policies_count": int(b.get("policiesCount") or 0), "subsidies_count": int(b.get("subsidiesCount") or 0),
             "loans_count": int(b.get("loansCount") or 0), "summary": str(b.get("summary", "") or "")[:200]}
    await asyncio.to_thread(db.insert, "policy_logs", entry)
    return {"ok": True, "id": entry["id"]}


@app.get("/api/policy-logs")
async def admin_policy(authorization: str = Header(default="")):
    require_auth(authorization)
    return {"ok": True, "list": await asyncio.to_thread(db.list_all, "policy_logs")}


@app.post("/api/plan-logs")
async def report_plan(req: Request):
    b = await req.json()
    if not b.get("region"):
        raise HTTPException(400, "缺少流程信息")
    entry = {"id": db.gen_id("L"), "created_at": now_iso(), "username": str(b.get("username", "")).strip(),
             "region": str(b.get("region", "")).strip(), "status": str(b.get("status", "")).strip(),
             "need_park": 1 if b.get("needPark") else 0, "has_staff": 1 if b.get("hasStaff") else 0,
             "steps_count": int(b.get("stepsCount") or 0)}
    await asyncio.to_thread(db.insert, "plan_logs", entry)
    return {"ok": True, "id": entry["id"]}


@app.get("/api/plan-logs")
async def admin_plan(authorization: str = Header(default="")):
    require_auth(authorization)
    return {"ok": True, "list": await asyncio.to_thread(db.list_all, "plan_logs")}


@app.post("/api/survey-logs")
async def report_survey(req: Request):
    b = await req.json()
    if not isinstance(b.get("answers"), dict):
        raise HTTPException(400, "缺少调研答案")
    entry = {"id": db.gen_id("S"), "created_at": now_iso(), "username": str(b.get("username", "")).strip(),
             "source": str(b.get("source", "")).strip() or "web", "answers": json.dumps(b.get("answers"), ensure_ascii=False)}
    await asyncio.to_thread(db.insert, "survey_logs", entry)
    return {"ok": True, "id": entry["id"], "createdAt": entry["created_at"]}


@app.get("/api/survey-logs")
async def admin_survey(authorization: str = Header(default="")):
    require_auth(authorization)
    return {"ok": True, "list": await asyncio.to_thread(db.list_all, "survey_logs")}


# ================= 运营统计 =================
@app.get("/api/ops/stats")
async def ops_stats(authorization: str = Header(default="")):
    require_auth(authorization)
    conn = await asyncio.to_thread(db.get_conn, )
    def count(t):
        return conn.execute(f"SELECT COUNT(*) c FROM {t}").fetchone()["c"]
    def count_where(t, where, val):
        return conn.execute(f"SELECT COUNT(*) c FROM {t} WHERE {where}", (val,)).fetchone()["c"]
    now_ms = int(__import__("time").time() * 1000)
    upcoming = sum(1 for e in await asyncio.to_thread(db.list_all, "events") if _parse_ms(e["start_at"]) >= now_ms)
    free = sum(1 for e in await asyncio.to_thread(db.list_all, "events") if e["type"] == "free" and _parse_ms(e["start_at"]) >= now_ms)
    salon = sum(1 for e in await asyncio.to_thread(db.list_all, "events") if e["type"] == "salon" and _parse_ms(e["start_at"]) >= now_ms)
    # 先全部算完再关连接
    stats = {
        "bookings": count("bookings"),
        "pending": await asyncio.to_thread(count_where, "bookings", "status=?", "pending"),
        "confirmed": await asyncio.to_thread(count_where, "bookings", "status=?", "confirmed"),
        "arrived": await asyncio.to_thread(count_where, "bookings", "status=?", "arrived"),
        "converted": await asyncio.to_thread(count_where, "bookings", "status=?", "converted"),
        "rejected": await asyncio.to_thread(count_where, "bookings", "status=?", "rejected"),
        "events": count("events"), "upcomingEvents": upcoming, "free": free, "salon": salon,
        "tests": count("tests"), "policyLogs": count("policy_logs"), "planLogs": count("plan_logs"),
        "surveyLogs": count("survey_logs"),
    }
    conn.close()
    return {"ok": True, "stats": stats}


def _parse_ms(iso):
    from datetime import datetime
    try:
        return datetime.fromisoformat(iso).timestamp() * 1000
    except Exception:
        return 0


# ================= 图像上传 =================
@app.post("/api/upload")
async def upload(request: Request, file: UploadFile = File(...), dir: str = Form("misc"), authorization: str = Header(default="")):
    # 管理端上传需登录（pine 后台）；公开端也可保留（依据调用方），这里允许管理端 token
    require_auth(authorization)
    # OSS 优先（已配置 OSS_* 时走桶），未配置则 save_media 内部落本地 uploads；统一返回 CDN/OSS 直链
    # dir 为 OSS 业务目录（avatar / park-admission / news / event / misc），按《OSS 路径规范》分目录
    from ..services.media_upload import save_media
    url = await save_media(file, dir=dir)
    return {"ok": True, "url": url}


# ================= 园区入驻申请 =================

def _park_tenants_public():
    """平台可见园区列表（C端选园区用）：id/name/intro。"""
    out = []
    for t in db.list_all("park_tenants"):
        try:
            intro = json.loads(t.get("intro_json") or "[]")
        except Exception:
            intro = []
        if not isinstance(intro, list):
            intro = []
        out.append({"id": t.get("id"), "name": t.get("name"), "intro": intro})
    return out


def _admission_view(row: dict) -> dict:
    v = dict(row)
    try:
        v["form"] = json.loads(v.get("form_json") or "{}")
    except Exception:
        v["form"] = {}
    try:
        v["docs"] = json.loads(v.get("docs_json") or "{}")
    except Exception:
        v["docs"] = {}
    v.pop("form_json", None)
    v.pop("docs_json", None)
    return v


@app.get("/api/park/tenants")
async def park_tenants_public():
    """选择园区：平台可见园区列表（公开，C端选园区）。"""
    return {"ok": True, "tenants": _park_tenants_public()}


@app.post("/api/park-admission")
async def park_admission_submit(request: Request, authorization: str = Header(default="")):
    """提交园区入驻申请（需登录）。form=申请人/企业/项目/需求，docs=上传资料URL集合。"""
    auth = require_auth(authorization)
    body = await request.json()
    tenant_id = str(body.get("tenant_id") or "").strip()
    if not tenant_id:
        raise HTTPException(400, "请选择意向园区")
    tenant = db.fetch_one("park_tenants", id=tenant_id)
    if not tenant:
        raise HTTPException(404, "园区不存在")
    form = body.get("form") if isinstance(body.get("form"), dict) else {}
    docs = body.get("docs") if isinstance(body.get("docs"), dict) else {}
    if docs:
        from ..infrastructure.oss import to_object_path
        docs = {k: to_object_path(v) for k, v in docs.items()}
    u = await _current_user_async(auth)
    now = now_iso()
    rec = {
        "id": f"adm_{secrets.token_hex(8)}",
        "tenant_id": tenant_id,
        "tenant_name": tenant.get("name", ""),
        "user_id": str(auth.get("user_id") or (u.get("id") if u else "") or ""),
        "username": auth.get("username", ""),
        "applicant_name": str(form.get("name") or (u.get("nickname") if u else "") or ""),
        "contact_phone": str(form.get("contact_phone") or (u.get("phone") if u else "") or ""),
        "status": "pending",
        "form_json": json.dumps(form, ensure_ascii=False),
        "docs_json": json.dumps(docs, ensure_ascii=False),
        "source": (request.headers.get("X-Client", "") or "miniprogram"),
        "review_comment": "", "reviewed_by": "", "reviewed_at": "",
        "created_at": now, "updated_at": now,
    }
    db.insert("park_admissions", rec)
    return {"ok": True, "id": rec["id"], "status": "pending"}


@app.get("/api/park-admission/mine")
async def park_admission_mine(authorization: str = Header(default="")):
    """我的入驻申请列表。"""
    auth = require_auth(authorization)
    mine = []
    for i in db.list_all("park_admissions"):
        if str(i.get("user_id")) == str(auth.get("user_id")) or str(i.get("username")) == str(auth.get("username")):
            mine.append(_admission_view(i))
    mine.sort(key=lambda x: x.get("created_at", ""), reverse=True)
    return {"ok": True, "items": mine}


@app.get("/api/park-admission")
async def park_admission_list(authorization: str = Header(default="")):
    """运营方：全部入驻申请（审核列表）。"""
    auth = require_auth(authorization)
    u = await _current_user_async(auth)
    if not u or u.get("role") != "operator":
        raise HTTPException(403, "仅运营方可查看")
    items = [_admission_view(i) for i in db.list_all("park_admissions")]
    items.sort(key=lambda x: x.get("created_at", ""), reverse=True)
    return {"ok": True, "items": items}


@app.post("/api/park-admission/{aid}/review")
async def park_admission_review(aid: str, request: Request, authorization: str = Header(default="")):
    """运营方：审核入驻申请（approved/rejected/reviewing + 意见）。"""
    auth = require_auth(authorization)
    u = await _current_user_async(auth)
    if not u or u.get("role") != "operator":
        raise HTTPException(403, "仅运营方可审核")
    target = db.fetch_by_id("park_admissions", aid)
    if not target:
        raise HTTPException(404, "申请不存在")
    body = await request.json()
    status = str(body.get("status") or "")
    if status not in ("approved", "rejected", "reviewing"):
        raise HTTPException(400, "status 需为 approved/rejected/reviewing")
    db.update_row("park_admissions", aid, {
        "status": status,
        "review_comment": str(body.get("comment", "")),
        "reviewed_by": auth.get("username", ""),
        "reviewed_at": now_iso(),
        "updated_at": now_iso(),
    })
    return {"ok": True, "id": aid, "status": status}