diff --git a/qa/view-statistics-oracle.cjs b/qa/view-statistics-oracle.cjs new file mode 100644 index 0000000..1992f97 --- /dev/null +++ b/qa/view-statistics-oracle.cjs @@ -0,0 +1,11 @@ +/* 仅测试使用：冻结的前端基线，不随产品展示模块改动。 */ +const fs = require('node:fs'); +const vm = require('node:vm'); +const {execFileSync} = require('node:child_process'); +const source = execFileSync('git', ['show', '497fd1c:src/mi_fitness_mcp/web_assets/static/js/health-views.js'], {encoding:'utf8'}); +const context = {module:{exports:{}}}; +vm.runInNewContext(source, context); +const input = JSON.parse(fs.readFileSync(0, 'utf8')); +const model = context.module.exports.build(input.kind, input.rows, input.range); +process.stdout.write(JSON.stringify({cards:model.cards.map(c=>c.raw ?? null), points:model.points, + references:Object.fromEntries(model.references.map(r=>[r.key,r.value])), coverage:model.coverage, stride:model.stride, count:model.records.length})); diff --git a/src/mi_fitness_mcp/services/view_query_service.py b/src/mi_fitness_mcp/services/view_query_service.py new file mode 100644 index 0000000..8b72912 --- /dev/null +++ b/src/mi_fitness_mcp/services/view_query_service.py @@ -0,0 +1,45 @@ +"""健康页面汇总、分页及流式导出的只读服务。""" + +from typing import Iterator + +from mi_fitness_mcp.storage import Database +from mi_fitness_mcp.storage.view_queries import ViewRepository, ViewSelection, _check_page + + +class ViewQueryService: + def __init__(self, db: Database, user_id: str): + self.repository = ViewRepository(db, user_id) + + @staticmethod + def _envelope(selection: ViewSelection) -> dict: + return dict(status='ok', kind=selection.kind, range=dict(start=selection.start_date, end=selection.end_date)) + + def get_view(self, selection: ViewSelection, page: int = 1) -> dict: + _check_page(page) + with self.repository.snapshot() as connection: + statistics = self.repository.statistics(selection, connection) + coverage = statistics.pop('coverage') + return self._envelope(selection) | dict(statistics=statistics, + points=self.repository.points(selection, connection), coverage=coverage, stride=selection.stride, + page=self.repository.page(selection, page, connection)) + + def get_page(self, selection: ViewSelection, page: int = 1) -> dict: + _check_page(page) + with self.repository.snapshot() as connection: + return self._envelope(selection) | dict(page=self.repository.page(selection, page, connection)) + + def get_overview(self, start_date: str, end_date: str) -> dict: + selections = {kind: ViewSelection(kind, start_date, end_date, + sample_type='resting' if kind == 'heartrate' else None) for kind in ('activity', 'heartrate', 'sleep', 'spo2', 'stress')} + with self.repository.snapshot() as connection: + statistics = {kind: self.repository.statistics(selection, connection) for kind, selection in selections.items()} + activity, heart, sleep, spo2, stress = (statistics[kind] for kind in selections) + return dict(steps=activity['primary_sum'], active_kcal=activity['total_active_kcal'], + distance_m=activity['total_distance_m'], resting_mean_bpm=heart['resting_mean_bpm'], + mean_daily_asleep_minutes=sleep['mean_daily_asleep_minutes'], mean_spo2_pct=spo2['primary_mean'], + mean_stress=stress['primary_mean'], activity_days=activity['coverage'], sleep_days=sleep['coverage'], + resting_count=heart['count'], spo2_count=spo2['count']) + + def iter_export(self, selection: ViewSelection) -> Iterator[dict]: + with self.repository.snapshot() as connection: + yield from self.repository.iter_records(selection, connection) diff --git a/src/mi_fitness_mcp/storage/view_queries.py b/src/mi_fitness_mcp/storage/view_queries.py new file mode 100644 index 0000000..12dd69f --- /dev/null +++ b/src/mi_fitness_mcp/storage/view_queries.py @@ -0,0 +1,272 @@ +"""健康视图的共享筛选、SQLite 聚合与有界明细读取。""" + +import json +import math +import sqlite3 +from contextlib import contextmanager +from dataclasses import dataclass +from datetime import date, timedelta +from typing import Iterator + +from mi_fitness_mcp.storage import Database + + +@dataclass(frozen=True) +class _Kind: + table: str + time: str + primary: str + fields: tuple[str, ...] + + +KINDS = { + 'activity': _Kind('daily_activity', 'date', 'steps', ('date', 'steps', 'distance_m', 'active_kcal', 'total_kcal', 'floors', 'active_minutes')), + 'heartrate': _Kind('heart_rate_samples', 'timestamp', 'bpm', ('timestamp', 'bpm', 'sample_type')), + 'sleep': _Kind('sleep_sessions', 'end_at', '_asleep', ('sleep_id', 'start_at', 'end_at', 'duration_minutes', 'time_asleep_minutes', 'time_awake_minutes', 'sleep_score', 'is_nap', 'stages')), + 'workouts': _Kind('workouts', 'start_at', 'duration_minutes', ('workout_id', 'activity_type', 'start_at', 'end_at', 'duration_minutes', 'distance_m', 'calories_kcal', 'avg_heart_rate_bpm', 'max_heart_rate_bpm', 'avg_pace_sec_per_km', 'max_pace_sec_per_km', 'total_steps')), + 'body': _Kind('body_measurements', 'timestamp', 'weight_kg', ('timestamp', 'weight_kg', 'bmi', 'body_fat_pct', 'muscle_mass_kg', 'water_pct', 'bone_mass_kg', 'visceral_fat_score', 'basal_metabolism_kcal', 'metabolic_age')), + 'spo2': _Kind('spo2_samples', 'timestamp', 'spo2_pct', ('timestamp', 'spo2_pct')), + 'stress': _Kind('stress_samples', 'timestamp', 'stress_score', ('timestamp', 'stress_score', 'level')), + 'abnormal': _Kind('abnormal_heart_beat_events', 'start_at', 'duration_seconds', ('event_id', 'start_at', 'end_at', 'duration_seconds')), +} +PRESSURE_BANDS = {'relaxed': (1, 25), 'mild': (26, 50), 'moderate': (51, 80), 'severe': (81, 100)} + + +@dataclass(frozen=True) +class ViewSelection: + 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: int | None = None + + def __post_init__(self): + if self.kind not in KINDS: + raise ValueError('不支持的查询类型') + try: + start, end = date.fromisoformat(self.start_date), date.fromisoformat(self.end_date) + except (TypeError, ValueError): + raise ValueError('日期必须为 YYYY-MM-DD') from None + if start.isoformat() != self.start_date or end.isoformat() != self.end_date: + raise ValueError('日期必须为 YYYY-MM-DD') + if not 1 <= (end - start).days + 1 <= 3660: + raise ValueError('请选择有效日期范围（最多3660天）') + if self.limit is not None and (type(self.limit) is not int or self.limit <= 0): + raise ValueError('数量限制必须为正整数') + if self.pressure_level is not None and self.pressure_level not in (*PRESSURE_BANDS, 'unclassified'): + raise ValueError('不支持的压力等级') + if self.sample_type is not None and (not isinstance(self.sample_type, str) or not self.sample_type or len(self.sample_type) > 100): + raise ValueError('采样类型必须为非空短字符串') + if type(self.include_naps) is not bool or type(self.latest_only) is not bool: + raise ValueError('小睡及最新记录筛选必须为布尔值') + + @property + def days(self) -> int: + return (date.fromisoformat(self.end_date) - date.fromisoformat(self.start_date)).days + 1 + + @property + def stride(self) -> int: + return math.ceil(self.days / 120) + + +def _check_page(page: int): + if type(page) is not int or not 1 <= page <= 1000000: + raise ValueError('页码必须为1至1000000的整数') + + +class ViewRepository: + def __init__(self, db: Database, user_id: str): + self.db = db + self.user_id = user_id + + @contextmanager + def snapshot(self): + with self.db._get_connection() as connection: + connection.execute('PRAGMA query_only = ON') + connection.execute('BEGIN') + try: + yield connection + finally: + connection.rollback() + + def _selection(self, selection: ViewSelection) -> tuple[str, list]: + """所有入口共用同一 CTE；先精确筛选，再按真实时刻和主键取最近 N 条。""" + spec = KINDS[selection.kind] + start, end = date.fromisoformat(selection.start_date), date.fromisoformat(selection.end_date) + if selection.kind == 'activity': + wide_start, wide_end = selection.start_date, selection.end_date + day_sql = 'date(date)' + moment_sql, fraction_sql = 'unixepoch(date)', '0' + else: + # ISO 文本粗筛保留时区偏移余量；无时区时间按 SQLite 的 UTC 语义解释。 + wide_start = (start - timedelta(days=min(2, (start - date.min).days))).isoformat() + wide_end = (end + timedelta(days=min(2, (date.max - end).days))).isoformat() + 'T23:59:59.999999' + # SQLite 的日期解析会把小数秒四舍五入到毫秒，跨午夜时甚至会进位。 + # 用整秒解析时区，另保留小数秒排序，避免微秒记录被主键错误重排。 + timestamp = spec.time + zone = f"CASE WHEN substr({timestamp}, -6, 1) IN ('+', '-') THEN substr({timestamp}, -6) WHEN substr({timestamp}, -1) = 'Z' THEN 'Z' ELSE '' END" + whole_second = f'(substr({timestamp}, 1, 19) || ({zone}))' + day_sql = f"date({whole_second}, '+8 hours')" + moment_sql = f'unixepoch({whole_second})' + fraction_sql = f"CASE WHEN substr({timestamp}, 20, 1) = '.' THEN CAST(ROUND(CAST('0.' || substr({timestamp}, 21) AS REAL) * 1000000000) AS INTEGER) ELSE 0 END" + fields = ', '.join(('id', *spec.fields)) + private = ', source_record_id, collected_at' if selection.kind == 'sleep' else '' + sql = f"""WITH candidates AS ( + SELECT {fields}{private}, {day_sql} AS _day, {moment_sql} AS _moment, {fraction_sql} AS _fraction + FROM {spec.table} WHERE user_id = ? AND {spec.time} >= ? AND {spec.time} <= ? + ), ranged AS ( + SELECT * FROM candidates WHERE _day BETWEEN ? AND ? + )""" + parameters = [self.user_id, wide_start, wide_end, selection.start_date, selection.end_date] + source = 'ranged' + if selection.kind == 'activity': + # 与旧 QueryService 一致：设备活动先合为日报，三个可选字段回退 0。 + sums = [] + for field in spec.fields: + if field == 'date': + continue + expression = f'SUM({field})' + if field in ('total_kcal', 'floors', 'active_minutes'): + expression = f'COALESCE({expression}, 0)' + sums.append(f'{expression} AS {field}') + sql += f", normalized AS (SELECT MAX(id) AS id, _day AS date, _day, MAX(_moment) AS _moment, 0 AS _fraction, {', '.join(sums)} FROM ranged GROUP BY _day)" + source = 'normalized' + elif selection.kind == 'sleep': + sql += """, reports AS ( + SELECT *, MAX(CASE WHEN source_record_id LIKE 'daily_report:%' THEN COALESCE(collected_at, '') END) + OVER (PARTITION BY _day) AS _report FROM ranged + ), sessions AS ( + SELECT * FROM reports WHERE _report IS NULL OR + (source_record_id LIKE 'daily_report:%' AND COALESCE(collected_at, '') = _report) + ), normalized AS ( + SELECT *, COALESCE(time_asleep_minutes, duration_minutes) AS _asleep""" + # JSON 只在 SQLite 内展开；缺失阶段和非法源 JSON 均保留 NULL。 + for phase in ('deep', 'light', 'rem'): + value = "CASE WHEN j.type = 'object' THEN j.value ELSE '{}' END" + sql += f""", (SELECT SUM(json_extract({value}, '$.minutes')) + FROM json_each(CASE WHEN json_valid(stages) THEN + CASE WHEN json_type(stages) = 'array' THEN stages ELSE '[]' END ELSE '[]' END) AS j + WHERE json_extract({value}, '$.stage') = '{phase}' + AND json_type({value}, '$.minutes') IN ('integer', 'real')) AS _{phase}""" + sql += ' FROM sessions)' + source = 'normalized' + filters = [] + if selection.kind == 'heartrate' and selection.sample_type is not None: + filters.append('sample_type = ?') + parameters.append(selection.sample_type) + if selection.kind == 'sleep' and not selection.include_naps: + filters.append('COALESCE(is_nap, 0) = 0') + if selection.kind == 'stress' and selection.pressure_level is not None: + if selection.pressure_level == 'unclassified': + filters.append('(stress_score IS NULL OR stress_score < 1 OR stress_score > 100)') + else: + filters.append('stress_score BETWEEN ? AND ?') + parameters.extend(PRESSURE_BANDS[selection.pressure_level]) + limit = 1 if selection.latest_only else selection.limit if selection.limit is not None else -1 + sql += f", selected AS (SELECT * FROM {source} {'WHERE ' + ' AND '.join(filters) if filters else ''} ORDER BY _moment DESC, _fraction DESC, id DESC LIMIT ?) " + parameters.append(limit) + return sql, parameters + + def _projection(self, selection: ViewSelection) -> str: + fields = ['id', *KINDS[selection.kind].fields] + if selection.kind == 'sleep': + fields.extend(('_asleep', '_deep', '_light', '_rem')) + return ', '.join(fields) + + def _record(self, row: sqlite3.Row, selection: ViewSelection) -> dict: + record = dict(row) + if selection.kind == 'sleep': + record['is_nap'] = bool(record['is_nap']) + try: + stages = json.loads(record['stages']) if record['stages'] else None + record['stages'] = stages if isinstance(stages, list) else None + except (TypeError, ValueError): + record['stages'] = None + return record + + def statistics(self, selection: ViewSelection, connection: sqlite3.Connection) -> dict: + sql, parameters = self._selection(selection) + field = KINDS[selection.kind].primary + expressions = ['COUNT(*) AS count', 'COUNT(DISTINCT _day) AS coverage', + f'SUM({field}) AS primary_sum', f'AVG({field}) AS primary_mean', + f'MIN({field}) AS primary_min', f'MAX({field}) AS primary_max'] + if selection.kind in ('activity', 'workouts'): + expressions.append('SUM(distance_m) AS total_distance_m') + if selection.kind == 'activity': + expressions.extend(('SUM(active_kcal) AS total_active_kcal', 'SUM(total_kcal) AS total_calories_kcal', + 'COUNT(DISTINCT _day) AS activity_days', + '(SELECT AVG(value) FROM (SELECT SUM(steps) AS value FROM selected GROUP BY _day)) AS mean_daily_steps')) + elif selection.kind == 'workouts': + expressions.append('SUM(calories_kcal) AS total_calories_kcal') + elif selection.kind == 'heartrate': + expressions.extend(("AVG(CASE WHEN sample_type = 'passive' THEN bpm END) AS passive_mean_bpm", + "AVG(CASE WHEN sample_type = 'resting' THEN bpm END) AS resting_mean_bpm")) + elif selection.kind == 'sleep': + expressions.extend(('SUM(CASE WHEN is_nap THEN 1 ELSE 0 END) AS nap_count', + '(SELECT AVG(value) FROM (SELECT SUM(_asleep) AS value FROM selected GROUP BY _day)) AS mean_daily_asleep_minutes', + '(SELECT AVG(value) FROM (SELECT SUM(_deep) AS value FROM selected GROUP BY _day)) AS mean_daily_deep_minutes')) + result = dict(connection.execute(sql + 'SELECT ' + ', '.join(expressions) + ' FROM selected', parameters).fetchone()) + if selection.kind == 'sleep' and result['nap_count'] is None: + result['nap_count'] = 0 + if selection.kind in ('body', 'abnormal'): + row = connection.execute(sql + f'SELECT {self._projection(selection)} FROM selected ORDER BY _moment DESC, _fraction DESC, id DESC LIMIT 1', parameters).fetchone() + result['latest'] = self._record(row, selection) if row is not None else None + return result + + def points(self, selection: ViewSelection, connection: sqlite3.Connection) -> list[dict]: + sql, parameters = self._selection(selection) + field = KINDS[selection.kind].primary + sql += ', bucketed AS (SELECT *, CAST((julianday(_day) - julianday(?)) / ? AS INTEGER) AS _bucket FROM selected) ' + parameters.extend((selection.start_date, selection.stride)) + if selection.kind == 'body': + sql += f', ranked AS (SELECT *, ROW_NUMBER() OVER (PARTITION BY _bucket ORDER BY _moment DESC, _fraction DESC, id DESC) AS _rank FROM bucketed WHERE {field} IS NOT NULL) SELECT _bucket, {field} AS v0 FROM ranked WHERE _rank = 1' + else: + if selection.kind == 'sleep': + values = ['SUM(_asleep)', 'SUM(_deep)', 'SUM(_light)', 'SUM(_rem)'] + elif selection.kind == 'heartrate': + values = ["AVG(CASE WHEN sample_type = 'passive' THEN bpm END)", + "AVG(CASE WHEN sample_type = 'resting' THEN bpm END)", + "AVG(CASE WHEN sample_type NOT IN ('passive', 'resting') OR sample_type IS NULL THEN bpm END)"] + else: + aggregate = 'SUM' if selection.kind in ('activity', 'workouts') else 'AVG' + values = ['COUNT(*)' if selection.kind == 'abnormal' else f'{aggregate}({field})'] + sql += 'SELECT _bucket, ' + ', '.join(f'{expression} AS v{n}' for n, expression in enumerate(values)) + ' FROM bucketed GROUP BY _bucket' + buckets = {row['_bucket']: dict(row) for row in connection.execute(sql, parameters)} + count = 4 if selection.kind == 'sleep' else 3 if selection.kind == 'heartrate' else 1 + start = date.fromisoformat(selection.start_date) + points = [] + for day in range(0, selection.days, selection.stride): + row = buckets.get(day // selection.stride) + values = [row[f'v{n}'] for n in range(count)] if row else [None] * count + if selection.kind == 'sleep': + total, *stages = values + phase_sum = sum(value or 0 for value in stages) + if total is None: + values = [None] * 4 + elif phase_sum > total: + values = [None, None, None, total / 60] + else: + values = [None if value is None else value / 60 for value in stages] + values.append(max(0, total - phase_sum) / 60) + points.append(dict(date=(start + timedelta(days=day)).isoformat(), values=values, missing=all(value is None for value in values))) + return points + + def page(self, selection: ViewSelection, page: int, connection: sqlite3.Connection) -> dict: + _check_page(page) + sql, parameters = self._selection(selection) + total = connection.execute(sql + 'SELECT COUNT(*) FROM selected', parameters).fetchone()[0] + rows = connection.execute(sql + f'SELECT {self._projection(selection)} FROM selected ORDER BY _moment DESC, _fraction DESC, id DESC LIMIT 10 OFFSET ?', [*parameters, (page - 1) * 10]) + return dict(number=page, size=10, total=total, rows=[self._record(row, selection) for row in rows]) + + def iter_records(self, selection: ViewSelection, connection: sqlite3.Connection) -> Iterator[dict]: + sql, parameters = self._selection(selection) + cursor = connection.execute(sql + f'SELECT {self._projection(selection)} FROM selected ORDER BY _moment DESC, _fraction DESC, id DESC', parameters) + try: + while batch := cursor.fetchmany(256): + for row in batch: + yield self._record(row, selection) + finally: + cursor.close() diff --git a/tests/test_view_queries.py b/tests/test_view_queries.py new file mode 100644 index 0000000..ad68035 --- /dev/null +++ b/tests/test_view_queries.py @@ -0,0 +1,232 @@ +import json +from contextlib import contextmanager + +import pytest + +from mi_fitness_mcp.storage import Database +from mi_fitness_mcp.storage.view_queries import ViewRepository, ViewSelection +from mi_fitness_mcp.services.view_query_service import ViewQueryService + + +@pytest.fixture +def db(tmp_path): + return Database(tmp_path / 'synthetic.db') + + +def insert(db, table, **fields): + fields = {'provider': 'synthetic', 'source_type': 'test', 'user_id': 'owner', **fields} + with db._get_connection() as connection: + connection.execute( + f"INSERT INTO {table} ({','.join(fields)}) VALUES ({','.join('?' for _ in fields)})", + tuple(fields.values()), + ) + connection.commit() + + +def test_selection_before_limit_and_paging(db): + for n in range(31): + insert(db, 'stress_samples', id=f's{n:02}', timestamp=f'2026-10-15T00:{n:02}:00Z', + stress_score=26 + n if n < 25 else 90, level='source-level') + service = ViewQueryService(db, 'owner') + selection = ViewSelection('stress', '2026-10-15', '2026-10-15', pressure_level='mild', limit=11) + result = service.get_view(selection) + assert result['statistics']['count'] == 11 + assert result['statistics']['primary_mean'] == pytest.approx(sum(range(40, 51)) / 11) + assert result['page']['total'] == 11 + assert len(result['page']['rows']) == 10 + page = service.get_page(selection, 2) + assert set(page) == {'status', 'kind', 'range', 'page'} + assert len(page['page']['rows']) == 1 + exported = list(service.iter_export(selection)) + assert exported == result['page']['rows'] + page['page']['rows'] + assert [r['stress_score'] for r in exported] == list(range(50, 39, -1)) + assert all(row['level'] == 'source-level' for row in exported) + + +def test_equal_timestamps_stable_order_and_owner_isolation(db): + for n in range(24): + insert(db, 'body_measurements', id=f'b{n:02}', timestamp='2026-10-15T00:00:00Z', + device_id=f'private-device-{n}', weight_kg=60 + n) + insert(db, 'body_measurements', id='outsider', user_id='other', device_id='other', + timestamp='2026-10-15T00:00:00Z', weight_kg=999) + service = ViewQueryService(db, 'owner') + selection = ViewSelection('body', '2026-10-15', '2026-10-15') + pages = [service.get_page(selection, n)['page']['rows'] for n in (1, 2, 3)] + records = sum(pages, []) + assert records == list(service.iter_export(selection)) + assert pages[0] == service.get_page(selection)['page']['rows'] + assert [r['weight_kg'] for r in records] == list(range(83, 59, -1)) + assert len({r['id'] for r in records}) == 24 + assert not any(set(row) & {'user_id', 'device_id', 'source_record_id', 'provider'} for row in records) + limited = service.get_view(ViewSelection('body', '2026-10-15', '2026-10-15', latest_only=True)) + assert limited['statistics']['count'] == 1 + assert limited['statistics']['latest']['weight_kg'] == 83 + for n, sample_type in enumerate(['resting', 'passive', 'active']): + insert(db, 'heart_rate_samples', id=f'h{n}', timestamp='2026-10-15T00:00:00Z', + bpm=60 + n * 10, sample_type=sample_type) + resting = service.get_view(ViewSelection('heartrate', '2026-10-15', '2026-10-15', sample_type='resting', limit=1)) + assert resting['statistics']['primary_mean'] == 60 + assert resting['page']['rows'][0]['sample_type'] == 'resting' + + +def test_china_midnight_sleep_and_nulls(db): + insert(db, 'sleep_sessions', id='night', sleep_id='night', start_at='2026-10-14T16:00:00Z', + end_at='2026-10-15T00:15:00Z', duration_minutes=495, time_asleep_minutes=480, + time_awake_minutes=15, stages=json.dumps([{'stage': 'deep', 'minutes': 120}])) + insert(db, 'sleep_sessions', id='nap', sleep_id='nap', start_at='2026-10-15T05:00:00Z', + end_at='2026-10-15T05:30:00Z', duration_minutes=30, time_asleep_minutes=30, + time_awake_minutes=0, is_nap=1, stages=None) + insert(db, 'sleep_sessions', id='next', sleep_id='next', start_at='2026-10-15T15:00:00Z', + end_at='2026-10-15T16:30:00Z', duration_minutes=90, time_asleep_minutes=90, + time_awake_minutes=0, stages='not-json') + service = ViewQueryService(db, 'owner') + result = service.get_view(ViewSelection('sleep', '2026-10-15', '2026-10-17')) + assert result['statistics']['primary_sum'] == 600 + assert result['statistics']['mean_daily_asleep_minutes'] == 300 + assert result['statistics']['mean_daily_deep_minutes'] == 120 + assert result['statistics']['nap_count'] == 1 + assert result['coverage'] == 2 + assert result['points'][0]['values'] == [2, None, None, 6.5] + assert result['points'][1]['values'] == [None, None, None, 1.5] + assert result['points'][2]['missing'] is True + assert service.get_view(ViewSelection('sleep', '2026-10-15', '2026-10-15', include_naps=False))['statistics']['primary_sum'] == 480 + for n, timestamp in enumerate(['2026-10-14T16:00:00Z', '2026-10-15T00:00:00+08:00', '2026-10-14 16:00:00']): + insert(db, 'heart_rate_samples', id=f't{n}', timestamp=timestamp, bpm=60 + n, sample_type='passive') + assert service.get_view(ViewSelection('heartrate', '2026-10-15', '2026-10-15'))['statistics']['count'] == 3 + insert(db, 'stress_samples', id='zero', timestamp='2026-10-14T16:00:00Z', stress_score=0, level='zero') + zero = service.get_view(ViewSelection('stress', '2026-10-15', '2026-10-16')) + assert zero['statistics']['primary_mean'] == 0 + assert zero['points'][1]['values'] == [None] + + +def test_sleep_daily_report_supersedes_raw_and_old_report(db): + for row_id, source, collected, minutes in [('raw', 'raw:1', '2026-10-15', 400), + ('old', 'daily_report:1', '2026-10-15T01:00:00Z', 450), + ('new', 'daily_report:2', '2026-10-15T02:00:00Z', 480)]: + insert(db, 'sleep_sessions', id=row_id, sleep_id=row_id, source_record_id=source, + collected_at=collected, start_at='2026-10-14T15:00:00Z', end_at='2026-10-14T23:00:00Z', + duration_minutes=minutes, time_asleep_minutes=minutes, time_awake_minutes=0) + result = ViewQueryService(db, 'owner').get_view(ViewSelection('sleep', '2026-10-15', '2026-10-15')) + assert result['statistics']['count'] == 1 + assert result['statistics']['primary_sum'] == 480 + + +def test_body_real_instant_order_precedes_text_and_id(db): + for row_id, timestamp, weight in [('z', '2026-10-15T09:00:00+08:00', 80), + ('a', '2026-10-15T02:00:00Z', 70), ('y', '2026-10-14T15:59:59Z', 999)]: + insert(db, 'body_measurements', id=row_id, timestamp=timestamp, weight_kg=weight) + service = ViewQueryService(db, 'owner') + result = service.get_view(ViewSelection('body', '2026-10-15', '2026-10-15', limit=1)) + assert result['statistics']['latest']['id'] == 'a' + assert result['statistics']['primary_mean'] == 70 + assert result['points'][0]['values'] == [70] + + +def test_submillisecond_timestamps_still_order_by_real_instant(db): + for row_id, timestamp, weight in [ + ('z', '2026-10-15T09:00:00.000100+08:00', 80), + ('a', '2026-10-15T01:00:00.000200Z', 70), + ('x', '2026-10-15T01:00:00.999900Z', 60), + ('b', '2026-10-15T01:00:01.000000Z', 50), + ]: + insert(db, 'body_measurements', id=row_id, timestamp=timestamp, weight_kg=weight) + rows = list(ViewQueryService(db, 'owner').iter_export(ViewSelection('body', '2026-10-15', '2026-10-15'))) + assert [row['id'] for row in rows] == ['b', 'x', 'a', 'z'] + + +@pytest.mark.parametrize('level, expected', [('relaxed', [25, 1]), ('mild', [50, 26]), + ('moderate', [80, 51]), ('severe', [100, 81]), ('unclassified', [101, 0])]) +def test_pressure_numeric_bands_ignore_source_level(db, level, expected): + for n, value in enumerate([0, 1, 25, 26, 50, 51, 80, 81, 100, 101]): + insert(db, 'stress_samples', id=str(n), timestamp=f'2026-10-15T00:{n:02}:00Z', + stress_score=value, level='unrelated-source-value') + rows = list(ViewQueryService(db, 'owner').iter_export(ViewSelection('stress', '2026-10-15', '2026-10-15', pressure_level=level))) + assert [row['stress_score'] for row in rows] == expected + + +def test_export_close_releases_short_read_transaction(db, monkeypatch): + insert(db, 'stress_samples', id='one', timestamp='2026-10-15T00:00:00Z', stress_score=26, level='source') + original = ViewRepository.snapshot + closed = [] + @contextmanager + def observed(repository): + with original(repository) as connection: + try: + yield connection + finally: + closed.append(True) + monkeypatch.setattr(ViewRepository, 'snapshot', observed) + generator = ViewQueryService(db, 'owner').iter_export(ViewSelection('stress', '2026-10-15', '2026-10-15')) + assert next(generator)['stress_score'] == 26 + assert not closed + generator.close() + assert closed == [True] + + +def test_first_view_uses_one_read_snapshot(db, monkeypatch): + service = ViewQueryService(db, 'owner') + original = ViewRepository.snapshot + connections = [] + + @contextmanager + def observed(repository): + with original(repository) as connection: + assert connection.in_transaction + connections.append(connection) + yield connection + + monkeypatch.setattr(ViewRepository, 'snapshot', observed) + traces = [] + for method in ['statistics', 'points', 'page']: + implementation = getattr(ViewRepository, method) + + def wrapped(self, *args, _method=implementation, **kwargs): + connection = kwargs.get('connection', args[-1]) + assert connection is connections[-1] + assert connection.in_transaction + traces.append(connection) + return _method(self, *args, **kwargs) + + monkeypatch.setattr(ViewRepository, method, wrapped) + service.get_view(ViewSelection('stress', '2026-10-15', '2026-10-15')) + assert len(connections) == 1 + assert len(traces) == 3 + assert all(connection is connections[0] for connection in traces) + + +def test_large_input_uses_sql_paging(db, monkeypatch): + with db._get_connection() as connection: + connection.executemany("INSERT INTO stress_samples(id,provider,source_type,user_id,timestamp,stress_score,level) VALUES(?,?,?,?,?,?,?)", + [(f's{n:05}', 'synthetic', 'test', 'owner', f'2026-10-15T00:00:{n % 60:02}.{n:06}Z', n % 101, 'source') for n in range(20000)]) + connection.commit() + def forbidden(*args, **kwargs): + raise AssertionError('must aggregate and page in SQLite') + monkeypatch.setattr(db, 'query_stress_samples', forbidden) + monkeypatch.setattr(ViewRepository, 'iter_records', forbidden) + statements = [] + original = ViewRepository.snapshot + @contextmanager + def observed(repository): + with original(repository) as connection: + connection.set_trace_callback(statements.append) + yield connection + monkeypatch.setattr(ViewRepository, 'snapshot', observed) + result = ViewQueryService(db, 'owner').get_view(ViewSelection('stress', '2026-10-15', '2026-10-15'), 2) + assert result['statistics']['count'] == 20000 + assert len(result['page']['rows']) == 10 + assert len(result['points']) == 1 + assert any('LIMIT 10 OFFSET 10' in sql for sql in statements) + assert not any('SELECT * FROM stress_samples' in sql for sql in statements) + + +@pytest.mark.parametrize('kwargs', [dict(kind='unknown'), dict(start_date='2026-02-30'), dict(end_date='2026-10-14'), dict(limit=0), dict(limit=True), dict(pressure_level='sql injection'), dict(start_date='2000-01-01')]) +def test_invalid_selections(kwargs): + values = dict(kind='stress', start_date='2026-10-15', end_date='2026-10-16') + with pytest.raises(ValueError): + ViewSelection(**(values | kwargs)) + + +@pytest.mark.parametrize('page', [0, -1, True, 1.5, 1000001]) +def test_invalid_pages(db, page): + with pytest.raises(ValueError): + ViewQueryService(db, 'owner').get_page(ViewSelection('stress', '2026-10-15', '2026-10-15'), page) diff --git a/tests/test_view_statistics.py b/tests/test_view_statistics.py new file mode 100644 index 0000000..116f3e0 --- /dev/null +++ b/tests/test_view_statistics.py @@ -0,0 +1,174 @@ +import json +import shutil +import subprocess +from pathlib import Path + +import pytest + +from mi_fitness_mcp.storage import Database +from mi_fitness_mcp.storage.view_queries import ViewSelection +from mi_fitness_mcp.services.view_query_service import ViewQueryService +from mi_fitness_mcp.services.query_service import QueryService + +from test_view_queries import insert + + +ROOT = Path(__file__).resolve().parents[1] +NODE = shutil.which('node') or 'D:/nodejs/node.exe' + + +def seed(db, kind): + table, time_field, base = { + 'activity': ('daily_activity', 'date', dict(steps=100, distance_m=80, active_kcal=5)), + 'heartrate': ('heart_rate_samples', 'timestamp', dict(bpm=72, sample_type='passive')), + 'sleep': ('sleep_sessions', 'end_at', dict(sleep_id='s', start_at='2026-10-14T14:00:00Z', duration_minutes=495, time_asleep_minutes=480, time_awake_minutes=15, stages=json.dumps([dict(stage='deep', minutes=100), dict(stage='light', minutes=300), dict(stage='rem', minutes=80)]))), + 'workouts': ('workouts', 'start_at', dict(workout_id='w', end_at='2026-10-15T02:00:00Z', activity_type='run', duration_minutes=50, distance_m=8000, calories_kcal=400)), + 'body': ('body_measurements', 'timestamp', dict(weight_kg=75, bmi=22, body_fat_pct=20)), + 'spo2': ('spo2_samples', 'timestamp', dict(spo2_pct=96)), + 'stress': ('stress_samples', 'timestamp', dict(stress_score=0, level='source')), + 'abnormal': ('abnormal_heart_beat_events', 'start_at', dict(event_id='a', end_at='2026-10-15T02:00:00Z', duration_seconds=30)), + }[kind] + rows = [] + for n in range(4): + row = dict(base, id=f'id{n}', **{time_field: f'2026-10-{15 + n // 3:02}' + ('' if kind == 'activity' else f'T0{n}:00:00Z')}) + if kind == 'activity': + row.update(device_id=f'd{n}', steps=n * 100, total_kcal=None, active_minutes=None) + elif kind == 'heartrate': + row.update(bpm=60 + n * 20, sample_type=['resting', 'passive', 'passive', 'active'][n]) + elif kind == 'sleep': + row.update(sleep_id=str(n), is_nap=n == 1) + if n == 1: + row.update(time_asleep_minutes=30, duration_minutes=30, stages=None) + if n == 2: + row['stages'] = json.dumps([dict(stage='deep', minutes=999)]) + elif kind == 'body': + row['weight_kg'] = 75 - n + elif kind == 'spo2': + row['spo2_pct'] = 94 + n + elif kind == 'stress': + row['stress_score'] = [0, 26, 81, 101][n] + elif kind == 'workouts': + row['workout_id'] = str(n) + elif kind == 'abnormal': + row['event_id'] = str(n) + insert(db, table, **row) + if kind == 'sleep': + row['stages'] = json.loads(row['stages']) if row['stages'] else None + rows.append(row) + if kind == 'activity': + grouped = {} + for row in rows: + day = grouped.setdefault(row['date'], dict(date=row['date'], steps=0, distance_m=0, active_kcal=0, total_kcal=0, floors=0, active_minutes=0)) + for key in ['steps', 'distance_m', 'active_kcal', 'total_kcal', 'floors', 'active_minutes']: + day[key] += row.get(key) or 0 + rows = list(grouped.values()) + return rows + + +@pytest.mark.parametrize('kind', ['activity', 'heartrate', 'sleep', 'workouts', 'body', 'spo2', 'stress', 'abnormal']) +@pytest.mark.parametrize('end_date', ['2026-10-17', '2027-10-15']) +def test_all_eight_kinds_match_frontend_baseline(tmp_path, kind, end_date): + db = Database(tmp_path / 'synthetic.db') + rows = seed(db, kind) + expected = json.loads(subprocess.run([NODE, str(ROOT / 'qa/view-statistics-oracle.cjs')], cwd=ROOT, + input=json.dumps(dict(kind=kind, rows=rows, range=dict(start='2026-10-15', end=end_date))), text=True, + encoding='utf-8', capture_output=True, check=True).stdout) + result = ViewQueryService(db, 'owner').get_view(ViewSelection(kind, '2026-10-15', end_date)) + s = result['statistics'] + cards = { + 'activity': [s.get('primary_sum'), s.get('total_active_kcal'), None if s.get('total_distance_m') is None else s['total_distance_m'] / 1000, result['coverage']], + 'heartrate': [s.get('passive_mean_bpm'), s.get('resting_mean_bpm'), s.get('primary_min'), s.get('primary_max')], + 'sleep': [s.get('mean_daily_asleep_minutes'), result['coverage'], s.get('mean_daily_deep_minutes'), s.get('nap_count')], + 'workouts': [s['count'], s.get('primary_sum'), None if s.get('total_distance_m') is None else s['total_distance_m'] / 1000, s.get('total_calories_kcal')], + 'body': [s.get('latest', {}).get('weight_kg'), s.get('latest', {}).get('bmi'), s.get('latest', {}).get('body_fat_pct'), s['count']], + 'spo2': [s.get('primary_mean'), s.get('primary_min'), s.get('primary_max'), result['coverage']], + 'stress': [s.get('primary_mean'), s.get('primary_min'), s.get('primary_max'), result['coverage']], + 'abnormal': [s['count'], s.get('primary_sum'), result['coverage'], None], + }[kind] + assert cards == pytest.approx(expected['cards'], abs=1e-9) + assert result['coverage'] == expected['coverage'] + assert result['stride'] == expected['stride'] + assert s['count'] == expected['count'] + assert len(result['points']) <= 120 + for actual, wanted in zip(result['points'], expected['points'], strict=True): + assert actual['date'] == wanted['date'] + assert actual['values'] == pytest.approx(wanted['values'], abs=1e-9) + assert actual['missing'] == wanted['missing'] + if 'average' in expected['references']: + reference = s['mean_daily_steps'] if kind == 'activity' else s['mean_daily_asleep_minutes'] / 60 if kind == 'sleep' else s['primary_mean'] + assert reference == pytest.approx(expected['references']['average'], abs=1e-9) + + +def test_overview_and_empty_statistics(tmp_path): + db = Database(tmp_path / 'synthetic.db') + service = ViewQueryService(db, 'owner') + empty = service.get_view(ViewSelection('stress', '2026-10-15', '2026-10-15')) + assert empty['statistics']['primary_mean'] is None + assert empty['statistics']['primary_sum'] is None + assert empty['page'] == dict(number=1, size=10, total=0, rows=[]) + for kind in ['activity', 'heartrate', 'sleep', 'spo2', 'stress']: + seed(db, kind) + overview = service.get_overview('2026-10-15', '2026-10-17') + assert set(overview) == {'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'} + assert overview['steps'] == 600 + assert overview['resting_mean_bpm'] == 60 + assert overview['mean_daily_asleep_minutes'] == 735 + assert overview['activity_days'] == 2 + assert overview['sleep_days'] == 2 + assert overview['resting_count'] == 1 + assert overview['spo2_count'] == 4 + + +def test_sleep_reports_match_existing_query_service_and_frontend(tmp_path): + db = Database(tmp_path / 'synthetic.db') + for row_id, source, collected, minutes, nap in [ + ('raw', 'source:raw', '2026-10-15T01:00:00Z', 420, False), + ('old', 'daily_report:old', '2026-10-15T02:00:00Z', 450, False), + ('night', 'daily_report:new-night', '2026-10-15T03:00:00Z', 480, False), + ('nap', 'daily_report:new-nap', '2026-10-15T03:00:00Z', 30, True), + ]: + insert(db, 'sleep_sessions', id=row_id, sleep_id=row_id, source_record_id=source, + collected_at=collected, start_at='2026-10-14T23:00:00+08:00', end_at='2026-10-15T07:00:00+08:00', + duration_minutes=minutes + 15, time_asleep_minutes=minutes, time_awake_minutes=15, is_nap=nap, + stages=json.dumps([dict(stage='deep', minutes=100)]) if not nap else None) + for include_naps in (True, False): + records = QueryService(db, 'owner').get_sleep_sessions('2026-10-15', '2026-10-15', include_naps=include_naps) + expected = json.loads(subprocess.run([NODE, str(ROOT / 'qa/view-statistics-oracle.cjs')], cwd=ROOT, + input=json.dumps(dict(kind='sleep', rows=records, range=dict(start='2026-10-15', end='2026-10-15'))), + text=True, encoding='utf-8', capture_output=True, check=True).stdout) + result = ViewQueryService(db, 'owner').get_view(ViewSelection('sleep', '2026-10-15', '2026-10-15', include_naps=include_naps)) + assert result['statistics']['count'] == expected['count'] + assert result['statistics']['mean_daily_asleep_minutes'] == expected['cards'][0] + assert result['points'] == expected['points'] + + +def test_legacy_sleep_actual_fallback_and_json_stage_types(tmp_path): + db = Database(tmp_path / 'synthetic.db') + # 仅此合成库放宽旧字段约束，以验证历史 NULL 值的回退与 JSON 容错。 + with db._get_connection() as connection: + connection.execute('CREATE TABLE legacy_sleep AS SELECT * FROM sleep_sessions') + connection.execute('DROP TABLE sleep_sessions') + connection.execute('ALTER TABLE legacy_sleep RENAME TO sleep_sessions') + connection.commit() + insert(db, 'sleep_sessions', id='legacy', sleep_id='legacy', start_at='2026-10-14T23:00:00+08:00', + end_at='2026-10-15T07:00:00+08:00', duration_minutes=480, time_asleep_minutes=None, + time_awake_minutes=0, stages=json.dumps(['ignored', dict(stage='deep', minutes='120'), + dict(stage='deep', minutes=60), dict(stage='deep', minutes=None), dict(stage='light', minutes=300)])) + result = ViewQueryService(db, 'owner').get_view(ViewSelection('sleep', '2026-10-15', '2026-10-15')) + assert result['statistics']['primary_sum'] == 480 + assert result['statistics']['mean_daily_deep_minutes'] == 60 + assert result['points'][0]['values'] == [1, 5, None, 2] + + +def test_activity_limit_applies_to_daily_summary_and_optional_zero_fallback(tmp_path): + db = Database(tmp_path / 'synthetic.db') + seed(db, 'activity') + service = ViewQueryService(db, 'owner') + selection = ViewSelection('activity', '2026-10-15', '2026-10-17', limit=1) + result = service.get_view(selection) + assert result['statistics']['count'] == 1 + assert result['statistics']['primary_sum'] == 300 + assert result['statistics']['mean_daily_steps'] == 300 + assert result['page']['rows'][0]['total_kcal'] == 0 + assert result['page']['rows'][0]['active_minutes'] == 0 + assert list(service.iter_export(selection)) == result['page']['rows']