erplibre/script/git/git_local_server.py

#!/usr/bin/env python3
# © 2025-2026 TechnoLibre (http://www.technolibre.ca)
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl)

import argparse
import logging
import os
import subprocess
import sys
import xml.etree.ElementTree as ET

new_path = os.path.normpath(
    os.path.join(os.path.dirname(__file__), "..", "..")
)
sys.path.append(new_path)

from script.execute.execute import Execute

_logger = logging.getLogger(__name__)

DEFAULT_ERPLIBRE_PATH = os.path.normpath(
    os.path.join(os.path.dirname(__file__), "..", "..")
)
DEFAULT_GIT_PATH = os.path.join(os.path.expanduser("~"), ".git-server")
PRODUCTION_GIT_PATH = "/srv/git"
DEFAULT_MANIFEST = ".repo/local_manifests/erplibre_manifest.xml"
DEFAULT_REMOTE_NAME = "local"
DEFAULT_PORT = 9418


def get_config():
    parser = argparse.ArgumentParser(
        formatter_class=argparse.RawDescriptionHelpFormatter,
        description="""\
Deploy a local git server using git daemon and configure
all repos from the ERPLibre manifest.

Actions (in order):
  1. init   - Create bare repos in the git server path
  2. remote - Add 'local' remote to each repo
  3. push   - Push all branches to the local remote
  4. serve  - Start git daemon

By default, all actions are executed sequentially.
The default path is ~/.git-server (user-level, no root needed).
Use --production-ready for /srv/git (requires root).
""",
    )
    parser.add_argument(
        "-p",
        "--path",
        default=None,
        help=(
            "Path for git server bare repos" f" (default: {DEFAULT_GIT_PATH})"
        ),
    )
    parser.add_argument(
        "--production-ready",
        action="store_true",
        help=(
            f"Use {PRODUCTION_GIT_PATH} as git server path"
            " (requires root privileges)"
        ),
    )
    parser.add_argument(
        "-m",
        "--manifest",
        default=DEFAULT_MANIFEST,
        help="Manifest XML file" f" (default: {DEFAULT_MANIFEST})",
    )
    parser.add_argument(
        "--remote-name",
        default=DEFAULT_REMOTE_NAME,
        help="Remote name to add" f" (default: {DEFAULT_REMOTE_NAME})",
    )
    parser.add_argument(
        "--port",
        type=int,
        default=DEFAULT_PORT,
        help=f"Git daemon port (default: {DEFAULT_PORT})",
    )
    parser.add_argument(
        "--action",
        choices=["init", "remote", "push", "serve", "all"],
        default="all",
        help="Action to perform (default: all)",
    )
    parser.add_argument(
        "--erplibre-root",
        default=None,
        help="ERPLibre root directory (default: auto-detect)",
    )
    parser.add_argument(
        "--push-all-branches",
        action="store_true",
        help="Push all local branches, not just the current one",
    )
    parser.add_argument(
        "--remote-url-type",
        choices=["file", "daemon"],
        default="file",
        help=(
            "Remote URL type: 'file' uses local path"
            " (push works without daemon), 'daemon'"
            " uses git:// protocol (default: file)"
        ),
    )
    parser.add_argument(
        "-v",
        "--verbose",
        action="store_true",
        help="Enable verbose output",
    )
    args = parser.parse_args()

    # Resolve git server path
    if args.path and args.production_ready:
        parser.error("--path and --production-ready are mutually exclusive")
    if args.production_ready:
        args.path = PRODUCTION_GIT_PATH
    elif args.path is None:
        args.path = DEFAULT_GIT_PATH

    return args


def check_path_permissions(path):
    """Check if we have write permissions on the target
    path. Try the path itself first, then its closest
    existing parent directory."""
    check = path
    while check and not os.path.exists(check):
        check = os.path.dirname(check)
    if not check:
        check = "/"
    if os.access(check, os.W_OK):
        return True
    return False


