#!/usr/bin/env python3
"""
考勤报表导出脚本 — 公共模块

[AI Agent 强制门禁] 本模块不可单独执行，且任何调用方脚本
   （attendance_report_detail/monthly/daily.py）执行前都必须先阅读：
   references/attendance-report.md

   工作流细节、报表类型判断、人员获取、列选择、错误处理等约束
   全部在 attendance-report.md，禁止凭本脚本源码或 --help 自行组装命令。

被 attendance_report_detail.py / attendance_report_monthly.py /
attendance_report_daily.py 三个粒度脚本共享。

职责：
  1. dws CLI 调用（run_dws / run_dws_raw）
  2. 接口分批 / 切片（chunk_users / slice_date_range）
  3. dws 返回值通用解析（unwrap_result / extract_records）
  4. 字段（columns）模糊匹配（match_columns_by_keywords）
  5. userId → name 映射（resolve_user_names）
  6. Excel 写入（write_excel）
  7. 错误处理 / stderr 进度日志

约束：
  - 不依赖 dws 命令的具体业务字段（除接口顶层 success/result/error）
  - 业务字段解析全部由各粒度脚本负责
  - 时间字段单位由各粒度脚本自行处理（attendance 接口多为毫秒时间戳）
"""

from __future__ import annotations

import hashlib
import json
import os
import subprocess
import sys
import tempfile
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from typing import Any, Iterable

# ─────────────────────────────────────────────────────────────────────────────
# 常量：dws 接口限制（来自 attendance.md）
# ─────────────────────────────────────────────────────────────────────────────

MAX_USERS_PER_BATCH = 5             # report query-data: --users 最多 5 人（降低单批人数避免超时）
MAX_DAYS_PER_SLICE = 32             # report query-data: --start 到 --end ≤ 32 天
DWS_TIMEOUT_SECONDS = 120           # 单次 dws 调用超时
DATETIME_FMT = "%Y-%m-%d %H:%M:%S"
DATE_FMT = "%Y-%m-%d"


# ─────────────────────────────────────────────────────────────────────────────
# stderr 日志（脚本所有进度信息打 stderr，stdout 留给最终摘要）
# ─────────────────────────────────────────────────────────────────────────────

def log(msg: str) -> None:
    """打印进度信息到 stderr，stdout 保留给最终摘要。"""
    print(msg, file=sys.stderr, flush=True)


def warn(msg: str) -> None:
    log(f"[WARN] {msg}")


def error(msg: str) -> None:
    log(f"[ERROR] {msg}")


# ─────────────────────────────────────────────────────────────────────────────
# dws 调用
# ─────────────────────────────────────────────────────────────────────────────

class DwsCallError(Exception):
    """dws 调用失败（含进程退出非零、超时、JSON 解析失败、业务 success=false）。"""

    def __init__(self, message: str, *, is_permission_error: bool = False) -> None:
        super().__init__(message)
        self.is_permission_error = is_permission_error


def _looks_like_permission_error(text: str) -> bool:
    """启发式：判断错误文本是否属于权限/管理员问题。"""
    if not text:
        return False
    lower = text.lower()
    keywords = ("403", "permission", "denied", "unauthorized",
                "forbidden", "无权限", "权限不足", "管理员")
    return any(k in lower for k in keywords)


def run_dws(args: list[str]) -> Any:
    """
    调用 `dws <args> --format json`，返回解析后的 JSON。

    自动追加 `--format json`（如果调用方没传），并解开顶层 `success/result/error`：
      - success=True  → 返回 result 内容
      - success=False → 抛 DwsCallError（含 is_permission_error 标记）
      - 进程退出非零或解析失败 → 抛 DwsCallError

    注意：本函数仅做"顶层解包"，业务字段解析由调用方负责。
    """
    if "--format" not in args:
        args = args + ["--format", "json"]

    cmd = ["dws"] + args
    try:
        result = subprocess.run(
            cmd,
            capture_output=True,
            text=True,
            timeout=DWS_TIMEOUT_SECONDS,
        )
    except subprocess.TimeoutExpired as e:
        raise DwsCallError(f"dws 调用超时（{DWS_TIMEOUT_SECONDS}s）：{' '.join(cmd)}") from e
    except FileNotFoundError as e:
        raise DwsCallError("未找到 dws 命令，请确认 dws CLI 已安装并在 PATH 中") from e

    stdout = result.stdout or ""
    stderr = result.stderr or ""

    if result.returncode != 0:
        is_perm = _looks_like_permission_error(stderr) or _looks_like_permission_error(stdout)
        raise DwsCallError(
            f"dws 调用失败（exit={result.returncode}）: {stderr.strip() or stdout.strip()}",
            is_permission_error=is_perm,
        )

    try:
        data = json.loads(stdout)
    except json.JSONDecodeError as e:
        raise DwsCallError(f"dws 返回非 JSON：{stdout[:200]!r}") from e

    return unwrap_result(data)


def unwrap_result(data: Any) -> Any:
    """
    解开 dws 返回的顶层 `{success, result, error}` 包装。

    success=True  → 返回 result（可能是 dict / list / None）
    success=False → 抛 DwsCallError
    其他形态     → 原样返回（兼容部分接口直接返回数据）
    """
    if not isinstance(data, dict):
        return data

    if "success" not in data:
        # 不是标准包装，原样返回
        return data

    if data.get("success") is True:
        return data.get("result")

    # success = False
    err = data.get("error") or {}
    if isinstance(err, dict):
        msg = err.get("message") or err.get("msg") or json.dumps(err, ensure_ascii=False)
    else:
        msg = str(err)
    raise DwsCallError(
        f"dws 业务失败：{msg}",
        is_permission_error=_looks_like_permission_error(msg),
    )


# ─────────────────────────────────────────────────────────────────────────────
# 通用记录提取（兼容多种数据嵌套形态）
# ─────────────────────────────────────────────────────────────────────────────

def extract_records(payload: Any) -> list[dict]:
    """
    从 dws 返回的 result 中提取"记录数组"。

    兼容多种常见嵌套：
      - 直接是 list[dict]                → 原样返回
      - {"data": [...]}                  → 取 data
      - {"records": [...]}               → 取 records
      - {"list": [...]}                  → 取 list
      - {"items": [...]}                 → 取 items
      - {"result": [...]} (双层包装)     → 递归一次
      - 其他 dict 但只有一个值是 list    → 取那个 list
      - 其他形态                         → 返回 []，并 warn

    业务字段不在本函数关心范围内。
    """
    if payload is None:
        return []
    if isinstance(payload, list):
        return [item for item in payload if isinstance(item, dict)]
    if isinstance(payload, dict):
        for key in ("data", "records", "list", "items", "result"):
            if key in payload and isinstance(payload[key], list):
                return [item for item in payload[key] if isinstance(item, dict)]
        # 兜底：dict 中只有一个 list 值
        list_values = [v for v in payload.values() if isinstance(v, list)]
        if len(list_values) == 1:
            return [item for item in list_values[0] if isinstance(item, dict)]
        warn(f"未能从返回中识别记录数组，顶层 keys={list(payload.keys())}")
        return []
    warn(f"未能识别返回类型：{type(payload).__name__}")
    return []


def flatten_query_data_records(
    records: list[dict],
    column_id_to_name: dict[str, str] | None = None,
) -> list[dict]:
    """
    展平 report query-data 返回的嵌套 values 结构。

    接口原始格式：
      {"userId":"xxx", "values":[{"termId":"173410778","value":"1"}, ...], "workDate":"2026-05-01"}

    展平后：
      {"userId":"xxx", "workDate":"2026-05-01", "173410778":"1", "节假日+出勤":"1", ...}

    Args:
        records: extract_records 返回的原始记录列表
        column_id_to_name: 可选的 columnId → columnName 映射，展平时同时写入字段名 key
    """
    flattened: list[dict] = []
    for record in records:
        values_list = record.get("values")
        if not isinstance(values_list, list):
            # 已经是平铺格式或无 values 字段，原样保留
            flattened.append(record)
            continue
        flat: dict[str, Any] = {}
        # 保留顶层非 values 字段（userId, workDate, corpId 等）
        for k, v in record.items():
            if k != "values":
                flat[k] = v
        # 展平 values 数组
        for entry in values_list:
            if not isinstance(entry, dict):
                continue
            term_id = str(entry.get("termId", entry.get("columnId", entry.get("id", ""))))
            value = entry.get("value", entry.get("data", ""))
            if term_id:
                flat[term_id] = value
                # 同时写入字段名 key（方便按名称取值）
                if column_id_to_name and term_id in column_id_to_name:
                    flat[column_id_to_name[term_id]] = value
        flattened.append(flat)
    return flattened


