"""询盘推送 UniService 服务(设计 2026-09-28-lead-push-to-uniservice)。

三个入口:
- dispatch_lead_push: 触发点调用(发信分支进入处),写 pending/skipped 日志。
  红线: 整体吞异常,绝不影响邮件主流程。
- fire_pending_pushes: 主事务 commit 之后调用,create_task 派生后台任务。
- sweep_stale_pending: lifespan 启动清扫(崩溃遗留的 pending → failed)。

后台任务红线(设计 8.1): 不读库只写结果,payload 内存传递;
结果写入用独立 session——session_factory 由派发方传入(Task 5b 起无全局
单例,写回派发它的那个租户的库;Task 9 收口: 转必填,漏传在调用点即
TypeError,不再拖到任务深处)。

Task 9 配置接线: PUSH 组(开关/URL/密钥/前缀/超时/重试)经
resolve_push_settings 从租户 config 解析,空槽回退 env settings——
各入口收 config_data(租户配置 dict),None = 平台默认(env)。
"""
import asyncio
import logging
from datetime import datetime, timedelta, timezone

import httpx
from sqlalchemy import func, select, update
from sqlalchemy.ext.asyncio import AsyncSession

from app.engine.config import settings
from app.engine.models import (
    Conversation, CustomerProfile, Evaluation, IntentLevel, LeadPushLog,
    Message, MessageRole,
)
from app.engine.services.cc_completion import cc_state_for, completion_note, outbound_e164
from app.engine.services.ip_service import resolve_country

logger = logging.getLogger(__name__)


def resolve_push_settings(config_data: dict | None) -> dict:
    """租户 config.push → 推送参数(Task 9 接线; env settings 回退)。

    出厂默认的空槽(False/空串)视同"未设置"→ 回退 env PUSH_*——部署级
    .env 开关是平台默认集的一部分,租户未配置时继承,不静默关停。
    已知边界: env 开 + 租户显式 false 无法区分于未配置(布尔空槽只有
    False),Phase 2 admin 落地时以"平台清 env、配置显式写 false"解决。
    """
    p = ((config_data or {}).get("push")) or {}
    return {
        "enabled": bool(p.get("enabled") or settings.PUSH_ENABLED),
        "url": p.get("url") or settings.PUSH_API_URL,
        "api_key": p.get("api_key") or settings.PUSH_API_KEY,
        "prefix": p.get("prefix") or settings.PUSH_EXTERNAL_KEY_PREFIX,
        "timeout": p.get("timeout") or settings.PUSH_TIMEOUT_SECONDS,
        "max_attempts": p.get("max_attempts") or settings.PUSH_MAX_ATTEMPTS,
        "retry_backoff_base": (p.get("retry_backoff_base")
                               or settings.PUSH_RETRY_BACKOFF_BASE),
    }

# 对方契约(docs/AI_INQUIRY_PUSH_API.md): phone/whatsapp/email 至少一个有效
_TRIO = ("phone", "whatsapp", "email")
_CST = timezone(timedelta(hours=8))
_INTENT_MAP = {
    "interested": "high",
    "pending": "medium",
    "not_interested": "low",
}


def _fmt_cst(dt: datetime) -> str:
    """UTC(-aware 或 naive)→ +08:00 ISO 串。对方明确拒收 UTC(契约 §2)。"""
    if dt.tzinfo is None:
        dt = dt.replace(tzinfo=timezone.utc)
    dt = dt.replace(microsecond=0)   # 归一整秒: 契约示例为整秒,亚秒精度无益且冒严格解析器拒收风险
    return dt.astimezone(_CST).isoformat()


def map_intent(level: str | None, evaluated: bool) -> str:
    """我方 IntentLevel → 对方 high/medium/low/unknown。

    evaluated=False(<3 条客户消息,D2)→ unknown: 未评估不是"待确认",
    与邮件标签 [未评估]/[待确认] 的区分同源。
    """
    if not evaluated or not level:
        return "unknown"
    return _INTENT_MAP.get(level, "unknown")


