"""admin CPS 数据访问:群/活动 CRUD + 美团订单拉单入库(对账) + 按群统计聚合。

美团 query_order 返回的金额是「元」字符串、时间是秒级时间戳,这里统一转「分」+ tz-aware。
统计在 Python 侧聚合(订单量级小、admin 低频),逻辑清晰、跨 PG/SQLite 无 SQL 方言坑。
"""
from __future__ import annotations

from datetime import datetime, timedelta, timezone
from decimal import Decimal, InvalidOperation
from uuid import uuid4

from sqlalchemy import desc, func, select
from sqlalchemy.orm import Session

from app.admin.repositories.queries import _as_utc, offset_paginate
from app.integrations import meituan
from app.repositories import cps_link as cps_link_repo
from app.models.cps_activity import CpsActivity
from app.models.cps_group import CpsGroup
from app.models.cps_link import CpsClick, CpsLink
from app.models.cps_order import CpsOrder
from app.models.cps_wx_user import CpsWxUser

# 美团订单状态:取消(4)/风控(5)不计佣金;结算(6)为佣金真正到账
_INVALID_STATUS = {"4", "5"}
_SETTLED_STATUS = "6"

# CPS 点击时序按北京时区分桶(运营看的是北京时间)
_BJ_TZ = timezone(timedelta(hours=8))


# ───────────── 单位换算 ─────────────
def _yuan_to_cents(v: object) -> int | None:
    """「元」字符串/数 → 分。None / 空 / 字面 "null" → None。"""
    if v is None:
        return None
    s = str(v).strip()
    if not s or s.lower() == "null":
        return None
    try:
        return int((Decimal(s) * 100).to_integral_value())
    except (InvalidOperation, ValueError):
        return None


def _ts_to_dt(ts: object) -> datetime | None:
    """秒级时间戳 → tz-aware UTC datetime(绝对时刻,前端按北京展示)。"""
    if not ts:
        return None
    try:
        return datetime.fromtimestamp(int(ts), tz=timezone.utc)
    except (ValueError, OSError, TypeError):
        return None


# ───────────── 群 ─────────────
def get_group(db: Session, group_id: int) -> CpsGroup | None:
    return db.get(CpsGroup, group_id)


def get_group_by_sid(db: Session, sid: str) -> CpsGroup | None:
    return db.execute(select(CpsGroup).where(CpsGroup.sid == sid)).scalar_one_or_none()


def list_groups(
    db: Session, *, keyword: str | None = None, status: str | None = None,
    limit: int = 20, cursor: int | None = None,
) -> tuple[list[CpsGroup], int | None, int]:
    stmt = select(CpsGroup)
    if keyword and keyword.strip():
        kw = f"%{keyword.strip()}%"
        stmt = stmt.where(CpsGroup.name.ilike(kw) | CpsGroup.sid.ilike(kw))
    if status:
        stmt = stmt.where(CpsGroup.status == status)
    return offset_paginate(db, stmt, (desc(CpsGroup.id),), limit=limit, cursor=cursor)


def create_group(
    db: Session, *, name: str, platforms: list[str], sid: str | None = None,
    member_count: int | None = None, remark: str | None = None, commit: bool = True,
) -> CpsGroup:
    """建群。含 meituan 才有 sid(留空自动 g<id>);纯淘宝/京东 sid=None(那俩不支持 sid)。"""
    has_meituan = "meituan" in (platforms or [])
    # 含美团:用填的 sid,或临时占位待回填 g<id>;不含美团:sid 恒 None
    seed_sid = (sid or f"tmp{uuid4().hex[:20]}") if has_meituan else None
    group = CpsGroup(
        name=name,
        platforms=list(platforms or []),
        sid=seed_sid,
        member_count=member_count,
        remark=remark,
    )
    db.add(group)
    db.flush()
    if has_meituan and not sid:
        group.sid = f"g{group.id}"
        db.flush()
    if commit:
        db.commit()
        db.refresh(group)
    return group


