"""合作方通讯录:同步、查询、待确认队列。

同步是「拉名册 → 打标」两个阶段,但对管理员是一个开关一个 cron ——
拆成两个任务的话,第一个成功第二个失败时,界面上会出现「名册是今天
的、标签是昨天的」这种没人能一眼看懂的状态。
"""
from __future__ import annotations

import logging
import uuid
from datetime import UTC, datetime
from typing import Any

from sqlalchemy import func, or_, select
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.coordinators.partner_tag_coordinator import sync_partner_tags
from app.core.exceptions import AppException, NotFoundError
from app.domains.partner_directory.models import (
    LINK_APPLIED,
    LINK_DISMISSED,
    LINK_PENDING_REVIEW,
    LINK_STALE,
    PartnerDirectoryLink,
    PartnerDirectoryMember,
)
from app.integrations.driver_base import (
    PartnerDirectoryDriver,
    PartnerDirectoryMemberRecord,
    PartnerDirectorySnapshot,
)
from app.integrations.models import Integration

logger = logging.getLogger(__name__)

PARTNER_DIRECTORY_KIND = "partner_directory"


def _now() -> datetime:
    return datetime.now(UTC)


def _local_part(email: str | None) -> str | None:
    if not email or "@" not in email:
        return None
    part = email.split("@", 1)[0].strip().lower()
    return part or None


def _norm_email(email: str | None) -> str | None:
    value = (email or "").strip().lower()
    return value or None


async def list_partner_integrations(db: AsyncSession) -> list[Integration]:
    """所有启用的合作方通讯录连接器。"""
    return list((await db.execute(
        select(Integration).where(
            Integration.kind == PARTNER_DIRECTORY_KIND,
            Integration.enabled.is_(True),
        ).order_by(Integration.name),
    )).scalars().all())


async def sync_roster(
    db: AsyncSession,
    *,
    integration_id: str,
    driver: PartnerDirectoryDriver,
) -> dict[str, Any]:
    """拉一次对方名册,写进 ``partner_directory_members``。

    退役规则:只有 ``snapshot.complete`` 为真时,本次没出现的行才置
    ``deleted_at``。一次因为权限过期而读到半份名册的同步会把几百个人
    标成"已离开",而下游的打标是按活跃名册算的 —— 那一天所有 CBG 标签
    会集体变成僵尸链接,第二天权限恢复后又全部复活。宁可留着过期的行。
    """
    snapshot: PartnerDirectorySnapshot = await driver.list_members()
    now = _now()

    existing = {
        row.external_id: row
        for row in (await db.execute(
            select(PartnerDirectoryMember).where(
                PartnerDirectoryMember.integration_id == integration_id,
            ),
        )).scalars().all()
    }

    created = updated = revived = 0
    seen: set[str] = set()
    for record in snapshot.members:
        seen.add(record.external_id)
        row = existing.get(record.external_id)
        if row is None:
            row = PartnerDirectoryMember(
                id=str(uuid.uuid4()),
                integration_id=integration_id,
                external_id=record.external_id,
                first_seen_at=now,
            )
            db.add(row)
            existing[record.external_id] = row
            created += 1
        else:
            if row.deleted_at is not None:
                revived += 1
            updated += 1
        _apply_record(row, record)
        row.last_seen_at = now
        row.deleted_at = None

    retired = 0
    if snapshot.complete:
        for external_id, row in existing.items():
            if external_id in seen or row.deleted_at is not None:
                continue
            row.deleted_at = now
            retired += 1
    elif snapshot.members:
        logger.warning(
            "partner_directory[%s]: incomplete snapshot (%d/%d departments "
            "unreadable) — skipping retirement of absent members",
            integration_id, snapshot.departments_unreadable,
            snapshot.departments_scanned,
        )

    await db.flush()
    return {
        "pulled": len(snapshot.members),
        "created": created,
        "updated": updated,
        "revived": revived,
        "retired": retired,
        "complete": snapshot.complete,
        "departments_scanned": snapshot.departments_scanned,
        "departments_unreadable": snapshot.departments_unreadable,
        "with_github_id": sum(
            1 for m in snapshot.members if m.github_login
        ),
        "with_email": sum(1 for m in snapshot.members if m.email),
        "active": sum(1 for m in snapshot.members if m.is_active),
        "notes": list(snapshot.notes),
    }


