vimax-studio/app.py at main · EnzoChen0618/vimax-studio · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
from flask import Flask, jsonify, request, send_file, session, redirect, render_template_string
import asyncio, glob, json, os, subprocess, sys, threading, time, uuid

app = Flask(__name__)
app.secret_key = os.environ.get('SECRET_KEY', 'vimax-secret-key')

APP_PASSWORD = os.environ.get('APP_PASSWORD', 'vimax2024')
OPENROUTER_API_KEY = os.environ.get('OPENROUTER_API_KEY', '')
GOOGLE_API_KEY = os.environ.get('GOOGLE_API_KEY', '')
WORKING_DIR = '/tmp/vimax_working'
os.makedirs(WORKING_DIR, exist_ok=True)

jobs = {}

LOGIN_HTML = '''<!DOCTYPE html>
<html><head>
<meta charset="UTF-8"><meta name="viewport" content="width=device-width,initial-scale=1">
<title>ViMax · 登录</title>
<link href="https://fonts.googleapis.com/css2?family=Syne:wght@700;800&family=DM+Mono:wght@400;500&display=swap" rel="stylesheet">
<style>
*{margin:0;padding:0;box-sizing:border-box}
body{background:#0a0a0f;color:#e8e8f0;font-family:"DM Mono",monospace;min-height:100vh;display:flex;align-items:center;justify-content:center}
.box{background:#12121a;border:1px solid #2a2a40;border-radius:20px;padding:48px 40px;width:360px;text-align:center}
.logo{width:56px;height:56px;background:linear-gradient(135deg,#7c6aff,#ff6a9e);border-radius:14px;display:flex;align-items:center;justify-content:center;font-family:"Syne",sans-serif;font-weight:800;font-size:22px;color:white;margin:0 auto 20px;box-shadow:0 0 28px rgba(124,106,255,.4)}
h1{font-family:"Syne",sans-serif;font-size:24px;font-weight:800;margin-bottom:6px;background:linear-gradient(90deg,#e8e8f0,#7c6aff);-webkit-background-clip:text;-webkit-text-fill-color:transparent;background-clip:text}
p{color:#6a6a88;font-size:12px;margin-bottom:32px}
input{width:100%;background:#0a0a0f;border:1px solid #2a2a40;border-radius:10px;color:#e8e8f0;font-family:"DM Mono",monospace;font-size:14px;padding:14px 16px;outline:none;transition:.2s;margin-bottom:14px}
input:focus{border-color:#7c6aff;box-shadow:0 0 0 3px rgba(124,106,255,.12)}
button{width:100%;padding:14px;background:linear-gradient(135deg,#7c6aff,#9b6aff);border:none;border-radius:10px;color:white;font-family:"Syne",sans-serif;font-size:15px;font-weight:700;cursor:pointer;transition:.2s}
button:hover{transform:translateY(-1px);box-shadow:0 8px 24px rgba(124,106,255,.4)}
.err{color:#ff5c5c;font-size:12px;margin-top:12px}
</style></head>
<body><div class="box">
<div class="logo">Vi</div>
<h1>ViMax Studio</h1>
<p>AI 视频生成平台 · 请输入访问密码</p>
<form method="post">
<input type="password" name="password" placeholder="输入密码" autofocus>
<button type="submit">进入</button>
{% if error %}<div class="err">密码错误，请重试</div>{% endif %}
</form>
</div></body></html>'''

def require_auth(f):
    from functools import wraps
    @wraps(f)
    def decorated(*args, **kwargs):
        if not session.get('authenticated'):
            return redirect('/login')
        return f(*args, **kwargs)
    return decorated

@app.route('/login', methods=['GET', 'POST'])
def login():
    error = False
    if request.method == 'POST':
        if request.form.get('password') == APP_PASSWORD:
            session['authenticated'] = True
            return redirect('/')
        error = True
    return render_template_string(LOGIN_HTML, error=error)

@app.route('/logout')
def logout():
    session.clear()
    return redirect('/login')

