"""用户工单路由。"""
from __future__ import annotations

import io
import re
import secrets
from pathlib import Path
from datetime import datetime, timezone
from typing import Any

import aiofiles
from fastapi import APIRouter, Depends, File, HTTPException, Query, Request, UploadFile
from fastapi.responses import FileResponse, Response
from PIL import Image, UnidentifiedImageError
from pydantic import BaseModel, Field, field_validator
from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession

from app.config import settings
from app.core.deps import get_current_user
from app.database import get_db
from app.models.member_level import MemberLevel
from app.models.instance import Instance
from app.models.ticket import (
    Ticket,
    TicketAttachment,
    TicketDepartment,
    TicketEvent,
    TicketEventType,
    TicketMessage,
    TicketPriority,
    TicketSenderType,
    TicketStatus,
)
from app.models.user import User, UserRole, UserStatus
from app.schemas.common import ApiResponse
from app.services import membership_service

router = APIRouter()
ALLOWED_TICKET_IMAGE_TYPES = {
    "image/jpeg": ".jpg",
    "image/png": ".png",
    "image/gif": ".gif",
    "image/webp": ".webp",
}

# Images wider than this are downscaled before saving to reduce
# bandwidth and frontend rendering cost. Height follows proportionally.
TICKET_IMAGE_MAX_WIDTH = 1920
TICKET_IMAGE_MAX_HEIGHT = 1920
TICKET_ATTACHMENT_URL_PREFIX = "/api/v1/tickets/attachments/images"
TICKET_ATTACHMENT_TOKEN_RE = re.compile(
    r"/attachments/images/(?:file|thumbnail)/([A-Za-z0-9_-]+)"
)


def _reject_embedded_images(value: str) -> str:
    if "data:image/" in value.lower():
        raise ValueError("请先上传图片，不要直接粘贴 base64 图片")
    return value


class TicketCreate(BaseModel):
    department_id: int
    subject: str = Field(min_length=2, max_length=160)
    content: str = Field(min_length=2, max_length=8000)
    priority: TicketPriority = TicketPriority.NORMAL
    instance_id: int | None = None
    attachments: list[dict[str, Any]] | None = None

    @field_validator("content")
    @classmethod
    def validate_content(cls, value: str) -> str:
        return _reject_embedded_images(value)


class TicketReplyCreate(BaseModel):
    content: str = Field(min_length=1, max_length=8000)
    attachments: list[dict[str, Any]] | None = None

    @field_validator("content")
    @classmethod
    def validate_content(cls, value: str) -> str:
        return _reject_embedded_images(value)


class TicketRatingCreate(BaseModel):
    rating: int = Field(ge=1, le=5)
    comment: str | None = Field(default=None, max_length=1000)

    @field_validator("comment")
    @classmethod
    def validate_comment(cls, value: str | None) -> str | None:
        if value is None:
            return None
        return value.strip() or None


def _dt(value: datetime | None) -> str | None:
    return value.isoformat() if value else None


def _enum_value(value: Any) -> str | None:
    if value is None:
        return None
    return value.value if hasattr(value, "value") else str(value)


def _display_user(user: User | None) -> dict[str, Any] | None:
    if not user:
        return None
    return {
        "id": user.id,
        "username": user.username,
        "nickname": user.nickname,
        "email": user.email,
        "phone": user.phone,
    }


async def _serialize_assignee(db: AsyncSession, assignee_id: int | None) -> dict[str, Any] | None:
    if not assignee_id:
        return None
    user = await db.scalar(select(User).where(User.id == assignee_id))
    if not user:
        return None
    return {
        "id": user.id,
        "username": user.username,
        "nickname": user.nickname,
        "email": None,
        "phone": None,
    }


def _attachment_public_url(token: str) -> str:
    return f"{TICKET_ATTACHMENT_URL_PREFIX}/file/{token}"


def _attachment_thumbnail_url(token: str) -> str:
    return f"{TICKET_ATTACHMENT_URL_PREFIX}/thumbnail/{token}"


def _serialize_attachment(item: TicketAttachment) -> dict[str, Any]:
    return {
        "id": item.id,
        "token": item.token,
        "name": item.original_name,
        "url": _attachment_public_url(item.token),
        "thumbnail_url": _attachment_thumbnail_url(item.token),
        "size": item.size,
        "content_type": item.content_type,
        "path": item.storage_path,
        "created_at": _dt(item.created_at),
    }


