| import os
|
| from flask import Flask, request, jsonify, render_template_string, redirect, url_for, session, render_template
|
| import config
|
| import data_manager
|
| import scheduler_jobs
|
| import api_client
|
| import requests
|
|
|
| app = Flask(__name__)
|
| app.secret_key = config.SECRET_KEY
|
| app.permanent_session_lifetime = config.PERMANENT_SESSION_LIFETIME
|
|
|
| LOGIN_FORM_HTML = """
|
| <!DOCTYPE html>
|
| <html lang="zh-CN">
|
| <head>
|
| <meta charset="UTF-8">
|
| <meta name="viewport" content="width=device-width, initial-scale=1.0">
|
| <title>访问授权</title>
|
| <style>
|
| body {
|
| font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", Roboto, "Helvetica Neue", Arial, sans-serif;
|
| display: flex;
|
| justify-content: center;
|
| align-items: center;
|
| height: 100vh;
|
| margin: 0;
|
| background-color: #fff;
|
| color: #000;
|
| }
|
| .login-container {
|
| background-color: #fff;
|
| padding: 40px;
|
| border-radius: 8px;
|
| box-shadow: 0 8px 30px rgba(0,0,0,0.1);
|
| width: 100%;
|
| max-width: 360px;
|
| border: 1px solid #eaeaea;
|
| }
|
| h2 {
|
| text-align: center;
|
| color: #000;
|
| font-weight: 600;
|
| margin-bottom: 30px;
|
| }
|
| label {
|
| display: block;
|
| margin-bottom: 8px;
|
| color: #444;
|
| font-size: 14px;
|
| }
|
| input[type="password"] {
|
| width: 100%;
|
| padding: 12px;
|
| margin-bottom: 20px;
|
| border: 1px solid #ccc;
|
| border-radius: 6px;
|
| box-sizing: border-box;
|
| background-color: #fff;
|
| color: #000;
|
| font-size: 16px;
|
| }
|
| input[type="password"]:focus {
|
| border-color: #999;
|
| outline: none;
|
| box-shadow: 0 0 0 2px rgba(0,0,0,0.05);
|
| }
|
| input[type="submit"] {
|
| width: 100%;
|
| padding: 12px;
|
| background-color: #000;
|
| color: #fff;
|
| border: none;
|
| border-radius: 6px;
|
| cursor: pointer;
|
| font-size: 16px;
|
| font-weight: 500;
|
| transition: background-color 0.2s ease;
|
| }
|
| input[type="submit"]:hover {
|
| background-color: #333;
|
| }
|
| .error {
|
| color: #e53e3e;
|
| text-align: center;
|
| margin-bottom: 15px;
|
| font-size: 14px;
|
| }
|
| </style>
|
| </head>
|
| <body>
|
| <div class="login-container">
|
| <h2>授权访问</h2>
|
| {% if error %}
|
| <p class="error">{{ error }}</p>
|
| {% endif %}
|
| <form method="post">
|
| <label for="password">访问密码:</label>
|
| <input type="password" id="password" name="password" required autofocus>
|
| <input type="submit" value="继续">
|
| </form>
|
| </div>
|
| </body>
|
| </html>
|
| """
|
|
|
| @app.route('/', methods=['GET', 'POST'])
|
| def login_frontend():
|
| if not config.FRONTEND_PASSWORD:
|
| return "错误: 前端密码 (PASSWORD 环境变量) 未设置。", 500
|
|
|
| if 'logged_in' in session and session['logged_in']:
|
| return redirect(url_for('dashboard'))
|
|
|
| error = None
|
| if request.method == 'POST':
|
| entered_password = request.form.get('password')
|
| if entered_password == config.FRONTEND_PASSWORD:
|
| accounts_list = data_manager.get_accounts_from_config()
|
| if not accounts_list:
|
| error = "配置错误：未找到账户信息。"
|
| return render_template_string(LOGIN_FORM_HTML, error=error)
|
|
|
| first_account_email = accounts_list[0]['email']
|
| first_account_password = accounts_list[0]['password']
|
|
|
| jwt_token, login_api_error = api_client.api_login(first_account_email, first_account_password)
|
|
|
| if login_api_error:
|
| error = f"API 登录失败: {login_api_error}"
|
| return render_template_string(LOGIN_FORM_HTML, error=error)
|
|
|
| if not jwt_token:
|
| error = "API 登录成功但未能获取 Token。"
|
| return render_template_string(LOGIN_FORM_HTML, error=error)
|
|
|
| session['logged_in'] = True
|
| session['jwt_token'] = jwt_token
|
| session.permanent = True
|
| return redirect(url_for('dashboard'))
|
| else:
|
| error = "密码错误！"
|
| return render_template_string(LOGIN_FORM_HTML, error=error)
|
|
|
| @app.route('/dashboard')
|
| def dashboard():
|
| if not ('logged_in' in session and session['logged_in']):
|
| return redirect(url_for('login_frontend'))
|
| return render_template_string(open('templates/dashboard.html', encoding='utf-8').read())
|
|
|
| @app.route('/dashboard/statements')
|
| def account_statements_page():
|
| if not ('logged_in' in session and session['logged_in']):
|
| return redirect(url_for('login_frontend'))
|
|
|
| account_email_param = request.args.get('email')
|
| if not account_email_param:
|
| return redirect(url_for('dashboard'))
|
|
|
| return render_template('account_statements.html', account_email=account_email_param)
|
|
|
| @app.route('/logout')
|
| def logout():
|
| session.pop('logged_in', None)
|
| return redirect(url_for('login_frontend'))
|
|
|
| @app.route('/api/data', methods=['GET'])
|
| def get_all_data():
|
| if not ('logged_in' in session and session['logged_in']):
|
| return jsonify({"error": "未授权访问"}), 401
|
|
|
| display_data = data_manager.get_accounts_data_for_display()
|
| return jsonify(display_data)
|
|
|
| @app.route('/api/account_statements', methods=['GET'])
|
| def get_account_statements():
|
| if not ('logged_in' in session and session['logged_in']):
|
| return jsonify({"error": "未授权访问"}), 401
|
|
|
| email_param = request.args.get('email')
|
| if not email_param:
|
| return jsonify({"error": "缺少 'email' 参数"}), 400
|
|
|
| page = request.args.get('page', 1, type=int)
|
| size = request.args.get('size', 10000, type=int)
|
|
|
| token_to_use = None
|
| target_account_password = None
|
|
|
| all_accounts = data_manager.get_accounts_from_config()
|
| for acc in all_accounts:
|
| if acc.get('email') == email_param:
|
| target_account_password = acc.get('password')
|
| break
|
|
|
| if not target_account_password:
|
| return jsonify({"error": f"未找到账户 {email_param} 的配置信息或密码。"}), 404
|
|
|
| print(f"为特定账户 {email_param} 获取账单，尝试重新登录获取专属 token...")
|
| specific_token, login_error = api_client.api_login(email_param, target_account_password)
|
|
|
| if login_error:
|
| print(f"为账户 {email_param} 获取专属 token 失败: {login_error}")
|
| return jsonify({"error": f"为账户 {email_param} 获取账单时登录失败: {login_error}"}), 500
|
|
|
| if not specific_token:
|
| return jsonify({"error": f"为账户 {email_param} 获取账单时未能获取专属 API token。"}), 500
|
|
|
| token_to_use = specific_token
|
| print(f"成功获取账户 {email_param} 的专属 token，将用于获取账单。")
|
|
|
| statement_data, error = api_client.get_statement_records(token_to_use, page, size)
|
|
|
| if error:
|
| return jsonify({"error": f"获取账户 {email_param} 的账单失败: {error}"}), 500
|
|
|
| return jsonify(statement_data)
|
|
|
| @app.route('/api/account_summary_statistics', methods=['GET'])
|
| def get_account_summary_statistics():
|
| if not ('logged_in' in session and session['logged_in']):
|
| return jsonify({"error": "未授权访问"}), 401
|
|
|
| email_param = request.args.get('email')
|
| if not email_param:
|
| return jsonify({"error": "缺少 'email' 参数"}), 400
|
|
|
| target_account_password = None
|
| all_accounts = data_manager.get_accounts_from_config()
|
| for acc in all_accounts:
|
| if acc.get('email') == email_param:
|
| target_account_password = acc.get('password')
|
| break
|
|
|
| if not target_account_password:
|
| return jsonify({"error": f"未找到账户 {email_param} 的配置信息或密码。"}), 404
|
|
|
| print(f"为特定账户 {email_param} 获取统计摘要，尝试重新登录获取专属 token...")
|
| specific_token, login_error = api_client.api_login(email_param, target_account_password)
|
|
|
| if login_error:
|
| print(f"为账户 {email_param} 获取专属 token 失败: {login_error}")
|
| return jsonify({"error": f"为账户 {email_param} 获取统计摘要时登录失败: {login_error}"}), 500
|
|
|
| if not specific_token:
|
| return jsonify({"error": f"为账户 {email_param} 获取统计摘要时未能获取专属 API token。"}), 500
|
|
|
| token_to_use = specific_token
|
| print(f"成功获取账户 {email_param} 的专属 token，将用于获取统计摘要。")
|
|
|
| external_api_url = "https://api-card.infini.money/user/statement/statistics"
|
| headers = {
|
| "Authorization": f"Bearer {token_to_use}",
|
| "User-Agent": "InfiniDashboard/1.0"
|
| }
|
|
|
| try:
|
| response = requests.get(external_api_url, headers=headers, timeout=10)
|
| response.raise_for_status()
|
|
|
| api_response_data = response.json()
|
|
|
| return jsonify(api_response_data)
|
|
|
| except requests.exceptions.HTTPError as http_err:
|
| print(f"HTTP error occurred while fetching statistics for {email_param}: {http_err}")
|
| try:
|
| error_content = http_err.response.json()
|
| return jsonify({"error": f"获取统计摘要失败 (API错误): {error_content.get('message', str(http_err))}", "details": error_content}), http_err.response.status_code
|
| except:
|
| return jsonify({"error": f"获取统计摘要失败: {str(http_err)}"}), getattr(http_err.response, 'status_code', 500)
|
| except requests.exceptions.RequestException as req_err:
|
| print(f"Request error occurred while fetching statistics for {email_param}: {req_err}")
|
| return jsonify({"error": f"获取统计摘要时发生网络请求错误: {str(req_err)}"}), 503
|
| except Exception as e:
|
| print(f"An unexpected error occurred while fetching statistics for {email_param}: {e}")
|
| return jsonify({"error": f"获取统计摘要时发生未知错误: {str(e)}"}), 500
|
|
|
|
|
| @app.route('/api/refresh', methods=['POST'])
|
| def manual_refresh_all_data():
|
| if not ('logged_in' in session and session['logged_in']):
|
| return jsonify({"error": "未授权访问"}), 401
|
|
|
| scheduler_jobs.manual_refresh_all_data_job()
|
| return jsonify({"message": "刷新任务已启动，请稍后查看数据。"}), 202
|
|
|
| if __name__ == '__main__':
|
| if not config.FRONTEND_PASSWORD:
|
| print("警告: PASSWORD 环境变量未设置，前端将无法登录。程序将尝试继续，但部分功能可能受限。")
|
|
|
| if scheduler_jobs.initialize_scheduler(is_gunicorn=False):
|
| is_hf_space = os.getenv("SPACE_ID") is not None
|
| if not is_hf_space:
|
| print("在本地环境运行 Flask 开发服务器...")
|
| app.run(host='0.0.0.0', port=int(os.getenv("PORT", 7860)), debug=False)
|
| else:
|
| print("检测到在 Hugging Face Space 环境中运行。假定由 Gunicorn 等 WSGI 服务器启动。")
|
| else:
|
| print("定时任务初始化失败，程序退出。")
|
|
|
|
|
| else:
|
| print("在 Gunicorn 环境下初始化应用...")
|
| if not config.FRONTEND_PASSWORD:
|
| print("Gunicorn警告: PASSWORD 环境变量未设置，前端将无法登录。")
|
|
|
| if not scheduler_jobs.initialize_scheduler(is_gunicorn=True):
|
| print("Gunicorn: 定时任务初始化失败。")
|
|
|
