import uuid
from dataclasses import dataclass
from pathlib import Path
from urllib.parse import quote

from fastapi import HTTPException, UploadFile
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from yuxi.agents.backends.sandbox import (
    ensure_thread_dirs,
    sandbox_uploads_dir,
)
from yuxi.agents.buildin import agent_manager
from yuxi.config import config as app_config
from yuxi.knowledge.parser import Parser
from yuxi.repositories.agent_repository import AgentRepository
from yuxi.repositories.conversation_repository import ConversationRepository
from yuxi.services.mention_search_service import invalidate_mention_cache
from yuxi.storage.minio import StorageError, get_minio_client
from yuxi.storage.postgres.models_business import User
from yuxi.utils.datetime_utils import utc_isoformat
from yuxi.utils.logging_config import logger
from yuxi.utils.paths import VIRTUAL_PATH_UPLOADS
from yuxi.utils.upload_utils import read_upload_with_limit, write_upload_to_path

ATTACHMENT_ALLOWED_EXTENSIONS: tuple[str, ...] = ()
MAX_ATTACHMENT_SIZE_BYTES = 5 * 1024 * 1024  # 5 MB
MAX_ATTACHMENT_MARKDOWN_CHARS = 32_000  # TODO: 转 MARKDOWN的时候，不应该裁剪
TMP_ATTACHMENT_PREFIX = "tmp/chat_attachments"
TMP_ATTACHMENT_PARSE_EXTENSIONS = (".pdf", ".jpg", ".jpeg", ".png", ".bmp", ".tiff", ".tif")
TMP_ATTACHMENT_IMAGE_EXTENSIONS = (".jpg", ".jpeg", ".png", ".bmp", ".tiff", ".tif")
TMP_ATTACHMENT_OCR_METHODS = (
    "rapid_ocr",
    "mineru_ocr",
    "mineru_official",
    "pp_structure_v3_ocr",
    "deepseek_ocr",
)
TMP_ATTACHMENT_PARSE_METHODS = ("disable", *TMP_ATTACHMENT_OCR_METHODS)


@dataclass(slots=True)
class ConversionResult:
    """Represents the normalized output of an uploaded attachment."""

    file_id: str
    file_name: str
    file_type: str | None
    file_size: int
    markdown: str
    truncated: bool


def _ensure_workdir() -> Path:
    workdir = Path(app_config.save_dir) / "uploads" / "chat_attachments"
    workdir.mkdir(parents=True, exist_ok=True)
    return workdir


async def _write_upload_to_disk(upload: UploadFile, dest: Path) -> int:
    return await write_upload_to_path(
        upload,
        dest,
        max_size_bytes=MAX_ATTACHMENT_SIZE_BYTES,
        too_large_message="附件过大，当前仅支持 5 MB 以内的文件",
    )


def _truncate_markdown(markdown: str) -> tuple[str, bool]:
    if len(markdown) <= MAX_ATTACHMENT_MARKDOWN_CHARS:
        return markdown, False

    truncated_content = markdown[: MAX_ATTACHMENT_MARKDOWN_CHARS - 100].rstrip()
    truncated_content = f"{truncated_content}\n\n[内容已截断，超出 {MAX_ATTACHMENT_MARKDOWN_CHARS} 字符限制]"
    return truncated_content, True


async def _convert_upload_to_markdown(upload: UploadFile) -> ConversionResult:
    """Persist an UploadFile temporarily, convert it to markdown, and clean up."""
    if not upload.filename:
        raise ValueError("无法识别的文件名")

    file_name = Path(upload.filename).name
    suffix = Path(file_name).suffix.lower()

    if ATTACHMENT_ALLOWED_EXTENSIONS and suffix not in ATTACHMENT_ALLOWED_EXTENSIONS:
        allowed = ", ".join(ATTACHMENT_ALLOWED_EXTENSIONS)
        raise ValueError(f"不支持的文件类型: {suffix or '未知'}，当前仅支持 {allowed}")

    temp_dir = _ensure_workdir()
    temp_path = temp_dir / f"{uuid.uuid4().hex}{suffix}"

    try:
        file_size = await _write_upload_to_disk(upload, temp_path)
        markdown = await Parser.aparse(str(temp_path))
        markdown, truncated = _truncate_markdown(markdown)
        return ConversionResult(
            file_id=uuid.uuid4().hex,
            file_name=file_name,
            file_type=upload.content_type,
            file_size=file_size,
            markdown=markdown,
            truncated=truncated,
        )
    except Exception as exc:  # noqa: BLE001
        logger.error(f"Attachment conversion failed: {exc}")
        raise


