from __future__ import annotations

import hashlib
import hmac
import json
import uuid
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any
from urllib.parse import urlsplit, urlunsplit

import httpx
from fastapi import HTTPException, Request, status
from jose import JWTError, jwt
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.config import settings
from app.models.collaboration import (
    CollaborationCallbackReceipt,
    CollaborationFile,
    CollaborationRevision,
    CollaborationSession,
    CollaborationShareLink,
)
from app.models.user import User
from app.schemas.collaboration import CollaborationCallbackPayload, CollaborationEditorConfigRead
from app.services import collaboration_service, onlyoffice_service


def collaboration_document_key(file_id: uuid.UUID, generation: int) -> str:
    fingerprint = f"{settings.ONLYOFFICE_INSTANCE_ID or ''}:collaboration:{file_id}:{generation}"
    return f"ctms-collab-{hashlib.sha256(fingerprint.encode('utf-8')).hexdigest()}"


def _content_url(session_id: uuid.UUID) -> str:
    return (
        f"{settings.ONLYOFFICE_STORAGE_BASE_URL.rstrip('/')}"
        f"/internal/onlyoffice/collaboration/sessions/{session_id}/content"
    )


def _callback_url(session_id: uuid.UUID) -> str:
    return (
        f"{settings.ONLYOFFICE_STORAGE_BASE_URL.rstrip('/')}"
        f"/internal/onlyoffice/collaboration/sessions/{session_id}/callback"
    )


async def _active_session(
    db: AsyncSession, item: CollaborationFile, user_id: uuid.UUID
) -> CollaborationSession:
    session = await db.scalar(
        select(CollaborationSession).where(
            CollaborationSession.file_id == item.id,
            CollaborationSession.generation == item.generation,
        ).order_by(CollaborationSession.created_at.desc())
    )
    if session:
        if session.status != "ACTIVE":
            session.status = "ACTIVE"
            session.closed_at = None
            await db.commit()
            await db.refresh(session)
        return session
    if not item.current_revision_id:
        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="协作文件尚无可编辑内容")
    session = CollaborationSession(
        file_id=item.id,
        base_revision_id=item.current_revision_id,
        document_key=collaboration_document_key(item.id, item.generation),
        generation=item.generation,
        started_by=user_id,
    )
    db.add(session)
    await db.commit()
    await db.refresh(session)
    return session


async def build_editor_config(
    db: AsyncSession, item: CollaborationFile, user
) -> CollaborationEditorConfigRead:
    await onlyoffice_service.ensure_onlyoffice_available()
    revision = await db.get(CollaborationRevision, item.current_revision_id)
    if not revision or not Path(revision.file_uri).exists():
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="协作文件内容不存在")
    can_edit = await collaboration_service.can_edit_file(db, item, user)
    can_request_edit = await collaboration_service.can_request_edit_file(db, item, user)
    can_download = await collaboration_service.can_export_file(db, item, user)
    can_save_as = can_download and await collaboration_service.can_create_file(db, item, user)
    session = await _active_session(db, item, user.id)
    now = datetime.now(timezone.utc)
    expires_at = now + timedelta(seconds=settings.ONLYOFFICE_CONFIG_TTL_SECONDS)
    config: dict[str, Any] = {
        "type": "desktop",
        "documentType": item.file_type,
        "document": {
            "fileType": item.extension,
            "key": session.document_key,
            "title": item.title,
            "url": _content_url(session.id),
            "permissions": {
                "chat": False,
                "copy": can_download,
                "comment": can_edit,
                "download": can_download,
                # In view mode ONLYOFFICE displays "Edit current file" only
                # when edit=true and onRequestEditRights is registered. CTMS
                # handles that event as an approval request, not an escalation.
                "edit": can_edit or can_request_edit,
                "fillForms": False,
                "modifyContentControl": can_edit,
                "modifyFilter": can_edit,
                "print": can_download,
                "protect": False,
                "review": False,
            },
        },
        "editorConfig": {
            "callbackUrl": _callback_url(session.id),
            "coEditing": {"mode": "fast", "change": False},
            "customization": {
                "autosave": True,
                "chat": False,
                "comments": can_edit,
                "forcesave": can_edit,
                "help": False,
                "plugins": False,
            },
            "lang": "zh-CN",
            "mode": "edit" if can_edit else "view",
            "user": {"id": str(user.id), "name": user.full_name},
        },
    }
    config["token"] = jwt.encode(
        {**config, "iat": int(now.timestamp()), "exp": int(expires_at.timestamp())},
        settings.ONLYOFFICE_JWT_SECRET or "",
        algorithm="HS256",
    )
    return CollaborationEditorConfigRead(
        file_id=item.id,
        file_name=item.title,
        access_mode="edit" if can_edit else "view",
        can_save_as=can_save_as,
        can_download=can_download,
        can_request_edit=can_request_edit,
        expires_at=expires_at,
        config=config,
    )