def _apply_record(
    row: PartnerDirectoryMember, record: PartnerDirectoryMemberRecord,
) -> None:
    row.union_id = record.union_id
    row.partner_user_id = record.partner_user_id
    row.employee_no = record.employee_no
    row.name = record.name
    row.en_name = record.en_name
    row.email = _norm_email(record.email)
    row.secondary_email = _norm_email(record.secondary_email)
    # 跨域匹配的载体。企业邮箱缺席时退到 secondary —— 但只在它也在
    # 对方域里时才有意义,这里不判断域,匹配器那边会判。
    row.email_name = _local_part(row.email) or _local_part(row.secondary_email)
    row.job_title = record.job_title
    row.department_ids = list(record.department_ids)
    row.department_path = list(record.department_path)
    login = (record.github_login or "").strip()
    row.github_login = login or None
    row.github_login_lower = login.lower() or None
    row.is_active = record.is_active
    row.is_suspended = record.is_suspended
    row.status_raw = dict(record.status_raw)
    row.custom_attrs = dict(record.custom_attrs)


async def sync_one(
    db: AsyncSession,
    *,
    integration: Integration,
    trace_id: str,
    apply_tags: bool = True,
) -> dict[str, Any]:
    """一个连接器的完整同步:拉名册 + 打标。"""
    from app.integrations.service import build_driver_by_id

    driver = await build_driver_by_id(integration.id, session=db)
    if not isinstance(driver, PartnerDirectoryDriver):
        raise AppException(
            code="INTEGRATION_KIND_MISMATCH",
            message=(
                f"Integration {integration.slug!r} is not a partner-directory "
                f"connector (kind={integration.kind!r})"
            ),
            status_code=400,
        )

    roster = await sync_roster(
        db, integration_id=str(integration.id), driver=driver,
    )
    local_domain = str(
        (integration.config_public or {}).get("local_email_domain") or "",
    )
    # 打标开始的时刻。用它区分「这一轮又被匹配到」和「这一轮没再出现」
    # —— 匹配器会把每条命中的链接 last_seen_at 推到当前时刻,所以严格
    # 早于这个时刻的就是本轮没命中的。不要在退役函数里再取一次 now:
    # 那会让判据依赖两次时钟调用之间的间隔,而不是依赖这一轮的事实。
    tag_started_at = _now()
    tags = await sync_partner_tags(
        db,
        integration_id=str(integration.id),
        local_email_domain=local_domain,
        trace_id=trace_id,
        apply=apply_tags,
    )
    stale = await _retire_orphan_links(
        db, integration_id=str(integration.id), before=tag_started_at,
    )
    return {
        "integration": integration.slug or str(integration.id),
        "roster": roster,
        "tags": {
            "matched": tags.matched,
            "applied": tags.applied,
            "pending_review": tags.pending_review,
            "skipped": tags.skipped,
            "by_surface": tags.by_surface,
            "by_method": tags.by_method,
            "errors": tags.errors,
        },
        "links_marked_stale": stale,
    }


async def _retire_orphan_links(
    db: AsyncSession, *, integration_id: str, before: datetime,
) -> int:
    """把这一轮没再被匹配到的 ``pending_review`` 链接标成 ``stale``。

    只动 ``pending_review``。已经 ``applied`` 的不动 —— 标签已经写进
    目标系统了,把链接标成 stale 只会让「这个标是怎么来的」失去解释,
    而标签本身还在。撤销标签是人的动作,不是同步的动作。
    """
    # ``before`` 向下取整到秒。``last_seen_at`` 是不带 fsp 的 MySQL
    # DATETIME,写 36.2 秒读回来是 36 秒 —— 一条本轮**刚刚**被匹配到的
    # 链接因此可能读出一个比 ``before``(36.1)更早的值,然后被误判成
    # 「本轮没再出现」而标成 stale。取整之后,任何 raw >= before 的写入
    # 读回来都 >= floor(before),这个误判就没有了。真正过期的链接
    # last_seen_at 来自上一轮,差着小时级,不受影响。
    cutoff = before.replace(microsecond=0)
    rows = (await db.execute(
        select(PartnerDirectoryLink).where(
            PartnerDirectoryLink.integration_id == integration_id,
            PartnerDirectoryLink.state == LINK_PENDING_REVIEW,
            PartnerDirectoryLink.last_seen_at < cutoff,
        ),
    )).scalars().all()
    count = 0
    for row in rows:
        row.state = LINK_STALE
        count += 1
    if count:
        await db.flush()
    return count


