Initial commit

Sources.py

@@ -0,0 +1,112 @@
+"""Getting program source from a git host (GitHub now, Gitea later) over
+plain HTTPS, standard library only, so the AppImage needs no git binary.
+
+- The latest commit comes from the smart-HTTP ref advertisement
+  (<repo>.git/info/refs?service=git-upload-pack), which GitHub and Gitea
+  serve identically, and which is what `git ls-remote` reads too.
+- The files come from <repo>/archive/<commit>.tar.gz, also the same on both.
+"""
+
+import shutil
+import tarfile
+import tempfile
+import urllib.error
+import urllib.request
+from pathlib import Path
+from typing import Callable
+
+from Core import AppName, DownloadsDir
+
+_Timeout = 20
+
+
+class SourceError(Exception):
+    pass
+
+
+def _Get(Url: str) -> bytes:
+    Request = urllib.request.Request(Url, headers={"User-Agent": f"{AppName} (git/2)"})
+    try:
+        with urllib.request.urlopen(Request, timeout=_Timeout) as Response:
+            return Response.read()
+    except urllib.error.HTTPError as Error:
+        if Error.code in (401, 403, 404):
+            # Hosts answer 401 rather than 404 for missing repos, so it's
+            # impossible to tell "private" from "doesn't exist" apart.
+            raise SourceError(f"{Url.split('.git/info/refs')[0].split('/archive/')[0]} doesn't exist or is private") from Error
+        raise SourceError(f"Couldn't reach {Url}: {Error}") from Error
+    except OSError as Error:  # URLError and timeouts are OSErrors too
+        raise SourceError(f"Couldn't reach {Url}: {Error}") from Error
+
+
+def _RepoUrl(GitUrl: str) -> str:
+    return GitUrl.rstrip("/").removesuffix(".git")
+
+
+def _PktLines(Data: bytes) -> list[bytes]:
+    """Split git's pkt-line format: 4 hex digits of length (including
+    themselves), "0000" being a flush packet between sections."""
+    Lines, Position = [], 0
+    while Position + 4 <= len(Data):
+        Length = int(Data[Position:Position + 4], 16)
+        if Length == 0:
+            Position += 4
+            continue
+        Lines.append(Data[Position + 4:Position + Length].rstrip(b"\n"))
+        Position += Length
+    return Lines
+
+
+def LatestCommit(GitUrl: str, Branch: str = "") -> str:
+    """The commit a branch points at (the repo's default branch if Branch is empty)."""
+    Data = _Get(f"{_RepoUrl(GitUrl)}.git/info/refs?service=git-upload-pack")
+    Refs: dict[str, str] = {}
+    DefaultBranch = ""
+    for Line in _PktLines(Data):
+        if Line.startswith(b"#"):
+            continue
+        RefPart, _, Capabilities = Line.partition(b"\0")
+        Sha, _, RefName = RefPart.decode().partition(" ")
+        if Sha.strip("0"):  # an empty repo advertises one all-zero "capabilities^{}" line
+            Refs[RefName] = Sha
+        for Capability in Capabilities.decode().split():
+            if Capability.startswith("symref=HEAD:"):
+                DefaultBranch = Capability.removeprefix("symref=HEAD:")
+    if not Refs:
+        raise SourceError(f"{_RepoUrl(GitUrl)} is empty (nothing pushed yet)")
+    Wanted = f"refs/heads/{Branch}" if Branch else (DefaultBranch or "HEAD")
+    if Wanted not in Refs:
+        raise SourceError(f"{GitUrl} has no branch {Branch or 'HEAD'}")
+    return Refs[Wanted]
+
+
+def Download(Name: str, GitUrl: str, Commit: str, Log: Callable[[str], None]) -> Path:
+    """Extract one commit's files into the download cache (reused if it's
+    already there) and drop older downloads of the same program."""
+    Target = DownloadsDir / f"{Name}-{Commit[:12]}"
+    if Target.is_dir():
+        return Target
+
+    Url = f"{_RepoUrl(GitUrl)}/archive/{Commit}.tar.gz"
+    Log(f"Downloading {Url}")
+    DownloadsDir.mkdir(parents=True, exist_ok=True)
+    with tempfile.TemporaryDirectory(dir=DownloadsDir) as Temporary:
+        Archive = Path(Temporary) / "Source.tar.gz"
+        Archive.write_bytes(_Get(Url))
+        Extracted = Path(Temporary) / "Extracted"
+        with tarfile.open(Archive) as Tar:
+            Tar.extractall(Extracted, filter="data")
+        # Archives wrap everything in one "<repo>-<commit>/" folder.
+        Inner = list(Extracted.iterdir())
+        Root = Inner[0] if len(Inner) == 1 and Inner[0].is_dir() else Extracted
+        Root.rename(Target)
+
+    for Old in DownloadsDir.glob(f"{Name}-*"):
+        if Old != Target:
+            shutil.rmtree(Old, ignore_errors=True)
+    return Target
+
+
+def ClearDownloads(Name: str) -> None:
+    for Old in DownloadsDir.glob(f"{Name}-*"):
+        shutil.rmtree(Old, ignore_errors=True)