_sleep = asyncio.sleep   # 模块级绑定,测试 monkeypatch 记录退避序列用


async def _build_payload(
    db: AsyncSession, ev: Evaluation, conv: Conversation,
    profile: CustomerProfile, confirmed: dict, push_type: str,
    evaluation: dict, evaluated: bool,
    config_data: dict | None = None,
) -> dict:
    """组装契约报文(设计 7.2 映射表)。仅读当前事务,不产生副作用。"""
    # 7.1 event 规则: 仅更正可能 update,且必须同会话此前推送成功过
    # (不依赖对方"update 未知键"的未定义行为)
    event = "new"
    if push_type == "correction":
        prior = await db.execute(
            select(func.count()).select_from(LeadPushLog).where(
                LeadPushLog.conversation_id == conv.id,
                LeadPushLog.status == "success",
            ))
        if (prior.scalar() or 0) > 0:
            event = "update"

    result = await db.execute(
        select(Message)
        .where(Message.conversation_id == conv.id)
        .order_by(Message.created_at)
    )
    msgs = result.scalars().all()

    raw_info = profile.raw_info or {}
    client_info = conv.client_info or {}
    # Task 6(spec §5 先消费后过滤,顺序钉死): 先读 raw_info 的 _cc 派生——
    # contact 的 phone/whatsapp 换成对应渠道 e164(首封即补全号,确认与否都给;
    # 无 e164 态用原文),标注进"号码补全说明"派生键;之后才做 _ 前缀过滤,
    # _cc 本体绝不进 payload,派生键(无 _ 前缀)存活。canon 不匹配的残留态
    # 在 cc_state_for 内判 None——旧补全永不跟随新号。
    cc_map = raw_info.get("_cc") or {}
    contact_out: dict = {}
    cc_notes: list[str] = []
    for k, v in (confirmed or {}).items():
        if k in ("phone", "whatsapp") and v:
            state = cc_state_for(cc_map, k, str(v))
            contact_out[k] = outbound_e164(str(v), state)
            note = completion_note(str(v), state)
            if note:
                cc_notes.append(note)
        else:
            contact_out[k] = v
    # _ 前缀内部状态(一次性计数等)不进推送画像(spec §4.3——渲染侧
    # 三处有过滤(enrich/reminder/email_service),推送端曾整体拷贝泄漏)
    profile_out = {k: v for k, v in raw_info.items()
                   if not str(k).startswith("_")}
    profile_out["country"] = resolve_country(client_info.get("ip", ""))
    profile_out["source"] = conv.source.value
    if client_info.get("lang"):
        profile_out["语言"] = client_info["lang"]
    if cc_notes:
        profile_out["号码补全说明"] = "；".join(cc_notes)

    reasoning = (evaluation.get("reasoning") or "") if evaluated else ""

    return {
        "event": event,
        "external_key": f"{resolve_push_settings(config_data)['prefix']}-{conv.id}",
        "occurred_at": _fmt_cst(datetime.now(timezone.utc)),
        "contact": {
            "name": None,   # 我方 contact_info 快照无称呼字段,契约可空
            "phone": contact_out.get("phone"),
            "whatsapp": contact_out.get("whatsapp"),
            "email": contact_out.get("email"),
            "wechat": contact_out.get("wechat"),
            "skype": None,
        },
        "intent_level": map_intent(evaluation.get("intent_level"), evaluated),
        "intent_reason": reasoning or None,
        "profile": profile_out,
        "conversation": [
            {
                "role": "customer" if m.role == MessageRole.user else "ai",
                "text": m.content,
                "at": _fmt_cst(m.created_at),
            }
            for m in msgs if m.role in (MessageRole.user, MessageRole.assistant)
        ],
    }