def extract_group_names_from_records(
    records: list[dict],
    user_ids: list[str],
) -> dict[str, str]:
    """
    获取每个用户的考勤组名称。

    优先从 report query-data 原始记录中提取（如果接口返回了 groupName 字段），
    否则回退到通过 `dws attendance group search` + `filtered-get --member`
    获取所有考勤组的成员列表，反向映射 userId → 考勤组名称。

    返回 {userId: groupName} 映射，未找到的用户映射为空字符串。
    """
    group_map: dict[str, str] = {}
    candidate_keys = ("groupName", "group_name", "attendanceGroupName",
                      "groupId", "group_id")

    # 1) 先尝试从原始记录中提取
    for record in records:
        uid = _first_nonempty(record, ("userId", "userid", "user_id", "targetUserId"))
        if uid is None:
            continue
        uid_str = str(uid)
        if uid_str in group_map:
            continue
        name = _first_nonempty(record, candidate_keys)
        if name is not None and str(name).strip():
            group_map[uid_str] = str(name).strip()

    # 2) 如果还有用户未匹配到考勤组，通过 group API 反向查找
    missing_uids = {uid for uid in user_ids if uid not in group_map}
    if missing_uids:
        api_map = _resolve_group_names_via_api(missing_uids)
        group_map.update(api_map)

    # 兜底：未找到的用户填空字符串
    for uid in user_ids:
        if uid not in group_map:
            group_map[uid] = ""

    return group_map


def _resolve_group_names_via_api(target_uids: set[str]) -> dict[str, str]:
    """
    通过 dws attendance group search + filtered-get --member
    反向映射 userId → 考勤组名称。

    流程：
      1. group search 获取所有考勤组（id + name）
      2. 对每个考勤组调用 filtered-get --member 获取成员 userId 列表
      3. 将 target_uids 中的用户与考勤组成员做交集映射
    """
    result_map: dict[str, str] = {}
    if not target_uids:
        return result_map

    # 获取所有考勤组
    try:
        search_payload = run_dws([
            "attendance", "group", "search",
            "--limit", "200",
        ])
    except DwsCallError as e:
        log(f"[group] 获取考勤组列表失败：{e}")
        return result_map

    search_result = unwrap_result(search_payload)
    groups: list[dict] = []
    if isinstance(search_result, dict):
        groups = search_result.get("items", [])
    elif isinstance(search_result, list):
        groups = search_result

    if not groups:
        log("[group] 未获取到任何考勤组")
        return result_map

    log(f"[group] 共 {len(groups)} 个考勤组，开始查询成员列表")

    remaining = set(target_uids)
    for group in groups:
        if not remaining:
            break
        group_id = group.get("id")
        group_name = group.get("name", "")
        if not group_id:
            continue

        try:
            detail_payload = run_dws([
                "attendance", "group", "filtered-get",
                "--group-id", str(group_id),
                "--member",
            ])
        except DwsCallError as e:
            log(f"[group] filtered-get 失败 (group={group_name}): {e}")
            continue

        detail = unwrap_result(detail_payload)
        if not isinstance(detail, dict):
            continue

        member_users = detail.get("memberUsers", [])
        if not isinstance(member_users, list):
            continue

        for member_uid in member_users:
            uid_str = str(member_uid)
            if uid_str in remaining:
                result_map[uid_str] = group_name
                remaining.discard(uid_str)

    if remaining:
        log(f"[group] {len(remaining)} 个用户未匹配到考勤组")

    return result_map


def dump_first_record_for_inspection(records: list[dict], label: str) -> None:
    """
    第一次跑脚本时，把第一条记录打到 stderr，方便用户/开发者
    看清真实字段结构后回来调优解析逻辑。
    """
    if not records:
        log(f"[inspect:{label}] 无记录")
        return
    sample = records[0]
    log(f"[inspect:{label}] 首条记录字段示例（用于核对真实结构）：")
    log(json.dumps(sample, ensure_ascii=False, indent=2))


# ─────────────────────────────────────────────────────────────────────────────
# 时间区间切片
# ─────────────────────────────────────────────────────────────────────────────

@dataclass
class DateSlice:
    start: datetime  # 含
    end: datetime    # 含

    @property
    def start_str(self) -> str:
        return self.start.strftime(DATETIME_FMT)

    @property
    def end_str(self) -> str:
        return self.end.strftime(DATETIME_FMT)

    @property
    def label(self) -> str:
        return f"{self.start.strftime(DATE_FMT)}~{self.end.strftime(DATE_FMT)}"


def parse_datetime_arg(s: str, *, end_of_day: bool = False) -> datetime:
    """
    解析用户输入的日期参数，支持：
      - YYYY-MM-DD                  → 00:00:00 或 23:59:59（取决于 end_of_day）
      - YYYY-MM-DD HH:mm:ss
    """
    s = s.strip()
    try:
        return datetime.strptime(s, DATETIME_FMT)
    except ValueError:
        pass
    try:
        d = datetime.strptime(s, DATE_FMT)
        if end_of_day:
            return d.replace(hour=23, minute=59, second=59)
        return d
    except ValueError as e:
        raise ValueError(
            f"无法解析日期 {s!r}，请使用 YYYY-MM-DD 或 YYYY-MM-DD HH:mm:ss"
        ) from e


def slice_date_range(
    start: datetime,
    end: datetime,
    max_days: int = MAX_DAYS_PER_SLICE,
) -> list[DateSlice]:
    """
    把 [start, end] 切成多个 ≤ max_days 天的小区间。

    每片含起含止；最后一片可能短于 max_days。
    """
    if end < start:
        raise ValueError(f"结束时间 {end} 早于开始时间 {start}")

    slices: list[DateSlice] = []
    cur_start = start
    while cur_start <= end:
        # 这片的最晚结束时间（不超 max_days，且不超 end）
        cur_end_limit = cur_start + timedelta(days=max_days - 1)
        cur_end_limit = cur_end_limit.replace(hour=23, minute=59, second=59)
        cur_end = min(cur_end_limit, end)
        slices.append(DateSlice(start=cur_start, end=cur_end))
        # 下一片从次日 00:00 开始
        next_day = (cur_end + timedelta(seconds=1)).replace(
            hour=0, minute=0, second=0, microsecond=0
        )
        cur_start = next_day
    return slices


# ─────────────────────────────────────────────────────────────────────────────
# 用户分批
# ─────────────────────────────────────────────────────────────────────────────

def chunk_users(users: list[str], size: int = MAX_USERS_PER_BATCH) -> list[list[str]]:
    """把 userId 列表切成每片 ≤ size 的小批。"""
    if size <= 0:
        raise ValueError(f"size 必须 > 0，得到 {size}")
    return [users[i: i + size] for i in range(0, len(users), size)]


# ─────────────────────────────────────────────────────────────────────────────
# columns 模糊匹配
# ─────────────────────────────────────────────────────────────────────────────

