solpbc.org
personal memory agent
1#!/usr/bin/env python3
2# SPDX-License-Identifier: AGPL-3.0-only
3# Copyright (c) 2026 sol pbc
4
5"""Backfill Apple Health workout distance/energy metadata from retained raws.
6
7The original Apple Health workout dedupe keys are stable identifiers once
8assigned. This migration intentionally updates only normalized row metadata
9from retained raw exports; it never recalculates or rewrites ``dedupe_key``,
10``normalized_ref``, or ``raw_ref``.
11
12Dry-run by default. Pass ``--apply`` to rewrite changed normalized shards.
13"""
14
15from __future__ import annotations
16
17import argparse
18import json
19import sys
20import zipfile
21from pathlib import Path
22from typing import Any, Iterable
23
24REPO_ROOT = Path(__file__).resolve().parent.parent
25if str(REPO_ROOT) not in sys.path:
26 sys.path.insert(0, str(REPO_ROOT))
27
28from solstone.think.importers.apple_health import ( # noqa: E402
29 _find_export_xml_in_directory,
30 _find_export_xml_in_zip,
31 _workout_statistics_by_raw_ref,
32)
33from solstone.think.importers.health_schema import SOURCE_APPLE_HEALTH # noqa: E402
34from solstone.think.importers.shared import write_jsonl_records # noqa: E402
35
36_WORKOUT_STATISTIC_KEYS = (
37 "totalDistance",
38 "totalDistanceUnit",
39 "totalDistanceType",
40 "totalEnergyBurned",
41 "totalEnergyBurnedUnit",
42 "totalEnergyBurnedType",
43)
44
45
46def _iter_import_dirs(journal_root: Path) -> Iterable[Path]:
47 imports_root = journal_root / "imports"
48 if not imports_root.is_dir():
49 return ()
50 return (
51 path
52 for path in sorted(imports_root.iterdir())
53 if path.is_dir() and not path.name.startswith("_")
54 )
55
56
57def _raw_sources(import_dir: Path) -> list[tuple[Path, str]]:
58 raw_dir = import_dir / "raw"
59 if not raw_dir.is_dir():
60 return []
61
62 sources: list[tuple[Path, str]] = []
63 export_xml = _find_export_xml_in_directory(raw_dir)
64 if export_xml is not None:
65 raw_ref = (
66 f"imports/{import_dir.name}/{export_xml.relative_to(import_dir).as_posix()}"
67 )
68 sources.append((raw_dir, raw_ref))
69
70 for candidate in sorted(raw_dir.rglob("*.zip")):
71 with zipfile.ZipFile(candidate) as archive:
72 if _find_export_xml_in_zip(archive.namelist()) is None:
73 continue
74 raw_ref = (
75 f"imports/{import_dir.name}/{candidate.relative_to(import_dir).as_posix()}"
76 )
77 sources.append((candidate, raw_ref))
78 return sources
79
80
81def _load_workout_statistics(import_dir: Path) -> dict[str, dict[str, str]]:
82 stats_by_ref: dict[str, dict[str, str]] = {}
83 for source_path, raw_ref in _raw_sources(import_dir):
84 stats_by_ref.update(_workout_statistics_by_raw_ref(source_path, raw_ref))
85 return stats_by_ref
86
87
88def _read_jsonl(path: Path) -> list[dict[str, Any]]:
89 rows: list[dict[str, Any]] = []
90 for line_number, line in enumerate(
91 path.read_text(encoding="utf-8").splitlines(), start=1
92 ):
93 if not line.strip():
94 continue
95 try:
96 rows.append(json.loads(line))
97 except json.JSONDecodeError as exc:
98 raise ValueError(
99 f"Could not parse {path} line {line_number}: {exc}"
100 ) from exc
101 return rows
102
103
104def _merge_workout_statistics(
105 row: dict[str, Any],
106 stats_by_ref: dict[str, dict[str, str]],
107) -> bool:
108 if row.get("source_family") != SOURCE_APPLE_HEALTH:
109 return False
110 if row.get("kind") != "workout":
111 return False
112 raw_ref = row.get("raw_ref")
113 if not isinstance(raw_ref, str):
114 return False
115 recovered = stats_by_ref.get(raw_ref)
116 if not recovered:
117 return False
118
119 metadata = row.get("metadata")
120 if not isinstance(metadata, dict):
121 metadata = {}
122 row["metadata"] = metadata
123
124 changed = False
125 for key in _WORKOUT_STATISTIC_KEYS:
126 value = recovered.get(key)
127 if value is not None and key not in metadata:
128 metadata[key] = value
129 changed = True
130 return changed
131
132
133def _process_import_dir(
134 import_dir: Path,
135 *,
136 journal_root: Path,
137 apply: bool,
138) -> tuple[int, int]:
139 stats_by_ref = _load_workout_statistics(import_dir)
140 if not stats_by_ref:
141 return (0, 0)
142
143 normalized_dir = import_dir / "normalized"
144 if not normalized_dir.is_dir():
145 return (0, 0)
146
147 rows_seen = 0
148 rows_changed = 0
149 for shard in sorted(normalized_dir.glob("*.jsonl")):
150 rows = _read_jsonl(shard)
151 shard_changed = 0
152 for row in rows:
153 if row.get("kind") == "workout":
154 rows_seen += 1
155 if _merge_workout_statistics(row, stats_by_ref):
156 shard_changed += 1
157 if shard_changed:
158 rows_changed += shard_changed
159 rel = shard.relative_to(journal_root)
160 action = "updated" if apply else "would update"
161 print(f"{rel}: {shard_changed} workout rows {action}")
162 if apply:
163 write_jsonl_records(shard, rows)
164 return (rows_seen, rows_changed)
165
166
167def main(argv: list[str] | None = None) -> int:
168 parser = argparse.ArgumentParser(
169 description=(
170 "Backfill Apple Health workout distance/energy metadata from retained "
171 "raw exports. Dry-run unless --apply is given."
172 )
173 )
174 parser.add_argument(
175 "journal_root",
176 type=Path,
177 help="Journal root to inspect (required; never defaults to a live journal)",
178 )
179 parser.add_argument(
180 "--apply",
181 action="store_true",
182 help="Rewrite changed normalized shards (default: dry-run report only)",
183 )
184 args = parser.parse_args(argv)
185
186 journal_root = args.journal_root.expanduser().resolve()
187 if not journal_root.is_dir():
188 parser.error(f"journal root is not a directory: {journal_root}")
189
190 workout_rows = 0
191 changed_rows = 0
192 for import_dir in _iter_import_dirs(journal_root):
193 seen, changed = _process_import_dir(
194 import_dir,
195 journal_root=journal_root,
196 apply=args.apply,
197 )
198 workout_rows += seen
199 changed_rows += changed
200
201 if args.apply:
202 print(f"{workout_rows} workout rows inspected; {changed_rows} rows updated")
203 else:
204 print(
205 f"{workout_rows} workout rows inspected; "
206 f"{changed_rows} rows would update "
207 "(dry-run; pass --apply to rewrite)"
208 )
209 return 0
210
211
212if __name__ == "__main__":
213 raise SystemExit(main())