@app.route('/')
@require_auth
def index():
    return send_file('vimax_ui.html')

@app.route('/api/generate', methods=['POST'])
@require_auth
def generate():
    data = request.json or {}
    idea = data.get('idea', '').strip()
    requirement = data.get('requirement', '').strip()
    style = data.get('style', 'Realistic').strip()
    if not idea:
        return jsonify({'error': 'Idea is required'}), 400
    job_id = str(uuid.uuid4())[:8]
    job_working_dir = os.path.join(WORKING_DIR, job_id)
    os.makedirs(job_working_dir, exist_ok=True)
    jobs[job_id] = {'id': job_id, 'status': 'running', 'logs': [], 'images': [], 'videos': [], 'working_dir': job_working_dir}
    config = f"""chat_model:
  init_args:
    model: google/gemini-2.5-flash-lite-preview-09-2025
    model_provider: openai
    api_key: {OPENROUTER_API_KEY}
    base_url: https://openrouter.ai/api/v1
  max_requests_per_minute: 500
  max_requests_per_day: 2000
image_generator:
  class_path: tools.ImageGeneratorNanobananaGoogleAPI
  init_args:
    api_key: {GOOGLE_API_KEY}
  max_requests_per_minute: 10
  max_requests_per_day: 500
video_generator:
  class_path: tools.VideoGeneratorVeoGoogleAPI
  init_args:
    api_key: {GOOGLE_API_KEY}
  max_requests_per_minute: 2
  max_requests_per_day: 10
working_dir: {job_working_dir}
"""
    config_path = f'/tmp/vimax_config_{job_id}.yaml'
    with open(config_path, 'w') as f:
        f.write(config)
    script = f'''import asyncio, sys
sys.path.insert(0, '/app')
from pipelines.idea2video_pipeline import Idea2VideoPipeline
async def main():
    pipeline = Idea2VideoPipeline.init_from_config("{config_path}")
    await pipeline(idea="""{idea}""", user_requirement="""{requirement}""", style="""{style}""")
asyncio.run(main())
'''
    script_path = f'/tmp/vimax_run_{job_id}.py'
    with open(script_path, 'w') as f:
        f.write(script)
    threading.Thread(target=_run_job, args=(job_id, script_path, config_path), daemon=True).start()
    return jsonify({'job_id': job_id, 'status': 'started'})

@app.route('/api/status')
@require_auth
def status():
    job_id = request.args.get('job_id')
    if job_id not in jobs:
        return jsonify({'error': 'Not found'}), 404
    return jsonify(jobs[job_id])

@app.route('/api/file')
@require_auth
def serve_file():
    path = request.args.get('path', '')
    if not os.path.abspath(path).startswith(os.path.abspath(WORKING_DIR)):
        return jsonify({'error': 'Forbidden'}), 403
    if not os.path.exists(path):
        return jsonify({'error': 'Not found'}), 404
    return send_file(path)

def _run_job(job_id, script_path, config_path):
    try:
        process = subprocess.Popen([sys.executable, script_path], stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, bufsize=1, cwd='/app')
        for line in process.stdout:
            line = line.rstrip()
            if line:
                jobs[job_id]['logs'].append(line)
                _scan_media(job_id)
        process.wait()
        _scan_media(job_id)
        jobs[job_id]['status'] = 'done' if process.returncode == 0 else 'error'
    except Exception as e:
        jobs[job_id]['status'] = 'error'
        jobs[job_id]['logs'].append(f'ERROR: {e}')
    finally:
        for p in [script_path, config_path]:
            try: os.remove(p)
            except: pass

def _scan_media(job_id):
    wd = jobs[job_id]['working_dir']
    jobs[job_id]['images'] = sorted(glob.glob(f'{wd}/**/*.png', recursive=True))
    jobs[job_id]['videos'] = sorted(glob.glob(f'{wd}/**/*.mp4', recursive=True))

if __name__ == '__main__':
    port = int(os.environ.get('PORT', 8080))
    app.run(host='0.0.0.0', port=port)