# -*- coding: utf-8 -*-
"""Export full intimacy JSON with member visits and card balances."""

from __future__ import annotations

import argparse
import json
import os
import sys
from datetime import date, datetime
from decimal import Decimal
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple

ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
if ROOT not in sys.path:
    sys.path.insert(0, ROOT)

from config.settings import AppConfig
from database.connection import DatabaseConnection
from database.operations import DatabaseOperations


def _as_int(v: Any) -> Optional[int]:
    if v is None:
        return None
    s = str(v).strip()
    if not s:
        return None
    return int(s)


def _to_float(v: Any, default: float = 0.0) -> float:
    if v is None:
        return default
    if isinstance(v, Decimal):
        return float(v)
    if isinstance(v, (int, float)):
        return float(v)
    s = str(v).strip()
    if not s:
        return default
    return float(s)


def _fmt_dt(v: Any) -> Optional[str]:
    if v is None:
        return None
    if isinstance(v, datetime):
        return v.isoformat()
    if isinstance(v, date):
        return v.isoformat()
    return str(v)


def _resolve_site_id(config: AppConfig, db: DatabaseOperations, cli_site_id: Optional[int]) -> int:
    if cli_site_id is not None:
        return int(cli_site_id)

    from_cfg = _as_int(config.get("app.store_id")) or _as_int(config.get("app.default_site_id"))
    if from_cfg is not None:
        return from_cfg

    rows = db.query(
        """
        SELECT site_id
        FROM dws.dws_member_assistant_intimacy
        WHERE site_id IS NOT NULL
        GROUP BY site_id
        ORDER BY COUNT(*) DESC
        LIMIT 1
        """
    )
    if rows:
        return int(dict(rows[0])["site_id"])

    raise RuntimeError("Unable to resolve site_id; pass --site-id explicitly.")


def _fetch_pairs(db: DatabaseOperations, site_id: int) -> List[Dict[str, Any]]:
    sql = """
    SELECT
        i.site_id,
        i.tenant_id,
        i.member_id,
        i.assistant_id,
        i.session_count,
        i.total_duration_minutes,
        i.basic_session_count,
        i.incentive_session_count,
        i.days_since_last_session,
        i.attributed_recharge_count,
        i.attributed_recharge_amount,
        i.score_frequency,
        i.score_recency,
        i.score_recharge,
        i.score_duration,
        i.burst_multiplier,
        i.raw_score,
        i.display_score,
        i.calc_time,
        COALESCE(m.nickname, CONCAT('member_', i.member_id::text)) AS member_nickname,
        COALESCE(a.nickname, CONCAT('assistant_', i.assistant_id::text)) AS assistant_nickname
    FROM dws.dws_member_assistant_intimacy i
    LEFT JOIN dwd.dim_member m
        ON i.member_id = m.member_id
       AND m.scd2_is_current = 1
    LEFT JOIN dwd.dim_assistant a
        ON i.assistant_id = a.assistant_id
       AND a.scd2_is_current = 1
    WHERE i.site_id = %s
    ORDER BY i.display_score DESC NULLS LAST, i.session_count DESC, i.member_id, i.assistant_id
    """
    rows = db.query(sql, (site_id,))
    return [dict(r) for r in (rows or [])]


