"""Files 路由(user-rooted,不绑 task)+ 用户磁盘用量 + pptx 在线预览。

文件面板的目录树 mutation 入口(DESIGN §7.4):顶层目录 rename 复用
core.working_dirs 的 DB-aware 服务，delete/move 走本路由状态闸。路径安全原语在
web/userfiles.py。
"""
from __future__ import annotations

import asyncio
from pathlib import Path
from uuid import UUID

from fastapi import Depends, File, Form, HTTPException, UploadFile
from fastapi.responses import FileResponse
from sqlalchemy import func, select

from core.paths import from_db_path, to_db_path
from core.storage import session_scope
from core.storage.models import Task
from core.working_dirs import (
    WorkingDirConflictError,
    WorkingDirRenameError,
    rename_working_dir,
)

from ..common import norm_path
from ..schemas import (
    FileDeleteRequest,
    FileMkdirRequest,
    FileRenameRequest,
    FileTransferRequest,
)
from ..userfiles import (
    enumerate_files,
    load_user_root,
    rel_to,
    safe_join,
    system_wd_names,
    validate_transfer,
)

# pptx→PDF 预览:按解析后的 pptx 绝对路径加锁,防同一文件并发重复转换(DESIGN §8.3)。
_pptx_preview_locks: dict[str, asyncio.Lock] = {}


def _pptx_lock_for(abs_path: str) -> asyncio.Lock:
    lock = _pptx_preview_locks.get(abs_path)
    if lock is None:
        lock = _pptx_preview_locks[abs_path] = asyncio.Lock()
    return lock


def _regular_file_response(target: Path, display_path: str) -> FileResponse:
    if not target.exists():
        raise HTTPException(404, f"file not found: {display_path}")
    if not target.is_file():
        raise HTTPException(400, f"not a file: {display_path}")
    media_type = "image/svg+xml" if target.suffix.lower() == ".svg" else None
    return FileResponse(
        path=str(target),
        filename=target.name,
        media_type=media_type,
        headers={"Cache-Control": "no-cache"},
    )


def _task_working_dir(task_id: str, user_id: UUID, root: Path) -> tuple[UUID, Path]:
    try:
        tid = UUID(task_id)
    except ValueError:
        raise HTTPException(404, f"invalid task id: {task_id!r}")
    with session_scope() as s:
        db_path = s.execute(
            select(Task.working_dir).where(Task.task_id == tid, Task.user_id == user_id)
        ).scalar_one_or_none()
    if not db_path:
        raise HTTPException(404, "task not found")
    working_dir = from_db_path(db_path).resolve()
    try:
        working_dir.relative_to(root.resolve())
    except ValueError:
        raise HTTPException(400, "task working_dir is outside user workspace")
    return tid, working_dir


def _task_file_target(root: Path, working_dir: Path, path: str, legacy: bool) -> Path:
    """Resolve canonical task refs, with a read-only fallback chain for old cards.

    Historical messages used user-root paths, while one known failure accidentally emitted
    a task-relative path with the same leading directory name.  Old cards therefore try the
    original user-root meaning first, then both task-relative interpretations.  New refs never
    use this branch and remain unambiguous.
    """
    if not legacy:
        return safe_join(working_dir, path)
    candidates = [safe_join(root, path), safe_join(working_dir, path)]
    normalized = str(path or "").replace("\\", "/")
    if "/" in normalized:
        candidates.append(safe_join(working_dir, normalized.split("/", 1)[1]))
    for candidate in candidates:
        if candidate.is_file():
            return candidate
    return candidates[0]


async def _pptx_preview_response(target: Path, display_path: str) -> FileResponse:
    from ..pptx_render import (
        PptxConvertError,
        SofficeNotFoundError,
        pptx_to_pdf,
    )

    if not target.exists():
        raise HTTPException(404, f"file not found: {display_path}")
    if not target.is_file():
        raise HTTPException(400, f"not a file: {display_path}")
    if target.suffix.lower() not in (".pptx", ".ppt"):
        raise HTTPException(400, f"not a pptx: {display_path}")
    abs_path = str(target.resolve())
    loop = asyncio.get_event_loop()
    async with _pptx_lock_for(abs_path):
        try:
            pdf_path = await loop.run_in_executor(None, pptx_to_pdf, target)
        except SofficeNotFoundError as e:
            raise HTTPException(501, str(e))
        except PptxConvertError as e:
            raise HTTPException(500, str(e))
    return FileResponse(
        path=str(pdf_path),
        media_type="application/pdf",
        headers={"Cache-Control": "no-cache"},
    )