def match_columns_by_keywords(
    all_columns: list[dict],
    keywords: list[str],
    *,
    name_keys: tuple[str, ...] = ("name", "columnName", "title", "label"),
    id_keys: tuple[str, ...] = ("id", "columnId", "code", "key"),
) -> list[dict]:
    """
    在 report columns 返回的字段列表里，按关键词匹配目标字段。

    匹配策略（按优先级）：
      1. 精确匹配：关键词 == 字段名（优先）
      2. 子串匹配：关键词是字段名的子串（仅当精确匹配无结果时回退）

    列名严格使用接口返回的原始字段名，不做任何修改。
    """
    matched: list[dict] = []
    matched_ids: set[str] = set()
    hit_keywords: set[str] = set()

    # 构建 name → (col_dict, cid_str) 索引
    col_index: list[tuple[str, str, dict]] = []
    for col in all_columns:
        name = _first_nonempty(col, name_keys)
        cid = _first_nonempty(col, id_keys)
        if not name or cid is None:
            continue
        col_index.append((str(name), str(cid), col))

    for kw in keywords:
        kw_stripped = kw.strip()
        if not kw_stripped:
            continue

        # 第一轮：精确匹配
        exact_hit = False
        for name, cid_str, col in col_index:
            if name == kw_stripped and cid_str not in matched_ids:
                enriched = dict(col)
                enriched["_column_id"] = cid_str
                enriched["_column_name"] = name
                matched.append(enriched)
                matched_ids.add(cid_str)
                hit_keywords.add(kw_stripped)
                exact_hit = True
                break

        if exact_hit:
            continue

        # 第二轮：子串匹配（回退），只取第一个命中
        kw_lower = kw_stripped.lower()
        for name, cid_str, col in col_index:
            if kw_lower in name.lower() and cid_str not in matched_ids:
                enriched = dict(col)
                enriched["_column_id"] = cid_str
                enriched["_column_name"] = name
                matched.append(enriched)
                matched_ids.add(cid_str)
                hit_keywords.add(kw_stripped)
                break

    missing = set(k.strip() for k in keywords if k.strip()) - hit_keywords
    if missing:
        warn(f"以下关键词未匹配到任何字段，已跳过：{sorted(missing)}")
    return matched


def _first_nonempty(d: dict, keys: Iterable[str]) -> Any:
    for k in keys:
        if k in d and d[k] not in (None, ""):
            return d[k]
    return None


# ─────────────────────────────────────────────────────────────────────────────
# userId → name 映射
# ─────────────────────────────────────────────────────────────────────────────

def resolve_users_from_input(raw_ids: list[str]) -> list[str]:
    """
    智能解析 --users 输入：自动区分部门ID和员工userId。

    Wukong Agent 经常把部门ID当 userId 传入。本函数尝试对每个ID调用
    `dws contact dept list-members` 获取成员列表：
      - 如果成功且返回了员工，说明该ID是部门ID，展开为员工userId列表
      - 如果失败或无结果，说明该ID本身就是userId，原样保留

    最终返回去重后的 userId 列表。
    """
    if not raw_ids:
        return []

    resolved: list[str] = []
    seen: set[str] = set()

    # 先尝试批量查部门成员（可能全是部门ID）
    try:
        result = run_dws([
            "contact", "dept", "list-members",
            "--ids", ",".join(raw_ids),
        ])
        members = extract_records(result)
        if members:
            # 成功获取到成员 → 输入是部门ID
            for member in members:
                uid = _first_nonempty(member, ("userId", "userid", "id"))
                if uid and str(uid) not in seen:
                    resolved.append(str(uid))
                    seen.add(str(uid))
            if resolved:
                log(f"[users] 检测到输入为部门ID，已展开为 {len(resolved)} 个员工userId")
                return resolved
    except DwsCallError:
        pass

    # 逐个ID尝试：可能混合了部门ID和userId
    for raw_id in raw_ids:
        if raw_id in seen:
            continue
        try:
            result = run_dws([
                "contact", "dept", "list-members",
                "--ids", raw_id,
            ])
            members = extract_records(result)
            if members:
                for member in members:
                    uid = _first_nonempty(member, ("userId", "userid", "id"))
                    if uid and str(uid) not in seen:
                        resolved.append(str(uid))
                        seen.add(str(uid))
                log(f"[users] 部门ID {raw_id} 展开为 {len(members)} 个员工")
                continue
        except DwsCallError:
            pass
        # 不是部门ID，当作userId保留
        if raw_id not in seen:
            resolved.append(raw_id)
            seen.add(raw_id)

    return resolved


@dataclass
class UserInfo:
    """用户基础信息，用于报表的姓名/部门/工号/职位列。"""
    name: str = ""
    dept_name: str = ""
    job_number: str = ""
    title: str = ""


def _extract_title_from_labels(labels: list) -> str:
    """从 orgEmployeeModel.labels 数组中提取职务名称。"""
    if not isinstance(labels, list):
        return ""
    for item in labels:
        if isinstance(item, dict) and item.get("groupName") == "职务":
            name = item.get("name", "")
            if name:
                return str(name)
    return ""


def _parse_user_record(record: dict) -> tuple[str, UserInfo] | None:
    """
    从 dws contact user get 返回的单条记录中解析用户信息。

    接口返回结构为嵌套格式：
      {"orgEmployeeModel": {"userId": "xxx", "orgUserName": "吾贤",
       "depts": [{"deptName": "技术部"}],
       "labels": [{"groupName": "职务", "name": "财务"}],
       ...}, "isAdmin": true}

    也兼容扁平格式（其他接口可能返回）：
      {"userId": "xxx", "name": "吾贤", "deptName": "技术部", ...}
    """
    # 优先从嵌套的 orgEmployeeModel 中提取
    model = record.get("orgEmployeeModel")
    if isinstance(model, dict):
        uid = model.get("userId") or model.get("orgUserId")
        if not uid:
            return None
        name = model.get("orgUserName") or model.get("name") or ""
        depts = model.get("depts") or []
        dept_name = depts[0].get("deptName", "") if depts and isinstance(depts[0], dict) else ""
        # 工号：尝试多个候选字段
        job_number = (model.get("jobNumber") or model.get("workNumber")
                      or model.get("empId") or "")
        # 职位：优先 title/position，回退到 labels 中 groupName=="职务" 的条目
        title = (model.get("title") or model.get("position")
                 or _extract_title_from_labels(model.get("labels", [])))
        return str(uid), UserInfo(
            name=str(name),
            dept_name=str(dept_name),
            job_number=str(job_number),
            title=str(title),
        )

    # 回退：扁平格式
    uid = _first_nonempty(record, ("userId", "userid", "id"))
    if not uid:
        return None
    return str(uid), UserInfo(
        name=str(_first_nonempty(record, ("name", "userName", "nick")) or ""),
        dept_name=str(_first_nonempty(record, ("deptName", "dept_name", "department")) or ""),
        job_number=str(_first_nonempty(record, ("jobNumber", "job_number", "workNumber")) or ""),
        title=str(_first_nonempty(record, ("title", "position", "jobTitle")) or ""),
    )


def resolve_user_info(user_ids: list[str]) -> dict[str, UserInfo]:
    """
    批量获取用户的完整基础信息（姓名、部门、工号、职位）。

    适配 dws contact user get 返回的嵌套 orgEmployeeModel 结构。
    分批处理（每批20人），失败时不抛错。
    """
    if not user_ids:
        return {}

    info_map: dict[str, UserInfo] = {}

    for i in range(0, len(user_ids), 20):
        batch = user_ids[i:i + 20]
        try:
            result = run_dws(["contact", "user", "get", "--ids", ",".join(batch)])
            for record in extract_records(result):
                parsed = _parse_user_record(record)
                if parsed:
                    uid_str, info = parsed
                    info_map[uid_str] = info
        except DwsCallError as e:
            log(f"[user-info] 批量 get 失败（batch {i // 20 + 1}）：{e}")

    # 兜底：未解析到的用户填充 userId 作为姓名
    for uid in user_ids:
        if uid not in info_map:
            info_map[uid] = UserInfo(name=uid)

    return info_map


def resolve_user_names(user_ids: list[str]) -> dict[str, str]:
    """
    给一组 userId 解析姓名映射（向后兼容接口）。

    内部调用 resolve_user_info，只返回 {userId: name} 映射。
    """
    info_map = resolve_user_info(user_ids)
    return {uid: info.name or uid for uid, info in info_map.items()}