async def require_user_conversation(conv_repo: ConversationRepository, thread_id: str, uid: str):
    conversation = await conv_repo.get_conversation_by_thread_id(thread_id)
    if not conversation or conversation.uid != str(uid) or conversation.status == "deleted":
        raise HTTPException(status_code=404, detail="对话线程不存在")
    return conversation


def _safe_file_name(file_name: str | None, default: str = "attachment.bin") -> str:
    safe_name = Path(file_name or "").name.replace("/", "_").replace("\\", "_").strip(" .")
    return safe_name or default


def _make_upload_virtual_path(file_name: str) -> str:
    return f"{VIRTUAL_PATH_UPLOADS}/{_safe_file_name(file_name)}"


def _make_attachment_path(file_name: str) -> str:
    """生成附件在沙盒用户目录中的统一路径。"""
    file_name = _safe_file_name(file_name)
    # 提取不带扩展名的部分
    base_name = file_name
    for ext in [".docx", ".txt", ".html", ".htm", ".pdf", ".md"]:
        if file_name.lower().endswith(ext):
            base_name = file_name[: -len(ext)]
            break

    # 替换路径分隔符
    safe_name = base_name.replace("/", "_").replace("\\", "_")
    return f"{safe_name}.md"


def _build_attachment_storage_path(*, uid: str, thread_id: str, file_name: str) -> tuple[str, Path]:
    """返回附件虚拟路径和宿主机落盘路径。"""
    relative_name = _make_attachment_path(file_name)
    virtual_path = f"{VIRTUAL_PATH_UPLOADS}/attachments/{relative_name}"

    host_dir = Path(app_config.save_dir) / "threads" / thread_id / "user-data" / "uploads" / "attachments"
    host_dir.mkdir(parents=True, exist_ok=True)
    host_path = host_dir / relative_name

    return virtual_path, host_path


def _artifact_url(thread_id: str, virtual_path: str) -> str:
    return f"/api/chat/thread/{thread_id}/artifacts/{virtual_path.lstrip('/')}"


def _tmp_attachment_prefix(uid: str, tmp_file_id: str) -> str:
    return f"{TMP_ATTACHMENT_PREFIX}/{uid}/{tmp_file_id}"


def _get_tmp_attachment_bucket() -> str:
    return get_minio_client().KB_BUCKETS["documents"]


def _make_tmp_attachment_object(uid: str, file_name: str) -> tuple[str, str]:
    """生成用户隔离的 tmp 对象路径。"""
    tmp_file_id = uuid.uuid4().hex
    safe_name = _safe_file_name(file_name)
    return tmp_file_id, f"{_tmp_attachment_prefix(uid, tmp_file_id)}/original/{safe_name}"


def _make_tmp_parsed_object(uid: str, tmp_file_id: str, file_name: str) -> str:
    stem = Path(_safe_file_name(file_name)).stem or "attachment"
    return f"{_tmp_attachment_prefix(uid, tmp_file_id)}/parsed/{stem}.md"


def _minio_source(bucket_name: str, object_name: str) -> str:
    return f"minio://{bucket_name}/{quote(object_name, safe='/')}"


def _parse_user_tmp_object(object_name: str, uid: str) -> tuple[str, str, str]:
    if not object_name or "\\" in object_name:
        raise HTTPException(status_code=400, detail="无效的临时附件路径")

    user_prefix = f"{TMP_ATTACHMENT_PREFIX}/{uid}/"
    if not object_name.startswith(user_prefix):
        raise HTTPException(status_code=403, detail="无权访问该临时附件")

    parts = object_name[len(user_prefix) :].split("/")
    if len(parts) != 3 or any(not part or part in {".", ".."} for part in parts):
        raise HTTPException(status_code=400, detail="无效的临时附件路径")

    return parts[0], parts[1], parts[2]


