"""PMFlow 团队负载服务。"""

from __future__ import annotations

import json
from datetime import date, datetime, timedelta
from typing import Optional

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

from ..config.ai_prompts import PROMPT_WORKLOAD_OPTIMIZATION
from ..models import AICallLog, Task, User, UserProject, UserSkillBinding, UserWorkload
from .llm import llm_service

LEVEL_HOURS_MULTIPLIER = {
    "expert": 0.8,
    "advanced": 0.9,
    "intermediate": 1.0,
    "beginner": 1.1,
}


def remaining_workdays_this_week(reference_date: Optional[date] = None) -> int:
    """计算本周剩余工作日（含当天）。"""
    today = reference_date or date.today()
    weekday = today.weekday()  # Monday = 0
    days_to_friday = max(0, 4 - weekday)
    count = 0
    for offset in range(days_to_friday + 1):
        d = today + timedelta(days=offset)
        if d.weekday() < 5:
            count += 1
    return max(count, 1)


def calculate_workload_ratio(
    assigned_hours: float,
    max_weekly_hours: float,
    reference_date: Optional[date] = None,
) -> float:
    """
    负载率公式（百分比）：
    ratio = assigned_hours / ((max_weekly_hours / 5) * remaining_workdays_this_week)
    """
    remaining_days = remaining_workdays_this_week(reference_date)
    denominator = (max_weekly_hours / 5.0) * remaining_days
    if denominator <= 0:
        return 0.0
    return round((assigned_hours / denominator) * 100.0, 2)


async def ensure_user_workload(
    db: AsyncSession,
    tenant_id: int,
    user_id: int,
) -> UserWorkload:
    result = await db.execute(
        select(UserWorkload).where(
            and_(
                UserWorkload.tenant_id == tenant_id,
                UserWorkload.user_id == user_id,
            )
        )
    )
    workload = result.scalar_one_or_none()
    if workload:
        return workload

    workload = UserWorkload(
        tenant_id=tenant_id,
        user_id=user_id,
        current_assigned_hours=0.0,
        max_available_hours_per_week=40.0,
        workload_ratio=0.0,
    )
    db.add(workload)
    await db.flush()
    return workload


async def recalculate_user_workload_ratio(
    db: AsyncSession,
    tenant_id: int,
    user_id: int,
    *,
    commit: bool = True,
) -> Optional[UserWorkload]:
    """重算单用户负载率并同步 users.workload。"""
    user_result = await db.execute(
        select(User).where(and_(User.tenant_id == tenant_id, User.id == user_id))
    )
    user = user_result.scalar_one_or_none()
    if not user:
        return None

    workload = await ensure_user_workload(db, tenant_id, user_id)
    ratio = calculate_workload_ratio(
        assigned_hours=float(workload.current_assigned_hours or 0),
        max_weekly_hours=float(workload.max_available_hours_per_week or 40),
    )
    workload.workload_ratio = ratio
    user.workload = int(max(0, min(200, ratio)))
    user.updated_at = datetime.utcnow()
    try:
        from . import workload_alert_service

        await workload_alert_service.on_user_workload_changed(
            db,
            tenant_id=tenant_id,
            user_id=user_id,
            workload_ratio=ratio,
        )
    except Exception:
        pass

    if commit:
        await db.commit()
        await db.refresh(workload)
    return workload


async def recalculate_all_users_workload_ratio(db: AsyncSession, tenant_id: int) -> int:
    """批量重算租户内所有用户负载率。"""
    result = await db.execute(select(User.id).where(User.tenant_id == tenant_id))
    user_ids = [int(uid) for uid in result.scalars().all()]
    for uid in user_ids:
        await recalculate_user_workload_ratio(db, tenant_id, uid, commit=False)
    await db.commit()
    return len(user_ids)


async def list_tenant_ids(db: AsyncSession) -> list[int]:
    """获取当前系统存在用户数据的租户 ID 列表。"""
    result = await db.execute(select(User.tenant_id).distinct())
    return [int(tid) for tid in result.scalars().all() if tid is not None]


