pre / routes /web.py
yangtb24's picture
Upload 70 files
2809c18 verified
raw
history blame
4.85 kB
"""
Web路由模块 - 处理所有页面请求和身份验证
"""
import re # 导入正则表达式模块
from flask import Blueprint, render_template, request, redirect, url_for, make_response, session
from datetime import datetime
from config import ADMIN_PASSWORD, PLATFORMS, PLATFORM_STYLES, TOKEN_EXPIRY_DAYS
from utils.auth import AuthManager
from models.api_key import ApiKeyManager
# 创建Web蓝图
web_bp = Blueprint('web', __name__)
# 辅助函数:为API密钥生成排序键
def get_sort_key(key):
"""
生成用于排序的元组 (balance, platform_priority)。
- balance: 主要排序键,降序。
- platform_priority: 次要排序键,仅在 balance 相同时生效,降序。
- Google: paid=1, free=0
- OpenAI/Claude: t5=5, t4=4, ..., t1=1, 其他=0
"""
balance = key.get('balance', 0)
platform_priority = 0
platform = key.get('platform', '').lower()
name = key.get('name', '').lower() # 用于 OpenAI/Claude tier 判断
states = key.get('states', '').lower() # 获取 states 字段并转小写,用于 Google 判断
if platform == 'google':
if 'paid' in states: # 检查 states 字段是否包含 'paid' (已转小写)
platform_priority = 1 # paid 优先
elif platform in ['openai', 'anthropic']: # 假设 Claude 的 platform id 是 'anthropic'
match = re.search(r't(\d)', name)
if match:
try:
tier = int(match.group(1))
if 1 <= tier <= 5: # 假设 tier 在 1-5 之间
platform_priority = tier
except ValueError:
pass # 如果转换失败,保持 priority 为 0
# 返回元组,Python 的 sorted 会按元组顺序比较
# 因为 sorted 默认升序,我们需要对 balance 和 priority 都取反来实现降序效果
# 或者直接在 sorted 中用 reverse=True,然后只对 platform_priority 取反(或用正数表示优先级)
# 这里我们用 reverse=True,所以 balance 自然降序,platform_priority 越大越优先
return (balance, platform_priority)
@web_bp.route('/')
def index():
"""首页 - 显示API密钥管理界面"""
api_keys_data = ApiKeyManager.get_all_keys()
all_keys_list = api_keys_data.get("api_keys", [])
# 按 balance 降序排序密钥
# 注意:确保 balance 存在且为数字类型,否则排序可能出错。使用 .get('balance', 0) 提供默认值。
# 使用新的排序逻辑
sorted_keys = sorted(all_keys_list, key=get_sort_key, reverse=True)
# 按平台分组密钥
grouped_keys = {}
for platform in PLATFORMS:
grouped_keys[platform["id"]] = {
"name": platform["name"],
"keys": []
}
# 将密钥加入对应的平台组
for key in sorted_keys: # 使用排序后的列表进行分组
platform_id = key.get("platform", "other").lower()
if platform_id not in grouped_keys:
platform_id = "other"
grouped_keys[platform_id]["keys"].append(key)
is_ajax = request.args.get('ajax', '0') == '1'
return render_template('index.html', platforms=PLATFORMS, grouped_keys=grouped_keys, platform_styles=PLATFORM_STYLES)
@web_bp.route('/update-test')
def update_test():
"""API密钥更新测试页面"""
return render_template('update_test.html')
@web_bp.route('/login', methods=['GET', 'POST'])
def login():
"""用户登录"""
error = None
if request.method == 'POST':
password = request.form.get('password')
if password == ADMIN_PASSWORD:
# 密码正确,生成令牌
token = AuthManager.generate_token()
AuthManager.store_token(token)
# 创建响应并设置Cookie
response = make_response(redirect(url_for('web.index')))
response.set_cookie(
'auth_token',
token,
max_age=TOKEN_EXPIRY_DAYS * 24 * 60 * 60,
httponly=True,
secure=request.is_secure
)
return response
else:
error = "密码错误,请重试"
# 传递当前年份以供页脚使用
current_year = datetime.now().year
return render_template('login.html', error=error, now={'year': current_year})
@web_bp.route('/logout')
def logout():
"""用户登出"""
# 从Cookie中获取令牌
token = request.cookies.get('auth_token')
# 删除令牌
if token:
AuthManager.remove_token(token)
# 返回响应并清除Cookie
response = make_response(redirect(url_for('web.login')))
response.delete_cookie('auth_token')
return response