sync.py

#!/usr/bin/python3

import sys
import os
import re
import subprocess
from pathlib import Path
from typing import List, Tuple


DEBUG = os.environ.get("DEBUG")
WORK_DIR = "/usr/local/lib/tunasync"

# Get environment variables
TO = os.environ.get("TO", "/data")
DELETE = os.environ.get("APTSYNC_UNLINK", "")
APTSYNC_URL = os.environ.get("APTSYNC_URL")
APTSYNC_DISTS = os.environ.get("APTSYNC_DISTS")

# Mirror list file path
MIRRORS_LIST_FILE = os.environ.get("MIRRORS_LIST", "/config/mirrors.list")


def parse_mirrors_list(file_path: str) -> List[Tuple[str, str, str, str, str]]:
    """
    Parse mirrors.list file format with mirror_path support
    Returns list of (url, dist, components, architectures, directory) tuples
    
    Supports:
    - mirror_path <url_prefix> <local_directory>
    - deb <url> <dist> <components>
    - deb [ arch=... ] <url> <dist> <components>
    """
    mirrors = []
    mirror_paths = {}  # url_prefix -> local_directory mapping
    
    if not os.path.exists(file_path):
        return mirrors
    
    with open(file_path, 'r') as f:
        for line_num, line in enumerate(f, 1):
            original_line = line
            line = line.strip()
            
            # Skip empty lines and comments
            if not line or line.startswith('#'):
                continue
            
            # Parse mirror_path lines
            if line.startswith('mirror_path '):
                parts = line.split()
                if len(parts) != 3:
                    print(f"Warning: Invalid mirror_path format at line {line_num}: {original_line.strip()}", file=sys.stderr)
                    print(f"  Expected: mirror_path <url_prefix> <local_directory>", file=sys.stderr)
                    continue
                
                _, url_prefix, local_directory = parts
                # Normalize URL prefix: ensure it ends with / for proper matching
                url_prefix = url_prefix.rstrip('/') + '/'
                mirror_paths[url_prefix] = local_directory
                print(f"Configured mirror_path: {url_prefix} -> {local_directory}")
                continue
            
            # Parse deb and deb-src lines
            if line.startswith(('deb ', 'deb-src ')):
                # Handle architecture constraints like [ arch=amd64,arm64 ]
                arch_constraints = None
                if '[ arch=' in line:
                    # Extract architecture constraints
                    start_bracket = line.find('[ arch=')
                    end_bracket = line.find(' ]', start_bracket)
                    if end_bracket != -1:
                        arch_part = line[start_bracket+7:end_bracket]  # Extract the arch list
                        arch_constraints = arch_part.replace(',', ' ')
                        # Remove the bracket part from the line
                        line = line[:start_bracket] + line[end_bracket+2:]
                        line = line.strip()
                
                parts = line.split()
                if len(parts) < 4:
                    print(f"Warning: Invalid format at line {line_num}: {original_line.strip()}", file=sys.stderr)
                    continue
                
                line_type = parts[0]  # deb or deb-src
                url = parts[1]
                dist = parts[2]
                components = ' '.join(parts[3:])
                
                # Handle architecture-specific repositories (e.g., deb-amd64)
                if line_type.startswith('deb-'):
                    arch_part = line_type[4:]  # Extract architecture from deb-amd64
                    architectures = arch_part
                elif arch_constraints:
                    # Use architecture constraints from [ arch=... ]
                    architectures = arch_constraints
                else:
                    # For regular 'deb' lines, use default architectures
                    architectures = os.environ.get("DEFAULT_ARCH", "amd64")
                
                # Find matching mirror_path for this URL
                directory = None
                matched_prefix = None
                for url_prefix, local_dir in mirror_paths.items():
                    if url.startswith(url_prefix):
                        # Calculate the relative path from the URL
                        relative_path = url[len(url_prefix):].rstrip('/')
                        if relative_path:
                            directory = f"{local_dir}/{relative_path}"
                        else:
                            directory = local_dir
                        matched_prefix = url_prefix
                        break
                
                if directory is None:
                    # No mirror_path found, use auto-generated directory
                    from urllib.parse import urlparse
                    parsed_url = urlparse(url.rstrip('/'))
                    path_parts = [p for p in parsed_url.path.split('/') if p]
                    if path_parts:
                        directory = '_'.join(path_parts[-2:]) if len(path_parts) >= 2 else path_parts[-1]
                    else:
                        directory = parsed_url.netloc.replace('.', '_')
                
                mirrors.append((url, dist, components, architectures, directory))
    
    return mirrors


