from __future__ import annotations

import json
import os
import re
import shutil
import tempfile
import zipfile
from datetime import datetime
from decimal import Decimal, ROUND_HALF_UP
from pathlib import Path, PurePosixPath
from typing import Any

from fastapi import APIRouter, BackgroundTasks, File, Header, HTTPException, Query, Request, UploadFile
from fastapi.concurrency import run_in_threadpool
from fastapi.responses import FileResponse
from pydantic import BaseModel, Field

from api.image_inputs import parse_image_edit_request, read_image_sources
from api.support import require_identity, resolve_image_base_url
from services.content_filter import check_request
from services.config import config
from services.auth_service import DAILY_GIFT_ACCOUNT_POOL_MODEL_ID, auth_service
from services.image_task_service import image_task_service
from services.log_service import LoggedCall
from services.pool_pricing_service import pool_pricing_service
from services.share_evidence_service import share_evidence_service
from services.third_party_image_service import third_party_image_service


def _credit_amount(value: object) -> int | float:
    try:
        amount = max(Decimal("0"), Decimal(str(value or 0))).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
    except Exception:
        amount = Decimal("0")
    return int(amount) if amount == amount.to_integral_value() else float(amount)


def _third_party_activity_quote(identity: dict[str, object], *, group_id: str, base_unit_price: object) -> dict[str, object]:
    pricing_config = config.normal_user_third_party_image_pricing
    if str(identity.get("role") or "").strip().lower() != "normal" or not bool(pricing_config.get("enabled")):
        base = _credit_amount(base_unit_price)
        return {
            "applies": False,
            "target_type": "third_party_group",
            "third_party_group_id": group_id,
            "base_unit_price": base,
            "pay_percentage": 100,
            "unit_price": base,
            "quantity": 1,
            "estimated_total": base,
            "rule_id": "",
            "rule_name": "",
        }
    return pool_pricing_service.quote(
        identity,
        quantity=1,
        pricing_config=pricing_config,
        target_type="third_party_group",
        third_party_group_id=group_id,
        base_unit_price=base_unit_price,
    )


class ImageGenerationTaskRequest(BaseModel):
    client_task_id: str = Field(..., min_length=1)
    prompt: str = Field(..., min_length=1)
    model: str = "gpt-image-2"
    size: str | None = None
    quality: str | None = None
    resolution: str | None = None
    image_resolution: str | None = None
    third_party_image_group_id: str | None = None
    account_pool_strategy: str = "own_first"
    memory: bool = False
    memory_conversation_id: str = ""
    memory_reset: bool = False


class ResumePollRequest(BaseModel):
    extra_timeout_secs: float | None = Field(default=None, ge=5.0, le=120.0)


class ShareEvidenceRequest(BaseModel):
    conversation_id: str = Field(..., min_length=1)


class ImageHistoryExportRequest(BaseModel):
    """导出请求体。

    conversations 由前端带上（本地会话结构：标题、分组、轮次），服务端原样
    写进 manifest；include_images 决定是否把原图一起打进 zip。
    """

    conversations: list[Any] = Field(default_factory=list)
    include_images: bool = True


def _parse_task_ids(value: str) -> list[str]:
    return [item.strip() for item in value.split(",") if item.strip()]


def _require_owned_task(identity: dict[str, object], conversation_id: str) -> dict[str, object]:
    """校验这条上游会话确实属于当前用户，返回内部任务字典。

    没有这层校验的话，任何人拿到一个别人的 conversation_id 就能把它变成公开分享链接。

    ⚠️ 生图任务记录可能已被清理，但官网取证台账里还有这条会话 → 也算「属于你」。
    否则用户点「查看官网」时只会看到一句像越权警告的 404，而真实原因是
    「这条会话已过保留期，程序已按策略把官网对话删了」。
    """
    task = image_task_service.find_owned_conversation(identity, conversation_id)
    if task is None:
        raise HTTPException(status_code=404, detail={
            "error": "这条会话的官网记录已不存在：可能已过保留期被自动清理，或不属于当前账号",
            "reason": "conversation_not_found",
        })
    return task


def _internal_model_limit_kwargs(identity: dict[str, object], model: object) -> dict[str, object]:
    if str(identity.get("role") or "").strip().lower() != "user":
        return {}
    model_id = str(model or "gpt-image-2").strip() or "gpt-image-2"
    return {
        "internal_model_limit_user_id": str(identity.get("id") or "").strip(),
        "internal_model_limit_model_id": model_id,
        "internal_model_limit_amount": 1,
    }


