"""同步共享:客户端工厂、顺序分页(含重试/延迟)、RIIN 商品拉取、SDS 物流渠道字典、日期辅助。

行为对齐 inkreach-bi-auth-mng 的 sync/tasks/clients.ts:
- 顺序分页(页间 500ms),非并发,对外部 API 友好。
- fetch_with_retry:仅对 5xx / 429 / 超时重试,最多 3 次,退避 3s*次数。
"""

from __future__ import annotations

import json
import threading
import time
from datetime import datetime, timedelta, timezone
from typing import Any, Callable, TypeVar

from prefect.blocks.system import Secret

from api import RiinDisClient, SdsMngClient
from api.riin.models import OrderPageRequest as RiinReq
from api.sds.models import LogisticsPageRequest, OrderPageRequest as SdsReq
from flows.shared.mappers import map_sds_channel_type

T = TypeVar("T")

GOODS_CONCURRENCY = 5
GOODS_MIN_INTERVAL_MS = 150  # 相邻两次 order/goods 调用全局最小间隔(峰值 ≤ ~6.7 次/秒)
PAGE_DELAY_MS = 500
MAX_RETRIES = 3
RETRY_DELAY_MS = 3000
TZ_CN = timezone(timedelta(hours=8))

_RIIN_FINISHED = ["已退款", "已关闭", "已发货"]
_SDS_FINISHED = ["已完成", "已取消"]


# ---------- 客户端工厂 ----------

def create_riin_client() -> RiinDisClient:
    cred = _load_secret("riin-credentials")
    return RiinDisClient(cred["username"], cred["password"])


def create_sds_client() -> SdsMngClient:
    cred = _load_secret("sds-credentials")
    return SdsMngClient(cred["username"], cred["password"])


def _load_secret(name: str) -> dict[str, str]:
    value = Secret.load(name).get()
    if isinstance(value, str):
        value = json.loads(value)
    return value


# ---------- 日期辅助 ----------

def format_date(d: datetime) -> str:
    return d.strftime("%Y-%m-%d %H:%M:%S")


def start_of_day(d: datetime) -> datetime:
    return d.replace(hour=0, minute=0, second=0, microsecond=0)


def floor_to_hour(d: datetime) -> datetime:
    return d.replace(minute=0, second=0, microsecond=0)


# ---------- 重试 ----------

def fetch_with_retry(fn: Callable[[], T], label: str) -> T:
    last_err: Exception | None = None
    for attempt in range(1, MAX_RETRIES + 1):
        try:
            return fn()
        except Exception as err:  # noqa: BLE001
            last_err = err
            status = _status_of(err)
            msg = str(err)
            is_timeout = "timeout" in msg.lower() or "timed out" in msg.lower()
            retryable = status == 429 or (status is not None and status >= 500) or is_timeout
            if attempt < MAX_RETRIES and retryable:
                print(f"[sync] {label} 第 {attempt}/{MAX_RETRIES} 次重试 ({status}: {msg[:120]})")
                time.sleep(RETRY_DELAY_MS * attempt / 1000)
                continue
            raise
    assert last_err is not None
    raise last_err


def _status_of(err: Exception) -> int | None:
    for attr in ("status_code", "statusCode", "status"):
        v = getattr(err, attr, None)
        if isinstance(v, int):
            return v
    resp = getattr(err, "response", None)
    return getattr(resp, "status_code", None) if resp is not None else None


# ---------- 分页 ----------

