Spaces:
Running
Running
| from __future__ import annotations | |
| import logging | |
| import os | |
| from .manual import discover_metadata_txt, ingest_manuals, resolve_local_mp4 | |
| from .metadata import parse_metadata_txt | |
| from .pdf_manual import ingest_pdf_manual | |
| from .pipeline import run_etl | |
| logger = logging.getLogger(__name__) | |
| def batch_build_index( | |
| *, | |
| data_root: str, | |
| runs_root: str = "runs", | |
| skip_existing_runs: bool = True, | |
| skip_vlm_if_no_ollama: bool = False, | |
| include_pdf: bool = False, | |
| pdf_path: str | None = None, | |
| vehicle_metadata: dict | None = None, | |
| pdf_pages_per_chunk: int = 2, | |
| **etl_kwargs, | |
| ) -> dict: | |
| """ | |
| For each metadata .txt under data_root: | |
| 1) Always upsert manual procedure doc (via full manual scan at end, or per-file ETL metadata) | |
| 2) If local .mp4 exists, run video ETL into runs/<safe_id>/ | |
| """ | |
| txt_paths = discover_metadata_txt(data_root) | |
| os.makedirs(runs_root, exist_ok=True) | |
| processed_videos = 0 | |
| skipped_videos = 0 | |
| missing_mp4 = 0 | |
| errors: list[str] = [] | |
| for meta_path in txt_paths: | |
| mp4 = resolve_local_mp4(meta_path) | |
| if not mp4: | |
| missing_mp4 += 1 | |
| continue | |
| meta = parse_metadata_txt(meta_path) | |
| safe_id = _safe_run_id(meta_path) | |
| out_dir = os.path.join(runs_root, safe_id) | |
| if skip_existing_runs and os.path.isfile(os.path.join(out_dir, "summary.json")): | |
| skipped_videos += 1 | |
| continue | |
| try: | |
| run_etl( | |
| video_path=mp4, | |
| metadata_path=meta_path, | |
| out_dir=out_dir, | |
| **etl_kwargs, | |
| ) | |
| processed_videos += 1 | |
| except Exception as e: | |
| msg = f"{meta_path}: {e}" | |
| logger.exception(msg) | |
| errors.append(msg) | |
| manual_summary = ingest_manuals( | |
| root_dir=data_root, | |
| embed_model=etl_kwargs.get("embed_model", "text-embedding-v3"), | |
| embed_backend=etl_kwargs.get("embed_backend", "cloud"), | |
| ollama_host=etl_kwargs.get("ollama_host"), | |
| qdrant_path=etl_kwargs.get("qdrant_path", ".fixsense_qdrant"), | |
| collection=etl_kwargs.get("collection", "fixsense_rag"), | |
| use_chroma=etl_kwargs.get("use_chroma", False), | |
| ) | |
| pdf_summary = None | |
| if include_pdf and pdf_path: | |
| pdf_summary = ingest_pdf_manual( | |
| pdf_path=pdf_path, | |
| vehicle_metadata=vehicle_metadata, | |
| pages_per_chunk=pdf_pages_per_chunk, | |
| embed_model=etl_kwargs.get("embed_model", "text-embedding-v3"), | |
| embed_backend=etl_kwargs.get("embed_backend", "cloud"), | |
| ollama_host=etl_kwargs.get("ollama_host"), | |
| qdrant_path=etl_kwargs.get("qdrant_path", ".fixsense_qdrant"), | |
| collection=etl_kwargs.get("collection", "fixsense_rag"), | |
| use_chroma=etl_kwargs.get("use_chroma", False), | |
| ) | |
| return { | |
| "data_root": data_root, | |
| "metadata_files": len(txt_paths), | |
| "processed_videos": processed_videos, | |
| "skipped_videos": skipped_videos, | |
| "missing_mp4": missing_mp4, | |
| "manual_ingest": manual_summary, | |
| "pdf_ingest": pdf_summary, | |
| "errors": errors, | |
| "skip_vlm_if_no_ollama": skip_vlm_if_no_ollama, | |
| } | |
| def _safe_run_id(metadata_path: str) -> str: | |
| base = os.path.splitext(os.path.basename(metadata_path))[0] | |
| safe = "".join(c if c.isalnum() or c in "-_" else "_" for c in base) | |
| return safe[:120] | |