CodeVetter · Repo Unpack · Data flow
harry0703/MoneyPrinterTurbo
A request becomes a pydantic VideoParams, gets a UUID task_id, a state record, and a storage/tasks/<task_id> workspace. The pipeline transforms subject→script→terms→audio→subtitle→materials→final MP4, writing intermediate files (script.json, audio.mp3, subtitle.srt, combined-N.mp4, final-N.mp4) into the task dir and progress/state into MemoryState or RedisState. Outputs leave via the /tasks static mount, /stream, /download, or cross-posting to upload-post.com. Configuration flows the other direction: config.toml → section dicts, with WebUI runtime writes funneled through a non-blocking pending-update queue and atomic file saves.
- Ingress: TaskVideoRequest (a VideoParams subclass) is validated by pydantic, dumped into a task record {task_id, request_id, params}, persisted via sm.state.update_task, then tm.start is scheduled through task_manager.add_task — with rollback of the state record if scheduling throws.
- Script/terms stage: generate_script uses params.video_script verbatim or calls llm.generate_script; generate_terms calls llm.generate_terms (5 terms, or 8 in match-script-order mode) and optionally reranks through twelvelabs.rerank_terms_by_subject; results are persisted to <task_dir>/script.json via task_artifacts.write_script_data's atomic temp-file+os.replace write.
- Audio stage: resolve_custom_audio_file confines submitted paths to the task directory unless allow_server_file_input=True (CLI only); otherwise voice.tts writes audio.mp3 and returns an edge-tts SubMaker timeline; audio_duration is measured from the written file, not cue end times.
- Subtitle stage: 'edge' provider serializes SubMaker cues to subtitle.srt (never auto-falling back to Whisper, to avoid surprise multi-GB model downloads); 'whisper' transcribes the audio file with faster-whisper and runs subtitle.correct() against the script.
- Materials stage: stock sources go through _search_videos_with_cache (24h per-term cache in material_cache, double-checked locking) then download_videos; paid generation sources take _download_videos_*_on_demand paths that generate clips one at a time and stop once audio_duration*video_count is covered; 'local' preprocesses user files; 'loomloom' executes a confirmed SkillBot run and persists run_id immediately since billing has already started.
- Render stage: generate_final_videos loops video_count times calling video.combine_videos (→ combined-N.mp4), optionally calls the AI music provider's generate_bgm, then video.generate_video (muxing narration, subtitles, BGM → final-N.mp4); progress is written to state at ~5/10/20/30/40/50→100 milestones.
- Egress: task state stores absolute file paths; GET /tasks/{id} converts them to /tasks/* URLs via _task_file_to_uri (which re-validates each path inside the task dir), and the StaticFiles mount plus /stream//download endpoints serve bytes from storage/tasks.
- Config read path: load_config() → _cfg TOML dict → section dicts (app, azure, whisper, proxy, ui, and per-TTS sections); reads anywhere use config.<section>.get(...); app.redis_host is overridable via MPT_APP_REDIS_HOST/REDIS_HOST env vars.
- Config write path (WebUI): controls call update_config_nonblocking/delete_config_nonblocking which enqueue into _pending_config_updates keyed by id(section); a background 'mpt-config-flush' thread applies them when runtime_config_lock is free; save_config writes config.toml atomically.
- In Redis mode the task queue is a Redis list of JSON payloads; enqueue serializes VideoParams via model_dump and stores the function name, and dequeue re-validates against FUNC_MAP and current VideoParams — dropping stale entries while marking their tasks failed so they never appear stuck in 'processing'.
- Cross-post flow: on success the task is marked COMPLETE with cross_post_state='pending' and cross_post_owner='hostname:pid:uuid'; _schedule_cross_post submits to a bounded ThreadPoolExecutor(max_workers=2) gated by a BoundedSemaphore, and terminal state is written via patch_task so a deleted task is never resurrected.
- Failure propagation: every stage funnels into _mark_task_failed(state=-1, failed_stage, error) which preserves remote IDs (volcengine_seedance_task_id, ofox_task_id, metaso_minimax_task_id, muapi_task_id, loomloom_run_id) so paid remote jobs remain recoverable from provider dashboards.