def require_permissions_or_exit(path):
    """Exit with instructions if we cannot write to path."""
    print(f"Error: No write permission on {path}")
    print(
        "Please run this script with appropriate"
        " privileges:"
        f"\n  sudo {sys.executable}"
        f" {' '.join(sys.argv)}"
    )
    sys.exit(1)


def parse_manifest(manifest_path):
    """Parse the manifest XML and return list of projects."""
    tree = ET.parse(manifest_path)
    root = tree.getroot()

    remotes = {}
    for remote in root.findall("remote"):
        remotes[remote.get("name")] = remote.get("fetch")

    projects = []
    for project in root.findall("project"):
        name = project.get("name")
        path = project.get("path")
        remote = project.get("remote")
        revision = project.get("revision")
        projects.append(
            {
                "name": name,
                "path": path,
                "remote": remote,
                "revision": revision,
                "fetch_url": remotes.get(remote, ""),
            }
        )

    return projects


def init_bare_repos(git_path, projects):
    """Create bare repos for all projects in the manifest."""
    os.makedirs(git_path, exist_ok=True)

    created = 0
    skipped = 0
    for project in projects:
        repo_name = project["name"]
        if not repo_name.endswith(".git"):
            repo_name += ".git"
        bare_path = os.path.join(git_path, repo_name)

        if os.path.exists(bare_path):
            _logger.info(f"  Exists: {bare_path}")
            skipped += 1
            continue

        _logger.info(f"  Creating: {bare_path}")
        subprocess.run(
            ["git", "init", "--bare", bare_path],
            check=True,
            capture_output=True,
        )

        # Enable git daemon export
        export_file = os.path.join(bare_path, "git-daemon-export-ok")
        open(export_file, "w").close()

        created += 1

    print(f"Bare repos: {created} created, {skipped} skipped (already exist)")


def build_remote_url(git_path, repo_name, port, remote_url_type):
    """Build the remote URL based on the type."""
    if remote_url_type == "daemon":
        if port != DEFAULT_PORT:
            return f"git://localhost:{port}/{repo_name}"
        return f"git://localhost/{repo_name}"
    # Default: file path
    return os.path.join(git_path, repo_name)


def add_remotes(
    erplibre_root,
    git_path,
    projects,
    remote_name,
    port,
    remote_url_type,
):
    """Add local remote to each repo.

    remote_url_type='file': uses local path, push works
      without daemon running.
    remote_url_type='daemon': uses git://localhost URL,
      requires daemon for push.
    """
    added = 0
    skipped = 0
    errors = 0
    for project in projects:
        repo_path = os.path.join(erplibre_root, project["path"])
        if not os.path.isdir(repo_path):
            _logger.warning(f"  Not found: {repo_path}")
            errors += 1
            continue

        repo_name = project["name"]
        if not repo_name.endswith(".git"):
            repo_name += ".git"

        remote_url = build_remote_url(
            git_path, repo_name, port, remote_url_type
        )

        # Check if remote already exists
        result = subprocess.run(
            ["git", "-C", repo_path, "remote"],
            capture_output=True,
            text=True,
        )
        existing_remotes = result.stdout.strip().split("\n")

        if remote_name in existing_remotes:
            # Update URL if remote exists
            subprocess.run(
                [
                    "git",
                    "-C",
                    repo_path,
                    "remote",
                    "set-url",
                    remote_name,
                    remote_url,
                ],
                check=True,
                capture_output=True,
            )
            _logger.info(f"  Updated: {project['path']}")
            skipped += 1
        else:
            subprocess.run(
                [
                    "git",
                    "-C",
                    repo_path,
                    "remote",
                    "add",
                    remote_name,
                    remote_url,
                ],
                check=True,
                capture_output=True,
            )
            _logger.info(f"  Added: {project['path']}")
            added += 1

    print(f"Remotes: {added} added, {skipped} updated," f" {errors} errors")


def is_detached_head(repo_path):
    """Check if a repo is in detached HEAD state."""
    result = subprocess.run(
        ["git", "-C", repo_path, "symbolic-ref", "HEAD"],
        capture_output=True,
        text=True,
    )
    return result.returncode != 0


