feat: 完善芽说英语品牌Logo与图标配置，完成跨平台云端同步服务开发与自动化部署

2026-09-16 09:15:53 +08:00
parent 293b838341
commit eda5dfce62
66 changed files with 3564 additions and 80 deletions
@@ -0,0 +1,18 @@
+FROM python:3.11-slim
+
+WORKDIR /app
+
+# Install dependencies
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+# Copy application source
+COPY main.py .
+
+# Storage volume
+VOLUME ["/data"]
+
+EXPOSE 8080
+
+# Run with uvicorn
+CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8080", "--workers", "2"]
@@ -0,0 +1,40 @@
+# 芽说英语 (SpeakSprout) - 自建跨平台多端同步服务
+
+轻量、高性能、本地优先的跨平台（Android / iOS / macOS）学习进度与艾宾浩斯复习同步服务。
+
+## 🌟 核心特性
+- **极简架构**：基于 Python 3.11 + FastAPI + aiosqlite，单容器内存占用 < 30MB。
+- **无损合并 (CRDT/LWW)**：课程通关自动取并集，艾宾浩斯复习阶段取更高 Checkpoint，绝不发生进度回退。
+- **离线优先 (Offline-First)**：弱网/断网下客户端完全正常学习，恢复网络后自动增量合并同步。
+- **安全认证**：bcrypt 密码加盐哈希 + JWT 鉴权。
+
+## 🚀 快速启动
+
+### 方案 1：Docker Compose 一键启动（推荐）
+```bash
+# 1. 复制目录到您的服务器
+cd sync_server
+
+# 2. 一键启动
+docker compose up -d --build
+
+# 3. 检查服务健康状态
+curl http://127.0.0.1:8080/api/v1/health
+```
+
+### 方案 2：直接 Python 运行
+```bash
+pip install -r requirements.txt
+python main.py
+```
+
+## 🧪 自动化测试
+```bash
+python test_server.py
+```
+
+## 📡 API 快速参考
+- `POST /api/v1/auth/register` : 注册新账号
+- `POST /api/v1/auth/login`    : 登录账号获取 Token
+- `GET  /api/v1/sync/pull`     : 增量拉取云端学习进度
+- `POST /api/v1/sync/push`     : 增量推送本地最新进度
@@ -0,0 +1,19 @@
+version: '3.8'
+
+services:
+  speaksprout-sync:
+    build: .
+    container_name: speaksprout-sync
+    restart: always
+    environment:
+      - DATABASE_PATH=/data/speaksprout_sync.db
+      - JWT_SECRET=speaksprout-production-jwt-secret-key-replace-with-random-hex-2026
+    volumes:
+      - ./data:/data
+    ports:
+      - "127.0.0.1:8080:8080"
+    healthcheck:
+      test: ["CMD", "python3", "-c", "import urllib.request; urllib.request.urlopen('http://127.0.0.1:8080/api/v1/health')"]
+      interval: 30s
+      timeout: 5s
+      retries: 3
@@ -0,0 +1,456 @@
+"""
+芽说英语 (SpeakSprout) - 自建跨平台多端同步服务
+"""
+import os
+import json
+import uuid
+import datetime
+import logging
+from typing import Optional, List, Dict, Any
+
+import jwt
+import bcrypt
+import aiosqlite
+from fastapi import FastAPI, HTTPException, Depends, Header, status
+from fastapi.middleware.cors import CORSMiddleware
+from pydantic import BaseModel, Field
+
+logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
+logger = logging.getLogger("speaksprout_sync")
+
+DATABASE_PATH = os.getenv("DATABASE_PATH", "./data/speaksprout_sync.db")
+JWT_SECRET = os.getenv("JWT_SECRET", "speaksprout-secure-sync-secret-key-2026-v1-production-secure")
+JWT_ALGORITHM = "HS256"
+JWT_EXPIRATION_DAYS = 60
+
+app = FastAPI(
+    title="芽说英语 · SpeakSprout 同步服务",
+    description="支持 Android / iOS / macOS 三端学习进度与艾宾浩斯复习的自建同步服务",
+    version="1.0.0"
+)
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+# ----------------- 数据契约 (DTO) -----------------
+class RegisterRequest(BaseModel):
+    username: str = Field(..., min_length=3, max_length=50)
+    password: str = Field(..., min_length=6)
+    device_name: Optional[str] = "Unknown Device"
+
+class LoginRequest(BaseModel):
+    username: str
+    password: str
+    device_name: Optional[str] = "Unknown Device"
+
+class LessonProgressDTO(BaseModel):
+    active_lesson_id: str
+    completed_lesson_ids: List[str] = []
+    completed_segment_ids: List[str] = []
+    active_step: Optional[str] = "preview"
+    streak_days: Optional[int] = 0
+    updated_at: str
+
+class MasteryItemDTO(BaseModel):
+    item_id: str
+    checkpoint: int
+    status: str
+    due_at: str
+    successful_reviews: int = 0
+    attempts: int = 0
+    payload: Optional[Dict[str, Any]] = {}
+    updated_at: str
+
+class ProfileDTO(BaseModel):
+    onboarding_complete: Optional[bool] = True
+    goal: Optional[str] = "travel"
+    placement: Optional[str] = "A0"
+    daily_minutes: Optional[int] = 20
+    show_chinese_hints: Optional[bool] = True
+    ai_endpoint: Optional[str] = ""
+    ai_model: Optional[str] = ""
+    ai_provider: Optional[str] = ""
+    settings_payload: Optional[Dict[str, Any]] = {}
+    updated_at: str
+
+class SyncPushRequest(BaseModel):
+    client_time: str
+    device_name: Optional[str] = None
+    progress: Optional[LessonProgressDTO] = None
+    mastery_updates: Optional[List[MasteryItemDTO]] = []
+    profile: Optional[ProfileDTO] = None
+
+# ----------------- 数据库连接与表结构 -----------------
+async def get_db():
+    async with aiosqlite.connect(DATABASE_PATH) as db:
+        db.row_factory = aiosqlite.Row
+        yield db
+
+async def init_db():
+    db_dir = os.path.dirname(os.path.abspath(DATABASE_PATH))
+    if db_dir:
+        os.makedirs(db_dir, exist_ok=True)
+    async with aiosqlite.connect(DATABASE_PATH) as db:
+        await db.execute("""
+            CREATE TABLE IF NOT EXISTS users (
+                id TEXT PRIMARY KEY,
+                username TEXT UNIQUE NOT NULL,
+                password_hash TEXT NOT NULL,
+                device_names TEXT DEFAULT '[]',
+                created_at TEXT NOT NULL,
+                updated_at TEXT NOT NULL
+            );
+        """)
+        await db.execute("""
+            CREATE TABLE IF NOT EXISTS user_lesson_progress (
+                user_id TEXT PRIMARY KEY,
+                active_lesson_id TEXT NOT NULL,
+                completed_lesson_ids TEXT NOT NULL,
+                completed_segment_ids TEXT NOT NULL,
+                active_step TEXT NOT NULL DEFAULT 'preview',
+                streak_days INTEGER NOT NULL DEFAULT 0,
+                client_updated_at TEXT NOT NULL,
+                server_updated_at TEXT NOT NULL
+            );
+        """)
+        await db.execute("""
+            CREATE TABLE IF NOT EXISTS user_mastery_records (
+                user_id TEXT NOT NULL,
+                item_id TEXT NOT NULL,
+                checkpoint INTEGER NOT NULL DEFAULT 0,
+                status TEXT NOT NULL,
+                due_at TEXT NOT NULL,
+                successful_reviews INTEGER NOT NULL DEFAULT 0,
+                attempts INTEGER NOT NULL DEFAULT 0,
+                payload TEXT NOT NULL DEFAULT '{}',
+                client_updated_at TEXT NOT NULL,
+                server_updated_at TEXT NOT NULL,
+                PRIMARY KEY(user_id, item_id)
+            );
+        """)
+        await db.execute("""
+            CREATE TABLE IF NOT EXISTS user_profiles (
+                user_id TEXT PRIMARY KEY,
+                onboarding_complete INTEGER NOT NULL DEFAULT 1,
+                goal TEXT DEFAULT 'travel',
+                placement TEXT DEFAULT 'A0',
+                daily_minutes INTEGER DEFAULT 20,
+                show_chinese_hints INTEGER DEFAULT 1,
+                ai_endpoint TEXT DEFAULT '',
+                ai_model TEXT DEFAULT '',
+                ai_provider TEXT DEFAULT '',
+                settings_payload TEXT NOT NULL DEFAULT '{}',
+                client_updated_at TEXT NOT NULL,
+                server_updated_at TEXT NOT NULL
+            );
+        """)
+        await db.commit()
+    logger.info("✅ 数据库表初始化完成: %s", DATABASE_PATH)
+
+@app.on_event("startup")
+async def on_startup():
+    await init_db()
+
+# ----------------- 鉴权逻辑 -----------------
+def create_jwt(user_id: str, username: str) -> str:
+    now = datetime.datetime.now(datetime.timezone.utc)
+    payload = {
+        "sub": user_id,
+        "username": username,
+        "iat": now,
+        "exp": now + datetime.timedelta(days=JWT_EXPIRATION_DAYS)
+    }
+    return jwt.encode(payload, JWT_SECRET, algorithm=JWT_ALGORITHM)
+
+async def get_current_user(authorization: Optional[str] = Header(None)) -> Dict[str, str]:
+    if not authorization or not authorization.startswith("Bearer "):
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="未提供有效的认证凭证")
+    token = authorization.split(" ")[1]
+    try:
+        payload = jwt.decode(token, JWT_SECRET, algorithms=[JWT_ALGORITHM])
+        return {"user_id": payload["sub"], "username": payload.get("username", "")}
+    except jwt.ExpiredSignatureError:
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="登录凭证已过期，请重新登录")
+    except jwt.PyJWTError:
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="无效的认证凭证")
+
+# ----------------- 控制器路由 -----------------
+@app.get("/")
+def root():
+    return {
+        "app": "芽说英语 · SpeakSprout Sync Service",
+        "status": "online",
+        "version": "1.0.0",
+        "server_time": datetime.datetime.now(datetime.timezone.utc).isoformat()
+    }
+
+@app.get("/api/v1/health")
+def health():
+    return {"status": "ok", "timestamp": datetime.datetime.now(datetime.timezone.utc).isoformat()}
+
+@app.post("/api/v1/auth/register")
+async def register(req: RegisterRequest, db: aiosqlite.Connection = Depends(get_db)):
+    clean_username = req.username.strip().lower()
+    cursor = await db.execute("SELECT id FROM users WHERE username = ?", (clean_username,))
+    if await cursor.fetchone():
+        raise HTTPException(status_code=400, detail="该账号已被注册，请直接登录")
+    
+    salt = bcrypt.gensalt()
+    pwd_hash = bcrypt.hashpw(req.password.encode('utf-8'), salt).decode('utf-8')
+    user_id = f"u_{uuid.uuid4().hex[:12]}"
+    now_iso = datetime.datetime.now(datetime.timezone.utc).isoformat()
+    devices = json.dumps([req.device_name] if req.device_name else [])
+
+    await db.execute(
+        "INSERT INTO users (id, username, password_hash, device_names, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?)",
+        (user_id, clean_username, pwd_hash, devices, now_iso, now_iso)
+    )
+    await db.commit()
+    token = create_jwt(user_id, clean_username)
+    logger.info("🎉 新用户注册: %s (%s)", clean_username, user_id)
+    return {
+        "code": 0,
+        "message": "注册成功",
+        "data": {
+            "user_id": user_id,
+            "username": clean_username,
+            "token": token,
+            "expires_in": JWT_EXPIRATION_DAYS * 86400
+        }
+    }
+
+@app.post("/api/v1/auth/login")
+async def login(req: LoginRequest, db: aiosqlite.Connection = Depends(get_db)):
+    clean_username = req.username.strip().lower()
+    cursor = await db.execute("SELECT id, password_hash, device_names FROM users WHERE username = ?", (clean_username,))
+    row = await cursor.fetchone()
+    if not row or not bcrypt.checkpw(req.password.encode('utf-8'), row["password_hash"].encode('utf-8')):
+        raise HTTPException(status_code=400, detail="账号或密码错误")
+
+    user_id = row["id"]
+    try:
+        devices = json.loads(row["device_names"])
+        if req.device_name and req.device_name not in devices:
+            devices.append(req.device_name)
+            await db.execute(
+                "UPDATE users SET device_names = ?, updated_at = ? WHERE id = ?",
+                (json.dumps(devices), datetime.datetime.now(datetime.timezone.utc).isoformat(), user_id)
+            )
+            await db.commit()
+    except Exception:
+        pass
+
+    token = create_jwt(user_id, clean_username)
+    logger.info("🔑 用户登录: %s (%s)", clean_username, user_id)
+    return {
+        "code": 0,
+        "message": "登录成功",
+        "data": {
+            "user_id": user_id,
+            "username": clean_username,
+            "token": token,
+            "expires_in": JWT_EXPIRATION_DAYS * 86400
+        }
+    }
+
+@app.get("/api/v1/sync/pull")
+async def sync_pull(
+    since: Optional[str] = None,
+    auth: Dict[str, str] = Depends(get_current_user),
+    db: aiosqlite.Connection = Depends(get_db)
+):
+    user_id = auth["user_id"]
+    now_iso = datetime.datetime.now(datetime.timezone.utc).isoformat()
+
+    # 1. 课程进度
+    cursor = await db.execute("SELECT * FROM user_lesson_progress WHERE user_id = ?", (user_id,))
+    p_row = await cursor.fetchone()
+    progress_data = None
+    if p_row:
+        progress_data = {
+            "active_lesson_id": p_row["active_lesson_id"],
+            "completed_lesson_ids": json.loads(p_row["completed_lesson_ids"]),
+            "completed_segment_ids": json.loads(p_row["completed_segment_ids"]),
+            "active_step": p_row["active_step"],
+            "streak_days": p_row["streak_days"],
+            "updated_at": p_row["server_updated_at"]
+        }
+
+    # 2. 复习掌握项 (增量)
+    if since:
+        cursor = await db.execute(
+            "SELECT * FROM user_mastery_records WHERE user_id = ? AND server_updated_at > ?",
+            (user_id, since)
+        )
+    else:
+        cursor = await db.execute("SELECT * FROM user_mastery_records WHERE user_id = ?", (user_id,))
+
+    m_rows = await cursor.fetchall()
+    mastery_list = []
+    for r in m_rows:
+        mastery_list.append({
+            "item_id": r["item_id"],
+            "checkpoint": r["checkpoint"],
+            "status": r["status"],
+            "due_at": r["due_at"],
+            "successful_reviews": r["successful_reviews"],
+            "attempts": r["attempts"],
+            "payload": json.loads(r["payload"]),
+            "updated_at": r["server_updated_at"]
+        })
+
+    # 3. 用户画像
+    cursor = await db.execute("SELECT * FROM user_profiles WHERE user_id = ?", (user_id,))
+    prof_row = await cursor.fetchone()
+    profile_data = None
+    if prof_row:
+        profile_data = {
+            "onboarding_complete": bool(prof_row["onboarding_complete"]),
+            "goal": prof_row["goal"],
+            "placement": prof_row["placement"],
+            "daily_minutes": prof_row["daily_minutes"],
+            "show_chinese_hints": bool(prof_row["show_chinese_hints"]),
+            "ai_endpoint": prof_row["ai_endpoint"],
+            "ai_model": prof_row["ai_model"],
+            "ai_provider": prof_row["ai_provider"],
+            "settings_payload": json.loads(prof_row["settings_payload"]),
+            "updated_at": prof_row["server_updated_at"]
+        }
+
+    return {
+        "code": 0,
+        "message": "拉取成功",
+        "data": {
+            "server_time": now_iso,
+            "progress": progress_data,
+            "mastery_updates": mastery_list,
+            "profile": profile_data
+        }
+    }
+
+@app.post("/api/v1/sync/push")
+async def sync_push(
+    req: SyncPushRequest,
+    auth: Dict[str, str] = Depends(get_current_user),
+    db: aiosqlite.Connection = Depends(get_db)
+):
+    user_id = auth["user_id"]
+    server_now = datetime.datetime.now(datetime.timezone.utc).isoformat()
+
+    # 1. 合并课程进度 (Union 并集)
+    if req.progress:
+        p = req.progress
+        cursor = await db.execute("SELECT * FROM user_lesson_progress WHERE user_id = ?", (user_id,))
+        existing = await cursor.fetchone()
+        if existing:
+            local_lessons = set(json.loads(existing["completed_lesson_ids"]))
+            incoming_lessons = set(p.completed_lesson_ids)
+            merged_lessons = list(local_lessons.union(incoming_lessons))
+
+            local_segments = set(json.loads(existing["completed_segment_ids"]))
+            incoming_segments = set(p.completed_segment_ids)
+            merged_segments = list(local_segments.union(incoming_segments))
+
+            active_lesson = p.active_lesson_id if p.updated_at >= existing["client_updated_at"] else existing["active_lesson_id"]
+            streak = max(p.streak_days or 0, existing["streak_days"] or 0)
+
+            await db.execute("""
+                UPDATE user_lesson_progress
+                SET active_lesson_id = ?, completed_lesson_ids = ?, completed_segment_ids = ?,
+                    active_step = ?, streak_days = ?, client_updated_at = ?, server_updated_at = ?
+                WHERE user_id = ?
+            """, (active_lesson, json.dumps(merged_lessons), json.dumps(merged_segments),
+                  p.active_step or "preview", streak, p.updated_at, server_now, user_id))
+        else:
+            await db.execute("""
+                INSERT INTO user_lesson_progress (user_id, active_lesson_id, completed_lesson_ids, completed_segment_ids,
+                                                 active_step, streak_days, client_updated_at, server_updated_at)
+                VALUES (?, ?, ?, ?, ?, ?, ?, ?)
+            """, (user_id, p.active_lesson_id, json.dumps(p.completed_lesson_ids), json.dumps(p.completed_segment_ids),
+                  p.active_step or "preview", p.streak_days or 0, p.updated_at, server_now))
+
+    # 2. 合并复习掌握项 (Max Checkpoint + Latest Timestamp)
+    updated_mastery_count = 0
+    if req.mastery_updates:
+        for m in req.mastery_updates:
+            cursor = await db.execute(
+                "SELECT checkpoint, client_updated_at FROM user_mastery_records WHERE user_id = ? AND item_id = ?",
+                (user_id, m.item_id)
+            )
+            existing_m = await cursor.fetchone()
+            
+            should_update = False
+            if not existing_m:
+                should_update = True
+            else:
+                if m.checkpoint > existing_m["checkpoint"]:
+                    should_update = True
+                elif m.checkpoint == existing_m["checkpoint"] and m.updated_at >= existing_m["client_updated_at"]:
+                    should_update = True
+
+            if should_update:
+                updated_mastery_count += 1
+                await db.execute("""
+                    INSERT INTO user_mastery_records (user_id, item_id, checkpoint, status, due_at,
+                                                     successful_reviews, attempts, payload, client_updated_at, server_updated_at)
+                    VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+                    ON CONFLICT(user_id, item_id) DO UPDATE SET
+                        checkpoint = excluded.checkpoint,
+                        status = excluded.status,
+                        due_at = excluded.due_at,
+                        successful_reviews = excluded.successful_reviews,
+                        attempts = excluded.attempts,
+                        payload = excluded.payload,
+                        client_updated_at = excluded.client_updated_at,
+                        server_updated_at = excluded.server_updated_at
+                """, (user_id, m.item_id, m.checkpoint, m.status, m.due_at,
+                      m.successful_reviews, m.attempts, json.dumps(m.payload or {}), m.updated_at, server_now))
+
+    # 3. 合并用户配置 (LWW)
+    if req.profile:
+        prof = req.profile
+        cursor = await db.execute("SELECT client_updated_at FROM user_profiles WHERE user_id = ?", (user_id,))
+        existing_prof = await cursor.fetchone()
+        
+        if not existing_prof or prof.updated_at >= existing_prof["client_updated_at"]:
+            await db.execute("""
+                INSERT INTO user_profiles (user_id, onboarding_complete, goal, placement, daily_minutes,
+                                          show_chinese_hints, ai_endpoint, ai_model, ai_provider, settings_payload,
+                                          client_updated_at, server_updated_at)
+                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+                ON CONFLICT(user_id) DO UPDATE SET
+                    onboarding_complete = excluded.onboarding_complete,
+                    goal = excluded.goal,
+                    placement = excluded.placement,
+                    daily_minutes = excluded.daily_minutes,
+                    show_chinese_hints = excluded.show_chinese_hints,
+                    ai_endpoint = excluded.ai_endpoint,
+                    ai_model = excluded.ai_model,
+                    ai_provider = excluded.ai_provider,
+                    settings_payload = excluded.settings_payload,
+                    client_updated_at = excluded.client_updated_at,
+                    server_updated_at = excluded.server_updated_at
+            """, (user_id, 1 if prof.onboarding_complete else 0, prof.goal or "travel", prof.placement or "A0",
+                  prof.daily_minutes or 20, 1 if prof.show_chinese_hints else 0, prof.ai_endpoint or "",
+                  prof.ai_model or "", prof.ai_provider or "", json.dumps(prof.settings_payload or {}),
+                  prof.updated_at, server_now))
+
+    await db.commit()
+    logger.info("🔄 数据推送完成 [user=%s]: 词汇=%d", user_id, updated_mastery_count)
+    return {
+        "code": 0,
+        "message": "同步成功",
+        "data": {
+            "server_time": server_now,
+            "synced_mastery_count": updated_mastery_count
+        }
+    }
+
+if __name__ == "__main__":
+    import uvicorn
+    uvicorn.run("main:app", host="0.0.0.0", port=8080, reload=True)
@@ -0,0 +1,39 @@
+# Nginx 反向代理配置示例
+server {
+    listen 80;
+    server_name sync.yourdomain.com;
+    
+    # 强制重定向至 HTTPS
+    return 301 https://$host$request_uri;
+}
+
+server {
+    listen 443 ssl http2;
+    server_name sync.yourdomain.com;
+
+    # SSL 证书路径（请根据实际证书位置配置，如 Certbot Let's Encrypt）
+    ssl_certificate /etc/letsencrypt/live/sync.yourdomain.com/fullchain.pem;
+    ssl_certificate_key /etc/letsencrypt/live/sync.yourdomain.com/privkey.pem;
+
+    ssl_protocols TLSv1.2 TLSv1.3;
+    ssl_ciphers HIGH:!aNULL:!MD5;
+    ssl_prefer_server_ciphers on;
+
+    # 同步接口反向代理
+    location / {
+        proxy_pass http://127.0.0.1:8080;
+        proxy_set_header Host $host;
+        proxy_set_header X-Real-IP $remote_addr;
+        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
+        proxy_set_header X-Forwarded-Proto $scheme;
+
+        # 支持 HTTP/1.1 长连接
+        proxy_http_version 1.1;
+        proxy_set_header Connection "";
+
+        # 超时设置
+        proxy_connect_timeout 30s;
+        proxy_read_timeout 60s;
+        proxy_send_timeout 60s;
+    }
+}
@@ -0,0 +1,8 @@
+fastapi>=0.100.0
+uvicorn[standard]>=0.23.0
+aiosqlite>=0.19.0
+pyjwt>=2.8.0
+bcrypt>=4.0.0
+pydantic>=2.0.0
+python-multipart>=0.0.6
+httpx>=0.24.0
@@ -0,0 +1,141 @@
+"""
+芽说英语 (SpeakSprout) - 同步服务自动化端到端测试
+"""
+import os
+import tempfile
+from httpx import AsyncClient, ASGITransport
+
+# Use temporary database for tests
+temp_db = tempfile.NamedTemporaryFile(suffix=".db", delete=False)
+os.environ["DATABASE_PATH"] = temp_db.name
+os.environ["JWT_SECRET"] = "test-secret-key-1234567890-secure-32b"
+
+from main import app, init_db
+
+async def test_full_sync_flow():
+    await init_db()
+    transport = ASGITransport(app=app)
+    async with AsyncClient(transport=transport, base_url="http://test") as ac:
+        # 1. Health check
+        res = await ac.get("/api/v1/health")
+        assert res.status_code == 200
+        assert res.json()["status"] == "ok"
+
+        # 2. Register
+        reg_res = await ac.post("/api/v1/auth/register", json={
+            "username": "test_user_01",
+            "password": "Password123!",
+            "device_name": "Test iPhone"
+        })
+        assert reg_res.status_code == 200
+        reg_data = reg_res.json()
+        assert reg_data["code"] == 0
+        token = reg_data["data"]["token"]
+        user_id = reg_data["data"]["user_id"]
+        assert token and user_id
+
+        # 3. Duplicate register should fail
+        dup_res = await ac.post("/api/v1/auth/register", json={
+            "username": "test_user_01",
+            "password": "Password123!"
+        })
+        assert dup_res.status_code == 400
+
+        # 4. Login
+        login_res = await ac.post("/api/v1/auth/login", json={
+            "username": "test_user_01",
+            "password": "Password123!",
+            "device_name": "Test Mac"
+        })
+        assert login_res.status_code == 200
+        assert login_res.json()["data"]["token"]
+
+        headers = {"Authorization": f"Bearer {token}"}
+
+        # 5. Push progress from Device 1 (iPhone)
+        push_1 = await ac.post("/api/v1/sync/push", headers=headers, json={
+            "client_time": "2026-09-16T10:00:00Z",
+            "device_name": "Test iPhone",
+            "progress": {
+                "active_lesson_id": "a0-02",
+                "completed_lesson_ids": ["a0-01"],
+                "completed_segment_ids": ["a0-01-a", "a0-01-b"],
+                "active_step": "speaking",
+                "streak_days": 1,
+                "updated_at": "2026-09-16T10:00:00Z"
+            },
+            "mastery_updates": [
+                {
+                    "item_id": "A0-W01",
+                    "checkpoint": 1,
+                    "status": "learning",
+                    "due_at": "2026-09-17T10:00:00Z",
+                    "successful_reviews": 1,
+                    "attempts": 1,
+                    "updated_at": "2026-09-16T10:00:00Z"
+                }
+            ],
+            "profile": {
+                "daily_minutes": 20,
+                "show_chinese_hints": True,
+                "updated_at": "2026-09-16T10:00:00Z"
+            }
+        })
+        assert push_1.status_code == 200
+        assert push_1.json()["data"]["synced_mastery_count"] == 1
+
+        # 6. Push progress from Device 2 (Mac) with non-overlapping lesson and higher checkpoint
+        push_2 = await ac.post("/api/v1/sync/push", headers=headers, json={
+            "client_time": "2026-09-16T12:00:00Z",
+            "device_name": "Test Mac",
+            "progress": {
+                "active_lesson_id": "a0-03",
+                "completed_lesson_ids": ["a0-02"],
+                "completed_segment_ids": ["a0-02-a"],
+                "active_step": "dialogue",
+                "streak_days": 2,
+                "updated_at": "2026-09-16T12:00:00Z"
+            },
+            "mastery_updates": [
+                {
+                    "item_id": "A0-W01",
+                    "checkpoint": 2, # Upgraded checkpoint
+                    "status": "reviewing",
+                    "due_at": "2026-09-19T12:00:00Z",
+                    "successful_reviews": 2,
+                    "attempts": 2,
+                    "updated_at": "2026-09-16T12:00:00Z"
+                },
+                {
+                    "item_id": "A0-W02",
+                    "checkpoint": 1,
+                    "status": "learning",
+                    "due_at": "2026-09-17T12:00:00Z",
+                    "successful_reviews": 1,
+                    "attempts": 1,
+                    "updated_at": "2026-09-16T12:00:00Z"
+                }
+            ]
+        })
+        assert push_2.status_code == 200
+
+        # 7. Pull merged progress
+        pull_res = await ac.get("/api/v1/sync/pull", headers=headers)
+        assert pull_res.status_code == 200
+        p_data = pull_res.json()["data"]
+
+        # Verify Union of completed lessons: ["a0-01", "a0-02"]
+        assert "a0-01" in p_data["progress"]["completed_lesson_ids"]
+        assert "a0-02" in p_data["progress"]["completed_lesson_ids"]
+        assert p_data["progress"]["active_lesson_id"] == "a0-03"
+
+        # Verify Mastery items have merged highest checkpoint
+        mastery_map = {item["item_id"]: item for item in p_data["mastery_updates"]}
+        assert mastery_map["A0-W01"]["checkpoint"] == 2
+        assert mastery_map["A0-W02"]["checkpoint"] == 1
+
+        print("\n✅ All Backend Sync API tests passed with flying colors!")
+
+if __name__ == "__main__":
+    import asyncio
+    asyncio.run(test_full_sync_flow())