Gitea_Client/gitea.py

#!/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()