speedclone.py

#!/usr/bin/env python3

# -- SpeedClone ------------------------------------------------- #
# speedclone.py on SpeedClone                                     #
# Made by DiamondGotCat, Licensed under MIT License               #
# Copyright (c) 2025 DiamondGotCat                                #
# ---------------------------------------------- DiamondGotCat -- #

from __future__ import annotations
import argparse, io, json, os, re, sys, tarfile, zipfile, time, subprocess, platform, shutil
from pathlib import Path
from urllib.parse import urlparse
from urllib.request import Request, urlopen
from rich import print

os_tmp = platform.system().strip().lower()
os_name = platform.system()[:4].ljust(5, "O").upper()
if os_tmp.startswith("linux"):
    os_name = "Linux"
elif os_tmp.startswith("windows"):
    os_name = "Microsoft Windows"
elif os_tmp.startswith("android"):
    os_name = "Android"
elif os_tmp.startswith("java"):
    os_name = "Java"
elif os_tmp.startswith("darwin"):
    os_name = "macOS"
elif os_tmp.startswith("ios"):
    os_name = "iOS"
elif os_tmp.startswith("ipados"):
    os_name = "iPadOS"

VERSION = "2.0"
UA = f"SpeedClone/{VERSION} (+https://github.com/DiamondGotCat/SpeedClone/) {os_tmp} ({os_name})"

def _http(url, accept=None, token=None):
    h = {"User-Agent": UA}
    if accept: h["Accept"] = accept
    if token: h["Authorization"] = f"Bearer {token}"
    req = Request(url, headers=h)
    return urlopen(req)

def _json(url, token=None):
    with _http(url, "application/vnd.github+json", token) as r:
        return json.loads(r.read().decode("utf-8"))

def _read(url):
    with _http(url) as r:
        return r.read()

def _owner_repo(url: str):
    u = urlparse(url)
    if "github.com" not in u.netloc.lower(): raise SystemExit("GitHub HTTPS URLのみ対応しています。")
    parts = [p for p in u.path.split("/") if p]
    if len(parts) < 2: raise SystemExit("不正なURLです。")
    owner, repo = parts[-2], parts[-1]
    if repo.endswith(".git"): repo = repo[:-4]
    return owner, repo

def _guess_default_sha_html(owner, repo, branch_hint=None):
    def scrape(path):
        html = _read(f"https://github.com/{owner}/{repo}/{path}").decode("utf-8","ignore")
        return html
    for b in ([branch_hint] if branch_hint else []) + ["main","master"]:
        try:
            html = scrape(f"commits/{b}")
            m = re.search(rf'href="/{re.escape(owner)}/{re.escape(repo)}/commit/([0-9a-f]{{40}})"', html, re.I)
            if m: return b, m.group(1)
        except Exception:
            pass
    for b in ([branch_hint] if branch_hint else []) + ["main","master"]:
        try:
            html = scrape(f"tree/{b}")
            m = re.search(r'data-test-selector="commit-tease-sha".*?>([0-9a-f]{7,40})<', html, re.I|re.S)
            if m: return b, m.group(1)
        except Exception:
            pass
    raise RuntimeError("HTMLから最新SHAを取得できませんでした。")

def _is_within(base: Path, target: Path) -> bool:
    try:
        base = base.resolve(strict=False)
        target = target.resolve(strict=False)
        return str(target) == str(base) or str(target).startswith(str(base) + os.sep)
    except Exception:
        return False

def _safe_symlink_supported() -> bool:
    if hasattr(os, "symlink"):
        if os.name == "nt":
            return True
        return True
    return False

def _make_symlink(dst: Path, link_target: str) -> bool:
    try:
        if os.path.isabs(link_target): return False
        norm = os.path.normpath(link_target)
        if norm.startswith(".."+os.sep) or norm == "..": return False
        dst.parent.mkdir(parents=True, exist_ok=True)
        os.symlink(link_target, dst)
        return True
    except Exception:
        return False

def _download_tar_stream(url: str, dst: Path, symlinks_mode: str):
    with _http(url) as resp:
        with tarfile.open(fileobj=resp, mode="r|gz") as tf:
            _extract_tar_to(tf, dst, symlinks_mode)

