feat: Python 参考实现(server.py)

- 零依赖 HTTP 服务器 + SQLite 存储

- 与 Go 版功能等价的历史/参考实现

Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent)

Co-authored-by: Sisyphus <clio-agent@sisyphuslabs.ai>
This commit is contained in:
2026-08-23 21:29:55 +08:00
parent f207a0d0bd
commit 532d117415
4 changed files with 686 additions and 0 deletions

270
server.py Normal file
View File

@@ -0,0 +1,270 @@
# -*- 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()

153
static/app.js Normal file
View File

@@ -0,0 +1,153 @@
// 宿舍电费监控 - 前端逻辑
const POLL_INTERVAL = 30000; // 自动刷新间隔 30s
let usageChart = null;
const $ = (id) => document.getElementById(id);
// ---------- 工具 ----------
function formatTime(ts) {
if (!ts) return "--";
const d = new Date(ts * 1000);
return d.toLocaleString("zh-CN", { hour12: false });
}
function formatDateStr(dateStr) {
const d = new Date(dateStr);
return d.toLocaleDateString("zh-CN", { month: "numeric", day: "numeric" });
}
function setBadge(state, text) {
const el = $("statusBadge");
el.className = "status-badge " + state;
el.textContent = text;
}
// ---------- 当前电费 ----------
async function fetchCurrent() {
setBadge("loading", "查询中...");
try {
const resp = await fetch("/api/electric");
const data = await resp.json();
if (data.ok) {
$("surplusValue").textContent = data.surplus.toFixed(2);
$("amountValue").textContent = data.amount.toFixed(2);
$("updateTime").textContent = formatTime(data.time);
$("roomInfo").textContent = "🏠 " + data.room;
setBadge("ok", "实时");
} else {
setBadge("err", "查询失败");
$("roomInfo").textContent = "⚠️ " + (data.error || "查询失败");
}
} catch (e) {
setBadge("err", "网络错误");
}
}
// ---------- 历史记录 ----------
function renderChart(daily) {
$("chartEmpty").style.display = daily.length ? "none" : "block";
if (!daily.length) return;
const labels = daily.map((d) => formatDateStr(d.date));
const usage = daily.map((d) => d.usage);
const surplus = daily.map((d) => d.surplus);
const ctx = $("usageChart").getContext("2d");
if (usageChart) usageChart.destroy();
usageChart = new Chart(ctx, {
data: {
labels,
datasets: [
{
type: "bar",
label: "日耗电(度)",
data: usage,
backgroundColor: "rgba(255, 167, 38, 0.75)",
borderRadius: 6,
yAxisID: "y",
},
{
type: "line",
label: "剩余电量(度)",
data: surplus,
borderColor: "#4fc3f7",
backgroundColor: "rgba(79, 195, 247, 0.15)",
tension: 0.3,
pointRadius: 4,
yAxisID: "y1",
},
],
},
options: {
responsive: true,
interaction: { mode: "index", intersect: false },
plugins: {
legend: {
labels: { color: "#e8edf2", usePointStyle: true },
},
},
scales: {
x: { ticks: { color: "#9fb3c8" }, grid: { color: "rgba(255,255,255,0.05)" } },
y: {
position: "left",
title: { display: true, text: "耗电(度)", color: "#ffa726" },
ticks: { color: "#9fb3c8" },
grid: { color: "rgba(255,255,255,0.05)" },
},
y1: {
position: "right",
title: { display: true, text: "剩余(度)", color: "#4fc3f7" },
ticks: { color: "#9fb3c8" },
grid: { drawOnChartArea: false },
},
},
},
});
}
function renderHistory(records) {
const body = $("historyBody");
if (!records.length) {
body.innerHTML = '<tr><td colspan="4" class="empty-td">暂无记录</td></tr>';
return;
}
body.innerHTML = records
.map((r) => {
const usageTxt = r.usage >= 0 ? r.usage.toFixed(2) : "--";
const tagCls = r.source === "auto" ? "auto" : "manual";
const tagTxt = r.source === "auto" ? "自动" : "手动";
return `<tr>
<td>${formatDateStr(r.date)}</td>
<td>${r.surplus.toFixed(2)}</td>
<td>${usageTxt}</td>
<td><span class="tag ${tagCls}">${tagTxt}</span></td>
</tr>`;
})
.join("");
}
async function fetchHistory() {
try {
const resp = await fetch("/api/history");
const data = await resp.json();
renderChart(data.daily || []);
renderHistory(data.records || []);
} catch (e) {
console.error("history fetch error", e);
}
}
// ---------- 主流程 ----------
async function refreshAll() {
await Promise.all([fetchCurrent(), fetchHistory()]);
}
$("refreshBtn").addEventListener("click", () => {
$("refreshBtn").disabled = true;
refreshAll().finally(() => {
$("refreshBtn").disabled = false;
});
});
refreshAll();
setInterval(refreshAll, POLL_INTERVAL);

63
static/index.html Normal file
View File