# ── 读面 ───────────────────────────────────────────────────────────


async def overview(db: AsyncSession) -> dict[str, Any]:
    """每个连接器一行的概览。"""
    integrations = await list_partner_integrations(db)
    out: list[dict[str, Any]] = []
    for integration in integrations:
        iid = str(integration.id)
        total = (await db.execute(
            select(func.count()).select_from(PartnerDirectoryMember).where(
                PartnerDirectoryMember.integration_id == iid,
                PartnerDirectoryMember.deleted_at.is_(None),
            ),
        )).scalar_one()
        active = (await db.execute(
            select(func.count()).select_from(PartnerDirectoryMember).where(
                PartnerDirectoryMember.integration_id == iid,
                PartnerDirectoryMember.deleted_at.is_(None),
                PartnerDirectoryMember.is_active.is_(True),
            ),
        )).scalar_one()
        with_github = (await db.execute(
            select(func.count()).select_from(PartnerDirectoryMember).where(
                PartnerDirectoryMember.integration_id == iid,
                PartnerDirectoryMember.deleted_at.is_(None),
                PartnerDirectoryMember.github_login_lower.is_not(None),
            ),
        )).scalar_one()
        last_seen = (await db.execute(
            select(func.max(PartnerDirectoryMember.last_seen_at)).where(
                PartnerDirectoryMember.integration_id == iid,
            ),
        )).scalar_one()
        by_state: dict[str, int] = {}
        rows = (await db.execute(
            select(
                PartnerDirectoryLink.state, func.count(),
            ).where(
                PartnerDirectoryLink.integration_id == iid,
            ).group_by(PartnerDirectoryLink.state),
        )).all()
        for state, count in rows:
            by_state[state] = count
        by_surface: dict[str, int] = {}
        rows = (await db.execute(
            select(
                PartnerDirectoryLink.surface, func.count(),
            ).where(
                PartnerDirectoryLink.integration_id == iid,
                PartnerDirectoryLink.state == LINK_APPLIED,
            ).group_by(PartnerDirectoryLink.surface),
        )).all()
        for surface, count in rows:
            by_surface[surface] = count
        out.append({
            "integration_id": iid,
            "slug": integration.slug,
            "display_name": integration.name,
            "enabled": integration.enabled,
            "last_test_status": integration.last_test_status,
            "last_test_message": integration.last_test_message,
            "members_total": total,
            "members_active": active,
            "members_with_github": with_github,
            "last_synced_at": last_seen.isoformat() if last_seen else None,
            "links_by_state": by_state,
            "applied_by_surface": by_surface,
        })
    return {"integrations": out}


async def list_members(
    db: AsyncSession,
    *,
    integration_id: str | None = None,
    q: str | None = None,
    only_active: bool | None = None,
    has_github: bool | None = None,
    include_deleted: bool = False,
    page: int = 1,
    page_size: int = 50,
) -> dict[str, Any]:
    stmt = select(PartnerDirectoryMember)
    count_stmt = select(func.count()).select_from(PartnerDirectoryMember)
    conditions = []
    if integration_id:
        conditions.append(PartnerDirectoryMember.integration_id == integration_id)
    if not include_deleted:
        conditions.append(PartnerDirectoryMember.deleted_at.is_(None))
    if only_active is not None:
        conditions.append(PartnerDirectoryMember.is_active.is_(only_active))
    if has_github is True:
        conditions.append(PartnerDirectoryMember.github_login_lower.is_not(None))
    elif has_github is False:
        conditions.append(PartnerDirectoryMember.github_login_lower.is_(None))
    if q:
        needle = f"%{q.strip().lower()}%"
        conditions.append(or_(
            func.lower(PartnerDirectoryMember.name).like(needle),
            func.lower(PartnerDirectoryMember.en_name).like(needle),
            func.lower(PartnerDirectoryMember.email).like(needle),
            func.lower(PartnerDirectoryMember.github_login).like(needle),
            func.lower(PartnerDirectoryMember.employee_no).like(needle),
        ))
    for condition in conditions:
        stmt = stmt.where(condition)
        count_stmt = count_stmt.where(condition)

    total = (await db.execute(count_stmt)).scalar_one()
    rows = (await db.execute(
        stmt.order_by(
            PartnerDirectoryMember.is_active.desc(),
            PartnerDirectoryMember.email.asc(),
        ).offset(max(page - 1, 0) * page_size).limit(page_size),
    )).scalars().all()
    return {
        "items": [member_json(r) for r in rows],
        "total": total,
        "page": page,
        "page_size": page_size,
    }