def update_group(
    db: Session, group: CpsGroup, *, name: str | None = None,
    platforms: list[str] | None = None,
    member_count: int | None = None, status: str | None = None,
    remark: str | None = None, commit: bool = True,
) -> CpsGroup:
    if name is not None:
        group.name = name
    if platforms is not None:
        group.platforms = list(platforms)
    if member_count is not None:
        group.member_count = member_count
    if status is not None:
        group.status = status
    if remark is not None:
        group.remark = remark
    if commit:
        db.commit()
        db.refresh(group)
    else:
        db.flush()
    return group


def delete_group(db: Session, group: CpsGroup, *, commit: bool = True) -> None:
    """硬删群。不级联删已生成的 cps_link(已发链接继续可用);该群历史点击/订单按 sid 仍在库。"""
    db.delete(group)
    if commit:
        db.commit()
    else:
        db.flush()


# ───────────── 活动 ─────────────
def get_activity(db: Session, activity_id: int) -> CpsActivity | None:
    return db.get(CpsActivity, activity_id)


def list_activities(
    db: Session, *, platform: str | None = None, status: str | None = None,
    limit: int = 20, cursor: int | None = None,
) -> tuple[list[CpsActivity], int | None, int]:
    stmt = select(CpsActivity)
    if platform:
        stmt = stmt.where(CpsActivity.platform == platform)
    if status:
        stmt = stmt.where(CpsActivity.status == status)
    return offset_paginate(db, stmt, (desc(CpsActivity.id),), limit=limit, cursor=cursor)


def list_activity_images(db: Session) -> list[str]:
    """已用过的活动落地页图(distinct 非空 image_url),供新建活动复用选择。"""
    rows = (
        db.execute(
            select(CpsActivity.image_url)
            .where(CpsActivity.image_url.is_not(None))
            .distinct()
        )
        .scalars()
        .all()
    )
    return [r for r in rows if r]


def create_activity(
    db: Session, *, name: str, platform: str = "meituan", act_id: str | None = None,
    product_view_sign: str | None = None, payload: str | None = None,
    image_url: str | None = None, remark: str | None = None, commit: bool = True,
) -> CpsActivity:
    activity = CpsActivity(
        name=name, platform=platform, act_id=act_id,
        product_view_sign=product_view_sign, payload=payload,
        image_url=image_url, remark=remark,
    )
    db.add(activity)
    if commit:
        db.commit()
        db.refresh(activity)
    else:
        db.flush()
    return activity


def update_activity(
    db: Session, activity: CpsActivity, *, name: str | None = None,
    platform: str | None = None, act_id: str | None = None,
    product_view_sign: str | None = None, payload: str | None = None,
    image_url: str | None = None, remark: str | None = None,
    status: str | None = None, commit: bool = True,
) -> CpsActivity:
    if name is not None:
        activity.name = name
    if platform is not None:
        activity.platform = platform
    if act_id is not None:
        activity.act_id = act_id
    if product_view_sign is not None:
        activity.product_view_sign = product_view_sign
    if payload is not None:
        activity.payload = payload
    if image_url is not None:
        activity.image_url = image_url
    if remark is not None:
        activity.remark = remark
    if status is not None:
        activity.status = status
    if commit:
        db.commit()
        db.refresh(activity)
    else:
        db.flush()
    return activity


def delete_activity(db: Session, activity: CpsActivity, *, commit: bool = True) -> None:
    """硬删活动。不级联删 cps_link(已发链接继续可用;淘宝落地页缺活动时取图走兜底默认图)。"""
    db.delete(activity)
    if commit:
        db.commit()
    else:
        db.flush()


