```
├── .dockerignore (100 tokens)
├── .gitattributes (omitted)
├── .github/
├── ISSUE_TEMPLATE/
├── bug_report.yml (600 tokens)
├── config.yml
├── feature_request.yml (200 tokens)
├── SECURITY.md (200 tokens)
├── workflows/
├── ci.yml (700 tokens)
├── docker-ghcr.yml (400 tokens)
├── .gitignore (100 tokens)
├── .python-version
├── Dockerfile (1000 tokens)
├── Dockerfile.claude (200 tokens)
├── Dockerfile.gpu (500 tokens)
├── LICENSE (omitted)
├── README-en.md (7.9k tokens)
├── README-ja.md (5.5k tokens)
├── README.md (5.5k tokens)
├── app/
├── __init__.py
├── asgi.py (2k tokens)
├── config/
├── __init__.py (100 tokens)
├── config.py (4.1k tokens)
├── controllers/
├── base.py (600 tokens)
├── manager/
├── base_manager.py (1100 tokens)
├── memory_manager.py (100 tokens)
├── redis_manager.py (1600 tokens)
├── ping.py
├── v1/
├── base.py
├── llm.py (500 tokens)
├── video.py (4.1k tokens)
├── models/
├── __init__.py
├── const.py (100 tokens)
├── exception.py (200 tokens)
├── llm_provider.py (3.7k tokens)
├── schema.py (3.3k tokens)
├── router.py (100 tokens)
├── services/
├── __init__.py
├── bgm.py (2.9k tokens)
├── cache_manager.py (1500 tokens)
├── data/
├── azure_voices.json (4.3k tokens)
├── elevenlabs_music.py (2.8k tokens)
├── llm.py (9.2k tokens)
├── loomloom.py (7.7k tokens)
├── material.py (22.4k tokens)
├── material_cache.py (3k tokens)
├── material_upload.py (2.4k tokens)
├── metaso_minimax.py (3.2k tokens)
├── muapi.py (3.2k tokens)
├── ofox.py (3.2k tokens)
├── sonilo.py (3k tokens)
├── state.py (2.1k tokens)
├── subtitle.py (2.4k tokens)
├── task.py (14.1k tokens)
├── task_artifacts.py (600 tokens)
├── twelvelabs.py (1400 tokens)
├── upload_post.py (3.2k tokens)
├── utils/
├── video_effects.py (900 tokens)
├── version_checker.py (1400 tokens)
├── video.py (15.1k tokens)
├── video_project.py (7.2k tokens)
├── voice.py (28.6k tokens)
├── volcengine_seedance.py (3.3k tokens)
├── webui_task.py (1400 tokens)
├── utils/
├── file_security.py (200 tokens)
├── logging_utils.py (800 tokens)
├── subtitle_reader.py (300 tokens)
├── subtitle_writer.py (400 tokens)
├── utils.py (3.2k tokens)
├── cli.py (12.6k tokens)
├── config.example.toml (6.2k tokens)
├── docker-compose.claude.yml (300 tokens)
├── docker-compose.gpu.yml (200 tokens)
├── docker-compose.release.yml (200 tokens)
├── docker-compose.yml (200 tokens)
├── docs/
├── MoneyPrinterTurbo.ipynb (1700 tokens)
├── api.jpg
├── loomloom/
├── moneyprinterturbo-ai-video-materials.template.json (500 tokens)
├── moneyprinterturbo-script-candidates.template.json (500 tokens)
├── skill/
├── SKILL.md (1700 tokens)
├── mpt_agent.py (6.3k tokens)
├── sponsors/
├── apimart-logo.png
├── astraflow-logo.png
├── byteplus-logo.svg (1300 tokens)
├── ccsub-logo.png
├── infistar-logo.svg (2.3k tokens)
├── metaso-logo.png
├── ofox-logo.svg (1000 tokens)
├── picwish-logo.svg (800 tokens)
├── reccloud-logo.svg (200 tokens)
├── shengsuanyun-logo.jpg
├── sidrune-logo.png
├── volcengine-logo.svg (400 tokens)
├── video-projects.md (1600 tokens)
├── voice-list.txt (2.6k tokens)
├── webui-en.jpg
├── webui.jpg
├── main.py (100 tokens)
├── pyproject.toml (400 tokens)
├── requirements.txt (100 tokens)
├── resource/
├── fonts/
├── BeVietnamPro-Bold.ttf
├── BeVietnamPro-Medium.ttf
├── Charm-Bold.ttf
├── Charm-Regular.ttf
├── MicrosoftYaHeiBold.ttc
├── MicrosoftYaHeiNormal.ttc
├── STHeitiLight.ttc
├── STHeitiMedium.ttc
├── UTM Kabel KT.ttf
├── public/
├── index.html (100 tokens)
├── songs/
├── output000.mp3
├── output001.mp3
├── output002.mp3
├── output003.mp3
├── output004.mp3
├── output005.mp3
├── output006.mp3
├── output007.mp3
├── output008.mp3
├── output009.mp3
├── output010.mp3
├── output011.mp3
├── output012.mp3
├── output013.mp3
├── output014.mp3
├── output015.mp3
├── output016.mp3
├── output017.mp3
├── output018.mp3
├── output019.mp3
├── output020.mp3
├── output021.mp3
├── output022.mp3
├── output023.mp3
├── output024.mp3
├── output025.mp3
├── output027.mp3
├── output028.mp3
├── output029.mp3
├── test/
├── README.md (400 tokens)
├── __init__.py
├── resources/
├── 1.png
├── 2.png
├── 3.png
├── 4.png
├── 5.png
├── 6.png
├── 7.png
├── 8.png
├── 9.png
├── services/
├── __init__.py
├── test_api_authentication.py (700 tokens)
├── test_api_validation_json.py (500 tokens)
├── test_artifact_download_mime.py (300 tokens)
├── test_artifact_uri_escaping.py (200 tokens)
├── test_asgi_cors.py (2000 tokens)
├── test_asgi_static_files.py (800 tokens)
├── test_audio_export_volume.py (900 tokens)
├── test_audio_stage_terms.py (300 tokens)
├── test_azure_boundary_duration.py (400 tokens)
├── test_azure_file_publication.py (1000 tokens)
├── test_azure_locale_escaping.py (300 tokens)
├── test_batch_material_allocation.py (1600 tokens)
├── test_bgm.py (3.6k tokens)
├── test_bgm_decoded_audio.py (200 tokens)
├── test_bgm_final_fade.py (400 tokens)
├── test_bgm_list_race.py (100 tokens)
├── test_blank_material_terms.py (300 tokens)
├── test_cache_manager.py (1300 tokens)
├── test_caption_timecodes.py (200 tokens)
├── test_cli.py (17.4k tokens)
├── test_cli_stage_files.py (300 tokens)
├── test_clip_speed.py (200 tokens)
├── test_combine_clip_isolation.py (500 tokens)
├── test_compatible_speech_transport.py (200 tokens)
├── test_concat_manifest_isolation.py (500 tokens)
├── test_concat_model_default.py (700 tokens)
├── test_concurrent_task_delete.py (600 tokens)
├── test_config.py (5k tokens)
├── test_config_flush_start.py (400 tokens)
├── test_config_initialization.py (1300 tokens)
├── test_controller_base.py (1200 tokens)
├── test_controller_llm.py (700 tokens)
├── test_controller_ping.py (100 tokens)
├── test_controller_video.py (7.3k tokens)
├── test_directory_creation.py (400 tokens)
├── test_download_complete_decode.py (400 tokens)
├── test_download_directory_race.py (300 tokens)
├── test_edge_atomic_audio.py (400 tokens)
├── test_edge_backpressure.py (900 tokens)
├── test_elevenlabs_catalog_pagination.py (900 tokens)
├── test_elevenlabs_music.py (4.8k tokens)
├── test_elevenlabs_speech_speed.py (1200 tokens)
├── test_finite_audio_parameters.py (300 tokens)
├── test_fish_audio.py (2.8k tokens)
├── test_fluxionai_llm.py (1000 tokens)
├── test_gemini_atomic_export.py (300 tokens)
├── test_generated_term_contract.py (200 tokens)
├── test_image_exif_orientation.py (300 tokens)
├── test_image_sanitize_identity.py (500 tokens)
├── test_image_submission_redirect.py (400 tokens)
├── test_kokoro.py (1400 tokens)
├── test_kokoro_integration.py (400 tokens)
├── test_llm.py (20.1k tokens)
├── test_llm_api_failures.py (200 tokens)
├── test_llm_client_cleanup.py (600 tokens)
├── test_loom_download_staging_native.py (600 tokens)
├── test_loom_poll_deadline_native.py (400 tokens)
├── test_loomloom.py (5.8k tokens)
├── test_loomloom_pagination_cycles.py (600 tokens)
├── test_loomloom_request_redirects.py (600 tokens)
├── test_material.py (20k tokens)
├── test_material_cache.py (4.4k tokens)
├── test_material_cache_duration.py (300 tokens)
├── test_material_cache_replacement.py (800 tokens)
├── test_material_cover_picture.py (400 tokens)
├── test_material_openai_image.py (7.3k tokens)
├── test_material_source_preflight.py (300 tokens)
├── test_material_upload.py (3.5k tokens)
├── test_media_accepted_audio_retries.py (400 tokens)
├── test_media_atomic_concat.py (400 tokens)
├── test_media_image_fallback.py (200 tokens)
├── test_media_mimo_publication.py (700 tokens)
├── test_media_srt_publication.py (1200 tokens)
├── test_metaso_minimax.py (3k tokens)
├── test_mimo_paid_retries.py (300 tokens)
├── test_minimax_success_retry.py (600 tokens)
├── test_mpt_agent_skill.py (9.4k tokens)
├── test_muapi.py (3.1k tokens)
├── test_music_provider_redirects.py (800 tokens)
├── test_nullable_narration_volume.py (300 tokens)
├── test_ofox.py (7.7k tokens)
├── test_paid_video_provider_redirects.py (800 tokens)
├── test_partial_batch_results.py (2.1k tokens)
├── test_pause_concat_completeness.py (400 tokens)
├── test_pexels_renditions.py (700 tokens)
├── test_posix_concat_paths.py (400 tokens)
├── test_qualified_voice_identity.py (700 tokens)
├── test_queue_signature.py (400 tokens)
├── test_queued_preview_snapshot.py (400 tokens)
├── test_redis_patch_ownership.py (400 tokens)
├── test_redis_queue_admission.py (700 tokens)
├── test_redis_short_form_requests.py (700 tokens)
├── test_remote_speech_complete_decode.py (800 tokens)
├── test_render_parameter_validation.py (300 tokens)
├── test_retry_diagnostics.py (400 tokens)
├── test_schema.py (1300 tokens)
├── test_script_patch_concurrency.py (400 tokens)
├── test_serial_material_search.py (300 tokens)
├── test_short_clip_transitions.py (500 tokens)
├── test_social_error_text.py (400 tokens)
├── test_sonilo.py (5k tokens)
├── test_sonilo_event_bound.py (300 tokens)
├── test_speech_post_redirects.py (500 tokens)
├── test_standalone_upload_containers.py (400 tokens)
├── test_state.py (3.1k tokens)
├── test_state_input_snapshot.py (200 tokens)
├── test_state_string_values.py (500 tokens)
├── test_state_wrongtype.py (500 tokens)
├── test_subtitle.py (3.9k tokens)
├── test_subtitle_background_settings.py (1100 tokens)
├── test_subtitle_color_preflight.py (700 tokens)
├── test_subtitle_manual_wrap.py (300 tokens)
├── test_subtitle_stage_result.py (500 tokens)
├── test_task.py (20.8k tokens)
├── test_task_artifacts.py (700 tokens)
├── test_task_manager.py (5.1k tokens)
├── test_twelvelabs.py (1400 tokens)
├── test_twelvelabs_client_lifecycle.py (400 tokens)
├── test_twelvelabs_invalid_embedding_native.py (600 tokens)
├── test_upload_post.py (3.9k tokens)
├── test_upload_post_account_snapshot.py (1100 tokens)
├── test_upload_post_missing_platform.py (500 tokens)
├── test_upload_post_privacy_fallback.py (400 tokens)
├── test_upload_post_timeout_id.py (700 tokens)
├── test_upload_redirect_handle_native.py (300 tokens)
├── test_upload_response_recovery.py (300 tokens)
├── test_utils_ffmpeg_check.py (500 tokens)
├── test_version_checker.py (1900 tokens)
├── test_video.py (16.3k tokens)
├── test_video_color_tags.py (500 tokens)
├── test_video_effects.py (1400 tokens)
├── test_video_project.py (5k tokens)
├── test_video_reader_stdout.py (600 tokens)
├── test_video_term_items.py (500 tokens)
├── test_voice.py (20.9k tokens)
├── test_volcengine_seedance.py (7.5k tokens)
├── test_voxcpm.py (2.5k tokens)
├── test_voxcpm_stream_bounds.py (1300 tokens)
├── test_voxcpm_truncated_reference.py (300 tokens)
├── test_webui_audio_staging.py (1000 tokens)
├── test_webui_bgm.py (4.8k tokens)
├── test_webui_catalan.py (700 tokens)
├── test_webui_custom_audio_upload.py (900 tokens)
├── test_webui_font_catalog.py (600 tokens)
├── test_webui_generation_defaults.py (3.8k tokens)
├── test_webui_grouped_select.py (1300 tokens)
├── test_webui_headless_task_actions.py (600 tokens)
├── test_webui_i18n.py (2.9k tokens)
├── test_webui_kokoro.py (700 tokens)
├── test_webui_llm_settings.py (1900 tokens)
├── test_webui_local_material_upload.py (900 tokens)
├── test_webui_loomloom.py (4.4k tokens)
├── test_webui_loomloom_regressions.py (2.9k tokens)
├── test_webui_metaso_minimax.py (1000 tokens)
├── test_webui_muapi.py (400 tokens)
├── test_webui_ofox.py (400 tokens)
├── test_webui_ofox_placement.py (600 tokens)
├── test_webui_seedance.py (500 tokens)
├── test_webui_settings_transfer.py (3.2k tokens)
├── test_webui_startup.py (500 tokens)
├── test_webui_task.py (5.3k tokens)
├── test_webui_task_history.py (1600 tokens)
├── test_webui_tts_settings.py (3k tokens)
├── test_webui_upload_post_settings.py (1100 tokens)
├── test_webui_voice_preview.py (4.6k tokens)
├── test_youtube_audience.py (1400 tokens)
├── test_zoom_alpha_alignment.py (300 tokens)
├── test_main.py (200 tokens)
├── uv.lock (omitted)
├── webui.bat (600 tokens)
├── webui.sh (500 tokens)
├── webui/
├── .streamlit/
├── config.toml
├── Main.py (66.3k tokens)
├── i18n/
├── az.json (7.2k tokens)
├── ca.json (7k tokens)
├── de.json (7.3k tokens)
├── en.json (12.6k tokens)
├── es.json (7.1k tokens)
├── fr.json (7.4k tokens)
├── id.json (6.8k tokens)
├── it.json (7.4k tokens)
├── ko.json (5.1k tokens)
├── pt.json (7k tokens)
├── ru.json (6.9k tokens)
├── tr.json (6.8k tokens)
├── vi.json (6.6k tokens)
├── zh.json (8.5k tokens)
├── styles.css (2.6k tokens)
```
## /.dockerignore
```dockerignore path="/.dockerignore"
# Exclude common Python files and directories
venv/
.venv/
__pycache__/
*.pyc
*.pyo
*.pyd
*.pyz
*.pyw
*.pyi
*.egg-info/
# Exclude development and local files
.env
.env.*
*.log
*.db
.DS_Store
.pytest_cache/
.ruff_cache/
.idea/
logs/
# Exclude version control system files
.git/
.gitignore
.svn/
storage/
config.toml
```
## /.github/ISSUE_TEMPLATE/bug_report.yml
```yml path="/.github/ISSUE_TEMPLATE/bug_report.yml"
name: 🐛 Bug | Bug Report
description: 报告错误或异常问题 | Report an error or unexpected behavior
title: "[Bug]: "
labels:
- bug
body:
- type: markdown
attributes:
value: |
**提交问题前，请确保您已阅读以下文档：[Getting Started (English)](https://github.com/harry0703/MoneyPrinterTurbo/blob/main/README-en.md#system-requirements-) 或 [快速开始 (中文)](https://github.com/harry0703/MoneyPrinterTurbo/blob/main/README.md#%E5%BF%AB%E9%80%9F%E5%BC%80%E5%A7%8B-)。**
**Before submitting an issue, please make sure you've read the following documentation: [Getting Started (English)](https://github.com/harry0703/MoneyPrinterTurbo/blob/main/README-en.md#system-requirements-) or [快速开始 (Chinese)](https://github.com/harry0703/MoneyPrinterTurbo/blob/main/README.md#%E5%BF%AB%E9%80%9F%E5%BC%80%E5%A7%8B-).**
- type: textarea
attributes:
label: 问题描述 | Current Behavior
description: |
描述您遇到的问题
Describe the issue you're experiencing
placeholder: |
当我执行...操作时，程序出现了...问题
When I perform..., the program shows...
validations:
required: true
- type: textarea
attributes:
label: 重现步骤 | Steps to Reproduce
description: |
详细描述如何重现此问题
Describe in detail how to reproduce this issue
placeholder: |
1. 打开...
2. 点击...
3. 出现错误...
1. Open...
2. Click on...
3. Error occurs...
validations:
required: true
- type: textarea
attributes:
label: 错误日志 | Error Logs
description: |
请提供相关错误信息或日志（注意不要包含敏感信息）
Please provide any error messages or logs (be careful not to include sensitive information)
placeholder: |
错误信息、日志或截图...
Error messages, logs, or screenshots...
validations:
required: true
- type: input
attributes:
label: Python 版本 | Python Version
description: |
您使用的 Python 版本
The Python version you're using
placeholder: v3.13.0, v3.10.0, etc.
validations:
required: true
- type: input
attributes:
label: 操作系统 | Operating System
description: |
您的操作系统信息
Your operating system information
placeholder: macOS 14.1, Windows 11, Ubuntu 22.04, etc.
validations:
required: true
- type: input
attributes:
label: MoneyPrinterTurbo 版本 | Version
description: |
您使用的 MoneyPrinterTurbo 版本
The version of MoneyPrinterTurbo you're using
placeholder: v1.2.2, etc.
validations:
required: true
- type: textarea
attributes:
label: 补充信息 | Additional Information
description: |
其他对解决问题有帮助的信息（如截图、视频等）
Any other information that might help solve the issue (screenshots, videos, etc.)
validations:
required: false
```
## /.github/ISSUE_TEMPLATE/config.yml
```yml path="/.github/ISSUE_TEMPLATE/config.yml"
blank_issues_enabled: false
```
## /.github/ISSUE_TEMPLATE/feature_request.yml
```yml path="/.github/ISSUE_TEMPLATE/feature_request.yml"
name: ✨ 增加功能 | Feature Request
description: 为此项目提出一个新想法或建议 | Suggest a new idea for this project
title: "[Feature]: "
labels:
- enhancement
body:
- type: textarea
attributes:
label: 需求描述 | Problem Statement
description: |
请描述您希望解决的问题或需求
Please describe the problem you want to solve
placeholder: |
我在使用过程中遇到了...
I encountered... when using this project
validations:
required: true
- type: textarea
attributes:
label: 建议的解决方案 | Proposed Solution
description: |
请描述您认为可行的解决方案或实现方式
Please describe your suggested solution or implementation
placeholder: |
可以考虑添加...功能来解决这个问题
Consider adding... feature to address this issue
validations:
required: true
```
## /.github/SECURITY.md
# Security Policy
## Supported Versions
Security fixes are applied on a best-effort basis to the latest `main` branch and the most recent published release line.
## Reporting a Vulnerability
Please do **not** disclose suspected vulnerabilities in public GitHub issues.
Preferred process:
1. Use GitHub private vulnerability reporting for this repository if it is available in the repository security settings.
2. If private reporting is not available, open a minimal public issue that only requests a private contact channel and does **not** include vulnerability details, proof-of-concept code, payloads, or sensitive file paths.
3. Wait for a maintainer response before sharing any technical details publicly.
When reporting a vulnerability privately, include:
- affected commit, tag, or release version
- attack surface or vulnerable endpoint
- impact summary
- reproduction conditions
- suggested remediation, if available
## Disclosure Expectations
- Please give maintainers reasonable time to investigate and prepare a fix before public disclosure.
- Once a fix is available, coordinated public disclosure is welcome.
## /.github/workflows/ci.yml
```yml path="/.github/workflows/ci.yml"
name: CI
on:
pull_request:
push:
branches:
- main
jobs:
tests:
name: Python ${{ matrix.python-version }} tests
runs-on: ubuntu-latest
timeout-minutes: 30
strategy:
fail-fast: false
matrix:
python-version: ["3.11", "3.13"]
services:
redis:
image: redis:7-alpine
ports:
- 6379:6379
options: >-
--health-cmd "redis-cli ping"
--health-interval 5s
--health-timeout 3s
--health-retries 10
env:
MPT_TEST_REDIS_HOST: 127.0.0.1
MPT_TEST_REDIS_PORT: 6379
MPT_TEST_REDIS_DB: 15
steps:
- name: Check out repository
uses: actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v5
with:
python-version: ${{ matrix.python-version }}
- name: Install uv
uses: astral-sh/setup-uv@v6
with:
enable-cache: true
- name: Install dependencies
run: uv sync --frozen --python ${{ matrix.python-version }}
- name: Compile Python sources
run: uv run --no-sync python -m compileall app cli.py main.py webui test docs/skill
- name: Lint Python sources
if: matrix.python-version == '3.11'
run: uv run --no-sync ruff check app cli.py main.py webui test docs/skill
- name: Run unit tests with coverage
# pytest 兼容现有 unittest.TestCase，同时会收集仓库中的 pytest 风格
# 测试，避免测试文件已存在但 CI 没有真正执行。
run: uv run --no-sync python -X utf8 -m coverage run -m pytest -q test
# Keep the legacy suite's portable media backend selection unchanged.
# Native project rendering is a separate mandatory pass after installing
# system tools; append its coverage before enforcing the shared threshold.
- name: Install FFmpeg for native revision-rendering tests
run: sudo apt-get update && sudo apt-get install -y ffmpeg
- name: Run native revision-rendering tests
run: >-
uv run --no-sync python -X utf8 -m coverage run --append -m pytest -q
test/services/test_video_project.py
- name: Report test coverage
run: uv run --no-sync python -m coverage report
- name: Export coverage report
run: uv run --no-sync python -m coverage xml
- name: Upload coverage report
uses: actions/upload-artifact@v4
with:
name: coverage-python-${{ matrix.python-version }}
path: coverage.xml
retention-days: 7
windows-smoke:
name: Windows smoke tests
runs-on: windows-latest
timeout-minutes: 30
steps:
- name: Check out repository
uses: actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v5
with:
python-version: "3.11"
- name: Install uv
uses: astral-sh/setup-uv@v6
with:
enable-cache: true
- name: Install dependencies
run: uv sync --frozen --python 3.11
- name: Compile Python sources
run: uv run --no-sync python -m compileall app cli.py main.py webui test docs/skill
- name: Run core service smoke tests
run: >-
uv run --no-sync python -X utf8 -m pytest -q
test/services/test_config.py
test/services/test_state.py
test/services/test_task.py
test/services/test_task_manager.py
test/services/test_upload_post.py
test/services/test_controller_video.py
test/services/test_webui_task.py
test/services/test_webui_headless_task_actions.py
```
## /.github/workflows/docker-ghcr.yml
```yml path="/.github/workflows/docker-ghcr.yml"
name: Publish Docker image
on:
push:
branches:
- main
tags:
- "v*"
paths-ignore:
- "README*.md"
- "docs/**"
- ".github/workflows/docker-ghcr.yml"
workflow_dispatch:
permissions:
contents: read
packages: write
concurrency:
group: docker-ghcr-${{ github.ref }}
cancel-in-progress: true
env:
IMAGE_NAME: ghcr.io/harry0703/moneyprinterturbo
jobs:
publish:
name: Build and publish
runs-on: ubuntu-latest
timeout-minutes: 90
steps:
- name: Check out repository
uses: actions/checkout@v4
- name: Set up QEMU
uses: docker/setup-qemu-action@v3
- name: Set up Docker Buildx
uses: docker/setup-buildx-action@v3
- name: Log in to GHCR
uses: docker/login-action@v3
with:
registry: ghcr.io
username: ${{ github.actor }}
password: ${{ secrets.GITHUB_TOKEN }}
- name: Extract Docker metadata
id: meta
uses: docker/metadata-action@v5
with:
images: ${{ env.IMAGE_NAME }}
tags: |
type=ref,event=branch
type=ref,event=tag
type=semver,pattern={{version}}
type=raw,value=latest,enable={{is_default_branch}}
- name: Build and push Docker image
uses: docker/build-push-action@v6
with:
context: .
file: ./Dockerfile
push: true
provenance: mode=max
sbom: true
platforms: linux/amd64,linux/arm64
build-args: |
DOCKER_BUILD_MIRROR=default
PIP_USE_OFFICIAL=1
tags: ${{ steps.meta.outputs.tags }}
labels: ${{ steps.meta.outputs.labels }}
cache-from: type=gha
cache-to: type=gha,mode=max
```
## /.gitignore
```gitignore path="/.gitignore"
.DS_Store
/config.toml
/config.toml.bak
/storage/
/.idea/
/app/services/__pycache__
/app/__pycache__/
/app/config/__pycache__/
/app/models/__pycache__/
/app/utils/__pycache__/
/*/__pycache__/*
.vscode
/**/.streamlit
__pycache__
logs/
node_modules
# VuePress 默认临时文件目录
/sites/docs/.vuepress/.temp
# VuePress 默认缓存目录
/sites/docs/.vuepress/.cache
# VuePress 默认构建生成的静态文件目录
/sites/docs/.vuepress/dist
# 模型目录
/models/
./models/*
venv/
.venv
# Debug and test files
CLAUDE.md
debug/
debug_*.py
test_*.py
!test/services/test_*.py
!test/test_*.py
streamlit.log
.coverage
coverage.xml
htmlcov/
.codegraph
/docs/superpowers
.understand-anything
# Claude Code 订阅令牌（docker compose 自动读取），禁止提交
.env
```
## /.python-version
```python-version path="/.python-version"
3.11
```
## /Dockerfile
``` path="/Dockerfile"
# Use an official Python runtime as a parent image
FROM python:3.11-slim-bullseye
# Set the working directory in the container
WORKDIR /MoneyPrinterTurbo
# 设置/MoneyPrinterTurbo目录权限为777
RUN chmod 777 /MoneyPrinterTurbo
ENV PYTHONPATH="/MoneyPrinterTurbo"
# 本地用户默认继续优先使用国内镜像；GitHub Actions 发布 GHCR 镜像时使用 default，
# 避免海外 runner 访问国内镜像过慢导致镜像发布长时间卡住。
ARG DOCKER_BUILD_MIRROR=china
ARG PIP_USE_OFFICIAL=0
# 系统依赖安装需要同时满足两点：国内环境保留镜像回退能力，所有镜像均
# 失败时必须让 Docker 构建立刻失败。旧循环最后执行的 sleep 总会返回 0，
# 导致 git/ffmpeg 未安装时仍生成不可用镜像。这里把“写入软件源”“安装”
# 和“三次重试”拆成边界清晰的 shell 函数，并用函数返回值决定是否继续。
# 所有软件源统一使用 HTTPS，避免部分网络环境直接拦截明文 HTTP 请求。
RUN set -u; \
write_debian_sources() { \
main_url="$1"; \
security_url="$2"; \
printf 'deb %s bullseye main\ndeb %s bullseye-updates main\ndeb %s bullseye-security main\n' \
"$main_url" "$main_url" "$security_url" > /etc/apt/sources.list; \
rm -rf /var/lib/apt/lists/*; \
}; \
install_system_dependencies() { \
apt-get update && \
apt-get install -y --no-install-recommends git ffmpeg; \
}; \
retry_system_dependencies() { \
attempt=1; \
while [ "$attempt" -le 3 ]; do \
echo "Attempt $attempt: installing system dependencies"; \
if install_system_dependencies; then \
return 0; \
fi; \
echo "Attempt $attempt failed" >&2; \
if [ "$attempt" -lt 3 ]; then \
echo "Retrying in 5 seconds..." >&2; \
sleep 5; \
fi; \
attempt=$((attempt + 1)); \
done; \
return 1; \
}; \
if [ "$DOCKER_BUILD_MIRROR" = "china" ]; then \
write_debian_sources \
"https://mirrors.aliyun.com/debian" \
"https://mirrors.aliyun.com/debian-security"; \
if ! retry_system_dependencies; then \
echo "Aliyun mirror failed, switching to Tsinghua mirror" >&2; \
write_debian_sources \
"https://mirrors.tuna.tsinghua.edu.cn/debian" \
"https://mirrors.tuna.tsinghua.edu.cn/debian-security"; \
if ! install_system_dependencies; then \
echo "Tsinghua mirror failed, switching to default Debian mirror" >&2; \
write_debian_sources \
"https://deb.debian.org/debian" \
"https://deb.debian.org/debian-security"; \
if ! install_system_dependencies; then \
echo "Failed to install system dependencies from all configured mirrors" >&2; \
exit 1; \
fi; \
fi; \
fi; \
else \
echo "Using default Debian mirrors"; \
write_debian_sources \
"https://deb.debian.org/debian" \
"https://deb.debian.org/debian-security"; \
if ! retry_system_dependencies; then \
echo "Failed to install system dependencies from the default Debian mirror" >&2; \
exit 1; \
fi; \
fi; \
rm -rf /var/lib/apt/lists/*
# Copy only the requirements.txt first to leverage Docker cache
COPY requirements.txt ./
# 本地默认优先国内 PyPI 镜像；GHCR 发布使用官方 PyPI，避免海外 runner 因跨境镜像访问变慢。
RUN if [ "$PIP_USE_OFFICIAL" = "1" ]; then \
pip install --no-cache-dir --retries 3 --timeout 60 -r requirements.txt; \
else \
pip install --no-cache-dir -i https://mirrors.aliyun.com/pypi/simple/ --trusted-host mirrors.aliyun.com --retries 3 --timeout 60 -r requirements.txt || \
pip install --no-cache-dir -i https://mirrors.tuna.tsinghua.edu.cn/pypi/web/simple/ --trusted-host mirrors.tuna.tsinghua.edu.cn --retries 3 --timeout 60 -r requirements.txt || \
pip install --no-cache-dir --retries 3 --timeout 60 -r requirements.txt; \
fi
# Now copy the rest of the codebase into the image
COPY . .
# Expose the port the app runs on
EXPOSE 8501
# 容器内部必须监听 0.0.0.0，宿主机仍通过 docker 端口映射限制为 127.0.0.1。
# browser.serverAddress 只决定浏览器展示的访问地址，不能替代 server.address。
CMD ["streamlit", "run", "./webui/Main.py", "--server.address=0.0.0.0", "--server.port=8501", "--browser.serverAddress=127.0.0.1", "--server.enableCORS=True", "--browser.gatherUsageStats=False", "--client.toolbarMode=minimal", "--logger.hideWelcomeMessage=True", "--server.showEmailPrompt=False"]
# 1. Build the Docker image using the following command
# docker build -t moneyprinterturbo .
# 2. Run the Docker container using the following command
## For Linux or MacOS:
# docker run -v $(pwd)/config.toml:/MoneyPrinterTurbo/config.toml -v $(pwd)/storage:/MoneyPrinterTurbo/storage -p 127.0.0.1:8501:8501 moneyprinterturbo
## For Windows:
# docker run -v ${PWD}/config.toml:/MoneyPrinterTurbo/config.toml -v ${PWD}/storage:/MoneyPrinterTurbo/storage -p 127.0.0.1:8501:8501 moneyprinterturbo
```
## /Dockerfile.claude
```claude path="/Dockerfile.claude"
# Claude Code 变体镜像：在官方发布镜像的基础上安装 claude CLI。
#
# Claude 订阅（Pro / Max / Team）不签发 API Key，只有 Claude Code 官方客户端
# 能使用订阅凭证，所以 `claude_code` Provider 依赖容器内存在可执行的 claude
# CLI。这里复用发布镜像，避免重新安装 Python 依赖。
FROM ghcr.io/harry0703/moneyprinterturbo:latest
# CLI 版本固定，避免镜像重建时行为漂移；升级时改这里的版本号即可。
ARG CLAUDE_CODE_VERSION=2.1.260
ARG NODE_MAJOR=22
# claude CLI 通过 npm 分发，发布镜像里没有 Node.js，需要先装 NodeSource 源。
RUN set -eu; \
apt-get update; \
apt-get install -y --no-install-recommends ca-certificates curl gnupg; \
curl -fsSL "https://deb.nodesource.com/setup_${NODE_MAJOR}.x" | bash -; \
apt-get install -y --no-install-recommends nodejs; \
npm install -g "@anthropic-ai/claude-code@${CLAUDE_CODE_VERSION}"; \
npm cache clean --force; \
apt-get purge -y gnupg; \
apt-get autoremove -y; \
apt-get clean; \
rm -rf /var/lib/apt/lists/*; \
claude --version
```
## /Dockerfile.gpu
```gpu path="/Dockerfile.gpu"
# Use NVIDIA CUDA runtime as parent image (includes CUDA 12.1 + cuDNN 8)
FROM nvidia/cuda:12.1.1-cudnn8-runtime-ubuntu22.04
# Avoid interactive timezone prompt
ENV DEBIAN_FRONTEND=noninteractive
# Set the working directory in the container
WORKDIR /MoneyPrinterTurbo
RUN chmod 777 /MoneyPrinterTurbo
ENV PYTHONPATH="/MoneyPrinterTurbo"
# 安装链中的任何前置步骤失败都必须终止构建。只有删除系统自带 blinker
# 属于可忽略操作，因此把容错限制在该子命令的括号内，避免末尾的
# `|| true` 把 apt、PPA、Python 或 pip 安装失败一并转换成成功状态。
RUN apt-get update && apt-get install -y --no-install-recommends \
software-properties-common \
git \
ffmpeg \
curl \
&& add-apt-repository ppa:deadsnakes/ppa \
&& apt-get update && apt-get install -y --no-install-recommends \
python3.11 \
python3.11-venv \
python3.11-dev \
python3-pip \
&& ln -sf /usr/bin/python3.11 /usr/bin/python3 \
&& ln -sf /usr/bin/python3.11 /usr/bin/python \
&& python3.11 -m pip install --upgrade pip \
&& (apt-get remove -y python3-blinker || \
echo "python3-blinker removal skipped; continuing") \
&& rm -rf /var/lib/apt/lists/*
# Copy only the requirements.txt first to leverage Docker cache
COPY requirements.txt ./
# Install Python dependencies
RUN python3 -m pip install --no-cache-dir \
-i https://mirrors.aliyun.com/pypi/simple/ --trusted-host mirrors.aliyun.com \
--retries 3 --timeout 60 -r requirements.txt || \
python3 -m pip install --no-cache-dir \
-i https://mirrors.tuna.tsinghua.edu.cn/pypi/web/simple/ --trusted-host mirrors.tuna.tsinghua.edu.cn \
--retries 3 --timeout 60 -r requirements.txt || \
python3 -m pip install --no-cache-dir \
--retries 3 --timeout 60 -r requirements.txt
# Now copy the rest of the codebase into the image
COPY . .
# Expose the port the app runs on
EXPOSE 8501
# 容器内部必须监听 0.0.0.0，宿主机仍通过 docker 端口映射限制为 127.0.0.1。
# browser.serverAddress 只决定浏览器展示的访问地址，不能替代 server.address。
CMD ["streamlit", "run", "./webui/Main.py", "--server.address=0.0.0.0", "--server.port=8501", "--browser.serverAddress=127.0.0.1", "--server.enableCORS=True", "--browser.gatherUsageStats=False", "--client.toolbarMode=minimal", "--logger.hideWelcomeMessage=True", "--server.showEmailPrompt=False"]
```
## /README-en.md
<div align="center">
# MoneyPrinterTurbo 💸
### An All-in-One AI Short Video Generator
Provide a video <b>topic</b> or <b>keyword</b>, and MoneyPrinterTurbo will generate the script, match footage, create subtitles and background music, and produce an HD short video.
[](https://github.com/harry0703/MoneyPrinterTurbo/releases)
[](https://github.com/harry0703/MoneyPrinterTurbo/releases/latest)
[](https://www.python.org/)
[](https://github.com/harry0703/MoneyPrinterTurbo/releases/latest)
<a href="https://trendshift.io/repositories/8731" target="_blank"><img src="https://trendshift.io/api/badge/repositories/8731" alt="harry0703%2FMoneyPrinterTurbo | Trendshift" style="width: 250px; height: 55px;" width="250" height="55"/></a>
<a href="https://www.star-history.com/harry0703/moneyprinterturbo"><img src="https://api.star-history.com/badge?repo=harry0703/MoneyPrinterTurbo" alt="Star History Rank" style="height: 55px;" height="55"/></a>
English | [简体中文](README.md) | [日本語](README-ja.md) | [Releases](https://github.com/harry0703/MoneyPrinterTurbo/releases) | [Issues](https://github.com/harry0703/MoneyPrinterTurbo/issues)
</div>
## Screenshots 🖥️
<h4 align="center">WebUI</h4>
<h4 align="center">API</h4>
---
## Special Thanks ❤️
<div align="center">
<a href="https://platform.kimi.ai?track_id=track-9e3b711aa2594e378f6fe5b8de718a76&aff=moneyprinterturbo" target="_blank"><img src="https://gcdn.moonshot.cn/growth-cdn/sponsor/kimi-en.png" alt="Kimi sponsors MoneyPrinterTurbo" width="100%"></a>
</div>
Thanks to [Kimi](https://platform.kimi.ai?track_id=track-9e3b711aa2594e378f6fe5b8de718a76&aff=moneyprinterturbo) for sponsoring this project! [Kimi K3](https://www.kimi.com/blog/kimi-k3?aff=moneyprinterturbo) is Moonshot AI's most capable model and the world's first open 3T-class model. With native vision and a 1-million-token context window, K3 delivers frontier performance across knowledge work, reasoning, and long-horizon tasks. Within MoneyPrinterTurbo, K3 powers video creation by writing scripts and extracting the search keywords that determine the final footage—the better it understands the content, the more relevant the results.
**Exclusive offer for MoneyPrinterTurbo users: new users who register through the dedicated link receive bonus API credit equal to 10% of their first successful top-up, up to CNY 1,000. The offer ends December 31, 2026. Visit the Kimi Open Platform ([中文站](https://platform.kimi.com?track_id=track-6eec1e56a4494e52adcaebbcbbefce59&aff=moneyprinterturbo) | [Global](https://platform.kimi.ai?track_id=track-9e3b711aa2594e378f6fe5b8de718a76&aff=moneyprinterturbo)) to try the API.**
<br>
<table align="center">
<tr>
<td align="center" width="120">
<a href="https://www.byteplus.com/en/product/modelark?utm_campaign=hw&utm_content=MoneyPrinterTurbo&utm_medium=devrel_tool_web&utm_source=OWO&utm_term=MoneyPrinterTurbo"><img src="docs/sponsors/byteplus-logo.svg" alt="BytePlus" height="25"></a><br>
<a href="https://www.byteplus.com/en/product/modelark?utm_campaign=hw&utm_content=MoneyPrinterTurbo&utm_medium=devrel_tool_web&utm_source=OWO&utm_term=MoneyPrinterTurbo"><strong>BytePlus ModelArk</strong></a>
</td>
<td align="left">
Thanks to ByteDance VolcEngine for sponsoring this project! VolcEngine Ark's Agent/Coding Plan for leading Chinese models starts at CNY 9.9 for first-time buyers and supports GLM-5.3, Kimi-K3, DeepSeek, MiniMax, Doubao, and more. New users receive 25 million free tokens. One unified API is designed for coding and agent development. <a href="https://www.byteplus.com/en/product/modelark?utm_campaign=hw&utm_content=MoneyPrinterTurbo&utm_medium=devrel_tool_web&utm_source=OWO&utm_term=MoneyPrinterTurbo">Visit now</a>
</td>
</tr>
<tr>
<td align="center" width="120">
<a href="https://go.apimart.ai/gh-moneyprinterturbo"><img src="docs/sponsors/apimart-logo.png" alt="APIMart" width="100"></a>
</td>
<td align="left">
Thanks to <a href="https://go.apimart.ai/gh-moneyprinterturbo">APIMart</a> for sponsoring this project! APIMart is a low-cost API platform for AI image & video generation — <strong>GPT-Image-2 from $0.006/image, 160+ images per dollar</strong>. <strong>One async API covers both image and video—switch models without changing code.</strong> Submit a task, get an ID, and fetch results via polling or callback. Batch tens of thousands of images without timeouts. Pay-as-you-go with no monthly fee — <a href="https://go.apimart.ai/gh-moneyprinterturbo">sign up here</a> to get started.
</td>
</tr>
<tr>
<td align="center" width="120">
<a href="https://metaso.cn/minimax-h3/?s=MPT"><img src="docs/sponsors/metaso-logo.png" alt="Metaso" width="100"></a><br>
<a href="https://metaso.cn/minimax-h3/?s=MPT"><strong>Metaso</strong></a>
</td>
<td align="left">
<strong>MiniMax H3 Video Generation API by Metaso</strong><br>
Metaso offers a cost-effective MiniMax H3 video generation service: <strong>768p for just CNY 0.09 per second and 2K for CNY 0.15 per second</strong>. It supports native 2K output, synchronized audio and video, an OpenAI-compatible API, and ComfyUI—all without requiring you to deploy or manage GPUs.<br>
🎁 Sign up through the <a href="https://metaso.cn/minimax-h3/?s=MPT">exclusive MoneyPrinterTurbo link</a> to receive bonus credits and special offers.
</td>
</tr>
<tr>
<td align="center" width="120">
<a href="https://infistar.cc/register?aff=6T4EYXP2&ref_source=link"><img src="docs/sponsors/infistar-logo.svg" alt="Infistar.cc" height="56"></a><br>
<a href="https://infistar.cc/register?aff=6T4EYXP2&ref_source=link"><strong>Infistar.cc</strong></a>
</td>
<td align="left">
Thanks to <a href="https://infistar.cc/register?aff=6T4EYXP2&ref_source=link"><strong>Infistar.cc</strong></a> for sponsoring this open-source project! Infistar.cc offers cost-effective AI API access: LLMs from <strong>1% of official rates</strong> and AI image generation from <strong>CNY 0.06 per image</strong>.<br>
<strong>One API key for text, image, and video creation</strong>, with access to leading models including GPT, Claude, Gemini, DeepSeek, Qwen, and Kling—no separate API groups required.<br>
<strong>Authenticity checks for every model</strong>, transparent pricing, an ICP filing in China, real-time billing in CNY, and <strong>business invoices available</strong>.<br>
🎁 Exclusive MPT offer: <a href="https://infistar.cc/register?aff=6T4EYXP2&ref_source=link">register through our dedicated link</a> to receive <strong>$5 in trial credits</strong> and a special first-top-up offer.
</td>
</tr>
<tr>
<td align="center" width="120">
<a href="https://ofox.ai/?utm_source=github&utm_medium=sponsorship&utm_content=moneyprinterturbo"><img src="docs/sponsors/ofox-logo.svg" alt="OfoxAI" width="120"></a>
</td>
<td align="left">
Thanks to <a href="https://ofox.ai/?utm_source=github&utm_medium=sponsorship&utm_content=moneyprinterturbo">OfoxAI</a> for sponsoring this project! MoneyPrinterTurbo already supports Ofox multi-model text-to-video generation—just configure your API key to get started. Create video assets with Seedance, MiniMax H3, and Wan; design cover images with GPT Image 2.5 and Seedream; and refine scripts or build applications with GPT, Claude, Gemini, and DeepSeek. <strong>One key and a shared balance for text, image, and video models</strong>, with OpenAI-compatible endpoints and native Anthropic and Gemini interfaces. <strong>Pay-as-you-go billing, transparent pricing, and official model-provider channels deliver stable, high-speed, unlimited access.</strong> Explore <a href="https://ofox.ai/?utm_source=github&utm_medium=sponsorship&utm_content=moneyprinterturbo">OfoxAI models and pricing</a>.
</td>
</tr>
<tr>
<td align="center" width="120">
<a href="https://www.shengsuanyun.com/?from=CH_XUQ4OTSK"><img src="docs/sponsors/shengsuanyun-logo.jpg" alt="Shengsuan Cloud" height="56"></a><br>
<a href="https://www.shengsuanyun.com/?from=CH_XUQ4OTSK"><strong>Shengsuan Cloud</strong></a>
</td>
<td align="left">
Thanks to <a href="https://www.shengsuanyun.com/?from=CH_XUQ4OTSK">Shengsuan Cloud</a> for sponsoring this project! Shengsuan Cloud is an API aggregation platform for AI-native teams, providing unified, usage-based access to leading language and multimodal models including Claude, ChatGPT, and Gemini.<br>
The platform focuses on compliant API services and also offers enterprise gateways with team cost and permission management, intelligent routing, security controls, BYOK credential management, and invoice support.<br>
🎁 New users who register through <a href="https://www.shengsuanyun.com/?from=CH_XUQ4OTSK">this link</a> can receive CNY 10 in trial credits.
</td>
</tr>
<tr>
<td align="center" width="120">
<a href="https://www.ucloud.cn/site/active/astraflow?ytag=geo_waituo_Money"><img src="docs/sponsors/astraflow-logo.png" alt="AstraFlow" height="56"></a><br>
<a href="https://www.ucloud.cn/site/active/astraflow?ytag=geo_waituo_Money"><strong>AstraFlow</strong></a>
</td>
<td align="left">
Thanks to <a href="https://www.ucloud.cn/site/active/astraflow?ytag=geo_waituo_Money">AstraFlow</a> for sponsoring this project!<br>
🎬 <strong>One platform for leading video models</strong>: access MiniMax-H3, Seedance-2.5, and more. Generate scripts and videos in one place, without separate accounts or integrations.<br>
🚀 <strong>200+ AI models, with new models available on release day</strong>: access DeepSeek V4.1, Kimi K3, Qwen 3.8 Max, GLM 5.3, and more through a single platform.<br>
💰 <strong>Built by publicly listed UCloud, with transparent billing and cost control</strong>: track costs by API key and view detailed usage records.<br>
🎁 <strong>Exclusive offer for MoneyPrinterTurbo users</strong>: register through <a href="https://www.ucloud.cn/site/active/astraflow?ytag=geo_waituo_Money">our referral link</a> to receive new-user credits and get started right away! <a href="https://f.howxm.com/xs/u/HVFW5X8EIYV1">Claim CNY 50 in compute credits</a>.
</td>
</tr>
<tr>
<td align="center" width="120">
<a href="https://sidrune.ai/register?source=github&campaign=moneyprinterturbo&promo=moneyprinterturbo"><img src="docs/sponsors/sidrune-logo.png" alt="Sidrune AI" width="120"></a>
</td>
<td align="left">
Thanks to <a href="https://sidrune.ai/register?source=github&campaign=moneyprinterturbo&promo=moneyprinterturbo">Sidrune AI</a> for sponsoring this project! <strong>One gateway to access and manage leading AI models worldwide.</strong> Built for individual developers, technical teams, and enterprises, Sidrune AI offers a unified API to access and manage leading models. It dynamically selects between multiple routes to improve availability, with transparent model performance, response times, and costs. Depending on the model and route, <strong>API costs can be 40%–98% lower than official or benchmark rates</strong>. Sign up through <a href="https://sidrune.ai/register?source=github&campaign=moneyprinterturbo&promo=moneyprinterturbo">our exclusive link</a> to receive <strong>$3 in API credits</strong>.
</td>
</tr>
<tr>
<td align="center" width="120">
<a href="https://reccloud.com"><img src="docs/sponsors/reccloud-logo.svg" alt="RecCloud" height="36"></a><br>
<a href="https://reccloud.com">RecCloud</a>
</td>
<td align="left">
Thanks to <a href="https://reccloud.com">RecCloud</a>, an AI-powered multimedia platform, for offering a free <strong>AI Video Generator</strong> based on this project. Use it online with no deployment required—a beginner-friendly way to get started.
</td>
</tr>
<tr>
<td align="center" width="120">
<a href="https://picwish.com"><img src="docs/sponsors/picwish-logo.svg" alt="Picwish" height="36"></a><br>
<a href="https://picwish.com">Picwish</a>
</td>
<td align="left">
Thanks to <a href="https://picwish.com">Picwish</a> for supporting and sponsoring this project! Picwish offers a wide range of easy-to-use <strong>online image editing tools</strong> to help users handle everyday image editing tasks with ease.
</td>
</tr>
</table>
## Another Open-Source Project from the Creator: MangoDisk ⭐
<p align="center">
<a href="https://mangodisk.app">
<picture>
<source media="(prefers-color-scheme: dark)" srcset="https://assets.mangodisk.app/images/screenshots/en/dark-01-deep-cleanup.jpg?ver=1.1.0">
<source media="(prefers-color-scheme: light)" srcset="https://assets.mangodisk.app/images/screenshots/en/light-01-deep-cleanup.jpg?ver=1.1.0">
<img src="https://assets.mangodisk.app/images/screenshots/en/light-01-deep-cleanup.jpg?ver=1.1.0" width="900" alt="MangoDisk Deep Cleanup interface">
</picture>
</a>
</p>
<p align="center">
<strong>An open-source disk cleaner, storage analyzer, and system optimizer for macOS and Windows</strong><br>
Clean caches, large files, duplicates, and app leftovers; analyze disk usage; manage apps and startup items; and maintain your system.
</p>
<p align="center">
<a href="https://mangodisk.app">Visit the MangoDisk Website</a> · <a href="https://github.com/harry0703/MangoDisk">View on GitHub</a>
</p>
---
## Features 🎯
### Creation Workflows
- [x] Use **AI Agent, WebUI, API, or CLI** workflows for quick creation or automated production
- [x] Go from a topic to script, voiceover, footage, subtitles, music, and editing automatically, while retaining control over every stage
- [x] Generate multiple output variants in batches, review task history, and import or export generation settings and API keys
### Scripts and Model Providers
- [x] Generate or rewrite **multilingual video scripts** with AI, or supply a complete custom script
- [x] Use leading providers including [Kimi / Moonshot AI](https://platform.kimi.ai?track_id=track-9e3b711aa2594e378f6fe5b8de718a76&aff=moneyprinterturbo), [OpenAI](https://platform.openai.com/api-keys), [Anthropic Claude](https://platform.claude.com/settings/keys), [Google Gemini](https://aistudio.google.com/app/apikey), [DeepSeek](https://platform.deepseek.com/api_keys), [Alibaba Cloud Qwen](https://qwen.ai/apiplatform), [Microsoft Azure OpenAI](https://portal.azure.com/#view/Microsoft_Azure_ProjectOxford/CognitiveServicesHub/~/OpenAI), [ByteDance VolcEngine Ark](https://console.volcengine.com/ark), [xAI Grok](https://console.x.ai/), [MiniMax](https://platform.minimax.io/), and [Xiaomi MiMo](https://platform.xiaomimimo.com/docs/zh-CN/quick-start/first-api-call)
- [x] Connect through [Shengsuan Cloud](https://www.shengsuanyun.com/?from=CH_XUQ4OTSK), [APIMart](https://go.apimart.ai/gh-moneyprinterturbo), [Cloudflare AI Gateway](https://dash.cloudflare.com/), [Alibaba ModelScope](https://modelscope.cn/docs/model-service/API-Inference/intro), [AIHubMix](https://aihubmix.com/), [AIML API](https://aimlapi.com/app/keys), [EvoLink](https://evolink.ai/dashboard/keys), [OpenRouter](https://openrouter.ai/settings/keys), [API Route](https://www.api-route.com/), [Fluxion AI](https://fluxionai.space/register?source=github&campaign=moneyprinterturbo&promo=MONEYPRINTERTURBO), [Ollama](https://ollama.com/), [Claude Code subscription](https://code.claude.com/docs), [OneAPI](https://github.com/songquanpeng/one-api), [LiteLLM](https://docs.litellm.ai/docs/providers), [Groq](https://console.groq.com/keys), [Pollinations AI](https://enter.pollinations.ai/), and other compatible gateways or local runtimes
### Video and Image Footage
- [x] Upload your own **local images and videos**, or get HD stock footage from [Pexels (free)](https://www.pexels.com/api/), [Pixabay (free)](https://pixabay.com/api/docs/), and [Coverr](https://coverr.co/developers?ctx=header_navigation)
- [x] Generate `768P` or `2K` source footage with [Metaso MiniMax H3](https://metaso.cn/minimax-h3/?s=MPT), with 4–15 second clips in `9:16`, `16:9`, or `1:1`
- [x] Create multiple AI video clips with [Shengsuan Cloud AI Video](https://www.shengsuanyun.com/?from=CH_XUQ4OTSK), then combine them with the project's voiceover, subtitle, and editing workflow
- [x] Use the native [Volcano Engine Ark Seedance](https://console.volcengine.com/ark/region:ark+cn-beijing/apikey) integration to generate cohesive visuals from individual script segments
- [x] Turn script keywords into original video footage with [WaveSpeed AI](https://wavespeed.ai)
- [x] Access Seedance, Wan, and other text-to-video models through [OFox](https://ofox.ai) with a single API key
- [x] Generate 3–12 second AI video materials through the asynchronous [MuAPI](https://muapi.ai) text-to-video API, with configurable endpoint, aspect ratio, resolution, and polling
- [x] Connect [OpenAI-compatible text-to-image](https://platform.openai.com/docs/guides/image-generation) services or custom image gateways and turn generated images into animated video clips
- [x] Adjust clip duration, frame fitting, and material order to suit different aspect ratios and storytelling styles
### Voiceover, Subtitles, and Background Music
- [x] Choose automatic voiceover, uploaded audio, or no voiceover, with voice samples and full narration previews
- [x] Use **Edge TTS (free, no API key required)**, Azure Speech, SiliconFlow, Google Gemini, Xiaomi MiMo, MiniMax, ElevenLabs, Chatterbox, Kokoro, Fish Audio, ModelBest VoxCPM, and other voice services
- [x] Generate subtitles and configure their font, position, color, size, outline, and background style
- [x] Use random, local, or AI-generated background music with independent volume control
### Output and Publishing
- [x] Export portrait `9:16 (1080×1920)`, landscape `16:9 (1920×1080)`, or square `1:1 (1080×1080)` videos
- [x] Publish completed videos directly to **TikTok, Instagram, and YouTube Shorts**
## Gallery 🎬
All examples below were generated with MoneyPrinterTurbo.
### Portrait 9:16
<table width="100%">
<tr>
<td align="center" width="25%"><a href="https://harry0703.github.io/mpt-assets/?video=03-zh-portrait-city-morning.mp4"><img src="https://github.com/harry0703/mpt-assets/releases/download/assets/03-zh-portrait-city-morning.jpg" width="180" alt="When the City Wakes"></a><br><strong>When the City Wakes</strong><br>Chinese · 14 sec</td>
<td align="center" width="25%"><a href="https://harry0703.github.io/mpt-assets/?video=05-zh-portrait-clean-energy.mp4"><img src="https://github.com/harry0703/mpt-assets/releases/download/assets/05-zh-portrait-clean-energy.jpg" width="180" alt="The Future of Clean Energy"></a><br><strong>The Future of Clean Energy</strong><br>Chinese · 24 sec</td>
<td align="center" width="25%"><a href="https://harry0703.github.io/mpt-assets/?video=07-zh-portrait-space-exploration.mp4"><img src="https://github.com/harry0703/mpt-assets/releases/download/assets/07-zh-portrait-space-exploration.jpg" width="180" alt="Why We Still Explore Space"></a><br><strong>Why We Still Explore Space</strong><br>Chinese · 27 sec</td>
<td align="center" width="25%"><a href="https://harry0703.github.io/mpt-assets/?video=17-zh-portrait-seed-journey.mp4"><img src="https://github.com/harry0703/mpt-assets/releases/download/assets/17-zh-portrait-seed-journey.jpg" width="180" alt="A Seed's Journey"></a><br><strong>A Seed's Journey</strong><br>Chinese · 44 sec</td>
</tr>
<tr>
<td align="center" width="25%"><a href="https://harry0703.github.io/mpt-assets/?video=09-en-portrait-future-robotics.mp4"><img src="https://github.com/harry0703/mpt-assets/releases/download/assets/09-en-portrait-future-robotics.jpg" width="180" alt="The Future of Everyday Robotics"></a><br><strong>The Future of Everyday Robotics</strong><br>English · 21 sec</td>
<td align="center" width="25%"><a href="https://harry0703.github.io/mpt-assets/?video=11-en-portrait-small-habits.mp4"><img src="https://github.com/harry0703/mpt-assets/releases/download/assets/11-en-portrait-small-habits.jpg" width="180" alt="Small Habits, Lasting Change"></a><br><strong>Small Habits, Lasting Change</strong><br>English · 19 sec</td>
<td align="center" width="25%"><a href="https://harry0703.github.io/mpt-assets/?video=13-en-portrait-creative-work.mp4"><img src="https://github.com/harry0703/mpt-assets/releases/download/assets/13-en-portrait-creative-work.jpg" width="180" alt="Making Space for Creative Work"></a><br><strong>Making Space for Creative Work</strong><br>English · 20 sec</td>
<td align="center" width="25%"><a href="https://harry0703.github.io/mpt-assets/?video=15-en-portrait-coffee-science.mp4"><img src="https://github.com/harry0703/mpt-assets/releases/download/assets/15-en-portrait-coffee-science.jpg" width="180" alt="The Science Inside Coffee"></a><br><strong>The Science Inside Coffee</strong><br>English · 23 sec</td>
</tr>
</table>
### Landscape 16:9
<table width="100%">
<tr>
<td align="center" width="25%"><a href="https://harry0703.github.io/mpt-assets/?video=02-zh-landscape-deep-ocean.mp4"><img src="https://github.com/harry0703/mpt-assets/releases/download/assets/02-zh-landscape-deep-ocean.jpg" width="280" alt="Light in the Deep Ocean"></a><br><strong>Light in the Deep Ocean</strong><br>Chinese · 23 sec</td>
<td align="center" width="25%"><a href="https://harry0703.github.io/mpt-assets/?video=04-zh-landscape-reading-power.mp4"><img src="https://github.com/harry0703/mpt-assets/releases/download/assets/04-zh-landscape-reading-power.jpg" width="280" alt="How Reading Shapes Us"></a><br><strong>How Reading Shapes Us</strong><br>Chinese · 23 sec</td>
<td align="center" width="25%"><a href="https://harry0703.github.io/mpt-assets/?video=06-zh-landscape-pour-over-coffee.mp4"><img src="https://github.com/harry0703/mpt-assets/releases/download/assets/06-zh-landscape-pour-over-coffee.jpg" width="280" alt="The Details of Pour-Over Coffee"></a><br><strong>The Details of Pour-Over Coffee</strong><br>Chinese · 23 sec</td>
<td align="center" width="25%"><a href="https://harry0703.github.io/mpt-assets/?video=08-zh-landscape-spring-journey.mp4"><img src="https://github.com/harry0703/mpt-assets/releases/download/assets/08-zh-landscape-spring-journey.jpg" width="280" alt="Spring Is Made for Travel"></a><br><strong>Spring Is Made for Travel</strong><br>Chinese · 14 sec</td>
</tr>
<tr>
<td align="center" width="25%"><a href="https://harry0703.github.io/mpt-assets/?video=10-en-landscape-ocean-conservation.mp4"><img src="https://github.com/harry0703/mpt-assets/releases/download/assets/10-en-landscape-ocean-conservation.jpg" width="280" alt="Why Ocean Conservation Matters"></a><br><strong>Why Ocean Conservation Matters</strong><br>English · 25 sec</td>
<td align="center" width="25%"><a href="https://harry0703.github.io/mpt-assets/?video=14-en-landscape-sustainable-cities.mp4"><img src="https://github.com/harry0703/mpt-assets/releases/download/assets/14-en-landscape-sustainable-cities.jpg" width="280" alt="Designing More Sustainable Cities"></a><br><strong>Designing More Sustainable Cities</strong><br>English · 27 sec</td>
<td align="center" width="25%"><a href="https://harry0703.github.io/mpt-assets/?video=16-en-landscape-mountain-perspective.mp4"><img src="https://github.com/harry0703/mpt-assets/releases/download/assets/16-en-landscape-mountain-perspective.jpg" width="280" alt="What Mountains Teach Us"></a><br><strong>What Mountains Teach Us</strong><br>English · 18 sec</td>
<td align="center" width="25%"><a href="https://harry0703.github.io/mpt-assets/?video=18-en-landscape-history-of-flight.mp4"><img src="https://github.com/harry0703/mpt-assets/releases/download/assets/18-en-landscape-history-of-flight.jpg" width="280" alt="A Brief History of Human Flight"></a><br><strong>A Brief History of Human Flight</strong><br>English · 59 sec</td>
</tr>
</table>
## System Requirements 📦
- Recommended platforms: Windows 10+, macOS 11+, or a mainstream Linux distribution
- Local deployment requires Python 3.11 or later; Python 3.11 is recommended
- A GPU is not required, but it is recommended if you want faster local transcription, faster video processing, or smoother batch generation
| Item | Minimum | Recommended | Optimal |
| ---- | ------------ | ------------ | ---------- |
| CPU | 4 cores | 6 to 8 cores | 8+ cores |
| RAM | 4 GB | 8 GB | 16+ GB |
| GPU | Not required | 4+ GB VRAM | 8+ GB VRAM |
- If you mainly rely on cloud LLMs, cloud TTS, and online material sources, CPU and RAM matter more than GPU
- If you use `faster-whisper`, batch generation, or heavier local processing, a GPU will improve throughput noticeably
## Quick Start 🚀
### Recommended Paths
- If you do not want to install or configure the project manually: generate videos with an AI Agent
- Windows users: use the one-click package first for the fastest local trial
- macOS / Linux users: use `uv` for the primary local setup path
- If you want a more isolated runtime: use Docker deployment
### Generate Videos with an AI Agent
If your AI Agent can read Skill documents and operate a local terminal, send it the prompt below. The Agent will install and configure MoneyPrinterTurbo, generate the video, and return the video file path. It will ask only for required API keys that are not already configured. This workflow currently supports macOS and Windows.
```text
Use this Skill: https://raw.githubusercontent.com/harry0703/MoneyPrinterTurbo/main/docs/skill/SKILL.md
Create a video with the topic "How AI is changing everyday life."
```
### Run in Google Colab
Want to try MoneyPrinterTurbo without setting up a local environment? Run it directly in Google Colab!
[](https://colab.research.google.com/github/harry0703/MoneyPrinterTurbo/blob/main/docs/MoneyPrinterTurbo.ipynb)
### Windows
Download the latest Windows one-click package from GitHub Releases, then extract it directly.
- [Download the latest Windows one-click package](https://github.com/harry0703/MoneyPrinterTurbo/releases/latest)
> Download the `.7z` archive from the **Assets** section of that page. The
> auto-generated `Source code (zip)` / `Source code (tar.gz)` archives contain
> source code only: after extracting them you get `webui.bat` but no `start.bat`
> or `update.bat`.
After downloading, it is recommended to **double-click** `update.bat` first to update to the **latest code**, then double-click `start.bat` to launch
After launching, the browser will open automatically (if it opens blank, it is recommended to use **Chrome** or **Edge**)
### macOS / Linux
Use the local setup or Docker instructions below.
## Installation & Deployment 📥
### Prerequisites
- Local deployment requires Python 3.11 or later
- On Windows, avoid project paths containing non-ASCII characters, special characters, or spaces
#### ① Clone the Project
```shell
git clone https://github.com/harry0703/MoneyPrinterTurbo.git
```
#### ② Complete the Initial Setup
On first launch, the project creates `config.toml` from `config.example.toml`, so you do not need to create the file manually. Before using cloud LLMs, online footage, or AI video services, add the corresponding API keys in the WebUI basic settings.
### Docker Deployment 🐳
#### ① Launch the Docker Container
If Docker is not installed, [download and install Docker Desktop](https://www.docker.com/products/docker-desktop/) first.
If you are using a Windows system, please refer to Microsoft's documentation:
1. [Install WSL](https://learn.microsoft.com/en-us/windows/wsl/install)
2. [Use Docker containers with WSL](https://learn.microsoft.com/en-us/windows/wsl/tutorials/wsl-containers)
```shell
cd MoneyPrinterTurbo
docker compose -f docker-compose.release.yml up
```
> The recommended default is `docker-compose.release.yml`, which pulls the prebuilt image from GitHub Container Registry: `ghcr.io/harry0703/moneyprinterturbo:latest`.
> If you need to build the image locally, you can still run `docker compose up`.
> Before the first start, copy `config.example.toml` to `config.toml` so it can be mounted into the containers.
#### ② Access the WebUI
Open your browser and visit http://127.0.0.1:8501
#### ③ Access the API Documentation
Open your browser and visit http://127.0.0.1:8080/docs or http://127.0.0.1:8080/redoc
> The API allows same-origin browser access by default. Set the `CORS_ALLOWED_ORIGINS` environment variable only when a separate browser frontend must call the API directly from another origin, for example `http://localhost:3000,https://frontend.example.com`. CORS does not affect curl, Postman, n8n, or other server-side clients.
### Manual Deployment 📦
#### ① Create a Python Virtual Environment
Use [uv](https://docs.astral.sh/uv/) to manage the Python environment and dependencies. The project supports Python 3.11 or later; the example below uses Python 3.11.
```shell
git clone https://github.com/harry0703/MoneyPrinterTurbo.git
cd MoneyPrinterTurbo
uv python install 3.11
uv sync --frozen
```
If you are not using `uv` yet, you can still use `venv + pip`.
```shell
python3.11 -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt
```
Notes:
- `pyproject.toml` is now the primary dependency manifest.
- `uv.lock` pins the resolved environment, so `uv sync --frozen` is recommended by default.
- `requirements.txt` is kept only for legacy `pip`-based installation.
#### ② Launch the WebUI 🌐
Note that you need to execute the following commands in the `root directory` of the MoneyPrinterTurbo project
###### Windows
```powershell
.\webui.bat
```
You can also run `webui.bat` in CMD.
`webui.bat` prefers the project `.venv` or bundled Python from the portable package. If no project Python is found but `uv` is installed, it automatically falls back to `uv run streamlit`.
To allow other devices on your LAN to access the WebUI, run `set MPT_WEBUI_HOST=0.0.0.0` before running `webui.bat`.
###### macOS or Linux
```shell
sh webui.sh
```
The script automatically uses the project virtual environment or `uv` and selects an available local port. To allow access from other devices on your LAN, run:
```shell
MPT_WEBUI_HOST=0.0.0.0 sh webui.sh
```
After launching, the browser will open automatically
#### ③ Launch the API Service 🚀
```shell
uv run python main.py
```
If you have already activated the virtual environment manually, you can still run:
```shell
python main.py
```
#### ④ Pure CLI Mode (No Browser) ⌨️
If you cannot use a browser or port forwarding, generate videos directly from the
command line. The simplest complete generation command is:
```shell
uv run python cli.py --video-subject "How AI is changing everyday life"
```
Subtitle style and voiceover options resolve in this order: **explicit CLI option >
saved `[ui]` value in `config.toml` > built-in default**. Other generation settings,
such as background music, video count, and paragraph count, are not inherited from
the WebUI. If the WebUI is set to use uploaded audio, pass `--custom-audio-file`
explicitly, since the uploaded path is not persisted.
For the complete command reference, parameter descriptions, and usage instructions,
run:
```shell
uv run python cli.py --help
```
To run several tasks sequentially, pass a UTF-8 JSON array or JSONL manifest. CLI
options act as defaults, and each object overrides fields from `VideoParams`:
```json
[
{ "video_subject": "How solar panels work" },
{ "video_subject": "How wind turbines work", "video_aspect": "16:9" }
]
```
```shell
uv run python cli.py --batch-file ./tasks.json --stop-at video
```
The manifest is resolved from the current working directory. Relative
`custom_audio_file` and local `video_materials[].url` values inside it are resolved
from the manifest's directory; file paths supplied as CLI defaults keep their normal
current-working-directory semantics. A manifest is limited to 100 tasks and 1 MiB.
All entries are validated before the first task starts, tasks continue after an
individual runtime failure, and the command prints one JSON summary when finished.
The summary contains `total`, `succeeded`, `failed`, and `tasks`; each task entry has
`index`, `task_id`, `status`, `result`, `failed_stage`, and `error`.
## Voiceover, Subtitles, and Background Music 🎙️
### Voice Synthesis
ElevenLabs speech speed depends on the selected model. The default `eleven_multilingual_v2` accepts 0.7×–1.2×; select `eleven_v3` for the wider 0.25×–4.0× REST range. MoneyPrinterTurbo reports unsupported selections before sending a generation request and preserves the selected rate without clamping. Non-default speeds on `eleven_v4`/`eleven_v4_turbo` are unsupported. Other model IDs use a conservative 0.7×–1.2× validation range; this does not guarantee that every model supports the speed setting. If the provider returns `invalid_voice_settings`, generation stops without retrying and shows its explanation. The [live v2/v3 validation reported by the maintainer](https://github.com/harry0703/MoneyPrinterTurbo/pull/1578#issuecomment-6040107858) takes precedence over the broader range described in the [official voice-settings reference](https://github.com/elevenlabs/skills/blob/main/text-to-speech/references/voice-settings.md).
**Azure TTS V1** in the WebUI is powered by **Edge TTS** and is free to use without an API key. MoneyPrinterTurbo also supports **Azure TTS V2**, **SiliconFlow TTS**, **Google Gemini TTS**, **Xiaomi MiMo TTS**, **MiniMax TTS**, **ElevenLabs TTS**, self-hosted **Chatterbox TTS**, self-hosted **Kokoro TTS**, **Fish Audio TTS**, [ModelBest VoxCPM TTS](https://platform.modelbest.cn/console/docs/api/audio), and a no-voice mode.
Select a provider and voice in the WebUI, then follow the on-screen instructions for any required credentials. Edge TTS does not require an API key; [Azure TTS V2](https://portal.azure.com/) and other cloud providers require credentials from their respective platforms. See the available Edge TTS voices in the [voice list](./docs/voice-list.txt).
ModelBest VoxCPM requires an API key and a model ID with the `speech_synthesis` capability. Its SSE response streams WAV audio, which MoneyPrinterTurbo automatically converts to the MP3 used by the video pipeline. In addition to standard text-to-speech, the WebUI can clone speaker identity from optional reference audio. High-fidelity delivery reuses that clip with its exact transcript by default, or accepts a separate performance example, and sends `prompt_audio` plus `prompt_text` to continue its pacing, emotion, and pronunciation. The WebUI can use the local Whisper configuration to create an editable transcript draft. Uploads are limited to 20 MiB and each converted WAV to 5 MiB. Reference material is scoped to the current browser session and task and is not written to configuration, presets, task history, or logs.
### Subtitle Generation
Two subtitle generation modes are available:
- **edge**: Uses TTS timestamps, runs quickly without a GPU, and is the default mode.
- **whisper**: Uses local `faster-whisper` transcription when a more accurate subtitle timeline is needed. The model is downloaded on first use.
Set `subtitle_provider` in `config.toml` to switch modes. Whisper uses the approximately 3 GB `large-v3` model by default. To use the smaller and faster, approximately 1.6 GB `large-v3-turbo` model:
```toml
[app]
subtitle_provider = "whisper"
[whisper]
model_size = "large-v3-turbo"
```
> On first use, Whisper automatically downloads the model from Hugging Face. If the automatic download fails, download `whisper-large-v3` manually from [Hugging Face](https://huggingface.co/Systran/faster-whisper-large-v3).
After extracting the model, place the entire directory in `.\MoneyPrinterTurbo\models`. The final path should be `.\MoneyPrinterTurbo\models\whisper-large-v3`:
```
MoneyPrinterTurbo
├─models
│ └─whisper-large-v3
│ config.json
│ model.bin
│ preprocessor_config.json
│ tokenizer.json
│ vocabulary.json
```
### Background Music
Background music for videos is located in the project's `resource/songs` directory.
> The current project includes some default music from YouTube videos. If there are copyright issues, please delete
> them.
### Subtitle Fonts
Fonts for rendering video subtitles are located in the project's `resource/fonts` directory, and you can also add your
own fonts.
## Common Questions 🤔
<details>
<summary>RuntimeError: No ffmpeg exe could be found</summary>
Normally, ffmpeg will be automatically downloaded and detected.
However, if your environment has issues preventing automatic downloads, you may encounter the following error:
```
RuntimeError: No ffmpeg exe could be found.
Install ffmpeg on your system, or set the IMAGEIO_FFMPEG_EXE environment variable.
```
In this case, download FFmpeg from [gyan.dev](https://www.gyan.dev/ffmpeg/builds/), extract it, and set `ffmpeg_path` to
the actual installation path.
```toml
[app]
# Please set according to your actual path, note that Windows path separators are \\
ffmpeg_path = "C:\\Users\\harry\\Downloads\\ffmpeg.exe"
```
</details>
<details>
<summary>OSError: [Errno 24] Too many open files</summary>
This issue is caused by the system's limit on the number of open files. You can solve it by modifying the system's file open limit.
Check the current limit:
```shell
ulimit -n
```
If it's too low, you can increase it, for example:
```shell
ulimit -n 10240
```
</details>
<details>
<summary>Whisper model download failed</summary>
```
LocalEntryNotFoundError: Cannot find an appropriate cached snapshot folder for the specified revision on the local disk and
outgoing traffic has been disabled.
To enable repo look-ups and downloads online, pass 'local_files_only=False' as input.
```
or
```
An error occurred while synchronizing the model Systran/faster-whisper-large-v3 from the Hugging Face Hub:
An error happened while trying to locate the files on the Hub and we cannot find the appropriate snapshot folder for the
specified revision on the local disk. Please check your internet connection and try again.
Trying to load the model directly from the local cache, if it exists.
```
Solution: [See how to download the model manually from Hugging Face](#subtitle-generation)
</details>
## Feedback & Suggestions 📢
- You can submit an [issue](https://github.com/harry0703/MoneyPrinterTurbo/issues) or a [pull request](https://github.com/harry0703/MoneyPrinterTurbo/pulls).
## License 📝
Click to view the [`LICENSE`](LICENSE) file
### Editable local video projects
Use [revision-aware local projects](docs/video-projects.md) to replace prepared scene media, compare revisions and rebuild only affected stages. The opt-in CLI preserves verified artifacts and the last successful export; it makes no provider calls or automatic posts.
## /README-ja.md
<div align="center">
# MoneyPrinterTurbo 💸
### オールインワン AI ショート動画生成ツール
動画の<b>テーマ</b>または<b>キーワード</b>を指定するだけで、MoneyPrinterTurbo が台本の生成、素材のマッチング、字幕と BGM の作成を行い、高画質のショート動画を出力します。
[](https://github.com/harry0703/MoneyPrinterTurbo/releases)
[](https://github.com/harry0703/MoneyPrinterTurbo/releases/latest)
[](https://www.python.org/)
[](https://github.com/harry0703/MoneyPrinterTurbo/releases/latest)
<a href="https://trendshift.io/repositories/8731" target="_blank"><img src="https://trendshift.io/api/badge/repositories/8731" alt="harry0703%2FMoneyPrinterTurbo | Trendshift" style="width: 250px; height: 55px;" width="250" height="55"/></a>
<a href="https://www.star-history.com/harry0703/moneyprinterturbo"><img src="https://api.star-history.com/badge?repo=harry0703/MoneyPrinterTurbo" alt="Star History Rank" style="height: 55px;" height="55"/></a>
日本語 | [English](README-en.md) | [简体中文](README.md) | [リリース](https://github.com/harry0703/MoneyPrinterTurbo/releases) | [Issues](https://github.com/harry0703/MoneyPrinterTurbo/issues)
</div>
## スクリーンショット 🖥️
<h4 align="center">WebUI</h4>
<h4 align="center">API</h4>
---
## スペシャルサンクス ❤️
<div align="center">
<a href="https://platform.kimi.ai?track_id=track-9e3b711aa2594e378f6fe5b8de718a76&aff=moneyprinterturbo" target="_blank"><img src="https://gcdn.moonshot.cn/growth-cdn/sponsor/kimi-en.png" alt="Kimi sponsors MoneyPrinterTurbo" width="100%"></a>
</div>
本プロジェクトをスポンサードしてくださっている [Kimi](https://platform.kimi.ai?track_id=track-9e3b711aa2594e378f6fe5b8de718a76&aff=moneyprinterturbo) に感謝します！ [Kimi K3](https://www.kimi.com/blog/kimi-k3?aff=moneyprinterturbo) は Moonshot AI の最も高性能なモデルであり、世界初の 3T クラスのオープンモデルです。ネイティブの視覚能力と 100 万トークンのコンテキストウィンドウを備えた K3 は、ナレッジワーク、推論、長期にわたるタスクにおいて最先端の性能を発揮します。MoneyPrinterTurbo では、K3 が台本の作成と、最終的な映像素材を左右する検索キーワードの抽出を担い、動画制作を支えています。内容をより深く理解するほど、より適切な素材が得られます。
**MoneyPrinterTurbo ユーザー限定特典: 専用リンクから新規登録すると、初回のチャージ成功額の 10% 相当（上限 1,000 元）の API クレジットがボーナスとして付与されます。特典の終了は 2026 年 12 月 31 日です。Kimi オープンプラットフォーム（[中国語サイト](https://platform.kimi.com?track_id=track-6eec1e56a4494e52adcaebbcbbefce59&aff=moneyprinterturbo) | [グローバル](https://platform.kimi.ai?track_id=track-9e3b711aa2594e378f6fe5b8de718a76&aff=moneyprinterturbo)）から API をお試しください。**
<br>
<table align="center">
<tr>
<td align="center" width="120">
<a href="https://www.byteplus.com/en/product/modelark?utm_campaign=hw&utm_content=MoneyPrinterTurbo&utm_medium=devrel_tool_web&utm_source=OWO&utm_term=MoneyPrinterTurbo"><img src="docs/sponsors/byteplus-logo.svg" alt="BytePlus" height="25"></a><br>
<a href="https://www.byteplus.com/en/product/modelark?utm_campaign=hw&utm_content=MoneyPrinterTurbo&utm_medium=devrel_tool_web&utm_source=OWO&utm_term=MoneyPrinterTurbo"><strong>BytePlus ModelArk</strong></a>
</td>
<td align="left">
本プロジェクトをスポンサードしてくださっている Dola Seed に感謝します！ <a href="https://www.byteplus.com/en/product/modelark?utm_campaign=hw&utm_content=MoneyPrinterTurbo&utm_medium=devrel_tool_web&utm_source=OWO&utm_term=MoneyPrinterTurbo">Dola Seed 2.0</a> は、ByteDance がグローバル市場向けに独自開発した全モーダル対応の汎用大規模モデルです。統合マルチモーダルアーキテクチャを基盤とし、テキスト・画像・音声・動画の統合的な理解と生成をサポートします。エージェント連携をネイティブに実現し、推論、長時間タスクの遂行、ツール連携、コーディングにおいて優れた能力を発揮します。こちらのリンクから登録すると、モデルごとに 50 万トークンの無料推論枠が付与されます。
</td>
</tr>
<tr>
<td align="center" width="120">
<a href="https://go.apimart.ai/gh-moneyprinterturbo"><img src="docs/sponsors/apimart-logo.png" alt="APIMart" width="100"></a>
</td>
<td align="left">
本プロジェクトをスポンサードしてくださっている <a href="https://go.apimart.ai/gh-moneyprinterturbo">APIMart</a> に感謝します！ APIMart は、AI 画像・動画生成に特化した低価格 API プラットフォームです。<strong>GPT-Image-2 は 1 枚あたり 0.006 米ドルから利用でき、1 米ドルで 160 枚以上の画像を生成できます</strong>。<strong>画像と動画を 1 つの非同期 API で扱い、コードを変更せずにモデルを切り替えられます</strong>。タスクを送信して ID を受け取った後、ポーリングまたはコールバックで結果を取得でき、数万枚規模のバッチ処理にも対応します。月額料金のない従量課金制で、<a href="https://go.apimart.ai/gh-moneyprinterturbo">こちらの登録リンク</a>からすぐに利用を開始できます。
</td>
</tr>
<tr>
<td align="center" width="120">
<a href="https://metaso.cn/minimax-h3/?s=MPT"><img src="docs/sponsors/metaso-logo.png" alt="Metaso" width="100"></a><br>
<a href="https://metaso.cn/minimax-h3/?s=MPT"><strong>Metaso</strong></a>
</td>
<td align="left">
<strong>Metaso の MiniMax H3 動画生成 API</strong><br>
Metaso は、コストパフォーマンスに優れた MiniMax H3 動画生成サービスを提供しています。料金は <strong>768P が 1 秒あたり 0.09 元、2K が 1 秒あたり 0.15 元</strong>です。ネイティブ 2K、映像と音声の同期、OpenAI 互換 API、ComfyUI に対応しており、GPU を自分で用意・運用する必要はありません。<br>
🎁 <a href="https://metaso.cn/minimax-h3/?s=MPT">MoneyPrinterTurbo 専用リンク</a>から登録すると、ボーナスクレジットや限定特典を受け取れます。
</td>
</tr>
<tr>
<td align="center" width="120">
<a href="https://infistar.cc/register?aff=6T4EYXP2&ref_source=link"><img src="docs/sponsors/infistar-logo.svg" alt="Infistar.cc" height="56"></a><br>
<a href="https://infistar.cc/register?aff=6T4EYXP2&ref_source=link"><strong>Infistar.cc</strong></a>
</td>
<td align="left">
本オープンソースプロジェクトを支援してくださっている <a href="https://infistar.cc/register?aff=6T4EYXP2&ref_source=link"><strong>Infistar.cc</strong></a> に感謝します！コストパフォーマンスに優れた AI API サービスで、大規模言語モデルは<strong>公式料金の 1% から</strong>、AI 画像生成は<strong>1 枚あたり 0.06 元から</strong>利用できます。<br>
<strong>API キー 1 つでテキスト・画像・動画を生成</strong>。GPT、Claude、Gemini、DeepSeek、Qwen、Kling などの主要モデルに対応し、API グループを分ける必要はありません。<br>
<strong>全モデルの真正性を検証可能</strong>。料金は透明で、中国での ICP 届出を行っており、人民元でのリアルタイム課金と<strong>法人向けの請求書発行</strong>に対応しています。<br>
🎁 MPT 限定特典：<a href="https://infistar.cc/register?aff=6T4EYXP2&ref_source=link">専用リンクから登録</a>すると、<strong>5 米ドル分のトライアルクレジット</strong>と初回チャージの優待を受けられます。
</td>
</tr>
<tr>
<td align="center" width="120">
<a href="https://www.shengsuanyun.com/?from=CH_XUQ4OTSK"><img src="docs/sponsors/shengsuanyun-logo.jpg" alt="Shengsuan Cloud" height="56"></a><br>
<a href="https://www.shengsuanyun.com/?from=CH_XUQ4OTSK"><strong>Shengsuan Cloud</strong></a>
</td>
<td align="left">
本プロジェクトをスポンサードしてくださっている <a href="https://www.shengsuanyun.com/?from=CH_XUQ4OTSK">Shengsuan Cloud</a> に感謝します！ Shengsuan Cloud は AI ネイティブなチーム向けの API 集約プラットフォームで、Claude、ChatGPT、Gemini などの主要な言語モデルやマルチモーダルモデルを、統一 API と従量課金で利用できます。<br>
コンプライアンスに配慮した API サービスに加え、チームのコスト・権限管理、インテリジェントルーティング、セキュリティ対策、BYOK による認証情報管理、請求書発行に対応するエンタープライズゲートウェイも提供しています。<br>
🎁 <a href="https://www.shengsuanyun.com/?from=CH_XUQ4OTSK">こちらのリンク</a>から新規登録すると、10 元分のトライアルクレジットを受け取れます。
</td>
</tr>
<tr>
<td align="center" width="120">
<a href="https://sidrune.ai/register?source=github&campaign=moneyprinterturbo&promo=moneyprinterturbo"><img src="docs/sponsors/sidrune-logo.png" alt="Sidrune AI" width="120"></a>
</td>
<td align="left">
本プロジェクトを支援してくださっている <a href="https://sidrune.ai/register?source=github&campaign=moneyprinterturbo&promo=moneyprinterturbo">Sidrune AI</a> に感謝します！ <strong>世界の主要 AI モデルへのアクセスと管理を、ひとつの窓口で。</strong>個人開発者、技術チーム、企業向けに、統一 API を通じて主要モデルを利用・管理できるプラットフォームです。複数の接続経路を動的に切り替えることで可用性を高め、モデルの性能、応答時間、利用料金を透明に確認できます。モデルや接続経路によって、<strong>API 利用コストを公式料金や基準価格より 40%～98% 削減できます</strong>。<a href="https://sidrune.ai/register?source=github&campaign=moneyprinterturbo&promo=moneyprinterturbo">専用リンクから登録</a>すると、<strong>3 米ドル分の API クレジット</strong>を受け取れます。
</td>
</tr>
<tr>
<td align="center" width="120">
<a href="https://reccloud.com"><img src="docs/sponsors/reccloud-logo.svg" alt="RecCloud" height="36"></a><br>
<a href="https://reccloud.com">RecCloud</a>
</td>
<td align="left">
本プロジェクトをベースにした無料の <strong>AI 動画生成ツール</strong>を提供してくださっている <a href="https://reccloud.com">RecCloud（AI マルチメディアプラットフォーム）</a> に感謝します。デプロイ不要でオンラインですぐに使えるため、初心者にも手軽にご利用いただけます。
</td>
</tr>
<tr>
<td align="center" width="120">
<a href="https://picwish.com"><img src="docs/sponsors/picwish-logo.svg" alt="Picwish" height="36"></a><br>
<a href="https://picwish.com">Picwish</a>
</td>
<td align="left">
本プロジェクトを支援・スポンサードしてくださっている <a href="https://picwish.com">Picwish</a> に感謝します！Picwish は豊富な<strong>オンライン画像編集ツール</strong>を提供し、さまざまな画像の加工を簡単な操作で行えます。
</td>
</tr>
</table>
## 作者によるもう 1 つのオープンソースプロジェクト: MangoDisk ⭐
<p align="center">
<a href="https://mangodisk.app/ja">
<picture>
<source media="(prefers-color-scheme: dark)" srcset="https://assets.mangodisk.app/images/screenshots/ja/dark-01-deep-cleanup.jpg?ver=1.1.0">
<source media="(prefers-color-scheme: light)" srcset="https://assets.mangodisk.app/images/screenshots/ja/light-01-deep-cleanup.jpg?ver=1.1.0">
<img src="https://assets.mangodisk.app/images/screenshots/ja/light-01-deep-cleanup.jpg?ver=1.1.0" width="900" alt="MangoDisk ディープクリーン画面">
</picture>
</a>
</p>
<p align="center">
<strong>macOS・Windows 向けのオープンソースのディスククリーンアップ、容量分析、システム最適化ツール</strong><br>
キャッシュ、大容量ファイル、重複ファイル、アプリの残骸を整理し、容量分析、アプリのアンインストール、スタートアップ管理、システム最適化・メンテナンスにも対応します。
</p>
<p align="center">
<a href="https://mangodisk.app/ja">MangoDisk 公式サイト</a> · <a href="https://github.com/harry0703/MangoDisk">GitHub で見る</a>
</p>
---
## 機能 🎯
### 制作フロー
- [x] **AI エージェント、WebUI、API、CLI** の 4 つの利用方法を提供し、手軽な動画制作から自動化ワークフローへの組み込みまで対応
- [x] テーマから台本、ナレーション、素材、字幕、BGM、編集までを自動化し、各工程で独自コンテンツを使用することも可能
- [x] **複数動画の一括生成**、タスク履歴、生成設定と API キーのインポート・エクスポート・復元に対応
### 台本とモデルプロバイダー
- [x] AI による**多言語動画台本**の生成・リライトに加え、独自台本の利用にも対応
- [x] [Kimi / Moonshot AI](https://platform.kimi.ai?track_id=track-9e3b711aa2594e378f6fe5b8de718a76&aff=moneyprinterturbo)、[OpenAI](https://platform.openai.com/api-keys)、[Anthropic Claude](https://platform.claude.com/settings/keys)、[Google Gemini](https://aistudio.google.com/app/apikey)、[DeepSeek](https://platform.deepseek.com/api_keys)、[Alibaba Cloud Qwen](https://qwen.ai/apiplatform)、[Microsoft Azure OpenAI](https://portal.azure.com/#view/Microsoft_Azure_ProjectOxford/CognitiveServicesHub/~/OpenAI)、[ByteDance VolcEngine Ark](https://console.volcengine.com/ark)、[xAI Grok](https://console.x.ai/)、[MiniMax](https://platform.minimax.io/)、[Xiaomi MiMo](https://platform.xiaomimimo.com/docs/zh-CN/quick-start/first-api-call) などの主要なモデルサービスに対応
- [x] [Shengsuan Cloud](https://www.shengsuanyun.com/?from=CH_XUQ4OTSK)、[APIMart](https://go.apimart.ai/gh-moneyprinterturbo)、[Cloudflare AI Gateway](https://dash.cloudflare.com/)、[Alibaba ModelScope](https://modelscope.cn/docs/model-service/API-Inference/intro)、[AIHubMix](https://aihubmix.com/)、[AIML API](https://aimlapi.com/app/keys)、[EvoLink](https://evolink.ai/dashboard/keys)、[OpenRouter](https://openrouter.ai/settings/keys)、[API Route](https://www.api-route.com/)、[Fluxion AI](https://fluxionai.space/register?source=github&campaign=moneyprinterturbo&promo=MONEYPRINTERTURBO)、[Ollama](https://ollama.com/)、[Claude Code サブスクリプション](https://code.claude.com/docs)、[OneAPI](https://github.com/songquanpeng/one-api)、[LiteLLM](https://docs.litellm.ai/docs/providers)、[Groq](https://console.groq.com/keys)、[Pollinations AI](https://enter.pollinations.ai/) などの統合ゲートウェイ、集約サービス、ローカルランタイムに対応
### 動画・画像素材
- [x] 手持ちの**ローカル画像・動画**をアップロードできるほか、[Pexels（無料）](https://www.pexels.com/api/)、[Pixabay（無料）](https://pixabay.com/api/docs/)、[Coverr](https://coverr.co/developers?ctx=header_navigation) の高画質ストック素材を利用可能
- [x] [Metaso MiniMax H3](https://metaso.cn/minimax-h3/?s=MPT) によるテキスト動画生成に対応し、`768P`／`2K`、4～15 秒の素材を `9:16`、`16:9`、`1:1` の 3 種類のアスペクト比で生成可能
- [x] [Shengsuan Cloud AI Video](https://www.shengsuanyun.com/?from=CH_XUQ4OTSK) で複数の AI 動画素材を生成し、プロジェクトのナレーション、字幕、編集フローで 1 本の動画に合成可能
- [x] [Volcano Engine Ark Seedance](https://console.volcengine.com/ark/region:ark+cn-beijing/apikey) をネイティブに統合し、台本の各セグメントから一貫した動画素材を生成可能
- [x] [WaveSpeed AI](https://wavespeed.ai) によるテキスト動画生成で、台本のキーワードからオリジナル素材を作成可能
- [x] [OFox](https://ofox.ai) のマルチモデルテキスト動画生成に対応し、1 つの API キーで Seedance、Wan などのモデルを利用可能
- [x] 非同期の [MuAPI](https://muapi.ai) テキスト動画 API で 3～12 秒の AI 素材を生成し、Endpoint、アスペクト比、解像度、ポーリングを設定可能
- [x] [OpenAI 互換の画像生成](https://platform.openai.com/docs/guides/image-generation)サービスや独自の画像ゲートウェイを接続し、生成画像を動きのある動画クリップに変換可能
- [x] クリップの長さ、フレームへの合わせ方、素材の使用順を調整し、アスペクト比や構成に合わせた編集が可能
### ナレーション・字幕・BGM
- [x] 自動ナレーション、音声ファイルのアップロード、音声なしの 3 つの方式に対応し、音声サンプルとナレーション全体をプレビュー可能
- [x] **Edge TTS（無料・API キー不要）**、Azure Speech、SiliconFlow、Google Gemini、Xiaomi MiMo、MiniMax、ElevenLabs、Chatterbox、Kokoro、Fish Audio、ModelBest VoxCPM などの音声サービスに対応
- [x] 字幕を自動生成し、フォント、位置、色、サイズ、縁取り、背景スタイルを調整可能
- [x] ランダム、ローカル、AI 生成の BGM に対応し、音量を個別に調整可能
### 出力と公開
- [x] 縦型 `9:16（1080×1920）`、横型 `16:9（1920×1080）`、正方形 `1:1（1080×1080）` に対応
- [x] 動画生成後、**TikTok、Instagram、YouTube Shorts** へワンクリックで自動公開可能
## ギャラリー 🎬
以下の例はすべて MoneyPrinterTurbo で生成されたものです。
### 縦型 9:16
<table width="100%">
<tr>
<td align="center" width="25%"><a href="https://harry0703.github.io/mpt-assets/?video=03-zh-portrait-city-morning.mp4"><img src="https://github.com/harry0703/mpt-assets/releases/download/assets/03-zh-portrait-city-morning.jpg" width="180" alt="When the City Wakes"></a><br><strong>When the City Wakes</strong><br>中国語 · 14 秒</td>
<td align="center" width="25%"><a href="https://harry0703.github.io/mpt-assets/?video=05-zh-portrait-clean-energy.mp4"><img src="https://github.com/harry0703/mpt-assets/releases/download/assets/05-zh-portrait-clean-energy.jpg" width="180" alt="The Future of Clean Energy"></a><br><strong>The Future of Clean Energy</strong><br>中国語 · 24 秒</td>
<td align="center" width="25%"><a href="https://harry0703.github.io/mpt-assets/?video=07-zh-portrait-space-exploration.mp4"><img src="https://github.com/harry0703/mpt-assets/releases/download/assets/07-zh-portrait-space-exploration.jpg" width="180" alt="Why We Still Explore Space"></a><br><strong>Why We Still Explore Space</strong><br>中国語 · 27 秒</td>
<td align="center" width="25%"><a href="https://harry0703.github.io/mpt-assets/?video=17-zh-portrait-seed-journey.mp4"><img src="https://github.com/harry0703/mpt-assets/releases/download/assets/17-zh-portrait-seed-journey.jpg" width="180" alt="A Seed's Journey"></a><br><strong>A Seed's Journey</strong><br>中国語 · 44 秒</td>
</tr>
<tr>
<td align="center" width="25%"><a href="https://harry0703.github.io/mpt-assets/?video=09-en-portrait-future-robotics.mp4"><img src="https://github.com/harry0703/mpt-assets/releases/download/assets/09-en-portrait-future-robotics.jpg" width="180" alt="The Future of Everyday Robotics"></a><br><strong>The Future of Everyday Robotics</strong><br>英語 · 21 秒</td>
<td align="center" width="25%"><a href="https://harry0703.github.io/mpt-assets/?video=11-en-portrait-small-habits.mp4"><img src="https://github.com/harry0703/mpt-assets/releases/download/assets/11-en-portrait-small-habits.jpg" width="180" alt="Small Habits, Lasting Change"></a><br><strong>Small Habits, Lasting Change</strong><br>英語 · 19 秒</td>
<td align="center" width="25%"><a href="https://harry0703.github.io/mpt-assets/?video=13-en-portrait-creative-work.mp4"><img src="https://github.com/harry0703/mpt-assets/releases/download/assets/13-en-portrait-creative-work.jpg" width="180" alt="Making Space for Creative Work"></a><br><strong>Making Space for Creative Work</strong><br>英語 · 20 秒</td>
<td align="center" width="25%"><a href="https://harry0703.github.io/mpt-assets/?video=15-en-portrait-coffee-science.mp4"><img src="https://github.com/harry0703/mpt-assets/releases/download/assets/15-en-portrait-coffee-science.jpg" width="180" alt="The Science Inside Coffee"></a><br><strong>The Science Inside Coffee</strong><br>英語 · 23 秒</td>
</tr>
</table>
### 横型 16:9
<table width="100%">
<tr>
<td align="center" width="25%"><a href="https://harry0703.github.io/mpt-assets/?video=02-zh-landscape-deep-ocean.mp4"><img src="https://github.com/harry0703/mpt-assets/releases/download/assets/02-zh-landscape-deep-ocean.jpg" width="280" alt="Light in the Deep Ocean"></a><br><strong>Light in the Deep Ocean</strong><br>中国語 · 23 秒</td>
<td align="center" width="25%"><a href="https://harry0703.github.io/mpt-assets/?video=04-zh-landscape-reading-power.mp4"><img src="https://github.com/harry0703/mpt-assets/releases/download/assets/04-zh-landscape-reading-power.jpg" width="280" alt="How Reading Shapes Us"></a><br><strong>How Reading Shapes Us</strong><br>中国語 · 23 秒</td>
<td align="center" width="25%"><a href="https://harry0703.github.io/mpt-assets/?video=06-zh-landscape-pour-over-coffee.mp4"><img src="https://github.com/harry0703/mpt-assets/releases/download/assets/06-zh-landscape-pour-over-coffee.jpg" width="280" alt="The Details of Pour-Over Coffee"></a><br><strong>The Details of Pour-Over Coffee</strong><br>中国語 · 23 秒</td>
<td align="center" width="25%"><a href="https://harry0703.github.io/mpt-assets/?video=08-zh-landscape-spring-journey.mp4"><img src="https://github.com/harry0703/mpt-assets/releases/download/assets/08-zh-landscape-spring-journey.jpg" width="280" alt="Spring Is Made for Travel"></a><br><strong>Spring Is Made for Travel</strong><br>中国語 · 14 秒</td>
</tr>
<tr>
<td align="center" width="25%"><a href="https://harry0703.github.io/mpt-assets/?video=10-en-landscape-ocean-conservation.mp4"><img src="https://github.com/harry0703/mpt-assets/releases/download/assets/10-en-landscape-ocean-conservation.jpg" width="280" alt="Why Ocean Conservation Matters"></a><br><strong>Why Ocean Conservation Matters</strong><br>英語 · 25 秒</td>
<td align="center" width="25%"><a href="https://harry0703.github.io/mpt-assets/?video=14-en-landscape-sustainable-cities.mp4"><img src="https://github.com/harry0703/mpt-assets/releases/download/assets/14-en-landscape-sustainable-cities.jpg" width="280" alt="Designing More Sustainable Cities"></a><br><strong>Designing More Sustainable Cities</strong><br>英語 · 27 秒</td>
<td align="center" width="25%"><a href="https://harry0703.github.io/mpt-assets/?video=16-en-landscape-mountain-perspective.mp4"><img src="https://github.com/harry0703/mpt-assets/releases/download/assets/16-en-landscape-mountain-perspective.jpg" width="280" alt="What Mountains Teach Us"></a><br><strong>What Mountains Teach Us</strong><br>英語 · 18 秒</td>
<td align="center" width="25%"><a href="https://harry0703.github.io/mpt-assets/?video=18-en-landscape-history-of-flight.mp4"><img src="https://github.com/harry0703/mpt-assets/releases/download/assets/18-en-landscape-history-of-flight.jpg" width="280" alt="A Brief History of Human Flight"></a><br><strong>A Brief History of Human Flight</strong><br>英語 · 59 秒</td>
</tr>
</table>
## 動作環境 📦
- 推奨プラットフォーム: Windows 10 以降、macOS 11 以降、または主要な Linux ディストリビューション
- ローカル環境へのデプロイには Python 3.11 以降が必要です。Python 3.11 を推奨します
- GPU は必須ではありませんが、ローカルでの文字起こしや動画処理を高速化したい場合、一括生成をより快適に行いたい場合には推奨されます
| 項目 | 最低 | 推奨 | 最適 |
| ------ | ------ | -------------- | -------------- |
| CPU | 4 コア | 6〜8 コア | 8 コア以上 |
| メモリ | 4 GB | 8 GB | 16 GB 以上 |
| GPU | 不要 | VRAM 4 GB 以上 | VRAM 8 GB 以上 |
- 主にクラウド LLM、クラウド TTS、オンライン素材ソースを利用する場合は、GPU よりも CPU とメモリのほうが重要です
- `faster-whisper`、一括生成、負荷の高いローカル処理を利用する場合は、GPU によってスループットが目に見えて向上します
## クイックスタート 🚀
### おすすめの手順
- 手動でのインストールや設定をしたくない場合: AI エージェントで動画を生成する
- Windows ユーザー: まずはワンクリックパッケージを使うのが最短で試せます
- macOS / Linux ユーザー: ローカル環境の主要な手順として `uv` を使用します
- より独立した実行環境が欲しい場合: Docker でデプロイします
### AI エージェントで動画を生成する
お使いの AI エージェントが Skill ドキュメントを読み、ローカルのターミナルを操作できるなら、以下のプロンプトを送ってください。エージェントが MoneyPrinterTurbo のインストールと設定を行い、動画を生成して、動画ファイルのパスを返します。未設定の必須 API キーだけを尋ねてきます。このワークフローは現在 macOS と Windows に対応しています。
```text
Use this Skill: https://raw.githubusercontent.com/harry0703/MoneyPrinterTurbo/main/docs/skill/SKILL.md
Create a video with the topic "How AI is changing everyday life."
```
### Google Colab で実行する
ローカル環境を用意せずに MoneyPrinterTurbo を試したいですか？ Google Colab で直接実行できます！
[](https://colab.research.google.com/github/harry0703/MoneyPrinterTurbo/blob/main/docs/MoneyPrinterTurbo.ipynb)
### Windows
GitHub Releases から最新の Windows 用ワンクリックパッケージをダウンロードし、そのまま展開してください。
- [最新の Windows 用ワンクリックパッケージをダウンロード](https://github.com/harry0703/MoneyPrinterTurbo/releases/latest)
> 同ページの **Assets** から `.7z` ファイルをダウンロードしてください。GitHub が
> 自動生成する `Source code (zip)` / `Source code (tar.gz)` はソースコードのみで、
> 展開しても `webui.bat` だけが現れ、`start.bat` と `update.bat` は含まれません。
ダウンロード後は、まず `update.bat` を**ダブルクリック**して**最新のコード**に更新し、その後 `start.bat` をダブルクリックして起動することを推奨します
起動するとブラウザーが自動的に開きます（真っ白な画面になる場合は **Chrome** または **Edge** の利用を推奨します）
### macOS / Linux
以下のローカルセットアップまたは Docker の手順を参照してください。
## インストールとデプロイ 📥
### 前提条件
- ローカル環境へのデプロイには Python 3.11 以降が必要です
- Windows では、プロジェクトのパスに非 ASCII 文字、特殊文字、スペースを含めないでください
#### ① プロジェクトをクローンする
```shell
git clone https://github.com/harry0703/MoneyPrinterTurbo.git
```
#### ② 初期設定を行う
初回起動時に `config.example.toml` から `config.toml` が自動作成されるため、設定ファイルを手動で作成する必要はありません。クラウド LLM、オンライン素材、AI 動画などのサービスを利用する前に、WebUI の基本設定で対応する API キーを入力してください。
### Docker でデプロイ 🐳
#### ① Docker コンテナを起動する
Docker が未インストールの場合は、先に [Docker Desktop をダウンロードしてインストール](https://www.docker.com/products/docker-desktop/)してください。
Windows をお使いの場合は、Microsoft のドキュメントを参照してください:
1. [WSL をインストールする](https://learn.microsoft.com/en-us/windows/wsl/install)
2. [WSL で Docker コンテナを使用する](https://learn.microsoft.com/en-us/windows/wsl/tutorials/wsl-containers)
```shell
cd MoneyPrinterTurbo
docker compose -f docker-compose.release.yml up
```
> 既定では `docker-compose.release.yml` を推奨します。GitHub Container Registry からビルド済みイメージ `ghcr.io/harry0703/moneyprinterturbo:latest` を取得します。
> ローカルでイメージをビルドする必要がある場合は、これまでどおり `docker compose up` を実行できます。
> 初回起動の前に、`config.example.toml` を `config.toml` にコピーしておくと、コンテナにマウントされます。
#### ② WebUI にアクセスする
ブラウザーで http://127.0.0.1:8501 を開きます
#### ③ API ドキュメントにアクセスする
ブラウザーで http://127.0.0.1:8080/docs または http://127.0.0.1:8080/redoc を開きます
> API はデフォルトで同一オリジンのブラウザーアクセスのみを許可します。別のブラウザーフロントエンドから API を直接呼び出す場合に限り、環境変数 `CORS_ALLOWED_ORIGINS` へ信頼するオリジンを設定してください（例: `http://localhost:3000,https://frontend.example.com`）。curl、Postman、n8n、その他のサーバー側クライアントは CORS の影響を受けません。
### 手動でデプロイ 📦
#### ① Python の仮想環境を作成する
[uv](https://docs.astral.sh/uv/) を使って Python 環境と依存関係を管理します。本プロジェクトは Python 3.11 以降に対応しており、以下の例では Python 3.11 を使用します。
```shell
git clone https://github.com/harry0703/MoneyPrinterTurbo.git
cd MoneyPrinterTurbo
uv python install 3.11
uv sync --frozen
```
`uv` をまだ使っていない場合は、これまでどおり `venv + pip` も利用できます。
```shell
python3.11 -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt
```
補足:
- 現在は `pyproject.toml` が主要な依存関係の定義ファイルです。
- `uv.lock` が解決済みの環境を固定するため、既定では `uv sync --frozen` を推奨します。
- `requirements.txt` は従来の `pip` によるインストールのためだけに残されています。
#### ② WebUI を起動する 🌐
以下のコマンドは MoneyPrinterTurbo プロジェクトの`ルートディレクトリ`で実行する必要がある点に注意してください
###### Windows
```powershell
.\webui.bat
```
`webui.bat` は CMD から実行することもできます。
`webui.bat` は、プロジェクトの `.venv` またはポータブルパッケージに同梱された Python を優先して使用します。プロジェクトの Python が見つからず、`uv` がインストールされている場合は、自動的に `uv run streamlit` にフォールバックします。
同じ LAN 内の別の端末から WebUI にアクセスできるようにするには、`webui.bat` を実行する前に `set MPT_WEBUI_HOST=0.0.0.0` を実行してください。
###### macOS または Linux
```shell
sh webui.sh
```
このスクリプトは、プロジェクトの仮想環境または `uv` を自動的に使用し、利用可能なローカルポートを選択します。同じ LAN 内の別の端末からアクセスできるようにするには、次のように実行します:
```shell
MPT_WEBUI_HOST=0.0.0.0 sh webui.sh
```
起動するとブラウザーが自動的に開きます
#### ③ API サービスを起動する 🚀
```shell
uv run python main.py
```
仮想環境を手動で有効化済みの場合は、これまでどおり次のように実行できます:
```shell
python main.py
```
#### ④ CLI のみのモード（ブラウザー不要） ⌨️
ブラウザーやポートフォワーディングが利用できない場合は、コマンドラインから直接動画を生成できます。最もシンプルな生成コマンドは次のとおりです:
```shell
uv run python cli.py --video-subject "How AI is changing everyday life"
```
コマンドの完全なリファレンス、パラメーターの説明、使い方については、次を実行してください:
```shell
uv run python cli.py --help
```
複数のタスクを順番に実行するには、`--batch-file` で UTF-8 の JSON 配列または
JSONL マニフェストを指定します。CLI オプションが共通の既定値となり、各オブジェクトで
`VideoParams` フィールドを上書きできます。
```shell
uv run python cli.py --batch-file ./tasks.json --stop-at video
```
マニフェストは最大 100 タスク、1 MiB までです。最初のタスクを開始する前に全項目と
ローカルファイルを検証し、個別タスクが失敗しても後続タスクを続行して最後に JSON
サマリーを出力します。マニフェスト内の相対音声・ローカル素材パスはマニフェストの
ディレクトリを基準に解決されます。
## ナレーション・字幕・BGM 🎙️
### 音声合成
WebUI の **Azure TTS V1** は **Edge TTS** を利用しており、無料かつ API キー不要です。MoneyPrinterTurbo は **Azure TTS V2**、**SiliconFlow TTS**、**Google Gemini TTS**、**Xiaomi MiMo TTS**、**ElevenLabs TTS**、セルフホストの **Chatterbox TTS**、セルフホストの **Kokoro TTS**、**Fish Audio TTS**、および音声なしモードにも対応しています。
WebUI でプロバイダーと音声を選択し、必要な認証情報については画面の案内に従ってください。Edge TTS に API キーは不要です。[Azure TTS V2](https://portal.azure.com/) やその他のクラウドプロバイダーでは、それぞれのプラットフォームで発行した認証情報が必要です。利用可能な Edge TTS の音声は[音声リスト](./docs/voice-list.txt)で確認できます。
### 字幕生成
字幕の生成モードは 2 種類あります:
- **edge**: TTS のタイムスタンプを利用します。GPU なしで高速に動作し、既定のモードです。
- **whisper**: より正確な字幕のタイムラインが必要な場合に、ローカルの `faster-whisper` による文字起こしを利用します。モデルは初回利用時にダウンロードされます。
モードを切り替えるには `config.toml` の `subtitle_provider` を設定します。Whisper は既定で約 3 GB の `large-v3` モデルを使用します。より小さく高速な約 1.6 GB の `large-v3-turbo` モデルを使う場合:
```toml
[app]
subtitle_provider = "whisper"
[whisper]
model_size = "large-v3-turbo"
```
> 初回利用時、Whisper は Hugging Face からモデルを自動的にダウンロードします。自動ダウンロードに失敗する場合は、[Hugging Face](https://huggingface.co/Systran/faster-whisper-large-v3) から `whisper-large-v3` を手動でダウンロードしてください。
モデルを展開したら、ディレクトリ全体を `.\MoneyPrinterTurbo\models` に配置します。最終的なパスは `.\MoneyPrinterTurbo\models\whisper-large-v3` になります:
```
MoneyPrinterTurbo
├─models
│ └─whisper-large-v3
│ config.json
│ model.bin
│ preprocessor_config.json
│ tokenizer.json
│ vocabulary.json
```
### BGM
動画の BGM は、プロジェクトの `resource/songs` ディレクトリにあります。
> 現在のプロジェクトには YouTube 動画由来の既定の音楽がいくつか含まれています。著作権上の問題がある場合は削除してください。
### 字幕フォント
動画の字幕描画に使うフォントは、プロジェクトの `resource/fonts` ディレクトリにあります。独自のフォントを追加することもできます。
## よくある質問 🤔
<details>
<summary>RuntimeError: No ffmpeg exe could be found</summary>
通常、ffmpeg は自動的にダウンロードされ、検出されます。
しかし、環境の問題で自動ダウンロードができない場合は、次のようなエラーが発生することがあります:
```
RuntimeError: No ffmpeg exe could be found.
Install ffmpeg on your system, or set the IMAGEIO_FFMPEG_EXE environment variable.
```
この場合は、[gyan.dev](https://www.gyan.dev/ffmpeg/builds/) から FFmpeg をダウンロードして展開し、`ffmpeg_path` に実際のインストールパスを設定してください。
```toml
[app]
# 実際のパスに合わせて設定してください。Windows のパス区切り文字は \\ である点に注意してください
ffmpeg_path = "C:\\Users\\harry\\Downloads\\ffmpeg.exe"
```
</details>
<details>
<summary>OSError: [Errno 24] Too many open files</summary>
これはシステムのオープンファイル数の上限が原因です。システムのファイルオープン数の上限を変更することで解決できます。
現在の上限を確認します:
```shell
ulimit -n
```
低すぎる場合は、例えば次のように引き上げます:
```shell
ulimit -n 10240
```
</details>
<details>
<summary>Whisper モデルのダウンロードに失敗する</summary>
```
LocalEntryNotFoundError: Cannot find an appropriate cached snapshot folder for the specified revision on the local disk and
outgoing traffic has been disabled.
To enable repo look-ups and downloads online, pass 'local_files_only=False' as input.
```
または
```
An error occurred while synchronizing the model Systran/faster-whisper-large-v3 from the Hugging Face Hub:
An error happened while trying to locate the files on the Hub and we cannot find the appropriate snapshot folder for the
specified revision on the local disk. Please check your internet connection and try again.
Trying to load the model directly from the local cache, if it exists.
```
解決方法: [Hugging Face からモデルを手動でダウンロードする方法を参照してください](#字幕生成)
</details>
## フィードバックと提案 📢
- [issue](https://github.com/harry0703/MoneyPrinterTurbo/issues) または [pull request](https://github.com/harry0703/MoneyPrinterTurbo/pulls) を送っていただけます。
## ライセンス 📝
[`LICENSE`](LICENSE) ファイルをクリックしてご覧ください
## /README.md
<div align="center">
# MoneyPrinterTurbo 💸
### 一站式 AI 短视频生成工具
只需提供视频<b>主题</b>或<b>关键词</b>，即可自动生成视频脚本、匹配素材、生成字幕和背景音乐，并合成高清短视频。
[](https://github.com/harry0703/MoneyPrinterTurbo/releases)
[](https://github.com/harry0703/MoneyPrinterTurbo/releases/latest)
[](https://www.python.org/)
[](https://github.com/harry0703/MoneyPrinterTurbo/releases/latest)
<a href="https://trendshift.io/repositories/8731" target="_blank"><img src="https://trendshift.io/api/badge/repositories/8731" alt="harry0703%2FMoneyPrinterTurbo | Trendshift" style="width: 250px; height: 55px;" width="250" height="55"/></a>
<a href="https://www.star-history.com/harry0703/moneyprinterturbo"><img src="https://api.star-history.com/badge?repo=harry0703/MoneyPrinterTurbo" alt="Star History Rank" style="height: 55px;" height="55"/></a>
简体中文 | [English](README-en.md) | [日本語](README-ja.md) | [版本发布](https://github.com/harry0703/MoneyPrinterTurbo/releases) | [问题反馈](https://github.com/harry0703/MoneyPrinterTurbo/issues)
</div>
## 界面预览 🖥️
<h4 align="center">WebUI</h4>
<h4 align="center">API</h4>
---
## 特别感谢 ❤️
<div align="center">
<a href="https://platform.kimi.com?track_id=track-6eec1e56a4494e52adcaebbcbbefce59&aff=moneyprinterturbo" target="_blank"><img src="https://gcdn.moonshot.cn/growth-cdn/sponsor/kimi-zh.png" alt="Kimi 赞助 MoneyPrinterTurbo" width="100%"></a>
</div>
感谢 [Kimi](https://platform.kimi.com?track_id=track-6eec1e56a4494e52adcaebbcbbefce59&aff=moneyprinterturbo) 赞助本项目！[Kimi K3](https://www.kimi.com/blog/kimi-k3?aff=moneyprinterturbo) 是 Moonshot AI 迄今能力最强的模型，也是全球首个开源 3T 级模型，拥有原生视觉能力与 100 万 Token 上下文，在知识工作、推理和长周期任务中展现前沿性能。在 MoneyPrinterTurbo 中，K3 能直接驱动视频创作，不仅撰写视频文案，还会提炼素材搜索关键词、决定成片画面；对内容理解越准确，匹配到的素材就越贴题。
**MoneyPrinterTurbo 用户专属优惠：新用户通过专属链接注册，首次成功充值可获充值金额 10% 的 API 额度，最高赠送 ¥1000。活动截至 2026 年 12 月 31 日。前往 Kimi 开放平台（[中文站](https://platform.kimi.com?track_id=track-6eec1e56a4494e52adcaebbcbbefce59&aff=moneyprinterturbo)｜[Global](https://platform.kimi.ai?track_id=track-9e3b711aa2594e378f6fe5b8de718a76&aff=moneyprinterturbo)）体验 API。**
<br>
<table align="center">
<tr>
<td align="center" width="120">
<a href="https://www.volcengine.com/activity/ai618?utm_campaign=hw&utm_content=hw&utm_medium=devrel_tool_web&utm_source=OWO&utm_term=MoneyPrinterTurbo"><img src="docs/sponsors/volcengine-logo.svg" alt="火山引擎" height="32"></a><br>
<a href="https://www.volcengine.com/activity/ai618?utm_campaign=hw&utm_content=hw&utm_medium=devrel_tool_web&utm_source=OWO&utm_term=MoneyPrinterTurbo"><strong>火山引擎</strong></a>
</td>
<td align="left">
感谢字节火山引擎赞助本项目！火山方舟 Agent/Coding Plan 国模套餐<strong>首购 9.9</strong>，支持 GLM-5.3、Kimi-K3、DeepSeek、MiniMax、Doubao 等，注册免费领 <strong>2500w Token</strong>，统一 API，适配编码与智能体开发。<a href="https://www.volcengine.com/activity/ai618?utm_campaign=hw&utm_content=hw&utm_medium=devrel_tool_web&utm_source=OWO&utm_term=MoneyPrinterTurbo">立即前往</a>
</td>
</tr>
<tr>
<td align="center" width="120">
<a href="https://go.apimart.ai/gh-moneyprinterturbo"><img src="docs/sponsors/apimart-logo.png" alt="APIMart" width="100"></a>
</td>
<td align="left">
感谢 <a href="https://go.apimart.ai/gh-moneyprinterturbo">APIMart</a> 赞助了本项目！APIMart 是专注 AI 图片/视频生成的低价 API 平台，<strong>GPT-Image-2 低至 $0.006/张，1 美元可出图 160+ 张</strong>。<strong>图片、视频一套异步 API 通吃，换模型不改代码</strong>；提交任务拿 ID，通过轮询或回调获取结果，支持万张级批量生成。按量付费、无月费，通过<a href="https://go.apimart.ai/gh-moneyprinterturbo">此注册链接</a>注册即可开用。
</td>
</tr>
<tr>
<td align="center" width="120">
<a href="https://metaso.cn/minimax-h3/?s=MPT"><img src="docs/sponsors/metaso-logo.png" alt="秘塔科技" width="100"></a><br>
<a href="https://metaso.cn/minimax-h3/?s=MPT"><strong>秘塔科技</strong></a>
</td>
<td align="left">
<strong>MiniMax H3 视频生成 API｜秘塔科技</strong><br>
秘塔科技提供高性价比的 MiniMax H3 视频生成服务：<strong>768P 仅 0.09 元/秒，2K 仅 0.15 元/秒</strong>。支持原生 2K、音画同步，API 兼容 <strong>OpenAI 协议</strong>，同时支持 <strong>ComfyUI</strong>，无需自行部署 GPU。<br>
🎁 通过 <a href="https://metaso.cn/minimax-h3/?s=MPT">MoneyPrinterTurbo专属链接注册</a>，即可领取赠送额度及专属优惠。
</td>
</tr>
<tr>
<td align="center" width="120">
<a href="https://infistar.cc/register?aff=6T4EYXP2&ref_source=link"><img src="docs/sponsors/infistar-logo.svg" alt="Infistar.cc 无限星河" height="56"></a><br>
<a href="https://infistar.cc/register?aff=6T4EYXP2&ref_source=link"><strong>Infistar.cc 无限星河</strong></a>
</td>
<td align="left">
感谢 <a href="https://infistar.cc/register?aff=6T4EYXP2&ref_source=link">Infistar.cc 无限星河</a> 赞助本项目！Infistar.cc 提供高性价比的大模型 API 服务：大模型低至<strong>官方 0.1 折</strong>，AI 生图低至 <strong>¥0.06/张</strong>。<br>
<strong>一个 API Key，覆盖文案、图片与视频创作</strong>，支持 GPT、Claude、Gemini、DeepSeek、通义千问、可灵等主流模型，无需分组。<br>
<strong>全模型支持验真</strong>，价格透明，正规备案，人民币实时计费，<strong>支持对公发票</strong>。<br>
🎁 MPT 专属福利：通过<a href="https://infistar.cc/register?aff=6T4EYXP2&ref_source=link">专属链接注册</a>即送 <strong>$5 体验额度</strong>，并享首充特惠。
</td>
</tr>
<tr>
<td align="center" width="120">
<a href="https://ofox.ai/?utm_source=github&utm_medium=sponsorship&utm_content=moneyprinterturbo"><img src="docs/sponsors/ofox-logo.svg" alt="OfoxAI" width="120"></a>
</td>
<td align="left">
感谢 <a href="https://ofox.ai/?utm_source=github&utm_medium=sponsorship&utm_content=moneyprinterturbo">OfoxAI</a> 赞助本项目！MoneyPrinterTurbo 已集成 Ofox 多模型文生视频，配置 API Key 即可使用。Seedance、MiniMax H3、Wan 生成视频素材，GPT Image 2.5、Seedream 制作封面配图，GPT、Claude、Gemini、DeepSeek 打磨脚本、辅助开发。<strong>一个 Key，文本、图像、视频共用余额</strong>；提供 OpenAI 兼容接口及 Anthropic、Gemini 原生接口。<strong>按量付费、价格公开、官方渠道接入，稳定高速不限量</strong>，前往 <a href="https://ofox.ai/?utm_source=github&utm_medium=sponsorship&utm_content=moneyprinterturbo">OfoxAI 查看模型与价格</a>。
</td>
</tr>
<tr>
<td align="center" width="120">
<a href="https://www.shengsuanyun.com/?from=CH_XUQ4OTSK"><img src="docs/sponsors/shengsuanyun-logo.jpg" alt="胜算云" height="56"></a><br>
<a href="https://www.shengsuanyun.com/?from=CH_XUQ4OTSK"><strong>胜算云</strong></a>
</td>
<td align="left">
感谢<a href="https://www.shengsuanyun.com/?from=CH_XUQ4OTSK">胜算云</a>对本项目的赞助！胜算云是面向 AI 原生团队的模型 API 聚合平台，汇集 Claude、ChatGPT、Gemini 等海内外大语言模型及多媒体模型，支持统一接入与按量调用。<br>
平台坚持合规 API 服务，杜绝逆向工程和资源稀释。此外平台提供企业级定制网关，包括团队成本与权限管理、智能路由、安全防护及 BYOK 密钥托管，并提供发票服务。<br>
🎁新用户通过<a href="https://www.shengsuanyun.com/?from=CH_XUQ4OTSK">此链接</a>注册，即可领取 10 元 Token 体验额度。
</td>
</tr>
<tr>
<td align="center" width="120">
<a href="https://www.ucloud.cn/site/active/astraflow?ytag=geo_waituo_Money"><img src="docs/sponsors/astraflow-logo.png" alt="星图 AstraFlow" height="56"></a><br>
<a href="https://www.ucloud.cn/site/active/astraflow?ytag=geo_waituo_Money"><strong>星图 AstraFlow</strong></a>
</td>
<td align="left">
感谢 <a href="https://www.ucloud.cn/site/active/astraflow?ytag=geo_waituo_Money">星图 AstraFlow</a> 赞助本项目！<br>
🎬 <strong>视频模型一站式调用</strong>：聚合 MiniMax-H3，Seedance-2.5 等主流视频生成模型，文案生成 + 视频生成一个平台全搞定，无需多端注册、多处对接<br>
🚀 <strong>200+ 大模型一键调用，新模型当日上架</strong>：聚合 DeepSeek V4.1、Kimi K3、Qwen 3.8 Max、GLM 5.3 等 200+ 主流大模型，新模型上线当天即可接入<br>
💰 <strong>UCloud 上市公司出品，调用计费透明可控</strong>：按 Key 级别计费，调用明细全程可查<br>
🎁 <strong>MoneyPrinterTurbo 用户专属福利</strong>：通过<a href="https://www.ucloud.cn/site/active/astraflow?ytag=geo_waituo_Money">专属推广链接</a>注册，即享新用户赠送额度，开箱即用！<a href="https://f.howxm.com/xs/u/HVFW5X8EIYV1">点击领取 50 元算力金</a>
</td>
</tr>
<tr>
<td align="center" width="120">
<a href="https://sidrune.ai/register?source=github&campaign=moneyprinterturbo&promo=moneyprinterturbo"><img src="docs/sponsors/sidrune-logo.png" alt="Sidrune AI" width="120"></a>
</td>
<td align="left">
感谢 <a href="https://sidrune.ai/register?source=github&campaign=moneyprinterturbo&promo=moneyprinterturbo">Sidrune AI</a> 赞助本项目！<strong>一个入口，接入并管理全球主流 AI 模型。</strong>Sidrune AI 面向个人开发者、技术团队与企业，通过统一 API 接入并管理全球主流 AI 模型，多线路动态调度提升可用性，模型表现、响应时间与费用透明可查。根据不同模型与线路，<strong>API 调用成本较官方或基准价格可降低 40%～98%</strong>。通过<a href="https://sidrune.ai/register?source=github&campaign=moneyprinterturbo&promo=moneyprinterturbo">专属链接注册</a>，即可获得 <strong>$3 API 额度</strong>。
</td>
</tr>
<tr>
<td align="center" width="120">
<a href="https://reccloud.cn"><img src="docs/sponsors/reccloud-logo.svg" alt="录咖" height="36"></a><br>
<a href="https://reccloud.cn">录咖 AI</a>
</td>
<td align="left">
感谢 <a href="https://reccloud.cn">录咖（AI 智能多媒体服务平台）</a> 基于本项目提供免费的 <strong>AI 视频生成器</strong>，无需部署，在线即可使用，对新手更加友好。
</td>
</tr>
<tr>
<td align="center" width="120">
<a href="https://picwish.cn"><img src="docs/sponsors/picwish-logo.svg" alt="佐糖" height="36"></a><br>
<a href="https://picwish.cn">佐糖</a>
</td>
<td align="left">
感谢 <a href="https://picwish.cn">佐糖</a> 对本项目的支持与赞助！佐糖提供丰富的<strong>在线图像处理工具</strong>，操作简单，帮助用户轻松完成各类图片处理。
</td>
</tr>
</table>
## 作者的另一个开源项目：MangoDisk ⭐
<p align="center">
<a href="https://mangodisk.app/zh">
<picture>
<source media="(prefers-color-scheme: dark)" srcset="https://assets.mangodisk.app/images/screenshots/zh/dark-01-deep-cleanup.jpg?ver=1.1.0">
<source media="(prefers-color-scheme: light)" srcset="https://assets.mangodisk.app/images/screenshots/zh/light-01-deep-cleanup.jpg?ver=1.1.0">
<img src="https://assets.mangodisk.app/images/screenshots/zh/light-01-deep-cleanup.jpg?ver=1.1.0" width="900" alt="MangoDisk 深度清理界面">
</picture>
</a>
</p>
<p align="center">
<strong>面向 macOS 和 Windows 的开源磁盘清理、空间分析与系统优化工具</strong><br>
一站式清理缓存、大文件、重复文件和应用残留，并提供磁盘空间分析、应用卸载、启动项管理、系统优化与维护
</p>
<p align="center">
<a href="https://mangodisk.app/zh">访问 MangoDisk 官网</a> · <a href="https://github.com/harry0703/MangoDisk">查看 GitHub 开源项目</a>
</p>
---
## 功能特性 🎯
### 创作入口与工作流
- [x] 提供 **AI Agent、WebUI、API 和 CLI** 四种使用方式，既能快速上手，也能接入自动化流程
- [x] 从主题自动完成脚本、配音、素材、字幕、配乐和剪辑，也支持在每个环节使用自定义内容
- [x] 支持 **批量生成多条成片**、任务历史记录，以及生成设置与 API Key 的导入、导出和恢复
### 脚本与模型服务
- [x] 支持 AI 自动生成或改写 **多语言视频脚本**，也可以直接使用自定义脚本
- [x] 支持 [Kimi / Moonshot AI](https://platform.kimi.com?track_id=track-6eec1e56a4494e52adcaebbcbbefce59&aff=moneyprinterturbo)、[OpenAI](https://platform.openai.com/api-keys)、[Anthropic Claude](https://platform.claude.com/settings/keys)、[Google Gemini](https://aistudio.google.com/app/apikey)、[DeepSeek](https://platform.deepseek.com/api_keys)、[阿里云通义千问](https://dashscope.console.aliyun.com/apiKey)、[Microsoft Azure OpenAI](https://portal.azure.com/#view/Microsoft_Azure_ProjectOxford/CognitiveServicesHub/~/OpenAI)、[火山引擎方舟](https://www.volcengine.com/activity/ai618?utm_campaign=hw&utm_content=hw&utm_medium=devrel_tool_web&utm_source=OWO&utm_term=MoneyPrinterTurbo)、[xAI Grok](https://console.x.ai/)、[MiniMax](https://platform.minimaxi.com/) 和 [小米 MiMo](https://platform.xiaomimimo.com/docs/zh-CN/quick-start/first-api-call) 等主流模型服务
- [x] 兼容 [胜算云](https://www.shengsuanyun.com/?from=CH_XUQ4OTSK)、[APIMart](https://go.apimart.ai/gh-moneyprinterturbo)、[Cloudflare AI Gateway](https://dash.cloudflare.com/)、[魔搭 ModelScope](https://modelscope.cn/docs/model-service/API-Inference/intro)、[AIHubMix](https://aihubmix.com/)、[AIML API](https://aimlapi.com/app/keys)、[EvoLink](https://evolink.ai/dashboard/keys)、[OpenRouter](https://openrouter.ai/settings/keys)、[API Route](https://www.api-route.com/)、[Fluxion AI](https://fluxionai.space/register?source=github&campaign=moneyprinterturbo&promo=MONEYPRINTERTURBO)、[Ollama](https://ollama.com/)、[Claude Code 订阅](https://code.claude.com/docs)、[OneAPI](https://github.com/songquanpeng/one-api)、[LiteLLM](https://docs.litellm.ai/docs/providers)、[Groq](https://console.groq.com/keys) 和 [Pollinations AI](https://enter.pollinations.ai/) 等统一网关、聚合平台和本地运行环境
### 视频与图片素材
- [x] 支持上传自己的 **本地图片和视频**，也可从 [Pexels（免费）](https://www.pexels.com/api/)、[Pixabay（免费）](https://pixabay.com/api/docs/) 和 [Coverr](https://coverr.co/developers?ctx=header_navigation) 获取高清库存素材
- [x] 支持 [秘塔 MiniMax H3](https://metaso.cn/minimax-h3/?s=MPT) 文生视频，可生成 `768P`/`2K`、4～15 秒的原始素材，并适配 `9:16`、`16:9` 和 `1:1` 三种画幅
- [x] 支持 [胜算云 AI 视频](https://www.shengsuanyun.com/?from=CH_XUQ4OTSK)，可生成多段 AI 视频素材，并沿用项目的配音、字幕和剪辑流程合成成片
- [x] 原生接入 [火山引擎方舟 Seedance](https://console.volcengine.com/ark/region:ark+cn-beijing/apikey)，可根据脚本片段生成连贯的视频画面
- [x] 支持 [WaveSpeed AI](https://wavespeed.ai) 文生视频，可根据脚本关键词快速生成原创素材
- [x] 支持 [OFox](https://ofox.ai) 多模型文生视频，一个 API Key 即可调用 Seedance、Wan 等模型
- [x] 通过异步 [MuAPI](https://muapi.ai) 文生视频接口生成 3～12 秒 AI 视频素材，并支持配置 Endpoint、画幅、分辨率和轮询参数
- [x] 支持 [OpenAI 兼容文生图](https://platform.openai.com/docs/guides/image-generation)，可连接云端服务或自定义图片网关，并将生成图片转换为动态视频片段
- [x] 支持调整片段时长、画面适配方式和素材匹配顺序，以适配不同画幅和叙事节奏
### 配音、字幕与配乐
- [x] 支持自动配音、上传配音和无配音三种方式，并提供音色试听与完整配音预览
- [x] 集成 **Edge TTS（免费、无需 API Key）**、Azure Speech、SiliconFlow、Google Gemini、小米 MiMo、MiniMax、ElevenLabs、Chatterbox、Kokoro、Fish Audio 和 ModelBest VoxCPM 等配音服务
- [x] 支持自动生成字幕，可调整字体、位置、颜色、大小、描边和背景样式
- [x] 支持随机、本地及 AI 生成背景音乐，并可独立控制音量
### 成片与发布
- [x] 支持竖屏 `9:16（1080×1920）`、横屏 `16:9（1920×1080）` 和方形 `1:1（1080×1080）`
- [x] 支持一键 **跨平台发布**，生成完成后可自动上传至 **TikTok、Instagram 和 YouTube Shorts**
## 作品展示 🎬
以下示例均由 MoneyPrinterTurbo 实际生成。
### 竖屏 9:16
<table width="100%">
<tr>
<td align="center" width="25%"><a href="https://harry0703.github.io/mpt-assets/?video=03-zh-portrait-city-morning.mp4"><img src="https://github.com/harry0703/mpt-assets/releases/download/assets/03-zh-portrait-city-morning.jpg" width="180" alt="城市醒来的时刻"></a><br><strong>城市醒来的时刻</strong><br>中文 · 14 秒</td>
<td align="center" width="25%"><a href="https://harry0703.github.io/mpt-assets/?video=05-zh-portrait-clean-energy.mp4"><img src="https://github.com/harry0703/mpt-assets/releases/download/assets/05-zh-portrait-clean-energy.jpg" width="180" alt="清洁能源的未来"></a><br><strong>清洁能源的未来</strong><br>中文 · 24 秒</td>
<td align="center" width="25%"><a href="https://harry0703.github.io/mpt-assets/?video=07-zh-portrait-space-exploration.mp4"><img src="https://github.com/harry0703/mpt-assets/releases/download/assets/07-zh-portrait-space-exploration.jpg" width="180" alt="为什么我们仍要探索太空"></a><br><strong>为什么我们仍要探索太空</strong><br>中文 · 27 秒</td>
<td align="center" width="25%"><a href="https://harry0703.github.io/mpt-assets/?video=17-zh-portrait-seed-journey.mp4"><img src="https://github.com/harry0703/mpt-assets/releases/download/assets/17-zh-portrait-seed-journey.jpg" width="180" alt="一粒种子的旅程"></a><br><strong>一粒种子的旅程</strong><br>中文 · 44 秒</td>
</tr>
<tr>
<td align="center" width="25%"><a href="https://harry0703.github.io/mpt-assets/?video=09-en-portrait-future-robotics.mp4"><img src="https://github.com/harry0703/mpt-assets/releases/download/assets/09-en-portrait-future-robotics.jpg" width="180" alt="The Future of Everyday Robotics"></a><br><strong>The Future of Everyday Robotics</strong><br>English · 21 sec</td>
<td align="center" width="25%"><a href="https://harry0703.github.io/mpt-assets/?video=11-en-portrait-small-habits.mp4"><img src="https://github.com/harry0703/mpt-assets/releases/download/assets/11-en-portrait-small-habits.jpg" width="180" alt="Small Habits, Lasting Change"></a><br><strong>Small Habits, Lasting Change</strong><br>English · 19 sec</td>
<td align="center" width="25%"><a href="https://harry0703.github.io/mpt-assets/?video=13-en-portrait-creative-work.mp4"><img src="https://github.com/harry0703/mpt-assets/releases/download/assets/13-en-portrait-creative-work.jpg" width="180" alt="Making Space for Creative Work"></a><br><strong>Making Space for Creative Work</strong><br>English · 20 sec</td>
<td align="center" width="25%"><a href="https://harry0703.github.io/mpt-assets/?video=15-en-portrait-coffee-science.mp4"><img src="https://github.com/harry0703/mpt-assets/releases/download/assets/15-en-portrait-coffee-science.jpg" width="180" alt="The Science Inside Coffee"></a><br><strong>The Science Inside Coffee</strong><br>English · 23 sec</td>
</tr>
</table>
### 横屏 16:9
<table width="100%">
<tr>
<td align="center" width="25%"><a href="https://harry0703.github.io/mpt-assets/?video=02-zh-landscape-deep-ocean.mp4"><img src="https://github.com/harry0703/mpt-assets/releases/download/assets/02-zh-landscape-deep-ocean.jpg" width="280" alt="深海里的微光"></a><br><strong>深海里的微光</strong><br>中文 · 23 秒</td>
<td align="center" width="25%"><a href="https://harry0703.github.io/mpt-assets/?video=04-zh-landscape-reading-power.mp4"><img src="https://github.com/harry0703/mpt-assets/releases/download/assets/04-zh-landscape-reading-power.jpg" width="280" alt="阅读如何塑造我们"></a><br><strong>阅读如何塑造我们</strong><br>中文 · 23 秒</td>
<td align="center" width="25%"><a href="https://harry0703.github.io/mpt-assets/?video=06-zh-landscape-pour-over-coffee.mp4"><img src="https://github.com/harry0703/mpt-assets/releases/download/assets/06-zh-landscape-pour-over-coffee.jpg" width="280" alt="一杯手冲咖啡的细节"></a><br><strong>一杯手冲咖啡的细节</strong><br>中文 · 23 秒</td>
<td align="center" width="25%"><a href="https://harry0703.github.io/mpt-assets/?video=08-zh-landscape-spring-journey.mp4"><img src="https://github.com/harry0703/mpt-assets/releases/download/assets/08-zh-landscape-spring-journey.jpg" width="280" alt="春天适合出发"></a><br><strong>春天适合出发</strong><br>中文 · 14 秒</td>
</tr>
<tr>
<td align="center" width="25%"><a href="https://harry0703.github.io/mpt-assets/?video=10-en-landscape-ocean-conservation.mp4"><img src="https://github.com/harry0703/mpt-assets/releases/download/assets/10-en-landscape-ocean-conservation.jpg" width="280" alt="Why Ocean Conservation Matters"></a><br><strong>Why Ocean Conservation Matters</strong><br>English · 25 sec</td>
<td align="center" width="25%"><a href="https://harry0703.github.io/mpt-assets/?video=14-en-landscape-sustainable-cities.mp4"><img src="https://github.com/harry0703/mpt-assets/releases/download/assets/14-en-landscape-sustainable-cities.jpg" width="280" alt="Designing More Sustainable Cities"></a><br><strong>Designing More Sustainable Cities</strong><br>English · 27 sec</td>
<td align="center" width="25%"><a href="https://harry0703.github.io/mpt-assets/?video=16-en-landscape-mountain-perspective.mp4"><img src="https://github.com/harry0703/mpt-assets/releases/download/assets/16-en-landscape-mountain-perspective.jpg" width="280" alt="What Mountains Teach Us"></a><br><strong>What Mountains Teach Us</strong><br>English · 18 sec</td>
<td align="center" width="25%"><a href="https://harry0703.github.io/mpt-assets/?video=18-en-landscape-history-of-flight.mp4"><img src="https://github.com/harry0703/mpt-assets/releases/download/assets/18-en-landscape-history-of-flight.jpg" width="280" alt="A Brief History of Human Flight"></a><br><strong>A Brief History of Human Flight</strong><br>English · 59 sec</td>
</tr>
</table>
## 配置要求 📦
- 建议系统：Windows 10、macOS 11.0 或更高版本，以及主流 Linux 发行版
- 本地部署需要 Python 3.11 或更高版本，推荐使用 Python 3.11
- GPU 不是必需项，但如果你希望本地转录、更快的视频处理或更顺畅的批量生成体验，建议使用带显存的独立显卡
| 项目 | 最低配置 | 推荐配置 | 理想配置 |
| ---- | -------- | --------------- | --------------- |
| CPU | 4 核 | 6 到 8 核 | 8 核及以上 |
| RAM | 4 GB | 8 GB | 16 GB 及以上 |
| GPU | 非必须 | 4 GB 显存及以上 | 8 GB 显存及以上 |
- 如果你主要依赖云端 LLM、云端 TTS 和在线素材源，CPU 与内存比 GPU 更重要
- 如果你启用 `faster-whisper`、批量生成或更重的本地处理链路，GPU 会明显提升速度
## 快速开始 🚀
### 推荐使用方式
- 不想手动安装和配置：直接使用 AI Agent 生成视频
- Windows 用户：优先使用一键启动包，适合快速体验
- macOS / Linux 用户：优先使用 `uv` 进行本地部署
- 想要隔离运行环境：优先使用 Docker 部署
### 使用 AI Agent 生成视频
如果你的 AI Agent 支持读取 Skill 文档并操作本地终端，可以直接发送下面这段话。Agent 会自动完成安装、配置和视频生成；只有缺少必要的 API Key 时才会向你询问，完成后会返回生成的视频文件路径。目前支持 macOS 和 Windows。
```text
使用这个 Skill：https://raw.githubusercontent.com/harry0703/MoneyPrinterTurbo/main/docs/skill/SKILL.md
帮我生成一个主题为“人工智能如何改变普通人的日常生活”的视频。
```
### 在 Google Colab 中运行
免去本地环境配置，点击直接在 Google Colab 中快速体验 MoneyPrinterTurbo
[](https://colab.research.google.com/github/harry0703/MoneyPrinterTurbo/blob/main/docs/MoneyPrinterTurbo.ipynb)
### Windows 一键启动包
下载一键启动包，解压直接使用（路径不要有 **中文**、**特殊字符**、**空格**）
- [下载最新 Windows 一键启动包](https://github.com/harry0703/MoneyPrinterTurbo/releases/latest)
> 请在页面下方的 **Assets** 区域下载 `.7z` 压缩包。GitHub 自动生成的
> `Source code (zip)` / `Source code (tar.gz)` 只是源码，解压后只有 `webui.bat`，
> 不含 `start.bat` 和 `update.bat`。
下载后，建议先**双击执行** `update.bat` 更新到**最新代码**，然后双击 `start.bat` 启动
启动后，会自动打开浏览器（如果打开是空白，建议换成 **Chrome** 或者 **Edge** 打开）
## 安装部署 📥
### 前提条件
- 本地部署需要 Python 3.11 或更高版本
- Windows 用户建议避免使用包含中文、特殊字符或空格的项目路径
#### ① 克隆代码
```shell
git clone https://github.com/harry0703/MoneyPrinterTurbo.git
```
#### ② 首次配置
首次启动时，项目会根据 `config.example.toml` 自动创建 `config.toml`，无需手动创建配置文件。使用云端大模型、在线素材或 AI 视频等服务前，请在 WebUI 的基础设置中填写对应的 API Key。
### Docker 部署 🐳
#### ① 启动 Docker
如果尚未安装 Docker，请先[下载并安装 Docker Desktop](https://www.docker.com/products/docker-desktop/)。
Windows 用户可以参考微软的文档：
1. [安装 WSL](https://learn.microsoft.com/zh-cn/windows/wsl/install)
2. [在 WSL 中使用 Docker 容器](https://learn.microsoft.com/zh-cn/windows/wsl/tutorials/wsl-containers)
```shell
cd MoneyPrinterTurbo
docker compose -f docker-compose.release.yml up
```
> 默认推荐使用 `docker-compose.release.yml`，它会直接拉取 GitHub Container Registry 上的预构建镜像：`ghcr.io/harry0703/moneyprinterturbo:latest`。
> 如果你需要本地重新构建镜像，可以继续使用 `docker compose up`。
> 首次启动前，请将 `config.example.toml` 复制为 `config.toml`，供容器挂载使用。
#### ② 访问 WebUI
打开浏览器，访问 http://127.0.0.1:8501
#### ③ 访问 API 文档
打开浏览器，访问 http://127.0.0.1:8080/docs 或者 http://127.0.0.1:8080/redoc
> API 默认仅允许同源网页访问。只有独立网页前端需要从其他来源直接调用 API 时，才应通过环境变量 `CORS_ALLOWED_ORIGINS` 配置可信来源，例如 `http://localhost:3000,https://frontend.example.com`。curl、Postman、n8n 和其他服务端调用不受 CORS 限制。
### 手动部署 📦
> 视频教程
- [完整的使用演示](https://v.douyin.com/iFhnwsKY/)
- [如何在 Windows 上部署](https://v.douyin.com/iFyjoW3M)
#### ① 创建虚拟环境
推荐使用 [uv](https://docs.astral.sh/uv/) 管理 Python 环境和依赖。项目支持 Python 3.11 或更高版本，以下示例使用 Python 3.11。
```shell
git clone https://github.com/harry0703/MoneyPrinterTurbo.git
cd MoneyPrinterTurbo
uv python install 3.11
uv sync --frozen
```
如果你暂时不使用 `uv`，也可以继续使用 `venv + pip`
```shell
python3.11 -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt
```
说明：
- `pyproject.toml` 是主依赖定义文件
- `uv.lock` 是锁文件，建议默认执行 `uv sync --frozen`
- `requirements.txt` 仅保留给旧的 `pip` 安装方式兼容使用
#### ② 启动 WebUI 🌐
注意需要到 MoneyPrinterTurbo 项目 `根目录` 下执行以下命令
###### Windows
```powershell
.\webui.bat
```
在 CMD 中也可以执行 `webui.bat`。
`webui.bat` 会优先使用项目 `.venv` 或一键包内置 Python；如果没有找到项目 Python，但已安装 `uv`，会自动切换为 `uv run streamlit`。
如需允许局域网内其他设备访问 WebUI，可以先执行 `set MPT_WEBUI_HOST=0.0.0.0`，再运行 `webui.bat`。
###### macOS 或 Linux
```shell
sh webui.sh
```
脚本会自动使用项目虚拟环境或 `uv`，并选择可用的本地端口。如需允许局域网内其他设备访问，可以执行：
```shell
MPT_WEBUI_HOST=0.0.0.0 sh webui.sh
```
启动后，会自动打开浏览器（如果打开是空白，建议换成 **Chrome** 或者 **Edge** 打开）
#### ③ 启动 API 服务 🚀
```shell
uv run python main.py
```
如果你已经手动激活了虚拟环境，也可以直接执行：
```shell
python main.py
```
#### ④ 纯命令行方式（无浏览器）⌨️
如果你无法使用浏览器或端口转发，可以直接在命令行生成视频。最简单的完整视频生成命令如下：
```shell
uv run python cli.py --video-subject "人工智能如何改变日常生活"
```
字幕样式和配音参数按以下优先级取值：**命令行显式参数 > `config.toml` 中
`[ui]` 保存的 WebUI 设置 > 内置默认值**。其余生成设置（如背景音乐、视频数量、
段落数量等）不会自动沿用 WebUI 的保存值。若 WebUI 中选择了上传自备音频，
命令行需要显式传入 `--custom-audio-file`，因为音频路径不会被保存。
如需查看完整命令、参数说明和使用方法，可以执行：
```shell
uv run python cli.py --help
```
如需顺序执行多个任务，可通过 `--batch-file` 提供 UTF-8 JSON 数组或 JSONL
清单。CLI 参数作为全局默认值，每个对象可覆盖 `VideoParams` 字段：
```shell
uv run python cli.py --batch-file ./tasks.json --stop-at video
```
清单最多包含 100 个任务且不超过 1 MiB。所有条目会在第一个任务启动前完成
参数与本地文件预检；单个任务运行失败不会阻止后续条目，结束后会输出统一的
JSON 汇总。清单中的相对自定义音频与本地素材路径以清单目录为基准。
## 配音、字幕与配乐 🎙️
### 语音合成
WebUI 中的 **Azure TTS V1** 基于 **Edge TTS**，免费且无需 API Key。项目同时支持 **Azure TTS V2**、**SiliconFlow TTS**、**Google Gemini TTS**、**小米 MiMo TTS**、**MiniMax TTS**、**ElevenLabs TTS**、自托管 **Chatterbox TTS**、自托管 **Kokoro TTS**、**Fish Audio TTS**、[ModelBest VoxCPM TTS](https://platform.modelbest.cn/console/docs/api/audio)，以及无配音模式。
可直接在 WebUI 中选择 Provider 和音色，并按照界面提示填写所需凭据。Edge TTS 不需要 API Key；[Azure TTS V2](https://portal.azure.com/) 及其他云端服务需要对应平台的凭据。Edge TTS 音色可查看：[音色列表](./docs/voice-list.txt)。
ModelBest VoxCPM 需要填写 API Key 和支持 `speech_synthesis` 的模型 ID。服务以 SSE 返回 WAV 音频，项目会自动转换为视频流水线使用的 MP3。除了普通文本合成，还可以上传参考音频复刻说话人音色；开启“保留参考音频的节奏与情绪”后，默认复用这段音频并要求填写其准确逐字稿，也可另选一段节奏与情绪示范，通过 `prompt_audio` 和 `prompt_text` 延续演绎。WebUI 可使用本地 Whisper 自动生成可编辑的逐字稿草稿。上传文件最大 20 MiB，转换后的单个 WAV 最大 5 MiB。参考素材仅用于当前浏览器会话和任务，不写入配置、预设、任务历史或日志。
### 字幕生成
当前支持两种字幕生成方式：
- **edge**：使用 TTS 时间戳生成字幕，速度快，不需要 GPU，默认使用该模式。
- **whisper**：使用本地 `faster-whisper` 转写音频，适用于需要更准确字幕时间轴的场景。首次使用时需要下载模型。
在 `config.toml` 中修改 `subtitle_provider` 即可切换模式。Whisper 默认使用约 3 GB 的 `large-v3`；如需更小、更快的模型，可以使用约 1.6 GB 的 `large-v3-turbo`：
```toml
[app]
subtitle_provider = "whisper"
[whisper]
model_size = "large-v3-turbo"
```
> 首次使用 Whisper 时，程序会自动从 Hugging Face 下载模型。如果当前网络无法自动下载，可以从 [Hugging Face](https://huggingface.co/Systran/faster-whisper-large-v3) 手动下载 `whisper-large-v3`。
下载并解压后，将整个目录放到 `.\MoneyPrinterTurbo\models`，最终路径应为 `.\MoneyPrinterTurbo\models\whisper-large-v3`：
```
MoneyPrinterTurbo
├─models
│ └─whisper-large-v3
│ config.json
│ model.bin
│ preprocessor_config.json
│ tokenizer.json
│ vocabulary.json
```
### 背景音乐
用于视频的背景音乐，位于项目的 `resource/songs` 目录下。
> 当前项目里面放了一些默认的音乐，来自于 YouTube 视频，如有侵权，请删除。
### 字幕字体
用于视频字幕的渲染，位于项目的 `resource/fonts` 目录下，你也可以放进去自己的字体。
## 常见问题 🤔
<details>
<summary>RuntimeError: No ffmpeg exe could be found</summary>
通常情况下，ffmpeg 会被自动下载，并且会被自动检测到。
但是如果你的环境有问题，无法自动下载，可能会遇到如下错误：
```
RuntimeError: No ffmpeg exe could be found.
Install ffmpeg on your system, or set the IMAGEIO_FFMPEG_EXE environment variable.
```
此时可以从 [gyan.dev](https://www.gyan.dev/ffmpeg/builds/) 下载 FFmpeg。解压后，将 `ffmpeg_path` 设置为实际安装路径即可。
```toml
[app]
# 请根据你的实际路径设置，注意 Windows 路径分隔符为 \\
ffmpeg_path = "C:\\Users\\harry\\Downloads\\ffmpeg.exe"
```
</details>
<details>
<summary>OSError: [Errno 24] Too many open files</summary>
这个问题是由于系统打开文件数限制导致的，可以通过修改系统的文件打开数限制来解决。
查看当前限制
```shell
ulimit -n
```
如果过低，可以调高一些，比如
```shell
ulimit -n 10240
```
</details>
<details>
<summary>Whisper 模型下载失败</summary>
```
LocalEntryNotFoundError: Cannot find an appropriate cached snapshot folder for the specified revision on the local disk and
outgoing traffic has been disabled.
To enable repo look-ups and downloads online, pass 'local_files_only=False' as input.
```
或者
```
An error occurred while synchronizing the model Systran/faster-whisper-large-v3 from the Hugging Face Hub:
An error happened while trying to locate the files on the Hub and we cannot find the appropriate snapshot folder for the
specified revision on the local disk. Please check your internet connection and try again.
Trying to load the model directly from the local cache, if it exists.
```
解决方法：[查看如何从 Hugging Face 手动下载模型](#字幕生成)
</details>
## 反馈建议 📢
- 可以提交 [issue](https://github.com/harry0703/MoneyPrinterTurbo/issues) 或者 [pull request](https://github.com/harry0703/MoneyPrinterTurbo/pulls)。
## 许可证 📝
点击查看 [`LICENSE`](LICENSE) 文件
## /app/__init__.py
```py path="/app/__init__.py"
"""MoneyPrinterTurbo 应用包元数据。"""
__version__ = "1.3.8"
```
## /app/asgi.py
```py path="/app/asgi.py"
"""Application implementation - ASGI."""
import math
import os
from contextlib import asynccontextmanager
from urllib.parse import urlsplit
from fastapi import FastAPI, Request
from fastapi.exceptions import RequestValidationError
from fastapi.encoders import jsonable_encoder
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
from fastapi.staticfiles import StaticFiles
from loguru import logger
from app.config import config
from app.controllers import base
from app.models.exception import HttpException
from app.router import root_api_router
from app.utils import utils
@asynccontextmanager
async def application_lifespan(_: FastAPI):
"""集中处理 API 进程启动恢复和关闭日志。"""
logger.info("startup event")
configured_api_key = config.app.get("api_key", "")
if configured_api_key in (None, ""):
logger.warning(
"API key authentication is disabled; keep the API on a trusted network"
)
elif isinstance(configured_api_key, str):
# 只记录保护范围，不得输出 Key、长度或摘要，避免凭据进入日志系统。
logger.info("API key authentication is enabled for /api/v1 and /tasks")
else:
logger.error(
"API key authentication is misconfigured: app.api_key must be a string"
)
# 跨平台发布由当前进程线程池执行，不会在服务重启后恢复。启动时把 Redis
# 中确认已失去执行进程的活动状态收敛为失败，避免任务永久无法删除。
from app.services import task as task_service
task_service.recover_interrupted_cross_posts()
# Redis queue entries persist across API restarts. No task_done callback
# exists in the new process to dispatch them, so fill its worker slots now.
from app.controllers.manager.redis_manager import RedisTaskManager
from app.controllers.v1 import video as video_controller
if isinstance(video_controller.task_manager, RedisTaskManager):
video_controller.task_manager.resume_queued_tasks()
try:
yield
finally:
logger.info("shutdown event")
def exception_handler(request: Request, e: HttpException):
return JSONResponse(
status_code=e.status_code,
content=utils.get_response(e.status_code, e.data, e.message),
)
def validation_exception_handler(request: Request, e: RequestValidationError):
# Rejected inputs can contain NaN/Infinity, and custom validators attach
# exception objects to ctx. Neither can be emitted by JSONResponse directly.
errors = jsonable_encoder(
e.errors(),
custom_encoder={
float: lambda value: value if math.isfinite(value) else str(value),
Exception: str,
},
)
return JSONResponse(
status_code=400,
content=utils.get_response(
status=400, data=errors, message="field required"
),
)
_DEFAULT_PORTS = {"http": 80, "https": 443}
def _normalize_allowed_origin(raw_origin: str) -> str | None:
"""把配置项折叠成浏览器实际发送的 Origin 形式。
Origin 请求头由 RFC 6454 固定为 ``scheme://host[:port]``：没有路径、没有尾
斜杠，scheme 与 host 已经规范化为小写，并且省略协议的默认端口。用户在浏览器
地址栏复制得到的 ``https://frontend.example/``，以及显式写出默认端口的
``https://frontend.example:443``，都与浏览器实际发送的
``https://frontend.example`` 逐字符不相等，白名单因此静默失效：前端只看到
CORS 报错，服务端只留下一行 blocked 日志，两边都指不到配置本身。
返回 ``None`` 表示该写法不可能匹配任何 Origin（缺少 scheme、scheme 不是
http/https、主机名为空或含空白、IPv6 字面量畸形、端口非法或越界），调用方应
丢弃并告警。解析失败必须收敛成 ``None``：本函数在模块导入期就会执行，让
``urlsplit()`` 的 ``ValueError`` 冒出去等于整条 API 无法启动。
"""
if raw_origin == "*":
return raw_origin
try:
parsed = urlsplit(raw_origin)
scheme = parsed.scheme.lower()
# 读取 hostname / port 时才校验 netloc：畸形 IPv6 字面量由 urlsplit()
# 抛出，非数字端口与超出 0-65535 的端口在读取 port 时抛出。
host = parsed.hostname
port = parsed.port
except ValueError:
return None
# 只有 http/https 具备浏览器 Origin 语义，且两者都有可省略的默认端口。
if scheme not in _DEFAULT_PORTS or not host:
return None
if any(character.isspace() for character in host):
return None
# 显式小写，不依赖 urlsplit 在不同 Python 版本上的大小写处理。
host = host.lower()
# 浏览器序列化 Origin 时给 IPv6 字面量保留方括号；``:443``、``:0443``、
# ``:80`` 这类默认端口写法必须折叠成不带端口的形态才能与真实前端匹配。
if ":" in host:
host = f"[{host}]"
if port is None or port == _DEFAULT_PORTS[scheme]:
return f"{scheme}://{host}"
return f"{scheme}://{host}:{port}"
def parse_cors_allowed_origins(raw_origins: str | None) -> list[str]:
"""解析浏览器跨域来源白名单。
CORS 只约束浏览器中的跨域 JavaScript，不影响 curl、Postman、n8n
或服务端 SDK。未配置时返回空列表，表示默认不开放跨域访问；用户确实
部署了独立网页前端时，再通过 ``CORS_ALLOWED_ORIGINS`` 显式开启。
每个来源都先折叠成 Origin 请求头的规范形式，因此 ``https://a.example/``、
``HTTPS://A.Example`` 与显式默认端口的 ``https://a.example:443`` 得到同一个
结果，无法匹配任何 Origin 的写法会被丢弃并留下告警。畸形条目只丢弃自身，
不会中断这里的解析——本函数在模块导入期调用，解析异常会让 API 无法启动。
"""
if not raw_origins:
return []
# 去除逗号分隔项两侧的空白，并忽略空项，避免常见的环境变量格式
# ``https://a.example, https://b.example,`` 产生永远无法匹配的来源。
origins: list[str] = []
for candidate in raw_origins.split(","):
item = candidate.strip()
if not item:
continue
origin = _normalize_allowed_origin(item)
if origin is None:
logger.warning(
f"ignoring configured CORS origin that cannot match a browser "
f"Origin header: {item!r}"
)
continue
if origin not in origins:
origins.append(origin)
return origins
def configure_cors(instance: FastAPI, allowed_origins: list[str]) -> None:
"""按显式白名单配置 CORS；空白名单保持默认同源策略。"""
if not allowed_origins:
logger.info(
"browser cross-origin API access is disabled; set "
"CORS_ALLOWED_ORIGINS to enable trusted origins"
)
return
allow_all_origins = "*" in allowed_origins
configured_api_key = config.app.get("api_key", "")
if allow_all_origins and configured_api_key in (None, ""):
# ``*`` 是用户显式选择的兼容模式，因此不强制拒绝启动；但在免认证
# 状态下它会允许任意网页读取和调用 API，必须留下可定位的安全告警。
logger.warning(
"CORS allows every browser origin while API key authentication is "
"disabled; configure app.api_key or restrict CORS_ALLOWED_ORIGINS"
)
instance.add_middleware(
CORSMiddleware,
allow_origins=allowed_origins,
# Starlette 在 ``*`` 与 credentials 同时启用时会反射任意 Origin。
# 通配符模式不需要 Cookie 认证，因此主动关闭 credentials；显式来源
# 仍保留旧行为，避免影响已有独立网页前端的 credentials 请求模式。
allow_credentials=not allow_all_origins,
allow_methods=["*"],
allow_headers=["*"],
# 远程 HTTPS 前端访问本机或局域网 API 时，现代浏览器会额外发送
# Private Network Access 预检。只有精确白名单来源可以获得许可；
# 通配符模式继续拒绝，避免任意网站探测用户的私有网络服务。
allow_private_network=not allow_all_origins,
)
def is_browser_origin_allowed(
request: Request, allowed_origins: list[str]
) -> bool:
"""判断浏览器请求来源是否为同源或显式白名单来源。"""
origin = request.headers.get("origin")
if not origin:
# curl、Postman、n8n 和服务端 SDK 通常不发送 Origin。保留这类请求，
# 避免安全修复错误地改变现有 API 客户端的调用契约。
return True
if "*" in allowed_origins or origin in allowed_origins:
return True
# 浏览器对同源 POST 也可能发送 Origin。仅比较 scheme + authority，忽略
# 路径和查询参数；反向代理部署若未正确转发公网 scheme/host，可通过显式
# CORS_ALLOWED_ORIGINS 声明外部来源，避免依赖不可信的转发 Header。
request_url = urlsplit(str(request.url))
request_origin = f"{request_url.scheme}://{request_url.netloc}"
return origin == request_origin
def configure_browser_access(instance: FastAPI, allowed_origins: list[str]) -> None:
"""同时配置服务端 Origin 防护与浏览器 CORS 响应策略。"""
@instance.middleware("http")
async def reject_untrusted_browser_origin(request: Request, call_next):
"""主动拒绝不可信浏览器来源，覆盖无需 CORS 预检的简单请求。"""
if not is_browser_origin_allowed(request, allowed_origins):
origin = request.headers.get("origin", "")
logger.warning(
f"blocked untrusted browser origin: method={request.method}, "
f"path={request.url.path}, origin={origin}"
)
return JSONResponse(
status_code=403,
content=utils.get_response(
status=403,
message="cross-origin browser request is not allowed",
),
)
return await call_next(request)
# CORS 中间件最后注册后位于 Origin 防护外层：可信预检可直接成功，
# 非可信预检由 CORS 拒绝；无需预检的实际请求仍会进入上面的 403 防护。
configure_cors(instance, allowed_origins)
def get_application() -> FastAPI:
"""Initialize FastAPI application.
Returns:
FastAPI: Application object instance.
"""
instance = FastAPI(
title=config.project_name,
description=config.project_description,
version=config.project_version,
debug=False,
lifespan=application_lifespan,
)
instance.include_router(root_api_router)
instance.add_exception_handler(HttpException, exception_handler)
instance.add_exception_handler(RequestValidationError, validation_exception_handler)
return instance
app = get_application()
@app.middleware("http")
async def protect_generated_task_files(request: Request, call_next):
"""保护任务产物静态路由，防止绕过 API 鉴权直接下载。
``/tasks`` 由 StaticFiles 独立挂载，无法复用 APIRouter 的依赖，
因此在中间件中调用同一个 verify_token。鉴权函数会在未配置
api_key 时放行；OPTIONS 预检请求也保留给 CORS 中间件处理。
"""
request_path = request.url.path
is_task_file = request_path == "/tasks" or request_path.startswith("/tasks/")
if is_task_file and request.method != "OPTIONS":
try:
base.verify_token(request)
except HttpException as exception:
return exception_handler(request, exception)
return await call_next(request)
# 默认遵循浏览器同源策略；仅在用户显式配置可信网页来源时开放跨域。
cors_allowed_origins = parse_cors_allowed_origins(
os.getenv("CORS_ALLOWED_ORIGINS", "")
)
configure_browser_access(app, cors_allowed_origins)
task_dir = utils.task_dir()
app.mount("/tasks", StaticFiles(directory=task_dir, html=True), name="")
public_dir = utils.public_dir()
app.mount("/", StaticFiles(directory=public_dir, html=True), name="")
```
## /app/config/__init__.py
```py path="/app/config/__init__.py"
import sys
from app.config import config
from app.utils.logging_utils import configure_terminal_logger
def __init_logger():
# _log_file = utils.storage_dir("logs/server.log")
_lvl = config.log_level
configure_terminal_logger(
sys.stdout,
level=_lvl,
colorize=True,
)
# logger.add(
# _log_file,
# level=_lvl,
# format=format_log_record,
# rotation="00:00",
# retention="3 days",
# backtrace=True,
# diagnose=True,
# enqueue=True,
# )
__init_logger()
```
## /app/config/config.py
```py path="/app/config/config.py"
import copy
import errno
import os
import shutil
import socket
import tempfile
import threading
from contextlib import contextmanager
import toml
from loguru import logger
from app import __version__
root_dir = os.path.dirname(os.path.dirname(os.path.dirname(os.path.realpath(__file__))))
config_file = f"{root_dir}/config.toml"
_CONTAINER_CGROUP_MARKERS = ("docker", "containerd", "kubepods", "libpod", "podman")
_DOCKER_HOST_GATEWAY_NAME = "host.docker.internal"
_config_save_lock = threading.RLock()
_pending_config_lock = threading.RLock()
_pending_config_updates = {}
_pending_config_save_requested = False
_pending_config_flush_scheduled = False
_MISSING = object()
_DELETE = object()
_UTF8_BOM = "\ufeff"
class _SynchronizedConfig(dict):
"""保持 dict 使用方式不变，同时让运行期配置写操作服从同一把锁。"""
def __setitem__(self, key, value):
# Streamlit 每次整页 rerun 都会把当前控件值重新写回配置。视频任务持有
# runtime_config_lock 时，如果值没有变化，这次写入没有任何副作用，也
# 不应让刷新后的页面卡在表单中途。真正改变配置的写入仍进入下方锁，
# 因而不能在正在生成的视频中途切换 Provider、密钥或其它全局设置。
current = super().get(key, _MISSING)
if current is not _MISSING and current == value:
return
with _config_save_lock:
super().__setitem__(key, value)
def __delitem__(self, key):
with _config_save_lock:
super().__delitem__(key)
def clear(self):
if not self:
return
with _config_save_lock:
super().clear()
def pop(self, key, default=_MISSING):
# ``pop(key, default)`` 在 key 不存在时同样不会改变配置。WebUI 使用
# 这种写法表达“采用默认策略”，刷新时必须允许它直接完成。
if key not in self:
if default is _MISSING:
raise KeyError(key)
return default
with _config_save_lock:
if default is _MISSING:
return super().pop(key)
return super().pop(key, default)
def setdefault(self, key, default=None):
# 与 __setitem__ 相同，已存在 key 的 setdefault 是只读操作。提前返回
# 可以让只读取默认配置的页面刷新不受长任务配置锁影响。
current = super().get(key, _MISSING)
if current is not _MISSING:
return current
with _config_save_lock:
return super().setdefault(key, default)
def update(self, *args, **kwargs):
changes = dict(*args, **kwargs)
if all(
(current := dict.get(self, key, _MISSING)) is not _MISSING
and current == value
for key, value in changes.items()
):
return
with _config_save_lock:
super().update(changes)
def _pending_update_key(config_section, key):
"""为进程内固定配置分区生成待更新键。"""
return id(config_section), key
def update_config_nonblocking(config_section, key, value):
"""
非阻塞更新 WebUI 的运行期配置。
视频生成会持有 ``runtime_config_lock``，确保同一任务不会在执行中途切换
Provider、密钥或语音配置。Streamlit 控件发生变化时不能等待这把长任务锁，
否则浏览器会表现为页面冻结。锁空闲时立即更新；锁繁忙时只保留每个配置项
的最新值，并在当前任务释放锁时统一应用。
返回 True 表示值已经生效，False 表示已进入待更新队列。
"""
# 所有更新都先进入同一队列，再尝试获取配置锁。这样多个页面同时修改同一
# 配置项时，写入队列的先后顺序就是最终顺序，不会出现较早线程在获取锁后
# 把较新线程已经排队的值误删掉。
with _pending_config_lock:
_pending_config_updates[_pending_update_key(config_section, key)] = (
config_section,
key,
copy.deepcopy(value),
)
acquired = _config_save_lock.acquire(blocking=False)
if not acquired:
# 调用方通常会在本次 Streamlit rerun 末尾请求保存，但不能依赖这一步
# 一定执行。例如页面中途异常或更新恰好发生在任务退出保存阶段时，仍需
# 有后台刷新线程保证排队值最终生效。
_schedule_deferred_config_flush()
return False
try:
_apply_pending_config_updates_locked()
return config_section.get(key, _MISSING) == value
finally:
_config_save_lock.release()
def delete_config_nonblocking(config_section, key):
"""
非阻塞删除 WebUI 配置项。
“使用默认值”需要真正移除配置项，而不是写入空字符串。视频任务占用配置
锁时，删除意图会覆盖同一配置项之前排队的更新，并在任务结束后执行。
"""
with _pending_config_lock:
_pending_config_updates[_pending_update_key(config_section, key)] = (
config_section,
key,
_DELETE,
)
acquired = _config_save_lock.acquire(blocking=False)
if not acquired:
_schedule_deferred_config_flush()
return False
try:
_apply_pending_config_updates_locked()
return key not in config_section
finally:
_config_save_lock.release()
def _apply_pending_config_updates_locked():
"""在持有配置写锁时应用 WebUI 暂存的最新配置值。"""
with _pending_config_lock:
updates = list(_pending_config_updates.values())
_pending_config_updates.clear()
# 应用配置时继续持有待更新锁。读取“当前值 + 待更新值”快照的线程由此
# 只能看到应用前或应用后的完整状态，不会读到只更新了一半的配置集合。
for config_section, key, value in updates:
if value is _DELETE:
config_section.pop(key, None)
else:
config_section[key] = value
return bool(updates)
def snapshot_config_with_pending(config_section):
"""
返回配置分区的有效快照，并合并尚未应用的 WebUI 更新。
视频任务持锁期间不能改写全局配置，但用户仍可准备下一条内容。LLM 请求
使用这个快照后，界面中刚选择的 Provider、模型和密钥会参与新请求，同时
不会改变正在执行的视频任务。
"""
with _pending_config_lock:
snapshot = dict(config_section)
section_id = id(config_section)
for (pending_section_id, key), (_, _, value) in _pending_config_updates.items():
if pending_section_id != section_id:
continue
if value is _DELETE:
snapshot.pop(key, None)
else:
snapshot[key] = copy.deepcopy(value)
return snapshot
def _flush_pending_config_locked(*, suppress_save_errors):
"""在持有配置写锁时应用并保存当前所有待处理配置。"""
global _pending_config_save_requested
updates_applied = _apply_pending_config_updates_locked()
with _pending_config_lock:
save_requested = _pending_config_save_requested
_pending_config_save_requested = False
if not updates_applied and not save_requested:
return True
try:
save_config()
return True
except Exception as exc:
# 内存中的配置已经成功应用，保存失败时只保留待保存标记。视频任务不应
# 因配置文件暂时不可写而被改判失败；下一次页面交互会再次触发保存。
with _pending_config_lock:
_pending_config_save_requested = True
if not suppress_save_errors:
raise
logger.exception(f"failed to save deferred runtime config: {exc}")
return False
def _run_deferred_config_flush():
"""等待长任务释放配置锁，并可靠清空期间积累的配置更新。"""
global _pending_config_flush_scheduled
while True:
with _config_save_lock:
flush_succeeded = _flush_pending_config_locked(
suppress_save_errors=True
)
with _pending_config_lock:
has_pending_work = bool(
_pending_config_updates or _pending_config_save_requested
)
if not flush_succeeded or not has_pending_work:
_pending_config_flush_scheduled = False
return
def _schedule_deferred_config_flush():
"""保证同一时间最多只有一个后台线程等待刷新配置。"""
global _pending_config_flush_scheduled
with _pending_config_lock:
if _pending_config_flush_scheduled:
return
_pending_config_flush_scheduled = True
try:
threading.Thread(
target=_run_deferred_config_flush,
name="mpt-config-flush",
daemon=True,
).start()
except Exception:
# No worker owns this reservation when construction/start fails.
# Leave queued updates and the save request intact so a later call
# can schedule another flush rather than silently waiting forever.
with _pending_config_lock:
_pending_config_flush_scheduled = False
raise
def try_save_config():
"""
非阻塞保存 WebUI 配置，锁繁忙时交由当前长任务结束后保存。
普通 API、CLI 和维护脚本仍可调用 ``save_config`` 获得原来的阻塞写入语义；
只有 Streamlit rerun 使用本函数，避免页面为等待视频任务而长时间无响应。
"""
global _pending_config_save_requested
with _pending_config_lock:
_pending_config_save_requested = True
acquired = _config_save_lock.acquire(blocking=False)
if not acquired:
_schedule_deferred_config_flush()
return False
try:
return _flush_pending_config_locked(suppress_save_errors=False)
finally:
_config_save_lock.release()
@contextmanager
def runtime_config_lock():
"""
在一次依赖全局配置的完整操作期间阻止其它 WebUI 会话改写配置。
当前项目默认绑定本地回环地址，配置仍然是单用户全局配置。这个轻量锁主要
保护生成、试听等长操作，避免另一个标签页在操作中途切换 Provider 或密钥。
"""
with _config_save_lock:
# 如果上一个短操作释放锁时后台刷新线程尚未获得调度，新任务必须在读取
# Provider、密钥等全局配置前先应用队列，不能继续使用旧配置执行整条流水线。
_flush_pending_config_locked(suppress_save_errors=True)
try:
yield
finally:
_flush_pending_config_locked(suppress_save_errors=True)
@contextmanager
def try_runtime_config_lock():
"""
尝试获取运行期配置锁，并立即返回是否成功。
WebUI 试听属于用户主动触发的短操作，不应在后台视频任务持锁时等待数分钟。
调用方可以在未获取锁时就近提示用户稍后重试；成功获取后仍能保证试听期间
Provider、密钥和模型配置不会被其它会话修改。
"""
acquired = _config_save_lock.acquire(blocking=False)
try:
if acquired:
_flush_pending_config_locked(suppress_save_errors=True)
yield acquired
finally:
if acquired:
_flush_pending_config_locked(suppress_save_errors=True)
_config_save_lock.release()
def is_running_in_container(
dockerenv_path: str = "/.dockerenv",
containerenv_path: str = "/run/.containerenv",
cgroup_path: str = "/proc/1/cgroup",
) -> bool:
"""
判断当前进程是否运行在容器内。
这个判断主要用于 Ollama 默认地址选择：
- 普通本机运行时，`localhost` 指向用户机器本身；
- Docker 容器内，`localhost` 指向容器自己，访问宿主机 Ollama
通常需要使用 `host.docker.internal`。
不能只判断 `/proc/1/cgroup` 是否存在，因为普通 Linux 也会有这个文件。
这里只在检测到明确的容器标记时返回 True，避免误伤非 Docker Linux 用户。
参数保留为可注入路径，便于单元测试覆盖不同运行环境。
"""
if os.path.isfile(dockerenv_path) or os.path.isfile(containerenv_path):
return True
try:
with open(cgroup_path, mode="r", encoding="utf-8") as fp:
cgroup_content = fp.read().lower()
except OSError:
return False
return any(marker in cgroup_content for marker in _CONTAINER_CGROUP_MARKERS)
def _can_resolve_hostname(hostname: str) -> bool:
try:
socket.gethostbyname(hostname)
except OSError:
return False
return True
def _decode_linux_route_gateway(hex_gateway: str) -> str:
# /proc/net/route 里的 Gateway 是 16 进制小端序，例如 010011AC 表示
# 172.17.0.1。这里单独解析，是为了在原生 Linux Docker 没有
# host.docker.internal DNS 记录时，还能尝试访问容器默认网关上的宿主机。
if len(hex_gateway) != 8:
raise ValueError("invalid gateway length")
octets = [
str(int(hex_gateway[index : index + 2], 16)) for index in range(6, -1, -2)
]
return ".".join(octets)
def get_container_default_gateway_ip(route_path: str = "/proc/net/route") -> str:
"""
读取 Linux 容器里的默认网关 IP。
Docker Desktop 通常提供 `host.docker.internal`，但原生 Linux Docker
默认不一定提供这个 DNS 名称。默认网关通常可以作为访问宿主机服务的
兜底地址；如果用户的 Ollama 只监听 127.0.0.1，则仍需要用户让
Ollama 监听宿主机网卡或手动配置 `ollama_base_url`。
"""
try:
with open(route_path, mode="r", encoding="utf-8") as fp:
route_lines = fp.readlines()
except OSError:
return ""
for line in route_lines[1:]:
fields = line.strip().split()
if len(fields) < 3:
continue
destination = fields[1]
gateway = fields[2]
if destination != "00000000" or gateway == "00000000":
continue
try:
return _decode_linux_route_gateway(gateway)
except ValueError:
logger.warning(f"invalid container gateway route entry: {line.strip()}")
return ""
return ""
def get_default_ollama_base_url() -> str:
"""
返回 Ollama 的默认 OpenAI-compatible base_url。
用户显式配置 `ollama_base_url` 时不会走这里；这里只处理“未配置时的
最佳默认值”。容器内默认指向宿主机，普通本机运行默认指向 localhost。
"""
if not is_running_in_container():
return "http://localhost:11434/v1"
if _can_resolve_hostname(_DOCKER_HOST_GATEWAY_NAME):
return f"http://{_DOCKER_HOST_GATEWAY_NAME}:11434/v1"
gateway_ip = get_container_default_gateway_ip()
if gateway_ip:
logger.info(
"host.docker.internal is not resolvable, fallback to container "
f"default gateway for Ollama: {gateway_ip}"
)
return f"http://{gateway_ip}:11434/v1"
logger.warning(
"failed to resolve host.docker.internal and container default gateway; "
"fallback to host.docker.internal for Ollama"
)
return f"http://{_DOCKER_HOST_GATEWAY_NAME}:11434/v1"
def _load_toml_config(config_path: str):
"""
加载 TOML，并兼容 Windows 编辑器可能写入的重复 UTF-8 BOM。
``utf-8-sig`` 只会移除文件开头的一个 BOM。部分 Windows 编辑器或
解压、保存流程可能再次写入 BOM，导致第二个不可见字符进入 TOML
解析器并在第一行报错。这里仅在标准解析失败后做一次只读归一化，
不回写原文件，避免意外覆盖用户已经填写的 API Key。
"""
try:
return toml.load(config_path)
except (toml.TomlDecodeError, UnicodeDecodeError) as exc:
logger.warning(
"load config failed, retry with UTF-8 BOM compatibility: "
f"path={config_path}, error={type(exc).__name__}: {exc}"
)
try:
with open(config_path, mode="r", encoding="utf-8-sig") as fp:
config_content = fp.read()
normalized_content = config_content.lstrip(_UTF8_BOM)
removed_bom_count = len(config_content) - len(normalized_content)
if removed_bom_count:
logger.warning(
"removed repeated UTF-8 BOM characters while loading config: "
f"path={config_path}, count={removed_bom_count}"
)
return toml.loads(normalized_content)
except (toml.TomlDecodeError, UnicodeDecodeError) as exc:
logger.error(
"config file is not valid TOML after UTF-8 BOM normalization: "
f"path={config_path}, error={type(exc).__name__}: {exc}"
)
raise
def _initialize_config_from_example(example_file):
"""Publish a complete first-boot config without replacing another writer."""
fd, temp_path = tempfile.mkstemp(
prefix=".config-init-", suffix=".toml.tmp",
dir=os.path.dirname(os.path.abspath(config_file)),
)
try:
with os.fdopen(fd, "wb") as destination:
with open(example_file, "rb") as source:
shutil.copyfileobj(source, destination)
destination.flush()
os.fsync(destination.fileno())
try:
if os.name == "nt":
# Windows rename refuses an existing destination. POSIX rename
# overwrites it, so use an exclusive hard-link publication there.
os.rename(temp_path, config_file)
else:
os.link(temp_path, config_file)
except FileExistsError:
return False
except OSError as exc:
raise OSError(
exc.errno,
"cannot safely initialize config.toml; copy config.example.toml "
"to config.toml before starting the application",
) from exc
return True
finally:
try:
os.unlink(temp_path)
except FileNotFoundError:
# A successful Windows rename consumed the temporary path.
pass
def load_config():
# Docker may create an empty directory when a missing config.toml is bind-mounted.
# Only remove that empty placeholder; a nonempty directory may contain user data.
if os.path.isdir(config_file):
try:
os.rmdir(config_file)
except FileNotFoundError:
# Another initializer already removed this empty bind-mount stub.
pass
except OSError as exc:
# Another initializer may already have replaced the empty Docker
# stub with its complete regular file. Accept only that race, not
# permission failures or a still-present/nonempty directory.
if not (
isinstance(exc, NotADirectoryError) and os.path.isfile(config_file)
):
raise IsADirectoryError(
f"{config_file} is a directory and cannot be used as the config file; "
"move or rename it, then restart"
) from exc
if not os.path.isfile(config_file):
example_file = f"{root_dir}/config.example.toml"
if os.path.isfile(example_file):
if _initialize_config_from_example(example_file):
logger.info("copy config.example.toml to config.toml")
logger.info(f"load config from file: {config_file}")
return _load_toml_config(config_file)
def save_config():
"""
原子保存运行时配置。
Streamlit 的不同会话可能在相近时间触发配置保存。直接覆盖 config.toml 时，
另一个线程可能读取到只写了一部分的 TOML 内容。这里使用进程内可重入锁串行化
保存，并先写入同目录临时文件，再通过 os.replace 原子替换目标文件。
Docker Desktop 单文件 bind mount 会把 config.toml 本身作为挂载点，
Linux 内核不允许通过 rename/replace 替换挂载点，因此会返回 EBUSY。
该场景下只能在锁内原地覆盖文件；其它异常仍然抛出，避免掩盖权限、磁盘
或路径错误。
这仍然保留项目现有的单用户全局配置语义，不额外引入复杂的多用户配置系统；
主要用于避免多标签页或快速 rerun 时损坏配置文件。
"""
with _config_save_lock:
config_to_save = dict(_cfg)
config_to_save["app"] = dict(app)
config_to_save["azure"] = dict(azure)
config_to_save["siliconflow"] = dict(siliconflow)
config_to_save["minimax_tts"] = dict(minimax_tts)
config_to_save["elevenlabs"] = dict(elevenlabs)
config_to_save["chatterbox"] = dict(chatterbox)
config_to_save["kokoro"] = dict(kokoro)
config_to_save["fish_audio"] = dict(fish_audio)
config_to_save["voxcpm"] = dict(voxcpm)
config_to_save["ui"] = dict(ui)
serialized_config = toml.dumps(config_to_save)
# WebUI 完整 rerun 结束时会调用保存。内容没有变化时直接返回，避免每次
# 点击普通控件都产生一次磁盘写入和 fsync。
try:
with open(config_file, mode="r", encoding="utf-8") as f:
if f.read() == serialized_config:
_cfg.clear()
_cfg.update(config_to_save)
return
except (OSError, UnicodeError):
pass
temp_path = ""
try:
fd, temp_path = tempfile.mkstemp(
prefix=".config-",
suffix=".toml.tmp",
dir=root_dir,
)
with os.fdopen(fd, mode="w", encoding="utf-8") as f:
f.write(serialized_config)
f.flush()
os.fsync(f.fileno())
try:
os.replace(temp_path, config_file)
except OSError as exc:
if exc.errno != errno.EBUSY:
raise
logger.warning(
"atomic config replacement is unavailable for the mounted "
f"file, fallback to in-place write: {config_file}"
)
with open(config_file, mode="w", encoding="utf-8") as f:
f.write(serialized_config)
f.flush()
os.fsync(f.fileno())
_cfg.clear()
_cfg.update(config_to_save)
finally:
if temp_path and os.path.exists(temp_path):
os.remove(temp_path)
_cfg = load_config()
app = _SynchronizedConfig(_cfg.get("app", {}))
whisper = _cfg.get("whisper", {})
proxy = _cfg.get("proxy", {})
azure = _SynchronizedConfig(_cfg.get("azure", {}))
siliconflow = _SynchronizedConfig(_cfg.get("siliconflow", {}))
minimax_tts = _SynchronizedConfig(_cfg.get("minimax_tts", {}))
elevenlabs = _SynchronizedConfig(_cfg.get("elevenlabs", {}))
chatterbox = _SynchronizedConfig(_cfg.get("chatterbox", {}))
kokoro = _SynchronizedConfig(_cfg.get("kokoro", {}))
fish_audio = _SynchronizedConfig(_cfg.get("fish_audio", {}))
voxcpm = _SynchronizedConfig(_cfg.get("voxcpm", {}))
ui = _SynchronizedConfig(
_cfg.get(
"ui",
{
"hide_log": False,
},
)
)
hostname = socket.gethostname()
log_level = _cfg.get("log_level", "DEBUG")
listen_host = _cfg.get("listen_host", "0.0.0.0")
listen_port = _cfg.get("listen_port", 8080)
project_name = _cfg.get("project_name", "MoneyPrinterTurbo")
project_description = _cfg.get(
"project_description",
"<a href='https://github.com/harry0703/MoneyPrinterTurbo'>https://github.com/harry0703/MoneyPrinterTurbo</a>",
)
project_version = _cfg.get("project_version", __version__)
reload_debug = False
app["redis_host"] = os.getenv(
"MPT_APP_REDIS_HOST",
os.getenv("REDIS_HOST", app.get("redis_host", "localhost")),
)
ffmpeg_path = app.get("ffmpeg_path", "")
if ffmpeg_path and os.path.isfile(ffmpeg_path):
os.environ["IMAGEIO_FFMPEG_EXE"] = ffmpeg_path
logger.info(f"{project_name} v{project_version}")
```
## /app/controllers/base.py
```py path="/app/controllers/base.py"
import secrets
from typing import Annotated
from uuid import uuid4
from fastapi import Header, Request
from app.config import config
from app.models.exception import HttpException
MAX_TASK_ID_LENGTH = 128
def normalize_task_id(value: object) -> str:
"""Return a log-safe request ID, replacing invalid client input with a UUID."""
if (
not isinstance(value, str)
or not value
or len(value) > MAX_TASK_ID_LENGTH
or not value.isprintable()
):
return str(uuid4())
return value
def get_task_id(request: Request) -> str:
return normalize_task_id(request.headers.get("x-task-id"))
def get_api_key(request: Request):
api_key = request.headers.get("x-api-key")
return api_key
def get_api_key_values(request: Request) -> list[str]:
"""返回请求中全部 API Key Header，保留重复值用于安全校验。"""
# Starlette Headers 提供 getlist()，可以区分代理或客户端发送的重复 Header。
# 单元测试中的轻量 Request 替身只使用普通 dict，因此保留兼容回退。
get_list = getattr(request.headers, "getlist", None)
if callable(get_list):
return [value for value in get_list("x-api-key") if isinstance(value, str)]
api_key = get_api_key(request)
return [api_key] if isinstance(api_key, str) else []
def verify_token(
request: Request,
x_api_key: Annotated[str | None, Header(alias="x-api-key")] = None,
):
"""按配置决定是否校验 API Key。
空 Key 保留现有的本地免认证模式；管理员显式配置非空 Key 后，API
路由和任务产物下载都会要求客户端通过 ``x-api-key`` 请求头提供同一
个值。参数声明同时让 Swagger 展示该请求头，便于受保护环境调试。
"""
configured_key = config.app.get("api_key", "")
if configured_key in (None, ""):
return None
# 配置项必须是字符串。这里拒绝列表、数字等错误类型，避免字符串隐式
# 转换产生难以发现的认证行为；错误信息也不包含实际 Key。
if not isinstance(configured_key, str):
raise HttpException(
task_id=get_task_id(request),
status_code=500,
message="API authentication is misconfigured",
)
# FastAPI 参数用于在 OpenAPI 中声明 x-api-key；实际校验始终读取 Request，
# 才能识别同名 Header 被重复发送的情况。普通客户端和反向代理对重复 Header
# 的取值顺序可能不同，因此必须拒绝，而不能隐式采用第一个或最后一个值。
token_values = get_api_key_values(request)
if not token_values and isinstance(x_api_key, str):
token_values = [x_api_key]
if len(token_values) != 1:
raise HttpException(
task_id=get_task_id(request),
status_code=401,
message="invalid API key",
)
# compare_digest 对 str 只支持 ASCII。请求 Header 属于不可信输入，攻击者
# 可以发送 Latin-1 字符触发 TypeError。统一编码为 UTF-8 bytes 后既保留
# 恒定时间比较，也支持 TOML 中合法的 Unicode Key。
token = token_values[0]
if not secrets.compare_digest(
token.encode("utf-8"), configured_key.encode("utf-8")
):
raise HttpException(
task_id=get_task_id(request),
status_code=401,
message="invalid API key",
)
return None
```
## /app/controllers/manager/base_manager.py
```py path="/app/controllers/manager/base_manager.py"
import math
import threading
from typing import Any, Callable, Dict
from loguru import logger
class TaskQueueFullError(ValueError):
pass
def _coerce_task_limit(value: Any, name: str) -> int:
"""
把配置里的并发 / 排队上限解析成整数。
TOML 的数值既可能写成 `max_concurrent_tasks = 5`，也可能被写成 `"5"`。字符串
会一路传到 `add_task` 的比较运算里，变成不指出配置键名的 TypeError，而且
`max_queued_tasks` 写错时只在并发名额用尽后才触发。`app/services/webui_task.py`
对同一个 `max_queued_tasks` 键已经做了同样的收敛，这里把它提到唯一的构造入口。
解析不出整数的写法一律报出配置键名：`0.5` 会被 `int()` 静默截断成 0（任务只入队、
没有 worker 执行），TOML 允许的 `inf` / `-inf` / `nan` 也不是有效上限。
0 与负数保持原样：前者表示暂时不执行任务，后者让请求直接进入排队分支。
"""
if isinstance(value, bool):
# bool 是 int 的子类，但 `true` 显然不是用户想要的上限。
raise ValueError(f"{name} must be an integer, got {value!r}")
if isinstance(value, float):
# 非有限值不是上限，小数则会在截断后退化成「0：只排队不执行」的语义。
if not math.isfinite(value):
raise ValueError(f"{name} must be a finite integer, got {value!r}")
if not value.is_integer():
raise ValueError(f"{name} must be an integer, got {value!r}")
return int(value)
try:
return int(value)
except (TypeError, ValueError, OverflowError):
raise ValueError(f"{name} must be an integer, got {value!r}") from None
class TaskManager:
def __init__(self, max_concurrent_tasks: int, max_queued_tasks: int = 100):
self.max_concurrent_tasks = _coerce_task_limit(
max_concurrent_tasks, "max_concurrent_tasks"
)
self.max_queued_tasks = _coerce_task_limit(max_queued_tasks, "max_queued_tasks")
self.current_tasks = 0
self.lock = threading.Lock()
self.queue = self.create_queue()
def create_queue(self):
raise NotImplementedError()
def add_task(self, func: Callable, *args: Any, **kwargs: Any):
with self.lock:
if self.current_tasks < self.max_concurrent_tasks:
logger.info(
f"add task: {func.__name__}, current_tasks: {self.current_tasks}"
)
# 在线程启动前先预占并发名额。原实现在线程内部递增，连续请求
# 可能都在子线程获得锁之前看到 current_tasks=0，从而突破并发
# 上限。启动失败时回滚名额，让后续请求仍可正常调度。
self.current_tasks += 1
try:
self.execute_task(func, *args, **kwargs)
except Exception:
self.current_tasks -= 1
raise
else:
queue_size = self.queue_size()
# 并发数已满时才进入排队。队列必须有上限，否则匿名接口可以持续
# 堆积任务对象和请求参数，最终造成内存耗尽或第三方 API 成本失控。
if queue_size >= self.max_queued_tasks:
logger.warning(
f"reject task: {func.__name__}, queue_size: {queue_size}, "
f"max_queued_tasks: {self.max_queued_tasks}"
)
raise TaskQueueFullError("task queue is full, please try again later")
logger.info(
f"enqueue task: {func.__name__}, current_tasks: {self.current_tasks}, "
f"queue_size: {queue_size}"
)
self.enqueue_new_task({"func": func, "args": args, "kwargs": kwargs})
def execute_task(self, func: Callable, *args: Any, **kwargs: Any):
thread = threading.Thread(
target=self.run_task, args=(func, *args), kwargs=kwargs
)
thread.start()
def run_task(self, func: Callable, *args: Any, **kwargs: Any):
try:
func(*args, **kwargs) # call the function here, passing *args and **kwargs.
finally:
self.task_done()
def check_queue(self):
with self.lock:
self._check_queue_locked()
def _check_queue_locked(self):
"""Dispatch one queued task while the caller holds ``self.lock``."""
if (
self.current_tasks < self.max_concurrent_tasks
and not self.is_queue_empty()
):
task_info = self.dequeue()
if task_info is None:
# dequeue() may skip and discard queue entries that no longer
# pass current validation (see RedisTaskManager.dequeue) and
# return None once nothing usable is left, even though
# is_queue_empty() was False a moment earlier.
return
func = task_info["func"]
args = task_info.get("args", ())
kwargs = task_info.get("kwargs", {})
# Reserve the freed slot before another add_task can claim it.
self.current_tasks += 1
try:
self.execute_task(func, *args, **kwargs)
except Exception:
self.current_tasks -= 1
self.enqueue(task_info)
raise
def task_done(self):
with self.lock:
self.current_tasks -= 1
self._check_queue_locked()
def enqueue(self, task: Dict):
raise NotImplementedError()
def enqueue_new_task(self, task: Dict):
"""Admit new work; shared queues may need their own atomic limit check."""
self.enqueue(task)
def dequeue(self):
raise NotImplementedError()
def is_queue_empty(self):
raise NotImplementedError()
def queue_size(self):
raise NotImplementedError()
```
## /app/controllers/manager/memory_manager.py
```py path="/app/controllers/manager/memory_manager.py"
from queue import Queue
from typing import Dict
from app.controllers.manager.base_manager import TaskManager
class InMemoryTaskManager(TaskManager):
def create_queue(self):
return Queue(maxsize=self.max_queued_tasks)
def enqueue(self, task: Dict):
self.queue.put(task)
def dequeue(self):
return self.queue.get()
def is_queue_empty(self):
return self.queue.empty()
def queue_size(self):
return self.queue.qsize()
```
## /app/controllers/manager/redis_manager.py
```py path="/app/controllers/manager/redis_manager.py"
import inspect
import json
from typing import Dict
import redis
from loguru import logger
from pydantic import ValidationError
from app.controllers.manager.base_manager import TaskManager, TaskQueueFullError
from app.models import const
from app.models.schema import AudioRequest, SubtitleRequest, VideoParams
from app.services import state as sm
from app.services import task as tm
FUNC_MAP = {
"start": tm.start,
# 'start_test': tm.start_test
}
_ADMIT_QUEUED_TASK_SCRIPT = """
if redis.call("LLEN", KEYS[1]) >= tonumber(ARGV[1]) then
return 0
end
redis.call("RPUSH", KEYS[1], ARGV[2])
return 1
"""
class RedisTaskManager(TaskManager):
def __init__(
self,
max_concurrent_tasks: int,
redis_url: str,
max_queued_tasks: int = 100,
):
self.redis_client = redis.Redis.from_url(redis_url)
super().__init__(max_concurrent_tasks, max_queued_tasks=max_queued_tasks)
def create_queue(self):
return "task_queue"
def resume_queued_tasks(self):
"""Dispatch persisted queue entries when a new API process starts.
Queued Redis items survive a restart, but no worker calls check_queue
until an active task finishes. Fill each available local worker slot so
an idle process does not leave old requests stuck in processing.
"""
for _ in range(max(0, self.max_concurrent_tasks - self.current_tasks)):
self.check_queue()
def enqueue(self, task: Dict):
self.redis_client.rpush(self.queue, self._serialize_task(task))
def enqueue_new_task(self, task: Dict):
# Manager locks are process-local. Keep capacity validation and admission
# in one Redis operation so other API workers cannot claim the same slot.
admitted = self.redis_client.eval(
_ADMIT_QUEUED_TASK_SCRIPT,
1,
self.queue,
self.max_queued_tasks,
self._serialize_task(task),
)
if not admitted:
raise TaskQueueFullError("task queue is full, please try again later")
@staticmethod
def _serialize_task(task: Dict) -> str:
task_with_serializable_params = task.copy()
# task.copy() 只复制最外层字典；如果直接改写嵌套 kwargs，会把调用方
# 持有的 VideoParams 同步替换成 dict。后续日志或重试仍可能读取原任务，
# 因此这里单独复制 kwargs，确保序列化过程没有意外副作用。
task_kwargs = task.get("kwargs", {})
task_with_serializable_params["kwargs"] = task_kwargs.copy()
if "params" in task_kwargs and isinstance(
task_kwargs["params"], (VideoParams, AudioRequest, SubtitleRequest)
):
task_with_serializable_params["kwargs"]["params"] = task_kwargs[
"params"
].model_dump(warnings=False)
# 将函数对象转换为其名称
task_with_serializable_params["func"] = task["func"].__name__
return json.dumps(task_with_serializable_params)
def dequeue(self):
# 循环而非单次弹出：某个任务在入队时可能满足当时的校验规则，但校验规则与
# FUNC_MAP 成员会随部署变化（例如 VideoParams 新增 ge=1 约束、某个入口
# 函数被移除），队列里因此可能残留按旧 schema 写入、或已无法解析的条目。
# lpop 是破坏性操作，一旦弹出就不能放回原位；这条任务已经从队列中永久
# 移除了，不能再假装它还在。与其让异常从这里往上抛（check_queue 持锁调用
# 本方法，异常会顺着 task_done → run_task 的 finally 把工作线程带崩；此后
# 没有任务在跑，就再也不会有人调用 check_queue，队列里后面的任务会永久
# 停在 processing），不如原地丢弃并继续尝试下一条，把"拿到一条可用任务
# 或者队列确实空了"这个约定维持住。
while True:
task_json = self.redis_client.lpop(self.queue)
# 只有 lpop 什么都没弹出来才代表队列空了。空字符串（或空 bytes）同样
# 是一条不可用条目，它后面可能还排着可用的任务，所以要走下面的丢弃
# 路径，而不是当成"队列结束"直接返回。
if task_json is None:
return None
task_info = None
try:
task_info = json.loads(task_json)
# 将函数名称转换回函数对象。名称缺失、或已不在 FUNC_MAP 中时不能
# 直接索引，否则 KeyError 会绕过下面针对 params 的丢弃策略。
task_info["func"] = FUNC_MAP[task_info["func"]]
task_kwargs = task_info["kwargs"]
if not isinstance(task_kwargs, dict):
raise ValueError("queued task has no keyword argument mapping")
# args 整体缺失时沿用 check_queue 的默认值；写成 null 或其它不是
# 数组的形态则会让 check_queue 展开 `*args` 时抛 TypeError，那里
# 会把条目重新入队并让异常逃出工作线程，必须在这里先拦下。
if not isinstance(task_info.get("args", []), list):
raise ValueError("queued task positional arguments are not a list")
# A persisted request can outlive a callable's signature. Fail
# it before dispatch: Python argument-binding errors happen
# before start() enters its pipeline failure handler.
inspect.signature(task_info["func"]).bind(
*task_info.get("args", []), **task_kwargs
)
except (TypeError, ValueError, KeyError) as e:
logger.error(f"dropping unusable queued task: {e}")
# 与下面的 params 校验失败路径一致：只要能读出可用的 task_id，就把
# 这条已经永久离开队列的任务收敛为失败，否则 API/WebUI 会一直显示
# 它在 processing。payload 本身没法解析、或 task_id 不是字符串
# （例如 JSON 数组）时则没有可回写的记录，只能丢弃 —— 把非字符串
# 直接交给 patch_task 会让 redis 抛 DataError，反过来打断丢弃循环。
stale_kwargs = (
task_info.get("kwargs") if isinstance(task_info, dict) else None
)
task_id = (
stale_kwargs.get("task_id")
if isinstance(stale_kwargs, dict)
else None
)
if isinstance(task_id, str) and task_id:
sm.state.patch_task(
task_id,
state=const.TASK_STATE_FAILED,
failed_stage="dequeue",
error=f"discarded stale queued task: {e}",
)
continue
if "params" in task_kwargs and isinstance(task_kwargs["params"], dict):
try:
params_model = VideoParams
# /audio and /subtitle use short-form request models without
# video_subject. Preserve those schemas at the queue boundary;
# full VideoParams can also stop at either of these stages.
if "video_subject" not in task_kwargs["params"]:
if task_kwargs.get("stop_at") == "audio":
params_model = AudioRequest
elif task_kwargs.get("stop_at") == "subtitle":
params_model = SubtitleRequest
task_kwargs["params"] = params_model(**task_kwargs["params"])
except ValidationError as e:
logger.error(
"dropping queued task with params that fail current "
f"request model validation (queued under an older, more "
f"permissive schema, or corrupted): {e}"
)
# 任务状态记录在入队前就已创建，且默认是 processing；如果只是
# 丢弃这条队列项而不动状态记录，API/WebUI 会一直显示任务在
# 运行，永远不会变成失败。用 patch_task 而不是 update_task，
# 这样如果用户已经删除了这个任务，我们不会又把它建回来。
# task_id 不是字符串时同上：没有可回写的记录，跳过状态更新。
task_id = task_kwargs.get("task_id")
if isinstance(task_id, str) and task_id:
sm.state.patch_task(
task_id,
state=const.TASK_STATE_FAILED,
failed_stage="dequeue",
error=f"discarded stale queued task: {e}",
)
continue
return task_info
def is_queue_empty(self):
return self.redis_client.llen(self.queue) == 0
def queue_size(self):
return self.redis_client.llen(self.queue)
```
## /app/controllers/ping.py
```py path="/app/controllers/ping.py"
from fastapi import APIRouter, Request
router = APIRouter()
@router.get(
"/ping",
tags=["Health Check"],
description="检查服务可用性",
response_description="pong",
)
def ping(request: Request) -> str:
return "pong"
```
## /app/controllers/v1/base.py
```py path="/app/controllers/v1/base.py"
from fastapi import APIRouter
def new_router(dependencies=None):
router = APIRouter()
router.tags = ["V1"]
router.prefix = "/api/v1"
# 将认证依赖项应用于所有路由
if dependencies:
router.dependencies = dependencies
return router
```
## /app/controllers/v1/llm.py
```py path="/app/controllers/v1/llm.py"
from fastapi import Depends, Request
from app.controllers import base
from app.controllers.v1.base import new_router
from app.models.exception import HttpException
from app.models.schema import (
VideoScriptRequest,
VideoScriptResponse,
VideoSocialMetadataRequest,
VideoSocialMetadataResponse,
VideoTermsRequest,
VideoTermsResponse,
)
from app.services import llm
from app.utils import utils
# LLM 接口与视频接口共用同一鉴权规则，避免新增端点时遗漏保护。
# api_key 为空时 verify_token 直接放行，不改变默认本地使用体验。
router = new_router(dependencies=[Depends(base.verify_token)])
@router.post(
"/scripts",
response_model=VideoScriptResponse,
summary="Create a script for the video",
)
def generate_video_script(request: Request, body: VideoScriptRequest):
video_script = llm.generate_script(
video_subject=body.video_subject,
language=body.video_language,
paragraph_number=body.paragraph_number,
video_script_prompt=body.video_script_prompt,
custom_system_prompt=body.custom_system_prompt,
)
if not video_script or video_script.startswith("Error: "):
raise HttpException(
task_id=base.get_task_id(request), status_code=503,
message="script generation is unavailable",
)
response = {"video_script": video_script}
return utils.get_response(200, response)
@router.post(
"/terms",
response_model=VideoTermsResponse,
summary="Generate video terms based on the video script",
)
def generate_video_terms(request: Request, body: VideoTermsRequest):
video_terms = llm.generate_terms(
video_subject=body.video_subject,
video_script=body.video_script,
amount=body.amount,
match_script_order=body.match_materials_to_script,
)
if not video_terms:
raise HttpException(
task_id=base.get_task_id(request), status_code=503,
message="search term generation is unavailable",
)
response = {"video_terms": video_terms}
return utils.get_response(200, response)
@router.post(
"/social-metadata",
response_model=VideoSocialMetadataResponse,
summary="Generate social publishing metadata",
)
def generate_video_social_metadata(
request: Request, body: VideoSocialMetadataRequest
):
metadata = llm.generate_social_metadata(
video_subject=body.video_subject,
video_script=body.video_script,
language=body.language,
platform=body.platform,
)
return utils.get_response(200, metadata)
```
## /app/controllers/v1/video.py
```py path="/app/controllers/v1/video.py"
import mimetypes
import os
import pathlib
import shutil
from typing import Union
from urllib.parse import quote
from fastapi import BackgroundTasks, Depends, Path, Query, Request, UploadFile
from fastapi.params import File
from fastapi.responses import FileResponse, StreamingResponse
from loguru import logger
from starlette.background import BackgroundTask
from app.config import config
from app.controllers import base
from app.controllers.manager.base_manager import TaskQueueFullError
from app.controllers.manager.memory_manager import InMemoryTaskManager
from app.controllers.manager.redis_manager import RedisTaskManager
from app.controllers.v1.base import new_router
from app.models.exception import HttpException
from app.models.schema import (
AudioRequest,
BgmRetrieveResponse,
BgmUploadResponse,
SubtitleRequest,
TaskDeletionResponse,
TaskListResponse,
TaskQueryRequest,
TaskQueryResponse,
TaskResponse,
TaskVideoRequest,
VideoMaterialUploadResponse,
VideoMaterialRetrieveResponse
)
from app.services import bgm as bgm_service
from app.services import material_upload as material_upload_service
from app.services import state as sm
from app.services import task as tm
from app.utils import file_security, utils
# 统一在 V1 视频路由入口执行鉴权。verify_token 会在 api_key 为空时
# 保留现有免认证行为，只有管理员显式配置后才会影响客户端。
router = new_router(dependencies=[Depends(base.verify_token)])
_enable_redis = config.app.get("enable_redis", False)
_redis_host = config.app.get("redis_host", "localhost")
_redis_port = config.app.get("redis_port", 6379)
_redis_db = config.app.get("redis_db", 0)
_redis_password = config.app.get("redis_password", None)
_max_concurrent_tasks = config.app.get("max_concurrent_tasks", 5)
_max_queued_tasks = config.app.get("max_queued_tasks", 100)
def _build_redis_url(host: str, port: int, db: int, password: str | None) -> str:
# Passwords are URL userinfo. Escape reserved characters so Redis receives
# the exact configured secret rather than parsing part of it as a host,
# port, path, query, or fragment.
auth = f":{quote(password, safe='')}@" if password else ""
# URL authorities require brackets around IPv6 literals. RedisState also
# accepts a plain IPv6 host, so keep the queue client compatible with it.
url_host = f"[{host}]" if ":" in host and not host.startswith("[") else host
return f"redis://{auth}{url_host}:{port}/{db}"
redis_url = _build_redis_url(_redis_host, _redis_port, _redis_db, _redis_password)
# 根据配置选择合适的任务管理器
if _enable_redis:
task_manager = RedisTaskManager(
max_concurrent_tasks=_max_concurrent_tasks,
redis_url=redis_url,
max_queued_tasks=_max_queued_tasks,
)
else:
task_manager = InMemoryTaskManager(
max_concurrent_tasks=_max_concurrent_tasks,
max_queued_tasks=_max_queued_tasks,
)
def _sanitize_upload_filename(filename: str, request_id: str) -> str:
# 浏览器或客户端有时会附带目录信息，甚至可能夹带 ../ 这类穿越片段。
# 这里只保留纯文件名，避免上传接口把文件写到目标目录之外。
normalized_name = (filename or "").replace("\\", "/").split("/")[-1].strip()
if not normalized_name or normalized_name in {".", ".."}:
raise HttpException(
task_id=request_id,
status_code=400,
message=f"{request_id}: invalid filename",
)
return normalized_name
def _resolve_path_within_directory(base_dir: str, unsafe_path: str, request_id: str) -> str:
try:
return file_security.resolve_path_within_directory(base_dir, unsafe_path)
except ValueError as exc:
logger.warning(
f"reject unsafe file path, request_id: {request_id}, path: {unsafe_path}, "
f"error: {str(exc)}"
)
raise HttpException(
task_id=request_id,
status_code=404 if str(exc) == "file does not exist" else 403,
message=f"{request_id}: invalid file path",
)
def _public_task_data(task: dict) -> dict:
"""复制任务状态并移除仅用于服务端进程协调的内部字段。"""
public_task = dict(task)
public_task.pop("cross_post_owner", None)
return public_task
def _task_file_to_uri(file: str, endpoint: str, task_dir: str, request_id: str) -> str:
if not isinstance(file, str):
return file
if file.startswith(("http://", "https://")):
return file
try:
resolved_path = file_security.resolve_path_within_directory(task_dir, file)
except ValueError as exc:
# 任务状态理论上只应保存任务目录内的产物路径。这里不再继续拼接 URL，
# 避免把异常路径包装成可访问链接；同时保留原值，便于排查历史脏数据。
logger.warning(
f"skip unsafe task output path, request_id: {request_id}, path: {file}, "
f"error: {str(exc)}"
)
return file
relative_path = os.path.relpath(
resolved_path, os.path.realpath(task_dir)
).replace("\\", "/")
uri_path = f"tasks/{quote(relative_path, safe='/')}"
if endpoint:
return f"{endpoint.rstrip('/')}/{uri_path}"
return f"/{uri_path}"
def _task_response_data(task: dict, endpoint: str, task_dir: str, request_id: str) -> dict:
response_task = _public_task_data(task)
for key in ("videos", "combined_videos"):
if key in task:
response_task[key] = [
_task_file_to_uri(file, endpoint, task_dir, request_id)
for file in task[key]
]
for key in ("audio_file", "subtitle_path"):
if task.get(key):
response_task[key] = _task_file_to_uri(
task[key], endpoint, task_dir, request_id
)
return response_task
def _parse_byte_range(
range_header: str | None, file_size: int, request_id: str
) -> tuple[int, int]:
"""解析单段 HTTP Range，并把无效或越界请求稳定转换成 416。"""
if file_size <= 0:
raise HttpException(
task_id=request_id,
status_code=416,
message=f"{request_id}: requested range is not satisfiable",
)
if not range_header:
return 0, file_size - 1
try:
# 视频播放器这里只需要单段 bytes range。拒绝多段请求可以避免返回体
# 与 Content-Range 不一致，也避免异常字符串落入 int() 产生 500。
if not range_header.startswith("bytes=") or "," in range_header:
raise ValueError("unsupported range format")
start_text, end_text = range_header[6:].split("-", 1)
if not start_text and not end_text:
raise ValueError("empty range")
if not start_text:
suffix_length = int(end_text)
if suffix_length <= 0:
raise ValueError("invalid suffix length")
start = max(file_size - suffix_length, 0)
end = file_size - 1
else:
start = int(start_text)
end = int(end_text) if end_text else file_size - 1
if start < 0 or start >= file_size or end < start:
raise ValueError("range outside file")
end = min(end, file_size - 1)
except (TypeError, ValueError) as exc:
logger.warning(
f"reject invalid video range, request_id: {request_id}, "
f"range: {range_header}, file_size: {file_size}, error: {str(exc)}"
)
raise HttpException(
task_id=request_id,
status_code=416,
message=f"{request_id}: requested range is not satisfiable",
) from exc
return start, end
@router.post("/videos", response_model=TaskResponse, summary="Generate a short video")
def create_video(
background_tasks: BackgroundTasks, request: Request, body: TaskVideoRequest
):
return create_task(request, body, stop_at="video")
@router.post("/subtitle", response_model=TaskResponse, summary="Generate subtitle only")
def create_subtitle(
background_tasks: BackgroundTasks, request: Request, body: SubtitleRequest
):
return create_task(request, body, stop_at="subtitle")
@router.post("/audio", response_model=TaskResponse, summary="Generate audio only")
def create_audio(
background_tasks: BackgroundTasks, request: Request, body: AudioRequest
):
return create_task(request, body, stop_at="audio")
def create_task(
request: Request,
body: Union[TaskVideoRequest, SubtitleRequest, AudioRequest],
stop_at: str,
):
task_id = utils.get_uuid()
request_id = base.get_task_id(request)
try:
if (
stop_at == "video"
and isinstance(body, TaskVideoRequest)
and body.subtitle_enabled
):
tm.video.validate_subtitle_colors(body)
# 字体名可由 API 客户端直接提交，不能等到后台渲染时才发现路径越界。
# 这里与渲染层共用目录边界校验，让非法请求在创建付费任务前返回 400。
file_security.resolve_path_within_directory(
utils.font_dir(), body.font_name or "STHeitiMedium.ttc"
)
task = {
"task_id": task_id,
"request_id": request_id,
"params": body.model_dump(),
}
sm.state.update_task(task_id)
try:
task_manager.add_task(
tm.start, task_id=task_id, params=body, stop_at=stop_at
)
except Exception:
# 状态记录在调度前创建，默认标记为 processing。如果调度器没能
# 接管任务（例如线程启动失败或 Redis 队列不可用），必须回滚该
# 记录，否则 API 和 WebUI 会永久展示一个实际从未运行的任务。
sm.state.delete_task(task_id)
raise
logger.success(f"Task created: {utils.to_json(task)}")
return utils.get_response(200, task)
except TaskQueueFullError as e:
logger.warning(
f"reject task because queue is full, request_id: {request_id}, task_id: {task_id}"
)
raise HttpException(
task_id=task_id, status_code=429, message=f"{request_id}: {str(e)}"
)
except ValueError as e:
raise HttpException(
task_id=task_id, status_code=400, message=f"{request_id}: {str(e)}"
)
@router.get("/tasks", response_model=TaskListResponse, summary="Get all tasks")
def get_all_tasks(
request: Request,
page: int = Query(1, ge=1),
page_size: int = Query(10, ge=1, le=1000),
):
tasks, total = sm.state.get_all_tasks(page, page_size)
request_id = base.get_task_id(request)
endpoint = config.app.get("endpoint", "").rstrip("/")
task_dir = utils.task_dir()
response = {
"tasks": [
_task_response_data(task, endpoint, task_dir, request_id) for task in tasks
],
"total": total,
"page": page,
"page_size": page_size,
}
return utils.get_response(200, response)
@router.get(
"/tasks/{task_id}", response_model=TaskQueryResponse, summary="Query task status"
)
def get_task(
request: Request,
task_id: str = Path(..., description="Task ID"),
query: TaskQueryRequest = Depends(),
):
request_id = base.get_task_id(request)
endpoint = config.app.get("endpoint", "").rstrip("/")
task = sm.state.get_task(task_id)
if task:
task_dir = utils.task_dir()
response_task = _task_response_data(task, endpoint, task_dir, request_id)
return utils.get_response(200, response_task)
raise HttpException(
task_id=task_id, status_code=404, message=f"{request_id}: task not found"
)
@router.delete(
"/tasks/{task_id}",
response_model=TaskDeletionResponse,
summary="Delete a generated short video task",
)
def delete_video(request: Request, task_id: str = Path(..., description="Task ID")):
request_id = base.get_task_id(request)
task = sm.state.get_task(task_id)
if task:
if tm.is_task_busy(task):
logger.warning(
f"refuse to delete busy task, request_id: {request_id}, "
f"task_id: {task_id}, state: {task.get('state')}, "
f"cross_post_state: {task.get('cross_post_state')}"
)
raise HttpException(
task_id=task_id,
status_code=409,
message=f"{request_id}: task is still running",
)
tasks_dir = utils.task_dir()
current_task_dir = os.path.join(tasks_dir, task_id)
if os.path.exists(current_task_dir):
try:
shutil.rmtree(current_task_dir)
except FileNotFoundError:
# Another completed-task deletion may remove this directory
# after our existence check. Only accept an absent root;
# partial deletion and other filesystem errors must keep state.
if os.path.lexists(current_task_dir):
raise
sm.state.delete_task(task_id)
logger.success(f"video deleted: {utils.to_json(task)}")
return utils.get_response(200)
raise HttpException(
task_id=task_id, status_code=404, message=f"{request_id}: task not found"
)
@router.get(
"/musics", response_model=BgmRetrieveResponse, summary="Retrieve local BGM files"
)
def get_bgm_list(request: Request):
bgm_list = []
for file in bgm_service.list_bgm_files():
filename = os.path.basename(file)
try:
size = os.path.getsize(file)
except OSError as exc:
logger.warning(f"skip unavailable background music: name={filename}, error={exc}")
continue
bgm_list.append(
{
"name": filename,
"size": size,
# 只返回文件名，避免把服务器绝对路径暴露给调用方。服务端会
# 在 storage/bgm 和 resource/songs 两个白名单目录中重新解析。
"file": filename,
}
)
response = {"files": bgm_list}
return utils.get_response(200, response)
@router.post(
"/musics",
response_model=BgmUploadResponse,
summary="Upload a background music file",
description=(
"Validate an MP3, M4A, AAC, WAV, FLAC, OGG, OPUS, or WMA file up to "
"30 MB and store it under an immutable UUID filename in storage/bgm."
),
responses={
400: {"description": "The filename, format, size, or audio stream is invalid"},
500: {"description": "FFmpeg validation or persistent storage is unavailable"},
},
)
def upload_bgm_file(request: Request, file: UploadFile = File(...)):
request_id = base.get_task_id(request)
try:
safe_filename = bgm_service.save_bgm_upload(file.filename, file.file)
except bgm_service.BgmUploadError as exc:
# 上传失败通常可以由用户更换文件后恢复，因此记录 request_id 和明确原因，
# 但不输出文件内容或绝对路径，避免日志泄露用户数据。
logger.warning(
f"background music upload rejected: request_id={request_id}, error={str(exc)}"
)
raise HttpException(
task_id=request_id,
status_code=400,
message=f"{request_id}: {str(exc)}",
)
except bgm_service.BgmServiceError as exc:
# 工具链或存储故障属于服务端问题，不能伪装成用户文件错误。日志保留
# request_id 和内部原因，HTTP 响应只返回稳定文案，避免暴露服务器路径。
logger.error(
f"background music upload failed: request_id={request_id}, error={str(exc)}"
)
raise HttpException(
task_id=request_id,
status_code=500,
message=f"{request_id}: background music validation is unavailable",
)
response = {"file": safe_filename}
return utils.get_response(200, response)
@router.get(
"/video_materials", response_model=VideoMaterialRetrieveResponse, summary="Retrieve local video materials"
)
def get_video_materials_list(request: Request):
allowed_suffixes = material_upload_service.SUPPORTED_MATERIAL_EXTENSIONS
local_videos_dir = utils.storage_dir("local_videos", create=True)
video_materials_list = []
with os.scandir(local_videos_dir) as entries:
for entry in entries:
if (
entry.name.startswith(".")
or pathlib.Path(entry.name).suffix.lower() not in allowed_suffixes
):
continue
try:
# Do not follow links outside local_videos or list an upload
# that disappeared while the directory was being scanned.
if not entry.is_file(follow_symlinks=False):
continue
size = entry.stat(follow_symlinks=False).st_size
except OSError as exc:
logger.warning(
f"skip unavailable local material: name={entry.name}, error={exc}"
)
continue
video_materials_list.append(
{"name": entry.name, "size": size, "file": entry.name}
)
# Keep ordered material selection stable across file systems and runs.
video_materials_list.sort(key=lambda item: (item["name"].casefold(), item["name"]))
response = {"files": video_materials_list}
return utils.get_response(200, response)
@router.post(
"/video_materials",
response_model=VideoMaterialUploadResponse,
summary="Upload the video material file to the local videos directory",
)
def upload_video_material_file(request: Request, file: UploadFile = File(...)):
request_id = base.get_task_id(request)
try:
# Keep accepting browser-supplied client paths, but persist an immutable
# UUID storage key so repeated names cannot overwrite queued task inputs.
safe_filename = _sanitize_upload_filename(file.filename, request_id)
stored_filename = material_upload_service.save_material_upload(
safe_filename, file.file
)
except material_upload_service.MaterialUploadError as exc:
logger.warning(
f"local material upload rejected: request_id={request_id}, "
f"error={str(exc)}"
)
raise HttpException(
task_id=request_id,
status_code=400,
message=f"{request_id}: {str(exc)}",
)
except material_upload_service.MaterialServiceError as exc:
logger.error(
f"local material upload failed: request_id={request_id}, "
f"error={str(exc)}"
)
raise HttpException(
task_id=request_id,
status_code=500,
message=f"{request_id}: local material validation is unavailable",
)
response = {"file": stored_filename}
return utils.get_response(200, response)
@router.get("/stream/{file_path:path}")
async def stream_video(request: Request, file_path: str):
request_id = base.get_task_id(request)
tasks_dir = utils.task_dir()
video_path = _resolve_path_within_directory(tasks_dir, file_path, request_id)
range_header = request.headers.get("Range")
# The body is produced after this handler returns. Open now so a task
# deletion or replacement between headers and iteration cannot make the
# stream fail or mismatch the size used for Content-Range.
try:
video_file = open(video_path, "rb")
except FileNotFoundError as exc:
raise HttpException(
task_id=request_id,
status_code=404,
message=f"{request_id}: file no longer exists",
) from exc
try:
video_size = os.fstat(video_file.fileno()).st_size
start, end = _parse_byte_range(range_header, video_size, request_id)
except Exception:
video_file.close()
raise
length = end - start + 1
def file_iterator():
try:
video_file.seek(start, os.SEEK_SET)
remaining = length
while remaining > 0:
data = video_file.read(min(4096, remaining))
if not data:
break
remaining -= len(data)
yield data
finally:
video_file.close()
response = StreamingResponse(
file_iterator(),
media_type="video/mp4",
background=BackgroundTask(video_file.close),
)
response.headers["Accept-Ranges"] = "bytes"
response.headers["Content-Length"] = str(length)
if range_header:
response.headers["Content-Range"] = f"bytes {start}-{end}/{video_size}"
response.status_code = 206 # Partial Content
return response
@router.get("/download/{file_path:path}")
async def download_video(request: Request, file_path: str):
"""
download video
:param request: Request request
:param file_path: video file path, eg: /cd1727ed-3473-42a2-a7da-4faafafec72b/final-1.mp4
:return: video file
"""
request_id = base.get_task_id(request)
tasks_dir = utils.task_dir()
video_path = _resolve_path_within_directory(tasks_dir, file_path, request_id)
file_path = pathlib.Path(video_path)
filename = file_path.name
media_type, _ = mimetypes.guess_type(filename)
return FileResponse(
path=video_path,
filename=filename,
media_type=media_type or "application/octet-stream",
)
```
## /app/models/__init__.py
```py path="/app/models/__init__.py"
```
## /app/models/const.py
```py path="/app/models/const.py"
PUNCTUATIONS = [
"?",
",",
".",
"、",
";",
":",
"!",
"…",
"？",
"，",
"。",
"、",
"；",
"：",
"！",
"...",
# 阿拉伯语常用标点也应作为自然断句点，避免脚本文本和 edge-tts
# 返回的字幕停顿边界不一致，导致后续逐行匹配失败。
"،",
"؛",
"؟",
]
TASK_STATE_FAILED = -1
TASK_STATE_COMPLETE = 1
TASK_STATE_PROCESSING = 4
CROSS_POST_STATE_PENDING = "pending"
CROSS_POST_STATE_PROCESSING = "processing"
CROSS_POST_STATE_COMPLETE = "complete"
CROSS_POST_STATE_FAILED = "failed"
FILE_TYPE_VIDEOS = ["mp4", "mov", "mkv", "webm"]
FILE_TYPE_IMAGES = ["jpg", "jpeg", "png", "bmp"]
```
## /app/models/exception.py
```py path="/app/models/exception.py"
import traceback
from typing import Any
from loguru import logger
class HttpException(Exception):
def __init__(
self, task_id: str, status_code: int, message: str = "", data: Any = None
):
self.message = message
self.status_code = status_code
self.data = data
# Retrieve the exception stack trace information.
tb_str = traceback.format_exc().strip()
if not tb_str or tb_str == "NoneType: None":
msg = f"HttpException: {status_code}, {task_id}, {message}"
else:
msg = f"HttpException: {status_code}, {task_id}, {message}\n{tb_str}"
# 400/401 都是可预期的客户端输入问题。尤其鉴权开启后，公网扫描可能
# 产生大量无效 Key；使用 WARNING 既保留定位信息，也避免污染 ERROR
# 告警。服务端配置错误和其它异常仍保持 ERROR。
if status_code in (400, 401):
logger.warning(msg)
else:
logger.error(msg)
class FileNotFoundException(Exception):
pass
```
## /app/models/llm_provider.py
```py path="/app/models/llm_provider.py"
from dataclasses import dataclass
DEFAULT_LLM_PROVIDER_ID = "moonshot"
@dataclass(frozen=True, slots=True)
class LLMProviderField:
"""描述 Provider 除 API Key、Base URL、模型名之外的额外配置字段。"""
config_suffix: str
label_key: str
required: bool = False
secret: bool = False
default_value: str = ""
@dataclass(frozen=True, slots=True)
class LLMProviderEndpoint:
"""描述同一 Provider 在不同服务区域使用的配套入口和 API 地址。"""
endpoint_id: str
default_label: str
base_url: str
api_key_url: str
model_docs_url: str = ""
@dataclass(frozen=True, slots=True)
class LLMProviderSpec:
"""
LLM Provider 的集中声明。
这里集中保存跨 WebUI、配置加载和服务调用都会使用的稳定元数据，包括默认
展示名称和 locale key，但不保存具体翻译文案，也不实现 API 请求。这样
Provider 的“是什么”由 Registry 维护，“怎么调用”仍由服务层适配器负责。
"""
provider_id: str
default_label: str
adapter: str = "openai_compatible"
api_key_url: str = ""
default_model: str = ""
default_base_url: str = ""
model_docs_url: str = ""
requires_api_key: bool = True
requires_model_name: bool = True
requires_base_url: bool = True
show_api_key: bool = True
show_base_url: bool = True
deprecated_models: tuple[str, ...] = ()
deprecated_base_urls: tuple[str, ...] = ()
extra_fields: tuple[LLMProviderField, ...] = ()
service_endpoints: tuple[LLMProviderEndpoint, ...] = ()
default_service_endpoint_id: str = ""
international_service_endpoint_id: str = ""
@property
def label_key(self) -> str:
return f"llm_provider_label.{self.provider_id}"
@property
def tips_key(self) -> str:
return f"llm_provider_tips.{self.provider_id}"
@property
def endpoint_selector_label_key(self) -> str:
return f"llm_provider_endpoint_selector.{self.provider_id}"
@property
def endpoint_selector_help_key(self) -> str:
return f"llm_provider_endpoint_selector_help.{self.provider_id}"
@property
def authentication_error_key(self) -> str:
return f"llm_provider_authentication_error.{self.provider_id}"
def endpoint_label_key(self, endpoint_id: str) -> str:
return f"llm_provider_endpoint.{self.provider_id}.{endpoint_id}"
def config_key(self, suffix: str) -> str:
return f"{self.provider_id}_{suffix}"
def resolve_model_name(self, configured_model: str | None) -> str:
"""将空值或已废弃的历史默认值统一解析为当前默认模型。"""
model_name = (configured_model or "").strip()
if not model_name or model_name in self.deprecated_models:
return self.default_model
return model_name
def resolve_base_url(self, configured_base_url: str | None) -> str:
"""解析 Base URL，并将已经停用的历史地址迁移到当前默认值。"""
base_url = (configured_base_url or "").strip()
deprecated_urls = {url.rstrip("/") for url in self.deprecated_base_urls}
if not base_url or base_url.rstrip("/") in deprecated_urls:
return self.effective_default_base_url
return base_url
def get_service_endpoint(self, endpoint_id: str) -> LLMProviderEndpoint | None:
"""按稳定 ID 获取服务区域，避免业务逻辑依赖可变化的推广链接。"""
return next(
(
endpoint
for endpoint in self.service_endpoints
if endpoint.endpoint_id == endpoint_id
),
None,
)
@property
def default_service_endpoint(self) -> LLMProviderEndpoint | None:
"""返回 Provider 声明的默认服务区域。"""
return self.get_service_endpoint(self.default_service_endpoint_id)
@property
def international_service_endpoint(self) -> LLMProviderEndpoint | None:
"""返回 Provider 声明的国际服务区域。"""
return self.get_service_endpoint(self.international_service_endpoint_id)
@property
def effective_default_base_url(self) -> str:
"""优先从默认服务区域读取 Base URL，普通 Provider 仍使用原字段。"""
endpoint = self.default_service_endpoint
return endpoint.base_url if endpoint else self.default_base_url
def preferred_service_endpoint(
self, *, prefer_international: bool
) -> LLMProviderEndpoint | None:
"""根据界面区域返回首选入口，缺少国际入口时安全回退默认入口。"""
if prefer_international and self.international_service_endpoint:
return self.international_service_endpoint
return self.default_service_endpoint
def effective_api_key_url(self, *, prefer_international: bool = False) -> str:
"""统一解析 API Key 申请入口，避免 Endpoint Provider 重复维护链接。"""
endpoint = self.preferred_service_endpoint(
prefer_international=prefer_international
)
return endpoint.api_key_url if endpoint else self.api_key_url
def effective_model_docs_url(self, *, prefer_international: bool = False) -> str:
"""统一解析模型列表与文档入口，避免 Endpoint Provider 重复维护链接。"""
endpoint = self.preferred_service_endpoint(
prefer_international=prefer_international
)
return endpoint.model_docs_url if endpoint and endpoint.model_docs_url else self.model_docs_url
def find_service_endpoint(
self, configured_base_url: str | None
) -> LLMProviderEndpoint | None:
"""根据已保存的 Base URL 识别 Provider 的标准服务区域。"""
normalized_url = (configured_base_url or "").strip().rstrip("/")
if not normalized_url:
return None
return next(
(
endpoint
for endpoint in self.service_endpoints
if endpoint.base_url.rstrip("/") == normalized_url
),
None,
)
def select_service_endpoint(
self,
configured_base_url: str | None,
*,
has_api_key: bool,
prefer_international: bool,
) -> LLMProviderEndpoint | None:
"""
选择 WebUI 应展示的标准服务区域。
已明确保存的标准地址优先；未知地址保留为自定义。历史配置可能只有
API Key 而没有 Base URL，这类用户继续使用 Registry 默认区域，避免
升级后因界面语言不同而切换服务。只有全新配置才根据界面语言选择
国际入口。
"""
configured_url = (configured_base_url or "").strip()
if configured_url:
return self.find_service_endpoint(configured_url)
default_endpoint = self.default_service_endpoint
if has_api_key or not prefer_international:
return default_endpoint
return self.preferred_service_endpoint(
prefer_international=prefer_international
)
# 元组顺序就是 WebUI 下拉框顺序。新增普通 OpenAI-compatible Provider 时，
# 通常只需要在这里增加一项并补充 locale；只有协议不同的 Provider 才需要在
# app/services/llm.py 中增加对应 adapter 实现。
LLM_PROVIDER_REGISTRY = (
# 推荐 Provider
LLMProviderSpec(
"moonshot",
"Kimi / Moonshot AI",
default_model="kimi-k3",
service_endpoints=(
LLMProviderEndpoint(
endpoint_id="china",
default_label="China",
base_url="https://api.moonshot.cn/v1",
api_key_url=(
"https://platform.kimi.com?"
"track_id=track-6eec1e56a4494e52adcaebbcbbefce59&"
"aff=moneyprinterturbo"
),
model_docs_url=(
"https://platform.kimi.com/docs/models?"
"track_id=track-6eec1e56a4494e52adcaebbcbbefce59&"
"aff=moneyprinterturbo"
),
),
LLMProviderEndpoint(
endpoint_id="global",
default_label="Global",
base_url="https://api.moonshot.ai/v1",
api_key_url=(
"https://platform.kimi.ai?"
"track_id=track-9e3b711aa2594e378f6fe5b8de718a76&"
"aff=moneyprinterturbo"
),
model_docs_url=(
"https://platform.kimi.ai/docs/models?"
"track_id=track-9e3b711aa2594e378f6fe5b8de718a76&"
"aff=moneyprinterturbo"
),
),
),
default_service_endpoint_id="china",
international_service_endpoint_id="global",
),
# 主流模型原厂与云厂商
LLMProviderSpec(
"openai",
"OpenAI",
api_key_url="https://platform.openai.com/api-keys",
default_model="gpt-5.5",
default_base_url="https://api.openai.com/v1",
),
LLMProviderSpec(
"anthropic",
"Anthropic Claude",
api_key_url="https://platform.claude.com/settings/keys",
default_model="claude-sonnet-5",
default_base_url="https://api.anthropic.com/v1/",
),
LLMProviderSpec(
"gemini",
"Google Gemini",
adapter="gemini",
api_key_url="https://aistudio.google.com/app/apikey",
default_model="gemini-3.1-pro-preview",
requires_base_url=False,
show_base_url=False,
deprecated_models=("gemini-pro", "gemini-1.0-pro"),
),
LLMProviderSpec(
"deepseek",
"DeepSeek",
api_key_url="https://platform.deepseek.com/api_keys",
default_model="deepseek-v4-pro",
default_base_url="https://api.deepseek.com",
),
LLMProviderSpec(
"qwen",
"Alibaba Cloud Qwen",
adapter="qwen",
api_key_url="https://dashscope.console.aliyun.com/apiKey",
default_model="qwen-max",
requires_base_url=False,
show_base_url=False,
),
LLMProviderSpec(
"azure",
"Microsoft Azure OpenAI",
adapter="azure",
api_key_url=(
"https://portal.azure.com/#view/"
"Microsoft_Azure_ProjectOxford/CognitiveServicesHub/~/OpenAI"
),
default_model="gpt-35-turbo",
),
LLMProviderSpec(
"volcengine",
"ByteDance VolcEngine Ark",
api_key_url=(
"https://www.volcengine.com/activity/ai618?utm_campaign=hw&"
"utm_content=hw&utm_medium=devrel_tool_web&utm_source=OWO&"
"utm_term=MoneyPrinterTurbo"
),
default_model="doubao-seed-2-1-turbo-260628",
default_base_url="https://ark.cn-beijing.volces.com/api/v3",
),
LLMProviderSpec(
"grok",
"xAI Grok",
api_key_url="https://console.x.ai/",
default_model="grok-4.3",
default_base_url="https://api.x.ai/v1",
),
LLMProviderSpec(
"minimax",
"MiniMax",
api_key_url="https://platform.minimax.io/",
default_model="MiniMax-M3",
default_base_url="https://api.minimax.io/v1",
),
LLMProviderSpec(
"mimo",
"Xiaomi MiMo",
api_key_url=(
"https://platform.xiaomimimo.com/docs/zh-CN/quick-start/first-api-call"
),
default_model="mimo-v2.5-pro",
default_base_url="https://api.xiaomimimo.com/v1",
),
# 讯飞星辰 MaaS 的按量付费和 Token Plan 使用不同的 OpenAI 兼容地址，
# 两边的 API Key 不能混用。这里默认按量付费，Token Plan 用户在 Base URL
# 中改填 maas-token-api 的地址即可。
LLMProviderSpec(
"iflytek",
"iFlytek Spark (Astron MaaS)",
api_key_url="https://maas.xfyun.cn/",
default_model="spark-x2.5",
default_base_url="https://maas-api.cn-huabei-1.xf-yun.com/v2",
model_docs_url="https://maas.xfyun.cn/modelSquare",
),
# 聚合与统一接入平台
LLMProviderSpec(
"shengsuanyun",
"Shengsuan Cloud",
api_key_url="https://www.shengsuanyun.com/?from=CH_XUQ4OTSK",
default_model="deepseek/deepseek-v4-flash",
default_base_url="https://router.shengsuanyun.com/api/v1",
),
# APIMart 同时提供 `/api/v1` 业务接口和 `/v1` OpenAI 兼容接口。
# 当前 LLM 服务层依赖 OpenAI SDK 直接读取 choices，因此必须使用不带
# code/data 外层包装的 `/v1` 入口，不能照搬异步业务接口的地址。
LLMProviderSpec(
"apimart",
"APIMart",
api_key_url="https://go.apimart.ai/gh-moneyprinterturbo",
default_model="gpt-5.6-terra",
default_base_url="https://api.apimart.ai/v1",
),
LLMProviderSpec(
"cloudflare",
"Cloudflare AI Gateway",
adapter="cloudflare_ai_gateway",
api_key_url="https://dash.cloudflare.com/",
default_model="openai/gpt-4.1-mini",
requires_base_url=False,
show_base_url=False,
deprecated_models=("@cf/meta/llama-3.1-8b-instruct",),
extra_fields=(
LLMProviderField("account_id", "Account ID", required=True),
LLMProviderField(
"gateway_id",
"Gateway ID",
default_value="default",
),
),
),
LLMProviderSpec(
"modelscope",
"Alibaba ModelScope",
adapter="modelscope",
api_key_url=("https://modelscope.cn/docs/model-service/API-Inference/intro"),
default_model="ZhipuAI/GLM-5.2",
default_base_url="https://api-inference.modelscope.cn/v1/",
),
LLMProviderSpec(
"aihubmix",
"AIHubMix",
api_key_url="https://aihubmix.com/",
default_model="gpt-5.4-mini",
default_base_url="https://aihubmix.com/v1",
),
LLMProviderSpec(
"aimlapi",
"AIML API",
api_key_url="https://aimlapi.com/app/keys",
default_model="openai/gpt-5-5",
default_base_url="https://api.aimlapi.com/v1",
),
LLMProviderSpec(
"evolink",
"EvoLink",
api_key_url="https://evolink.ai/dashboard/keys",
default_model="gpt-5.5",
default_base_url="https://direct.evolink.ai/v1",
),
LLMProviderSpec(
"openrouter",
"OpenRouter",
api_key_url="https://openrouter.ai/settings/keys",
default_model="minimax/minimax-m3:free",
default_base_url="https://openrouter.ai/api/v1",
),
LLMProviderSpec(
"api_route",
"API Route",
api_key_url="https://www.api-route.com",
default_model="gpt-5.4-mini",
default_base_url="https://www.api-route.com/v1",
model_docs_url="https://www.api-route.com/pricing",
),
# Fluxion 的 OpenAI 分组使用 /v1 接口，复用现有 Chat Completions 适配器。
# 分组决定可用模型和地址，因此这里只提供默认值，保留用户覆盖配置的能力；
# Anthropic/Gemini 原生分组不能直接使用这个入口，避免协议不匹配。
LLMProviderSpec(
"fluxionai",
"Fluxion AI",
api_key_url=(
"https://fluxionai.space/register?source=github"
"&campaign=moneyprinterturbo&promo=MONEYPRINTERTURBO"
),
default_model="gpt-5.5",
default_base_url="https://fluxionai.space/v1",
model_docs_url="https://fluxionai.space/model-plaza",
),
LLMProviderSpec(
"cheaperinference",
"Cheaper Inference",
api_key_url="https://cheaperinference.com/signup",
default_model="gpt-5.4-mini",
default_base_url="https://api.cheaperinference.com/v1",
model_docs_url="https://cheaperinference.com/#models",
),
LLMProviderSpec(
"requesty",
"Requesty",
api_key_url="https://app.requesty.ai/api-keys",
default_model="openai/gpt-5.4-mini",
default_base_url="https://router.requesty.ai/v1",
model_docs_url="https://www.requesty.ai/models",
),
LLMProviderSpec(
"futureinfra",
"FutureInfra",
api_key_url="https://futureinfra.ai/console/?screen=ai-router",
default_model="openai/gpt-4o-mini",
default_base_url="https://futureinfra.ai/v1/ai",
model_docs_url="https://futureinfra.ai/ai/",
),
LLMProviderSpec(
"yapi",
"Y-API",
api_key_url="https://y-api.bestvirtualgoods.com/app/keys",
default_model="deepseek/deepseek-v4-flash",
default_base_url="https://api.y-api.bestvirtualgoods.com/v1",
model_docs_url="https://y-api.bestvirtualgoods.com/models",
),
LLMProviderSpec(
"opper",
"Opper",
api_key_url="https://platform.opper.ai",
default_model="gpt-5.4-mini",
default_base_url="https://api.opper.ai/v3/compat",
model_docs_url="https://opper.ai/models",
),
# 本地部署与通用网关
LLMProviderSpec(
"ollama",
"Ollama",
requires_api_key=False,
show_api_key=False,
),
# Claude 订阅（Pro / Max / Team）不签发 API Key，凭证只能由 Claude Code
# 官方客户端使用，因此这个 Provider 不走 HTTP 接口，而是调用本机已登录
# 的 claude CLI。模型名留空即沿用 CLI 当前的默认模型。
LLMProviderSpec(
"claude_code",
"Claude Code (Claude subscription)",
adapter="claude_code",
requires_api_key=False,
show_api_key=False,
requires_base_url=False,
show_base_url=False,
requires_model_name=False,
extra_fields=(
LLMProviderField("cli_path", "Claude CLI Path"),
LLMProviderField("timeout", "Timeout (seconds)", default_value="300"),
),
),
LLMProviderSpec(
"oneapi",
"OneAPI",
api_key_url="https://github.com/songquanpeng/one-api",
),
LLMProviderSpec(
"litellm",
"LiteLLM",
adapter="litellm",
default_model="openai/gpt-4o-mini",
requires_api_key=False,
requires_base_url=False,
show_api_key=False,
show_base_url=False,
),
# 其它推理与公共服务
LLMProviderSpec(
"groq",
"Groq",
api_key_url="https://console.groq.com/keys",
default_model="openai/gpt-oss-120b",
default_base_url="https://api.groq.com/openai/v1",
),
LLMProviderSpec(
"pollinations",
"Pollinations AI",
api_key_url="https://enter.pollinations.ai/",
default_model="openai-fast",
default_base_url="https://gen.pollinations.ai/v1",
deprecated_models=("default",),
deprecated_base_urls=("https://text.pollinations.ai/openai",),
),
)
LLM_PROVIDERS = {provider.provider_id: provider for provider in LLM_PROVIDER_REGISTRY}
if len(LLM_PROVIDERS) != len(LLM_PROVIDER_REGISTRY):
raise RuntimeError("duplicate LLM provider id in registry")
def get_llm_provider(provider_id: str) -> LLMProviderSpec | None:
return LLM_PROVIDERS.get((provider_id or "").lower())
def normalize_provider_override(value: str | None, default_value: str | None) -> str:
"""
只保留与 Registry 默认值不同的用户覆盖值。
WebUI 需要把默认值展示在输入框中，但不能因此把默认值固化到 config.toml；
否则后续升级 Registry 默认模型或地址时，旧配置会继续覆盖新默认值。
"""
normalized_value = (value or "").strip()
normalized_default = (default_value or "").strip()
if normalized_value == normalized_default:
return ""
return normalized_value
```
## /app/models/schema.py
```py path="/app/models/schema.py"
import warnings
from enum import Enum
from typing import Any, List, Literal, Optional, Union
import pydantic
from pydantic import BaseModel, ConfigDict, Field
from app.config import config
# 忽略 Pydantic 的特定警告
warnings.filterwarnings(
"ignore",
category=UserWarning,
message="Field name.*shadows an attribute in parent.*",
)
class VideoConcatMode(str, Enum):
random = "random"
sequential = "sequential"
class VideoTransitionMode(str, Enum):
none = None
shuffle = "Shuffle"
fade_in = "FadeIn"
fade_out = "FadeOut"
slide_in = "SlideIn"
slide_out = "SlideOut"
zoom_in = "ZoomIn"
zoom_out = "ZoomOut"
class VideoAspect(str, Enum):
landscape = "16:9"
portrait = "9:16"
square = "1:1"
def to_resolution(self):
if self == VideoAspect.landscape:
return 1920, 1080
elif self == VideoAspect.portrait:
return 1080, 1920
elif self == VideoAspect.square:
return 1080, 1080
raise ValueError(f"unsupported video aspect: {self}")
class VideoFitMode(str, Enum):
"""How source clips with a different aspect ratio fill the output canvas."""
cover = "cover"
contain = "contain"
SubtitleDisplayMode = Literal["sentence", "word_by_word"]
SubtitleAnimation = Literal["none", "pop_spring"]
_SUBTITLE_DISPLAY_MODES = ("sentence", "word_by_word")
_SUBTITLE_ANIMATIONS = ("none", "pop_spring")
def _get_valid_ui_choice(key: str, allowed_values: tuple[str, ...], default: str) -> str:
"""
读取经过校验的 WebUI 枚举配置，兼容旧用户可能残留的无效值。
请求体由 Pydantic 的 Literal 严格校验，拼写错误会返回明确的字段校验错误；
配置文件则需要宽容处理，避免用户升级后因为历史手工配置错误导致整个服务
无法启动。HTTP 状态码由应用统一的校验异常处理器决定，这里不绑定具体数值。
"""
configured_value = config.ui.get(key, default)
return configured_value if configured_value in allowed_values else default
_Config = ConfigDict(
arbitrary_types_allowed=True,
# Note: ensure your key names match renamed V2 parameters if needed
)
@pydantic.dataclasses.dataclass(config=_Config)
class MaterialInfo:
provider: str = "pexels"
url: str = ""
duration: int = 0
# 在线素材搜索会附带经过筛选的公开来源信息，供搜索缓存和任务记录复用。
# 本地上传素材不需要填写；写入任务文件前仍会按字段白名单重新构造，
# 避免外部请求传入的签名 URL、凭据或无关字段进入持久化数据。
source_info: Optional[dict[str, Any]] = None
class VideoParams(BaseModel):
"""
{
"video_subject": "",
"video_aspect": "横屏 16:9（西瓜视频）",
"voice_name": "女生-晓晓",
"bgm_name": "random",
"font_name": "STHeitiMedium 黑体-中",
"text_color": "#FFFFFF",
"font_size": 60,
"stroke_color": "#000000",
"stroke_width": 1.5
}
"""
video_subject: str
video_script: str = "" # Script used to generate the video
video_terms: Optional[str | List[str]] = None # Keywords used to generate the video
video_aspect: Optional[VideoAspect] = VideoAspect.portrait.value
video_fit_mode: VideoFitMode = VideoFitMode.cover
video_concat_mode: Optional[VideoConcatMode] = VideoConcatMode.random.value
video_transition_mode: Optional[VideoTransitionMode] = None
video_clip_duration: int = Field(default=5, ge=1)
video_clip_speed: Optional[float] = 1.0
match_materials_to_script: bool = False
video_count: int = Field(default=1, ge=1)
video_source: Optional[str] = "pexels"
video_materials: Optional[List[MaterialInfo]] = (
None # Materials used to generate the video
)
custom_audio_file: Optional[str] = (
None # Custom audio file path, will ignore TTS and can still use Whisper subtitles
)
video_language: Optional[str] = "" # auto detect
voice_name: Optional[str] = ""
voice_volume: Optional[float] = Field(default=1.0, allow_inf_nan=False)
voice_rate: Optional[float] = Field(default=1.0, allow_inf_nan=False)
bgm_type: Optional[str] = "random"
bgm_file: Optional[str] = ""
bgm_volume: Optional[float] = Field(default=0.2, allow_inf_nan=False)
# 视频配乐供应商共用提示词，WebUI 新任务统一写入该字段。保留下面的
# Sonilo 专用字段以兼容旧任务记录和现有 CLI 参数。
video_music_prompt: str = Field(default="", max_length=2000)
sonilo_bgm_prompt: str = Field(default="", max_length=2000)
subtitle_enabled: Optional[bool] = True
subtitle_position: Optional[str] = config.ui.get(
"subtitle_position", "bottom"
) # top, bottom, center, custom, two_thirds_bottom
subtitle_display_mode: SubtitleDisplayMode = _get_valid_ui_choice(
"subtitle_display_mode", _SUBTITLE_DISPLAY_MODES, "sentence"
)
subtitle_animation: SubtitleAnimation = _get_valid_ui_choice(
"subtitle_animation", _SUBTITLE_ANIMATIONS, "none"
)
custom_position: float = Field(
default=config.ui.get("custom_position", 70.0), ge=0, le=100, allow_inf_nan=False
)
font_name: Optional[str] = "STHeitiMedium.ttc"
text_fore_color: Optional[str] = "#FFFFFF"
text_background_color: Union[bool, str] = False
rounded_subtitle_background: bool = False
font_size: int = Field(default=60, ge=1)
stroke_color: Optional[str] = "#000000"
stroke_width: float = Field(default=1.5, ge=0, allow_inf_nan=False)
n_threads: Optional[int] = 2
paragraph_number: int = Field(default=1, ge=1, le=10)
video_script_prompt: str = Field(default="", max_length=2000)
custom_system_prompt: str = Field(default="", max_length=8000)
class SubtitleRequest(BaseModel):
video_script: str
video_language: Optional[str] = ""
voice_name: Optional[str] = "zh-CN-XiaoxiaoNeural-Female"
voice_volume: Optional[float] = Field(default=1.0, allow_inf_nan=False)
voice_rate: Optional[float] = Field(default=1.2, allow_inf_nan=False)
bgm_type: Optional[str] = "random"
bgm_file: Optional[str] = ""
bgm_volume: Optional[float] = Field(default=0.2, allow_inf_nan=False)
subtitle_position: Optional[str] = config.ui.get("subtitle_position", "bottom")
subtitle_display_mode: SubtitleDisplayMode = _get_valid_ui_choice(
"subtitle_display_mode", _SUBTITLE_DISPLAY_MODES, "sentence"
)
subtitle_animation: SubtitleAnimation = _get_valid_ui_choice(
"subtitle_animation", _SUBTITLE_ANIMATIONS, "none"
)
font_name: Optional[str] = "STHeitiMedium.ttc"
text_fore_color: Optional[str] = "#FFFFFF"
text_background_color: Union[bool, str] = False
rounded_subtitle_background: bool = False
font_size: int = Field(default=60, ge=1)
stroke_color: Optional[str] = "#000000"
stroke_width: float = Field(default=1.5, ge=0, allow_inf_nan=False)
video_source: Optional[str] = "local"
subtitle_enabled: Optional[bool] = True
class AudioRequest(BaseModel):
video_script: str
video_language: Optional[str] = ""
voice_name: Optional[str] = "zh-CN-XiaoxiaoNeural-Female"
voice_volume: Optional[float] = Field(default=1.0, allow_inf_nan=False)
voice_rate: Optional[float] = Field(default=1.2, allow_inf_nan=False)
bgm_type: Optional[str] = "random"
bgm_file: Optional[str] = ""
bgm_volume: Optional[float] = Field(default=0.2, allow_inf_nan=False)
video_source: Optional[str] = "local"
class VideoScriptParams:
"""
{
"video_subject": "春天的花海",
"video_language": "",
"paragraph_number": 1,
"video_script_prompt": "",
"custom_system_prompt": ""
}
"""
video_subject: Optional[str] = "春天的花海"
video_language: Optional[str] = ""
paragraph_number: int = Field(default=1, ge=1, le=10)
video_script_prompt: str = Field(default="", max_length=2000)
custom_system_prompt: str = Field(default="", max_length=8000)
class VideoTermsParams:
"""
{
"video_subject": "",
"video_script": "",
"amount": 5,
"match_materials_to_script": false
}
"""
video_subject: Optional[str] = "春天的花海"
video_script: Optional[str] = (
"春天的花海，如诗如画般展现在眼前。万物复苏的季节里，大地披上了一袭绚丽多彩的盛装。金黄的迎春、粉嫩的樱花、洁白的梨花、艳丽的郁金香……"
)
amount: Optional[int] = 5
match_materials_to_script: bool = False
class VideoSocialMetadataParams:
"""
{
"video_subject": "A day in Shanghai",
"video_script": "",
"language": "auto",
"platform": "tiktok"
}
"""
video_subject: Optional[str] = Field(default="A day in Shanghai", max_length=500)
video_script: Optional[str] = Field(default="", max_length=8000)
language: Optional[str] = Field(default="auto", max_length=64)
platform: Optional[str] = Field(default="tiktok", max_length=64)
class TaskVideoRequest(VideoParams, BaseModel):
# Bound API resource requests while leaving the CLI's explicit batch and
# FFmpeg thread controls in VideoParams unchanged.
video_count: int = Field(default=1, ge=1, le=5)
video_clip_duration: int = Field(default=5, ge=1, le=15)
n_threads: Optional[int] = Field(default=2, ge=1, le=16)
class TaskQueryRequest(BaseModel):
pass
class VideoScriptRequest(VideoScriptParams, BaseModel):
pass
class VideoTermsRequest(VideoTermsParams, BaseModel):
# Ordered term generation allocates an example list proportional to amount
# before contacting the model. Reject malformed or excessive API requests.
amount: int = Field(default=5, ge=1, le=50)
class VideoSocialMetadataRequest(VideoSocialMetadataParams, BaseModel):
pass
# ---------------------------
# ----- RESPONSE MODELS -----
# ---------------------------
class BaseResponse(BaseModel):
status: int = 200
message: Optional[str] = "success"
data: Any = None
# ---- DATA MODELS ----
class TaskResponseData(BaseModel):
task_id: str
class TaskStatusData(BaseModel):
"""任务查询对外保证的稳定字段；历史和扩展字段继续原样透传。"""
model_config = ConfigDict(extra="allow")
task_id: str
state: int
progress: int = 0
videos: Optional[List[str]] = None
combined_videos: Optional[List[str]] = None
failed_stage: Optional[str] = None
error: Optional[str] = None
cross_post_state: Optional[
Literal["pending", "processing", "complete", "failed"]
] = None
cross_post_results: Optional[List[dict[str, Any]]] = None
cross_post_error: Optional[str] = None
class TaskListData(BaseModel):
"""分页任务列表结构。"""
tasks: List[TaskStatusData]
total: int
page: int
page_size: int
class VideoScriptData(BaseModel):
video_script: str
class VideoTermsData(BaseModel):
video_terms: List[str]
class VideoSocialMetadataData(BaseModel):
title: str
caption: str
hashtags: List[str]
class FileData(BaseModel):
name: str
size: int
file: str
class BgmRetrieveData(BaseModel):
files: List[FileData]
class BgmUploadData(BaseModel):
file: str
class VideoMaterialRetrieveData(BaseModel):
files: List[FileData]
class VideoMaterialUploadData(BaseModel):
file: str
# ---- RESPONSE MODELS ----
class TaskResponse(BaseResponse):
data: TaskResponseData
model_config = ConfigDict(
json_schema_extra={
"example": {
"status": 200,
"message": "success",
"data": {
"task_id": "6c85c8cc-a77a-42b9-bc30-947815aa0558",
},
},
}
)
class TaskQueryResponse(BaseResponse):
"""
任务查询会返回生成状态和可选的跨平台发布状态。
生成失败时包含 `failed_stage` 和 `error`；生成完成后如果启用了自动发布，
`cross_post_state` 会依次进入 pending、processing、complete 或 failed。
"""
data: TaskStatusData
model_config = ConfigDict(
json_schema_extra={
"examples": [
{
"status": 200,
"message": "success",
"data": {
"task_id": "6c85c8cc-a77a-42b9-bc30-947815aa0558",
"state": 1,
"progress": 100,
"videos": ["/tasks/example/final-1.mp4"],
"cross_post_state": "complete",
"cross_post_results": [{"success": True}],
},
},
{
"status": 200,
"message": "success",
"data": {
"task_id": "6c85c8cc-a77a-42b9-bc30-947815aa0558",
"state": -1,
"progress": 30,
"failed_stage": "audio",
"error": "TTS request timed out",
},
},
],
}
)
class TaskListResponse(BaseResponse):
"""任务列表使用独立响应模型，避免与单任务查询混用文档结构。"""
data: TaskListData
model_config = ConfigDict(
json_schema_extra={
"example": {
"status": 200,
"message": "success",
"data": {
"tasks": [
{
"task_id": "6c85c8cc-a77a-42b9-bc30-947815aa0558",
"state": 4,
"progress": 50,
}
],
"total": 1,
"page": 1,
"page_size": 10,
},
}
}
)
class TaskDeletionResponse(BaseResponse):
data: None = None
model_config = ConfigDict(
json_schema_extra={
"example": {
"status": 200,
"message": "success",
"data": None,
},
}
)
class VideoScriptResponse(BaseResponse):
data: VideoScriptData
model_config = ConfigDict(
json_schema_extra={
"example": {
"status": 200,
"message": "success",
"data": {
"video_script": "春天的花海，是大自然的一幅美丽画卷。在这个季节里，大地复苏，万物生长，花朵争相绽放，形成了一片五彩斑斓的花海..."
},
},
}
)
class VideoTermsResponse(BaseResponse):
data: VideoTermsData
model_config = ConfigDict(
json_schema_extra={
"example": {
"status": 200,
"message": "success",
"data": {"video_terms": ["sky", "tree"]},
},
}
)
class VideoSocialMetadataResponse(BaseResponse):
data: VideoSocialMetadataData
model_config = ConfigDict(
json_schema_extra={
"example": {
"status": 200,
"message": "success",
"data": {
"title": "A Day in Shanghai You Should Not Miss",
"caption": "Save this quick Shanghai inspiration and follow for more short travel ideas.",
"hashtags": ["#shorts", "#travel", "#shanghai", "#viral", "#fyp"],
},
},
}
)
class BgmRetrieveResponse(BaseResponse):
data: BgmRetrieveData
model_config = ConfigDict(
json_schema_extra={
"example": {
"status": 200,
"message": "success",
"data": {
"files": [
{
"name": "4fca18fce7344f3aa824777a40d45c8c.mp3",
"size": 1891269,
"file": "4fca18fce7344f3aa824777a40d45c8c.mp3",
}
]
},
},
}
)
class BgmUploadResponse(BaseResponse):
data: BgmUploadData
model_config = ConfigDict(
json_schema_extra={
"example": {
"status": 200,
"message": "success",
"data": {"file": "4fca18fce7344f3aa824777a40d45c8c.mp3"},
},
}
)
class VideoMaterialRetrieveResponse(BaseResponse):
data: VideoMaterialRetrieveData
model_config = ConfigDict(
json_schema_extra={
"example": {
"status": 200,
"message": "success",
"data": {
"files": [
{
"name": "example.mp4",
"size": 12345678,
"file": "/MoneyPrinterTurbo/resource/videos/example.mp4",
}
]
},
},
}
)
class VideoMaterialUploadResponse(BaseResponse):
data: VideoMaterialUploadData
model_config = ConfigDict(
json_schema_extra={
"example": {
"status": 200,
"message": "success",
"data": {
"file": "/MoneyPrinterTurbo/resource/videos/example.mp4",
},
},
}
)
```
## /app/router.py
```py path="/app/router.py"
"""Application configuration - root APIRouter
Defines all FastAPI application endpoints.
Resources:
1. https://fastapi.tiangolo.com/tutorial/bigger-applications
"""
from fastapi import APIRouter
from app.controllers import ping
from app.controllers.v1 import llm, video
root_api_router = APIRouter()
root_api_router.include_router(ping.router)
# v1
root_api_router.include_router(video.router)
root_api_router.include_router(llm.router)
```
## /app/services/__init__.py
```py path="/app/services/__init__.py"
```
## /app/services/bgm.py
```py path="/app/services/bgm.py"
import math
import os
import subprocess
import tempfile
from pathlib import Path
from typing import BinaryIO
from uuid import uuid4
from loguru import logger
from app.utils import file_security, utils
# Streamlit 默认允许较大的上传文件，但背景音乐通常只有几 MB。这里设置明确的
# 服务端上限，避免 API 或 WebUI 把超大文件完整写入磁盘，影响同一进程中的视频任务。
MAX_BGM_UPLOAD_BYTES = 30 * 1024 * 1024
_COPY_CHUNK_BYTES = 1024 * 1024
_INTERNAL_UPLOAD_PREFIX = ".bgm-upload-"
_WINDOWS_INVALID_FILENAME_CHARS = frozenset('<>:"|?*')
_WINDOWS_RESERVED_FILENAMES = frozenset(
{"CON", "PRN", "AUX", "NUL"}
| {f"{prefix}{number}" for prefix in ("COM", "LPT") for number in range(1, 10)}
# Win32 把 Latin-1 上标数字 ¹、²、³ 也识别为设备编号，因此与普通数字保留名
# 同等处理，避免 COM¹.mp3 绕过保护。清单与 webui/Main.py 的下载文件名规则一致。
| {f"{prefix}{number}" for prefix in ("COM", "LPT") for number in ("¹", "²", "³")}
)
# 文件名会原样进入日志、API 响应和 WebUI 界面，因此除路径分隔符与 Win32 保留名外，
# 还要拒绝所有控制符和会改变显示形态的字符。此前只拦截 ord < 32（C0 控制符），
# 漏掉了同一类问题的另一半：
# * C1 控制符 U+007F-U+009F：U+0085 会被部分日志查看器渲染成换行，一个文件名就能
# 伪造出额外的、看起来独立的日志行。
# * 双向文本控制符 U+200E/U+200F/U+202A-U+202E/U+2066-U+2069：U+202E 之后的字符会被
# 反向渲染，"photo\u202egnp.mp3" 在资源管理器和日志里看起来像另一个扩展名。
# * U+2028/U+2029（Unicode 行/段分隔符）同样能在一行日志里制造视觉换行。
# 这些字符都无法在文件名输入框里键入，正常上传不受影响；判定与
# app/controllers/base.py 的 normalize_task_id 一致（那边直接用 isprintable）。
_UNSAFE_FILENAME_CHARACTERS = frozenset(
chr(code)
for code in (
*range(0x00, 0x20),
*range(0x7F, 0xA0),
0x200E,
0x200F,
0x2028,
0x2029,
*range(0x202A, 0x202F),
*range(0x2066, 0x206A),
)
)
# MoviePy 最终通过 FFmpeg 解码背景音乐，因此不需要人为限制为 MP3。这里仅开放
# 主流且语义明确的音频扩展名，避免把 MP4 等带视频容器误当作背景音乐上传。
# 元组同时作为 WebUI 上传控件的单一数据源，后续增删格式时不会出现前后端不一致。
SUPPORTED_BGM_EXTENSIONS = (
".mp3",
".m4a",
".aac",
".wav",
".flac",
".ogg",
".opus",
".wma",
)
class BgmUploadError(ValueError):
"""表示上传文件不满足背景音乐的安全或格式要求。"""
class BgmServiceError(RuntimeError):
"""表示 FFmpeg 或文件系统不可用等服务端执行故障。"""
def should_use_bgm(bgm_type: str | None, bgm_volume: float | None) -> bool:
"""
统一判断当前任务是否需要处理任何背景音乐。
该规则与具体来源无关：没有选择来源、音量不合法或音量不大于 0 时，随机、
自定义、Sonilo 以及未来新增的提供商都必须跳过文件解析、外部生成和最终混音。
放在通用 BGM 服务中可以避免每增加一个提供商就复制一套 0 音量判断。
"""
if not str(bgm_type or "").strip():
return False
try:
normalized_volume = float(bgm_volume or 0)
except (TypeError, ValueError):
return False
return math.isfinite(normalized_volume) and normalized_volume > 0
def uploaded_bgm_dir(create: bool = True) -> str:
"""
返回用户背景音乐的持久化目录。
内置歌曲属于代码资源，继续放在 resource/songs；用户上传内容属于运行时数据，
必须放在 Docker 已挂载的 storage 下，容器重建后才能保留，也不会污染 Git 工作区。
"""
return utils.storage_dir("bgm", create=create)
def _remove_staged_file(file_path: str) -> None:
"""尽力清理上传临时文件，且不覆盖调用方正在处理的原始异常。"""
if not file_path or not os.path.exists(file_path):
return
try:
os.remove(file_path)
except OSError as exc:
# 临时文件使用保留前缀，不会进入 BGM 列表；清理失败不应把“音频非法”
# 等更准确的原始异常覆盖掉，但必须留下路径和系统错误供运维定位。
logger.warning(
f"failed to remove staged background music: path={file_path}, "
f"error={str(exc)}"
)
def sanitize_upload_filename(filename: str) -> str:
"""提取可跨平台展示的音频文件名，并拒绝非法名称与不支持的扩展名。"""
safe_name = (filename or "").replace("\\", "/").split("/")[-1].strip()
if (
not safe_name
or safe_name in {".", ".."}
or len(safe_name) > 255
or any(character in _UNSAFE_FILENAME_CHARACTERS for character in safe_name)
or any(character in _WINDOWS_INVALID_FILENAME_CHARS for character in safe_name)
or safe_name.lower().startswith(_INTERNAL_UPLOAD_PREFIX)
):
raise BgmUploadError("invalid background music filename")
# Windows 会把扩展名前的首段识别为设备名，例如 CON.mp3、LPT1.wav 都
# 不能作为普通文件创建。即使服务端最终使用 UUID，提前拒绝这类名称也能
# 保证 API 在不同平台上的输入行为一致。
windows_basename = safe_name.split(".", 1)[0].rstrip(" .").upper()
if windows_basename in _WINDOWS_RESERVED_FILENAMES:
raise BgmUploadError("invalid background music filename")
if Path(safe_name).suffix.lower() not in SUPPORTED_BGM_EXTENSIONS:
supported_formats = ", ".join(
extension.removeprefix(".").upper()
for extension in SUPPORTED_BGM_EXTENSIONS
)
raise BgmUploadError(
f"unsupported background music format; supported formats: {supported_formats}"
)
return safe_name
def _validate_audio(file_path: str, timeout_seconds: int = 30) -> None:
"""
仅使用项目当前配置的 FFmpeg 验证文件包含可完整解码的音频流。
项目允许 imageio-ffmpeg 提供便携 FFmpeg，该安装方式不保证同时存在
FFprobe，因此不能新增独立二进制依赖。`-map 0:a:0` 会在没有音频流时失败，
`-xerror` 会把解码错误提升为失败；完整解码还能拦截加密文件或随机数据偶然
命中音频帧头的误判。文件可以包含专辑封面等附加流，但只校验第一条音频流。
"""
try:
decoded = subprocess.run(
[
utils.get_ffmpeg_binary(),
"-nostdin",
"-v",
"error",
"-xerror",
# Uploaded files must be self-contained media, not playlists
# that cause validation/rendering to read adjacent files or URLs.
"-protocol_whitelist",
"file,pipe",
"-format_whitelist",
"mp3,mov,aac,wav,flac,ogg,asf",
"-i",
file_path,
"-map",
"0:a:0",
# Normalize timestamps from decoded samples. A header-only
# WAV can exit successfully without producing any audio.
"-af",
"asetpts=N/SR/TB",
"-progress",
"pipe:1",
"-nostats",
"-f",
"null",
"-",
],
capture_output=True,
timeout=timeout_seconds,
check=False,
)
except subprocess.TimeoutExpired as exc:
raise BgmServiceError("FFmpeg background music validation timed out") from exc
except OSError as exc:
raise BgmServiceError("failed to run FFmpeg for background music validation") from exc
if decoded.returncode != 0:
raise BgmUploadError("uploaded file must contain a decodable audio stream")
decoded_audio = False
for line in decoded.stdout.splitlines():
key, separator, value = line.partition(b"=")
if separator and key == b"out_time_us":
try:
decoded_audio = decoded_audio or int(value) > 0
except ValueError:
continue
if not decoded_audio:
raise BgmUploadError("uploaded file must contain nonempty decodable audio")
def validate_audio_file(file_path: str, timeout_seconds: int = 120) -> None:
"""
校验磁盘上的音频文件可由项目 FFmpeg 完整解码。
上传预检通常只需 30 秒；Sonilo 生成的配乐最长可达 6 分钟，因此对外提供
可调整超时的复用入口。服务只依赖 FFmpeg，不要求系统额外安装 FFprobe。
"""
if not os.path.isfile(file_path) or os.path.getsize(file_path) <= 0:
raise BgmUploadError("background music file is empty or missing")
_validate_audio(file_path, timeout_seconds=timeout_seconds)
def _stage_bgm_upload(filename: str, source: BinaryIO) -> tuple[str, str, int]:
"""
将上传流写入同目录临时文件，并返回安全文件名、临时路径和字节数。
WebUI 的上传预检和最终持久化必须使用完全相同的分块读取、大小限制与文件名
规则，否则可能出现界面显示可用、点击生成后却被服务端拒绝的状态分裂。
临时文件由调用方在完成音频探测后删除或原子替换。
"""
safe_name = sanitize_upload_filename(filename)
try:
target_dir = uploaded_bgm_dir(create=True)
except OSError as exc:
raise BgmServiceError("failed to prepare background music storage") from exc
temp_path = ""
total_bytes = 0
try:
try:
source.seek(0)
except (AttributeError, OSError) as exc:
raise BgmUploadError("background music upload is not seekable") from exc
# 保留原始扩展名便于 FFmpeg 针对无容器头的 AAC 等格式选择正确的
# demuxer；临时文件仍放在目标目录，以保证最终 os.replace 是原子操作。
descriptor, temp_path = tempfile.mkstemp(
prefix=_INTERNAL_UPLOAD_PREFIX,
suffix=Path(safe_name).suffix.lower(),
dir=target_dir,
)
with os.fdopen(descriptor, "wb") as output:
while True:
chunk = source.read(_COPY_CHUNK_BYTES)
if not chunk:
break
if not isinstance(chunk, (bytes, bytearray, memoryview)):
raise BgmUploadError("background music upload must be binary")
total_bytes += len(chunk)
if total_bytes > MAX_BGM_UPLOAD_BYTES:
raise BgmUploadError("background music file exceeds the 30 MB limit")
output.write(chunk)
output.flush()
os.fsync(output.fileno())
if total_bytes == 0:
raise BgmUploadError("background music file is empty")
return safe_name, temp_path, total_bytes
except Exception as exc:
_remove_staged_file(temp_path)
if isinstance(exc, BgmUploadError):
raise
if isinstance(exc, OSError):
raise BgmServiceError("failed to stage background music upload") from exc
raise
finally:
# Streamlit 还需要使用同一个 UploadedFile 做浏览器试听；恢复文件指针可
# 避免校验后播放器或最终保存读取到空内容。
try:
source.seek(0)
except (AttributeError, OSError):
pass
def validate_bgm_upload(filename: str, source: BinaryIO) -> str:
"""完整校验上传音频但不持久化，用于 WebUI 在显示“已就绪”前预检。"""
safe_name, temp_path, total_bytes = _stage_bgm_upload(filename, source)
try:
_validate_audio(temp_path)
logger.debug(
f"background music upload validated: name={safe_name}, "
f"size={total_bytes} bytes"
)
return safe_name
finally:
_remove_staged_file(temp_path)
def save_bgm_upload(filename: str, source: BinaryIO) -> str:
"""
以分块、限量和原子替换的方式保存用户背景音乐。
使用场景包括 FastAPI UploadFile 和 Streamlit UploadedFile，两者都提供二进制
文件接口。先写同目录临时文件并验证，再通过 os.replace 原子落盘，既能避免
并发上传或进程中断留下半个音频文件，也会让同名上传获得不同的 UUID 存储键，
已排队或运行中的任务因此始终引用原来的不可变文件。
"""
safe_name, temp_path, total_bytes = _stage_bgm_upload(filename, source)
stored_name = f"{uuid4().hex}{Path(safe_name).suffix.lower()}"
target_path = os.path.join(os.path.dirname(temp_path), stored_name)
try:
_validate_audio(temp_path)
try:
os.replace(temp_path, target_path)
except OSError as exc:
raise BgmServiceError("failed to persist background music upload") from exc
temp_path = ""
logger.info(
f"background music uploaded: original_name={safe_name}, "
f"stored_name={stored_name}, size={total_bytes} bytes"
)
return stored_name
finally:
_remove_staged_file(temp_path)
def _list_bgm_files(directories: tuple[str, ...]) -> list[str]:
"""按目录优先级枚举安全且受支持的背景音乐文件。"""
files_by_name: dict[str, str] = {}
for directory in directories:
if not os.path.isdir(directory):
continue
for name in sorted(os.listdir(directory), key=str.lower):
# 上传预检和最终保存都会短暂创建同目录文件。临时文件虽然带有合法
# 音频扩展名，但尚未完成校验，不能被随机 BGM 列表提前选中。
if name.startswith(_INTERNAL_UPLOAD_PREFIX):
continue
if Path(name).suffix.lower() not in SUPPORTED_BGM_EXTENSIONS:
continue
file_path = os.path.join(directory, name)
try:
# 枚举结果同样需要真实路径校验。否则攻击者可在允许目录中放置
# 指向外部文件的音频符号链接，再借随机 BGM 路径交给 MoviePy。
resolved_path = file_security.resolve_path_within_directory(
directory, file_path
)
except ValueError as exc:
logger.warning(
f"skip unsafe background music file: name={name}, error={str(exc)}"
)
continue
files_by_name[name] = resolved_path
return [files_by_name[name] for name in sorted(files_by_name, key=str.lower)]
def list_builtin_bgm_files() -> list[str]:
"""
列出随项目分发的内置背景音乐。
WebUI 的“预设歌曲”和设置预设导入导出只使用这一列表，确保保存的文件名
可以在另一台使用相同版本的设备上恢复；用户上传文件仍由自定义音乐管理。
"""
return _list_bgm_files((utils.song_dir(),))
def list_bgm_files() -> list[str]:
"""列出用户上传和内置的可用背景音乐，重名时优先使用上传文件。"""
return _list_bgm_files((utils.song_dir(), uploaded_bgm_dir(create=True)))
def resolve_builtin_bgm_file(unsafe_path: str) -> str:
"""按文件名解析内置背景音乐，并拒绝路径、未知文件和用户上传文件。"""
if not unsafe_path:
raise ValueError("background music filename is required")
filename = str(unsafe_path)
if filename != os.path.basename(filename):
raise ValueError("preset background music must use a filename")
files_by_name = {
os.path.basename(file_path): file_path for file_path in list_builtin_bgm_files()
}
if filename not in files_by_name:
raise ValueError("preset background music is not available")
return files_by_name[filename]
def resolve_bgm_file(unsafe_path: str) -> str:
"""
在用户上传目录和内置歌曲目录中解析 BGM，并拒绝两个白名单之外的路径与上传
中断遗留的暂存文件。
文件名优先命中用户目录，同时保留 `output000.mp3`、绝对白名单路径和
`./resource/songs/output000.mp3` 等旧用法。新上传文件使用 UUID，正常情况下
不会与内置歌曲或历史上传发生重名。
"""
if (
not unsafe_path
or Path(unsafe_path).suffix.lower() not in SUPPORTED_BGM_EXTENSIONS
):
raise ValueError("unsupported background music path")
if os.path.basename(str(unsafe_path)).lower().startswith(_INTERNAL_UPLOAD_PREFIX):
# 上传预检与最终保存都会在同一个目录里短暂创建 ``.bgm-upload-`` 前缀的
# 中间文件。它们带有合法的音频扩展名，但尚未完成校验，``_list_bgm_files``
# 已经把它们排除在随机 BGM 之外；解析入口必须给出同一个结论，否则
# API/CLI 传入的名称可以命中一个写入中断、内容不完整的中间文件。
# 用户传入的名称按不区分大小写比较，与 Windows/macOS 的文件系统一致。
raise ValueError("background music upload staging files are not selectable")
candidates = [unsafe_path]
if not os.path.isabs(unsafe_path):
candidates.append(os.path.join(utils.root_dir(), unsafe_path))
last_error = ValueError("background music file does not exist")
for directory in (uploaded_bgm_dir(create=True), utils.song_dir()):
for candidate in candidates:
try:
return file_security.resolve_path_within_directory(directory, candidate)
except ValueError as exc:
last_error = exc
raise ValueError(str(last_error)) from last_error
```
## /app/services/cache_manager.py
```py path="/app/services/cache_manager.py"
"""视频素材缓存的统计、预览和清理服务。"""
from __future__ import annotations
import os
import re
import stat
import time
from dataclasses import dataclass
from typing import Iterator
from loguru import logger
from app.utils import utils
# 在线素材使用 URL 的 MD5 作为稳定文件名。缓存管理只接受该命名格式，避免把
# 用户误放到目录中的视频、说明文件或其它业务文件当作缓存删除。
_VIDEO_CACHE_FILE_PATTERN = re.compile(r"^vid-[0-9a-f]{32}\.mp4{{contextString}}quot;)
# save_video writes to this same-directory name before atomically publishing a
# validated clip. A forced process exit can leave one behind, so include only
# aged temporary files in cache cleanup and never touch an active download.
_VIDEO_CACHE_TEMP_FILE_PATTERN = re.compile(
r"^\.vid-[0-9a-f]{32}-[a-z0-9_]+\.mp4{{contextString}}quot;
)
_SECONDS_PER_DAY = 24 * 60 * 60
@dataclass(frozen=True)
class VideoCacheStats:
"""缓存目录的轻量统计结果，只包含文件系统元数据。"""
file_count: int = 0
total_size: int = 0
oldest_mtime: float | None = None
newest_mtime: float | None = None
@dataclass(frozen=True)
class VideoCacheCleanupResult:
"""一次清理的执行结果，允许部分文件删除失败。"""
deleted_count: int = 0
deleted_size: int = 0
failed_count: int = 0
@dataclass(frozen=True)
class _VideoCacheEntry:
"""扫描阶段保存的最小文件信息，避免清理时打开或解析视频。"""
path: str
name: str
size: int
mtime: float
device: int
inode: int
mtime_ns: int
def video_cache_dir() -> str:
"""返回项目管理的默认视频缓存目录。"""
return os.path.realpath(utils.storage_dir("cache_videos"))
def _iter_video_cache_entries(include_temp: bool = False) -> Iterator[_VideoCacheEntry]:
"""
顺序扫描默认缓存目录第一层。
使用 ``os.scandir`` 是为了在缓存达到数万文件时复用目录遍历返回的元数据，
避免 ``Path.iterdir`` 后再次查询文件类型。这里不递归、不打开视频，也不调用
FFmpeg，因此耗时主要与文件数量线性相关，而不是与视频总容量相关。
"""
cache_dir = video_cache_dir()
try:
entries = os.scandir(cache_dir)
except FileNotFoundError:
return
except OSError as exc:
logger.warning(
f"failed to scan video cache directory: path={cache_dir}, error={exc}"
)
return
with entries:
for entry in entries:
if not _VIDEO_CACHE_FILE_PATTERN.fullmatch(entry.name) and not (
include_temp and _VIDEO_CACHE_TEMP_FILE_PATTERN.fullmatch(entry.name)
):
continue
try:
# 不跟随符号链接，确保清理逻辑不会越过默认缓存目录边界。
if not entry.is_file(follow_symlinks=False):
continue
stat_result = entry.stat(follow_symlinks=False)
except OSError as exc:
logger.warning(
f"failed to inspect video cache file: file={entry.name}, error={exc}"
)
continue
yield _VideoCacheEntry(
path=entry.path,
name=entry.name,
size=stat_result.st_size,
mtime=stat_result.st_mtime,
device=stat_result.st_dev,
inode=stat_result.st_ino,
mtime_ns=stat_result.st_mtime_ns,
)
def _is_cleanup_candidate(
entry: _VideoCacheEntry,
max_age_days: int | None,
now: float,
) -> bool:
if (
_VIDEO_CACHE_TEMP_FILE_PATTERN.fullmatch(entry.name)
and entry.mtime >= now - _SECONDS_PER_DAY
):
return False
if max_age_days is None:
return True
return entry.mtime < now - max_age_days * _SECONDS_PER_DAY
def _validate_max_age_days(max_age_days: int | None) -> None:
"""即使缓存目录为空，也应稳定拒绝无效清理参数。"""
if max_age_days is None:
return
if (
isinstance(max_age_days, bool)
or not isinstance(max_age_days, int)
or max_age_days <= 0
):
raise ValueError("max_age_days must be a positive integer or None")
def get_video_cache_stats(max_age_days: int | None = None) -> VideoCacheStats:
"""
统计全部缓存，或预览修改时间早于指定天数的可清理缓存。
``max_age_days=None`` 表示全部缓存。统计过程只读取目录项的大小和修改时间，
不读取视频内容，因此即使缓存总容量很大也不会产生与容量成比例的 I/O。
"""
_validate_max_age_days(max_age_days)
now = time.time()
file_count = 0
total_size = 0
oldest_mtime = None
newest_mtime = None
for entry in _iter_video_cache_entries(include_temp=True):
if not _is_cleanup_candidate(entry, max_age_days, now):
continue
file_count += 1
total_size += entry.size
oldest_mtime = (
entry.mtime if oldest_mtime is None else min(oldest_mtime, entry.mtime)
)
newest_mtime = (
entry.mtime if newest_mtime is None else max(newest_mtime, entry.mtime)
)
return VideoCacheStats(
file_count=file_count,
total_size=total_size,
oldest_mtime=oldest_mtime,
newest_mtime=newest_mtime,
)
def clean_video_cache(max_age_days: int | None = None) -> VideoCacheCleanupResult:
"""
清理默认视频缓存，并返回可向用户展示的汇总结果。
页面预览与真正点击清理之间可能间隔较久，所以执行时必须重新扫描和判断，
不能复用旧候选列表。删除采用逐文件容错：单个文件被占用或权限不足时记录
警告并继续，避免几百个文件中一个异常文件导致整次清理失败。
"""
_validate_max_age_days(max_age_days)
now = time.time()
logger.info(
f"start cleaning video cache: max_age_days={max_age_days}"
)
candidate_count = 0
candidate_size = 0
deleted_count = 0
deleted_size = 0
failed_count = 0
cache_dir = video_cache_dir()
# 边扫描边删除，不在内存中保留完整候选列表。即使目录增长到几十万个文件，
# 清理过程的额外内存仍保持常量级；执行时使用统一 now，避免长清理过程中
# 截止时间不断移动而产生不可预测的候选范围。
for entry in _iter_video_cache_entries(include_temp=True):
if not _is_cleanup_candidate(entry, max_age_days, now):
continue
candidate_count += 1
candidate_size += entry.size
try:
# entry.path 来自默认目录的第一层 scandir；删除前再次校验父目录和
# 文件名，防止未来修改扫描逻辑时意外扩大可删除范围。
if (
os.path.realpath(os.path.dirname(entry.path)) != cache_dir
or not (
_VIDEO_CACHE_FILE_PATTERN.fullmatch(entry.name)
or _VIDEO_CACHE_TEMP_FILE_PATTERN.fullmatch(entry.name)
)
or os.path.islink(entry.path)
):
raise ValueError("cache file is outside the managed directory")
try:
current = os.stat(entry.path, follow_symlinks=False)
except FileNotFoundError:
# Another cleanup already removed the scanned candidate.
continue
if not stat.S_ISREG(current.st_mode) or (
current.st_dev,
current.st_ino,
current.st_mtime_ns,
current.st_size,
) != (entry.device, entry.inode, entry.mtime_ns, entry.size):
# A downloader may atomically publish a new file at this name
# after the scan. Do not delete the fresh replacement.
continue
os.unlink(entry.path)
deleted_count += 1
deleted_size += entry.size
except (OSError, ValueError) as exc:
failed_count += 1
logger.warning(
f"failed to delete video cache file: file={entry.name}, error={exc}"
)
logger.info(
"finished cleaning video cache: "
f"candidates={candidate_count}, candidate_bytes={candidate_size}, "
f"deleted={deleted_count}, deleted_bytes={deleted_size}, failed={failed_count}"
)
return VideoCacheCleanupResult(
deleted_count=deleted_count,
deleted_size=deleted_size,
failed_count=failed_count,
)
```
## /app/services/task_artifacts.py
```py path="/app/services/task_artifacts.py"
"""任务目录中持久化文件的安全读写。"""
from __future__ import annotations
import json
import os
import tempfile
import threading
from pathlib import Path
from typing import Any, Mapping
from loguru import logger
from app.utils import utils
_SCRIPT_LOCKS = tuple(threading.RLock() for _ in range(256))
def _script_lock(task_id: str):
return _SCRIPT_LOCKS[hash(task_id) % len(_SCRIPT_LOCKS)]
def _script_file(task_id: str) -> Path:
"""返回任务脚本清单路径，并复用统一的任务目录创建逻辑。"""
return Path(utils.task_dir(task_id)) / "script.json"
def _write_json_atomic(target: Path, payload: Mapping[str, Any]) -> None:
"""
在目标目录内原子写入 JSON，避免进程中断留下半个文件。
临时文件和目标文件必须位于同一目录，才能保证 ``os.replace`` 在常见
本地文件系统和 Docker 挂载目录中保持原子替换语义。写入成功前不会修改
现有文件；异常时只清理本次创建的临时文件，并把错误交给调用方决定是否
影响主流程。
"""
temp_path: Path | None = None
try:
with tempfile.NamedTemporaryFile(
mode="w",
encoding="utf-8",
dir=target.parent,
prefix=f".{target.name}.",
suffix=".tmp",
delete=False,
) as temp_file:
temp_path = Path(temp_file.name)
json.dump(
payload,
temp_file,
ensure_ascii=False,
indent=4,
default=lambda value: value.__dict__,
)
temp_file.write("\n")
temp_file.flush()
os.fsync(temp_file.fileno())
os.replace(temp_path, target)
temp_path = None
finally:
if temp_path is not None:
temp_path.unlink(missing_ok=True)
def write_script_data(task_id: str, payload: Mapping[str, Any]) -> None:
"""创建或完整替换任务的 ``script.json`` 清单。"""
with _script_lock(task_id):
_write_json_atomic(_script_file(task_id), payload)
def patch_script_data(task_id: str, **updates: Any) -> bool:
"""
在保留原有字段的前提下补充任务清单，失败时返回 ``False``。
素材来源属于辅助诊断信息，不能因为文件权限、磁盘瞬时异常或历史文件损坏
阻断视频生成。因此该入口会记录完整异常并降级；首次创建任务清单仍使用
``write_script_data``，由主流程决定基础任务数据写入失败时如何处理。
"""
try:
with _script_lock(task_id):
target = _script_file(task_id)
with target.open("r", encoding="utf-8") as script_file:
payload = json.load(script_file)
if not isinstance(payload, dict):
raise ValueError("task script data must be a JSON object")
payload.update(updates)
_write_json_atomic(target, payload)
return True
except FileNotFoundError:
# ``download_videos`` 也可能被测试、脚本或第三方代码独立调用，此时没有
# 任务清单属于正常场景，不应制造警告或为了辅助记录创建残缺文件。
logger.debug(
f"skip task script update because script.json does not exist: "
f"task_id={task_id}"
)
return False
except Exception as exc:
logger.warning(
"failed to update task script data: "
f"task_id={task_id}, fields={sorted(updates)}, "
f"error={type(exc).__name__}, detail={exc}"
)
return False
```
## /docs/api.jpg
Binary file available at https://raw.githubusercontent.com/harry0703/MoneyPrinterTurbo/refs/heads/main/docs/api.jpg
## /test/__init__.py
```py path="/test/__init__.py"
# Unit test package for test
```
The content has been capped at 50000 tokens. The user could consider applying other filters to refine the result. The better and more specific the context, the better the LLM can follow instructions. If the context seems verbose, the user can refine the filter using uithub. Thank you for using https://uithub.com - Perfect LLM context for any GitHub repo.