def _fetch_member_cards(
    db: DatabaseOperations,
    site_id: int,
    member_ids: List[int],
) -> Dict[int, Dict[str, Any]]:
    if not member_ids:
        return {}

    member_ids_str = ",".join(str(int(x)) for x in sorted(set(member_ids)))
    sql = f"""
    SELECT
        tenant_member_id AS member_id,
        member_card_id,
        card_type_id,
        member_card_grade_code,
        member_card_grade_code_name,
        member_card_type_name,
        member_name,
        member_mobile,
        balance,
        principal_balance,
        status,
        start_time,
        end_time,
        last_consume_time
    FROM dwd.dim_member_card_account
    WHERE register_site_id = %s
      AND scd2_is_current = 1
      AND COALESCE(is_delete, 0) = 0
      AND tenant_member_id IN ({member_ids_str})
    ORDER BY tenant_member_id, balance DESC NULLS LAST, member_card_id
    """
    rows = db.query(sql, (site_id,)) or []

    result: Dict[int, Dict[str, Any]] = {}
    for r in rows:
        d = dict(r)
        mid = int(d["member_id"])
        balance = _to_float(d.get("balance"), 0.0)
        card = {
            "member_card_id": _as_int(d.get("member_card_id")),
            "card_type_id": _as_int(d.get("card_type_id")),
            "member_card_grade_code": _as_int(d.get("member_card_grade_code")),
            "member_card_grade_code_name": d.get("member_card_grade_code_name"),
            "member_card_type_name": d.get("member_card_type_name"),
            "member_name": d.get("member_name"),
            "member_mobile": d.get("member_mobile"),
            "balance": round(balance, 2),
            "principal_balance": round(_to_float(d.get("principal_balance"), 0.0), 2),
            "status": _as_int(d.get("status")),
            "start_time": _fmt_dt(d.get("start_time")),
            "end_time": _fmt_dt(d.get("end_time")),
            "last_consume_time": _fmt_dt(d.get("last_consume_time")),
        }

        bucket = result.setdefault(
            mid,
            {
                "member_id": mid,
                "cards_all": [],
                "cards_balance_ge_10": [],
                "total_card_balance_all": 0.0,
            },
        )
        bucket["cards_all"].append(card)
        bucket["total_card_balance_all"] = round(bucket["total_card_balance_all"] + balance, 2)
        if balance >= 10.0:
            bucket["cards_balance_ge_10"].append(card)

    return result


def _fetch_visit_rows(
    db: DatabaseOperations,
    site_id: int,
    member_ids: List[int],
) -> Dict[Tuple[int, int], Dict[str, Any]]:
    if not member_ids:
        return {}

    member_ids_str = ",".join(str(int(x)) for x in sorted(set(member_ids)))
    sql = f"""
    SELECT
        member_id,
        order_settle_id,
        visit_date,
        visit_time,
        table_name,
        area_name,
        area_category,
        table_duration_min,
        assistant_duration_min,
        table_fee,
        goods_amount,
        assistant_amount,
        total_consume,
        total_discount,
        actual_pay,
        cash_pay,
        cash_card_pay,
        gift_card_pay,
        groupbuy_pay
    FROM dws.dws_member_visit_detail
    WHERE site_id = %s
      AND member_id IN ({member_ids_str})
    ORDER BY member_id, visit_time DESC, order_settle_id DESC
    """
    rows = db.query(sql, (site_id,)) or []

    result: Dict[Tuple[int, int], Dict[str, Any]] = {}
    for r in rows:
        d = dict(r)
        key = (int(d["member_id"]), int(d["order_settle_id"]))
        result[key] = {
            "member_id": int(d["member_id"]),
            "order_settle_id": int(d["order_settle_id"]),
            "visit_date": _fmt_dt(d.get("visit_date")),
            "visit_time": _fmt_dt(d.get("visit_time")),
            "table_name": d.get("table_name"),
            "area_name": d.get("area_name"),
            "area_category": d.get("area_category"),
            "table_duration_min": _as_int(d.get("table_duration_min")) or 0,
            "assistant_duration_min_total": _as_int(d.get("assistant_duration_min")) or 0,
            "table_fee": round(_to_float(d.get("table_fee"), 0.0), 2),
            "goods_amount": round(_to_float(d.get("goods_amount"), 0.0), 2),
            "assistant_amount": round(_to_float(d.get("assistant_amount"), 0.0), 2),
            "total_consume": round(_to_float(d.get("total_consume"), 0.0), 2),
            "total_discount": round(_to_float(d.get("total_discount"), 0.0), 2),
            "actual_pay": round(_to_float(d.get("actual_pay"), 0.0), 2),
            "cash_pay": round(_to_float(d.get("cash_pay"), 0.0), 2),
            "cash_card_pay": round(_to_float(d.get("cash_card_pay"), 0.0), 2),
            "gift_card_pay": round(_to_float(d.get("gift_card_pay"), 0.0), 2),
            "groupbuy_pay": round(_to_float(d.get("groupbuy_pay"), 0.0), 2),
        }
    return result