async def dispatch_lead_push(
    db: AsyncSession, ev: Evaluation, conv: Conversation,
    profile: CustomerProfile, confirmed: dict, push_type: str,
    evaluation: dict, evaluated: bool,
    config_data: dict | None = None,
) -> LeadPushLog | None:
    """触发点唯一入口: 返回待派发的 pending 行(加入当前事务,随主事务提交)。

    红线(设计 6.1): 整体吞异常——任何失败只记日志,绝不影响邮件主流程;
    触发条件 = 调用方进入发信分支,与 SMTP 成败解耦。
    DB 层失败用 SAVEPOINT 隔离: 只回滚推送写入,不毒化调用方事务
    (否则 flush 失败 → PendingRollbackError → email_sent 落不了库 → 重复发邮件)。
    config_data = 租户配置(Task 9): 开关/前缀经 resolve_push_settings,
    None 走 env settings(平台默认)。
    """
    if not resolve_push_settings(config_data)["enabled"]:
        return None
    try:
        async with db.begin_nested():
            # 三选一预检(设计 §4 skipped 语义): 对方必拒,不发 HTTP 只留审计痕迹
            if not any(confirmed.get(k) for k in _TRIO):
                db.add(LeadPushLog(
                    evaluation_id=ev.id, conversation_id=conv.id,
                    push_type=push_type, status="skipped", attempts=0,
                    error_message="no valid contact (phone/whatsapp/email)",
                ))
                return None
            payload = await _build_payload(
                db, ev, conv, profile, confirmed, push_type, evaluation,
                evaluated, config_data)
            log = LeadPushLog(
                evaluation_id=ev.id, conversation_id=conv.id,
                push_type=push_type, status="pending", attempts=0,
                request_payload=payload,
            )
            db.add(log)
            await db.flush()   # 取 id 供后台任务写结果
        return log
    except Exception:
        logger.exception("dispatch_lead_push failed (type=%s, conv=%s)",
                         push_type, conv.id)
        return None


async def _do_post(payload: dict,
                   config_data: dict | None = None) -> tuple[httpx.Response | None, str | None]:
    """实际 HTTP 调用(独立函数便于测试 monkeypatch)。

    返回 (resp, transport_error): 传输层异常不抛出,转成错误串由重试矩阵消化。
    URL/密钥/超时经 resolve_push_settings(Task 9: 租户配置, None 回退 env)。
    """
    cfg = resolve_push_settings(config_data)
    try:
        async with httpx.AsyncClient(
                timeout=cfg["timeout"],
                follow_redirects=True) as client:   # 307/308 不落终态失败
            resp = await client.post(
                cfg["url"],
                json={"messages": [payload]},   # 契约 §2: 批量结构,我方单条
                headers={"X-Api-Key": cfg["api_key"]},
            )
        return resp, None
    except httpx.HTTPError as exc:
        return None, repr(exc)


def _retryable(resp: httpx.Response | None) -> bool:
    """契约 §3: 5xx/超时可重试;401/400/其他 4xx 重试无益。200 不在此判定。"""
    return resp is None or resp.status_code >= 500


async def _update_log(session_factory, push_log_id: str, **fields) -> None:
    """后台任务写结果——独立 session + 小事务(设计 6.3)。
    session_factory: 派发租户的工厂(Task 5b),写回同一租户库。"""
    async with session_factory() as session:
        await session.execute(
            update(LeadPushLog)
            .where(LeadPushLog.id == push_log_id)
            .values(**fields)
        )
        await session.commit()


