AI Music Generator Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: Add a free MiniMax-powered music generator to the blog without exposing the owner's MiniMax token.

Architecture: Add a standalone Flask backend under pywork/music-generator/ and expose it through Nginx at /tools/music-generator/api/. The VuePress page calls only same-origin backend APIs; MiniMax credentials stay on the server. SQLite stores tasks and usage counters so long-running music jobs survive polling across requests.

Tech Stack: VuePress 1 + Vue 2 single-file component, Flask, Python standard library urllib, SQLite, Node assert wiring tests, pytest backend tests, existing shell deployment scripts.

---

File Structure

Create backend service:

• pywork/music-generator/backend/config.py: env var parsing, paths, public style presets, numeric limits.
• pywork/music-generator/backend/minimax_client.py: MiniMax lyrics/music calls, SSE parsing, token-redacted errors.
• pywork/music-generator/backend/task_store.py: SQLite schema, task CRUD, usage counters.
• pywork/music-generator/backend/rate_limit.py: per-IP/global quota and queue checks.
• pywork/music-generator/backend/cleanup.py: delete expired MP3 files and old task rows.
• pywork/music-generator/backend/app.py: Flask routes, request validation, worker dispatch, downloads.
• pywork/music-generator/requirements.txt: Flask runtime dependency.
• pywork/music-generator/README.md: local run and production env instructions.

Create backend tests:

• pywork/music-generator/tests/test_minimax_client.py
• pywork/music-generator/tests/test_task_store_rate_limit.py
• pywork/music-generator/tests/test_app_behavior.py

Create frontend and content:

• docs/.vuepress/components/MusicGeneratorTool.vue
• docs/07.工具/20.学习工具/50.AI音乐生成器.md

Modify navigation and related pages:

• docs/.vuepress/config.ts
• docs/07.工具/00.概览.md
• docs/07.工具/20.学习工具/20.AI播客生成器.md
• docs/07.工具/10.测试工具/71.样例音频库.md

Modify deployment:

• scripts/remote_configure_music_nginx.sh
• scripts/deploy_prod.sh
• scripts/remote_restart.sh
• scripts/smoke_test.sh

Create/modify wiring tests:

• scripts/test_music_generator_wiring.js
• scripts/test_tools_entry_pages.js

---

Task 1: Add MiniMax Client Tests

Files:

• Create: pywork/music-generator/tests/test_minimax_client.py

• Later create: pywork/music-generator/backend/minimax_client.py

• [ ] Step 1: Write the failing MiniMax client tests

Create pywork/music-generator/tests/test_minimax_client.py with:

import json
from pathlib import Path
import sys

ROOT = Path(__file__).resolve().parents[1]
BACKEND = ROOT / "backend"
sys.path.insert(0, str(BACKEND))

from minimax_client import MiniMaxClient, MiniMaxError, parse_sse_audio, redact_sensitive_text


def test_parse_sse_audio_uses_final_audio_chunk():
    content = "\n".join([
        'data: {"data":{"status":1,"audio":"aaaa"}}',
        'data: {"data":{"status":2,"audio":"68656c6c6f"},"extra_info":{"music_duration":42800}}',
        "data: [DONE]",
    ])

    audio_hex, extra_info = parse_sse_audio(content)

    assert audio_hex == "68656c6c6f"
    assert extra_info["music_duration"] == 42800


def test_parse_sse_audio_ignores_bad_lines():
    content = "\n".join([
        "event: message",
        "data: not-json",
        'data: {"data":{"status":2,"audio":"00ff"}}',
    ])

    audio_hex, extra_info = parse_sse_audio(content)

    assert audio_hex == "00ff"
    assert extra_info == {}


def test_redact_sensitive_text_removes_bearer_and_key_values():
    raw = "Authorization: Bearer abc MINIMAX_API_KEY=unit"

    redacted = redact_sensitive_text(raw)

    assert "Bearer abc" not in redacted
    assert "MINIMAX_API_KEY=unit" not in redacted
    assert "Bearer [REDACTED]" in redacted
    assert "MINIMAX_API_KEY=[REDACTED]" in redacted


def test_client_requires_api_key_for_network_calls():
    client = MiniMaxClient(api_key="", api_host="https://api.minimaxi.com")

    try:
        client.generate_lyrics(prompt="春天", title="")
    except MiniMaxError as exc:
        assert str(exc) == "服务暂不可用"
    else:
        raise AssertionError("expected MiniMaxError")


def test_build_music_payload_for_vocal_and_instrumental():
    client = MiniMaxClient(api_key="unit-test-key", api_host="https://api.minimaxi.com")

    vocal = client.build_music_payload(lyrics="[Verse]\n你好", prompt="pop", instrumental=False)
    instrumental = client.build_music_payload(lyrics="", prompt="ambient", instrumental=True)

    assert vocal == {
        "model": "music-2.6",
        "lyrics": "[Verse]\n你好",
        "prompt": "pop",
        "stream": True,
        "output_format": "hex",
    }
    assert instrumental == {
        "model": "music-2.6",
        "prompt": "ambient",
        "stream": True,
        "output_format": "hex",
        "is_instrumental": True,
    }

• [ ] Step 2: Run the tests and verify they fail for the missing module

Run:

python -m pytest pywork/music-generator/tests/test_minimax_client.py -q

Expected: FAIL with ModuleNotFoundError: No module named 'minimax_client'.

Task 2: Implement MiniMax Client

Files:

• Create: pywork/music-generator/backend/minimax_client.py

• Test: pywork/music-generator/tests/test_minimax_client.py

• [ ] Step 1: Implement the MiniMax client module

Create pywork/music-generator/backend/minimax_client.py with:

"""MiniMax API client for the blog music generator."""

import json
import re
from urllib.error import HTTPError, URLError
from urllib.request import Request, urlopen


class MiniMaxError(Exception):
    """User-safe MiniMax error."""


def redact_sensitive_text(value: str) -> str:
    text = str(value or "")
    text = re.sub(r"Bearer\s+[A-Za-z0-9._-]+", "Bearer [REDACTED]", text)
    text = re.sub(r"(MINIMAX_API_KEY\s*=\s*)[A-Za-z0-9._-]+", r"\1[REDACTED]", text)
    text = re.sub(r"(Authorization:\s*)[^\n\r]+", r"\1[REDACTED]", text, flags=re.I)
    return text


def parse_sse_audio(content: str):
    audio_hex = ""
    extra_info = {}
    for raw_line in str(content or "").splitlines():
        line = raw_line.strip()
        if not line.startswith("data:"):
            continue
        payload_text = line[5:].strip()
        if not payload_text or payload_text == "[DONE]":
            continue
        try:
            payload = json.loads(payload_text)
        except json.JSONDecodeError:
            continue
        if isinstance(payload.get("extra_info"), dict):
            extra_info = payload["extra_info"]
        data = payload.get("data") if isinstance(payload.get("data"), dict) else {}
        if data.get("status") == 2 and data.get("audio"):
            audio_hex = data["audio"]
    return audio_hex, extra_info