def _fetch_assistant_service_rows(
    db: DatabaseOperations,
    site_id: int,
    member_ids: List[int],
) -> Dict[Tuple[int, int], List[Dict[str, Any]]]:
    if not member_ids:
        return {}

    member_ids_str = ",".join(str(int(x)) for x in sorted(set(member_ids)))
    sql = f"""
    SELECT
        s.tenant_member_id AS member_id,
        s.order_settle_id,
        d.assistant_id,
        COALESCE(d.nickname, s.nickname) AS assistant_nickname,
        SUM(COALESCE(s.income_seconds, 0)) / 60.0 AS duration_min,
        SUM(COALESCE(s.ledger_amount, 0)) AS amount
    FROM dwd.dwd_assistant_service_log s
    JOIN dwd.dim_assistant d
        ON s.user_id = d.user_id
       AND d.scd2_is_current = 1
    WHERE s.site_id = %s
      AND s.is_delete = 0
      AND s.tenant_member_id IN ({member_ids_str})
      AND s.order_settle_id IS NOT NULL
    GROUP BY
        s.tenant_member_id,
        s.order_settle_id,
        d.assistant_id,
        COALESCE(d.nickname, s.nickname)
    ORDER BY s.tenant_member_id, s.order_settle_id
    """
    rows = db.query(sql, (site_id,)) or []

    result: Dict[Tuple[int, int], List[Dict[str, Any]]] = {}
    for r in rows:
        d = dict(r)
        key = (int(d["member_id"]), int(d["order_settle_id"]))
        rec = {
            "assistant_id": int(d["assistant_id"]),
            "assistant_nickname": d.get("assistant_nickname"),
            "duration_min": round(_to_float(d.get("duration_min"), 0.0), 2),
            "amount": round(_to_float(d.get("amount"), 0.0), 2),
        }
        result.setdefault(key, []).append(rec)

    return result


def _pk_key(assistant_nickname: Optional[str], member_nickname: Optional[str]) -> str:
    a = (assistant_nickname or "").strip() or "assistant_unknown"
    m = (member_nickname or "").strip() or "member_unknown"
    return f"{a}__{m}"