def _credit_kwargs(identity: dict[str, object], model: object) -> dict[str, object]:
    if str(identity.get("role") or "").strip().lower() != "normal":
        return _internal_model_limit_kwargs(identity, model)
    pricing = pool_pricing_service.quote(identity, 1)
    return {
        "credit_user_id": str(identity.get("id") or "").strip(),
        "credit_amount": pricing["estimated_total"],
        "credit_model_id": DAILY_GIFT_ACCOUNT_POOL_MODEL_ID,
        "credit_pricing": pricing,
    }


def _third_party_wallet_remaining(identity: dict[str, object], group_id: str, model_id: str) -> int | float:
    try:
        summary = auth_service.credit_summary(identity)
    except Exception:
        return 0
    wallets = summary.get("third_party_image_wallets") if isinstance(summary, dict) else []
    wallet_key = f"{group_id}:{model_id}"
    wallet_items = wallets if isinstance(wallets, list) else []
    balances: list[int | float] = []
    for wallet in wallet_items:
        if not isinstance(wallet, dict):
            continue
        if str(wallet.get("id") or "") == wallet_key or (
            str(wallet.get("group_id") or "") == group_id
            and str(wallet.get("model_id") or "").strip() == model_id
        ):
            try:
                balances.append(_credit_amount(wallet.get("remaining")))
            except (TypeError, ValueError):
                balances.append(0)
    return max(balances) if balances else 0


def _image_api_billing_kwargs(
    identity: dict[str, object],
    *,
    model: object,
    quality: object = "",
    resolution: object = "",
    image_resolution: object = "",
    third_party_image_group_id: object = "",
) -> dict[str, object]:
    resolution = str(resolution or image_resolution or "").strip().lower()
    group_id = str(third_party_image_group_id or "").strip()
    explicit_third_party = resolution in {"1k", "2k", "4k"} or bool(group_id)
    route = None
    if not explicit_third_party:
        find_named_route = getattr(third_party_image_service, "find_named_route", None)
        route = find_named_route(model, advance=False) if callable(find_named_route) else None
    if not explicit_third_party and route is None:
        return _credit_kwargs(identity, model)
    require_4k = resolution == "4k"
    if route is None and str(identity.get("role") or "").strip().lower() == "normal":
        find_wallet_route = getattr(third_party_image_service, "find_wallet_route_for_identity", None)
        if callable(find_wallet_route):
            route = find_wallet_route(
                identity,
                model,
                group_id=group_id,
                require_4k=require_4k,
                resolution=resolution,
                advance=False,
            )
    if route is None:
        route = third_party_image_service.find_route(
            model,
            group_id=group_id,
            require_4k=require_4k,
            resolution=resolution or "1k",
            advance=False,
        )
    if route is None:
        raise ValueError("No enabled image API model route")
    try:
        third_party_image_service.assert_access(identity, route)
    except PermissionError as exc:
        raise PermissionError(str(exc)) from exc
    cost = third_party_image_service.cost_for_route(route, quality, 1)
    common = {
        "resolution": str(route.model.get("resolution") or resolution or "1k"),
        "third_party_image_group_id": str(route.group.get("id") or ""),
    }
    route_model_id = str(route.model.get("id") or "")
    if str(identity.get("role") or "").strip().lower() != "normal":
        return {
            **common,
            **_internal_model_limit_kwargs(identity, route_model_id),
        }
    billing_mode = str(route.group.get("billing_mode") or "")
    route_group_id = str(route.group.get("id") or "")
    if billing_mode == "general_multiplier" and _third_party_wallet_remaining(identity, route_group_id, route_model_id) > 0:
        # Keep existing model-specific dedicated balances usable even if an
        # older/stale group setting still says general-only.
        billing_mode = "both"
    wallet_pricing = _third_party_activity_quote(identity, group_id=route_group_id, base_unit_price=1)
    wallet_cost = _credit_amount(wallet_pricing.get("estimated_total", 1))
    if billing_mode == "dedicated_wallet" or (
        billing_mode == "both" and cost > 0 and _third_party_wallet_remaining(identity, route_group_id, route_model_id) >= wallet_cost
    ):
        wallet_kwargs: dict[str, object] = {
            **common,
            "third_party_wallet_user_id": str(identity.get("id") or "").strip(),
            "third_party_wallet_group_id": route_group_id,
            "third_party_wallet_model_id": route_model_id,
            "third_party_wallet_amount": wallet_cost,
            "third_party_wallet_cost_per_image": _credit_amount(wallet_pricing.get("unit_price", 1)),
            "credit_pricing": wallet_pricing,
        }
        if billing_mode == "both" and cost > 0:
            fallback_pricing = _third_party_activity_quote(identity, group_id=route_group_id, base_unit_price=cost)
            wallet_kwargs["third_party_wallet_fallback_credit_user_id"] = str(identity.get("id") or "").strip()
            wallet_kwargs["third_party_wallet_fallback_credit_amount"] = fallback_pricing["estimated_total"]
            wallet_kwargs["third_party_wallet_fallback_credit_model_id"] = f"{route_group_id}:{route_model_id}"
            wallet_kwargs["third_party_wallet_fallback_credit_pricing"] = fallback_pricing
        return wallet_kwargs
    pricing = _third_party_activity_quote(identity, group_id=route_group_id, base_unit_price=cost)
    return {
        **common,
        "credit_user_id": str(identity.get("id") or "").strip(),
        "credit_amount": pricing["estimated_total"],
        "credit_model_id": f"{route_group_id}:{route_model_id}",
        "credit_pricing": pricing,
    }