def _extract_tar_to(tf: tarfile.TarFile, dst: Path, symlinks_mode: str):
    prefix = None
    for m in tf.getmembers():
        if m.isdir():
            prefix = m.name.rstrip('/') + '/'
            break
    if prefix is None: raise RuntimeError("tarの解凍時にエラー: 構造が不正です。")
    for m in tf.getmembers():
        if not m.name.startswith(prefix): continue
        rel = m.name[len(prefix):]
        if not rel: continue
        out = (dst / rel).resolve(strict=False)
        if not _is_within(dst, out):
            continue

        if m.issym():
            if symlinks_mode == "skip":
                continue
            elif symlinks_mode == "text":
                out.parent.mkdir(parents=True, exist_ok=True)
                with open(out, "wb") as f:
                    f.write((m.linkname or "").encode("utf-8"))
            else:
                if _safe_symlink_supported() and _make_symlink(out, m.linkname or ""):
                    pass
                else:
                    out.parent.mkdir(parents=True, exist_ok=True)
                    with open(out, "wb") as f:
                        f.write((m.linkname or "").encode("utf-8"))
            continue

        if m.islnk():
            continue
        if m.isdir():
            out.mkdir(parents=True, exist_ok=True)
        elif m.isfile():
            out.parent.mkdir(parents=True, exist_ok=True)
            src = tf.extractfile(m)
            if src is None: continue
            with src, open(out, "wb") as f:
                shutil.copyfileobj(src, f, length=1024*1024)
            try:
                os.chmod(out, m.mode & 0o777)
            except Exception:
                pass

def _extract_zip_to(zf: zipfile.ZipFile, dst: Path, symlinks_mode: str):
    roots = sorted({name.split('/',1)[0]+'/' for name in zf.namelist() if '/' in name})
    prefix = roots[0] if roots else ''
    for name in zf.namelist():
        if prefix and not name.startswith(prefix):
            continue
        rel = name[len(prefix):]
        if not rel:
            continue
        out = (dst / rel).resolve(strict=False)
        if not _is_within(dst, out):
            continue
        if name.endswith('/'):
            out.mkdir(parents=True, exist_ok=True)
        else:
            out.parent.mkdir(parents=True, exist_ok=True)
            with zf.open(name) as src, open(out,"wb") as f:
                shutil.copyfileobj(src, f, length=1024*1024)
            try:
                info = zf.getinfo(name)
                perm = (info.external_attr >> 16) & 0o777
                if perm:
                    os.chmod(out, perm)
            except Exception:
                pass

def _download_snapshot(dst: Path, owner: str, repo: str, ref_sha: str, branch_name: str, symlinks_mode: str):
    errs = []
    try:
        _download_tar_stream(f"https://codeload.github.com/{owner}/{repo}/tar.gz/{ref_sha}", dst, symlinks_mode)
        return
    except Exception as e:
        errs.append(f"tar.gz(stream)@sha: {e}")
    try:
        _download_tar_stream(f"https://codeload.github.com/{owner}/{repo}/tar.gz/{branch_name}", dst, symlinks_mode)
        return
    except Exception as e:
        errs.append(f"tar.gz(stream)@branch: {e}")
    def fetch_bytes(url: str) -> bytes:
        data = _read(url)
        if not data: raise RuntimeError("空の応答が返されました。")
        return data

    for url, label in [
        (f"https://codeload.github.com/{owner}/{repo}/tar.gz/{ref_sha}", "tar.gz@sha"),
        (f"https://codeload.github.com/{owner}/{repo}/tar.gz/{branch_name}", "tar.gz@branch"),
    ]:
        try:
            data = fetch_bytes(url)
            with tarfile.open(fileobj=io.BytesIO(data), mode="r:gz") as tf:
                _extract_tar_to(tf, dst, symlinks_mode); return
        except Exception as e:
            errs.append(f"{label}: {e}")

    for url, label in [
        (f"https://codeload.github.com/{owner}/{repo}/zip/{ref_sha}", "zip@sha"),
        (f"https://codeload.github.com/{owner}/{repo}/zip/{branch_name}", "zip@branch"),
    ]:
        try:
            data = fetch_bytes(url)
            with zipfile.ZipFile(io.BytesIO(data)) as zf:
                _extract_zip_to(zf, dst, symlinks_mode); return
        except Exception as e:
            errs.append(f"{label}: {e}")

    raise RuntimeError("スナップショットのダウンロードに失敗しました: " + "; ".join(errs))

