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, ShotReplicateLogEventEnum, 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=ShotReplicateLogEventEnum.ANALYSIS_STARTED.value, 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", task_set_id=task_set_id, trace_id=f"shot-task-set-analysis:{task_set_id}") 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=ShotReplicateLogEventEnum.ANALYSIS_SUCCESS.value, 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=ShotReplicateLogEventEnum.ANALYSIS_SUCCESS.value, 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=ShotReplicateLogEventEnum.ANALYSIS_FAILED.value, 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=ShotReplicateLogEventEnum.SEGMENT_ANALYSIS_STARTED.value, 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", task_set_id=task_set_id, segment_id=segment_id, trace_id=f"shot-segment-analysis:{segment_id}") 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=ShotReplicateLogEventEnum.SEGMENT_ANALYSIS_SUCCESS.value, 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=ShotReplicateLogEventEnum.SEGMENT_ANALYSIS_SUCCESS.value, 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=ShotReplicateLogEventEnum.SEGMENT_ANALYSIS_FAILED.value, 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()