import os import aiosqlite from datetime import date, datetime, timedelta from fastapi import FastAPI, HTTPException from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel from contextlib import asynccontextmanager import config _ACTIONS = ["pushups", "squats", "crunches", "mountain", "plank"] _ACTION_LABELS = { "pushups": "俯卧撑", "squats": "深蹲", "crunches": "卷腹", "mountain": "登山跑", "plank": "平板支撑", } async def init_db(): async with aiosqlite.connect(config.DB_PATH) as db: await db.execute(""" CREATE TABLE IF NOT EXISTS workouts ( id INTEGER PRIMARY KEY AUTOINCREMENT, date TEXT NOT NULL UNIQUE, pushups INTEGER NOT NULL DEFAULT 0, squats INTEGER NOT NULL DEFAULT 0, crunches INTEGER NOT NULL DEFAULT 0, mountain INTEGER NOT NULL DEFAULT 0, plank INTEGER NOT NULL DEFAULT 0, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) """) await db.execute(""" CREATE TABLE IF NOT EXISTS weights ( id INTEGER PRIMARY KEY AUTOINCREMENT, date TEXT NOT NULL UNIQUE, weight REAL NOT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) """) await db.commit() @asynccontextmanager async def lifespan(app: FastAPI): await init_db() yield app = FastAPI(title="Workout Five 五件套", lifespan=lifespan) app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) class WorkoutCreate(BaseModel): date: str | None = None pushups: int = 0 squats: int = 0 crunches: int = 0 mountain: int = 0 plank: int = 0 class WeightCreate(BaseModel): date: str | None = None weight: float # ---------- Workouts ---------- @app.get("/api/workouts") async def list_workouts(): async with aiosqlite.connect(config.DB_PATH) as db: db.row_factory = aiosqlite.Row cursor = await db.execute( "SELECT * FROM workouts ORDER BY date DESC" ) return [dict(r) for r in await cursor.fetchall()] @app.get("/api/workouts/today") async def get_today(): async with aiosqlite.connect(config.DB_PATH) as db: db.row_factory = aiosqlite.Row cursor = await db.execute( "SELECT * FROM workouts WHERE date = ?", (date.today().isoformat(),), ) row = await cursor.fetchone() return dict(row) if row else None @app.post("/api/workouts") async def upsert_workout(req: WorkoutCreate): wdate = req.date or date.today().isoformat() async with aiosqlite.connect(config.DB_PATH) as db: db.row_factory = aiosqlite.Row cursor = await db.execute("SELECT id FROM workouts WHERE date = ?", (wdate,)) row = await cursor.fetchone() if row: await db.execute( """UPDATE workouts SET pushups=?, squats=?, crunches=?, mountain=?, plank=? WHERE id=?""", (req.pushups, req.squats, req.crunches, req.mountain, req.plank, row["id"]), ) wid = row["id"] else: cursor = await db.execute( """INSERT INTO workouts (date, pushups, squats, crunches, mountain, plank) VALUES (?, ?, ?, ?, ?, ?)""", (wdate, req.pushups, req.squats, req.crunches, req.mountain, req.plank), ) wid = cursor.lastrowid await db.commit() return {"id": wid, "ok": True} @app.delete("/api/workouts/{wid}") async def delete_workout(wid: int): async with aiosqlite.connect(config.DB_PATH) as db: cursor = await db.execute("DELETE FROM workouts WHERE id = ?", (wid,)) await db.commit() if cursor.rowcount == 0: raise HTTPException(404, "记录不存在") return {"ok": True} # ---------- Weights ---------- @app.get("/api/weights") async def list_weights(): async with aiosqlite.connect(config.DB_PATH) as db: db.row_factory = aiosqlite.Row cursor = await db.execute("SELECT * FROM weights ORDER BY date DESC") return [dict(r) for r in await cursor.fetchall()] @app.post("/api/weights") async def upsert_weight(req: WeightCreate): wdate = req.date or date.today().isoformat() async with aiosqlite.connect(config.DB_PATH) as db: db.row_factory = aiosqlite.Row cursor = await db.execute("SELECT id FROM weights WHERE date = ?", (wdate,)) row = await cursor.fetchone() if row: await db.execute("UPDATE weights SET weight=? WHERE id=?", (req.weight, row["id"])) wid = row["id"] else: cursor = await db.execute( "INSERT INTO weights (date, weight) VALUES (?, ?)", (wdate, req.weight) ) wid = cursor.lastrowid await db.commit() return {"id": wid, "ok": True} @app.delete("/api/weights/{wid}") async def delete_weight(wid: int): async with aiosqlite.connect(config.DB_PATH) as db: cursor = await db.execute("DELETE FROM weights WHERE id = ?", (wid,)) await db.commit() if cursor.rowcount == 0: raise HTTPException(404, "记录不存在") return {"ok": True} # ---------- Stats ---------- @app.get("/api/stats") async def get_stats(): async with aiosqlite.connect(config.DB_PATH) as db: db.row_factory = aiosqlite.Row cursor = await db.execute("SELECT * FROM workouts ORDER BY date ASC") rows = [dict(r) for r in await cursor.fetchall()] wcursor = await db.execute("SELECT * FROM weights ORDER BY date ASC") wrows = [dict(r) for r in await wcursor.fetchall()] # 累计总量 totals = {a: sum(r[a] for r in rows) for a in _ACTIONS} # 连续打卡天数（从今天往回数，或从最近一次往回数） done_dates = {r["date"] for r in rows} streak = 0 d = date.today() # 今天没练就从昨天开始算 if d.isoformat() not in done_dates: d = d - timedelta(days=1) while d.isoformat() in done_dates: streak += 1 d = d - timedelta(days=1) # 本周（周一起）练了多少天 today = date.today() monday = today - timedelta(days=today.weekday()) week_days = sum( 1 for r in rows if monday <= datetime.strptime(r["date"], "%Y-%m-%d").date() <= today ) # 最近一次体重变化 weight_trend = None if len(wrows) >= 2: last = wrows[-1]["weight"] prev = wrows[-2]["weight"] weight_trend = {"latest": last, "prev": prev, "delta": round(last - prev, 1)} return { "totals": totals, "streak": streak, "week_days": week_days, "total_days": len(rows), "weight_trend": weight_trend, "last_weight": wrows[-1]["weight"] if wrows else None, } # ---------- Static ---------- _static_dir = os.path.join(os.path.dirname(__file__), "static") if os.path.isdir(_static_dir): from fastapi.staticfiles import StaticFiles app.mount("/", StaticFiles(directory=_static_dir, html=True), name="static") if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=config.PORT)