def _generation_billing_kwargs(identity: dict[str, object], body: ImageGenerationTaskRequest) -> dict[str, object]:
    return _image_api_billing_kwargs(
        identity,
        model=body.model,
        quality=body.quality,
        resolution=body.resolution,
        image_resolution=body.image_resolution,
        third_party_image_group_id=body.third_party_image_group_id,
    )


def _effective_account_pool_strategy(identity: dict[str, object], value: object) -> str:
    if identity.get("role") == "admin" or bool(identity.get("account_pool_enabled")):
        return str(value or "own_first")
    return "admin_only"


async def filter_or_log(call: LoggedCall, text: str) -> None:
    try:
        await run_in_threadpool(check_request, text)
    except HTTPException as exc:
        call.log("调用失败", status="failed", error=str(exc.detail))
        raise


# ---------------------------------------------------------------------------
# 历史记录导出 / 导入
#
# 包结构（zip）：
#   manifest.json          记录本体（tasks 为原始任务结构，不做 _public_task 裁剪）
#   images/<相对路径>      任务里引用到的原图，路径与 /images/<rel> 完全一致
#
# 不写 image_index.json：/images/{path} 直接读文件，历史记录靠 task.data[].url
# 就能显示；索引只影响「图片管理」页的列表。刻意不碰索引，避免与
# image_storage_service 的写锁互相干扰。
# ---------------------------------------------------------------------------

# 导出包的 format 字段会随 zip 离开本机（用户分享/网盘备份），不能带项目名，
# 新导出用中性值；导入同时认旧值，老包还能导回来。
HISTORY_FORMAT = "image-history-package"
HISTORY_FORMATS = ("image-history-package", "chatgpt2api-image-history")
HISTORY_VERSION = 1
_IMAGE_URL_RE = re.compile(r"/images/(.+)$")
_ZIP_MAX_UNCOMPRESSED = 8 * 1024 * 1024 * 1024  # 防 zip bomb：解压后上限 8GB


def _rel_from_image_url(url: object) -> str:
    """从 `<任意域名>/images/<rel>` 里取出安全的相对路径，非法则返回空串。"""
    text = str(url or "").strip()
    if not text:
        return ""
    match = _IMAGE_URL_RE.search(text)
    if not match:
        return ""
    rel = match.group(1).split("?", 1)[0].split("#", 1)[0].strip()
    if not rel:
        return ""
    pure = PurePosixPath(rel)
    if pure.is_absolute() or ".." in pure.parts:
        return ""
    return pure.as_posix()


def _collect_task_image_rels(tasks: list[object]) -> list[str]:
    rels: set[str] = set()
    for task in tasks:
        if not isinstance(task, dict):
            continue
        data = task.get("data")
        if not isinstance(data, list):
            continue
        for item in data:
            if not isinstance(item, dict):
                continue
            rel = _rel_from_image_url(item.get("url"))
            if rel:
                rels.add(rel)
    return sorted(rels)


