424 lines
17 KiB
Python
424 lines
17 KiB
Python
"""FastAPI 应用：认证、平台/账号 CRUD、设置、静态前端。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import logging
|
|
import secrets
|
|
import sqlite3
|
|
from contextlib import asynccontextmanager
|
|
from pathlib import Path
|
|
|
|
from fastapi import Depends, FastAPI, Header, HTTPException, Query
|
|
from fastapi.responses import FileResponse
|
|
from fastapi.staticfiles import StaticFiles
|
|
from pydantic import ValidationError
|
|
|
|
from app import db
|
|
from app.config import Config, load_config, save_config
|
|
from app.models import (
|
|
AccountCreate,
|
|
AccountUpdate,
|
|
LoginRequest,
|
|
PasswordChange,
|
|
PlatformCreate,
|
|
PlatformUpdate,
|
|
SettingsUpdate,
|
|
)
|
|
from app.monitor import Monitor
|
|
from app.providers import get_provider, list_providers as list_providers_svc
|
|
|
|
logger = logging.getLogger("monitor.api")
|
|
|
|
STATIC_DIR = Path(__file__).resolve().parent / "static"
|
|
|
|
# 内存 token 表：{token: 过期时间戳}，重启即失效（单用户场景足够）
|
|
_tokens: dict[str, float] = {}
|
|
TOKEN_TTL = 7 * 24 * 3600
|
|
|
|
|
|
def _b64(v: str) -> str:
|
|
return base64.b64encode(v.encode("utf-8")).decode("ascii")
|
|
|
|
|
|
def _unb64(v: str) -> str:
|
|
try:
|
|
return base64.b64decode(v.encode("ascii")).decode("utf-8")
|
|
except Exception:
|
|
return v
|
|
|
|
|
|
def _account_out(row: dict) -> dict:
|
|
out = dict(row)
|
|
out["api_key"] = _unb64(out["api_key"]) # 前端展示/编辑需要明文（本地工具约定）
|
|
return out
|
|
|
|
|
|
def require_auth(authorization: str | None = Header(default=None)) -> None:
|
|
if authorization is None or not authorization.startswith("Bearer "):
|
|
raise HTTPException(status_code=401, detail="未登录")
|
|
token = authorization.removeprefix("Bearer ").strip()
|
|
ts = _tokens.get(token)
|
|
if ts is None:
|
|
raise HTTPException(status_code=401, detail="登录已失效")
|
|
if ts < __import__("time").time():
|
|
_tokens.pop(token, None)
|
|
raise HTTPException(status_code=401, detail="登录已失效")
|
|
|
|
|
|
WINDOW_OFFSETS = {
|
|
"1h": "-1 hour",
|
|
"1d": "-1 day",
|
|
"1w": "-7 days",
|
|
"1mo": "-1 month",
|
|
"1yr": "-1 year",
|
|
}
|
|
|
|
|
|
def _window_time(window: str) -> str | None:
|
|
"""把窗口键转成 SQLite 时间边界（localtime 字符串）。未知窗口返回 None。"""
|
|
offset = WINDOW_OFFSETS.get(window)
|
|
if offset is None:
|
|
return None
|
|
with db.get_conn() as conn:
|
|
row = conn.execute(
|
|
"SELECT datetime('now', 'localtime', ?) AS t", (offset,)
|
|
).fetchone()
|
|
return row["t"]
|
|
|
|
|
|
def _downsample(rows: list[sqlite3.Row], limit: int) -> list[sqlite3.Row]:
|
|
"""按索引均匀抽样到 limit 条，保证首尾尽量覆盖。"""
|
|
if len(rows) <= limit:
|
|
return rows
|
|
step = len(rows) / limit
|
|
picked = [rows[int(i * step)] for i in range(limit)]
|
|
if picked[-1] is not rows[-1]:
|
|
picked[-1] = rows[-1]
|
|
return picked
|
|
|
|
|
|
def create_app(cfg: Config | None = None) -> FastAPI:
|
|
cfg = cfg or load_config()
|
|
db.init_db()
|
|
db.sync_platforms()
|
|
monitor = Monitor(cfg)
|
|
|
|
@asynccontextmanager
|
|
async def lifespan(app: FastAPI):
|
|
monitor.start()
|
|
yield
|
|
monitor.stop()
|
|
|
|
app = FastAPI(title="AI Balance Monitor", lifespan=lifespan)
|
|
|
|
# ---------- 认证 ----------
|
|
|
|
@app.post("/api/login")
|
|
def login(body: LoginRequest):
|
|
import time
|
|
if body.password != cfg.password:
|
|
logger.warning("登录失败：密码错误")
|
|
raise HTTPException(status_code=401, detail="密码错误")
|
|
token = secrets.token_urlsafe(32)
|
|
_tokens[token] = time.time() + TOKEN_TTL
|
|
logger.info("登录成功")
|
|
return {"token": token}
|
|
|
|
@app.post("/api/logout")
|
|
def logout(authorization: str | None = Header(default=None)):
|
|
if authorization and authorization.startswith("Bearer "):
|
|
_tokens.pop(authorization.removeprefix("Bearer ").strip(), None)
|
|
return {"ok": True}
|
|
|
|
# ---------- 平台 ----------
|
|
|
|
@app.get("/api/providers", dependencies=[Depends(require_auth)])
|
|
def list_providers():
|
|
"""代码内置的平台适配器列表（新增平台 = 代码扩展）。"""
|
|
return list_providers_svc()
|
|
|
|
@app.get("/api/platforms", dependencies=[Depends(require_auth)])
|
|
def list_platforms():
|
|
with db.get_conn() as conn:
|
|
rows = conn.execute(
|
|
"""SELECT p.id, p.name, p.currency, p.icon, p.provider_id,
|
|
p.interval_seconds, p.retry_count, p.timeout_seconds,
|
|
p.enabled, p.note, p.created_at, p.updated_at,
|
|
(SELECT COUNT(*) FROM accounts a WHERE a.platform_id = p.id) AS account_count
|
|
FROM platforms p ORDER BY p.id"""
|
|
).fetchall()
|
|
return [dict(r) for r in rows]
|
|
|
|
@app.post("/api/platforms", dependencies=[Depends(require_auth)])
|
|
def create_platform(body: PlatformCreate):
|
|
provider = get_provider(body.provider_id)
|
|
if provider is None:
|
|
raise HTTPException(status_code=400, detail=f"不支持的平台提供方: {body.provider_id}")
|
|
currency = body.currency or provider.currency
|
|
icon = body.icon or provider.icon
|
|
try:
|
|
with db.get_conn() as conn:
|
|
cur = conn.execute(
|
|
"""INSERT INTO platforms (provider_id, name, currency, icon,
|
|
interval_seconds, retry_count, timeout_seconds, enabled, note)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""",
|
|
(
|
|
body.provider_id, body.name, currency, icon,
|
|
body.interval_seconds, body.retry_count, body.timeout_seconds,
|
|
int(body.enabled), body.note,
|
|
),
|
|
)
|
|
pid = cur.lastrowid
|
|
except Exception as exc:
|
|
if "UNIQUE" in str(exc):
|
|
raise HTTPException(status_code=409, detail="平台名称已存在")
|
|
raise HTTPException(status_code=400, detail=str(exc))
|
|
logger.info("创建平台: %s (id=%s, provider=%s)", body.name, pid, body.provider_id)
|
|
return {"id": pid}
|
|
|
|
@app.put("/api/platforms/{pid}", dependencies=[Depends(require_auth)])
|
|
def update_platform(pid: int, body: PlatformUpdate):
|
|
fields = {}
|
|
if body.provider_id is not None:
|
|
provider = get_provider(body.provider_id)
|
|
if provider is None:
|
|
raise HTTPException(status_code=400, detail=f"不支持的平台提供方: {body.provider_id}")
|
|
fields["provider_id"] = body.provider_id
|
|
# 换 provider 时若未显式给货币/图标，则重置为 provider 默认
|
|
if "currency" not in body.model_fields_set:
|
|
fields["currency"] = provider.currency
|
|
if "icon" not in body.model_fields_set:
|
|
fields["icon"] = provider.icon
|
|
if body.name is not None:
|
|
fields["name"] = body.name
|
|
if body.currency is not None:
|
|
fields["currency"] = body.currency
|
|
if body.icon is not None:
|
|
fields["icon"] = body.icon
|
|
if body.interval_seconds is not None:
|
|
fields["interval_seconds"] = body.interval_seconds
|
|
if body.retry_count is not None:
|
|
fields["retry_count"] = body.retry_count
|
|
if body.timeout_seconds is not None:
|
|
fields["timeout_seconds"] = body.timeout_seconds
|
|
if body.enabled is not None:
|
|
fields["enabled"] = int(body.enabled)
|
|
if body.note is not None:
|
|
fields["note"] = body.note
|
|
if not fields:
|
|
raise HTTPException(status_code=400, detail="没有可更新的字段")
|
|
try:
|
|
with db.get_conn() as conn:
|
|
cur = conn.execute(
|
|
"UPDATE platforms SET "
|
|
+ ", ".join(f"{k}=?" for k in fields)
|
|
+ ", updated_at=datetime('now','localtime') WHERE id=?",
|
|
(*fields.values(), pid),
|
|
)
|
|
except Exception as exc:
|
|
if "UNIQUE" in str(exc):
|
|
raise HTTPException(status_code=409, detail="平台名称已存在")
|
|
raise HTTPException(status_code=400, detail=str(exc))
|
|
if cur.rowcount == 0:
|
|
raise HTTPException(status_code=404, detail="平台不存在")
|
|
logger.info("更新平台 id=%s 字段: %s", pid, ", ".join(fields))
|
|
return {"id": pid}
|
|
|
|
@app.delete("/api/platforms/{pid}", dependencies=[Depends(require_auth)])
|
|
def delete_platform(pid: int):
|
|
with db.get_conn() as conn:
|
|
cur = conn.execute("DELETE FROM platforms WHERE id=?", (pid,))
|
|
if cur.rowcount == 0:
|
|
raise HTTPException(status_code=404, detail="平台不存在")
|
|
logger.info("删除平台 id=%s", pid)
|
|
return {"ok": True}
|
|
|
|
# ---------- 账号 ----------
|
|
|
|
@app.get("/api/accounts", dependencies=[Depends(require_auth)])
|
|
def list_accounts(
|
|
platform_id: int | None = Query(default=None),
|
|
include_disabled: bool = Query(default=True),
|
|
):
|
|
sql = """SELECT a.*, p.name AS platform_name, p.currency, p.icon,
|
|
p.interval_seconds AS platform_interval
|
|
FROM accounts a JOIN platforms p ON p.id = a.platform_id"""
|
|
params: list = []
|
|
if platform_id is not None:
|
|
sql += " WHERE a.platform_id = ?"
|
|
params.append(platform_id)
|
|
if not include_disabled:
|
|
sql += " AND a.enabled = 1" if "WHERE" in sql else " WHERE a.enabled = 1"
|
|
sql += " ORDER BY a.id"
|
|
with db.get_conn() as conn:
|
|
rows = conn.execute(sql, params).fetchall()
|
|
return [_account_out(dict(r)) for r in rows]
|
|
|
|
@app.post("/api/accounts", dependencies=[Depends(require_auth)])
|
|
def create_account(body: AccountCreate):
|
|
with db.get_conn() as conn:
|
|
plat = conn.execute("SELECT id FROM platforms WHERE id=?", (body.platform_id,)).fetchone()
|
|
if plat is None:
|
|
raise HTTPException(status_code=404, detail="平台不存在")
|
|
try:
|
|
with db.get_conn() as conn:
|
|
cur = conn.execute(
|
|
"""INSERT INTO accounts (platform_id, name, api_key, threshold, enabled, note)
|
|
VALUES (?, ?, ?, ?, ?, ?)""",
|
|
(body.platform_id, body.name, _b64(body.api_key),
|
|
body.threshold, int(body.enabled), body.note),
|
|
)
|
|
aid = cur.lastrowid
|
|
except Exception as exc:
|
|
raise HTTPException(status_code=400, detail=str(exc))
|
|
logger.info("创建账号: %s (id=%s, 平台id=%s)", body.name, aid, body.platform_id)
|
|
return {"id": aid}
|
|
|
|
@app.put("/api/accounts/{aid}", dependencies=[Depends(require_auth)])
|
|
def update_account(aid: int, body: AccountUpdate):
|
|
fields = {}
|
|
if body.platform_id is not None:
|
|
fields["platform_id"] = body.platform_id
|
|
if body.name is not None:
|
|
fields["name"] = body.name
|
|
if body.api_key is not None:
|
|
fields["api_key"] = _b64(body.api_key)
|
|
if body.threshold is not None:
|
|
fields["threshold"] = body.threshold
|
|
if body.enabled is not None:
|
|
fields["enabled"] = int(body.enabled)
|
|
if body.note is not None:
|
|
fields["note"] = body.note
|
|
if not fields:
|
|
raise HTTPException(status_code=400, detail="没有可更新的字段")
|
|
if "platform_id" in fields:
|
|
with db.get_conn() as conn:
|
|
plat = conn.execute("SELECT id FROM platforms WHERE id=?", (fields["platform_id"],)).fetchone()
|
|
if plat is None:
|
|
raise HTTPException(status_code=404, detail="平台不存在")
|
|
with db.get_conn() as conn:
|
|
cur = conn.execute(
|
|
"UPDATE accounts SET "
|
|
+ ", ".join(f"{k}=?" for k in fields)
|
|
+ ", updated_at=datetime('now','localtime') WHERE id=?",
|
|
(*fields.values(), aid),
|
|
)
|
|
if cur.rowcount == 0:
|
|
raise HTTPException(status_code=404, detail="账号不存在")
|
|
logger.info("更新账号 id=%s 字段: %s", aid, ", ".join(fields))
|
|
return {"id": aid}
|
|
|
|
@app.delete("/api/accounts/{aid}", dependencies=[Depends(require_auth)])
|
|
def delete_account(aid: int):
|
|
with db.get_conn() as conn:
|
|
cur = conn.execute("DELETE FROM accounts WHERE id=?", (aid,))
|
|
if cur.rowcount == 0:
|
|
raise HTTPException(status_code=404, detail="账号不存在")
|
|
logger.info("删除账号 id=%s", aid)
|
|
return {"ok": True}
|
|
|
|
@app.post("/api/accounts/{aid}/check", dependencies=[Depends(require_auth)])
|
|
def check_account_now(aid: int):
|
|
if not monitor.check_now(aid):
|
|
raise HTTPException(status_code=409, detail="账号未启用或正在检查中")
|
|
return {"ok": True}
|
|
|
|
@app.get("/api/accounts/{aid}/history", dependencies=[Depends(require_auth)])
|
|
def account_history(
|
|
aid: int,
|
|
limit: int = Query(default=120, ge=2, le=500),
|
|
window: str = Query(default="all", pattern="^(all|1h|1d|1w|1mo|1yr)$"),
|
|
):
|
|
where = "account_id=?"
|
|
params: list = [aid]
|
|
t0 = _window_time(window)
|
|
if t0 is not None:
|
|
where += " AND checked_at >= ?"
|
|
params.append(t0)
|
|
with db.get_conn() as conn:
|
|
rows = conn.execute(
|
|
f"SELECT balance, checked_at FROM balance_history WHERE {where} ORDER BY id",
|
|
params,
|
|
).fetchall()
|
|
return [
|
|
{"balance": r["balance"], "checked_at": r["checked_at"]}
|
|
for r in _downsample(rows, limit)
|
|
]
|
|
|
|
@app.get("/api/history", dependencies=[Depends(require_auth)])
|
|
def all_history(
|
|
limit: int = Query(default=30, ge=2, le=200),
|
|
window: str = Query(default="all", pattern="^(all|1h|1d|1w|1mo|1yr)$"),
|
|
):
|
|
"""批量返回所有账号的余额历史（窗口内均匀降采样），供监控页趋势图一次拉取。"""
|
|
where = "1=1"
|
|
params: list = []
|
|
t0 = _window_time(window)
|
|
if t0 is not None:
|
|
where += " AND checked_at >= ?"
|
|
params.append(t0)
|
|
with db.get_conn() as conn:
|
|
rows = conn.execute(
|
|
f"""SELECT account_id, balance, checked_at FROM (
|
|
SELECT h.*, ROW_NUMBER() OVER (PARTITION BY account_id ORDER BY id) AS rn,
|
|
COUNT(*) OVER (PARTITION BY account_id) AS total
|
|
FROM balance_history h WHERE {where}
|
|
) WHERE rn = 1 OR rn = total OR rn % MAX(1, total / ?) = 0
|
|
ORDER BY account_id, rn""",
|
|
(*params, max(limit, 1)),
|
|
).fetchall()
|
|
result: dict[int, list] = {}
|
|
for r in rows:
|
|
result.setdefault(r["account_id"], []).append(
|
|
{"balance": r["balance"], "checked_at": r["checked_at"]}
|
|
)
|
|
return result
|
|
|
|
# ---------- 设置 ----------
|
|
|
|
@app.get("/api/settings", dependencies=[Depends(require_auth)])
|
|
def get_settings():
|
|
return {
|
|
"port": cfg.port,
|
|
"global_interval_seconds": cfg.global_interval_seconds,
|
|
"retry_count": cfg.retry_count,
|
|
"timeout_seconds": cfg.timeout_seconds,
|
|
"max_workers": cfg.max_workers,
|
|
"telegram_bot_token": cfg.telegram_bot_token,
|
|
"telegram_chat_id": cfg.telegram_chat_id,
|
|
}
|
|
|
|
@app.put("/api/settings", dependencies=[Depends(require_auth)])
|
|
def update_settings(body: SettingsUpdate):
|
|
changed = False
|
|
for field in ("global_interval_seconds", "retry_count", "timeout_seconds",
|
|
"max_workers", "telegram_bot_token", "telegram_chat_id"):
|
|
value = getattr(body, field)
|
|
if value is not None and getattr(cfg, field) != value:
|
|
setattr(cfg, field, value)
|
|
changed = True
|
|
if changed:
|
|
save_config(cfg)
|
|
logger.info("设置已更新: %s", ", ".join(f for f in ("global_interval_seconds", "retry_count", "timeout_seconds", "max_workers", "telegram_bot_token", "telegram_chat_id") if getattr(body, f) is not None))
|
|
return {"ok": True}
|
|
|
|
@app.post("/api/settings/password", dependencies=[Depends(require_auth)])
|
|
def change_password(body: PasswordChange):
|
|
if body.old_password != cfg.password:
|
|
raise HTTPException(status_code=401, detail="旧密码错误")
|
|
cfg.password = body.new_password
|
|
save_config(cfg)
|
|
logger.info("管理密码已修改")
|
|
return {"ok": True}
|
|
|
|
# ---------- 静态前端 ----------
|
|
|
|
@app.get("/")
|
|
def index():
|
|
return FileResponse(STATIC_DIR / "index.html")
|
|
|
|
app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static")
|
|
|
|
return app
|