#!/usr/bin/env python3 """ 账单查询服务 - Flask版 从 zhangdan.db 读取数据，提供网页展示和API接口 """ import sqlite3 import os import secrets import threading from flask import Flask, render_template_string, jsonify, request, session, redirect, url_for from flask_cors import CORS # ===== 🔧 常量 ===== PORT = 5001 PID_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), ".flask.pid") # ===== 🔧 全局状态 ===== _server = None # werkzeug server 实例 _server_thread = None # 服务线程 _server_lock = threading.Event() # 启动 def runwebui(web_port,web_passwd,datadb): global AUTH_PASSWORD,DB_PATH,AUTH_COOKIE_MAX_AGE try: AUTH_PASSWORD = web_passwd AUTH_COOKIE_MAX_AGE = 86400 * 1 # 1天免登录 DB_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), datadb) start_server(host="0.0.0.0", port=web_port, debug=False ) except EOFError as e: print(e) def start_server(host, port, debug): """在子线程中启动 Flask 服务，主线程不受影响""" global _server, _server_thread if _server_thread and _server_thread.is_alive(): print("⚠️ 服务已在运行中") return _server from werkzeug.serving import make_server _server = make_server(host, port, app, threaded=True) _server_thread = threading.Thread(target=_server.serve_forever, daemon=True) _server_thread.start() _server_lock.set() # 写 PID with open(PID_FILE, "w") as f: f.write(str(os.getpid())) print(f"🚀 服务已启动 (子线程) → http://{host}:{port}") return _server def stop_server(): """只关闭 Flask 子线程，主线程继续运行""" global _server, _server_thread if _server: try: _server.shutdown() # 优雅关闭 werkzeug server except Exception: pass _server = None if _server_thread and _server_thread.is_alive(): _server_thread.join(timeout=3) _server_thread = None _server_lock.clear() # 清理 PID 文件 if os.path.exists(PID_FILE): try: os.remove(PID_FILE) except OSError: pass print("🛑 Flask 服务已关闭（主线程仍在运行）") def server_status(): """检查 Flask 子线程是否在运行""" alive = _server_thread is not None and _server_thread.is_alive() return {"running": alive, "port": PORT} app = Flask(__name__) app.secret_key = secrets.token_hex(32) CORS(app) def check_auth(): """检查是否已登录""" return session.get("authenticated") is True def login_required(f): """登录装饰器""" from functools import wraps @wraps(f) def decorated(*args, **kwargs): if not check_auth(): return redirect(url_for("login_page")) return f(*args, **kwargs) return decorated def get_db(): conn = sqlite3.connect(DB_PATH) conn.row_factory = sqlite3.Row return conn @app.route("/login", methods=["GET", "POST"]) def login_page(): error = None if request.method == "POST": pwd = request.form.get("password", "") if pwd == AUTH_PASSWORD: session["authenticated"] = True session.permanent = True app.permanent_session_lifetime = __import__("datetime").timedelta(seconds=AUTH_COOKIE_MAX_AGE) return redirect(url_for("index")) error = "密码不对哦，再试试 🤔" return render_template_string(LOGIN_TEMPLATE, error=error) @app.route("/logout") def logout(): session.clear() return redirect(url_for("login_page")) # ===== 🔴 关机密令 ===== SHUTDOWN_TOKEN = "shutdown_2026" @app.route("/shutdown") def shutdown_server(): """通过密令关闭服务，访问 /shutdown?token=xxx 即可关机""" token = request.args.get("token", "") if token != SHUTDOWN_TOKEN: return "❌ 无效密令", 403 # 在新线程中关闭，避免阻塞当前请求 threading.Thread(target=stop_server, daemon=True).start() return "🛑 服务已关闭" @app.route("/") @login_required def index(): return render_template_string(HTML_TEMPLATE, shutdown_token=SHUTDOWN_TOKEN) @app.route("/api/bills") @login_required def api_bills(): """返回所有账单数据，支持排序和筛选""" conn = get_db() cur = conn.cursor() # 筛选参数 payto = request.args.get("payto") # 收入/支出 payqd = request.args.get("payqd") # 渠道 paybox = request.args.get("paybox") # 钱包 paytype = request.args.get("paytype") # 交易类型 keyword = request.args.get("keyword") # 搜索关键词 date_from = request.args.get("date_from") # 起始日期 date_to = request.args.get("date_to") # 截止日期 # 排序参数 sort = request.args.get("sort", "paydate") order = request.args.get("order", "DESC") allowed_sorts = { "paydate", "paysize", "paytype", "payname", "payto", "paystu", "paybox", "payqd", "id" } if sort not in allowed_sorts: sort = "paydate" order = "ASC" if order.upper() == "ASC" else "DESC" query = "SELECT * FROM zhangdan WHERE 1=1" params = [] if payto: query += " AND payto = ?" params.append(payto) if payqd: query += " AND payqd = ?" params.append(payqd) if paybox: query += " AND paybox = ?" params.append(paybox) if paytype: query += " AND paytype = ?" params.append(paytype) if keyword: query += " AND (payname LIKE ? OR paytype LIKE ?)" params.extend([f"%{keyword}%", f"%{keyword}%"]) if date_from: query += " AND paydate >= ?" params.append(date_from) if date_to: query += " AND paydate <= ?" params.append(date_to + " 23:59:59") # paysize 存的是文本，需要特殊处理排序 if sort == "paysize": query += f" ORDER BY CAST({sort} AS REAL) {order}" else: query += f" ORDER BY {sort} {order}" cur.execute(query, params) rows = [dict(r) for r in cur.fetchall()] conn.close() return jsonify(rows) @app.route("/api/summary") @login_required def api_summary(): """汇总统计 — 支持与 /api/bills 相同的筛选参数""" conn = get_db() cur = conn.cursor() # 收集筛选条件（与 /api/bills 一致） payto = request.args.get("payto") payqd = request.args.get("payqd") paybox = request.args.get("paybox") paytype = request.args.get("paytype") keyword = request.args.get("keyword") date_from = request.args.get("date_from") date_to = request.args.get("date_to") where = "WHERE 1=1" params = [] if payto: where += " AND payto = ?" params.append(payto) if payqd: where += " AND payqd = ?" params.append(payqd) if paybox: where += " AND paybox = ?" params.append(paybox) if paytype: where += " AND paytype = ?" params.append(paytype) if keyword: where += " AND (payname LIKE ? OR paytype LIKE ?)" params.extend([f"%{keyword}%", f"%{keyword}%"]) if date_from: where += " AND paydate >= ?" params.append(date_from) if date_to: where += " AND paydate <= ?" params.append(date_to + " 23:59:59") # 总收入/支出 cur.execute(f""" SELECT payto, COUNT(*) as count, SUM(CAST(paysize AS REAL)) as total FROM zhangdan {where} GROUP BY payto """, params) overview = [dict(r) for r in cur.fetchall()] # 按类型汇总 cur.execute(f""" SELECT paytype, payto, COUNT(*) as count, SUM(CAST(paysize AS REAL)) as total FROM zhangdan {where} GROUP BY paytype, payto ORDER BY total DESC """, params) by_type = [dict(r) for r in cur.fetchall()] # 按渠道汇总 cur.execute(f""" SELECT payqd, payto, COUNT(*) as count, SUM(CAST(paysize AS REAL)) as total FROM zhangdan {where} GROUP BY payqd, payto ORDER BY total DESC """, params) by_channel = [dict(r) for r in cur.fetchall()] # 按钱包汇总 cur.execute(f""" SELECT paybox, payto, COUNT(*) as count, SUM(CAST(paysize AS REAL)) as total FROM zhangdan {where} GROUP BY paybox, payto ORDER BY total DESC """, params) by_wallet = [dict(r) for r in cur.fetchall()] conn.close() return jsonify({ "overview": overview, "by_type": by_type, "by_channel": by_channel, "by_wallet": by_wallet, }) @app.route("/api/filters") @login_required def api_filters(): """获取所有可选的筛选值""" conn = get_db() cur = conn.cursor() result = {} for col in ["payto", "payqd", "paybox", "paytype"]: cur.execute(f"SELECT DISTINCT {col} FROM zhangdan ORDER BY {col}") result[col] = [r[0] for r in cur.fetchall()] conn.close() return jsonify(result) LOGIN_TEMPLATE = r""" 🔐 验证身份
🔒

账单查询

请输入密码以继续

{% if error %}
{{ error }}
{% endif %}

数据已加密保护 🔐

""" HTML_TEMPLATE = r""" 💰 账单查询

💰 账单查询

数据来源: zhangdan.db · Flask API 驱动

按类型
按渠道
按钱包
· Flask + SQLite · 0 条记录 · 退出登录 🔒 · 关闭服务 ⏻
"""