def _write_git_skeleton(dst: Path, remote_url: str, default_branch: str, head_sha: str):
    git = dst / ".git"
    (git / "refs/heads").mkdir(parents=True, exist_ok=True)
    (git / "refs/remotes/origin").mkdir(parents=True, exist_ok=True)
    (git / "objects/info").mkdir(parents=True, exist_ok=True)
    (git / "HEAD").write_text(f"ref: refs/heads/{default_branch}\n", encoding="utf-8")
    (git / "refs/heads" / default_branch).write_text(head_sha + "\n", encoding="ascii")
    (git / "refs/remotes/origin" / default_branch).write_text(head_sha + "\n", encoding="ascii")
    cfg = f"""
[core]
\trepositoryformatversion = 0
\tfilemode = true
\tbare = false
\tlogallrefupdates = true
[remote "origin"]
\turl = {remote_url}
\tfetch = +refs/heads/*:refs/remotes/origin/*
\tpromisor = true
\tpartialclonefilter = blob:none
[branch "{default_branch}"]
\tremote = origin
\tmerge = refs/heads/{default_branch}
[extensions]
\tpartialClone = origin
[fetch]
\tprune = true
""".lstrip()
    (git / "config").write_text(cfg, encoding="utf-8")
    (git / "objects/info/promisor").write_text("promisor\n", encoding="utf-8")

def _run_git(cwd: Path, *args: str, check=True, input_bytes: bytes|None=None, extra_env: dict|None=None):
    env = os.environ.copy()
    if extra_env: env.update(extra_env)
    cmd = ["git", "-c", "protocol.version=2"]
    cmd.extend(args)
    return subprocess.run(cmd, cwd=str(cwd), input=input_bytes, check=check, stdout=subprocess.PIPE, stderr=subprocess.PIPE)

def _get_tree_recursive(owner: str, repo: str, sha: str, token: str|None):
    def fetch_one(tree_sha: str):
        url = f"https://api.github.com/repos/{owner}/{repo}/git/trees/{tree_sha}?recursive=1"
        return _json(url, token)

    out = []
    root = fetch_one(sha)
    if not root or "tree" not in root: return out
    out.extend(root["tree"])
    if root.get("truncated"):
        queue = [t["sha"] for t in root["tree"] if t.get("type") == "tree"]
        seen = set(queue)
        while queue:
            t_sha = queue.pop()
            node = fetch_one(t_sha)
            for e in node.get("tree", []):
                out.append(e)
                if e.get("type") == "tree" and e["sha"] not in seen:
                    seen.add(e["sha"]); queue.append(e["sha"])
    return out

def _write_index_from_tree(dst: Path, entries: list[dict]):
    lines = []
    for e in entries:
        if e.get("type") != "blob":
            continue
        mode = e.get("mode", "100644")
        sha1 = e.get("sha")
        path = e.get("path")
        if not (mode and sha1 and path):
            continue
        if not (dst / path).exists():
            continue
        line = f"{mode} blob {sha1}\t{path}\0"
        lines.append(line)
    if not lines:
        return
    payload = "".join(lines).encode("utf-8", "surrogatepass")
    _run_git(dst, "update-index", "-z", "--index-info", input_bytes=payload)

def _prime_metadata(dst: Path, branch: str, depth: int, fetch_tags: bool):
    args = ["fetch", "-q", "--filter=tree:0", f"--depth={depth}"]
    if not fetch_tags:
        args.append("--no-tags")
    args.extend(["origin", f"{branch}:{branch}"])
    try:
        _run_git(dst, *args)
    except subprocess.CalledProcessError as e:
        print(f"[yellow](!)[/yellow] メタデータの事前取得に失敗しました: {e.stderr.decode('utf-8','ignore').strip()}")