async def get_team_workload_board(
    db: AsyncSession,
    tenant_id: int = 1,
    project_id: Optional[int] = None,
) -> list[dict]:
    """团队负载看板数据。"""
    query = (
        select(UserWorkload, User)
        .join(User, UserWorkload.user_id == User.id)
        .where(
            and_(
                UserWorkload.tenant_id == tenant_id,
                User.tenant_id == tenant_id,
            )
        )
    )
    if project_id is not None:
        query = query.join(
            UserProject,
            and_(
                UserProject.tenant_id == tenant_id,
                UserProject.user_id == User.id,
            ),
        ).where(UserProject.project_id == project_id)

    query = query.order_by(UserWorkload.workload_ratio.desc(), User.id.asc())
    rows = (await db.execute(query)).all()

    return [
        {
            "user_id": str(w.user_id),
            "user_name": u.name,
            "username": u.username,
            "role": u.role,
            "is_active": u.is_active,
            "current_assigned_hours": float(w.current_assigned_hours or 0),
            "max_weekly_hours": float(w.max_available_hours_per_week or 40),
            "workload_ratio": float(w.workload_ratio or 0),
            "remaining_hours": round(
                max(
                    0.0,
                    float(w.max_available_hours_per_week or 40)
                    - float(w.current_assigned_hours or 0),
                ),
                2,
            ),
            "updated_at": w.updated_at.isoformat() if w.updated_at else None,
        }
        for w, u in rows
    ]


async def get_user_personal_workload(
    db: AsyncSession,
    tenant_id: int,
    user_id: int,
) -> Optional[dict]:
    """个人负载详情。"""
    row = (
        await db.execute(
            select(UserWorkload, User)
            .join(User, UserWorkload.user_id == User.id)
            .where(
                and_(
                    UserWorkload.tenant_id == tenant_id,
                    UserWorkload.user_id == user_id,
                    User.tenant_id == tenant_id,
                )
            )
        )
    ).first()
    if row:
        w, u = row
        return {
            "user_id": str(u.id),
            "user_name": u.name,
            "username": u.username,
            "is_active": u.is_active,
            "current_assigned_hours": float(w.current_assigned_hours or 0),
            "max_weekly_hours": float(w.max_available_hours_per_week or 40),
            "workload_ratio": float(w.workload_ratio or 0),
            "remaining_hours": round(
                max(
                    0.0,
                    float(w.max_available_hours_per_week or 40)
                    - float(w.current_assigned_hours or 0),
                ),
                2,
            ),
            "updated_at": w.updated_at.isoformat() if w.updated_at else None,
        }

    user = (
        await db.execute(
            select(User).where(and_(User.tenant_id == tenant_id, User.id == user_id))
        )
    ).scalar_one_or_none()
    if not user:
        return None

    workload = await ensure_user_workload(db, tenant_id, user_id)
    await db.commit()
    return {
        "user_id": str(user.id),
        "user_name": user.name,
        "username": user.username,
        "is_active": user.is_active,
        "current_assigned_hours": float(workload.current_assigned_hours or 0),
        "max_weekly_hours": float(workload.max_available_hours_per_week or 40),
        "workload_ratio": float(workload.workload_ratio or 0),
        "remaining_hours": round(
            max(
                0.0,
                float(workload.max_available_hours_per_week or 40)
                - float(workload.current_assigned_hours or 0),
            ),
            2,
        ),
        "updated_at": workload.updated_at.isoformat() if workload.updated_at else None,
    }


async def update_user_max_weekly_hours(
    db: AsyncSession,
    tenant_id: int,
    user_id: int,
    max_weekly_hours: float,
) -> Optional[dict]:
    """更新每周可用工时并重算负载率。"""
    user = (
        await db.execute(
            select(User).where(and_(User.tenant_id == tenant_id, User.id == user_id))
        )
    ).scalar_one_or_none()
    if not user:
        return None

    workload = await ensure_user_workload(db, tenant_id, user_id)
    workload.max_available_hours_per_week = max(1.0, float(max_weekly_hours))
    await recalculate_user_workload_ratio(db, tenant_id, user_id, commit=False)
    await db.commit()
    await db.refresh(workload)

    return {
        "user_id": str(user.id),
        "max_weekly_hours": float(workload.max_available_hours_per_week),
        "workload_ratio": float(workload.workload_ratio or 0),
    }


async def list_user_assigned_tasks(
    db: AsyncSession,
    tenant_id: int,
    user_id: int,
    *,
    status: Optional[str] = None,
    priority: Optional[str] = None,
) -> list[dict]:
    """查询用户任务列表。"""
    query = select(Task).where(
        and_(
            Task.tenant_id == tenant_id,
            Task.assignee_id == user_id,
        )
    )
    if status:
        query = query.where(Task.status == status)
    if priority:
        query = query.where(Task.priority == priority)

    rows = (await db.execute(query.order_by(Task.updated_at.desc()))).scalars().all()
    return [
        {
            "id": str(t.id),
            "title": t.title,
            "project_id": str(t.project_id),
            "priority": t.priority,
            "status": t.status,
            "estimated_hours": float(t.estimated_hours or 0),
            "due_date": t.due_date.isoformat() if t.due_date else None,
        }
        for t in rows
    ]