ACTIVE_TICKET_STATUSES = {
    TicketStatus.OPEN,
    TicketStatus.PENDING,
    TicketStatus.PROCESSING,
    TicketStatus.WAITING_USER,
    TicketStatus.ANSWERED,
}


async def _new_ticket_no(db: AsyncSession) -> str:
    prefix = datetime.now(timezone.utc).strftime("T%Y%m%d")
    for _ in range(10):
        ticket_no = f"{prefix}{secrets.token_hex(3).upper()}"
        exists = await db.scalar(select(Ticket.id).where(Ticket.ticket_no == ticket_no))
        if not exists:
            return ticket_no
    raise HTTPException(status_code=500, detail="工单编号生成失败，请重试")


async def _message_count(db: AsyncSession, ticket_id: int) -> int:
    count = await db.scalar(
        select(func.count()).select_from(TicketMessage).where(TicketMessage.ticket_id == ticket_id)
    )
    return int(count or 0)


async def _message_counts_batch(db: AsyncSession, ticket_ids: list[int]) -> dict[int, int]:
    """Fetch message counts for multiple tickets in one query."""
    if not ticket_ids:
        return {}
    rows = (await db.execute(
        select(TicketMessage.ticket_id, func.count())
        .where(TicketMessage.ticket_id.in_(ticket_ids))
        .group_by(TicketMessage.ticket_id)
    )).all()
    return {row.ticket_id: int(row.count) for row in rows}


async def _serialize_instance(db: AsyncSession, instance_id: int | None, user_id: int | None = None) -> dict[str, Any] | None:
    if not instance_id:
        return None
    stmt = select(Instance).where(Instance.id == instance_id)
    if user_id is not None:
        stmt = stmt.where(Instance.user_id == user_id)
    instance = (await db.execute(stmt)).scalar_one_or_none()
    if not instance:
        return None
    return {
        "id": instance.id,
        "status": instance.status,
        "hostname": instance.hostname,
        "main_ip": instance.main_ip,
        "cpu": instance.cpu,
        "memory": instance.memory,
        "in_bw": instance.in_bw,
        "out_bw": instance.out_bw,
        "expires_at": _dt(instance.expires_at),
    }


def _serialize_satisfaction(ticket: Ticket) -> dict[str, Any] | None:
    if ticket.satisfaction_rating is None:
        return None
    return {
        "rating": ticket.satisfaction_rating,
        "comment": ticket.satisfaction_comment,
        "rated_at": _dt(ticket.satisfaction_rated_at),
    }


async def _serialize_ticket_summary(
    db: AsyncSession, ticket: Ticket, user_id: int | None = None,
    message_count: int | None = None,
) -> dict[str, Any]:
    if message_count is None:
        message_count = await _message_count(db, ticket.id)
    return {
        "id": ticket.id,
        "ticket_no": ticket.ticket_no,
        "subject": ticket.subject,
        "status": ticket.status,
        "priority": ticket.priority,
        "department_id": ticket.department_id,
        "department_name": ticket.department_name,
        "instance": await _serialize_instance(db, ticket.instance_id, user_id),
        "assignee": await _serialize_assignee(db, ticket.assignee_id),
        "assigned_at": _dt(ticket.assigned_at),
        "last_message_at": _dt(ticket.last_message_at),
        "last_reply_by": ticket.last_reply_by,
        "unread_count": ticket.user_unread_count if user_id else ticket.admin_unread_count,
        "message_count": message_count,
        "satisfaction": _serialize_satisfaction(ticket),
        "created_at": _dt(ticket.created_at),
        "updated_at": _dt(ticket.updated_at),
    }


def _serialize_department(department: TicketDepartment) -> dict[str, Any]:
    return {
        "id": department.id,
        "code": department.code,
        "name": department.name,
        "description": department.description,
        "is_enabled": department.is_enabled,
        "sort_order": department.sort_order,
        "min_member_level": department.min_member_level,
        "auto_reply_enabled": department.auto_reply_enabled,
        "auto_reply_content": department.auto_reply_content,
        "created_at": _dt(department.created_at),
        "updated_at": _dt(department.updated_at),
    }