# ─────────────────────────────────────────────────────────────────────────────
# Excel 写入
# ─────────────────────────────────────────────────────────────────────────────

# ─────────────────────────────────────────────────────────────────────────────
# Excel 样式常量与辅助函数
# ─────────────────────────────────────────────────────────────────────────────

# 配色（参考钉钉考勤报表风格：青绿标题 + 浅黄表头 + 白色数据区）
_TITLE_FILL_COLOR = "D5EAEA"       # 浅青绿 — 主标题背景
_TITLE_FONT_COLOR = "1F6E6E"       # 深青 — 主标题字体
_SUBTITLE_FILL_COLOR = "DAEEF3"    # 浅蓝 — 副标题（生成时间）背景
_SUBTITLE_FONT_COLOR = "31708F"    # 深蓝 — 副标题字体

_HEADER_FILL_COLOR = "FFF2CC"      # 浅黄 — 表头背景
_HEADER_FONT_COLOR = "333333"      # 深灰近黑 — 表头字体
_DATA_FONT_COLOR = "333333"        # 数据字体颜色

_BORDER_COLOR = "BFBFBF"           # 浅灰 — 单元格边框
_FONT_NAME = "微软雅黑"

# 日历表人员交替配色（奇数人白底，偶数人浅灰蓝底，便于区分不同人员）
_CALENDAR_BAND_COLORS = ("FFFFFF", "EDF2F9")

# 考勤结果单元格条件配色（参考钉钉 previewStyleByValue 配置）
# 规则按优先级排列，首个匹配命中即停止；无 color 键表示不填充背景色。
import re as _re

_ATTEND_RESULT_STYLE_RULES: list[tuple["_re.Pattern[str]", str | None]] = [
    # 白底（红字加粗由字体控制）：周末 — POI index 9 WHITE (255,255,255)
    (_re.compile(r".*(星期六|星期日|星期天)[\s\S]*"), "FFFFFF"),
    # 浅黄(TAN)：补卡审批通过/举证打卡/加班/外出/假/调休 — POI index 47 (255,204,153)
    (_re.compile(r".*(补卡审批通过|举证打卡审批通过|加班|外出|假|调休)[\s\S]*"), "FFCC99"),
    # 浅青(LIGHT_TURQUOISE)：外勤/出差 — POI index 41 (204,255,255)
    (_re.compile(r".*(外勤|出差)[\s\S]*"), "CCFFFF"),
    # 浅黄(TAN)：管理员改为正常 — POI index 47 (255,204,153)
    (_re.compile(r"(?=.*管理员)(?=.*改为正常)^[\s\S]*$"), "FFCC99"),
    # 水蓝(AQUA)：旷工迟到（须在"旷工""迟到"之前匹配）— POI index 49 (51,204,204)
    (_re.compile(r".*(旷工迟到)[\s\S]*"), "33CCCC"),
    # 玫红(ROSE)：旷工 — POI index 45 (255,153,204)
    (_re.compile(r".*(旷工)[\s\S]*"), "FF99CC"),
    # 淡蓝(PALE_BLUE)：严重迟到（须在普通"迟到"之前匹配）— POI index 44 (153,204,255)
    (_re.compile(r".*(严重迟到)[\s\S]*"), "99CCFF"),
    # 浅绿(LIGHT_GREEN)：迟到 — POI index 42 (204,255,204)
    (_re.compile(r".*(迟到)[\s\S]*"), "CCFFCC"),
    # 柠檬黄(LEMON_CHIFFON)：早退 — POI index 26 (255,255,153)
    (_re.compile(r".*(早退)[\s\S]*"), "FFFF99"),
    # 珊瑚粉(CORAL)：缺卡 — POI index 29 (255,128,128)
    (_re.compile(r".*(缺卡)[\s\S]*"), "FF8080"),
    # 无色：未排班/休息/正常 — 不填充
    (_re.compile(r".*(未排班|休息|正常)[\s\S]*"), None),
]


def _match_attend_result_color(text: str) -> str | None:
    """
    根据考勤结果文本匹配颜色（6 位 hex，不含 #），无匹配或匹配到"无色"规则时返回 None。
    """
    if not text:
        return None
    for pattern, color in _ATTEND_RESULT_STYLE_RULES:
        if pattern.fullmatch(text):
            return color
    return None

# 行高
_TITLE_ROW_HEIGHT = 30             # 主标题行高
_SUBTITLE_ROW_HEIGHT = 22          # 副标题行高
_HEADER_ROW_HEIGHT = 32            # 表头行高（更高，配合浅黄底色）
_DATA_ROW_HEIGHT = 22              # 数据行高

# 列宽估算：中文字符权重，宽度上下限
_CJK_CHAR_WEIGHT = 2.0
_ASCII_CHAR_WEIGHT = 1.1
_MIN_COL_WIDTH = 10
_MAX_COL_WIDTH = 40


def _is_cjk(ch: str) -> bool:
    """判断一个字符是否为中日韩字符（用于估算 Excel 列宽）。"""
    if not ch:
        return False
    code = ord(ch)
    return (
        0x4E00 <= code <= 0x9FFF       # CJK 统一汉字
        or 0x3000 <= code <= 0x303F    # CJK 符号和标点
        or 0xFF00 <= code <= 0xFFEF    # 全角字符
    )


def _estimate_text_width(text: str) -> float:
    """估算字符串在 Excel 中显示所占的列宽（中文 ≈ 2，ASCII ≈ 1）。"""
    if not text:
        return 0.0
    width = 0.0
    for ch in text:
        width += _CJK_CHAR_WEIGHT if _is_cjk(ch) else _ASCII_CHAR_WEIGHT
    return width


def _is_numeric_value(value: Any) -> bool:
    """判断一个值是否为可对齐到右侧的数字（或纯数字字符串）。"""
    if isinstance(value, bool):
        return False
    if isinstance(value, (int, float)):
        return True
    if isinstance(value, str):
        s = value.strip()
        if not s:
            return False
        try:
            float(s)
            return True
        except ValueError:
            return False
    return False


def _build_styles():
    """构建 Excel 样式对象集合，避免每个单元格重复创建。"""
    from openpyxl.styles import (
        Alignment, Border, Font, PatternFill, Side,
    )

    thin_side = Side(border_style="thin", color=_BORDER_COLOR)
    border = Border(left=thin_side, right=thin_side, top=thin_side, bottom=thin_side)

    return {
        # 主标题（"月度汇总展示  统计日期：xxx 至 xxx"）
        "title_font": Font(name=_FONT_NAME, bold=True, color=_TITLE_FONT_COLOR, size=14),
        "title_fill": PatternFill(fill_type="solid", fgColor=_TITLE_FILL_COLOR),
        "title_align": Alignment(horizontal="left", vertical="center", indent=1),
        # 副标题（"报表生成时间：xxx"）
        "subtitle_font": Font(name=_FONT_NAME, color=_SUBTITLE_FONT_COLOR, size=10),
        "subtitle_fill": PatternFill(fill_type="solid", fgColor=_SUBTITLE_FILL_COLOR),
        "subtitle_align": Alignment(horizontal="left", vertical="center", indent=1),
        # 表头
        "header_font": Font(name=_FONT_NAME, bold=True, color=_HEADER_FONT_COLOR, size=11),
        "header_fill": PatternFill(fill_type="solid", fgColor=_HEADER_FILL_COLOR),
        "header_align": Alignment(horizontal="center", vertical="center", wrap_text=True),
        # 数据
        "data_font": Font(name=_FONT_NAME, color=_DATA_FONT_COLOR, size=10),
        "align_left": Alignment(horizontal="left", vertical="center", wrap_text=True),
        "align_right": Alignment(horizontal="right", vertical="center", wrap_text=False),
        "align_center": Alignment(horizontal="center", vertical="center", wrap_text=True),
        "border": border,
        # 日历表人员交替配色（按 merge_groups 块交替）
        "band_fills": tuple(
            PatternFill(fill_type="solid", fgColor=c)
            for c in _CALENDAR_BAND_COLORS
        ),
    }


