diff --git a/.gitignore b/.gitignore index 2243596..c08019d 100644 --- a/.gitignore +++ b/.gitignore @@ -40,3 +40,4 @@ config.json *.sqlite-* *.db-* *.zip +.backup/ diff --git a/src/mi_fitness_mcp/adapters/mi_fitness_cloud.py b/src/mi_fitness_mcp/adapters/mi_fitness_cloud.py index 7b22a01..5aa5b58 100644 --- a/src/mi_fitness_mcp/adapters/mi_fitness_cloud.py +++ b/src/mi_fitness_mcp/adapters/mi_fitness_cloud.py @@ -2,7 +2,9 @@ import asyncio import base64 import hashlib import json +import logging import os +import random import struct from collections import defaultdict from collections.abc import AsyncIterator @@ -27,6 +29,28 @@ from mi_fitness_mcp.models import ( LOGIN_PREFIX = b"&&&START&&&" KNOWN_REGIONS = ["ru", "cn", "de", "i2", "sg", "us"] +AUTH_ERROR_MARKERS = ( + "authentication failed", + "invalid credential", + "invalid pass token", + "invalid passtoken", + "login required", + "not logged in", + "session expired", + "unauthorized", +) +logger = logging.getLogger(__name__) + + +class MiFitnessAuthenticationError(RuntimeError): + """The Xiaomi cloud session is no longer authenticated.""" + + +def _is_authentication_error(code: Any, message: str) -> bool: + normalized = message.casefold() + return code in {401, 403, -6, -10001} or any( + marker in normalized for marker in AUTH_ERROR_MARKERS + ) def _read_login_payload(text: str) -> dict: @@ -92,21 +116,52 @@ class MiFitnessCloudAdapter(DataAdapter): self._client: httpx.AsyncClient | None = None self._connected = False self._available_types: list[str] = [] + self._connect_lock = asyncio.Lock() + self.last_error: str | None = None + self.last_health_check_at: datetime | None = None + self.max_pages = 200 + self.request_retries = 3 + self.http_timeout = 20.0 async def connect(self) -> bool: - if not self.user_id or not self.pass_token: - return False - - self._client = httpx.AsyncClient(timeout=30.0, follow_redirects=False) + async with self._connect_lock: + if not self.user_id or not self.pass_token: + self.last_error = "Missing Mi Fitness credentials" + self._connected = False + return False + await self._close_client() + self._client = httpx.AsyncClient(timeout=self.http_timeout, follow_redirects=False) + try: + await self._login_with_token(self.user_id, self.pass_token) + # Trust an explicitly configured region. Expensive cross-region discovery + # made MCP startup exceed host initialization timeouts. + if not self.region: + self.region = await self._discover_region("cn") + self._available_types = await self._discover_data_types() + self._connected = True + self.last_error = None + return True + except Exception as exc: + self.last_error = f"{type(exc).__name__}: {exc}" + logger.warning("Mi Fitness connection failed: %s", self.last_error) + self._connected = False + await self._close_client() + return False + async def health_check(self) -> bool: + self.last_health_check_at = datetime.now(UTC) + if not self.is_connected(): + return await self.connect() try: - await self._login_with_token(self.user_id, self.pass_token) - self.region = await self._discover_region(self.region) - self._available_types = await self._discover_data_types() - self._connected = True + # A one-day query is small and verifies authentication and the data API. + today = datetime.now(self._request_timezone()).date().isoformat() + await self._fetch_key("steps", today, today) + self.last_error = None return True - except Exception: - self._connected = False + except Exception as exc: + self.last_error = f"{type(exc).__name__}: {exc}" + # Request handling already marks explicit HTTP authentication failures + # disconnected. Keep a valid session on transient network/API failures. return False def is_connected(self) -> bool: @@ -147,7 +202,7 @@ class MiFitnessCloudAdapter(DataAdapter): raise RuntimeError("client not initialized") last_error: Exception | None = None - for attempt in range(3): + for attempt in range(self.request_retries): try: form = {"data": json.dumps(payload, separators=(",", ":"))} nonce = _gen_nonce() @@ -175,15 +230,38 @@ class MiFitnessCloudAdapter(DataAdapter): plaintext = _rc4_crypt(signed_nonce, base64.b64decode(response.text)) body = json.loads(plaintext) if body.get("code") != 0: - raise RuntimeError(body.get("message", "unknown mi fitness error")) + message = str(body.get("message", "unknown mi fitness error")) + if _is_authentication_error(body.get("code"), message): + raise MiFitnessAuthenticationError(message) + raise RuntimeError(message) return body.get("result", {}) except Exception as exc: last_error = exc - if attempt == 2: + retryable = isinstance(exc, (httpx.TimeoutException, httpx.NetworkError)) + authentication_error = isinstance(exc, MiFitnessAuthenticationError) + if isinstance(exc, httpx.HTTPStatusError): + status = exc.response.status_code + authentication_error = status in (401, 403) + if not authentication_error: + retryable = status == 429 or status >= 500 + if authentication_error: + self._connected = False + if attempt < self.request_retries - 1 and self.user_id and self.pass_token: + try: + await self._login_with_token(self.user_id, self.pass_token) + self._connected = True + retryable = True + except Exception as login_exc: + self.last_error = f"Authentication refresh failed: {login_exc}" + retryable = False + else: + retryable = False + if attempt == self.request_retries - 1 or not retryable: break - await asyncio.sleep(0.5 * (attempt + 1)) - raise RuntimeError(f"Mi Fitness request failed: {last_error}") - + await asyncio.sleep(min(4.0, 0.5 * (2**attempt)) + random.random() * 0.1) + if isinstance(last_error, MiFitnessAuthenticationError): + raise last_error + raise RuntimeError(f"Mi Fitness request failed: {last_error}") from last_error def _request_timezone(self, region: str | None = None) -> timezone: region_name = self.region if region is None else region @@ -212,7 +290,12 @@ class MiFitnessCloudAdapter(DataAdapter): next_key = None items: list[dict] = [] + seen_keys: set[str] = set() + page = 0 while True: + page += 1 + if page > self.max_pages: + raise RuntimeError("Mi Fitness pagination exceeded safety limit") payload = { "start_time": start_time, "end_time": end_time, @@ -225,7 +308,11 @@ class MiFitnessCloudAdapter(DataAdapter): items.extend(result.get("data_list", [])) if not result.get("has_more") or not result.get("next_key"): break - next_key = result.get("next_key") + candidate = str(result.get("next_key")) + if candidate in seen_keys: + raise RuntimeError("Mi Fitness pagination cursor loop detected") + seen_keys.add(candidate) + next_key = candidate return items @@ -288,7 +375,12 @@ class MiFitnessCloudAdapter(DataAdapter): next_key = None items: list[dict] = [] + seen_keys: set[str] = set() + page = 0 while True: + page += 1 + if page > self.max_pages: + raise RuntimeError("Mi Fitness sport pagination exceeded safety limit") payload: dict[str, Any] = { "start_time": start_time, "end_time": end_time, @@ -303,7 +395,11 @@ class MiFitnessCloudAdapter(DataAdapter): items.extend(result.get("sport_records", [])) if not result.get("has_more") or not result.get("next_key"): break - next_key = result.get("next_key") + candidate = str(result.get("next_key")) + if candidate in seen_keys: + raise RuntimeError("Mi Fitness sport pagination cursor loop detected") + seen_keys.add(candidate) + next_key = candidate return items @@ -359,7 +455,10 @@ class MiFitnessCloudAdapter(DataAdapter): daily[date_str]["distance_m"] += float(payload.get("distance", 0)) daily[date_str]["active_kcal"] += float(payload.get("calories", 0)) daily[date_str]["timezone"] = item.get("zone_name") or daily[date_str]["timezone"] - if daily[date_str]["collected_at"] is None or collected_at > daily[date_str]["collected_at"]: + if ( + daily[date_str]["collected_at"] is None + or collected_at > daily[date_str]["collected_at"] + ): daily[date_str]["collected_at"] = collected_at calorie_records = await self._fetch_key("calories", start_date, end_date) @@ -370,7 +469,10 @@ class MiFitnessCloudAdapter(DataAdapter): date_str = collected_at.strftime("%Y-%m-%d") calorie_totals[date_str] += float(payload.get("calories", 0)) daily[date_str]["timezone"] = item.get("zone_name") or daily[date_str]["timezone"] - if daily[date_str]["collected_at"] is None or collected_at > daily[date_str]["collected_at"]: + if ( + daily[date_str]["collected_at"] is None + or collected_at > daily[date_str]["collected_at"] + ): daily[date_str]["collected_at"] = collected_at for date_str, total in calorie_totals.items(): @@ -420,13 +522,10 @@ class MiFitnessCloudAdapter(DataAdapter): start_at = self._timestamp_to_datetime(sleep_start, zone_offset) end_at = self._timestamp_to_datetime(sleep_end, zone_offset) duration_minutes = int( - payload.get("duration") - or max(0, (int(sleep_end) - int(sleep_start)) // 60) + payload.get("duration") or max(0, (int(sleep_end) - int(sleep_start)) // 60) ) awake_minutes = int( - payload.get("awake_duration") - or payload.get("sleep_awake_duration") - or 0 + payload.get("awake_duration") or payload.get("sleep_awake_duration") or 0 ) asleep_minutes = max(0, duration_minutes - awake_minutes) @@ -438,7 +537,9 @@ class MiFitnessCloudAdapter(DataAdapter): minutes = max(0, (seg_end - seg_start) // 60) if minutes: stages.append( - SleepStage(stage=self._sleep_stage_name(segment.get("state")), minutes=minutes) + SleepStage( + stage=self._sleep_stage_name(segment.get("state")), minutes=minutes + ) ) except Exception: continue @@ -500,12 +601,19 @@ class MiFitnessCloudAdapter(DataAdapter): timezone=item.get("zone_name") or "UTC", collected_at=self._record_datetime(item), workout_id=workout_id, - activity_type=str(item.get("category") or item.get("key") or payload.get("sport_type") or "workout"), + activity_type=str( + item.get("category") + or item.get("key") + or payload.get("sport_type") + or "workout" + ), start_at=start_at, end_at=end_at, duration_minutes=duration_minutes, distance_m=self._optional_float(payload.get("distance")), - calories_kcal=self._optional_float(payload.get("calories") or payload.get("total_cal")), + calories_kcal=self._optional_float( + payload.get("calories") or payload.get("total_cal") + ), avg_heart_rate_bpm=self._optional_int(payload.get("avg_hrm")), max_heart_rate_bpm=self._optional_int(payload.get("max_hrm")), avg_pace_sec_per_km=self._optional_float(payload.get("avg_pace")), @@ -581,12 +689,13 @@ class MiFitnessCloudAdapter(DataAdapter): user_id=self.user_id or "unknown", timezone=item.get("zone_name") or "UTC", collected_at=self._record_datetime(item), - timestamp=self._timestamp_to_datetime(timestamp or item.get("time"), int(item.get("zone_offset", 0) or 0)), + timestamp=self._timestamp_to_datetime( + timestamp or item.get("time"), int(item.get("zone_offset", 0) or 0) + ), bpm=int(payload.get("bpm", 0)), sample_type="resting", ) - def _stress_level(self, score: int) -> str: if score < 30: return "low" @@ -618,7 +727,9 @@ class MiFitnessCloudAdapter(DataAdapter): user_id=self.user_id or "unknown", timezone=item.get("zone_name") or "UTC", collected_at=self._record_datetime(item), - timestamp=self._timestamp_to_datetime(timestamp, int(item.get("zone_offset", 0) or 0)), + timestamp=self._timestamp_to_datetime( + timestamp, int(item.get("zone_offset", 0) or 0) + ), spo2_pct=int(float(spo2)), ) @@ -647,7 +758,9 @@ class MiFitnessCloudAdapter(DataAdapter): user_id=self.user_id or "unknown", timezone=item.get("zone_name") or "UTC", collected_at=self._record_datetime(item), - timestamp=self._timestamp_to_datetime(timestamp, int(item.get("zone_offset", 0) or 0)), + timestamp=self._timestamp_to_datetime( + timestamp, int(item.get("zone_offset", 0) or 0) + ), stress_score=score, level=self._stress_level(score), ) diff --git a/src/mi_fitness_mcp/config.py b/src/mi_fitness_mcp/config.py index b24c48f..327babe 100644 --- a/src/mi_fitness_mcp/config.py +++ b/src/mi_fitness_mcp/config.py @@ -25,7 +25,13 @@ class Config(BaseModel): auto_sync_on_start: bool = True stale_after_minutes: int = 60 store_raw_payloads: bool = True - default_lookback_days: int = 30 + default_lookback_days: int = Field(default=30, ge=1, le=3650) + sync_chunk_days: int = Field(default=7, ge=1, le=90) + http_timeout_seconds: float = Field(default=20.0, gt=0, le=120) + request_retries: int = Field(default=3, ge=1, le=10) + health_check_timeout_seconds: float = Field(default=10.0, gt=0, le=60) + sync_type_timeout_seconds: float = Field(default=180.0, gt=0, le=3600) + max_pages: int = Field(default=200, ge=1, le=5000) model_config = ConfigDict(arbitrary_types_allowed=True) diff --git a/src/mi_fitness_mcp/main.py b/src/mi_fitness_mcp/main.py index efa378e..6e92d25 100644 --- a/src/mi_fitness_mcp/main.py +++ b/src/mi_fitness_mcp/main.py @@ -145,9 +145,7 @@ def cmd_sync(args): def main(): - parser = argparse.ArgumentParser( - prog=PROGRAM_NAME, description="小米运动健康数据 MCP Server" - ) + parser = argparse.ArgumentParser(prog=PROGRAM_NAME, description="小米运动健康数据 MCP Server") subparsers = parser.add_subparsers(dest="command", help="可用命令") subparsers.add_parser("serve", help="运行 MCP Server") diff --git a/src/mi_fitness_mcp/models/__init__.py b/src/mi_fitness_mcp/models/__init__.py index 528f6a4..afce6b0 100644 --- a/src/mi_fitness_mcp/models/__init__.py +++ b/src/mi_fitness_mcp/models/__init__.py @@ -102,6 +102,7 @@ class SpO2Sample(BaseEntity): timestamp: datetime = Field(description="Measurement time") spo2_pct: int = Field(ge=0, le=100, description="SpO2 percentage") + class StressSample(BaseEntity): """Stress level measurement.""" @@ -119,7 +120,6 @@ class AbnormalHeartBeatEvent(BaseEntity): duration_seconds: int = Field(ge=0, description="Event duration in seconds") - class UserProfile(BaseModel): """User profile information.""" diff --git a/src/mi_fitness_mcp/server.py b/src/mi_fitness_mcp/server.py index 2bfaa6d..dff4cbb 100644 --- a/src/mi_fitness_mcp/server.py +++ b/src/mi_fitness_mcp/server.py @@ -1,9 +1,10 @@ """MCP server implementation for Mi Fitness.""" +import asyncio import json import logging import uuid -from datetime import datetime +from datetime import UTC, datetime from typing import Any from mcp.server import Server @@ -18,7 +19,6 @@ from mi_fitness_mcp.services.query_service import QueryService from mi_fitness_mcp.services.sync_service import SyncService from mi_fitness_mcp.storage import Database -logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) app = Server("mi-fitness-mcp") @@ -28,6 +28,20 @@ db = None adapter = None sync_service = None query_service = None +sync_tasks: dict[str, dict[str, Any]] = {} +sync_active = False +MAX_SYNC_TASKS = 100 + + +def _prune_sync_tasks() -> None: + completed = [ + sync_id + for sync_id, state in sync_tasks.items() + if state.get("status") not in {"queued", "running"} + ] + excess = max(0, len(sync_tasks) - MAX_SYNC_TASKS + 1) + for sync_id in completed[:excess]: + sync_tasks.pop(sync_id, None) @app.list_tools() @@ -48,9 +62,19 @@ async def list_tools() -> list[Tool]: "start_date": {"type": "string"}, "end_date": {"type": "string"}, "force_full_sync": {"type": "boolean"}, + "background": {"type": "boolean", "default": False}, }, }, ), + Tool( + name="get_sync_status", + description="Get background synchronization status", + inputSchema={ + "type": "object", + "properties": {"sync_id": {"type": "string"}}, + "required": ["sync_id"], + }, + ), Tool( name="get_profile", description="Get user profile information", @@ -218,6 +242,8 @@ async def call_tool(name: str, arguments: dict[str, Any]) -> list[TextContent]: result = await _handle_get_connection_status() elif name == "sync_data": result = await _handle_sync_data(arguments) + elif name == "get_sync_status": + result = _handle_get_sync_status(arguments) elif name == "get_profile": result = await _handle_get_profile() elif name == "get_daily_summary": @@ -255,7 +281,18 @@ async def _handle_get_connection_status() -> dict: mode="not_configured", connected=False, message="Server not configured." ).model_dump() - connected = adapter is not None and adapter.is_connected() + connected = False + if adapter is not None: + if sync_service and sync_service.sync_in_progress: + connected = adapter.is_connected() + else: + try: + connected = await asyncio.wait_for( + adapter.health_check(), timeout=config.health_check_timeout_seconds + ) + except Exception as exc: + logger.warning("Connection health check failed: %s", exc) + connected = False last_sync = None available_types = [] if db: @@ -275,48 +312,164 @@ async def _handle_get_connection_status() -> dict: sync_time = datetime.fromisoformat(state["last_sync_at"]) if last_sync is None or sync_time > last_sync: last_sync = sync_time - return ConnectionStatus( + result = ConnectionStatus( mode=config.mode, connected=connected, last_sync_at=last_sync, - available_data_types=available_types, + available_data_types=(adapter.get_available_data_types() if connected else available_types), + message=getattr(adapter, "last_error", None) if not connected else None, ).model_dump() + result.update( + { + "connection_state": "connected" if connected else "disconnected", + "region": config.region, + "last_health_check_at": getattr(adapter, "last_health_check_at", None), + "last_connection_error": getattr(adapter, "last_error", None), + "sync_in_progress": bool(sync_service and sync_service.sync_in_progress), + } + ) + return result + + +async def _background_sync(sync_id: str, arguments: dict) -> None: + global sync_active + try: + sync_tasks[sync_id].update( + status="running", started_at=datetime.now(UTC).isoformat() + ) + sync_tasks[sync_id] = await _run_sync_data(arguments, sync_id) + except asyncio.CancelledError: + sync_tasks[sync_id] = {"sync_id": sync_id, "status": "cancelled"} + raise + except Exception as exc: + logger.exception("Background synchronization failed") + sync_tasks[sync_id] = { + "sync_id": sync_id, + "status": "error", + "error_code": type(exc).__name__, + "error": str(exc), + } + finally: + sync_active = False async def _handle_sync_data(arguments: dict) -> dict: + global sync_active + # No await between check and assignment: foreground/background reservation is atomic. + if sync_active: + return {"status": "error", "error": "Another synchronization is in progress"} + sync_active = True + + if arguments.get("background"): + try: + _prune_sync_tasks() + sync_id = str(uuid.uuid4()) + sync_tasks[sync_id] = { + "sync_id": sync_id, + "status": "queued", + "created_at": datetime.now(UTC).isoformat(), + } + task = asyncio.create_task( + _background_sync(sync_id, {**arguments, "background": False}) + ) + sync_tasks[sync_id]["task"] = task + return {"status": "accepted", "sync_id": sync_id} + except Exception: + sync_active = False + raise + + try: + return await _run_sync_data(arguments) + finally: + sync_active = False + + +def _handle_get_sync_status(arguments: dict) -> dict: + state = sync_tasks.get(arguments.get("sync_id")) + if state is None: + return {"status": "error", "error": "Unknown sync_id"} + return {key: value for key, value in state.items() if key != "task"} + + +async def _run_sync_data(arguments: dict, sync_id: str | None = None) -> dict: + if not sync_service: return {"status": "error", "error": "Sync service not initialized"} - data_types = arguments.get("data_types") or sync_service.adapter.get_available_data_types() - sync_id = str(uuid.uuid4()) - started_at = datetime.utcnow() - total_added = 0 - total_updated = 0 - total_skipped = 0 - types_synced = [] + if (not adapter or not adapter.is_connected()) and (not adapter or not await adapter.connect()): + return {"status": "error", "error": getattr(adapter, "last_error", "Not connected")} + data_types_arg = arguments.get("data_types") + if data_types_arg == []: + return {"status": "error", "error": "data_types must not be empty"} + data_types = data_types_arg or sync_service.adapter.get_available_data_types() + supported = set(sync_service.adapter.get_available_data_types()) + unknown = sorted(set(data_types) - supported) + if unknown: + return {"status": "error", "error": f"Unsupported data types: {', '.join(unknown)}"} + sync_id = sync_id or str(uuid.uuid4()) + started_at = datetime.now(UTC) + totals = {"added": 0, "updated": 0, "skipped": 0} + details = [] for data_type in data_types: + type_started = datetime.now(UTC) try: - result = await sync_service.sync_data_type( - data_type=data_type, - start_date=arguments.get("start_date"), - end_date=arguments.get("end_date"), - force_full=arguments.get("force_full_sync", False), + result = await asyncio.wait_for( + sync_service.sync_data_type( + data_type=data_type, + start_date=arguments.get("start_date"), + end_date=arguments.get("end_date"), + force_full=arguments.get("force_full_sync", False), + ), + timeout=config.sync_type_timeout_seconds, ) - total_added += result.get("added", 0) - total_updated += result.get("updated", 0) - total_skipped += result.get("skipped", 0) - types_synced.append(data_type) - except Exception as e: - logger.error(f"Failed to sync {data_type}: {e}") - finished_at = datetime.utcnow() + result_status = result.get("status", "ok") + for key in totals: + totals[key] += result.get(key, 0) + details.append( + { + "data_type": data_type, + "status": result_status, + **result, + "duration_seconds": (datetime.now(UTC) - type_started).total_seconds(), + } + ) + except TimeoutError: + details.append( + { + "data_type": data_type, + "status": "error", + "error_code": "timeout", + "error": "Data type synchronization timed out", + } + ) + except Exception as exc: + logger.exception("Failed to sync %s", data_type) + details.append( + { + "data_type": data_type, + "status": "error", + "error_code": type(exc).__name__, + "error": str(exc), + } + ) + succeeded = [item["data_type"] for item in details if item["status"] == "ok"] + has_partial = any(item["status"] == "partial" for item in details) + status = ( + "ok" + if len(succeeded) == len(details) + else "partial" + if succeeded or has_partial + else "error" + ) return { - "status": "ok", + "status": status, "sync_id": sync_id, "started_at": started_at.isoformat(), - "finished_at": finished_at.isoformat(), - "records_added": total_added, - "records_updated": total_updated, - "records_skipped": total_skipped, - "data_types_synced": types_synced, + "finished_at": datetime.now(UTC).isoformat(), + "records_added": totals["added"], + "records_updated": totals["updated"], + "records_skipped": totals["skipped"], + "data_types_synced": succeeded, + "results": details, } @@ -476,14 +629,29 @@ async def main(): adapter = MiFitnessCloudAdapter( user_id=user_id, pass_token=pass_token, region=config.region ) - if await adapter.connect(): - logger.info("Connected to Mi Fitness cloud API") - else: - logger.warning("Failed to connect to Mi Fitness cloud API") + adapter.http_timeout = config.http_timeout_seconds + adapter.request_retries = config.request_retries + adapter.max_pages = config.max_pages + # Do not connect here: MCP stdio must become available even when Xiaomi + # authentication or networking is slow. Status/sync tools connect on demand. if adapter: - sync_service = SyncService(adapter, db) + sync_service = SyncService( + adapter, db, config.default_lookback_days, config.sync_chunk_days + ) query_service = QueryService(db, adapter.get_user_id() or "unknown") else: query_service = QueryService(db, "unknown") - async with stdio_server() as (read_stream, write_stream): - await app.run(read_stream, write_stream, app.create_initialization_options()) + try: + async with stdio_server() as (read_stream, write_stream): + await app.run(read_stream, write_stream, app.create_initialization_options()) + finally: + pending = [ + state.get("task") for state in sync_tasks.values() if state.get("task") is not None + ] + for task in pending: + if not task.done(): + task.cancel() + if pending: + await asyncio.gather(*pending, return_exceptions=True) + if adapter: + await adapter.close() diff --git a/src/mi_fitness_mcp/services/query_service.py b/src/mi_fitness_mcp/services/query_service.py index 249e376..ec9b097 100644 --- a/src/mi_fitness_mcp/services/query_service.py +++ b/src/mi_fitness_mcp/services/query_service.py @@ -301,7 +301,6 @@ class QueryService: return samples[:limit] return samples - def get_spo2_samples( self, start_date: str, diff --git a/src/mi_fitness_mcp/services/sync_service.py b/src/mi_fitness_mcp/services/sync_service.py index 5be98b7..c2fb6da 100644 --- a/src/mi_fitness_mcp/services/sync_service.py +++ b/src/mi_fitness_mcp/services/sync_service.py @@ -3,7 +3,7 @@ import asyncio import logging from collections.abc import AsyncIterator -from datetime import datetime +from datetime import datetime, timedelta from typing import Any from mi_fitness_mcp.adapters.base import DataAdapter @@ -15,7 +15,13 @@ logger = logging.getLogger(__name__) class SyncService: """Service for synchronizing data from adapters to local database.""" - def __init__(self, adapter: DataAdapter, db: Database): + def __init__( + self, + adapter: DataAdapter, + db: Database, + default_lookback_days: int = 30, + chunk_days: int = 7, + ): """Initialize sync service. Args: @@ -24,6 +30,14 @@ class SyncService: """ self.adapter = adapter self.db = db + self.default_lookback_days = default_lookback_days + self.chunk_days = chunk_days + self._sync_lock = asyncio.Lock() + self._sync_active = False + + @property + def sync_in_progress(self) -> bool: + return self._sync_active or self._sync_lock.locked() async def _iterate_records(self, records: Any) -> AsyncIterator[Any]: if hasattr(records, "__aiter__"): @@ -40,6 +54,26 @@ class SyncService: start_date: str | None = None, end_date: str | None = None, force_full: bool = False, + ) -> dict: + """Synchronize one type while preventing overlapping sync operations.""" + # Check and reserve without awaiting, making this atomic for tasks on this loop. + if self._sync_active or self._sync_lock.locked(): + raise RuntimeError("Another synchronization is already in progress") + self._sync_active = True + try: + async with self._sync_lock: + return await self._sync_data_type_unlocked( + data_type, start_date, end_date, force_full + ) + finally: + self._sync_active = False + + async def _sync_data_type_unlocked( + self, + data_type: str, + start_date: str | None = None, + end_date: str | None = None, + force_full: bool = False, ) -> dict: """Synchronize a specific data type. @@ -64,9 +98,77 @@ class SyncService: if not end_date: end_date = datetime.now().strftime("%Y-%m-%d") - if not start_date: - start_date = last_record_ts.strftime("%Y-%m-%d") if last_record_ts else "2020-01-01" + try: + end_dt = datetime.strptime(end_date, "%Y-%m-%d") + start_dt = datetime.strptime(start_date, "%Y-%m-%d") if start_date else None + except ValueError as exc: + raise ValueError("Dates must use YYYY-MM-DD format") from exc + if start_dt is None: + start_dt = ( + last_record_ts.replace(tzinfo=None) + if last_record_ts + else end_dt - timedelta(days=self.default_lookback_days - 1) + ) + start_date = start_dt.strftime("%Y-%m-%d") + if start_dt > end_dt: + raise ValueError("start_date must not be after end_date") + chunk_days = getattr(self, "chunk_days", 7) + totals = {"added": 0, "updated": 0, "skipped": 0} + chunks = [] + cursor = start_dt + while cursor <= end_dt: + chunk_end = min(cursor + timedelta(days=chunk_days - 1), end_dt) + chunk_start_text = cursor.strftime("%Y-%m-%d") + chunk_end_text = chunk_end.strftime("%Y-%m-%d") + try: + result = await self._sync_range(data_type, chunk_start_text, chunk_end_text) + except asyncio.CancelledError: + raise + except Exception as exc: + chunks.append( + { + "start_date": chunk_start_text, + "end_date": chunk_end_text, + "status": "error", + "error_code": type(exc).__name__, + "error": str(exc), + } + ) + return { + "status": "partial" if any(c.get("status") == "ok" for c in chunks) else "error", + "data_type": data_type, + **totals, + "start_date": start_date, + "end_date": end_date, + "chunks": chunks, + "error_code": type(exc).__name__, + "error": str(exc), + } + for key in totals: + totals[key] += result[key] + chunks.append( + { + "start_date": chunk_start_text, + "end_date": chunk_end_text, + "status": "ok", + "added": result["added"], + "updated": result["updated"], + "skipped": result["skipped"], + } + ) + cursor = chunk_end + timedelta(days=1) + + return { + "status": "ok", + "data_type": data_type, + **totals, + "start_date": start_date, + "end_date": end_date, + "chunks": chunks, + } + + async def _sync_range(self, data_type: str, start_date: str, end_date: str) -> dict: added = 0 updated = 0 skipped = 0 diff --git a/src/mi_fitness_mcp/storage/__init__.py b/src/mi_fitness_mcp/storage/__init__.py index 85c2a09..8fc4f11 100644 --- a/src/mi_fitness_mcp/storage/__init__.py +++ b/src/mi_fitness_mcp/storage/__init__.py @@ -581,7 +581,6 @@ class Database: conn.commit() return cursor.rowcount > 0 - def update_sync_state(self, data_type: str, last_record_ts: datetime | None = None) -> None: """Update sync state for a data type.""" with self._get_connection() as conn: @@ -857,7 +856,6 @@ class Database: } ) - for data_type, table_name, date_expr in [ ("spo2", "spo2_samples", "timestamp"), ("stress", "stress_samples", "timestamp"), diff --git a/tests/test_mi_fitness_cloud.py b/tests/test_mi_fitness_cloud.py index 0d9f2e4..23202e7 100644 --- a/tests/test_mi_fitness_cloud.py +++ b/tests/test_mi_fitness_cloud.py @@ -1,6 +1,9 @@ import pytest -from mi_fitness_mcp.adapters.mi_fitness_cloud import MiFitnessCloudAdapter +from mi_fitness_mcp.adapters.mi_fitness_cloud import ( + MiFitnessCloudAdapter, + _is_authentication_error, +) async def _collect(async_iterable): @@ -29,6 +32,7 @@ def test_record_datetime_uses_zone_offset(): dt = adapter._record_datetime({"time": 0, "zone_offset": 10800}) assert dt.isoformat().startswith("1970-01-01T03:00:00") + @pytest.mark.asyncio async def test_iter_daily_activity_aggregates_steps_and_calories(monkeypatch): adapter = MiFitnessCloudAdapter(user_id="u1", pass_token="p1") @@ -63,3 +67,36 @@ async def test_iter_daily_activity_aggregates_steps_and_calories(monkeypatch): assert items[0].steps == 30 assert items[0].distance_m == 24 assert items[0].active_kcal == 12 + + +@pytest.mark.asyncio +async def test_fetch_key_rejects_repeated_pagination_cursor(monkeypatch): + adapter = MiFitnessCloudAdapter(user_id="u1", pass_token="p1", region="cn") + adapter._client = object() + + async def fake_request(base_url, api_path, payload): + return {"data_list": [], "has_more": True, "next_key": "same"} + + monkeypatch.setattr(adapter, "_request", fake_request) + with pytest.raises(RuntimeError, match="cursor loop"): + await adapter._fetch_key("steps", "2026-07-06", "2026-07-12") + + +@pytest.mark.asyncio +async def test_connect_failure_closes_client(monkeypatch): + adapter = MiFitnessCloudAdapter(user_id="u1", pass_token="bad", region="cn") + + async def failed_login(user_id, pass_token): + raise RuntimeError("invalid credentials") + + monkeypatch.setattr(adapter, "_login_with_token", failed_login) + assert await adapter.connect() is False + assert adapter._client is None + assert "invalid credentials" in adapter.last_error + + +def test_authentication_error_detection(): + assert _is_authentication_error(401, "denied") + assert _is_authentication_error(0, "session expired") + assert _is_authentication_error(-10001, "unknown") + assert not _is_authentication_error(500, "temporary server failure") diff --git a/tests/test_server_sync.py b/tests/test_server_sync.py new file mode 100644 index 0000000..35e4cc2 --- /dev/null +++ b/tests/test_server_sync.py @@ -0,0 +1,77 @@ +import asyncio + +import pytest + +from mi_fitness_mcp import server + + +@pytest.mark.asyncio +async def test_background_sync_can_be_polled(monkeypatch): + server.sync_tasks.clear() + + async def fake_run(arguments, sync_id=None): + await asyncio.sleep(0) + return {"status": "ok", "sync_id": sync_id, "records_added": 1} + + monkeypatch.setattr(server, "_run_sync_data", fake_run) + accepted = await server._handle_sync_data({"background": True}) + assert accepted["status"] == "accepted" + await asyncio.sleep(0.01) + state = server._handle_get_sync_status({"sync_id": accepted["sync_id"]}) + assert state["status"] == "ok" + assert state["records_added"] == 1 + assert "task" not in state + + +def test_unknown_background_sync_id(): + assert server._handle_get_sync_status({"sync_id": "missing"})["status"] == "error" + + +@pytest.mark.asyncio +async def test_second_background_sync_is_rejected_while_queued(monkeypatch): + server.sync_tasks.clear() + gate = asyncio.Event() + + async def fake_run(arguments, sync_id=None): + await gate.wait() + return {"status": "ok", "sync_id": sync_id} + + monkeypatch.setattr(server, "_run_sync_data", fake_run) + first = await server._handle_sync_data({"background": True}) + task = server.sync_tasks[first["sync_id"]]["task"] + second = await server._handle_sync_data({"background": True}) + assert first["status"] == "accepted" + assert second["status"] == "error" + gate.set() + await task + + +@pytest.mark.asyncio +async def test_connection_status_skips_health_check_during_sync(monkeypatch): + class Adapter: + last_error = None + last_health_check_at = None + + def is_connected(self): + return True + + def get_available_data_types(self): + return [] + + async def health_check(self): + raise AssertionError("health check must not run while syncing") + + class Service: + sync_in_progress = True + + class Config: + mode = "mi_fitness_cloud" + region = "cn" + health_check_timeout_seconds = 1 + + monkeypatch.setattr(server, "adapter", Adapter()) + monkeypatch.setattr(server, "sync_service", Service()) + monkeypatch.setattr(server, "config", Config()) + monkeypatch.setattr(server, "db", None) + result = await server._handle_get_connection_status() + assert result["connected"] is True diff --git a/tests/test_sync_service.py b/tests/test_sync_service.py new file mode 100644 index 0000000..3d13a7d --- /dev/null +++ b/tests/test_sync_service.py @@ -0,0 +1,130 @@ +import pytest + +from mi_fitness_mcp.services.sync_service import SyncService + + +class FakeAdapter: + def is_connected(self): + return True + + async def iter_daily_activity(self, start_date, end_date): + if False: + yield None + + +class FakeDatabase: + def get_sync_state(self, data_type): + return None + + def update_sync_state(self, data_type, timestamp): + raise AssertionError("empty sync must not advance cursor") + + +@pytest.mark.asyncio +async def test_default_lookback_and_date_validation(): + service = SyncService(FakeAdapter(), FakeDatabase(), default_lookback_days=7) + result = await service.sync_data_type("daily_activity", end_date="2026-07-12") + assert result["start_date"] == "2026-07-06" + assert result["end_date"] == "2026-07-12" + + with pytest.raises(ValueError, match="start_date"): + await service.sync_data_type( + "daily_activity", start_date="2026-07-13", end_date="2026-07-12" + ) + + +@pytest.mark.asyncio +async def test_sync_lock_rejects_overlap(): + service = SyncService(FakeAdapter(), FakeDatabase()) + await service._sync_lock.acquire() + try: + assert service.sync_in_progress is True + with pytest.raises(RuntimeError, match="already in progress"): + await service.sync_data_type( + "daily_activity", start_date="2026-07-12", end_date="2026-07-12" + ) + finally: + service._sync_lock.release() + assert service.sync_in_progress is False + + +@pytest.mark.asyncio +async def test_sync_splits_date_range_into_chunks(monkeypatch): + service = SyncService(FakeAdapter(), FakeDatabase(), chunk_days=3) + calls = [] + + async def fake_range(data_type, start_date, end_date): + calls.append((start_date, end_date)) + return { + "data_type": data_type, + "added": 1, + "updated": 0, + "skipped": 0, + "start_date": start_date, + "end_date": end_date, + } + + monkeypatch.setattr(service, "_sync_range", fake_range) + result = await service.sync_data_type( + "daily_activity", start_date="2026-07-06", end_date="2026-07-12" + ) + assert calls == [ + ("2026-07-06", "2026-07-08"), + ("2026-07-09", "2026-07-11"), + ("2026-07-12", "2026-07-12"), + ] + assert result["added"] == 3 + assert len(result["chunks"]) == 3 + + +@pytest.mark.asyncio +async def test_chunk_failure_returns_partial_statistics(monkeypatch): + service = SyncService(FakeAdapter(), FakeDatabase(), chunk_days=2) + calls = 0 + + async def fake_range(data_type, start_date, end_date): + nonlocal calls + calls += 1 + if calls == 2: + raise RuntimeError("cloud unavailable") + return { + "data_type": data_type, + "added": 2, + "updated": 0, + "skipped": 0, + "start_date": start_date, + "end_date": end_date, + } + + monkeypatch.setattr(service, "_sync_range", fake_range) + result = await service.sync_data_type( + "daily_activity", start_date="2026-07-06", end_date="2026-07-09" + ) + assert result["status"] == "partial" + assert result["added"] == 2 + assert result["chunks"][0]["status"] == "ok" + assert result["chunks"][1]["status"] == "error" + + +@pytest.mark.asyncio +async def test_concurrent_sync_is_rejected_instead_of_queued(monkeypatch): + service = SyncService(FakeAdapter(), FakeDatabase()) + entered = __import__("asyncio").Event() + release = __import__("asyncio").Event() + + async def blocked_range(data_type, start_date, end_date): + entered.set() + await release.wait() + return {"data_type": data_type, "added": 0, "updated": 0, "skipped": 0} + + monkeypatch.setattr(service, "_sync_range", blocked_range) + first = __import__("asyncio").create_task( + service.sync_data_type("daily_activity", "2026-07-12", "2026-07-12") + ) + await entered.wait() + with pytest.raises(RuntimeError, match="already in progress"): + await service.sync_data_type( + "daily_activity", "2026-07-12", "2026-07-12" + ) + release.set() + await first