# ───────────── 订单(对账) ─────────────
def _map_order_fields(r: dict) -> dict:
    """美团 query_order 单条 dataList → CpsOrder 字段(金额转分、时间转 datetime)。"""
    pn = r.get("productName")
    if pn and len(pn) > 500:
        pn = pn[:500]
    return {
        "sid": (r.get("sid") or None),
        "act_id": str(r["actId"]) if r.get("actId") is not None else None,
        "biz_line": r.get("businessLine"),
        "trade_type": r.get("tradeType"),
        "pay_price_cents": _yuan_to_cents(r.get("payPrice")),
        "commission_cents": _yuan_to_cents(r.get("profit")),
        "commission_rate": str(r["commissionRate"]) if r.get("commissionRate") is not None else None,
        "refund_price_cents": _yuan_to_cents(r.get("refundPrice")),
        "refund_profit_cents": _yuan_to_cents(r.get("refundProfit")),
        "mt_status": str(r["status"]) if r.get("status") is not None else None,
        "invalid_reason": (r.get("invalidReason") or None),
        "product_name": pn or None,
        "pay_time": _ts_to_dt(r.get("payTime")),
        "mt_update_time": _ts_to_dt(r.get("updateTime")),
        "raw": r,
    }


def reconcile_orders(
    db: Session, *, start_time: int, end_time: int,
    query_time_type: int = 1, sid: str | None = None, max_pages: int = 200,
) -> dict:
    """调美团 query_order 分页拉单 → 按 order_id upsert。返回 {fetched, inserted, updated, pages}。

    订单状态会随时间变(付款→完成→结算/退款),重复拉同一单则更新。max_pages 防异常死循环。
    """
    fetched = inserted = updated = pages = 0
    page = 1
    while page <= max_pages:
        resp = meituan.query_order(
            sid=sid, start_time=start_time, end_time=end_time,
            query_time_type=query_time_type, page=page, limit=100,
        )
        rows = ((resp.get("data") or {}).get("dataList")) or []
        if not rows:
            break
        pages += 1
        for r in rows:
            order_id = str(r.get("orderId") or "").strip()
            if not order_id:
                continue
            fetched += 1
            fields = _map_order_fields(r)
            existing = db.execute(
                select(CpsOrder).where(CpsOrder.order_id == order_id)
            ).scalar_one_or_none()
            if existing is None:
                db.add(CpsOrder(order_id=order_id, **fields))
                inserted += 1
            else:
                for k, v in fields.items():
                    setattr(existing, k, v)
                updated += 1
        if len(rows) < 100:
            break
        page += 1
    db.commit()
    return {"fetched": fetched, "inserted": inserted, "updated": updated, "pages": pages}


def list_orders(
    db: Session, *, sid: str | None = None, mt_status: str | None = None,
    limit: int = 20, cursor: int | None = None,
) -> tuple[list[CpsOrder], int | None, int]:
    stmt = select(CpsOrder)
    if sid:
        stmt = stmt.where(CpsOrder.sid == sid)
    if mt_status:
        stmt = stmt.where(CpsOrder.mt_status == mt_status)
    return offset_paginate(db, stmt, (desc(CpsOrder.id),), limit=limit, cursor=cursor)