class MiniMaxClient:
    def __init__(self, api_key: str, api_host: str, timeout: int = 300):
        self.api_key = str(api_key or "").strip()
        self.api_host = str(api_host or "https://api.minimaxi.com").rstrip("/")
        self.timeout = timeout

    def _require_key(self):
        if not self.api_key:
            raise MiniMaxError("服务暂不可用")

    def _post_json(self, endpoint: str, payload: dict, timeout: int | None = None) -> dict:
        self._require_key()
        url = f"{self.api_host}{endpoint}"
        request = Request(
            url,
            data=json.dumps(payload).encode("utf-8"),
            headers={
                "Authorization": f"Bearer {self.api_key}",
                "Content-Type": "application/json",
            },
            method="POST",
        )
        try:
            with urlopen(request, timeout=timeout or self.timeout) as response:
                return json.loads(response.read().decode("utf-8"))
        except HTTPError as exc:
            body = exc.read().decode("utf-8", errors="replace")
            safe_body = redact_sensitive_text(body[:500])
            raise MiniMaxError(f"上游服务返回错误 HTTP {exc.code}: {safe_body}") from exc
        except (URLError, TimeoutError, json.JSONDecodeError) as exc:
            raise MiniMaxError("上游服务暂不可用") from exc

    def _post_streaming_text(self, endpoint: str, payload: dict, timeout: int | None = None) -> str:
        self._require_key()
        url = f"{self.api_host}{endpoint}"
        request = Request(
            url,
            data=json.dumps(payload).encode("utf-8"),
            headers={
                "Authorization": f"Bearer {self.api_key}",
                "Content-Type": "application/json",
            },
            method="POST",
        )
        try:
            with urlopen(request, timeout=timeout or self.timeout) as response:
                return response.read().decode("utf-8", errors="replace")
        except HTTPError as exc:
            body = exc.read().decode("utf-8", errors="replace")
            safe_body = redact_sensitive_text(body[:500])
            raise MiniMaxError(f"上游服务返回错误 HTTP {exc.code}: {safe_body}") from exc
        except (URLError, TimeoutError) as exc:
            raise MiniMaxError("上游服务暂不可用") from exc

    def generate_lyrics(self, prompt: str, title: str = "") -> dict:
        payload = {
            "mode": "write_full_song",
            "prompt": prompt,
        }
        if title:
            payload["title"] = title
        result = self._post_json("/v1/lyrics_generation", payload, timeout=60)
        return {
            "title": result.get("song_title", ""),
            "style": result.get("style_tags", ""),
            "lyrics": result.get("lyrics", ""),
        }

    def build_music_payload(self, lyrics: str, prompt: str, instrumental: bool) -> dict:
        payload = {
            "model": "music-2.6",
            "prompt": prompt,
            "stream": True,
            "output_format": "hex",
        }
        if instrumental:
            payload["is_instrumental"] = True
        else:
            payload["lyrics"] = lyrics
        return payload

    def generate_music(self, lyrics: str, prompt: str, instrumental: bool) -> tuple[bytes, dict]:
        payload = self.build_music_payload(lyrics=lyrics, prompt=prompt, instrumental=instrumental)
        content = self._post_streaming_text("/v1/music_generation", payload, timeout=300)
        audio_hex, extra_info = parse_sse_audio(content)
        if not audio_hex:
            raise MiniMaxError("音乐生成失败")
        try:
            return bytes.fromhex(audio_hex), extra_info
        except ValueError as exc:
            raise MiniMaxError("音乐生成失败") from exc

• [ ] Step 2: Run MiniMax client tests

Run:

python -m pytest pywork/music-generator/tests/test_minimax_client.py -q

Expected: PASS.

• [ ] Step 3: Commit MiniMax client

Run:

git add pywork/music-generator/backend/minimax_client.py pywork/music-generator/tests/test_minimax_client.py
git commit -m "feat: add minimax music client"

Task 3: Add Store and Rate Limit Tests

Files:

• Create: pywork/music-generator/tests/test_task_store_rate_limit.py

• Later create: pywork/music-generator/backend/config.py

• Later create: pywork/music-generator/backend/task_store.py

• Later create: pywork/music-generator/backend/rate_limit.py

• Later create: pywork/music-generator/backend/cleanup.py

• [ ] Step 1: Write failing tests for persistence, quotas, and cleanup

Create pywork/music-generator/tests/test_task_store_rate_limit.py with:

import time
from pathlib import Path
import sys

ROOT = Path(__file__).resolve().parents[1]
BACKEND = ROOT / "backend"
sys.path.insert(0, str(BACKEND))

from cleanup import cleanup_expired_outputs
from config import AppConfig, hash_ip
from rate_limit import RateLimitResult, check_music_limits
from task_store import TaskStore


def make_config(tmp_path):
    return AppConfig(
        api_key="",
        api_host="https://api.minimaxi.com",
        output_dir=tmp_path / "output",
        db_path=tmp_path / "music.sqlite",
        daily_global_limit=2,
        daily_ip_limit=1,
        daily_lyrics_ip_limit=3,
        max_queue_size=2,
        max_concurrent_jobs=1,
        output_ttl_hours=24,
        usage_salt="unit-test-salt",
    )


def test_hash_ip_is_stable_and_does_not_store_raw_ip(tmp_path):
    config = make_config(tmp_path)

    digest = hash_ip("203.0.113.9", config, day="2026-04-21")

    assert digest == hash_ip("203.0.113.9", config, day="2026-04-21")
    assert "203.0.113.9" not in digest
    assert len(digest) == 64


def test_task_store_creates_and_updates_task(tmp_path):
    config = make_config(tmp_path)
    store = TaskStore(config.db_path)
    store.init_db()

    task_id = store.create_task(
        ip_hash="abc",
        lyrics="[Verse]\nHi",
        prompt="pop",
        instrumental=False,
    )
    store.update_task(task_id, status="processing", progress=40)
    task = store.get_task(task_id)

    assert len(task_id) == 16
    assert task["status"] == "processing"
    assert task["progress"] == 40
    assert task["lyrics"] == "[Verse]\nHi"


def test_usage_counters_and_limits(tmp_path):
    config = make_config(tmp_path)
    store = TaskStore(config.db_path)
    store.init_db()
    ip_hash = "same-ip"

    first = check_music_limits(store, config, ip_hash)
    assert first.allowed is True
    store.increment_usage(day=first.day, ip_hash=ip_hash, usage_type="music")

    second = check_music_limits(store, config, ip_hash)
    assert second == RateLimitResult(
        allowed=False,
        status_code=429,
        message="今日免费额度已用完，明天再来试试",
        day=first.day,
    )


def test_queue_limit_blocks_when_pending_and_processing_are_full(tmp_path):
    config = make_config(tmp_path)
    store = TaskStore(config.db_path)
    store.init_db()
    store.create_task(ip_hash="a", lyrics="1", prompt="pop", instrumental=False)
    task_id = store.create_task(ip_hash="b", lyrics="2", prompt="pop", instrumental=False)
    store.update_task(task_id, status="processing", progress=20)

    result = check_music_limits(store, config, ip_hash="c")

    assert result.allowed is False
    assert result.status_code == 429
    assert result.message == "当前排队人数较多，请稍后再试"


def test_cleanup_deletes_old_outputs(tmp_path):
    config = make_config(tmp_path)
    config.output_dir.mkdir(parents=True)
    old_file = config.output_dir / "aaaaaaaaaaaaaaaa.mp3"
    fresh_file = config.output_dir / "bbbbbbbbbbbbbbbb.mp3"
    old_file.write_bytes(b"old")
    fresh_file.write_bytes(b"fresh")
    old_mtime = time.time() - (25 * 3600)
    old_file.touch(times=(old_mtime, old_mtime))

    deleted = cleanup_expired_outputs(config.output_dir, ttl_hours=24)

    assert old_file.name in deleted
    assert not old_file.exists()
    assert fresh_file.exists()

• [ ] Step 2: Run the tests and verify they fail for missing modules

Run:

python -m pytest pywork/music-generator/tests/test_task_store_rate_limit.py -q

Expected: FAIL with missing cleanup, config, rate_limit, or task_store.

Task 4: Implement Config, Store, Rate Limit, and Cleanup

Files:

• Create: pywork/music-generator/backend/config.py

• Create: pywork/music-generator/backend/task_store.py

• Create: pywork/music-generator/backend/rate_limit.py

• Create: pywork/music-generator/backend/cleanup.py

• Create: pywork/music-generator/requirements.txt

• Create: pywork/music-generator/README.md

• Test: pywork/music-generator/tests/test_task_store_rate_limit.py

• [ ] Step 1: Add configuration module

Create pywork/music-generator/backend/config.py with:

"""Configuration for the blog music generator."""

import hashlib
import os
from dataclasses import dataclass
from pathlib import Path


PROJECT_DIR = Path(__file__).resolve().parents[1]
OUTPUT_DIR = PROJECT_DIR / "output"
DB_PATH = PROJECT_DIR / "music_generator.sqlite"

