124 lines
3.7 KiB
Python
124 lines
3.7 KiB
Python
from flask import Flask, render_template, jsonify
|
|
from flask_jwt_extended import JWTManager
|
|
from flask_cors import CORS
|
|
from config import Config
|
|
from models import db
|
|
from api.users import users_bp
|
|
from api.tasks import tasks_bp
|
|
from api.statistics import statistics_bp
|
|
from api.user_management import user_mgmt_bp
|
|
app = Flask(__name__)
|
|
app.config.from_object(Config)
|
|
|
|
jwt = JWTManager(app)
|
|
|
|
# 添加 JWT 错误处理
|
|
from flask import request
|
|
import traceback
|
|
@jwt.invalid_token_loader
|
|
def invalid_token_callback(callback):
|
|
print(f"=== JWT Token 验证失败 ===")
|
|
print(f"完整错误信息: {callback}")
|
|
print(f"请求头: {dict(request.headers)}")
|
|
print(f"Authorization 头: {request.headers.get('Authorization')}")
|
|
import traceback
|
|
traceback.print_exc() # 打印堆栈跟踪
|
|
print("=======================")
|
|
return jsonify({'error': '无效的认证令牌', 'detail': callback}), 422
|
|
@jwt.unauthorized_loader
|
|
def unauthorized_callback(callback):
|
|
print(f"=== 未授权访问 ===")
|
|
print(f"错误信息: {callback}")
|
|
print(f"请求头: {dict(request.headers)}")
|
|
print(f"认证信息: {request.authorization}")
|
|
print("==================")
|
|
return jsonify({'error': '缺少认证令牌'}), 401
|
|
@jwt.expired_token_loader
|
|
def expired_token_callback(jwt_header, jwt_payload):
|
|
print(f"=== Token 已过期 ===")
|
|
print(f"JWT Header: {jwt_header}")
|
|
print(f"JWT Payload: {jwt_payload}")
|
|
print("====================")
|
|
return jsonify({'error': '认证令牌已过期'}), 401
|
|
# 添加一个 token 加载失败的回调（有助于调试更多错误）
|
|
@jwt.token_in_blocklist_loader
|
|
def check_if_token_revoked(jwt_header, jwt_payload):
|
|
# 如果你实现了 token 黑名单，这里会有逻辑
|
|
return False # 默认返回 False
|
|
# 初始化扩展
|
|
db.init_app(app)
|
|
CORS(app)
|
|
# 注册蓝图
|
|
app.register_blueprint(users_bp, url_prefix='/api')
|
|
app.register_blueprint(tasks_bp, url_prefix='/api')
|
|
app.register_blueprint(statistics_bp, url_prefix='/api')
|
|
app.register_blueprint(user_mgmt_bp, url_prefix='/api')
|
|
# 前端路由
|
|
@app.route('/')
|
|
def index():
|
|
return render_template('login.html')
|
|
|
|
@app.route('/tasks')
|
|
def tasks():
|
|
return render_template('task_list.html')
|
|
|
|
@app.route('/tasks/create')
|
|
def create_task_page():
|
|
return render_template('task_create.html')
|
|
|
|
@app.route('/statistics')
|
|
def statistics():
|
|
return render_template('statistics.html')
|
|
|
|
@app.route('/register')
|
|
def register_page():
|
|
return render_template('register.html')
|
|
|
|
@app.route('/users')
|
|
def users_page():
|
|
return render_template('users.html')
|
|
|
|
@app.route('/profile')
|
|
def profile_page():
|
|
return render_template('profile.html')
|
|
|
|
# 初始化数据库
|
|
@app.cli.command()
|
|
def init_db():
|
|
"""初始化数据库"""
|
|
db.create_all()
|
|
print('数据库初始化完成')
|
|
|
|
@app.cli.command()
|
|
def create_admin():
|
|
"""创建管理员账户"""
|
|
from models import User, Group, UserGroup
|
|
|
|
# 创建管理员
|
|
admin = User.query.filter_by(username='admin').first()
|
|
if not admin:
|
|
admin = User(username='admin', email='admin@example.com', role='admin')
|
|
admin.set_password('admin123')
|
|
db.session.add(admin)
|
|
print('管理员账户创建成功: admin / admin123')
|
|
else:
|
|
print('管理员账户已存在')
|
|
|
|
# 创建发布组
|
|
group = Group.query.filter_by(group_key='release').first()
|
|
if not group:
|
|
group = Group(
|
|
group_name='发布组',
|
|
group_key='release',
|
|
description='负责PT资源的发布'
|
|
)
|
|
db.session.add(group)
|
|
print('发布组创建成功')
|
|
else:
|
|
print('发布组已存在')
|
|
|
|
db.session.commit()
|
|
|
|
if __name__ == '__main__':
|
|
app.run(debug=True, host='0.0.0.0', port=5000)
|