# ───────────── 统计(按群聚合) ─────────────
def group_stats(
    db: Session, *, date_from: datetime | None = None, date_to: datetime | None = None,
) -> list[dict]:
    """按 sid 聚合订单 + join 群信息。已建但本期无单的活跃群也列出(全 0)。
    未归群的 sid(历史/其它来源)单独成行 group_id=None。按预估佣金降序。
    """
    ostmt = select(CpsOrder)
    if date_from is not None:
        ostmt = ostmt.where(CpsOrder.pay_time >= _as_utc(date_from))
    if date_to is not None:
        ostmt = ostmt.where(CpsOrder.pay_time <= _as_utc(date_to))
    orders_by_sid: dict[str | None, list[CpsOrder]] = {}
    for o in db.execute(ostmt).scalars().all():
        orders_by_sid.setdefault(o.sid, []).append(o)

    clicks = cps_link_repo.click_stats_by_group(db, date_from=date_from, date_to=date_to)
    groups = list(db.execute(select(CpsGroup)).scalars().all())

    def _order_agg(items: list[CpsOrder]) -> dict:
        valid = [o for o in items if o.mt_status not in _INVALID_STATUS]
        settled = [o for o in items if o.mt_status == _SETTLED_STATUS]
        canceled = [o for o in items if o.mt_status in _INVALID_STATUS]
        return {
            "order_count": len(valid),
            "settled_count": len(settled),
            "canceled_count": len(canceled),
            "gmv_cents": sum(o.pay_price_cents or 0 for o in valid),
            "est_commission_cents": sum(o.commission_cents or 0 for o in valid),
            "settled_commission_cents": sum(o.commission_cents or 0 for o in settled),
        }

    # 淘宝/京东无法对账 → 对账字段全 None(前端显示 "-")
    _no_recon = {
        "order_count": None, "settled_count": None, "canceled_count": None,
        "gmv_cents": None, "est_commission_cents": None, "settled_commission_cents": None,
    }

    rows: list[dict] = []
    seen_sids: set[str] = set()
    for g in groups:
        if g.status != "active":
            continue
        c = clicks.get(g.id) or {}
        row = {
            "group_id": g.id, "sid": g.sid, "name": g.name,
            "platforms": list(g.platforms or []), "member_count": g.member_count,
            "click_pv": c.get("pv", 0), "click_uv": c.get("uv", 0),
            "copy_pv": c.get("copy_pv", 0), "copy_uv": c.get("copy_uv", 0),
        }
        # 对账只对美团群(有 sid);淘宝/京东 → None
        if "meituan" in (g.platforms or []) and g.sid:
            row.update(_order_agg(orders_by_sid.get(g.sid, [])))
            seen_sids.add(g.sid)
        else:
            row.update(_no_recon)
        rows.append(row)

    # 未归群的历史美团 sid(如 wonderableai):单列,有对账无点击
    for sid, items in orders_by_sid.items():
        if sid is None or sid in seen_sids:
            continue
        row = {
            "group_id": None, "sid": sid, "name": sid, "platforms": ["meituan"],
            "member_count": None, "click_pv": 0, "click_uv": 0, "copy_pv": 0, "copy_uv": 0,
        }
        row.update(_order_agg(items))
        rows.append(row)

    # 佣金降序(None 当 0),次按点击
    rows.sort(key=lambda x: ((x["est_commission_cents"] or 0), x["click_pv"]), reverse=True)
    return rows


def group_click_timeseries(
    db: Session, *, group_id: int, granularity: str,
    start: datetime, end: datetime,
) -> list[dict]:
    """该群点击时序(北京时区分桶,granularity=day|hour)。

    每桶:click_pv/uv(visit)+copy_pv/uv;UV=distinct(ip,ua)。**生成连续桶(无数据补 0)**,
    让折线图 x 轴按所选范围铺满、不断裂(CPS 数据稀疏,补零是关键)。
    """
    rows = db.execute(
        select(CpsClick.clicked_at, CpsClick.event_type, CpsClick.ip, CpsClick.ua)
        .where(CpsClick.group_id == group_id)
        .where(CpsClick.clicked_at >= _as_utc(start))
        .where(CpsClick.clicked_at <= _as_utc(end))
    ).all()

    agg: dict = {}  # bucket_key -> {vp, vu(set), cp, cu(set)}
    for clicked_at, event_type, ip, ua in rows:
        # naive(本地 SQLite 读回无 tz)按 UTC 兜底,再转北京——与 queries._as_utc 约定一致,
        # 否则 astimezone 会把 UTC 瞬刻当系统本地时区解释,分桶整体偏移。
        ca = clicked_at if clicked_at.tzinfo else clicked_at.replace(tzinfo=timezone.utc)
        bj = ca.astimezone(_BJ_TZ)
        key = bj.strftime("%Y-%m-%d") if granularity == "day" else bj.hour
        b = agg.setdefault(key, {"vp": 0, "vu": set(), "cp": 0, "cu": set()})
        uid = f"{ip or ''}|{ua or ''}"
        if event_type == "copy":
            b["cp"] += 1
            b["cu"].add(uid)
        else:
            b["vp"] += 1
            b["vu"].add(uid)

    def _point(label: str, b: dict | None) -> dict:
        return {
            "label": label,
            "click_pv": b["vp"] if b else 0,
            "click_uv": len(b["vu"]) if b else 0,
            "copy_pv": b["cp"] if b else 0,
            "copy_uv": len(b["cu"]) if b else 0,
        }

    points: list[dict] = []
    if granularity == "day":
        cur = start.astimezone(_BJ_TZ).date()
        last = end.astimezone(_BJ_TZ).date()
        while cur <= last:
            points.append(_point(cur.strftime("%m-%d"), agg.get(cur.strftime("%Y-%m-%d"))))
            cur += timedelta(days=1)
    else:
        for h in range(24):
            points.append(_point(f"{h:02d}:00", agg.get(h)))
    return points