def build_export_payload(db: DatabaseOperations, site_id: int) -> Dict[str, Any]:
    pairs = _fetch_pairs(db, site_id)
    member_ids = sorted({int(p["member_id"]) for p in pairs})

    cards_by_member = _fetch_member_cards(db, site_id, member_ids)
    visits_by_key = _fetch_visit_rows(db, site_id, member_ids)
    service_by_key = _fetch_assistant_service_rows(db, site_id, member_ids)

    visits_by_member: Dict[int, List[Tuple[Tuple[int, int], Dict[str, Any]]]] = {}
    for k, v in visits_by_key.items():
        visits_by_member.setdefault(k[0], []).append((k, v))

    data_by_pk: Dict[str, Dict[str, Any]] = {}
    collisions: List[str] = []

    for p in pairs:
        member_id = int(p["member_id"])
        assistant_id = int(p["assistant_id"])
        assistant_nickname = p.get("assistant_nickname")
        member_nickname = p.get("member_nickname")

        visit_items: List[Dict[str, Any]] = []
        for key, visit in visits_by_member.get(member_id, []):

            service_list = service_by_key.get(key, [])
            if not service_list:
                continue

            matched = [x for x in service_list if x["assistant_id"] == assistant_id]
            if not matched:
                continue

            matched_duration = round(sum(x["duration_min"] for x in matched), 2)
            matched_amount = round(sum(x["amount"] for x in matched), 2)
            matched_nicknames = sorted({x.get("assistant_nickname") for x in matched if x.get("assistant_nickname")})

            visit_items.append(
                {
                    "order_settle_id": visit.get("order_settle_id"),
                    "visit_date": visit.get("visit_date"),
                    "visit_time": visit.get("visit_time"),
                    "table_name": visit.get("table_name"),
                    "area_name": visit.get("area_name"),
                    "area_category": visit.get("area_category"),
                    "table_duration_min": visit.get("table_duration_min"),
                    "assistant_duration_min_total": visit.get("assistant_duration_min_total"),
                    "table_fee": visit.get("table_fee"),
                    "goods_amount": visit.get("goods_amount"),
                    "assistant_amount": visit.get("assistant_amount"),
                    "total_consume": visit.get("total_consume"),
                    "total_discount": visit.get("total_discount"),
                    "actual_pay": visit.get("actual_pay"),
                    "cash_pay": visit.get("cash_pay"),
                    "cash_card_pay": visit.get("cash_card_pay"),
                    "gift_card_pay": visit.get("gift_card_pay"),
                    "groupbuy_pay": visit.get("groupbuy_pay"),
                    "target_assistant_nickname": ", ".join(matched_nicknames) if matched_nicknames else p.get("assistant_nickname"),
                    "target_assistant_duration_min": matched_duration,
                    "target_assistant_amount": matched_amount,
                }
            )

        visit_items.sort(
            key=lambda x: (x.get("visit_time") or "", x.get("order_settle_id") or 0),
            reverse=True,
        )

        member_cards = cards_by_member.get(
            member_id,
            {
                "member_id": member_id,
                "cards_all": [],
                "cards_balance_ge_10": [],
                "total_card_balance_all": 0.0,
            },
        )

        pk = _pk_key(assistant_nickname, member_nickname)
        item = {
            "primary_key": {
                "assistant_nickname": assistant_nickname,
                "member_nickname": member_nickname,
            },
            "intimacy": {
                "display_score": round(_to_float(p.get("display_score"), 0.0), 2),
                "raw_score": round(_to_float(p.get("raw_score"), 0.0), 6),
                "session_count": _as_int(p.get("session_count")) or 0,
                "total_duration_minutes": _as_int(p.get("total_duration_minutes")) or 0,
                "basic_session_count": _as_int(p.get("basic_session_count")) or 0,
                "incentive_session_count": _as_int(p.get("incentive_session_count")) or 0,
                "days_since_last_session": _as_int(p.get("days_since_last_session")),
                "attributed_recharge_count": _as_int(p.get("attributed_recharge_count")) or 0,
                "attributed_recharge_amount": round(_to_float(p.get("attributed_recharge_amount"), 0.0), 2),
                "score_frequency": round(_to_float(p.get("score_frequency"), 0.0), 4),
                "score_recency": round(_to_float(p.get("score_recency"), 0.0), 4),
                "score_recharge": round(_to_float(p.get("score_recharge"), 0.0), 4),
                "score_duration": round(_to_float(p.get("score_duration"), 0.0), 4),
                "burst_multiplier": round(_to_float(p.get("burst_multiplier"), 1.0), 4),
                "calc_time": _fmt_dt(p.get("calc_time")),
            },
            "member_cards": {
                "cards_balance_ge_10": member_cards.get("cards_balance_ge_10", []),
                "total_card_balance_all": round(_to_float(member_cards.get("total_card_balance_all"), 0.0), 2),
            },
            "visit_consumptions": visit_items,
        }
        if pk in data_by_pk:
            collisions.append(pk)
            existing = data_by_pk[pk]
            existing["collision_items"] = existing.get("collision_items", [])
            existing["collision_items"].append(item)
        else:
            data_by_pk[pk] = item

    payload = {
        "meta": {
            "site_id": site_id,
            "generated_at": datetime.now().isoformat(),
            "pair_count": len(pairs),
            "primary_key_count": len(data_by_pk),
            "member_count": len(member_ids),
            "primary_key_rule": "assistant_nickname + member_nickname",
            "collision_count": len(collisions),
        },
        "data": data_by_pk,
    }
    return payload


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Export full intimacy JSON")
    parser.add_argument("--site-id", type=int, default=None, help="site_id, defaults to app.store_id")
    parser.add_argument(
        "--output",
        default="tmp/intimacy_full_export.json",
        help="output JSON file path",
    )
    return parser.parse_args()


def main() -> None:
    args = parse_args()

    config = AppConfig.load()
    db_conn = DatabaseConnection(config.config["db"]["dsn"])
    db = DatabaseOperations(db_conn)

    try:
        site_id = _resolve_site_id(config, db, args.site_id)
        payload = build_export_payload(db, site_id)
    finally:
        db_conn.close()

    output_path = Path(args.output)
    output_path.parent.mkdir(parents=True, exist_ok=True)
    output_path.write_text(
        json.dumps(payload, ensure_ascii=False, indent=2),
        encoding="utf-8",
    )

    print(f"Exported intimacy JSON: {output_path}")
    print(f"pair_count={payload['meta']['pair_count']}, member_count={payload['meta']['member_count']}")


if __name__ == "__main__":
    main()
