from __future__ import annotations

import logging
from datetime import datetime, timedelta, timezone
from typing import Any

from sqlalchemy import select

from app.config import settings
from app.enums.credit_record import CreditRecordBillingScene, CreditRecordOwnerType
from app.enums.shot_replicate import (
    ModuleCodeEnum,
    ShotAnalysisStatusEnum,
    ShotSegmentAnalysisStatusEnum,
    ShotSegmentSourceModeEnum,
    ShotSplitStatusEnum,
    ShotTaskSetStatusEnum,
)
from app.models.base import async_session
from app.models.shot_replicate_segment import ShotReplicateSegment
from app.models.shot_replicate_task_set import ShotReplicateTaskSet
from app.services.module_generation_log_service import log_module_error, log_module_event_file, log_module_prompt_event
from app.services.redis_registry_service import get_registry_redis, redis_acquire_lock, redis_release_lock
from app.services.module_async_recovery_service import (
    OBJECT_SHOT_SEGMENT_ANALYSIS,
    OBJECT_SHOT_SPLIT_SEGMENT,
    OBJECT_SHOT_TASK_SET_ANALYSIS,
    acquire_object_lock,
    cleanup_active_if_terminal,
    mark_active_started,
    postpone_active_task,
    register_shot_segment_analysis_task,
    register_shot_split_task,
    register_shot_task_set_analysis_task,
    release_object_lock,
    remove_active_task,
)
from app.services.shot_replicate_taskset_service import refresh_task_set_split_summary
from app.services.shot_video_analysis_service import analyze_video_for_shot_split
from app.services.generation_billing_service import charge_shot_video_analysis_usage
from app.services.shot_video_split_service import split_video_segment_async
from app.services.upload_video_asset_service import validate_split_range
from app.tasks.async_runner import run_async
from app.tasks.celery_app import celery_app

logger = logging.getLogger("video_gen")

MODULE = ModuleCodeEnum.SHOT_REPLICATE.value
SPLIT_QUEUE = "gen_result_download"
ANALYSIS_QUEUE = "gen_chatapi_create"


def _now() -> datetime:
    return datetime.now(timezone.utc)


def _lease_until(now: datetime | None = None) -> datetime:
    return (now or _now()) + timedelta(seconds=int(settings.SHOT_SPLIT_LEASE_SECONDS or 600))


def _retry_at(attempt: int, now: datetime | None = None) -> datetime:
    base = int(settings.SHOT_SPLIT_RETRY_BACKOFF_SECONDS or settings.DOWNLOAD_TASK_RETRY_BACKOFF_SECONDS or 30)
    return (now or _now()) + timedelta(seconds=max(1, base * max(1, attempt)))


async def _acquire_split_semaphore(segment_id: str) -> str | None:
    """简单 Redis 并发闸门：用固定槽位锁限制 ffmpeg 同时运行数量。"""
    max_concurrent = max(1, int(settings.SHOT_SPLIT_MAX_CONCURRENT or 1))
    ttl = int(settings.SHOT_SPLIT_LEASE_SECONDS or 600)
    for slot in range(max_concurrent):
        key = f"{settings.SHOT_SPLIT_SEMAPHORE_KEY_PREFIX}:{slot}"
        token = await redis_acquire_lock(lock_key=key, ttl_seconds=ttl, token=segment_id, log_context="shot_split_semaphore")
        if token:
            return key
    return None


async def _release_split_semaphore(lock_key: str | None, segment_id: str) -> None:
    if lock_key:
        await redis_release_lock(lock_key=lock_key, token=segment_id, log_context="shot_split_semaphore")


