services/image_task_service.py

from __future__ import annotations

import json
import threading
import time
from collections.abc import Callable
from datetime import datetime
from pathlib import Path
from typing import Any

from services.config import DATA_DIR, config
from services.content_filter import request_text
from services.log_service import LOG_TYPE_CALL, log_service
from services.protocol import openai_v1_image_edit, openai_v1_image_generations
from services.time_utils import china_from_timestamp_string, china_now_string

TASK_STATUS_QUEUED = "queued"
TASK_STATUS_RUNNING = "running"
TASK_STATUS_SUCCESS = "success"
TASK_STATUS_ERROR = "error"
TERMINAL_STATUSES = {TASK_STATUS_SUCCESS, TASK_STATUS_ERROR}
UNFINISHED_STATUSES = {TASK_STATUS_QUEUED, TASK_STATUS_RUNNING}
TASK_CLEANUP_INTERVAL_SECONDS = 60 * 60


def _now_iso() -> str:
    return china_now_string()


def _timestamp(value: object) -> float:
    if not isinstance(value, str) or not value.strip():
        return 0.0
    for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%dT%H:%M:%S.%f", "%Y-%m-%dT%H:%M:%S"):
        try:
            return datetime.strptime(value[:26], fmt).timestamp()
        except ValueError:
            continue
    try:
        return datetime.fromisoformat(value.replace("Z", "+00:00")).timestamp()
    except Exception:
        return 0.0


def _clean(value: object, default: str = "") -> str:
    return str(value or default).strip()


def _owner_id(identity: dict[str, object]) -> str:
    return _clean(identity.get("id")) or "anonymous"


def _memory_conversation_id(value: object) -> str:
    return _clean(value) or "default"


def _account_pool_strategy(value: object) -> str:
    normalized = _clean(value).lower().replace("-", "_")
    return normalized if normalized in {"own_first", "admin_first", "shared_first", "own_only", "admin_only", "shared_only"} else "own_first"


def _task_key(owner_id: str, task_id: str) -> str:
    return f"{owner_id}:{task_id}"


def _collect_image_urls(data: list[Any]) -> list[str]:
    urls: list[str] = []
    for item in data:
        if isinstance(item, dict):
            url = item.get("url")
            if isinstance(url, str) and url:
                urls.append(url)
    return urls


def _count_image_items(data: object) -> int:
    if not isinstance(data, list):
        return 0
    count = 0
    for item in data:
        if isinstance(item, dict) and (_clean(item.get("url")) or _clean(item.get("b64_json"))):
            count += 1
    return count


def _collect_third_party_image_api(result: object) -> dict[str, Any]:
    if isinstance(result, dict) and isinstance(result.get("_third_party_image_api"), dict):
        return result["_third_party_image_api"]
    return {}


def _public_task(task: dict[str, Any]) -> dict[str, Any]:
    item = {
        "id": task.get("id"),
        "status": task.get("status"),
        "mode": task.get("mode"),
        "model": task.get("model"),
        "size": task.get("size"),
        "created_at": task.get("created_at"),
        "started_at": task.get("started_at"),
        "updated_at": task.get("updated_at"),
    }
    if task.get("progress"):
        item["progress"] = task.get("progress")
    if isinstance(task.get("progress_events"), list):
        item["progress_events"] = [
            event
            for event in task.get("progress_events", [])
            if isinstance(event, dict) and _clean(event.get("step"))
        ]
    if task.get("duration_ms") is not None:
        item["duration_ms"] = task.get("duration_ms")
    if task.get("status") in UNFINISHED_STATUSES:
        base_ts = task.get("started_ts") if task.get("status") == TASK_STATUS_RUNNING else task.get("created_ts")
        if not base_ts:
            base_ts = _timestamp(task.get("started_at") if task.get("status") == TASK_STATUS_RUNNING else task.get("created_at"))
        if base_ts:
            item["elapsed_secs"] = round(time.time() - float(base_ts), 1)
    if task.get("data") is not None:
        item["data"] = task.get("data")
    if task.get("error"):
        item["error"] = task.get("error")
    return item