def group_order_daily(
    db: Session, *, sid: str, start: datetime, end: datetime,
) -> dict[str, dict]:
    """美团群订单按天(北京时区,key=YYYY-MM-DD)聚合。口径与 group_stats._order_agg 一致:
    取消(4)/风控(5)不计有效;结算只数 status=6。退款佣金取当天全部单的 refund_profit_cents 之和。
    返回 {date: {order_count, canceled_count, gmv_cents, est_commission_cents,
    refund_profit_cents, settled_commission_cents}};无单的天不在 dict 里(端点侧补零)。
    """
    rows = (
        db.execute(
            select(CpsOrder)
            .where(CpsOrder.sid == sid)
            .where(CpsOrder.pay_time >= _as_utc(start))
            .where(CpsOrder.pay_time <= _as_utc(end))
        )
        .scalars()
        .all()
    )
    by_day: dict[str, list] = {}
    for o in rows:
        if not o.pay_time:
            continue
        pt = o.pay_time if o.pay_time.tzinfo else o.pay_time.replace(tzinfo=timezone.utc)
        d = pt.astimezone(_BJ_TZ).strftime("%Y-%m-%d")
        by_day.setdefault(d, []).append(o)

    out: dict[str, dict] = {}
    for d, items in by_day.items():
        valid = [o for o in items if o.mt_status not in _INVALID_STATUS]
        settled = [o for o in items if o.mt_status == _SETTLED_STATUS]
        canceled = [o for o in items if o.mt_status in _INVALID_STATUS]
        out[d] = {
            "order_count": len(valid),
            "canceled_count": len(canceled),
            "gmv_cents": sum(o.pay_price_cents or 0 for o in valid),
            "est_commission_cents": sum(o.commission_cents or 0 for o in valid),
            "refund_profit_cents": sum(o.refund_profit_cents or 0 for o in items),
            "settled_commission_cents": sum(o.commission_cents or 0 for o in settled),
        }
    return out


def group_wx_users(db: Session, *, group_id: int, limit: int = 200) -> list[dict]:
    """该群来过的微信用户:在本群有带 openid 点击的所有用户 + 本群 visit/copy 次数 + 画像。

    用「本群点击」判定归属,而非 cps_wx_user.first_group_id —— 同一用户会在多个群活动,
    first_group_id 只记首次授权群,会漏掉"先在别处授权、cookie 已存、又来本群"的用户。
    copy=本群点「复制口令」次数(领券意向),visit=进落地页次数。按本群首次点击倒序。
    昵称/头像来自授权画像;美团/京东群只 302 跳转、无 userinfo 触发点,故多为空。
    """
    rows = db.execute(
        select(CpsClick.openid, CpsClick.event_type, func.count())
        .where(CpsClick.group_id == group_id)
        .where(CpsClick.openid.is_not(None))
        .group_by(CpsClick.openid, CpsClick.event_type)
    ).all()
    if not rows:
        return []
    stat: dict[str, dict] = {}
    for openid, event_type, cnt in rows:
        s = stat.setdefault(openid, {"visit": 0, "copy": 0})
        if event_type == "copy":
            s["copy"] = cnt
        else:
            s["visit"] = cnt
    openids = list(stat.keys())
    # 本群每个 openid 的首次点击时间(排序 + 展示"首次进入")
    first_seen = dict(
        db.execute(
            select(CpsClick.openid, func.min(CpsClick.clicked_at))
            .where(CpsClick.group_id == group_id)
            .where(CpsClick.openid.is_not(None))
            .group_by(CpsClick.openid)
        ).all()
    )
    # 关联画像(未授权 userinfo 的昵称/头像为空)
    users = {
        u.openid: u
        for u in db.execute(select(CpsWxUser).where(CpsWxUser.openid.in_(openids))).scalars().all()
    }
    result = [
        {
            "openid": openid,
            "nickname": users[openid].nickname if openid in users else None,
            "headimgurl": users[openid].headimgurl if openid in users else None,
            "first_seen": first_seen.get(openid),
            "visit_count": stat[openid]["visit"],
            "copy_count": stat[openid]["copy"],
        }
        for openid in openids
    ]
    result.sort(key=lambda x: x["first_seen"], reverse=True)
    return result[:limit]