async def _run_analyze_original_video(task_set_id: str) -> None:
    lock_token = await acquire_object_lock(object_type=OBJECT_SHOT_TASK_SET_ANALYSIS, object_id=task_set_id)
    if not lock_token:
        return
    await register_shot_task_set_analysis_task(task_set_id)
    await mark_active_started(object_type=OBJECT_SHOT_TASK_SET_ANALYSIS, object_id=task_set_id)

    task_set_user_id: str | None = None
    video_url: str | None = None
    try:
        async with async_session() as db:
            result = await db.execute(
                select(ShotReplicateTaskSet)
                .where(ShotReplicateTaskSet.id == task_set_id, ShotReplicateTaskSet.deleted_at.is_(None))
                .with_for_update()
                .limit(1)
            )
            task_set = result.scalar_one_or_none()
            if not task_set:
                await remove_active_task(object_type=OBJECT_SHOT_TASK_SET_ANALYSIS, object_id=task_set_id)
                return
            if task_set.analysis_status == ShotAnalysisStatusEnum.COMPLETED.value:
                await remove_active_task(object_type=OBJECT_SHOT_TASK_SET_ANALYSIS, object_id=task_set_id)
                return
            task_set_user_id = task_set.user_id
            video_url = task_set.video_url
            task_set.status = ShotTaskSetStatusEnum.ANALYZING.value
            task_set.analysis_status = ShotAnalysisStatusEnum.PROCESSING.value
            task_set.analysis_error_message = None
            await db.commit()

        log_module_event_file(
            module=MODULE,
            event_type="SHOT_ANALYSIS_STARTED",
            project_id=task_set_id,
            user_id=task_set_user_id,
            message="原视频拆镜分析开始",
            detail={"task_set_id": task_set_id, "video_url": video_url, "analysis_mode": "full_breakdown"},
        )

        async with async_session() as db:
            analyzed = await analyze_video_for_shot_split(db, video_url or "", user_id=task_set_user_id, mode="full_breakdown")
            result = await db.execute(
                select(ShotReplicateTaskSet)
                .where(ShotReplicateTaskSet.id == task_set_id, ShotReplicateTaskSet.deleted_at.is_(None))
                .with_for_update()
                .limit(1)
            )
            task_set = result.scalar_one_or_none()
            if not task_set:
                await db.rollback()
                await remove_active_task(object_type=OBJECT_SHOT_TASK_SET_ANALYSIS, object_id=task_set_id)
                return
            result_json = analyzed.result
            task_set.original_video_content = str(result_json.get("原视频内容") or "无")
            task_set.original_video_category = str(result_json.get("原视频分类") or "无")
            task_set.original_video_audience = str(result_json.get("原视频受众人群") or "无")
            task_set.ai_suggestion_json = result_json.get("拆镜内容剖析") or []
            task_set.analysis_raw_json = analyzed.raw_response
            task_set.analysis_result_json = result_json
            task_set.analysis_status = ShotAnalysisStatusEnum.COMPLETED.value
            task_set.status = ShotTaskSetStatusEnum.ANALYSIS_COMPLETED.value
            task_set.analysis_error_message = None
            await charge_shot_video_analysis_usage(
                db,
                user_id=task_set.user_id,
                owner_type=CreditRecordOwnerType.SHOT_REPLICATE_TASK_SET.value,
                owner_id=task_set.id,
                usage=analyzed.usage,
                description="拆镜复刻-原视频分析",
                billing_scene=CreditRecordBillingScene.SHOT_ORIGINAL_VIDEO_ANALYSIS.value,
                source_project_id=task_set.id,
            )
            await db.commit()
            await cleanup_active_if_terminal(db, object_type=OBJECT_SHOT_TASK_SET_ANALYSIS, object_id=task_set_id)

        log_module_prompt_event(
            event_type="SHOT_ANALYSIS_SUCCESS",
            project_id=task_set_id,
            step_id=task_set_id,
            user_id=task_set_user_id or "",
            module=MODULE,
            prompt_type="shot_video_analysis",
            request=analyzed.usage.get("log_request") if isinstance(analyzed.usage, dict) else {},
            response=analyzed.result,
            token_usage=analyzed.usage,
        )
        log_module_event_file(
            module=MODULE,
            event_type="SHOT_ANALYSIS_SUCCESS",
            project_id=task_set_id,
            user_id=task_set_user_id,
            message="原视频拆镜分析成功",
            detail={"suggestion_count": len(analyzed.result.get("拆镜内容剖析") or []), "token_usage": analyzed.usage},
        )
    except Exception as exc:
        async with async_session() as db:
            result = await db.execute(
                select(ShotReplicateTaskSet)
                .where(ShotReplicateTaskSet.id == task_set_id, ShotReplicateTaskSet.deleted_at.is_(None))
                .with_for_update()
                .limit(1)
            )
            task_set = result.scalar_one_or_none()
            if task_set:
                task_set_user_id = task_set_user_id or task_set.user_id
                task_set.status = ShotTaskSetStatusEnum.ANALYSIS_FAILED.value
                task_set.analysis_status = ShotAnalysisStatusEnum.FAILED.value
                task_set.analysis_error_message = str(exc)
                await db.commit()
                await cleanup_active_if_terminal(db, object_type=OBJECT_SHOT_TASK_SET_ANALYSIS, object_id=task_set_id)
            else:
                await remove_active_task(object_type=OBJECT_SHOT_TASK_SET_ANALYSIS, object_id=task_set_id)
        log_module_error(
            module=MODULE,
            event_type="SHOT_ANALYSIS_FAILED",
            project_id=task_set_id,
            user_id=task_set_user_id,
            message="原视频拆镜分析失败",
            detail={"task_set_id": task_set_id, "video_url": video_url, "analysis_mode": "full_breakdown"},
            exc=exc,
        )
    finally:
        await release_object_lock(object_type=OBJECT_SHOT_TASK_SET_ANALYSIS, object_id=task_set_id, token=lock_token)


