#!/usr/bin/env python3 import argparse import getpass import json import os import subprocess import ssl import sys import urllib.error import urllib.parse import urllib.request from pathlib import Path import certifi SERVER = "https://git.absdev.net" KEYCHAIN_SERVICE = "ABSDEV-Gitea" KEYCHAIN_ACCOUNT = "mag" EXPORT_DIR = Path.home() / "Downloads" / "GiteaExports" def get_token(): result = subprocess.run( [ "security", "find-generic-password", "-s", KEYCHAIN_SERVICE, "-a", KEYCHAIN_ACCOUNT, "-w", ], capture_output=True, text=True, ) if result.returncode != 0: sys.exit( "Gitea token not configured. " "Run: python3 gitea.py login" ) return result.stdout.strip() def login(): token = getpass.getpass("Gitea API token: ").strip() if not token: sys.exit("Token cannot be empty.") request = urllib.request.Request( f"{SERVER}/api/v1/user", headers={ "Authorization": f"token {token}", "Accept": "application/json", }, ) try: with urllib.request.urlopen(request, timeout=30, context=ssl.create_default_context(cafile=certifi.where())) as response: user = json.load(response) except urllib.error.HTTPError as exc: sys.exit(f"Authentication failed: HTTP {exc.code}") except urllib.error.URLError as exc: sys.exit(f"Connection failed: {exc.reason}") subprocess.run( [ "security", "delete-generic-password", "-s", KEYCHAIN_SERVICE, "-a", KEYCHAIN_ACCOUNT, ], capture_output=True, ) subprocess.run( [ "security", "add-generic-password", "-U", "-s", KEYCHAIN_SERVICE, "-a", KEYCHAIN_ACCOUNT, "-w", token, ], check=True, ) print(f"Authenticated as: {user['login']}") print("Token stored in macOS Keychain.") def api(path, binary=False): token = get_token() request = urllib.request.Request( f"{SERVER}/api/v1/{path.lstrip('/')}", headers={ "Authorization": f"token {token}", "Accept": "application/octet-stream" if binary else "application/json", }, ) try: with urllib.request.urlopen(request, timeout=120, context=ssl.create_default_context(cafile=certifi.where())) as response: if binary: return response.read() return json.load(response) except urllib.error.HTTPError as exc: sys.exit(f"Gitea API error: HTTP {exc.code}") except urllib.error.URLError as exc: sys.exit(f"Connection failed: {exc.reason}") def repositories(): page = 1 results = [] while True: batch = api( f"user/repos?limit=50&page={page}" ) results.extend(batch) if len(batch) < 50: break page += 1 return results def list_repositories(): repos = repositories() print(f"\nRepositories: {len(repos)}\n") for repo in repos: visibility = ( "PRIVATE" if repo["private"] else "PUBLIC" ) print( f"{repo['full_name']:<55} " f"{visibility}" ) def branches(repository): owner, name = parse_repository(repository) result = api( f"repos/{owner}/{name}/branches?limit=100" ) for branch in result: print(branch["name"]) def commits(repository): owner, name = parse_repository(repository) result = api( f"repos/{owner}/{name}/commits?limit=20" ) for commit in result: sha = commit["sha"][:10] message = commit["commit"]["message"].splitlines()[0] print(f"{sha} {message}") def parse_repository(repository): parts = repository.split("/") if len(parts) != 2 or not all(parts): sys.exit("Repository must be owner/name") return tuple( urllib.parse.quote(part, safe="") for part in parts ) def export_repository(repository): owner, name = parse_repository(repository) details = api(f"repos/{owner}/{name}") branch = details["default_branch"] if not branch: sys.exit("Repository has no default branch.") encoded_branch = urllib.parse.quote(branch, safe="") print(f"Exporting {repository} ({branch})") content = api( f"repos/{owner}/{name}/archive/" f"{encoded_branch}.zip", binary=True, ) EXPORT_DIR.mkdir(parents=True, exist_ok=True) filename = ( f"{owner}-{name}-{branch.replace('/', '-')}.zip" ) destination = EXPORT_DIR / filename destination.write_bytes(content) print(f"Saved: {destination}") def export_all(): repos = repositories() for repo in repos: try: export_repository(repo["full_name"]) except Exception as exc: print( f"Failed: {repo['full_name']}: {exc}", file=sys.stderr, ) def download_file(repository, filepath): owner, name = parse_repository(repository) encoded_path = urllib.parse.quote(filepath, safe="/") content = api( f"repos/{owner}/{name}/raw/{encoded_path}", binary=True, ) destination = EXPORT_DIR / owner / name / filepath destination.parent.mkdir( parents=True, exist_ok=True, ) destination.write_bytes(content) print(f"Saved: {destination}") # ============================================================ # Gitea Client 1.1 # Repository browsing, searching and selective exports # ============================================================ GITEA_CLIENT_VERSION = '1.2.0' import fnmatch import io import zipfile import base64 SENSITIVE_PATTERNS = [ ".env", ".env.*", "*.pem", "*.key", "*.p12", "*.pfx", "id_rsa", "id_ed25519", "*.sqlite", "*.sqlite3", "*.db", "credentials.json", "secrets.json", "service-account*.json", "*.mobileprovision", "*.provisionprofile", "*.keystore", "*.jks", ] EXCLUDED_DIRECTORIES = { ".git", ".venv", "venv", "node_modules", "vendor", "__pycache__", ".idea", ".build", "DerivedData", } def is_sensitive(path): """Return True when a path should not be exported.""" parts = Path(path).parts for part in parts: if part in EXCLUDED_DIRECTORIES: return True filename = Path(path).name.lower() for pattern in SENSITIVE_PATTERNS: if fnmatch.fnmatch(filename, pattern.lower()): return True return False def repository_tree(repository, branch=None): """Retrieve the complete repository tree.""" owner, name = parse_repository(repository) details = api(f"repos/{owner}/{name}") if branch is None: branch = details["default_branch"] if not branch: raise RuntimeError("Repository has no default branch.") encoded_branch = urllib.parse.quote(branch, safe="") result = api( f"repos/{owner}/{name}/git/trees/" f"{encoded_branch}?recursive=true" ) if result.get("truncated"): raise RuntimeError( "Gitea returned a truncated repository tree." ) return result.get("tree", []), branch def browse_repository(repository, directory=""): """Display repository files and directories.""" tree, branch = repository_tree(repository) directory = directory.strip("/") print() print(f"Repository: {repository}") print(f"Branch: {branch}") print(f"Directory: /{directory}") print() entries = {} prefix = directory + "/" if directory else "" for item in tree: path = item.get("path", "") if not path.startswith(prefix): continue relative = path[len(prefix):] if not relative: continue first = relative.split("/")[0] if "/" in relative: entries[first] = "DIR" else: entries[first] = ( "DIR" if item.get("type") == "tree" else "FILE" ) for name, kind in sorted(entries.items()): print(f"{kind:<6} {name}") print() print(f"{len(entries)} entries") def search_repository(repository, query): """Search repository filenames and source content.""" tree, branch = repository_tree(repository) query_lower = query.lower() owner, name = parse_repository(repository) matches = 0 scanned = 0 skipped = 0 MAX_FILES = 500 MAX_FILE_SIZE = 512 * 1024 for item in tree: if item.get("type") != "blob": continue path = item.get("path", "") if is_sensitive(path): continue if query_lower in path.lower(): print(f"PATH {path}") matches += 1 if item.get("size", 0) > MAX_FILE_SIZE: continue if scanned >= MAX_FILES: skipped += 1 continue scanned += 1 encoded = urllib.parse.quote(path, safe="/") try: content = api( f"repos/{owner}/{name}/raw/{encoded}", binary=True, ) except SystemExit: skipped += 1 continue if b"\\x00" in content: continue try: text = content.decode("utf-8") except UnicodeDecodeError: continue for line_number, line in enumerate( text.splitlines(), start=1 ): if query_lower in line.lower(): print( f"{path}:{line_number}: " f"{line.strip()[:180]}" ) matches += 1 print() print(f"Matches: {matches}") print(f"Files scanned: {scanned}") if skipped: print(f"Files skipped: {skipped}") if scanned >= MAX_FILES: print( "Search limit reached; results may be incomplete." ) def export_directory(repository, directory): """Export a selected directory into a ZIP archive.""" tree, branch = repository_tree(repository) owner, name = parse_repository(repository) directory = directory.strip("/") if not directory: raise RuntimeError( "Specify a directory for selective export." ) prefix = directory + "/" selected = [ item for item in tree if item.get("type") == "blob" and item.get("path", "").startswith(prefix) and not is_sensitive(item.get("path", "")) ] if not selected: raise RuntimeError( "No exportable files found in directory." ) EXPORT_DIR.mkdir(parents=True, exist_ok=True) safe_directory = directory.replace("/", "-") destination = EXPORT_DIR / ( f"{owner}-{name}-{safe_directory}.zip" ) print(f"Exporting {repository}/{directory}") print(f"Files: {len(selected)}") exported = 0 with zipfile.ZipFile( destination, "w", compression=zipfile.ZIP_DEFLATED, ) as archive: for item in selected: path = item["path"] encoded = urllib.parse.quote(path, safe="/") try: content = api( f"repos/{owner}/{name}/raw/{encoded}", binary=True, ) except SystemExit: print(f"SKIPPED: {path}") continue archive.writestr(path, content) exported += 1 print(f"Exported: {exported}") print(f"Saved: {destination}") def safe_export_repository(repository): """Export repository while excluding sensitive files.""" tree, branch = repository_tree(repository) owner, name = parse_repository(repository) selected = [ item for item in tree if item.get("type") == "blob" and not is_sensitive(item.get("path", "")) ] EXPORT_DIR.mkdir(parents=True, exist_ok=True) destination = EXPORT_DIR / ( f"{owner}-{name}-safe.zip" ) print(f"Safe exporting: {repository}") exported = 0 with zipfile.ZipFile( destination, "w", compression=zipfile.ZIP_DEFLATED, ) as archive: for item in selected: path = item["path"] encoded = urllib.parse.quote(path, safe="/") try: content = api( f"repos/{owner}/{name}/raw/{encoded}", binary=True, ) except SystemExit: print(f"SKIPPED: {path}") continue archive.writestr(path, content) exported += 1 print(f"Exported: {exported}") print(f"Saved: {destination}") # ============================================================ # Gitea Client 1.2.0 # Controlled Git integration # ============================================================ import shutil import tempfile import zipfile import subprocess import re import stat from pathlib import PurePosixPath WORKSPACE_DIR = ( Path.home() / "Developer" / "GiteaWorkspace" ) def git(*arguments, cwd=None, check=True, capture=True): """Execute Git without using a shell.""" command = ["git", *arguments] result = subprocess.run( command, cwd=cwd, text=True, capture_output=capture, ) if check and result.returncode: raise RuntimeError( result.stderr.strip() or result.stdout.strip() or "Git command failed" ) return result def repository_path(repository): owner, name = parse_repository(repository) return WORKSPACE_DIR / owner / name def ensure_repository(repository): path = repository_path(repository) if not path.is_dir(): raise RuntimeError( "Repository not cloned. " f"Run: python3 gitea.py clone {repository}" ) result = git( "rev-parse", "--show-toplevel", cwd=path, ) if Path(result.stdout.strip()).resolve() != path.resolve(): raise RuntimeError("Unexpected Git repository location.") return path def current_branch(path): return git( "branch", "--show-current", cwd=path, ).stdout.strip() def protected_branch(path): branch = current_branch(path) if branch in ("main", "master", "develop"): raise RuntimeError( f"Cannot modify protected branch: {branch}. " "Create a development branch first." ) if not branch: raise RuntimeError("Detached HEAD is not supported.") return branch def ensure_clean(path): result = git( "status", "--porcelain", "--untracked-files=all", cwd=path, ) if result.stdout.strip(): raise RuntimeError( "Working tree contains changes. " "Review or commit them before continuing." ) def clone_repository(repository): """Clone a repository using local Git credentials.""" path = repository_path(repository) if path.exists(): raise RuntimeError( f"Destination already exists: {path}" ) details = api( "repos/" + "/".join(parse_repository(repository)) ) clone_url = details["clone_url"] if not clone_url.startswith(SERVER + "/"): raise RuntimeError( "Unexpected clone URL returned by Gitea." ) path.parent.mkdir( parents=True, exist_ok=True, ) print(f"Cloning {repository}...") git( "clone", "--", clone_url, str(path), ) print(f"Cloned: {path}") def create_branch(repository, branch): """Create a development branch.""" path = ensure_repository(repository) ensure_clean(path) if not re.fullmatch( r"[A-Za-z0-9][A-Za-z0-9._/-]*", branch, ): raise RuntimeError("Invalid branch name.") if branch in ("main", "master", "develop"): raise RuntimeError("Protected branch name.") git( "check-ref-format", "--branch", branch, ) git( "switch", "-c", branch, cwd=path, ) print(f"Created branch: {branch}") def git_status(repository): """Show repository changes.""" path = ensure_repository(repository) print(f"Repository: {repository}") print(f"Branch: {current_branch(path)}") print() result = git( "status", "--short", cwd=path, ) print(result.stdout or "Working tree clean.") def git_diff(repository): """Preview working-tree changes.""" path = ensure_repository(repository) result = git( "diff", "--", cwd=path, ) print(result.stdout or "No tracked changes.") def safe_archive_members(archive): """Validate archive paths before extraction.""" members = [] for info in archive.infolist(): if info.is_dir(): continue name = info.filename.replace("\\", "/") path = PurePosixPath(name) if ( path.is_absolute() or ".." in path.parts or not path.parts ): raise RuntimeError( f"Unsafe archive path: {name}" ) if info.external_attr >> 16 & stat.S_IFMT( info.external_attr >> 16 ) == stat.S_IFLNK: raise RuntimeError( f"Archive contains symbolic link: {name}" ) if is_sensitive(name): print(f"SKIPPED sensitive file: {name}") continue members.append((info, path)) return members def import_archive(repository, archive_file): """Import reviewed files into a development branch.""" path = ensure_repository(repository) branch = protected_branch(path) ensure_clean(path) archive_file = Path(archive_file).expanduser() if not archive_file.is_file(): raise RuntimeError( f"Archive not found: {archive_file}" ) if not zipfile.is_zipfile(archive_file): raise RuntimeError("File is not a ZIP archive.") print(f"Repository: {repository}") print(f"Branch: {branch}") print(f"Archive: {archive_file}") with zipfile.ZipFile(archive_file) as archive: members = safe_archive_members(archive) if not members: raise RuntimeError( "Archive contains no importable files." ) with tempfile.TemporaryDirectory() as temp: staging = Path(temp) for info, relative in members: destination = staging.joinpath( *relative.parts ) destination.parent.mkdir( parents=True, exist_ok=True, ) with archive.open(info) as source: with destination.open("wb") as target: shutil.copyfileobj(source, target) imported = 0 for _, relative in members: source = staging.joinpath( *relative.parts ) destination = path.joinpath( *relative.parts ) if destination.is_symlink(): raise RuntimeError( f"Refusing symbolic link: {destination}" ) destination.parent.mkdir( parents=True, exist_ok=True, ) shutil.copy2(source, destination) imported += 1 print(f"Imported files: {imported}") print() print("Review changes before committing:") print(f"python3 gitea.py status {repository}") print(f"python3 gitea.py diff {repository}") def commit_changes(repository, message): """Commit reviewed changes.""" path = ensure_repository(repository) protected_branch(path) if not message.strip(): raise RuntimeError("Commit message cannot be empty.") print(f"Repository: {repository}") print(f"Branch: {current_branch(path)}") confirmation = input( "Stage and commit ALL working-tree changes? " "Type COMMIT: " ) if confirmation != "COMMIT": print("Commit cancelled.") return git( "add", "-A", cwd=path, ) result = git( "diff", "--cached", "--quiet", cwd=path, check=False, ) if result.returncode == 0: print("No changes to commit.") return if result.returncode != 1: raise RuntimeError( result.stderr.strip() or "Unable to inspect staged changes." ) git( "commit", "-m", message, cwd=path, ) print("Commit created successfully.") def push_changes(repository): """Push development branch after confirmation.""" path = ensure_repository(repository) branch = protected_branch(path) ensure_clean(path) print(f"Repository: {repository}") print(f"Branch: {branch}") print() confirmation = input( f"Push branch '{branch}' to Gitea? " "Type PUSH: " ) if confirmation != "PUSH": print("Push cancelled.") return git( "push", "-u", "origin", branch, cwd=path, ) print("Branch pushed successfully.") def main(): parser = argparse.ArgumentParser( description="ABSDEV Gitea API Client" ) sub = parser.add_subparsers( dest="command", required=True, ) sub.add_parser("login") sub.add_parser("repos") sub.add_parser("export-all") browse_parser = sub.add_parser("browse") browse_parser.add_argument("repository") browse_parser.add_argument("directory", nargs="?", default="") search_parser = sub.add_parser("search") search_parser.add_argument("repository") search_parser.add_argument("query") export_dir_parser = sub.add_parser("export-dir") export_dir_parser.add_argument("repository") export_dir_parser.add_argument("directory") safe_export_parser = sub.add_parser("safe-export") safe_export_parser.add_argument("repository") sub.add_parser("version") clone_parser = sub.add_parser("clone") clone_parser.add_argument("repository") branch_parser = sub.add_parser("branch") branch_parser.add_argument("repository") branch_parser.add_argument("branch") status_parser = sub.add_parser("status") status_parser.add_argument("repository") diff_parser = sub.add_parser("diff") diff_parser.add_argument("repository") import_parser = sub.add_parser("import") import_parser.add_argument("repository") import_parser.add_argument("archive") commit_parser = sub.add_parser("commit") commit_parser.add_argument("repository") commit_parser.add_argument("message") push_parser = sub.add_parser("push") push_parser.add_argument("repository") for command in ("branches", "commits", "export"): p = sub.add_parser(command) p.add_argument("repository") p = sub.add_parser("file") p.add_argument("repository") p.add_argument("filepath") args = parser.parse_args() if args.command == "clone": clone_repository(args.repository) elif args.command == "branch": create_branch(args.repository, args.branch) elif args.command == "status": git_status(args.repository) elif args.command == "diff": git_diff(args.repository) elif args.command == "import": import_archive(args.repository, args.archive) elif args.command == "commit": commit_changes(args.repository, args.message) elif args.command == "push": push_changes(args.repository) elif args.command == "version": print(f"Gitea Client {GITEA_CLIENT_VERSION}") elif args.command == "browse": browse_repository(args.repository, args.directory) elif args.command == "search": search_repository(args.repository, args.query) elif args.command == "export-dir": export_directory(args.repository, args.directory) elif args.command == "safe-export": safe_export_repository(args.repository) elif args.command == "login": login() elif args.command == "repos": list_repositories() elif args.command == "branches": branches(args.repository) elif args.command == "commits": commits(args.repository) elif args.command == "export": export_repository(args.repository) elif args.command == "export-all": export_all() elif args.command == "file": download_file( args.repository, args.filepath, ) if __name__ == "__main__": main()