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 = """
访问授权
授权访问
{% if error %}
{{ error }}
{% endif %}
"""
@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: 定时任务初始化失败。")