hand-motion-pipeline/scripts/package_source_release.py

"""Build a source-only, history-free snapshot for an internal Git platform."""
import argparse
import hashlib
import json
from pathlib import Path
import re
import shutil
import subprocess
import tarfile

ROOT = Path(__file__).resolve().parents[1]
EXTENSIONS = {'.py','.pyi','.sh','.bash','.md','.rst','.txt','.json','.yaml','.yml',
              '.toml','.cfg','.ini','.xml','.urdf','.cpp','.cc','.c','.h','.hpp','.cu',
              '.cuh','.cmake','.in','.lock','.gitignore','.gitmodules','.bib','.css',
              '.html','.js','.ipynb','.pxd','.pyx','.bat','.ps1','.m','.glsl','.vert','.frag'}
SPECIAL = {'LICENSE','LICENCE','COPYING','NOTICE','AUTHORS','Makefile','CMakeLists.txt',
           'Dockerfile','.gitignore','.gitattributes','environment.yml'}
EXCLUDE_PARTS = {'.git','__pycache__','.dynhamr','.venv','venv','outputs','output',
                 '_DATA','checkpoints','weights','.pytest_cache','.mypy_cache'}


def git(repo, *args):
    return subprocess.check_output(['git','-C',str(repo),*args], stderr=subprocess.DEVNULL).decode().strip()


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--output', type=Path, default=ROOT/'dist/hand-motion-pipeline')
    args = parser.parse_args()
    dest = args.output.resolve()
    if dest.exists():
        raise SystemExit(f'Refusing to overwrite existing release: {dest}')
    dest.mkdir(parents=True)
    excluded, repos, portability = [], [], []

    def copy_file(src, relative):
        reason = None
        if src.is_symlink():
            reason = 'symlink: restore dependency explicitly'
        elif any(p in EXCLUDE_PARTS for p in relative.parts):
            reason = 'runtime/data/cache directory'
        elif relative.parts[0] == 'third_party' and ('build' in relative.parts or '.cache' in relative.parts or src.suffix == '.part' or any(x.endswith('.egg-info') for x in relative.parts)):
            reason = 'build artifact or partial download'
        elif relative.name == '.gitmodules' or (relative.parts[0] == 'third_party' and relative.name in {'.gitignore','.gitattributes'}):
            reason = 'vendored snapshot: omit nested Git filters and metadata'
        elif src.suffix not in EXTENSIONS and src.name not in SPECIAL and not src.name.startswith(('LICENSE','COPYING','NOTICE')) \
                and not (relative.parts[0] == 'docs' and src.suffix.lower() == '.stl' and src.stat().st_size < 1024*1024):
            reason = 'non-source file; restore separately when needed'
        elif src.stat().st_size > 5*1024*1024:
            reason = 'file exceeds 5 MiB source limit'
        if reason:
            excluded.append({'path':relative.as_posix(),'reason':reason})
            return
        raw = src.read_bytes()
        if b'\0' in raw and src.suffix.lower() != '.stl':
            excluded.append({'path':relative.as_posix(),'reason':'binary contents'})
            return
        target = dest/relative
        target.parent.mkdir(parents=True, exist_ok=True)
        shutil.copy2(src,target)
        if src.suffix in {'.py','.sh','.yaml','.yml','.json','.toml'}:
            for i,line in enumerate(raw.decode('utf-8',errors='replace').splitlines(),1):
                if re.search(r'/home/|/tmp/|2047635068|15886123|20260915_171525|20260916_104026',line):
                    portability.append({'path':relative.as_posix(),'line':i,
                                        'kind':'machine path or fixed example identifier'})

    def snapshot(repo, relative, ancestors=()):
        real = repo.resolve()
        if real in ancestors:
            raise RuntimeError(f'Recursive repository: {repo}')
        records = subprocess.check_output(['git','-C',str(repo),'ls-files','--stage','-z']).split(b'\0')
        repos.append({'path':relative.as_posix(),'commit':git(repo,'rev-parse','HEAD'),
                      'origin':git(repo,'remote','get-url','origin'),
                      'local_changes':git(repo,'status','--porcelain'),
                      'snapshot':'working tree content; tracked files plus source additions'})
        visited = set()
        for record in records:
            if not record:
                continue
            metadata,name = record.decode().split('\t',1)
            visited.add(name)
            src = repo/name
            if metadata.startswith('160000'):
                if src.exists() and (src/'.git').exists():
                    snapshot(src,relative/name,(*ancestors,real))
                else:
                    excluded.append({'path':(relative/name).as_posix(),'reason':'unavailable submodule; restore pinned dependency', 'commit':metadata.split()[1]})
            elif src.is_file() or src.is_symlink():
                copy_file(src,relative/name)
        additions = subprocess.check_output(['git','-C',str(repo),'ls-files','--others','--exclude-standard','-z']).split(b'\0')
        for item in additions:
            if item:
                name=item.decode()
                if name not in visited and (repo/name).is_file():
                    copy_file(repo/name,relative/name)

    for folder in ['scripts','docs','model','utils','preprocessing','visualization','configs','requirements']:
        for src in sorted((ROOT/folder).rglob('*')):
            if src.is_file():
                copy_file(src,src.relative_to(ROOT))
    for name in ['README.md','LICENSE','.gitignore','requirements.txt','setup_env.sh','setup_vipe_env.sh']:
        copy_file(ROOT/name,Path(name))
    for name in ['hamer','vipe','Dyn-HaMR','spider','FoundationPose','nvdiffrast']:
        snapshot(ROOT/'third_party'/name,Path('third_party')/name)
    repos.insert(0,{'path':'.','commit':git(ROOT,'rev-parse','HEAD'),
                    'origin':git(ROOT,'remote','get-url','origin'),
                    'snapshot':'selected integration source and documentation from current working tree'})
    # Broad upstream patterns such as data/ would hide vendored Python source.
    (dest/'.gitignore').write_text(
        '# Source snapshot: runtime exclusions anchored to repository root.\n'
        '/output/\n/results/\n/logs/\n/weights/\n/dist/\n/.venv/\n/venv/\n'
        '/.dex/\n/.spider/\n/.cuda/\n/.hf_cache/\n/.torch_cache/\n/.spider_cache/\n'
        '/.uv_cache_spider/\n/.env\n__pycache__/\n*.pyc\n'
        '/third_party/FoundationPose/weights/\n/third_party/*/build/\n/.cache/\n*.part\n'
        '/docs/20260915_171525/\n/docs/20260916_104026/\n')
    assets=[]
    for rel in ['third_party/l20_assets','third_party/bottle_model_parametric','third_party/dex-assets',
                'third_party/FoundationPose/weights','docs/20260915_171525']:
        folder=ROOT/rel
        files=[{'path':p.relative_to(ROOT).as_posix(),'bytes':p.stat().st_size,
                'sha256':hashlib.sha256(p.read_bytes()).hexdigest()}
               for p in sorted(folder.rglob('*')) if p.is_file() and not p.is_symlink()]
        assets.append({'path':folder.relative_to(ROOT).as_posix(),'bundled':False,'files':files})
    manifests={'UPSTREAM_SOURCES.json':repos,'EXTERNAL_ASSETS.json':assets,
               'EXCLUDED_FILES.json':excluded,'PORTABILITY_REPORT.json':portability}
    for name,content in manifests.items():
        (dest/name).write_text(json.dumps(content,indent=2,ensure_ascii=False)+'\n')
    # Validate local Python syntax without importing scripts (some execute on import).
    checked=[]
    for folder in ['scripts','model','utils','preprocessing','visualization']:
        for p in (dest/folder).rglob('*.py'):
            compile(p.read_bytes(),str(p),'exec')
            checked.append(p.relative_to(dest).as_posix())
    for p in (dest/'scripts').glob('*.sh'):
        subprocess.run(['bash','-n',str(p)],check=True)
    allfiles=[p for p in dest.rglob('*') if p.is_file()]
    assert not any(p.is_symlink() or '.git' in p.relative_to(dest).parts for p in allfiles)
    report={'local_python_syntax_checked':len(checked),'shell_syntax':'PASS',
            'source_files':len(allfiles),'bytes':sum(p.stat().st_size for p in allfiles),
            'portability_findings':len(portability),'excluded_files':len(excluded),
            'end_to_end_reconstruction_rerun':False,'remote_upload_performed':False}
    (dest/'RELEASE_VALIDATION.json').write_text(json.dumps(report,indent=2)+'\n')
    hashes=[f'{hashlib.sha256(p.read_bytes()).hexdigest()}  {p.relative_to(dest).as_posix()}'
            for p in sorted(dest.rglob('*')) if p.is_file()]
    (dest/'SHA256SUMS').write_text('\n'.join(hashes)+'\n')
    archive=dest.with_suffix('.tar.gz')
    with tarfile.open(archive,'w:gz') as tar:
        tar.add(dest,arcname=dest.name)
    with tarfile.open(archive) as tar:
        members=tar.getmembers()
        assert all(not m.issym() and not m.islnk() and '..' not in Path(m.name).parts for m in members)
    print(json.dumps({**report,'directory':str(dest),'archive':str(archive),
                      'archive_bytes':archive.stat().st_size},indent=2))


if __name__ == '__main__':
    main()