def _apply_sheet_styles(
    ws,
    headers: list[str],
    rows: list[list[Any]],
    styles: dict,
    *,
    title: str | None = None,
    subtitle: str | None = None,
    freeze_first_col: bool = True,
    merge_groups: list[tuple[int, int, int]] | None = None,
    attend_result_columns: set[int] | None = None,
    attend_result_rows: set[int] | None = None,
    image_columns: list[str] | None = None,
    image_size: tuple[int, int] = (80, 120),
) -> None:
    """
    给一个已写入数据的 sheet 应用统一样式（钉钉风格）。

    布局（自上而下）：
      [可选] 第 1 行：主标题（青绿底，跨所有列，写"xxx展示  统计日期：A 至 B"）
      [可选] 第 2 行：副标题（浅蓝底，跨所有列，写"报表生成时间：xxx"）
      表头行：浅黄底 + 加粗黑字 + 居中 + 加高
      数据行：白底 + 居中 + 灰色细边框

    参数:
      title:    可选主标题文本（如"月度汇总展示  统计日期：2026-01-01 至 2026-01-31"）
      subtitle: 可选副标题文本（如"报表生成时间：2026-01-20 15:46"）
      freeze_first_col: 是否冻结首列（姓名）
      attend_result_columns: 考勤结果列的 0-based 列索引集合（月度汇总场景）
      attend_result_rows:    考勤结果行的 0-based 行偏移集合（日历表场景）
    """
    from openpyxl.utils import get_column_letter

    n_cols = len(headers)
    n_rows = len(rows)
    last_col_letter = get_column_letter(n_cols) if n_cols >= 1 else "A"

    # ── 标题区（如果有）── 占据 1~2 行，跨所有列 ───────────────────────
    title_row_count = 0
    if title:
        title_row_count += 1
        title_row = title_row_count
        ws.cell(row=title_row, column=1, value=title)
        if n_cols >= 2:
            ws.merge_cells(
                start_row=title_row, start_column=1,
                end_row=title_row, end_column=n_cols,
            )
        cell = ws.cell(row=title_row, column=1)
        cell.font = styles["title_font"]
        cell.fill = styles["title_fill"]
        cell.alignment = styles["title_align"]
        ws.row_dimensions[title_row].height = _TITLE_ROW_HEIGHT
    if subtitle:
        title_row_count += 1
        sub_row = title_row_count
        ws.cell(row=sub_row, column=1, value=subtitle)
        if n_cols >= 2:
            ws.merge_cells(
                start_row=sub_row, start_column=1,
                end_row=sub_row, end_column=n_cols,
            )
        cell = ws.cell(row=sub_row, column=1)
        cell.font = styles["subtitle_font"]
        cell.fill = styles["subtitle_fill"]
        cell.alignment = styles["subtitle_align"]
        ws.row_dimensions[sub_row].height = _SUBTITLE_ROW_HEIGHT

    header_row = title_row_count + 1
    first_data_row = header_row + 1

    # ── 表头样式 ─────────────────────────────────────────────────────────
    for col_idx in range(1, n_cols + 1):
        cell = ws.cell(row=header_row, column=col_idx)
        cell.font = styles["header_font"]
        cell.fill = styles["header_fill"]
        cell.alignment = styles["header_align"]
        cell.border = styles["border"]
    ws.row_dimensions[header_row].height = _HEADER_ROW_HEIGHT

    # ── 人员交替配色映射（仅在有 merge_groups 时生效）─────────────────
    # 为每个数据行偏移预计算应使用的背景 fill（按人员块交替）
    band_fill_map: dict[int, Any] = {}
    if merge_groups:
        band_fills = styles.get("band_fills", ())
        if band_fills:
            for group_idx, (start_off, end_off, _n_base) in enumerate(merge_groups):
                fill = band_fills[group_idx % len(band_fills)]
                for off in range(start_off, end_off + 1):
                    band_fill_map[off] = fill

    # ── 考勤结果条件配色准备 ────────────────────────────────────────────
    # attend_result_columns: 月度汇总场景，指定哪些列（0-based）是考勤结果列
    # attend_result_rows:    日历表场景，指定哪些行偏移（0-based）是考勤结果行
    # 两者都需要结合日期数据列（跳过基础信息列）来判断是否需要配色
    from openpyxl.styles import PatternFill as _PF
    _attend_fill_cache: dict[str, _PF] = {}

    def _get_attend_fill(color_hex: str) -> _PF:
        """按颜色值缓存 PatternFill，避免重复创建。"""
        if color_hex not in _attend_fill_cache:
            _attend_fill_cache[color_hex] = _PF(fill_type="solid", fgColor=color_hex)
        return _attend_fill_cache[color_hex]

    _ar_cols = attend_result_columns or set()
    _ar_rows = attend_result_rows or set()

    # ── 数据行样式 ───────────────────────────────────────────────────────
    for row_offset in range(n_rows):
        excel_row = first_data_row + row_offset
        row_fill = band_fill_map.get(row_offset)
        is_attend_row = row_offset in _ar_rows
        for col_idx in range(1, n_cols + 1):
            cell = ws.cell(row=excel_row, column=col_idx)
            cell.font = styles["data_font"]
            cell.border = styles["border"]

            # 背景色优先级：考勤结果条件配色 > 人员交替配色 > 默认无色
            col_zero = col_idx - 1
            value = rows[row_offset][col_zero] if col_zero < len(rows[row_offset]) else None
            is_attend_cell = (col_zero in _ar_cols) or (is_attend_row and col_zero >= 4)
            attend_color = None
            if is_attend_cell and value not in (None, ""):
                attend_color = _match_attend_result_color(str(value))

            if attend_color:
                cell.fill = _get_attend_fill(attend_color)
            elif row_fill is not None:
                cell.fill = row_fill

            # 数据区统一居中（参考图风格），仅对长文本（>10 字符）的非首列左对齐避免拥挤
            if _is_numeric_value(value):
                cell.alignment = styles["align_right"]
            else:
                text = str(value) if value not in (None, "") else ""
                if _estimate_text_width(text) > 14 and col_idx > 1:
                    cell.alignment = styles["align_left"]
                else:
                    cell.alignment = styles["align_center"]
        ws.row_dimensions[excel_row].height = _DATA_ROW_HEIGHT

    # ── 列宽自适应 ───────────────────────────────────────────────────────
    for col_idx in range(1, n_cols + 1):
        header_text = str(headers[col_idx - 1])
        max_width = _estimate_text_width(header_text) + 2
        for row in rows:
            if col_idx - 1 < len(row):
                cell_value = row[col_idx - 1]
                if cell_value is None or cell_value == "":
                    continue
                w = _estimate_text_width(str(cell_value))
                if w > max_width:
                    max_width = w
        max_width = min(max(max_width + 1, _MIN_COL_WIDTH), _MAX_COL_WIDTH)
        ws.column_dimensions[get_column_letter(col_idx)].width = max_width

    # ── 冻结窗格 ─────────────────────────────────────────────────────────
    # 冻结到首个数据行 + 第二列（保留标题/表头/姓名列常驻）
    freeze_col_letter = "B" if (freeze_first_col and n_cols >= 2) else "A"
    ws.freeze_panes = f"{freeze_col_letter}{first_data_row}"

    # ── 自动筛选器（覆盖表头到最后一行数据）──────────────────────────
    # 注意：合并单元格时不应用 auto_filter（会和合并冲突）
    if n_cols >= 1 and n_rows >= 1 and not merge_groups:
        ws.auto_filter.ref = (
            f"A{header_row}:{last_col_letter}{first_data_row + n_rows - 1}"
        )

    # ── 单元格纵向合并（仅指定的基础列）──────────────────────────
    # merge_groups: [(start_row_offset, end_row_offset, n_base_cols), ...]
    # 其中 row_offset 是基于数据区第 0 行的偏移
    if merge_groups:
        for start_offset, end_offset, n_base_cols in merge_groups:
            if end_offset <= start_offset:
                continue
            excel_start = first_data_row + start_offset
            excel_end = first_data_row + end_offset
            for col_idx in range(1, n_base_cols + 1):
                # 取首行的值，合并后只保留首行内容
                top_value = ws.cell(row=excel_start, column=col_idx).value
                ws.merge_cells(
                    start_row=excel_start, start_column=col_idx,
                    end_row=excel_end, end_column=col_idx,
                )
                top_cell = ws.cell(row=excel_start, column=col_idx)
                top_cell.value = top_value
                top_cell.alignment = styles["align_center"]
                top_cell.border = styles["border"]

    # ── 图片嵌入（下载 URL 列指向的图片，转 PNG，嵌入对应单元格）────
    if image_columns:
        _embed_images_in_columns(
            ws, headers, rows,
            image_column_names=image_columns,
            header_row=header_row,
            first_data_row=first_data_row,
            image_size=image_size,
        )