def parse_aptsync_dists(aptsync_dists: str) -> List[Tuple[str, str, str, str]]:
    """
    Parse APTSYNC_DISTS format
    Returns list of (dist, components, architectures, directory) tuples
    """
    dists = []
    for dist in re.split(r"[:\n]+", aptsync_dists):
        if not dist.strip():
            continue
        parts = dist.strip().split("|")
        if len(parts) != 4:
            print(f"Warning: Invalid APTSYNC_DISTS format: {dist}", file=sys.stderr)
            continue
        
        apt_dist, apt_comp, apt_arch, apt_dir = parts
        dists.append((apt_dist, apt_comp, apt_arch, apt_dir))
    
    return dists


def main():
    os.chdir(WORK_DIR)
    
    cmds = []
    rets = []
    
    # Check which configuration method to use
    if os.path.exists(MIRRORS_LIST_FILE):
        print(f"Using mirrors.list configuration from: {MIRRORS_LIST_FILE}")
        
        mirrors = parse_mirrors_list(MIRRORS_LIST_FILE)
        if not mirrors:
            print(f"Error: No valid mirrors found in {MIRRORS_LIST_FILE}", file=sys.stderr)
            sys.exit(1)
        
        print(f"Found {len(mirrors)} repositories to sync:")
        for i, (url, dist, components, architectures, directory) in enumerate(mirrors, 1):
            print(f"  {i}. {url} -> {dist} [{components}] ({architectures})")
        print()
        
        for url, dist, components, architectures, directory in mirrors:
            # Convert components and architectures to comma-separated format
            comp_list = components.replace(' ', ',')
            arch_list = architectures.replace(' ', ',')
            
            # Normalize URL: remove trailing slash to avoid double slashes
            # apt-sync.py expects base_url and will add paths like "/dists/..." 
            base_url = url.rstrip('/')
            
            # Use the directory calculated by parse_mirrors_list
            output_dir = os.path.join(TO, directory)
            
            print(f"Syncing: {base_url} -> {dist}/{comp_list}/{arch_list} to {output_dir}")
            
            cmd = [sys.executable, "apt-sync.py"]
            if DELETE:
                cmd.append("--delete")
            
            cmd += [base_url, dist, comp_list, arch_list, output_dir]
            cmds.append(cmd[:])
            
            if DEBUG:
                print(f"Debug: Running command: {' '.join(cmd)}")
            
            cp = subprocess.run(cmd)
            rets.append(cp.returncode)
            
            if cp.returncode != 0:
                print(f"Warning: Sync failed for {base_url} -> {dist} (exit code: {cp.returncode})")
            else:
                print(f"Success: Completed sync for {base_url} -> {dist}")
            print()
    
    elif APTSYNC_URL and APTSYNC_DISTS:
        print("Using environment variables configuration (APTSYNC_URL + APTSYNC_DISTS)")
        
        dists = parse_aptsync_dists(APTSYNC_DISTS)
        
        print(f"Found {len(dists)} distribution configurations:")
        for i, (apt_dist, apt_comp, apt_arch, apt_dir) in enumerate(dists, 1):
            print(f"  {i}. {APTSYNC_URL}{apt_dir} -> {apt_dist} [{apt_comp}] ({apt_arch})")
        print()
        
        for apt_dist, apt_comp, apt_arch, apt_dir in dists:
            apt_arch = apt_arch.replace(" ", ",")
            apt_comp = apt_comp.replace(" ", ",")
            
            base_url = APTSYNC_URL + apt_dir
            output_dir = os.path.join(TO, apt_dir)
            
            print(f"Syncing: {base_url} -> {apt_dist}/{apt_comp}/{apt_arch} to {output_dir}")
            
            cmd = [sys.executable, "apt-sync.py"]
            if DELETE:
                cmd.append("--delete")
            
            cmd += [base_url, apt_dist, apt_comp, apt_arch, output_dir]
            cmds.append(cmd[:])
            
            if DEBUG:
                print(f"Debug: Running command: {' '.join(cmd)}")
            
            cp = subprocess.run(cmd)
            rets.append(cp.returncode)
            
            if cp.returncode != 0:
                print(f"Warning: Sync failed for {base_url} -> {apt_dist} (exit code: {cp.returncode})")
            else:
                print(f"Success: Completed sync for {base_url} -> {apt_dist}")
            print()
    
    else:
        print("Error: No configuration found!", file=sys.stderr)
        print("Please provide either:", file=sys.stderr)
        print(f"1. A mirrors.list file at {MIRRORS_LIST_FILE}", file=sys.stderr)
        print("2. Environment variables APTSYNC_URL and APTSYNC_DISTS", file=sys.stderr)
        sys.exit(1)
    
    if DEBUG:
        for cmd, ret in zip(cmds, rets):
            print(f"CMD {cmd} = {ret}", file=sys.stderr)
    
    sys.exit(sum(rets))


if __name__ == "__main__":
    main()