def _require_tmp_object_section(
    object_name: str,
    uid: str,
    section: str,
    tmp_file_id: str | None = None,
) -> tuple[str, str]:
    current_tmp_file_id, current_section, object_file_name = _parse_user_tmp_object(object_name, uid)
    if current_section != section or (tmp_file_id is not None and current_tmp_file_id != tmp_file_id):
        raise HTTPException(status_code=400, detail="无效的临时附件路径")
    if section == "parsed" and Path(object_file_name).suffix.lower() != ".md":
        raise HTTPException(status_code=400, detail="无效的解析附件路径")
    return current_tmp_file_id, object_file_name


def _normalize_parse_method(file_name: str, parse_method: str | None) -> str:
    suffix = Path(file_name).suffix.lower()
    if suffix not in TMP_ATTACHMENT_PARSE_EXTENSIONS:
        raise HTTPException(status_code=400, detail="当前仅支持 PDF 和图片附件解析")

    method = parse_method or ("rapid_ocr" if suffix in TMP_ATTACHMENT_IMAGE_EXTENSIONS else "disable")
    if suffix in TMP_ATTACHMENT_IMAGE_EXTENSIONS:
        allowed_methods = TMP_ATTACHMENT_OCR_METHODS
    else:
        allowed_methods = TMP_ATTACHMENT_PARSE_METHODS

    if method not in allowed_methods:
        allowed = ", ".join(allowed_methods)
        raise HTTPException(status_code=400, detail=f"不支持的解析方法: {method}，可选: {allowed}")
    return method


def _build_state_uploads(attachments: list[dict]) -> list[dict]:
    uploads: list[dict] = []
    for attachment in attachments:
        path = attachment.get("path")
        if not isinstance(path, str) or not path.strip():
            continue

        uploads.append(
            {
                "file_id": attachment.get("file_id"),
                "file_name": attachment.get("file_name"),
                "file_type": attachment.get("file_type"),
                "file_size": attachment.get("file_size", 0),
                "status": attachment.get("status", "uploaded"),
                "uploaded_at": attachment.get("uploaded_at"),
                "path": path,
                "artifact_url": attachment.get("artifact_url"),
                "request_id": attachment.get("request_id"),
            }
        )
    return uploads


async def _sync_thread_upload_state(
    *,
    thread_id: str,
    uid: str,
    agent_id: str,
    backend_id: str | None,
    attachments: list[dict],
) -> None:
    try:
        agent = agent_manager.get_agent(backend_id or agent_id)
        if not agent:
            logger.warning(f"Skip upload state sync: agent not found ({agent_id})")
            return

        graph = await agent.get_graph()
        config = {"configurable": {"thread_id": thread_id, "uid": str(uid)}}

        await graph.aupdate_state(
            config=config,
            values={
                "uploads": _build_state_uploads(attachments),
            },
        )
    except Exception as exc:  # noqa: BLE001
        logger.warning(f"Failed to sync upload state for thread {thread_id}: {exc}")


def serialize_attachment(record: dict) -> dict:
    path = record.get("path")
    return {
        "file_id": record.get("file_id"),
        "file_name": record.get("file_name"),
        "file_type": record.get("file_type"),
        "file_size": record.get("file_size", 0),
        "status": record.get("status", "uploaded"),
        "uploaded_at": record.get("uploaded_at"),
        "path": path,
        "artifact_url": record.get("artifact_url"),
        "original_path": record.get("original_path"),
        "original_artifact_url": record.get("original_artifact_url"),
        "minio_url": record.get("minio_url"),
        "request_id": record.get("request_id"),
    }


async def _materialize_attachment_files(
    *,
    thread_id: str,
    uid: str,
    upload: UploadFile,
    file_name: str,
    file_content: bytes,
) -> dict:
    """将原始附件与可选 markdown 副本落盘到线程 user-data。"""
    ensure_thread_dirs(thread_id, uid)

    upload_virtual_path = _make_upload_virtual_path(file_name)
    uploads_dir = sandbox_uploads_dir(thread_id)
    upload_actual_path = uploads_dir / Path(upload_virtual_path).name
    upload_actual_path.write_bytes(file_content)

    record = {
        "status": "uploaded",
        "path": upload_virtual_path,
        "artifact_url": _artifact_url(thread_id, upload_virtual_path),
        "storage_path": str(upload_actual_path),
        "original_path": upload_virtual_path,
        "original_artifact_url": _artifact_url(thread_id, upload_virtual_path),
        "original_storage_path": str(upload_actual_path),
        "minio_url": None,
    }

    try:
        await upload.seek(0)
        conversion = await _convert_upload_to_markdown(upload)
    except ValueError:
        return record
    except Exception as exc:  # noqa: BLE001
        logger.warning(f"Attachment markdown materialization failed for {file_name}: {exc}")
        return record

    markdown_virtual_path, markdown_host_path = _build_attachment_storage_path(
        uid=uid,
        thread_id=thread_id,
        file_name=file_name,
    )
    markdown_host_path.write_text(conversion.markdown, encoding="utf-8")

    record.update(
        {
            "status": "parsed",
            "path": markdown_virtual_path,
            "artifact_url": _artifact_url(thread_id, markdown_virtual_path),
            "storage_path": str(markdown_host_path),
            "file_path": markdown_virtual_path,
            "markdown": conversion.markdown,
            "truncated": conversion.truncated,
            "markdown_storage_path": str(markdown_host_path),
        }
    )
    return record


