File size: 5,142 Bytes
3d2357d 7769832 3d2357d 7769832 3d2357d 7769832 3d2357d 7769832 3d2357d 7769832 3d2357d 7769832 4f5d5ba 7769832 4f5d5ba 3d2357d |
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 |
import os
import requests
import urllib.parse
from flask import Flask, render_template, request, redirect, url_for, flash, jsonify, Response
from flask_login import LoginManager, UserMixin, login_user, logout_user, login_required, current_user
from werkzeug.security import check_password_hash, generate_password_hash
from dotenv import load_dotenv
import time
# 加载环境变量
load_dotenv()
app = Flask(__name__)
app.secret_key = os.environ.get('SECRET_KEY', 'starsky_secret_key')
# 登录管理配置
login_manager = LoginManager()
login_manager.init_app(app)
login_manager.login_view = 'login'
# 用户类
class User(UserMixin):
def __init__(self, id, username, password_hash):
self.id = id
self.username = username
self.password_hash = password_hash
# 默认用户
default_username = os.environ.get('ADMIN_USERNAME', 'admin')
default_password = os.environ.get('ADMIN_PASSWORD', 'admin123')
# 用户存储
users = {
default_username: User(
default_username,
default_username,
generate_password_hash(default_password)
)
}
# 检查 subconverter 是否启动
def check_subconverter():
max_retries = 10
for i in range(max_retries):
try:
response = requests.get('http://localhost:25500/version', timeout=2)
if response.status_code == 200:
print(f"Subconverter 服务已启动: {response.text}")
return True
except:
pass
print(f"等待 Subconverter 服务启动... ({i+1}/{max_retries})")
time.sleep(2)
print("警告: Subconverter 服务可能未正常启动")
return False
# 应用启动时检查 subconverter 服务
check_subconverter()
@login_manager.user_loader
def load_user(user_id):
return users.get(user_id)
@app.route('/login', methods=['GET', 'POST'])
def login():
if request.method == 'POST':
username = request.form.get('username')
password = request.form.get('password')
user = users.get(username)
if user and check_password_hash(user.password_hash, password):
login_user(user)
return redirect(url_for('index'))
else:
flash('用户名或密码错误')
return render_template('login.html')
@app.route('/logout')
@login_required
def logout():
logout_user()
return redirect(url_for('login'))
@app.route('/')
@login_required
def index():
return render_template('index.html')
@app.route('/convert', methods=['POST'])
@login_required
def convert():
# 获取参数
backend_url = request.form.get('backend_url', 'https://raw.githubusercontent.com/yuanwangokk-1/subscribe/refs/heads/main/ACL4SSR/ACL4SSR.ini')
target = request.form.get('target', 'clash')
original_url = request.form.get('original_url', '')
if not original_url:
return jsonify({"status": "error", "message": "订阅链接不能为空"})
try:
# 检查subconverter服务是否可用
try:
version_check = requests.get('http://localhost:25500/version', timeout=2)
if version_check.status_code != 200:
return jsonify({"status": "error", "message": "订阅转换服务暂时不可用，请稍后再试"})
except:
return jsonify({"status": "error", "message": "订阅转换服务暂时不可用，请稍后再试"})
# 构建转换请求参数
params = {
'target': target,
'url': original_url,
'insert': 'false',
'config': backend_url,
'emoji': 'true',
'list': 'false',
'xudp': 'false',
'udp': 'false',
'tfo': 'false',
'expand': 'true',
'scv': 'false',
'fdn': 'false',
'new_name': 'true'
}
# 通过URL参数构建转换链接
base_url = request.host_url.rstrip('/')
convert_url = f"{base_url}/api/sub?{urllib.parse.urlencode(params)}"
return jsonify({"status": "success", "result": convert_url})
except Exception as e:
return jsonify({"status": "error", "message": f"处理失败: {str(e)}"})
@app.route('/api/sub')
def subscribe_api():
"""代理 subconverter 的转换请求"""
try:
# 获取所有URL参数
params = request.args.to_dict()
# 调用本地 subconverter 服务
sub_url = f"http://localhost:25500/sub?{urllib.parse.urlencode(params)}"
response = requests.get(sub_url, timeout=30)
# 返回原始响应
return Response(
response.content,
status=response.status_code,
content_type=response.headers.get('Content-Type', 'text/plain')
)
except Exception as e:
return f"转换失败: {str(e)}", 500
# 不要在这里调用 app.run()
# 让 Hugging Face 的 WSGI 服务器来运行应用
# 仅在本地开发时使用以下代码
if __name__ == '__main__' and os.environ.get('DEVELOPMENT') == 'true':
app.run(host='0.0.0.0', port=7860, debug=True)
|