def register_file_routes(app, *, require_user) -> None:
    @app.get("/v1/user/storage", tags=["user"])
    def user_storage(user_id: UUID = Depends(require_user)):
        """当前用户磁盘用量 + 配额。

        数据来自后台扫描落库的 user_disk_usage(默 15min 一次),非实时;
        无扫描记录(新用户 / 首次扫描前)bytes_used/file_count=0、scanned_at=None。
        limit_bytes<=0 或 None → 不限(前端不画进度条)。
        """
        from core.agent_builder import load_config as _load_cfg
        from core.storage.disk_quota import get_user_usage, parse_bytes
        _quotas_cfg = (_load_cfg().get("quotas") or {})
        _limit = parse_bytes(_quotas_cfg.get("disk_bytes_per_user"))
        usage = get_user_usage(user_id)
        if usage is None:
            bytes_used, file_count, scanned_at = 0, 0, None
        else:
            bytes_used, file_count, scanned_at = usage
        return {
            "bytes_used": bytes_used,
            "file_count": file_count,
            "limit_bytes": (_limit if (_limit and _limit > 0) else None),
            "scanned_at": scanned_at.isoformat() if scanned_at else None,
        }

    @app.get("/v1/files", tags=["files"])
    def list_files(
        path: str = "",
        include_hidden: bool = False,
        task_id: str = "",
        user_id: UUID = Depends(require_user),
    ):
        """列 user_root 下子目录条目 + 面包屑。`path` 留空 → user_root;
        `../` / 绝对 → 400。dotfile/dotdir 默认隐藏；`include_hidden=true`
        仅在 `task_id` 属于当前用户且 current 位于该 task 工作目录子树时展示
        dotdir，永不从 user_root 展示平台目录。根目录层再隐藏系统工作目录
        (定时任务 / 渠道对话,见 system_wd_names)。点文件始终隐藏。
        """
        root = load_user_root(user_id)
        current = safe_join(root, path)
        include_task_hidden = False
        if include_hidden and task_id:
            _tid, working_dir = _task_working_dir(task_id, user_id, root)
            try:
                current.relative_to(working_dir)
                include_task_hidden = True
            except ValueError:
                pass
        entries, crumbs, exists = enumerate_files(
            root,
            current,
            include_hidden_dirs=include_task_hidden,
        )
        if not rel_to(root, current):
            hidden = system_wd_names(user_id)
            if hidden:
                entries = [
                    e for e in entries
                    if not (e["is_dir"] and e["name"] in hidden)
                ]
        return {
            "root": norm_path(str(root)),
            "current": rel_to(root, current),
            "exists": exists,
            "crumbs": crumbs,
            "entries": entries,
            "hidden_dirs_included": include_task_hidden,
        }

    @app.get("/v1/files/download", tags=["files"])
    def download_file(
        path: str,
        user_id: UUID = Depends(require_user),
    ):
        """下载 user_root 下单个 regular file(目录 → 400 / 不存在 → 404)。"""
        root = load_user_root(user_id)
        target = safe_join(root, path)
        # workspace 文件可变, 禁浏览器启发式缓存 (RFC 7234 默认能缓数小时)
        # 否则文件改了 SPA 预览还是旧内容
        # (Starlette FileResponse 不实现 304, 总是 200 全量; workspace 文件小, 可接受)
        # .svg 显式给 image/svg+xml: 部分部署环境 mimetypes 未注册 svg, FileResponse
        # 会猜成 octet-stream, 前端 <img> 就渲染不出 SVG 预览
        return _regular_file_response(target, path)

    @app.get("/v1/tasks/{task_id}/files/download", tags=["files"])
    def download_task_file(
        task_id: str,
        path: str,
        legacy: bool = False,
        user_id: UUID = Depends(require_user),
    ):
        """Download a file addressed relative to the task's current working_dir."""
        root = load_user_root(user_id)
        _tid, working_dir = _task_working_dir(task_id, user_id, root)
        target = _task_file_target(root, working_dir, path, legacy)
        return _regular_file_response(target, path)

    @app.get("/v1/files/preview_pdf", tags=["files"])
    async def preview_pdf(
        path: str,
        user_id: UUID = Depends(require_user),
    ):
        """把 user_root 下的 .pptx 转成 PDF 返回,供前端复用 PDF iframe 在线预览。

        转换跑在 backend host(不进沙盒),按需触发 + 缓存到 `.preview/`(DESIGN §8.3)。
        soffice 缺失 → 501;转换失败/超时 → 500;前端据此回退到下载。
        """
        root = load_user_root(user_id)
        target = safe_join(root, path)
        return await _pptx_preview_response(target, path)

    @app.get("/v1/tasks/{task_id}/files/preview_pdf", tags=["files"])
    async def preview_task_pdf(
        task_id: str,
        path: str,
        legacy: bool = False,
        user_id: UUID = Depends(require_user),
    ):
        """Preview a PPT addressed relative to the task's current working_dir."""
        root = load_user_root(user_id)
        _tid, working_dir = _task_working_dir(task_id, user_id, root)
        target = _task_file_target(root, working_dir, path, legacy)
        return await _pptx_preview_response(target, path)

    @app.post("/v1/files/upload", tags=["files"])
    async def upload_files(
        path: str = Form(""),
        files: list[UploadFile] = File(...),
        user_id: UUID = Depends(require_user),
    ):
        """multipart 多文件上传到 `<user_root>/<path>/`。
        路径不存在自动 mkdir(parents=True);重名直接覆盖。
        文件名严格校验(含 `/ \\ ..` 或为空 → 400)。
        """
        # 磁盘配额 gate(§7.5 #4):超额 413 阻止上传,提示 user 清旧产物
        from core.agent_builder import load_config as _load_cfg
        from core.storage.disk_quota import check_disk_quota, parse_bytes
        _quotas_cfg = (_load_cfg().get("quotas") or {})
        _limit = parse_bytes(_quotas_cfg.get("disk_bytes_per_user"))
        if _limit is not None and _limit > 0:
            _err = check_disk_quota(user_id, _limit)
            if _err is not None:
                raise HTTPException(413, _err)

        root = load_user_root(user_id)
        dest_dir = safe_join(root, path)
        if dest_dir.exists() and not dest_dir.is_dir():
            raise HTTPException(400, f"upload target is a file, not a directory: {path}")
        dest_dir.mkdir(parents=True, exist_ok=True)

        saved: list[dict] = []
        for up in files or []:
            raw_name = up.filename or ""
            if (
                not raw_name
                or raw_name in (".", "..")
                or "/" in raw_name or "\\" in raw_name
                or any(part in (".", "..") for part in Path(raw_name).parts)
            ):
                raise HTTPException(400, f"invalid filename: {raw_name!r}")
            dest = dest_dir / raw_name
            try:
                dest.resolve().relative_to(root.resolve())
            except ValueError:
                raise HTTPException(400, f"path escapes user_root: {raw_name!r}")
            data = await up.read()
            dest.write_bytes(data)
            saved.append({"name": raw_name, "size": len(data), "rel": rel_to(root, dest)})
        if not saved:
            raise HTTPException(400, "no files uploaded")
        return {"count": len(saved), "saved": saved}

    @app.post("/v1/files/mkdir", tags=["files"])
    def create_directory(
        body: FileMkdirRequest,
        user_id: UUID = Depends(require_user),
    ):
        """在 `<user_root>/<path>/` 下创建一个直接子目录。

        name 与 working_dir 共用 leaf 校验，禁止路径分隔符、NUL 和点目录；
        parent 必须已存在且为目录，目标已存在返回 409，不递归补建 parent。
        """
        from core.agent_builder import InvalidTaskName, validate_task_name

        root = load_user_root(user_id)
        parent = safe_join(root, body.path)
        if not parent.exists():
            raise HTTPException(404, f"parent directory not found: {body.path!r}")
        if not parent.is_dir():
            raise HTTPException(400, f"parent is not a directory: {body.path!r}")
        try:
            name = validate_task_name(body.name)
        except InvalidTaskName as e:
            raise HTTPException(400, f"name 不合法: {e}")

        target = parent / name
        if target.exists():
            raise HTTPException(409, f"target already exists: {rel_to(root, target)!r}")
        try:
            target.mkdir()
        except FileExistsError:
            raise HTTPException(
                409, f"target already exists: {rel_to(root, target)!r}"
            )
        except OSError as e:
            raise HTTPException(400, f"mkdir failed: {e}")
        return {"ok": True, "path": rel_to(root, target), "name": name}

    @app.post("/v1/files/delete", tags=["files"])
    def delete_file(
        body: FileDeleteRequest,
        user_id: UUID = Depends(require_user),
    ):
        """删 user_root 下文件或目录。

        - `recursive=False`(默认):目录必须为空(`rmdir`),非空 → 400
        - `recursive=True`:`shutil.rmtree`;若目标是顶层目录且被某 task.working_dir
          引用 → 409,提示先 DELETE task(避免 DB 还引用、FS artifacts 已清的错位)
        - 顶层空目录 / 子级空目录无论 recursive 与否都可删:task.working_dir 字段不动,
          下次 build_agent 按需 mkdir 重建,FS 目录视为可重生
        - root → 400;不存在 → 404
        """
        root = load_user_root(user_id)
        target = safe_join(root, body.path)
        if target.resolve() == root.resolve():
            raise HTTPException(400, "cannot delete user_root")
        if not target.exists():
            raise HTTPException(404, f"path not found: {body.path}")

        if target.is_dir() and body.recursive:
            is_top_level = target.parent.resolve() == root.resolve()
            if is_top_level:
                db_form = to_db_path(target)
                with session_scope() as s:
                    n = s.execute(
                        select(func.count()).select_from(Task).where(
                            Task.user_id == user_id,
                            Task.working_dir == db_form,
                            Task.deleted_at.is_(None),  # 软删 task 不再算引用
                        )
                    ).scalar_one() or 0
                if n:
                    raise HTTPException(
                        409,
                        f"该顶层目录正被 {n} 个 task 引用,不能递归删除;"
                        f"请先 DELETE task,再清残留文件",
                    )

        try:
            if target.is_dir():
                if body.recursive:
                    import shutil
                    shutil.rmtree(target)
                else:
                    target.rmdir()  # 非空目录会触发 OSError
            else:
                target.unlink()
        except OSError as e:
            raise HTTPException(400, f"delete failed: {e}")
        return {"ok": True, "path": body.path}

    @app.post("/v1/files/rename", tags=["files"])
    def rename_path(
        body: FileRenameRequest,
        user_id: UUID = Depends(require_user),
    ):
        """重命名 user_root 下文件或目录(任意深度)。

        - `path` 必填,指被重命名对象;不能为 user_root
        - `new_name` 是新 leaf 名(`validate_task_name`:非空 / 不含 `/\\..` / 非 dotfile / ≤255);
          不是路径,parent 自动取自原 path
        - 目标 sibling `<parent>/<new_name>` 不能已存在(防覆盖)
        - **path 是顶层目录**(user_root 直接子项,且为目录)→ DB-aware:
          * 同事务内 `SELECT ... FOR UPDATE` 锁该目录对应 task;任一 run_status 在
            running/cancelling → 409(避免 BG 线程握旧路径而 DB 已指新路径)
          * `check_no_subtask(new_db, exclude=被改名 tids)` 防止改名后跟其它 task 形成嵌套
          * `UPDATE tasks SET working_dir=new_db WHERE task_id IN (...)` 先写 DB
          * 再 `os.rename` FS;失败 → 抛错 → session_scope 回滚 DB
          * 唯一不一致窗口是 "FS 已改名 + commit 阶段失败"(PG 单事务 commit 极少失败)
        - 非顶层(子目录 / 文件)→ 纯 FS rename,不动 DB
        """
        from core.agent_builder import InvalidTaskName, validate_task_name

        root = load_user_root(user_id)
        target = safe_join(root, body.path)
        if target.resolve() == root.resolve():
            raise HTTPException(400, "cannot rename user_root")
        if not target.exists():
            raise HTTPException(404, f"path not found: {body.path}")

        try:
            new_name = validate_task_name(body.new_name)
        except InvalidTaskName as e:
            raise HTTPException(400, f"new_name 不合法: {e}")

        if new_name == target.name:
            raise HTTPException(400, f"new_name 与原名相同: {new_name!r}")

        new_target = target.parent / new_name
        if new_target.exists():
            raise HTTPException(
                409, f"target already exists: {rel_to(root, new_target)!r}"
            )

        is_top_level_dir = (
            target.is_dir() and target.parent.resolve() == root.resolve()
        )

        if not is_top_level_dir:
            try:
                target.rename(new_target)
            except OSError as e:
                raise HTTPException(400, f"rename failed: {e}")
            return {
                "ok": True,
                "old": body.path,
                "new": rel_to(root, new_target),
                "tasks_updated": 0,
            }

        # 顶层目录:网页和对话内延迟动作共用同一个 DB-aware 服务。
        try:
            renamed = rename_working_dir(
                user_id=user_id,
                old_path=target,
                new_path=new_target,
            )
        except WorkingDirConflictError as e:
            raise HTTPException(409, str(e))
        except WorkingDirRenameError as e:
            # 保持原 files API 的普通 FS 失败 400 语义。
            raise HTTPException(400, str(e))

        return {
            "ok": True,
            "old": body.path,
            "new": rel_to(root, new_target),
            "tasks_updated": renamed.tasks_updated,
        }

    @app.post("/v1/files/copy", tags=["files"])
    def copy_files(
        body: FileTransferRequest,
        user_id: UUID = Depends(require_user),
    ):
        """批量拷贝 paths → dest_dir/<name>(目录递归)。

        - 不覆盖(任一目标已存在 → 409)
        - 不能拷到自己 / 自身子树
        - 顶层目录(可能是某 task 的 working_dir)可以拷:新副本无 task 关联,不动 DB
        - 部分失败语义:任一 FS 拷贝抛错 → 抛 HTTPException,**前面已成功的拷贝保留**
          (无 FS 事务可回滚;预检通过后通常不会失败,失败也是磁盘满 / 权限这类不能恢复的)
        """
        import shutil
        root = load_user_root(user_id)
        sources, dest = validate_transfer(root, body.paths, body.dest_dir)
        transferred: list[dict] = []
        for src in sources:
            target = dest / src.name
            try:
                if src.is_dir():
                    shutil.copytree(src, target)
                else:
                    shutil.copy2(src, target)
            except OSError as e:
                raise HTTPException(
                    500,
                    f"copy failed at {src.name!r}: {e} "
                    f"(已成功 {len(transferred)} 项,剩余未处理)",
                )
            transferred.append({
                "old": rel_to(root, src),
                "new": rel_to(root, target),
            })
        return {"ok": True, "count": len(transferred), "transferred": transferred}

    @app.post("/v1/files/move", tags=["files"])
    def move_files(
        body: FileTransferRequest,
        user_id: UUID = Depends(require_user),
    ):
        """批量移动 paths → dest_dir/<name>。

        - 不覆盖、不自嵌套(同 /copy)
        - **顶层目录是某 task 的 working_dir → 409**,维持 "working_dir = 顶层目录" invariant
          (允许的话 task working_dir 沉到子目录会让 rename 顶层的 DB-aware 逻辑失效;
          用户想归档:先 DELETE task)
        - 拷贝(`/copy`)无此限制,因为新副本无 task 关联
        - 部分失败:同 /copy,前面成功的不回滚(`shutil.move` 失败几乎只发生在
          跨卷拷贝中断,workspace 都在同一磁盘下罕见)
        """
        import shutil
        root = load_user_root(user_id)
        sources, dest = validate_transfer(root, body.paths, body.dest_dir)

        # 顶层目录-是-某 task.working_dir → 闸
        top_level_dir_srcs = [
            s for s in sources
            if s.is_dir() and s.parent.resolve() == root.resolve()
        ]
        if top_level_dir_srcs:
            db_forms = [to_db_path(s) for s in top_level_dir_srcs]
            with session_scope() as s:
                rows = s.execute(
                    select(Task.working_dir, func.count())
                    .where(
                        Task.user_id == user_id,
                        Task.working_dir.in_(db_forms),
                    )
                    .group_by(Task.working_dir)
                ).all()
            occupied = {wd: n for wd, n in rows}
            if occupied:
                # 反查 db_form → src.name 给报错文案
                form2name = {to_db_path(s): s.name for s in top_level_dir_srcs}
                names = ", ".join(
                    f"{form2name[wd]!r}({n} 个 task)"
                    for wd, n in occupied.items()
                )
                raise HTTPException(
                    409,
                    f"以下顶层目录正被 task 引用,不能移动:{names};"
                    f"请先删 task,或改用复制",
                )

        transferred: list[dict] = []
        for src in sources:
            target = dest / src.name
            try:
                shutil.move(str(src), str(target))
            except OSError as e:
                raise HTTPException(
                    500,
                    f"move failed at {src.name!r}: {e} "
                    f"(已成功 {len(transferred)} 项,剩余未处理)",
                )
            transferred.append({
                "old": rel_to(root, src),
                "new": rel_to(root, target),
            })
        return {"ok": True, "count": len(transferred), "transferred": transferred}