Key Source Files in MoneyPrinterTurbo: A Complete Architecture Guide

MoneyPrinterTurbo organizes its FastAPI microservice into seven distinct layers—bootstrap, routing, models, services, utilities, configuration, and task managers—with app/asgi.py, app/controllers/v1/video.py, and app/services/task.py serving as the primary orchestration points.

MoneyPrinterTurbo is an open-source FastAPI microservice that automates short-form video generation, subtitle rendering, and text-to-speech synthesis. Understanding the key source files in MoneyPrinterTurbo is essential for developers who want to extend the pipeline, debug generation failures, or integrate custom voice providers. The codebase follows a clean separation of concerns, with distinct modules handling HTTP routing, Pydantic validation, state management, and asynchronous task execution.

1. Application Bootstrap and Entry Points

The bootstrap layer initializes the ASGI server and constructs the FastAPI application instance.

main.py

Located at the repository root, main.py serves as the entry point that launches the Uvicorn server. It imports the application factory from app.asgi and starts the server with configuration values loaded from config.example.toml.

# main.py (simplified)

if __name__ == "__main__":
    uvicorn.run(app="app.asgi:app", host=config.listen_host,
                port=config.listen_port, reload=config.reload_debug)

app/asgi.py

The app/asgi.py module contains the get_application() factory function. This function creates the FastAPI instance, registers the root router, attaches custom exception handlers for HttpException and RequestValidationError, mounts static file directories for generated media, and configures CORS middleware.

instance.include_router(root_api_router)          # Registers all API endpoints

instance.add_exception_handler(HttpException, ...) # Custom error handling

instance.mount("/tasks", StaticFiles(...))        # Serves generated media files

instance.mount("/", StaticFiles(...))             # Serves public UI assets

2. API Routing and Controllers

The routing layer maps HTTP endpoints to controller functions and handles request validation.

app/router.py

This file declares the root APIRouter and includes version-specific sub-routers. It acts as the central hub that aggregates all endpoint definitions before they are registered in the ASGI application.

app/controllers/v1/video.py

Located at app/controllers/v1/video.py, this is the primary controller implementing the public API for video creation, subtitle generation, audio synthesis, media upload/download, and task management. It defines endpoints such as POST /videos for creating generation tasks.

@router.post("/videos", response_model=TaskResponse)
def create_video(background_tasks: BackgroundTasks, request: Request,
                 body: TaskVideoRequest):
    return create_task(request, body, stop_at="video")

The create_task() function generates a UUID, records the request ID, updates the shared state, and delegates to the task manager to run app.services.task.start asynchronously.

app/controllers/base.py

This utility module provides helper functions used by controllers, such as extracting the request-id header and handling API key validation.

3. Data Models and Validation

Pydantic schemas enforce type safety and generate OpenAPI documentation automatically.

app/models/schema.py

This central file defines all request and response models, including TaskVideoRequest, TaskResponse, and VideoMaterialUploadResponse. These models serve as the core contract between the API layer and the service layer.

app/models/exception.py

Defines the custom HttpException class used throughout the API for consistent error handling.

app/models/const.py

Contains global constants such as punctuation characters used in text processing.

4. Core Services and Business Logic

The service layer implements the heavy-lifting for video generation, state management, and media processing.

app/services/state.py

This module provides an in-memory task state store that tracks task status, generated file paths, and pagination metadata. Functions like update_task() and get_task() manage the lifecycle of generation jobs.

app/services/task.py

Located at app/services/task.py, this is the orchestration engine that coordinates the end-to-end video generation pipeline. It handles text-to-speech synthesis, video clipping, concatenation, and subtitle rendering.

app/services/video.py

Contains helper functions for video-specific processing, such as selecting background music and measuring text dimensions for subtitle placement.

app/services/voice.py

Manages retrieval of available voice models from providers like Azure, Gemini, and SiliconFlow.

app/services/material.py

Reads API keys for external services from the configuration file.

5. Utility Functions and Helpers

app/utils/utils.py

This core utility library provides generic helper functions including JSON response builders, UUID generation, filesystem path helpers (task_dir, song_dir, public_dir), background thread runners, string manipulation, and locale loading.

