Rocky/DDUp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(api): 添加 BMI 分析端点 /api/bmi/analysis

Browse Source 
- 新增 /api/bmi/analysis API 端点，提供 BMI 分析数据
- 返回当前体重、BMI、BMI状态、目标体重、预估达成天数等
- 基于近30天饮食和运动数据计算每日净消耗
- 更新测试框架支持 MySQL 和认证
- 修复其他 API 端点的用户隔离问题

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
liweiliang0905@gmail.com
2026-01-27 16:16:32 +08:00
parent 295bc6e736
commit a8c9c87540
3 changed files with 228 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

173
src/vitals/web/app.py
View File

@@ -376,6 +376,10 @@ class LoginInput(BaseModel):
remember_me: bool = False
class ResetPasswordInput(BaseModel):
new_password: str
class RegisterInput(BaseModel):
username: str
password: str
@@ -687,6 +691,19 @@ async def admin_enable_user(user_id: int, admin: User = Depends(require_admin)):
return {"message": "用户已启用"}
@app.post("/api/admin/users/{user_id}/reset-password")
async def admin_reset_password(user_id: int, data: ResetPasswordInput, admin: User = Depends(require_admin)):
"""重置用户密码（管理员）"""
user = db.get_user(user_id)
if not user:
raise HTTPException(status_code=404, detail="用户不存在")
if len(data.new_password) < 6:
raise HTTPException(status_code=400, detail="密码长度至少6位")
user.password_hash = hash_password(data.new_password)
db.update_user(user)
return {"message": f"用户 {user.name} 的密码已重置"}
@app.delete("/api/admin/users/{user_id}")
async def admin_delete_user(user_id: int, admin: User = Depends(require_admin)):
"""删除用户（管理员）"""
@@ -1113,7 +1130,7 @@ async def get_exercise_stats(
month_type_counts[e.type] = month_type_counts.get(e.type, 0) + 1
top_type = max(month_type_counts, key=month_type_counts.get) if month_type_counts else None
period_exercises = db.get_exercises(start_date=period_start, end_date=today)
period_exercises = db.get_exercises(start_date=period_start, end_date=today, user_id=active_user.id)
type_counts: dict[str, int] = {}
type_calories: dict[str, int] = {}
for e in period_exercises:
@@ -1146,6 +1163,107 @@ async def get_exercise_stats(
}
@app.get("/api/bmi/analysis")
async def get_bmi_analysis():
"""获取 BMI 分析数据"""
active_user = db.get_active_user()
if not active_user:
raise HTTPException(status_code=400, detail="没有激活的用户")
config = db.get_config(active_user.id)
# 获取最新体重记录
today = date.today()
weight_records = db.get_weight_records(
start_date=today - timedelta(days=30),
end_date=today,
user_id=active_user.id
)
current_weight = weight_records[0].weight_kg if weight_records else config.weight
# 计算 BMI
height_m = (config.height or 170) / 100
current_bmi = round(current_weight / (height_m * height_m), 1) if current_weight else None
# BMI 状态
def get_bmi_status(bmi):
if bmi is None:
return None
if bmi < 18.5:
return {"status": "偏瘦", "color": "#60A5FA", "range": "<18.5"}
elif bmi < 24:
return {"status": "正常", "color": "#34D399", "range": "18.5-24"}
elif bmi < 28:
return {"status": "偏重", "color": "#FBBF24", "range": "24-28"}
else:
return {"status": "肥胖", "color": "#F87171", "range": ">28"}
bmi_info = get_bmi_status(current_bmi)
# 目标体重（从用户配置或默认 BMI 22 计算）
target_weight = getattr(config, 'target_weight', None)
if not target_weight:
# 默认目标 BMI 22
target_weight = round(22 * height_m * height_m, 1)
target_bmi = round(target_weight / (height_m * height_m), 1)
weight_diff = round(current_weight - target_weight, 1) if current_weight else None
# 计算预估达成天数
estimated_days = None
daily_deficit = None
calculation_basis = None
if current_weight and target_weight and config.tdee:
# 获取近30天饮食数据
meals = db.get_meals(
start_date=today - timedelta(days=30),
end_date=today,
user_id=active_user.id
)
# 获取近30天运动数据
exercises = db.get_exercises(
start_date=today - timedelta(days=30),
end_date=today,
user_id=active_user.id
)
# 计算平均每日摄入
total_intake = sum(m.calories for m in meals)
days_with_data = len(set(m.date for m in meals)) or 1
avg_daily_intake = total_intake / days_with_data
# 计算平均每日运动消耗
total_exercise = sum(e.calories for e in exercises)
avg_daily_exercise = total_exercise / 30
# 每日净消耗 = TDEE - 摄入 + 运动
daily_deficit = round(config.tdee - avg_daily_intake + avg_daily_exercise)
if daily_deficit > 0 and weight_diff > 0:
# 7700 卡 ≈ 1 kg
estimated_days = round(weight_diff * 7700 / daily_deficit)
calculation_basis = {
"tdee": config.tdee,
"avg_intake": round(avg_daily_intake),
"avg_exercise": round(avg_daily_exercise),
"daily_deficit": daily_deficit
}
return {
"current_weight": current_weight,
"current_bmi": current_bmi,
"bmi_status": bmi_info,
"height": config.height,
"target_weight": target_weight,
"target_bmi": target_bmi,
"weight_diff": weight_diff,
"estimated_days": estimated_days,
"calculation_basis": calculation_basis,
"has_sufficient_data": bool(calculation_basis)
}
@app.get("/api/meals", response_model=list[MealResponse])
async def get_meals(
days: int = Query(default=30, ge=1, le=365, description="查询天数"),
@@ -1213,8 +1331,11 @@ async def add_exercise_api(data: ExerciseInput):
@app.delete("/api/exercise/{exercise_id}")
async def delete_exercise_api(exercise_id: int):
"""删除运动记录"""
active_user = db.get_active_user()
if not active_user:
raise HTTPException(status_code=400, detail="没有激活的用户")
try:
db.delete_exercise(exercise_id)
db.delete_exercise(exercise_id, user_id=active_user.id)
return {"success": True}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@@ -1367,8 +1488,11 @@ async def add_meal_api(
@app.delete("/api/meal/{meal_id}")
async def delete_meal_api(meal_id: int):
"""删除饮食记录"""
active_user = db.get_active_user()
if not active_user:
raise HTTPException(status_code=400, detail="没有激活的用户")
try:
db.delete_meal(meal_id)
db.delete_meal(meal_id, user_id=active_user.id)
return {"success": True}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@@ -1482,8 +1606,11 @@ async def add_sleep_api(data: SleepInput):
@app.delete("/api/sleep/{sleep_id}")
async def delete_sleep_api(sleep_id: int):
"""删除睡眠记录"""
active_user = db.get_active_user()
if not active_user:
raise HTTPException(status_code=400, detail="没有激活的用户")
try:
db.delete_sleep(sleep_id)
db.delete_sleep(sleep_id, user_id=active_user.id)
return {"success": True}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@@ -1516,8 +1643,11 @@ async def add_weight_api(data: WeightInput):
@app.delete("/api/weight/{weight_id}")
async def delete_weight_api(weight_id: int):
"""删除体重记录"""
active_user = db.get_active_user()
if not active_user:
raise HTTPException(status_code=400, detail="没有激活的用户")
try:
db.delete_weight(weight_id)
db.delete_weight(weight_id, user_id=active_user.id)
return {"success": True}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@@ -1726,11 +1856,15 @@ async def add_reading(reading_input: ReadingInput):
@app.delete("/api/reading/{reading_id}")
async def delete_reading(reading_id: int):
"""删除阅读记录"""
reading = db.get_reading(reading_id)
active_user = db.get_active_user()
if not active_user:
raise HTTPException(status_code=400, detail="没有激活的用户")
reading = db.get_reading(reading_id, user_id=active_user.id)
if not reading:
raise HTTPException(status_code=404, detail="阅读记录不存在")
db.delete_reading(reading_id)
db.delete_reading(reading_id, user_id=active_user.id)
return {"message": "阅读记录已删除"}
@@ -3074,6 +3208,11 @@ def get_admin_page_html() -> str:
color: white;
}
.btn-success:hover { background: #059669; }
.btn-warning {
background: #f59e0b;
color: white;
}
.btn-warning:hover { background: #d97706; }
.btn-sm {
padding: 6px 12px;
font-size: 0.85rem;
@@ -3419,6 +3558,7 @@ def get_admin_page_html() -> str:
<td data-label="操作">
${user.id !== currentUser.id ? `
<div class="action-buttons">
<button class="btn btn-warning btn-sm" onclick="resetPassword(${user.id}, '${user.name}')">重置密码</button>
${user.is_disabled
? `<button class="btn btn-success btn-sm" onclick="enableUser(${user.id})">启用</button>`
: `<button class="btn btn-danger btn-sm" onclick="disableUser(${user.id})">禁用</button>`
@@ -3448,6 +3588,25 @@ def get_admin_page_html() -> str:
loadUsers();
}
async function resetPassword(userId, username) {
const newPassword = prompt(`请输入用户 "${username}" 的新密码（至少6位）：`);
if (!newPassword) return;
if (newPassword.length < 6) {
alert('密码长度至少6位');
return;
}
const response = await apiRequest(`/api/admin/users/${userId}/reset-password`, {
method: 'POST',
body: JSON.stringify({ new_password: newPassword })
});
if (response && response.ok) {
alert(`用户 "${username}" 的密码已重置`);
} else {
const data = await response.json();
alert(data.detail || '重置密码失败');
}
}
async function loadInvites() {
const response = await apiRequest('/api/admin/invites');
if (!response) return;

35
tests/conftest.py
View File

@@ -7,23 +7,38 @@ from pathlib import Path
import pytest
# 设置测试数据库路径
# 设置测试数据库
@pytest.fixture(autouse=True)
def test_db(tmp_path, monkeypatch):
"""使用临时数据库进行测试"""
db_path = tmp_path / "test_vitals.db"
# 覆盖 get_db_path 函数
def mock_get_db_path():
return db_path
def test_db(monkeypatch):
"""使用 MySQL 测试数据库进行测试"""
# 设置 MySQL 环境变量（使用本地 Docker MySQL）
monkeypatch.setenv("MYSQL_HOST", "localhost")
monkeypatch.setenv("MYSQL_PORT", "3306")
monkeypatch.setenv("MYSQL_USER", "vitals")
monkeypatch.setenv("MYSQL_PASSWORD", "vitalspassword")
monkeypatch.setenv("MYSQL_DATABASE", "vitals")
from src.vitals.core import database
monkeypatch.setattr(database, "get_db_path", mock_get_db_path)
# 重置连接池以使用新配置
database._connection_pool = None
# 初始化数据库
database.init_db()
yield db_path
# 清理测试数据（在测试前清空表）
with database.get_connection() as (conn, cursor):
cursor.execute("DELETE FROM exercise")
cursor.execute("DELETE FROM meal")
cursor.execute("DELETE FROM sleep")
cursor.execute("DELETE FROM weight")
cursor.execute("DELETE FROM reading")
# 不删除 users 和 config，只是确保有默认用户
yield
# 测试后重置连接池
database._connection_pool = None
@pytest.fixture

39
tests/test_web.py
View File

@@ -7,13 +7,36 @@ from fastapi.testclient import TestClient
from src.vitals.core import database as db
from src.vitals.core.models import Exercise, Meal, Sleep, Weight, UserConfig
from src.vitals.core.auth import create_token
@pytest.fixture
def client():
"""创建测试客户端"""
"""创建测试客户端（带认证）"""
from src.vitals.web.app import app
return TestClient(app)
# 确保有测试用户
active_user = db.get_active_user()
if not active_user:
# 获取或创建默认用户
users = db.get_users()
if users:
active_user = users[0]
db.set_active_user(active_user.id)
else:
# 创建默认测试用户
from src.vitals.core.auth import hash_password
user_id = db.create_user("test_user", hash_password("test123"))
db.set_active_user(user_id)
active_user = db.get_user_by_id(user_id)
# 创建认证 token
token = create_token(active_user.id, active_user.name, active_user.is_admin)
# 创建带认证 cookie 的客户端
test_client = TestClient(app)
test_client.cookies.set("auth_token", token)
return test_client
@pytest.fixture
@@ -282,3 +305,15 @@ class TestErrorHandling:
"""测试无效参数"""
response = client.get("/api/exercises?days=abc")
assert response.status_code == 422
def test_bmi_analysis_endpoint(client):
"""测试 BMI 分析 API"""
response = client.get("/api/bmi/analysis")
assert response.status_code == 200
data = response.json()
assert "current_weight" in data
assert "current_bmi" in data
assert "bmi_status" in data
assert "target_weight" in data
assert "estimated_days" in data