#!/usr/bin/env python3 """ 账单查询服务 - Flask版 从 zhangdan.db 读取数据,提供网页展示和API接口 """ import sqlite3 import os import sys import signal import hashlib import secrets import threading import time from urllib.request import urlopen from urllib.error import URLError 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="0.0.0.0", port=PORT, debug=False): """在子线程中启动 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()] # 按交易名称汇总 cur.execute(f""" SELECT payname, payto, COUNT(*) as count, SUM(CAST(paysize AS REAL)) as total FROM zhangdan {where} GROUP BY payname, payto ORDER BY total DESC """, params) by_name = [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, "by_name": by_name, }) @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 驱动

按类型
按渠道
按钱包
按名称
""" # if __name__ == "__main__": # command = sys.argv[1] if len(sys.argv) > 1 else "start" # # ---------- stop ---------- # if command == "stop": # # 尝试通过 HTTP 调 shutdown 接口 # try: # urlopen(f"http://localhost:{PORT}/shutdown?token={SHUTDOWN_TOKEN}", timeout=3) # print("🛑 已发送关闭指令") # except URLError: # print("⚠️ 服务未运行或无法连接") # sys.exit(0) # # ---------- status ---------- # elif command == "status": # info = server_status() # if info["running"]: # print(f"✅ Flask 服务运行中 → http://localhost:{info['port']}") # else: # print("❌ Flask 服务未运行") # sys.exit(0) # # ---------- restart ---------- # elif command == "restart": # stop_server() # time.sleep(0.5) # print("🔄 正在重启...") # # ---------- start (默认) ---------- # elif command != "start": # print(f"用法: python3 {sys.argv[0]} [start|stop|restart|status]") # sys.exit(1) # # 启动 Flask 子线程 # start_server(host="0.0.0.0", port=PORT, debug=True) # # 主线程继续做其他事(不会被 Flask 阻塞) # print("📌 主线程继续运行中... (Ctrl+C 退出)") # try: # while True: # time.sleep(1) # except KeyboardInterrupt: # print("\n👋 主线程退出") # stop_server()