dekun
e5a586f903
Move business code under modules/, env template to config/, PM2 single qihuo process, and _legacy shims for old imports. Co-authored-by: Cursor <cursoragent@cursor.com>
109 lines
3.4 KiB
Python
109 lines
3.4 KiB
Python
# Copyright (c) 2025-2026 马建军. All rights reserved.
|
|
"""HTTP routes for web module."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import date, datetime
|
|
|
|
from flask import (
|
|
Response,
|
|
flash,
|
|
jsonify,
|
|
redirect,
|
|
render_template,
|
|
request,
|
|
send_file,
|
|
session,
|
|
stream_with_context,
|
|
url_for,
|
|
)
|
|
|
|
|
|
def register(deps) -> None:
|
|
app = deps.app
|
|
login_required = deps.login_required
|
|
require_nav = deps.require_nav
|
|
get_db = deps.get_db
|
|
get_setting = deps.get_setting
|
|
set_setting = deps.set_setting
|
|
fetch_price = deps.fetch_price
|
|
send_wechat_msg = deps.send_wechat_msg
|
|
touch_stats_cache = deps.touch_stats_cache
|
|
get_stats_data = deps.get_stats_data
|
|
build_market_quote_payload = deps.build_market_quote_payload
|
|
today_str = deps.today_str
|
|
expire_old_plans = deps.expire_old_plans
|
|
TZ = deps.tz
|
|
DB_PATH = deps.db_path
|
|
UPLOAD_DIR = deps.upload_dir
|
|
OPEN_TYPES = deps.open_types
|
|
EXIT_TRIGGERS = deps.exit_triggers
|
|
BEHAVIOR_TAGS = deps.behavior_tags
|
|
KLINE_PERIODS = deps.kline_periods
|
|
KLINE_CUTOFFS = deps.kline_cutoffs
|
|
calc_holding_duration = deps.calc_holding_duration
|
|
holding_to_minutes = deps.holding_to_minutes
|
|
classify_close_result = deps.classify_close_result
|
|
calc_rr_ratio = deps.calc_rr_ratio
|
|
calc_theoretical_pnl = deps.calc_theoretical_pnl
|
|
parse_review_date_filter = deps.parse_review_date_filter
|
|
_trading_mode = deps.trading_mode
|
|
_ua_is_phone = deps.ua_is_phone
|
|
_static_asset_v = deps.static_asset_v
|
|
|
|
from werkzeug.security import check_password_hash
|
|
import json
|
|
|
|
|
|
@app.route("/")
|
|
def index():
|
|
if session.get("logged_in"):
|
|
return redirect(url_for("positions"))
|
|
return redirect(url_for("login"))
|
|
|
|
|
|
@app.route("/manifest.webmanifest")
|
|
def web_manifest():
|
|
import json
|
|
|
|
manifest_path = os.path.join(app.static_folder, "manifest.json")
|
|
with open(manifest_path, encoding="utf-8") as fh:
|
|
data = json.load(fh)
|
|
if _ua_is_phone(request.headers.get("User-Agent", "")):
|
|
data["orientation"] = "portrait-primary"
|
|
else:
|
|
data["orientation"] = "any"
|
|
response = app.make_response(json.dumps(data, ensure_ascii=False))
|
|
response.mimetype = "application/manifest+json"
|
|
response.headers["Cache-Control"] = "no-cache"
|
|
return response
|
|
|
|
|
|
@app.route("/sw.js")
|
|
def service_worker():
|
|
response = app.send_static_file("sw.js")
|
|
response.headers["Cache-Control"] = "no-cache"
|
|
response.headers["Service-Worker-Allowed"] = "/"
|
|
return response
|
|
|
|
|
|
@app.route("/login", methods=["GET", "POST"])
|
|
def login():
|
|
if request.method == "POST":
|
|
u = request.form.get("username", "").strip()
|
|
p = request.form.get("password", "")
|
|
admin_u = get_setting("admin_username")
|
|
admin_hash = get_setting("admin_password_hash")
|
|
if u == admin_u and check_password_hash(admin_hash, p):
|
|
session["logged_in"] = True
|
|
session["username"] = u
|
|
return redirect(url_for("positions"))
|
|
flash("账号或密码错误")
|
|
return render_template("login.html")
|
|
|
|
|
|
@app.route("/logout")
|
|
def logout():
|
|
session.clear()
|
|
return redirect(url_for("login"))
|