def group_day_users(
    db: Session, *, group_id: int, start: datetime, end: datetime, limit: int = 200,
) -> list[dict]:
    """该群某天(北京)以用户为单位的领券/点击 + 每人 visit 过的券。

    时间窗为半开区间 [start, end)(end=次日 00:00),避免午夜双计。只统计 openid 非空
    (可归属到人)的点击 —— 匿名点击(美团/京东 302 多为匿名)不计入。券名 = 该点击 link
    对应活动名;活动被硬删则兜底 活动#{id}。copy=领券次数、visit=点击次数;coupons 仅
    取 visit 事件按活动分组、按次数倒序(合计 = visit_count)。排序:领券 desc、再点击 desc。
    与 group_wx_users 同风格(Python 侧聚合,跨 PG/SQLite 无方言坑)。

    注:每日明细行的 click_pv/copy_pv 计全部点击(含匿名、UV 按 ip,ua);本函数只计 openid
    用户,故各用户求和 <= 当天行总数,二者口径不同、不必相等。
    """
    rows = db.execute(
        select(CpsClick.openid, CpsClick.event_type, CpsClick.link_id)
        .where(CpsClick.group_id == group_id)
        .where(CpsClick.clicked_at >= _as_utc(start))
        .where(CpsClick.clicked_at < _as_utc(end))
        .where(CpsClick.openid.is_not(None))
    ).all()
    if not rows:
        return []

    # link_id -> activity_id -> 券名(活动名)
    link_ids = {r.link_id for r in rows}
    link_to_act = dict(
        db.execute(
            select(CpsLink.id, CpsLink.activity_id).where(CpsLink.id.in_(link_ids))
        ).all()
    )
    act_ids = {aid for aid in link_to_act.values() if aid is not None}
    act_name = (
        dict(
            db.execute(
                select(CpsActivity.id, CpsActivity.name).where(CpsActivity.id.in_(act_ids))
            ).all()
        )
        if act_ids
        else {}
    )

    def _coupon_name(link_id: int) -> str:
        aid = link_to_act.get(link_id)
        if aid is None:
            return f"链接#{link_id}"
        return act_name.get(aid) or f"活动#{aid}"

    stat: dict[str, dict] = {}
    for openid, event_type, link_id in rows:
        s = stat.setdefault(openid, {"copy": 0, "visit": 0, "coupons": {}})
        if event_type == "copy":
            s["copy"] += 1
        else:
            s["visit"] += 1
            name = _coupon_name(link_id)
            s["coupons"][name] = s["coupons"].get(name, 0) + 1

    openids = list(stat.keys())
    users = {
        u.openid: u
        for u in db.execute(
            select(CpsWxUser).where(CpsWxUser.openid.in_(openids))
        ).scalars().all()
    }

    result = [
        {
            "openid": openid,
            "nickname": users[openid].nickname if openid in users else None,
            "headimgurl": users[openid].headimgurl if openid in users else None,
            "copy_count": s["copy"],
            "visit_count": s["visit"],
            "coupons": [
                {"name": name, "count": cnt}
                # 次数倒序;同次数按券名升序兜底,保证 PG 无 ORDER BY 行序下输出稳定
                for name, cnt in sorted(
                    s["coupons"].items(), key=lambda kv: (-kv[1], kv[0])
                )
            ],
        }
        for openid, s in stat.items()
    ]
    result.sort(key=lambda x: (x["copy_count"], x["visit_count"]), reverse=True)
    return result[:limit]