def _rewrite_task_image_urls(task: dict, base_url: str) -> None:
    """把记录里的图片地址改写到当前部署，否则换环境后图片全部指向旧域名。"""
    data = task.get("data")
    if not isinstance(data, list):
        return
    for item in data:
        if not isinstance(item, dict):
            continue
        rel = _rel_from_image_url(item.get("url"))
        if rel:
            item["url"] = f"{base_url}/images/{rel}" if base_url else f"/images/{rel}"


def _build_history_archive(
    identity: dict[str, object],
    include_images: bool,
    conversations: list[object],
) -> tuple[str, str, dict[str, object]]:
    tasks = image_task_service.export_owned_tasks(identity)
    rels = _collect_task_image_rels(tasks) if include_images else []
    manifest = {
        "format": HISTORY_FORMAT,
        "version": HISTORY_VERSION,
        "exported_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
        "owner_id": str(identity.get("id") or ""),
        "count": len(tasks),
        "conversation_count": len(conversations),
        "includes_images": bool(include_images),
        # 历史记录的「会话」结构（标题/分组/轮次）存在浏览器本地，服务端不知道，
        # 由前端在导出时一并带上来，导入时原样还回去，否则导入后标题和分组会丢。
        "conversations": conversations,
        "tasks": tasks,
    }
    images_dir = config.images_dir
    handle, archive_path = tempfile.mkstemp(prefix="image-history-", suffix=".zip")
    os.close(handle)
    written = 0
    missing = 0
    # 图片本身已是压缩格式，用 STORED 省 CPU（线上只有 2 个共享 vCPU）
    with zipfile.ZipFile(archive_path, "w", zipfile.ZIP_STORED) as archive:
        archive.writestr("manifest.json", json.dumps(manifest, ensure_ascii=False, indent=2))
        for rel in rels:
            source = images_dir / rel
            try:
                if not source.is_file():
                    missing += 1
                    continue
                archive.write(source, f"images/{rel}")
                written += 1
            except OSError:
                missing += 1
    filename = f"image-history-{datetime.now().strftime('%Y%m%d-%H%M%S')}.zip"
    stats = {"tasks": len(tasks), "images": written, "missing_images": missing}
    return archive_path, filename, stats


def _history_prefix(archive: zipfile.ZipFile) -> str:
    """返回 manifest.json 在包内的目录前缀（含结尾斜杠，根目录时为空串）。

    正常导出的包 manifest.json 就在根目录。但用解压工具或部分下载器重新打包时，
    会给整包套一层目录，变成 `image-history-xxx/manifest.json`。这时按根目录找
    会直接报「缺少 manifest.json」，用户会以为包坏了，实际只是路径多了一层。
    只接受正好一层目录，避免误命中 images/ 内部碰巧叫 manifest.json 的文件。
    """
    names = set(archive.namelist())
    if "manifest.json" in names:
        return ""
    candidates = [name[: -len("manifest.json")] for name in names if name.endswith("/manifest.json")]
    candidates = [item for item in candidates if item.count("/") == 1]
    return candidates[0] if len(candidates) == 1 else ""


def _apply_history_archive(identity: dict[str, object], file_obj) -> dict[str, object]:
    try:
        archive = zipfile.ZipFile(file_obj)
    except zipfile.BadZipFile as exc:
        raise HTTPException(status_code=400, detail={"error": "文件不是有效的 zip 包"}) from exc

    with archive:
        prefix = _history_prefix(archive)
        try:
            raw_manifest = archive.read(f"{prefix}manifest.json")
        except KeyError as exc:
            raise HTTPException(status_code=400, detail={"error": "包里缺少 manifest.json，不是本系统导出的历史记录"}) from exc
        try:
            manifest = json.loads(raw_manifest.decode("utf-8"))
        except (UnicodeDecodeError, json.JSONDecodeError) as exc:
            raise HTTPException(status_code=400, detail={"error": "manifest.json 无法解析"}) from exc
        if not isinstance(manifest, dict) or manifest.get("format") not in HISTORY_FORMATS:
            raise HTTPException(status_code=400, detail={"error": "包格式不匹配，不是图片历史记录包"})

        tasks = manifest.get("tasks")
        if not isinstance(tasks, list):
            tasks = []

        images_dir = config.images_dir
        total_uncompressed = sum(info.file_size for info in archive.infolist())
        if total_uncompressed > _ZIP_MAX_UNCOMPRESSED:
            raise HTTPException(status_code=400, detail={"error": "包解压后过大，已拒绝导入"})

        images_written = 0
        images_skipped = 0
        images_prefix = f"{prefix}images/"
        for info in archive.infolist():
            if info.is_dir() or not info.filename.startswith(images_prefix):
                continue
            rel = info.filename[len(images_prefix):]
            pure = PurePosixPath(rel)
            if not rel or pure.is_absolute() or ".." in pure.parts:
                continue  # 防 zip slip
            target = images_dir / pure
            try:
                if target.exists():
                    images_skipped += 1
                    continue
                target.parent.mkdir(parents=True, exist_ok=True)
                with archive.open(info) as source, target.open("wb") as destination:
                    shutil.copyfileobj(source, destination, length=1024 * 256)
                images_written += 1
            except OSError:
                images_skipped += 1

    base_url = config.base_url
    for task in tasks:
        if isinstance(task, dict):
            _rewrite_task_image_urls(task, base_url)

    result = image_task_service.import_tasks(identity, tasks)
    result["images"] = images_written
    result["images_skipped"] = images_skipped
    conversations = manifest.get("conversations")
    if isinstance(conversations, list):
        # 会话结构还给前端，由它写回本地存储（服务端不持有这些数据）
        result["conversations"] = conversations
    return result