def bootstrap(repo_url: str, dst: Path, depth: int, do_prime: bool, sparse: list[str], symlinks_mode: str, fetch_tags: bool):
    print("[blue](i)[/blue] 引数等の問題はありません。開始します...")
    owner, repo = _owner_repo(repo_url)
    print(f"[purple](i)[/purple] owner: {owner}, repo: {repo}")
    token = os.environ.get("GITHUB_TOKEN")
    default_branch = "master"
    head_sha = None
    try:
        meta = _json(f"https://api.github.com/repos/{owner}/{repo}", token)
        default_branch = meta.get("default_branch") or "master"
        tip = _json(f"https://api.github.com/repos/{owner}/{repo}/commits/{default_branch}", token)
        head_sha = tip["sha"]
        print(f"[purple](i)[/purple] default_branch: {default_branch}, head: {head_sha[:12]}… (APIで取得)")
    except Exception:
        print("[yellow](!)[/yellow] API経由でのダウンロード情報の取得に失敗しました。HTMLでの取得を試行します...")
        b, sha = _guess_default_sha_html(owner, repo)
        default_branch, head_sha = b, sha
        print(f"[purple](i)[/purple] default_branch: {default_branch}, head: {head_sha[:12]}… (HTMLで取得)")

    _write_git_skeleton(dst, repo_url, default_branch, head_sha)

    if sparse:
        try:
            _run_git(dst, "sparse-checkout", "init", "--cone")
            _run_git(dst, "sparse-checkout", "set", *sparse)
            print(f"[blue](i)[/blue] sparse-checkout を設定しました: {', '.join(sparse)}")
        except Exception as e:
            print(f"[yellow](!)[/yellow] sparse-checkout 設定に失敗しました: {e}")

    print("[blue](i)[/blue] スナップショットのダウンロードを開始します。")
    t0 = time.time()
    _download_snapshot(dst, owner, repo, head_sha, default_branch, symlinks_mode)
    print(f"[blue](i)[/blue] スナップショットはcodeloadを使用して{time.time()-t0:.1f}秒でダウンロードされました。")

    try:
        print("[blue](i)[/blue] インデックスを構築しています…")
        commit = _json(f"https://api.github.com/repos/{owner}/{repo}/git/commits/{head_sha}", token)
        tree_sha = commit["tree"]["sha"]
        entries = _get_tree_recursive(owner, repo, tree_sha, token)
        _write_index_from_tree(dst, entries)
    except Exception as e:
        print(f"[yellow](!)[/yellow] インデックス構築に失敗しました: {e}")

    if do_prime and depth > 0:
        _prime_metadata(dst, default_branch, depth, fetch_tags)

    print("[green](✓)[/green] 完了しました。")

def main():
    ap = argparse.ArgumentParser(prog="speedclone", description="ちょっと速い気がしなくもないgitのクローンツール")
    ap.add_argument("url")
    ap.add_argument("target")
    ap.add_argument("--force", action="store_true", help="既存ディレクトリを強制再作成")
    ap.add_argument("--depth", type=int, default=1, help="メタデータの浅い履歴（コミット/ツリーのみ）を取得する深さ（0で無効）")
    ap.add_argument("--no-prime", action="store_true", help="メタデータの極小フェッチを行わない（最速優先）")
    ap.add_argument("--sparse", type=str, default="", help='coneモードのsparseパス（空白区切り） 例: "src tools"')
    ap.add_argument("--symlinks", choices=["auto","skip","text"], default="auto", help="アーカイブ内のシンボリックリンクの扱い")
    ap.add_argument("--tags", action="store_true", help="タグも取得する（既定は取得しない）")

    args = ap.parse_args()
    dst = Path(args.target)
    if dst.exists():
        if not args.force:
            print(f"[yellow](!)[/yellow] ディレクトリが既に存在しています: {dst}", file=sys.stderr); sys.exit(1)
        for p in sorted(dst.rglob("*"), reverse=True):
            try:
                if p.is_file() or p.is_symlink(): p.unlink()
            except IsADirectoryError:
                pass
        for p in sorted(dst.rglob("*"), reverse=True):
            try: p.rmdir()
            except Exception: pass
    dst.mkdir(parents=True, exist_ok=True)
    try:
        sparse = [s for s in args.sparse.split() if s]
        bootstrap(
            args.url, dst,
            depth=max(args.depth, 0),
            do_prime=(not args.no_prime),
            sparse=sparse,
            symlinks_mode=args.symlinks,
            fetch_tags=args.tags,
        )
    except subprocess.CalledProcessError as e:
        print("[red](!)[/red] gitコマンドの実行に失敗しました:", e.stderr.decode("utf-8","ignore"), file=sys.stderr); sys.exit(3)
    except Exception as e:
        print("[red](!)[/red] 例外が発生しました:", e, file=sys.stderr); sys.exit(2)

if __name__ == "__main__":
    main()