async def _run_analyze_custom_segment_video(segment_id: str) -> None:
    lock_token = await acquire_object_lock(object_type=OBJECT_SHOT_SEGMENT_ANALYSIS, object_id=segment_id)
    if not lock_token:
        return
    await register_shot_segment_analysis_task(segment_id)
    await mark_active_started(object_type=OBJECT_SHOT_SEGMENT_ANALYSIS, object_id=segment_id)

    user_id: str | None = None
    task_set_id: str | None = None
    video_url: str | None = None
    try:
        async with async_session() as db:
            result = await db.execute(
                select(ShotReplicateSegment)
                .where(ShotReplicateSegment.id == segment_id, ShotReplicateSegment.deleted_at.is_(None))
                .with_for_update()
                .limit(1)
            )
            segment = result.scalar_one_or_none()
            if not segment or not segment.segment_video_url:
                await remove_active_task(object_type=OBJECT_SHOT_SEGMENT_ANALYSIS, object_id=segment_id)
                return
            if segment.analysis_status == ShotSegmentAnalysisStatusEnum.COMPLETED.value:
                await remove_active_task(object_type=OBJECT_SHOT_SEGMENT_ANALYSIS, object_id=segment_id)
                return
            user_id = segment.user_id
            task_set_id = segment.task_set_id
            video_url = segment.segment_video_url
            await register_shot_segment_analysis_task(segment_id, task_set_id=task_set_id)
            segment.analysis_status = ShotSegmentAnalysisStatusEnum.PROCESSING.value
            segment.analysis_error_message = None
            await db.commit()

        log_module_event_file(
            module=MODULE,
            event_type="SHOT_SEGMENT_ANALYSIS_STARTED",
            project_id=task_set_id,
            step_id=segment_id,
            user_id=user_id,
            message="自定义拆镜片段分析开始",
            detail={"segment_id": segment_id, "task_set_id": task_set_id, "video_url": video_url, "analysis_mode": "summary_only"},
        )

        async with async_session() as db:
            analyzed = await analyze_video_for_shot_split(db, video_url or "", user_id=user_id, mode="summary_only")
            result = await db.execute(
                select(ShotReplicateSegment)
                .where(ShotReplicateSegment.id == segment_id, ShotReplicateSegment.deleted_at.is_(None))
                .with_for_update()
                .limit(1)
            )
            segment = result.scalar_one_or_none()
            if not segment:
                await db.rollback()
                await remove_active_task(object_type=OBJECT_SHOT_SEGMENT_ANALYSIS, object_id=segment_id)
                return
            result_json = analyzed.result
            segment.original_video_content = str(result_json.get("原视频内容") or "无")
            segment.original_video_category = str(result_json.get("原视频分类") or "无")
            segment.original_video_audience = str(result_json.get("原视频受众人群") or "无")
            segment.segment_content = segment.original_video_content
            segment.segment_category = segment.original_video_category
            segment.segment_audience = segment.original_video_audience
            segment.analysis_json = result_json
            segment.analysis_status = ShotSegmentAnalysisStatusEnum.COMPLETED.value
            segment.analysis_error_message = None
            await charge_shot_video_analysis_usage(
                db,
                user_id=segment.user_id,
                owner_type=CreditRecordOwnerType.SHOT_REPLICATE_SEGMENT.value,
                owner_id=segment.id,
                usage=analyzed.usage,
                description="拆镜复刻-片段视频分析",
                billing_scene=CreditRecordBillingScene.SHOT_SEGMENT_VIDEO_ANALYSIS.value,
                source_project_id=segment.task_set_id,
                source_step_id=segment.id,
            )
            await db.commit()
            await cleanup_active_if_terminal(db, object_type=OBJECT_SHOT_SEGMENT_ANALYSIS, object_id=segment_id)

        log_module_prompt_event(
            event_type="SHOT_SEGMENT_ANALYSIS_SUCCESS",
            project_id=task_set_id,
            step_id=segment_id,
            user_id=user_id or "",
            module=MODULE,
            prompt_type="shot_segment_analysis",
            request=analyzed.usage.get("log_request") if isinstance(analyzed.usage, dict) else {},
            response=analyzed.result,
            token_usage=analyzed.usage,
        )
        log_module_event_file(
            module=MODULE,
            event_type="SHOT_SEGMENT_ANALYSIS_SUCCESS",
            project_id=task_set_id,
            step_id=segment_id,
            user_id=user_id,
            message="自定义拆镜片段分析成功",
            detail={"segment_id": segment_id, "task_set_id": task_set_id, "token_usage": analyzed.usage},
        )
    except Exception as exc:
        async with async_session() as db:
            result = await db.execute(
                select(ShotReplicateSegment)
                .where(ShotReplicateSegment.id == segment_id, ShotReplicateSegment.deleted_at.is_(None))
                .with_for_update()
                .limit(1)
            )
            segment = result.scalar_one_or_none()
            if segment:
                user_id = user_id or segment.user_id
                task_set_id = task_set_id or segment.task_set_id
                segment.analysis_status = ShotSegmentAnalysisStatusEnum.FAILED.value
                segment.analysis_error_message = str(exc)
                await db.commit()
                await cleanup_active_if_terminal(db, object_type=OBJECT_SHOT_SEGMENT_ANALYSIS, object_id=segment_id)
            else:
                await remove_active_task(object_type=OBJECT_SHOT_SEGMENT_ANALYSIS, object_id=segment_id)
        log_module_error(
            module=MODULE,
            event_type="SHOT_SEGMENT_ANALYSIS_FAILED",
            project_id=task_set_id,
            step_id=segment_id,
            user_id=user_id,
            message="自定义拆镜片段分析失败",
            detail={"segment_id": segment_id, "task_set_id": task_set_id, "video_url": video_url, "analysis_mode": "summary_only"},
            exc=exc,
        )
    finally:
        await release_object_lock(object_type=OBJECT_SHOT_SEGMENT_ANALYSIS, object_id=segment_id, token=lock_token)