async def _run_push(push_log_id: str, payload: dict,
                    session_factory, config_data: dict | None = None) -> None:
    """后台任务体: 分类重试 + 200 逐条结果解析(设计 6.3/§10)。

    红线: 全套 try/except,异常只记日志绝不逃逸(fire-and-forget)。
    session_factory 必填(Task 9 收口): 派发租户的工厂,写回同一租户库。
    重试矩阵参数(上限/退避)经 resolve_push_settings(租户配置, None 回退 env)。
    """
    cfg = resolve_push_settings(config_data)
    try:
        attempts = 0
        while True:
            attempts += 1
            resp, transport_error = await _do_post(payload, config_data)
            if _retryable(resp) and attempts < cfg["max_attempts"]:
                await _update_log(
                    session_factory, push_log_id, attempts=attempts,
                    error_message=transport_error
                    or (f"http {resp.status_code}" if resp is not None else "unknown"),
                )
                await _sleep(
                    cfg["retry_backoff_base"] * 2 ** (attempts - 1))
                continue
            break

        if resp is not None and resp.status_code == 200:
            try:
                raw = resp.json()
            except ValueError:
                # 非 JSON 的 200(如网关重定向到 HTML 登录页)——
                # 无法确认对方已接受,诚实落 failed(幂等键保证补推安全)
                await _update_log(session_factory, push_log_id, attempts=attempts,
                                  status="failed",
                                  response_status=200,
                                  error_message="non-JSON 200 body")
                return
            data = raw.get("data") if isinstance(raw, dict) else None
            data = data if isinstance(data, dict) else {}
            failed_mine = [
                f for f in (data.get("failed") or [])
                if f.get("external_key") == payload["external_key"]
            ]
            if failed_mine:
                await _update_log(
                    session_factory, push_log_id, attempts=attempts,
                    status="failed", response_status=200,
                    error_message=str(failed_mine[0].get("reason") or "rejected"),
                )
            else:
                # 单条推送: accepted/duplicated 计数即本条结论
                outcome = ("duplicated"
                           if (data.get("duplicated") or 0) > 0 else "accepted")
                await _update_log(
                    session_factory, push_log_id, attempts=attempts,
                    status="success", response_status=200, outcome=outcome,
                )
            return

        # 终态失败: 重试耗尽(5xx/传输)或不可重试(401/400/其他)
        if resp is not None and resp.status_code in (400, 401):
            logger.error("lead push rejected (log_id=%s, http %s) — "
                         "密钥/报文问题需人工介入,不重试",
                         push_log_id, resp.status_code)
        await _update_log(
            session_factory, push_log_id, attempts=attempts, status="failed",
            response_status=resp.status_code if resp is not None else None,
            error_message=transport_error
            or (f"http {resp.status_code}" if resp is not None else "unknown"),
        )
    except Exception:
        logger.exception("lead push task crashed (log_id=%s)", push_log_id)


# 后台任务引用集: 防 GC(asyncio 官方要求),done 即移除
_tasks: set[asyncio.Task] = set()


def fire_pending_pushes(pushes: list[LeadPushLog], session_factory,
                        config_data: dict | None = None) -> None:
    """主事务 commit 之后调用(设计 8.1): payload 内存传递,任务不读库。

    同步函数、纳秒级返回——这就是"不阻塞"的派发点。session_factory
    必填(Task 9 收口): 派发租户的工厂,随任务传递(后台写结果落回同一
    租户库);config_data 同随任务传递(重试参数走租户配置)。
    """
    for log in pushes:
        if log.status != "pending":
            continue
        task = asyncio.create_task(
            _run_push(log.id, log.request_payload, session_factory,
                      config_data=config_data))
        _tasks.add(task)
        task.add_done_callback(_tasks.discard)


async def sweep_stale_pending(session_factory, stale_minutes: int = 30) -> int:
    """启动清扫(设计 8.2): 崩溃遗留的 pending → failed。

    session_factory 必填(Task 9 收口): 目标租户的工厂(lifespan 逐租户调用)。
    已知限制(设计 §14): 推送成功但结果写回失败也会被标 failed——
    幂等键保证将来补推安全(对方返回 duplicated),概率极低,接受。
    """
    cutoff = datetime.now(timezone.utc) - timedelta(minutes=stale_minutes)
    async with session_factory() as session:
        result = await session.execute(
            update(LeadPushLog).where(
                LeadPushLog.status == "pending",
                LeadPushLog.updated_at < cutoff,
            ).values(status="failed", error_message="interrupted by restart")
        )
        await session.commit()
        return result.rowcount or 0