async def build_shared_editor_config(
    db: AsyncSession,
    item: CollaborationFile,
    link: CollaborationShareLink,
    *,
    client_id: str,
    display_name: str,
) -> CollaborationEditorConfigRead:
    await onlyoffice_service.ensure_onlyoffice_available()
    revision = await db.get(CollaborationRevision, item.current_revision_id)
    if not revision or not Path(revision.file_uri).exists():
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="共享文件内容不存在")
    can_edit = link.access_mode == "EDIT"
    session = await _active_session(db, item, item.owner_id)
    now = datetime.now(timezone.utc)
    expires_at = now + timedelta(seconds=settings.ONLYOFFICE_CONFIG_TTL_SECONDS)
    if link.expires_at and link.expires_at < expires_at:
        expires_at = link.expires_at
    external_user_id = f"share-{link.id.hex[:12]}-{client_id[:32]}"
    config: dict[str, Any] = {
        "type": "desktop",
        "documentType": item.file_type,
        "document": {
            "fileType": item.extension,
            "key": session.document_key,
            "title": item.title,
            "url": _content_url(session.id),
            "permissions": {
                "chat": False,
                "copy": item.allow_export,
                "comment": can_edit,
                "download": item.allow_export,
                "edit": can_edit,
                "fillForms": False,
                "modifyContentControl": can_edit,
                "modifyFilter": can_edit,
                "print": item.allow_export,
                "protect": False,
                "review": False,
            },
        },
        "editorConfig": {
            "callbackUrl": _callback_url(session.id),
            "coEditing": {"mode": "fast", "change": False},
            "customization": {
                "autosave": can_edit,
                "chat": False,
                "comments": can_edit,
                "forcesave": can_edit,
                "help": False,
                "plugins": False,
            },
            "lang": "zh-CN",
            "mode": "edit" if can_edit else "view",
            "user": {"id": external_user_id, "name": display_name},
        },
    }
    config["token"] = jwt.encode(
        {**config, "iat": int(now.timestamp()), "exp": int(expires_at.timestamp())},
        settings.ONLYOFFICE_JWT_SECRET or "",
        algorithm="HS256",
    )
    return CollaborationEditorConfigRead(
        file_id=item.id,
        file_name=item.title,
        access_mode="edit" if can_edit else "view",
        can_save_as=False,
        can_download=item.allow_export,
        expires_at=expires_at,
        config=config,
    )


async def get_session_content(
    db: AsyncSession, session_id: uuid.UUID, authorization: str | None
) -> tuple[CollaborationRevision, CollaborationFile]:
    session = await db.get(CollaborationSession, session_id)
    if not session:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="协作会话不存在")
    onlyoffice_service.validate_outbox_token(authorization, _content_url(session_id))
    revision = await db.get(CollaborationRevision, session.base_revision_id)
    item = await db.get(CollaborationFile, session.file_id)
    if not revision or not item or not Path(revision.file_uri).exists():
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="协作文件内容不存在")
    return revision, item


def validate_callback_token(token: str | None, payload: CollaborationCallbackPayload) -> dict[str, Any]:
    if not token:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="缺少 ONLYOFFICE 回调签名")
    value = token.strip()
    if " " in value:
        scheme, credential = value.split(" ", 1)
        if scheme.lower() != "bearer":
            raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="ONLYOFFICE 回调签名格式无效")
        value = credential.strip()
    try:
        decoded = onlyoffice_service.decode_onlyoffice_token(value)
    except JWTError as exc:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="ONLYOFFICE 回调签名无效") from exc
    signed = decoded.get("payload") if isinstance(decoded.get("payload"), dict) else decoded
    if not isinstance(signed, dict):
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="ONLYOFFICE 回调载荷无效")
    signed_key = signed.get("key")
    signed_status = signed.get("status")
    if not isinstance(signed_key, str) or not hmac.compare_digest(signed_key, payload.key):
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="ONLYOFFICE 回调 key 不匹配")
    if not isinstance(signed_status, int) or signed_status != payload.status:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="ONLYOFFICE 回调状态不匹配")
    if payload.url:
        signed_url = signed.get("url")
        if not isinstance(signed_url, str) or not hmac.compare_digest(signed_url, payload.url):
            raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="ONLYOFFICE 回调文件地址不匹配")
    return decoded


def _callback_fingerprint(payload: CollaborationCallbackPayload) -> str:
    normalized = payload.model_dump(mode="json", exclude_none=True)
    return hashlib.sha256(json.dumps(normalized, sort_keys=True, separators=(",", ":")).encode()).hexdigest()


def _url_origin_matches(actual, expected) -> bool:
    actual_port = actual.port or (443 if actual.scheme == "https" else 80)
    expected_port = expected.port or (443 if expected.scheme == "https" else 80)
    return (
        actual.scheme == expected.scheme
        and actual.hostname
        and actual.hostname.lower() == (expected.hostname or "").lower()
        and actual_port == expected_port
    )