STYLE_PRESETS = [
    {"id": "pop", "name": "流行", "prompt": "pop, catchy, upbeat, modern production"},
    {"id": "folk", "name": "民谣", "prompt": "folk, acoustic guitar, warm, storytelling"},
    {"id": "rock", "name": "摇滚", "prompt": "rock, energetic, powerful drums, electric guitar"},
    {"id": "jazz", "name": "爵士", "prompt": "jazz, smooth, saxophone, relaxing, slow tempo"},
    {"id": "electronic", "name": "电子", "prompt": "electronic, synth, energetic beat"},
    {"id": "ambient", "name": "氛围", "prompt": "ambient, cinematic, calm, atmospheric"},
]


@dataclass
class AppConfig:
    api_key: str
    api_host: str
    output_dir: Path
    db_path: Path
    daily_global_limit: int
    daily_ip_limit: int
    daily_lyrics_ip_limit: int
    max_queue_size: int
    max_concurrent_jobs: int
    output_ttl_hours: int
    usage_salt: str


def read_int_env(name: str, default: int) -> int:
    raw = os.getenv(name, "").strip()
    if not raw:
        return default
    try:
        return int(raw)
    except ValueError:
        return default


def load_config() -> AppConfig:
    output_dir = Path(os.getenv("MUSIC_OUTPUT_DIR", str(OUTPUT_DIR)))
    db_path = Path(os.getenv("MUSIC_DB_PATH", str(DB_PATH)))
    output_dir.mkdir(parents=True, exist_ok=True)
    db_path.parent.mkdir(parents=True, exist_ok=True)
    return AppConfig(
        api_key=os.getenv("MINIMAX_API_KEY", "").strip(),
        api_host=os.getenv("MINIMAX_API_HOST", "https://api.minimaxi.com").strip(),
        output_dir=output_dir,
        db_path=db_path,
        daily_global_limit=read_int_env("MUSIC_DAILY_GLOBAL_LIMIT", 20),
        daily_ip_limit=read_int_env("MUSIC_DAILY_IP_LIMIT", 2),
        daily_lyrics_ip_limit=read_int_env("MUSIC_DAILY_LYRICS_IP_LIMIT", 20),
        max_queue_size=read_int_env("MUSIC_MAX_QUEUE_SIZE", 5),
        max_concurrent_jobs=read_int_env("MUSIC_MAX_CONCURRENT_JOBS", 1),
        output_ttl_hours=read_int_env("MUSIC_OUTPUT_TTL_HOURS", 24),
        usage_salt=os.getenv("MUSIC_USAGE_SALT", "music-generator-default-salt"),
    )


def hash_ip(ip_address: str, config: AppConfig, day: str) -> str:
    source = f"{day}:{ip_address}:{config.usage_salt}".encode("utf-8")
    return hashlib.sha256(source).hexdigest()

• [ ] Step 2: Add SQLite task store

Create pywork/music-generator/backend/task_store.py with:

"""SQLite task and usage persistence."""

import sqlite3
import time
import uuid
from pathlib import Path


class TaskStore:
    def __init__(self, db_path: Path):
        self.db_path = Path(db_path)

    def connect(self):
        conn = sqlite3.connect(self.db_path)
        conn.row_factory = sqlite3.Row
        return conn

    def init_db(self):
        with self.connect() as conn:
            conn.execute("""
                CREATE TABLE IF NOT EXISTS music_tasks (
                    task_id TEXT PRIMARY KEY,
                    ip_hash TEXT NOT NULL,
                    status TEXT NOT NULL,
                    progress INTEGER NOT NULL,
                    lyrics TEXT NOT NULL,
                    prompt TEXT NOT NULL,
                    instrumental INTEGER NOT NULL,
                    output_file TEXT NOT NULL DEFAULT '',
                    duration REAL NOT NULL DEFAULT 0,
                    error TEXT NOT NULL DEFAULT '',
                    created_at REAL NOT NULL,
                    updated_at REAL NOT NULL
                )
            """)
            conn.execute("""
                CREATE TABLE IF NOT EXISTS music_usage (
                    day TEXT NOT NULL,
                    ip_hash TEXT NOT NULL,
                    usage_type TEXT NOT NULL,
                    count INTEGER NOT NULL,
                    PRIMARY KEY (day, ip_hash, usage_type)
                )
            """)
            conn.commit()

    def create_task(self, ip_hash: str, lyrics: str, prompt: str, instrumental: bool) -> str:
        task_id = uuid.uuid4().hex[:16]
        now = time.time()
        with self.connect() as conn:
            conn.execute(
                """
                INSERT INTO music_tasks (
                    task_id, ip_hash, status, progress, lyrics, prompt,
                    instrumental, created_at, updated_at
                ) VALUES (?, ?, 'pending', 0, ?, ?, ?, ?, ?)
                """,
                (task_id, ip_hash, lyrics, prompt, 1 if instrumental else 0, now, now),
            )
            conn.commit()
        return task_id

    def update_task(self, task_id: str, **fields):
        if not fields:
            return
        fields["updated_at"] = time.time()
        assignments = ", ".join(f"{key} = ?" for key in fields)
        values = list(fields.values()) + [task_id]
        with self.connect() as conn:
            conn.execute(f"UPDATE music_tasks SET {assignments} WHERE task_id = ?", values)
            conn.commit()

    def get_task(self, task_id: str):
        with self.connect() as conn:
            row = conn.execute("SELECT * FROM music_tasks WHERE task_id = ?", (task_id,)).fetchone()
        return dict(row) if row else None

    def count_active_tasks(self) -> int:
        with self.connect() as conn:
            row = conn.execute(
                "SELECT COUNT(*) AS total FROM music_tasks WHERE status IN ('pending', 'processing')"
            ).fetchone()
        return int(row["total"])

    def count_processing_tasks(self) -> int:
        with self.connect() as conn:
            row = conn.execute(
                "SELECT COUNT(*) AS total FROM music_tasks WHERE status = 'processing'"
            ).fetchone()
        return int(row["total"])

    def next_pending_task(self):
        with self.connect() as conn:
            row = conn.execute(
                """
                SELECT * FROM music_tasks
                WHERE status = 'pending'
                ORDER BY created_at ASC
                LIMIT 1
                """
            ).fetchone()
        return dict(row) if row else None

    def increment_usage(self, day: str, ip_hash: str, usage_type: str):
        with self.connect() as conn:
            conn.execute(
                """
                INSERT INTO music_usage (day, ip_hash, usage_type, count)
                VALUES (?, ?, ?, 1)
                ON CONFLICT(day, ip_hash, usage_type)
                DO UPDATE SET count = count + 1
                """,
                (day, ip_hash, usage_type),
            )
            conn.commit()

    def usage_count(self, day: str, ip_hash: str, usage_type: str) -> int:
        with self.connect() as conn:
            row = conn.execute(
                "SELECT count FROM music_usage WHERE day = ? AND ip_hash = ? AND usage_type = ?",
                (day, ip_hash, usage_type),
            ).fetchone()
        return int(row["count"]) if row else 0

    def global_usage_count(self, day: str, usage_type: str) -> int:
        with self.connect() as conn:
            row = conn.execute(
                "SELECT COALESCE(SUM(count), 0) AS total FROM music_usage WHERE day = ? AND usage_type = ?",
                (day, usage_type),
            ).fetchone()
        return int(row["total"])

• [ ] Step 3: Add rate-limit checks

Create pywork/music-generator/backend/rate_limit.py with:

"""Quota checks for free music generation."""

from dataclasses import dataclass
from datetime import datetime, timezone


@dataclass(frozen=True)
class RateLimitResult:
    allowed: bool
    status_code: int
    message: str
    day: str


def today_utc() -> str:
    return datetime.now(timezone.utc).strftime("%Y-%m-%d")


