|
#!/usr/bin/env python3 |
|
"""Bulk-add local git repos to the GitHub Copilot desktop app. |
|
|
|
Observed against the local ~/.copilot/data.db schema where repo registration lives in: |
|
- projects |
|
- project_checkouts |
|
|
|
Usage: |
|
python3 copilot_bulk_add_projects.py /path/to/folder/of/repos |
|
python3 copilot_bulk_add_projects.py /path/to/folder/of/repos --dry-run |
|
|
|
Notes: |
|
- Close GitHub Copilot before running this, or SQLite may be locked. |
|
- A timestamped backup of the DB is created automatically unless --dry-run is used. |
|
- Direct child directories with a .git folder are treated as candidate repos. |
|
""" |
|
|
|
from __future__ import annotations |
|
|
|
import argparse |
|
import collections |
|
import re |
|
import sqlite3 |
|
import subprocess |
|
import sys |
|
import uuid |
|
from datetime import datetime |
|
from pathlib import Path |
|
from typing import Iterable |
|
|
|
GITHUB_REMOTE_RE = re.compile(r"github\.com[:/]([^/]+)/([^/]+?)(?:\.git)?$") |
|
DEFAULT_DB = Path.home() / ".copilot" / "data.db" |
|
|
|
|
|
class BulkAddError(RuntimeError): |
|
pass |
|
|
|
|
|
def parse_args() -> argparse.Namespace: |
|
parser = argparse.ArgumentParser(description=__doc__) |
|
parser.add_argument("root", nargs="?", type=Path, help="Folder whose direct children are local repos") |
|
parser.add_argument("--db", type=Path, default=DEFAULT_DB, help=f"Copilot SQLite DB (default: {DEFAULT_DB})") |
|
parser.add_argument("--host-id", default="local", help="Value for project_checkouts.host_id (default: local)") |
|
parser.add_argument("--dry-run", action="store_true", help="Print what would be added, but do not change the DB") |
|
parser.add_argument( |
|
"--include-duplicate-remotes", |
|
action="store_true", |
|
help="Do not auto-skip secondary clones of the same GitHub remote", |
|
) |
|
return parser.parse_args() |
|
|
|
|
|
def git(repo: Path, *args: str) -> str: |
|
return subprocess.check_output( |
|
["git", "-C", str(repo), *args], |
|
text=True, |
|
stderr=subprocess.DEVNULL, |
|
).strip() |
|
|
|
|
|
def parse_github_remote(url: str) -> tuple[str | None, str | None]: |
|
match = GITHUB_REMOTE_RE.search(url.strip()) if url else None |
|
if not match: |
|
return None, None |
|
owner, repo = match.groups() |
|
return owner, repo |
|
|
|
|
|
def get_origin_remote(repo: Path) -> str: |
|
try: |
|
return git(repo, "config", "--get", "remote.origin.url") |
|
except Exception: |
|
return "" |
|
|
|
|
|
def default_branch(repo: Path) -> str: |
|
for args in ( |
|
("symbolic-ref", "--short", "refs/remotes/origin/HEAD"), |
|
("branch", "--show-current"), |
|
("rev-parse", "--abbrev-ref", "HEAD"), |
|
): |
|
try: |
|
value = git(repo, *args) |
|
except Exception: |
|
continue |
|
if value: |
|
return value.split("/", 1)[1] if value.startswith("origin/") else value |
|
return "main" |
|
|
|
|
|
def find_git_repos(root: Path) -> list[Path]: |
|
if not root.exists(): |
|
raise BulkAddError(f"Root does not exist: {root}") |
|
if not root.is_dir(): |
|
raise BulkAddError(f"Root is not a directory: {root}") |
|
return sorted(path.resolve() for path in root.iterdir() if path.is_dir() and (path / ".git").exists()) |
|
|
|
|
|
def duplicate_remote_skips(repos: Iterable[Path]) -> dict[Path, str]: |
|
by_remote: dict[tuple[str, str], list[Path]] = collections.defaultdict(list) |
|
for repo in repos: |
|
owner, name = parse_github_remote(get_origin_remote(repo)) |
|
if owner and name: |
|
by_remote[(owner.lower(), name.lower())].append(repo) |
|
|
|
skipped: dict[Path, str] = {} |
|
for (_, repo_name), clones in by_remote.items(): |
|
if len(clones) < 2: |
|
continue |
|
clones = sorted( |
|
clones, |
|
key=lambda path: ( |
|
path.name.lower() != repo_name, |
|
path.name.lower().endswith("-bak"), |
|
len(path.name), |
|
path.name.lower(), |
|
), |
|
) |
|
keep = clones[0] |
|
for clone in clones[1:]: |
|
skipped[clone] = f"duplicate remote of {keep.name}" |
|
return skipped |
|
|
|
|
|
def latest_project_defaults(con: sqlite3.Connection) -> dict[str, object]: |
|
row = con.execute("SELECT * FROM projects ORDER BY created_at DESC LIMIT 1").fetchone() |
|
if row is None: |
|
return { |
|
"setup_script": "", |
|
"run_script": "", |
|
"scripts_json": "", |
|
"archive_script": "", |
|
"pull_request_prompt": "", |
|
"instructions": "", |
|
"remote_control_enabled": 0, |
|
"server_ready_pattern": "", |
|
"auto_open_in_browser": 1, |
|
"auto_approve": 1, |
|
"trusted_config_sha256": None, |
|
"trusted_config_accepted_at": None, |
|
"branch_prefix": None, |
|
"sparse_checkout": None, |
|
"sandbox_enabled": 0, |
|
"reusable_worktrees_enabled": 0, |
|
"reusable_worktrees_max_count": 2, |
|
"reusable_worktrees_expiry_days": 7, |
|
"checkout_is_managed": 0, |
|
"reusable_worktrees_user_enabled": 1, |
|
"show_in_sidebar": 1, |
|
} |
|
return {key: row[key] for key in ( |
|
"setup_script", |
|
"run_script", |
|
"scripts_json", |
|
"archive_script", |
|
"pull_request_prompt", |
|
"instructions", |
|
"remote_control_enabled", |
|
"server_ready_pattern", |
|
"auto_open_in_browser", |
|
"auto_approve", |
|
"trusted_config_sha256", |
|
"trusted_config_accepted_at", |
|
"branch_prefix", |
|
"sparse_checkout", |
|
"sandbox_enabled", |
|
"reusable_worktrees_enabled", |
|
"reusable_worktrees_max_count", |
|
"reusable_worktrees_expiry_days", |
|
"checkout_is_managed", |
|
"reusable_worktrees_user_enabled", |
|
"show_in_sidebar", |
|
)} |
|
|
|
|
|
def default_account_id(con: sqlite3.Connection) -> str | None: |
|
row = con.execute( |
|
"SELECT id FROM accounts WHERE host='github.com' AND is_default=1 ORDER BY created_at DESC LIMIT 1" |
|
).fetchone() |
|
return None if row is None else row["id"] |
|
|
|
|
|
def backup_path(db: Path) -> Path: |
|
stamp = datetime.now().strftime("%Y%m%d-%H%M%S") |
|
return db.with_name(f"{db.name}.bak.bulk-add-{stamp}") |
|
|
|
|
|
def make_backup(con: sqlite3.Connection, db: Path) -> Path: |
|
target = backup_path(db) |
|
bcon = sqlite3.connect(target) |
|
try: |
|
con.backup(bcon) |
|
finally: |
|
bcon.close() |
|
return target |
|
|
|
|
|
def insert_project( |
|
con: sqlite3.Connection, |
|
repo: Path, |
|
host_id: str, |
|
account_id: str | None, |
|
defaults: dict[str, object], |
|
) -> None: |
|
remote = get_origin_remote(repo) |
|
owner, github_repo = parse_github_remote(remote) |
|
project_id = str(uuid.uuid4()) |
|
con.execute( |
|
""" |
|
INSERT INTO projects ( |
|
id, name, container_kind, main_repo_path, default_branch, |
|
github_owner, github_repo, github_account_id, |
|
setup_script, run_script, scripts_json, archive_script, |
|
pull_request_prompt, instructions, |
|
remote_control_enabled, server_ready_pattern, |
|
auto_open_in_browser, auto_approve, |
|
trusted_config_sha256, trusted_config_accepted_at, |
|
branch_prefix, sparse_checkout, |
|
sandbox_enabled, reusable_worktrees_enabled, |
|
reusable_worktrees_max_count, reusable_worktrees_expiry_days, |
|
checkout_is_managed, reusable_worktrees_user_enabled, show_in_sidebar |
|
) VALUES ( |
|
?, ?, 'repository', ?, ?, |
|
?, ?, ?, |
|
?, ?, ?, ?, |
|
?, ?, |
|
?, ?, |
|
?, ?, |
|
?, ?, |
|
?, ?, |
|
?, ?, |
|
?, ?, |
|
?, ?, ? |
|
) |
|
""", |
|
( |
|
project_id, |
|
repo.name, |
|
str(repo), |
|
default_branch(repo), |
|
owner, |
|
github_repo, |
|
account_id if owner and github_repo else None, |
|
defaults["setup_script"], |
|
defaults["run_script"], |
|
defaults["scripts_json"], |
|
defaults["archive_script"], |
|
defaults["pull_request_prompt"], |
|
defaults["instructions"], |
|
defaults["remote_control_enabled"], |
|
defaults["server_ready_pattern"], |
|
defaults["auto_open_in_browser"], |
|
defaults["auto_approve"], |
|
defaults["trusted_config_sha256"], |
|
defaults["trusted_config_accepted_at"], |
|
defaults["branch_prefix"], |
|
defaults["sparse_checkout"], |
|
defaults["sandbox_enabled"], |
|
defaults["reusable_worktrees_enabled"], |
|
defaults["reusable_worktrees_max_count"], |
|
defaults["reusable_worktrees_expiry_days"], |
|
defaults["checkout_is_managed"], |
|
defaults["reusable_worktrees_user_enabled"], |
|
defaults["show_in_sidebar"], |
|
), |
|
) |
|
con.execute( |
|
"INSERT INTO project_checkouts (project_id, host_id, root_resource_uri) VALUES (?, ?, ?)", |
|
(project_id, host_id, repo.as_uri()), |
|
) |
|
|
|
|
|
def run(root: Path, db: Path, host_id: str, dry_run: bool, include_duplicate_remotes: bool) -> int: |
|
repos = find_git_repos(root) |
|
if not repos: |
|
raise BulkAddError(f"No git repos found under: {root}") |
|
|
|
con = sqlite3.connect(db, timeout=30) |
|
con.row_factory = sqlite3.Row |
|
try: |
|
con.execute("PRAGMA foreign_keys = ON") |
|
con.execute("PRAGMA busy_timeout = 30000") |
|
|
|
existing_paths = {row["main_repo_path"] for row in con.execute("SELECT main_repo_path FROM projects")} |
|
defaults = latest_project_defaults(con) |
|
account_id = default_account_id(con) |
|
auto_skips = {} if include_duplicate_remotes else duplicate_remote_skips(repos) |
|
|
|
added: list[Path] = [] |
|
skipped: list[tuple[Path, str]] = [] |
|
no_remote: list[Path] = [] |
|
|
|
if dry_run: |
|
for repo in repos: |
|
if repo in auto_skips: |
|
skipped.append((repo, auto_skips[repo])) |
|
continue |
|
if str(repo) in existing_paths: |
|
skipped.append((repo, "already present")) |
|
continue |
|
if not get_origin_remote(repo): |
|
no_remote.append(repo) |
|
added.append(repo) |
|
print("dry_run=yes") |
|
print(f"would_add={len(added)}") |
|
for repo in added: |
|
print(f"+ {repo}") |
|
print(f"skipped={len(skipped)}") |
|
for repo, reason in skipped: |
|
print(f"- {repo}: {reason}") |
|
if no_remote: |
|
print(f"no_origin_remote={len(no_remote)}") |
|
for repo in no_remote: |
|
print(f"! {repo}") |
|
return 0 |
|
|
|
backup = make_backup(con, db) |
|
try: |
|
con.execute("BEGIN IMMEDIATE") |
|
except sqlite3.OperationalError as exc: |
|
raise BulkAddError("Could not lock the Copilot DB. Close GitHub Copilot and try again.") from exc |
|
|
|
for repo in repos: |
|
if repo in auto_skips: |
|
skipped.append((repo, auto_skips[repo])) |
|
continue |
|
if str(repo) in existing_paths: |
|
skipped.append((repo, "already present")) |
|
continue |
|
if not get_origin_remote(repo): |
|
no_remote.append(repo) |
|
insert_project(con, repo, host_id, account_id, defaults) |
|
added.append(repo) |
|
|
|
con.commit() |
|
print(f"backup={backup}") |
|
print(f"added={len(added)}") |
|
for repo in added: |
|
print(f"+ {repo}") |
|
print(f"skipped={len(skipped)}") |
|
for repo, reason in skipped: |
|
print(f"- {repo}: {reason}") |
|
if no_remote: |
|
print(f"no_origin_remote={len(no_remote)}") |
|
for repo in no_remote: |
|
print(f"! {repo}") |
|
return 0 |
|
finally: |
|
con.close() |
|
|
|
|
|
def main() -> int: |
|
args = parse_args() |
|
if args.root is None: |
|
print("error: root is required", file=sys.stderr) |
|
return 2 |
|
try: |
|
return run( |
|
root=args.root.expanduser().resolve(), |
|
db=args.db.expanduser().resolve(), |
|
host_id=args.host_id, |
|
dry_run=args.dry_run, |
|
include_duplicate_remotes=args.include_duplicate_remotes, |
|
) |
|
except BulkAddError as exc: |
|
print(f"error: {exc}", file=sys.stderr) |
|
return 1 |
|
|
|
|
|
if __name__ == "__main__": |
|
raise SystemExit(main()) |