def write_excel(
    out_path: str,
    headers: list[str],
    rows: list[list[Any]],
    *,
    sheet_name: str = "考勤报表",
    title: str | None = None,
    subtitle: str | None = None,
    image_columns: list[str] | None = None,
    image_size: tuple[int, int] = (80, 120),
) -> None:
    """
    用 openpyxl 写一个钉钉风格的美化版 Excel。

    布局：
      [可选] 第 1 行：主标题（青绿底色，跨所有列）
      [可选] 第 2 行：副标题（浅蓝底色，跨所有列，常用于"报表生成时间"）
      表头行：浅黄底色 + 黑字加粗 + 居中 + 加高
      数据行：白底 + 文本居中 / 数字右对齐 + 灰色细边框

    交互：
      - 冻结首列 + 表头（含标题区域）
      - 自动筛选器（覆盖表头到最后一行）
      - 列宽自适应（中文字符按 2 宽度估算）

    参数:
      title:    可选主标题（如"月度汇总展示  统计日期：2026-01-01 至 2026-01-31"）
      subtitle: 可选副标题（如"报表生成时间：2026-01-20 15:46"）
    """
    try:
        from openpyxl import Workbook
    except ImportError as e:
        raise RuntimeError(
            "缺少 openpyxl 依赖，请执行：pip install openpyxl"
        ) from e

    wb = Workbook()
    ws = wb.active
    ws.title = sheet_name[:31]  # openpyxl sheet 名 ≤ 31 字符

    # 计算标题占用行数
    title_row_count = (1 if title else 0) + (1 if subtitle else 0)
    header_row = title_row_count + 1
    first_data_row = header_row + 1

    # 写入表头
    for col_idx, header in enumerate(headers, start=1):
        ws.cell(row=header_row, column=col_idx, value=header)

    # 写入数据
    for row_offset, row in enumerate(rows):
        excel_row = first_data_row + row_offset
        for col_idx, value in enumerate(row, start=1):
            ws.cell(row=excel_row, column=col_idx, value=_excel_safe(value))

    # 应用统一样式（含标题区）
    styles = _build_styles()
    _apply_sheet_styles(
        ws, headers, rows, styles,
        title=title, subtitle=subtitle,
        image_columns=image_columns, image_size=image_size,
    )

    out_abs = os.path.abspath(out_path)
    wb.save(out_abs)


def _excel_safe(value: Any) -> Any:
    """
    把 dict / list 等复杂类型序列化为 JSON 字符串，避免 openpyxl 写入失败。
    """
    if value is None:
        return ""
    if isinstance(value, (str, int, float, bool)):
        return value
    if isinstance(value, datetime):
        return value.strftime(DATETIME_FMT)
    try:
        return json.dumps(value, ensure_ascii=False)
    except (TypeError, ValueError):
        return str(value)


def write_excel_multi_sheets(
    out_path: str,
    sheets: list[dict],
) -> None:
    """
    用 openpyxl 写一个多 sheet 的 Excel 文件，每个 sheet 应用与 write_excel 相同的钉钉风格美化样式。

    每个 sheet 用一个 dict 描述：
      {
        "name":          str,             # sheet 标题
        "headers":       list[str],
        "rows":          list[list[Any]],
        "title":         str | None,      # 可选主标题（青绿底）
        "subtitle":      str | None,      # 可选副标题（浅蓝底）
        "merge_groups":  list[tuple[int,int,int]] | None,
                                          # 可选纵向合并配置（基础列）
                                          # 每项: (start_row_offset, end_row_offset, n_base_cols)
        "attend_result_columns": set[int] | None,
                                          # 可选：考勤结果列的 0-based 列索引集合（月度汇总场景）
        "attend_result_rows":    set[int] | None,
                                          # 可选：考勤结果行的 0-based 行偏移集合（日历表场景）
        "image_columns": list[str] | None,
                                          # 可选：哪些列名的 URL 要嵌入为图片
        "image_size":    tuple[int,int] | None,
                                          # 可选：嵌入图片像素尺寸 (width, height)，默认 (80,120)
      }
    """
    if not sheets:
        raise ValueError("sheets 不能为空")

    try:
        from openpyxl import Workbook
    except ImportError as e:
        raise RuntimeError(
            "缺少 openpyxl 依赖，请执行：pip install openpyxl"
        ) from e

    wb = Workbook()
    # 删除默认 sheet，由 sheets 描述完全决定
    default_ws = wb.active
    wb.remove(default_ws)

    styles = _build_styles()

    for sheet_def in sheets:
        name = sheet_def.get("name") or "Sheet"
        headers = sheet_def.get("headers") or []
        rows = sheet_def.get("rows") or []
        title = sheet_def.get("title")
        subtitle = sheet_def.get("subtitle")
        merge_groups = sheet_def.get("merge_groups")
        attend_result_columns = sheet_def.get("attend_result_columns")
        attend_result_rows = sheet_def.get("attend_result_rows")
        image_columns = sheet_def.get("image_columns")
        image_size = sheet_def.get("image_size") or (80, 120)

        ws = wb.create_sheet(title=name[:31])

        title_row_count = (1 if title else 0) + (1 if subtitle else 0)
        header_row = title_row_count + 1
        first_data_row = header_row + 1

        # 写入表头
        for col_idx, header in enumerate(headers, start=1):
            ws.cell(row=header_row, column=col_idx, value=header)

        # 写入数据
        for row_offset, row in enumerate(rows):
            excel_row = first_data_row + row_offset
            for col_idx, value in enumerate(row, start=1):
                ws.cell(row=excel_row, column=col_idx, value=_excel_safe(value))

        # 应用统一样式
        if headers:
            _apply_sheet_styles(
                ws, headers, rows, styles,
                title=title, subtitle=subtitle,
                merge_groups=merge_groups,
                attend_result_columns=attend_result_columns,
                attend_result_rows=attend_result_rows,
                image_columns=image_columns,
                image_size=image_size,
            )

    out_abs = os.path.abspath(out_path)
    wb.save(out_abs)


# ─────────────────────────────────────────────────────────────────────────────
# 输出文件命名
# ─────────────────────────────────────────────────────────────────────────────

def build_output_filename(start: datetime, end: datetime, *, suffix: str = "") -> str:
    """
    生成 attendance_report_<startDate>_<endDate>[_suffix].xlsx 形式的文件名，
    落在当前工作目录。
    """
    base = f"attendance_report_{start.strftime(DATE_FMT)}_{end.strftime(DATE_FMT)}"
    if suffix:
        base = f"{base}_{suffix}"
    return f"{base}.xlsx"


# ─────────────────────────────────────────────────────────────────────────────
# 请假数据查询（query-leave）
# ─────────────────────────────────────────────────────────────────────────────

# 默认关注的 4 类假期
DEFAULT_LEAVE_NAMES: tuple[str, ...] = ("事假", "调休", "病假", "年假")


