# ========= Copyright 2025-2026 @ Eigent.ai All Rights Reserved. =========
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# ========= Copyright 2025-2026 @ Eigent.ai All Rights Reserved. =========

import asyncio
import logging
import mimetypes
import re
import time
from functools import partial
from pathlib import Path
from typing import Annotated
from urllib.parse import quote

from fastapi import (
    APIRouter,
    Depends,
    File,
    Header,
    HTTPException,
    Query,
    UploadFile,
)
from fastapi.responses import FileResponse
from starlette.concurrency import run_in_threadpool

from app.artifacts import (
    discover_task_changed_files,
    scan_task_changed_files,
    task_modification_windows,
)
from app.auth import require_local_control_principal
from app.component.environment import env
from app.run_journal import get_default_run_journal
from app.utils.file_utils import list_files, resolve_under_base
from app.utils.workspace_paths import runtime_owner_key, task_dir_name
from app.utils.workspace_resolver import get_workspace_resolver

router = APIRouter()
file_logger = logging.getLogger("file_controller")

# Config
MAX_FILE_SIZE_BYTES = 50 * 1024 * 1024  # 50MB
MAX_FILES_PER_SESSION = 20
WORKSPACE_ROOT = env("EIGENT_WORKSPACE", "~/.eigent/workspace")
SESSION_ID_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$")
FILE_LIST_SEMAPHORE = asyncio.Semaphore(4)
SLOW_FILE_LIST_LOG_MS = 300


def _get_eigent_root() -> Path:
    """Base root for eigent storage (~/eigent). Do NOT use env file_save_path
    here: chat overwrites it to task path, which would break list/stream."""
    eigent = Path.home() / "eigent"
    if eigent.exists():
        return eigent
    dot_eigent = Path.home() / ".eigent"
    if dot_eigent.exists():
        return dot_eigent
    return eigent  # default to ~/eigent


def _get_workspace_root() -> Path:
    return Path(WORKSPACE_ROOT).expanduser()


def _validate_session_id(session_id: str) -> str:
    normalized = (session_id or "").strip()
    if not SESSION_ID_PATTERN.fullmatch(normalized):
        raise ValueError("Invalid X-Session-ID")
    return normalized


def _get_session_uploads_dir(session_id: str) -> Path:
    root = _get_workspace_root().resolve()
    validated = _validate_session_id(session_id)
    uploads_dir = (root / validated / "uploads").resolve()
    try:
        uploads_dir.relative_to(root)
    except ValueError as exc:
        raise ValueError("Invalid X-Session-ID") from exc
    return uploads_dir


def _count_session_uploads(session_id: str) -> int:
    uploads_dir = _get_session_uploads_dir(session_id)
    if not uploads_dir.exists():
        return 0
    return len(list(uploads_dir.iterdir()))


def _redacted_path_suffix(path: Path) -> str:
    parts = path.parts[-2:]
    return (".../" + "/".join(parts)) if parts else "..."


@router.post("/files")
async def upload_file(
    file: Annotated[UploadFile, File()],
    x_session_id: Annotated[str | None, Header(alias="X-Session-ID")] = None,
) -> dict:
    """
    Upload file. Requires X-Session-ID header.
    Returns file_id for message attachments reference.
    """
    if not x_session_id:
        raise HTTPException(
            status_code=400,
            detail="X-Session-ID header is required for file upload",
        )
    try:
        validated_session_id = _validate_session_id(x_session_id)
    except ValueError as exc:
        raise HTTPException(
            status_code=400, detail="Invalid X-Session-ID"
        ) from exc

    # Check session file count limit
    count = _count_session_uploads(validated_session_id)
    if count >= MAX_FILES_PER_SESSION:
        raise HTTPException(
            status_code=429,
            detail=f"Maximum {MAX_FILES_PER_SESSION} files per session",
        )

    # Read and validate size
    content = await file.read()
    if len(content) > MAX_FILE_SIZE_BYTES:
        raise HTTPException(
            status_code=413,
            detail=f"File size exceeds {MAX_FILE_SIZE_BYTES // (1024 * 1024)}MB limit",
        )

    # Generate safe filename
    timestamp = int(time.time() * 1000)
    safe_name = "".join(
        c if c.isalnum() or c in "._-" else "_"
        for c in (file.filename or "file")
    )
    stored_name = f"{safe_name}_{timestamp}"

    # Write to disk
    uploads_dir = _get_session_uploads_dir(validated_session_id)
    uploads_dir.mkdir(parents=True, exist_ok=True)
    target_path = uploads_dir / stored_name
    target_path.write_bytes(content)

    file_id = f"upload://{stored_name}"
    file_logger.info(
        f"File uploaded: session={validated_session_id}, file_id={file_id}, size={len(content)}"
    )

    return {
        "file_id": file_id,
        "filename": file.filename or "file",
        "size": len(content),
    }