async def get_user_workload_trend(
    db: AsyncSession,
    tenant_id: int,
    user_id: int,
    days: int = 7,
) -> list[dict]:
    """
    用户负载趋势。

    当前无历史快照表时，返回最近 N 天的平滑序列用于前端趋势图展示。
    """
    workload = (
        await db.execute(
            select(UserWorkload).where(
                and_(
                    UserWorkload.tenant_id == tenant_id,
                    UserWorkload.user_id == user_id,
                )
            )
        )
    ).scalar_one_or_none()
    ratio_now = float(workload.workload_ratio or 0) if workload else 0.0

    days = max(1, min(days, 30))
    start = date.today() - timedelta(days=days - 1)
    series: list[dict] = []
    for idx in range(days):
        current_day = start + timedelta(days=idx)
        # 缺少历史表时做线性平滑，保留可视化可用性。
        ratio = round(max(0.0, ratio_now * (0.92 + 0.08 * (idx + 1) / days)), 2)
        series.append(
            {
                "date": current_day.isoformat(),
                "workload_ratio": ratio,
            }
        )
    return series


async def refresh_user_assigned_hours_from_tasks(
    db: AsyncSession,
    tenant_id: int,
    user_id: int,
) -> Optional[UserWorkload]:
    """从任务表汇总 assigned_hours 并重算 ratio。"""
    workload = await ensure_user_workload(db, tenant_id, user_id)

    task_rows = (
        await db.execute(
            select(Task).where(
                and_(
                    Task.tenant_id == tenant_id,
                    Task.assignee_id == user_id,
                    func.lower(Task.status).notin_(["done", "completed", "closed"]),
                )
            )
        )
    ).scalars().all()

    skill_rows = (
        await db.execute(
            select(
                UserSkillBinding.skill_tag_id,
                UserSkillBinding.level,
                UserSkillBinding.proficiency_level,
                UserSkillBinding.proficiency,
            ).where(
                and_(
                    UserSkillBinding.tenant_id == tenant_id,
                    UserSkillBinding.user_id == user_id,
                )
            )
        )
    ).all()

    def normalize_level(
        level: Optional[str],
        proficiency_level: Optional[str],
        proficiency: Optional[int],
    ) -> str:
        raw = (level or proficiency_level or "").strip().lower()
        if raw in LEVEL_HOURS_MULTIPLIER:
            return raw
        score = int(proficiency or 0)
        if score >= 85:
            return "expert"
        if score >= 70:
            return "advanced"
        if score >= 50:
            return "intermediate"
        return "beginner"

    level_map: dict[int, str] = {
        int(skill_tag_id): normalize_level(level, proficiency_level, proficiency)
        for skill_tag_id, level, proficiency_level, proficiency in skill_rows
    }

    def effective_estimated_hours(task: Task) -> float:
        base = float(task.estimated_hours or 0.0)
        if base <= 0:
            return 0.0
        required_ids = _safe_int_list(task.required_skill_tag_ids)
        if not required_ids:
            return base
        multipliers = [
            LEVEL_HOURS_MULTIPLIER.get(level_map[sid], 1.0)
            for sid in required_ids
            if sid in level_map
        ]
        if not multipliers:
            return base
        factor = sum(multipliers) / len(multipliers)
        return round(base * factor, 2)

    workload.current_assigned_hours = round(
        sum(effective_estimated_hours(task) for task in task_rows),
        2,
    )

    await recalculate_user_workload_ratio(db, tenant_id, user_id, commit=False)
    await db.commit()
    await db.refresh(workload)
    return workload


def _skill_match_score(task_skill_ids: list[int], user_skill_ids: set[int]) -> float:
    if not task_skill_ids:
        return 0.0
    matched = sum(1 for sid in task_skill_ids if sid in user_skill_ids)
    return matched / len(task_skill_ids)


