#!/usr/bin/env python3 """Import validated, repository-owned Labyricorn project records.""" from __future__ import annotations import argparse import configparser import hashlib import ipaddress import io import json import os from pathlib import Path, PurePosixPath import re import shutil import subprocess import sys import tempfile from datetime import date, datetime, timezone from typing import Any from urllib.error import HTTPError, URLError from urllib.parse import quote, urlparse from urllib.request import Request, urlopen from lektor.metaformat import tokenize SCHEMA_VERSION = "1" MAX_FILES = 100 MAX_FILE_SIZE = 5 * 1024 * 1024 MAX_TOTAL_SIZE = 20 * 1024 * 1024 ALLOWED_IMAGE_EXTENSIONS = {".jpg", ".jpeg", ".png", ".webp"} IGNORED_INSTRUCTION_FILES = {"AGENTS.md", "README.md"} SLUG_RE = re.compile(r"^[a-z0-9]+(?:-[a-z0-9]+)*$") COMMIT_RE = re.compile(r"^[0-9a-f]{40}$") RAW_HTML_RE = re.compile(r"<\s*(?:!|/?[A-Za-z])[\s\S]*?>") PROJECT_REQUIRED_FIELDS = { "_model", "schema_version", "project_id", "title", "summary", "status", "started", "author", "repository_url", "default_branch", "tags", "body", } DEVLOG_REQUIRED_FIELDS = {"_model", "schema_version", "title", "summary"} ENTRY_REQUIRED_FIELDS = { "_model", "schema_version", "title", "date", "author", "summary", "tags", "source_commit", "body", } class ProjectSourceError(RuntimeError): """Raised when a remote project cannot be imported safely.""" class RepositoryNotPublicError(ProjectSourceError): """Raised when the provider reports that an allowlisted source is not public.""" def log(message: str) -> None: print(f"project-sync: {message}", flush=True) def public_git_environment(cache_root: Path) -> dict[str, str]: public_home = cache_root / "public-git-home" public_home.mkdir(parents=True, exist_ok=True) environment = os.environ.copy() environment.update( { "HOME": str(public_home), "XDG_CONFIG_HOME": str(public_home / ".config"), "GIT_CONFIG_NOSYSTEM": "1", "GIT_CONFIG_GLOBAL": os.devnull, "GIT_TERMINAL_PROMPT": "0", "GCM_INTERACTIVE": "Never", } ) environment.pop("GIT_ASKPASS", None) environment.pop("SSH_ASKPASS", None) return environment def run_git(git_dir: Path, *args: str, text: bool = True) -> str | bytes: cache_root = git_dir.parent.parent command = ["git", "--git-dir", str(git_dir), *args] result = subprocess.run( command, check=True, capture_output=True, text=text, env=public_git_environment(cache_root), ) return result.stdout def validate_url(value: str, field: str) -> str: parsed = urlparse(value) if parsed.scheme != "https" or not parsed.hostname or parsed.username or parsed.password: raise ProjectSourceError(f"{field} must be a credential-free HTTPS URL") if parsed.hostname.lower() == "localhost": raise ProjectSourceError(f"{field} must not target localhost") try: address = ipaddress.ip_address(parsed.hostname) except ValueError: pass else: if not address.is_global: raise ProjectSourceError(f"{field} must not target a private or local address") return value.rstrip("/") def load_registry(site_root: Path) -> list[dict[str, str | int]]: registry_path = site_root / "configs" / "project-sources.ini" parser = configparser.ConfigParser(interpolation=None) if not parser.read(registry_path, encoding="utf-8"): raise ProjectSourceError(f"project source registry is missing: {registry_path}") sources: list[dict[str, str | int]] = [] for project_id in parser.sections(): if not SLUG_RE.fullmatch(project_id): raise ProjectSourceError(f"invalid project id in registry: {project_id}") section = parser[project_id] required = {"repository", "web_url", "api_url", "branch"} missing = required - set(section) if missing: raise ProjectSourceError(f"{project_id}: missing registry fields {sorted(missing)}") branch = section["branch"].strip() if not re.fullmatch(r"[A-Za-z0-9._/-]+", branch) or ".." in branch: raise ProjectSourceError(f"{project_id}: invalid branch") source: dict[str, str | int] = { "project_id": project_id, "repository": validate_url(section["repository"].strip(), "repository"), "web_url": validate_url(section["web_url"].strip(), "web_url"), "api_url": validate_url(section["api_url"].strip(), "api_url"), "branch": branch, } if "featured_order" in section: try: source["featured_order"] = int(section["featured_order"]) except ValueError as exc: raise ProjectSourceError( f"{project_id}: featured_order must be an integer" ) from exc sources.append(source) if not sources: raise ProjectSourceError("project source registry is empty") return sources def load_state(cache_root: Path) -> dict[str, Any]: path = cache_root / "state.json" if not path.exists(): return {"version": 1, "projects": {}} try: state = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as exc: raise ProjectSourceError(f"cannot read project cache state: {exc}") from exc if state.get("version") != 1 or not isinstance(state.get("projects"), dict): raise ProjectSourceError("unsupported project cache state") return state def write_state(cache_root: Path, state: dict[str, Any]) -> None: cache_root.mkdir(parents=True, exist_ok=True) fd, temporary_name = tempfile.mkstemp(prefix="state.", suffix=".json", dir=cache_root) try: with os.fdopen(fd, "w", encoding="utf-8", newline="\n") as stream: json.dump(state, stream, indent=2, sort_keys=True) stream.write("\n") os.replace(temporary_name, cache_root / "state.json") finally: if os.path.exists(temporary_name): os.unlink(temporary_name) def ensure_mirror(source: dict[str, str], cache_root: Path) -> tuple[Path, bool]: repos_root = cache_root / "repos" repos_root.mkdir(parents=True, exist_ok=True) mirror = repos_root / f"{source['project_id']}.git" fetched = True try: if not mirror.exists(): subprocess.run( ["git", "clone", "--mirror", source["repository"], str(mirror)], check=True, capture_output=True, text=True, env=public_git_environment(cache_root), ) else: configured_url = str(run_git(mirror, "remote", "get-url", "origin")).strip() if configured_url.rstrip("/") != source["repository"]: raise ProjectSourceError( f"{source['project_id']}: cached mirror URL does not match registry" ) run_git(mirror, "fetch", "--prune", "origin") except (subprocess.CalledProcessError, OSError) as exc: fetched = False if not mirror.exists(): raise ProjectSourceError( f"{source['project_id']}: initial repository fetch failed" ) from exc log(f"{source['project_id']}: fetch unavailable; considering last-known-good snapshot") return mirror, fetched def resolve_branch(mirror: Path, branch: str) -> str: candidates = (f"refs/heads/{branch}", f"refs/remotes/origin/{branch}") for candidate in candidates: try: commit = str(run_git(mirror, "rev-parse", "--verify", f"{candidate}^{{commit}}")).strip() except subprocess.CalledProcessError: continue if COMMIT_RE.fullmatch(commit): return commit raise ProjectSourceError(f"remote branch does not resolve to a commit: {branch}") def git_file(mirror: Path, commit: str, path: str) -> bytes: try: return bytes(run_git(mirror, "show", f"{commit}:{path}", text=False)) except subprocess.CalledProcessError as exc: raise ProjectSourceError(f"missing repository file: {path}") from exc def list_publishable_files(mirror: Path, commit: str) -> dict[str, int]: raw = bytes( run_git( mirror, "-c", "core.quotepath=false", "ls-tree", "-r", "-l", "-z", commit, "--", ".labyricorn", text=False, ) ) accepted: dict[str, int] = {} total_size = 0 for row in raw.split(b"\0"): if not row: continue header, encoded_path = row.split(b"\t", 1) mode, object_type, _object_id, size_text = header.decode("ascii").split(" ", 3) path = encoded_path.decode("utf-8") parts = PurePosixPath(path).parts if not parts or parts[0] != ".labyricorn" or ".." in parts: raise ProjectSourceError(f"unsafe publishing path: {path}") if object_type != "blob" or mode not in {"100644", "100664"}: raise ProjectSourceError(f"unsupported Git object or file mode: {path} ({mode})") try: size = int(size_text) except ValueError as exc: raise ProjectSourceError(f"unknown file size: {path}") from exc if size > MAX_FILE_SIZE: raise ProjectSourceError(f"publishing file exceeds {MAX_FILE_SIZE} bytes: {path}") relative = PurePosixPath(*parts[1:]) if relative.name in IGNORED_INSTRUCTION_FILES: continue is_project_record = relative == PurePosixPath("project/contents.lr") is_devlog_index = relative == PurePosixPath("devlog/contents.lr") is_devlog_entry = ( len(relative.parts) == 3 and relative.parts[0] == "devlog" and SLUG_RE.fullmatch(relative.parts[1]) is not None and relative.parts[2] == "contents.lr" ) extension = relative.suffix.lower() is_project_image = len(relative.parts) == 2 and relative.parts[0] == "project" is_entry_image = ( len(relative.parts) == 3 and relative.parts[0] == "devlog" and SLUG_RE.fullmatch(relative.parts[1]) is not None ) is_image = extension in ALLOWED_IMAGE_EXTENSIONS and (is_project_image or is_entry_image) if not (is_project_record or is_devlog_index or is_devlog_entry or is_image): raise ProjectSourceError(f"unsupported publishing file: {path}") accepted[path] = size total_size += size if len(accepted) > MAX_FILES: raise ProjectSourceError(f"publishing tree exceeds {MAX_FILES} imported files") if total_size > MAX_TOTAL_SIZE: raise ProjectSourceError(f"publishing tree exceeds {MAX_TOTAL_SIZE} imported bytes") required_paths = { ".labyricorn/project/contents.lr", ".labyricorn/devlog/contents.lr", } missing = required_paths - set(accepted) if missing: raise ProjectSourceError(f"publishing tree is missing {sorted(missing)}") return accepted def parse_record(data: bytes, path: str) -> tuple[dict[str, str], str]: try: text = data.decode("utf-8") except UnicodeDecodeError as exc: raise ProjectSourceError(f"record is not UTF-8: {path}") from exc tokens = list(tokenize(io.StringIO(text))) keys = [key for key, _lines in tokens] if len(keys) != len(set(keys)): raise ProjectSourceError(f"record contains duplicate fields: {path}") values = {key: "".join(lines).strip() for key, lines in tokens} for key, value in values.items(): if key not in {"repository_url"} and RAW_HTML_RE.search(value): raise ProjectSourceError(f"raw HTML is not allowed in {path}:{key}") return values, text def load_tag_vocabulary(site_root: Path) -> set[str]: tags_root = site_root / "content" / "tags" vocabulary: set[str] = set() if not tags_root.is_dir(): raise ProjectSourceError("controlled tag vocabulary is missing") for record_path in tags_root.glob("*/contents.lr"): slug = record_path.parent.name if not SLUG_RE.fullmatch(slug): raise ProjectSourceError(f"invalid controlled tag slug: {slug}") record, _text = parse_record(record_path.read_bytes(), str(record_path)) if record.get("_model") != "tag" or not record.get("title"): raise ProjectSourceError(f"invalid controlled tag record: {record_path}") vocabulary.add(slug) if not vocabulary: raise ProjectSourceError("controlled tag vocabulary is empty") return vocabulary def normalize_tag_value(value: str) -> str: return re.sub(r"[^a-z0-9]+", "-", value.casefold()).strip("-") def approved_tag_slugs( project_id: str, source_name: str, values: list[str], vocabulary: set[str] ) -> list[str]: approved: list[str] = [] for value in values: slug = normalize_tag_value(value) if slug not in vocabulary: log(f"{project_id}: ignoring unapproved {source_name} tag {value!r}") continue if slug not in approved: approved.append(slug) return approved def taxonomy_label_items(values: list[str], approved_slugs: list[str]) -> list[str]: approved = set(approved_slugs) return [ f"{normalize_tag_value(value) if normalize_tag_value(value) in approved else '-'}\t{value.replace(chr(9), ' ')}" for value in values ] def validate_image(path: str, data: bytes) -> None: suffix = PurePosixPath(path).suffix.lower() signatures = { ".png": (b"\x89PNG\r\n\x1a\n",), ".jpg": (b"\xff\xd8\xff",), ".jpeg": (b"\xff\xd8\xff",), ".webp": (b"RIFF",), } if not any(data.startswith(prefix) for prefix in signatures[suffix]): raise ProjectSourceError(f"image signature does not match extension: {path}") if suffix == ".webp" and data[8:12] != b"WEBP": raise ProjectSourceError(f"image signature does not match extension: {path}") def validate_snapshot( source: dict[str, str], mirror: Path, commit: str ) -> tuple[dict[str, bytes], dict[str, dict[str, str]]]: paths = list_publishable_files(mirror, commit) files = {path: git_file(mirror, commit, path) for path in paths} records: dict[str, dict[str, str]] = {} for path, data in files.items(): if path.endswith("contents.lr"): records[path], _text = parse_record(data, path) else: validate_image(path, data) project_path = ".labyricorn/project/contents.lr" project = records[project_path] missing = PROJECT_REQUIRED_FIELDS - set(project) if missing: raise ProjectSourceError(f"project record is missing {sorted(missing)}") if project["_model"] != "project" or project["schema_version"] != SCHEMA_VERSION: raise ProjectSourceError("project record model or schema version is unsupported") if project["project_id"] != source["project_id"]: raise ProjectSourceError("project record ID does not match registry") if project["repository_url"].rstrip("/") != source["web_url"]: raise ProjectSourceError("project repository URL does not match registry") if project["default_branch"] != source["branch"]: raise ProjectSourceError("project default branch does not match registry") date.fromisoformat(project["started"]) logo = project.get("logo") if logo: logo_path = f".labyricorn/project/{logo}" if logo_path not in files or PurePosixPath(logo).name != logo: raise ProjectSourceError("project logo is missing or not a local attachment") devlog_path = ".labyricorn/devlog/contents.lr" devlog = records[devlog_path] missing = DEVLOG_REQUIRED_FIELDS - set(devlog) if missing: raise ProjectSourceError(f"devlog record is missing {sorted(missing)}") if devlog["_model"] != "devlog" or devlog["schema_version"] != SCHEMA_VERSION: raise ProjectSourceError("devlog record model or schema version is unsupported") for path, record in records.items(): if path in {project_path, devlog_path}: continue missing = ENTRY_REQUIRED_FIELDS - set(record) if missing: raise ProjectSourceError(f"{path}: missing fields {sorted(missing)}") if record["_model"] != "devlog-entry" or record["schema_version"] != SCHEMA_VERSION: raise ProjectSourceError(f"{path}: model or schema version is unsupported") date.fromisoformat(record["date"]) source_commit = record["source_commit"] if not COMMIT_RE.fullmatch(source_commit): raise ProjectSourceError(f"{path}: source_commit must be a full commit ID") try: run_git(mirror, "merge-base", "--is-ancestor", source_commit, commit) except subprocess.CalledProcessError as exc: raise ProjectSourceError(f"{path}: source_commit is not in the imported history") from exc return files, records def fetch_json( url: str, *, allow_not_found: bool = False, require_public: bool = False ) -> Any: request = Request(url, headers={"Accept": "application/json", "User-Agent": "LabyricornProjectSync/1"}) try: with urlopen(request, timeout=10) as response: return json.load(response) except HTTPError as exc: if allow_not_found and exc.code == 404: return None if require_public and exc.code in {401, 403, 404}: raise RepositoryNotPublicError( "repository API is not anonymously accessible" ) from exc raise ProjectSourceError(f"metadata request failed with HTTP {exc.code}: {url}") from exc except (URLError, TimeoutError, json.JSONDecodeError) as exc: raise ProjectSourceError(f"metadata request failed: {url}") from exc def commit_metadata(mirror: Path, commit: str) -> dict[str, str]: raw = str(run_git(mirror, "show", "-s", "--format=%H%x00%an%x00%aI%x00%s", commit)).rstrip("\n") commit_id, author, committed_at, subject = raw.split("\0", 3) lektor_datetime = datetime.fromisoformat(committed_at).strftime("%Y-%m-%d %H:%M:%S %z") return { "commit": commit_id, "commit_short": commit_id[:10], "commit_author": author, "commit_date": lektor_datetime, "commit_message": subject, } def detect_license(mirror: Path, commit: str) -> str: try: package = json.loads(git_file(mirror, commit, "package.json")) value = package.get("license") if isinstance(value, str) and value.strip(): return value.strip() except (ProjectSourceError, json.JSONDecodeError, UnicodeDecodeError): pass return "Not declared" def collect_metadata( source: dict[str, str], mirror: Path, commit: str, previous: dict[str, Any] | None ) -> tuple[dict[str, Any], bool]: metadata: dict[str, Any] = commit_metadata(mirror, commit) metadata.update( { "repository_url": source["web_url"], "readme_url": f"{source['web_url']}/src/branch/{quote(source['branch'], safe='')}/README.md", "commit_url": f"{source['web_url']}/commit/{commit}", "default_branch": source["branch"], "license": detect_license(mirror, commit), "languages": [], "open_issues": 0, "stars": 0, "forks": 0, "latest_release": "", "latest_release_url": "", } ) provider_current = True try: repository = fetch_json(source["api_url"], require_public=True) if repository.get("private") is not False: raise RepositoryNotPublicError( f"{source['project_id']}: source repository is not public" ) if repository.get("default_branch") != source["branch"]: raise ProjectSourceError(f"{source['project_id']}: API default branch differs from registry") metadata["open_issues"] = int(repository.get("open_issues_count") or 0) metadata["stars"] = int(repository.get("stars_count") or 0) metadata["forks"] = int(repository.get("forks_count") or 0) languages = fetch_json(f"{source['api_url']}/languages") if isinstance(languages, dict): metadata["languages"] = [ name for name, _size in sorted(languages.items(), key=lambda item: item[1], reverse=True)[:5] ] release = fetch_json(f"{source['api_url']}/releases/latest", allow_not_found=True) if release: metadata["latest_release"] = str(release.get("tag_name") or release.get("name") or "") metadata["latest_release_url"] = str(release.get("html_url") or "") except RepositoryNotPublicError: raise except ProjectSourceError: provider_current = False cached = (previous or {}).get("metadata") if isinstance(cached, dict): for key in ( "open_issues", "stars", "forks", "languages", "latest_release", "latest_release_url", ): if key in cached: metadata[key] = cached[key] log(f"{source['project_id']}: provider metadata unavailable; using Git and cached metadata") return metadata, provider_current def scalar(value: Any) -> str: result = str(value).replace("\r", " ").replace("\n", " ").strip() if result == "---": result = "—" return result def split_list(value: str) -> list[str]: return [item.strip() for item in re.split(r"[,\n]", value) if item.strip()] def serialize_field(key: str, value: Any) -> str: if isinstance(value, list): lines = "\n".join(scalar(item) for item in value) return f"\n---\n{key}:\n\n{lines}" return f"\n---\n{key}: {scalar(value)}" def append_fields(original: bytes, fields: dict[str, Any]) -> bytes: text = original.decode("utf-8").rstrip("\r\n") additions = "".join(serialize_field(key, value) for key, value in fields.items()) return (text + additions + "\n").encode("utf-8") def materialize( source: dict[str, str | int], destination_root: Path, files: dict[str, bytes], records: dict[str, dict[str, str]], metadata: dict[str, Any], synchronized_at: str, sync_status: str, tag_vocabulary: set[str], ) -> None: project_root = destination_root / "content" / "projects" / source["project_id"] if project_root.exists(): raise ProjectSourceError(f"refusing to overwrite existing project path: {project_root}") (project_root / "devlog").mkdir(parents=True) technology_values = split_list( records[".labyricorn/project/contents.lr"]["tags"] ) language_values = list(metadata["languages"]) technology_tag_slugs = approved_tag_slugs( source["project_id"], "technology", technology_values, tag_vocabulary, ) repository_language_tags = approved_tag_slugs( source["project_id"], "language", language_values, tag_vocabulary, ) taxonomy_tags = list( dict.fromkeys(technology_tag_slugs + repository_language_tags) ) project_fields = { "kicker": "Project", "date": records[".labyricorn/project/contents.lr"]["started"], "repository_readme_url": metadata["readme_url"], "repository_commit": metadata["commit"], "repository_commit_short": metadata["commit_short"], "repository_commit_url": metadata["commit_url"], "repository_commit_message": metadata["commit_message"], "repository_commit_author": metadata["commit_author"], "repository_commit_date": metadata["commit_date"], "repository_license": metadata["license"], "repository_languages": metadata["languages"], "repository_open_issues": metadata["open_issues"], "repository_stars": metadata["stars"], "repository_forks": metadata["forks"], "repository_latest_release": metadata["latest_release"], "repository_latest_release_url": metadata["latest_release_url"], "synchronized_at": synchronized_at, "sync_status": sync_status, "technology_tags": technology_values, "technology_labels": taxonomy_label_items( technology_values, technology_tag_slugs ), "technology_tag_slugs": technology_tag_slugs, "repository_language_labels": taxonomy_label_items( language_values, repository_language_tags ), "repository_language_tags": repository_language_tags, "taxonomy_tags": taxonomy_tags, } if "featured_order" in source: project_fields["featured"] = "yes" project_fields["featured_order"] = source["featured_order"] for source_path, data in files.items(): relative = PurePosixPath(source_path).relative_to(".labyricorn") if relative.parts[0] == "project": target_relative = PurePosixPath(*relative.parts[1:]) target = project_root.joinpath(*target_relative.parts) else: target = project_root.joinpath(*relative.parts) target.parent.mkdir(parents=True, exist_ok=True) output = data if source_path == ".labyricorn/project/contents.lr": output = append_fields(data, project_fields) elif source_path.startswith(".labyricorn/devlog/") and source_path.endswith("/contents.lr"): record = records[source_path] if record.get("_model") == "devlog-entry": entry_commit = record["source_commit"] entry_meta = commit_metadata(Path(metadata["mirror_path"]), entry_commit) topic_values = split_list(record["tags"]) topic_tag_slugs = approved_tag_slugs( source["project_id"], "devlog topic", topic_values, tag_vocabulary, ) output = append_fields( data, { "kicker": "Devlog", "source_commit_url": f"{source['web_url']}/commit/{entry_commit}", "source_commit_time": entry_meta["commit_date"], "topic_tags": topic_values, "topic_labels": taxonomy_label_items( topic_values, topic_tag_slugs ), "topic_tag_slugs": topic_tag_slugs, }, ) target.write_bytes(output) def metadata_digest(metadata: dict[str, Any]) -> str: public_metadata = {key: value for key, value in metadata.items() if key != "mirror_path"} payload = json.dumps(public_metadata, sort_keys=True, separators=(",", ":")).encode("utf-8") return hashlib.sha256(payload).hexdigest() def sync_projects(site_root: Path, destination_root: Path, cache_root: Path) -> dict[str, Any]: sources = load_registry(site_root) tag_vocabulary = load_tag_vocabulary(site_root) cache_root.mkdir(parents=True, exist_ok=True) state = load_state(cache_root) next_projects_state = dict(state["projects"]) manifest_projects: dict[str, Any] = {} for source in sources: project_id = source["project_id"] previous = state["projects"].get(project_id) mirror, fetched = ensure_mirror(source, cache_root) candidate: str try: candidate = resolve_branch(mirror, source["branch"]) except ProjectSourceError: if previous and COMMIT_RE.fullmatch(str(previous.get("commit", ""))): candidate = str(previous["commit"]) fetched = False else: raise sync_status = "current" if fetched else "last-known-good" try: files, records = validate_snapshot(source, mirror, candidate) except (ProjectSourceError, ValueError) as exc: previous_commit = str((previous or {}).get("commit", "")) if not previous_commit or previous_commit == candidate or not COMMIT_RE.fullmatch(previous_commit): raise ProjectSourceError(f"{project_id}: no valid snapshot is available: {exc}") from exc log(f"{project_id}: new snapshot rejected; preserving {previous_commit[:10]}") candidate = previous_commit files, records = validate_snapshot(source, mirror, candidate) sync_status = "last-known-good" metadata, provider_current = collect_metadata(source, mirror, candidate, previous) metadata["mirror_path"] = str(mirror) if not provider_current: sync_status = "last-known-good" if not fetched else "metadata-cached" digest = metadata_digest(metadata) previous_digest = str((previous or {}).get("metadata_digest", "")) if previous and previous.get("commit") == candidate and previous_digest == digest: synchronized_at = str(previous["synchronized_at"]) else: synchronized_at = datetime.now(timezone.utc).replace(microsecond=0).strftime( "%Y-%m-%d %H:%M:%S %z" ) materialize( source, destination_root, files, records, metadata, synchronized_at, sync_status, tag_vocabulary, ) public_metadata = {key: value for key, value in metadata.items() if key != "mirror_path"} next_projects_state[project_id] = { "commit": candidate, "metadata": public_metadata, "metadata_digest": digest, "synchronized_at": synchronized_at, } manifest_projects[project_id] = { "repository": source["web_url"], "branch": source["branch"], "commit": candidate, "metadata_digest": digest, "synchronized_at": synchronized_at, "status": sync_status, } log(f"{project_id}: {sync_status} commit={candidate[:10]}") next_state = {"version": 1, "projects": next_projects_state} write_state(cache_root, next_state) return {"version": 1, "projects": manifest_projects} def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--site-root", type=Path, default=Path(__file__).resolve().parents[1]) parser.add_argument("--destination-root", type=Path, required=True) parser.add_argument("--cache-root", type=Path, required=True) parser.add_argument("--manifest", type=Path) args = parser.parse_args() try: manifest = sync_projects( args.site_root.resolve(), args.destination_root.resolve(), args.cache_root.resolve(), ) if args.manifest: args.manifest.write_text(json.dumps(manifest, indent=2, sort_keys=True) + "\n", encoding="utf-8") except (ProjectSourceError, subprocess.CalledProcessError, OSError, ValueError) as exc: print(f"project-sync: ERROR: {exc}", file=sys.stderr) return 1 return 0 if __name__ == "__main__": raise SystemExit(main())