async def _serialize_message(db: AsyncSession, message: TicketMessage, user_cache: dict[int, str] | None = None) -> dict[str, Any]:
    sender_name = None
    if message.sender_id:
        if user_cache is not None:
            if message.sender_id not in user_cache:
                user = await db.scalar(select(User.nickname, User.username).where(User.id == message.sender_id))
                user_cache[message.sender_id] = (user.nickname or user.username) if user else ""
            sender_name = user_cache[message.sender_id] or None
        else:
            user = await db.scalar(select(User).where(User.id == message.sender_id))
            sender_name = (user.nickname or user.username) if user else None
    return {
        "id": message.id,
        "ticket_id": message.ticket_id,
        "sender_type": message.sender_type,
        "sender_id": message.sender_id,
        "sender_name": sender_name,
        "content": message.content,
        "attachments": message.attachments or [],
        "is_internal": message.is_internal,
        "created_at": _dt(message.created_at),
    }


async def _serialize_messages_batch(db: AsyncSession, messages: list[TicketMessage]) -> list[dict[str, Any]]:
    """Batch-serialize messages with a single user lookup query to avoid N+1."""
    if not messages:
        return []

    # Collect unique sender IDs
    sender_ids = {m.sender_id for m in messages if m.sender_id}
    user_cache: dict[int, str] = {}
    if sender_ids:
        rows = (await db.execute(
            select(User.id, User.nickname, User.username).where(User.id.in_(sender_ids))
        )).all()
        for user_id, nickname, username in rows:
            user_cache[user_id] = nickname or username or ""

    return [await _serialize_message(db, m, user_cache) for m in messages]


async def _serialize_ticket_event(db: AsyncSession, event: TicketEvent, user_cache: dict[int, str] | None = None) -> dict[str, Any]:
    actor_name = None
    if event.actor_id:
        if user_cache is not None:
            actor_name = user_cache.get(event.actor_id)
        else:
            user = await db.scalar(select(User).where(User.id == event.actor_id))
            actor_name = (user.nickname or user.username) if user else None
    return {
        "id": event.id,
        "ticket_id": event.ticket_id,
        "actor_type": event.actor_type,
        "actor_id": event.actor_id,
        "actor_name": actor_name,
        "event_type": event.event_type,
        "from_value": event.from_value,
        "to_value": event.to_value,
        "message": event.message,
        "meta": event.meta or {},
        "created_at": _dt(event.created_at),
    }


async def _serialize_ticket_events_batch(db: AsyncSession, events: list[TicketEvent]) -> list[dict[str, Any]]:
    if not events:
        return []
    actor_ids = {event.actor_id for event in events if event.actor_id}
    user_cache: dict[int, str] = {}
    if actor_ids:
        rows = (await db.execute(
            select(User.id, User.nickname, User.username).where(User.id.in_(actor_ids))
        )).all()
        for user_id, nickname, username in rows:
            user_cache[user_id] = nickname or username or ""
    return [await _serialize_ticket_event(db, event, user_cache) for event in events]


async def _load_ticket_events(db: AsyncSession, ticket_id: int, *, include_internal: bool = False) -> list[dict[str, Any]]:
    stmt = select(TicketEvent).where(TicketEvent.ticket_id == ticket_id).order_by(TicketEvent.id.asc())
    if not include_internal:
        stmt = stmt.where(~TicketEvent.event_type.in_([
            TicketEventType.INTERNAL_NOTE_ADDED,
            TicketEventType.ASSIGNED,
        ]))
    events = (await db.execute(stmt)).scalars().all()
    return await _serialize_ticket_events_batch(db, events)


async def _add_ticket_event(
    db: AsyncSession,
    ticket: Ticket,
    actor_type: TicketSenderType,
    actor_id: int | None,
    event_type: TicketEventType,
    *,
    from_value: Any = None,
    to_value: Any = None,
    message: str | None = None,
    meta: dict[str, Any] | None = None,
) -> TicketEvent:
    event = TicketEvent(
        ticket_id=ticket.id,
        actor_type=actor_type,
        actor_id=actor_id,
        event_type=event_type,
        from_value=_enum_value(from_value),
        to_value=_enum_value(to_value),
        message=message,
        meta=meta or {},
    )
    db.add(event)
    await db.flush()
    await db.refresh(event)
    return event


def _extract_attachment_tokens(content: str | None, attachments: list[dict[str, Any]] | None = None) -> set[str]:
    tokens = set(TICKET_ATTACHMENT_TOKEN_RE.findall(content or ""))
    for item in attachments or []:
        token = item.get("token")
        if isinstance(token, str) and token.strip():
            tokens.add(token.strip())
        for key in ("url", "thumbnail_url"):
            value = item.get(key)
            if isinstance(value, str):
                tokens.update(TICKET_ATTACHMENT_TOKEN_RE.findall(value))
    return tokens