def _materialize_tmp_attachment_files(
    *,
    thread_id: str,
    uid: str,
    file_id: str,
    file_name: str,
    file_content: bytes,
    parsed_markdown: str | None = None,
    truncated: bool = False,
) -> dict:
    """将 tmp 附件复制到线程目录，不主动删除 tmp 对象。"""
    ensure_thread_dirs(thread_id, uid)

    storage_name = f"{file_id}_{file_name}"
    upload_virtual_path = _make_upload_virtual_path(storage_name)
    uploads_dir = sandbox_uploads_dir(thread_id)
    upload_actual_path = uploads_dir / Path(upload_virtual_path).name
    upload_actual_path.write_bytes(file_content)

    record = {
        "status": "uploaded",
        "path": upload_virtual_path,
        "artifact_url": _artifact_url(thread_id, upload_virtual_path),
        "storage_path": str(upload_actual_path),
        "original_path": upload_virtual_path,
        "original_artifact_url": _artifact_url(thread_id, upload_virtual_path),
        "original_storage_path": str(upload_actual_path),
        "minio_url": None,
    }

    if parsed_markdown is None:
        return record

    markdown_virtual_path, markdown_host_path = _build_attachment_storage_path(
        uid=uid,
        thread_id=thread_id,
        file_name=storage_name,
    )
    markdown_host_path.write_text(parsed_markdown, encoding="utf-8")
    record.update(
        {
            "status": "parsed",
            "path": markdown_virtual_path,
            "artifact_url": _artifact_url(thread_id, markdown_virtual_path),
            "storage_path": str(markdown_host_path),
            "file_path": markdown_virtual_path,
            "markdown": parsed_markdown,
            "truncated": truncated,
            "markdown_storage_path": str(markdown_host_path),
        }
    )
    return record


async def create_thread_view(
    *,
    agent_id: str,
    title: str | None,
    metadata: dict | None,
    db: AsyncSession,
    current_uid: str,
) -> dict:
    user_result = await db.execute(select(User).where(User.uid == str(current_uid)))
    current_user = user_result.scalar_one_or_none()
    if not current_user:
        raise HTTPException(status_code=404, detail="用户不存在")

    agent_repo = AgentRepository(db)
    agent_item = await agent_repo.get_visible_by_slug(slug=agent_id, user=current_user)
    if not agent_item:
        raise HTTPException(status_code=404, detail="智能体不存在")

    thread_id = str(uuid.uuid4())
    conv_repo = ConversationRepository(db)
    thread_metadata = dict(metadata or {})
    thread_metadata["backend_id"] = agent_item.backend_id
    conversation = await conv_repo.create_conversation(
        uid=str(current_uid),
        agent_id=agent_item.slug,
        title=title or "新的对话",
        thread_id=thread_id,
        metadata=thread_metadata,
    )

    return {
        "id": conversation.thread_id,
        "uid": conversation.uid,
        "agent_id": conversation.agent_id,
        "title": conversation.title,
        "created_at": conversation.created_at.isoformat(),
        "updated_at": conversation.updated_at.isoformat(),
        "metadata": conversation.extra_metadata or {},
    }


