| """Use pinned upstream preprocessing while excluding every known v1 asset/group.""" |
| import importlib.util |
| import json |
| from pathlib import Path |
| from audit_data import audit, identity_sets, conflicts |
|
|
| ROOT = Path(__file__).resolve().parent |
|
|
| def main(): |
| upstream = ROOT / 'upstream' |
| old = json.loads((upstream / 'ATTRIBUTIONS.json').read_text()) |
| blocked = identity_sets(old) |
| spec = importlib.util.spec_from_file_location('pinned_prepare', upstream / 'scripts/prepare_hf.py') |
| module = importlib.util.module_from_spec(spec) |
| spec.loader.exec_module(module) |
| for name in ('scenario_candidates', 'objaverse_candidates'): |
| original = getattr(module, name) |
| def filtered(*args, _original=original, **kwargs): |
| return [r for r in _original(*args, **kwargs) if not conflicts(r, blocked)] |
| setattr(module, name, filtered) |
| downloader = module.Downloader |
| class FilteredDownloader(downloader): |
| def get(self, *args, **kwargs): |
| path = super().get(*args, **kwargs) |
| if module.digest(path) in blocked['sha256']: |
| raise ValueError('Excluded byte-identical v1 mesh') |
| return path |
| module.Downloader = FilteredDownloader |
| original_prepare = module.prepare |
| def checked_prepare(args): |
| out = Path(args.out) |
| if (out / 'sources.json').exists(): |
| for row in json.loads((out / 'sources.json').read_text()): |
| if conflicts(row, blocked): |
| raise ValueError('Existing dataset overlaps v1; choose a fresh --out directory') |
| original_prepare(args) |
| audit(out, upstream / 'ATTRIBUTIONS.json') |
| module.prepare = checked_prepare |
| module.main() |
|
|
| if __name__ == '__main__': |
| main() |
|
|