def fetch_riin_all_pages(client: RiinDisClient, params: dict[str, Any], max_pages: int = 1000) -> list[Any]:
    page_size = params.get("page_size", 200)
    base = {**params, "page_size": page_size}

    def fetch(idx: int):
        return fetch_with_retry(lambda: client.get_orders(RiinReq(**{**base, "page_index": idx})), f"RIIN page {idx}")

    first = fetch(1)
    total = first.data.total
    records = list(first.data.records)
    total_pages = min(-(-total // page_size), max_pages)
    print(f"[RIIN sync] 分页开始: 共 {total} 条 / {total_pages} 页 (page_size={page_size})")
    for page in range(2, total_pages + 1):
        resp = fetch(page)
        records.extend(resp.data.records)
        if page % 10 == 0 or page == total_pages:
            print(f"[RIIN sync] 分页进度: {page}/{total_pages} 页, 累计 {len(records)} 条")
        time.sleep(PAGE_DELAY_MS / 1000)
    if total_pages == max_pages:
        print(f"[RIIN sync] 达到最大页数限制 {max_pages},截断到 {len(records)} 条")
    print(f"[RIIN sync] 分页完成: 共拉取 {len(records)} 条")
    return records


def fetch_sds_all_pages(client: SdsMngClient, params: dict[str, Any]) -> list[Any]:
    page_size = params.get("size", 500)
    base = {**params, "size": page_size}

    def fetch(idx: int):
        return fetch_with_retry(lambda: client.get_orders(SdsReq(**{**base, "page": idx})), f"SDS page {idx}")

    first = fetch(1)
    records = list(first.records)
    total_pages = -(-first.total_count // page_size)
    print(f"[SDS sync] 分页开始: 共 {first.total_count} 条 / {total_pages} 页 (page_size={page_size})")
    for page in range(2, total_pages + 1):
        resp = fetch(page)
        records.extend(resp.records)
        if page % 10 == 0 or page == total_pages:
            print(f"[SDS sync] 分页进度: {page}/{total_pages} 页, 累计 {len(records)} 条")
        time.sleep(PAGE_DELAY_MS / 1000)
    print(f"[SDS sync] 分页完成: 共拉取 {len(records)} 条")
    return records


# ---------- RIIN 商品 ----------

# 全局限速:RIIN 分销侧反馈 distribution/order/goods 逐单调用对其服务器压力过大,
# 所有线程共享节拍,相邻两次调用至少间隔 GOODS_MIN_INTERVAL_MS。
_goods_pace_lock = threading.Lock()
_goods_last_call_at = 0.0


def _pace_goods_call() -> None:
    global _goods_last_call_at
    with _goods_pace_lock:
        wait = _goods_last_call_at + GOODS_MIN_INTERVAL_MS / 1000 - time.monotonic()
        if wait > 0:
            time.sleep(wait)
        _goods_last_call_at = time.monotonic()


def fetch_goods_for_order(client: RiinDisClient, order_id: str) -> list[Any]:
    try:
        _pace_goods_call()
        resp = client.get_order_goods(order_id)
        return list(resp.data)
    except Exception as err:  # noqa: BLE001
        print(f"[RIIN sync] 订单 {order_id} 商品拉取失败: {err}")
        return []


def fetch_goods_concurrent(client: RiinDisClient, order_ids: list[str]) -> dict[str, list[Any]]:
    """以 GOODS_CONCURRENCY 并发拉取多订单商品,返回 {order_id: [goods...]}。"""
    from collections.abc import Callable
    from concurrent.futures import ThreadPoolExecutor

    results: dict[str, list[Any]] = {}
    if not order_ids:
        return results
    concurrency = min(GOODS_CONCURRENCY, len(order_ids))
    print(f"[RIIN sync] 商品拉取开始: {len(order_ids)} 个订单 (并发={concurrency})")

    def make(oid: str) -> Callable[[], None]:
        def run() -> None:
            results[oid] = fetch_goods_for_order(client, oid)
        return run

    with ThreadPoolExecutor(max_workers=concurrency) as pool:
        list(pool.map(lambda fn: fn(), [make(oid) for oid in order_ids]))
    total_goods = sum(len(v) for v in results.values())
    empty_orders = sum(1 for v in results.values() if not v)
    print(f"[RIIN sync] 商品拉取完成: 共 {total_goods} 条商品, {empty_orders} 个订单无商品")
    return results


# ---------- SDS 物流渠道字典 ----------

_channel_cache: dict[str, Any] = {}
_CHANNEL_TTL = 24 * 3600
_CHANNEL_KEY = "all"


def get_channel_type_map(client: SdsMngClient) -> dict[int, str | None]:
    cached = _channel_cache.get(_CHANNEL_KEY)
    if cached and cached["expire_at"] > time.time():
        return cached["data"]
    mapping: dict[int, str | None] = {}
    try:
        areas = client.get_logistics_areas()
        area_ids = [a.id for a in areas.items if a.id is not None]
    except Exception as err:  # noqa: BLE001
        print(f"[SDS sync] 物流区域拉取失败: {err}")
        area_ids = []
    for area_id in area_ids:
        try:
            resp = client.get_logistics_page(LogisticsPageRequest(issuing_bay_area_id=area_id, all=True, size=200))
            for item in resp.items:
                if item.id is None:
                    continue
                raw = getattr(item, "channelType", None) or getattr(item, "channel_type", None)
                label = map_sds_channel_type(raw)
                if label is None:
                    continue
                mapping[int(item.id)] = label
        except Exception as err:  # noqa: BLE001
            print(f"[SDS sync] 区域 {area_id} 物流字典拉取失败: {err}")
    _channel_cache[_CHANNEL_KEY] = {"data": mapping, "expire_at": time.time() + _CHANNEL_TTL}
    return mapping


def refresh_channel_type_map(client: SdsMngClient) -> dict[int, str | None]:
    _channel_cache.pop(_CHANNEL_KEY, None)
    return get_channel_type_map(client)


def resolve_courier_source(
    raw_logistics_id: Any,
    mapping: dict[int, str | None],
    refresh: Callable[[], dict[int, str | None]],
    refreshed_ids: set[int],
) -> str | None:
    if raw_logistics_id is None:
        return None
    try:
        lid = int(raw_logistics_id)
    except (TypeError, ValueError):
        return None
    if lid in mapping:
        return mapping[lid]
    if lid in refreshed_ids:
        return None
    refreshed_ids.add(lid)
    try:
        new_map = refresh()
        mapping.update(new_map)
    except Exception:  # noqa: BLE001
        pass
    return mapping.get(lid)