def _safe_int_list(values: object) -> list[int]:
    """兼容历史脏数据：数组中可能混入技能名称等非数字值。"""
    source = values
    if source is None:
        return []
    if isinstance(source, str):
        text = source.strip()
        if not text:
            return []
        try:
            source = json.loads(text)
        except (TypeError, ValueError, json.JSONDecodeError):
            return []
    if not isinstance(source, list):
        return []

    parsed: list[int] = []
    for item in source:
        if item is None or isinstance(item, bool):
            continue
        if isinstance(item, int):
            parsed.append(item)
            continue
        if isinstance(item, float):
            if item.is_integer():
                parsed.append(int(item))
            continue
        if isinstance(item, str):
            text = item.strip()
            if text and text.lstrip("-").isdigit():
                parsed.append(int(text))
    return parsed


def _parse_json_response(text: str) -> dict:
    body = (text or "").strip()
    if body.startswith("```json"):
        body = body[7:].strip()
    if body.startswith("```"):
        body = body[3:].strip()
    if body.endswith("```"):
        body = body[:-3].strip()
    return json.loads(body)


async def build_ai_optimize_preview(
    db: AsyncSession,
    tenant_id: int,
    *,
    project_id: Optional[int] = None,
    max_moves: int = 20,
    unassigned_only: bool = False,
) -> list[dict]:
    """生成任务重分配建议（预览，不落库）。"""
    task_query = select(Task).where(
        and_(
            Task.tenant_id == tenant_id,
            func.lower(Task.status).notin_(["done", "completed", "closed"]),
        )
    )
    if unassigned_only:
        task_query = task_query.where(Task.assignee_id.is_(None))
    if project_id is not None:
        task_query = task_query.where(Task.project_id == project_id)

    task_rows = (await db.execute(task_query.order_by(Task.updated_at.desc()))).scalars().all()
    if not task_rows:
        return []

    active_users_query = select(User).where(
        and_(
            User.tenant_id == tenant_id,
            User.status == "active",
            User.is_active.is_(True),
        )
    )
    users = (await db.execute(active_users_query)).scalars().all()
    if not users:
        return []

    workload_rows = (
        await db.execute(
            select(UserWorkload).where(
                and_(
                    UserWorkload.tenant_id == tenant_id,
                    UserWorkload.user_id.in_([int(u.id) for u in users]),
                )
            )
        )
    ).scalars().all()
    workload_map = {int(w.user_id): float(w.workload_ratio or 0.0) for w in workload_rows}

    skill_rows = (
        await db.execute(
            select(UserSkillBinding.user_id, UserSkillBinding.skill_tag_id).where(
                UserSkillBinding.tenant_id == tenant_id
            )
        )
    ).all()
    user_skill_map: dict[int, set[int]] = {}
    for uid, sid in skill_rows:
        user_skill_map.setdefault(int(uid), set()).add(int(sid))

    users_by_project: dict[int, list[User]] = {}
    for task in task_rows:
        if int(task.project_id) in users_by_project:
            continue
        links = (
            await db.execute(
                select(UserProject.user_id).where(
                    and_(
                        UserProject.tenant_id == tenant_id,
                        UserProject.project_id == int(task.project_id),
                    )
                )
            )
        ).scalars().all()
        allowed_ids = {int(uid) for uid in links}
        users_by_project[int(task.project_id)] = [u for u in users if int(u.id) in allowed_ids]

    llm_suggestion_map: dict[int, dict] = {}
    members_payload = []
    for user in users:
        uid = int(user.id)
        members_payload.append(
            {
                "user_id": uid,
                "user_name": user.name,
                "workload_ratio": round(workload_map.get(uid, float(user.workload or 0)), 2),
                "skill_tag_ids": sorted(list(user_skill_map.get(uid, set()))),
            }
        )
    tasks_payload = [
        {
            "task_id": int(task.id),
            "project_id": int(task.project_id),
            "title": task.title,
            "assignee_id": int(task.assignee_id) if task.assignee_id else None,
            "required_skill_tag_ids": _safe_int_list(task.required_skill_tag_ids),
            "estimated_hours": float(task.estimated_hours or 0),
            "status": task.status,
        }
        for task in task_rows
    ]
    llm_prompt = PROMPT_WORKLOAD_OPTIMIZATION.format(
        members_json=json.dumps(members_payload, ensure_ascii=False),
        tasks_json=json.dumps(tasks_payload, ensure_ascii=False),
    )
    llm_text = ""
    llm_status = "success"
    try:
        llm_text = await llm_service.generate(
            llm_prompt,
            function_name="workload_optimize_prompt",
            tenant_id=tenant_id,
        )
        parsed = _parse_json_response(llm_text)
        for item in parsed.get("proposals", []) if isinstance(parsed, dict) else []:
            try:
                tid = int(item.get("task_id"))
                llm_suggestion_map[tid] = {
                    "new_assignee_id": int(item.get("new_assignee_id")),
                    "reason": str(item.get("reason") or "llm_suggested"),
                }
            except (TypeError, ValueError):
                continue
    except Exception as exc:
        llm_status = "failed"
        llm_text = str(exc)

    db.add(
        AICallLog(
            tenant_id=tenant_id,
            user_id=None,
            function_name="workload_optimize_prompt",
            prompt=llm_prompt,
            response=(llm_text or "")[:8000],
            tokens_used=max(len(llm_prompt) // 4, 1) + max(len(llm_text) // 4, 0),
            cost=((max(len(llm_prompt) // 4, 1) * 0.00015 + max(len(llm_text) // 4, 0) * 0.0006) / 1000),
            status=llm_status,
        )
    )
    await db.commit()

    proposals: list[dict] = []
    for task in task_rows:
        current_assignee = int(task.assignee_id or 0)
        is_unassigned = current_assignee <= 0
        candidates = users_by_project.get(int(task.project_id), [])
        if not candidates:
            continue

        required_skill_ids = _safe_int_list(task.required_skill_tag_ids)
        ranked = []
        llm_target_id = llm_suggestion_map.get(int(task.id), {}).get("new_assignee_id")
        llm_reason = llm_suggestion_map.get(int(task.id), {}).get("reason")
        for candidate in candidates:
            uid = int(candidate.id)
            if not is_unassigned and uid == current_assignee:
                continue
            ratio = workload_map.get(uid, float(candidate.workload or 0))
            skill_score = _skill_match_score(required_skill_ids, user_skill_map.get(uid, set()))
            # workload 越低越优，skill_score 越高越优
            total_score = (skill_score * 0.7) + ((max(0.0, 120.0 - ratio) / 120.0) * 0.3)
            if llm_target_id and uid == int(llm_target_id):
                total_score += 0.2
            ranked.append((total_score, ratio, candidate))

        ranked.sort(key=lambda x: (-x[0], x[1]))
        if not ranked:
            continue

        best_score, _best_ratio, best_user = ranked[0]
        current_ratio = workload_map.get(current_assignee, 0.0) if not is_unassigned else 0.0
        if is_unassigned:
            if best_score < 0.2:
                continue
            reason = f"initial_assignment_score={best_score:.2f}"
        else:
            if current_ratio <= 100 and best_score < 0.5:
                continue
            reason = f"skill_match+workload_score={best_score:.2f}"
        if llm_target_id and int(best_user.id) == int(llm_target_id):
            reason = f"{reason};llm={llm_reason or 'suggested'}"

        proposals.append(
            {
                "task_id": int(task.id),
                "task_title": task.title,
                "project_id": int(task.project_id),
                "old_assignee_id": current_assignee if not is_unassigned else None,
                "new_assignee_id": int(best_user.id),
                "old_ratio": round(current_ratio, 2) if not is_unassigned else None,
                "new_ratio": round(workload_map.get(int(best_user.id), 0.0), 2),
                "reason": reason,
            }
        )
        if len(proposals) >= max_moves:
            break

    return proposals


async def apply_ai_optimize_plan(
    db: AsyncSession,
    tenant_id: int,
    proposals: list[dict],
) -> dict:
    """应用优化方案并重算涉及成员负载。"""
    touched_user_ids: set[int] = set()
    applied = 0

    for item in proposals:
        task_id = int(item.get("task_id", 0))
        new_assignee_id = int(item.get("new_assignee_id", 0))
        if task_id <= 0 or new_assignee_id <= 0:
            continue

        task = (
            await db.execute(
                select(Task).where(
                    and_(
                        Task.tenant_id == tenant_id,
                        Task.id == task_id,
                    )
                )
            )
        ).scalar_one_or_none()
        if not task:
            continue

        if task.assignee_id:
            touched_user_ids.add(int(task.assignee_id))
        task.assignee_id = new_assignee_id
        touched_user_ids.add(new_assignee_id)
        applied += 1

    if touched_user_ids:
        for uid in touched_user_ids:
            await refresh_user_assigned_hours_from_tasks(db, tenant_id, uid)
    else:
        await db.commit()

    return {"applied_count": applied, "affected_users": len(touched_user_ids)}