async def list_threads_view(
    *,
    agent_id: str | None,
    db: AsyncSession,
    current_uid: str,
    limit: int | None = None,
    offset: int = 0,
) -> list[dict]:
    conv_repo = ConversationRepository(db)
    conversations = await conv_repo.list_conversations(
        uid=str(current_uid),
        agent_id=agent_id,
        status="active",
        limit=limit,
        offset=offset,
    )

    return [
        {
            "id": conv.thread_id,
            "uid": conv.uid,
            "agent_id": conv.agent_id,
            "title": conv.title,
            "is_pinned": bool(conv.is_pinned),
            "created_at": conv.created_at.isoformat(),
            "updated_at": conv.updated_at.isoformat(),
            "metadata": conv.extra_metadata or {},
        }
        for conv in conversations
    ]


async def delete_thread_view(
    *,
    thread_id: str,
    db: AsyncSession,
    current_uid: str,
) -> dict:
    conv_repo = ConversationRepository(db)
    await require_user_conversation(conv_repo, thread_id, str(current_uid))
    deleted = await conv_repo.delete_conversation(thread_id, soft_delete=True)
    if not deleted:
        raise HTTPException(status_code=404, detail="对话线程不存在")
    return {"message": "删除成功"}


async def update_thread_view(
    *,
    thread_id: str,
    title: str | None = None,
    is_pinned: bool | None = None,
    db: AsyncSession,
    current_uid: str,
) -> dict:
    conv_repo = ConversationRepository(db)
    await require_user_conversation(conv_repo, thread_id, str(current_uid))
    updated_conv = await conv_repo.update_conversation(thread_id, title=title, is_pinned=is_pinned)
    if not updated_conv:
        raise HTTPException(status_code=500, detail="更新失败")
    return {
        "id": updated_conv.thread_id,
        "uid": updated_conv.uid,
        "agent_id": updated_conv.agent_id,
        "title": updated_conv.title,
        "is_pinned": bool(updated_conv.is_pinned),
        "created_at": updated_conv.created_at.isoformat(),
        "updated_at": updated_conv.updated_at.isoformat(),
        "metadata": updated_conv.extra_metadata or {},
    }


async def upload_tmp_attachment_view(*, file: UploadFile, current_uid: str) -> dict:
    """上传附件到用户隔离的 MinIO tmp 路径。"""
    if not file.filename:
        raise HTTPException(status_code=400, detail="无法识别的文件名")

    file_name = _safe_file_name(file.filename)
    try:
        file_content = await read_upload_with_limit(
            file,
            max_size_bytes=MAX_ATTACHMENT_SIZE_BYTES,
            too_large_message="附件过大，当前仅支持 5 MB 以内的文件",
        )
    except ValueError as exc:
        raise HTTPException(status_code=400, detail=str(exc)) from exc

    file_size = len(file_content)
    tmp_file_id, object_name = _make_tmp_attachment_object(str(current_uid), file_name)
    minio_client = get_minio_client()
    bucket_name = _get_tmp_attachment_bucket()
    try:
        upload_result = await minio_client.aupload_file(
            bucket_name=bucket_name,
            object_name=object_name,
            data=file_content,
            content_type=file.content_type,
        )
    except StorageError as exc:
        raise HTTPException(status_code=500, detail=f"临时附件上传失败: {exc}") from exc

    suffix = Path(file_name).suffix.lower()
    if suffix == ".pdf":
        parse_methods = list(TMP_ATTACHMENT_PARSE_METHODS)
    elif suffix in TMP_ATTACHMENT_IMAGE_EXTENSIONS:
        parse_methods = list(TMP_ATTACHMENT_OCR_METHODS)
    else:
        parse_methods = []

    return {
        "tmp_file_id": tmp_file_id,
        "file_name": file_name,
        "file_type": file.content_type,
        "file_size": file_size,
        "bucket_name": upload_result.bucket_name,
        "object_name": upload_result.object_name,
        "minio_url": upload_result.url,
        "uploaded_at": utc_isoformat(),
        "parse_supported": bool(parse_methods),
        "parse_methods": parse_methods,
    }