class ImageTaskService:
    def __init__(
        self,
        path: Path,
        *,
        generation_handler: Callable[[dict[str, Any]], dict[str, Any]] = openai_v1_image_generations.handle,
        edit_handler: Callable[[dict[str, Any]], dict[str, Any]] = openai_v1_image_edit.handle,
        retention_days_getter: Callable[[], int] | None = None,
    ):
        self.path = path
        self.generation_handler = generation_handler
        self.edit_handler = edit_handler
        self.retention_days_getter = retention_days_getter or (lambda: config.image_retention_days)
        self._lock = threading.RLock()
        self._tasks: dict[str, dict[str, Any]] = {}
        self._last_cleanup_at = time.monotonic()
        self.path.parent.mkdir(parents=True, exist_ok=True)
        with self._lock:
            self._tasks = self._load_locked()
            changed = self._recover_unfinished_locked()
            changed = self._cleanup_locked() or changed
            if changed:
                self._save_locked()

    def submit_generation(
        self,
        identity: dict[str, object],
        *,
        client_task_id: str,
        prompt: str,
        model: str,
        size: str | None,
        base_url: str,
        memory_enabled: bool = False,
        memory_conversation_id: str = "",
        memory_reset: bool = False,
        account_pool_strategy: str = "own_first",
        credit_user_id: str = "",
        credit_amount: int = 0,
        quality: str = "",
        resolution: str = "",
        third_party_image_group_id: str = "",
        third_party_wallet_user_id: str = "",
        third_party_wallet_group_id: str = "",
        third_party_wallet_model_id: str = "",
        third_party_wallet_amount: int = 0,
        third_party_wallet_cost_per_image: int = 1,
        third_party_wallet_fallback_credit_user_id: str = "",
        third_party_wallet_fallback_credit_amount: int = 0,
    ) -> dict[str, Any]:
        payload = {
            "prompt": prompt,
            "model": model,
            "n": 1,
            "size": size,
            "quality": quality,
            "resolution": resolution,
            "third_party_image_group_id": third_party_image_group_id,
            "response_format": "url",
            "base_url": base_url,
            "account_owner_id": _owner_id(identity),
            "account_pool_strategy": _account_pool_strategy(account_pool_strategy),
            "account_allow_shared": not memory_enabled,
        }
        if memory_enabled:
            payload["memory"] = True
            payload["memory_owner_id"] = _owner_id(identity)
            payload["memory_conversation_id"] = _memory_conversation_id(memory_conversation_id)
            if memory_reset:
                payload["memory_reset"] = True
        return self._submit(
            identity,
            client_task_id=client_task_id,
            mode="generate",
            payload=payload,
            credit_user_id=credit_user_id,
            credit_amount=credit_amount,
            third_party_wallet_user_id=third_party_wallet_user_id,
            third_party_wallet_group_id=third_party_wallet_group_id,
            third_party_wallet_model_id=third_party_wallet_model_id,
            third_party_wallet_amount=third_party_wallet_amount,
            third_party_wallet_cost_per_image=third_party_wallet_cost_per_image,
            third_party_wallet_fallback_credit_user_id=third_party_wallet_fallback_credit_user_id,
            third_party_wallet_fallback_credit_amount=third_party_wallet_fallback_credit_amount,
        )

    def submit_edit(
        self,
        identity: dict[str, object],
        *,
        client_task_id: str,
        prompt: str,
        model: str,
        size: str | None,
        base_url: str,
        images: list[tuple[bytes, str, str]],
        memory_enabled: bool = False,
        memory_conversation_id: str = "",
        memory_reset: bool = False,
        account_pool_strategy: str = "own_first",
        credit_user_id: str = "",
        credit_amount: int = 0,
        quality: str = "",
        resolution: str = "",
        third_party_image_group_id: str = "",
        third_party_wallet_user_id: str = "",
        third_party_wallet_group_id: str = "",
        third_party_wallet_model_id: str = "",
        third_party_wallet_amount: int = 0,
        third_party_wallet_cost_per_image: int = 1,
        third_party_wallet_fallback_credit_user_id: str = "",
        third_party_wallet_fallback_credit_amount: int = 0,
    ) -> dict[str, Any]:
        payload = {
            "prompt": prompt,
            "images": images,
            "model": model,
            "n": 1,
            "size": size,
            "quality": quality,
            "resolution": resolution,
            "third_party_image_group_id": third_party_image_group_id,
            "response_format": "url",
            "base_url": base_url,
            "account_owner_id": _owner_id(identity),
            "account_pool_strategy": _account_pool_strategy(account_pool_strategy),
            "account_allow_shared": not memory_enabled,
        }
        if memory_enabled:
            payload["memory"] = True
            payload["memory_owner_id"] = _owner_id(identity)
            payload["memory_conversation_id"] = _memory_conversation_id(memory_conversation_id)
            if memory_reset:
                payload["memory_reset"] = True
        return self._submit(
            identity,
            client_task_id=client_task_id,
            mode="edit",
            payload=payload,
            credit_user_id=credit_user_id,
            credit_amount=credit_amount,
            third_party_wallet_user_id=third_party_wallet_user_id,
            third_party_wallet_group_id=third_party_wallet_group_id,
            third_party_wallet_model_id=third_party_wallet_model_id,
            third_party_wallet_amount=third_party_wallet_amount,
            third_party_wallet_cost_per_image=third_party_wallet_cost_per_image,
            third_party_wallet_fallback_credit_user_id=third_party_wallet_fallback_credit_user_id,
            third_party_wallet_fallback_credit_amount=third_party_wallet_fallback_credit_amount,
        )

    def list_tasks(self, identity: dict[str, object], task_ids: list[str]) -> dict[str, Any]:
        owner = _owner_id(identity)
        requested_ids = [_clean(task_id) for task_id in task_ids if _clean(task_id)]
        with self._lock:
            if self._cleanup_if_due_locked():
                self._save_locked()
            items = []
            missing_ids = []
            for task_id in requested_ids:
                task = self._tasks.get(_task_key(owner, task_id))
                if task is None:
                    missing_ids.append(task_id)
                else:
                    items.append(_public_task(task))
            if not requested_ids:
                items = [
                    _public_task(task)
                    for task in self._tasks.values()
                    if task.get("owner_id") == owner
                ]
                items.sort(key=lambda item: str(item.get("updated_at") or ""), reverse=True)
                missing_ids = []
            return {"items": items, "missing_ids": missing_ids}

    def _submit(
        self,
        identity: dict[str, object],
        *,
        client_task_id: str,
        mode: str,
        payload: dict[str, Any],
        credit_user_id: str = "",
        credit_amount: int = 0,
        third_party_wallet_user_id: str = "",
        third_party_wallet_group_id: str = "",
        third_party_wallet_model_id: str = "",
        third_party_wallet_amount: int = 0,
        third_party_wallet_cost_per_image: int = 1,
        third_party_wallet_fallback_credit_user_id: str = "",
        third_party_wallet_fallback_credit_amount: int = 0,
    ) -> dict[str, Any]:
        task_id = _clean(client_task_id)
        if not task_id:
            raise ValueError("client_task_id is required")
        owner = _owner_id(identity)
        key = _task_key(owner, task_id)
        now = _now_iso()
        should_start = False
        with self._lock:
            cleaned = self._cleanup_if_due_locked()
            task = self._tasks.get(key)
            if task is not None:
                if cleaned:
                    self._save_locked()
                return _public_task(task)
            normalized_credit_user_id = _clean(credit_user_id)
            normalized_credit_amount = max(0, int(credit_amount or 0))
            normalized_fallback_credit_user_id = _clean(third_party_wallet_fallback_credit_user_id)
            normalized_fallback_credit_amount = max(0, int(third_party_wallet_fallback_credit_amount or 0))
            daily_reserved = 0
            paid_reserved = 0
            package_reservations: list[dict[str, object]] = []
            grant_reservations: list[dict[str, object]] = []
            normalized_wallet_user_id = _clean(third_party_wallet_user_id)
            normalized_wallet_group_id = _clean(third_party_wallet_group_id)
            normalized_wallet_model_id = _clean(third_party_wallet_model_id)
            normalized_wallet_amount = max(0, int(third_party_wallet_amount or 0))
            normalized_wallet_cost_per_image = max(0, int(third_party_wallet_cost_per_image or 0))
            if normalized_wallet_user_id and normalized_wallet_group_id and normalized_wallet_model_id and normalized_wallet_amount > 0:
                from services.auth_service import auth_service

                try:
                    auth_service.reserve_third_party_image_wallet(
                        normalized_wallet_user_id,
                        group_id=normalized_wallet_group_id,
                        model_id=normalized_wallet_model_id,
                        amount=normalized_wallet_amount,
                    )
                except ValueError:
                    if not normalized_fallback_credit_user_id or normalized_fallback_credit_amount <= 0:
                        raise
                    normalized_wallet_user_id = ""
                    normalized_wallet_group_id = ""
                    normalized_wallet_model_id = ""
                    normalized_wallet_amount = 0
                    normalized_wallet_cost_per_image = 0
                    normalized_credit_user_id = normalized_fallback_credit_user_id
                    normalized_credit_amount = normalized_fallback_credit_amount
            if normalized_credit_user_id and normalized_credit_amount > 0:
                from services.auth_service import auth_service

                reservation = auth_service.reserve_credits(normalized_credit_user_id, normalized_credit_amount).get("_credit_reservation", {})
                if not isinstance(reservation, dict):
                    reservation = {}
                daily_reserved = max(0, int(reservation.get("daily") or 0))
                paid_reserved = max(0, int(reservation.get("permanent_paid") if "permanent_paid" in reservation else reservation.get("paid") or 0))
                package_reservations = reservation.get("packages") if isinstance(reservation.get("packages"), list) else []
                grant_reservations = reservation.get("grants") if isinstance(reservation.get("grants"), list) else []
            task = {
                "id": task_id,
                "owner_id": owner,
                "status": TASK_STATUS_QUEUED,
                "mode": mode,
                "model": _clean(payload.get("model"), "gpt-image-2"),
                "size": _clean(payload.get("size")),
                "created_at": now,
                "updated_at": now,
                "created_ts": time.time(),
                "updated_ts": time.time(),
                "progress_events": [],
            }
            if normalized_credit_user_id and normalized_credit_amount > 0:
                task["credit_user_id"] = normalized_credit_user_id
                task["credit_reserved"] = normalized_credit_amount
                task["credit_daily_reserved"] = daily_reserved
                task["credit_paid_reserved"] = paid_reserved
                task["credit_package_reservations"] = package_reservations
                task["credit_grant_reservations"] = grant_reservations
                task["credit_settled"] = False
            if normalized_wallet_user_id and normalized_wallet_amount > 0:
                task["third_party_wallet_user_id"] = normalized_wallet_user_id
                task["third_party_wallet_group_id"] = normalized_wallet_group_id
                task["third_party_wallet_model_id"] = normalized_wallet_model_id
                task["third_party_wallet_reserved"] = normalized_wallet_amount
                task["third_party_wallet_cost_per_image"] = normalized_wallet_cost_per_image
                task["third_party_wallet_settled"] = False
            self._tasks[key] = task
            self._save_locked()
            should_start = True

        if should_start:
            thread = threading.Thread(
                target=self._run_task,
                args=(key, mode, payload, dict(identity), _clean(payload.get("model"), "gpt-image-2")),
                name=f"image-task-{task_id[:16]}",
                daemon=True,
            )
            thread.start()
        return _public_task(task)

    def _run_task(
        self,
        key: str,
        mode: str,
        payload: dict[str, Any],
        identity: dict[str, object],
        model: str,
    ) -> None:
        started = time.time()
        self._update_task(key, status=TASK_STATUS_RUNNING, started_at=_now_iso(), error="")
        account_email = ""

        def progress_callback(step: str) -> None:
            cleaned_step = _clean(step) or "generating"
            self._record_progress(key, cleaned_step, set_started_ts=cleaned_step == "image_stream_resolve_start")

        payload_with_progress = {**payload, "progress_callback": progress_callback}
        try:
            handler = self.edit_handler if mode == "edit" else self.generation_handler
            result = handler(payload_with_progress)
            if not isinstance(result, dict):
                raise RuntimeError("image task returned streaming result unexpectedly")
            data = result.get("data")
            account_email = _clean(result.get("_account_email") or result.get("account_email"))
            if not isinstance(data, list) or not data:
                upstream = _clean(result.get("message"))
                if upstream:
                    message = upstream
                else:
                    message = "号池中没有可用账号或所有账号均被限流，请检查号池状态（账号额度、是否被封禁、是否到达生图上限）"
                raise RuntimeError(message)
            duration_ms = int((time.time() - started) * 1000)
            self._update_task(key, status=TASK_STATUS_SUCCESS, data=data, error="", duration_ms=duration_ms)
            self._settle_credit(key, success=True)
            self._settle_third_party_wallet(key, success_count=_count_image_items(data))
            self._log_call(
                identity,
                mode,
                model,
                started,
                "调用完成",
                request_preview=request_text(payload.get("prompt")),
                urls=_collect_image_urls(data),
                account_email=account_email,
                image_api=_collect_third_party_image_api(result),
            )
        except Exception as exc:
            error_message = str(exc) or "image task failed"
            account_email = _clean(getattr(exc, "account_email", account_email))
            duration_ms = int((time.time() - started) * 1000)
            self._update_task(key, status=TASK_STATUS_ERROR, error=error_message, data=[], duration_ms=duration_ms)
            self._settle_credit(key, success=False)
            self._settle_third_party_wallet(key, success_count=0)
            self._log_call(
                identity,
                mode,
                model,
                started,
                "调用失败",
                request_preview=request_text(payload.get("prompt")),
                status="failed",
                error=error_message,
                account_email=account_email,
            )

    def _log_call(
        self,
        identity: dict[str, object],
        mode: str,
        model: str,
        started: float,
        suffix: str,
        *,
        request_preview: str = "",
        status: str = "success",
        error: str = "",
        urls: list[str] | None = None,
        account_email: str = "",
        image_api: dict[str, Any] | None = None,
    ) -> None:
        endpoint = "/v1/images/edits" if mode == "edit" else "/v1/images/generations"
        summary_prefix = "图生图" if mode == "edit" else "文生图"
        detail = {
            "key_id": identity.get("id"),
            "key_name": identity.get("name"),
            "role": identity.get("role"),
            "endpoint": endpoint,
            "model": model,
            "started_at": china_from_timestamp_string(started),
            "ended_at": _now_iso(),
            "duration_ms": int((time.time() - started) * 1000),
            "status": status,
        }
        if request_preview:
            detail["request_text"] = request_preview
        if error:
            detail["error"] = error
        if account_email:
            detail["account_email"] = account_email
        if image_api:
            detail["image_api_name"] = " / ".join(
                item
                for item in [
                    _clean(image_api.get("group_name") or image_api.get("group_id")),
                    _clean(image_api.get("connection_name") or image_api.get("connection_id")),
                ]
                if item
            )
            detail["image_api_group"] = image_api.get("group_name") or image_api.get("group_id")
            detail["image_api_connection"] = image_api.get("connection_name") or image_api.get("connection_id")
            detail["image_api_model"] = image_api.get("model_id")
            detail["image_api_upstream_model"] = image_api.get("upstream_model")
            detail["image_api_resolution"] = image_api.get("resolution")
            if image_api.get("response_format"):
                detail["image_api_response_format"] = image_api.get("response_format")
            if image_api.get("size"):
                detail["image_api_size"] = image_api.get("size")
        if urls:
            detail["urls"] = list(dict.fromkeys(urls))
        try:
            log_service.add(LOG_TYPE_CALL, f"{summary_prefix}{suffix}", detail)
        except Exception:
            pass

    def _update_task(self, key: str, **updates: Any) -> None:
        with self._lock:
            task = self._tasks.get(key)
            if task is None:
                return
            task.update(updates)
            task["updated_at"] = _now_iso()
            task["updated_ts"] = time.time()
            self._save_locked()

    def _record_progress(self, key: str, step: str, *, set_started_ts: bool = False) -> None:
        cleaned_step = _clean(step) or "generating"
        now_ts = time.time()
        with self._lock:
            task = self._tasks.get(key)
            if task is None:
                return
            events = task.get("progress_events") if isinstance(task.get("progress_events"), list) else []
            normalized_events = [
                event for event in events
                if isinstance(event, dict) and _clean(event.get("step"))
            ]
            if not normalized_events or _clean(normalized_events[-1].get("step")) != cleaned_step:
                normalized_events.append({
                    "step": cleaned_step,
                    "at": _now_iso(),
                    "ts": now_ts,
                })
            task["progress_events"] = normalized_events[-20:]
            task["progress"] = cleaned_step
            if set_started_ts:
                task["started_ts"] = now_ts
            task["updated_at"] = _now_iso()
            task["updated_ts"] = now_ts
            self._save_locked()

    def _settle_credit(self, key: str, *, success: bool) -> None:
        with self._lock:
            task = self._tasks.get(key)
            if task is None or bool(task.get("credit_settled")):
                return
            user_id = _clean(task.get("credit_user_id"))
            amount = max(0, int(task.get("credit_reserved") or 0))
            daily_amount = max(0, int(task.get("credit_daily_reserved") or 0))
            paid_amount = max(0, int(task.get("credit_paid_reserved") or 0))
            package_reservations = task.get("credit_package_reservations") if isinstance(task.get("credit_package_reservations"), list) else []
            grant_reservations = task.get("credit_grant_reservations") if isinstance(task.get("credit_grant_reservations"), list) else []
            package_amount = sum(max(0, int(item.get("amount") or 0)) for item in package_reservations if isinstance(item, dict))
            grant_amount = sum(max(0, int(item.get("amount") or 0)) for item in grant_reservations if isinstance(item, dict))
            if amount > 0 and daily_amount + paid_amount + package_amount + grant_amount == 0:
                paid_amount = amount
            if not user_id or amount <= 0:
                task["credit_settled"] = True
                self._save_locked()
                return
            try:
                from services.auth_service import auth_service

                if success:
                    auth_service.consume_reserved_credits(
                        user_id,
                        amount,
                        daily_amount=daily_amount,
                        paid_amount=paid_amount,
                        package_reservations=package_reservations,
                        grant_reservations=grant_reservations,
                    )
                else:
                    auth_service.refund_reserved_credits(
                        user_id,
                        amount,
                        daily_amount=daily_amount,
                        paid_amount=paid_amount,
                        package_reservations=package_reservations,
                        grant_reservations=grant_reservations,
                    )
            finally:
                task["credit_settled"] = True
                task["updated_at"] = _now_iso()
                self._save_locked()

    def _settle_third_party_wallet(self, key: str, *, success_count: int) -> None:
        with self._lock:
            task = self._tasks.get(key)
            if task is None or bool(task.get("third_party_wallet_settled")):
                return
            user_id = _clean(task.get("third_party_wallet_user_id"))
            group_id = _clean(task.get("third_party_wallet_group_id"))
            model_id = _clean(task.get("third_party_wallet_model_id"))
            amount = max(0, int(task.get("third_party_wallet_reserved") or 0))
            cost_per_image = max(0, int(task.get("third_party_wallet_cost_per_image") or 0))
            if not user_id or not group_id or not model_id or amount <= 0:
                task["third_party_wallet_settled"] = True
                self._save_locked()
                return
            consume_amount = min(amount, max(0, int(success_count or 0)) * cost_per_image)
            refund_amount = amount - consume_amount
            try:
                from services.auth_service import auth_service

                if consume_amount > 0:
                    auth_service.consume_third_party_image_wallet(
                        user_id,
                        group_id=group_id,
                        model_id=model_id,
                        amount=consume_amount,
                    )
                if refund_amount > 0:
                    auth_service.refund_third_party_image_wallet(
                        user_id,
                        group_id=group_id,
                        model_id=model_id,
                        amount=refund_amount,
                    )
            finally:
                task["third_party_wallet_settled"] = True
                task["updated_at"] = _now_iso()
                self._save_locked()

    def _load_locked(self) -> dict[str, dict[str, Any]]:
        if not self.path.exists():
            return {}
        try:
            raw = json.loads(self.path.read_text(encoding="utf-8"))
        except Exception:
            return {}
        raw_items = raw.get("tasks") if isinstance(raw, dict) else raw
        if not isinstance(raw_items, list):
            return {}
        tasks: dict[str, dict[str, Any]] = {}
        for item in raw_items:
            if not isinstance(item, dict):
                continue
            task_id = _clean(item.get("id"))
            owner = _clean(item.get("owner_id"))
            if not task_id or not owner:
                continue
            status = _clean(item.get("status"))
            if status not in {TASK_STATUS_QUEUED, TASK_STATUS_RUNNING, TASK_STATUS_SUCCESS, TASK_STATUS_ERROR}:
                status = TASK_STATUS_ERROR
            task = {
                "id": task_id,
                "owner_id": owner,
                "status": status,
                "mode": "edit" if item.get("mode") == "edit" else "generate",
                "model": _clean(item.get("model"), "gpt-image-2"),
                "size": _clean(item.get("size")),
                "created_at": _clean(item.get("created_at"), _now_iso()),
                "started_at": _clean(item.get("started_at")),
                "updated_at": _clean(item.get("updated_at"), _clean(item.get("created_at"), _now_iso())),
                "created_ts": item.get("created_ts"),
                "updated_ts": item.get("updated_ts"),
                "started_ts": item.get("started_ts"),
                "duration_ms": item.get("duration_ms"),
            }
            progress = _clean(item.get("progress"))
            if progress:
                task["progress"] = progress
            if isinstance(item.get("progress_events"), list):
                task["progress_events"] = [
                    event
                    for event in item.get("progress_events", [])[-20:]
                    if isinstance(event, dict) and _clean(event.get("step"))
                ]
            credit_user_id = _clean(item.get("credit_user_id"))
            credit_reserved = max(0, int(item.get("credit_reserved") or 0))
            if credit_user_id and credit_reserved > 0:
                task["credit_user_id"] = credit_user_id
                task["credit_reserved"] = credit_reserved
                task["credit_daily_reserved"] = max(0, int(item.get("credit_daily_reserved") or 0))
                task["credit_paid_reserved"] = max(0, int(item.get("credit_paid_reserved") or 0))
                task["credit_package_reservations"] = item.get("credit_package_reservations") if isinstance(item.get("credit_package_reservations"), list) else []
                task["credit_grant_reservations"] = item.get("credit_grant_reservations") if isinstance(item.get("credit_grant_reservations"), list) else []
                task["credit_settled"] = bool(item.get("credit_settled"))
            wallet_user_id = _clean(item.get("third_party_wallet_user_id"))
            wallet_reserved = max(0, int(item.get("third_party_wallet_reserved") or 0))
            if wallet_user_id and wallet_reserved > 0:
                task["third_party_wallet_user_id"] = wallet_user_id
                task["third_party_wallet_group_id"] = _clean(item.get("third_party_wallet_group_id"))
                task["third_party_wallet_model_id"] = _clean(item.get("third_party_wallet_model_id"))
                task["third_party_wallet_reserved"] = wallet_reserved
                task["third_party_wallet_cost_per_image"] = max(0, int(item.get("third_party_wallet_cost_per_image") or 0))
                task["third_party_wallet_settled"] = bool(item.get("third_party_wallet_settled"))
            data = item.get("data")
            if isinstance(data, list):
                task["data"] = data
            error = _clean(item.get("error"))
            if error:
                task["error"] = error
            tasks[_task_key(owner, task_id)] = task
        return tasks

    def _save_locked(self) -> None:
        items = sorted(self._tasks.values(), key=lambda item: str(item.get("updated_at") or ""), reverse=True)
        tmp_path = self.path.with_suffix(self.path.suffix + ".tmp")
        tmp_path.write_text(json.dumps({"tasks": items}, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
        tmp_path.replace(self.path)

    def _recover_unfinished_locked(self) -> bool:
        changed = False
        for task in self._tasks.values():
            if task.get("status") in UNFINISHED_STATUSES:
                self._refund_task_credit_locked(task)
                self._refund_task_third_party_wallet_locked(task)
                task["status"] = TASK_STATUS_ERROR
                task["error"] = "服务已重启，未完成的图片任务已中断"
                task["updated_at"] = _now_iso()
                changed = True
        return changed

    def _refund_task_credit_locked(self, task: dict[str, Any]) -> None:
        if bool(task.get("credit_settled")):
            return
        user_id = _clean(task.get("credit_user_id"))
        amount = max(0, int(task.get("credit_reserved") or 0))
        daily_amount = max(0, int(task.get("credit_daily_reserved") or 0))
        paid_amount = max(0, int(task.get("credit_paid_reserved") or 0))
        package_reservations = task.get("credit_package_reservations") if isinstance(task.get("credit_package_reservations"), list) else []
        grant_reservations = task.get("credit_grant_reservations") if isinstance(task.get("credit_grant_reservations"), list) else []
        package_amount = sum(max(0, int(item.get("amount") or 0)) for item in package_reservations if isinstance(item, dict))
        grant_amount = sum(max(0, int(item.get("amount") or 0)) for item in grant_reservations if isinstance(item, dict))
        if amount > 0 and daily_amount + paid_amount + package_amount + grant_amount == 0:
            paid_amount = amount
        if not user_id or amount <= 0:
            task["credit_settled"] = True
            return
        try:
            from services.auth_service import auth_service

            auth_service.refund_reserved_credits(
                user_id,
                amount,
                daily_amount=daily_amount,
                paid_amount=paid_amount,
                package_reservations=package_reservations,
                grant_reservations=grant_reservations,
            )
        finally:
            task["credit_settled"] = True

    def _refund_task_third_party_wallet_locked(self, task: dict[str, Any]) -> None:
        if bool(task.get("third_party_wallet_settled")):
            return
        user_id = _clean(task.get("third_party_wallet_user_id"))
        group_id = _clean(task.get("third_party_wallet_group_id"))
        model_id = _clean(task.get("third_party_wallet_model_id"))
        amount = max(0, int(task.get("third_party_wallet_reserved") or 0))
        if not user_id or not group_id or not model_id or amount <= 0:
            task["third_party_wallet_settled"] = True
            return
        try:
            from services.auth_service import auth_service

            auth_service.refund_third_party_image_wallet(
                user_id,
                group_id=group_id,
                model_id=model_id,
                amount=amount,
            )
        finally:
            task["third_party_wallet_settled"] = True

    def _cleanup_locked(self) -> bool:
        try:
            retention_days = max(1, int(self.retention_days_getter()))
        except Exception:
            retention_days = 30
        cutoff = time.time() - retention_days * 86400
        removed_keys = [
            key
            for key, task in self._tasks.items()
            if task.get("status") in TERMINAL_STATUSES and _timestamp(task.get("updated_at")) < cutoff
        ]
        for key in removed_keys:
            self._tasks.pop(key, None)
        return bool(removed_keys)

    def _cleanup_if_due_locked(self) -> bool:
        now = time.monotonic()
        if now - self._last_cleanup_at < TASK_CLEANUP_INTERVAL_SECONDS:
            return False
        self._last_cleanup_at = now
        return self._cleanup_locked()


image_task_service = ImageTaskService(DATA_DIR / "image_tasks.json")