1499 lines
62 KiB
Go
1499 lines
62 KiB
Go
package worker
|
||
|
||
import (
|
||
"bufio"
|
||
"bytes"
|
||
"context"
|
||
"crypto/sha256"
|
||
"encoding/hex"
|
||
"encoding/json"
|
||
"errors"
|
||
"fmt"
|
||
"io"
|
||
"log/slog"
|
||
"net/http"
|
||
"os"
|
||
"os/exec"
|
||
"regexp"
|
||
"sort"
|
||
"strconv"
|
||
"strings"
|
||
"time"
|
||
|
||
"juhe-factory/api/internal/billing"
|
||
mediakey "juhe-factory/api/internal/media"
|
||
"juhe-factory/api/internal/model"
|
||
"juhe-factory/api/internal/provider/apimart"
|
||
queuepkg "juhe-factory/api/internal/queue"
|
||
"juhe-factory/api/internal/security"
|
||
"juhe-factory/api/internal/storage"
|
||
|
||
"github.com/google/uuid"
|
||
"github.com/hibiken/asynq"
|
||
"gorm.io/gorm"
|
||
"gorm.io/gorm/clause"
|
||
)
|
||
|
||
type Media struct {
|
||
DB *gorm.DB
|
||
COS *storage.COS
|
||
Encryptor *security.Encryptor
|
||
Provider apimart.Client
|
||
HTTP *http.Client
|
||
FFmpeg string
|
||
FFprobe string
|
||
Python string
|
||
ASRScript string
|
||
localSlot chan struct{}
|
||
}
|
||
|
||
type transcriptWord struct {
|
||
Start float64 `json:"start"`
|
||
End float64 `json:"end"`
|
||
Text string `json:"text"`
|
||
Probability *float64 `json:"probability"`
|
||
}
|
||
|
||
type transcriptSegment struct {
|
||
Start float64 `json:"start"`
|
||
End float64 `json:"end"`
|
||
Text string `json:"text"`
|
||
Words []transcriptWord `json:"words"`
|
||
Confidence float64 `json:"confidence"`
|
||
NeedsReview bool `json:"needs_review"`
|
||
}
|
||
|
||
type transcriptResult struct {
|
||
Language string `json:"language"`
|
||
Error string `json:"error"`
|
||
Segments []transcriptSegment `json:"segments"`
|
||
}
|
||
|
||
type reverseAsset struct {
|
||
Type string `json:"type"`
|
||
Name string `json:"name"`
|
||
ImagePrompt string `json:"image_prompt"`
|
||
}
|
||
|
||
type reverseSegment struct {
|
||
Title string `json:"title"`
|
||
Prompt string `json:"prompt"`
|
||
AssetNames []string `json:"asset_names"`
|
||
Dialogue []map[string]any `json:"dialogue"`
|
||
}
|
||
|
||
type reverseResponse struct {
|
||
Assets []reverseAsset `json:"assets"`
|
||
Segment reverseSegment `json:"segment"`
|
||
Raw json.RawMessage `json:"-"`
|
||
InputTokens int64 `json:"-"`
|
||
OutputTokens int64 `json:"-"`
|
||
UsageRaw json.RawMessage `json:"-"`
|
||
}
|
||
|
||
type analysisSnapshot struct {
|
||
TaskID uuid.UUID
|
||
UserID uuid.UUID
|
||
ProjectID uuid.UUID
|
||
EpisodeID *uuid.UUID
|
||
EpisodeNo int
|
||
StoryboardID *uuid.UUID
|
||
SourceLanguage string
|
||
AudioSource string
|
||
SourceVideoURL string
|
||
SubtitleURL string
|
||
AspectRatio string
|
||
EraType string
|
||
CustomEra string
|
||
Localization string
|
||
StyleName string
|
||
ModelName string
|
||
BaseURL string
|
||
APIKeyCiphertext string
|
||
TaskStatus string
|
||
Mode string
|
||
ScriptPrompt string
|
||
PromptSnapshot string
|
||
BillingSnapshot json.RawMessage
|
||
}
|
||
|
||
type fixedSegment struct {
|
||
Index int
|
||
Start float64
|
||
End float64
|
||
}
|
||
|
||
func NewMedia(db *gorm.DB, cos *storage.COS, encryptor *security.Encryptor, client *http.Client, ffmpeg, ffprobe, python, asrScript string) *Media {
|
||
return &Media{DB: db, COS: cos, Encryptor: encryptor, Provider: apimart.NewClient(client), HTTP: client, FFmpeg: ffmpeg, FFprobe: ffprobe, Python: python, ASRScript: asrScript, localSlot: make(chan struct{}, 1)}
|
||
}
|
||
|
||
func (w *Media) Register(mux *asynq.ServeMux) {
|
||
mux.HandleFunc(queuepkg.TypeAnalyzeEpisode, w.analyzeEpisode)
|
||
}
|
||
|
||
func (w *Media) analyzeEpisode(ctx context.Context, raw *asynq.Task) error {
|
||
taskID, err := decodeID(raw)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
w.localSlot <- struct{}{}
|
||
defer func() { <-w.localSlot }()
|
||
snapshot, err := w.loadAnalysisSnapshot(ctx, taskID)
|
||
if err != nil {
|
||
var status string
|
||
_ = w.DB.WithContext(ctx).Model(&model.GenerationTask{}).Select("status").Where("id=?", taskID).Scan(&status).Error
|
||
if status == "cancel_requested" || status == "cancelled" {
|
||
return w.finishAnalysisCancellation(ctx, taskID, uuid.Nil)
|
||
}
|
||
return w.failAnalysis(ctx, taskID, uuid.Nil, err)
|
||
}
|
||
if snapshot.TaskStatus == "cancel_requested" {
|
||
return w.finishAnalysisCancellation(ctx, taskID, snapshot.ProjectID)
|
||
}
|
||
if snapshot.Mode == "script" {
|
||
return w.reverseProjectScript(ctx, snapshot)
|
||
}
|
||
if snapshot.SourceVideoURL == "" || w.COS == nil {
|
||
return w.failAnalysis(ctx, taskID, snapshot.ProjectID, errors.New("原视频或 COS 配置不可用"))
|
||
}
|
||
w.DB.WithContext(ctx).Model(&model.GenerationTask{}).Where("id=? AND task_type='prompt_reverse' AND status='submitted'", taskID).Update("status", "processing")
|
||
if snapshot.StoryboardID == nil {
|
||
w.updateRedrawScope(ctx, snapshot, "analyzing", "正在下载原视频")
|
||
}
|
||
videoPath, cleanup, err := w.downloadTemporary(ctx, snapshot.SourceVideoURL, ".video")
|
||
if err != nil {
|
||
return w.failAnalysis(ctx, taskID, snapshot.ProjectID, err)
|
||
}
|
||
defer cleanup()
|
||
duration, err := w.probeDuration(ctx, videoPath)
|
||
mediaDuration := duration
|
||
duration = normalizeAnalysisDuration(duration)
|
||
if err != nil || duration < 5 {
|
||
if err == nil {
|
||
err = errors.New("视频时长不足 5 秒")
|
||
}
|
||
return w.failAnalysis(ctx, taskID, snapshot.ProjectID, err)
|
||
}
|
||
segments := buildFixedSegments(duration)
|
||
if snapshot.StoryboardID != nil {
|
||
var storyboard model.EpisodeStoryboard
|
||
query := w.DB.WithContext(ctx).Where("id=? AND deleted_at IS NULL", *snapshot.StoryboardID)
|
||
if snapshot.EpisodeID != nil {
|
||
query = query.Where("episode_id=?", *snapshot.EpisodeID)
|
||
} else {
|
||
query = query.Where("project_id=?", snapshot.ProjectID)
|
||
}
|
||
if err := query.Take(&storyboard).Error; err != nil {
|
||
return w.failAnalysis(ctx, taskID, snapshot.ProjectID, err)
|
||
}
|
||
segments = []fixedSegment{{Index: storyboard.SequenceNo, Start: float64(storyboard.StartMS) / 1000, End: float64(storyboard.EndMS) / 1000}}
|
||
} else if err := w.prepareAnalysisStoryboards(ctx, snapshot, segments); err != nil {
|
||
return w.failAnalysis(ctx, taskID, snapshot.ProjectID, err)
|
||
}
|
||
var transcript transcriptResult
|
||
if snapshot.SubtitleURL != "" && (snapshot.AudioSource == "subtitle_file" || snapshot.StoryboardID != nil) {
|
||
transcript, err = w.loadSubtitle(ctx, snapshot.SubtitleURL, duration)
|
||
} else {
|
||
if snapshot.StoryboardID == nil {
|
||
w.updateRedrawScope(ctx, snapshot, "analyzing", "正在识别视频台词")
|
||
}
|
||
transcript, err = w.transcribe(ctx, videoPath, snapshot.SourceLanguage)
|
||
if err == nil {
|
||
err = w.storeTranscript(ctx, snapshot, transcript)
|
||
}
|
||
}
|
||
if err != nil {
|
||
return w.failAnalysis(ctx, taskID, snapshot.ProjectID, err)
|
||
}
|
||
apiKey, err := w.Encryptor.Decrypt(snapshot.APIKeyCiphertext)
|
||
if err != nil {
|
||
return w.failAnalysis(ctx, taskID, snapshot.ProjectID, errors.New("反推渠道密钥不可用"))
|
||
}
|
||
knownAssets := make([]map[string]any, 0)
|
||
_ = w.DB.WithContext(ctx).Table("project_assets").Select("id,asset_type AS type,name,image_prompt").Where("project_id=? AND deleted_at IS NULL", snapshot.ProjectID).Order("created_at").Find(&knownAssets).Error
|
||
for position, segment := range segments {
|
||
if w.analysisCancellationRequested(ctx, taskID) {
|
||
return w.finishAnalysisCancellation(ctx, taskID, snapshot.ProjectID)
|
||
}
|
||
if err := w.startAnalysisSegment(ctx, snapshot, segment); err != nil {
|
||
return w.failAnalysis(ctx, taskID, snapshot.ProjectID, err)
|
||
}
|
||
if snapshot.StoryboardID == nil {
|
||
message := fmt.Sprintf("正在抽取分镜 %d/%d 的画面", position+1, len(segments))
|
||
w.updateRedrawScope(ctx, snapshot, "analyzing", message)
|
||
}
|
||
samplingSegment := segment
|
||
if samplingSegment.End > mediaDuration {
|
||
samplingSegment.End = mediaDuration
|
||
}
|
||
frames, err := w.extractAndStoreFrames(ctx, snapshot, videoPath, samplingSegment)
|
||
if err != nil {
|
||
return w.failAnalysis(ctx, taskID, snapshot.ProjectID, err)
|
||
}
|
||
if w.analysisCancellationRequested(ctx, taskID) {
|
||
return w.finishAnalysisCancellation(ctx, taskID, snapshot.ProjectID)
|
||
}
|
||
message := "正在反推当前分镜"
|
||
if snapshot.StoryboardID == nil {
|
||
message = fmt.Sprintf("正在反推分镜 %d/%d", position+1, len(segments))
|
||
}
|
||
if err := w.startSegmentInference(ctx, snapshot, message); err != nil {
|
||
return w.failAnalysis(ctx, taskID, snapshot.ProjectID, err)
|
||
}
|
||
response, err := w.reverseSegment(ctx, snapshot, segment, frames, transcriptForRange(transcript.Segments, segment.Start, segment.End), knownAssets, apiKey)
|
||
if err != nil {
|
||
return w.failAnalysis(ctx, taskID, snapshot.ProjectID, err)
|
||
}
|
||
if w.analysisCancellationRequested(ctx, taskID) {
|
||
return w.finishAnalysisCancellation(ctx, taskID, snapshot.ProjectID)
|
||
}
|
||
persisted, err := w.persistReverseResult(ctx, snapshot, segment, frames[0].ID, response)
|
||
if err != nil {
|
||
return w.failAnalysis(ctx, taskID, snapshot.ProjectID, err)
|
||
}
|
||
knownAssets = persisted
|
||
}
|
||
return w.DB.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
||
var task model.GenerationTask
|
||
if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Where("id=?", taskID).First(&task).Error; err != nil {
|
||
return err
|
||
}
|
||
if task.Status == "cancel_requested" || task.Status == "cancelled" {
|
||
return w.finishAnalysisCancellationLocked(tx, &task, snapshot.ProjectID)
|
||
}
|
||
if err := tx.Model(&task).Updates(map[string]any{"status": "succeeded", "finished_at": time.Now(), "actual_points": task.PrepaidPoints, "error_code": nil, "error_message": nil}).Error; err != nil {
|
||
return err
|
||
}
|
||
if task.StoryboardID == nil {
|
||
if snapshot.EpisodeID != nil {
|
||
return tx.Model(&model.ProjectEpisode{}).Where("id=?", *snapshot.EpisodeID).Updates(map[string]any{"status": "review", "analysis_message": "全部分镜提示词已完成，可以反推为剧本"}).Error
|
||
}
|
||
return tx.Model(&model.CreativeProject{}).Where("id=?", snapshot.ProjectID).Updates(map[string]any{"redraw_status": "review", "analysis_message": "全部分镜提示词已完成，可以反推为剧本"}).Error
|
||
}
|
||
return w.activateRedrawScriptIfReady(tx, snapshot)
|
||
})
|
||
}
|
||
|
||
// isWholeVideoAnalysis 判断当前任务是否为整段视频反推，单分镜重推不得改写剧集或项目的总状态。
|
||
func isWholeVideoAnalysis(snapshot analysisSnapshot) bool {
|
||
return snapshot.StoryboardID == nil
|
||
}
|
||
|
||
// 记录分镜正式进入 AI 反推的时间；仅整段视频反推更新剧集或项目的总处理阶段。
|
||
func (w *Media) startSegmentInference(ctx context.Context, snapshot analysisSnapshot, message string) error {
|
||
if err := w.DB.WithContext(ctx).Model(&model.GenerationTask{}).Where("id=?", snapshot.TaskID).Update(
|
||
"input_data",
|
||
gorm.Expr("jsonb_set(jsonb_set(coalesce(input_data,'{}'::jsonb), '{phase}', to_jsonb(?::text), true), '{phase_started_at}', to_jsonb(CURRENT_TIMESTAMP), true)", "inference"),
|
||
).Error; err != nil {
|
||
return err
|
||
}
|
||
if isWholeVideoAnalysis(snapshot) {
|
||
w.updateRedrawScope(ctx, snapshot, "analyzing", message)
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// redrawStoryboardsReady 判断当前范围内是否存在分镜，且每个分镜都具备反推提示词。
|
||
func redrawStoryboardsReady(total, incomplete int64) bool {
|
||
return total > 0 && incomplete == 0
|
||
}
|
||
|
||
// activateRedrawScriptIfReady 在单分镜反推成功后复核全部分镜，完整时恢复剧本反推入口。
|
||
// 分镜抽帧和资产图片均不属于剧本反推前置条件，完整性只依据实际参与模型调用的提示词。
|
||
func (w *Media) activateRedrawScriptIfReady(tx *gorm.DB, snapshot analysisSnapshot) error {
|
||
query := tx.Model(&model.EpisodeStoryboard{}).Where("deleted_at IS NULL")
|
||
if snapshot.EpisodeID != nil {
|
||
query = query.Where("episode_id=?", *snapshot.EpisodeID)
|
||
} else {
|
||
query = query.Where("project_id=?", snapshot.ProjectID)
|
||
}
|
||
var readiness struct {
|
||
Total int64
|
||
Incomplete int64
|
||
}
|
||
if err := query.Select(`count(*) AS total,
|
||
count(*) FILTER (WHERE trim(coalesce(prompt_content,''))='') AS incomplete`).
|
||
Scan(&readiness).Error; err != nil {
|
||
return err
|
||
}
|
||
if !redrawStoryboardsReady(readiness.Total, readiness.Incomplete) {
|
||
return nil
|
||
}
|
||
updates := map[string]any{"status": "review", "analysis_message": "全部分镜提示词已完成，可以反推为剧本"}
|
||
if snapshot.EpisodeID != nil {
|
||
return tx.Model(&model.ProjectEpisode{}).Where("id=?", *snapshot.EpisodeID).Updates(updates).Error
|
||
}
|
||
return tx.Model(&model.CreativeProject{}).Where("id=?", snapshot.ProjectID).Updates(map[string]any{
|
||
"redraw_status": "review", "analysis_message": updates["analysis_message"],
|
||
}).Error
|
||
}
|
||
|
||
func (w *Media) loadAnalysisSnapshot(ctx context.Context, taskID uuid.UUID) (analysisSnapshot, error) {
|
||
var result analysisSnapshot
|
||
err := w.DB.WithContext(ctx).Raw(`SELECT gt.id AS task_id,gt.user_id,gt.project_id,gt.episode_id,coalesce(e.episode_no,0) AS episode_no,gt.storyboard_id,
|
||
CASE WHEN gt.episode_id IS NOT NULL THEN coalesce(e.source_language,'') ELSE coalesce(p.source_language,'') END AS source_language,
|
||
CASE WHEN gt.episode_id IS NOT NULL THEN e.audio_source ELSE p.audio_source END AS audio_source,source.public_url AS source_video_url,
|
||
coalesce(sub.public_url,'') AS subtitle_url,p.aspect_ratio,p.era_type,coalesce(p.custom_era,'') AS custom_era,
|
||
p.localization,ps.name AS style_name,m.name AS model_name,c.base_url,c.api_key_ciphertext,gt.status AS task_status,
|
||
coalesce(gt.input_data->>'mode','') AS mode,coalesce(gt.input_data->>'prompt','') AS script_prompt,gt.prompt_snapshot,gt.billing_snapshot
|
||
FROM generation_tasks gt
|
||
JOIN creative_projects p ON p.id=gt.project_id AND p.user_id=gt.user_id
|
||
LEFT JOIN project_episodes e ON e.id=gt.episode_id AND e.project_id=p.id AND e.deleted_at IS NULL
|
||
JOIN project_styles ps ON ps.id=p.style_id
|
||
JOIN media_assets source ON source.id=CASE WHEN gt.episode_id IS NOT NULL THEN e.source_video_asset_id ELSE p.source_video_asset_id END
|
||
LEFT JOIN media_assets sub ON sub.id=CASE WHEN gt.episode_id IS NOT NULL THEN e.subtitle_asset_id ELSE p.subtitle_asset_id END
|
||
JOIN models m ON m.id=gt.model_id AND m.enabled=true AND m.deleted_at IS NULL
|
||
JOIN channels c ON c.id=m.channel_id AND c.enabled=true AND c.deleted_at IS NULL
|
||
WHERE gt.id=? AND p.project_type='video_redraw' AND gt.task_type='prompt_reverse' AND gt.status IN ('submitted','processing','cancel_requested')`, taskID).Scan(&result).Error
|
||
if err != nil {
|
||
return result, err
|
||
}
|
||
if result.TaskID == uuid.Nil {
|
||
return result, gorm.ErrRecordNotFound
|
||
}
|
||
return result, nil
|
||
}
|
||
|
||
func (w *Media) reverseProjectScript(ctx context.Context, snapshot analysisSnapshot) error {
|
||
type scriptStoryboard struct {
|
||
SequenceNo int `json:"sequence_no"`
|
||
StartMS int64 `json:"start_ms"`
|
||
EndMS int64 `json:"end_ms"`
|
||
PromptContent string `json:"prompt_content"`
|
||
}
|
||
storyboards := make([]scriptStoryboard, 0)
|
||
if err := json.Unmarshal([]byte(snapshot.PromptSnapshot), &storyboards); err != nil {
|
||
return w.failScriptReverse(ctx, snapshot, fmt.Errorf("分镜提示词快照无法解析: %w", err))
|
||
}
|
||
if len(storyboards) == 0 {
|
||
return w.failScriptReverse(ctx, snapshot, errors.New("暂无可用于反推剧本的分镜提示词"))
|
||
}
|
||
for _, storyboard := range storyboards {
|
||
if strings.TrimSpace(storyboard.PromptContent) == "" {
|
||
return w.failScriptReverse(ctx, snapshot, fmt.Errorf("分镜 %d 的提示词为空", storyboard.SequenceNo))
|
||
}
|
||
}
|
||
storyboardsJSON, _ := json.Marshal(storyboards)
|
||
promptTemplate := strings.TrimSpace(snapshot.ScriptPrompt)
|
||
if promptTemplate == "" {
|
||
return w.failScriptReverse(ctx, snapshot, errors.New("剧本反推提示词快照为空"))
|
||
}
|
||
userPrompt := promptTemplate + "\n\n" + string(storyboardsJSON)
|
||
apiKey, err := w.Encryptor.Decrypt(snapshot.APIKeyCiphertext)
|
||
if err != nil {
|
||
return w.failScriptReverse(ctx, snapshot, errors.New("反推渠道密钥不可用"))
|
||
}
|
||
pricing, err := billing.ParseTextPricingSnapshot(snapshot.BillingSnapshot)
|
||
if err != nil {
|
||
return w.failScriptReverse(ctx, snapshot, err)
|
||
}
|
||
if err := w.DB.WithContext(ctx).Model(&model.GenerationTask{}).
|
||
Where("id=? AND task_type='prompt_reverse' AND status='submitted'", snapshot.TaskID).
|
||
Update("status", "processing").Error; err != nil {
|
||
return w.failScriptReverse(ctx, snapshot, err)
|
||
}
|
||
if pricing.Mode == billing.TextBillingPerToken {
|
||
estimatedInput := billing.EstimateTextTokens(userPrompt)
|
||
if err := w.DB.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
||
_, reserveErr := billing.ReserveTextCall(tx, snapshot.UserID, snapshot.TaskID.String(), "script", pricing, estimatedInput, "剧本反推")
|
||
return reserveErr
|
||
}); err != nil {
|
||
return w.failScriptReverse(ctx, snapshot, err)
|
||
}
|
||
}
|
||
payload := map[string]any{
|
||
"model": snapshot.ModelName,
|
||
"messages": []map[string]any{
|
||
{"role": "user", "content": userPrompt},
|
||
},
|
||
"temperature": 0.2,
|
||
"response_format": map[string]any{"type": "json_object"},
|
||
}
|
||
var chat apimart.ChatResult
|
||
var result struct {
|
||
Script string `json:"script"`
|
||
}
|
||
var responseErr error
|
||
for attempt := 0; attempt < 2; attempt++ {
|
||
if attempt > 0 {
|
||
retryPrompt := userPrompt + "\n\n上次响应未能形成完整 JSON。请进一步合并重复描写，但仍需覆盖全部分镜、关键动作和台词，并优先保证 JSON 完整闭合。"
|
||
payload["messages"] = []map[string]any{
|
||
{"role": "user", "content": retryPrompt},
|
||
}
|
||
payload["temperature"] = 0
|
||
}
|
||
requestCtx, cancel := context.WithTimeout(ctx, 8*time.Minute)
|
||
chat, err = w.Provider.ChatWithUsage(requestCtx, snapshot.BaseURL, apiKey, payload)
|
||
cancel()
|
||
if err != nil {
|
||
return w.failScriptReverse(ctx, snapshot, err)
|
||
}
|
||
result.Script = ""
|
||
responseErr = json.Unmarshal([]byte(stripJSONFence(chat.Content)), &result)
|
||
result.Script = strings.TrimSpace(result.Script)
|
||
if responseErr == nil && result.Script == "" {
|
||
responseErr = errors.New("未返回完整剧本")
|
||
}
|
||
if responseErr == nil {
|
||
break
|
||
}
|
||
}
|
||
if responseErr != nil {
|
||
finishReason := strings.TrimSpace(chat.FinishReason)
|
||
if finishReason == "" {
|
||
finishReason = "未知"
|
||
}
|
||
return w.failScriptReverse(ctx, snapshot, fmt.Errorf("剧本模型连续两次返回无法解析（结束原因：%s）: %w", finishReason, responseErr))
|
||
}
|
||
if pricing.Mode == billing.TextBillingPerToken {
|
||
if err := w.DB.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
||
settled, settleErr := billing.SettleTextCall(tx, snapshot.UserID, snapshot.TaskID.String(), "script", pricing, billing.TextUsage{Input: chat.InputTokens, Output: chat.OutputTokens}, "剧本反推")
|
||
if settleErr != nil {
|
||
return settleErr
|
||
}
|
||
return tx.Model(&model.GenerationTask{}).Where("id=?", snapshot.TaskID).Updates(map[string]any{
|
||
"estimated_points": gorm.Expr("estimated_points+?::numeric", settled),
|
||
"prepaid_points": gorm.Expr("prepaid_points+?::numeric", settled),
|
||
"input_tokens": chat.InputTokens, "output_tokens": chat.OutputTokens,
|
||
"total_tokens": chat.InputTokens + chat.OutputTokens, "token_count_source": "upstream", "usage_raw": chat.UsageRaw,
|
||
}).Error
|
||
}); err != nil {
|
||
return w.failScriptReverse(ctx, snapshot, err)
|
||
}
|
||
}
|
||
return w.DB.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
||
var task model.GenerationTask
|
||
if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Where("id=?", snapshot.TaskID).Take(&task).Error; err != nil {
|
||
return err
|
||
}
|
||
if task.Status == "cancel_requested" || task.Status == "cancelled" {
|
||
return w.finishAnalysisCancellationLocked(tx, &task, snapshot.ProjectID)
|
||
}
|
||
if err := tx.Model(&task).Updates(map[string]any{
|
||
"status": "succeeded", "finished_at": time.Now(), "actual_points": task.PrepaidPoints,
|
||
"error_code": nil, "error_message": nil,
|
||
}).Error; err != nil {
|
||
return err
|
||
}
|
||
if snapshot.EpisodeID != nil {
|
||
return tx.Model(&model.ProjectEpisode{}).Where("id=?", *snapshot.EpisodeID).Updates(map[string]any{
|
||
"redraw_script": result.Script, "status": "review", "analysis_message": "完整剧本反推完成",
|
||
}).Error
|
||
}
|
||
return tx.Model(&model.CreativeProject{}).Where("id=?", snapshot.ProjectID).Updates(map[string]any{
|
||
"redraw_script": result.Script, "redraw_status": "review", "analysis_message": "完整剧本反推完成",
|
||
}).Error
|
||
})
|
||
}
|
||
|
||
func (w *Media) failScriptReverse(ctx context.Context, snapshot analysisSnapshot, taskErr error) error {
|
||
slog.ErrorContext(ctx, "完整剧本反推失败", "task_id", snapshot.TaskID, "project_id", snapshot.ProjectID, "error", taskErr)
|
||
return w.DB.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
||
var task model.GenerationTask
|
||
if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Where("id=?", snapshot.TaskID).Take(&task).Error; err != nil {
|
||
return err
|
||
}
|
||
if task.Status == "cancel_requested" || task.Status == "cancelled" {
|
||
return w.finishAnalysisCancellationLocked(tx, &task, snapshot.ProjectID)
|
||
}
|
||
if task.Status == "succeeded" || task.Status == "failed" {
|
||
return nil
|
||
}
|
||
refunded, err := billing.RefundTextGenerationTask(tx, &task, "剧本反推失败返还")
|
||
if err != nil {
|
||
return err
|
||
}
|
||
updates := map[string]any{"status": "failed", "actual_points": "0.00", "error_code": "script_reverse_failed", "error_message": truncate(taskErr.Error(), 4000), "finished_at": time.Now()}
|
||
if refunded {
|
||
updates["cost_refunded"] = true
|
||
}
|
||
if err := tx.Model(&task).Updates(updates).Error; err != nil {
|
||
return err
|
||
}
|
||
if snapshot.EpisodeID != nil {
|
||
return tx.Model(&model.ProjectEpisode{}).Where("id=?", *snapshot.EpisodeID).Updates(map[string]any{
|
||
"status": "review", "analysis_message": taskErr.Error(),
|
||
}).Error
|
||
}
|
||
return tx.Model(&model.CreativeProject{}).Where("id=?", snapshot.ProjectID).Updates(map[string]any{
|
||
"redraw_status": "review", "analysis_message": taskErr.Error(),
|
||
}).Error
|
||
})
|
||
}
|
||
|
||
func (w *Media) downloadTemporary(ctx context.Context, sourceURL, suffix string) (string, func(), error) {
|
||
request, err := http.NewRequestWithContext(ctx, http.MethodGet, sourceURL, nil)
|
||
if err != nil {
|
||
return "", func() {}, err
|
||
}
|
||
response, err := w.HTTP.Do(request)
|
||
if err != nil {
|
||
return "", func() {}, err
|
||
}
|
||
defer response.Body.Close()
|
||
if response.StatusCode < 200 || response.StatusCode >= 300 {
|
||
return "", func() {}, fmt.Errorf("读取 COS 媒体失败: HTTP %d", response.StatusCode)
|
||
}
|
||
file, err := os.CreateTemp("", "jcf-analysis-*"+suffix)
|
||
if err != nil {
|
||
return "", func() {}, err
|
||
}
|
||
path := file.Name()
|
||
cleanup := func() { _ = os.Remove(path) }
|
||
_, copyErr := io.Copy(file, response.Body)
|
||
closeErr := file.Close()
|
||
if copyErr != nil {
|
||
cleanup()
|
||
return "", func() {}, copyErr
|
||
}
|
||
if closeErr != nil {
|
||
cleanup()
|
||
return "", func() {}, closeErr
|
||
}
|
||
return path, cleanup, nil
|
||
}
|
||
|
||
func (w *Media) probeDuration(ctx context.Context, path string) (float64, error) {
|
||
command := exec.CommandContext(ctx, w.FFprobe, "-v", "error", "-show_entries", "format=duration", "-of", "default=noprint_wrappers=1:nokey=1", path)
|
||
output, err := command.Output()
|
||
if err != nil {
|
||
return 0, fmt.Errorf("无法读取视频时长: %w", err)
|
||
}
|
||
duration, err := strconv.ParseFloat(strings.TrimSpace(string(output)), 64)
|
||
if err != nil || duration <= 0 {
|
||
return 0, errors.New("视频时长无效")
|
||
}
|
||
return duration, nil
|
||
}
|
||
|
||
func (w *Media) transcribe(ctx context.Context, videoPath, sourceLanguage string) (transcriptResult, error) {
|
||
audio, err := os.CreateTemp("", "jcf-asr-*.wav")
|
||
if err != nil {
|
||
return transcriptResult{}, err
|
||
}
|
||
audioPath := audio.Name()
|
||
audio.Close()
|
||
defer os.Remove(audioPath)
|
||
filter := "highpass=f=80,lowpass=f=7600,afftdn=nf=-25,dynaudnorm=f=150:g=7"
|
||
args := []string{"-y", "-i", videoPath, "-vn", "-af", filter, "-ac", "1", "-ar", "16000", "-f", "wav", audioPath}
|
||
if output, err := exec.CommandContext(ctx, w.FFmpeg, args...).CombinedOutput(); err != nil {
|
||
fallback := []string{"-y", "-i", videoPath, "-vn", "-ac", "1", "-ar", "16000", "-f", "wav", audioPath}
|
||
if retryOutput, retryErr := exec.CommandContext(ctx, w.FFmpeg, fallback...).CombinedOutput(); retryErr != nil {
|
||
return transcriptResult{}, fmt.Errorf("音频提取失败: %s / %s", truncate(string(output), 300), truncate(string(retryOutput), 300))
|
||
}
|
||
}
|
||
command := exec.CommandContext(ctx, w.Python, w.ASRScript, audioPath, sourceLanguage)
|
||
var stderr bytes.Buffer
|
||
command.Stderr = &stderr
|
||
output, err := command.Output()
|
||
if err != nil {
|
||
detail := strings.TrimSpace(stderr.String())
|
||
if detail == "" {
|
||
detail = err.Error()
|
||
}
|
||
return transcriptResult{}, fmt.Errorf("音频识别失败: %s", truncate(detail, 500))
|
||
}
|
||
var result transcriptResult
|
||
if err := json.Unmarshal(output, &result); err != nil {
|
||
return result, errors.New("音频识别结果格式无效")
|
||
}
|
||
if result.Error != "" {
|
||
return result, fmt.Errorf("音频识别失败: %s", result.Error)
|
||
}
|
||
return result, nil
|
||
}
|
||
|
||
func (w *Media) storeTranscript(ctx context.Context, snapshot analysisSnapshot, transcript transcriptResult) error {
|
||
var content strings.Builder
|
||
content.WriteString("WEBVTT\n\n")
|
||
for index, segment := range transcript.Segments {
|
||
if segment.End <= segment.Start || strings.TrimSpace(segment.Text) == "" {
|
||
continue
|
||
}
|
||
fmt.Fprintf(&content, "%d\n%s --> %s\n%s\n\n", index+1, vttTimestamp(segment.Start), vttTimestamp(segment.End), strings.TrimSpace(segment.Text))
|
||
}
|
||
data := []byte(content.String())
|
||
mediaID := uuid.New()
|
||
key := mediakey.ProjectSubtitle(snapshot.ProjectID, mediaID, "transcript.vtt", "text/vtt")
|
||
if snapshot.EpisodeID != nil {
|
||
key = mediakey.EpisodeSubtitle(snapshot.ProjectID, snapshot.EpisodeNo, *snapshot.EpisodeID, mediaID, "transcript.vtt", "text/vtt")
|
||
}
|
||
url, err := w.COS.Put(ctx, key, "text/vtt", bytes.NewReader(data), int64(len(data)))
|
||
if err != nil {
|
||
return err
|
||
}
|
||
ownerID := snapshot.UserID
|
||
asset := model.MediaAsset{
|
||
ID: mediaID, OwnerUserID: &ownerID, StorageProvider: "cos", ObjectKey: key, PublicURL: url,
|
||
DisplayName: "视频识别字幕", MimeType: "text/vtt",
|
||
SizeBytes: int64(len(data)), SHA256: fmt.Sprintf("%x", sha256.Sum256(data)),
|
||
}
|
||
err = w.DB.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
||
if err := tx.Create(&asset).Error; err != nil {
|
||
return err
|
||
}
|
||
if snapshot.EpisodeID != nil {
|
||
return tx.Model(&model.ProjectEpisode{}).Where("id=?", *snapshot.EpisodeID).Update("subtitle_asset_id", mediaID).Error
|
||
}
|
||
return tx.Model(&model.CreativeProject{}).Where("id=?", snapshot.ProjectID).Update("subtitle_asset_id", mediaID).Error
|
||
})
|
||
if err != nil {
|
||
_ = w.COS.Delete(ctx, key)
|
||
}
|
||
return err
|
||
}
|
||
|
||
func vttTimestamp(seconds float64) string {
|
||
if seconds < 0 {
|
||
seconds = 0
|
||
}
|
||
totalMilliseconds := int64(seconds * 1000)
|
||
hours := totalMilliseconds / 3_600_000
|
||
minutes := (totalMilliseconds % 3_600_000) / 60_000
|
||
wholeSeconds := (totalMilliseconds % 60_000) / 1000
|
||
milliseconds := totalMilliseconds % 1000
|
||
return fmt.Sprintf("%02d:%02d:%02d.%03d", hours, minutes, wholeSeconds, milliseconds)
|
||
}
|
||
|
||
func buildFixedSegments(duration float64) []fixedSegment {
|
||
const minSeconds, baseSeconds = 5.0, 15.0
|
||
if duration <= baseSeconds {
|
||
return []fixedSegment{{Index: 1, Start: 0, End: duration}}
|
||
}
|
||
count := int(duration / baseSeconds)
|
||
remainder := duration - float64(count)*baseSeconds
|
||
boundaries := []float64{0}
|
||
if remainder == 0 || remainder >= minSeconds {
|
||
for index := 1; index <= count; index++ {
|
||
boundaries = append(boundaries, float64(index)*baseSeconds)
|
||
}
|
||
if remainder > 0 {
|
||
boundaries = append(boundaries, duration)
|
||
}
|
||
} else {
|
||
for index := 1; index < count; index++ {
|
||
boundaries = append(boundaries, float64(index)*baseSeconds)
|
||
}
|
||
lastStart := boundaries[len(boundaries)-1]
|
||
half := (duration - lastStart) / 2
|
||
boundaries = append(boundaries, lastStart+half, duration)
|
||
}
|
||
segments := make([]fixedSegment, 0, len(boundaries)-1)
|
||
for index := 0; index < len(boundaries)-1; index++ {
|
||
segments = append(segments, fixedSegment{Index: index + 1, Start: boundaries[index], End: boundaries[index+1]})
|
||
}
|
||
return segments
|
||
}
|
||
|
||
func normalizeAnalysisDuration(duration float64) float64 {
|
||
if duration > 15 && duration < 16 {
|
||
return 15
|
||
}
|
||
if duration > 4 && duration < 5 {
|
||
return 5
|
||
}
|
||
return duration
|
||
}
|
||
|
||
func storyboardDuration(segment fixedSegment) int {
|
||
duration := int(segment.End - segment.Start + 0.5)
|
||
if duration < 5 {
|
||
return 5
|
||
}
|
||
if duration > 15 {
|
||
return 15
|
||
}
|
||
return duration
|
||
}
|
||
|
||
func (w *Media) prepareAnalysisStoryboards(ctx context.Context, snapshot analysisSnapshot, segments []fixedSegment) error {
|
||
return w.DB.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
||
scopeID := analysisScopeID(snapshot)
|
||
for _, segment := range segments {
|
||
storyboardID := stableStoryboardID(scopeID, segment.Index)
|
||
if snapshot.EpisodeID != nil {
|
||
if err := tx.Exec(`INSERT INTO episode_storyboards(id,episode_id,sequence_no,stable_key,start_ms,end_ms,duration_seconds,dialogue,asset_refs,status)
|
||
VALUES(?,?,?,?,?,?,?,'[]'::jsonb,'[]'::jsonb,'queued')
|
||
ON CONFLICT(id) DO UPDATE SET start_ms=excluded.start_ms,end_ms=excluded.end_ms,duration_seconds=excluded.duration_seconds,
|
||
status='queued',updated_at=CURRENT_TIMESTAMP WHERE episode_storyboards.locked=false`,
|
||
storyboardID, *snapshot.EpisodeID, segment.Index, fmt.Sprintf("segment-%04d", segment.Index), int64(segment.Start*1000), int64(segment.End*1000), storyboardDuration(segment)).Error; err != nil {
|
||
return err
|
||
}
|
||
continue
|
||
}
|
||
if err := tx.Exec(`INSERT INTO episode_storyboards(id,project_id,sequence_no,stable_key,start_ms,end_ms,duration_seconds,dialogue,asset_refs,status)
|
||
VALUES(?,?,?,?,?,?,?,'[]'::jsonb,'[]'::jsonb,'queued')
|
||
ON CONFLICT(id) DO UPDATE SET start_ms=excluded.start_ms,end_ms=excluded.end_ms,duration_seconds=excluded.duration_seconds,
|
||
status='queued',updated_at=CURRENT_TIMESTAMP WHERE episode_storyboards.locked=false`,
|
||
storyboardID, snapshot.ProjectID, segment.Index, fmt.Sprintf("segment-%04d", segment.Index), int64(segment.Start*1000), int64(segment.End*1000), storyboardDuration(segment)).Error; err != nil {
|
||
return err
|
||
}
|
||
}
|
||
return nil
|
||
})
|
||
}
|
||
|
||
func (w *Media) startAnalysisSegment(ctx context.Context, snapshot analysisSnapshot, segment fixedSegment) error {
|
||
storyboardID := stableStoryboardID(analysisScopeID(snapshot), segment.Index)
|
||
if snapshot.StoryboardID != nil {
|
||
storyboardID = *snapshot.StoryboardID
|
||
}
|
||
return w.DB.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
||
if err := tx.Model(&model.GenerationTask{}).Where("id=?", snapshot.TaskID).
|
||
Update("input_data", gorm.Expr("jsonb_set(coalesce(input_data,'{}'::jsonb), '{current_sequence}', to_jsonb(?::int), true)", segment.Index)).Error; err != nil {
|
||
return err
|
||
}
|
||
return tx.Model(&model.EpisodeStoryboard{}).Where("id=?", storyboardID).Update("status", "generating").Error
|
||
})
|
||
}
|
||
|
||
func frameTimestamps(segment fixedSegment) []float64 {
|
||
duration := segment.End - segment.Start
|
||
values := []float64{segment.Start + 0.05, segment.End - 0.05}
|
||
for index := 0; index < 15; index++ {
|
||
values = append(values, segment.Start+(float64(index)+0.5)*(duration/15))
|
||
}
|
||
sort.Float64s(values)
|
||
unique := make([]float64, 0, 16)
|
||
for _, value := range values {
|
||
if value < segment.Start {
|
||
value = segment.Start
|
||
}
|
||
if value > segment.End-0.01 {
|
||
value = segment.End - 0.01
|
||
}
|
||
if len(unique) == 0 || value-unique[len(unique)-1] > 0.02 {
|
||
unique = append(unique, value)
|
||
}
|
||
}
|
||
if len(unique) <= 16 {
|
||
return unique
|
||
}
|
||
// Preserve both boundaries and distribute the remaining fourteen samples.
|
||
result := []float64{unique[0]}
|
||
for index := 1; index <= 14; index++ {
|
||
position := int(float64(index) * float64(len(unique)-1) / 15)
|
||
result = append(result, unique[position])
|
||
}
|
||
return append(result, unique[len(unique)-1])
|
||
}
|
||
|
||
func (w *Media) extractAndStoreFrames(ctx context.Context, snapshot analysisSnapshot, videoPath string, segment fixedSegment) ([]model.MediaAsset, error) {
|
||
frames := make([]model.MediaAsset, 0, 16)
|
||
for _, timestamp := range frameTimestamps(segment) {
|
||
file, err := os.CreateTemp("", "jcf-frame-*.jpg")
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
path := file.Name()
|
||
file.Close()
|
||
extract := func(at float64) ([]byte, error) {
|
||
args := []string{"-hide_banner", "-loglevel", "error", "-y", "-ss", fmt.Sprintf("%.3f", at), "-i", videoPath, "-frames:v", "1", "-vf", "scale=1024:1024:force_original_aspect_ratio=decrease", "-q:v", "3", "-update", "1", path}
|
||
return exec.CommandContext(ctx, w.FFmpeg, args...).CombinedOutput()
|
||
}
|
||
output, commandErr := extract(timestamp)
|
||
if commandErr != nil {
|
||
if retryTimestamp, ok := frameRetryTimestamp(segment, timestamp); ok {
|
||
output, commandErr = extract(retryTimestamp)
|
||
}
|
||
}
|
||
if commandErr != nil {
|
||
os.Remove(path)
|
||
detail := strings.TrimSpace(string(output))
|
||
if detail == "" {
|
||
detail = commandErr.Error()
|
||
}
|
||
return nil, fmt.Errorf("抽帧失败（%.3f秒）: %s", timestamp, truncate(detail, 1000))
|
||
}
|
||
info, err := os.Stat(path)
|
||
if err != nil {
|
||
os.Remove(path)
|
||
return nil, err
|
||
}
|
||
mediaID := uuid.New()
|
||
storyboardID := stableStoryboardID(analysisScopeID(snapshot), segment.Index)
|
||
key := mediakey.ProjectStoryboardFrame(snapshot.ProjectID, storyboardID, mediaID, segment.Index, int64(timestamp*1000))
|
||
if snapshot.EpisodeID != nil {
|
||
key = mediakey.StoryboardFrame(snapshot.ProjectID, *snapshot.EpisodeID, storyboardID, mediaID, segment.Index, int64(timestamp*1000))
|
||
}
|
||
frame, err := os.Open(path)
|
||
if err != nil {
|
||
os.Remove(path)
|
||
return nil, err
|
||
}
|
||
hash := sha256.New()
|
||
if _, err := io.Copy(hash, frame); err != nil {
|
||
frame.Close()
|
||
os.Remove(path)
|
||
return nil, err
|
||
}
|
||
_, _ = frame.Seek(0, io.SeekStart)
|
||
url, err := w.COS.Put(ctx, key, "image/jpeg", frame, info.Size())
|
||
frame.Close()
|
||
os.Remove(path)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if url == "" {
|
||
return nil, errors.New("COS 公网地址未配置，反推模型无法读取抽帧")
|
||
}
|
||
asset := model.MediaAsset{ID: mediaID, OwnerUserID: &snapshot.UserID, StorageProvider: "cos", ObjectKey: key, PublicURL: url, DisplayName: fmt.Sprintf("分镜%04d-%010d帧", segment.Index, int64(timestamp*1000)), MimeType: "image/jpeg", SizeBytes: info.Size(), SHA256: hex.EncodeToString(hash.Sum(nil))}
|
||
if err := w.DB.WithContext(ctx).Create(&asset).Error; err != nil {
|
||
_ = w.COS.Delete(ctx, key)
|
||
return nil, err
|
||
}
|
||
frames = append(frames, asset)
|
||
}
|
||
if len(frames) == 0 {
|
||
return nil, errors.New("未能抽取视频画面")
|
||
}
|
||
return frames, nil
|
||
}
|
||
|
||
func frameRetryTimestamp(segment fixedSegment, timestamp float64) (float64, bool) {
|
||
retry := timestamp + 0.5
|
||
if timestamp >= (segment.Start+segment.End)/2 {
|
||
retry = timestamp - 0.5
|
||
}
|
||
if retry <= segment.Start || retry >= segment.End {
|
||
return 0, false
|
||
}
|
||
return retry, true
|
||
}
|
||
|
||
func stableStoryboardID(scopeID uuid.UUID, sequence int) uuid.UUID {
|
||
return uuid.NewSHA1(scopeID, []byte(fmt.Sprintf("storyboard:%d", sequence)))
|
||
}
|
||
|
||
func analysisScopeID(snapshot analysisSnapshot) uuid.UUID {
|
||
if snapshot.EpisodeID != nil {
|
||
return *snapshot.EpisodeID
|
||
}
|
||
return snapshot.ProjectID
|
||
}
|
||
|
||
func (w *Media) reverseSegment(ctx context.Context, snapshot analysisSnapshot, segment fixedSegment, frames []model.MediaAsset, transcript []transcriptSegment, knownAssets []map[string]any, apiKey string) (reverseResponse, error) {
|
||
region := map[string]string{"china": "中国", "uk_us": "英美", "korea": "韩国", "japan": "日本", "france": "法国", "russia": "俄罗斯", "vietnam": "越南", "thailand": "泰国", "india": "印度"}[snapshot.Localization]
|
||
era := snapshot.EraType
|
||
if snapshot.CustomEra != "" {
|
||
era = snapshot.CustomEra
|
||
}
|
||
transcriptJSON, _ := json.Marshal(transcript)
|
||
knownJSON, _ := json.Marshal(knownAssets)
|
||
text := fmt.Sprintf(`分析第 %d 段视频，时间 %.3f-%.3f 秒。
|
||
项目时代/题材：%s；目标本土化地区：%s；项目风格：%s；画面比例：%s。
|
||
台词事实：%s
|
||
项目已有资产：%s
|
||
先逐帧确认可见角色、场景和关键道具。相同角色必须复用已有资产名称，不创建重复角色。
|
||
本土化必须保留原时代和题材，但人物外貌、服装、建筑和道具转换为目标地区等价设定。
|
||
资产只返回 type 和 name，禁止返回 description、image_prompt、appearances 或任何文生图提示词。角色 name 和台词使用目标地区语言；scene 和 prop 的 name 使用中文。
|
||
分镜 prompt 必须详细还原原视频，并严格使用以下结构：
|
||
分镜 %d（%.1f-%.1f秒）· 场景=场景资产名称
|
||
【第一帧】
|
||
- 画面：明确景别、构图、角色外观与环境视觉信息。
|
||
- 站位：明确每个角色及关键道具在画面中的前后左右关系、朝向和视线。
|
||
【画面内容】
|
||
连续镜头直接按时间顺序描述；存在明显镜头变化时拆成【镜头 1】（起止秒数）、【镜头 2】（起止秒数）。每个镜头必须写清景别与构图、运镜方式、角色动作与表情、环境变化、光影色调以及对应时间内实际发生的台词。台词必须放入实际发生的镜头段落，禁止在末尾集中罗列。
|
||
禁止只写“保持原视频构图”“自然运镜”“角色互动”等空泛内容；必须从关键帧反推出具体、可执行的视觉细节，并保持人物身份、服装、空间方位和动作连续。
|
||
只返回 JSON：{"assets":[{"type":"character|scene|prop","name":"","image_prompt":""}],"segment":{"title":"","prompt":"","asset_names":[],"dialogue":[]}}。`, segment.Index, segment.Start, segment.End, era, region, snapshot.StyleName, snapshot.AspectRatio, transcriptJSON, knownJSON, segment.Index, segment.Start, segment.End)
|
||
text += `
|
||
|
||
追加格式要求（必须遵守）：角色资产的 name 使用目标本土语言；scene 和 prop 资产的 name 必须使用中文，不得本土化翻译。assets 每项只能包含 type、name 两个字段，严禁输出任何文生图提示词或资产视觉描述。segment.prompt 中的视觉描述和镜头说明必须使用中文；只有角色 name、台词中的 text 和 speaker 可以使用目标本土语言。分镜 prompt 必须使用“分镜 N（起止秒）· 场景=…”、“【第一帧】”、“【画面内容】”、“【镜头 N】（起止秒数）”结构。每个镜头必须写明景别/机位/运镜、角色站位、动作、表情、环境、光线色调、道具交互和该时间段实际台词；台词必须写入发生它的镜头段落。dialogue 中保留 source_text、corrected_source_text、localized_text、text、speaker、start、end、needs_review 字段；不确定台词不得臆造。`
|
||
content := []map[string]any{{"type": "text", "text": text}}
|
||
for _, frame := range frames {
|
||
content = append(content, map[string]any{"type": "image_url", "image_url": map[string]any{"url": frame.PublicURL}})
|
||
}
|
||
payload := map[string]any{
|
||
"model": snapshot.ModelName,
|
||
"messages": []map[string]any{
|
||
{"role": "system", "content": "你是商业视频转绘的结构化反推服务。必须忠实使用画面与台词事实，严格返回可解析 JSON，不输出 Markdown。"},
|
||
{"role": "user", "content": content},
|
||
},
|
||
"temperature": 0.2,
|
||
"response_format": map[string]any{"type": "json_object"},
|
||
}
|
||
pricing, err := billing.ParseTextPricingSnapshot(snapshot.BillingSnapshot)
|
||
if err != nil {
|
||
return reverseResponse{}, err
|
||
}
|
||
if pricing.Mode == billing.TextBillingPerToken {
|
||
if err := w.DB.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
||
estimatedInput := billing.EstimateTextTokens(text) + billing.EstimateTextTokens("你是商业视频转绘的结构化反推服务。必须忠实使用画面与台词事实，严格返回可解析 JSON，不输出 Markdown。")
|
||
_, reserveErr := billing.ReserveTextCall(tx, snapshot.UserID, snapshot.TaskID.String(), strconv.Itoa(segment.Index), pricing, estimatedInput, "视频反推")
|
||
return reserveErr
|
||
}); err != nil {
|
||
return reverseResponse{}, err
|
||
}
|
||
}
|
||
var chat apimart.ChatResult
|
||
var result reverseResponse
|
||
var responseErr error
|
||
for attempt := 0; attempt < 2; attempt++ {
|
||
if attempt > 0 {
|
||
retryContent := append([]map[string]any(nil), content...)
|
||
retryContent = append(retryContent, map[string]any{"type": "text", "text": "上次响应不是完整、可解析的 JSON。请压缩重复措辞，但保留画面、动作、台词和资产事实；优先确保 JSON 完整闭合且字段结构严格正确。只输出 JSON。"})
|
||
payload["messages"] = []map[string]any{
|
||
{"role": "system", "content": "你是商业视频转绘的结构化反推服务。必须忠实使用画面与台词事实，严格返回可解析 JSON，不输出 Markdown。"},
|
||
{"role": "user", "content": retryContent},
|
||
}
|
||
payload["temperature"] = 0
|
||
}
|
||
requestCtx, cancel := context.WithTimeout(ctx, 4*time.Minute)
|
||
chat, err = w.Provider.ChatWithUsage(requestCtx, snapshot.BaseURL, apiKey, payload)
|
||
cancel()
|
||
if err != nil {
|
||
return reverseResponse{}, err
|
||
}
|
||
result = reverseResponse{}
|
||
stripped := stripJSONFence(chat.Content)
|
||
if strings.TrimSpace(stripped) == "" {
|
||
// 上游返回空内容（常见于 content_filter 触发），给出明确错误而非 JSON 解析错误
|
||
if chat.FinishReason == "content_filter" {
|
||
responseErr = errors.New("上游模型触发内容安全过滤，未返回内容")
|
||
} else {
|
||
responseErr = fmt.Errorf("上游模型返回空内容（结束原因：%s）", chat.FinishReason)
|
||
}
|
||
} else {
|
||
responseErr = json.Unmarshal([]byte(stripped), &result)
|
||
if responseErr == nil && strings.TrimSpace(result.Segment.Prompt) == "" {
|
||
responseErr = errors.New("未返回分镜提示词")
|
||
}
|
||
}
|
||
if responseErr == nil {
|
||
break
|
||
}
|
||
}
|
||
if responseErr != nil {
|
||
finishReason := strings.TrimSpace(chat.FinishReason)
|
||
if finishReason == "" {
|
||
finishReason = "未知"
|
||
}
|
||
return result, fmt.Errorf("反推模型连续两次返回无法解析（结束原因：%s）: %w", finishReason, responseErr)
|
||
}
|
||
result.Raw = chat.Raw
|
||
result.InputTokens = chat.InputTokens
|
||
result.OutputTokens = chat.OutputTokens
|
||
result.UsageRaw = chat.UsageRaw
|
||
if pricing.Mode == billing.TextBillingPerToken {
|
||
if err := w.DB.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
||
settled, settleErr := billing.SettleTextCall(tx, snapshot.UserID, snapshot.TaskID.String(), strconv.Itoa(segment.Index), pricing, billing.TextUsage{Input: chat.InputTokens, Output: chat.OutputTokens}, "视频反推")
|
||
if settleErr != nil {
|
||
return settleErr
|
||
}
|
||
return tx.Model(&model.GenerationTask{}).Where("id=?", snapshot.TaskID).Updates(map[string]any{
|
||
"estimated_points": gorm.Expr("estimated_points+?::numeric", settled),
|
||
"prepaid_points": gorm.Expr("prepaid_points+?::numeric", settled),
|
||
"input_tokens": gorm.Expr("coalesce(input_tokens,0)+?", chat.InputTokens),
|
||
"output_tokens": gorm.Expr("coalesce(output_tokens,0)+?", chat.OutputTokens),
|
||
"total_tokens": gorm.Expr("coalesce(total_tokens,0)+?", chat.InputTokens+chat.OutputTokens),
|
||
"token_count_source": "upstream",
|
||
"usage_raw": gorm.Expr("coalesce(usage_raw,'{}'::jsonb) || jsonb_build_object(?::text, ?::jsonb)", strconv.Itoa(segment.Index), string(chat.UsageRaw)),
|
||
}).Error
|
||
}); err != nil {
|
||
return reverseResponse{}, err
|
||
}
|
||
}
|
||
return result, nil
|
||
}
|
||
|
||
func (w *Media) persistReverseResult(ctx context.Context, snapshot analysisSnapshot, segment fixedSegment, thumbnailID uuid.UUID, response reverseResponse) ([]map[string]any, error) {
|
||
mediaID := uuid.New()
|
||
storyboardID := stableStoryboardID(analysisScopeID(snapshot), segment.Index)
|
||
key := mediakey.ProjectStoryboardAnalysis(snapshot.ProjectID, storyboardID, snapshot.TaskID, mediaID, segment.Index)
|
||
if snapshot.EpisodeID != nil {
|
||
key = mediakey.StoryboardAnalysis(snapshot.ProjectID, *snapshot.EpisodeID, storyboardID, snapshot.TaskID, mediaID, segment.Index)
|
||
}
|
||
if len(response.Raw) == 0 {
|
||
response.Raw, _ = json.Marshal(response)
|
||
}
|
||
url, err := w.COS.Put(ctx, key, "application/json", bytes.NewReader(response.Raw), int64(len(response.Raw)))
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
ownerID := snapshot.UserID
|
||
payloadAsset := model.MediaAsset{ID: mediaID, OwnerUserID: &ownerID, StorageProvider: "cos", ObjectKey: key, PublicURL: url, DisplayName: fmt.Sprintf("分镜%04d反推结果", segment.Index), MimeType: "application/json", SizeBytes: int64(len(response.Raw)), SHA256: fmt.Sprintf("%x", sha256.Sum256(response.Raw))}
|
||
err = w.DB.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
||
var task model.GenerationTask
|
||
if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Select("id", "status").Where("id=?", snapshot.TaskID).Take(&task).Error; err != nil {
|
||
return err
|
||
}
|
||
if task.Status != "submitted" && task.Status != "processing" {
|
||
return errors.New("反推任务已停止")
|
||
}
|
||
// 幂等检查：asynq 重试时若该分镜产出已存在，跳过 mediaAsset 与 output 创建，避免违反 (task_id, sequence_no) 唯一约束；后续封面、project_assets、storyboards 均为幂等写入，可安全重复执行
|
||
var existingOutput model.GenerationOutput
|
||
if existingErr := tx.Where("task_id=? AND sequence_no=?", snapshot.TaskID, segment.Index).Take(&existingOutput).Error; existingErr == nil {
|
||
// 产出已存在，跳过创建
|
||
} else if errors.Is(existingErr, gorm.ErrRecordNotFound) {
|
||
if err := tx.Create(&payloadAsset).Error; err != nil {
|
||
return err
|
||
}
|
||
output := model.GenerationOutput{TaskID: snapshot.TaskID, MediaAssetID: mediaID, OutputType: "text", SequenceNo: segment.Index, Metadata: json.RawMessage(`{"kind":"prompt_reverse"}`)}
|
||
if err := tx.Create(&output).Error; err != nil {
|
||
return err
|
||
}
|
||
} else {
|
||
return existingErr
|
||
}
|
||
// 第一张原片抽帧作为剧本封面。
|
||
if snapshot.StoryboardID == nil && segment.Index == 1 {
|
||
if snapshot.EpisodeID != nil {
|
||
if err := tx.Exec("UPDATE project_episodes SET cover_asset_id=? WHERE id=? AND cover_asset_id IS NULL", thumbnailID, *snapshot.EpisodeID).Error; err != nil {
|
||
return err
|
||
}
|
||
}
|
||
if err := tx.Exec("UPDATE creative_projects SET cover_asset_id=? WHERE id=? AND cover_asset_id IS NULL", thumbnailID, snapshot.ProjectID).Error; err != nil {
|
||
return err
|
||
}
|
||
}
|
||
nameIDs := map[string]uuid.UUID{}
|
||
assetNames := map[string]string{}
|
||
var existing []model.ProjectAsset
|
||
if err := tx.Where("project_id=? AND deleted_at IS NULL", snapshot.ProjectID).Find(&existing).Error; err != nil {
|
||
return err
|
||
}
|
||
for _, asset := range existing {
|
||
nameKey := strings.ToLower(strings.TrimSpace(asset.Name))
|
||
nameIDs[nameKey] = asset.ID
|
||
assetNames[nameKey] = strings.TrimSpace(asset.Name)
|
||
}
|
||
for _, incoming := range response.Assets {
|
||
assetType := strings.ToLower(strings.TrimSpace(incoming.Type))
|
||
if !map[string]bool{"character": true, "scene": true, "prop": true}[assetType] || strings.TrimSpace(incoming.Name) == "" {
|
||
continue
|
||
}
|
||
key := strings.ToLower(strings.TrimSpace(incoming.Name))
|
||
if _, exists := nameIDs[key]; exists {
|
||
continue
|
||
}
|
||
asset := model.ProjectAsset{ProjectID: snapshot.ProjectID, AssetType: assetType, Name: strings.TrimSpace(incoming.Name), ImagePrompt: strings.TrimSpace(incoming.ImagePrompt), Appearances: json.RawMessage("[]")}
|
||
if err := tx.Create(&asset).Error; err != nil {
|
||
return err
|
||
}
|
||
nameIDs[key] = asset.ID
|
||
assetNames[key] = asset.Name
|
||
}
|
||
candidateNames := make([]string, 0, len(assetNames))
|
||
for _, name := range assetNames {
|
||
candidateNames = append(candidateNames, name)
|
||
}
|
||
candidateNames = sortedMentionNames(candidateNames)
|
||
promptContent := mentionAssets(response.Segment.Prompt, candidateNames)
|
||
refs := make([]map[string]any, 0)
|
||
refKeys := map[string]bool{}
|
||
addRef := func(name string) {
|
||
key := strings.ToLower(strings.TrimSpace(name))
|
||
id, exists := nameIDs[key]
|
||
if !exists || refKeys[key] {
|
||
return
|
||
}
|
||
refKeys[key] = true
|
||
refs = append(refs, map[string]any{"id": id, "name": assetNames[key]})
|
||
}
|
||
for _, name := range response.Segment.AssetNames {
|
||
addRef(name)
|
||
}
|
||
for _, name := range mentionedAssetNames(promptContent, candidateNames) {
|
||
addRef(name)
|
||
}
|
||
dialogue, _ := json.Marshal(response.Segment.Dialogue)
|
||
assetRefs, _ := json.Marshal(refs)
|
||
duration := storyboardDuration(segment)
|
||
if snapshot.EpisodeID != nil {
|
||
return tx.Exec(`INSERT INTO episode_storyboards(id,episode_id,sequence_no,stable_key,start_ms,end_ms,duration_seconds,title,thumbnail_asset_id,prompt_content,dialogue,asset_refs,status)
|
||
VALUES(?,?,?,?,?,?,?,?,?,?,?::jsonb,?::jsonb,'idle')
|
||
ON CONFLICT(id) DO UPDATE SET start_ms=excluded.start_ms,end_ms=excluded.end_ms,duration_seconds=excluded.duration_seconds,title=excluded.title,
|
||
thumbnail_asset_id=excluded.thumbnail_asset_id,prompt_content=excluded.prompt_content,dialogue=excluded.dialogue,asset_refs=excluded.asset_refs,
|
||
status='idle',updated_at=CURRENT_TIMESTAMP WHERE episode_storyboards.locked=false`, storyboardID, *snapshot.EpisodeID, segment.Index, fmt.Sprintf("segment-%04d", segment.Index), int64(segment.Start*1000), int64(segment.End*1000), duration, response.Segment.Title, thumbnailID, promptContent, string(dialogue), string(assetRefs)).Error
|
||
}
|
||
return tx.Exec(`INSERT INTO episode_storyboards(id,project_id,sequence_no,stable_key,start_ms,end_ms,duration_seconds,title,thumbnail_asset_id,prompt_content,dialogue,asset_refs,status)
|
||
VALUES(?,?,?,?,?,?,?,?,?,?,?::jsonb,?::jsonb,'idle')
|
||
ON CONFLICT(id) DO UPDATE SET start_ms=excluded.start_ms,end_ms=excluded.end_ms,duration_seconds=excluded.duration_seconds,title=excluded.title,
|
||
thumbnail_asset_id=excluded.thumbnail_asset_id,prompt_content=excluded.prompt_content,dialogue=excluded.dialogue,asset_refs=excluded.asset_refs,
|
||
status='idle',updated_at=CURRENT_TIMESTAMP WHERE episode_storyboards.locked=false`, storyboardID, snapshot.ProjectID, segment.Index, fmt.Sprintf("segment-%04d", segment.Index), int64(segment.Start*1000), int64(segment.End*1000), duration, response.Segment.Title, thumbnailID, promptContent, string(dialogue), string(assetRefs)).Error
|
||
})
|
||
if err != nil {
|
||
_ = w.COS.Delete(ctx, key)
|
||
return nil, err
|
||
}
|
||
known := make([]map[string]any, 0)
|
||
err = w.DB.WithContext(ctx).Table("project_assets").Select("id,asset_type AS type,name,image_prompt").Where("project_id=? AND deleted_at IS NULL", snapshot.ProjectID).Order("created_at").Find(&known).Error
|
||
return known, err
|
||
}
|
||
|
||
func mentionAssets(prompt string, names []string) string {
|
||
var result strings.Builder
|
||
segmentStart := 0
|
||
var quoteEnd rune
|
||
for index, current := range prompt {
|
||
if quoteEnd == 0 {
|
||
if current != '“' && current != '"' {
|
||
continue
|
||
}
|
||
result.WriteString(mentionAssetNames(prompt[segmentStart:index], names))
|
||
segmentStart = index
|
||
if current == '“' {
|
||
quoteEnd = '”'
|
||
} else {
|
||
quoteEnd = '"'
|
||
}
|
||
continue
|
||
}
|
||
if current != quoteEnd || (current == '"' && escapedQuote(prompt, index)) {
|
||
continue
|
||
}
|
||
end := index + len(string(current))
|
||
result.WriteString(prompt[segmentStart:end])
|
||
segmentStart = end
|
||
quoteEnd = 0
|
||
}
|
||
if quoteEnd != 0 {
|
||
result.WriteString(prompt[segmentStart:])
|
||
} else {
|
||
result.WriteString(mentionAssetNames(prompt[segmentStart:], names))
|
||
}
|
||
return result.String()
|
||
}
|
||
|
||
func mentionAssetNames(content string, names []string) string {
|
||
result := content
|
||
for _, raw := range sortedMentionNames(names) {
|
||
name := strings.TrimSpace(raw)
|
||
if name == "" {
|
||
continue
|
||
}
|
||
var mentioned strings.Builder
|
||
remaining := result
|
||
for {
|
||
index := strings.Index(remaining, name)
|
||
if index < 0 {
|
||
mentioned.WriteString(remaining)
|
||
break
|
||
}
|
||
mentioned.WriteString(remaining[:index])
|
||
if index == 0 || remaining[index-1] != '@' {
|
||
mentioned.WriteByte('@')
|
||
}
|
||
mentioned.WriteString(name)
|
||
remaining = remaining[index+len(name):]
|
||
}
|
||
result = mentioned.String()
|
||
}
|
||
return result
|
||
}
|
||
|
||
func sortedMentionNames(names []string) []string {
|
||
seen := map[string]bool{}
|
||
result := make([]string, 0, len(names))
|
||
for _, raw := range names {
|
||
name := strings.TrimSpace(raw)
|
||
key := strings.ToLower(name)
|
||
if name == "" || seen[key] {
|
||
continue
|
||
}
|
||
seen[key] = true
|
||
result = append(result, name)
|
||
}
|
||
sort.SliceStable(result, func(left, right int) bool {
|
||
if len(result[left]) == len(result[right]) {
|
||
return result[left] < result[right]
|
||
}
|
||
return len(result[left]) > len(result[right])
|
||
})
|
||
return result
|
||
}
|
||
|
||
func mentionedAssetNames(content string, names []string) []string {
|
||
names = sortedMentionNames(names)
|
||
seen := map[string]bool{}
|
||
result := make([]string, 0)
|
||
for index := 0; index < len(content); {
|
||
if content[index] != '@' {
|
||
index++
|
||
continue
|
||
}
|
||
remaining := content[index+1:]
|
||
matched := ""
|
||
for _, name := range names {
|
||
if strings.HasPrefix(remaining, name) {
|
||
matched = name
|
||
break
|
||
}
|
||
}
|
||
if matched == "" {
|
||
index++
|
||
continue
|
||
}
|
||
key := strings.ToLower(matched)
|
||
if !seen[key] {
|
||
seen[key] = true
|
||
result = append(result, matched)
|
||
}
|
||
index += 1 + len(matched)
|
||
}
|
||
return result
|
||
}
|
||
|
||
func escapedQuote(content string, quoteIndex int) bool {
|
||
backslashes := 0
|
||
for index := quoteIndex - 1; index >= 0 && content[index] == '\\'; index-- {
|
||
backslashes++
|
||
}
|
||
return backslashes%2 == 1
|
||
}
|
||
|
||
func transcriptForRange(items []transcriptSegment, start, end float64) []transcriptSegment {
|
||
result := make([]transcriptSegment, 0)
|
||
for _, item := range items {
|
||
if item.End > start && item.Start < end {
|
||
result = append(result, item)
|
||
}
|
||
}
|
||
return result
|
||
}
|
||
|
||
func (w *Media) loadSubtitle(ctx context.Context, sourceURL string, duration float64) (transcriptResult, error) {
|
||
request, err := http.NewRequestWithContext(ctx, http.MethodGet, sourceURL, nil)
|
||
if err != nil {
|
||
return transcriptResult{}, err
|
||
}
|
||
response, err := w.HTTP.Do(request)
|
||
if err != nil {
|
||
return transcriptResult{}, err
|
||
}
|
||
defer response.Body.Close()
|
||
data, err := io.ReadAll(io.LimitReader(response.Body, 2*1024*1024+1))
|
||
if err != nil || len(data) > 2*1024*1024 {
|
||
return transcriptResult{}, errors.New("字幕文件读取失败或超过 2MB")
|
||
}
|
||
segments := parseSubtitle(string(data))
|
||
if len(segments) == 0 && strings.TrimSpace(string(data)) != "" {
|
||
segments = []transcriptSegment{{Start: 0, End: duration, Text: strings.TrimSpace(string(data)), Confidence: 1}}
|
||
}
|
||
return transcriptResult{Segments: segments}, nil
|
||
}
|
||
|
||
var subtitleTime = regexp.MustCompile(`(?m)(\d{1,2}):(\d{2}):(\d{2})[,.](\d{3})\s*-->\s*(\d{1,2}):(\d{2}):(\d{2})[,.](\d{3})`)
|
||
|
||
func parseSubtitle(value string) []transcriptSegment {
|
||
value = strings.ReplaceAll(value, "\r\n", "\n")
|
||
blocks := regexp.MustCompile(`\n\s*\n`).Split(value, -1)
|
||
result := make([]transcriptSegment, 0, len(blocks))
|
||
for _, block := range blocks {
|
||
match := subtitleTime.FindStringSubmatchIndex(block)
|
||
if len(match) == 0 {
|
||
continue
|
||
}
|
||
parts := subtitleTime.FindStringSubmatch(block[match[0]:match[1]])
|
||
if len(parts) != 9 {
|
||
continue
|
||
}
|
||
start := subtitleSeconds(parts[1:5])
|
||
end := subtitleSeconds(parts[5:9])
|
||
text := strings.TrimSpace(block[match[1]:])
|
||
if text != "" && end > start {
|
||
result = append(result, transcriptSegment{Start: start, End: end, Text: text, Confidence: 1})
|
||
}
|
||
}
|
||
return result
|
||
}
|
||
|
||
func subtitleSeconds(parts []string) float64 {
|
||
values := make([]float64, len(parts))
|
||
for index, part := range parts {
|
||
values[index], _ = strconv.ParseFloat(part, 64)
|
||
}
|
||
return values[0]*3600 + values[1]*60 + values[2] + values[3]/1000
|
||
}
|
||
|
||
func stripJSONFence(value string) string {
|
||
value = strings.TrimSpace(value)
|
||
value = strings.TrimPrefix(value, "```json")
|
||
value = strings.TrimPrefix(value, "```")
|
||
value = strings.TrimSuffix(value, "```")
|
||
return strings.TrimSpace(value)
|
||
}
|
||
|
||
func (w *Media) updateRedrawProject(ctx context.Context, projectID uuid.UUID, status, message string) {
|
||
_ = w.DB.WithContext(ctx).Model(&model.CreativeProject{}).Where("id=?", projectID).Updates(map[string]any{"redraw_status": status, "analysis_message": message}).Error
|
||
}
|
||
|
||
func (w *Media) updateRedrawScope(ctx context.Context, snapshot analysisSnapshot, status, message string) {
|
||
if snapshot.EpisodeID != nil {
|
||
_ = w.DB.WithContext(ctx).Model(&model.ProjectEpisode{}).Where("id=?", *snapshot.EpisodeID).Updates(map[string]any{"status": status, "analysis_message": message}).Error
|
||
return
|
||
}
|
||
w.updateRedrawProject(ctx, snapshot.ProjectID, status, message)
|
||
}
|
||
|
||
func (w *Media) analysisCancellationRequested(ctx context.Context, taskID uuid.UUID) bool {
|
||
var status string
|
||
_ = w.DB.WithContext(ctx).Model(&model.GenerationTask{}).Select("status").Where("id=?", taskID).Scan(&status).Error
|
||
return status == "cancel_requested" || status == "cancelled"
|
||
}
|
||
|
||
func (w *Media) finishAnalysisCancellation(ctx context.Context, taskID, projectID uuid.UUID) error {
|
||
return w.DB.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
||
var task model.GenerationTask
|
||
if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Where("id=?", taskID).First(&task).Error; err != nil {
|
||
return err
|
||
}
|
||
if projectID == uuid.Nil && task.ProjectID != nil {
|
||
projectID = *task.ProjectID
|
||
}
|
||
return w.finishAnalysisCancellationLocked(tx, &task, projectID)
|
||
})
|
||
}
|
||
|
||
func (w *Media) finishAnalysisCancellationLocked(tx *gorm.DB, task *model.GenerationTask, projectID uuid.UUID) error {
|
||
if task.Status == "cancelled" || task.Status == "succeeded" || task.Status == "failed" {
|
||
return nil
|
||
}
|
||
if projectID == uuid.Nil && task.ProjectID != nil {
|
||
projectID = *task.ProjectID
|
||
}
|
||
updates := map[string]any{
|
||
"status": "cancelled", "actual_points": task.PrepaidPoints, "finished_at": time.Now(),
|
||
"error_code": nil, "error_message": "用户取消，已扣积分不退",
|
||
}
|
||
if err := tx.Model(task).Updates(updates).Error; err != nil {
|
||
return err
|
||
}
|
||
if task.StoryboardID != nil {
|
||
if err := tx.Model(&model.EpisodeStoryboard{}).Where("id=?", *task.StoryboardID).Update("status", "idle").Error; err != nil {
|
||
return err
|
||
}
|
||
}
|
||
if task.EpisodeID != nil && task.StoryboardID == nil {
|
||
return w.updateEpisodeAfterAnalysisCancellation(tx, *task.EpisodeID)
|
||
}
|
||
if projectID != uuid.Nil && task.StoryboardID == nil {
|
||
return w.updateProjectAfterAnalysisCancellation(tx, projectID)
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func (w *Media) updateEpisodeAfterAnalysisCancellation(tx *gorm.DB, episodeID uuid.UUID) error {
|
||
if err := tx.Model(&model.EpisodeStoryboard{}).Where("episode_id=? AND status IN ?", episodeID, []string{"queued", "generating"}).Update("status", "idle").Error; err != nil {
|
||
return err
|
||
}
|
||
var storyboardCount int64
|
||
if err := tx.Model(&model.EpisodeStoryboard{}).Where("episode_id=? AND deleted_at IS NULL", episodeID).Count(&storyboardCount).Error; err != nil {
|
||
return err
|
||
}
|
||
status := "uploaded"
|
||
if storyboardCount > 0 {
|
||
status = "review"
|
||
}
|
||
return tx.Model(&model.ProjectEpisode{}).Where("id=?", episodeID).Updates(map[string]any{"status": status, "analysis_message": "视频分析已取消"}).Error
|
||
}
|
||
|
||
func (w *Media) updateProjectAfterAnalysisCancellation(tx *gorm.DB, projectID uuid.UUID) error {
|
||
if err := tx.Model(&model.EpisodeStoryboard{}).Where("project_id=? AND status IN ?", projectID, []string{"queued", "generating"}).Update("status", "idle").Error; err != nil {
|
||
return err
|
||
}
|
||
var storyboardCount int64
|
||
if err := tx.Model(&model.EpisodeStoryboard{}).Where("project_id=? AND deleted_at IS NULL", projectID).Count(&storyboardCount).Error; err != nil {
|
||
return err
|
||
}
|
||
status := "uploaded"
|
||
if storyboardCount > 0 {
|
||
status = "review"
|
||
}
|
||
return tx.Model(&model.CreativeProject{}).Where("id=?", projectID).Updates(map[string]any{"redraw_status": status, "analysis_message": "视频分析已取消"}).Error
|
||
}
|
||
|
||
func (w *Media) failAnalysis(ctx context.Context, taskID, projectID uuid.UUID, err error) error {
|
||
// 服务关闭时 ctx 会被取消，结算事务随之失败导致 asynq 重试。检测到 ctx 已取消时改用不继承取消信号的 context 完成结算，并加 30 秒超时防止卡死
|
||
if errors.Is(ctx.Err(), context.Canceled) {
|
||
var cancel context.CancelFunc
|
||
ctx, cancel = context.WithTimeout(context.WithoutCancel(ctx), 30*time.Second)
|
||
defer cancel()
|
||
}
|
||
if taskID != uuid.Nil && w.analysisCancellationRequested(ctx, taskID) {
|
||
return w.finishAnalysisCancellation(ctx, taskID, projectID)
|
||
}
|
||
slog.ErrorContext(ctx, "视频分析失败", "task_id", taskID, "project_id", projectID, "error", err)
|
||
now := time.Now()
|
||
updateProject := projectID != uuid.Nil
|
||
var episodeID *uuid.UUID
|
||
if taskID != uuid.Nil {
|
||
billingErr := w.DB.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
||
var task model.GenerationTask
|
||
if lockErr := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Where("id=?", taskID).First(&task).Error; lockErr != nil {
|
||
// 任务记录已被删除（如用户删除项目时连带清理），结算无意义，直接返回避免 asynq 无限重试
|
||
if errors.Is(lockErr, gorm.ErrRecordNotFound) {
|
||
return nil
|
||
}
|
||
return lockErr
|
||
}
|
||
if task.Status == "cancel_requested" {
|
||
return w.finishAnalysisCancellationLocked(tx, &task, projectID)
|
||
}
|
||
if task.Status == "succeeded" || task.Status == "failed" || task.Status == "cancelled" {
|
||
return nil
|
||
}
|
||
if task.StoryboardID != nil {
|
||
updateProject = false
|
||
}
|
||
episodeID = task.EpisodeID
|
||
refunded, refundErr := billing.RefundTextGenerationTask(tx, &task, "视频反推失败返还")
|
||
if refundErr != nil {
|
||
return refundErr
|
||
}
|
||
updates := map[string]any{"status": "failed", "actual_points": "0.00", "error_code": "analysis_failed", "error_message": truncate(err.Error(), 4000), "finished_at": now}
|
||
if refunded {
|
||
updates["cost_refunded"] = true
|
||
}
|
||
if updateErr := tx.Model(&task).Updates(updates).Error; updateErr != nil {
|
||
return updateErr
|
||
}
|
||
storyboardID := task.StoryboardID
|
||
if storyboardID == nil {
|
||
var input struct {
|
||
CurrentSequence int `json:"current_sequence"`
|
||
}
|
||
if json.Unmarshal(task.InputData, &input) == nil && input.CurrentSequence > 0 {
|
||
scopeID := projectID
|
||
if episodeID != nil {
|
||
scopeID = *episodeID
|
||
}
|
||
id := stableStoryboardID(scopeID, input.CurrentSequence)
|
||
storyboardID = &id
|
||
}
|
||
}
|
||
if storyboardID != nil {
|
||
if updateErr := tx.Model(&model.EpisodeStoryboard{}).Where("id=?", *storyboardID).Update("status", "failed").Error; updateErr != nil {
|
||
return updateErr
|
||
}
|
||
}
|
||
if updateProject {
|
||
if episodeID != nil {
|
||
return tx.Model(&model.ProjectEpisode{}).Where("id=?", *episodeID).Updates(map[string]any{"status": "failed", "analysis_message": err.Error()}).Error
|
||
}
|
||
return tx.Model(&model.CreativeProject{}).Where("id=?", projectID).Updates(map[string]any{"redraw_status": "failed", "analysis_message": err.Error()}).Error
|
||
}
|
||
return nil
|
||
})
|
||
if billingErr != nil {
|
||
slog.ErrorContext(ctx, "反推失败结算失败", "task_id", taskID, "error", billingErr)
|
||
}
|
||
return billingErr
|
||
}
|
||
if updateProject {
|
||
return w.DB.WithContext(ctx).Model(&model.CreativeProject{}).Where("id=?", projectID).Updates(map[string]any{"redraw_status": "failed", "analysis_message": err.Error()}).Error
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func scanLines(value string) []string {
|
||
result := make([]string, 0)
|
||
scanner := bufio.NewScanner(strings.NewReader(value))
|
||
for scanner.Scan() {
|
||
if line := strings.TrimSpace(scanner.Text()); line != "" {
|
||
result = append(result, line)
|
||
}
|
||
}
|
||
return result
|
||
}
|