async def parse_tmp_attachment_view(
    *,
    object_name: str,
    file_name: str,
    parse_method: str | None,
    bucket_name: str | None,
    current_uid: str,
) -> dict:
    """解析用户 tmp 附件并把 markdown 写回 tmp。"""
    minio_client = get_minio_client()
    expected_bucket = _get_tmp_attachment_bucket()
    bucket_name = bucket_name or expected_bucket
    if bucket_name != expected_bucket:
        raise HTTPException(status_code=400, detail="无效的临时附件 bucket")

    tmp_file_id, safe_name = _require_tmp_object_section(object_name, str(current_uid), "original")
    method = _normalize_parse_method(safe_name, parse_method)

    try:
        markdown = await Parser.aparse(_minio_source(bucket_name, object_name), params={"ocr_engine": method})
        markdown, truncated = _truncate_markdown(markdown)
        parsed_object_name = _make_tmp_parsed_object(str(current_uid), tmp_file_id, safe_name)
        upload_result = await minio_client.aupload_file(
            bucket_name=bucket_name,
            object_name=parsed_object_name,
            data=markdown.encode("utf-8"),
            content_type="text/markdown; charset=utf-8",
        )
    except StorageError as exc:
        raise HTTPException(status_code=400, detail=f"读取临时附件失败: {exc}") from exc
    except Exception as exc:  # noqa: BLE001
        logger.warning(f"Tmp attachment parse failed for {safe_name}: {exc}")
        raise HTTPException(status_code=400, detail=f"附件解析失败: {exc}") from exc

    return {
        "tmp_file_id": tmp_file_id,
        "file_name": safe_name,
        "bucket_name": upload_result.bucket_name,
        "object_name": object_name,
        "parsed_object_name": upload_result.object_name,
        "parsed_minio_url": upload_result.url,
        "parse_method": method,
        "status": "parsed",
        "truncated": truncated,
    }


async def confirm_tmp_thread_attachments_view(
    *,
    thread_id: str,
    attachments: list[dict],
    db: AsyncSession,
    current_uid: str,
) -> dict:
    """将选中的 tmp 附件正式关联到对话线程。"""
    if not attachments:
        raise HTTPException(status_code=400, detail="请选择要添加的附件")

    conv_repo = ConversationRepository(db)
    conversation = await require_user_conversation(conv_repo, thread_id, str(current_uid))
    minio_client = get_minio_client()
    expected_bucket = _get_tmp_attachment_bucket()
    prepared_items: list[dict] = []

    for item in attachments:
        object_name = str(item.get("object_name") or "")
        bucket_name = str(item.get("bucket_name") or expected_bucket)
        if bucket_name != expected_bucket:
            raise HTTPException(status_code=400, detail="无效的临时附件 bucket")

        tmp_file_id, file_name = _require_tmp_object_section(object_name, str(current_uid), "original")
        try:
            file_content = await minio_client.adownload_file(bucket_name, object_name)
        except StorageError as exc:
            raise HTTPException(status_code=400, detail=f"读取临时附件失败: {exc}") from exc

        if len(file_content) > MAX_ATTACHMENT_SIZE_BYTES:
            max_size_mb = MAX_ATTACHMENT_SIZE_BYTES // (1024 * 1024)
            raise HTTPException(status_code=400, detail=f"附件过大，当前仅支持 {max_size_mb} MB 以内的文件")

        parsed_markdown = None
        parsed_object_name = str(item.get("parsed_object_name") or "")
        if parsed_object_name:
            _require_tmp_object_section(parsed_object_name, str(current_uid), "parsed", tmp_file_id)
            expected_parsed_object = _make_tmp_parsed_object(str(current_uid), tmp_file_id, file_name)
            if parsed_object_name != expected_parsed_object:
                raise HTTPException(status_code=400, detail="解析附件路径无效")
            try:
                parsed_bytes = await minio_client.adownload_file(bucket_name, parsed_object_name)
                parsed_markdown = parsed_bytes.decode("utf-8")
            except StorageError as exc:
                raise HTTPException(status_code=400, detail=f"读取解析附件失败: {exc}") from exc
            except UnicodeDecodeError as exc:
                raise HTTPException(status_code=400, detail="解析附件内容不是有效的 Markdown 文本") from exc

        prepared_items.append(
            {
                "file_name": file_name,
                "file_type": item.get("file_type"),
                "file_content": file_content,
                "parsed_markdown": parsed_markdown,
                "truncated": bool(item.get("truncated")),
            }
        )

    added_records: list[dict] = []
    for prepared in prepared_items:
        file_id = uuid.uuid4().hex
        materialized = _materialize_tmp_attachment_files(
            thread_id=thread_id,
            uid=str(conversation.uid),
            file_id=file_id,
            file_name=prepared["file_name"],
            file_content=prepared["file_content"],
            parsed_markdown=prepared["parsed_markdown"],
            truncated=prepared["truncated"],
        )
        attachment_record = {
            "file_id": file_id,
            "file_name": prepared["file_name"],
            "file_type": prepared["file_type"],
            "file_size": len(prepared["file_content"]),
            "status": materialized["status"],
            "uploaded_at": utc_isoformat(),
            "path": materialized["path"],
            "artifact_url": materialized["artifact_url"],
            "storage_path": materialized["storage_path"],
            "original_path": materialized["original_path"],
            "original_artifact_url": materialized["original_artifact_url"],
            "original_storage_path": materialized["original_storage_path"],
            "minio_url": materialized["minio_url"],
        }
        for optional_key in ("file_path", "markdown", "truncated", "markdown_storage_path"):
            if optional_key in materialized:
                attachment_record[optional_key] = materialized[optional_key]
        added_records.append(attachment_record)

    await conv_repo.add_attachments(conversation.id, added_records)
    all_attachments = await conv_repo.get_attachments(conversation.id)
    await _sync_thread_upload_state(
        thread_id=thread_id,
        uid=str(current_uid),
        agent_id=conversation.agent_id,
        backend_id=(conversation.extra_metadata or {}).get("backend_id"),
        attachments=all_attachments,
    )
    await invalidate_mention_cache(thread_id)

    return {"attachments": [serialize_attachment(item) for item in added_records]}


