One-time migration of the old single-file traces.json into daily JSONL. Returns True if migration ran, False if it was skipped (target dir already populated, no legacy file, or legacy file is not parseable).
(legacy_path: Path, target_dir: Path)
| 295 | |
| 296 | |
| 297 | def migrate_legacy_json(legacy_path: Path, target_dir: Path) -> bool: |
| 298 | """One-time migration of the old single-file traces.json into daily JSONL. |
| 299 | |
| 300 | Returns True if migration ran, False if it was skipped (target dir already |
| 301 | populated, no legacy file, or legacy file is not parseable). |
| 302 | """ |
| 303 | if target_dir.exists() and any(target_dir.glob("*.jsonl")): |
| 304 | return False |
| 305 | if not legacy_path.exists(): |
| 306 | return False |
| 307 | try: |
| 308 | records = json.loads(legacy_path.read_text(encoding="utf-8")) |
| 309 | except Exception: |
| 310 | return False |
| 311 | if not isinstance(records, list): |
| 312 | return False |
| 313 | |
| 314 | target_dir.mkdir(parents=True, exist_ok=True, mode=0o700) |
| 315 | for rec in records: |
| 316 | if not isinstance(rec, dict): |
| 317 | continue |
| 318 | ts = float(rec.get("timestamp", 0.0)) |
| 319 | day = _date_str(ts) |
| 320 | path = target_dir / f"{day}.jsonl" |
| 321 | try: |
| 322 | line = json.dumps(rec, default=str, ensure_ascii=False) |
| 323 | with open(path, "a", encoding="utf-8") as f: |
| 324 | f.write(line + "\n") |
| 325 | except Exception: |
| 326 | continue |
| 327 | |
| 328 | try: |
| 329 | legacy_path.rename(legacy_path.with_suffix(".json.bak")) |
| 330 | except Exception: |
| 331 | pass |
| 332 | return True |
| 333 | |
| 334 | |
| 335 | class TraceStore: |