def checkout_manifest_branch(repo_path, revision):
    """Checkout the branch from the manifest revision.

    When Google Repo syncs, repos end up in detached HEAD
    on the exact commit. We create/checkout a local branch
    matching the manifest revision so git push works.
    """
    # Extract branch name — revision can be
    # "18.0", "18.0_dev", "ERPLibre/18.0", "main", etc.
    branch = revision.split("/")[-1] if "/" in revision else revision

    # Check if local branch already exists
    result = subprocess.run(
        [
            "git",
            "-C",
            repo_path,
            "show-ref",
            "--verify",
            f"refs/heads/{branch}",
        ],
        capture_output=True,
    )
    if result.returncode == 0:
        # Branch exists, checkout it
        subprocess.run(
            ["git", "-C", repo_path, "checkout", branch],
            capture_output=True,
        )
    else:
        # Create branch from current HEAD
        subprocess.run(
            [
                "git",
                "-C",
                repo_path,
                "checkout",
                "-b",
                branch,
            ],
            capture_output=True,
        )
    return branch


def update_bare_head(git_path, project):
    """Update the HEAD of the bare repo to point to the
    manifest branch, so git clone checks out the right
    branch by default."""
    repo_name = project["name"]
    if not repo_name.endswith(".git"):
        repo_name += ".git"
    bare_path = os.path.join(git_path, repo_name)
    if not os.path.isdir(bare_path):
        return

    revision = project.get("revision", "")
    if not revision:
        return
    branch = revision.split("/")[-1] if "/" in revision else revision
    subprocess.run(
        [
            "git",
            "-C",
            bare_path,
            "symbolic-ref",
            "HEAD",
            f"refs/heads/{branch}",
        ],
        capture_output=True,
    )


def push_to_local(
    erplibre_root,
    git_path,
    projects,
    remote_name,
    push_all_branches,
):
    """Push to local remote for each repo."""
    pushed = 0
    errors = 0
    checkouts = 0
    for project in projects:
        repo_path = os.path.join(erplibre_root, project["path"])
        if not os.path.isdir(repo_path):
            _logger.warning(f"  Not found: {repo_path}")
            errors += 1
            continue

        # Unshallow if needed (clone-depth in manifest)
        shallow_file = os.path.join(repo_path, ".git", "shallow")
        if os.path.exists(shallow_file):
            _logger.info(f"  Unshallowing {project['path']}...")
            # Try each remote until unshallow succeeds
            result = subprocess.run(
                ["git", "-C", repo_path, "remote"],
                capture_output=True,
                text=True,
            )
            remotes = [
                r
                for r in result.stdout.strip().split("\n")
                if r and r != remote_name
            ]
            # Try the manifest remote first
            manifest_remote = project.get("remote", "")
            if manifest_remote in remotes:
                remotes.remove(manifest_remote)
                remotes.insert(0, manifest_remote)

            for try_remote in remotes:
                result = subprocess.run(
                    [
                        "git",
                        "-C",
                        repo_path,
                        "fetch",
                        "--unshallow",
                        try_remote,
                    ],
                    capture_output=True,
                    text=True,
                    timeout=300,
                )
                if not os.path.exists(shallow_file):
                    _logger.info(f"  Unshallowed via" f" {try_remote}")
                    break
            else:
                if os.path.exists(shallow_file):
                    _logger.warning(
                        f"  Unshallow failed for"
                        f" {project['path']},"
                        " push may fail"
                    )

        # Handle detached HEAD: checkout manifest branch
        if is_detached_head(repo_path):
            revision = project.get("revision", "")
            if revision:
                branch = checkout_manifest_branch(repo_path, revision)
                _logger.info(
                    f"  Checkout {branch} for"
                    f" {project['path']}"
                    " (was detached HEAD)"
                )
                checkouts += 1
            else:
                _logger.warning(
                    f"  Detached HEAD with no revision"
                    f" for {project['path']}, skipping"
                )
                errors += 1
                continue

        try:
            if push_all_branches:
                cmd = [
                    "git",
                    "-C",
                    repo_path,
                    "push",
                    remote_name,
                    "--all",
                ]
            else:
                cmd = [
                    "git",
                    "-C",
                    repo_path,
                    "push",
                    remote_name,
                ]
            result = subprocess.run(
                cmd,
                capture_output=True,
                text=True,
                timeout=120,
            )
            if result.returncode != 0:
                _logger.warning(
                    f"  Push failed for"
                    f" {project['path']}:"
                    f" {result.stderr.strip()}"
                )
                errors += 1
            else:
                update_bare_head(git_path, project)
                _logger.info(f"  Pushed: {project['path']}")
                pushed += 1
        except subprocess.TimeoutExpired:
            _logger.warning(f"  Timeout pushing {project['path']}")
            errors += 1

    print(
        f"Push: {pushed} pushed, {checkouts} branch"
        f" checkouts, {errors} errors"
    )