def _validate_result_url(url: str) -> str:
    actual = urlsplit(url)
    if (
        actual.scheme not in {"http", "https"}
        or actual.username
        or actual.password
        or actual.fragment
        or not actual.hostname
    ):
        raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="ONLYOFFICE 保存地址不受信任")

    internal = urlsplit(settings.ONLYOFFICE_INTERNAL_URL.rstrip("/"))
    if _url_origin_matches(actual, internal):
        return urlunsplit((internal.scheme, internal.netloc, actual.path, actual.query, ""))

    public = urlsplit(settings.FRONTEND_PUBLIC_URL.rstrip("/"))
    proxy_prefix = "/onlyoffice/"
    if not _url_origin_matches(actual, public) or not actual.path.startswith(proxy_prefix):
        raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="ONLYOFFICE 保存地址不受信任")
    internal_path = f"{internal.path.rstrip('/')}/{actual.path[len(proxy_prefix):]}"
    return urlunsplit((internal.scheme, internal.netloc, internal_path, actual.query, ""))


async def _download_result(url: str) -> bytes:
    download_url = _validate_result_url(url)
    try:
        async with httpx.AsyncClient(timeout=30.0, follow_redirects=False) as client:
            async with client.stream("GET", download_url) as response:
                if response.status_code != status.HTTP_200_OK:
                    raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail="ONLYOFFICE 保存文件下载失败")
                content = bytearray()
                async for chunk in response.aiter_bytes():
                    content.extend(chunk)
                    if len(content) > settings.COLLABORATION_MAX_FILE_BYTES:
                        raise HTTPException(status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE, detail="ONLYOFFICE 保存文件超出限制")
    except httpx.HTTPError as exc:
        raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail="ONLYOFFICE 保存文件下载失败") from exc
    return bytes(content)


async def _callback_user(
    db: AsyncSession, payload: CollaborationCallbackPayload, session: CollaborationSession
) -> User | None:
    has_public_share_user = False
    for value in payload.users:
        if value.startswith("share-"):
            has_public_share_user = True
            continue
        try:
            user = await db.get(User, uuid.UUID(value))
        except (ValueError, TypeError):
            user = None
        if user:
            return user
    if has_public_share_user:
        return None
    user = await db.get(User, session.started_by)
    if not user:
        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="协作会话用户不存在")
    return user


async def process_callback(
    db: AsyncSession,
    session_id: uuid.UUID,
    payload: CollaborationCallbackPayload,
) -> dict[str, int]:
    session = await db.scalar(
        select(CollaborationSession).where(CollaborationSession.id == session_id).with_for_update()
    )
    if not session:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="协作会话不存在")
    if not hmac.compare_digest(session.document_key, payload.key):
        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="协作会话 key 不匹配")
    fingerprint = _callback_fingerprint(payload)
    duplicate = await db.scalar(select(CollaborationCallbackReceipt.id).where(
        CollaborationCallbackReceipt.session_id == session.id,
        CollaborationCallbackReceipt.fingerprint == fingerprint,
    ))
    if duplicate:
        return {"error": 0}

    item = await db.get(CollaborationFile, session.file_id)
    if not item:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="协作文件不存在")
    session.last_callback_at = datetime.now(timezone.utc)
    session.active_users = json.dumps(payload.users, ensure_ascii=True)
    result = "ACKNOWLEDGED"
    saved_revision_id = None

    if payload.status in {2, 6}:
        if not payload.url:
            raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="ONLYOFFICE 保存回调缺少文件地址")
        if session.generation != item.generation:
            result = "STALE"
        else:
            content = await _download_result(payload.url)
            actor = await _callback_user(db, payload, session)
            source = "SESSION_CLOSE" if payload.status == 2 else "FORCE_SAVE"
            if actor is None:
                source = "SHARE_SESSION_CLOSE" if payload.status == 2 else "SHARE_FORCE_SAVE"
            revision, created = await collaboration_service.append_revision(
                db, item, content, source=source, created_by=actor.id if actor else None
            )
            saved_revision_id = revision.id
            result = "SAVED" if created else "UNCHANGED"
            if payload.status == 2:
                item.generation += 1
                session.status = "CLOSED"
                session.closed_at = datetime.now(timezone.utc)
            else:
                # 强制保存不结束当前共同编辑会话；同步基线可保证 Document
                # Server 缓存重建时仍从最近一次持久化内容恢复。
                session.base_revision_id = revision.id
    elif payload.status == 4:
        session.status = "CLOSED"
        session.closed_at = datetime.now(timezone.utc)
        result = "UNCHANGED"
    elif payload.status in {3, 7}:
        session.status = "ERROR"
        result = "ERROR"

    db.add(CollaborationCallbackReceipt(
        session_id=session.id,
        fingerprint=fingerprint,
        callback_status=payload.status,
        result=result,
        revision_id=saved_revision_id,
    ))
    await db.commit()
    # ONLYOFFICE 要求回调处理器在接收并记录状态后固定确认成功。
    # status 3/7 表示文档服务自身保存失败，不应通过 error=1 制造重试环。
    return {"error": 0}