6. Configuration and Logging

app/config/init.py

This module loads configuration from config.example.toml and initializes the Loguru logger with colored console output. It exposes configuration values such as listen_host, listen_port, and project_name to the rest of the application.

7. Pluggable Task Managers

The application supports multiple concurrency backends through an abstract manager interface.

app/controllers/manager/base_manager.py

Defines the abstract base class specifying the interface for add_task, cancel_task, and other management operations.

app/controllers/manager/memory_manager.py

Implements a simple in-process task queue using Python threading. This is the default manager when Redis is not enabled.

app/controllers/manager/redis_manager.py

Provides a Redis-backed task queue for distributed deployments. Enabled via the enable_redis configuration option.

The router automatically selects the appropriate manager based on config.app["enable_redis"], ensuring the rest of the codebase remains agnostic to the underlying concurrency mechanism.

8. Web Interface and Documentation

webui/Main.py

An optional Streamlit-based web interface located at webui/Main.py for manual testing and interaction with the API.

README.md

The primary documentation file containing project overview, installation steps, and quick-start instructions.

Practical Code Examples

Starting the Server

To launch the development server, run the entry point script:

# Install dependencies

pip install -r requirements.txt

# Run with auto-reload enabled

python main.py

The service will be available at http://127.0.0.1:8000 (or the host/port defined in your TOML config). Interactive API documentation is automatically generated at /docs.

Creating a Video Generation Task

Submit a POST request to the videos endpoint defined in app/controllers/v1/video.py:

curl -X POST "http://127.0.0.1:8000/videos" \
     -H "Content-Type: application/json" \
     -d '{
           "script": "Hello, this is a demo video generated by Money Printer Turbo.",
           "bgm_type": "random",
           "voice": "en-US-Standard-A"
         }'

The controller's create_video function returns a TaskResponse containing a task_id that you can use to track progress.

Checking Task Status

Poll the task state managed by app/services/state.py:

curl "http://127.0.0.1:8000/tasks/a3f9c2d4-1b6e-4c7a-9f3e-7d5b6a2c1e0f"

When the orchestration in app/services/task.py completes, the response includes download URLs for the final video files stored in the static tasks directory.

Summary

• Entry Point: main.py launches the Uvicorn server, while app/asgi.py constructs the FastAPI application and registers routers.
• Routing Layer: app/router.py aggregates endpoints, with app/controllers/v1/video.py handling the core video generation API.
• Data Validation: app/models/schema.py defines Pydantic models for type-safe request/response handling.
• Business Logic: app/services/task.py orchestrates the generation pipeline, supported by state.py, video.py, and voice.py.
• Concurrency: Pluggable task managers in app/controllers/manager/ support both in-memory and Redis-backed queues.
• Configuration: app/config/__init__.py loads TOML settings and initializes Loguru logging.

Frequently Asked Questions

What is the main entry point to start the MoneyPrinterTurbo server?

The server entry point is main.py in the repository root. This script imports the ASGI application factory from app.asgi and launches Uvicorn with configuration values loaded from config.example.toml, including host, port, and debug reload settings.

How does MoneyPrinterTurbo handle asynchronous video generation tasks?

The system uses a pluggable task manager architecture defined in app/controllers/manager/base_manager.py. By default, app/controllers/manager/memory_manager.py handles tasks using Python threading, while app/controllers/manager/redis_manager.py provides distributed queue capabilities when enable_redis is set to true in the configuration.

Where are the API request and response models defined?

All Pydantic schemas are centralized in app/models/schema.py. This file defines the core data contracts including TaskVideoRequest, TaskResponse, and VideoMaterialUploadResponse, which enforce type safety and automatically generate OpenAPI documentation at the /docs endpoint.

How can I switch from in-memory task management to Redis?

To enable Redis-backed task queues, modify the enable_redis setting in your configuration file loaded by app/config/__init__.py. When set to true, the router in app/router.py automatically instantiates RedisManager from app/controllers/manager/redis_manager.py instead of the default MemoryManager, allowing distributed task processing across multiple worker instances.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}