def member_json(row: PartnerDirectoryMember) -> dict[str, Any]:
    return {
        "id": row.id,
        "integration_id": row.integration_id,
        "external_id": row.external_id,
        "name": row.name,
        "en_name": row.en_name,
        "email": row.email,
        "email_name": row.email_name,
        "secondary_email": row.secondary_email,
        "employee_no": row.employee_no,
        "job_title": row.job_title,
        "department_path": row.department_path or [],
        "github_login": row.github_login,
        "is_active": row.is_active,
        "is_suspended": row.is_suspended,
        "last_seen_at": row.last_seen_at.isoformat() if row.last_seen_at else None,
        "deleted_at": row.deleted_at.isoformat() if row.deleted_at else None,
    }


async def list_links(
    db: AsyncSession,
    *,
    integration_id: str | None = None,
    state: str | None = None,
    surface: str | None = None,
    page: int = 1,
    page_size: int = 50,
) -> dict[str, Any]:
    stmt = select(PartnerDirectoryLink, PartnerDirectoryMember).join(
        PartnerDirectoryMember,
        PartnerDirectoryLink.member_id == PartnerDirectoryMember.id,
        isouter=True,
    )
    count_stmt = select(func.count()).select_from(PartnerDirectoryLink)
    conditions = []
    if integration_id:
        conditions.append(PartnerDirectoryLink.integration_id == integration_id)
    if state:
        conditions.append(PartnerDirectoryLink.state == state)
    if surface:
        conditions.append(PartnerDirectoryLink.surface == surface)
    for condition in conditions:
        stmt = stmt.where(condition)
        count_stmt = count_stmt.where(condition)

    total = (await db.execute(count_stmt)).scalar_one()
    rows = (await db.execute(
        stmt.order_by(PartnerDirectoryLink.updated_at.desc())
        .offset(max(page - 1, 0) * page_size)
        .limit(page_size),
    )).all()
    return {
        "items": [_link_json(link, member) for link, member in rows],
        "total": total,
        "page": page,
        "page_size": page_size,
    }


def _link_json(
    link: PartnerDirectoryLink, member: PartnerDirectoryMember | None,
) -> dict[str, Any]:
    return {
        "id": link.id,
        "surface": link.surface,
        "target_key": link.target_key,
        "target_label": link.target_label,
        "match_method": link.match_method,
        "state": link.state,
        "applied_at": link.applied_at.isoformat() if link.applied_at else None,
        "apply_skipped_reason": link.apply_skipped_reason,
        "note": link.note,
        "member": member_json(member) if member is not None else None,
        "updated_at": link.updated_at.isoformat() if link.updated_at else None,
    }


async def review_link(
    db: AsyncSession,
    *,
    link_id: str,
    decision: str,
    actor_person_id: str | None,
    trace_id: str,
    note: str | None = None,
) -> dict[str, Any]:
    """人对一条待确认匹配拍板。

    ``confirm`` 走和自动路径完全相同的写入函数 —— 人确认的标签和机器
    打的标签在目标系统里必须是同一种东西,否则「已确认」和「自动」两
    条路会在离职豁免那边表现不同。
    """
    from app.core.coordinators.partner_tag_coordinator import (
        apply_reviewed_link,
    )

    link = (await db.execute(
        select(PartnerDirectoryLink).where(PartnerDirectoryLink.id == link_id),
    )).scalar_one_or_none()
    if link is None:
        raise NotFoundError("PartnerDirectoryLink", link_id)

    now = _now()
    link.reviewed_by = actor_person_id
    link.reviewed_at = now
    if note:
        link.note = note[:2000]

    if decision == "dismiss":
        link.state = LINK_DISMISSED
        await db.flush()
        return {"id": link.id, "state": link.state}
    if decision != "confirm":
        raise AppException(
            code="INVALID_DECISION",
            message="decision must be 'confirm' or 'dismiss'",
            status_code=400,
        )

    applied, reason = await apply_reviewed_link(
        db, link, trace_id=trace_id,
    )
    link.state = LINK_APPLIED
    link.apply_skipped_reason = reason
    if applied:
        link.applied_at = link.applied_at or now
    await db.flush()
    return {
        "id": link.id,
        "state": link.state,
        "applied": applied,
        "reason": reason,
    }