"""Read-only Feishu (Lark) 考勤 / attendance client.

Wraps the shared :class:`~app.approval.feishu_client.FeishuApprovalClient`
so we reuse its process-wide ``tenant_access_token`` cache and uniform
error-code handling, and only add the attendance-specific endpoints:

* :meth:`batch_resolve_user_ids` — work-email → Feishu ``user_id``
  (the ``employee_id`` the attendance API keys on), chunked + cached.
* :meth:`query_user_tasks` — ``POST /attendance/v1/user_tasks/query``,
  the per-day 打卡结果 (check-in/out result + supplement) that the
  monthly report is built from.

Helix never *writes* attendance. 补卡 happens in Feishu; we re-pull.

The attendance scope (``attendance:task:readonly``) is separate from the
contact/approval scopes the shared app already holds — until it is
granted and the app is added to the relevant 考勤组 management scope,
``query_user_tasks`` raises :class:`AttendanceFeishuError` (a clean 503)
rather than crashing the generation run.
"""
from __future__ import annotations

import logging
from dataclasses import dataclass, field
from datetime import date, datetime, timedelta

from sqlalchemy.ext.asyncio import AsyncSession

from app.approval.feishu_client import (
    FeishuApprovalClient,
    is_systemic_feishu_failure,
)
from app.core.exceptions import AppException
from app.domains.attendance.cycle import COMPANY_TZ
from app.integrations.compat import get_feishu_config

logger = logging.getLogger(__name__)

# Feishu caps user_tasks/query (and batch_get_id) at 50 users per call.
_FEISHU_USER_CHUNK = 50


class AttendanceFeishuError(AppException):
    """Raised when the Feishu attendance API can't be reached / read —
    typically a missing ``attendance:task:readonly`` scope or the app not
    being in the 考勤组 management scope. Carries a 503 so the generation
    flow degrades to a clear "couldn't pull from Feishu" rather than a
    500."""

    def __init__(self, message: str) -> None:
        super().__init__(
            code="ATTENDANCE_FEISHU_UNAVAILABLE",
            message=message,
            status_code=503,
        )


@dataclass
class UserTasksResult:
    """Aggregated result of a (possibly multi-chunk) user_tasks query.

    ``results`` are raw Feishu ``user_task_result`` dicts (one per
    user×day). ``invalid_user_ids`` / ``unauthorized_user_ids`` are the
    ids Feishu reported it could not resolve or was not permitted to read
    — surfaced upward as data-quality, never silently dropped.
    """

    results: list[dict] = field(default_factory=list)
    invalid_user_ids: list[str] = field(default_factory=list)
    unauthorized_user_ids: list[str] = field(default_factory=list)


def _chunks(items: list[str], size: int) -> list[list[str]]:
    return [items[i : i + size] for i in range(0, len(items), size)]


def _yyyymmdd(d: date) -> int:
    return int(d.strftime("%Y%m%d"))


# user_approvals/query caps the window at 30 days; the 21→20 cycle is 31,
# so split into ≤30-day windows.
def _date_windows(d_from: date, d_to: date, max_days: int = 30) -> list[tuple[date, date]]:
    out: list[tuple[date, date]] = []
    cur = d_from
    while cur <= d_to:
        wend = min(d_to, cur + timedelta(days=max_days - 1))
        out.append((cur, wend))
        cur = wend + timedelta(days=1)
    return out


def _approval_date(val: object) -> str | None:
    """Feishu approval start_time/end_time → ISO date. Accepts a
    ``"yyyy-MM-dd HH:mm:ss"`` string or an epoch (s/ms)."""
    if val is None:
        return None
    s = str(val).strip()
    if not s:
        return None
    if "-" in s:  # date-ish string
        return s[:10]
    try:
        n = int(s)
    except ValueError:
        return None
    if n > 10_000_000_000:
        n //= 1000
    try:
        return datetime.fromtimestamp(n, tz=COMPANY_TZ).date().isoformat()
    except (ValueError, OSError, OverflowError):
        return None


