# -*- coding: utf-8 -*- """ 宿舍电费监控 Web 服务 - 零依赖版 - 提供 /api/electric 实时电费查询接口 - 提供 /api/history 历史记录与每日耗电统计 - 每天 24:00(00:00) 定时查询并记录，失败重试 5 次 - 静态前端: static/ (index.html / style.css / app.js) 运行: python server.py (默认 http://127.0.0.1:8080) """ import json import os import sqlite3 import sys import threading import time from datetime import datetime, timedelta from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer from urllib.parse import urlparse # 将 final/ 目录加入 sys.path，复用 lzu_electric_full 模块 BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) # final/ sys.path.insert(0, BASE_DIR) import lzu_electric_full as lef STATIC_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "static") DB_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "electric_history.db") HOST = "127.0.0.1" PORT = 8000 RETRY_TIMES = 5 RETRY_INTERVAL = 60 # 秒 CHECK_INTERVAL = 30 # 定时循环检查间隔(秒) # 查询互斥锁：防止定时任务与手动刷新并发登录 _query_lock = threading.Lock() # ---------- 数据库 ---------- def db_init(): conn = sqlite3.connect(DB_PATH) conn.execute(""" CREATE TABLE IF NOT EXISTS electric_records ( id INTEGER PRIMARY KEY AUTOINCREMENT, date TEXT NOT NULL UNIQUE, -- YYYY-MM-DD surplus REAL NOT NULL, -- 剩余电量(度) amount REAL NOT NULL, -- 预估金额(元) source TEXT NOT NULL, -- auto=定时记录 / manual=手动记录 created_at INTEGER NOT NULL -- Unix 时间戳 ) """) conn.commit() conn.close() def db_record(date_str, surplus, amount, source): conn = sqlite3.connect(DB_PATH) try: conn.execute( "INSERT OR REPLACE INTO electric_records (date, surplus, amount, source, created_at) VALUES (?,?,?,?,?)", (date_str, surplus, amount, source, int(time.time())), ) conn.commit() finally: conn.close() def db_get_records(): conn = sqlite3.connect(DB_PATH) try: cur = conn.execute( "SELECT date, surplus, amount, source FROM electric_records ORDER BY date DESC LIMIT 90" ) rows = [{"date": r[0], "surplus": r[1], "amount": r[2], "source": r[3]} for r in cur.fetchall()] return rows finally: conn.close() def db_has_today(date_str): conn = sqlite3.connect(DB_PATH) try: cur = conn.execute("SELECT 1 FROM electric_records WHERE date=?", (date_str,)) return cur.fetchone() is not None finally: conn.close() # ---------- 电费查询 ---------- def query_electric(): """实时查询电费。返回 dict 或 None。 优先用缓存会话，失效则完整登录，并回写缓存。""" with _query_lock: result = lef.try_query_with_cache() if result is not None: item = result["list"][0] return { "ok": True, "room": result["display"], "surplus": round(item["surplus"], 2), "amount": round(item["amount"], 2), "status": item.get("roomStatus", ""), "source": "cache", "time": int(time.time()), } # 缓存失效 -> 完整登录 ret = lef.full_login_flow() if ret is None: return None app_shiro, room, result = ret lef.save_cache(app_shiro, room) item = result["list"][0] return { "ok": True, "room": result["display"], "surplus": round(item["surplus"], 2), "amount": round(item["amount"], 2), "status": item.get("roomStatus", ""), "source": "login", "time": int(time.time()), } def query_and_record(source="auto"): """查询电费并写入当日记录。成功返回 True。""" data = query_electric() if not data or not data.get("ok"): return False today = datetime.now().strftime("%Y-%m-%d") db_record(today, data["surplus"], data["amount"], source) print("[记录] %s 剩余 %.2f 度 (%.2f 元) [%s]" % (today, data["surplus"], data["amount"], source)) return True # ---------- 每日耗电统计 ---------- def build_daily_stats(records): """records 按日期降序。返回升序的每日统计，含日耗电。""" recs = sorted(records, key=lambda r: r["date"]) # 升序 daily = [] prev_surplus = None for r in recs: usage = None if prev_surplus is not None: usage = round(prev_surplus - r["surplus"], 2) # 负数说明充值/数据异常，归零处理 if usage < 0: usage = 0.0 daily.append({ "date": r["date"], "surplus": r["surplus"], "amount": r["amount"], "usage": usage, }) prev_surplus = r["surplus"] return daily # ---------- 定时任务 ---------- def scheduler_loop(): """每 CHECK_INTERVAL 秒检查一次，到达次日 00:00 且当日无记录时执行。""" last_try_date = None while True: try: now = datetime.now() today = now.strftime("%Y-%m-%d") # 00:00 ~ 00:05 窗口内触发 if now.hour == 0 and now.minute < 5 and last_try_date != today: last_try_date = today if not db_has_today(today): print("[定时] %s 00:00 触发每日电费记录" % today) run_daily_record_with_retry() time.sleep(CHECK_INTERVAL * 2) # 避免窗口内重复触发 else: # 若某天任务失败，每分钟补记一次直至成功或窗口关闭 if now.hour == 0 and not db_has_today(today) and last_try_date != today: last_try_date = today print("[补记] 今日无记录，尝试补记") run_daily_record_with_retry() except Exception as e: print("[定时] 异常:", e) time.sleep(CHECK_INTERVAL) def run_daily_record_with_retry(): """执行一次记录，失败重试 RETRY_TIMES 次。""" for i in range(1, RETRY_TIMES + 1): if query_and_record("auto"): return print("[定时] 第 %d 次尝试失败，%d 秒后重试" % (i, RETRY_INTERVAL)) time.sleep(RETRY_INTERVAL) print("[定时] %d 次重试均失败，等待下轮" % RETRY_TIMES) # ---------- HTTP Handler ---------- class Handler(SimpleHTTPRequestHandler): def __init__(self, *args, **kwargs): super().__init__(*args, directory=STATIC_DIR, **kwargs) def log_message(self, fmt, *args): sys.stdout.write("[%s] %s\n" % (datetime.now().strftime("%H:%M:%S"), fmt % args)) def do_GET(self): parsed = urlparse(self.path) path = parsed.path if path == "/api/electric": self._send_json(self._handle_electric()) elif path == "/api/history": self._send_json(self._handle_history()) elif path == "/": self.path = "/index.html" super().do_GET() elif path.startswith("/static/"): self.path = path[len("/static/"):] super().do_GET() else: self.send_error(404) # -- API 实现 -- def _handle_electric(self): try: data = query_electric() if data is None: return {"ok": False, "error": "电费查询失败，请查看服务器日志"} return data except Exception as e: return {"ok": False, "error": str(e)} def _handle_history(self): records = db_get_records() daily = build_daily_stats(records) return {"ok": True, "records": records, "daily": daily} def _send_json(self, obj): body = json.dumps(obj, ensure_ascii=False).encode("utf-8") self.send_response(200) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(body))) self.send_header("Cache-Control", "no-store") self.end_headers() self.wfile.write(body) def main(): db_init() # 启动定时任务线程（守护） threading.Thread(target=scheduler_loop, daemon=True).start() # 启动时立即记录一次当日数据（若缺失），确保首日有基线 today = datetime.now().strftime("%Y-%m-%d") if not db_has_today(today): print("[启动] 今日无记录，立即查询并记录") threading.Thread(target=run_daily_record_with_retry, daemon=True).start() server = ThreadingHTTPServer((HOST, PORT), Handler) print("=" * 50) print("宿舍电费监控服务已启动") print(" 访问地址: http://%s:%d" % (HOST, PORT)) print(" 定时记录: 每天 00:00，失败重试 %d 次" % RETRY_TIMES) print(" 数据库: %s" % DB_PATH) print("=" * 50) try: server.serve_forever() except KeyboardInterrupt: print("\n服务已停止") server.shutdown() if __name__ == "__main__": main()