# Vitals H5/MySQL/权限增强 实施计划
> **For Claude:** REQUIRED SUB-SKILL: Use superpowers:executing-plans to implement this plan task-by-task.
**Goal:** 将 Vitals 应用改造为 H5 移动端友好、MySQL 数据库、增强用户权限隔离
**Architecture:**
- 数据库层:mysql-connector-python 替换 sqlite3,保持裸 SQL 查询结构
- API 层:添加用户权限过滤,非管理员仅能访问自己数据
- UI 层:响应式 CSS 适配移动端,底部 Tab 导航
**Tech Stack:** Python 3.10+, FastAPI, mysql-connector-python, 原生 CSS 媒体查询
---
## 阶段 1: MySQL 数据库迁移
### Task 1.1: 添加 mysql-connector-python 依赖
**Files:**
- Modify: `pyproject.toml`
**Step 1: 添加依赖**
在 `pyproject.toml` 的 `dependencies` 列表中添加:
```toml
"mysql-connector-python>=8.0.0",
```
**Step 2: 安装依赖**
Run: `pip install -e .`
Expected: Successfully installed mysql-connector-python
**Step 3: Commit**
```bash
git add pyproject.toml
git commit -m "deps: 添加 mysql-connector-python 依赖"
```
---
### Task 1.2: 重构 database.py - 连接管理
**Files:**
- Modify: `src/vitals/core/database.py:1-40`
**Step 1: 添加 MySQL 导入和配置**
将文件开头改为:
```python
"""数据库操作 - 支持 MySQL"""
import json
import os
from contextlib import contextmanager
from datetime import date, time, datetime, timedelta
from pathlib import Path
from typing import Optional
import mysql.connector
from mysql.connector import pooling
from .models import Exercise, Meal, Sleep, Weight, UserConfig, User, Reading, Invite
# 数据库连接池(全局)
_connection_pool = None
def get_mysql_config() -> dict:
"""获取 MySQL 配置"""
return {
"host": os.environ.get("MYSQL_HOST", "localhost"),
"port": int(os.environ.get("MYSQL_PORT", "3306")),
"user": os.environ.get("MYSQL_USER", "vitals"),
"password": os.environ.get("MYSQL_PASSWORD", ""),
"database": os.environ.get("MYSQL_DATABASE", "vitals"),
}
def init_connection_pool():
"""初始化数据库连接池"""
global _connection_pool
if _connection_pool is None:
config = get_mysql_config()
_connection_pool = pooling.MySQLConnectionPool(
pool_name="vitals_pool",
pool_size=5,
pool_reset_session=True,
**config
)
return _connection_pool
@contextmanager
def get_connection():
"""获取数据库连接"""
pool = init_connection_pool()
conn = pool.get_connection()
try:
cursor = conn.cursor(dictionary=True)
yield conn, cursor
conn.commit()
except Exception:
conn.rollback()
raise
finally:
cursor.close()
conn.close()
```
**Step 2: 验证语法**
Run: `python -c "from vitals.core.database import get_connection; print('OK')"`
Expected: 语法正确(连接会失败因为还没有 MySQL 服务器,这是预期的)
**Step 3: Commit**
```bash
git add src/vitals/core/database.py
git commit -m "refactor: database.py 添加 MySQL 连接池管理"
```
---
### Task 1.3: 重构 database.py - 建表语句
**Files:**
- Modify: `src/vitals/core/database.py` - `init_db()` 函数
**Step 1: 改写 init_db 函数**
```python
def init_db():
"""初始化数据库表"""
with get_connection() as (conn, cursor):
# 运动记录表
cursor.execute("""
CREATE TABLE IF NOT EXISTS exercise (
id INT PRIMARY KEY AUTO_INCREMENT,
user_id INT DEFAULT 1,
date DATE NOT NULL,
type VARCHAR(100) NOT NULL,
duration INT NOT NULL,
calories INT DEFAULT 0,
distance FLOAT,
heart_rate_avg INT,
source VARCHAR(50) DEFAULT '手动',
raw_data TEXT,
notes TEXT,
INDEX idx_exercise_date (date)
)
""")
# 饮食记录表
cursor.execute("""
CREATE TABLE IF NOT EXISTS meal (
id INT PRIMARY KEY AUTO_INCREMENT,
user_id INT DEFAULT 1,
date DATE NOT NULL,
meal_type VARCHAR(50) NOT NULL,
description TEXT,
calories INT DEFAULT 0,
protein FLOAT,
carbs FLOAT,
fat FLOAT,
photo_path VARCHAR(500),
food_items TEXT,
INDEX idx_meal_date (date)
)
""")
# 睡眠记录表
cursor.execute("""
CREATE TABLE IF NOT EXISTS sleep (
id INT PRIMARY KEY AUTO_INCREMENT,
user_id INT DEFAULT 1,
date DATE NOT NULL,
bedtime VARCHAR(20),
wake_time VARCHAR(20),
duration FLOAT NOT NULL,
quality INT DEFAULT 3,
deep_sleep_mins INT,
source VARCHAR(50) DEFAULT '手动',
notes TEXT,
INDEX idx_sleep_date (date)
)
""")
# 体重记录表
cursor.execute("""
CREATE TABLE IF NOT EXISTS weight (
id INT PRIMARY KEY AUTO_INCREMENT,
user_id INT DEFAULT 1,
date DATE NOT NULL,
weight_kg FLOAT NOT NULL,
body_fat_pct FLOAT,
muscle_mass FLOAT,
notes TEXT,
INDEX idx_weight_date (date)
)
""")
# 用户配置表
cursor.execute("""
CREATE TABLE IF NOT EXISTS config (
`key` VARCHAR(100) PRIMARY KEY,
value TEXT
)
""")
# 用户表
cursor.execute("""
CREATE TABLE IF NOT EXISTS users (
id INT PRIMARY KEY AUTO_INCREMENT,
name VARCHAR(255) NOT NULL UNIQUE,
created_at DATETIME NOT NULL,
is_active TINYINT DEFAULT 0,
gender VARCHAR(10),
height_cm FLOAT,
weight_kg FLOAT,
age INT,
password_hash VARCHAR(255),
email VARCHAR(255),
is_admin TINYINT DEFAULT 0,
is_disabled TINYINT DEFAULT 0
)
""")
# 邀请码表
cursor.execute("""
CREATE TABLE IF NOT EXISTS invites (
id INT PRIMARY KEY AUTO_INCREMENT,
code VARCHAR(100) NOT NULL UNIQUE,
created_by INT NOT NULL,
used_by INT,
created_at DATETIME NOT NULL,
expires_at DATETIME
)
""")
# 阅读记录表
cursor.execute("""
CREATE TABLE IF NOT EXISTS reading (
id INT PRIMARY KEY AUTO_INCREMENT,
user_id INT DEFAULT 1,
date DATE NOT NULL,
title VARCHAR(500) NOT NULL,
author VARCHAR(255),
cover_url VARCHAR(1000),
duration INT DEFAULT 0,
mood VARCHAR(50),
notes TEXT,
INDEX idx_reading_date (date)
)
""")
```
**Step 2: Commit**
```bash
git add src/vitals/core/database.py
git commit -m "refactor: init_db 改为 MySQL 建表语句"
```
---
### Task 1.4: 重构 database.py - 所有查询占位符
**Files:**
- Modify: `src/vitals/core/database.py` - 所有 CRUD 函数
**Step 1: 批量替换占位符**
将所有 SQL 查询中的 `?` 替换为 `%s`。
主要改动点(示例):
```python
# 改前
cursor.execute("SELECT * FROM users WHERE id = ?", (user_id,))
# 改后
cursor.execute("SELECT * FROM users WHERE id = %s", (user_id,))
```
**Step 2: 修改所有函数的连接获取方式**
将所有函数中的:
```python
with get_connection() as conn:
cursor = conn.cursor()
```
改为:
```python
with get_connection() as (conn, cursor):
```
**Step 3: 修改 row 访问方式**
MySQL 返回 dictionary,直接用 `row["field"]` 访问(与现有代码兼容)。
移除所有 `conn.row_factory = sqlite3.Row` 相关代码。
**Step 4: 修改 lastrowid 获取**
```python
# MySQL 方式
cursor.execute(...)
return cursor.lastrowid
```
**Step 5: Commit**
```bash
git add src/vitals/core/database.py
git commit -m "refactor: 所有 SQL 查询改为 MySQL 语法"
```
---
### Task 1.5: 移除 SQLite 特有代码
**Files:**
- Modify: `src/vitals/core/database.py`
**Step 1: 移除 SQLite 导入和函数**
删除:
- `import sqlite3`
- `get_db_path()` 函数
- `PRAGMA` 相关代码
- `migrate_auth_fields()` 中的 SQLite 特有逻辑
**Step 2: 简化 migrate_auth_fields**
```python
def migrate_auth_fields():
"""迁移:为现有 users 表添加认证字段(MySQL 版本)"""
with get_connection() as (conn, cursor):
# 检查列是否存在,MySQL 方式
cursor.execute("""
SELECT COLUMN_NAME FROM INFORMATION_SCHEMA.COLUMNS
WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'users'
""")
columns = [row["COLUMN_NAME"] for row in cursor.fetchall()]
if "password_hash" not in columns:
cursor.execute("ALTER TABLE users ADD COLUMN password_hash VARCHAR(255)")
if "email" not in columns:
cursor.execute("ALTER TABLE users ADD COLUMN email VARCHAR(255)")
if "is_admin" not in columns:
cursor.execute("ALTER TABLE users ADD COLUMN is_admin TINYINT DEFAULT 0")
if "is_disabled" not in columns:
cursor.execute("ALTER TABLE users ADD COLUMN is_disabled TINYINT DEFAULT 0")
```
**Step 3: Commit**
```bash
git add src/vitals/core/database.py
git commit -m "refactor: 移除 SQLite 特有代码,完成 MySQL 迁移"
```
---
### Task 1.6: 创建数据迁移脚本
**Files:**
- Create: `scripts/migrate_sqlite_to_mysql.py`
**Step 1: 编写迁移脚本**
```python
#!/usr/bin/env python3
"""SQLite 到 MySQL 数据迁移脚本"""
import sqlite3
import os
from pathlib import Path
import mysql.connector
def get_sqlite_path() -> Path:
"""获取 SQLite 数据库路径"""
env_path = os.environ.get("VITALS_DB_PATH")
if env_path:
return Path(env_path)
return Path.home() / ".vitals" / "vitals.db"
def migrate():
# SQLite 连接
sqlite_path = get_sqlite_path()
if not sqlite_path.exists():
print(f"SQLite 数据库不存在: {sqlite_path}")
return
sqlite_conn = sqlite3.connect(sqlite_path)
sqlite_conn.row_factory = sqlite3.Row
sqlite_cursor = sqlite_conn.cursor()
# MySQL 连接
mysql_conn = mysql.connector.connect(
host=os.environ.get("MYSQL_HOST", "localhost"),
port=int(os.environ.get("MYSQL_PORT", "3306")),
user=os.environ.get("MYSQL_USER", "vitals"),
password=os.environ.get("MYSQL_PASSWORD", ""),
database=os.environ.get("MYSQL_DATABASE", "vitals"),
)
mysql_cursor = mysql_conn.cursor()
tables = ["users", "exercise", "meal", "sleep", "weight", "reading", "invites", "config"]
for table in tables:
print(f"迁移表: {table}")
try:
sqlite_cursor.execute(f"SELECT * FROM {table}")
rows = sqlite_cursor.fetchall()
if not rows:
print(f" - 无数据")
continue
columns = [desc[0] for desc in sqlite_cursor.description]
placeholders = ", ".join(["%s"] * len(columns))
column_names = ", ".join([f"`{c}`" for c in columns])
insert_sql = f"INSERT INTO {table} ({column_names}) VALUES ({placeholders})"
for row in rows:
values = tuple(row)
try:
mysql_cursor.execute(insert_sql, values)
except mysql.connector.IntegrityError as e:
print(f" - 跳过重复记录: {e}")
mysql_conn.commit()
print(f" - 迁移 {len(rows)} 条记录")
except sqlite3.OperationalError as e:
print(f" - 表不存在或错误: {e}")
sqlite_conn.close()
mysql_conn.close()
print("迁移完成!")
if __name__ == "__main__":
migrate()
```
**Step 2: Commit**
```bash
git add scripts/migrate_sqlite_to_mysql.py
git commit -m "feat: 添加 SQLite 到 MySQL 数据迁移脚本"
```
---
### Task 1.7: 更新 Docker 配置
**Files:**
- Modify: `docker-compose.yml`
- Modify: `.env.example`
**Step 1: 更新 docker-compose.yml**
添加 MySQL 服务:
```yaml
services:
mysql:
image: mysql:8.0
environment:
MYSQL_ROOT_PASSWORD: ${MYSQL_ROOT_PASSWORD:-rootpassword}
MYSQL_DATABASE: ${MYSQL_DATABASE:-vitals}
MYSQL_USER: ${MYSQL_USER:-vitals}
MYSQL_PASSWORD: ${MYSQL_PASSWORD:-vitalspassword}
volumes:
- mysql_data:/var/lib/mysql
ports:
- "3306:3306"
healthcheck:
test: ["CMD", "mysqladmin", "ping", "-h", "localhost"]
interval: 10s
timeout: 5s
retries: 5
vitals:
# ... 现有配置 ...
depends_on:
mysql:
condition: service_healthy
environment:
MYSQL_HOST: mysql
MYSQL_PORT: 3306
MYSQL_USER: ${MYSQL_USER:-vitals}
MYSQL_PASSWORD: ${MYSQL_PASSWORD:-vitalspassword}
MYSQL_DATABASE: ${MYSQL_DATABASE:-vitals}
volumes:
mysql_data:
```
**Step 2: 更新 .env.example**
```bash
# MySQL 配置
MYSQL_HOST=localhost
MYSQL_PORT=3306
MYSQL_USER=vitals
MYSQL_PASSWORD=your_password_here
MYSQL_DATABASE=vitals
MYSQL_ROOT_PASSWORD=rootpassword
```
**Step 3: Commit**
```bash
git add docker-compose.yml .env.example
git commit -m "config: 添加 MySQL Docker 配置"
```
---
## 阶段 2: 用户权限增强
### Task 2.1: 修改 GET /api/users 接口
**Files:**
- Modify: `src/vitals/web/app.py:1699-1717`
**Step 1: 添加权限过滤**
```python
@app.get("/api/users", response_model=list[UserResponse])
async def get_users(current_user: User = Depends(get_current_user)):
"""获取用户列表(非管理员仅返回自己)"""
if current_user.is_admin:
users = db.get_users()
else:
users = [db.get_user(current_user.id)]
return [
UserResponse(
id=u.id,
name=u.name,
created_at=u.created_at.isoformat(),
is_active=u.is_active,
gender=u.gender,
height_cm=u.height_cm,
weight_kg=u.weight_kg,
age=u.age,
bmi=u.bmi,
bmi_status=u.bmi_status,
)
for u in users if u
]
```
**Step 2: Commit**
```bash
git add src/vitals/web/app.py
git commit -m "feat: GET /api/users 非管理员仅返回自己"
```
---
### Task 2.2: 修改 GET /api/users/{user_id} 接口
**Files:**
- Modify: `src/vitals/web/app.py:1740-1758`
**Step 1: 添加权限检查**
```python
@app.get("/api/users/{user_id}", response_model=UserResponse)
async def get_user(user_id: int, current_user: User = Depends(get_current_user)):
"""获取指定用户(非管理员只能查自己)"""
if not current_user.is_admin and user_id != current_user.id:
raise HTTPException(status_code=403, detail="无权查看其他用户")
user = db.get_user(user_id)
if not user:
raise HTTPException(status_code=404, detail="用户不存在")
return UserResponse(
id=user.id,
name=user.name,
created_at=user.created_at.isoformat(),
is_active=user.is_active,
gender=user.gender,
height_cm=user.height_cm,
weight_kg=user.weight_kg,
age=user.age,
bmi=user.bmi,
bmi_status=user.bmi_status,
)
```
**Step 2: Commit**
```bash
git add src/vitals/web/app.py
git commit -m "feat: GET /api/users/{id} 非管理员只能查自己"
```
---
### Task 2.3: 隐藏非管理员的管理入口
**Files:**
- Modify: `src/vitals/web/app.py` - 导航栏 HTML 生成函数
**Step 1: 找到导航栏生成代码**
搜索 `get_navbar_html` 或类似函数,修改管理链接的显示条件。
**Step 2: 添加 is_admin 参数**
在导航栏生成函数中,仅当 `is_admin=True` 时显示管理链接:
```python
def get_navbar_html(active_page: str = "", is_admin: bool = False) -> str:
# ... 现有代码 ...
admin_link = ""
if is_admin:
admin_link = f'管理'
# 在导航栏 HTML 中使用 admin_link
```
**Step 3: 更新所有调用处**
在所有页面渲染时传入当前用户的 `is_admin` 状态。
**Step 4: Commit**
```bash
git add src/vitals/web/app.py
git commit -m "feat: 非管理员隐藏导航栏管理入口"
```
---
## 阶段 3: H5 移动端适配
### Task 3.1: 添加全局移动端基础样式
**Files:**
- Modify: `src/vitals/web/app.py` - 通用 CSS 函数
**Step 1: 创建移动端基础 CSS**
在 `get_common_styles()` 或类似函数中添加:
```css
/* 移动端基础 */
* {
-webkit-tap-highlight-color: transparent;
touch-action: manipulation;
}
html {
font-size: 16px;
-webkit-text-size-adjust: 100%;
}
/* 响应式断点 */
@media (max-width: 768px) {
html { font-size: 14px; }
.container {
padding: 0 16px;
max-width: 100%;
}
/* 隐藏桌面导航 */
.desktop-nav { display: none; }
/* 显示移动端底部导航 */
.mobile-nav { display: flex; }
/* 内容区域留出底部导航空间 */
.main-content {
padding-bottom: 80px;
}
}
@media (min-width: 769px) {
.mobile-nav { display: none; }
.desktop-nav { display: flex; }
}
/* 触摸目标 */
.touch-target {
min-height: 44px;
min-width: 44px;
display: flex;
align-items: center;
justify-content: center;
}
/* 安全区域 */
.safe-area-bottom {
padding-bottom: env(safe-area-inset-bottom, 0);
}
```
**Step 2: Commit**
```bash
git add src/vitals/web/app.py
git commit -m "style: 添加移动端基础 CSS 样式"
```
---
### Task 3.2: 创建底部 Tab 导航组件
**Files:**
- Modify: `src/vitals/web/app.py`
**Step 1: 创建底部导航 HTML 函数**
```python
def get_mobile_nav_html(active_page: str = "", is_admin: bool = False) -> str:
"""生成移动端底部导航栏"""
def nav_item(href: str, icon: str, label: str, page: str) -> str:
active_class = "active" if active_page == page else ""
return f'''
'''
# SVG 图标路径
icons = {
"home": '