def check_music_limits(store, config, ip_hash: str) -> RateLimitResult:
    day = today_utc()
    if store.count_active_tasks() >= config.max_queue_size:
        return RateLimitResult(False, 429, "当前排队人数较多，请稍后再试", day)
    if store.global_usage_count(day, "music") >= config.daily_global_limit:
        return RateLimitResult(False, 429, "今日免费额度已用完，明天再来试试", day)
    if store.usage_count(day, ip_hash, "music") >= config.daily_ip_limit:
        return RateLimitResult(False, 429, "今日免费额度已用完，明天再来试试", day)
    return RateLimitResult(True, 200, "", day)


def check_lyrics_limits(store, config, ip_hash: str) -> RateLimitResult:
    day = today_utc()
    if store.usage_count(day, ip_hash, "lyrics") >= config.daily_lyrics_ip_limit:
        return RateLimitResult(False, 429, "今日歌词生成额度已用完，明天再来试试", day)
    return RateLimitResult(True, 200, "", day)

• [ ] Step 4: Add cleanup helper

Create pywork/music-generator/backend/cleanup.py with:

"""Cleanup generated MP3 files."""

import time
from pathlib import Path


def cleanup_expired_outputs(output_dir: Path, ttl_hours: int) -> list[str]:
    output_dir = Path(output_dir)
    if not output_dir.exists():
        return []
    threshold = time.time() - ttl_hours * 3600
    deleted = []
    for path in output_dir.glob("*.mp3"):
        if path.stat().st_mtime < threshold:
            path.unlink()
            deleted.append(path.name)
    return deleted

• [ ] Step 5: Add requirements and README

Create pywork/music-generator/requirements.txt with:

flask
pytest

Create pywork/music-generator/README.md with:

# Music Generator Backend

Flask backend for the blog AI music generator.

## Local run

```bash
cd pywork/music-generator/backend
export MINIMAX_API_KEY="your-server-side-key"
python app.py
```

## Production notes

Do not commit `.env` files or API keys. Production must provide `MINIMAX_API_KEY` outside the rsync-managed repository directory.

• [ ] Step 6: Run store/rate-limit tests

Run:

python -m pytest pywork/music-generator/tests/test_task_store_rate_limit.py -q

Expected: PASS.

• [ ] Step 7: Commit persistence and rate-limit modules

Run:

git add pywork/music-generator/backend/config.py \
  pywork/music-generator/backend/task_store.py \
  pywork/music-generator/backend/rate_limit.py \
  pywork/music-generator/backend/cleanup.py \
  pywork/music-generator/requirements.txt \
  pywork/music-generator/README.md \
  pywork/music-generator/tests/test_task_store_rate_limit.py
git commit -m "feat: add music task store and rate limits"

Task 5: Add Flask API Behavior Tests

Files:

• Create: pywork/music-generator/tests/test_app_behavior.py

• Later create: pywork/music-generator/backend/app.py

• [ ] Step 1: Write failing Flask app tests

Create pywork/music-generator/tests/test_app_behavior.py with:

import importlib
from pathlib import Path
import sys

ROOT = Path(__file__).resolve().parents[1]
BACKEND = ROOT / "backend"
sys.path.insert(0, str(BACKEND))


def load_app(tmp_path, monkeypatch):
    monkeypatch.setenv("MUSIC_OUTPUT_DIR", str(tmp_path / "output"))
    monkeypatch.setenv("MUSIC_DB_PATH", str(tmp_path / "music.sqlite"))
    monkeypatch.setenv("MUSIC_DAILY_GLOBAL_LIMIT", "2")
    monkeypatch.setenv("MUSIC_DAILY_IP_LIMIT", "1")
    monkeypatch.setenv("MUSIC_DAILY_LYRICS_IP_LIMIT", "2")
    monkeypatch.delenv("MINIMAX_API_KEY", raising=False)
    module = importlib.import_module("app")
    module = importlib.reload(module)
    return module


def test_config_has_no_token_details(tmp_path, monkeypatch):
    module = load_app(tmp_path, monkeypatch)
    client = module.app.test_client()

    response = client.get("/api/config")
    body = response.get_json()

    assert response.status_code == 200
    assert body["success"] is True
    text = response.get_data(as_text=True)
    assert "MINIMAX_API_KEY" not in text
    assert "Bearer" not in text
    assert "sk-" not in text


def test_lyrics_validation_rejects_missing_prompt(tmp_path, monkeypatch):
    module = load_app(tmp_path, monkeypatch)
    client = module.app.test_client()

    response = client.post("/api/lyrics", json={"prompt": ""})

    assert response.status_code == 400
    assert response.get_json()["error"] == "请输入歌曲主题"


def test_lyrics_missing_token_returns_safe_503(tmp_path, monkeypatch):
    module = load_app(tmp_path, monkeypatch)
    client = module.app.test_client()

    response = client.post("/api/lyrics", json={"prompt": "一首关于春天的歌"})

    assert response.status_code == 503
    assert response.get_json()["error"] == "服务暂不可用"
    assert "MINIMAX_API_KEY" not in response.get_data(as_text=True)


def test_music_start_rejects_missing_lyrics_for_vocal(tmp_path, monkeypatch):
    module = load_app(tmp_path, monkeypatch)
    client = module.app.test_client()

    response = client.post("/api/music/start", json={"lyrics": "", "prompt": "pop", "instrumental": False})

    assert response.status_code == 400
    assert response.get_json()["error"] == "请输入歌词"


def test_music_start_accepts_instrumental_without_token_until_service_check(tmp_path, monkeypatch):
    module = load_app(tmp_path, monkeypatch)
    client = module.app.test_client()

    response = client.post("/api/music/start", json={"lyrics": "", "prompt": "ambient", "instrumental": True})

    assert response.status_code == 503
    assert response.get_json()["error"] == "服务暂不可用"


def test_download_rejects_path_traversal(tmp_path, monkeypatch):
    module = load_app(tmp_path, monkeypatch)
    client = module.app.test_client()

    response = client.get("/api/download/../../secret.mp3")

    assert response.status_code == 404


def test_download_serves_valid_mp3(tmp_path, monkeypatch):
    module = load_app(tmp_path, monkeypatch)
    module.config.output_dir.mkdir(parents=True, exist_ok=True)
    filename = "aaaaaaaaaaaaaaaa.mp3"
    (module.config.output_dir / filename).write_bytes(b"mp3")
    client = module.app.test_client()

    response = client.get(f"/api/download/{filename}")

    assert response.status_code == 200
    assert response.data == b"mp3"

• [ ] Step 2: Run test and verify it fails for missing app module

Run:

python -m pytest pywork/music-generator/tests/test_app_behavior.py -q

Expected: FAIL with ModuleNotFoundError: No module named 'app'.

Task 6: Implement Flask App and Worker

Files:

• Create: pywork/music-generator/backend/app.py

• Test: pywork/music-generator/tests/test_app_behavior.py

• Test: pywork/music-generator/tests/test_minimax_client.py

• Test: pywork/music-generator/tests/test_task_store_rate_limit.py

• [ ] Step 1: Implement Flask routes and worker loop

Create pywork/music-generator/backend/app.py with:

"""Flask API for the blog music generator."""

import re
import threading
import time
from pathlib import Path

from flask import Flask, jsonify, request, send_file

from cleanup import cleanup_expired_outputs
from config import STYLE_PRESETS, hash_ip, load_config
from minimax_client import MiniMaxClient, MiniMaxError
from rate_limit import check_lyrics_limits, check_music_limits, today_utc
from task_store import TaskStore


app = Flask(__name__)
config = load_config()
store = TaskStore(config.db_path)
store.init_db()
worker_lock = threading.Lock()


def json_error(message: str, status_code: int):
    return jsonify({"success": False, "error": message}), status_code


def request_ip() -> str:
    forwarded = request.headers.get("X-Forwarded-For", "")
    if forwarded:
        return forwarded.split(",", 1)[0].strip()
    return request.remote_addr or "0.0.0.0"


def current_ip_hash(day: str) -> str:
    return hash_ip(request_ip(), config, day)