def _normalize_leave_date(raw: Any) -> str | None:
    """把 query-leave 返回的 date 字段（毫秒时间戳字符串/数字）归一化为 YYYY-MM-DD。"""
    if raw is None or raw == "":
        return None
    try:
        ts = int(str(raw).strip())
    except (TypeError, ValueError):
        # 也可能本来就是 YYYY-MM-DD
        s = str(raw).strip()
        if len(s) >= 10 and s[4] == "-" and s[7] == "-":
            return s[:10]
        return None
    # 毫秒级
    if ts >= 1_000_000_000_000:
        ts = ts // 1000
    try:
        return datetime.fromtimestamp(ts).strftime(DATE_FMT)
    except (OSError, ValueError, OverflowError):
        return None


def query_leave_data(
    user_ids: list[str],
    start: datetime,
    end: datetime,
    leave_names: Iterable[str] = DEFAULT_LEAVE_NAMES,
    *,
    stats: "CallStats | None" = None,
) -> dict[str, dict[str, dict[str, float]]]:
    """
    分批分段调用 `dws attendance report query-leave`，聚合每个用户每天的假期数据。

    返回结构：
      {
        userId: {
          "YYYY-MM-DD": {
            "事假": 1.0,
            "调休": 0.5,
            ...
          },
          ...
        },
        ...
      }

    若同一 (userId, date, leaveName) 在多次返回中出现（理论不会），按 sum 累加。
    分批规则与 query-data 一致：≤ MAX_USERS_PER_BATCH 人/次、≤ MAX_DAYS_PER_SLICE 天/次。
    """
    result: dict[str, dict[str, dict[str, float]]] = {}
    if not user_ids:
        return result

    leave_names_list = [n for n in leave_names if n]
    if not leave_names_list:
        return result

    user_batches = chunk_users(user_ids)
    date_slices = slice_date_range(start, end)
    leave_arg = ",".join(leave_names_list)

    log(
        f"[leave] 查询 {len(user_ids)} 人 × {len(leave_names_list)} 类假期 × "
        f"{len(date_slices)} 个时间片"
    )

    for bi, batch in enumerate(user_batches, start=1):
        for si, dslice in enumerate(date_slices, start=1):
            log(
                f"[leave] [batch {bi}/{len(user_batches)}] "
                f"[slice {si}/{len(date_slices)}] users={len(batch)} "
                f"slice={dslice.label}"
            )
            try:
                payload = run_dws([
                    "attendance", "report", "query-leave",
                    "--users", ",".join(batch),
                    "--leave-names", leave_arg,
                    "--start", dslice.start_str,
                    "--end", dslice.end_str,
                ])
                if stats is not None:
                    stats.total_dws_calls += 1
            except DwsCallError as e:
                if stats is not None:
                    stats.total_dws_calls += 1
                    stats.failed_calls += 1
                if e.is_permission_error:
                    error("权限错误：当前账号无管理员权限，无法查询请假数据。")
                    raise SystemExit(2) from e
                if stats is not None:
                    stats.add_warning(f"[leave query failed] {dslice.label}: {e}")
                else:
                    warn(f"[leave query failed] {dslice.label}: {e}")
                continue

            records = extract_records(payload)
            for record in records:
                uid = _first_nonempty(record, ("userId", "userid", "user_id"))
                if uid is None:
                    continue
                uid_str = str(uid)
                leave_vals = record.get("leaveVals")
                if not isinstance(leave_vals, list):
                    continue
                user_bucket = result.setdefault(uid_str, {})
                for entry in leave_vals:
                    if not isinstance(entry, dict):
                        continue
                    date_str = _normalize_leave_date(entry.get("date"))
                    if not date_str:
                        continue
                    leave_name = entry.get("leaveName") or entry.get("name")
                    if not leave_name:
                        continue
                    leave_name = str(leave_name)
                    raw_value = entry.get("value", entry.get("data", 0))
                    try:
                        num = float(str(raw_value).strip())
                    except (TypeError, ValueError):
                        continue
                    day_bucket = user_bucket.setdefault(date_str, {})
                    day_bucket[leave_name] = day_bucket.get(leave_name, 0.0) + num

    return result


def build_vacation_filename(
    start: datetime | None,
    end: datetime | None,
    *,
    as_of: datetime | None = None,
) -> str:
    """
    生成 vacation_export_<startDate>_<endDate>.xlsx；
    无时间区间时退化为 vacation_export_<asOfDate>.xlsx（asOfDate 默认为今天）。
    """
    if start is not None and end is not None:
        return f"vacation_export_{start.strftime(DATE_FMT)}_{end.strftime(DATE_FMT)}.xlsx"
    snapshot_date = (as_of or datetime.now()).strftime(DATE_FMT)
    return f"vacation_export_{snapshot_date}.xlsx"


# ─────────────────────────────────────────────────────────────────────────────
# 通用报告骨架
# ─────────────────────────────────────────────────────────────────────────────

@dataclass
class CallStats:
    """记录一次脚本运行中的 dws 调用统计，用于最终摘要。"""
    user_batches: int = 0
    date_slices: int = 0
    total_dws_calls: int = 0
    failed_calls: int = 0
    warnings: list[str] = field(default_factory=list)

    def add_warning(self, msg: str) -> None:
        self.warnings.append(msg)
        warn(msg)


def print_summary(
    *,
    granularity_label: str,
    out_path: str,
    user_count: int,
    column_names: list[str],
    start: datetime,
    end: datetime,
    rows_count: int,
    stats: CallStats,
    extra_tail: str = "",
) -> None:
    """
    把最终摘要打到 stdout，供调用方（Agent / 终端用户）查看。
    格式与 SKILL.md 输出模板对齐。
    """
    abs_path = os.path.abspath(out_path)
    print("[完成] 考勤报表已导出")
    print()
    print(f"[文件] {abs_path}")
    print(f"[粒度] {granularity_label}")
    print(f"[用户] {user_count} 人（共 {stats.user_batches} 批）")
    print(f"[时间] {start.strftime(DATE_FMT)} ~ {end.strftime(DATE_FMT)}"
          f"（共 {stats.date_slices} 个时间片）")
    print(f"[调用] 共调用 dws 接口：{stats.total_dws_calls} 次"
          + (f"（其中 {stats.failed_calls} 次失败）" if stats.failed_calls else ""))
    print(f"[字段] {' / '.join(column_names) if column_names else '（默认）'}")
    print(f"[行数] {rows_count} 行")
    if stats.warnings:
        print()
        print("[警告]")
        for w in stats.warnings[:10]:
            print(f"  - {w}")
        if len(stats.warnings) > 10:
            print(f"  - ...（共 {len(stats.warnings)} 条警告）")
    if extra_tail:
        print()
        print(extra_tail)


# ─────────────────────────────────────────────────────────────────────────────
# 图片下载 + Excel 嵌入（detail 报表的"打卡图片"列专用）
# ─────────────────────────────────────────────────────────────────────────────

# 全局缓存：URL → 本地 PNG 文件路径，避免同一张图重复下载/转换
_IMAGE_CACHE_DIR = os.path.join(
    tempfile.gettempdir(), "dws_attendance_report_images"
)
_image_url_to_local: dict[str, str] = {}
# 下载/转换失败的 URL 黑名单，避免反复重试
_image_failed_urls: set[str] = set()


def _ensure_image_cache_dir() -> str:
    """确保图片缓存目录存在并返回路径。"""
    os.makedirs(_IMAGE_CACHE_DIR, exist_ok=True)
    return _IMAGE_CACHE_DIR


def _is_likely_url(value: Any) -> bool:
    """简单判断一个值是不是 http(s) URL。"""
    if not isinstance(value, str):
        return False
    s = value.strip()
    return s.startswith("http://") or s.startswith("https://")


