"""Bounded public LeRobot audit. No video files or ambient HF credentials.""" import json import os from pathlib import Path, PurePosixPath import re import subprocess import sys import tempfile MAX_BYTES = 128 * 1024 * 1024 MAX_FILES = 256 MAX_EPISODES = 500 TIMEOUT = 180 REF = re.compile(r"[A-Za-z0-9][A-Za-z0-9_.-]*/[A-Za-z0-9][A-Za-z0-9_.-]*\Z") def parse_ref(raw): ref = (raw or '').strip().removeprefix('https://huggingface.co/datasets/').rstrip('/') if not REF.fullmatch(ref) or '..' in ref: raise ValueError('Enter a public dataset as org/name or its HF dataset URL.') return ref def allowed_file(name): path = PurePosixPath(name) return (not path.is_absolute() and '..' not in path.parts and ((name.startswith('data/') and name.endswith('.parquet')) or (name.startswith('meta/') and path.suffix in {'.json', '.jsonl', '.parquet'}))) def check_metadata(meta): if not str(meta.get('codebase_version', '')).startswith(('v2', 'v3')): raise ValueError('Quick mode supports LeRobot v2/v3. Use the CLI for this format.') if not 0 < meta.get('total_episodes', 0) <= MAX_EPISODES: raise ValueError(f'Quick mode accepts up to {MAX_EPISODES} episodes. Use the CLI for this dataset.') if any(f.get('dtype') == 'image' for f in meta.get('features', {}).values()): raise ValueError('Embedded images require a CLI audit; quick mode excludes media.') for key in ('data_path', 'video_path'): path = str(meta.get(key) or '') if '://' in path or PurePosixPath(path).is_absolute() or '..' in PurePosixPath(path).parts: raise ValueError('Dataset contains unsupported file paths. Use the CLI.') def grade_local(root, ref, revision, license_name, byte_count, file_count): from kalanos import grade, SourceInfo from publisher_grader.convert import convert from publisher_grader.provenance import Provenance # All selected numeric shards are assessed; never silently sample episodes. report = grade(root, bundle='numeric-core-v1', sidecar=False, full_frame_scan=False) report.source = SourceInfo(uri=f'hf://datasets/{ref}@{revision}', protocol='hf', repo_id=ref, revision=revision, size_bytes=byte_count, file_count=file_count) report.run.source.revision = revision card = convert(report, Provenance(kind='hf', ref=ref, url=f'https://huggingface.co/datasets/{ref}', license=license_name or 'unknown', gated=False)).model_dump(mode='json') card['report_id'] = report.run.id card['notes'].append('HF Space quick audit: all downloaded numeric shards assessed; ' 'video and image files excluded. Video: not assessed. ' 'Storage size describes downloaded numeric/metadata files, not the full HF repository.') return card def worker(ref, destination): import requests from huggingface_hub import HfApi, hf_hub_url api = HfApi(token=False) info = api.dataset_info(ref, token=False) if info.private or info.gated: raise ValueError('Quick mode supports public, ungated datasets. Use the CLI for private or gated data.') revision = info.sha if not revision: raise ValueError('Could not resolve a dataset revision.') selected = [] # Cap listing work too; a huge media tree should not exhaust Space memory. for index, entry in enumerate(api.list_repo_tree(ref, repo_type='dataset', revision=revision, recursive=True, token=False)): if index >= 10000: raise ValueError('Dataset tree exceeds the quick-mode limit. Use the CLI.') if allowed_file(entry.path) and hasattr(entry, 'size'): selected.append(entry) if len(selected) > MAX_FILES or sum(e.size for e in selected) > MAX_BYTES: raise ValueError('Numeric data exceeds 128 MiB / 256 files. Use the CLI.') if not any(e.path == 'meta/info.json' for e in selected): raise ValueError('No LeRobot meta/info.json found.') with tempfile.TemporaryDirectory(prefix='kalanos-numeric-') as tmp: root = Path(tmp) downloaded = 0 # Validate the manifest before downloading shards. selected.sort(key=lambda e: e.path != 'meta/info.json') for entry in selected: target = root / entry.path target.parent.mkdir(parents=True, exist_ok=True) with requests.get(hf_hub_url(ref, entry.path, repo_type='dataset', revision=revision), stream=True, timeout=(10, 25)) as response: response.raise_for_status() with target.open('wb') as output: for chunk in response.iter_content(1024 * 1024): downloaded += len(chunk) if downloaded > MAX_BYTES: raise ValueError('Download exceeds the quick-mode limit. Use the CLI.') output.write(chunk) if entry.path == 'meta/info.json': check_metadata(json.loads(target.read_text())) card_data = info.card_data or {} card = grade_local(root, ref, revision, card_data.get('license', 'unknown'), downloaded, len(selected)) Path(destination).write_text(json.dumps(card)) def quick_grade(ref): ref = parse_ref(ref) with tempfile.TemporaryDirectory(prefix='kalanos-job-') as tmp: output = Path(tmp) / 'card.json' env = {k: v for k, v in os.environ.items() if not k.startswith(('HF_', 'HUGGING_FACE_', 'KALANOS_')) and k != 'INGEST_API_KEY'} env.update(HF_HUB_DISABLE_IMPLICIT_TOKEN='1', HF_HOME=tmp, KALANOS_PLUGIN_DIR=str(Path(tmp) / 'plugins'), OMP_NUM_THREADS='2') try: result = subprocess.run([sys.executable, str(Path(__file__).resolve()), ref, str(output)], env=env, capture_output=True, text=True, timeout=TIMEOUT) except subprocess.TimeoutExpired: raise ValueError('Quick audit exceeded 3 minutes. Run this dataset with the CLI.') from None if result.returncode: # Only intentional validation errors are user-facing; never expose tracebacks or paths. error_file = Path(str(output) + '.error') raise ValueError(error_file.read_text() if error_file.exists() else 'The quick audit could not complete. Use the CLI for details.') return json.loads(output.read_text()) if __name__ == '__main__': try: worker(parse_ref(sys.argv[1]), sys.argv[2]) except ValueError as exc: Path(sys.argv[2] + '.error').write_text(str(exc)) sys.exit(1)