diff --git a/video-gen-api/alembic/versions/20da1d353914_repair_llm_billing_optimize_idempotency.py b/video-gen-api/alembic/versions/20da1d353914_repair_llm_billing_optimize_idempotency.py new file mode 100644 index 00000000..99a667ba --- /dev/null +++ b/video-gen-api/alembic/versions/20da1d353914_repair_llm_billing_optimize_idempotency.py @@ -0,0 +1,364 @@ +"""repair llm billing optimize idempotency + +Revision ID: 20da1d353914 +Revises: 6a3ea8d0b4c8 +Create Date: 2026-07-24 13:35:48.513710 +""" +from __future__ import annotations + +import re +from typing import Any, Sequence, Union + +from alembic import op +import sqlalchemy as sa +from sqlalchemy.engine import Connection + + +# revision identifiers, used by Alembic. +revision: str = "20da1d353914" +down_revision: Union[str, None] = "6a3ea8d0b4c8" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +_PROMPT_USAGE_COLUMN = "prompt_usage_snapshot_json" + +_INDEX_SPECS: tuple[dict[str, Any], ...] = ( + { + "name": "uq_credit_records_user_refund_target", + "table": "credit_records", + "columns": ("user_id", "refund_for_biz_key"), + "where_sql": "type = 'refund' AND refund_for_biz_key IS NOT NULL", + "definition_fragments": ( + "(user_id, refund_for_biz_key)", + "type", + "refund", + "refund_for_biz_key is not null", + ), + "duplicate_sql": """ + SELECT + user_id, + refund_for_biz_key, + COUNT(*) AS duplicate_count + FROM credit_records + WHERE type = 'refund' + AND refund_for_biz_key IS NOT NULL + GROUP BY user_id, refund_for_biz_key + HAVING COUNT(*) > 1 + ORDER BY duplicate_count DESC, user_id, refund_for_biz_key + LIMIT 20 + """, + }, + { + "name": "uq_genrec_user_idempotency_active", + "table": "generation_records", + "columns": ("user_id", "idempotency_key"), + "where_sql": "idempotency_key IS NOT NULL AND deleted_at IS NULL", + "definition_fragments": ( + "(user_id, idempotency_key)", + "idempotency_key is not null", + "deleted_at is null", + ), + "duplicate_sql": """ + SELECT + user_id, + idempotency_key, + COUNT(*) AS duplicate_count + FROM generation_records + WHERE idempotency_key IS NOT NULL + AND deleted_at IS NULL + GROUP BY user_id, idempotency_key + HAVING COUNT(*) > 1 + ORDER BY duplicate_count DESC, user_id, idempotency_key + LIMIT 20 + """, + }, + { + "name": "uq_token_usage_user_biz_key", + "table": "token_usage", + "columns": ("user_id", "biz_key"), + "where_sql": "biz_key IS NOT NULL", + "definition_fragments": ( + "(user_id, biz_key)", + "biz_key is not null", + ), + "duplicate_sql": """ + SELECT + user_id, + biz_key, + COUNT(*) AS duplicate_count + FROM token_usage + WHERE user_id IS NOT NULL + AND biz_key IS NOT NULL + GROUP BY user_id, biz_key + HAVING COUNT(*) > 1 + ORDER BY duplicate_count DESC, user_id, biz_key + LIMIT 20 + """, + }, +) + + +def _bind() -> Connection: + bind = op.get_bind() + if bind.dialect.name != "postgresql": + raise RuntimeError( + "Migration 20da1d353914 requires PostgreSQL; " + f"current dialect is {bind.dialect.name!r}." + ) + return bind + + +def _current_schema(bind: Connection) -> str: + schema = bind.execute(sa.text("SELECT current_schema()")).scalar_one_or_none() + if not schema: + raise RuntimeError("Unable to resolve PostgreSQL current_schema().") + return str(schema) + + +def _require_table(bind: Connection, schema: str, table_name: str) -> None: + exists = bind.execute( + sa.text( + """ + SELECT EXISTS ( + SELECT 1 + FROM information_schema.tables + WHERE table_schema = :schema + AND table_name = :table_name + ) + """ + ), + {"schema": schema, "table_name": table_name}, + ).scalar_one() + if not bool(exists): + raise RuntimeError( + f"Required table {schema}.{table_name} does not exist; " + "refusing to apply migration on an unexpected schema baseline." + ) + + +def _column_exists( + bind: Connection, + schema: str, + table_name: str, + column_name: str, +) -> bool: + return bool( + bind.execute( + sa.text( + """ + SELECT EXISTS ( + SELECT 1 + FROM information_schema.columns + WHERE table_schema = :schema + AND table_name = :table_name + AND column_name = :column_name + ) + """ + ), + { + "schema": schema, + "table_name": table_name, + "column_name": column_name, + }, + ).scalar_one() + ) + + +def _index_info( + bind: Connection, + schema: str, + index_name: str, +) -> dict[str, Any] | None: + row = bind.execute( + sa.text( + """ + SELECT + table_rel.relname AS table_name, + index_meta.indisvalid AS is_valid, + index_meta.indisunique AS is_unique, + pg_get_indexdef(index_rel.oid) AS index_definition + FROM pg_class AS index_rel + JOIN pg_namespace AS namespace_rel + ON namespace_rel.oid = index_rel.relnamespace + JOIN pg_index AS index_meta + ON index_meta.indexrelid = index_rel.oid + JOIN pg_class AS table_rel + ON table_rel.oid = index_meta.indrelid + WHERE namespace_rel.nspname = :schema + AND index_rel.relname = :index_name + """ + ), + {"schema": schema, "index_name": index_name}, + ).mappings().one_or_none() + return dict(row) if row is not None else None + + +def _normalize_index_definition(value: str) -> str: + normalized = value.lower().replace('"', "") + normalized = re.sub(r"::[a-z_ ]+(?:\[\])?", "", normalized) + normalized = re.sub(r"[()]", lambda match: match.group(0), normalized) + normalized = re.sub(r"\s+", " ", normalized) + return normalized.strip() + + +def _assert_existing_index_matches(spec: dict[str, Any], info: dict[str, Any]) -> None: + if str(info["table_name"]) != str(spec["table"]): + raise RuntimeError( + f"Index {spec['name']} already exists on table {info['table_name']}, " + f"expected table {spec['table']}." + ) + if not bool(info["is_unique"]): + raise RuntimeError( + f"Index {spec['name']} already exists but is not UNIQUE." + ) + + definition = _normalize_index_definition(str(info["index_definition"] or "")) + missing = [ + fragment + for fragment in spec["definition_fragments"] + if fragment not in definition + ] + if missing: + raise RuntimeError( + f"Index {spec['name']} already exists with an unexpected definition; " + f"missing expected fragments: {missing}. Actual definition: " + f"{info['index_definition']}" + ) + + +def _assert_no_duplicates(bind: Connection, spec: dict[str, Any]) -> None: + rows = bind.execute(sa.text(spec["duplicate_sql"])).mappings().all() + if not rows: + return + + samples = "; ".join( + ", ".join(f"{key}={value!r}" for key, value in row.items()) + for row in rows + ) + raise RuntimeError( + f"Cannot create unique index {spec['name']}: duplicate historical data " + f"exists. Resolve the conflicts first. Samples: {samples}" + ) + + +def _quote_identifier(bind: Connection, value: str) -> str: + return bind.dialect.identifier_preparer.quote(value) + + +def _create_missing_indexes_concurrently( + bind: Connection, + schema: str, +) -> None: + missing_specs: list[dict[str, Any]] = [] + + for spec in _INDEX_SPECS: + _require_table(bind, schema, str(spec["table"])) + info = _index_info(bind, schema, str(spec["name"])) + if info is None: + _assert_no_duplicates(bind, spec) + missing_specs.append(spec) + continue + + if bool(info["is_valid"]): + _assert_existing_index_matches(spec, info) + continue + + # A failed CREATE INDEX CONCURRENTLY may leave an invalid index behind. + # Remove only the invalid index with this migration-owned name, then retry. + missing_specs.append(spec) + + if not missing_specs: + return + + context = op.get_context() + with context.autocommit_block(): + for spec in missing_specs: + current = _index_info(bind, schema, str(spec["name"])) + quoted_index = _quote_identifier(bind, str(spec["name"])) + quoted_schema = _quote_identifier(bind, schema) + if current is not None: + if bool(current["is_valid"]): + _assert_existing_index_matches(spec, current) + continue + op.execute( + sa.text( + f"DROP INDEX CONCURRENTLY IF EXISTS " + f"{quoted_schema}.{quoted_index}" + ) + ) + + quoted_table = _quote_identifier(bind, str(spec["table"])) + quoted_columns = ", ".join( + _quote_identifier(bind, str(column)) + for column in spec["columns"] + ) + op.execute( + sa.text( + f"CREATE UNIQUE INDEX CONCURRENTLY {quoted_index} " + f"ON {quoted_schema}.{quoted_table} ({quoted_columns}) " + f"WHERE {spec['where_sql']}" + ) + ) + + +def _drop_indexes_concurrently(bind: Connection, schema: str) -> None: + existing_names = [ + str(spec["name"]) + for spec in _INDEX_SPECS + if _index_info(bind, schema, str(spec["name"])) is not None + ] + if not existing_names: + return + + quoted_schema = _quote_identifier(bind, schema) + context = op.get_context() + with context.autocommit_block(): + for index_name in existing_names: + quoted_index = _quote_identifier(bind, index_name) + op.execute( + sa.text( + f"DROP INDEX CONCURRENTLY IF EXISTS " + f"{quoted_schema}.{quoted_index}" + ) + ) + + +def upgrade() -> None: + bind = _bind() + schema = _current_schema(bind) + + _require_table(bind, schema, "generation_records") + if not _column_exists( + bind, + schema, + "generation_records", + _PROMPT_USAGE_COLUMN, + ): + op.add_column( + "generation_records", + sa.Column(_PROMPT_USAGE_COLUMN, sa.Text(), nullable=True), + schema=schema, + ) + + _create_missing_indexes_concurrently(bind, schema) + + +def downgrade() -> None: + bind = _bind() + schema = _current_schema(bind) + + _drop_indexes_concurrently(bind, schema) + + _require_table(bind, schema, "generation_records") + if _column_exists( + bind, + schema, + "generation_records", + _PROMPT_USAGE_COLUMN, + ): + op.drop_column( + "generation_records", + _PROMPT_USAGE_COLUMN, + schema=schema, + ) diff --git a/video-gen-api/app/api/v1/auth.py b/video-gen-api/app/api/v1/auth.py index 05b2cc34..b68a7ea3 100644 --- a/video-gen-api/app/api/v1/auth.py +++ b/video-gen-api/app/api/v1/auth.py @@ -360,10 +360,10 @@ async def get_site_info(db: AsyncSession = Depends(get_db)): return f"{base_url}{path}" return { - "site_name": info.get("site_name", "VideoGen.AI"), + "site_name": info.get("site_name", "智创"), "site_logo": to_full_url(info.get("site_logo")), "user_agreement_privacy_url": to_full_url(info.get("user_agreement_privacy_url")), - "site_copyright": info.get("site_copyright", "© 2024 民众智创 版权所有"), + "site_copyright": info.get("site_copyright", "© 2026 智创 版权所有"), "operation_manual": info.get("operation_manual", ""), "login_bg_video": to_full_url(info.get("login_bg_video")) if info.get("login_bg_video") else "", "optimize_hold_credits": int(info.get("optimize_hold_credits") or 5), diff --git a/video-gen-api/app/api/v1/generation.py b/video-gen-api/app/api/v1/generation.py index 5fc3a4fe..7605d454 100644 --- a/video-gen-api/app/api/v1/generation.py +++ b/video-gen-api/app/api/v1/generation.py @@ -26,7 +26,6 @@ from app.services.generation.pipeline.db_lock_service import ( DatabaseRowLockBusy, execute_with_lock_timeout, ) -from app.services.llm import optimize_prompt from app.services.video_url import validate_and_get_record_id, get_video_stream_url from app.services.private_portrait.reference_resolver import batch_resolve_private_portrait_reference_display_urls, resolve_private_portrait_reference_display_urls from app.services.resource_signed_url_service import build_resource_signed_url @@ -43,21 +42,6 @@ from app.enums.generation_status import ( GenerationType, ) from app.enums.common import LogEventStatusEnum -from app.enums.credit_record import ( - CreditRecordBillingScene, - CreditRecordChargeKind, - CreditRecordSourceModule, -) -from app.enums.llm_billing import LlmBillingConfigKey -from app.services.llm_billing import ( - LlmBillingContext, - log_provider_failure, - log_provider_start, - log_provider_success, - release_on_failure, - settle_success, - start_hold, -) from app.enums.generation_record import ( GenerationRecordConfigSourceEnum, GenerationRecordEventTypeEnum, @@ -70,12 +54,9 @@ from app.services.generation.billing_service import ( from app.services.generation.ai.engine_service import ( get_image_engine, get_video_engine, - image_supported_sizes, - parse_json_list, ) from app.services.generation.pipeline.generation_record_config_service import ( ensure_generation_record_config_frozen, - freeze_generation_record_config_with_log, frozen_generation_record_engine_view, generation_record_config_fallback_hint, generation_record_engine_snapshot, @@ -87,12 +68,12 @@ from app.services.generation.media_reference_service import ( calculate_media_reference_usage, validate_media_reference_usage_for_engine, ) +from app.services.generation.prompt_optimize_service import optimize_generation_prompt from app.enums.audio_reference import ( AUDIO_ALLOWED_EXTENSIONS, AUDIO_ALLOWED_MIME_TYPES, AUDIO_MAX_FILE_SIZE_MB, ) -from app.utils.id_gen import generate_id from app.utils.exceptions import RecordNotFoundError, InvalidStatusError router = APIRouter(prefix="/generation-records", tags=["generation"]) @@ -107,43 +88,12 @@ def _record_config_complete(record: GenerationRecord) -> bool: return is_generation_record_config_complete(record) -def _canonical_references(value: object) -> str: - return json.dumps(value or [], ensure_ascii=False, sort_keys=True, separators=(",", ":")) - - -def _idempotency_config_matches(record: GenerationRecord, req: OptimizeParams) -> bool: - try: - existing_references = json.loads(record.media_references) if record.media_references else [] - except (TypeError, json.JSONDecodeError): - return False - if ( - str(record.project_id) != str(req.project_id) - or record.original_prompt != req.prompt - or record.gen_type != req.gen_type.value - or str(record.engine_id or "") != str(req.engine_id) - or bool(record.include_media_references) != bool(req.include_media_references) - or _canonical_references(existing_references) != _canonical_references(req.references) - ): - return False - if req.gen_type == GenerationType.video: - return ( - record.duration == req.duration - and record.aspect_ratio == req.aspect_ratio - and record.resolution == req.resolution - ) - return ( - record.image_size == req.image_size - and record.image_proportion == req.image_proportion - and record.image_px == req.image_px - ) - - def _record_status_view(record: GenerationRecord) -> dict[str, object]: config_complete = _record_config_complete(record) config_recoverable = is_generation_record_config_recoverable(record) prompt_failure = record.status == "failed" and record.resource_generation_started_at is None resource_failure = record.status == "failed" and record.resource_generation_started_at is not None - if record.status in {"pending", "optimizing"}: + if record.status in {"pending", "optimizing", "settlement_pending"}: client_status = "prompt_processing" operation_phase = "prompt" elif record.status == "prompt_optimized": @@ -164,7 +114,7 @@ def _record_status_view(record: GenerationRecord) -> dict[str, object]: "config_fallback_hint": generation_record_config_fallback_hint(record), "can_generate": record.status == "prompt_optimized" and (config_complete or config_recoverable), "can_retry": resource_failure and config_complete and record.pipeline_stage != GenerationRecordPipelineStage.UPSCALE_FAILED.value, - "should_poll": record.status in {"optimizing", "generating"}, + "should_poll": record.status in {"optimizing", "settlement_pending", "generating"}, "client_status": client_status, "operation_phase": operation_phase, } @@ -174,26 +124,6 @@ def _frozen_engine_view(record: GenerationRecord) -> SimpleNamespace: return frozen_generation_record_engine_view(record) -def _validate_video_engine_selection(engine, *, aspect_ratio: str, resolution: str, duration: int) -> None: - ratios = [str(item) for item in parse_json_list(engine.supported_ratios, [])] - resolutions = [str(item) for item in parse_json_list(engine.supported_resolutions, [])] - durations = [int(item) for item in parse_json_list(engine.supported_durations, []) if str(item).isdigit()] - if ratios and aspect_ratio not in ratios: - raise HTTPException(status_code=400, detail="当前视频引擎不支持所选画面比例") - if resolutions and resolution not in resolutions: - raise HTTPException(status_code=400, detail="当前视频引擎不支持所选分辨率") - if durations and duration not in durations: - raise HTTPException(status_code=400, detail="当前视频引擎不支持所选时长") - if int(engine.max_duration or 0) > 0 and duration > int(engine.max_duration): - raise HTTPException(status_code=400, detail="生成时长超过当前视频引擎上限") - - -def _validate_image_engine_selection(engine, *, image_size: str) -> None: - sizes = image_supported_sizes(engine) - if sizes and image_size not in sizes: - raise HTTPException(status_code=400, detail="当前图片引擎不支持所选画面分辨率") - - def _record_to_out(record: GenerationRecord, project_name: str, refs_override: list[dict] | None = None) -> GenerationRecordOut: refs = refs_override if refs is None and record.media_references: @@ -286,6 +216,8 @@ async def list_records( db: AsyncSession = Depends(get_db), ): allowed_statuses = { + "optimizing", + "settlement_pending", "prompt_optimized", "generating", "failed", @@ -295,7 +227,7 @@ async def list_records( if status and status not in allowed_statuses: raise HTTPException( status_code=400, - detail="状态参数错误，仅支持：prompt_optimized、generating、failed、completed", + detail="状态参数错误，仅支持：optimizing、settlement_pending、prompt_optimized、generating、failed、completed", ) offset = (page - 1) * page_size @@ -356,324 +288,20 @@ async def optimize( db: AsyncSession = Depends(get_db), ): user_id_snapshot = str(current_user.id) - # Validate project, engine and the complete generation configuration before - # charging prompt credits or invoking the LLM. - proj_result = await db.execute( - select(Project).where( - Project.id == req.project_id, - Project.user_id == user_id_snapshot, - Project.deleted_at.is_(None), - ).limit(1) - ) - project = proj_result.scalar_one_or_none() - if not project: - raise HTTPException(status_code=404, detail="项目不存在") - - log_generation_record_config_event( - event_type=GenerationRecordEventTypeEnum.PROMPT_CONFIG_VALIDATE_START, - event_status=LogEventStatusEnum.STARTED, - source=GenerationRecordConfigSourceEnum.PROMPT_OPTIMIZE, - record=GenerationRecord( - id=req.idempotency_key or "pending", - user_id=user_id_snapshot, - project_id=req.project_id, - original_prompt=req.prompt, - gen_type=req.gen_type.value, - duration=req.duration if req.gen_type == GenerationType.video else None, - aspect_ratio=req.aspect_ratio if req.gen_type == GenerationType.video else None, - resolution=req.resolution if req.gen_type == GenerationType.video else None, - image_size=req.image_size if req.gen_type == GenerationType.image else None, - image_proportion=req.image_proportion if req.gen_type == GenerationType.image else None, - image_px=req.image_px if req.gen_type == GenerationType.image else None, - include_media_references=bool(req.include_media_references), - media_references=json.dumps(req.references, ensure_ascii=False) if req.references else None, - engine_id=req.engine_id, - ), - detail={ - "project_id": req.project_id, - "gen_type": req.gen_type.value, - "engine_id": req.engine_id, - "include_media_references": bool(req.include_media_references), - "reference_count": len(req.references or []), - "source": GenerationRecordConfigSourceEnum.PROMPT_OPTIMIZE.value, - }, - ) - - # Idempotency must be checked before the temporary prompt-credit hold. The - # key is bound to one immutable generation configuration; reusing it with a - # different engine, parameter set or attachment selection is rejected. - if req.idempotency_key: - existing = await db.execute( - select(GenerationRecord, Project.name) - .join(Project, GenerationRecord.project_id == Project.id) - .where( - GenerationRecord.user_id == user_id_snapshot, - GenerationRecord.deleted_at.is_(None), - Project.deleted_at.is_(None), - GenerationRecord.idempotency_key == req.idempotency_key, - ) - .order_by(GenerationRecord.created_at.desc()) - .limit(1) - ) - row = existing.first() - if row: - existing_record, project_name = row - if not _idempotency_config_matches(existing_record, req): - raise HTTPException( - status_code=409, - detail="幂等键已绑定其他生成配置，请重新提交", - ) - if not existing_record.optimized_prompt or not _record_config_complete(existing_record): - raise HTTPException( - status_code=409, - detail="幂等记录配置不完整，请使用新的幂等键重新提交", - ) - refs = await resolve_private_portrait_reference_display_urls( - db, - json.loads(existing_record.media_references) if existing_record.media_references else None, - user_id=user_id_snapshot, - ) - return OptimizeResult( - optimized_prompt=existing_record.optimized_prompt, - text_credits_cost=existing_record.text_credits_cost or 0.0, - text_tokens_used=existing_record.text_tokens_used or 0, - record=_record_to_out(existing_record, project_name, refs_override=refs), - ) - - - if req.gen_type == GenerationType.video: - if req.duration not in DURATIONS: - raise HTTPException(status_code=400, detail=f"视频时长必须为{DURATIONS}秒之一") - if req.aspect_ratio not in ASPECT_RATIOS: - raise HTTPException(status_code=400, detail="不支持的画面比例") - if req.resolution not in RESOLUTIONS: - raise HTTPException(status_code=400, detail="不支持的分辨率") - engine = await get_video_engine(db, req.engine_id) - _validate_video_engine_selection( - engine, - aspect_ratio=req.aspect_ratio, - resolution=req.resolution, - duration=int(req.duration), - ) - else: - if req.image_size not in IMAGE_SIZES: - raise HTTPException(status_code=400, detail=f"图片分辨率必须为{IMAGE_SIZES}之一") - if not req.image_proportion or not req.image_px: - raise HTTPException(status_code=400, detail="图片生成需要指定比例和像素尺寸") - engine = await get_image_engine(db, req.engine_id) - _validate_image_engine_selection(engine, image_size=req.image_size) - - project_name_snapshot = str(project.name) - project_industry_snapshot = str(project.industry or "") - engine_snapshot_source = SimpleNamespace( - **{ - key: value - for key, value in vars(engine).items() - if key != "_sa_instance_state" - } - ) - - reference_usage = calculate_media_reference_usage( - json.dumps(req.references, ensure_ascii=False) if req.references else None, - include=bool(req.include_media_references), - ) - validate_media_reference_usage_for_engine( - reference_usage, - gen_type=req.gen_type.value, - engine=engine, - ) - - - log_generation_record_config_event( - event_type=GenerationRecordEventTypeEnum.PROMPT_CONFIG_VALIDATE_SUCCESS, - event_status=LogEventStatusEnum.SUCCESS, - source=GenerationRecordConfigSourceEnum.PROMPT_OPTIMIZE, - record=GenerationRecord( - id=req.idempotency_key or "pending", - user_id=user_id_snapshot, - project_id=req.project_id, - original_prompt=req.prompt, - gen_type=req.gen_type.value, - duration=req.duration if req.gen_type == GenerationType.video else None, - aspect_ratio=req.aspect_ratio if req.gen_type == GenerationType.video else None, - resolution=req.resolution if req.gen_type == GenerationType.video else None, - image_size=req.image_size if req.gen_type == GenerationType.image else None, - image_proportion=req.image_proportion if req.gen_type == GenerationType.image else None, - image_px=req.image_px if req.gen_type == GenerationType.image else None, - include_media_references=bool(req.include_media_references), - media_references=json.dumps(req.references, ensure_ascii=False) if req.references else None, - engine_id=req.engine_id, - ), - detail={ - "project_id": req.project_id, - "gen_type": req.gen_type.value, - "engine_id": req.engine_id, - "include_media_references": bool(req.include_media_references), - "reference_count": len(req.references or []), - "media_reference_usage": getattr(reference_usage, "__dict__", None) or str(reference_usage), - "source": GenerationRecordConfigSourceEnum.PROMPT_OPTIMIZE.value, - }, - ) - - - record_id_value = generate_id() - prompt_attempt_no = 1 - llm_billing_context = LlmBillingContext( + service_result = await optimize_generation_prompt( + db, + req=req, user_id=user_id_snapshot, - owner_type=OWNER_GENERATION_RECORD, - owner_id=record_id_value, - attempt_no=prompt_attempt_no, - charge_kind=CreditRecordChargeKind.TEXT_PROMPT.value, - billing_scene=CreditRecordBillingScene.GENERATION_RECORD_TEXT_PROMPT_OPTIMIZE.value, - source_module=CreditRecordSourceModule.GENERATION_RECORD.value, - related_id=record_id_value, - hold_config_key=LlmBillingConfigKey.HOLD_GENERATION_RECORD_PROMPT.value, - description_prefix="AI创作提示词优化", - trace_id=f"generation-optimize:{record_id_value}", - request_id=req.idempotency_key, ) - await start_hold(db, llm_billing_context) - await db.commit() - - log_provider_start(llm_billing_context, detail={"gen_type": req.gen_type.value}) - try: - optimized, token_usage = await optimize_prompt( - db, - req.prompt, - user_id=user_id_snapshot, - industry_key=project_industry_snapshot, - duration=req.duration if req.gen_type == GenerationType.video else None, - image_size=req.image_size if req.gen_type == GenerationType.image else None, - image_proportion=req.image_proportion if req.gen_type == GenerationType.image else None, - image_px=req.image_px if req.gen_type == GenerationType.image else None, - references=req.references, - gen_type=req.gen_type.value, - log_module="generation_record", - log_step="prompt_optimize", - log_project_id=req.project_id, - log_owner_type=OWNER_GENERATION_RECORD, - log_owner_id=record_id_value, - generation_attempt_no=prompt_attempt_no, - ) - log_provider_success(llm_billing_context, usage=token_usage) - except Exception as exc: - from app.services.error_codes import extract_error_message - - await db.rollback() - log_provider_failure(llm_billing_context, error=str(exc)) - await release_on_failure(db, llm_billing_context, error=str(exc)) - await db.commit() - raise HTTPException( - status_code=502, - detail=f"AI模型调用失败: {extract_error_message(exc, '提示词')}", - ) from exc - - async def _persist_optimized_result() -> str: - existing_result = await db.execute( - select(GenerationRecord) - .where(GenerationRecord.id == record_id_value) - .with_for_update() - .limit(1) - ) - record = existing_result.scalar_one_or_none() - if record is None: - record = GenerationRecord( - id=record_id_value, - user_id=user_id_snapshot, - project_id=req.project_id, - original_prompt=req.prompt, - optimized_prompt=optimized, - gen_type=req.gen_type.value, - duration=req.duration if req.gen_type == GenerationType.video else None, - aspect_ratio=req.aspect_ratio if req.gen_type == GenerationType.video else None, - resolution=req.resolution if req.gen_type == GenerationType.video else None, - image_size=req.image_size if req.gen_type == GenerationType.image else None, - image_proportion=req.image_proportion if req.gen_type == GenerationType.image else None, - image_px=req.image_px if req.gen_type == GenerationType.image else None, - status="prompt_optimized", - pipeline_stage=None, - credits_cost=0, - text_credits_cost=0, - text_tokens_used=int(token_usage.get("total_tokens", 0) or 0), - media_references=json.dumps(req.references, ensure_ascii=False) if req.references else None, - include_media_references=bool(req.include_media_references), - idempotency_key=req.idempotency_key, - ) - db.add(record) - else: - # commit 结果不确定或本地持久化重试时，复用同一主键和同一账务 attempt。 - record.optimized_prompt = optimized - record.status = "prompt_optimized" - record.pipeline_stage = None - record.error_message = None - record.text_credits_cost = 0 - record.text_tokens_used = int(token_usage.get("total_tokens", 0) or 0) - - if req.gen_type == GenerationType.video: - from app.services.video_upscale.snapshot_service import build_video_upscale_snapshot - - provider_resolution, upscale_enabled, upscale_snapshot_json = await build_video_upscale_snapshot( - db, - target_resolution=req.resolution, - aspect_ratio=req.aspect_ratio, - supported_provider_resolutions=parse_json_list( - engine_snapshot_source.supported_resolutions, - [], - ), - ) - record.provider_generation_resolution = provider_resolution - record.video_upscale_enabled_snapshot = upscale_enabled - record.video_upscale_snapshot_json = upscale_snapshot_json - else: - record.provider_generation_resolution = None - record.video_upscale_enabled_snapshot = False - record.video_upscale_snapshot_json = None - - freeze_generation_record_config_with_log( - record, - engine=engine_snapshot_source, - source=GenerationRecordConfigSourceEnum.PROMPT_OPTIMIZE, - ) - await db.flush() - billing = await settle_success( - db, - llm_billing_context, - usage=token_usage, - description=f"提示词优化 - {project_name_snapshot}", - ) - charge_item = next( - (item for item in billing.items if item.biz_key == llm_billing_context.charge_biz_key), - None, - ) - if charge_item: - record.text_credits_cost = round(charge_item.amount, 2) - record_id_snapshot = str(record.id) - await db.commit() - return record_id_snapshot - - try: - record_id_snapshot = await _persist_optimized_result() - except Exception as first_exc: - await db.rollback() - logger.exception( - "prompt optimize local persistence/settlement failed after provider success; retry once: record_id=%s", - record_id_value, - ) - try: - record_id_snapshot = await _persist_optimized_result() - except Exception: - await db.rollback() - logger.exception( - "prompt optimize idempotent persistence retry failed; active HOLD retained for repair: record_id=%s", - record_id_value, - ) - raise first_exc - - refreshed = await db.execute( select(GenerationRecord, Project.name) .join(Project, GenerationRecord.project_id == Project.id) - .where(GenerationRecord.id == record_id_snapshot) + .where( + GenerationRecord.id == service_result.record_id, + GenerationRecord.user_id == user_id_snapshot, + GenerationRecord.deleted_at.is_(None), + Project.deleted_at.is_(None), + ) .limit(1) ) refreshed_row = refreshed.first() diff --git a/video-gen-api/app/api/v1/shot_replicate.py b/video-gen-api/app/api/v1/shot_replicate.py index 604d43f7..2ca75228 100644 --- a/video-gen-api/app/api/v1/shot_replicate.py +++ b/video-gen-api/app/api/v1/shot_replicate.py @@ -18,6 +18,7 @@ from app.enums.credit_record import ( CreditRecordBillingScene, CreditRecordChargeKind, CreditRecordOwnerType, + CreditRecordSourceStepCode, ) from app.enums.llm_billing import LlmBillingConfigKey from app.enums.shot_replicate import ( @@ -274,7 +275,7 @@ def _analysis_dispatch_billing_context( source_module=MODULE, source_project_id=task_set_id, source_step_id=owner_id, - source_step_code=ShotReplicateStepCodeEnum.VIDEO_ANALYSIS.value, + source_step_code=CreditRecordSourceStepCode.VIDEO_ANALYSIS.value, related_id=owner_id, hold_config_key=LlmBillingConfigKey.HOLD_SHOT_VIDEO_ANALYSIS.value, description_prefix=( @@ -569,7 +570,7 @@ async def create_shot_task_set( from app.tasks.shot_replicate_tasks import analyze_original_video analyze_original_video.apply_async( - args=[task_set_id], + args=[task_set_id, analysis_attempt_no], queue=CeleryQueue.GEN_SHOT_ANALYSIS.value, countdown=0, task_id=celery_task_id, @@ -590,6 +591,7 @@ async def create_shot_task_set( db, current_user=_user_context(current_user), task_set_id=task_set_id, + expected_attempt_no=analysis_attempt_no, error_message=f"拆镜分析任务投递失败: {exc}", ) await db.commit() @@ -672,7 +674,7 @@ async def get_shot_task_set( "/task-sets/{task_set_id}/reanalyze", response_model=ShotReanalyzeOut, summary="重新投递原视频 AI 分析任务", - description="用于处理原视频分析失败或待处理的异常数据；重置分析状态后重新投递 analyze_original_video。", + description="仅用于重新处理原视频分析失败的数据；处理中、待处理或已完成状态均拒绝重复投递。", ) async def reanalyze_task_set( task_set_id: str = Path(..., description="拆镜总任务集ID，即 shot_replicate_task_sets.id"), @@ -686,7 +688,6 @@ async def reanalyze_task_set( db, current_user=current_user, task_set_id=task_set_id, - force=req.force, reason=req.reason, ) analysis_attempt_no = int(out.analysis_attempt_no) @@ -729,7 +730,7 @@ async def reanalyze_task_set( from app.tasks.shot_replicate_tasks import analyze_original_video analyze_original_video.apply_async( - args=[task_set_id], + args=[task_set_id, analysis_attempt_no], queue=CeleryQueue.GEN_SHOT_ANALYSIS.value, countdown=0, task_id=celery_task_id, @@ -758,6 +759,7 @@ async def reanalyze_task_set( db, current_user=_user_context(current_user), task_set_id=task_set_id, + expected_attempt_no=analysis_attempt_no, error_message=f"原视频再次分析任务投递失败: {exc}", ) await db.commit() @@ -923,7 +925,7 @@ async def get_segment( "/segments/{segment_id}/reanalyze", response_model=ShotReanalyzeOut, summary="重新投递切片视频 AI 分析任务", - description="用于处理自定义切片视频分析失败或待处理的异常数据；重置分析状态后重新投递 analyze_custom_segment_video。", + description="仅用于重新处理自定义切片视频分析失败的数据；处理中、待处理或已完成状态均拒绝重复投递。", ) async def reanalyze_segment( segment_id: str = Path(..., description="拆镜片段ID，即 shot_replicate_segments.id"), @@ -937,7 +939,6 @@ async def reanalyze_segment( db, current_user=current_user, segment_id=segment_id, - force=req.force, reason=req.reason, ) task_set_id = str(out.task_set_id) @@ -981,7 +982,7 @@ async def reanalyze_segment( from app.tasks.shot_replicate_tasks import analyze_custom_segment_video analyze_custom_segment_video.apply_async( - args=[segment_id], + args=[segment_id, analysis_attempt_no], queue=CeleryQueue.GEN_SHOT_ANALYSIS.value, countdown=0, task_id=celery_task_id, @@ -1012,6 +1013,7 @@ async def reanalyze_segment( db, current_user=_user_context(current_user), segment_id=segment_id, + expected_attempt_no=analysis_attempt_no, error_message=f"切片视频再次分析任务投递失败: {exc}", ) await db.commit() diff --git a/video-gen-api/app/config.py b/video-gen-api/app/config.py index 1cd8dc85..eb2eb956 100644 --- a/video-gen-api/app/config.py +++ b/video-gen-api/app/config.py @@ -170,7 +170,7 @@ class Settings(BaseSettings): CELERY_RUNTIME_GC_INTERVAL_SECONDS: int = 600 CELERY_RUNTIME_GC_BATCH_SIZE: int = 500 - CHATAPI_REQUEST_TIMEOUT_SECONDS: int = 180 + CHATAPI_REQUEST_TIMEOUT_SECONDS: int = 600 CHATAPI_VIDEO_FPS: float = 0.5 CHATAPI_ASYNC_MAX_RETRIES: int = 3 CHATAPI_ASYNC_RETRY_BACKOFF_SECONDS: int = 30 @@ -307,7 +307,7 @@ class Settings(BaseSettings): SHOT_ANALYSIS_ACTIVE_REDIS_ZSET_KEY: str = "vg:celery:shot_analysis:active_index" SHOT_ANALYSIS_LOCK_KEY_PREFIX: str = "vg:lock:shot_analysis" SHOT_ANALYSIS_TEMPERATURE: float = 0.1 - SHOT_ANALYSIS_MAX_TOKENS: int = 5000 + SHOT_ANALYSIS_MAX_TOKENS: int = 10000 SHOT_ANALYSIS_VIDEO_FPS: float = 1.0 SHOT_ANALYSIS_MAX_LOCAL_VIDEO_MB: int = 45 diff --git a/video-gen-api/app/enums/common.py b/video-gen-api/app/enums/common.py index 04d87f99..b4d43789 100644 --- a/video-gen-api/app/enums/common.py +++ b/video-gen-api/app/enums/common.py @@ -26,6 +26,7 @@ class LogSourceEnum(StrEnum): CELERY = "celery" RECOVERY = "recovery" REMOTE_API = "remote_api" + CLI = "cli" class ModuleGenerationFlowVersionEnum(StrEnum): """模块生成项目流程版本。""" diff --git a/video-gen-api/app/enums/generation_record.py b/video-gen-api/app/enums/generation_record.py index 3e908e5b..99c03ca3 100644 --- a/video-gen-api/app/enums/generation_record.py +++ b/video-gen-api/app/enums/generation_record.py @@ -20,6 +20,12 @@ class GenerationRecordEventTypeEnum(StrEnum): PROMPT_CONFIG_FREEZE_START = "PROMPT_CONFIG_FREEZE_START" PROMPT_CONFIG_FREEZE_SUCCESS = "PROMPT_CONFIG_FREEZE_SUCCESS" PROMPT_CONFIG_FREEZE_FAILED = "PROMPT_CONFIG_FREEZE_FAILED" + PROMPT_OPTIMIZE_PLACEHOLDER_CREATED = "PROMPT_OPTIMIZE_PLACEHOLDER_CREATED" + PROMPT_OPTIMIZE_IDEMPOTENCY_HIT = "PROMPT_OPTIMIZE_IDEMPOTENCY_HIT" + PROMPT_OPTIMIZE_PROVIDER_RESULT_STAGED = "PROMPT_OPTIMIZE_PROVIDER_RESULT_STAGED" + PROMPT_OPTIMIZE_SETTLEMENT_PENDING = "PROMPT_OPTIMIZE_SETTLEMENT_PENDING" + PROMPT_OPTIMIZE_SETTLEMENT_SUCCESS = "PROMPT_OPTIMIZE_SETTLEMENT_SUCCESS" + PROMPT_OPTIMIZE_FAILED_RELEASED = "PROMPT_OPTIMIZE_FAILED_RELEASED" LEGACY_CONFIG_FALLBACK_START = "LEGACY_CONFIG_FALLBACK_START" LEGACY_CONFIG_FALLBACK_SUCCESS = "LEGACY_CONFIG_FALLBACK_SUCCESS" LEGACY_CONFIG_FALLBACK_FAILED = "LEGACY_CONFIG_FALLBACK_FAILED" diff --git a/video-gen-api/app/enums/generation_status.py b/video-gen-api/app/enums/generation_status.py index 6cfc1db9..3d1272e5 100644 --- a/video-gen-api/app/enums/generation_status.py +++ b/video-gen-api/app/enums/generation_status.py @@ -3,6 +3,8 @@ from enum import Enum class GenerationStatus(str, Enum): """生成状态。""" + optimizing = "optimizing" + settlement_pending = "settlement_pending" prompt_optimized = "prompt_optimized" generating = "generating" completed = "completed" diff --git a/video-gen-api/app/enums/llm_billing.py b/video-gen-api/app/enums/llm_billing.py index a8f5c4ec..4b1f0626 100644 --- a/video-gen-api/app/enums/llm_billing.py +++ b/video-gen-api/app/enums/llm_billing.py @@ -50,6 +50,12 @@ class LlmBillingEvent(StrEnum): CELERY_DISPATCH_SUCCESS = "LLM_CELERY_DISPATCH_SUCCESS" CELERY_DISPATCH_FAILURE = "LLM_CELERY_DISPATCH_FAILURE" CELERY_DISPATCH_COMPENSATED = "LLM_CELERY_DISPATCH_COMPENSATED" + RETRY_PREVIOUS_ATTEMPT_VALIDATE_START = "LLM_RETRY_PREVIOUS_ATTEMPT_VALIDATE_START" + RETRY_PREVIOUS_ATTEMPT_VALIDATE_SUCCESS = "LLM_RETRY_PREVIOUS_ATTEMPT_VALIDATE_SUCCESS" + RETRY_PREVIOUS_ATTEMPT_BLOCKED = "LLM_RETRY_PREVIOUS_ATTEMPT_BLOCKED" + USAGE_INVALID = "LLM_USAGE_INVALID" + TOKEN_USAGE_CREATED = "LLM_TOKEN_USAGE_CREATED" + TOKEN_USAGE_REUSED = "LLM_TOKEN_USAGE_REUSED" class LlmBillingDomain(StrEnum): diff --git a/video-gen-api/app/enums/shot_replicate.py b/video-gen-api/app/enums/shot_replicate.py index d238fd7e..4870aad7 100644 --- a/video-gen-api/app/enums/shot_replicate.py +++ b/video-gen-api/app/enums/shot_replicate.py @@ -118,6 +118,7 @@ class ShotReplicateLogEventEnum(StrEnum): ANALYSIS_REMOTE_API_FAILED = "SHOT_ANALYSIS_REMOTE_API_FAILED" ANALYSIS_RESPONSE_PARSE_FAILED = "SHOT_ANALYSIS_RESPONSE_PARSE_FAILED" ANALYSIS_RESPONSE_EMPTY = "SHOT_ANALYSIS_RESPONSE_EMPTY" + ANALYSIS_STALE_ATTEMPT_SKIPPED = "SHOT_ANALYSIS_STALE_ATTEMPT_SKIPPED" SEGMENT_REANALYZE_RECEIVED = "SHOT_SEGMENT_REANALYZE_RECEIVED" SEGMENT_REANALYZE_SUBMITTED = "SHOT_SEGMENT_REANALYZE_SUBMITTED" @@ -131,6 +132,7 @@ class ShotReplicateLogEventEnum(StrEnum): SEGMENT_ANALYSIS_REMOTE_API_FAILED = "SHOT_SEGMENT_ANALYSIS_REMOTE_API_FAILED" SPLIT_STATUS_CHANGED = "SHOT_SPLIT_STATUS_CHANGED" + SPLIT_SUMMARY_REPAIRED = "SHOT_SPLIT_SUMMARY_REPAIRED" SPLIT_BY_AI_SUBMITTED = "SHOT_SPLIT_BY_AI_SUBMITTED" SPLIT_CUSTOM_SUBMITTED = "SHOT_SPLIT_CUSTOM_SUBMITTED" SEGMENT_DELETED = "SHOT_SEGMENT_DELETED" diff --git a/video-gen-api/app/models/credit_record.py b/video-gen-api/app/models/credit_record.py index 45d4322b..3f0928b3 100644 --- a/video-gen-api/app/models/credit_record.py +++ b/video-gen-api/app/models/credit_record.py @@ -1,4 +1,4 @@ -from sqlalchemy import Float, ForeignKey, Index, Integer, String +from sqlalchemy import Float, ForeignKey, Index, Integer, String, text from sqlalchemy.orm import Mapped, mapped_column from app.models.base import Base, TimestampMixin @@ -11,6 +11,13 @@ class CreditRecord(Base, TimestampMixin): # PostgreSQL/MySQL/SQLite 对 nullable unique 的处理都允许多条 NULL，兼容历史数据。 Index("uq_credit_records_user_biz_key", "user_id", "biz_key", unique=True), Index("ix_credit_records_user_refund_for_biz_key", "user_id", "refund_for_biz_key"), + Index( + "uq_credit_records_user_refund_target", + "user_id", + "refund_for_biz_key", + unique=True, + postgresql_where=text("type = 'refund' AND refund_for_biz_key IS NOT NULL"), + ), Index("ix_credit_records_related_type", "related_id", "type"), Index("ix_credit_records_owner", "owner_type", "owner_id"), Index("ix_credit_records_subject_media", "credit_subject", "media_type"), diff --git a/video-gen-api/app/models/generation_record.py b/video-gen-api/app/models/generation_record.py index 0f6db213..57324b1f 100644 --- a/video-gen-api/app/models/generation_record.py +++ b/video-gen-api/app/models/generation_record.py @@ -18,6 +18,7 @@ class GenerationRecord(Base, TimestampMixin, SoftDeleteMixin): ) original_prompt: Mapped[str] = mapped_column(Text) optimized_prompt: Mapped[str | None] = mapped_column(Text, nullable=True) + prompt_usage_snapshot_json: Mapped[str | None] = mapped_column(Text, nullable=True) gen_type: Mapped[str] = mapped_column(String(16), default="video") duration: Mapped[int | None] = mapped_column(Integer, nullable=True) aspect_ratio: Mapped[str | None] = mapped_column(String(8), nullable=True) @@ -95,6 +96,13 @@ class GenerationRecord(Base, TimestampMixin, SoftDeleteMixin): __table_args__ = ( Index('idx_genrec_user_status_created', 'user_id', 'status', 'created_at'), Index('idx_genrec_project_status', 'project_id', 'status'), + Index( + 'uq_genrec_user_idempotency_active', + 'user_id', + 'idempotency_key', + unique=True, + postgresql_where=text("idempotency_key IS NOT NULL AND deleted_at IS NULL"), + ), Index( 'idx_genrec_next_poll_at', 'next_poll_at', diff --git a/video-gen-api/app/models/token_usage.py b/video-gen-api/app/models/token_usage.py index de403537..cdbcefed 100644 --- a/video-gen-api/app/models/token_usage.py +++ b/video-gen-api/app/models/token_usage.py @@ -1,4 +1,4 @@ -from sqlalchemy import ForeignKey, Index, Integer, String +from sqlalchemy import ForeignKey, Index, Integer, String, text from sqlalchemy.orm import Mapped, mapped_column from app.models.base import Base, TimestampMixin @@ -9,6 +9,13 @@ class TokenUsage(Base, TimestampMixin): __table_args__ = ( Index("ix_token_usage_owner", "owner_type", "owner_id"), Index("ix_token_usage_biz_key", "biz_key"), + Index( + "uq_token_usage_user_biz_key", + "user_id", + "biz_key", + unique=True, + postgresql_where=text("biz_key IS NOT NULL"), + ), ) id: Mapped[str] = mapped_column(String(32), primary_key=True) diff --git a/video-gen-api/app/schemas/shot_replicate.py b/video-gen-api/app/schemas/shot_replicate.py index c858ba61..d282fe3d 100644 --- a/video-gen-api/app/schemas/shot_replicate.py +++ b/video-gen-api/app/schemas/shot_replicate.py @@ -729,7 +729,7 @@ class ShotSegmentListOut(BaseModel): class ShotReanalyzeRequest(BaseModel): - force: bool = Field(False, description="是否强制重跑；默认 false。当前仅允许失败/待处理数据重试，已完成数据不建议强制覆盖") + force: bool = Field(False, description="兼容旧客户端保留字段；后端不再支持强制重跑，仅分析失败状态允许重新分析") reason: str | None = Field(None, max_length=200, description="再次分析原因，会写入模块日志") @field_validator("reason", mode="before") diff --git a/video-gen-api/app/services/generation/prompt_optimize_service.py b/video-gen-api/app/services/generation/prompt_optimize_service.py new file mode 100644 index 00000000..e7bb75d2 --- /dev/null +++ b/video-gen-api/app/services/generation/prompt_optimize_service.py @@ -0,0 +1,647 @@ +from __future__ import annotations + +import json +import logging +from dataclasses import dataclass +from typing import Any + +from fastapi import HTTPException +from sqlalchemy import select +from sqlalchemy.exc import IntegrityError +from sqlalchemy.ext.asyncio import AsyncSession + +from app.enums.common import LogEventStatusEnum +from app.enums.credit_record import ( + CreditRecordBillingScene, + CreditRecordChargeKind, + CreditRecordSourceModule, +) +from app.enums.generation_record import ( + GenerationRecordConfigSourceEnum, + GenerationRecordEventTypeEnum, +) +from app.enums.generation_status import ( + ASPECT_RATIOS, + DURATIONS, + IMAGE_SIZES, + RESOLUTIONS, + GenerationStatus, + GenerationType, +) +from app.enums.llm_billing import LlmBillingConfigKey +from app.models.generation_record import GenerationRecord +from app.models.project import Project +from app.schemas.generation import OptimizeParams +from app.services.error_codes import extract_error_message +from app.services.generation.ai.engine_service import ( + get_image_engine, + get_video_engine, + image_supported_sizes, + parse_json_list, +) +from app.services.generation.billing_service import OWNER_GENERATION_RECORD +from app.services.generation.media_reference_service import ( + calculate_media_reference_usage, + validate_media_reference_usage_for_engine, +) +from app.services.generation.pipeline.generation_record_config_service import ( + freeze_generation_record_config_with_log, + is_generation_record_config_complete, +) +from app.services.llm import optimize_prompt +from app.services.llm_billing import ( + LlmBillingContext, + log_provider_failure, + log_provider_start, + log_provider_success, + release_on_failure, + settle_success, + start_hold, +) +from app.services.operation_log_service import log_operation_error, log_operation_event +from app.services.video_upscale.snapshot_service import build_video_upscale_snapshot +from app.utils.id_gen import generate_id + +logger = logging.getLogger("videogen") + +_PROMPT_ATTEMPT_NO = 1 +_LOG_DOMAIN = "generation_record" +_LOG_MODULE = "generation_record" +_LOG_SOURCE = GenerationRecordConfigSourceEnum.PROMPT_OPTIMIZE.value + + +@dataclass(slots=True, frozen=True) +class PromptOptimizeServiceResult: + record_id: str + idempotent: bool = False + + +def _log_event( + event: GenerationRecordEventTypeEnum, + *, + status: LogEventStatusEnum = LogEventStatusEnum.SUCCESS, + user_id: str, + project_id: str | None, + record_id: str | None, + detail: dict[str, Any] | None = None, + error: str | None = None, +) -> None: + log_operation_event( + domain=_LOG_DOMAIN, + module=_LOG_MODULE, + event_type=event.value, + event_status=status.value, + source=_LOG_SOURCE, + user_id=user_id, + project_id=project_id, + task_id=record_id, + detail=detail, + error=error, + ) + + +def _billing_context(*, user_id: str, record_id: str, request_id: str | None) -> LlmBillingContext: + return LlmBillingContext( + user_id=user_id, + owner_type=OWNER_GENERATION_RECORD, + owner_id=record_id, + attempt_no=_PROMPT_ATTEMPT_NO, + charge_kind=CreditRecordChargeKind.TEXT_PROMPT.value, + billing_scene=CreditRecordBillingScene.GENERATION_RECORD_TEXT_PROMPT_OPTIMIZE.value, + source_module=CreditRecordSourceModule.GENERATION_RECORD.value, + related_id=record_id, + hold_config_key=LlmBillingConfigKey.HOLD_GENERATION_RECORD_PROMPT.value, + description_prefix="AI创作提示词优化", + trace_id=f"generation-optimize:{record_id}", + request_id=request_id, + ) + + +def _canonical_references(value: object) -> str: + return json.dumps(value or [], ensure_ascii=False, sort_keys=True, separators=(",", ":")) + + +def _idempotency_config_matches(record: GenerationRecord, req: OptimizeParams) -> bool: + try: + existing_references = json.loads(record.media_references) if record.media_references else [] + except (TypeError, json.JSONDecodeError): + return False + if ( + str(record.project_id) != str(req.project_id) + or record.original_prompt != req.prompt + or record.gen_type != req.gen_type.value + or str(record.engine_id or "") != str(req.engine_id) + or bool(record.include_media_references) != bool(req.include_media_references) + or _canonical_references(existing_references) != _canonical_references(req.references) + ): + return False + if req.gen_type == GenerationType.video: + return ( + record.duration == req.duration + and record.aspect_ratio == req.aspect_ratio + and record.resolution == req.resolution + ) + return ( + record.image_size == req.image_size + and record.image_proportion == req.image_proportion + and record.image_px == req.image_px + ) + + +def _validate_video_engine_selection(engine: Any, *, aspect_ratio: str, resolution: str, duration: int) -> None: + ratios = [str(item) for item in parse_json_list(engine.supported_ratios, [])] + resolutions = [str(item) for item in parse_json_list(engine.supported_resolutions, [])] + durations = [int(item) for item in parse_json_list(engine.supported_durations, []) if str(item).isdigit()] + if ratios and aspect_ratio not in ratios: + raise HTTPException(status_code=400, detail="当前视频引擎不支持所选画面比例") + if resolutions and resolution not in resolutions: + raise HTTPException(status_code=400, detail="当前视频引擎不支持所选分辨率") + if durations and duration not in durations: + raise HTTPException(status_code=400, detail="当前视频引擎不支持所选时长") + if int(engine.max_duration or 0) > 0 and duration > int(engine.max_duration): + raise HTTPException(status_code=400, detail="生成时长超过当前视频引擎上限") + + +def _validate_image_engine_selection(engine: Any, *, image_size: str) -> None: + sizes = image_supported_sizes(engine) + if sizes and image_size not in sizes: + raise HTTPException(status_code=400, detail="当前图片引擎不支持所选画面分辨率") + + +async def _find_idempotency_record( + db: AsyncSession, + *, + user_id: str, + idempotency_key: str | None, +) -> GenerationRecord | None: + if not idempotency_key: + return None + stmt = ( + select(GenerationRecord) + .where( + GenerationRecord.user_id == user_id, + GenerationRecord.idempotency_key == idempotency_key, + GenerationRecord.deleted_at.is_(None), + ) + .order_by(GenerationRecord.created_at.desc(), GenerationRecord.id.desc()) + .limit(1) + ) + result = await db.execute(stmt) + return result.scalar_one_or_none() + + +async def _settle_staged_result( + db: AsyncSession, + *, + record_id: str, + user_id: str, + project_name: str, + request_id: str | None, +) -> PromptOptimizeServiceResult: + result = await db.execute( + select(GenerationRecord) + .where( + GenerationRecord.id == record_id, + GenerationRecord.user_id == user_id, + GenerationRecord.deleted_at.is_(None), + ) + .with_for_update() + .limit(1) + ) + record = result.scalar_one_or_none() + if record is None: + raise HTTPException(status_code=404, detail="生成记录不存在") + if record.status in { + GenerationStatus.prompt_optimized.value, + GenerationStatus.generating.value, + GenerationStatus.completed.value, + }: + await db.rollback() + return PromptOptimizeServiceResult(record_id=record_id, idempotent=True) + if record.status != GenerationStatus.settlement_pending.value: + await db.rollback() + raise HTTPException(status_code=409, detail=f"当前提词状态不可结算：{record.status}") + if not record.optimized_prompt or not record.prompt_usage_snapshot_json: + await db.rollback() + raise HTTPException(status_code=409, detail="提词结果或计费快照缺失，需人工排查") + + try: + usage = json.loads(record.prompt_usage_snapshot_json) + except (TypeError, json.JSONDecodeError) as exc: + await db.rollback() + raise HTTPException(status_code=409, detail="提词计费快照损坏，需人工排查") from exc + if not isinstance(usage, dict): + await db.rollback() + raise HTTPException(status_code=409, detail="提词计费快照格式错误，需人工排查") + + project_id_snapshot = str(record.project_id) + status_snapshot = str(record.status) + ctx = _billing_context(user_id=user_id, record_id=record_id, request_id=request_id) + _log_event( + GenerationRecordEventTypeEnum.PROMPT_OPTIMIZE_SETTLEMENT_PENDING, + status=LogEventStatusEnum.STARTED, + user_id=user_id, + project_id=project_id_snapshot, + record_id=record_id, + detail={"status": status_snapshot, "attempt_no": _PROMPT_ATTEMPT_NO}, + ) + try: + billing = await settle_success( + db, + ctx, + usage=usage, + description=f"提示词优化 - {project_name}", + ) + charge_item = next( + (item for item in billing.items if item.biz_key == ctx.charge_biz_key), + None, + ) + record.text_credits_cost = round(float(charge_item.amount if charge_item else 0.0), 2) + record.text_tokens_used = int(usage.get("total_tokens", 0) or 0) + record.status = GenerationStatus.prompt_optimized.value + record.pipeline_stage = None + record.error_message = None + credits_snapshot = float(record.text_credits_cost or 0.0) + tokens_snapshot = int(record.text_tokens_used or 0) + await db.commit() + except Exception as exc: + await db.rollback() + _log_event( + GenerationRecordEventTypeEnum.PROMPT_OPTIMIZE_SETTLEMENT_PENDING, + status=LogEventStatusEnum.FAILED, + user_id=user_id, + project_id=project_id_snapshot, + record_id=record_id, + detail={"attempt_no": _PROMPT_ATTEMPT_NO, "error_type": type(exc).__name__}, + error=str(exc), + ) + raise HTTPException(status_code=503, detail="提词已生成，积分结算暂未完成，请使用相同幂等键重试") from exc + + _log_event( + GenerationRecordEventTypeEnum.PROMPT_OPTIMIZE_SETTLEMENT_SUCCESS, + user_id=user_id, + project_id=project_id_snapshot, + record_id=record_id, + detail={ + "attempt_no": _PROMPT_ATTEMPT_NO, + "text_credits_cost": credits_snapshot, + "text_tokens_used": tokens_snapshot, + }, + ) + return PromptOptimizeServiceResult(record_id=record_id, idempotent=False) + + +async def _handle_existing_record( + db: AsyncSession, + *, + record: GenerationRecord, + req: OptimizeParams, + user_id: str, + project_name: str, +) -> PromptOptimizeServiceResult: + record_id = str(record.id) + project_id = str(record.project_id) + status = str(record.status) + if not _idempotency_config_matches(record, req): + await db.rollback() + raise HTTPException(status_code=409, detail="幂等键已绑定其他生成配置，请重新提交") + + _log_event( + GenerationRecordEventTypeEnum.PROMPT_OPTIMIZE_IDEMPOTENCY_HIT, + status=LogEventStatusEnum.SKIPPED, + user_id=user_id, + project_id=project_id, + record_id=record_id, + detail={"record_status": status, "idempotency_key": req.idempotency_key}, + ) + if status == GenerationStatus.settlement_pending.value: + await db.rollback() + return await _settle_staged_result( + db, + record_id=record_id, + user_id=user_id, + project_name=project_name, + request_id=req.idempotency_key, + ) + if status == GenerationStatus.optimizing.value: + await db.rollback() + raise HTTPException(status_code=409, detail="相同幂等请求正在处理，请勿重复提交") + if status == GenerationStatus.failed.value: + await db.rollback() + raise HTTPException(status_code=409, detail="该幂等请求已失败，请使用新的幂等键重新提交") + if record.optimized_prompt and is_generation_record_config_complete(record): + await db.rollback() + return PromptOptimizeServiceResult(record_id=record_id, idempotent=True) + await db.rollback() + raise HTTPException(status_code=409, detail="幂等记录配置不完整，需人工排查或使用新的幂等键") + + +async def optimize_generation_prompt( + db: AsyncSession, + *, + req: OptimizeParams, + user_id: str, +) -> PromptOptimizeServiceResult: + project_result = await db.execute( + select(Project).where( + Project.id == req.project_id, + Project.user_id == user_id, + Project.deleted_at.is_(None), + ).limit(1) + ) + project = project_result.scalar_one_or_none() + if project is None: + raise HTTPException(status_code=404, detail="项目不存在") + project_name = str(project.name) + project_industry = str(project.industry or "") + project_id = str(project.id) + + existing = await _find_idempotency_record( + db, + user_id=user_id, + idempotency_key=req.idempotency_key, + ) + if existing is not None: + return await _handle_existing_record( + db, + record=existing, + req=req, + user_id=user_id, + project_name=project_name, + ) + + if req.gen_type == GenerationType.video: + if req.duration not in DURATIONS: + raise HTTPException(status_code=400, detail=f"视频时长必须为{DURATIONS}秒之一") + if req.aspect_ratio not in ASPECT_RATIOS: + raise HTTPException(status_code=400, detail="不支持的画面比例") + if req.resolution not in RESOLUTIONS: + raise HTTPException(status_code=400, detail="不支持的分辨率") + engine = await get_video_engine(db, req.engine_id) + _validate_video_engine_selection( + engine, + aspect_ratio=str(req.aspect_ratio), + resolution=str(req.resolution), + duration=int(req.duration), + ) + else: + if req.image_size not in IMAGE_SIZES: + raise HTTPException(status_code=400, detail=f"图片分辨率必须为{IMAGE_SIZES}之一") + if not req.image_proportion or not req.image_px: + raise HTTPException(status_code=400, detail="图片生成需要指定比例和像素尺寸") + engine = await get_image_engine(db, req.engine_id) + _validate_image_engine_selection(engine, image_size=str(req.image_size)) + + reference_usage = calculate_media_reference_usage( + json.dumps(req.references, ensure_ascii=False) if req.references else None, + include=bool(req.include_media_references), + ) + validate_media_reference_usage_for_engine( + reference_usage, + gen_type=req.gen_type.value, + engine=engine, + ) + + record_id = generate_id() + record = GenerationRecord( + id=record_id, + user_id=user_id, + project_id=project_id, + original_prompt=req.prompt, + optimized_prompt=None, + prompt_usage_snapshot_json=None, + gen_type=req.gen_type.value, + duration=req.duration if req.gen_type == GenerationType.video else None, + aspect_ratio=req.aspect_ratio if req.gen_type == GenerationType.video else None, + resolution=req.resolution if req.gen_type == GenerationType.video else None, + image_size=req.image_size if req.gen_type == GenerationType.image else None, + image_proportion=req.image_proportion if req.gen_type == GenerationType.image else None, + image_px=req.image_px if req.gen_type == GenerationType.image else None, + status=GenerationStatus.optimizing.value, + pipeline_stage=None, + credits_cost=0, + text_credits_cost=0, + text_tokens_used=0, + media_references=json.dumps(req.references, ensure_ascii=False) if req.references else None, + include_media_references=bool(req.include_media_references), + idempotency_key=req.idempotency_key, + engine_id=req.engine_id, + ) + if req.gen_type == GenerationType.video: + provider_resolution, upscale_enabled, upscale_snapshot_json = await build_video_upscale_snapshot( + db, + target_resolution=str(req.resolution), + aspect_ratio=str(req.aspect_ratio), + supported_provider_resolutions=parse_json_list(engine.supported_resolutions, []), + ) + record.provider_generation_resolution = provider_resolution + record.video_upscale_enabled_snapshot = upscale_enabled + record.video_upscale_snapshot_json = upscale_snapshot_json + else: + record.provider_generation_resolution = None + record.video_upscale_enabled_snapshot = False + record.video_upscale_snapshot_json = None + freeze_generation_record_config_with_log( + record, + engine=engine, + source=GenerationRecordConfigSourceEnum.PROMPT_OPTIMIZE, + ) + db.add(record) + try: + await db.flush() + except IntegrityError: + await db.rollback() + conflicting = await _find_idempotency_record( + db, + user_id=user_id, + idempotency_key=req.idempotency_key, + ) + if conflicting is None: + raise + return await _handle_existing_record( + db, + record=conflicting, + req=req, + user_id=user_id, + project_name=project_name, + ) + + ctx = _billing_context(user_id=user_id, record_id=record_id, request_id=req.idempotency_key) + try: + await start_hold(db, ctx) + await db.commit() + except Exception: + await db.rollback() + raise + + _log_event( + GenerationRecordEventTypeEnum.PROMPT_OPTIMIZE_PLACEHOLDER_CREATED, + user_id=user_id, + project_id=project_id, + record_id=record_id, + detail={ + "idempotency_key": req.idempotency_key, + "gen_type": req.gen_type.value, + "engine_id": req.engine_id, + "attempt_no": _PROMPT_ATTEMPT_NO, + "reference_count": len(req.references or []), + }, + ) + + log_provider_start(ctx, detail={"gen_type": req.gen_type.value}) + try: + optimized_prompt, usage = await optimize_prompt( + db, + req.prompt, + user_id=user_id, + industry_key=project_industry, + duration=req.duration if req.gen_type == GenerationType.video else None, + image_size=req.image_size if req.gen_type == GenerationType.image else None, + image_proportion=req.image_proportion if req.gen_type == GenerationType.image else None, + image_px=req.image_px if req.gen_type == GenerationType.image else None, + references=req.references, + gen_type=req.gen_type.value, + log_module="generation_record", + log_step="prompt_optimize", + log_project_id=project_id, + log_owner_type=OWNER_GENERATION_RECORD, + log_owner_id=record_id, + generation_attempt_no=_PROMPT_ATTEMPT_NO, + ) + log_provider_success(ctx, usage=usage) + except Exception as exc: + await db.rollback() + log_provider_failure(ctx, error=str(exc)) + compensated = False + try: + failed_result = await db.execute( + select(GenerationRecord) + .where(GenerationRecord.id == record_id, GenerationRecord.deleted_at.is_(None)) + .with_for_update() + .limit(1) + ) + failed_record = failed_result.scalar_one_or_none() + if failed_record is not None and failed_record.status == GenerationStatus.optimizing.value: + failed_record.status = GenerationStatus.failed.value + failed_record.error_message = extract_error_message(exc, "提示词") + await release_on_failure(db, ctx, error=str(exc)) + compensated = True + await db.commit() + except Exception: + await db.rollback() + logger.exception("prompt optimize failure compensation failed: record_id=%s", record_id) + raise + _log_event( + GenerationRecordEventTypeEnum.PROMPT_OPTIMIZE_FAILED_RELEASED, + status=(LogEventStatusEnum.FAILED if compensated else LogEventStatusEnum.SKIPPED), + user_id=user_id, + project_id=project_id, + record_id=record_id, + detail={ + "attempt_no": _PROMPT_ATTEMPT_NO, + "error_type": type(exc).__name__, + "compensated": compensated, + }, + error=str(exc), + ) + raise HTTPException( + status_code=502, + detail=f"AI模型调用失败: {extract_error_message(exc, '提示词')}", + ) from exc + + usage_snapshot = dict(usage or {}) + try: + input_tokens = int(usage_snapshot.get("input_tokens", 0) or 0) + output_tokens = int(usage_snapshot.get("output_tokens", 0) or 0) + except (TypeError, ValueError): + # 保留原始 usage 交给统一账务校验拒绝；这里只避免展示字段写入异常。 + input_tokens = 0 + output_tokens = 0 + if input_tokens >= 0 and output_tokens >= 0: + reported_total = usage_snapshot.get("total_tokens") + normalized_total = input_tokens + output_tokens + if reported_total not in (None, ""): + try: + if int(reported_total) != normalized_total: + usage_snapshot["reported_total_tokens"] = int(reported_total) + except (TypeError, ValueError): + usage_snapshot["reported_total_tokens"] = reported_total + usage_snapshot["total_tokens"] = normalized_total + usage_snapshot.setdefault("source_module", "generation_record") + usage_snapshot.setdefault("source_step_code", "prompt_optimize") + staged = False + last_stage_error: Exception | None = None + for _ in range(2): + try: + await db.rollback() + stage_result = await db.execute( + select(GenerationRecord) + .where( + GenerationRecord.id == record_id, + GenerationRecord.user_id == user_id, + GenerationRecord.deleted_at.is_(None), + ) + .with_for_update() + .limit(1) + ) + staged_record = stage_result.scalar_one_or_none() + if staged_record is None: + raise RuntimeError("prompt optimize owner record missing") + if staged_record.status in { + GenerationStatus.prompt_optimized.value, + GenerationStatus.generating.value, + GenerationStatus.completed.value, + }: + await db.rollback() + return PromptOptimizeServiceResult(record_id=record_id, idempotent=True) + staged_record.optimized_prompt = optimized_prompt + staged_record.prompt_usage_snapshot_json = json.dumps( + usage_snapshot, + ensure_ascii=False, + sort_keys=True, + default=str, + ) + staged_record.text_tokens_used = int(usage_snapshot.get("total_tokens", 0) or 0) + staged_record.status = GenerationStatus.settlement_pending.value + staged_record.pipeline_stage = None + staged_record.error_message = None + await db.commit() + staged = True + break + except Exception as exc: + last_stage_error = exc + await db.rollback() + logger.exception("prompt optimize provider result staging failed: record_id=%s", record_id) + if not staged: + log_operation_error( + domain=_LOG_DOMAIN, + event_type=GenerationRecordEventTypeEnum.PROMPT_OPTIMIZE_PROVIDER_RESULT_STAGED.value, + module=_LOG_MODULE, + source=_LOG_SOURCE, + user_id=user_id, + project_id=project_id, + task_id=record_id, + detail={"attempt_no": _PROMPT_ATTEMPT_NO, "stage": "provider_result_persistence"}, + exc=last_stage_error or RuntimeError("unknown staging failure"), + ) + raise HTTPException(status_code=503, detail="提词已生成但本地暂存失败，请联系管理员根据模型日志处理") + + _log_event( + GenerationRecordEventTypeEnum.PROMPT_OPTIMIZE_PROVIDER_RESULT_STAGED, + user_id=user_id, + project_id=project_id, + record_id=record_id, + detail={ + "attempt_no": _PROMPT_ATTEMPT_NO, + "input_tokens": usage_snapshot.get("input_tokens"), + "output_tokens": usage_snapshot.get("output_tokens"), + "total_tokens": usage_snapshot.get("total_tokens"), + }, + ) + return await _settle_staged_result( + db, + record_id=record_id, + user_id=user_id, + project_name=project_name, + request_id=req.idempotency_key, + ) diff --git a/video-gen-api/app/services/generation/refund_service.py b/video-gen-api/app/services/generation/refund_service.py index 9274a41d..9a14b80c 100644 --- a/video-gen-api/app/services/generation/refund_service.py +++ b/video-gen-api/app/services/generation/refund_service.py @@ -8,7 +8,7 @@ from app.models.chat_generation_task import ChatGenerationTask from app.models.credit_record import CreditRecord from app.models.generation_record import GenerationRecord from app.services.generation.pipeline.db_lock_service import execute_with_lock_timeout -from app.services.credits import refund_credits +from app.services.credits import add_credits_result from app.services.credit_record_meta_service import build_refund_meta_from_charge from app.services.generation.billing_service import ( CHARGE_MEDIA, @@ -113,17 +113,19 @@ async def refund_unrefunded_media_charges( amount = abs(_round2(charge.amount)) if amount <= 0: continue - await refund_credits( + mutation = await add_credits_result( db, user_id=user_id, amount=amount, description=f"{description_prefix}失败积分回退", related_id=owner_id, + record_type="refund", biz_key=refund_biz_key, refund_for_biz_key=charge.biz_key, record_meta=build_refund_meta_from_charge(charge, attempt_no=attempt_no), ) - total_refunded = round(total_refunded + amount, 2) + if mutation.created: + total_refunded = round(total_refunded + mutation.amount, 2) return total_refunded diff --git a/video-gen-api/app/services/hot_opening_replicate_service.py b/video-gen-api/app/services/hot_opening_replicate_service.py index b8476710..94b4dfa8 100644 --- a/video-gen-api/app/services/hot_opening_replicate_service.py +++ b/video-gen-api/app/services/hot_opening_replicate_service.py @@ -370,6 +370,16 @@ async def _soft_delete_steps_from_index( def _step_to_out(step: ModuleGenerationStep) -> HotOpeningStepOut: + error_message = step.error_message + if error_message: + from app.services.error_codes import ARK_ERRORS + import re + # 兼容两种格式: code='xxx' 或 'code': 'xxx' + match = re.search(r"""code['"]?\s*[:=]\s*['"]([^'"]+)['"]""", error_message) + if match: + code = match.group(1) + if code in ARK_ERRORS: + error_message = ARK_ERRORS[code] return HotOpeningStepOut( id=step.id, project_id=step.project_id, @@ -384,7 +394,7 @@ def _step_to_out(step: ModuleGenerationStep) -> HotOpeningStepOut: chat_task_id=step.chat_task_id, input=_parse_json(step.input_json, {}), output=_parse_json(step.output_json, {}), - error_message=step.error_message, + error_message=error_message, created_at=step.created_at, updated_at=step.updated_at, completed_at=step.completed_at, @@ -1152,10 +1162,10 @@ async def run_image_prompt_optimize( await db.commit() return None step.status = ModuleStepStatusEnum.FAILED.value - step.error_message = str(exc) + step.error_message = str(exc) if str(exc) else type(exc).__name__ step.completed_at = _now() project.status = ModuleProjectStatusEnum.FAILED.value - project.error_message = f"图片 AI 提词生成失败: {exc}" + project.error_message = f"图片 AI 提词生成失败: {str(exc) if str(exc) else type(exc).__name__}" log_module_prompt_event( event_type="module_prompt_error", project_id=project.id, @@ -1623,10 +1633,10 @@ async def run_video_prompt_optimize( await db.commit() return None step.status = ModuleStepStatusEnum.FAILED.value - step.error_message = str(exc) + step.error_message = str(exc) if str(exc) else type(exc).__name__ step.completed_at = _now() project.status = ModuleProjectStatusEnum.FAILED.value - project.error_message = f"视频 AI 提词生成失败: {exc}" + project.error_message = f"视频 AI 提词生成失败: {str(exc) if str(exc) else type(exc).__name__}" log_module_prompt_event( event_type="module_prompt_error", project_id=project.id, diff --git a/video-gen-api/app/services/hot_opening_video_prompt_service.py b/video-gen-api/app/services/hot_opening_video_prompt_service.py index 7d6d3ada..98445d03 100644 --- a/video-gen-api/app/services/hot_opening_video_prompt_service.py +++ b/video-gen-api/app/services/hot_opening_video_prompt_service.py @@ -25,9 +25,8 @@ from app.enums.common import ( ) from app.enums.video_prompt_schema import PromptSchemaVersionEnum, VideoPromptSchemaUsageEnum from app.models.model_config import ModelConfig -from app.models.token_usage import TokenUsage -from app.utils.id_gen import generate_id from app.services.resource_signed_url_service import build_resource_signed_url +from app.utils.id_gen import generate_id DEFAULT_FRAME_RATE = "30fps" DEFAULT_REFERENCE_VIDEO_FPS = 1 @@ -1599,7 +1598,13 @@ async def optimize_hot_opening_video_prompt( await db.rollback() if not config: result = normalize_video_prompt_schema_from_ai(_mock_result(video_config, target_platform), video_config, schema_config_snapshot) - return result, build_final_video_prompt(result), {"input_tokens": 0, "output_tokens": 0, "total_tokens": 0} + return result, build_final_video_prompt(result), { + "input_tokens": 0, + "output_tokens": 0, + "total_tokens": 0, + "usage_reported": True, + "billing_free": True, + } if use_base64: video_url_final = await media_to_base64(material_video_url, "video/mp4") @@ -1628,7 +1633,7 @@ async def optimize_hot_opening_video_prompt( request_data = { "model": config.model_name, "messages": [{"role": "system", "content": build_system_prompt()}, user_message], - "max_tokens": 6000, + "max_tokens": 10000, "temperature": 0.15, "response_format": {"type": "json_object"}, } @@ -1739,27 +1744,20 @@ async def optimize_hot_opening_video_prompt( error=str(exc), ) raise - usage = data.get("usage", {}) or {} + raw_usage = data.get("usage") + usage_reported = bool( + isinstance(raw_usage, dict) + and any(key in raw_usage for key in ("prompt_tokens", "completion_tokens", "total_tokens")) + ) + usage = raw_usage if isinstance(raw_usage, dict) else {} token_usage = { "input_tokens": int(usage.get("prompt_tokens") or 0), "output_tokens": int(usage.get("completion_tokens") or 0), "total_tokens": int(usage.get("total_tokens") or 0), + "usage_reported": usage_reported, # "log_user_message": log_user_message, } - token_usage_id = generate_id() - db.add( - TokenUsage( - id=token_usage_id, - model_config_id=config.id, - user_id=user_id, - input_tokens=token_usage["input_tokens"], - output_tokens=token_usage["output_tokens"], - total_tokens=token_usage["total_tokens"], - ) - ) - await db.flush() token_usage.update({ - "token_usage_id": token_usage_id, "model_config_id": config.id, "model_config_name": config.name, "model_provider": config.provider, diff --git a/video-gen-api/app/services/llm.py b/video-gen-api/app/services/llm.py index e321812e..fe7ac339 100644 --- a/video-gen-api/app/services/llm.py +++ b/video-gen-api/app/services/llm.py @@ -8,9 +8,8 @@ from sqlalchemy.ext.asyncio import AsyncSession from app.config import settings from app.models.model_config import ModelConfig -from app.models.token_usage import TokenUsage -from app.utils.id_gen import generate_id from app.services.operation_log_service import build_exception_detail, log_ai_model_event +from app.utils.id_gen import generate_id @@ -47,7 +46,13 @@ def _get_default_prompt(prompt: str, gen_type: str = "video") -> tuple[str, dict f"smooth camera movements. Theme: {prompt}. Cinematic shooting techniques with " f"rich lighting layers and strong visual impact, suitable for commercial distribution." ) - return optimized, {"input_tokens": 0, "output_tokens": 0, "total_tokens": 0} + return optimized, { + "input_tokens": 0, + "output_tokens": 0, + "total_tokens": 0, + "usage_reported": True, + "billing_free": True, + } async def optimize_prompt( @@ -135,7 +140,13 @@ def _mock_optimize(prompt: str, gen_type: str = "video") -> tuple[str, dict]: """Return a keyword-matched mock optimized prompt.""" for keyword, optimized in MOCK_OPTIMIZED_PROMPTS.items(): if keyword in prompt: - return optimized, {"input_tokens": 0, "output_tokens": 0, "total_tokens": 0} + return optimized, { + "input_tokens": 0, + "output_tokens": 0, + "total_tokens": 0, + "usage_reported": True, + "billing_free": True, + } return _get_default_prompt(prompt, gen_type) @@ -425,7 +436,15 @@ async def _call_openai_compatible( raise LLMProviderCallError(f"{type(exc).__name__}: {exc}") from exc try: - usage = data.get("usage", {}) + raw_usage = data.get("usage") + usage_reported = bool( + isinstance(raw_usage, dict) + and any( + key in raw_usage + for key in ("prompt_tokens", "completion_tokens", "total_tokens") + ) + ) + usage = raw_usage if isinstance(raw_usage, dict) else {} input_tokens = int(usage.get("prompt_tokens", 0) or 0) output_tokens = int(usage.get("completion_tokens", 0) or 0) total_tokens = int(usage.get("total_tokens", input_tokens + output_tokens) or 0) @@ -444,40 +463,7 @@ async def _call_openai_compatible( ) raise LLMProviderCallError(f"模型响应解析失败: {exc}") from exc - token_usage_id = None - if db is not None: - try: - token_usage_id = generate_id() - record = TokenUsage( - id=token_usage_id, - model_config_id=config.id, - user_id=user_id, - input_tokens=input_tokens, - output_tokens=output_tokens, - total_tokens=total_tokens, - source_module=log_module, - source_step_code=log_step, - owner_type=log_owner_type, - owner_id=log_owner_id, - ) - db.add(record) - await db.flush() - except Exception as exc: - log_ai_model_event( - event_type="ERROR", - event_phase="ERROR", - event_status="failed", - latency_ms=int((time.perf_counter() - started) * 1000), - detail=build_exception_detail(exc, {"stage": "token_usage_persistence"}), - error=str(exc), - **common_log, - ) - # A local transaction failure must not call a second provider after - # the first provider has already returned a valid response. - raise - token_usage = { - "token_usage_id": token_usage_id, "model_config_id": config.id, "model_config_name": config.name, "model_provider": config.provider, @@ -487,5 +473,6 @@ async def _call_openai_compatible( "input_tokens": input_tokens, "output_tokens": output_tokens, "total_tokens": total_tokens, + "usage_reported": usage_reported, } return content, token_usage diff --git a/video-gen-api/app/services/llm_billing/__init__.py b/video-gen-api/app/services/llm_billing/__init__.py index e83b01f3..5e49ee5d 100644 --- a/video-gen-api/app/services/llm_billing/__init__.py +++ b/video-gen-api/app/services/llm_billing/__init__.py @@ -19,6 +19,7 @@ from app.services.llm_billing.service import ( release_on_failure, settle_success, start_hold, + validate_retryable_previous_attempt, ) __all__ = [ @@ -31,6 +32,7 @@ __all__ = [ "start_hold", "ensure_hold_exists", "get_llm_ledger_states", + "validate_retryable_previous_attempt", "log_provider_start", "log_provider_success", "log_provider_failure", diff --git a/video-gen-api/app/services/llm_billing/service.py b/video-gen-api/app/services/llm_billing/service.py index 2e663d2d..21860a2f 100644 --- a/video-gen-api/app/services/llm_billing/service.py +++ b/video-gen-api/app/services/llm_billing/service.py @@ -4,6 +4,7 @@ from dataclasses import dataclass from typing import Any, Iterable, Mapping from sqlalchemy import select +from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession from app.enums.credit_record import ( @@ -210,8 +211,21 @@ def _action_valid(record: CreditRecord | None, expected: CreditRecordAction) -> +def _record_meta_matches_context(record: CreditRecord | None, ctx: LlmBillingContext) -> bool: + if record is None: + return True + checks = ( + (record.owner_type, ctx.owner_type), + (record.owner_id, ctx.owner_id), + (record.attempt_no, ctx.attempt_no), + (record.charge_kind, ctx.charge_kind), + ) + return all(actual is None or str(actual) == str(expected) for actual, expected in checks) + + def _classify_ledger( *, + ctx: LlmBillingContext, hold: CreditRecord | None, release: CreditRecord | None, charge: CreditRecord | None, @@ -222,17 +236,39 @@ def _classify_ledger( return _LedgerRecords(LlmBillingLedgerState.INVALID, hold, release, charge, "release_action_mismatch") if not _action_valid(charge, CreditRecordAction.CHARGE): return _LedgerRecords(LlmBillingLedgerState.INVALID, hold, release, charge, "charge_action_mismatch") + if not _record_meta_matches_context(hold, ctx): + return _LedgerRecords(LlmBillingLedgerState.INVALID, hold, release, charge, "hold_context_mismatch") + if not _record_meta_matches_context(release, ctx): + return _LedgerRecords(LlmBillingLedgerState.INVALID, hold, release, charge, "release_context_mismatch") + if not _record_meta_matches_context(charge, ctx): + return _LedgerRecords(LlmBillingLedgerState.INVALID, hold, release, charge, "charge_context_mismatch") if hold is None: if release is not None or charge is not None: return _LedgerRecords(LlmBillingLedgerState.INVALID, hold, release, charge, "hold_missing_with_followup") return _LedgerRecords(LlmBillingLedgerState.MISSING) if release is None and charge is None: + if hold.type != "consume" or _round2(float(hold.amount or 0)) >= 0: + return _LedgerRecords(LlmBillingLedgerState.INVALID, hold, release, charge, "hold_type_or_sign_invalid") if _round2(abs(float(hold.amount or 0))) <= 0: return _LedgerRecords(LlmBillingLedgerState.INVALID, hold, release, charge, "hold_amount_not_positive") return _LedgerRecords(LlmBillingLedgerState.ACTIVE, hold) if release is not None and charge is None: + if release.type != "refund" or _round2(float(release.amount or 0)) <= 0: + return _LedgerRecords(LlmBillingLedgerState.INVALID, hold, release, charge, "release_type_or_sign_invalid") + if str(release.refund_for_biz_key or "") != str(ctx.hold_biz_key): + return _LedgerRecords(LlmBillingLedgerState.INVALID, hold, release, charge, "release_target_mismatch") + if _round2(release.amount) != _round2(abs(float(hold.amount or 0))): + return _LedgerRecords(LlmBillingLedgerState.INVALID, hold, release, charge, "release_amount_mismatch") return _LedgerRecords(LlmBillingLedgerState.RELEASED, hold, release) if release is not None and charge is not None: + if release.type != "refund" or _round2(float(release.amount or 0)) <= 0: + return _LedgerRecords(LlmBillingLedgerState.INVALID, hold, release, charge, "release_type_or_sign_invalid") + if str(release.refund_for_biz_key or "") != str(ctx.hold_biz_key): + return _LedgerRecords(LlmBillingLedgerState.INVALID, hold, release, charge, "release_target_mismatch") + if _round2(release.amount) != _round2(abs(float(hold.amount or 0))): + return _LedgerRecords(LlmBillingLedgerState.INVALID, hold, release, charge, "release_amount_mismatch") + if charge.type != "consume" or _round2(float(charge.amount or 0)) > 0: + return _LedgerRecords(LlmBillingLedgerState.INVALID, hold, release, charge, "charge_type_or_sign_invalid") return _LedgerRecords(LlmBillingLedgerState.CHARGED, hold, release, charge) return _LedgerRecords(LlmBillingLedgerState.INVALID, hold, release, charge, "charge_without_release") @@ -264,7 +300,7 @@ async def _load_ledgers( hold = record_map.get((ctx.user_id, ctx.hold_biz_key)) release = record_map.get((ctx.user_id, ctx.hold_release_biz_key)) charge = record_map.get((ctx.user_id, ctx.charge_biz_key)) - output[ctx.hold_biz_key] = _classify_ledger(hold=hold, release=release, charge=charge) + output[ctx.hold_biz_key] = _classify_ledger(ctx=ctx, hold=hold, release=release, charge=charge) return output @@ -291,6 +327,104 @@ async def get_llm_ledger_states( } +async def validate_retryable_previous_attempt( + db: AsyncSession, + ctx: LlmBillingContext, +) -> LlmHoldValidation: + """校验失败 attempt 的账务是否已关闭，供业务创建下一 attempt 前调用。 + + 只有已释放 HOLD，或当前计费明确关闭且旧 attempt 没有流水时，才允许创建 + 新 attempt。这里不主动退款，避免重试入口承担失败补偿职责。 + """ + _log( + ctx, + LlmBillingEvent.RETRY_PREVIOUS_ATTEMPT_VALIDATE_START, + status="started", + ) + ledger = await _load_ledger(db, ctx) + + if ledger.state == LlmBillingLedgerState.RELEASED: + result = LlmHoldValidation( + True, + ledger.hold_amount, + ledger.state, + hold_record_id=ledger.hold.id if ledger.hold else None, + ) + _log( + ctx, + LlmBillingEvent.RETRY_PREVIOUS_ATTEMPT_VALIDATE_SUCCESS, + detail=_context_detail( + ctx, + ledger_state=result.state.value, + hold_credits=result.amount, + hold_record_id=result.hold_record_id, + ), + ) + return result + + if ledger.state == LlmBillingLedgerState.MISSING: + policy = await get_llm_billing_policy( + db, + config_key=ctx.hold_config_key, + explicit_hold_credits=None, + ) + if policy.bypassed: + result = LlmHoldValidation( + True, + 0.0, + LlmBillingLedgerState.BILLING_BYPASSED, + "billing_disabled", + ) + _log( + ctx, + LlmBillingEvent.RETRY_PREVIOUS_ATTEMPT_VALIDATE_SUCCESS, + status="skipped", + detail=_context_detail( + ctx, + ledger_state=result.state.value, + billing_bypassed=True, + ), + ) + return result + if not policy.valid: + result = LlmHoldValidation( + False, + 0.0, + LlmBillingLedgerState.INVALID, + policy.error or "billing_config_invalid", + ) + else: + result = LlmHoldValidation( + False, + 0.0, + ledger.state, + "previous_attempt_hold_missing", + ) + else: + result = LlmHoldValidation( + False, + ledger.hold_amount, + ledger.state, + ledger.reason or f"previous_attempt_ledger_{ledger.state.value}", + ledger.hold.id if ledger.hold else None, + ) + + _log( + ctx, + LlmBillingEvent.RETRY_PREVIOUS_ATTEMPT_BLOCKED, + status="failed", + detail=_context_detail( + ctx, + ledger_state=result.state.value, + hold_credits=result.amount, + hold_record_id=result.hold_record_id, + skip_reason=result.reason, + ), + error="旧attempt账务尚未关闭，拒绝创建新的分析attempt", + ) + return result + + async def start_hold(db: AsyncSession, ctx: LlmBillingContext) -> LlmHoldResult: # 幂等/异常 attempt 优先由已落库流水判定；只有全新 attempt 才读取配置。 ledger = await _load_ledger(db, ctx) @@ -659,6 +793,147 @@ async def release_on_failure(db: AsyncSession, ctx: LlmBillingContext, *, error: return BillingSummary(record_id=ctx.owner_id, user_id=ctx.user_id, items=[item]) +def _normalize_settlement_usage( + ctx: LlmBillingContext, + usage: Mapping[str, Any], +) -> dict[str, Any]: + snapshot = dict(usage or {}) + try: + if snapshot.get("usage_reported") is False and not bool(snapshot.get("billing_free")): + raise ValueError("provider_usage_missing") + if "input_tokens" not in snapshot or "output_tokens" not in snapshot: + raise ValueError("input_or_output_tokens_missing") + input_tokens = int(snapshot.get("input_tokens")) + output_tokens = int(snapshot.get("output_tokens")) + if input_tokens < 0 or output_tokens < 0: + raise ValueError("token_count_negative") + normalized_total = input_tokens + output_tokens + raw_total = snapshot.get("total_tokens") + if raw_total not in (None, ""): + total_tokens = int(raw_total) + if total_tokens < 0: + raise ValueError("total_tokens_negative") + else: + total_tokens = normalized_total + if total_tokens != normalized_total: + snapshot["reported_total_tokens"] = total_tokens + total_tokens = normalized_total + snapshot["input_tokens"] = input_tokens + snapshot["output_tokens"] = output_tokens + snapshot["total_tokens"] = total_tokens + return snapshot + except (TypeError, ValueError) as exc: + _log( + ctx, + LlmBillingEvent.USAGE_INVALID, + status="failed", + detail=_context_detail(ctx, usage=snapshot, reason=str(exc)), + error="LLM usage 无效，拒绝释放 HOLD 和创建真实扣费", + ) + raise LlmBillingStateError(f"LLM usage 无效：{exc}") from exc + + +async def _ensure_token_usage_once( + db: AsyncSession, + ctx: LlmBillingContext, + usage: dict[str, Any], +) -> TokenUsage: + supplied_id = str(usage.get("token_usage_id") or "").strip() or None + if supplied_id: + supplied_result = await db.execute( + select(TokenUsage).where(TokenUsage.id == supplied_id).limit(1) + ) + supplied = supplied_result.scalar_one_or_none() + if supplied is not None: + if supplied.user_id not in (None, ctx.user_id): + raise LlmBillingStateError("TokenUsage 用户归属与当前账务上下文不一致") + if supplied.biz_key not in (None, ctx.charge_biz_key): + raise LlmBillingStateError("TokenUsage biz_key 与当前 charge 不一致") + supplied.user_id = supplied.user_id or ctx.user_id + supplied.owner_type = supplied.owner_type or ctx.owner_type + supplied.owner_id = supplied.owner_id or ctx.owner_id + supplied.biz_key = supplied.biz_key or ctx.charge_biz_key + supplied.source_module = supplied.source_module or ctx.source_module + supplied.source_step_code = supplied.source_step_code or ctx.source_step_code + usage["token_usage_id"] = supplied.id + ctx.token_usage_id = supplied.id + _log( + ctx, + LlmBillingEvent.TOKEN_USAGE_REUSED, + detail=_context_detail(ctx, token_usage_id=supplied.id, source="supplied_id"), + ) + return supplied + + result = await db.execute( + select(TokenUsage) + .where( + TokenUsage.user_id == ctx.user_id, + TokenUsage.biz_key == ctx.charge_biz_key, + ) + .limit(1) + ) + existing = result.scalar_one_or_none() + if existing is not None: + usage["token_usage_id"] = existing.id + ctx.token_usage_id = existing.id + _log( + ctx, + LlmBillingEvent.TOKEN_USAGE_REUSED, + detail=_context_detail(ctx, token_usage_id=existing.id, source="biz_key"), + ) + return existing + + token_usage = TokenUsage( + id=generate_id(), + model_config_id=usage.get("model_config_id"), + user_id=ctx.user_id, + input_tokens=int(usage["input_tokens"]), + output_tokens=int(usage["output_tokens"]), + total_tokens=int(usage["total_tokens"]), + owner_type=ctx.owner_type, + owner_id=ctx.owner_id, + biz_key=ctx.charge_biz_key, + source_module=ctx.source_module, + source_step_code=ctx.source_step_code, + ) + try: + async with db.begin_nested(): + db.add(token_usage) + await db.flush() + except IntegrityError: + if token_usage in db.sync_session: + db.sync_session.expunge(token_usage) + result = await db.execute( + select(TokenUsage) + .where( + TokenUsage.user_id == ctx.user_id, + TokenUsage.biz_key == ctx.charge_biz_key, + ) + .limit(1) + ) + token_usage = result.scalar_one_or_none() + if token_usage is None: + raise + event = LlmBillingEvent.TOKEN_USAGE_REUSED + else: + event = LlmBillingEvent.TOKEN_USAGE_CREATED + + usage["token_usage_id"] = token_usage.id + ctx.token_usage_id = token_usage.id + _log( + ctx, + event, + detail=_context_detail( + ctx, + token_usage_id=token_usage.id, + input_tokens=token_usage.input_tokens, + output_tokens=token_usage.output_tokens, + total_tokens=token_usage.total_tokens, + ), + ) + return token_usage + + async def _build_charge_meta(db: AsyncSession, ctx: LlmBillingContext, usage: Mapping[str, Any]) -> CreditRecordMeta: usage_snapshot = dict(usage or {}) ctx.provider = str(usage_snapshot.get("provider") or usage_snapshot.get("model_provider") or "") or ctx.provider @@ -680,26 +955,6 @@ async def _build_charge_meta(db: AsyncSession, ctx: LlmBillingContext, usage: Ma usage=usage_snapshot, ) if ctx.charge_kind == CreditRecordChargeKind.VIDEO_ANALYSIS.value: - if not usage_snapshot.get("token_usage_id"): - input_tokens = _safe_int(usage_snapshot.get("input_tokens")) - output_tokens = _safe_int(usage_snapshot.get("output_tokens")) - token_usage = TokenUsage( - id=generate_id(), - model_config_id=usage_snapshot.get("model_config_id"), - user_id=ctx.user_id, - input_tokens=input_tokens, - output_tokens=output_tokens, - total_tokens=_safe_int(usage_snapshot.get("total_tokens"), input_tokens + output_tokens), - owner_type=ctx.owner_type, - owner_id=ctx.owner_id, - biz_key=ctx.charge_biz_key, - source_module=ctx.source_module or CreditRecordSourceModule.SHOT_REPLICATE.value, - source_step_code=ctx.source_step_code, - ) - db.add(token_usage) - await db.flush() - usage_snapshot["token_usage_id"] = token_usage.id - ctx.token_usage_id = token_usage.id return await build_shot_video_analysis_meta( db, owner_type=ctx.owner_type, @@ -791,12 +1046,27 @@ async def _settle_success_impl( f"当前attempt账务状态为{ledger.state.value}，不能执行成功结算" ) - _log(ctx, LlmBillingEvent.SETTLE_START, status="started", detail=_context_detail(ctx, ledger_state=ledger.state.value, usage=dict(usage or {}))) - release_item = await _release_active_hold(db, ctx, hold_record=ledger.hold, reason="success") - input_tokens = _safe_int((usage or {}).get("input_tokens")) - output_tokens = _safe_int((usage or {}).get("output_tokens")) + normalized_usage = _normalize_settlement_usage(ctx, usage) + await _ensure_token_usage_once(db, ctx, normalized_usage) + _log( + ctx, + LlmBillingEvent.SETTLE_START, + status="started", + detail=_context_detail( + ctx, + ledger_state=ledger.state.value, + input_tokens=normalized_usage["input_tokens"], + output_tokens=normalized_usage["output_tokens"], + total_tokens=normalized_usage["total_tokens"], + provider=normalized_usage.get("provider") or normalized_usage.get("model_provider"), + model_name=normalized_usage.get("model_name") or normalized_usage.get("model"), + ), + ) + input_tokens = int(normalized_usage["input_tokens"]) + output_tokens = int(normalized_usage["output_tokens"]) amount = await calc_text_credits(db, input_tokens, output_tokens) - meta = await _build_charge_meta(db, ctx, usage) + meta = await _build_charge_meta(db, ctx, normalized_usage) + release_item = await _release_active_hold(db, ctx, hold_record=ledger.hold, reason="success") if meta.charge_action is None: meta.charge_action = CreditRecordAction.CHARGE.value meta.billing_scene = meta.billing_scene or ctx.billing_scene @@ -831,7 +1101,7 @@ async def _settle_success_impl( step = result.scalar_one_or_none() if step: step.token_usage_id = meta.token_usage_id - step.model_config_id = (usage or {}).get("model_config_id") + step.model_config_id = normalized_usage.get("model_config_id") step.input_tokens = meta.input_tokens step.output_tokens = meta.output_tokens step.total_tokens = meta.total_tokens diff --git a/video-gen-api/app/services/module_generation_v2/flow_service.py b/video-gen-api/app/services/module_generation_v2/flow_service.py index ed5d7d86..0be13454 100644 --- a/video-gen-api/app/services/module_generation_v2/flow_service.py +++ b/video-gen-api/app/services/module_generation_v2/flow_service.py @@ -1066,10 +1066,10 @@ async def run_video_prompt_optimize_v2( project, step = row if step.status == ModuleStepStatusEnum.PROCESSING.value: step.status = ModuleStepStatusEnum.FAILED.value - step.error_message = str(exc) + step.error_message = str(exc) if str(exc) else type(exc).__name__ step.completed_at = utc_now() project.status = ModuleProjectStatusEnum.FAILED.value - project.error_message = str(exc) + project.error_message = str(exc) if str(exc) else type(exc).__name__ # provider 已成功时 settle_success 已在当前事务写入 RELEASE/CHARGE； # 即使业务步骤已不存在或已不是 processing，也必须提交账务结算。 await db.commit() diff --git a/video-gen-api/app/services/shot_replicate_flow_service.py b/video-gen-api/app/services/shot_replicate_flow_service.py index 8ac13c50..bfe232f8 100644 --- a/video-gen-api/app/services/shot_replicate_flow_service.py +++ b/video-gen-api/app/services/shot_replicate_flow_service.py @@ -381,6 +381,16 @@ async def _soft_delete_steps_from_index( def _step_to_out(step: ModuleGenerationStep) -> ShotReplicateStepOut: + error_message = step.error_message + if error_message: + from app.services.error_codes import ARK_ERRORS + import re + # 兼容两种格式: code='xxx' 或 'code': 'xxx' + match = re.search(r"""code['"]?\s*[:=]\s*['"]([^'"]+)['"]""", error_message) + if match: + code = match.group(1) + if code in ARK_ERRORS: + error_message = ARK_ERRORS[code] return ShotReplicateStepOut( id=step.id, project_id=step.project_id, @@ -395,7 +405,7 @@ def _step_to_out(step: ModuleGenerationStep) -> ShotReplicateStepOut: chat_task_id=step.chat_task_id, input=_parse_json(step.input_json, {}), output=_parse_json(step.output_json, {}), - error_message=step.error_message, + error_message=error_message, created_at=step.created_at, updated_at=step.updated_at, completed_at=step.completed_at, @@ -1100,10 +1110,10 @@ async def run_image_prompt_optimize( await db.commit() return None step.status = ModuleStepStatusEnum.FAILED.value - step.error_message = str(exc) + step.error_message = str(exc) if str(exc) else type(exc).__name__ step.completed_at = _now() project.status = ModuleProjectStatusEnum.FAILED.value - project.error_message = f"图片 AI 提词生成失败: {exc}" + project.error_message = f"图片 AI 提词生成失败: {str(exc) if str(exc) else type(exc).__name__}" log_module_prompt_event( event_type="module_prompt_error", project_id=project.id, @@ -1581,10 +1591,10 @@ async def run_video_prompt_optimize( await db.commit() return None step.status = ModuleStepStatusEnum.FAILED.value - step.error_message = str(exc) + step.error_message = str(exc) if str(exc) else type(exc).__name__ step.completed_at = _now() project.status = ModuleProjectStatusEnum.FAILED.value - project.error_message = f"视频 AI 提词生成失败: {exc}" + project.error_message = f"视频 AI 提词生成失败: {str(exc) if str(exc) else type(exc).__name__}" log_module_prompt_event( event_type="module_prompt_error", project_id=project.id, diff --git a/video-gen-api/app/services/shot_replicate_recovery_service.py b/video-gen-api/app/services/shot_replicate_recovery_service.py index 3a9d3739..ff828bb9 100644 --- a/video-gen-api/app/services/shot_replicate_recovery_service.py +++ b/video-gen-api/app/services/shot_replicate_recovery_service.py @@ -327,7 +327,7 @@ async def recover_shot_analysis_tasks_once(db: AsyncSession) -> dict[str, Any]: try: if owner_type == "task_set": analyze_original_video.apply_async( - args=[owner_id], + args=[owner_id, attempt], queue=CeleryQueue.GEN_SHOT_ANALYSIS.value, countdown=0, priority=settings.DOWNLOAD_TASK_PRIORITY_RECOVER, @@ -335,7 +335,7 @@ async def recover_shot_analysis_tasks_once(db: AsyncSession) -> dict[str, Any]: ) else: analyze_custom_segment_video.apply_async( - args=[owner_id], + args=[owner_id, attempt], queue=CeleryQueue.GEN_SHOT_ANALYSIS.value, countdown=0, priority=settings.DOWNLOAD_TASK_PRIORITY_RECOVER, diff --git a/video-gen-api/app/services/shot_replicate_taskset_service.py b/video-gen-api/app/services/shot_replicate_taskset_service.py index 107a0366..ea30cd4a 100644 --- a/video-gen-api/app/services/shot_replicate_taskset_service.py +++ b/video-gen-api/app/services/shot_replicate_taskset_service.py @@ -54,7 +54,12 @@ from app.schemas.shot_replicate import ( ShotTaskSetOut, ) from app.services.module_generation_log_service import log_module_event_file -from app.services.llm_billing import LlmBillingContext, release_on_failure, start_hold +from app.services.llm_billing import ( + LlmBillingContext, + release_on_failure, + start_hold, + validate_retryable_previous_attempt, +) from app.services.resource_accounting_service import SOURCE_MODEL_SHOT_SEGMENT, soft_delete_resources_by_source from app.enums.upload_resource import UploadResourceModuleEnum, UploadResourceSourceModelEnum from app.services.upload_resource import release_upload_resources_by_source @@ -398,10 +403,15 @@ async def _next_segment_index(db: AsyncSession, task_set_id: str) -> int: return int(result.scalar() or 0) + 1 -async def refresh_task_set_split_summaries(db: AsyncSession, task_set_ids: set[str] | list[str]) -> None: +async def refresh_task_set_split_summaries( + db: AsyncSession, + task_set_ids: set[str] | list[str], + *, + log_changes: bool = True, +) -> list[dict[str, Any]]: ids = sorted({str(item) for item in task_set_ids if item}) if not ids: - return + return [] task_set_result = await db.execute( select(ShotReplicateTaskSet) @@ -411,7 +421,11 @@ async def refresh_task_set_split_summaries(db: AsyncSession, task_set_ids: set[s ) task_sets = list(task_set_result.scalars().all()) if not task_sets: - return + return [] + + # 项目 AsyncSession 关闭了 autoflush。聚合查询前必须把当前事务中刚修改的 + # segment.split_status/deleted_at 等字段落到数据库，否则最后一个片段会少统计一次。 + await db.flush() count_result = await db.execute( select( @@ -441,14 +455,18 @@ async def refresh_task_set_split_summaries(db: AsyncSession, task_set_ids: set[s for row in count_result.all() } + changes: list[dict[str, Any]] = [] for task_set in task_sets: total, completed, failed = count_map.get(str(task_set.id), (0, 0, 0)) + old_status = task_set.status + old_split_status = task_set.split_status + old_total = int(task_set.segment_count or 0) + old_completed = int(task_set.completed_segment_count or 0) + old_failed = int(task_set.failed_segment_count or 0) task_set.segment_count = total task_set.completed_segment_count = completed task_set.failed_segment_count = failed - old_status = task_set.status - old_split_status = task_set.split_status if total <= 0: task_set.split_status = ShotSplitStatusEnum.NONE.value if task_set.analysis_status == ShotAnalysisStatusEnum.COMPLETED.value: @@ -466,24 +484,39 @@ async def refresh_task_set_split_summaries(db: AsyncSession, task_set_ids: set[s task_set.split_status = ShotSplitStatusEnum.PROCESSING.value task_set.status = ShotTaskSetStatusEnum.SPLITTING.value - if old_status != task_set.status or old_split_status != task_set.split_status: - log_module_event_file( - module=MODULE, - event_type="SHOT_SPLIT_STATUS_CHANGED", - project_id=task_set.id, - user_id=task_set.user_id, - message="拆镜总任务集拆分状态变更", - detail={ - "task_set_id": task_set.id, - "from_status": old_status, - "to_status": task_set.status, - "from_split_status": old_split_status, - "to_split_status": task_set.split_status, - "segment_count": total, - "completed_segment_count": completed, - "failed_segment_count": failed, - }, - ) + changed = ( + old_status != task_set.status + or old_split_status != task_set.split_status + or old_total != total + or old_completed != completed + or old_failed != failed + ) + if changed: + change = { + "task_set_id": str(task_set.id), + "user_id": str(task_set.user_id), + "from_status": old_status, + "to_status": task_set.status, + "from_split_status": old_split_status, + "to_split_status": task_set.split_status, + "from_segment_count": old_total, + "segment_count": total, + "from_completed_segment_count": old_completed, + "completed_segment_count": completed, + "from_failed_segment_count": old_failed, + "failed_segment_count": failed, + } + changes.append(change) + if log_changes: + log_module_event_file( + module=MODULE, + event_type=ShotReplicateLogEventEnum.SPLIT_STATUS_CHANGED.value, + project_id=task_set.id, + user_id=task_set.user_id, + message="拆镜总任务集拆分状态变更", + detail=change, + ) + return changes async def refresh_task_set_split_summary(db: AsyncSession, task_set_id: str) -> None: @@ -511,20 +544,20 @@ async def create_segments_by_ai( if not suggestions: raise HTTPException(status_code=400, detail="selected_indices 没有匹配到可用 AI 建议") - old_result = await db.execute( - select(ShotReplicateSegment).where( - ShotReplicateSegment.task_set_id == task_set.id, - ShotReplicateSegment.source_mode == ShotSegmentSourceModeEnum.AI_SUGGESTION.value, - ShotReplicateSegment.deleted_at.is_(None), - ) - ) - old_segments = list(old_result.scalars().all()) - if old_segments and not req.replace_existing: - raise HTTPException(status_code=409, detail="已存在 AI 建议拆镜片段，如需重拆请传 replace_existing=true") - if old_segments and req.replace_existing: - now = _now() - for segment in old_segments: - segment.deleted_at = now + # old_result = await db.execute( + # select(ShotReplicateSegment).where( + # ShotReplicateSegment.task_set_id == task_set.id, + # ShotReplicateSegment.source_mode == ShotSegmentSourceModeEnum.AI_SUGGESTION.value, + # ShotReplicateSegment.deleted_at.is_(None), + # ) + # ) + # old_segments = list(old_result.scalars().all()) + # if old_segments and not req.replace_existing: + # raise HTTPException(status_code=409, detail="已存在 AI 建议拆镜片段，如需重拆请传 replace_existing=true") + # if old_segments and req.replace_existing: + # now = _now() + # for segment in old_segments: + # segment.deleted_at = now created: list[ShotReplicateSegment] = [] next_index = await _next_segment_index(db, task_set.id) @@ -576,7 +609,7 @@ async def create_segments_by_ai( detail={ "task_set_id": task_set.id, "selected_indices": req.selected_indices, - "replace_existing": req.replace_existing, + # "replace_existing": req.replace_existing, "created_segment_count": len(created), "segment_ids": [segment.id for segment in created], }, @@ -1100,39 +1133,60 @@ async def prepare_reanalyze_task_set( *, current_user: User, task_set_id: str, - force: bool = False, reason: str | None = None, ) -> ShotReanalyzeOut: - """重置原视频分析状态，供 API 重新投递 Celery。""" + """仅对已失败且旧账务已关闭的原视频分析创建新 attempt。""" task_set = await get_task_set_for_user(db, task_set_id=task_set_id, user=current_user, for_update=True) - if task_set.analysis_status == ShotAnalysisStatusEnum.PROCESSING.value: + previous_attempt_no = max(1, int(task_set.analysis_attempt_no or 1)) + if task_set.analysis_status != ShotAnalysisStatusEnum.FAILED.value: log_module_event_file( module=MODULE, event_type=ShotReplicateLogEventEnum.TASK_SET_REANALYZE_REJECTED.value, project_id=task_set.id, user_id=task_set.user_id, - message="原视频分析正在处理中，拒绝再次分析", - detail={"task_set_id": task_set.id, "analysis_status": task_set.analysis_status, "reason": reason}, + message="原视频分析不是失败终态，拒绝再次分析", + detail={ + "task_set_id": task_set.id, + "analysis_status": task_set.analysis_status, + "analysis_attempt_no": previous_attempt_no, + "reason": reason, + }, event_status="rejected", ) - raise HTTPException(status_code=409, detail="原视频分析正在处理中，不能重复投递") - if task_set.analysis_status == ShotAnalysisStatusEnum.COMPLETED.value and not force: - raise HTTPException(status_code=409, detail="原视频分析已完成，如确需重跑请传 force=true") - if force: - active_segments_result = await db.execute( - select(func.count()) - .select_from(ShotReplicateSegment) - .where( - ShotReplicateSegment.task_set_id == task_set.id, - ShotReplicateSegment.deleted_at.is_(None), - ) + raise HTTPException( + status_code=409, + detail=f"只有分析失败的原视频任务才能重新分析，当前状态：{task_set.analysis_status}", + ) + + previous_context = build_task_set_analysis_billing_context(task_set) + previous_validation = await validate_retryable_previous_attempt(db, previous_context) + if not previous_validation.can_execute: + log_module_event_file( + module=MODULE, + event_type=ShotReplicateLogEventEnum.TASK_SET_REANALYZE_REJECTED.value, + project_id=task_set.id, + user_id=task_set.user_id, + message="原视频旧分析 attempt 账务未关闭，拒绝再次分析", + detail={ + "task_set_id": task_set.id, + "analysis_attempt_no": previous_attempt_no, + "ledger_state": previous_validation.state.value, + "ledger_reason": previous_validation.reason, + "reason": reason, + }, + event_status="rejected", + ) + raise HTTPException( + status_code=409, + detail=( + "上一次原视频分析的冻结积分尚未完成释放或账务状态异常，" + f"当前账务状态：{previous_validation.state.value}" + ), ) - if int(active_segments_result.scalar() or 0) > 0: - raise HTTPException(status_code=409, detail="当前总任务集已存在拆镜片段，不能强制重跑原视频分析") task_set.status = ShotTaskSetStatusEnum.PENDING_ANALYSIS.value task_set.analysis_status = ShotAnalysisStatusEnum.PENDING.value - task_set.analysis_attempt_no = max(1, int(task_set.analysis_attempt_no or 1)) + 1 + task_set.analysis_attempt_no = previous_attempt_no + 1 task_set.analysis_claim_token = None task_set.analysis_started_at = None task_set.analysis_lease_until = None @@ -1151,7 +1205,14 @@ async def prepare_reanalyze_task_set( project_id=task_set.id, user_id=task_set.user_id, message="原视频再次分析已重置状态", - detail={"task_set_id": task_set.id, "force": force, "reason": reason, "video_url": task_set.video_url}, + detail={ + "task_set_id": task_set.id, + "previous_analysis_attempt_no": previous_attempt_no, + "analysis_attempt_no": int(task_set.analysis_attempt_no), + "previous_ledger_state": previous_validation.state.value, + "reason": reason, + "video_url": task_set.video_url, + }, ) return ShotReanalyzeOut( message="原视频再次分析任务已准备投递", @@ -1168,34 +1229,69 @@ async def prepare_reanalyze_segment( *, current_user: User, segment_id: str, - force: bool = False, reason: str | None = None, ) -> ShotReanalyzeOut: - """重置自定义切片视频分析状态，供 API 重新投递 Celery。""" + """仅对已失败且旧账务已关闭的自定义切片分析创建新 attempt。""" segment = await get_segment_for_user(db, segment_id=segment_id, user=current_user, for_update=True) if segment.split_status != ShotSplitStatusEnum.COMPLETED.value: raise HTTPException(status_code=409, detail="当前片段还未切割完成，不能再次分析") if not segment.segment_video_url: raise HTTPException(status_code=409, detail="当前片段缺少 segment_video_url，不能再次分析") - if segment.analysis_status == ShotSegmentAnalysisStatusEnum.PROCESSING.value: + if segment.source_mode != ShotSegmentSourceModeEnum.CUSTOM.value: + raise HTTPException(status_code=409, detail="只有自定义切片视频支持重新分析") + + previous_attempt_no = max(1, int(segment.analysis_attempt_no or 1)) + if segment.analysis_status != ShotSegmentAnalysisStatusEnum.FAILED.value: log_module_event_file( module=MODULE, event_type=ShotReplicateLogEventEnum.SEGMENT_REANALYZE_REJECTED.value, project_id=segment.task_set_id, step_id=segment.id, user_id=segment.user_id, - message="切片视频分析正在处理中，拒绝再次分析", - detail={"segment_id": segment.id, "analysis_status": segment.analysis_status, "reason": reason}, + message="切片视频分析不是失败终态，拒绝再次分析", + detail={ + "segment_id": segment.id, + "analysis_status": segment.analysis_status, + "analysis_attempt_no": previous_attempt_no, + "reason": reason, + }, event_status="rejected", ) - raise HTTPException(status_code=409, detail="切片视频分析正在处理中，不能重复投递") - if segment.analysis_status == ShotSegmentAnalysisStatusEnum.COMPLETED.value and not force: - raise HTTPException(status_code=409, detail="切片视频分析已完成，如确需重跑请传 force=true") - if segment.source_mode != ShotSegmentSourceModeEnum.CUSTOM.value and not force: - raise HTTPException(status_code=409, detail="AI 建议片段默认无需单独分析，如确需重跑请传 force=true") + raise HTTPException( + status_code=409, + detail=f"只有分析失败的切片视频才能重新分析，当前状态：{segment.analysis_status}", + ) + + previous_context = build_segment_analysis_billing_context(segment) + previous_validation = await validate_retryable_previous_attempt(db, previous_context) + if not previous_validation.can_execute: + log_module_event_file( + module=MODULE, + event_type=ShotReplicateLogEventEnum.SEGMENT_REANALYZE_REJECTED.value, + project_id=segment.task_set_id, + step_id=segment.id, + user_id=segment.user_id, + message="切片视频旧分析 attempt 账务未关闭，拒绝再次分析", + detail={ + "segment_id": segment.id, + "task_set_id": segment.task_set_id, + "analysis_attempt_no": previous_attempt_no, + "ledger_state": previous_validation.state.value, + "ledger_reason": previous_validation.reason, + "reason": reason, + }, + event_status="rejected", + ) + raise HTTPException( + status_code=409, + detail=( + "上一次切片视频分析的冻结积分尚未完成释放或账务状态异常，" + f"当前账务状态：{previous_validation.state.value}" + ), + ) segment.analysis_status = ShotSegmentAnalysisStatusEnum.PENDING.value - segment.analysis_attempt_no = max(1, int(segment.analysis_attempt_no or 1)) + 1 + segment.analysis_attempt_no = previous_attempt_no + 1 segment.analysis_claim_token = None segment.analysis_started_at = None segment.analysis_lease_until = None @@ -1216,7 +1312,15 @@ async def prepare_reanalyze_segment( step_id=segment.id, user_id=segment.user_id, message="切片视频再次分析已重置状态", - detail={"segment_id": segment.id, "task_set_id": segment.task_set_id, "force": force, "reason": reason, "video_url": segment.segment_video_url}, + detail={ + "segment_id": segment.id, + "task_set_id": segment.task_set_id, + "previous_analysis_attempt_no": previous_attempt_no, + "analysis_attempt_no": int(segment.analysis_attempt_no), + "previous_ledger_state": previous_validation.state.value, + "reason": reason, + "video_url": segment.segment_video_url, + }, ) return ShotReanalyzeOut( message="切片视频再次分析任务已准备投递", @@ -1233,9 +1337,12 @@ async def mark_task_set_analysis_dispatch_failed( *, current_user: User, task_set_id: str, + expected_attempt_no: int, error_message: str, ) -> bool: task_set = await get_task_set_for_user(db, task_set_id=task_set_id, user=current_user, for_update=True) + if int(task_set.analysis_attempt_no or 1) != int(expected_attempt_no): + return False if ( task_set.analysis_status == ShotAnalysisStatusEnum.PROCESSING.value and task_set.analysis_claim_token @@ -1245,12 +1352,11 @@ async def mark_task_set_analysis_dispatch_failed( return False if task_set.analysis_status in (ShotAnalysisStatusEnum.COMPLETED.value, ShotAnalysisStatusEnum.FAILED.value): return False - if task_set.analysis_status not in (ShotAnalysisStatusEnum.COMPLETED.value, ShotAnalysisStatusEnum.FAILED.value): - task_set.status = ShotTaskSetStatusEnum.ANALYSIS_FAILED.value - task_set.analysis_status = ShotAnalysisStatusEnum.FAILED.value - task_set.analysis_claim_token = None - task_set.analysis_lease_until = None - task_set.analysis_error_message = error_message + task_set.status = ShotTaskSetStatusEnum.ANALYSIS_FAILED.value + task_set.analysis_status = ShotAnalysisStatusEnum.FAILED.value + task_set.analysis_claim_token = None + task_set.analysis_lease_until = None + task_set.analysis_error_message = error_message await release_on_failure(db, build_task_set_analysis_billing_context(task_set), error=error_message) log_module_event_file( module=MODULE, @@ -1269,9 +1375,12 @@ async def mark_segment_analysis_dispatch_failed( *, current_user: User, segment_id: str, + expected_attempt_no: int, error_message: str, ) -> bool: segment = await get_segment_for_user(db, segment_id=segment_id, user=current_user, for_update=True) + if int(segment.analysis_attempt_no or 1) != int(expected_attempt_no): + return False if ( segment.analysis_status == ShotSegmentAnalysisStatusEnum.PROCESSING.value and segment.analysis_claim_token @@ -1281,11 +1390,10 @@ async def mark_segment_analysis_dispatch_failed( return False if segment.analysis_status in (ShotSegmentAnalysisStatusEnum.COMPLETED.value, ShotSegmentAnalysisStatusEnum.FAILED.value): return False - if segment.analysis_status not in (ShotSegmentAnalysisStatusEnum.COMPLETED.value, ShotSegmentAnalysisStatusEnum.FAILED.value): - segment.analysis_status = ShotSegmentAnalysisStatusEnum.FAILED.value - segment.analysis_claim_token = None - segment.analysis_lease_until = None - segment.analysis_error_message = error_message + segment.analysis_status = ShotSegmentAnalysisStatusEnum.FAILED.value + segment.analysis_claim_token = None + segment.analysis_lease_until = None + segment.analysis_error_message = error_message await release_on_failure(db, build_segment_analysis_billing_context(segment), error=error_message) log_module_event_file( module=MODULE, diff --git a/video-gen-api/app/services/shot_video_analysis_service.py b/video-gen-api/app/services/shot_video_analysis_service.py index 2c72779f..255c5edf 100644 --- a/video-gen-api/app/services/shot_video_analysis_service.py +++ b/video-gen-api/app/services/shot_video_analysis_service.py @@ -756,7 +756,12 @@ async def analyze_video_for_shot_split( result = ensure_result_schema(result) result = filter_and_normalize_breakdown(result, mode=mode) - usage = raw.get("usage") or {} + raw_usage = raw.get("usage") + usage_reported = bool( + isinstance(raw_usage, dict) + and any(key in raw_usage for key in ("prompt_tokens", "completion_tokens", "input_tokens", "output_tokens", "total_tokens")) + ) + usage = raw_usage if isinstance(raw_usage, dict) else {} token_usage = { "input_tokens": _int_usage(usage.get("prompt_tokens") or usage.get("input_tokens")), "output_tokens": _int_usage(usage.get("completion_tokens") or usage.get("output_tokens")), @@ -771,6 +776,7 @@ async def analyze_video_for_shot_split( "split_max_seconds": _split_max_seconds(), "analysis_mode": mode, "trace_id": trace_id, + "usage_reported": usage_reported, } if not token_usage["total_tokens"]: token_usage["total_tokens"] = token_usage["input_tokens"] + token_usage["output_tokens"] diff --git a/video-gen-api/app/tasks/shot_replicate_tasks.py b/video-gen-api/app/tasks/shot_replicate_tasks.py index b3b04f46..ab7372dd 100644 --- a/video-gen-api/app/tasks/shot_replicate_tasks.py +++ b/video-gen-api/app/tasks/shot_replicate_tasks.py @@ -33,7 +33,10 @@ from app.services.redis_registry_service import ( ) from app.services.celery_runtime.recovery_service import guard_periodic_recovery from app.services.celery_runtime.runtime_service import CeleryRuntimeLease, RuntimeIdentity -from app.services.shot_replicate_taskset_service import refresh_task_set_split_summary +from app.services.shot_replicate_taskset_service import ( + build_segment_analysis_billing_context, + refresh_task_set_split_summary, +) from app.services.shot_video_analysis_service import analyze_video_for_shot_split from app.services.llm_billing import ( LlmBillingContext, @@ -126,9 +129,237 @@ def _analysis_lock_key(owner_type: str, owner_id: str, attempt_no: int) -> str: return f"{settings.SHOT_ANALYSIS_LOCK_KEY_PREFIX}:{owner_type}:{owner_id}:attempt:{attempt_no}" -async def _run_analyze_original_video(task_set_id: str) -> None: +def _log_stale_analysis_attempt( + *, + owner_type: str, + owner_id: str, + expected_attempt_no: int | None, + actual_attempt_no: int | None, + analysis_status: str | None, + task_set_id: str | None = None, + user_id: str | None = None, + reason: str, +) -> None: + project_id = task_set_id or (owner_id if owner_type == "shot_task_set" else None) + log_module_event_file( + module=MODULE, + event_type=ShotReplicateLogEventEnum.ANALYSIS_STALE_ATTEMPT_SKIPPED.value, + project_id=project_id, + step_id=owner_id if owner_type == "shot_segment" else None, + user_id=user_id, + message="拆镜分析任务 attempt 已失效，跳过执行", + detail={ + "owner_type": owner_type, + "owner_id": owner_id, + "expected_attempt_no": expected_attempt_no, + "actual_attempt_no": actual_attempt_no, + "analysis_status": analysis_status, + "reason": reason, + }, + event_status="skipped", + ) + + +async def _persist_original_analysis_result( + *, + task_set_id: str, + attempt_no: int, + token: str, + video_url: str | None, + analyzed: Any, + billing_context: LlmBillingContext, + allow_business_write: bool, +) -> str: + """持久化原视频分析结果并完成账务；供应商成功后禁止再次调用模型。""" + 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() + is_current = bool( + allow_business_write + and task_set + and int(task_set.analysis_attempt_no or 1) == attempt_no + and task_set.analysis_claim_token == token + and str(task_set.video_url) == str(video_url) + and task_set.analysis_status == ShotAnalysisStatusEnum.PROCESSING.value + ) + if is_current and task_set is not None: + 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_claim_token = None + task_set.analysis_lease_until = None + task_set.analysis_error_message = None + description = "拆镜复刻-原视频分析" + outcome = "completed" + else: + description = "拆镜复刻-原视频分析（失效结果结算）" + outcome = "stale_settled" + await settle_success( + db, + billing_context, + usage=analyzed.usage, + description=description, + ) + await db.commit() + return outcome + + +async def _persist_segment_analysis_result( + *, + segment_id: str, + attempt_no: int, + token: str, + video_url: str | None, + analyzed: Any, + billing_context: LlmBillingContext, + allow_business_write: bool, +) -> str: + """持久化自定义切片分析结果并完成账务；供应商成功后禁止再次调用模型。""" + 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() + is_current = bool( + allow_business_write + and segment + and int(segment.analysis_attempt_no or 1) == attempt_no + and segment.analysis_claim_token == token + and str(segment.segment_video_url) == str(video_url) + and segment.analysis_status == ShotSegmentAnalysisStatusEnum.PROCESSING.value + ) + if is_current and segment is not None: + 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_claim_token = None + segment.analysis_lease_until = None + segment.analysis_error_message = None + description = "拆镜复刻-片段视频分析" + outcome = "completed" + else: + description = "拆镜复刻-片段视频分析（失效结果结算）" + outcome = "stale_settled" + await settle_success( + db, + billing_context, + usage=analyzed.usage, + description=description, + ) + await db.commit() + return outcome + + +async def _mark_original_provider_success_pending_manual( + *, + task_set_id: str, + attempt_no: int, + token: str, + error_message: str, +) -> bool: + """供应商已成功但本地结算失败：终止自动恢复，保留 HOLD 等待人工对账。""" + 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 + and int(task_set.analysis_attempt_no or 1) == attempt_no + and task_set.analysis_claim_token == token + and task_set.analysis_status == ShotAnalysisStatusEnum.PROCESSING.value + ): + await db.rollback() + return False + task_set.status = ShotTaskSetStatusEnum.ANALYSIS_FAILED.value + task_set.analysis_status = ShotAnalysisStatusEnum.FAILED.value + task_set.analysis_claim_token = None + task_set.analysis_lease_until = None + task_set.analysis_error_message = error_message + await db.commit() + return True + + +async def _mark_segment_provider_success_pending_manual( + *, + segment_id: str, + attempt_no: int, + token: str, + error_message: str, +) -> bool: + """供应商已成功但本地结算失败：终止自动恢复，保留 HOLD 等待人工对账。""" + 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 + and int(segment.analysis_attempt_no or 1) == attempt_no + and segment.analysis_claim_token == token + and segment.analysis_status == ShotSegmentAnalysisStatusEnum.PROCESSING.value + ): + await db.rollback() + return False + segment.analysis_status = ShotSegmentAnalysisStatusEnum.FAILED.value + segment.analysis_claim_token = None + segment.analysis_lease_until = None + segment.analysis_error_message = error_message + await db.commit() + return True + + +async def _run_analyze_original_video(task_set_id: str, expected_attempt_no: int | None) -> None: token = uuid.uuid4().hex - attempt_no = 1 + if expected_attempt_no is None: + _log_stale_analysis_attempt( + owner_type="shot_task_set", + owner_id=task_set_id, + expected_attempt_no=None, + actual_attempt_no=None, + analysis_status=None, + reason="missing_expected_attempt_no", + ) + return + attempt_no = max(1, int(expected_attempt_no)) async with async_session() as db: row = await db.execute( @@ -137,9 +368,25 @@ async def _run_analyze_original_video(task_set_id: str) -> None: .limit(1) ) initial = row.scalar_one_or_none() - if not initial or initial.analysis_status == ShotAnalysisStatusEnum.COMPLETED.value: + if not initial: + return + actual_attempt_no = int(initial.analysis_attempt_no or 1) + if ( + actual_attempt_no != attempt_no + or initial.analysis_status + not in (ShotAnalysisStatusEnum.PENDING.value, ShotAnalysisStatusEnum.PROCESSING.value) + ): + _log_stale_analysis_attempt( + owner_type="shot_task_set", + owner_id=task_set_id, + expected_attempt_no=attempt_no, + actual_attempt_no=actual_attempt_no, + analysis_status=initial.analysis_status, + user_id=str(initial.user_id), + reason="attempt_or_status_mismatch_before_lock", + ) + await db.rollback() return - attempt_no = int(initial.analysis_attempt_no or 1) await db.rollback() lease = await CeleryRuntimeLease.acquire( @@ -174,7 +421,24 @@ async def _run_analyze_original_video(task_set_id: str) -> None: .limit(1) ) task_set = result.scalar_one_or_none() - if not task_set or task_set.analysis_status == ShotAnalysisStatusEnum.COMPLETED.value: + if not task_set: + await db.rollback() + return + actual_attempt_no = int(task_set.analysis_attempt_no or 1) + if ( + actual_attempt_no != attempt_no + or task_set.analysis_status + not in (ShotAnalysisStatusEnum.PENDING.value, ShotAnalysisStatusEnum.PROCESSING.value) + ): + _log_stale_analysis_attempt( + owner_type="shot_task_set", + owner_id=task_set_id, + expected_attempt_no=attempt_no, + actual_attempt_no=actual_attempt_no, + analysis_status=task_set.analysis_status, + user_id=str(task_set.user_id), + reason="attempt_or_status_mismatch_after_lock", + ) await db.rollback() return current_lease = task_set.analysis_lease_until @@ -186,9 +450,6 @@ async def _run_analyze_original_video(task_set_id: str) -> None: ): await db.rollback() return - if int(task_set.analysis_attempt_no or 1) != attempt_no: - await db.rollback() - return task_set_user_id = str(task_set.user_id) video_url = str(task_set.video_url) task_set.status = ShotTaskSetStatusEnum.ANALYZING.value @@ -263,52 +524,25 @@ async def _run_analyze_original_video(task_set_id: str) -> None: task_set_id=task_set_id, trace_id=f"shot-task-set-analysis:{task_set_id}:attempt:{attempt_no}", ) - provider_succeeded = True - log_provider_success(llm_billing_context, usage=analyzed.usage) + provider_succeeded = True + log_provider_success(llm_billing_context, usage=analyzed.usage) + try: await lease.ensure_owned() - result = await call_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 - or int(task_set.analysis_attempt_no or 1) != attempt_no - or task_set.analysis_claim_token != token - or str(task_set.video_url) != str(video_url) - or task_set.analysis_status != ShotAnalysisStatusEnum.PROCESSING.value - ): - await call_db.rollback() - # Provider 已成功，旧业务对象失效也必须按真实 usage 结算。 - await settle_success( - call_db, - llm_billing_context, - usage=analyzed.usage, - description="拆镜复刻-原视频分析（失效结果结算）", - ) - await call_db.commit() - 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_claim_token = None - task_set.analysis_lease_until = None - task_set.analysis_error_message = None - await settle_success( - call_db, - llm_billing_context, - usage=analyzed.usage, - description="拆镜复刻-原视频分析", - ) - await call_db.commit() + allow_business_write = True + except RedisExecutionLockError: + # Provider 已成功后不能让 Celery retry 再次调用模型；失去执行权时仅结算。 + allow_business_write = False + persist_outcome = await _persist_original_analysis_result( + task_set_id=task_set_id, + attempt_no=attempt_no, + token=token, + video_url=video_url, + analyzed=analyzed, + billing_context=llm_billing_context, + allow_business_write=allow_business_write, + ) + if persist_outcome != "completed": + return log_module_prompt_event( event_type=ShotReplicateLogEventEnum.ANALYSIS_SUCCESS.value, @@ -334,40 +568,67 @@ async def _run_analyze_original_video(task_set_id: str) -> None: except Exception as exc: if "llm_billing_context" in locals() and not locals().get("provider_succeeded", False): log_provider_failure(llm_billing_context, error=str(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() - task_set_is_current = bool( - task_set - and int(task_set.analysis_attempt_no or 1) == attempt_no - and task_set.analysis_claim_token == token - and task_set.analysis_status == ShotAnalysisStatusEnum.PROCESSING.value - ) - if task_set_is_current and task_set is not None: - task_set_user_id = task_set_user_id or str(task_set.user_id) - task_set.status = ShotTaskSetStatusEnum.ANALYSIS_FAILED.value - task_set.analysis_status = ShotAnalysisStatusEnum.FAILED.value - task_set.analysis_claim_token = None - task_set.analysis_lease_until = None - task_set.analysis_error_message = str(exc) - # 业务对象是否仍有效，不影响本 attempt 的账务终态：provider 已成功必须结算， - # provider 未成功则幂等释放。这样人工删库/异常换 attempt 也不会遗留 active HOLD。 - if "llm_billing_context" in locals(): - if locals().get("provider_succeeded", False) and locals().get("analyzed") is not None: - await settle_success( - db, - llm_billing_context, - usage=analyzed.usage, - description="拆镜复刻-原视频分析（本地失败结算）", + if "llm_billing_context" in locals() and locals().get("provider_succeeded", False) and locals().get("analyzed") is not None: + try: + try: + await lease.ensure_owned() + allow_business_write = True + except RedisExecutionLockError: + allow_business_write = False + persist_outcome = await _persist_original_analysis_result( + task_set_id=task_set_id, + attempt_no=attempt_no, + token=token, + video_url=video_url, + analyzed=analyzed, + billing_context=llm_billing_context, + allow_business_write=allow_business_write, + ) + if persist_outcome == "completed": + 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={"task_set_id": task_set_id, "analysis_attempt_no": attempt_no}, ) - else: - await release_on_failure(db, llm_billing_context, error=str(exc)) - await db.commit() + return + except Exception as settlement_exc: + manual_error = ( + "供应商已成功，但本地结果保存或积分结算失败；已终止自动恢复并保留冻结积分，" + f"需人工对账。error={settlement_exc}" + ) + await _mark_original_provider_success_pending_manual( + task_set_id=task_set_id, + attempt_no=attempt_no, + token=token, + error_message=manual_error, + ) + exc = settlement_exc + elif "llm_billing_context" in locals(): + 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 + and int(task_set.analysis_attempt_no or 1) == attempt_no + and task_set.analysis_claim_token == token + and task_set.analysis_status == ShotAnalysisStatusEnum.PROCESSING.value + ): + task_set_user_id = task_set_user_id or str(task_set.user_id) + task_set.status = ShotTaskSetStatusEnum.ANALYSIS_FAILED.value + task_set.analysis_status = ShotAnalysisStatusEnum.FAILED.value + task_set.analysis_claim_token = None + task_set.analysis_lease_until = None + task_set.analysis_error_message = str(exc) + await release_on_failure(db, llm_billing_context, error=str(exc)) + await db.commit() log_module_error( module=MODULE, event_type=ShotReplicateLogEventEnum.ANALYSIS_FAILED.value, @@ -381,9 +642,19 @@ async def _run_analyze_original_video(task_set_id: str) -> None: await lease.close() -async def _run_analyze_custom_segment_video(segment_id: str) -> None: +async def _run_analyze_custom_segment_video(segment_id: str, expected_attempt_no: int | None) -> None: token = uuid.uuid4().hex - attempt_no = 1 + if expected_attempt_no is None: + _log_stale_analysis_attempt( + owner_type="shot_segment", + owner_id=segment_id, + expected_attempt_no=None, + actual_attempt_no=None, + analysis_status=None, + reason="missing_expected_attempt_no", + ) + return + attempt_no = max(1, int(expected_attempt_no)) async with async_session() as db: row = await db.execute( select(ShotReplicateSegment) @@ -391,9 +662,26 @@ async def _run_analyze_custom_segment_video(segment_id: str) -> None: .limit(1) ) initial = row.scalar_one_or_none() - if not initial or not initial.segment_video_url or initial.analysis_status == ShotSegmentAnalysisStatusEnum.COMPLETED.value: + if not initial or not initial.segment_video_url: + return + actual_attempt_no = int(initial.analysis_attempt_no or 1) + if ( + actual_attempt_no != attempt_no + or initial.analysis_status + not in (ShotSegmentAnalysisStatusEnum.PENDING.value, ShotSegmentAnalysisStatusEnum.PROCESSING.value) + ): + _log_stale_analysis_attempt( + owner_type="shot_segment", + owner_id=segment_id, + expected_attempt_no=attempt_no, + actual_attempt_no=actual_attempt_no, + analysis_status=initial.analysis_status, + task_set_id=str(initial.task_set_id), + user_id=str(initial.user_id), + reason="attempt_or_status_mismatch_before_lock", + ) + await db.rollback() return - attempt_no = int(initial.analysis_attempt_no or 1) await db.rollback() lease = await CeleryRuntimeLease.acquire( @@ -429,16 +717,31 @@ async def _run_analyze_custom_segment_video(segment_id: str) -> None: .limit(1) ) segment = result.scalar_one_or_none() - if not segment or not segment.segment_video_url or segment.analysis_status == ShotSegmentAnalysisStatusEnum.COMPLETED.value: + if not segment or not segment.segment_video_url: + await db.rollback() + return + actual_attempt_no = int(segment.analysis_attempt_no or 1) + if ( + actual_attempt_no != attempt_no + or segment.analysis_status + not in (ShotSegmentAnalysisStatusEnum.PENDING.value, ShotSegmentAnalysisStatusEnum.PROCESSING.value) + ): + _log_stale_analysis_attempt( + owner_type="shot_segment", + owner_id=segment_id, + expected_attempt_no=attempt_no, + actual_attempt_no=actual_attempt_no, + analysis_status=segment.analysis_status, + task_set_id=str(segment.task_set_id), + user_id=str(segment.user_id), + reason="attempt_or_status_mismatch_after_lock", + ) await db.rollback() return current_lease = segment.analysis_lease_until if segment.analysis_claim_token and segment.analysis_claim_token != token and current_lease and current_lease > _now(): await db.rollback() return - if int(segment.analysis_attempt_no or 1) != attempt_no: - await db.rollback() - return user_id = str(segment.user_id) task_set_id = str(segment.task_set_id) video_url = str(segment.segment_video_url) @@ -509,51 +812,24 @@ async def _run_analyze_custom_segment_video(segment_id: str) -> None: segment_id=segment_id, trace_id=f"shot-segment-analysis:{segment_id}:attempt:{attempt_no}", ) - provider_succeeded = True - log_provider_success(llm_billing_context, usage=analyzed.usage) + provider_succeeded = True + log_provider_success(llm_billing_context, usage=analyzed.usage) + try: await lease.ensure_owned() - result = await call_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 int(segment.analysis_attempt_no or 1) != attempt_no - or segment.analysis_claim_token != token - or str(segment.segment_video_url) != str(video_url) - or segment.analysis_status != ShotSegmentAnalysisStatusEnum.PROCESSING.value - ): - await call_db.rollback() - await settle_success( - call_db, - llm_billing_context, - usage=analyzed.usage, - description="拆镜复刻-片段视频分析（失效结果结算）", - ) - await call_db.commit() - 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_claim_token = None - segment.analysis_lease_until = None - segment.analysis_error_message = None - await settle_success( - call_db, - llm_billing_context, - usage=analyzed.usage, - description="拆镜复刻-片段视频分析", - ) - await call_db.commit() + allow_business_write = True + except RedisExecutionLockError: + allow_business_write = False + persist_outcome = await _persist_segment_analysis_result( + segment_id=segment_id, + attempt_no=attempt_no, + token=token, + video_url=video_url, + analyzed=analyzed, + billing_context=llm_billing_context, + allow_business_write=allow_business_write, + ) + if persist_outcome != "completed": + return log_module_prompt_event( event_type=ShotReplicateLogEventEnum.SEGMENT_ANALYSIS_SUCCESS.value, @@ -580,38 +856,68 @@ async def _run_analyze_custom_segment_video(segment_id: str) -> None: except Exception as exc: if "llm_billing_context" in locals() and not locals().get("provider_succeeded", False): log_provider_failure(llm_billing_context, error=str(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() - segment_is_current = bool( - segment - and int(segment.analysis_attempt_no or 1) == attempt_no - and segment.analysis_claim_token == token - and segment.analysis_status == ShotSegmentAnalysisStatusEnum.PROCESSING.value - ) - if segment_is_current and segment is not None: - user_id = user_id or str(segment.user_id) - task_set_id = task_set_id or str(segment.task_set_id) - segment.analysis_status = ShotSegmentAnalysisStatusEnum.FAILED.value - segment.analysis_claim_token = None - segment.analysis_lease_until = None - segment.analysis_error_message = str(exc) - if "llm_billing_context" in locals(): - if locals().get("provider_succeeded", False) and locals().get("analyzed") is not None: - await settle_success( - db, - llm_billing_context, - usage=analyzed.usage, - description="拆镜复刻-片段视频分析（本地失败结算）", + if "llm_billing_context" in locals() and locals().get("provider_succeeded", False) and locals().get("analyzed") is not None: + try: + try: + await lease.ensure_owned() + allow_business_write = True + except RedisExecutionLockError: + allow_business_write = False + persist_outcome = await _persist_segment_analysis_result( + segment_id=segment_id, + attempt_no=attempt_no, + token=token, + video_url=video_url, + analyzed=analyzed, + billing_context=llm_billing_context, + allow_business_write=allow_business_write, + ) + if persist_outcome == "completed": + 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, "analysis_attempt_no": attempt_no}, ) - else: - await release_on_failure(db, llm_billing_context, error=str(exc)) - await db.commit() + return + except Exception as settlement_exc: + manual_error = ( + "供应商已成功，但本地结果保存或积分结算失败；已终止自动恢复并保留冻结积分，" + f"需人工对账。error={settlement_exc}" + ) + await _mark_segment_provider_success_pending_manual( + segment_id=segment_id, + attempt_no=attempt_no, + token=token, + error_message=manual_error, + ) + exc = settlement_exc + elif "llm_billing_context" in locals(): + 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 + and int(segment.analysis_attempt_no or 1) == attempt_no + and segment.analysis_claim_token == token + and segment.analysis_status == ShotSegmentAnalysisStatusEnum.PROCESSING.value + ): + user_id = user_id or str(segment.user_id) + task_set_id = task_set_id or str(segment.task_set_id) + segment.analysis_status = ShotSegmentAnalysisStatusEnum.FAILED.value + segment.analysis_claim_token = None + segment.analysis_lease_until = None + segment.analysis_error_message = str(exc) + await release_on_failure(db, llm_billing_context, error=str(exc)) + await db.commit() log_module_error( module=MODULE, event_type=ShotReplicateLogEventEnum.SEGMENT_ANALYSIS_FAILED.value, @@ -643,6 +949,65 @@ async def _renew_split_lease(segment_id: str, attempt_no: int, token: str) -> bo return bool(result.rowcount == 1) +async def _mark_auto_segment_analysis_dispatch_failed( + *, + segment_id: str, + expected_attempt_no: int, + error_message: str, +) -> bool: + """切片成功后的自动分析投递失败补偿；不回滚或清理已完成的切片文件。""" + task_set_id: str | None = None + user_id: str | None = None + 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 int(segment.analysis_attempt_no or 1) != int(expected_attempt_no) + or segment.analysis_status != ShotSegmentAnalysisStatusEnum.PENDING.value + ): + await db.rollback() + return False + + task_set_id = str(segment.task_set_id) + user_id = str(segment.user_id) + segment.analysis_status = ShotSegmentAnalysisStatusEnum.FAILED.value + segment.analysis_claim_token = None + segment.analysis_started_at = None + segment.analysis_lease_until = None + segment.analysis_error_message = error_message + await release_on_failure( + db, + build_segment_analysis_billing_context(segment), + error=error_message, + ) + await db.commit() + + log_module_error( + module=MODULE, + event_type=ShotReplicateLogEventEnum.CELERY_DISPATCH_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, + "analysis_attempt_no": expected_attempt_no, + }, + exc=RuntimeError(error_message), + ) + return True + + async def _run_split_one_segment(segment_id: str) -> None: async with async_session() as db: row = await db.execute( @@ -826,6 +1191,7 @@ async def _run_split_one_segment(segment_id: str) -> None: source_mode = str(segment.source_mode) final_task_set_id = str(segment.task_set_id) final_user_id = str(segment.user_id) + analysis_attempt_no = max(1, int(segment.analysis_attempt_no or 1)) await refresh_task_set_split_summary(db, segment.task_set_id) await db.commit() @@ -847,12 +1213,19 @@ async def _run_split_one_segment(segment_id: str) -> None: ) if source_mode == ShotSegmentSourceModeEnum.CUSTOM.value and celery_app: - analyze_custom_segment_video.apply_async( - args=[segment_id], - queue=ANALYSIS_QUEUE, - countdown=0, - task_id=f"shot-analysis:segment:{segment_id}:attempt:1", - ) + try: + analyze_custom_segment_video.apply_async( + args=[segment_id, analysis_attempt_no], + queue=ANALYSIS_QUEUE, + countdown=0, + task_id=f"shot-analysis:segment:{segment_id}:attempt:{analysis_attempt_no}", + ) + except Exception as dispatch_exc: + await _mark_auto_segment_analysis_dispatch_failed( + segment_id=segment_id, + expected_attempt_no=analysis_attempt_no, + error_message=f"自定义切片分析任务投递失败: {dispatch_exc}", + ) except RedisExecutionLockError: cleanup_split_result(split_result) raise @@ -980,9 +1353,9 @@ if celery_app: time_limit=settings.SHOT_ANALYSIS_TIME_LIMIT_SECONDS, ignore_result=True, ) - def analyze_original_video(self, task_set_id: str) -> None: + def analyze_original_video(self, task_set_id: str, expected_attempt_no: int | None = None) -> None: try: - return run_async(_run_analyze_original_video(task_set_id)) + return run_async(_run_analyze_original_video(task_set_id, expected_attempt_no)) except RedisExecutionLockError as exc: raise self.retry(exc=exc, countdown=60) @@ -1004,9 +1377,9 @@ if celery_app: time_limit=settings.SHOT_ANALYSIS_TIME_LIMIT_SECONDS, ignore_result=True, ) - def analyze_custom_segment_video(self, segment_id: str) -> None: + def analyze_custom_segment_video(self, segment_id: str, expected_attempt_no: int | None = None) -> None: try: - return run_async(_run_analyze_custom_segment_video(segment_id)) + return run_async(_run_analyze_custom_segment_video(segment_id, expected_attempt_no)) except RedisExecutionLockError as exc: raise self.retry(exc=exc, countdown=60) diff --git a/video-gen-app/src/api/index.ts b/video-gen-app/src/api/index.ts index 6841d19f..4f7d81fb 100644 --- a/video-gen-app/src/api/index.ts +++ b/video-gen-app/src/api/index.ts @@ -302,7 +302,7 @@ export async function verifyCaptcha(captchaId: string, x: number): Promise { - if (USE_MOCK) return { siteName: 'VideoGen.AI', siteLogo: '', userAgreementPrivacyUrl: '', siteCopyright: '© 2024 民众智创 版权所有', operationManual: '', loginBgVideo: '' }; + if (USE_MOCK) return { siteName: '智创', siteLogo: '', userAgreementPrivacyUrl: '', siteCopyright: '© 2026 智创 版权所有', operationManual: '', loginBgVideo: '' }; return api.get('/auth/site-info', false); } // ── Video Engines ───────────────────────────────────────── diff --git a/video-gen-app/src/components/Layout/AppLayout.tsx b/video-gen-app/src/components/Layout/AppLayout.tsx index cff81175..fc2247de 100644 --- a/video-gen-app/src/components/Layout/AppLayout.tsx +++ b/video-gen-app/src/components/Layout/AppLayout.tsx @@ -385,7 +385,7 @@ const AppLayout: React.FC = () => { useEffect(() => { getSiteInfo().then(info => { - const name = info.siteName || '民众智创'; + const name = info.siteName || '智创'; const logo = info.siteLogo || ''; if (name !== siteName) { diff --git a/video-gen-app/src/pages/LoginPage.tsx b/video-gen-app/src/pages/LoginPage.tsx index 0b46fe30..60784104 100644 --- a/video-gen-app/src/pages/LoginPage.tsx +++ b/video-gen-app/src/pages/LoginPage.tsx @@ -36,12 +36,12 @@ const LoginPage: React.FC = () => { if (cached) { const info = JSON.parse(cached); return { - siteName: info.siteName || 'VideoGen.AI', + siteName: info.siteName || '智创', siteLogo: info.siteLogo || '', }; } } catch { } - return { siteName: 'VideoGen.AI', siteLogo: '' }; + return { siteName: '智创', siteLogo: '' }; }; const initialInfo = getInitialSiteInfo();