def _i18n_zh(obj: object) -> str:
    """Pull the Chinese label out of a Feishu ``i18n_names`` object."""
    if isinstance(obj, dict):
        for k in ("ch", "zh", "zh_cn", "zh-CN"):
            v = obj.get(k)
            if isinstance(v, str) and v.strip():
                return v.strip()
        for v in obj.values():
            if isinstance(v, str) and v.strip():
                return v.strip()
    elif isinstance(obj, str):
        return obj.strip()
    return ""


def _approval_epoch(val: object) -> int | None:
    """Preserve approval time bounds; date-only values cannot prove partial leave."""
    try:
        s = str(val).strip()
        if "-" in s:
            if len(s) <= 10:
                return None
            dt = datetime.fromisoformat(s)
            if dt.tzinfo is None:
                dt = dt.replace(tzinfo=COMPANY_TZ)
            n = int(dt.timestamp())
        else:
            n = int(s)
            if n > 10_000_000_000:
                n //= 1000
        return n if n > 0 else None
    except (TypeError, ValueError, OSError, OverflowError):
        return None


def _leave_seconds(val: object) -> int | None:
    # Feishu interval is seconds regardless of unit (day/hour/half-day).
    if isinstance(val, bool) or not isinstance(val, (str, int, float)):
        return None
    try:
        n = int(val)
        return n if n > 0 else None
    except (TypeError, ValueError, OverflowError):
        return None


@dataclass
class ApprovalSpan:
    """One approved 假勤 record (leave / business-trip / outing) reduced to
    a date span + (for leave) the 假别 label, exact bounds and net duration."""

    kind: str            # 'leave' | 'trip' | 'out'
    start_date: str      # ISO
    end_date: str        # ISO
    name: str = ""       # 假别 (zh) for leaves
    start_ts: int | None = None
    end_ts: int | None = None
    leave_seconds: int | None = None  # whole approval, NOT a per-date allocation


