diff --git a/src/mi_fitness_mcp/api.py b/src/mi_fitness_mcp/api.py index d2c5b45..7e99656 100644 --- a/src/mi_fitness_mcp/api.py +++ b/src/mi_fitness_mcp/api.py @@ -18,12 +18,15 @@ login_url 确认 → GET /api/auth/qr/poll 轮询到 confirmed 并自动发放 A """ import asyncio +import csv +import io import json import os import secrets import threading import time import uuid +from concurrent.futures import ThreadPoolExecutor from contextlib import asynccontextmanager from dataclasses import dataclass from datetime import UTC, datetime @@ -32,7 +35,7 @@ from typing import Annotated, Any import httpx from fastapi import Depends, FastAPI, Header, HTTPException, Query, Request from fastapi.middleware.cors import CORSMiddleware -from fastapi.responses import Response +from fastapi.responses import Response, StreamingResponse from pydantic import BaseModel from mi_fitness_mcp.adapters.mi_fitness_cloud import MiFitnessCloudAdapter @@ -48,7 +51,9 @@ from mi_fitness_mcp.security import ( ) from mi_fitness_mcp.services.query_service import QueryService from mi_fitness_mcp.services.sync_service import SyncService +from mi_fitness_mcp.services.view_query_service import ViewQueryService from mi_fitness_mcp.storage import Database +from mi_fitness_mcp.storage.view_queries import ViewSelection DATA_TYPES = [ "daily_activity", @@ -857,6 +862,164 @@ def get_coverage( return {"status": "ok", "count": len(data), "data": data} +# ---------------------------------------------------------------- 健康视图只读查询 + + +def _view_selection( + kind: str, + start_date: str, + end_date: str, + sample_type: str | None = None, + include_naps: bool = True, + pressure_level: str | None = None, + latest_only: bool = False, + limit: Annotated[int | None, Query(ge=1, le=9223372036854775807)] = None, +) -> ViewSelection: + try: + return ViewSelection(kind, start_date, end_date, sample_type or None, + include_naps, pressure_level or None, latest_only, limit) + except ValueError as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc + + +def _view_service(request: Request) -> ViewQueryService: + context = _resolve_context(request) + return ViewQueryService(request.app.state.db, context.user_id) + + +@app.get("/api/views/{kind}") +def get_view( + request: Request, + selection: Annotated[ViewSelection, Depends(_view_selection)], + page: Annotated[int, Query(ge=1, le=1000000)] = 1, +) -> dict: + return _view_service(request).get_view(selection, page) + + +@app.get("/api/views/{kind}/records") +def get_view_records( + request: Request, + selection: Annotated[ViewSelection, Depends(_view_selection)], + page: Annotated[int, Query(ge=1, le=1000000)] = 1, +) -> dict: + return _view_service(request).get_page(selection, page) + + +@app.get("/api/overview") +def get_overview(request: Request, start_date: str, end_date: str) -> dict: + # 复用相同日期约束;总览始终读取完整范围。 + selection = _view_selection("activity", start_date, end_date) + return {"status": "ok", "range": {"start": start_date, "end": end_date}, + "data": _view_service(request).get_overview(selection.start_date, selection.end_date)} + + +# 仅导出既有规范化业务字段;稳定分页 id 以及账号、设备、鉴权字段不进入 CSV。 +VIEW_CSV_COLUMNS = { + "activity": [("date", "日期"), ("steps", "步数(步)"), ("distance_m", "距离(米)"), + ("active_kcal", "活动热量(kcal)"), ("total_kcal", "总热量(kcal)"), + ("floors", "楼层数"), ("active_minutes", "活动时长(分钟)")], + "heartrate": [("timestamp", "记录时间"), ("bpm", "心率(bpm)"), ("sample_type", "采样类型")], + "sleep": [("sleep_id", "睡眠记录标识"), ("start_at", "入睡时间"), ("end_at", "醒来时间"), + ("duration_minutes", "源时长(分钟)"), ("time_asleep_minutes", "源实睡时长(分钟)"), + ("time_awake_minutes", "清醒(分钟)"), ("sleep_score", "评分"), ("is_nap", "小睡"), + ("stages", "睡眠阶段(源字段)"), ("_asleep", "实睡时长(分钟)"), + ("_deep", "深睡(分钟)"), ("_light", "浅睡(分钟)"), ("_rem", "REM(分钟)")], + "workouts": [("workout_id", "运动记录标识"), ("activity_type", "运动类型"), + ("start_at", "开始时间"), ("end_at", "结束时间"), ("duration_minutes", "时长(分钟)"), + ("distance_m", "距离(米)"), ("calories_kcal", "热量(kcal)"), + ("avg_heart_rate_bpm", "平均心率(bpm)"), ("max_heart_rate_bpm", "最高心率(bpm)"), + ("avg_pace_sec_per_km", "平均配速(秒/公里)"), ("max_pace_sec_per_km", "最快配速(秒/公里)"), + ("total_steps", "总步数")], + "body": [("timestamp", "测量时间"), ("weight_kg", "体重(kg)"), ("bmi", "BMI"), + ("body_fat_pct", "体脂率(%)"), ("muscle_mass_kg", "肌肉量(kg,源字段)"), + ("water_pct", "水分(%)"), ("bone_mass_kg", "骨量(kg)"), + ("visceral_fat_score", "内脏脂肪等级"), ("basal_metabolism_kcal", "基础代谢(kcal/日)"), + ("metabolic_age", "代谢年龄(岁)")], + "spo2": [("timestamp", "记录时间"), ("spo2_pct", "血氧(%)")], + "stress": [("timestamp", "记录时间"), ("stress_score", "压力指数"), + ("level", "源等级(原值)"), ("_pressureBand", "压力等级(当前分界)")], + "abnormal": [("event_id", "事件记录标识"), ("start_at", "开始时间"), + ("end_at", "结束时间"), ("duration_seconds", "持续时长(秒)")], +} + + +def _pressure_label(value) -> str: + if not isinstance(value, (int, float)) or value < 1 or value > 100: + return "未分类" + return "放松" if value <= 25 else "轻度" if value <= 50 else "中度" if value <= 80 else "重度" + + +def _csv_cell(value): + if isinstance(value, (list, dict)): + return json.dumps(value, ensure_ascii=False) + if isinstance(value, str) and value.lstrip().startswith(("=", "+", "-", "@")): + return "'" + value + return value + + +def _view_csv_chunks(service: ViewQueryService, selection: ViewSelection): + columns = VIEW_CSV_COLUMNS[selection.kind] + buffer = io.StringIO(newline="") + writer = csv.writer(buffer) + writer.writerow([label for _, label in columns]) + yield ("\ufeff" + buffer.getvalue()).encode("utf-8") + records = service.iter_export(selection) + try: + for record in records: + buffer.seek(0) + buffer.truncate(0) + writer.writerow([_csv_cell(_pressure_label(record.get("stress_score")) + if field == "_pressureBand" else record.get(field)) for field, _ in columns]) + yield buffer.getvalue().encode("utf-8") + finally: + records.close() + + +async def _view_csv_stream(service: ViewQueryService, selection: ViewSelection): + # sqlite3 游标有线程亲和性:创建、读取、关闭均由本次导出的单一线程执行。 + executor = ThreadPoolExecutor(max_workers=1, thread_name_prefix="health-csv") + loop = asyncio.get_running_loop() + state = {} + + def advance(): + if "iterator" not in state: + state["iterator"] = _view_csv_chunks(service, selection) + return next(state["iterator"], None) + + def close(): + if "iterator" in state: + state["iterator"].close() + + try: + while (chunk := await loop.run_in_executor(executor, advance)) is not None: + yield chunk + finally: + # 先排入关闭任务;即使当前请求被取消,任务也不能被 cancel_futures 丢弃。 + closing = executor.submit(close) + executor.shutdown(wait=False) + await asyncio.shield(asyncio.wrap_future(closing)) + + +class _ViewCSVResponse(StreamingResponse): + async def stream_response(self, send): + try: + await super().stream_response(send) + finally: + # 发送失败或断开时 async for 不保证及时 aclose,显式释放数据库迭代器。 + await self.body_iterator.aclose() + + +@app.get("/api/views/{kind}/export") +def get_view_export( + request: Request, + selection: Annotated[ViewSelection, Depends(_view_selection)], +) -> StreamingResponse: + return _ViewCSVResponse(_view_csv_stream(_view_service(request), selection), + media_type="text/csv; charset=utf-8", + headers={"Cache-Control": "no-store", "Content-Disposition": + f'attachment; filename="mi_fitness_{selection.kind}_{selection.start_date}_{selection.end_date}.csv"'}) + + def _to_csv(rows: list[dict[str, Any]]) -> str: """把查询结果扁平化为 CSV(嵌套结构转 JSON 字符串,加 BOM 便于 Excel 打开中文)。""" import csv diff --git a/src/mi_fitness_mcp/web.py b/src/mi_fitness_mcp/web.py index ea8e318..c68051f 100644 --- a/src/mi_fitness_mcp/web.py +++ b/src/mi_fitness_mcp/web.py @@ -170,6 +170,27 @@ def create_app( started = time.perf_counter() try: + parts = clean_subpath.split("/") + if request.method == "GET" and len(parts) == 4 and parts[:2] == ["api", "views"] and parts[3] == "export": + upstream_request = http_client.build_request("GET", target_url, headers=forward_headers) + upstream_resp = http_client.send(upstream_request, stream=True) + elapsed_ms = round((time.perf_counter() - started) * 1000, 2) + # 导出只保留下载必需的响应头,不能把服务端鉴权头带回浏览器。 + response_headers = [(key, value) for key, value in upstream_resp.headers.items() + if key.lower() in {"content-type", "content-disposition"}] + response_headers.append(("X-Upstream-Time-Ms", str(int(elapsed_ms)))) + + def chunks(): + try: + yield from upstream_resp.iter_bytes() + finally: + upstream_resp.close() + + response = Response(chunks(), status=upstream_resp.status_code, headers=response_headers) + # 也覆盖尚未开始消费就关闭响应的路径。 + response.call_on_close(upstream_resp.close) + return response + upstream_resp = http_client.request( method=request.method, url=target_url, diff --git a/tests/test_view_api.py b/tests/test_view_api.py new file mode 100644 index 0000000..00ab002 --- /dev/null +++ b/tests/test_view_api.py @@ -0,0 +1,74 @@ +import pytest +from fastapi.testclient import TestClient + +from mi_fitness_mcp import api +from mi_fitness_mcp.services.query_service import QueryService +from mi_fitness_mcp.storage import Database +from test_view_queries import insert + + +@pytest.fixture +def api_client(tmp_path, monkeypatch): + db = Database(tmp_path / "synthetic.db") + monkeypatch.setenv("MI_FITNESS_API_KEY", "synthetic-api-key") + context = api.UserContext("owner", "synthetic-unused", "cn", query_service=QueryService(db, "owner")) + monkeypatch.setattr(api.app.state, "db", db, raising=False) + monkeypatch.setattr(api.app.state, "default_context", context, raising=False) + client = TestClient(api.app, base_url="http://localhost", headers={"X-API-Key": "synthetic-api-key"}) + yield client, db + client.close() + + +def test_view_auth_validation_and_legacy_compatibility(api_client): + client, db = api_client + insert(db, "heart_rate_samples", id="sample", timestamp="2026-10-15T02:00:00Z", bpm=72, sample_type="passive") + query = "?start_date=2026-10-15&end_date=2026-10-15" + for path in ["/api/views/heartrate", "/api/views/heartrate/records", "/api/views/heartrate/export", "/api/overview"]: + request = client.build_request("GET", path + query) + request.headers.pop("X-API-Key") + assert client.send(request).status_code == 401 + result = client.get("/api/views/heartrate" + query) + assert result.status_code == 200 + assert result.json()["statistics"]["primary_mean"] == 72 + assert result.json()["page"]["size"] == 10 + assert result.json()["range"] == {"start": "2026-10-15", "end": "2026-10-15"} + old = client.get("/api/heart-rate" + query) + assert old.status_code == 200 + assert old.json() == {"status": "ok", "count": 1, "data": [dict(timestamp="2026-10-15T02:00:00Z", bpm=72, sample_type="passive")]} + page = client.get("/api/views/heartrate/records" + query).json() + assert set(page) == {"status", "kind", "range", "page"} + overview = client.get("/api/overview" + query).json() + assert overview["status"] == "ok" + assert overview["range"] == {"start": "2026-10-15", "end": "2026-10-15"} + assert overview["data"]["resting_mean_bpm"] is None + assert set(overview["data"]) == {"steps", "active_kcal", "distance_m", "resting_mean_bpm", "mean_daily_asleep_minutes", "mean_spo2_pct", "mean_stress", "activity_days", "sleep_days", "resting_count", "spo2_count"} + + +@pytest.mark.parametrize("changes", [dict(kind="unknown"), dict(start_date="2026-02-30"), dict(start_date="2026-10-16"), + dict(start_date="2000-01-01"), dict(page=0), dict(page=1000001), dict(limit=0), dict(limit=-1), + dict(pressure_level="invalid"), dict(page="1.5"), dict(include_naps="invalid")]) +def test_invalid_view_query_rejected(api_client, changes): + client, _ = api_client + values = dict(kind="stress", start_date="2026-10-15", end_date="2026-10-15") | changes + kind = values.pop("kind") + response = client.get(f"/api/views/{kind}", params=values) + assert response.status_code in (400, 422) + assert "detail" in response.json() + + +def test_blank_filters_and_empty_values_are_supported(api_client): + client, _ = api_client + response = client.get("/api/views/stress", params=dict(start_date="2026-10-15", end_date="2026-10-15", sample_type="", pressure_level="")) + assert response.status_code == 200 + assert response.json()["statistics"]["primary_mean"] is None + assert response.json()["page"]["rows"] == [] + + +def test_records_endpoint_does_not_reaggregate(api_client, monkeypatch): + client, _ = api_client + from mi_fitness_mcp.storage.view_queries import ViewRepository + def forbidden(*args, **kwargs): + raise AssertionError("翻页不能重新聚合") + monkeypatch.setattr(ViewRepository, "statistics", forbidden) + monkeypatch.setattr(ViewRepository, "points", forbidden) + assert client.get("/api/views/stress/records", params=dict(start_date="2026-10-15", end_date="2026-10-15", page=2)).status_code == 200 diff --git a/tests/test_view_export.py b/tests/test_view_export.py new file mode 100644 index 0000000..af61253 --- /dev/null +++ b/tests/test_view_export.py @@ -0,0 +1,150 @@ +import asyncio +import csv +import io +import threading +from contextlib import contextmanager + +import pytest +from starlette.requests import Request + +from mi_fitness_mcp import api +from mi_fitness_mcp.storage.view_queries import ViewRepository, ViewSelection +from test_view_api import api_client as api_client +from test_view_queries import insert + + +def test_export_matches_filtered_set_and_escapes_formula(api_client): + client, db = api_client + for n in range(31): + insert(db, "stress_samples", id=str(n), timestamp=f"2026-10-15T00:{n:02}:00Z", stress_score=26 + n if n < 25 else 99, level="=1+1") + params = dict(start_date="2026-10-15", end_date="2026-10-15", pressure_level="mild", limit=11, page=2) + response = client.get("/api/views/stress/export", params=params) + assert response.status_code == 200 + assert response.content.startswith(b"\xef\xbb\xbf") + assert response.headers["content-type"].startswith("text/csv") + assert "attachment" in response.headers["content-disposition"] + assert "content-length" not in response.headers + records = list(csv.DictReader(io.StringIO(response.content.decode("utf-8-sig")))) + assert len(records) == 11 + assert [int(row["压力指数"]) for row in records] == list(range(50, 39, -1)) + assert all(row["源等级(原值)"] == "'=1+1" for row in records) + assert all(row["压力等级(当前分界)"] == "轻度" for row in records) + assert not set(records[0]) & {"id", "user_id", "device_id", "pass_token", "api_key"} + assert response.headers["cache-control"] == "no-store" + + +@pytest.mark.parametrize("text", ["=1+1", "+SUM(A1)", "-1+1", "@SUM(A1)", "\t =1+1", "\n=1+1"]) +def test_formula_prefixes_in_source_text_are_safe(api_client, text): + client, db = api_client + insert(db, "heart_rate_samples", id="sample", timestamp="2026-10-15T00:00:00Z", bpm=72, sample_type=text) + response = client.get("/api/views/heartrate/export", params=dict(start_date="2026-10-15", end_date="2026-10-15")) + rows = list(csv.DictReader(io.StringIO(response.content.decode("utf-8-sig")))) + assert rows[0]["采样类型"] == "'" + text + assert rows[0]["心率(bpm)"] == "72" + + +def test_sleep_export_keeps_source_and_normalized_fields(api_client): + client, db = api_client + insert(db, "sleep_sessions", id="sample", sleep_id="synthetic-sleep", start_at="2026-10-14T23:00:00+08:00", end_at="2026-10-15T07:15:00+08:00", duration_minutes=495, time_asleep_minutes=480, time_awake_minutes=15, stages='[{"stage":"deep","minutes":120}]') + response = client.get("/api/views/sleep/export", params=dict(start_date="2026-10-15", end_date="2026-10-15")) + rows = list(csv.DictReader(io.StringIO(response.content.decode("utf-8-sig")))) + assert rows[0]["实睡时长(分钟)"] == "480" + assert rows[0]["源时长(分钟)"] == "495" + assert rows[0]["深睡(分钟)"] == "120" + assert '"deep"' in rows[0]["睡眠阶段(源字段)"] + + +@pytest.mark.asyncio +async def test_stream_disconnect_closes_cursor_and_proxy_response(api_client, monkeypatch): + _, db = api_client + insert(db, "stress_samples", id="sample", timestamp="2026-10-15T00:00:00Z", stress_score=26, level="source") + original = ViewRepository.snapshot + opened, closed, threads = [], [], [] + @contextmanager + def observed(repository): + with original(repository) as connection: + threads.append(threading.get_ident()) + opened.append(connection) + try: + yield connection + finally: + threads.append(threading.get_ident()) + closed.append(connection) + monkeypatch.setattr(ViewRepository, "snapshot", observed) + request = Request({"type": "http", "app": api.app, "headers": [(b"x-api-key", b"synthetic-api-key")]}) + response = api.get_view_export(request, ViewSelection("stress", "2026-10-15", "2026-10-15")) + assert not opened + await response.body_iterator.__anext__() + await response.body_iterator.__anext__() + assert len(opened) == 1 + assert not closed + await response.body_iterator.aclose() + assert closed == opened + assert len(set(threads)) == 1 + assert threads[0] != threading.get_ident() + + +@pytest.mark.asyncio +async def test_stream_send_error_closes_database_in_original_worker(api_client, monkeypatch): + _, db = api_client + insert(db, "stress_samples", id="sample", timestamp="2026-10-15T00:00:00Z", stress_score=26, level="source") + original = ViewRepository.snapshot + opened, closed = [], [] + @contextmanager + def observed(repository): + with original(repository) as connection: + opened.append(threading.get_ident()) + try: + yield connection + finally: + closed.append(threading.get_ident()) + monkeypatch.setattr(ViewRepository, "snapshot", observed) + request = Request({"type": "http", "app": api.app, "headers": [(b"x-api-key", b"synthetic-api-key")]}) + response = api.get_view_export(request, ViewSelection("stress", "2026-10-15", "2026-10-15")) + chunks = 0 + async def send(message): + nonlocal chunks + if message["type"] == "http.response.body": + chunks += 1 + if chunks == 2: + raise OSError("synthetic disconnect") + with pytest.raises(OSError, match="synthetic disconnect"): + await response.stream_response(send) + assert closed == opened + assert len(opened) == 1 + assert opened[0] != threading.get_ident() + + +@pytest.mark.asyncio +async def test_cancelled_read_still_queues_close_on_original_worker(api_client, monkeypatch): + _, db = api_client + original = ViewRepository.iter_records + started, released, closed = threading.Event(), threading.Event(), threading.Event() + threads = [] + def slow_records(repository, selection, connection): + threads.append(threading.get_ident()) + try: + started.set() + released.wait(5) + yield from original(repository, selection, connection) + finally: + threads.append(threading.get_ident()) + closed.set() + monkeypatch.setattr(ViewRepository, "iter_records", slow_records) + insert(db, "stress_samples", id="sample", timestamp="2026-10-15T00:00:00Z", stress_score=26, level="source") + request = Request({"type": "http", "app": api.app, "headers": [(b"x-api-key", b"synthetic-api-key")]}) + response = api.get_view_export(request, ViewSelection("stress", "2026-10-15", "2026-10-15")) + await response.body_iterator.__anext__() + pending = asyncio.create_task(response.body_iterator.__anext__()) + try: + assert await asyncio.to_thread(started.wait, 3) + pending.cancel() + released.set() + with pytest.raises(asyncio.CancelledError): + await pending + assert await asyncio.to_thread(closed.wait, 3) + assert len(set(threads)) == 1 + assert threads[0] != threading.get_ident() + finally: + released.set() + await response.body_iterator.aclose() diff --git a/tests/test_web_streaming.py b/tests/test_web_streaming.py new file mode 100644 index 0000000..3c3b493 --- /dev/null +++ b/tests/test_web_streaming.py @@ -0,0 +1,80 @@ +import httpx +import pytest +import respx + +from mi_fitness_mcp.web import create_app + + +class TrackedStream(httpx.SyncByteStream): + def __init__(self, fail=False): + self.consumed = 0 + self.closed = False + self.fail = fail + + def __iter__(self): + for n in range(30): + if self.fail and n == 1: + raise httpx.ReadError("synthetic read failure") + self.consumed += 1 + yield f"synthetic-row-{n}\r\n".encode() + + def close(self): + self.closed = True + + +def make_client(monkeypatch): + monkeypatch.setenv("MI_FITNESS_WEB_PASSWORD", "synthetic-web-password") + app = create_app(backend_url="http://synthetic-backend", api_key="synthetic-api-key", admin_key="synthetic-admin-key") + app.config["TESTING"] = True + return app.test_client() + + +@respx.mock +def test_stream_disconnect_closes_cursor_and_proxy_response(monkeypatch): + stream = TrackedStream() + route = respx.get("http://synthetic-backend/api/views/stress/export").mock(return_value=httpx.Response(200, + stream=stream, headers={"Content-Type": "text/csv; charset=utf-8", "Content-Disposition": 'attachment; filename="synthetic.csv"', "X-API-Key": "must-not-leak", "X-Admin-Key": "must-not-leak", "Authorization": "must-not-leak", "Set-Cookie": "must-not-leak"})) + client = make_client(monkeypatch) + response = client.get("/proxy/api/views/stress/export", auth=("admin", "synthetic-web-password"), buffered=False) + assert response.status_code == 200 + assert response.is_streamed + assert 0 < stream.consumed < 30 + assert not stream.closed + assert response.headers["Content-Disposition"].startswith("attachment") + assert not any(name.lower() in {"x-api-key", "x-admin-key", "authorization", "set-cookie"} for name in response.headers.keys()) + assert route.calls[0].request.headers["X-API-Key"] == "synthetic-api-key" + response.close() + assert stream.closed + assert stream.consumed < 30 + + +@respx.mock +def test_stream_normal_completion_and_error_release_upstream(monkeypatch): + client = make_client(monkeypatch) + for fail in [False, True]: + stream = TrackedStream(fail) + respx.get("http://synthetic-backend/api/views/body/export").mock(return_value=httpx.Response(200, stream=stream, headers={"Content-Type": "text/csv"})) + response = client.get("/proxy/api/views/body/export", auth=("admin", "synthetic-web-password"), buffered=False) + if fail: + with pytest.raises(httpx.ReadError): + list(response.response) + else: + assert len(b"".join(response.response).splitlines()) == 30 + assert stream.closed + response.close() + + +@respx.mock +def test_export_still_requires_web_auth_and_same_origin(monkeypatch): + client = make_client(monkeypatch) + assert client.get("/proxy/api/views/stress/export").status_code == 401 + assert client.get("/proxy/api/views/stress/export", auth=("admin", "synthetic-web-password"), headers={"Origin": "https://untrusted.example"}).status_code == 403 + + +@respx.mock +def test_export_connection_failure_uses_existing_error_shape(monkeypatch): + respx.get("http://synthetic-backend/api/views/stress/export").mock(side_effect=httpx.ConnectError("synthetic unavailable")) + response = make_client(monkeypatch).get("/proxy/api/views/stress/export", auth=("admin", "synthetic-web-password")) + assert response.status_code == 502 + assert response.json["error"] == "Flask 代理无法访问后端" + assert "X-Upstream-Time-Ms" in response.headers