def download_and_convert_image(
    url: str,
    *,
    timeout: int = 10,
) -> str | None:
    """
    下载图片 URL → PIL 转 PNG → 缓存到本地，返回本地 PNG 文件路径。

    特性：
      - 磁盘缓存（同一 URL 只下载一次）
      - 支持 webp/jpg/jpeg/png 等格式（PIL 自动识别）
      - 失败的 URL 加黑名单，避免反复重试
      - 失败返回 None，调用方应保留原 URL 文本

    依赖: requests + Pillow（PIL）
    """
    if url in _image_url_to_local:
        return _image_url_to_local[url]
    if url in _image_failed_urls:
        return None

    cache_dir = _ensure_image_cache_dir()
    url_hash = hashlib.md5(url.encode("utf-8")).hexdigest()
    local_path = os.path.join(cache_dir, f"{url_hash}.png")

    if os.path.exists(local_path) and os.path.getsize(local_path) > 0:
        _image_url_to_local[url] = local_path
        return local_path

    try:
        import requests
    except ImportError:
        warn("缺少 requests 依赖，无法下载图片，请执行: pip install requests")
        _image_failed_urls.add(url)
        return None
    try:
        from PIL import Image as PILImage
    except ImportError:
        warn("缺少 Pillow 依赖，无法转换图片格式，请执行: pip install Pillow")
        _image_failed_urls.add(url)
        return None

    try:
        resp = requests.get(url, timeout=timeout)
        resp.raise_for_status()
        raw_bytes = resp.content
        if not raw_bytes:
            raise ValueError("下载结果为空")
    except Exception as e:
        warn(f"[image] 下载失败: {url[:80]}... 原因: {e}")
        _image_failed_urls.add(url)
        return None

    try:
        from io import BytesIO
        with PILImage.open(BytesIO(raw_bytes)) as img:
            # webp 等可能是 RGBA / P 模式，统一转 RGB 再存 PNG
            if img.mode not in ("RGB", "RGBA"):
                img = img.convert("RGBA")
            img.save(local_path, format="PNG")
    except Exception as e:
        warn(f"[image] 转换失败: {url[:80]}... 原因: {e}")
        _image_failed_urls.add(url)
        return None

    _image_url_to_local[url] = local_path
    return local_path


def _set_image_hyperlink(ws, excel_row: int, col_idx: int, url: str) -> None:
    """把单元格设为可点击的超链接，文案显示"打卡图片"，避免直接暴露裸 URL。"""
    from openpyxl.styles import Font
    cell = ws.cell(row=excel_row, column=col_idx)
    cell.value = "打卡图片"
    cell.hyperlink = url
    cell.font = Font(color="0563C1", underline="single")


def _replace_all_image_urls_with_hyperlinks(
    ws, headers: list[str], rows: list[list[Any]],
    image_column_names: list[str], first_data_row: int,
) -> None:
    """Pillow 不可用时的兜底：把所有图片列的 URL 替换为"打卡图片"超链接。"""
    name_to_col_idx: dict[str, int] = {}
    for i, h in enumerate(headers, start=1):
        if h in image_column_names and h not in name_to_col_idx:
            name_to_col_idx[h] = i
    for row_offset, row in enumerate(rows):
        excel_row = first_data_row + row_offset
        for col_idx in name_to_col_idx.values():
            if col_idx - 1 >= len(row):
                continue
            if _is_likely_url(row[col_idx - 1]):
                _set_image_hyperlink(ws, excel_row, col_idx, str(row[col_idx - 1]).strip())


def _embed_images_in_columns(
    ws,
    headers: list[str],
    rows: list[list[Any]],
    *,
    image_column_names: list[str],
    header_row: int,
    first_data_row: int,
    image_size: tuple[int, int] = (80, 120),
) -> None:
    """
    把指定列里的 URL 替换为嵌入的图片：
      1. 找到 image_column_names 命中的列索引
      2. 遍历每行该列的值，若是 http(s) URL 则下载 + 转 PNG + add_image
      3. 同步调整列宽（与图片宽匹配）和行高（与图片高匹配）
      4. 下载/转换失败时将 URL 替换为可点击的"打卡图片"超链接，避免暴露裸 URL

    image_size: (width_px, height_px)，控制嵌入图片尺寸，默认 80×120 像素
    """
    if not image_column_names:
        return
    try:
        from openpyxl.drawing.image import Image as OpenpyxlImage
        from openpyxl.utils import get_column_letter
    except ImportError as e:
        warn(f"openpyxl 不完整，无法嵌入图片: {e}")
        return

    # openpyxl 的 Image 类内部依赖 Pillow（模块加载时检测），
    # 如果 Pillow 不可用，OpenpyxlImage() 会抛出：
    #   ImportError: You must install Pillow to fetch image objects
    # 这里提前检测，不可用时尝试自动安装，避免逐张图片重复报错。
    from openpyxl.drawing.image import PILImage as _openpyxl_pil_check
    if not _openpyxl_pil_check:
        warn(
            "[image] openpyxl 检测到 Pillow 未安装，尝试自动安装..."
        )
        import subprocess, sys
        try:
            subprocess.check_call(
                [sys.executable, "-m", "pip", "install", "Pillow"],
                stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
                timeout=60,
            )
            log("[image] Pillow 安装成功，重新加载 openpyxl.drawing.image...")
            # 安装后需要重新加载模块，让 openpyxl 重新检测 Pillow
            import importlib
            import openpyxl.drawing.image as _img_mod
            importlib.reload(_img_mod)
            from openpyxl.drawing.image import Image as OpenpyxlImage  # noqa: F811
            from openpyxl.drawing.image import PILImage as _recheck
            if not _recheck:
                warn(
                    "[image] Pillow 安装后 openpyxl 仍无法检测到，"
                    "图片将显示为可点击链接。请手动执行: pip install Pillow"
                )
                _replace_all_image_urls_with_hyperlinks(
                    ws, headers, rows, image_column_names, first_data_row,
                )
                return
        except Exception as install_err:
            warn(
                f"[image] Pillow 自动安装失败: {install_err}\n"
                "图片将显示为可点击链接。请手动执行: pip install Pillow"
            )
            _replace_all_image_urls_with_hyperlinks(
                ws, headers, rows, image_column_names, first_data_row,
            )
            return

    # 找到目标列索引（1-based）
    name_to_col_idx: dict[str, int] = {}
    for i, h in enumerate(headers, start=1):
        if h in image_column_names and h not in name_to_col_idx:
            name_to_col_idx[h] = i
    if not name_to_col_idx:
        return

    width_px, height_px = image_size
    # Excel 列宽单位 ≈ 字符数，1 字符 ≈ 7px；行高单位为点，1 点 ≈ 1.33px
    col_width = max(width_px / 7.0, 12.0)
    row_height = max(height_px * 0.78, 60.0)

    # 调列宽
    for col_idx in name_to_col_idx.values():
        ws.column_dimensions[get_column_letter(col_idx)].width = col_width

    # 收集所有要嵌入的 (excel_row, col_idx, url)，统一处理
    embed_tasks: list[tuple[int, int, str]] = []
    for row_offset, row in enumerate(rows):
        excel_row = first_data_row + row_offset
        row_has_image = False
        for col_name, col_idx in name_to_col_idx.items():
            if col_idx - 1 >= len(row):
                continue
            value = row[col_idx - 1]
            if not _is_likely_url(value):
                continue
            url = str(value).strip()
            embed_tasks.append((excel_row, col_idx, url))
            row_has_image = True
        if row_has_image:
            ws.row_dimensions[excel_row].height = row_height

    if not embed_tasks:
        return

    log(f"[image] 准备嵌入 {len(embed_tasks)} 张图片到 Excel...")
    success_count = 0
    failed_count = 0

    for excel_row, col_idx, url in embed_tasks:
        local_path = download_and_convert_image(url)
        if not local_path:
            _set_image_hyperlink(ws, excel_row, col_idx, url)
            failed_count += 1
            continue
        try:
            img = OpenpyxlImage(local_path)
            img.width = width_px
            img.height = height_px
            anchor = f"{get_column_letter(col_idx)}{excel_row}"
            # 清空原 URL 单元格内容（图片浮在格子上，但保留文本会重叠）
            ws.cell(row=excel_row, column=col_idx, value="")
            ws.add_image(img, anchor)
            success_count += 1
        except Exception as e:
            warn(f"[image] 嵌入失败 ({excel_row},{col_idx}): {e}")
            _set_image_hyperlink(ws, excel_row, col_idx, url)
            failed_count += 1

    log(f"[image] 嵌入完成: 成功 {success_count} 张，失败 {failed_count} 张")