async def upload_thread_attachment_view(
    *,
    thread_id: str,
    file: UploadFile,
    db: AsyncSession,
    current_uid: str,
) -> dict:
    conv_repo = ConversationRepository(db)
    conversation = await require_user_conversation(conv_repo, thread_id, str(current_uid))
    if not file.filename:
        raise HTTPException(status_code=400, detail="无法识别的文件名")

    file_name = Path(file.filename).name
    await file.seek(0)
    file_content = await file.read()
    file_size = len(file_content)
    if file_size > MAX_ATTACHMENT_SIZE_BYTES:
        max_size_mb = MAX_ATTACHMENT_SIZE_BYTES // (1024 * 1024)
        raise HTTPException(status_code=400, detail=f"附件过大，当前仅支持 {max_size_mb} MB 以内的文件")
    materialized = await _materialize_attachment_files(
        thread_id=thread_id,
        uid=str(conversation.uid),
        upload=file,
        file_name=file_name,
        file_content=file_content,
    )

    attachment_record = {
        "file_id": uuid.uuid4().hex,
        "file_name": file_name,
        "file_type": file.content_type,
        "file_size": file_size,
        "status": materialized["status"],
        "uploaded_at": utc_isoformat(),
        "path": materialized["path"],
        "artifact_url": materialized["artifact_url"],
        "storage_path": materialized["storage_path"],
        "original_path": materialized["original_path"],
        "original_artifact_url": materialized["original_artifact_url"],
        "original_storage_path": materialized["original_storage_path"],
        "minio_url": materialized["minio_url"],
    }
    for optional_key in ("file_path", "markdown", "truncated", "markdown_storage_path"):
        if optional_key in materialized:
            attachment_record[optional_key] = materialized[optional_key]

    await conv_repo.add_attachment(conversation.id, attachment_record)
    all_attachments = await conv_repo.get_attachments(conversation.id)
    await _sync_thread_upload_state(
        thread_id=thread_id,
        uid=str(current_uid),
        agent_id=conversation.agent_id,
        backend_id=(conversation.extra_metadata or {}).get("backend_id"),
        attachments=all_attachments,
    )

    await invalidate_mention_cache(thread_id)

    return serialize_attachment(attachment_record)


async def list_thread_attachments_view(
    *,
    thread_id: str,
    db: AsyncSession,
    current_uid: str,
) -> dict:
    conv_repo = ConversationRepository(db)
    conversation = await require_user_conversation(conv_repo, thread_id, str(current_uid))
    attachments = await conv_repo.get_attachments(conversation.id)
    return {
        "attachments": [serialize_attachment(item) for item in attachments],
        "limits": {
            "allowed_extensions": sorted(ATTACHMENT_ALLOWED_EXTENSIONS),
            "max_size_bytes": MAX_ATTACHMENT_SIZE_BYTES,
        },
    }