class FeishuAttendanceClient:
    """One instance per request / generation run."""

    def __init__(self, approval_client: FeishuApprovalClient) -> None:
        self._client = approval_client

    @classmethod
    async def from_config(
        cls, *, session: AsyncSession | None = None,
    ) -> FeishuAttendanceClient:
        config = await get_feishu_config(session=session)
        return cls(
            FeishuApprovalClient(
                app_id=str(config.get("app_id") or ""),
                app_secret=str(config.get("app_secret") or ""),
            )
        )

    async def batch_resolve_user_ids(
        self, emails: list[str],
    ) -> dict[str, str]:
        """Resolve lowercased work-emails to Feishu ``user_id`` values.

        Returns ``{email_lower: user_id}`` for every email Feishu knows;
        unmapped emails are simply absent from the map (the caller treats
        their absence as a data-quality signal). Chunked at 50/call.
        """
        norm = sorted({(e or "").strip().lower() for e in emails if e and e.strip()})
        out: dict[str, str] = {}
        for chunk in _chunks(norm, _FEISHU_USER_CHUNK):
            try:
                data = await self._client.request(
                    "POST",
                    "/contact/v3/users/batch_get_id?user_id_type=user_id",
                    json_body={"emails": chunk},
                )
            except AppException as exc:
                # Contact scope is already held by the shared app, so this
                # is a transient/systemic failure, not a config gap —
                # log and continue with a partial map rather than aborting
                # the whole run.
                logger.warning(
                    "attendance: batch_get_id failed for %d emails: %s",
                    len(chunk), exc,
                )
                if is_systemic_feishu_failure(exc):
                    raise AttendanceFeishuError(
                        f"Feishu 身份解析失败（系统性错误）：{exc.message}"
                    ) from exc
                continue
            for user in data.get("user_list", []):
                email = (user.get("email") or "").lower()
                user_id = user.get("user_id")
                if email and user_id:
                    out[email] = user_id
        return out

    async def query_user_tasks(
        self,
        user_ids: list[str],
        date_from: date,
        date_to: date,
    ) -> UserTasksResult:
        """Query per-day attendance results for ``user_ids`` in a range.

        ``user_ids`` are Feishu ``user_id`` values (``employee_type=
        employee_id``). Chunked at 50/call; results aggregated. Invalid /
        unauthorized ids reported by Feishu are accumulated for
        data-quality surfacing.
        """
        agg = UserTasksResult()
        wanted = [u for u in user_ids if u]
        if not wanted:
            return agg
        for chunk in _chunks(wanted, _FEISHU_USER_CHUNK):
            try:
                data = await self._client.request(
                    "POST",
                    "/attendance/v1/user_tasks/query?employee_type=employee_id"
                    "&ignore_invalid_users=true&include_terminated_user=true",
                    json_body={
                        "user_ids": chunk,
                        "check_date_from": _yyyymmdd(date_from),
                        "check_date_to": _yyyymmdd(date_to),
                        "need_overtime_result": False,
                    },
                )
            except AppException as exc:
                # First failed chunk aborts: a scope gap / systemic error
                # affects every chunk equally, so there's no value in
                # burning rate limit on the rest. Surface as a clean 503.
                logger.warning(
                    "attendance: user_tasks query failed (%d users %s..%s): %s",
                    len(chunk), date_from, date_to, exc,
                )
                raise AttendanceFeishuError(
                    "拉取飞书考勤数据失败，请确认应用已开通「考勤」"
                    f"（attendance:task:readonly）权限并加入考勤组管理范围：{exc.message}"
                ) from exc
            for item in data.get("user_task_results") or []:
                if isinstance(item, dict):
                    agg.results.append(item)
            for uid in data.get("invalid_user_ids") or []:
                if uid:
                    agg.invalid_user_ids.append(uid)
            for uid in data.get("unauthorized_user_ids") or []:
                if uid:
                    agg.unauthorized_user_ids.append(uid)
        return agg

    # ── Intern roster via Feishu department (contact API) ───────────
    async def _find_department_open_id(self, name: str) -> str | None:
        """Resolve a department *name* to its ``open_department_id`` by
        walking the org tree from the root (``fetch_child=true``).

        Returns None when no department with that exact name exists.
        Department names are matched case-insensitively + stripped.
        """
        target = (name or "").strip().lower()
        if not target:
            return None
        page_token: str | None = None
        while True:
            qs = (
                "parent_department_id=0&fetch_child=true"
                "&department_id_type=open_department_id&page_size=50"
            )
            if page_token:
                qs += f"&page_token={page_token}"
            data = await self._client.request(
                "GET", f"/contact/v3/departments?{qs}",
            )
            for item in data.get("items") or []:
                if (item.get("name") or "").strip().lower() == target:
                    return item.get("open_department_id") or item.get("department_id")
            if not data.get("has_more"):
                return None
            page_token = data.get("page_token") or None
            if not page_token:
                return None

    async def list_intern_members(
        self, department_name: str,
    ) -> tuple[list[dict], bool]:
        """List the members of the named Feishu department.

        Returns ``(members, department_found)`` where each member is
        ``{"email": <lower>, "name": str, "open_id": str}``. Recurses into
        sub-departments (``find_by_department`` is direct-only, so we pass
        the resolved parent and rely on Feishu returning the subtree via
        the department membership semantics). Any Feishu error (missing
        ``contact:*`` scope, connectivity) raises :class:`AttendanceFeishuError`
        so generation degrades to a clean 503 rather than an empty roster
        silently masquerading as "no interns".
        """
        try:
            dept_id = await self._find_department_open_id(department_name)
            if not dept_id:
                return [], False
            members: list[dict] = []
            seen: set[str] = set()
            page_token: str | None = None
            while True:
                qs = (
                    f"department_id={dept_id}"
                    "&department_id_type=open_department_id"
                    "&user_id_type=open_id&page_size=50"
                )
                if page_token:
                    qs += f"&page_token={page_token}"
                data = await self._client.request(
                    "GET", f"/contact/v3/users/find_by_department?{qs}",
                )
                for u in data.get("items") or []:
                    open_id = u.get("open_id") or ""
                    if open_id and open_id in seen:
                        continue
                    if open_id:
                        seen.add(open_id)
                    email = (
                        u.get("enterprise_email") or u.get("email") or ""
                    ).strip().lower()
                    members.append({
                        "email": email,
                        "name": u.get("name") or "",
                        "open_id": open_id,
                    })
                if not data.get("has_more"):
                    break
                page_token = data.get("page_token") or None
                if not page_token:
                    break
            return members, True
        except AppException as exc:
            logger.warning(
                "attendance: feishu department roster failed for %r: %s",
                department_name, exc,
            )
            raise AttendanceFeishuError(
                "解析飞书实习生部门成员失败，请确认应用已开通通讯录读取权限"
                "（contact:department.base:readonly + contact:user.base:readonly）"
                f"且该部门在应用可用范围内：{exc.message}"
            ) from exc

    async def query_user_approvals(
        self,
        user_ids: list[str],
        date_from: date,
        date_to: date,
    ) -> dict[str, list[ApprovalSpan]]:
        """Approved 假勤 (leave / trip / outing) per user over the range.

        Returns ``{user_id: [ApprovalSpan, ...]}`` (default status=approved).
        This is an **enrichment** layer for authoritative leave detection —
        if it fails (scope/transient), we degrade to whatever was already
        gathered and let the caller fall back to the打卡-result supplement,
        rather than blocking generation. Chunked at 50 users × ≤30-day
        windows (the API's hard limits)."""
        out: dict[str, list[ApprovalSpan]] = {}
        wanted = [u for u in user_ids if u]
        if not wanted:
            return out
        for uchunk in _chunks(wanted, _FEISHU_USER_CHUNK):
            for wfrom, wto in _date_windows(date_from, date_to, 30):
                try:
                    data = await self._client.request(
                        "POST",
                        "/attendance/v1/user_approvals/query?employee_type=employee_id",
                        json_body={
                            "user_ids": uchunk,
                            "check_date_from": _yyyymmdd(wfrom),
                            "check_date_to": _yyyymmdd(wto),
                        },
                    )
                except AppException as exc:
                    logger.warning(
                        "attendance: user_approvals query failed — degrading to "
                        "punch-result supplement for leave detection: %s", exc,
                    )
                    return out  # partial (or empty) — never blocks generation
                for ua in data.get("user_approvals") or []:
                    uid = ua.get("user_id")
                    if not uid:
                        continue
                    spans = out.setdefault(uid, [])
                    for lv in ua.get("leaves") or []:
                        sd = _approval_date(lv.get("start_time"))
                        ed = _approval_date(lv.get("end_time"))
                        if sd and ed:
                            spans.append(ApprovalSpan(
                                "leave", sd, ed, _i18n_zh(lv.get("i18n_names")),
                                start_ts=_approval_epoch(lv.get("start_time")),
                                end_ts=_approval_epoch(lv.get("end_time")),
                                leave_seconds=_leave_seconds(lv.get("interval")),
                            ))
                    for o in ua.get("outs") or []:
                        sd = _approval_date(o.get("start_time"))
                        ed = _approval_date(o.get("end_time"))
                        if sd and ed:
                            spans.append(ApprovalSpan("out", sd, ed))
                    for tp in ua.get("trips") or []:
                        sd = _approval_date(tp.get("start_time"))
                        ed = _approval_date(tp.get("end_time"))
                        if sd and ed:
                            spans.append(ApprovalSpan("trip", sd, ed))
        return out