async def _run_split_one_segment(segment_id: str) -> None:
    segment_lock_key = f"{settings.SHOT_SPLIT_LOCK_KEY_PREFIX}:{segment_id}"
    segment_lock_token = await redis_acquire_lock(
        lock_key=segment_lock_key,
        ttl_seconds=int(settings.SHOT_SPLIT_LEASE_SECONDS or 600),
        log_context="shot_split_segment_lock",
    )
    if not segment_lock_token:
        return

    await register_shot_split_task(segment_id)
    await mark_active_started(object_type=OBJECT_SHOT_SPLIT_SEGMENT, object_id=segment_id)

    semaphore_key: str | None = None
    user_id: str | None = None
    task_set_id: str | None = None
    source_path: str | None = None
    try:
        semaphore_key = await _acquire_split_semaphore(segment_id)
        if not semaphore_key:
            log_module_event_file(
                module=MODULE,
                event_type="SHOT_SEGMENT_SPLIT_RETRY_WAITING",
                step_id=segment_id,
                message="拆镜 ffmpeg 并发闸门已满，稍后重试",
                detail={"segment_id": segment_id, "reason": "semaphore_full"},
            )
            delay = max(1, int(settings.MODULE_ASYNC_REQUEUE_DELAY_SECONDS or 10))
            await postpone_active_task(
                object_type=OBJECT_SHOT_SPLIT_SEGMENT,
                object_id=segment_id,
                delay_seconds=delay,
                reason="semaphore_full",
            )
            if celery_app:
                split_one_segment.apply_async(args=[segment_id], queue=SPLIT_QUEUE, countdown=delay, priority=settings.DOWNLOAD_TASK_PRIORITY_NORMAL)
            return

        async with async_session() as db:
            result = await db.execute(
                select(ShotReplicateSegment)
                .where(ShotReplicateSegment.id == segment_id, ShotReplicateSegment.deleted_at.is_(None))
                .with_for_update()
                .limit(1)
            )
            segment = result.scalar_one_or_none()
            if not segment:
                await remove_active_task(object_type=OBJECT_SHOT_SPLIT_SEGMENT, object_id=segment_id)
                return
            user_id = segment.user_id
            task_set_id = segment.task_set_id
            task_set_result = await db.execute(
                select(ShotReplicateTaskSet)
                .where(ShotReplicateTaskSet.id == segment.task_set_id, ShotReplicateTaskSet.deleted_at.is_(None))
                .with_for_update()
                .limit(1)
            )
            task_set = task_set_result.scalar_one_or_none()
            if not task_set:
                await remove_active_task(object_type=OBJECT_SHOT_SPLIT_SEGMENT, object_id=segment_id)
                return
            if segment.split_status == ShotSplitStatusEnum.COMPLETED.value and segment.segment_video_url:
                await remove_active_task(object_type=OBJECT_SHOT_SPLIT_SEGMENT, object_id=segment_id)
                return

            validate_split_range(
                start_second=segment.start_second,
                end_second=segment.end_second,
                video_duration_seconds=task_set.video_duration_seconds,
            )

            now = _now()
            segment.split_status = ShotSplitStatusEnum.PROCESSING.value
            segment.split_started_at = now
            segment.split_lease_until = _lease_until(now)
            segment.split_retry_count = int(segment.split_retry_count or 0) + 1
            segment.split_next_retry_at = None
            segment.split_last_error = None
            task_set.status = ShotTaskSetStatusEnum.SPLITTING.value
            task_set.split_status = ShotSplitStatusEnum.PROCESSING.value
            await db.commit()
            await register_shot_split_task(segment_id, task_set_id=segment.task_set_id)
            await mark_active_started(object_type=OBJECT_SHOT_SPLIT_SEGMENT, object_id=segment_id)

            source_path = task_set.video_path
            date_dir = (segment.created_at or now).strftime("%Y/%m/%d")
            start_second = segment.start_second
            end_second = segment.end_second
            attempt = segment.split_retry_count

        log_module_event_file(
            module=MODULE,
            event_type="SHOT_SEGMENT_SPLIT_STARTED",
            project_id=task_set_id,
            step_id=segment_id,
            user_id=user_id,
            message="拆镜片段 ffmpeg 切割开始",
            detail={
                "segment_id": segment_id,
                "task_set_id": task_set_id,
                "source_path": source_path,
                "start_second": start_second,
                "end_second": end_second,
                "attempt": attempt,
            },
        )

        split_result = await split_video_segment_async(
            source_path=source_path,
            segment_id=segment_id,
            start_second=start_second,
            end_second=end_second,
            date_dir=date_dir,
        )

        async with async_session() as db:
            result = await db.execute(
                select(ShotReplicateSegment)
                .where(ShotReplicateSegment.id == segment_id, ShotReplicateSegment.deleted_at.is_(None))
                .with_for_update()
                .limit(1)
            )
            segment = result.scalar_one_or_none()
            if not segment:
                await remove_active_task(object_type=OBJECT_SHOT_SPLIT_SEGMENT, object_id=segment_id)
                return
            segment.segment_video_url = split_result.url
            segment.segment_video_path = split_result.path
            segment.split_status = ShotSplitStatusEnum.COMPLETED.value
            segment.split_completed_at = _now()
            segment.split_lease_until = None
            segment.split_next_retry_at = None
            segment.split_last_error = None
            await refresh_task_set_split_summary(db, segment.task_set_id)
            await db.commit()
            await cleanup_active_if_terminal(db, object_type=OBJECT_SHOT_SPLIT_SEGMENT, object_id=segment_id)

            log_module_event_file(
                module=MODULE,
                event_type="SHOT_SEGMENT_SPLIT_SUCCESS",
                project_id=segment.task_set_id,
                step_id=segment.id,
                user_id=segment.user_id,
                message="拆镜片段 ffmpeg 切割成功",
                detail={
                    "segment_id": segment.id,
                    "task_set_id": segment.task_set_id,
                    "segment_video_url": split_result.url,
                    "segment_video_path": split_result.path,
                    "source_mode": segment.source_mode,
                },
            )

            if segment.source_mode == ShotSegmentSourceModeEnum.CUSTOM.value and celery_app:
                await register_shot_segment_analysis_task(segment.id, task_set_id=segment.task_set_id)
                analyze_custom_segment_video.apply_async(args=[segment.id], queue=ANALYSIS_QUEUE, countdown=0)

    except Exception as exc:
        next_retry_delay: int | None = None
        final_failed = False
        async with async_session() as db:
            result = await db.execute(
                select(ShotReplicateSegment)
                .where(ShotReplicateSegment.id == segment_id, ShotReplicateSegment.deleted_at.is_(None))
                .with_for_update()
                .limit(1)
            )
            segment = result.scalar_one_or_none()
            if not segment:
                await remove_active_task(object_type=OBJECT_SHOT_SPLIT_SEGMENT, object_id=segment_id)
                return
            user_id = user_id or segment.user_id
            task_set_id = task_set_id or segment.task_set_id
            attempt = int(segment.split_retry_count or 0)
            segment.split_last_error = str(exc)
            segment.split_lease_until = None
            if attempt >= int(settings.SHOT_SPLIT_MAX_RETRY_COUNT or 3):
                segment.split_status = ShotSplitStatusEnum.FAILED.value
                segment.split_next_retry_at = None
                final_failed = True
            else:
                segment.split_status = ShotSplitStatusEnum.RETRY_WAITING.value
                segment.split_next_retry_at = _retry_at(attempt)
            await refresh_task_set_split_summary(db, segment.task_set_id)
            await db.commit()

            if segment.split_status == ShotSplitStatusEnum.RETRY_WAITING.value:
                next_retry_delay = max(1, int(((segment.split_next_retry_at or _now()) - _now()).total_seconds()))
                await postpone_active_task(
                    object_type=OBJECT_SHOT_SPLIT_SEGMENT,
                    object_id=segment_id,
                    delay_seconds=next_retry_delay,
                    reason="split_retry_waiting",
                )
                if celery_app:
                    split_one_segment.apply_async(args=[segment_id], queue=SPLIT_QUEUE, countdown=next_retry_delay, priority=settings.DOWNLOAD_TASK_PRIORITY_RECOVER)
            elif final_failed:
                await cleanup_active_if_terminal(db, object_type=OBJECT_SHOT_SPLIT_SEGMENT, object_id=segment_id)

        log_module_error(
            module=MODULE,
            event_type="SHOT_SEGMENT_SPLIT_FAILED" if final_failed else "SHOT_SEGMENT_SPLIT_RETRY_WAITING",
            project_id=task_set_id,
            step_id=segment_id,
            user_id=user_id,
            message="拆镜片段 ffmpeg 切割失败" if final_failed else "拆镜片段 ffmpeg 切割失败，等待重试",
            detail={
                "segment_id": segment_id,
                "task_set_id": task_set_id,
                "source_path": source_path,
                "next_retry_delay_seconds": next_retry_delay,
                "final_failed": final_failed,
            },
            exc=exc,
        )
    finally:
        await _release_split_semaphore(semaphore_key, segment_id)
        await redis_release_lock(lock_key=segment_lock_key, token=segment_lock_token, log_context="shot_split_segment_lock")