def print_clone_commands(git_path, projects, port):
    """Print git clone commands for all available repos."""
    print("=== Available repos to clone ===")
    base_url = (
        f"git://localhost:{port}"
        if port != DEFAULT_PORT
        else "git://localhost"
    )
    count = 0
    for project in projects:
        repo_name = project["name"]
        if not repo_name.endswith(".git"):
            repo_name += ".git"
        bare_path = os.path.join(git_path, repo_name)
        if os.path.isdir(bare_path):
            print(f"  git clone {base_url}/{repo_name}" f" {project['path']}")
            count += 1
    print(f"\nTotal: {count} repos available")
    print()


def serve_git_daemon(git_path, projects, port):
    """Start git daemon to serve repos."""
    print_clone_commands(git_path, projects, port)

    cmd = (
        f"git daemon --reuseaddr"
        f" --base-path={git_path}"
        f" --port={port}"
        f" --export-all"
        f" --enable=receive-pack"
        f" {git_path}"
    )
    print(f"Starting git daemon on port {port}...")
    print(f"  Base path: {git_path}")
    print(f"  URL: git://localhost:{port}/")
    print("  Press Ctrl+C to stop")

    execute = Execute()
    execute.exec_command_live(cmd, source_erplibre=False)


def main():
    config = get_config()

    log_level = logging.DEBUG if config.verbose else logging.INFO
    logging.basicConfig(
        level=log_level,
        format="%(levelname)s: %(message)s",
    )

    # Check write permissions on target path
    if not check_path_permissions(config.path):
        require_permissions_or_exit(config.path)

    # Detect ERPLibre root
    if config.erplibre_root:
        erplibre_root = os.path.abspath(config.erplibre_root)
    else:
        erplibre_root = DEFAULT_ERPLIBRE_PATH

    manifest_path = os.path.join(erplibre_root, config.manifest)
    if not os.path.exists(manifest_path):
        print(f"Error: Manifest not found: {manifest_path}")
        sys.exit(1)

    print(f"ERPLibre root: {erplibre_root}")
    print(f"Manifest: {manifest_path}")
    print(f"Git server path: {config.path}")
    print(f"Remote name: {config.remote_name}")
    if config.production_ready:
        print("Mode: production (/srv/git)")
    else:
        print("Mode: development (~/.git-server)")
    print(f"Remote URL type: {config.remote_url_type}")
    print()

    projects = parse_manifest(manifest_path)
    print(f"Found {len(projects)} projects in manifest")
    print()

    action = config.action

    if action in ("init", "all"):
        print("=== Creating bare repos ===")
        init_bare_repos(config.path, projects)
        print()

    if action in ("remote", "all"):
        print("=== Adding remotes ===")
        add_remotes(
            erplibre_root,
            config.path,
            projects,
            config.remote_name,
            config.port,
            config.remote_url_type,
        )
        print()

    if action in ("push", "all"):
        print("=== Pushing to local ===")
        push_to_local(
            erplibre_root,
            config.path,
            projects,
            config.remote_name,
            config.push_all_branches,
        )
        print()

    if action in ("serve", "all"):
        print("=== Starting git daemon ===")
        serve_git_daemon(config.path, projects, config.port)


if __name__ == "__main__":
    main()