def validate_text(value, min_length, max_length, empty_message, too_long_message):
    text = str(value or "").strip()
    if len(text) < min_length:
        raise ValueError(empty_message)
    if len(text) > max_length:
        raise ValueError(too_long_message)
    return text


def public_download_url(filename: str) -> str:
    return f"/tools/music-generator/api/download/{filename}"


def run_worker_once():
    if not worker_lock.acquire(blocking=False):
        return
    try:
        while store.count_processing_tasks() < config.max_concurrent_jobs:
            task = store.next_pending_task()
            if not task:
                break
            process_task(task)
    finally:
        worker_lock.release()


def process_task(task):
    task_id = task["task_id"]
    store.update_task(task_id, status="processing", progress=20)
    client = MiniMaxClient(api_key=config.api_key, api_host=config.api_host)
    try:
        audio_bytes, extra_info = client.generate_music(
            lyrics=task["lyrics"],
            prompt=task["prompt"],
            instrumental=bool(task["instrumental"]),
        )
        filename = f"{task_id}.mp3"
        config.output_dir.mkdir(parents=True, exist_ok=True)
        output_path = config.output_dir / filename
        output_path.write_bytes(audio_bytes)
        duration = float(extra_info.get("music_duration", 0) or 0) / 1000
        store.update_task(
            task_id,
            status="completed",
            progress=100,
            output_file=filename,
            duration=duration,
            error="",
        )
    except MiniMaxError as exc:
        store.update_task(task_id, status="failed", progress=100, error=str(exc))
    except Exception:
        store.update_task(task_id, status="failed", progress=100, error="音乐生成失败")


def start_worker_thread():
    thread = threading.Thread(target=run_worker_once, daemon=True)
    thread.start()


@app.route("/api/config")
def get_config():
    return jsonify({
        "success": True,
        "limits": {
            "daily_global_limit": config.daily_global_limit,
            "daily_ip_limit": config.daily_ip_limit,
            "daily_lyrics_ip_limit": config.daily_lyrics_ip_limit,
            "max_queue_size": config.max_queue_size,
        },
        "styles": STYLE_PRESETS,
    })


@app.route("/api/lyrics", methods=["POST"])
def generate_lyrics():
    data = request.get_json(silent=True) or {}
    try:
        prompt = validate_text(data.get("prompt"), 5, 500, "请输入歌曲主题", "歌曲主题不能超过 500 字")
        title = str(data.get("title") or "").strip()
        if len(title) > 80:
            return json_error("歌名不能超过 80 字", 400)
    except ValueError as exc:
        return json_error(str(exc), 400)
    day = today_utc()
    ip_hash = current_ip_hash(day)
    limit = check_lyrics_limits(store, config, ip_hash)
    if not limit.allowed:
        return json_error(limit.message, limit.status_code)
    if not config.api_key:
        return json_error("服务暂不可用", 503)
    client = MiniMaxClient(api_key=config.api_key, api_host=config.api_host)
    try:
        result = client.generate_lyrics(prompt=prompt, title=title)
    except MiniMaxError as exc:
        return json_error(str(exc), 503)
    store.increment_usage(day=limit.day, ip_hash=ip_hash, usage_type="lyrics")
    return jsonify({"success": True, **result})


@app.route("/api/music/start", methods=["POST"])
def start_music():
    data = request.get_json(silent=True) or {}
    instrumental = bool(data.get("instrumental", False))
    lyrics = str(data.get("lyrics") or "").strip()
    prompt = str(data.get("prompt") or "").strip()
    if not instrumental and not lyrics:
        return json_error("请输入歌词", 400)
    if not instrumental and len(lyrics) > 3500:
        return json_error("歌词不能超过 3500 字", 400)
    if instrumental and not prompt:
        return json_error("请输入纯音乐风格描述", 400)
    if len(prompt) > 500:
        return json_error("风格描述不能超过 500 字", 400)
    day = today_utc()
    ip_hash = current_ip_hash(day)
    limit = check_music_limits(store, config, ip_hash)
    if not limit.allowed:
        return json_error(limit.message, limit.status_code)
    if not config.api_key:
        return json_error("服务暂不可用", 503)
    task_id = store.create_task(ip_hash=ip_hash, lyrics=lyrics, prompt=prompt, instrumental=instrumental)
    store.increment_usage(day=limit.day, ip_hash=ip_hash, usage_type="music")
    start_worker_thread()
    return jsonify({"success": True, "task_id": task_id, "status": "pending"})


@app.route("/api/music/status/<task_id>")
def music_status(task_id):
    if not re.fullmatch(r"[a-f0-9]{16}", task_id or ""):
        return json_error("任务不存在", 404)
    task = store.get_task(task_id)
    if not task:
        return json_error("任务不存在", 404)
    response = {
        "success": True,
        "task_id": task_id,
        "status": task["status"],
        "progress": task["progress"],
    }
    if task["status"] == "completed":
        response["download_url"] = public_download_url(task["output_file"])
        response["duration"] = task["duration"]
    if task["status"] == "failed":
        response["error"] = task["error"] or "音乐生成失败"
    return jsonify(response)


@app.route("/api/download/<path:filename>")
def download(filename):
    if not re.fullmatch(r"[a-f0-9]{16}\.mp3", filename or ""):
        return json_error("文件不存在", 404)
    path = (config.output_dir / filename).resolve()
    output_root = config.output_dir.resolve()
    if output_root not in path.parents or not path.exists():
        return json_error("文件不存在", 404)
    return send_file(str(path), mimetype="audio/mpeg", as_attachment=True, download_name=filename)


@app.route("/api/cleanup", methods=["POST"])
def cleanup():
    deleted = cleanup_expired_outputs(config.output_dir, config.output_ttl_hours)
    return jsonify({"success": True, "deleted": len(deleted)})


if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5003, debug=True)

• [ ] Step 2: Run Flask app tests

Run:

python -m pytest pywork/music-generator/tests/test_app_behavior.py -q

Expected: PASS.

• [ ] Step 3: Run all backend tests

Run:

python -m pytest pywork/music-generator/tests -q

Expected: PASS.

• [ ] Step 4: Commit Flask backend

Run:

git add pywork/music-generator/backend/app.py pywork/music-generator/tests/test_app_behavior.py
git commit -m "feat: add music generator api"

Task 7: Add Frontend Wiring Tests

Files:

• Create: scripts/test_music_generator_wiring.js

• Modify: scripts/test_tools_entry_pages.js

• Later create: docs/.vuepress/components/MusicGeneratorTool.vue

• Later create: docs/07.工具/20.学习工具/50.AI音乐生成器.md

• Later modify: docs/.vuepress/config.ts

• Later modify: docs/07.工具/00.概览.md

• Later modify: docs/07.工具/20.学习工具/20.AI播客生成器.md

• Later modify: docs/07.工具/10.测试工具/71.样例音频库.md

• [ ] Step 1: Add failing music generator wiring test

Create scripts/test_music_generator_wiring.js with:

const assert = require('assert')
const fs = require('fs')
const path = require('path')

const root = path.resolve(__dirname, '..')

function read(relativePath) {
  return fs.readFileSync(path.join(root, relativePath), 'utf8')
}

function exists(relativePath) {
  return fs.existsSync(path.join(root, relativePath))
}

assert.ok(exists('docs/.vuepress/components/MusicGeneratorTool.vue'), 'MusicGeneratorTool.vue should exist')
assert.ok(exists('docs/07.工具/20.学习工具/50.AI音乐生成器.md'), 'music generator page should exist')

const component = read('docs/.vuepress/components/MusicGeneratorTool.vue')
assert.match(component, /name:\s*'MusicGeneratorTool'/)
assert.match(component, /const API_BASE = '\/tools\/music-generator\/api'/)
assert.match(component, /tools_music_generator_view/)
assert.match(component, /generateLyrics/)
assert.match(component, /startMusicGeneration/)
assert.match(component, /pollTaskStatus/)
assert.doesNotMatch(component, /MINIMAX_API_KEY/)
assert.doesNotMatch(component, /Bearer\s+[A-Za-z0-9._-]+/)
assert.doesNotMatch(component, /sk-[A-Za-z0-9._-]+/)