def create_router() -> APIRouter:
    router = APIRouter()

    @router.get("/api/image-tasks")
    async def list_image_tasks(
        ids: str = Query(default=""),
        authorization: str | None = Header(default=None),
    ):
        identity = require_identity(authorization)
        return await run_in_threadpool(image_task_service.list_tasks, identity, _parse_task_ids(ids))

    @router.post("/api/image-tasks/export")
    async def export_image_history(
        body: ImageHistoryExportRequest,
        background_tasks: BackgroundTasks,
        authorization: str | None = Header(default=None),
    ):
        """把当前用户的全部图片历史导出成 zip（记录 + 本地会话结构 + 可选原图）。"""
        identity = require_identity(authorization)
        archive_path, filename, _stats = await run_in_threadpool(
            _build_history_archive,
            identity,
            bool(body.include_images),
            list(body.conversations or []),
        )

        def _drop_archive(path: str = archive_path) -> None:
            try:
                os.unlink(path)
            except OSError:
                pass

        background_tasks.add_task(_drop_archive)
        return FileResponse(
            archive_path,
            media_type="application/zip",
            filename=filename,
            headers={"Cache-Control": "no-store"},
        )

    @router.post("/api/image-tasks/import")
    async def import_image_history(
        file: UploadFile = File(...),
        authorization: str | None = Header(default=None),
    ):
        """导入历史记录 zip：同 ID 跳过、其余合并，图片落到本机 images 目录。"""
        identity = require_identity(authorization)
        name = str(file.filename or "")
        if not name.lower().endswith(".zip"):
            raise HTTPException(status_code=400, detail={"error": "请选择 .zip 格式的历史记录包"})
        try:
            return await run_in_threadpool(_apply_history_archive, identity, file.file)
        finally:
            await file.close()

    @router.post("/api/image-tasks/generations")
    async def create_generation_task(
        body: ImageGenerationTaskRequest,
        request: Request,
        authorization: str | None = Header(default=None),
    ):
        identity = require_identity(authorization)
        await filter_or_log(LoggedCall(identity, "/api/image-tasks/generations", body.model, "文生图任务", request_text=body.prompt), body.prompt)
        try:
            return await run_in_threadpool(
                image_task_service.submit_generation,
                identity,
                client_task_id=body.client_task_id,
                prompt=body.prompt,
                model=body.model,
                size=body.size,
                quality=str(body.quality or ""),
                base_url=resolve_image_base_url(request),
                memory_enabled=body.memory,
                memory_conversation_id=body.memory_conversation_id,
                memory_reset=body.memory_reset,
                account_pool_strategy=_effective_account_pool_strategy(identity, body.account_pool_strategy),
                **_generation_billing_kwargs(identity, body),
            )
        except PermissionError as exc:
            raise HTTPException(status_code=403, detail={"error": str(exc)}) from exc
        except ValueError as exc:
            error_text = str(exc)
            status_code = 402 if "次数不足" in error_text or "not enough" in error_text.lower() or "balance" in error_text.lower() else 400
            raise HTTPException(status_code=status_code, detail={"error": str(exc)}) from exc

    @router.post("/api/image-tasks/edits")
    async def create_edit_task(
        request: Request,
        authorization: str | None = Header(default=None),
    ):
        identity = require_identity(authorization)
        payload, image_sources, mask_sources = await parse_image_edit_request(request)
        client_task_id = str(payload.get("client_task_id") or "").strip()
        if not client_task_id:
            raise HTTPException(status_code=400, detail={"error": "client_task_id is required"})
        prompt = str(payload["prompt"])
        model = str(payload["model"])
        await filter_or_log(LoggedCall(identity, "/api/image-tasks/edits", model, "图生图任务", request_text=prompt), prompt)
        images = await read_image_sources(image_sources)
        masks = await read_image_sources(mask_sources) if mask_sources else None
        try:
            return await run_in_threadpool(
                image_task_service.submit_edit,
                identity,
                client_task_id=client_task_id,
                prompt=prompt,
                model=model,
                size=payload["size"],
                base_url=resolve_image_base_url(request),
                images=images,
                masks=masks,
                memory_enabled=bool(payload.get("memory")),
                memory_conversation_id=str(payload.get("memory_conversation_id") or ""),
                memory_reset=bool(payload.get("memory_reset")),
                account_pool_strategy=_effective_account_pool_strategy(identity, payload.get("account_pool_strategy") or "own_first"),
                quality=str(payload.get("quality") or ""),
                **_image_api_billing_kwargs(
                    identity,
                    model=model,
                    quality=payload.get("quality"),
                    resolution=payload.get("resolution"),
                    image_resolution=payload.get("image_resolution"),
                    third_party_image_group_id=payload.get("third_party_image_group_id"),
                ),
            )
        except PermissionError as exc:
            raise HTTPException(status_code=403, detail={"error": str(exc)}) from exc
        except ValueError as exc:
            status_code = 402 if "次数不足" in str(exc) else 400
            raise HTTPException(status_code=status_code, detail={"error": str(exc)}) from exc

    @router.post("/api/image-tasks/{task_id}/resume-poll")
    async def resume_image_poll(
        task_id: str,
        body: ResumePollRequest,
        authorization: str | None = Header(default=None),
    ):
        identity = require_identity(authorization)
        try:
            return await run_in_threadpool(
                image_task_service.resume_poll,
                identity,
                task_id,
                body.extra_timeout_secs or config.image_timeout_retry_secs,
            )
        except ValueError as exc:
            raise HTTPException(status_code=400, detail={"error": str(exc)}) from exc

    @router.get("/api/image-tasks/share-evidence")
    async def share_evidence_status(
        conversation_id: str = Query(..., min_length=1),
        authorization: str | None = Header(default=None),
    ):
        """查询某条会话的「官网来源」取证状态（链接、剩余有效秒数）。"""
        identity = require_identity(authorization)
        _require_owned_task(identity, conversation_id)
        return share_evidence_service.describe(conversation_id)

    @router.post("/api/image-tasks/share-evidence")
    async def create_share_evidence(
        body: ShareEvidenceRequest,
        authorization: str | None = Header(default=None),
    ):
        """把一条会话取成为 ChatGPT 官方匿名分享链接（幂等）。"""
        identity = require_identity(authorization)
        if not config.image_share_evidence_enabled:
            raise HTTPException(status_code=403, detail={"error": "后台未开启「官网来源取证」，无法生成分享链接"})
        task = _require_owned_task(identity, body.conversation_id)
        try:
            await run_in_threadpool(
                share_evidence_service.create_share,
                conversation_id=body.conversation_id,
                owner_id=str(identity.get("id") or ""),
                account_email=str(task.get("account_email") or ""),
            )
        except ValueError as exc:
            raise HTTPException(status_code=400, detail={"error": str(exc)}) from exc
        except RuntimeError as exc:
            raise HTTPException(status_code=502, detail={"error": str(exc)}) from exc
        return share_evidence_service.describe(body.conversation_id)

    @router.delete("/api/image-tasks/share-evidence")
    async def revoke_share_evidence(
        conversation_id: str = Query(..., min_length=1),
        authorization: str | None = Header(default=None),
    ):
        """提前撤销分享链接并回收会话。"""
        identity = require_identity(authorization)
        _require_owned_task(identity, conversation_id)
        try:
            await run_in_threadpool(share_evidence_service.revoke, conversation_id)
        except ValueError as exc:
            raise HTTPException(status_code=400, detail={"error": str(exc)}) from exc
        except RuntimeError as exc:
            raise HTTPException(status_code=502, detail={"error": str(exc)}) from exc
        return share_evidence_service.describe(conversation_id)

    return router