async def _claim_message_attachments(
    db: AsyncSession,
    *,
    ticket: Ticket,
    message: TicketMessage,
    sender_type: TicketSenderType,
    sender_id: int | None,
    content: str,
    attachments: list[dict[str, Any]] | None,
) -> list[dict[str, Any]]:
    tokens = _extract_attachment_tokens(content, attachments)
    if not tokens:
        return attachments or []

    rows = (await db.execute(
        select(TicketAttachment).where(TicketAttachment.token.in_(tokens))
    )).scalars().all()
    found = {item.token: item for item in rows}
    missing = sorted(tokens - set(found))
    if missing:
        raise HTTPException(status_code=400, detail="附件不存在或已失效")

    claimed: list[dict[str, Any]] = []
    for item in rows:
        if item.uploaded_by_type != sender_type or item.uploaded_by_id != sender_id:
            raise HTTPException(status_code=403, detail="附件不属于当前操作人")
        if item.is_claimed and item.ticket_id not in (None, ticket.id):
            raise HTTPException(status_code=403, detail="附件已绑定到其他工单")
        item.ticket_id = ticket.id
        item.message_id = message.id
        item.is_claimed = True
        claimed.append(_serialize_attachment(item))
    await db.flush()
    return claimed


async def _notify_admins_for_ticket(
    db: AsyncSession,
    ticket: Ticket,
    *,
    title: str,
    content: str,
    level: str = "info",
    prefer_assignee: bool = True,
) -> None:
    from app.services import notification_service

    try:
        user_ids: list[int] = []
        if prefer_assignee and ticket.assignee_id:
            user_ids = [int(ticket.assignee_id)]
        else:
            rows = (await db.execute(
                select(User.id).where(
                    User.role.in_([UserRole.ADMIN, UserRole.SUPER_ADMIN]),
                    User.status == UserStatus.ACTIVE,
                )
            )).scalars().all()
            user_ids = [int(uid) for uid in rows]
        if not user_ids:
            return
        await notification_service.notify_many(
            db,
            user_ids=user_ids,
            category="ticket",
            title=title,
            content=content,
            link=f"/admin/tickets/{ticket.id}",
            level=level,
            action_label="处理工单",
            action_url=f"/admin/tickets/{ticket.id}",
        )
    except Exception:
        return


async def _add_message(
    db: AsyncSession,
    ticket: Ticket,
    sender_type: TicketSenderType,
    sender_id: int | None,
    content: str,
    attachments: list[dict[str, Any]] | None = None,
    status_after: TicketStatus | None = None,
    is_internal: bool = False,
) -> TicketMessage:
    now = datetime.now(timezone.utc)
    message = TicketMessage(
        ticket_id=ticket.id,
        sender_type=sender_type,
        sender_id=sender_id,
        content=content.strip(),
        attachments=attachments or [],
        is_internal=is_internal,
    )
    db.add(message)
    ticket.last_message_at = now
    ticket.last_reply_by = sender_type
    if sender_type == TicketSenderType.USER:
        ticket.admin_unread_count += 1
    elif sender_type in (TicketSenderType.ADMIN, TicketSenderType.SYSTEM) and not is_internal:
        ticket.user_unread_count += 1
    if status_after:
        ticket.status = status_after
        if status_after not in (TicketStatus.CLOSED, TicketStatus.RESOLVED):
            ticket.closed_at = None
    await db.flush()
    await db.refresh(ticket)
    await db.refresh(message)
    return message


def _safe_upload_name(filename: str | None) -> str:
    if not filename:
        return "image"
    name = Path(filename).name.strip().replace("\\", "_").replace("/", "_")
    return name[:120] or "image"


