refactor: extract/ingest facade, merge_one, deduplicate ops constants
- Add bincio/extract/ingest.py as a facade over the extract internals (ingest_parsed, strava_sync), reducing coupling from 6+ imports to one - Add merge_one() to merge.py — fast single-activity path for interactive edits (rewrites one file + index, skips full directory rebuild) - Rewrite edit/ops.py to delegate to the new facade; fix broken run_strava_sync return (was referencing undefined locals) - Remove duplicated SPORTS, STAT_PANELS, VALID_ACTIVITY_ID from edit/server.py — now imported from ops.py
This commit is contained in:
@@ -74,6 +74,84 @@ def _apply_sidecar_summary(summary: dict, fm: dict) -> dict:
|
||||
return s
|
||||
|
||||
|
||||
def merge_one(data_dir: Path, activity_id: str) -> None:
|
||||
"""Apply (or remove) sidecar overrides for a single activity.
|
||||
|
||||
Updates data_dir/_merged/activities/{id}.json and rewrites
|
||||
_merged/index.json. Faster than merge_all() for interactive edits
|
||||
because it touches only one activity file instead of rebuilding the
|
||||
whole _merged/activities/ directory.
|
||||
|
||||
Use merge_all() for bulk operations (first run, Strava sync, etc.).
|
||||
"""
|
||||
edits_dir = data_dir / "edits"
|
||||
acts_dir = data_dir / "activities"
|
||||
merged_dir = data_dir / "_merged"
|
||||
merged_acts = merged_dir / "activities"
|
||||
merged_acts.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
src = acts_dir / f"{activity_id}.json"
|
||||
if not src.exists():
|
||||
return
|
||||
|
||||
dest = merged_acts / f"{activity_id}.json"
|
||||
|
||||
# Determine if a sidecar or image list applies to this activity
|
||||
sidecar_path = edits_dir / f"{activity_id}.md" if edits_dir.exists() else None
|
||||
images_dir = edits_dir / "images" / activity_id if edits_dir.exists() else None
|
||||
has_sidecar = sidecar_path is not None and sidecar_path.exists()
|
||||
image_files: list[str] = []
|
||||
if images_dir and images_dir.exists():
|
||||
image_files = sorted(
|
||||
p.name for p in images_dir.iterdir()
|
||||
if p.is_file() and not p.name.startswith(".")
|
||||
)
|
||||
|
||||
needs_merge = has_sidecar or bool(image_files)
|
||||
|
||||
# Remove the old dest (symlink or file) before writing the new one
|
||||
if dest.exists() or dest.is_symlink():
|
||||
dest.unlink()
|
||||
|
||||
if needs_merge:
|
||||
detail = json.loads(src.read_text(encoding="utf-8"))
|
||||
if has_sidecar:
|
||||
fm, body = parse_sidecar(sidecar_path) # type: ignore[arg-type]
|
||||
detail = apply_sidecar(detail, fm, body)
|
||||
if image_files:
|
||||
detail["custom"] = dict(detail.get("custom") or {})
|
||||
detail["custom"]["images"] = image_files
|
||||
dest.write_text(json.dumps(detail, indent=2, ensure_ascii=False))
|
||||
else:
|
||||
dest.symlink_to(src.resolve())
|
||||
|
||||
# Rewrite index — load the full sidecar map so all summaries stay consistent
|
||||
index_path = data_dir / "index.json"
|
||||
if not index_path.exists():
|
||||
return
|
||||
|
||||
all_sidecars: dict[str, tuple[dict, str]] = {}
|
||||
if edits_dir and edits_dir.exists():
|
||||
for md_path in edits_dir.glob("*.md"):
|
||||
all_sidecars[md_path.stem] = parse_sidecar(md_path)
|
||||
|
||||
index = json.loads(index_path.read_text(encoding="utf-8"))
|
||||
activities = []
|
||||
for s in index.get("activities", []):
|
||||
aid = s.get("id", "")
|
||||
if aid in all_sidecars:
|
||||
fm, _ = all_sidecars[aid]
|
||||
s = _apply_sidecar_summary(s, fm)
|
||||
activities.append(s)
|
||||
|
||||
activities = [a for a in activities if a.get("privacy") != "private"]
|
||||
activities.sort(key=lambda a: a.get("started_at", ""), reverse=True)
|
||||
activities.sort(key=lambda a: 0 if a.get("custom", {}).get("highlight") else 1)
|
||||
|
||||
index["activities"] = activities
|
||||
(merged_dir / "index.json").write_text(json.dumps(index, indent=2, ensure_ascii=False))
|
||||
|
||||
|
||||
def merge_all(data_dir: Path) -> int:
|
||||
"""Build data_dir/_merged/ with all sidecar overrides applied.
|
||||
|
||||
|
||||
Reference in New Issue
Block a user