@@ -0,0 +1,63 @@
<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>宿舍电费监控</title>
<link rel="stylesheet" href="/static/style.css">
</head>
<body>
<div class="container">
<header>
<h1>⚡ 宿舍电费监控</h1>
<div class="room-info" id="roomInfo">加载中...</div>
</header>
<section class="card current-card">
<div class="card-header">
<span class="label">当前剩余电量</span>
<span class="status-badge" id="statusBadge">--</span>
</div>
<div class="surplus-display">
<span class="surplus-value" id="surplusValue">--</span>
<span class="surplus-unit"></span>
</div>
<div class="sub-info">
<span>预估金额: ¥<span id="amountValue">--</span></span>
<span>更新于: <span id="updateTime">--</span></span>
</div>
<button id="refreshBtn" class="refresh-btn">🔄 立即刷新</button>
</section>
<section class="card">
<div class="card-header">
<span class="label">近 7 天耗电量</span>
</div>
<canvas id="usageChart" height="220"></canvas>
<div class="chart-empty" id="chartEmpty">暂无历史数据</div>
</section>
<section class="card">
<div class="card-header">
<span class="label">每日电费记录</span>
</div>
<table class="history-table">
<thead>
<tr>
<th>日期</th>
<th>剩余电量(度)</th>
<th>日耗电(度)</th>
<th>记录方式</th>
</tr>
</thead>
<tbody id="historyBody">
<tr><td colspan="4" class="empty-td">加载中...</td></tr>
</tbody>
</table>
</section>
</div>
<script src="https://cdn.jsdelivr.net/npm/chart.js@4.4.0/dist/chart.umd.min.js"></script>
<script src="/static/app.js"></script>
</body>
</html>

200
static/style.css Normal file
View File

@@ -0,0 +1,200 @@
* {
margin: 0;
padding: 0;
box-sizing: border-box;
}
body {
font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", "PingFang SC",
"Microsoft YaHei", sans-serif;
background: linear-gradient(135deg, #0f2027 0%, #203a43 50%, #2c5364 100%);
min-height: 100vh;
color: #e8edf2;
}
.container {
max-width: 720px;
margin: 0 auto;
padding: 24px 16px 48px;
}
header {
text-align: center;
margin-bottom: 24px;
}
header h1 {
font-size: 26px;
font-weight: 700;
letter-spacing: 1px;
background: linear-gradient(90deg, #ffd700, #ffaa00);
-webkit-background-clip: text;
-webkit-text-fill-color: transparent;
}
.room-info {
margin-top: 8px;
font-size: 14px;
color: #9fb3c8;
}
.card {
background: rgba(255, 255, 255, 0.06);
backdrop-filter: blur(10px);
border: 1px solid rgba(255, 255, 255, 0.1);
border-radius: 16px;
padding: 20px;
margin-bottom: 20px;
box-shadow: 0 8px 32px rgba(0, 0, 0, 0.3);
}
.card-header {
display: flex;
justify-content: space-between;
align-items: center;
margin-bottom: 12px;
}
.label {
font-size: 14px;
color: #9fb3c8;
font-weight: 500;
}
.status-badge {
padding: 3px 12px;
border-radius: 20px;
font-size: 12px;
font-weight: 600;
}
.status-badge.ok {
background: rgba(46, 204, 113, 0.2);
color: #2ecc71;
}
.status-badge.err {
background: rgba(231, 76, 60, 0.2);
color: #e74c3c;
}
.status-badge.loading {
background: rgba(241, 196, 15, 0.2);
color: #f1c40f;
}
.surplus-display {
display: flex;
align-items: baseline;
justify-content: center;
padding: 16px 0 8px;
}
.surplus-value {
font-size: 64px;
font-weight: 800;
background: linear-gradient(90deg, #4fc3f7, #00e5ff);
-webkit-background-clip: text;
-webkit-text-fill-color: transparent;
}
.surplus-unit {
font-size: 20px;
color: #9fb3c8;
margin-left: 8px;
}
.sub-info {
display: flex;
justify-content: center;
gap: 24px;
font-size: 13px;
color: #9fb3c8;
margin-bottom: 16px;
}
.refresh-btn {
display: block;
width: 100%;
padding: 12px;
border: none;
border-radius: 10px;
background: linear-gradient(90deg, #00b4d8, #0096c7);
color: #fff;
font-size: 15px;
font-weight: 600;
cursor: pointer;
transition: transform 0.15s, opacity 0.15s;
}
.refresh-btn:hover {
transform: translateY(-1px);
opacity: 0.9;
}
.refresh-btn:disabled {
opacity: 0.5;
cursor: not-allowed;
}
.chart-wrap {
position: relative;
}
.chart-empty {
display: none;
text-align: center;
padding: 40px 0;
color: #9fb3c8;
font-size: 14px;
}
.history-table {
width: 100%;
border-collapse: collapse;
font-size: 13px;
}
.history-table th,
.history-table td {
padding: 10px 8px;
text-align: center;
border-bottom: 1px solid rgba(255, 255, 255, 0.08);
}
.history-table th {
color: #9fb3c8;
font-weight: 500;
font-size: 12px;
}
.history-table td:first-child {
text-align: left;
}
.empty-td {
color: #9fb3c8;
text-align: center !important;
padding: 24px 0 !important;
}
.tag {
padding: 2px 8px;
border-radius: 12px;
font-size: 11px;
}
.tag.auto {
background: rgba(0, 180, 216, 0.2);
color: #4fc3f7;
}
.tag.manual {
background: rgba(241, 196, 15, 0.2);
color: #f1c40f;
}
.tag.cache {
background: rgba(46, 204, 113, 0.2);
color: #2ecc71;
}