This commit is contained in:
@@ -0,0 +1,692 @@
|
||||
#!/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]]:
|
||||
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]] = []
|
||||
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")
|
||||
sources.append(
|
||||
{
|
||||
"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 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 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],
|
||||
destination_root: Path,
|
||||
files: dict[str, bytes],
|
||||
records: dict[str, dict[str, str]],
|
||||
metadata: dict[str, Any],
|
||||
synchronized_at: str,
|
||||
sync_status: 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)
|
||||
|
||||
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": split_list(records[".labyricorn/project/contents.lr"]["tags"]),
|
||||
}
|
||||
|
||||
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)
|
||||
output = append_fields(
|
||||
data,
|
||||
{
|
||||
"source_commit_url": f"{source['web_url']}/commit/{entry_commit}",
|
||||
"source_commit_time": entry_meta["commit_date"],
|
||||
"topic_tags": split_list(record["tags"]),
|
||||
},
|
||||
)
|
||||
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)
|
||||
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,
|
||||
)
|
||||
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())
|
||||
Reference in New Issue
Block a user