async def _save_ticket_image_upload(
    db: AsyncSession,
    file: UploadFile,
    *,
    uploaded_by_type: TicketSenderType,
    uploaded_by_id: int | None,
) -> dict[str, Any]:
    content_type = (file.content_type or "").split(";")[0].strip().lower()
    ext = ALLOWED_TICKET_IMAGE_TYPES.get(content_type)
    if not ext:
        raise HTTPException(status_code=400, detail="仅支持 JPG、PNG、GIF、WebP 图片")

    # Read entire upload into memory
    body = await file.read()
    await file.close()
    if len(body) > settings.TICKET_IMAGE_MAX_BYTES:
        raise HTTPException(status_code=413, detail="图片不能超过 5MB")

    # Resize large images before saving to reduce bandwidth & frontend cost
    original_size = len(body)
    try:
        img = Image.open(io.BytesIO(body))
        img.load()
        w, h = img.size
        if w > TICKET_IMAGE_MAX_WIDTH or h > TICKET_IMAGE_MAX_HEIGHT:
            img.thumbnail((TICKET_IMAGE_MAX_WIDTH, TICKET_IMAGE_MAX_HEIGHT), Image.LANCZOS)
            buf = io.BytesIO()
            save_kwargs: dict[str, Any] = {"format": img.format or "JPEG"}
            if img.format == "JPEG" or (img.mode == "RGB" and content_type == "image/jpeg"):
                save_kwargs["quality"] = 85
                save_kwargs["optimize"] = True
            elif img.format == "PNG":
                save_kwargs["optimize"] = True
            elif img.format == "WEBP":
                save_kwargs["quality"] = 80
            img.save(buf, **save_kwargs)
            body = buf.getvalue()
    except (UnidentifiedImageError, OSError):
        # If we can't read the image, keep the original bytes
        pass

    now = datetime.now(timezone.utc)
    relative_dir = Path("ticket-images") / now.strftime("%Y") / now.strftime("%m") / now.strftime("%d")
    target_dir = settings.MEDIA_ROOT_PATH / relative_dir
    target_dir.mkdir(parents=True, exist_ok=True)

    filename = f"{now.strftime('%H%M%S')}_{secrets.token_hex(8)}{ext}"
    target = target_dir / filename
    size = len(body)
    try:
        async with aiofiles.open(target, "wb") as out:
            await out.write(body)
    except Exception:
        if target.exists():
            target.unlink(missing_ok=True)
        raise

    media_url = settings.MEDIA_URL.rstrip("/")
    path = "/".join([*relative_dir.parts, filename])
    token = secrets.token_urlsafe(24)
    attachment = TicketAttachment(
        token=token,
        storage_path=path,
        original_name=_safe_upload_name(file.filename),
        content_type=content_type,
        size=size,
        uploaded_by_type=uploaded_by_type,
        uploaded_by_id=uploaded_by_id,
    )
    db.add(attachment)
    await db.flush()
    await db.refresh(attachment)

    return {
        "id": attachment.id,
        "attachment_id": attachment.id,
        "token": attachment.token,
        "name": _safe_upload_name(file.filename),
        "url": _attachment_public_url(attachment.token),
        "thumbnail_url": _attachment_thumbnail_url(attachment.token),
        "legacy_url": f"{media_url}/{path}",
        "path": path,
        "size": size,
        "content_type": content_type,
    }


async def _get_user_ticket(db: AsyncSession, ticket_id: int, user_id: int) -> Ticket:
    ticket = await db.scalar(select(Ticket).where(Ticket.id == ticket_id, Ticket.user_id == user_id))
    if not ticket:
        raise HTTPException(status_code=404, detail="工单不存在")
    return ticket


async def _current_member_level(db: AsyncSession, user_id: int) -> int:
    membership = await membership_service.get_or_create_membership(db, user_id)
    if not membership.member_level_id:
        return 0
    level = await db.scalar(select(MemberLevel.level).where(MemberLevel.id == membership.member_level_id))
    return int(level or 0)


def _can_use_department(department: TicketDepartment, member_level: int) -> bool:
    return department.min_member_level is None or member_level >= department.min_member_level