def _normalize_relative_path(path: str) -> str:
    """Normalize relative path to URL-safe POSIX style."""
    return path.replace("\\", "/")


def _get_project_root(
    email: str, project_id: str, user_id: str | int | None = None
) -> Path:
    """Get project root path under the canonical user_id or legacy email key."""
    return (
        _get_eigent_root()
        / runtime_owner_key(email, user_id)
        / f"project_{project_id}"
    )


def _resolve_project_root(
    email: str, project_id: str, user_id: str | int | None = None
) -> Path:
    """
    Resolve project root, preferring the canonical user_id-scoped path when
    available, then legacy email-scoped storage.
    """
    preferred = _get_project_root(email, project_id, user_id)
    if preferred.exists():
        return preferred

    if user_id is not None and str(user_id).strip():
        legacy_preferred = _get_project_root(email, project_id)
        if legacy_preferred.exists():
            return legacy_preferred

    return preferred


def _resolve_file_root(
    email: str,
    project_id: str,
    space_id: str | None = None,
    user_id: str | int | None = None,
) -> Path:
    if space_id:
        resolver = get_workspace_resolver()
        space_root = resolver.space_root(
            space_id=space_id,
            project_id=project_id,
            email=email,
            user_id=user_id,
        )
        if space_root is not None:
            return space_root
    return _resolve_project_root(email, project_id, user_id)


def _list_task_changed_files(
    snapshot,
    max_entries: int = 500,
    modification_windows: tuple[tuple[float, float | None], ...] | None = None,
) -> list[dict]:
    """List files generated or modified by one Run without uploading them."""
    return scan_task_changed_files(
        snapshot,
        max_entries,
        modification_windows,
        list_files_fn=list_files,
    )


def _task_modification_windows(
    task_id: str,
    project_id: str,
) -> tuple[tuple[tuple[float, float | None], ...] | None, bool]:
    """Return filesystem-mtime windows owned by this Run's attempts.

    ``None`` means the RunJournal has no matching canonical Run and callers
    should retain the legacy start-time-only behavior.
    """
    return task_modification_windows(
        get_default_run_journal(), task_id, project_id
    )


@router.get(
    "/files/changes",
    dependencies=[Depends(require_local_control_principal)],
)
async def list_task_changed_files(
    task_id: str = Query(..., description="Run/task ID"),
    project_id: str = Query(..., description="Project ID"),
    email: str = Query(..., description="User email"),
    user_id: str | None = Query(
        None, description="Optional canonical user ID"
    ),
) -> dict:
    """Return the Desktop-local preview index for one Run's changed files."""
    manifest_event = await run_in_threadpool(
        get_default_run_journal().get_run_artifact_manifest_event,
        task_id,
    )
    if manifest_event is not None:
        artifacts = manifest_event.payload.get("artifacts", [])
        if isinstance(artifacts, list):
            return {
                "artifacts": [
                    dict(item) for item in artifacts if isinstance(item, dict)
                ],
                "scan_status": str(
                    manifest_event.payload.get("scan_status") or "complete"
                ),
                "truncated": bool(manifest_event.payload.get("truncated")),
            }

    snapshot = get_workspace_resolver().store.get_snapshot(
        email, task_id, user_id
    )
    if snapshot is None or snapshot.project_id != project_id:
        raise HTTPException(status_code=404, detail="Task workspace not found")
    artifact_manifest = getattr(snapshot, "artifact_manifest", None)
    if artifact_manifest is not None:
        return {
            "artifacts": [dict(item) for item in artifact_manifest],
            "scan_status": "complete",
            "truncated": False,
        }

    windows, permanently_terminal = await run_in_threadpool(
        _task_modification_windows, task_id, project_id
    )
    scan_result = await run_in_threadpool(
        discover_task_changed_files,
        snapshot,
        max_entries=500,
        modification_windows=windows,
    )
    artifacts = scan_result.artifacts
    if permanently_terminal:
        await run_in_threadpool(
            get_workspace_resolver().store.freeze_artifact_manifest,
            email,
            snapshot,
            artifacts,
        )
    return {
        "artifacts": artifacts,
        "scan_status": scan_result.scan_status,
        "truncated": scan_result.truncated,
    }


