"""Mirror repos from a Forgejo/Gitea instance to a local target directory. Fetches all repos owned by each user in OWNERS and stores per-commit zip snapshots under `TARGET_DIR//`. Runs unattended via systemd timer. Local staging when the target is unavailable. Configuration precedence (for each key): env var > config file > hardcoded default. Config file: `~/.config/coe-mirror/config.toml` (see `python coe_mirror.py config init`). Requires Python 3.11+ (for stdlib `tomllib`). """ import json import logging import logging.handlers import os import re import shutil import subprocess import sys import time import tomllib import zipfile from datetime import datetime, timedelta, timezone from pathlib import Path from fcntl import flock, LOCK_EX, LOCK_NB from typing import List, Optional, Tuple from urllib.parse import quote import requests # --- Config file (optional) ------------------------------------------------ CONFIG_PATH = Path( os.getenv("COE_MIRROR_CONFIG") or (Path(os.getenv("XDG_CONFIG_HOME") or Path.home() / ".config") / "coe-mirror" / "config.toml") ) def _load_config_file() -> dict: """Load config.toml if present. Never raises; malformed file → empty dict + stderr warning.""" try: if CONFIG_PATH.is_file(): with open(CONFIG_PATH, "rb") as f: return tomllib.load(f) except Exception as exc: print(f"coe-mirror: warning: could not parse {CONFIG_PATH}: {exc}", file=sys.stderr) return {} _CFG = _load_config_file() _CFG_MATRIX = _CFG.get("matrix", {}) if isinstance(_CFG.get("matrix"), dict) else {} def _resolve(env_key: str, cfg_value, default): """env var > config value > hardcoded default. Empty env/config values are ignored.""" env_val = os.getenv(env_key) if env_val: return env_val if cfg_value not in (None, ""): return cfg_value return default # --- Module constants (resolved via env > config > default) --------------- API_BASE: str = _resolve("COE_MIRROR_API_BASE", _CFG.get("api_base"), "https://git.churchofmalware.org") OWNERS: List[str] = ( [o.strip() for o in os.getenv("COE_MIRROR_OWNERS", "").split(",") if o.strip()] or (_CFG.get("owners") if isinstance(_CFG.get("owners"), list) else None) or ["Nightmare_Eclipse", "shai_hulud", "bikini_mirror", "ek0mssavi0r"] ) TARGET_DIR: Path = Path( _resolve( "COE_MIRROR_TARGET_DIR", _CFG.get("target_dir"), str(Path.home() / "coe-mirror"), ) ).expanduser() PENDING_DIR: Path = Path( _resolve( "COE_MIRROR_PENDING_DIR", _CFG.get("pending_dir"), str(Path.home() / ".local" / "share" / "coe-mirror" / "pending"), ) ).expanduser() _MOUNT_ROOT_VAL = _resolve("COE_MIRROR_MOUNT_ROOT", _CFG.get("mount_root"), "") MOUNT_ROOT: Optional[Path] = Path(_MOUNT_ROOT_VAL).expanduser() if _MOUNT_ROOT_VAL else None LOG_DIR = Path.home() / ".local" / "share" / "coe-mirror" LOG_FILE = LOG_DIR / "coe-mirror.log" LOCK_NAME = "coe-mirror.lock" SHORT_SHA_LEN = 8 HTTP_TIMEOUT = 30 DOWNLOAD_TIMEOUT = 60 RETRY_BACKOFF = [0, 5, 30] # seconds between attempts; len = max attempts USER_AGENT = "coe-mirror/1.0" REPO_NAME_RE = re.compile(r"^[a-zA-Z0-9._-]+$") # Matrix configuration (fully opt-in; blank defaults => notifications silently skipped) MATRIX_HOMESERVER: str = _resolve("COE_MIRROR_MATRIX_HOMESERVER", _CFG_MATRIX.get("homeserver"), "") MATRIX_ROOM: str = _resolve("COE_MIRROR_MATRIX_ROOM", _CFG_MATRIX.get("room"), "") MATRIX_TOKEN_CACHE = LOG_DIR / "matrix-token.json" MATRIX_TIMEOUT = 8 MATRIX_REFRESH_THRESHOLD_SECONDS = 1800 # refresh if <30 min left COE_MIRROR_BOT_USER_KEY = "COE_MIRROR_BOT_USER" COE_MIRROR_BOT_PASSWORD_KEY = "COE_MIRROR_BOT_PASSWORD" logger: logging.Logger = None # Initialized in setup_logging() def _safe_repo_name(name: str) -> bool: """Validate repo name against whitelist to prevent path traversal.""" return bool(REPO_NAME_RE.fullmatch(name)) def setup_logging() -> None: """Configure logging to file and stderr with rotation.""" global logger LOG_DIR.mkdir(parents=True, exist_ok=True) PENDING_DIR.mkdir(parents=True, exist_ok=True) logger = logging.getLogger("coe-mirror") logger.setLevel(logging.DEBUG) # Rotating file handler file_handler = logging.handlers.RotatingFileHandler( LOG_FILE, maxBytes=5 * 1024 * 1024, backupCount=3 ) file_handler.setLevel(logging.DEBUG) # Stderr handler for systemd journal stderr_handler = logging.StreamHandler(sys.stderr) stderr_handler.setLevel(logging.DEBUG) # Formatter formatter = logging.Formatter( "%(asctime)s %(levelname)s %(name)s: %(message)s" ) file_handler.setFormatter(formatter) stderr_handler.setFormatter(formatter) logger.addHandler(file_handler) logger.addHandler(stderr_handler) def acquire_lock() -> Optional[int]: """Acquire exclusive lock. Return lock fd on success, None if contended.""" xdg_runtime = os.getenv("XDG_RUNTIME_DIR") lock_dir = Path(xdg_runtime) if xdg_runtime else Path("/tmp") lock_file = lock_dir / LOCK_NAME try: fd = os.open(str(lock_file), os.O_CREAT | os.O_WRONLY, 0o600) try: flock(fd, LOCK_EX | LOCK_NB) return fd except OSError: os.close(fd) return None except OSError as exc: logger.error("Failed to open lock file: %s", exc) return None def release_lock(lock_fd: int) -> None: """Release and close the lock file descriptor.""" try: os.close(lock_fd) except OSError: pass def probe_target() -> bool: """Probe target directory health. Write tiny file, read back, delete. Return True if healthy. When MOUNT_ROOT is configured, first verify it is a live mountpoint and TARGET_DIR is under it; return False (queue to pending) otherwise. """ try: if MOUNT_ROOT is not None: if not MOUNT_ROOT.exists() or not os.path.ismount(str(MOUNT_ROOT)): logger.warning("mount_root %s is not a live mountpoint; target unavailable; queuing locally", MOUNT_ROOT) return False if not TARGET_DIR.is_relative_to(MOUNT_ROOT): logger.warning("target_dir %s is not under configured mount_root %s; misconfig; queuing locally", TARGET_DIR, MOUNT_ROOT) return False probe_file = TARGET_DIR / ".write_probe" TARGET_DIR.mkdir(parents=True, exist_ok=True) probe_file.write_text("probe") # fsync on file descriptor, not stat with open(probe_file, "r") as f: os.fsync(f.fileno()) content = probe_file.read_text() probe_file.unlink() if content != "probe": logger.warning("target directory write verification failed (content mismatch); target unavailable; queuing locally") return False return True except Exception as exc: logger.warning("target directory probe failed: %s; target unavailable; queuing locally", exc) return False def place_into_target(src: Path, dst: Path) -> bool: """Copy src → dst with size verification; unlink src only on success. Returns True on success, False on any failure. Never raises. """ # Defense in depth: if src and dst are the same file, it's a no-op success if src.resolve() == dst.resolve(): logger.debug("src and dst are the same file, no-op: %s", src) return True try: shutil.copy2(src, dst) src_size = src.stat().st_size dst_size = dst.stat().st_size if src_size != dst_size: logger.warning( "size mismatch placing %s → %s (src=%d dst=%d)", src, dst, src_size, dst_size ) try: dst.unlink() except OSError: pass return False src.unlink() return True except Exception as exc: logger.warning("placement failed %s → %s: %s", src, dst, exc) return False def fetch_repos(session: requests.Session, owner: str) -> Optional[list]: """Fetch list of repos owned by `owner`. Return list of repo dicts or None on fatal error.""" url = f"{API_BASE}/api/v1/users/{owner}/repos" for attempt, backoff in enumerate(RETRY_BACKOFF): if attempt > 0: logger.info("Retrying repo list fetch (attempt %d), backoff %ds", attempt + 1, backoff) time.sleep(backoff) try: resp = session.get(url, timeout=HTTP_TIMEOUT) if resp.status_code == 429: logger.debug("Got 429 on repo list, will retry") continue if resp.status_code >= 500: logger.debug("Got %d on repo list, will retry", resp.status_code) continue resp.raise_for_status() return resp.json() except requests.RequestException as exc: logger.debug("Attempt %d: %s", attempt + 1, exc) logger.error("Failed to fetch repos after %d attempts", len(RETRY_BACKOFF)) return None def fetch_branch_sha(session: requests.Session, owner: str, repo_name: str, default_branch: str) -> Optional[str]: """Fetch commit SHA for repo's default branch. Return full hex SHA or None on error.""" url = f"{API_BASE}/api/v1/repos/{owner}/{quote(repo_name, safe='')}/branches/{quote(default_branch, safe='')}" for attempt, backoff in enumerate(RETRY_BACKOFF): if attempt > 0: logger.debug("Retrying branch fetch for %s (attempt %d), backoff %ds", repo_name, attempt + 1, backoff) time.sleep(backoff) try: resp = session.get(url, timeout=HTTP_TIMEOUT) if resp.status_code == 429: logger.debug("Got 429 on branch %s, will retry", repo_name) continue if resp.status_code >= 500: logger.debug("Got %d on branch %s, will retry", resp.status_code, repo_name) continue resp.raise_for_status() data = resp.json() return data.get("commit", {}).get("id") except requests.RequestException as exc: logger.debug("Attempt %d for %s: %s", attempt + 1, repo_name, exc) logger.error("Failed to fetch branch SHA for %s after %d attempts", repo_name, len(RETRY_BACKOFF)) return None def download_repo_zip(session: requests.Session, owner: str, repo_name: str, full_sha: str) -> Optional[Path]: """Download repo zip for given SHA. Return Path to downloaded file (in PENDING_DIR) or None. Pending filename is `{owner}__{repo}_{sha8}.zip` so owners with identical repo names don't collide in the shared PENDING_DIR. flush_pending() strips the owner prefix when routing to the per-owner target subdir. """ # Defense in depth: validate repo_name even though main() should have done so if not _safe_repo_name(repo_name) or not _safe_repo_name(owner): logger.error("refusing unsafe owner/repo name in download_repo_zip: %r/%r", owner, repo_name) return None filename = f"{owner}__{repo_name}_{full_sha[:SHORT_SHA_LEN]}.zip" partial_path = PENDING_DIR / f"{filename}.partial" final_path = PENDING_DIR / filename # URL-encode repo_name and full_sha for safety (defense in depth after main-level validation) url = f"{API_BASE}/{owner}/{quote(repo_name, safe='')}/archive/{quote(full_sha, safe='')}.zip" for attempt, backoff in enumerate(RETRY_BACKOFF): if attempt > 0: logger.debug("Retrying download for %s (attempt %d), backoff %ds", filename, attempt + 1, backoff) time.sleep(backoff) try: resp = session.get(url, timeout=DOWNLOAD_TIMEOUT, stream=True) if resp.status_code == 429: logger.debug("Got 429 on download %s, will retry", filename) continue if resp.status_code >= 500: logger.debug("Got %d on download %s, will retry", resp.status_code, filename) continue resp.raise_for_status() # Stream to temp file partial_path.parent.mkdir(parents=True, exist_ok=True) with open(partial_path, "wb") as f: for chunk in resp.iter_content(chunk_size=65536): if chunk: f.write(chunk) # Validate zip if not zipfile.is_zipfile(partial_path): logger.error("Downloaded file is not a valid zip: %s", filename) partial_path.unlink() return None # Rename to final partial_path.rename(final_path) logger.info("Downloaded %s", filename) return final_path except requests.RequestException as exc: logger.debug("Attempt %d for %s: %s", attempt + 1, filename, exc) except Exception as exc: logger.error("Error downloading %s: %s", filename, exc) if partial_path.exists(): partial_path.unlink() return None logger.error("Failed to download %s after %d attempts", filename, len(RETRY_BACKOFF)) if partial_path.exists(): partial_path.unlink() return None def flush_pending(target_healthy: bool) -> list: """Move all *.zip files from PENDING_DIR to TARGET_DIR// if healthy. Pending filenames are `{owner}__{repo}_{sha8}.zip`. The owner prefix is stripped when placing into target (final name is `{repo}_{sha8}.zip` inside owner subdir). Return list of (owner, repo, short_sha, final_location) tuples for files moved. """ moved = [] if not target_healthy or not PENDING_DIR.exists(): return moved for pending_file in PENDING_DIR.glob("*.zip"): # Expect `{owner}__{repo}_{sha8}.zip` if "__" not in pending_file.name: logger.warning("skipping pending file without owner prefix: %s", pending_file.name) continue owner, rest = pending_file.name.split("__", 1) if not _safe_repo_name(owner): logger.warning("refusing unsafe owner prefix in pending file: %r", owner) continue owner_dir = TARGET_DIR / owner owner_dir.mkdir(parents=True, exist_ok=True) dst = owner_dir / rest if place_into_target(pending_file, dst): # Extract repo name and short sha from remainder (`{repo}_{sha8}.zip`) name_parts = Path(rest).stem.rsplit("_", 1) if len(name_parts) == 2: repo, short_sha = name_parts moved.append((owner, repo, short_sha, str(dst))) logger.info("Flushed %s to target directory", pending_file.name) return moved def _read_token_cache() -> Optional[dict]: """Return cached token dict or None on any failure. Never raises.""" try: if not MATRIX_TOKEN_CACHE.exists(): return None with open(MATRIX_TOKEN_CACHE) as f: data = json.load(f) # Shape check if "access_token" in data and "expires_at" in data: return data except Exception as exc: logger.debug("token cache read failed: %s", exc) return None def _write_token_cache(access_token: str, expires_at_iso: str) -> None: """Write token cache with 0600 permissions. Never raises.""" try: MATRIX_TOKEN_CACHE.parent.mkdir(parents=True, exist_ok=True) payload = {"access_token": access_token, "expires_at": expires_at_iso} # Write then chmod for atomicity-ish behavior tmp = MATRIX_TOKEN_CACHE.with_suffix(".tmp") with open(tmp, "w") as f: json.dump(payload, f) os.chmod(tmp, 0o600) os.replace(tmp, MATRIX_TOKEN_CACHE) except Exception as exc: logger.warning("token cache write failed: %s", exc) def _fetch_creds() -> Optional[Tuple[str, str]]: """Fetch (user, password) for Matrix. Order: `creds` binary (Infisical) → `COE_MIRROR_MATRIX_USER` / `COE_MIRROR_MATRIX_PASSWORD` env vars → None. Distributed users typically set the env vars in the systemd EnvironmentFile; the Infisical path is convenience for hosts that already have it. """ try: user = subprocess.run( ["creds", "get", COE_MIRROR_BOT_USER_KEY, "matrix"], capture_output=True, text=True, timeout=5 ) pw = subprocess.run( ["creds", "get", COE_MIRROR_BOT_PASSWORD_KEY, "matrix"], capture_output=True, text=True, timeout=5 ) if user.returncode == 0 and pw.returncode == 0: u, p = user.stdout.strip(), pw.stdout.strip() if u and p: return u, p logger.debug("creds returned non-zero for %s/%s; trying env fallback", COE_MIRROR_BOT_USER_KEY, COE_MIRROR_BOT_PASSWORD_KEY) except FileNotFoundError: logger.debug("creds binary not on PATH; trying env fallback") except subprocess.TimeoutExpired: logger.warning("creds timed out; trying env fallback") except Exception as exc: logger.warning("creds fetch failed: %s; trying env fallback", exc) env_u = os.getenv("COE_MIRROR_MATRIX_USER", "").strip() env_p = os.getenv("COE_MIRROR_MATRIX_PASSWORD", "").strip() if env_u and env_p: return env_u, env_p logger.warning("no Matrix credentials available (creds + env both empty); notifications disabled") return None def _matrix_login(session: requests.Session) -> Optional[str]: """Login via password, cache token, return access_token. None on failure.""" creds = _fetch_creds() if not creds: return None user, password = creds url = f"{MATRIX_HOMESERVER}/_matrix/client/r0/login" body = {"type": "m.login.password", "user": user, "password": password} try: resp = session.post(url, json=body, timeout=MATRIX_TIMEOUT) if resp.status_code != 200: logger.warning("matrix login failed: HTTP %d", resp.status_code) return None data = resp.json() token = data.get("access_token") expires_in_ms = data.get("expires_in_ms", 0) if not token: logger.warning("matrix login returned no access_token") return None expires_at = datetime.now(timezone.utc) + timedelta(milliseconds=expires_in_ms) _write_token_cache(token, expires_at.isoformat()) return token except requests.RequestException as exc: logger.warning("matrix login network error: %s", exc) except Exception as exc: logger.warning("matrix login unexpected error: %s", exc) return None def _get_matrix_token(session: requests.Session, force_refresh: bool = False) -> Optional[str]: """Return a valid token. Read cache → check expiry → login if needed.""" if not force_refresh: cached = _read_token_cache() if cached: try: expires_at = datetime.fromisoformat(cached["expires_at"]) remaining = (expires_at - datetime.now(timezone.utc)).total_seconds() if remaining > MATRIX_REFRESH_THRESHOLD_SECONDS: return cached["access_token"] except Exception as exc: logger.debug("cache expiry parse failed: %s", exc) return _matrix_login(session) def _send_matrix(message: str) -> bool: """Send a Matrix message to MATRIX_ROOM. Return True on success. Non-fatal: any failure logs WARN and returns False. Never raises. Implements one 401 retry with token refresh. Silently skips (returns False) if MATRIX_HOMESERVER or MATRIX_ROOM is unconfigured. """ if not MATRIX_HOMESERVER or not MATRIX_ROOM: logger.debug("matrix notifications disabled (homeserver or room unset)") return False with requests.Session() as session: session.headers.update({"User-Agent": USER_AGENT}) encoded_room = quote(MATRIX_ROOM, safe="") url = f"{MATRIX_HOMESERVER}/_matrix/client/r0/rooms/{encoded_room}/send/m.room.message" body = {"msgtype": "m.text", "body": message} for attempt in (1, 2): token = _get_matrix_token(session, force_refresh=(attempt == 2)) if not token: logger.warning("matrix send aborted: no token (attempt %d)", attempt) return False try: resp = session.post( url, json=body, headers={"Authorization": f"Bearer {token}"}, timeout=MATRIX_TIMEOUT, ) if resp.status_code == 200: logger.info("matrix sent: %s", message[:80]) return True if resp.status_code == 401 and attempt == 1: logger.info("matrix 401 on send; refreshing token and retrying") continue logger.warning("matrix send failed: HTTP %d body=%s", resp.status_code, resp.text[:200]) return False except requests.RequestException as exc: logger.warning("matrix send network error: %s", exc) return False except Exception as exc: logger.warning("matrix send unexpected error: %s", exc) return False def notify(summary: List[Tuple[str, str, str, str]], failed_owners: Optional[List[str]] = None) -> None: """Build a Matrix message and POST it. Non-fatal. `summary` is a list of (owner, repo, short_sha, final_location). `failed_owners` is a list of owners whose repo-list fetch failed this run. """ failed_owners = failed_owners or [] if not summary and not failed_owners: return parts = [] if summary: by_owner: dict = {} for owner, repo, short_sha, _ in summary: by_owner.setdefault(owner, []).append(f"{repo}@{short_sha}") chunks = [f"{owner}: {', '.join(items)}" for owner, items in by_owner.items()] parts.append(f"coe-mirror: {len(summary)} new — " + " | ".join(chunks)) if failed_owners: parts.append(f"{len(failed_owners)} owner(s) failed: {', '.join(failed_owners)}") text = " ; ".join(parts) # Always log the intended message so behavior is visible even when Matrix is down logger.info("notify: %s", text) _send_matrix(text) def main() -> int: """Main entry point.""" setup_logging() lock_fd = acquire_lock() if lock_fd is None: logger.debug("Lock already held; another instance running, exiting gracefully") return 0 try: logger.info("coe-mirror starting") # Probe target directory health target_healthy = probe_target() logger.info("target directory health check: %s", "healthy" if target_healthy else "unavailable") # Flush pending at start initial_flush = flush_pending(target_healthy) # Create session and iterate owners session = requests.Session() session.headers.update({"User-Agent": USER_AGENT}) new_files = list(initial_flush) failed_owners: List[str] = [] for owner in OWNERS: if not _safe_repo_name(owner): logger.warning("refusing unsafe owner name: %r", owner) failed_owners.append(owner) continue repos = fetch_repos(session, owner) if repos is None: logger.error("Could not fetch repo list for %s; continuing to next owner", owner) failed_owners.append(owner) continue logger.info("Fetched %d repos for %s", len(repos), owner) # Ensure per-owner target subdir exists (best-effort; probe already covered parent) owner_target = TARGET_DIR / owner if target_healthy: try: owner_target.mkdir(parents=True, exist_ok=True) except OSError as exc: logger.warning("could not create owner subdir %s: %s", owner_target, exc) # Process each repo for this owner for repo in repos: repo_name = repo.get("name") default_branch = repo.get("default_branch", "main") if not repo_name: logger.warning("Repo object missing 'name' (owner=%s)", owner) continue # Reject unsafe repo names to prevent path traversal if not _safe_repo_name(repo_name): logger.warning("refusing unsafe repo name: %r (owner=%s)", repo_name, owner) continue sha = fetch_branch_sha(session, owner, repo_name, default_branch) if sha is None: logger.error("Skipping %s/%s (could not fetch branch SHA)", owner, repo_name) continue short_sha = sha[:SHORT_SHA_LEN] # final name in target: {repo}_{sha8}.zip inside owner subdir target_final = f"{repo_name}_{short_sha}.zip" target_path = owner_target / target_final # pending name carries owner prefix to avoid cross-owner collisions pending_name = f"{owner}__{target_final}" pending_path = PENDING_DIR / pending_name already_have = (target_path.exists() and target_path.stat().st_size > 0) \ or (pending_path.exists() and pending_path.stat().st_size > 0) if already_have: logger.info("Skipping %s (already present)", pending_name) continue downloaded = download_repo_zip(session, owner, repo_name, sha) if downloaded: # Try to place it if target_healthy: dst = owner_target / target_final if place_into_target(downloaded, dst): new_files.append((owner, repo_name, short_sha, str(dst))) else: logger.info("File %s left in pending (placement failed)", downloaded.name) new_files.append((owner, repo_name, short_sha, str(downloaded))) else: logger.info("target directory unhealthy; %s queued in pending", downloaded.name) new_files.append((owner, repo_name, short_sha, str(downloaded))) # Flush pending at end final_flush = flush_pending(target_healthy) new_files.extend(final_flush) # Notify notify(new_files, failed_owners=failed_owners) logger.info("coe-mirror complete") return 0 finally: release_lock(lock_fd) EXAMPLE_CONFIG = """\ # coe-mirror configuration # Precedence for every key: env var > this file > hardcoded default. # Env var names are COE_MIRROR_; matrix keys are prefixed COE_MIRROR_MATRIX_. api_base = "https://git.churchofmalware.org" # Where mirrored zips are stored. One subdirectory per owner is created here. target_dir = "~/coe-mirror" # Optional: require this path to be a live mountpoint before writing. If set and # not currently mounted, the run queues to pending instead of creating a phantom # local directory at the mountpoint. Leave blank to disable. # mount_root = "~/pCloudDrive" # Optional: local staging when target_dir is unavailable. # pending_dir = "~/.local/share/coe-mirror/pending" # Forgejo owners (users or orgs) to mirror. Edit this list to add/remove. owners = [ "Nightmare_Eclipse", "shai_hulud", "bikini_mirror", "ek0mssavi0r", ] # Optional Matrix notifications. Leave both blank to disable. # User/password come from COE_MIRROR_MATRIX_USER / COE_MIRROR_MATRIX_PASSWORD env vars # (or the `creds` binary if you use Infisical). [matrix] homeserver = "" room = "" """ def cmd_config_init() -> int: """Write an example config to CONFIG_PATH. Won't overwrite existing files.""" if CONFIG_PATH.exists(): print(f"coe-mirror: config already exists at {CONFIG_PATH}; not overwriting", file=sys.stderr) return 1 CONFIG_PATH.parent.mkdir(parents=True, exist_ok=True) CONFIG_PATH.write_text(EXAMPLE_CONFIG) print(f"coe-mirror: wrote example config to {CONFIG_PATH}") return 0 def cmd_config_path() -> int: """Print the config file path (resolved), regardless of whether it exists.""" print(str(CONFIG_PATH)) return 0 USAGE = """\ Usage: coe_mirror.py [command] Commands: (no args) Run the mirror (default; used by systemd timer) run Same as no args config init Write an example config.toml to the standard path config path Print the config file path -h, --help Show this message Config file: {path} Precedence: env var > config file > hardcoded default """ def dispatch(argv: List[str]) -> int: if not argv or argv[0] == "run": return main() if argv[0] in ("-h", "--help", "help"): print(USAGE.format(path=CONFIG_PATH)) return 0 if argv[0] == "config": sub = argv[1] if len(argv) > 1 else "" if sub == "init": return cmd_config_init() if sub == "path": return cmd_config_path() print(USAGE.format(path=CONFIG_PATH), file=sys.stderr) return 2 print(f"coe-mirror: unknown command {argv[0]!r}", file=sys.stderr) print(USAGE.format(path=CONFIG_PATH), file=sys.stderr) return 2 if __name__ == "__main__": sys.exit(dispatch(sys.argv[1:]))