Spaces:
Running
Running
| from pathlib import Path | |
| from typing import List, Dict, Any | |
| from .file_detector import detect_file_type | |
| from .parser_tcx import parse_tcx_file | |
| from .parser_gpx import parse_gpx_file | |
| from .parser_fit import parse_fit_file | |
| from .validation import validate_uploads, safe_gzip_decompress, is_xml_heuristic, ValidationError | |
| import tempfile | |
| import shutil | |
| import logging | |
| logger = logging.getLogger(__name__) | |
| def load_runs_from_folder(folder: str) -> List[Dict[str, Any]]: | |
| """ | |
| Parse all supported files from a local folder. Returns list of run dicts. | |
| """ | |
| runs = [] | |
| p = Path(folder) | |
| if not p.exists(): | |
| return runs | |
| for f in sorted(p.iterdir()): | |
| file_type = detect_file_type(str(f)) | |
| if file_type == "tcx": | |
| parsed = parse_tcx_file(str(f)) | |
| elif file_type == "gpx": | |
| parsed = parse_gpx_file(str(f)) | |
| elif file_type == "fit": | |
| parsed = parse_fit_file(str(f)) | |
| else: | |
| parsed = None | |
| if parsed: | |
| runs.append(parsed) | |
| # sort by start_time | |
| runs.sort(key=lambda r: r.get("start_time") or r.get("id")) | |
| return runs | |
| def load_runs_from_uploaded_files(uploaded_files) -> List[Dict[str, Any]]: | |
| """ | |
| Accepts Gradio-style uploaded files (list). Writes to temp folder and parses. | |
| """ | |
| tmpdir = Path(tempfile.mkdtemp(prefix="runner_ingest_")) | |
| runs = [] | |
| validate_uploads(uploaded_files) | |
| try: | |
| saved = [] | |
| for f in uploaded_files or []: | |
| src_path = getattr(f, "name", None) or getattr(f, "filename", None) | |
| if not src_path: | |
| continue | |
| src_path = str(src_path) | |
| filename = Path(src_path).name | |
| dest = tmpdir / filename | |
| # Special handling for .tcx.gz and .fit.gz -> safe decompression | |
| if filename.lower().endswith((".tcx.gz", ".fit.gz")): | |
| # We decompress it to a .tcx or .fit file in the temp dir | |
| ext_len = 3 # .gz | |
| decompressed_dest = tmpdir / (filename[:-ext_len]) | |
| safe_gzip_decompress(src_path, str(decompressed_dest)) | |
| saved.append(decompressed_dest) | |
| else: | |
| try: | |
| shutil.copyfile(src_path, dest) | |
| except Exception: | |
| # fallback: manual binary read/write | |
| with open(src_path, "rb") as sf, open(dest, "wb") as df: | |
| df.write(sf.read()) | |
| saved.append(dest) | |
| for dest in saved: | |
| file_type = detect_file_type(str(dest)) | |
| # XML heuristic check (only for TCX/GPX) | |
| if file_type in ("tcx", "gpx") and not is_xml_heuristic(str(dest)): | |
| logger.warning(f"File {dest.name} failed XML heuristic check, skipping.") | |
| continue | |
| if file_type == "tcx": | |
| parsed = parse_tcx_file(str(dest)) | |
| elif file_type == "gpx": | |
| parsed = parse_gpx_file(str(dest)) | |
| elif file_type == "fit": | |
| parsed = parse_fit_file(str(dest)) | |
| else: | |
| parsed = None | |
| if parsed: | |
| runs.append(parsed) | |
| finally: | |
| try: | |
| shutil.rmtree(tmpdir) | |
| except Exception: | |
| pass | |
| runs.sort(key=lambda r: r.get("start_time") or r.get("id")) | |
| return runs | |