mz/video-gen
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

交易流水对账明细导出完成

Browse Source
This commit is contained in:
GinHa
2026-06-25 16:53:04 +08:00
parent 8aafd7de32
commit 30db7d3bf5
34 changed files with 2491 additions and 327 deletions
Download Patch File Download Diff File Expand all files Collapse all files
video-gen-api/app/services/admin_credit_record_service.py
+299
View File
@@ -0,0 +1,299 @@
from __future__ import annotations
from datetime import datetime
from typing import Any
from sqlalchemy import and_, case, distinct, func, or_, select
from sqlalchemy.ext.asyncio import AsyncSession
from app.enums.credit_record import (
CREDIT_RECORD_BILLING_SCENE_LABELS,
CREDIT_RECORD_CHARGE_KIND_LABELS,
CREDIT_RECORD_MEDIA_TYPE_LABELS,
CREDIT_RECORD_SOURCE_MODULE_LABELS,
CREDIT_RECORD_SOURCE_STEP_CODE_LABELS,
CREDIT_RECORD_SUBJECT_LABELS,
CREDIT_RECORD_TYPE_LABELS,
CreditRecordSubject,
)
from app.enums.user import FRONTEND_USER_KIND_LABELS, USER_TYPE_LABELS, UserType
from app.models.chat_generation_task import ChatGenerationTask
from app.models.credit_record import CreditRecord
from app.models.generation_record import GenerationRecord
from app.models.module_generation_project import ModuleGenerationProject
from app.models.module_generation_step import ModuleGenerationStep
from app.models.shot_replicate_segment import ShotReplicateSegment
from app.models.shot_replicate_task_set import ShotReplicateTaskSet
from app.models.user import User
def _iso(dt: Any) -> str | None:
if dt is None:
return None
try:
return dt.isoformat()
except Exception:
return str(dt)
def _round2(value: Any) -> float:
try:
return round(float(value or 0), 2)
except Exception:
return 0.0
def _as_date_start(value: str | None) -> datetime | None:
if not value:
return None
try:
return datetime.strptime(value, "%Y-%m-%d")
except Exception:
return None
def _as_date_end(value: str | None) -> datetime | None:
if not value:
return None
try:
return datetime.strptime(value, "%Y-%m-%d").replace(hour=23, minute=59, second=59, microsecond=999999)
except Exception:
return None
def _label(mapping: dict[str, str], value: str | None) -> str:
if not value:
return "-"
return mapping.get(value, value)
def _build_filters(
*,
user_id: str | None = None,
user_name: str | None = None,
user_type: str | None = None,
frontend_user_kind: str | None = None,
record_type: str | None = None,
credit_subject: str | None = None,
media_type: str | None = None,
charge_kind: str | None = None,
source_module: str | None = None,
source_step_code: str | None = None,
billing_scene: str | None = None,
start_date: str | None = None,
end_date: str | None = None,
) -> list[Any]:
filters: list[Any] = []
if user_id:
filters.append(CreditRecord.user_id == user_id)
if user_name:
like = f"%{user_name}%"
filters.append(or_(User.username.ilike(like), User.phone.ilike(like), User.email.ilike(like)))
if user_type:
filters.append(CreditRecord.user_type_snapshot == user_type)
if frontend_user_kind:
filters.append(CreditRecord.frontend_user_kind_snapshot == frontend_user_kind)
filters.append(CreditRecord.user_type_snapshot == UserType.FRONTEND.value)
if record_type:
filters.append(CreditRecord.type == record_type)
if credit_subject:
filters.append(CreditRecord.credit_subject == credit_subject)
if media_type:
filters.append(CreditRecord.media_type == media_type)
if charge_kind:
filters.append(CreditRecord.charge_kind == charge_kind)
if source_module:
filters.append(CreditRecord.source_module == source_module)
if source_step_code:
filters.append(CreditRecord.source_step_code == source_step_code)
if billing_scene:
filters.append(CreditRecord.billing_scene == billing_scene)
start = _as_date_start(start_date)
end = _as_date_end(end_date)
if start:
filters.append(CreditRecord.created_at >= start)
if end:
filters.append(CreditRecord.created_at <= end)
return filters
async def _load_deleted_map(db: AsyncSession, records: list[CreditRecord]) -> dict[tuple[str, str], tuple[bool, str | None]]:
grouped: dict[str, set[str]] = {}
for record in records:
if record.owner_type and record.owner_id:
grouped.setdefault(record.owner_type, set()).add(record.owner_id)
model_map: dict[str, Any] = {
"chat_generation_task": ChatGenerationTask,
"generation_record": GenerationRecord,
"module_generation_project": ModuleGenerationProject,
"module_generation_step": ModuleGenerationStep,
"shot_replicate_task_set": ShotReplicateTaskSet,
"shot_replicate_segment": ShotReplicateSegment,
}
deleted_map: dict[tuple[str, str], tuple[bool, str | None]] = {}
for owner_type, ids in grouped.items():
model = model_map.get(owner_type)
if not model or not ids:
continue
result = await db.execute(select(model.id, model.deleted_at).where(model.id.in_(ids)))
found = {row[0]: row[1] for row in result.all()}
for owner_id in ids:
deleted_at = found.get(owner_id)
deleted_map[(owner_type, owner_id)] = (bool(deleted_at), _iso(deleted_at)) if owner_id in found else (False, None)
return deleted_map
def _record_to_item(record: CreditRecord, user: User | None, deleted_map: dict[tuple[str, str], tuple[bool, str | None]]) -> dict[str, Any]:
owner_deleted = False
owner_deleted_at = None
if record.owner_type and record.owner_id:
owner_deleted, owner_deleted_at = deleted_map.get((record.owner_type, record.owner_id), (False, None))
user_type = record.user_type_snapshot or (user.user_type if user else None)
frontend_kind = record.frontend_user_kind_snapshot or (getattr(user, "frontend_user_kind", None) if user else None)
return {
"id": record.id,
"user_id": record.user_id,
"username": user.username if user else None,
"phone": user.phone if user else None,
"email": user.email if user else None,
"user_type": user_type,
"user_type_label": _label(USER_TYPE_LABELS, user_type),
"frontend_user_kind": frontend_kind,
"frontend_user_kind_label": _label(FRONTEND_USER_KIND_LABELS, frontend_kind),
"type": record.type,
"record_type": record.type,
"record_type_label": _label(CREDIT_RECORD_TYPE_LABELS, record.type),
"amount": _round2(record.amount),
"balance_after": _round2(record.balance_after),
"description": record.description,
"related_id": record.related_id,
"biz_key": record.biz_key,
"refund_for_biz_key": record.refund_for_biz_key,
"owner_type": record.owner_type,
"owner_id": record.owner_id,
"owner_deleted": owner_deleted,
"owner_deleted_at": owner_deleted_at,
"attempt_no": record.attempt_no,
"charge_kind": record.charge_kind,
"charge_kind_label": _label(CREDIT_RECORD_CHARGE_KIND_LABELS, record.charge_kind),
"charge_action": record.charge_action,
"credit_subject": record.credit_subject,
"credit_subject_label": _label(CREDIT_RECORD_SUBJECT_LABELS, record.credit_subject),
"media_type": record.media_type,
"media_type_label": _label(CREDIT_RECORD_MEDIA_TYPE_LABELS, record.media_type),
"billing_scene": record.billing_scene,
"billing_scene_label": _label(CREDIT_RECORD_BILLING_SCENE_LABELS, record.billing_scene),
"source_module": record.source_module,
"source_module_label": _label(CREDIT_RECORD_SOURCE_MODULE_LABELS, record.source_module),
"source_project_id": record.source_project_id,
"source_step_id": record.source_step_id,
"source_step_code": record.source_step_code,
"source_step_code_label": _label(CREDIT_RECORD_SOURCE_STEP_CODE_LABELS, record.source_step_code),
"token_usage_id": record.token_usage_id,
"input_tokens": record.input_tokens or 0,
"output_tokens": record.output_tokens or 0,
"total_tokens": record.total_tokens or 0,
"engine_type": record.engine_type,
"engine_id": record.engine_id,
"engine_name": record.engine_name,
"engine_provider": record.engine_provider,
"engine_model_name": record.engine_model_name,
"created_at": _iso(record.created_at),
}
async def list_admin_credit_records(
db: AsyncSession,
*,
page: int,
page_size: int,
user_id: str | None = None,
user_name: str | None = None,
user_type: str | None = None,
frontend_user_kind: str | None = None,
record_type: str | None = None,
credit_subject: str | None = None,
media_type: str | None = None,
charge_kind: str | None = None,
source_module: str | None = None,
source_step_code: str | None = None,
billing_scene: str | None = None,
start_date: str | None = None,
end_date: str | None = None,
) -> dict[str, Any]:
page = max(int(page or 1), 1)
page_size = min(max(int(page_size or 20), 1), 1000)
filters = _build_filters(
user_id=user_id,
user_name=user_name,
user_type=user_type,
frontend_user_kind=frontend_user_kind,
record_type=record_type,
credit_subject=credit_subject,
media_type=media_type,
charge_kind=charge_kind,
source_module=source_module,
source_step_code=source_step_code,
billing_scene=billing_scene,
start_date=start_date,
end_date=end_date,
)
where_clause = and_(*filters) if filters else None
base_query = select(CreditRecord, User).join(User, CreditRecord.user_id == User.id, isouter=True)
count_query = select(func.count(CreditRecord.id)).select_from(CreditRecord).join(User, CreditRecord.user_id == User.id, isouter=True)
if where_clause is not None:
base_query = base_query.where(where_clause)
count_query = count_query.where(where_clause)
total = (await db.execute(count_query)).scalar() or 0
result = await db.execute(
base_query.order_by(CreditRecord.created_at.desc(), CreditRecord.id.desc())
.offset((page - 1) * page_size)
.limit(page_size)
)
rows = result.all()
records = [row[0] for row in rows]
deleted_map = await _load_deleted_map(db, records)
items = [_record_to_item(record, user, deleted_map) for record, user in rows]
summary_query = select(
func.coalesce(func.sum(case((CreditRecord.type == "recharge", CreditRecord.amount), else_=0)), 0),
func.coalesce(func.sum(case((CreditRecord.type == "consume", func.abs(CreditRecord.amount)), else_=0)), 0),
func.coalesce(func.sum(case((CreditRecord.type == "refund", CreditRecord.amount), else_=0)), 0),
func.count(CreditRecord.id),
func.count(distinct(case((and_(CreditRecord.credit_subject == CreditRecordSubject.MEDIA.value, CreditRecord.type == "consume"), func.concat(CreditRecord.owner_type, ":", CreditRecord.owner_id)), else_=None))),
func.count(case((and_(CreditRecord.credit_subject == CreditRecordSubject.MEDIA.value, CreditRecord.type == "consume"), 1), else_=None)),
func.count(distinct(case((and_(CreditRecord.credit_subject == CreditRecordSubject.MEDIA.value, CreditRecord.media_type == "image", CreditRecord.type == "consume"), func.concat(CreditRecord.owner_type, ":", CreditRecord.owner_id)), else_=None))),
func.count(distinct(case((and_(CreditRecord.credit_subject == CreditRecordSubject.MEDIA.value, CreditRecord.media_type == "video", CreditRecord.type == "consume"), func.concat(CreditRecord.owner_type, ":", CreditRecord.owner_id)), else_=None))),
func.coalesce(func.sum(case((and_(CreditRecord.credit_subject == CreditRecordSubject.MEDIA.value, CreditRecord.media_type == "image", CreditRecord.type == "consume"), func.abs(CreditRecord.amount)), else_=0)), 0),
func.coalesce(func.sum(case((and_(CreditRecord.credit_subject == CreditRecordSubject.MEDIA.value, CreditRecord.media_type == "video", CreditRecord.type == "consume"), func.abs(CreditRecord.amount)), else_=0)), 0),
func.coalesce(func.sum(case((and_(CreditRecord.credit_subject == CreditRecordSubject.TEXT.value, CreditRecord.type == "consume"), func.abs(CreditRecord.amount)), else_=0)), 0),
func.coalesce(func.sum(case((and_(CreditRecord.credit_subject == CreditRecordSubject.ANALYSIS.value, CreditRecord.type == "consume"), func.abs(CreditRecord.amount)), else_=0)), 0),
func.coalesce(func.sum(CreditRecord.total_tokens), 0),
func.coalesce(func.sum(CreditRecord.input_tokens), 0),
func.coalesce(func.sum(CreditRecord.output_tokens), 0),
).select_from(CreditRecord).join(User, CreditRecord.user_id == User.id, isouter=True)
if where_clause is not None:
summary_query = summary_query.where(where_clause)
s = (await db.execute(summary_query)).one()
summary = {
"total_recharge": _round2(s[0]),
"total_consume": _round2(s[1]),
"total_refund": _round2(s[2]),
"transaction_count": int(s[3] or 0),
"generation_count": int(s[4] or 0),
"generation_attempt_count": int(s[5] or 0),
"image_generation_count": int(s[6] or 0),
"video_generation_count": int(s[7] or 0),
"image_consume": _round2(s[8]),
"video_consume": _round2(s[9]),
"text_consume": _round2(s[10]),
"analysis_consume": _round2(s[11]),
"total_tokens": int(s[12] or 0),
"input_tokens": int(s[13] or 0),
"output_tokens": int(s[14] or 0),
}
return {"items": items, "total": total, "summary": summary}
video-gen-api/app/services/credit_record_meta_service.py
+360
View File
@@ -0,0 +1,360 @@
from __future__ import annotations
from dataclasses import asdict, dataclass
from typing import Any, Mapping
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.enums.credit_record import (
CreditRecordAction,
CreditRecordBillingScene,
CreditRecordChargeKind,
CreditRecordOwnerType,
CreditRecordSourceModule,
CreditRecordSourceStepCode,
CreditRecordSubject,
)
from app.enums.user import FrontendUserKind
from app.models.image_engine import ImageEngine
from app.models.model_config import ModelConfig
from app.models.module_generation_step import ModuleGenerationStep
from app.models.user import User
from app.models.video_engine import VideoEngine
@dataclass(slots=True)
class CreditRecordMeta:
owner_type: str | None = None
owner_id: str | None = None
attempt_no: int | None = None
charge_kind: str | None = None
charge_action: str | None = None
credit_subject: str | None = None
media_type: str | None = None
billing_scene: str | None = None
source_module: str | None = None
source_project_id: str | None = None
source_step_id: str | None = None
source_step_code: str | None = None
token_usage_id: str | None = None
input_tokens: int | None = None
output_tokens: int | None = None
total_tokens: int | None = None
# 统一执行配置冷备字段：
# - engine_type=model 时，engine_* 来源于 model_configs
# - engine_type=image 时，engine_* 来源于 image_engines
# - engine_type=video 时，engine_* 来源于 video_engines
engine_type: str | None = None
engine_id: str | None = None
engine_name: str | None = None
engine_provider: str | None = None
engine_model_name: str | None = None
user_type_snapshot: str | None = None
frontend_user_kind_snapshot: str | None = None
def to_record_kwargs(self) -> dict[str, Any]:
return {k: v for k, v in asdict(self).items() if v is not None}
def _safe_int(value: Any, default: int = 0) -> int:
try:
if value is None or value == "":
return default
return int(value)
except Exception:
return default
def _normalize_frontend_kind(value: str | None) -> str:
return value or FrontendUserKind.EXTERNAL.value
async def with_user_snapshot(db: AsyncSession, meta: CreditRecordMeta, user_id: str) -> CreditRecordMeta:
result = await db.execute(select(User).where(User.id == user_id).limit(1))
user = result.scalar_one_or_none()
if user:
meta.user_type_snapshot = user.user_type
meta.frontend_user_kind_snapshot = _normalize_frontend_kind(getattr(user, "frontend_user_kind", None))
return meta
async def get_model_snapshot(db: AsyncSession, model_config_id: str | None) -> dict[str, str | None]:
"""把 ModelConfig 冷备为统一 engine_* 快照。"""
if not model_config_id:
return {"engine_type": "model"}
result = await db.execute(select(ModelConfig).where(ModelConfig.id == model_config_id).limit(1))
config = result.scalar_one_or_none()
if not config:
return {"engine_type": "model", "engine_id": model_config_id}
return {
"engine_type": "model",
"engine_id": config.id,
"engine_name": config.name,
"engine_provider": config.provider,
"engine_model_name": config.model_name,
}
def _model_snapshot_from_usage(usage: Mapping[str, Any]) -> dict[str, str | None]:
model_config_id = usage.get("model_config_id")
return {
"engine_type": "model",
"engine_id": model_config_id,
"engine_name": usage.get("model_config_name") or usage.get("engine_name"),
"engine_provider": usage.get("model_provider") or usage.get("provider") or usage.get("engine_provider"),
"engine_model_name": usage.get("model_name") or usage.get("model") or usage.get("engine_model_name"),
}
async def _apply_model_snapshot(db: AsyncSession, meta: CreditRecordMeta, usage: Mapping[str, Any]) -> CreditRecordMeta:
for key, value in _model_snapshot_from_usage(usage).items():
setattr(meta, key, value)
if meta.engine_id and not meta.engine_model_name:
for key, value in (await get_model_snapshot(db, meta.engine_id)).items():
setattr(meta, key, value)
return meta
async def get_engine_snapshot(db: AsyncSession, *, gen_type: str, engine_id: str | None) -> dict[str, str | None]:
if not engine_id:
return {"engine_type": (gen_type or None)}
gen_type = (gen_type or "").lower().strip()
if gen_type == "image":
result = await db.execute(select(ImageEngine).where(ImageEngine.id == engine_id).limit(1))
else:
result = await db.execute(select(VideoEngine).where(VideoEngine.id == engine_id).limit(1))
engine = result.scalar_one_or_none()
if not engine:
return {"engine_type": gen_type or None, "engine_id": engine_id}
return {
"engine_type": gen_type or None,
"engine_id": engine.id,
"engine_name": getattr(engine, "name", None),
"engine_provider": getattr(engine, "provider", None),
"engine_model_name": getattr(engine, "model_name", None),
}
def infer_module_billing_scene(*, source_module: str | None, source_step_code: str | None, media_type: str | None = None) -> str | None:
if source_module == CreditRecordSourceModule.HOT_OPENING_REPLICATE.value:
mapping = {
"image_prompt_optimize": CreditRecordBillingScene.HOT_OPENING_IMAGE_PROMPT_OPTIMIZE.value,
"image_generate": CreditRecordBillingScene.HOT_OPENING_IMAGE_GENERATE.value,
"video_prompt_optimize": CreditRecordBillingScene.HOT_OPENING_VIDEO_PROMPT_OPTIMIZE.value,
"video_generate": CreditRecordBillingScene.HOT_OPENING_VIDEO_GENERATE.value,
}
return mapping.get(source_step_code)
if source_module == CreditRecordSourceModule.SHOT_REPLICATE.value:
mapping = {
"image_prompt_optimize": CreditRecordBillingScene.SHOT_IMAGE_PROMPT_OPTIMIZE.value,
"image_generate": CreditRecordBillingScene.SHOT_IMAGE_GENERATE.value,
"video_prompt_optimize": CreditRecordBillingScene.SHOT_VIDEO_PROMPT_OPTIMIZE.value,
"video_generate": CreditRecordBillingScene.SHOT_VIDEO_GENERATE.value,
CreditRecordSourceStepCode.VIDEO_ANALYSIS.value: CreditRecordBillingScene.SHOT_VIDEO_ANALYSIS.value,
}
return mapping.get(source_step_code)
if source_module == CreditRecordSourceModule.AI_CREATION.value:
return CreditRecordBillingScene.AI_CREATION_IMAGE_GENERATE.value if media_type == "image" else CreditRecordBillingScene.AI_CREATION_VIDEO_GENERATE.value
if source_module == CreditRecordSourceModule.GENERATION_RECORD.value:
return CreditRecordBillingScene.GENERATION_RECORD_IMAGE_GENERATE.value if media_type == "image" else CreditRecordBillingScene.GENERATION_RECORD_VIDEO_GENERATE.value
return None
def build_admin_adjust_meta(*, action: str = CreditRecordAction.CHARGE.value) -> CreditRecordMeta:
return CreditRecordMeta(
owner_type=CreditRecordOwnerType.ADMIN_ADJUST.value,
charge_kind=CreditRecordChargeKind.ADMIN_ADJUST.value,
charge_action=action,
credit_subject=CreditRecordSubject.ADMIN_ADJUST.value,
source_module=CreditRecordSourceModule.ADMIN.value,
billing_scene=CreditRecordBillingScene.ADMIN_ADJUST.value,
)
def build_recharge_meta(*, owner_id: str | None = None) -> CreditRecordMeta:
return CreditRecordMeta(
owner_type=CreditRecordOwnerType.PAYMENT_ORDER.value if owner_id else None,
owner_id=owner_id,
charge_kind=CreditRecordChargeKind.RECHARGE.value,
charge_action=CreditRecordAction.CHARGE.value,
credit_subject=CreditRecordSubject.RECHARGE.value,
source_module=CreditRecordSourceModule.PAYMENT.value,
billing_scene=CreditRecordBillingScene.RECHARGE.value,
)
def build_payment_refund_meta(*, owner_id: str | None = None) -> CreditRecordMeta:
return CreditRecordMeta(
owner_type=CreditRecordOwnerType.PAYMENT_ORDER.value if owner_id else None,
owner_id=owner_id,
charge_kind=CreditRecordChargeKind.REFUND.value,
charge_action=CreditRecordAction.REFUND.value,
credit_subject=CreditRecordSubject.REFUND.value,
source_module=CreditRecordSourceModule.PAYMENT.value,
billing_scene=CreditRecordBillingScene.REFUND.value,
)
async def build_generation_media_meta(
db: AsyncSession,
*,
owner_type: str,
owner_id: str,
attempt_no: int,
gen_type: str,
engine_id: str | None,
source_module: str | None = None,
source_project_id: str | None = None,
source_step_id: str | None = None,
source_step_code: str | None = None,
billing_scene: str | None = None,
) -> CreditRecordMeta:
media_type = (gen_type or "").lower().strip() or None
if source_module is None:
source_module = CreditRecordSourceModule.GENERATION_RECORD.value if owner_type == CreditRecordOwnerType.GENERATION_RECORD.value else CreditRecordSourceModule.AI_CREATION.value
if billing_scene is None:
billing_scene = infer_module_billing_scene(source_module=source_module, source_step_code=source_step_code, media_type=media_type)
meta = CreditRecordMeta(
owner_type=owner_type,
owner_id=owner_id,
attempt_no=attempt_no,
charge_kind=CreditRecordChargeKind.MEDIA.value,
charge_action=CreditRecordAction.CHARGE.value,
credit_subject=CreditRecordSubject.MEDIA.value,
media_type=media_type,
billing_scene=billing_scene,
source_module=source_module,
source_project_id=source_project_id,
source_step_id=source_step_id,
source_step_code=source_step_code,
)
for key, value in (await get_engine_snapshot(db, gen_type=media_type or "", engine_id=engine_id)).items():
setattr(meta, key, value)
return meta
async def build_generation_record_prompt_meta(
db: AsyncSession,
*,
record_id: str,
attempt_no: int,
charge_kind: str,
usage: Mapping[str, Any],
) -> CreditRecordMeta:
scene_map = {
CreditRecordChargeKind.TEXT_PROMPT.value: CreditRecordBillingScene.GENERATION_RECORD_TEXT_PROMPT_OPTIMIZE.value,
CreditRecordChargeKind.FILE_PARSE.value: CreditRecordBillingScene.GENERATION_RECORD_FILE_PARSE.value,
CreditRecordChargeKind.VISION_INPUT.value: CreditRecordBillingScene.GENERATION_RECORD_VISION_INPUT.value,
}
meta = CreditRecordMeta(
owner_type=CreditRecordOwnerType.GENERATION_RECORD.value,
owner_id=record_id,
attempt_no=attempt_no,
charge_kind=charge_kind,
charge_action=CreditRecordAction.CHARGE.value,
credit_subject=CreditRecordSubject.TEXT.value,
billing_scene=scene_map.get(charge_kind),
source_module=CreditRecordSourceModule.GENERATION_RECORD.value,
token_usage_id=usage.get("token_usage_id"),
input_tokens=_safe_int(usage.get("input_tokens")),
output_tokens=_safe_int(usage.get("output_tokens")),
total_tokens=_safe_int(usage.get("total_tokens"), _safe_int(usage.get("input_tokens")) + _safe_int(usage.get("output_tokens"))),
)
return await _apply_model_snapshot(db, meta, usage)
async def build_module_step_prompt_meta(
db: AsyncSession,
*,
step_id: str,
attempt_no: int,
usage: Mapping[str, Any],
) -> CreditRecordMeta:
result = await db.execute(select(ModuleGenerationStep).where(ModuleGenerationStep.id == step_id).limit(1))
step = result.scalar_one_or_none()
source_module = getattr(step, "module", None) if step else None
source_step_code = getattr(step, "step_code", None) if step else None
meta = CreditRecordMeta(
owner_type=CreditRecordOwnerType.MODULE_GENERATION_STEP.value,
owner_id=step_id,
attempt_no=attempt_no,
charge_kind=CreditRecordChargeKind.TEXT_PROMPT.value,
charge_action=CreditRecordAction.CHARGE.value,
credit_subject=CreditRecordSubject.TEXT.value,
billing_scene=infer_module_billing_scene(source_module=source_module, source_step_code=source_step_code),
source_module=source_module,
source_project_id=getattr(step, "project_id", None) if step else None,
source_step_id=step_id,
source_step_code=source_step_code,
token_usage_id=usage.get("token_usage_id"),
input_tokens=_safe_int(usage.get("input_tokens")),
output_tokens=_safe_int(usage.get("output_tokens")),
total_tokens=_safe_int(usage.get("total_tokens"), _safe_int(usage.get("input_tokens")) + _safe_int(usage.get("output_tokens"))),
)
return await _apply_model_snapshot(db, meta, usage)
async def build_shot_video_analysis_meta(
db: AsyncSession,
*,
owner_type: str,
owner_id: str,
attempt_no: int,
usage: Mapping[str, Any],
billing_scene: str,
source_project_id: str | None = None,
source_step_id: str | None = None,
) -> CreditRecordMeta:
meta = CreditRecordMeta(
owner_type=owner_type,
owner_id=owner_id,
attempt_no=attempt_no,
charge_kind=CreditRecordChargeKind.VIDEO_ANALYSIS.value,
charge_action=CreditRecordAction.CHARGE.value,
credit_subject=CreditRecordSubject.ANALYSIS.value,
media_type="video",
billing_scene=billing_scene,
source_module=CreditRecordSourceModule.SHOT_REPLICATE.value,
source_project_id=source_project_id,
source_step_id=source_step_id,
source_step_code=CreditRecordSourceStepCode.VIDEO_ANALYSIS.value,
token_usage_id=usage.get("token_usage_id"),
input_tokens=_safe_int(usage.get("input_tokens")),
output_tokens=_safe_int(usage.get("output_tokens")),
total_tokens=_safe_int(usage.get("total_tokens"), _safe_int(usage.get("input_tokens")) + _safe_int(usage.get("output_tokens"))),
)
return await _apply_model_snapshot(db, meta, usage)
def build_refund_meta_from_charge(charge: Any, *, attempt_no: int | None = None) -> CreditRecordMeta:
return CreditRecordMeta(
owner_type=getattr(charge, "owner_type", None),
owner_id=getattr(charge, "owner_id", None),
attempt_no=attempt_no or getattr(charge, "attempt_no", None),
charge_kind=getattr(charge, "charge_kind", None),
charge_action=CreditRecordAction.REFUND.value,
credit_subject=getattr(charge, "credit_subject", None) or CreditRecordSubject.REFUND.value,
media_type=getattr(charge, "media_type", None),
billing_scene=CreditRecordBillingScene.REFUND.value,
source_module=getattr(charge, "source_module", None),
source_project_id=getattr(charge, "source_project_id", None),
source_step_id=getattr(charge, "source_step_id", None),
source_step_code=getattr(charge, "source_step_code", None),
token_usage_id=getattr(charge, "token_usage_id", None),
input_tokens=getattr(charge, "input_tokens", None),
output_tokens=getattr(charge, "output_tokens", None),
total_tokens=getattr(charge, "total_tokens", None),
engine_type=getattr(charge, "engine_type", None),
engine_id=getattr(charge, "engine_id", None),
engine_name=getattr(charge, "engine_name", None),
engine_provider=getattr(charge, "engine_provider", None),
engine_model_name=getattr(charge, "engine_model_name", None),
user_type_snapshot=getattr(charge, "user_type_snapshot", None),
frontend_user_kind_snapshot=getattr(charge, "frontend_user_kind_snapshot", None),
)
video-gen-api/app/services/credits.py
+21
View File
@@ -11,6 +11,7 @@ from app.models.image_engine import ImageEngine
from app.models.credit_ratio import CreditRatio
from app.utils.id_gen import generate_id
from app.utils.exceptions import InsufficientCreditsError
from app.services.credit_record_meta_service import CreditRecordMeta, with_user_snapshot
async def calc_text_credits(db: AsyncSession, input_tokens: int, output_tokens: int) -> float:
@@ -167,6 +168,7 @@ async def deduct_credits(
*,
biz_key: str | None = None,
refund_for_biz_key: str | None = None,
record_meta: CreditRecordMeta | dict | None = None,
) -> User:
"""扣减用户积分，并写入消费流水。
@@ -196,6 +198,13 @@ async def deduct_credits(
raise InsufficientCreditsError()
user.credits = round(float(user.credits or 0) - amount, 2)
meta_kwargs = {}
if record_meta:
if isinstance(record_meta, CreditRecordMeta):
record_meta = await with_user_snapshot(db, record_meta, user_id)
meta_kwargs = record_meta.to_record_kwargs()
elif isinstance(record_meta, dict):
meta_kwargs = {k: v for k, v in record_meta.items() if v is not None}
record = CreditRecord(
id=generate_id(),
user_id=user_id,
@@ -206,6 +215,7 @@ async def deduct_credits(
related_id=related_id,
biz_key=biz_key,
refund_for_biz_key=refund_for_biz_key,
**meta_kwargs,
)
db.add(record)
await db.flush()
@@ -222,6 +232,7 @@ async def add_credits(
record_type: str = "recharge",
biz_key: str | None = None,
refund_for_biz_key: str | None = None,
record_meta: CreditRecordMeta | dict | None = None,
) -> User:
"""增加用户积分，并写入流水。
@@ -243,6 +254,13 @@ async def add_credits(
return user
user.credits = round(float(user.credits or 0) + amount, 2)
meta_kwargs = {}
if record_meta:
if isinstance(record_meta, CreditRecordMeta):
record_meta = await with_user_snapshot(db, record_meta, user_id)
meta_kwargs = record_meta.to_record_kwargs()
elif isinstance(record_meta, dict):
meta_kwargs = {k: v for k, v in record_meta.items() if v is not None}
record = CreditRecord(
id=generate_id(),
user_id=user_id,
@@ -253,6 +271,7 @@ async def add_credits(
related_id=related_id,
biz_key=biz_key,
refund_for_biz_key=refund_for_biz_key,
**meta_kwargs,
)
db.add(record)
await db.flush()
@@ -268,6 +287,7 @@ async def refund_credits(
*,
biz_key: str | None = None,
refund_for_biz_key: str | None = None,
record_meta: CreditRecordMeta | dict | None = None,
) -> User:
"""生成失败积分回退。"""
return await add_credits(
@@ -279,6 +299,7 @@ async def refund_credits(
record_type="refund",
biz_key=biz_key,
refund_for_biz_key=refund_for_biz_key,
record_meta=record_meta,
)
video-gen-api/app/services/generation_billing_service.py
+157 -13
View File
@@ -4,23 +4,36 @@ import re
from dataclasses import asdict, dataclass
from typing import Any, Mapping
from sqlalchemy import func, select
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.enums.credit_record import CreditRecordBillingScene, CreditRecordChargeKind, CreditRecordOwnerType, CreditRecordSourceModule
from app.models.credit_record import CreditRecord
from app.models.generation_record import GenerationRecord
from app.models.module_generation_step import ModuleGenerationStep
from app.models.token_usage import TokenUsage
from app.models.system_config import SystemConfig
from app.services.credit_record_meta_service import (
CreditRecordMeta,
build_generation_media_meta,
build_generation_record_prompt_meta,
build_module_step_prompt_meta,
build_shot_video_analysis_meta,
)
from app.services.credits import calc_image_credits, calc_text_credits, calc_video_credits, deduct_credits
CHARGE_TEXT_PROMPT = "text_prompt"
CHARGE_FILE_PARSE = "file_parse"
CHARGE_VISION_INPUT = "vision_input"
CHARGE_MEDIA = "media"
CHARGE_TEXT_PROMPT = CreditRecordChargeKind.TEXT_PROMPT.value
CHARGE_FILE_PARSE = CreditRecordChargeKind.FILE_PARSE.value
CHARGE_VISION_INPUT = CreditRecordChargeKind.VISION_INPUT.value
CHARGE_MEDIA = CreditRecordChargeKind.MEDIA.value
CHARGE_VIDEO_ANALYSIS = CreditRecordChargeKind.VIDEO_ANALYSIS.value
OWNER_GENERATION_RECORD = "generation_record"
OWNER_CHAT_GENERATION_TASK = "chat_generation_task"
OWNER_MODULE_GENERATION_STEP = "module_generation_step"
OWNER_GENERATION_RECORD = CreditRecordOwnerType.GENERATION_RECORD.value
OWNER_CHAT_GENERATION_TASK = CreditRecordOwnerType.CHAT_GENERATION_TASK.value
OWNER_MODULE_GENERATION_STEP = CreditRecordOwnerType.MODULE_GENERATION_STEP.value
OWNER_SHOT_REPLICATE_TASK_SET = CreditRecordOwnerType.SHOT_REPLICATE_TASK_SET.value
OWNER_SHOT_REPLICATE_SEGMENT = CreditRecordOwnerType.SHOT_REPLICATE_SEGMENT.value
_BIZ_KEY_PATTERN = re.compile(
r"^(?P<owner_type>[^:]+):(?P<owner_id>[^:]+):attempt:(?P<attempt_no>\d+):(?P<charge_kind>[^:]+):(?P<action>charge|refund)$"
@@ -169,10 +182,12 @@ async def deduct_credits_locked_once(
charge_key: str,
biz_key: str | None = None,
attempt_no: int | None = None,
record_meta: CreditRecordMeta | dict | None = None,
) -> BillingItem:
"""按 biz_key 做幂等扣费。
charge_key 只保留为业务分类；正式幂等以 biz_key 为准。
record_meta 负责把业务归属、模块、步骤、token、模型快照写入 CreditRecord。
"""
amount = _round2(amount)
if amount <= 0:
@@ -197,6 +212,7 @@ async def deduct_credits_locked_once(
description=description,
related_id=related_id,
biz_key=biz_key,
record_meta=record_meta,
)
return BillingItem(charge_key=charge_key, amount=amount, charged=True, biz_key=biz_key, attempt_no=attempt_no)
@@ -208,7 +224,7 @@ async def charge_chatapi_prompt_usage(
usage: Mapping[str, Any],
project_name: str | None = None,
) -> BillingSummary:
"""提示词整理扣费仍按记录维度一次性幂等，不参与生成失败媒体退款。"""
"""项目记录提示词整理扣费；不参与生成失败媒体退款。"""
project_name = project_name or "AI生成任务"
items: list[BillingItem] = []
@@ -219,12 +235,19 @@ async def charge_chatapi_prompt_usage(
input_tokens = _safe_int(usage.get("input_tokens"))
output_tokens = _safe_int(usage.get("output_tokens"))
text_credits = await calc_text_credits(db, input_tokens, output_tokens)
text_meta = await build_generation_record_prompt_meta(
db,
record_id=record.id,
attempt_no=attempt_no,
charge_kind=CHARGE_TEXT_PROMPT,
usage=usage,
)
items.append(
await deduct_credits_locked_once(
db,
user_id=record.user_id,
amount=text_credits,
description=f"ChatAPI提示词整理",
description=f"提示词优化 - {project_name}",
related_id=record.id,
charge_key=CHARGE_TEXT_PROMPT,
biz_key=build_credit_biz_key(
@@ -235,17 +258,25 @@ async def charge_chatapi_prompt_usage(
action="charge",
),
attempt_no=attempt_no,
record_meta=text_meta,
)
)
file_tokens = usage.get("file_parse_tokens") or usage.get("file_tokens") or usage.get("document_tokens") or 0
file_parse_credits = await _calc_optional_token_credits(db, _safe_int(file_tokens), "file_parse_credits_per_1000_tokens")
file_meta = await build_generation_record_prompt_meta(
db,
record_id=record.id,
attempt_no=attempt_no,
charge_kind=CHARGE_FILE_PARSE,
usage={**dict(usage), "total_tokens": _safe_int(file_tokens), "input_tokens": _safe_int(file_tokens), "output_tokens": 0},
)
items.append(
await deduct_credits_locked_once(
db,
user_id=record.user_id,
amount=file_parse_credits,
description=f"文件解析Token",
description="文件解析Token",
related_id=record.id,
charge_key=CHARGE_FILE_PARSE,
biz_key=build_credit_biz_key(
@@ -256,17 +287,25 @@ async def charge_chatapi_prompt_usage(
action="charge",
),
attempt_no=attempt_no,
record_meta=file_meta,
)
)
vision_tokens = usage.get("vision_input_tokens") or usage.get("image_input_tokens") or usage.get("image_tokens") or 0
vision_input_credits = await _calc_optional_token_credits(db, _safe_int(vision_tokens), "vision_input_credits_per_1000_tokens")
vision_meta = await build_generation_record_prompt_meta(
db,
record_id=record.id,
attempt_no=attempt_no,
charge_kind=CHARGE_VISION_INPUT,
usage={**dict(usage), "total_tokens": _safe_int(vision_tokens), "input_tokens": _safe_int(vision_tokens), "output_tokens": 0},
)
items.append(
await deduct_credits_locked_once(
db,
user_id=record.user_id,
amount=vision_input_credits,
description=f"图片理解Token",
description="图片理解Token",
related_id=record.id,
charge_key=CHARGE_VISION_INPUT,
biz_key=build_credit_biz_key(
@@ -277,6 +316,7 @@ async def charge_chatapi_prompt_usage(
action="charge",
),
attempt_no=attempt_no,
record_meta=vision_meta,
)
)
@@ -297,7 +337,7 @@ async def charge_module_prompt_usage(
description: str,
attempt_no: int = 1,
) -> BillingSummary:
"""爆款开头复刻模块图片/视频 AI 提词扣文本积分。
"""爆款开头复刻/拆镜复刻模块图片/视频 AI 提词扣文本积分。
文本提词属于已经发生的 LLM 消费：
- 调用成功后按 input_tokens + output_tokens 扣费。
@@ -314,6 +354,7 @@ async def charge_module_prompt_usage(
charge_kind=CHARGE_TEXT_PROMPT,
action="charge",
)
record_meta = await build_module_step_prompt_meta(db, step_id=step_id, attempt_no=attempt_no, usage=usage)
item = await deduct_credits_locked_once(
db,
user_id=user_id,
@@ -323,10 +364,92 @@ async def charge_module_prompt_usage(
charge_key=CHARGE_TEXT_PROMPT,
biz_key=biz_key,
attempt_no=attempt_no,
record_meta=record_meta,
)
result = await db.execute(select(ModuleGenerationStep).where(ModuleGenerationStep.id == step_id).limit(1))
step = result.scalar_one_or_none()
if step:
step.token_usage_id = record_meta.token_usage_id
step.model_config_id = usage.get("model_config_id")
step.input_tokens = record_meta.input_tokens
step.output_tokens = record_meta.output_tokens
step.total_tokens = record_meta.total_tokens
step.text_credits_cost = text_credits
return BillingSummary(record_id=step_id, user_id=user_id, items=[item])
async def charge_shot_video_analysis_usage(
db: AsyncSession,
*,
user_id: str,
owner_type: str,
owner_id: str,
usage: Mapping[str, Any],
description: str,
billing_scene: str,
source_project_id: str | None = None,
source_step_id: str | None = None,
attempt_no: int | None = None,
) -> BillingSummary:
"""拆镜复刻视频分析扣分析积分。
视频分析属于“文字提示词 + 视频素材”的模型调用类消费，
按 input_tokens + output_tokens 参考文本积分规则计费，
但账务归类为 analysis/video_analysis，避免混入提词优化统计。
"""
attempt_no = attempt_no or await get_next_credit_attempt_no(
db,
owner_type=owner_type,
owner_id=owner_id,
charge_kind=CHARGE_VIDEO_ANALYSIS,
)
input_tokens = _safe_int(usage.get("input_tokens"))
output_tokens = _safe_int(usage.get("output_tokens"))
amount = await calc_text_credits(db, input_tokens, output_tokens)
biz_key = build_credit_biz_key(
owner_type=owner_type,
owner_id=owner_id,
attempt_no=attempt_no,
charge_kind=CHARGE_VIDEO_ANALYSIS,
action="charge",
)
record_meta = await build_shot_video_analysis_meta(
db,
owner_type=owner_type,
owner_id=owner_id,
attempt_no=attempt_no,
usage=usage,
billing_scene=billing_scene,
source_project_id=source_project_id,
source_step_id=source_step_id,
)
item = await deduct_credits_locked_once(
db,
user_id=user_id,
amount=amount,
description=description,
related_id=owner_id,
charge_key=CHARGE_VIDEO_ANALYSIS,
biz_key=biz_key,
attempt_no=attempt_no,
record_meta=record_meta,
)
if record_meta.token_usage_id:
result = await db.execute(select(TokenUsage).where(TokenUsage.id == record_meta.token_usage_id).limit(1))
token_usage = result.scalar_one_or_none()
if token_usage:
token_usage.owner_type = token_usage.owner_type or owner_type
token_usage.owner_id = token_usage.owner_id or owner_id
token_usage.biz_key = token_usage.biz_key or biz_key
token_usage.source_module = token_usage.source_module or CreditRecordSourceModule.SHOT_REPLICATE.value
token_usage.source_step_code = token_usage.source_step_code or "video_analysis"
return BillingSummary(record_id=owner_id, user_id=user_id, items=[item])
async def charge_generation_media_by_params(
db: AsyncSession,
*,
@@ -341,6 +464,11 @@ async def charge_generation_media_by_params(
description_prefix: str = "AI创作-",
owner_type: str = OWNER_CHAT_GENERATION_TASK,
attempt_no: int | None = None,
source_module: str | None = None,
source_project_id: str | None = None,
source_step_id: str | None = None,
source_step_code: str | None = None,
billing_scene: str | None = None,
) -> BillingSummary:
"""图片/视频媒体生成扣费。
@@ -363,6 +491,19 @@ async def charge_generation_media_by_params(
action="charge",
)
items: list[BillingItem] = []
record_meta = await build_generation_media_meta(
db,
owner_type=owner_type,
owner_id=record_id,
attempt_no=attempt_no,
gen_type=gen_type,
engine_id=engine_id,
source_module=source_module,
source_project_id=source_project_id,
source_step_id=source_step_id,
source_step_code=source_step_code,
billing_scene=billing_scene,
)
if gen_type == "image":
size = image_size or "2K"
@@ -377,6 +518,7 @@ async def charge_generation_media_by_params(
charge_key=CHARGE_MEDIA,
biz_key=biz_key,
attempt_no=attempt_no,
record_meta=record_meta,
)
)
elif gen_type == "video":
@@ -391,6 +533,7 @@ async def charge_generation_media_by_params(
charge_key=CHARGE_MEDIA,
biz_key=biz_key,
attempt_no=attempt_no,
record_meta=record_meta,
)
)
else:
@@ -419,4 +562,5 @@ async def charge_generation_media_for_record(
description_prefix=description_prefix,
owner_type=OWNER_GENERATION_RECORD,
attempt_no=attempt_no,
source_module=CreditRecordSourceModule.GENERATION_RECORD.value,
)
video-gen-api/app/services/generation_prompt_service.py
+9 -1
View File
@@ -152,10 +152,13 @@ async def build_prompt_with_chatapi(db: AsyncSession, record: ChatGenerationTask
if not content:
raise RuntimeError("ChatAPI未返回有效prompt")
token_usage_id = generate_id()
db.add(TokenUsage(
id=generate_id(),
id=token_usage_id,
model_config_id=config.id,
user_id=record.user_id,
owner_type="generation_record",
owner_id=record.id,
input_tokens=input_tokens,
output_tokens=output_tokens,
total_tokens=total_tokens,
@@ -178,6 +181,11 @@ async def build_prompt_with_chatapi(db: AsyncSession, record: ChatGenerationTask
total_tokens=total_tokens,
)
return content, {
"token_usage_id": token_usage_id,
"model_config_id": config.id,
"model_config_name": config.name,
"model_provider": config.provider,
"model_name": config.model_name,
"input_tokens": input_tokens,
"output_tokens": output_tokens,
"total_tokens": total_tokens,
video-gen-api/app/services/generation_refund_service.py
+2
View File
@@ -10,6 +10,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.credits import refund_credits
from app.services.credit_record_meta_service import build_refund_meta_from_charge
from app.services.generation_billing_service import (
CHARGE_MEDIA,
OWNER_CHAT_GENERATION_TASK,
@@ -113,6 +114,7 @@ async def refund_unrefunded_media_charges(
related_id=owner_id,
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)
return total_refunded
video-gen-api/app/services/generation_task_factory_service.py
+15
View File
@@ -63,6 +63,11 @@ async def create_chat_generation_task_for_module(
resolution: str | None = None,
billing_project_name: str = "模块生成任务",
billing_description_prefix: str = "模块生成-",
billing_source_module: str | None = None,
billing_source_project_id: str | None = None,
billing_source_step_id: str | None = None,
billing_source_step_code: str | None = None,
billing_scene: str | None = None,
) -> ChatGenerationTask:
"""创建可复用的 ChatGenerationTask 子任务。
@@ -106,6 +111,11 @@ async def create_chat_generation_task_for_module(
description_prefix=billing_description_prefix,
owner_type=OWNER_CHAT_GENERATION_TASK,
attempt_no=1,
source_module=billing_source_module,
source_project_id=billing_source_project_id,
source_step_id=billing_source_step_id,
source_step_code=billing_source_step_code,
billing_scene=billing_scene,
)
snapshot = _build_image_snapshot(engine, size, proportion, px)
task = ChatGenerationTask(
@@ -155,6 +165,11 @@ async def create_chat_generation_task_for_module(
description_prefix=billing_description_prefix,
owner_type=OWNER_CHAT_GENERATION_TASK,
attempt_no=1,
source_module=billing_source_module,
source_project_id=billing_source_project_id,
source_step_id=billing_source_step_id,
source_step_code=billing_source_step_code,
billing_scene=billing_scene,
)
snapshot = _build_video_snapshot(engine, ratio, selected_resolution, selected_duration)
task = ChatGenerationTask(
video-gen-api/app/services/hot_opening_replicate_service.py
+86 -35
View File
@@ -951,6 +951,24 @@ async def generate_image_from_prompt(
{"type": "image", "url": material.get("material_image_url"), "name": "新产品图片"},
]
step = await _create_step(
db,
project=project,
step_code=HotOpeningStepCodeEnum.IMAGE_GENERATE.value,
status=ModuleStepStatusEnum.PROCESSING.value,
parent_step_id=prompt_step.id,
source_step_id=prompt_step.id,
input_data={
"engine_id": req.engine_id,
"params": {
"image_size": req.image_size,
"image_proportion": req.image_proportion,
"image_px": req.image_px,
},
"prompt": optimized_prompt,
"media_references": refs,
},
)
chat_task = await create_chat_generation_task_for_module(
db,
current_user=current_user,
@@ -965,21 +983,30 @@ async def generate_image_from_prompt(
image_px=req.image_px,
billing_project_name=project.title or "爆款开头复刻",
billing_description_prefix="爆款开头复刻图片生成",
billing_source_module=project.module,
billing_source_project_id=project.id,
billing_source_step_id=step.id,
billing_source_step_code=HotOpeningStepCodeEnum.IMAGE_GENERATE.value,
)
step = await _create_step(
db,
project=project,
step_code=HotOpeningStepCodeEnum.IMAGE_GENERATE.value,
status=ModuleStepStatusEnum.PROCESSING.value,
parent_step_id=prompt_step.id,
source_step_id=prompt_step.id,
chat_task_id=chat_task.id,
input_data={
"engine_id": chat_task.engine_id,
"params": {"image_size": chat_task.image_size, "image_proportion": chat_task.image_proportion, "image_px": chat_task.image_px},
"prompt": optimized_prompt,
"media_references": refs,
},
step.chat_task_id = chat_task.id
_force_set_json(
step,
"input_json",
_step_input(
step_code=HotOpeningStepCodeEnum.IMAGE_GENERATE.value,
source_step_id=prompt_step.id,
parent_step_id=prompt_step.id,
payload={
"engine_id": chat_task.engine_id,
"params": {
"image_size": chat_task.image_size,
"image_proportion": chat_task.image_proportion,
"image_px": chat_task.image_px,
},
"prompt": optimized_prompt,
"media_references": refs,
},
),
)
project.status = ModuleProjectStatusEnum.PROCESSING.value
project.current_step_code = HotOpeningStepCodeEnum.IMAGE_GENERATE.value
@@ -1261,6 +1288,25 @@ async def generate_video_from_prompt(
{"type": "image", "url": _build_file_url_or_data_uri(generated_image_url), "name": "新项目图片"},
]
step = await _create_step(
db,
project=project,
step_code=HotOpeningStepCodeEnum.VIDEO_GENERATE.value,
status=ModuleStepStatusEnum.PROCESSING.value,
parent_step_id=prompt_step.id,
source_step_id=prompt_step.id,
input_data={
"engine_id": req.engine_id or prompt_params.get("engine_id"),
"params": {
"duration": duration,
"aspect_ratio": aspect_ratio,
"resolution": resolution,
},
"prompt_schema": prompt_schema,
"final_prompt": final_prompt,
"media_references": refs,
},
)
chat_task = await create_chat_generation_task_for_module(
db,
current_user=current_user,
@@ -1275,29 +1321,34 @@ async def generate_video_from_prompt(
resolution=resolution,
billing_project_name=project.title or "爆款开头复刻",
billing_description_prefix="爆款开头复刻视频生成",
billing_source_module=project.module,
billing_source_project_id=project.id,
billing_source_step_id=step.id,
billing_source_step_code=HotOpeningStepCodeEnum.VIDEO_GENERATE.value,
)
step = await _create_step(
db,
project=project,
step_code=HotOpeningStepCodeEnum.VIDEO_GENERATE.value,
status=ModuleStepStatusEnum.PROCESSING.value,
parent_step_id=prompt_step.id,
source_step_id=prompt_step.id,
chat_task_id=chat_task.id,
input_data={
"engine_id": chat_task.engine_id,
"params": {
"duration": chat_task.duration,
"aspect_ratio": chat_task.aspect_ratio,
"resolution": chat_task.resolution,
"image_size": chat_task.image_size,
"image_proportion": chat_task.image_proportion,
"image_px": chat_task.image_px,
step.chat_task_id = chat_task.id
_force_set_json(
step,
"input_json",
_step_input(
step_code=HotOpeningStepCodeEnum.VIDEO_GENERATE.value,
source_step_id=prompt_step.id,
parent_step_id=prompt_step.id,
payload={
"engine_id": chat_task.engine_id,
"params": {
"duration": chat_task.duration,
"aspect_ratio": chat_task.aspect_ratio,
"resolution": chat_task.resolution,
"image_size": chat_task.image_size,
"image_proportion": chat_task.image_proportion,
"image_px": chat_task.image_px,
},
"prompt_schema": prompt_schema,
"final_prompt": final_prompt,
"media_references": refs,
},
"prompt_schema": prompt_schema,
"final_prompt": final_prompt,
"media_references": refs,
},
),
)
project.status = ModuleProjectStatusEnum.PROCESSING.value
project.current_step_code = HotOpeningStepCodeEnum.VIDEO_GENERATE.value
video-gen-api/app/services/hot_opening_video_prompt_service.py
+9 -1
View File
@@ -1488,9 +1488,10 @@ async def optimize_hot_opening_video_prompt(
"total_tokens": int(usage.get("total_tokens") or 0),
"log_user_message": log_user_message,
}
token_usage_id = generate_id()
db.add(
TokenUsage(
id=generate_id(),
id=token_usage_id,
model_config_id=config.id,
user_id=user_id,
input_tokens=token_usage["input_tokens"],
@@ -1499,6 +1500,13 @@ async def optimize_hot_opening_video_prompt(
)
)
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,
"model_name": config.model_name,
})
result = parse_model_json(content)
result = normalize_video_prompt_schema_from_ai(result, video_config, schema_config_snapshot)
video-gen-api/app/services/llm.py
+8 -1
View File
@@ -370,9 +370,11 @@ async def _call_openai_compatible(
output_tokens = usage.get("completion_tokens", 0)
total_tokens = usage.get("total_tokens", input_tokens + output_tokens)
token_usage_id = None
if db is not None:
token_usage_id = generate_id()
record = TokenUsage(
id=generate_id(),
id=token_usage_id,
model_config_id=config.id,
user_id=user_id,
input_tokens=input_tokens,
@@ -384,6 +386,11 @@ async def _call_openai_compatible(
content = data["choices"][0]["message"]["content"].strip()
token_usage = {
"token_usage_id": token_usage_id,
"model_config_id": config.id,
"model_config_name": config.name,
"model_provider": config.provider,
"model_name": config.model_name,
"input_tokens": input_tokens,
"output_tokens": output_tokens,
"total_tokens": total_tokens,
video-gen-api/app/services/payment.py
+65
View File
@@ -17,9 +17,45 @@ from sqlalchemy.ext.asyncio import AsyncSession
from app.config import settings
from app.models.payment_order import PaymentOrder
from app.models.system_config import SystemConfig
from app.enums.credit_record import (
CreditRecordAction,
CreditRecordBillingScene,
CreditRecordChargeKind,
CreditRecordOwnerType,
CreditRecordSourceModule,
CreditRecordSubject,
)
from app.services.credit_record_meta_service import CreditRecordMeta, build_recharge_meta
from app.services.credits import add_credits, deduct_credits
from app.utils.id_gen import generate_id, generate_order_no
def _payment_biz_key(order: PaymentOrder, *, charge_kind: str, action: str) -> str:
return (
f"{CreditRecordOwnerType.PAYMENT_ORDER.value}:{order.id}:"
f"attempt:1:{charge_kind}:{action}"
)
def _payment_recharge_meta(order: PaymentOrder) -> CreditRecordMeta:
meta = build_recharge_meta(owner_id=order.id)
meta.attempt_no = 1
return meta
def _payment_refund_meta(order: PaymentOrder) -> CreditRecordMeta:
return CreditRecordMeta(
owner_type=CreditRecordOwnerType.PAYMENT_ORDER.value,
owner_id=order.id,
attempt_no=1,
charge_kind=CreditRecordChargeKind.REFUND.value,
charge_action=CreditRecordAction.REFUND.value,
credit_subject=CreditRecordSubject.REFUND.value,
billing_scene=CreditRecordBillingScene.REFUND.value,
source_module=CreditRecordSourceModule.PAYMENT.value,
)
# ---------------------------------------------------------------------------
# Payment logger → log/payment/YYYY-MM-DD.log (one file per day, no cleanup)
# ---------------------------------------------------------------------------
@@ -321,6 +357,12 @@ async def create_recharge_order(
total_credits,
desc,
related_id=order.id,
biz_key=_payment_biz_key(
order,
charge_kind=CreditRecordChargeKind.RECHARGE.value,
action=CreditRecordAction.CHARGE.value,
),
record_meta=_payment_recharge_meta(order),
)
await db.flush()
else:
@@ -1094,6 +1136,12 @@ async def process_payment_success(db: AsyncSession, order_id: str):
order.credits,
f"充值成功({order.credits}积分)",
related_id=order.id,
biz_key=_payment_biz_key(
order,
charge_kind=CreditRecordChargeKind.RECHARGE.value,
action=CreditRecordAction.CHARGE.value,
),
record_meta=_payment_recharge_meta(order),
)
await db.commit()
@@ -1152,6 +1200,12 @@ async def process_payment_success_by_order_no(
order.credits,
f"充值成功({order.credits}积分), 订单号: {order_no}, 金额: {order.amount}",
related_id=order.id,
biz_key=_payment_biz_key(
order,
charge_kind=CreditRecordChargeKind.RECHARGE.value,
action=CreditRecordAction.CHARGE.value,
),
record_meta=_payment_recharge_meta(order),
)
await db.commit()
logger.info(
@@ -1220,6 +1274,17 @@ async def process_refund(
order.credits,
refund_reason,
related_id=order.id,
biz_key=_payment_biz_key(
order,
charge_kind=CreditRecordChargeKind.REFUND.value,
action=CreditRecordAction.REFUND.value,
),
refund_for_biz_key=_payment_biz_key(
order,
charge_kind=CreditRecordChargeKind.RECHARGE.value,
action=CreditRecordAction.CHARGE.value,
),
record_meta=_payment_refund_meta(order),
)
except Exception as e:
logger.exception(f"Failed to deduct credits for refund: {e}")
video-gen-api/app/services/shot_replicate_flow_service.py
+86 -35
View File
@@ -903,6 +903,24 @@ async def generate_image_from_prompt(
{"type": "image", "url": material.get("material_image_url"), "name": "新产品图片"},
]
step = await _create_step(
db,
project=project,
step_code=ShotReplicateStepCodeEnum.IMAGE_GENERATE.value,
status=ModuleStepStatusEnum.PROCESSING.value,
parent_step_id=prompt_step.id,
source_step_id=prompt_step.id,
input_data={
"engine_id": req.engine_id,
"params": {
"image_size": req.image_size,
"image_proportion": req.image_proportion,
"image_px": req.image_px,
},
"prompt": optimized_prompt,
"media_references": refs,
},
)
chat_task = await create_chat_generation_task_for_module(
db,
current_user=current_user,
@@ -917,21 +935,30 @@ async def generate_image_from_prompt(
image_px=req.image_px,
billing_project_name=project.title or "拆镜复刻",
billing_description_prefix="拆镜复刻图片生成",
billing_source_module=project.module,
billing_source_project_id=project.id,
billing_source_step_id=step.id,
billing_source_step_code=ShotReplicateStepCodeEnum.IMAGE_GENERATE.value,
)
step = await _create_step(
db,
project=project,
step_code=ShotReplicateStepCodeEnum.IMAGE_GENERATE.value,
status=ModuleStepStatusEnum.PROCESSING.value,
parent_step_id=prompt_step.id,
source_step_id=prompt_step.id,
chat_task_id=chat_task.id,
input_data={
"engine_id": chat_task.engine_id,
"params": {"image_size": chat_task.image_size, "image_proportion": chat_task.image_proportion, "image_px": chat_task.image_px},
"prompt": optimized_prompt,
"media_references": refs,
},
step.chat_task_id = chat_task.id
_force_set_json(
step,
"input_json",
_step_input(
step_code=ShotReplicateStepCodeEnum.IMAGE_GENERATE.value,
source_step_id=prompt_step.id,
parent_step_id=prompt_step.id,
payload={
"engine_id": chat_task.engine_id,
"params": {
"image_size": chat_task.image_size,
"image_proportion": chat_task.image_proportion,
"image_px": chat_task.image_px,
},
"prompt": optimized_prompt,
"media_references": refs,
},
),
)
project.status = ModuleProjectStatusEnum.PROCESSING.value
project.current_step_code = ShotReplicateStepCodeEnum.IMAGE_GENERATE.value
@@ -1223,6 +1250,25 @@ async def generate_video_from_prompt(
{"type": "image", "url": _build_file_url_or_data_uri(generated_image_url), "name": "新项目图片"},
]
step = await _create_step(
db,
project=project,
step_code=ShotReplicateStepCodeEnum.VIDEO_GENERATE.value,
status=ModuleStepStatusEnum.PROCESSING.value,
parent_step_id=prompt_step.id,
source_step_id=prompt_step.id,
input_data={
"engine_id": req.engine_id or prompt_params.get("engine_id"),
"params": {
"duration": duration,
"aspect_ratio": aspect_ratio,
"resolution": resolution,
},
"prompt_schema": prompt_schema,
"final_prompt": final_prompt,
"media_references": refs,
},
)
chat_task = await create_chat_generation_task_for_module(
db,
current_user=current_user,
@@ -1237,29 +1283,34 @@ async def generate_video_from_prompt(
resolution=resolution,
billing_project_name=project.title or "拆镜复刻",
billing_description_prefix="拆镜复刻视频生成",
billing_source_module=project.module,
billing_source_project_id=project.id,
billing_source_step_id=step.id,
billing_source_step_code=ShotReplicateStepCodeEnum.VIDEO_GENERATE.value,
)
step = await _create_step(
db,
project=project,
step_code=ShotReplicateStepCodeEnum.VIDEO_GENERATE.value,
status=ModuleStepStatusEnum.PROCESSING.value,
parent_step_id=prompt_step.id,
source_step_id=prompt_step.id,
chat_task_id=chat_task.id,
input_data={
"engine_id": chat_task.engine_id,
"params": {
"duration": chat_task.duration,
"aspect_ratio": chat_task.aspect_ratio,
"resolution": chat_task.resolution,
"image_size": chat_task.image_size,
"image_proportion": chat_task.image_proportion,
"image_px": chat_task.image_px,
step.chat_task_id = chat_task.id
_force_set_json(
step,
"input_json",
_step_input(
step_code=ShotReplicateStepCodeEnum.VIDEO_GENERATE.value,
source_step_id=prompt_step.id,
parent_step_id=prompt_step.id,
payload={
"engine_id": chat_task.engine_id,
"params": {
"duration": chat_task.duration,
"aspect_ratio": chat_task.aspect_ratio,
"resolution": chat_task.resolution,
"image_size": chat_task.image_size,
"image_proportion": chat_task.image_proportion,
"image_px": chat_task.image_px,
},
"prompt_schema": prompt_schema,
"final_prompt": final_prompt,
"media_references": refs,
},
"prompt_schema": prompt_schema,
"final_prompt": final_prompt,
"media_references": refs,
},
),
)
project.status = ModuleProjectStatusEnum.PROCESSING.value
project.current_step_code = ShotReplicateStepCodeEnum.VIDEO_GENERATE.value
video-gen-api/app/services/shot_video_analysis_service.py
+9 -1
View File
@@ -505,9 +505,10 @@ async def analyze_video_for_shot_split(
if not token_usage["total_tokens"]:
token_usage["total_tokens"] = token_usage["input_tokens"] + token_usage["output_tokens"]
token_usage_id = generate_id()
db.add(
TokenUsage(
id=generate_id(),
id=token_usage_id,
model_config_id=config.id,
user_id=user_id,
input_tokens=token_usage["input_tokens"],
@@ -516,6 +517,13 @@ async def analyze_video_for_shot_split(
)
)
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,
"model_name": config.model_name,
})
return ShotVideoAnalysisResult(result=result, raw_response=raw, usage=token_usage)