@router.get("/departments", response_model=ApiResponse)
async def list_departments(
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    member_level = await _current_member_level(db, current_user.id)
    res = await db.execute(
        select(TicketDepartment)
        .where(TicketDepartment.is_enabled == True)  # noqa: E712
        .order_by(TicketDepartment.sort_order.asc(), TicketDepartment.id.asc())
    )
    return ApiResponse(data=[
        _serialize_department(item)
        for item in res.scalars().all()
        if _can_use_department(item, member_level)
    ])


@router.post("/attachments/images", response_model=ApiResponse)
async def upload_ticket_image(
    file: UploadFile = File(...),
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    return ApiResponse(data=await _save_ticket_image_upload(
        db,
        file,
        uploaded_by_type=TicketSenderType.USER,
        uploaded_by_id=current_user.id,
    ))


@router.get("", response_model=ApiResponse)
async def list_tickets(
    status: TicketStatus | None = Query(None),
    limit: int = Query(50, ge=1, le=100),
    offset: int = Query(0, ge=0),
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    stmt = (
        select(Ticket)
        .where(Ticket.user_id == current_user.id)
        .order_by(Ticket.last_message_at.desc(), Ticket.id.desc())
        .limit(limit)
        .offset(offset)
    )
    if status:
        stmt = stmt.where(Ticket.status == status)
    tickets = (await db.execute(stmt)).scalars().all()
    counts = await _message_counts_batch(db, [t.id for t in tickets])
    return ApiResponse(data=[
        await _serialize_ticket_summary(db, ticket, current_user.id, message_count=counts.get(ticket.id, 0))
        for ticket in tickets
    ])


@router.post("", response_model=ApiResponse)
async def create_ticket(
    body: TicketCreate,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    department = await db.scalar(
        select(TicketDepartment).where(
            TicketDepartment.id == body.department_id,
            TicketDepartment.is_enabled == True,  # noqa: E712
        )
    )
    if not department:
        raise HTTPException(status_code=404, detail="工单部门不存在或已停用")
    if not _can_use_department(department, await _current_member_level(db, current_user.id)):
        raise HTTPException(status_code=403, detail="当前会员等级暂不可提交该工单部门")

    if body.instance_id:
        instance = await db.scalar(
            select(Instance).where(Instance.id == body.instance_id, Instance.user_id == current_user.id)
        )
        if not instance:
            raise HTTPException(status_code=404, detail="关联实例不存在")

    ticket = Ticket(
        ticket_no=await _new_ticket_no(db),
        user_id=current_user.id,
        department_id=department.id,
        department_name=department.name,
        instance_id=body.instance_id,
        subject=body.subject.strip(),
        priority=body.priority,
        status=TicketStatus.OPEN,
    )
    db.add(ticket)
    await db.flush()

    await _add_ticket_event(
        db,
        ticket,
        TicketSenderType.USER,
        current_user.id,
        TicketEventType.CREATED,
        message="用户创建了工单",
        meta={"subject": ticket.subject, "department_name": ticket.department_name},
    )

    user_message = await _add_message(
        db,
        ticket,
        TicketSenderType.USER,
        current_user.id,
        body.content,
        body.attachments,
        TicketStatus.OPEN,
    )
    user_message.attachments = await _claim_message_attachments(
        db,
        ticket=ticket,
        message=user_message,
        sender_type=TicketSenderType.USER,
        sender_id=current_user.id,
        content=body.content,
        attachments=body.attachments,
    )
    await _add_ticket_event(
        db,
        ticket,
        TicketSenderType.USER,
        current_user.id,
        TicketEventType.USER_REPLIED,
        message="用户提交了初始问题描述",
    )
    if department.auto_reply_enabled and department.auto_reply_content:
        auto_message = await _add_message(
            db,
            ticket,
            TicketSenderType.SYSTEM,
            None,
            department.auto_reply_content,
            None,
            TicketStatus.OPEN,
        )
        auto_message.attachments = []

    await _notify_admins_for_ticket(
        db,
        ticket,
        title="有新的用户工单",
        content=f"用户提交了工单「{ticket.subject}」，请及时处理。",
        level="warning" if ticket.priority in (TicketPriority.HIGH, TicketPriority.URGENT) else "info",
        prefer_assignee=False,
    )

    await db.refresh(ticket)
    return ApiResponse(data=await _serialize_ticket_summary(db, ticket, current_user.id))


@router.get("/{ticket_id}", response_model=ApiResponse)
async def get_ticket(
    ticket_id: int,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    ticket = await _get_user_ticket(db, ticket_id, current_user.id)
    ticket.user_unread_count = 0
    await db.flush()
    await db.refresh(ticket)
    messages = (await db.execute(
        select(TicketMessage)
        .where(TicketMessage.ticket_id == ticket.id, TicketMessage.is_internal == False)  # noqa: E712
        .order_by(TicketMessage.id.asc())
    )).scalars().all()
    return ApiResponse(data={
        **await _serialize_ticket_summary(db, ticket, current_user.id),
        "messages": await _serialize_messages_batch(db, messages),
        "events": await _load_ticket_events(db, ticket.id, include_internal=False),
    })


@router.post("/{ticket_id}/messages", response_model=ApiResponse)
async def reply_ticket(
    ticket_id: int,
    body: TicketReplyCreate,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    ticket = await _get_user_ticket(db, ticket_id, current_user.id)
    if ticket.status in (TicketStatus.CLOSED, TicketStatus.RESOLVED):
        raise HTTPException(status_code=400, detail="工单已结束，请先重新打开")
    old_status = ticket.status
    message = await _add_message(
        db,
        ticket,
        TicketSenderType.USER,
        current_user.id,
        body.content,
        body.attachments,
        TicketStatus.PENDING,
    )
    message.attachments = await _claim_message_attachments(
        db,
        ticket=ticket,
        message=message,
        sender_type=TicketSenderType.USER,
        sender_id=current_user.id,
        content=body.content,
        attachments=body.attachments,
    )
    await _add_ticket_event(
        db,
        ticket,
        TicketSenderType.USER,
        current_user.id,
        TicketEventType.USER_REPLIED,
        from_value=old_status,
        to_value=TicketStatus.PENDING,
        message="用户补充了回复",
    )
    await _notify_admins_for_ticket(
        db,
        ticket,
        title="工单有新的用户回复",
        content=f"工单「{ticket.subject}」收到用户补充，请继续处理。",
        level="warning" if ticket.priority in (TicketPriority.HIGH, TicketPriority.URGENT) else "info",
    )
    return ApiResponse(data=await _serialize_message(db, message))


@router.post("/{ticket_id}/close", response_model=ApiResponse)
async def close_ticket(
    ticket_id: int,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    ticket = await _get_user_ticket(db, ticket_id, current_user.id)
    old_status = ticket.status
    ticket.status = TicketStatus.CLOSED
    ticket.closed_at = datetime.now(timezone.utc)
    await _add_ticket_event(
        db,
        ticket,
        TicketSenderType.USER,
        current_user.id,
        TicketEventType.CLOSED,
        from_value=old_status,
        to_value=TicketStatus.CLOSED,
        message="用户关闭了工单",
    )
    await _notify_admins_for_ticket(
        db,
        ticket,
        title="用户关闭了工单",
        content=f"工单「{ticket.subject}」已由用户关闭。",
    )
    await db.flush()
    await db.refresh(ticket)
    return ApiResponse(data=await _serialize_ticket_summary(db, ticket, current_user.id))


@router.post("/{ticket_id}/rating", response_model=ApiResponse)
async def rate_ticket(
    ticket_id: int,
    body: TicketRatingCreate,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    ticket = await _get_user_ticket(db, ticket_id, current_user.id)
    if ticket.status not in (TicketStatus.CLOSED, TicketStatus.RESOLVED):
        raise HTTPException(status_code=400, detail="工单关闭后才能评价")
    old_rating = ticket.satisfaction_rating
    ticket.satisfaction_rating = body.rating
    ticket.satisfaction_comment = body.comment
    ticket.satisfaction_rated_at = datetime.now(timezone.utc)
    await _add_ticket_event(
        db,
        ticket,
        TicketSenderType.USER,
        current_user.id,
        TicketEventType.RATED,
        from_value=str(old_rating) if old_rating else None,
        to_value=str(body.rating),
        message="用户提交了满意度评价",
        meta={"rating": body.rating, "has_comment": bool(body.comment)},
    )
    await db.flush()
    await db.refresh(ticket)
    return ApiResponse(data=await _serialize_ticket_summary(db, ticket, current_user.id))


@router.post("/{ticket_id}/reopen", response_model=ApiResponse)
async def reopen_ticket(
    ticket_id: int,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    ticket = await _get_user_ticket(db, ticket_id, current_user.id)
    if ticket.status in ACTIVE_TICKET_STATUSES:
        return ApiResponse(data=await _serialize_ticket_summary(db, ticket, current_user.id))
    old_status = ticket.status
    ticket.status = TicketStatus.OPEN
    ticket.closed_at = None
    await _add_message(
        db,
        ticket,
        TicketSenderType.SYSTEM,
        None,
        "用户重新打开了工单。",
        None,
        TicketStatus.OPEN,
    )
    await _add_ticket_event(
        db,
        ticket,
        TicketSenderType.USER,
        current_user.id,
        TicketEventType.REOPENED,
        from_value=old_status,
        to_value=TicketStatus.OPEN,
        message="用户重新打开了工单",
    )
    await _notify_admins_for_ticket(
        db,
        ticket,
        title="用户重新打开了工单",
        content=f"工单「{ticket.subject}」被用户重新打开，请继续处理。",
        level="warning",
    )
    return ApiResponse(data=await _serialize_ticket_summary(db, ticket, current_user.id))


THUMBNAIL_WIDTH = 480
THUMBNAIL_HEIGHT = 360


def _resolve_ticket_image_path(path: str) -> Path:
    normalized = path.replace("\\", "/").strip("/")
    if not normalized.startswith("ticket-images/"):
        raise HTTPException(status_code=400, detail="仅支持工单图片")

    full_path = (settings.MEDIA_ROOT_PATH / normalized).resolve()
    try:
        full_path.relative_to(settings.MEDIA_ROOT_PATH.resolve())
    except ValueError:
        raise HTTPException(status_code=400, detail="图片路径不合法")
    if not full_path.exists() or not full_path.is_file():
        raise HTTPException(status_code=404, detail="图片不存在")
    return full_path


async def _get_attachment_by_token(db: AsyncSession, token: str) -> TicketAttachment:
    attachment = await db.scalar(select(TicketAttachment).where(TicketAttachment.token == token))
    if not attachment:
        raise HTTPException(status_code=404, detail="附件不存在")
    return attachment


def _image_cache_headers(path: Path) -> dict[str, str]:
    stat = path.stat()
    return {
        "ETag": f'"{stat.st_mtime_ns}:{stat.st_size}"',
        "Cache-Control": "private, max-age=31536000, immutable",
    }


@router.api_route("/attachments/images/file/{token}", methods=["GET", "HEAD"])
async def ticket_image_file(
    token: str,
    db: AsyncSession = Depends(get_db),
):
    attachment = await _get_attachment_by_token(db, token)
    full_path = _resolve_ticket_image_path(attachment.storage_path)
    return FileResponse(
        full_path,
        media_type=attachment.content_type,
        filename=attachment.original_name or full_path.name,
        content_disposition_type="inline",
        headers=_image_cache_headers(full_path),
    )


def _thumbnail_response(request: Request, full_path: Path) -> Response:
    # ETag based on file mtime + size — enables 304 Not Modified responses
    stat = full_path.stat()
    etag = f'"{stat.st_mtime_ns}:{stat.st_size}"'
    if_none_match = request.headers.get("if-none-match")
    if if_none_match and if_none_match == etag:
        return Response(status_code=304)

    try:
        img = Image.open(full_path)
        img.load()
        w, h = img.size
        if w <= THUMBNAIL_WIDTH and h <= THUMBNAIL_HEIGHT:
            buf = io.BytesIO()
            img.save(buf, format=img.format or "JPEG")
            content_type = f"image/{img.format.lower()}" if img.format else "image/jpeg"
            return Response(
                content=buf.getvalue(), media_type=content_type,
                headers={"ETag": etag, "Cache-Control": "private, max-age=31536000, immutable"},
            )

        img.thumbnail((THUMBNAIL_WIDTH, THUMBNAIL_HEIGHT), Image.LANCZOS)
        buf = io.BytesIO()
        fmt = img.format or "JPEG"
        save_kwargs: dict[str, Any] = {"format": fmt}
        if fmt == "JPEG":
            save_kwargs["quality"] = 75
            save_kwargs["optimize"] = True
        elif fmt == "PNG":
            save_kwargs["optimize"] = True
        elif fmt == "WEBP":
            save_kwargs["quality"] = 70
        img.save(buf, **save_kwargs)
        content_type = f"image/{fmt.lower()}" if fmt else "image/jpeg"
        return Response(
            content=buf.getvalue(), media_type=content_type,
            headers={"ETag": etag, "Cache-Control": "private, max-age=31536000, immutable"},
        )
    except (UnidentifiedImageError, OSError):
        raise HTTPException(status_code=400, detail="图片无法处理")


@router.api_route("/attachments/images/thumbnail/{token}", methods=["GET", "HEAD"])
async def ticket_image_thumbnail_by_token(
    token: str,
    request: Request,
    db: AsyncSession = Depends(get_db),
):
    attachment = await _get_attachment_by_token(db, token)
    full_path = _resolve_ticket_image_path(attachment.storage_path)
    return _thumbnail_response(request, full_path)


@router.api_route("/attachments/images/thumbnail", methods=["GET", "HEAD"])
async def ticket_image_thumbnail(
    request: Request, path: str = Query(..., description="Relative media path"),
):
    """Serve a downscaled thumbnail of a ticket image for inline display."""
    return _thumbnail_response(request, _resolve_ticket_image_path(path))
