fallensigh
532d117415
- 零依赖 HTTP 服务器 + SQLite 存储 - 与 Go 版功能等价的历史/参考实现 Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent) Co-authored-by: Sisyphus <clio-agent@sisyphuslabs.ai>
271 lines
9.1 KiB
Python
271 lines
9.1 KiB
Python
# -*- 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()
|