@router.get("/files")
async def list_project_files(
    project_id: str = Query(..., description="Project ID"),
    email: str = Query(..., description="User email"),
    space_id: str | None = Query(None, description="Optional Space ID"),
    user_id: str | None = Query(
        None, description="Optional canonical user ID"
    ),
    task_id: str | None = Query(
        None, description="Optional task ID to scope listing"
    ),
) -> list[dict]:
    """
    List files in project working directory (Brain storage).
    Used by Web mode when ipcRenderer is unavailable.
    Returns [{filename, url}] where url can be used to fetch file content.
    """
    if not project_id or not email:
        raise HTTPException(
            status_code=400,
            detail="project_id and email are required",
        )
    project_root = _resolve_file_root(email, project_id, space_id, user_id)
    list_dir = str(project_root)
    if task_id:
        list_dir = str(project_root / task_dir_name(task_id))
    if not Path(list_dir).exists():
        file_logger.debug(
            "list_project_files: path does not exist: %s",
            list_dir,
        )
        return []
    base_path = str(project_root.resolve())
    stats: dict[str, float | int] = {}
    started = time.perf_counter()
    try:
        async with FILE_LIST_SEMAPHORE:
            paths = await run_in_threadpool(
                partial(
                    list_files,
                    list_dir,
                    base=base_path,
                    max_entries=500,
                    stats=stats,
                )
            )
    except Exception as e:
        file_logger.warning("list_project_files failed: %s", e)
        return []
    elapsed_ms = (time.perf_counter() - started) * 1000
    log = (
        file_logger.info
        if elapsed_ms >= SLOW_FILE_LIST_LOG_MS
        else file_logger.debug
    )
    log(
        "list_project_files: project_id=%s space_id=%s task_id=%s count=%d "
        "elapsed_ms=%.1f scan_ms=%.1f realpath_ms=%.1f symlinks=%d root=%s",
        project_id,
        space_id,
        task_id,
        len(paths),
        elapsed_ms,
        float(stats.get("scan_elapsed_ms", 0)),
        float(stats.get("realpath_elapsed_ms", 0)),
        int(stats.get("symlink_count", 0)),
        _redacted_path_suffix(project_root),
    )
    result: list[dict] = []
    for abs_path in paths:
        try:
            rel = _normalize_relative_path(
                Path(abs_path).relative_to(base_path).as_posix()
            )
            # URL-encode the relative path for stream endpoint
            path_param = quote(rel, safe="")
            stat_result = Path(abs_path).stat()
            result.append(
                {
                    "filename": Path(abs_path).name,
                    "url": (
                        f"/files/stream?path={path_param}"
                        f"&project_id={quote(project_id)}"
                        f"&email={quote(email)}"
                        + (f"&space_id={quote(space_id)}" if space_id else "")
                        + (f"&user_id={quote(user_id)}" if user_id else "")
                    ),
                    "relativePath": rel,
                    "size": stat_result.st_size,
                    "modifiedAt": stat_result.st_mtime * 1000,
                    "supportsRanges": True,
                }
            )
        except (ValueError, OSError):
            continue
    return result


@router.api_route("/files/stream", methods=["GET", "HEAD"])
async def stream_file(
    path: str = Query(..., description="Relative path from project root"),
    project_id: str = Query(..., description="Project ID"),
    email: str = Query(..., description="User email"),
    space_id: str | None = Query(None, description="Optional Space ID"),
    user_id: str | None = Query(
        None, description="Optional canonical user ID"
    ),
):
    """
    Stream file content. Path must be relative to project root.
    Used by Web mode to fetch file content for display.
    """
    if not path or not project_id or not email:
        raise HTTPException(
            status_code=400,
            detail="path, project_id and email are required",
        )
    project_root = _resolve_file_root(email, project_id, space_id, user_id)
    # Resolve path and ensure it stays under project root (security)
    try:
        resolved = resolve_under_base(path, str(project_root.resolve()))
    except Exception as e:
        file_logger.warning("stream_file path validation failed: %s", e)
        raise HTTPException(status_code=400, detail="Invalid path") from e
    p = Path(resolved)
    if not p.is_file():
        raise HTTPException(status_code=404, detail="File not found")
    media_type, _ = mimetypes.guess_type(str(p))
    if not media_type:
        media_type = "application/octet-stream"
    # content_disposition_type=inline: display in iframe instead of triggering download
    return FileResponse(
        path=str(p),
        filename=p.name,
        media_type=media_type,
        content_disposition_type="inline",
    )


@router.get("/files/preview/{email}/{project_id}/{file_path:path}")
async def preview_file(
    email: str,
    project_id: str,
    file_path: str,
    space_id: str | None = Query(None, description="Optional Space ID"),
    user_id: str | None = Query(
        None, description="Optional canonical user ID"
    ),
):
    """
    Preview file content with a path-based URL so relative references inside
    HTML/CSS/JS resolve against the project directory structure.
    """
    if not file_path or not project_id or not email:
        raise HTTPException(
            status_code=400,
            detail="file_path, project_id and email are required",
        )

    project_root = _resolve_file_root(email, project_id, space_id, user_id)
    try:
        resolved = resolve_under_base(file_path, str(project_root.resolve()))
    except Exception as e:
        file_logger.warning("preview_file path validation failed: %s", e)
        raise HTTPException(status_code=400, detail="Invalid path") from e

    p = Path(resolved)
    if not p.is_file():
        raise HTTPException(status_code=404, detail="File not found")

    media_type, _ = mimetypes.guess_type(str(p))
    if not media_type:
        media_type = "application/octet-stream"

    return FileResponse(
        path=str(p),
        filename=p.name,
        media_type=media_type,
        content_disposition_type="inline",
    )