async def delete_thread_attachment_view(
    *,
    thread_id: str,
    file_id: str,
    db: AsyncSession,
    current_uid: str,
) -> dict:
    conv_repo = ConversationRepository(db)
    conversation = await require_user_conversation(conv_repo, thread_id, str(current_uid))

    existing_attachments = await conv_repo.get_attachments(conversation.id)
    target_attachment = next((item for item in existing_attachments if item.get("file_id") == file_id), None)

    removed = await conv_repo.remove_attachment(conversation.id, file_id)
    if not removed:
        raise HTTPException(status_code=404, detail="附件不存在或已被删除")

    if target_attachment:
        delete_candidates = {
            str(value).strip()
            for value in (
                target_attachment.get("storage_path"),
                target_attachment.get("original_storage_path"),
                target_attachment.get("markdown_storage_path"),
            )
            if isinstance(value, str) and value.strip()
        }
        for candidate in delete_candidates:
            try:
                file_path = Path(candidate)
                if file_path.exists():
                    file_path.unlink()
            except Exception as exc:  # noqa: BLE001
                logger.warning(f"Failed to remove attachment file {candidate}: {exc}")

    all_attachments = await conv_repo.get_attachments(conversation.id)
    await _sync_thread_upload_state(
        thread_id=thread_id,
        uid=str(current_uid),
        agent_id=conversation.agent_id,
        backend_id=(conversation.extra_metadata or {}).get("backend_id"),
        attachments=all_attachments,
    )

    await invalidate_mention_cache(thread_id)

    return {"message": "附件已删除"}


async def get_thread_history_view(
    *,
    thread_id: str,
    current_uid: str,
    db: AsyncSession,
) -> dict:
    """获取对话历史消息，包含用户反馈状态"""
    conv_repo = ConversationRepository(db)
    conversation = await conv_repo.get_conversation_by_thread_id(thread_id)
    if not conversation or conversation.uid != str(current_uid) or conversation.status == "deleted":
        raise HTTPException(status_code=404, detail="对话线程不存在")

    messages = await conv_repo.get_messages_by_thread_id(thread_id)
    message_request_ids = set()
    for msg in messages:
        request_id = (msg.extra_metadata or {}).get("request_id")
        if msg.role == "user" and request_id:
            message_request_ids.add(str(request_id))
    attachments_by_request_id: dict[str, list[dict]] = {}
    if message_request_ids:
        for attachment in await conv_repo.get_attachments(conversation.id):
            request_id = attachment.get("request_id")
            if not request_id or str(request_id) not in message_request_ids:
                continue
            attachments_by_request_id.setdefault(str(request_id), []).append(serialize_attachment(attachment))

    history: list[dict] = []
    role_type_map = {"user": "human", "assistant": "ai", "tool": "tool", "system": "system"}

    for msg in messages:
        user_feedback = None
        if msg.feedbacks:
            for feedback in msg.feedbacks:
                if feedback.uid == str(current_uid):
                    user_feedback = {
                        "id": feedback.id,
                        "rating": feedback.rating,
                        "reason": feedback.reason,
                        "created_at": feedback.created_at.isoformat() if feedback.created_at else None,
                    }
                    break

        extra_metadata = dict(msg.extra_metadata or {})
        request_id = extra_metadata.get("request_id")
        if msg.role == "user" and request_id and not extra_metadata.get("attachments"):
            extra_metadata["attachments"] = attachments_by_request_id.get(str(request_id), [])

        msg_dict = {
            "id": msg.id,
            "type": role_type_map.get(msg.role, msg.role),
            "content": msg.content,
            "created_at": msg.created_at.isoformat() if msg.created_at else None,
            "error_type": extra_metadata.get("error_type"),
            "error_message": extra_metadata.get("error_message"),
            "extra_metadata": extra_metadata,
            "message_type": msg.message_type,
            "image_content": msg.image_content,
            "feedback": user_feedback,
        }

        if msg.tool_calls:
            msg_dict["tool_calls"] = [
                {
                    "id": tc.langgraph_tool_call_id or str(tc.id),
                    "name": tc.tool_name,
                    "function": {"name": tc.tool_name},
                    "args": tc.tool_input or {},
                    "tool_call_result": {"content": (tc.tool_output or "")} if tc.status == "success" else None,
                    "status": tc.status,
                    "error_message": tc.error_message,
                }
                for tc in msg.tool_calls
            ]

        history.append(msg_dict)

    logger.info(f"Loaded {len(history)} messages with feedback for thread {thread_id}")
    return {"history": history}
