hz4th_coder
3a6d4fc683
- 多轨道时间线: 视频/图片/音频/文字, 拖拽/修剪/分割/复制/删除 - 12款滤镜特效 + 4种转场 + 4种文字动画 - 6款程序化原创配乐(Web Audio 实时合成, 零版权风险) - Canvas 合成 + MediaRecorder 实时导出 WebM(VP9+Opus) - 账号体系 + 每日导出额度 + 企业/AI短剧付费升级 + 管理后台 - 项目自动保存/缩略图, 移动端自适应 - 个人免费每天30次导出, 企业¥299/月 不限量, AI短剧¥499/月 不限量
479 lines
18 KiB
Python
479 lines
18 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""
|
|
FreeCut 自由剪辑 - 免费开源视频编辑软件 (Web)
|
|
个人自媒体免费 / 企业 & AI短剧付费
|
|
"""
|
|
import os
|
|
import sqlite3
|
|
import time
|
|
import json
|
|
import functools
|
|
import threading
|
|
from datetime import datetime, date
|
|
|
|
from flask import (Flask, request, session, redirect, jsonify, send_from_directory,
|
|
render_template_string, g, abort)
|
|
|
|
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
|
|
DATA_DIR = os.path.join(BASE_DIR, 'data')
|
|
DB_PATH = os.path.join(DATA_DIR, 'freecut.db')
|
|
STATIC_DIR = os.path.join(BASE_DIR, 'static')
|
|
|
|
os.makedirs(DATA_DIR, exist_ok=True)
|
|
|
|
app = Flask(__name__, static_folder=STATIC_DIR, static_url_path='')
|
|
app.secret_key = 'freecut-secret-2026-change-me'
|
|
app.config['MAX_CONTENT_LENGTH'] = 800 * 1024 * 1024 # 800MB 上传
|
|
|
|
_db_lock = threading.RLock()
|
|
|
|
# ---------------- 配额配置 ----------------
|
|
# 免费个人用户：每天 N 次导出（足够自媒体日常使用）
|
|
# 企业 / AI短剧：付费后不限量
|
|
QUOTA = {
|
|
'free': {'exports_per_day': 30, 'label': '个人免费版', 'desc': '无水印、全功能，每天 30 次导出，够自媒体日常使用'},
|
|
'pro': {'exports_per_day': 100, 'label': '个人专业版', 'desc': '更高额度'},
|
|
'business': {'exports_per_day': 0, 'label': '企业版', 'desc': '付费 · 不限量导出', 'paid': True},
|
|
'ai_short': {'exports_per_day': 0, 'label': 'AI短剧版', 'desc': '付费 · 不限量导出', 'paid': True},
|
|
}
|
|
TIER_NAMES = {'free': '个人免费版', 'pro': '个人专业版', 'business': '企业版', 'ai_short': 'AI短剧版'}
|
|
|
|
|
|
def get_db():
|
|
if 'db' not in g:
|
|
g.db = sqlite3.connect(DB_PATH, timeout=30)
|
|
g.db.row_factory = sqlite3.Row
|
|
return g.db
|
|
|
|
|
|
@app.teardown_appcontext
|
|
def close_db(exc):
|
|
db = g.pop('db', None)
|
|
if db is not None:
|
|
db.close()
|
|
|
|
|
|
def init_db():
|
|
with _db_lock:
|
|
db = sqlite3.connect(DB_PATH, timeout=30)
|
|
db.executescript('''
|
|
CREATE TABLE IF NOT EXISTS users (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
username TEXT UNIQUE NOT NULL,
|
|
password TEXT NOT NULL, -- sha256
|
|
nickname TEXT DEFAULT '',
|
|
tier TEXT DEFAULT 'free', -- free/pro/business/ai_short
|
|
paid_expire INTEGER DEFAULT 0, -- 0=未付费, 否则到期时间戳
|
|
created_at INTEGER DEFAULT 0
|
|
);
|
|
CREATE TABLE IF NOT EXISTS projects (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
user_id INTEGER NOT NULL,
|
|
name TEXT DEFAULT '未命名项目',
|
|
data TEXT DEFAULT '{}', -- 工程JSON
|
|
thumb TEXT DEFAULT '',
|
|
updated_at INTEGER DEFAULT 0,
|
|
created_at INTEGER DEFAULT 0
|
|
);
|
|
CREATE TABLE IF NOT EXISTS export_logs (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
user_id INTEGER NOT NULL,
|
|
project_id INTEGER DEFAULT 0,
|
|
date TEXT NOT NULL,
|
|
duration INTEGER DEFAULT 0,
|
|
size INTEGER DEFAULT 0,
|
|
created_at INTEGER DEFAULT 0
|
|
);
|
|
CREATE TABLE IF NOT EXISTS upgrade_requests (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
user_id INTEGER NOT NULL,
|
|
target_tier TEXT NOT NULL,
|
|
contact TEXT DEFAULT '',
|
|
status TEXT DEFAULT 'pending', -- pending/approved/rejected
|
|
created_at INTEGER DEFAULT 0
|
|
);
|
|
CREATE TABLE IF NOT EXISTS assets (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
user_id INTEGER DEFAULT 0,
|
|
name TEXT DEFAULT '',
|
|
kind TEXT DEFAULT 'video', -- video/image/audio
|
|
path TEXT DEFAULT '',
|
|
size INTEGER DEFAULT 0,
|
|
created_at INTEGER DEFAULT 0
|
|
);
|
|
''')
|
|
# 默认管理员
|
|
import hashlib
|
|
cur = db.execute("SELECT id FROM users WHERE username='admin'")
|
|
if not cur.fetchone():
|
|
db.execute("INSERT INTO users(username,password,nickname,tier,created_at) VALUES(?,?,?,?,?)",
|
|
('admin', hashlib.sha256('admin123'.encode()).hexdigest(), '管理员', 'business', int(time.time())))
|
|
db.commit()
|
|
db.close()
|
|
|
|
|
|
init_db()
|
|
|
|
|
|
def hash_pw(pw):
|
|
import hashlib
|
|
return hashlib.sha256(pw.encode()).hexdigest()
|
|
|
|
|
|
def now():
|
|
return int(time.time())
|
|
|
|
|
|
def login_required(fn):
|
|
@functools.wraps(fn)
|
|
def wrapper(*a, **kw):
|
|
if not session.get('uid'):
|
|
return jsonify({'ok': False, 'error': '未登录', 'code': 401})
|
|
return fn(*a, **kw)
|
|
return wrapper
|
|
|
|
|
|
def admin_required(fn):
|
|
@functools.wraps(fn)
|
|
def wrapper(*a, **kw):
|
|
if session.get('uid') != 1:
|
|
return jsonify({'ok': False, 'error': '需要管理员权限'})
|
|
return fn(*a, **kw)
|
|
return wrapper
|
|
|
|
|
|
def get_user(uid=None):
|
|
db = get_db()
|
|
row = db.execute("SELECT * FROM users WHERE id=?", (uid or session.get('uid'),)).fetchone()
|
|
return dict(row) if row else None
|
|
|
|
|
|
def today_usage(uid, d=None):
|
|
d = d or date.today().isoformat()
|
|
db = get_db()
|
|
row = db.execute("SELECT COUNT(*), COALESCE(SUM(size),0) FROM export_logs WHERE user_id=? AND date=?",
|
|
(uid, d)).fetchone()
|
|
return row[0], row[1] or 0
|
|
|
|
|
|
def quota_status(uid):
|
|
u = get_user(uid)
|
|
if not u:
|
|
return {'ok': False}
|
|
tier = u['tier']
|
|
conf = QUOTA.get(tier, QUOTA['free'])
|
|
used, size = today_usage(uid)
|
|
paid = u.get('paid_expire', 0) > now()
|
|
if conf['exports_per_day'] == 0:
|
|
# 付费版本：已付费则不限量，否则回退到免费额度（保留等级名）
|
|
limit = 0 if paid else QUOTA['free']['exports_per_day']
|
|
else:
|
|
limit = conf['exports_per_day']
|
|
return {
|
|
'ok': True,
|
|
'username': u['username'],
|
|
'nickname': u['nickname'] or u['username'],
|
|
'tier': tier,
|
|
'tier_name': conf['label'],
|
|
'tier_desc': conf['desc'],
|
|
'paid': paid,
|
|
'exports_per_day': limit,
|
|
'exports_used': used,
|
|
'exports_left': 0 if limit == 0 else max(0, limit - used),
|
|
'today_bytes': size,
|
|
'date': date.today().isoformat(),
|
|
}
|
|
|
|
|
|
# ---------------- 页面 ----------------
|
|
@app.route('/')
|
|
def index():
|
|
return send_from_directory(STATIC_DIR, 'index.html')
|
|
|
|
|
|
@app.route('/editor')
|
|
@login_required
|
|
def editor_page():
|
|
return send_from_directory(STATIC_DIR, 'editor.html')
|
|
|
|
|
|
@app.route('/admin')
|
|
@admin_required
|
|
def admin_page():
|
|
return send_from_directory(STATIC_DIR, 'admin.html')
|
|
|
|
|
|
# ---------------- 认证 ----------------
|
|
@app.route('/api/auth/register', methods=['POST'])
|
|
def register():
|
|
d = request.get_json(force=True, silent=True) or {}
|
|
username = (d.get('username') or '').strip()
|
|
password = d.get('password') or ''
|
|
nickname = (d.get('nickname') or '').strip() or username
|
|
if len(username) < 2 or len(password) < 4:
|
|
return jsonify({'ok': False, 'error': '用户名至少2位，密码至少4位'})
|
|
db = get_db()
|
|
if db.execute("SELECT id FROM users WHERE username=?", (username,)).fetchone():
|
|
return jsonify({'ok': False, 'error': '用户名已被占用'})
|
|
db.execute("INSERT INTO users(username,password,nickname,tier,created_at) VALUES(?,?,?,?,?)",
|
|
(username, hash_pw(password), nickname, 'free', now()))
|
|
db.commit()
|
|
return jsonify({'ok': True})
|
|
|
|
|
|
@app.route('/api/auth/login', methods=['POST'])
|
|
def login():
|
|
d = request.get_json(force=True, silent=True) or {}
|
|
username = (d.get('username') or '').strip()
|
|
password = d.get('password') or ''
|
|
db = get_db()
|
|
row = db.execute("SELECT * FROM users WHERE username=?", (username,)).fetchone()
|
|
if not row or row['password'] != hash_pw(password):
|
|
return jsonify({'ok': False, 'error': '用户名或密码错误'})
|
|
session['uid'] = row['id']
|
|
session['username'] = row['username']
|
|
return jsonify({'ok': True, 'user': {'username': row['username'], 'nickname': row['nickname'] or row['username'], 'tier': row['tier']}})
|
|
|
|
|
|
@app.route('/api/auth/logout', methods=['POST'])
|
|
def logout():
|
|
session.clear()
|
|
return jsonify({'ok': True})
|
|
|
|
|
|
@app.route('/api/me')
|
|
@login_required
|
|
def me():
|
|
u = get_user()
|
|
if not u:
|
|
session.clear()
|
|
return jsonify({'ok': False, 'error': '账号不存在，请重新登录', 'code': 401})
|
|
return jsonify({'ok': True, 'quota': quota_status(session['uid'])})
|
|
|
|
|
|
# ---------------- 项目 ----------------
|
|
@app.route('/api/projects', methods=['GET', 'POST'])
|
|
@login_required
|
|
def projects():
|
|
uid = session['uid']
|
|
db = get_db()
|
|
if request.method == 'POST':
|
|
d = request.get_json(force=True, silent=True) or {}
|
|
name = (d.get('name') or '').strip() or '未命名项目'
|
|
cur = db.execute("INSERT INTO projects(user_id,name,data,updated_at,created_at) VALUES(?,?,?,?,?)",
|
|
(uid, name, '{}', now(), now()))
|
|
db.commit()
|
|
return jsonify({'ok': True, 'id': cur.lastrowid})
|
|
rows = db.execute("SELECT id,name,thumb,updated_at,created_at FROM projects WHERE user_id=? ORDER BY updated_at DESC", (uid,)).fetchall()
|
|
return jsonify({'ok': True, 'projects': [dict(r) for r in rows]})
|
|
|
|
|
|
@app.route('/api/projects/<int:pid>', methods=['GET', 'PUT', 'DELETE'])
|
|
@login_required
|
|
def project_detail(pid):
|
|
uid = session['uid']
|
|
db = get_db()
|
|
row = db.execute("SELECT * FROM projects WHERE id=? AND user_id=?", (pid, uid)).fetchone()
|
|
if not row:
|
|
return jsonify({'ok': False, 'error': '项目不存在'})
|
|
if request.method == 'GET':
|
|
return jsonify({'ok': True, 'project': {'id': row['id'], 'name': row['name'], 'data': json.loads(row['data'] or '{}'), 'thumb': row['thumb']}})
|
|
if request.method == 'DELETE':
|
|
db.execute("DELETE FROM projects WHERE id=?", (pid,))
|
|
db.commit()
|
|
return jsonify({'ok': True})
|
|
d = request.get_json(force=True, silent=True) or {}
|
|
data = json.dumps(d.get('data', {}), ensure_ascii=False)
|
|
name = (d.get('name') or '').strip() or row['name']
|
|
thumb = d.get('thumb') or row['thumb']
|
|
db.execute("UPDATE projects SET name=?, data=?, thumb=?, updated_at=? WHERE id=?", (name, data, thumb, now(), pid))
|
|
db.commit()
|
|
return jsonify({'ok': True})
|
|
|
|
|
|
# ---------------- 配额 ----------------
|
|
@app.route('/api/quota')
|
|
@login_required
|
|
def quota_api():
|
|
return jsonify({'ok': True, 'quota': quota_status(session['uid'])})
|
|
|
|
|
|
@app.route('/api/quota/check-export', methods=['POST'])
|
|
@login_required
|
|
def check_export():
|
|
uid = session['uid']
|
|
q = quota_status(uid)
|
|
if q['exports_per_day'] != 0 and q['exports_left'] <= 0:
|
|
return jsonify({'ok': False, 'error': f"今日导出额度已用完（{q['exports_per_day']}次/天）。个人免费版每天额度充足，明天自动恢复；企业/AI短剧请付费解锁不限量。", 'quota': q})
|
|
return jsonify({'ok': True, 'quota': q})
|
|
|
|
|
|
@app.route('/api/quota/record-export', methods=['POST'])
|
|
@login_required
|
|
def record_export():
|
|
d = request.get_json(force=True, silent=True) or {}
|
|
pid = int(d.get('project_id') or 0)
|
|
duration = int(d.get('duration') or 0)
|
|
size = int(d.get('size') or 0)
|
|
db = get_db()
|
|
db.execute("INSERT INTO export_logs(user_id,project_id,date,duration,size,created_at) VALUES(?,?,?,?,?,?)",
|
|
(session['uid'], pid, date.today().isoformat(), duration, size, now()))
|
|
db.commit()
|
|
return jsonify({'ok': True, 'quota': quota_status(session['uid'])})
|
|
|
|
|
|
# ---------------- 升级/付费 ----------------
|
|
@app.route('/api/upgrade/plans')
|
|
def plans():
|
|
return jsonify({'ok': True, 'plans': {
|
|
'business': {'name': '企业版', 'price': '¥299/月', 'features': ['不限量导出', '团队协作(规划中)', '优先渲染(规划中)', '商用授权']},
|
|
'ai_short': {'name': 'AI短剧版', 'price': '¥499/月', 'features': ['不限量导出', '批量生产(规划中)', '商用授权', '独享高速通道(规划中)']},
|
|
}})
|
|
|
|
|
|
@app.route('/api/upgrade/apply', methods=['POST'])
|
|
@login_required
|
|
def apply_upgrade():
|
|
d = request.get_json(force=True, silent=True) or {}
|
|
target = d.get('tier')
|
|
if target not in ('business', 'ai_short'):
|
|
return jsonify({'ok': False, 'error': '只能申请企业版/AI短剧版'})
|
|
contact = (d.get('contact') or '').strip()
|
|
db = get_db()
|
|
# 避免重复 pending
|
|
dup = db.execute("SELECT id FROM upgrade_requests WHERE user_id=? AND target_tier=? AND status='pending'",
|
|
(session['uid'], target)).fetchone()
|
|
if dup:
|
|
return jsonify({'ok': False, 'error': '已有待审核申请，请耐心等待管理员开通'})
|
|
db.execute("INSERT INTO upgrade_requests(user_id,target_tier,contact,created_at) VALUES(?,?,?,?)",
|
|
(session['uid'], target, contact, now()))
|
|
db.commit()
|
|
return jsonify({'ok': True})
|
|
|
|
|
|
# ---------------- 上传资源 ----------------
|
|
@app.route('/api/assets', methods=['GET'])
|
|
@login_required
|
|
def list_assets():
|
|
db = get_db()
|
|
rows = db.execute("SELECT * FROM assets WHERE user_id=? ORDER BY id DESC", (session['uid'],)).fetchall()
|
|
out = []
|
|
for r in rows:
|
|
d = dict(r)
|
|
d['url'] = r['path']
|
|
out.append(d)
|
|
return jsonify({'ok': True, 'assets': out})
|
|
|
|
|
|
@app.route('/api/assets', methods=['POST'])
|
|
@login_required
|
|
def upload_asset():
|
|
f = request.files.get('file')
|
|
if not f or not f.filename:
|
|
return jsonify({'ok': False, 'error': '未选择文件'})
|
|
kind = request.form.get('kind', 'video')
|
|
ext = os.path.splitext(f.filename)[1].lower() or '.bin'
|
|
if kind == 'video' and ext not in ('.mp4', '.webm', '.mov', '.m4v', '.mkv'):
|
|
return jsonify({'ok': False, 'error': '不支持的视频格式'})
|
|
if kind == 'image' and ext not in ('.jpg', '.jpeg', '.png', '.gif', '.webp'):
|
|
return jsonify({'ok': False, 'error': '不支持的图片格式'})
|
|
if kind == 'audio' and ext not in ('.mp3', '.wav', '.m4a', '.ogg', '.aac'):
|
|
return jsonify({'ok': False, 'error': '不支持的音频格式'})
|
|
fname = f'{int(time.time()*1000)}_{os.urandom(3).hex()}{ext}'
|
|
fpath = os.path.join(DATA_DIR, 'uploads')
|
|
os.makedirs(fpath, exist_ok=True)
|
|
f.save(os.path.join(fpath, fname))
|
|
db = get_db()
|
|
cur = db.execute("INSERT INTO assets(user_id,name,kind,path,size,created_at) VALUES(?,?,?,?,?,?)",
|
|
(session['uid'], f.filename, kind, f'/uploads/{fname}', os.path.getsize(os.path.join(fpath, fname)), now()))
|
|
db.commit()
|
|
return jsonify({'ok': True, 'asset': {'id': cur.lastrowid, 'name': f.filename, 'kind': kind, 'url': f'/uploads/{fname}', 'size': os.path.getsize(os.path.join(fpath, fname))}})
|
|
|
|
|
|
@app.route('/uploads/<path:fn>')
|
|
def uploads(fn):
|
|
return send_from_directory(os.path.join(DATA_DIR, 'uploads'), fn)
|
|
|
|
|
|
# ---------------- 管理后台 ----------------
|
|
@app.route('/api/admin/users')
|
|
@admin_required
|
|
def admin_users():
|
|
db = get_db()
|
|
rows = db.execute("SELECT id,username,nickname,tier,paid_expire,created_at FROM users ORDER BY id").fetchall()
|
|
out = []
|
|
for r in rows:
|
|
used, _ = today_usage(r['id'])
|
|
out.append({**dict(r), 'today_exports': used})
|
|
return jsonify({'ok': True, 'users': out})
|
|
|
|
|
|
@app.route('/api/admin/user', methods=['POST'])
|
|
@admin_required
|
|
def admin_set_user():
|
|
d = request.get_json(force=True, silent=True) or {}
|
|
uid = int(d.get('id') or 0)
|
|
tier = d.get('tier')
|
|
if tier not in TIER_NAMES:
|
|
return jsonify({'ok': False, 'error': '非法等级'})
|
|
paid_expire = int(d.get('paid_expire') or 0)
|
|
db = get_db()
|
|
db.execute("UPDATE users SET tier=?, paid_expire=? WHERE id=?", (tier, paid_expire, uid))
|
|
db.commit()
|
|
return jsonify({'ok': True})
|
|
|
|
|
|
@app.route('/api/admin/requests')
|
|
@admin_required
|
|
def admin_requests():
|
|
db = get_db()
|
|
rows = db.execute("SELECT r.*, u.username FROM upgrade_requests r LEFT JOIN users u ON u.id=r.user_id ORDER BY r.id DESC").fetchall()
|
|
return jsonify({'ok': True, 'requests': [dict(r) for r in rows]})
|
|
|
|
|
|
@app.route('/api/admin/request', methods=['POST'])
|
|
@admin_required
|
|
def admin_handle_request():
|
|
d = request.get_json(force=True, silent=True) or {}
|
|
rid = int(d.get('id') or 0)
|
|
action = d.get('action')
|
|
db = get_db()
|
|
r = db.execute("SELECT * FROM upgrade_requests WHERE id=?", (rid,)).fetchone()
|
|
if not r:
|
|
return jsonify({'ok': False, 'error': '申请不存在'})
|
|
if action == 'approve':
|
|
# 付费 1 年
|
|
expire = now() + 365 * 24 * 3600
|
|
db.execute("UPDATE users SET tier=?, paid_expire=? WHERE id=?", (r['target_tier'], expire, r['user_id']))
|
|
db.execute("UPDATE upgrade_requests SET status='approved' WHERE id=?", (rid,))
|
|
else:
|
|
db.execute("UPDATE upgrade_requests SET status='rejected' WHERE id=?", (rid,))
|
|
db.commit()
|
|
return jsonify({'ok': True})
|
|
|
|
|
|
@app.route('/api/admin/stats')
|
|
@admin_required
|
|
def admin_stats():
|
|
db = get_db()
|
|
u = db.execute("SELECT COUNT(*) c FROM users").fetchone()['c']
|
|
p = db.execute("SELECT COUNT(*) c FROM projects").fetchone()['c']
|
|
e = db.execute("SELECT COUNT(*) c FROM export_logs").fetchone()['c']
|
|
# 近7天导出
|
|
days = []
|
|
for i in range(6, -1, -1):
|
|
d = (date.today().fromordinal(date.today().toordinal() - i)).isoformat()
|
|
c = db.execute("SELECT COUNT(*) c FROM export_logs WHERE date=?", (d,)).fetchone()['c']
|
|
days.append({'date': d, 'count': c})
|
|
return jsonify({'ok': True, 'users': u, 'projects': p, 'exports': e, 'days': days})
|
|
|
|
|
|
@app.route('/api/ping')
|
|
def ping():
|
|
return jsonify({'ok': True, 'name': 'FreeCut 自由剪辑', 'version': 'v1.0.0', 'time': now()})
|
|
|
|
|
|
if __name__ == '__main__':
|
|
port = int(os.environ.get('PORT', '16106'))
|
|
app.run(host='0.0.0.0', port=port, debug=False, threaded=True)
|