async def _run_recover_split_tasks_once() -> dict[str, Any]:
    from app.services.shot_replicate_recovery_service import recover_shot_split_tasks_once

    redis = await get_registry_redis()
    token: str | None = None
    if redis is not None:
        token = await redis_acquire_lock(
            lock_key=settings.SHOT_SPLIT_RECOVERY_LOCK_KEY,
            ttl_seconds=int(settings.CELERY_RECOVERY_TASK_LOCK_TTL_SECONDS or 600),
            log_context="shot_split_recovery",
        )
        if not token:
            return {"skipped": "lock_held", "lock_key": settings.SHOT_SPLIT_RECOVERY_LOCK_KEY}

    try:
        async with async_session() as db:
            result = await recover_shot_split_tasks_once(db)
            result["execution_lock"] = "lock_acquired" if token else "redis_unavailable_run_db_fallback"
            return result
    finally:
        if token:
            await redis_release_lock(
                lock_key=settings.SHOT_SPLIT_RECOVERY_LOCK_KEY,
                token=token,
                log_context="shot_split_recovery",
            )


if celery_app:

    @celery_app.task(name="shot_replicate.analyze_original_video")
    def analyze_original_video(task_set_id: str) -> None:
        return run_async(_run_analyze_original_video(task_set_id))


    @celery_app.task(name="shot_replicate.split_one_segment", bind=True, max_retries=0)
    def split_one_segment(self, segment_id: str) -> None:
        return run_async(_run_split_one_segment(segment_id))


    @celery_app.task(name="shot_replicate.analyze_custom_segment_video")
    def analyze_custom_segment_video(segment_id: str) -> None:
        return run_async(_run_analyze_custom_segment_video(segment_id))


    @celery_app.task(
        name="shot_replicate.recover_split_tasks_once",
        bind=True,
        soft_time_limit=settings.CELERY_RECOVERY_SOFT_TIME_LIMIT_SECONDS,
        time_limit=settings.CELERY_RECOVERY_TIME_LIMIT_SECONDS,
    )
    def recover_split_tasks_once(self) -> dict[str, Any]:
        return run_async(_run_recover_split_tasks_once())

else:

    class _DisabledTask:
        def delay(self, *args: Any, **kwargs: Any) -> None:
            raise RuntimeError("Celery is disabled")

        def apply_async(self, *args: Any, **kwargs: Any) -> None:
            raise RuntimeError("Celery is disabled")

    analyze_original_video = _DisabledTask()
    split_one_segment = _DisabledTask()
    analyze_custom_segment_video = _DisabledTask()
    recover_split_tasks_once = _DisabledTask()
