taler-typescript-core

Wallet core logic and WebUIs for various components
Log | Files | Refs | Submodules | README | LICENSE

archive.py (4907B)


      1 #!/usr/bin/env python3
      2 
      3 # This file has been placed in the public domain.
      4 
      5 """Create a source archive containing this repository and its submodules."""
      6 
      7 from __future__ import annotations
      8 
      9 import argparse
     10 import os
     11 from pathlib import Path
     12 import subprocess
     13 import sys
     14 import tarfile
     15 
     16 
     17 def run_git(repo: Path, *args: str, input_data: bytes | None = None) -> bytes:
     18     try:
     19         return subprocess.run(
     20             ["git", *args],
     21             cwd=repo,
     22             input=input_data,
     23             check=True,
     24             stdout=subprocess.PIPE,
     25         ).stdout
     26     except subprocess.CalledProcessError as exc:
     27         command = " ".join(("git", *args))
     28         raise RuntimeError(f"'{command}' failed in {repo}") from exc
     29 
     30 
     31 def tracked_paths(repo: Path) -> list[Path]:
     32     output = run_git(repo, "ls-files", "-z", "--cached")
     33     return [
     34         Path(os.fsdecode(item)) for item in output.rstrip(b"\0").split(b"\0") if item
     35     ]
     36 
     37 
     38 def submodule_paths(repo: Path) -> set[Path]:
     39     gitmodules = repo / ".gitmodules"
     40     if not gitmodules.exists():
     41         return set()
     42     result = subprocess.run(
     43         [
     44             "git",
     45             "config",
     46             "-f",
     47             ".gitmodules",
     48             "--get-regexp",
     49             r"^submodule\..*\.path$",
     50         ],
     51         cwd=repo,
     52         check=False,
     53         stdout=subprocess.PIPE,
     54         text=True,
     55     )
     56     if result.returncode not in (0, 1):
     57         raise RuntimeError(f"could not read submodules from {gitmodules}")
     58     return {Path(line.split(maxsplit=1)[1]) for line in result.stdout.splitlines()}
     59 
     60 
     61 def export_ignored(repo: Path, paths: list[Path]) -> set[Path]:
     62     if not paths:
     63         return set()
     64     query = b"\0".join(os.fsencode(path) for path in paths) + b"\0"
     65     output = run_git(
     66         repo, "check-attr", "-z", "--stdin", "export-ignore", input_data=query
     67     )
     68     fields = output.rstrip(b"\0").split(b"\0")
     69     ignored: set[Path] = set()
     70     for index in range(0, len(fields), 3):
     71         if fields[index + 2] == b"set":
     72             ignored.add(Path(os.fsdecode(fields[index])))
     73     return ignored
     74 
     75 
     76 def repository_entries(repo: Path) -> list[tuple[Path, Path]]:
     77     """Return (filesystem path, archive-relative path) pairs for one repository."""
     78     submodules = submodule_paths(repo)
     79     candidates: list[tuple[Path, Path]] = []
     80 
     81     for relative in tracked_paths(repo):
     82         if relative not in submodules:
     83             candidates.append((repo / relative, relative))
     84 
     85     for submodule in sorted(submodules):
     86         submodule_dir = repo / submodule
     87         if not (submodule_dir / ".git").exists():
     88             raise RuntimeError(
     89                 f"submodule '{submodule}' is not initialized; run './bootstrap' first"
     90             )
     91         for source, relative in repository_entries(submodule_dir):
     92             candidates.append((source, submodule / relative))
     93 
     94     ignored = export_ignored(repo, [relative for _, relative in candidates])
     95     return [
     96         (source, relative) for source, relative in candidates if relative not in ignored
     97     ]
     98 
     99 
    100 def archive_prefix(output: Path) -> Path:
    101     name = output.name
    102     for suffix in (".tar.gz", ".tgz"):
    103         if name.endswith(suffix):
    104             prefix = name[: -len(suffix)]
    105             if prefix:
    106                 return Path(prefix)
    107             break
    108     raise ValueError("output file must end in .tar.gz or .tgz")
    109 
    110 
    111 def parse_args() -> argparse.Namespace:
    112     parser = argparse.ArgumentParser(description=__doc__)
    113     parser.add_argument(
    114         "--include",
    115         action="append",
    116         default=[],
    117         metavar="FILE",
    118         help="include an untracked or export-ignored file (may be repeated)",
    119     )
    120     parser.add_argument("output", type=Path, help="output .tar.gz file")
    121     return parser.parse_args()
    122 
    123 
    124 def main() -> int:
    125     args = parse_args()
    126     root = Path(run_git(Path.cwd(), "rev-parse", "--show-toplevel").decode().strip())
    127     output = args.output.resolve()
    128     prefix = archive_prefix(output)
    129     entries = repository_entries(root)
    130 
    131     included: list[tuple[Path, Path]] = []
    132     for name in args.include:
    133         relative = Path(name)
    134         if relative.is_absolute() or ".." in relative.parts:
    135             raise ValueError(
    136                 f"included path must be relative to the repository: {name}"
    137             )
    138         source = (root / relative).absolute()
    139         if not source.is_file():
    140             raise FileNotFoundError(f"included file does not exist: {name}")
    141         included.append((source, relative))
    142 
    143     with tarfile.open(output, "w:gz") as archive:
    144         for source, relative in [*included, *entries]:
    145             archive.add(source, arcname=prefix / relative, recursive=False)
    146 
    147     print(f"created {output}")
    148     return 0
    149 
    150 
    151 if __name__ == "__main__":
    152     try:
    153         sys.exit(main())
    154     except (OSError, RuntimeError, ValueError) as exc:
    155         print(f"archive: {exc}", file=sys.stderr)
    156         sys.exit(1)