const page = read('docs/07.工具/20.学习工具/50.AI音乐生成器.md')
assert.match(page, /title:\s*AI音乐生成器/)
assert.match(page, /permalink:\s*\/tools\/music-generator\//)
assert.match(page, /<MusicGeneratorTool\s*\/>/)

const config = read('docs/.vuepress/config.ts')
assert.match(config, /AI Music Generator/)
assert.match(config, /\/tools\/music-generator\//)

const overview = read('docs/07.工具/00.概览.md')
assert.match(overview, /\[AI音乐生成器\]\(\/tools\/music-generator\/\)/)

const podcast = read('docs/07.工具/20.学习工具/20.AI播客生成器.md')
assert.match(podcast, /\[AI音乐生成器\]\(\/tools\/music-generator\/\)/)

const audioSamples = read('docs/07.工具/10.测试工具/71.样例音频库.md')
assert.match(audioSamples, /\[AI音乐生成器\]\(\/tools\/music-generator\/\)/)

console.log('music generator wiring tests passed')

• [ ] Step 2: Extend tools entry page test

Modify scripts/test_tools_entry_pages.js to include:

const musicPagePath = 'docs/07.工具/20.学习工具/50.AI音乐生成器.md'
assert.ok(exists(musicPagePath), 'music generator page should exist')
const musicPage = read(musicPagePath)
assert.match(musicPage, /permalink:\s*\/tools\/music-generator\//)
assert.match(musicPage, /<MusicGeneratorTool\s*\/>/)

• [ ] Step 3: Run wiring tests and verify they fail

Run:

node scripts/test_music_generator_wiring.js
node scripts/test_tools_entry_pages.js

Expected: first command FAIL because component/page do not exist yet; second command FAIL because the new page is not present yet.

Task 8: Implement VuePress Music Tool Page

Files:

• Create: docs/.vuepress/components/MusicGeneratorTool.vue

• Create: docs/07.工具/20.学习工具/50.AI音乐生成器.md

• Modify: docs/.vuepress/config.ts

• Modify: docs/07.工具/00.概览.md

• Modify: docs/07.工具/20.学习工具/20.AI播客生成器.md

• Modify: docs/07.工具/10.测试工具/71.样例音频库.md

• Test: scripts/test_music_generator_wiring.js

• Test: scripts/test_tools_entry_pages.js

• [ ] Step 1: Add MusicGeneratorTool.vue

Create docs/.vuepress/components/MusicGeneratorTool.vue with this component structure and methods:

<template>
  <section class="music-generator-tool">
    <header class="tool-hero">
      <p class="eyebrow">Free AI Music Tool</p>
      <h1>AI音乐生成器</h1>
      <p class="summary">输入主题或歌词，免费生成可播放、可下载的 AI 音乐。生成通常需要 1-5 分钟，免费额度按天限制。</p>
    </header>

    <section class="mode-switch" aria-label="生成模式">
      <button :class="{ active: mode === 'song' }" type="button" @click="mode = 'song'">主题写歌</button>
      <button :class="{ active: mode === 'instrumental' }" type="button" @click="mode = 'instrumental'">纯音乐</button>
    </section>

    <section class="tool-panel">
      <div v-if="mode === 'song'" class="form-stack">
        <label for="musicTheme">歌曲主题</label>
        <textarea id="musicTheme" v-model.trim="theme" rows="4" placeholder="例如：一首关于春天、通勤和重新出发的流行歌" />

        <label for="musicTitle">歌名（可选）</label>
        <input id="musicTitle" v-model.trim="title" type="text" placeholder="留空则自动生成" />

        <button class="primary-btn" type="button" :disabled="lyricsLoading" @click="generateLyrics">
          {{ lyricsLoading ? '歌词生成中...' : '生成歌词' }}
        </button>

        <label for="lyricsText">歌词</label>
        <textarea id="lyricsText" v-model.trim="lyrics" rows="12" placeholder="[Verse]\n在这里编辑歌词" />
      </div>

      <div class="form-stack">
        <label for="stylePreset">风格预设</label>
        <select id="stylePreset" v-model="selectedStyle" @change="applySelectedStyle">
          <option value="">自定义风格</option>
          <option v-for="style in styles" :key="style.id" :value="style.id">{{ style.name }}</option>
        </select>

        <label for="stylePrompt">风格描述</label>
        <textarea id="stylePrompt" v-model.trim="stylePrompt" rows="3" placeholder="例如：pop, catchy, upbeat, modern production" />

        <button class="primary-btn" type="button" :disabled="musicLoading" @click="startMusicGeneration">
          {{ musicLoading ? '音乐生成中...' : mode === 'instrumental' ? '生成纯音乐' : '生成音乐' }}
        </button>
      </div>
    </section>

    <section v-if="statusText" class="status-panel" :class="statusKind">
      <p>{{ statusText }}</p>
      <div v-if="musicLoading" class="progress-bar">
        <div class="progress-fill" :style="{ width: `${progress}%` }" />
      </div>
    </section>

    <section v-if="audioUrl" class="result-panel">
      <h2>生成完成</h2>
      <audio :src="audioUrl" controls preload="metadata" />
      <a class="download-btn" :href="audioUrl" download>下载 MP3</a>
    </section>
  </section>
</template>

<script>
const API_BASE = '/tools/music-generator/api'
const TRACKING_ENDPOINT = 'https://wangmouren.online:9000/save_data_pv'

export default {
  name: 'MusicGeneratorTool',
  data() {
    return {
      mode: 'song',
      theme: '',
      title: '',
      lyrics: '',
      stylePrompt: '',
      selectedStyle: '',
      styles: [],
      lyricsLoading: false,
      musicLoading: false,
      taskId: '',
      progress: 0,
      statusText: '',
      statusKind: '',
      audioUrl: '',
      pollTimer: null,
    }
  },
  mounted() {
    this.loadConfig()
    this.trackView()
  },
  beforeDestroy() {
    if (this.pollTimer) {
      clearTimeout(this.pollTimer)
    }
  },
  methods: {
    async loadConfig() {
      const response = await fetch(`${API_BASE}/config`)
      const data = await response.json()
      if (data.success) {
        this.styles = data.styles || []
      }
    },
    async trackView() {
      try {
        await fetch(TRACKING_ENDPOINT, {
          method: 'POST',
          headers: { 'Content-Type': 'application/json' },
          body: JSON.stringify({ event_type: 'tools_music_generator_view', page_name: 'tools/music-generator' }),
        })
      } catch (error) {
        // Tracking must not block tool usage.
      }
    },
    applySelectedStyle() {
      const style = this.styles.find((item) => item.id === this.selectedStyle)
      if (style) {
        this.stylePrompt = style.prompt
      }
    },
    setStatus(text, kind = 'info') {
      this.statusText = text
      this.statusKind = kind
    },
    async generateLyrics() {
      if (!this.theme) {
        this.setStatus('请输入歌曲主题', 'error')
        return
      }
      this.lyricsLoading = true
      this.setStatus('歌词生成中...', 'info')
      try {
        const response = await fetch(`${API_BASE}/lyrics`, {
          method: 'POST',
          headers: { 'Content-Type': 'application/json' },
          body: JSON.stringify({ prompt: this.theme, title: this.title, mode: 'write_full_song' }),
        })
        const data = await response.json()
        if (!response.ok || !data.success) {
          throw new Error(data.error || '歌词生成失败，请稍后重试')
        }
        this.title = data.title || this.title
        this.stylePrompt = data.style || this.stylePrompt
        this.lyrics = data.lyrics || ''
        this.setStatus('歌词已生成，可继续编辑后生成音乐', 'success')
      } catch (error) {
        this.setStatus(error.message || '歌词生成失败，请稍后重试', 'error')
      } finally {
        this.lyricsLoading = false
      }
    },
    async startMusicGeneration() {
      if (this.mode === 'song' && !this.lyrics) {
        this.setStatus('请输入歌词', 'error')
        return
      }
      if (this.mode === 'instrumental' && !this.stylePrompt) {
        this.setStatus('请输入纯音乐风格描述', 'error')
        return
      }
      this.musicLoading = true
      this.audioUrl = ''
      this.progress = 5
      this.setStatus('任务已提交，正在排队生成...', 'info')
      try {
        const response = await fetch(`${API_BASE}/music/start`, {
          method: 'POST',
          headers: { 'Content-Type': 'application/json' },
          body: JSON.stringify({
            lyrics: this.mode === 'song' ? this.lyrics : '',
            prompt: this.stylePrompt,
            instrumental: this.mode === 'instrumental',
          }),
        })
        const data = await response.json()
        if (!response.ok || !data.success) {
          throw new Error(data.error || '音乐生成失败，请稍后重试')
        }
        this.taskId = data.task_id
        this.pollTaskStatus()
      } catch (error) {
        this.musicLoading = false
        this.setStatus(error.message || '音乐生成失败，请稍后重试', 'error')
      }
    },
    async pollTaskStatus() {
      if (!this.taskId) return
      try {
        const response = await fetch(`${API_BASE}/music/status/${this.taskId}`)
        const data = await response.json()
        if (!response.ok || !data.success) {
          throw new Error(data.error || '任务状态查询失败')
        }
        this.progress = data.progress || this.progress
        if (data.status === 'completed') {
          this.musicLoading = false
          this.audioUrl = data.download_url
          this.setStatus('音乐生成完成', 'success')
          return
        }
        if (data.status === 'failed') {
          throw new Error(data.error || '音乐生成失败，请调整歌词或风格后重试')
        }
        this.setStatus('音乐生成中，通常需要 1-5 分钟...', 'info')
        this.pollTimer = setTimeout(() => this.pollTaskStatus(), 5000)
      } catch (error) {
        this.musicLoading = false
        this.setStatus(error.message || '生成时间较长，请稍后刷新任务状态', 'error')
      }
    },
  },
}
</script>

<style scoped>
.music-generator-tool {
  max-width: 980px;
  margin: 0 auto;
  padding: 24px 0 48px;
  color: #1f2937;
}
.tool-hero {
  padding: 28px 0 18px;
}
.eyebrow {
  color: #0f766e;
  font-weight: 700;
  letter-spacing: 0;
}
.tool-hero h1 {
  margin: 0 0 12px;
  font-size: 36px;
}
.summary {
  max-width: 720px;
  color: #4b5563;
  line-height: 1.7;
}
.mode-switch {
  display: flex;
  gap: 8px;
  margin: 18px 0;
}
.mode-switch button,
.primary-btn,
.download-btn {
  border: 1px solid #0f766e;
  background: #fff;
  color: #0f766e;
  border-radius: 8px;
  padding: 10px 14px;
  cursor: pointer;
  font-weight: 700;
}
.mode-switch button.active,
.primary-btn,
.download-btn {
  background: #0f766e;
  color: #fff;
}
.tool-panel,
.status-panel,
.result-panel {
  border: 1px solid #e5e7eb;
  border-radius: 8px;
  padding: 20px;
  margin: 16px 0;
  background: #fff;
}
.form-stack {
  display: grid;
  gap: 10px;
  margin-bottom: 18px;
}
textarea,
input,
select {
  width: 100%;
  border: 1px solid #cbd5e1;
  border-radius: 8px;
  padding: 10px 12px;
  font: inherit;
}
.status-panel.error {
  border-color: #fecaca;
  background: #fef2f2;
  color: #991b1b;
}
.status-panel.success {
  border-color: #bbf7d0;
  background: #f0fdf4;
  color: #166534;
}
.progress-bar {
  height: 10px;
  border-radius: 999px;
  background: #e5e7eb;
  overflow: hidden;
}
.progress-fill {
  height: 100%;
  background: #0f766e;
  transition: width 0.3s ease;
}
.result-panel audio {
  width: 100%;
  margin-bottom: 12px;
}
@media (max-width: 640px) {
  .tool-hero h1 {
    font-size: 28px;
  }
  .mode-switch {
    flex-direction: column;
  }
}
</style>

• [ ] Step 2: Add Markdown page

Create docs/07.工具/20.学习工具/50.AI音乐生成器.md with:

---
title: AI音乐生成器
date: 2026-04-21T00:00:00.000Z
permalink: /tools/music-generator/
sidebar: auto
article: false
comment: false
author:
  name: wangyang
description: 免费在线 AI 音乐生成器，支持主题生成歌词、歌词生成歌曲和纯音乐生成，可在线播放与下载 MP3。
keywords: AI音乐生成器,在线音乐生成,免费AI音乐工具,歌词生成歌曲,纯音乐生成
head:
  - - script
    - type: application/ld+json
    - '{"@context":"https://schema.org","@type":"SoftwareApplication","name":"AI音乐生成器","applicationCategory":"WebApplication","operatingSystem":"Web","offers":{"@type":"Offer","price":"0","priceCurrency":"CNY","availability":"https://schema.org/InStock"},"isAccessibleForFree":true,"url":"https://wangmouren.online/tools/music-generator/","description":"免费在线 AI 音乐生成器，支持主题生成歌词、歌词生成歌曲和纯音乐生成，可在线播放与下载 MP3。","keywords":"AI音乐生成器,在线音乐生成,免费AI音乐工具,歌词生成歌曲,纯音乐生成"}'
---

输入主题或歌词生成 AI 音乐。当前免费开放使用，但每日额度有限；音乐生成通常需要 1-5 分钟。

<MusicGeneratorTool />

## 使用场景
- 给短视频、课程或播客片头快速生成音乐草稿。
- 根据主题生成歌词，再继续调整成可分享的歌曲 Demo。

## 场景案例
- 场景案例：内容创作者输入“春天、通勤、重新出发”，生成一首轻快流行歌作为视频草稿配乐。

## 延伸阅读
- [AI播客生成器](/tools/podcast-generator/)
- [音频测试文件下载](/tools/test-audio-files/)

## 常见问题
### 可以免费用吗？
当前免费开放，但每日额度有限，用完后需要第二天再试。

### 会暴露我的 API Key 吗？
访客不需要输入 API Key。站点密钥只保存在服务器端，不会出现在页面代码里。

### 生成要多久？
通常需要 1-5 分钟，取决于排队情况和上游生成速度。

• [ ] Step 3: Update nav and related pages

In docs/.vuepress/config.ts, add this item in the Tools menu after AI Podcast Generator:

{text: 'AI Music Generator', link: '/tools/music-generator/'},

In docs/07.工具/00.概览.md, add under 学习工具:

- [AI音乐生成器](/tools/music-generator/)（免费生成歌词、歌曲和纯音乐，每日额度有限）

In docs/07.工具/20.学习工具/20.AI播客生成器.md, add under 延伸阅读:

- [AI音乐生成器](/tools/music-generator/)

In docs/07.工具/10.测试工具/71.样例音频库.md, add under 延伸阅读:

- [AI音乐生成器](/tools/music-generator/)

• [ ] Step 4: Run frontend wiring tests

Run:

node scripts/test_music_generator_wiring.js
node scripts/test_tools_entry_pages.js

Expected: PASS.

• [ ] Step 5: Commit frontend page

Run:

git add docs/.vuepress/components/MusicGeneratorTool.vue \
  docs/07.工具/20.学习工具/50.AI音乐生成器.md \
  docs/.vuepress/config.ts \
  docs/07.工具/00.概览.md \
  docs/07.工具/20.学习工具/20.AI播客生成器.md \
  docs/07.工具/10.测试工具/71.样例音频库.md \
  scripts/test_music_generator_wiring.js \
  scripts/test_tools_entry_pages.js
git commit -m "feat: add music generator page"

Task 9: Add Deployment Scripts and Smoke Tests

Files:

• Create: scripts/remote_configure_music_nginx.sh

• Modify: scripts/deploy_prod.sh

• Modify: scripts/remote_restart.sh

• Modify: scripts/smoke_test.sh

• [ ] Step 1: Add music nginx configuration script

Create scripts/remote_configure_music_nginx.sh with:

#!/usr/bin/env bash

set -euo pipefail

NGINX_CONF="/opt/conf/nginx.conf"

cp "${NGINX_CONF}" "${NGINX_CONF}.bak.$(date +%Y%m%d%H%M%S)"

python3 - "${NGINX_CONF}" <<'PY'
import re
import sys
from pathlib import Path

conf_path = Path(sys.argv[1])
text = conf_path.read_text(encoding="utf-8")

api_block = """        location /tools/music-generator/api/ {
            proxy_pass http://127.0.0.1:5003/api/;
            proxy_set_header Host $host;
            proxy_set_header X-Real-IP $remote_addr;
            proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
            proxy_set_header X-Forwarded-Proto $scheme;
            proxy_connect_timeout 30s;
            proxy_send_timeout 360s;
            proxy_read_timeout 360s;
            send_timeout 360s;
        }"""

def replace_or_insert(src: str, location_pattern: str, desired_block: str) -> str:
    block_regex = re.compile(location_pattern, re.S)
    if block_regex.search(src):
        return block_regex.sub(desired_block, src, count=1)
    anchor_regex = re.compile(r"(location\s+/\s*\{[^}]*\}\s*)", re.S)
    match = anchor_regex.search(src)
    if not match:
        raise SystemExit("Failed to locate nginx root location block for insertion")
    insert_at = match.end()
    return src[:insert_at] + "\n" + desired_block + "\n\n" + src[insert_at:]

text = replace_or_insert(
    text,
    r"location\s+/tools/music-generator/api/\s*\{[^}]*\}",
    api_block,
)

conf_path.write_text(text, encoding="utf-8")
PY

/opt/sbin/nginx -t
echo "Music generator nginx route configured/updated."

• [ ] Step 2: Update deploy script

In scripts/deploy_prod.sh, after podcast nginx configuration, add:

echo "[5/7] Configuring music generator nginx proxy routes (idempotent)..."
remote_run_script "scripts/remote_configure_music_nginx.sh"

Then renumber later echo labels so restart is [6/7] and smoke tests are [7/7].

• [ ] Step 3: Update remote restart script

In scripts/remote_restart.sh, add variables near the podcast variables:

MUSIC_DIR="${PYWORK_DIR}/music-generator/backend"
MUSIC_BIND="127.0.0.1:5003"
MUSIC_APP="app:app"
MUSIC_LOG="${PYWORK_DIR}/music-generator/output/music_gunicorn.log"
MUSIC_TIMEOUT="${MUSIC_TIMEOUT:-360}"
MUSIC_CONDA_ENV="${MUSIC_CONDA_ENV:-podcast310}"
MUSIC_GUNICORN_BIN="${MUSIC_GUNICORN_BIN:-/root/miniconda3/envs/${MUSIC_CONDA_ENV}/bin/gunicorn}"

After starting podcast gunicorn, add a matching stop/start block:

if [ ! -x "${MUSIC_GUNICORN_BIN}" ]; then
  echo "WARN: ${MUSIC_GUNICORN_BIN} not found, fallback to ${GUNICORN_BIN}"
  MUSIC_GUNICORN_BIN="${GUNICORN_BIN}"
fi

echo "Stopping existing music gunicorn process (if any)..."
MUSIC_OLD_PIDS="$(pgrep -f "gunicorn.*${MUSIC_BIND}.*${MUSIC_APP}" || true)"
if [ -z "${MUSIC_OLD_PIDS}" ]; then
  MUSIC_OLD_PIDS="$(pgrep -f "gunicorn.*${MUSIC_APP}.*${MUSIC_BIND}" || true)"
fi
if [ -n "${MUSIC_OLD_PIDS}" ]; then
  kill ${MUSIC_OLD_PIDS}
  sleep 2
fi

MUSIC_REMAINING_PIDS="$(pgrep -f "gunicorn.*${MUSIC_BIND}.*${MUSIC_APP}" || true)"
if [ -z "${MUSIC_REMAINING_PIDS}" ]; then
  MUSIC_REMAINING_PIDS="$(pgrep -f "gunicorn.*${MUSIC_APP}.*${MUSIC_BIND}" || true)"
fi
if [ -n "${MUSIC_REMAINING_PIDS}" ]; then
  kill -9 ${MUSIC_REMAINING_PIDS}
  sleep 1
fi

echo "Starting music generator gunicorn..."
mkdir -p "$(dirname "${MUSIC_LOG}")"
cd "${MUSIC_DIR}"
nohup "${MUSIC_GUNICORN_BIN}" -w 1 -b "${MUSIC_BIND}" --timeout "${MUSIC_TIMEOUT}" "${MUSIC_APP}" > "${MUSIC_LOG}" 2>&1 &

sleep 2
if ! pgrep -f "gunicorn.*${MUSIC_BIND}.*${MUSIC_APP}" >/dev/null 2>&1; then
  if ! pgrep -f "gunicorn.*${MUSIC_APP}.*${MUSIC_BIND}" >/dev/null 2>&1; then
    echo "ERROR: music generator gunicorn did not start correctly."
    exit 1
  fi
fi

• [ ] Step 4: Update smoke test

In scripts/smoke_test.sh, add:

MUSIC_PAGE_URL="https://wangmouren.online/tools/music-generator/"
MUSIC_API_URL="https://wangmouren.online/tools/music-generator/api/config"

curl -fsS "${MUSIC_PAGE_URL}" >/dev/null
MUSIC_CONFIG="$(curl -fsS "${MUSIC_API_URL}")"
echo "${MUSIC_CONFIG}" | node -e "
const fs = require('fs')
const body = fs.readFileSync(0, 'utf8')
const json = JSON.parse(body)
if (!json.success) process.exit(1)
if (/MINIMAX_API_KEY|Bearer|sk-/.test(body)) process.exit(1)
"

• [ ] Step 5: Make script executable and run shell syntax checks

Run:

chmod +x scripts/remote_configure_music_nginx.sh
bash -n scripts/remote_configure_music_nginx.sh
bash -n scripts/deploy_prod.sh
bash -n scripts/remote_restart.sh
bash -n scripts/smoke_test.sh

Expected: all commands exit 0.

• [ ] Step 6: Commit deployment changes

Run:

git add scripts/remote_configure_music_nginx.sh \
  scripts/deploy_prod.sh \
  scripts/remote_restart.sh \
  scripts/smoke_test.sh
git commit -m "chore: deploy music generator service"

Task 10: Final Verification and Token Safety Gate

Files:

• All files changed by Tasks 1-9.

• [ ] Step 1: Run backend tests

Run:

python -m pytest pywork/music-generator/tests -q

Expected: PASS.

• [ ] Step 2: Run Node wiring tests

Run:

node scripts/test_music_generator_wiring.js
node scripts/test_tools_entry_pages.js
node scripts/test_tool_interaction_wiring.js
node scripts/test_schulte_grid_wiring.js

Expected: PASS.

• [ ] Step 3: Run shell syntax checks

Run:

bash -n scripts/remote_configure_music_nginx.sh
bash -n scripts/deploy_prod.sh
bash -n scripts/remote_restart.sh
bash -n scripts/smoke_test.sh

Expected: all commands exit 0.

• [ ] Step 4: Run VuePress build

Run:

npm run build

Expected: build completes successfully.

• [ ] Step 5: Run token scan

Run:

git grep -nE 'sk-[A-Za-z0-9._-]{12,}|Bearer [A-Za-z0-9._-]{12,}' -- . ':!docs/superpowers/plans/2026-04-21-music-generator.md'

Expected: no output.

Run:

git grep -n 'MINIMAX_API_KEY' -- .

Expected: only source code, tests, README, spec, and this plan mention the environment variable name. No line should contain an actual token value.

• [ ] Step 6: Verify clean git state

Run:

git status --short

Expected: clean after all task commits.