agripreserve/sync_repos.py

#!/usr/bin/env python3
"""
Sync GitHub repos' portfolio.json into the portfolio backend.

Features
- Reads repos from repos.json (or --config)
- Fetches portfolio.json via GitHub raw
- Optionally downloads referenced assets to assets/projects/ and rewrites URLs
- Upserts to backend via POST /api/projects or PUT /api/projects/{id}

Environment variables
- PORTFOLIO_API_BASE: API base URL (default: http://localhost:8000/api)
- PORTFOLIO_API_TOKEN: Optional bearer token for backend auth
- REPOS_CONFIG: Path to repos.json (default: repos.json)
- SYNC_DOWNLOAD_ASSETS: If 'true', download assets locally (default: false)
- ASSET_BASE_PATH: Local dir to store assets (default: assets/projects)
- ASSET_PREFIX_URL: URL prefix to reference local assets (default: /assets/projects)
- REQUEST_TIMEOUT: HTTP timeout seconds (default: 20)

repos.json schema
[
  { "repo": "https://github.com/owner/name", "branch": "main", "path": "portfolio.json" }
]

Assumptions
- Project identifier is `slug` if present, else derived from `title`.
- Backend supports POST /api/projects and PUT /api/projects/{id} where id is slug.
"""
from __future__ import annotations

import argparse
import json
import os
import re
import sys
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple

import requests

DEFAULT_API_BASE = os.environ.get("PORTFOLIO_API_BASE", "http://localhost:8000/api")
DEFAULT_CONFIG = os.environ.get("REPOS_CONFIG", "repos.json")
DEFAULT_TIMEOUT = float(os.environ.get("REQUEST_TIMEOUT", "20"))
DOWNLOAD_ASSETS = os.environ.get("SYNC_DOWNLOAD_ASSETS", "false").lower() == "true"
ASSET_BASE_PATH = Path(os.environ.get("ASSET_BASE_PATH", "assets/projects"))
ASSET_PREFIX_URL = os.environ.get("ASSET_PREFIX_URL", "/assets/projects")
API_TOKEN = os.environ.get("PORTFOLIO_API_TOKEN")

SESSION = requests.Session()
if API_TOKEN:
    SESSION.headers.update({"Authorization": f"Bearer {API_TOKEN}"})
SESSION.headers.update({"Content-Type": "application/json"})


def slugify(text: str) -> str:
    text = text.strip().lower()
    text = re.sub(r"[^a-z0-9\-\s_]", "", text)
    text = re.sub(r"[\s_]+", "-", text)
    return text


def read_repos_config(path: Path) -> List[Dict[str, Any]]:
    with path.open("r", encoding="utf-8") as f:
        data = json.load(f)
    if not isinstance(data, list):
        raise ValueError("repos.json must be a list")
    return data


def make_raw_url(repo_url: str, branch: str, file_path: str) -> str:
    repo_url = repo_url.rstrip("/")
    if repo_url.startswith("https://github.com/"):
        owner_name = repo_url[len("https://github.com/") :]
        return f"https://raw.githubusercontent.com/{owner_name}/{branch}/{file_path}"
    if repo_url.startswith("git@github.com:"):
        owner_name = repo_url[len("git@github.com:") :].removesuffix(".git")
        return f"https://raw.githubusercontent.com/{owner_name}/{branch}/{file_path}"
    # Fallback: assume raw already
    return repo_url


def fetch_portfolio_json(raw_url: str, timeout: float = DEFAULT_TIMEOUT) -> Dict[str, Any]:
    resp = SESSION.get(raw_url, timeout=timeout)
    resp.raise_for_status()
    return resp.json()


def ensure_slug(project: Dict[str, Any]) -> str:
    slug = project.get("slug")
    if slug:
        return slug
    title = project.get("title")
    if not title:
        raise ValueError("Project must include either 'slug' or 'title'")
    slug = slugify(title)
    project["slug"] = slug
    return slug


def infer_assets(project: Dict[str, Any]) -> List[Tuple[str, str]]:
    """Return list of (field_path, url) for known asset fields."""
    assets: List[Tuple[str, str]] = []
    preview = project.get("brassFramedPreview")
    if isinstance(preview, str) and preview.startswith("http"):
        assets.append(("brassFramedPreview", preview))
    if isinstance(project.get("cogwheelSchema"), list):
        for i, url in enumerate(project["cogwheelSchema"]):
            if isinstance(url, str) and url.startswith("http"):
                assets.append((f"cogwheelSchema[{i}]", url))
    return assets


def download_asset(url: str, dest_dir: Path, name_hint: Optional[str] = None) -> str:
    dest_dir.mkdir(parents=True, exist_ok=True)
    filename = name_hint or url.split("/")[-1]
    target = dest_dir / filename
    r = SESSION.get(url, timeout=DEFAULT_TIMEOUT)
    r.raise_for_status()
    with target.open("wb") as f:
        f.write(r.content)
    rel_url = f"{ASSET_PREFIX_URL}/{target.name}"
    return rel_url


def maybe_localize_assets(project: Dict[str, Any]) -> None:
    if not DOWNLOAD_ASSETS:
        return
    assets = infer_assets(project)
    for field_path, url in assets:
        name_hint = None
        if "brassFramedPreview" in field_path:
            name_hint = f"{project.get('slug','preview')}_preview{Path(url).suffix or ''}"
        elif field_path.startswith("cogwheelSchema["):
            idx = re.search(r"\[(\d+)\]", field_path)
            suffix = Path(url).suffix or ""
            name_hint = f"{project.get('slug','schema')}_schema_{idx.group(1) if idx else '0'}{suffix}"
        new_url = download_asset(url, ASSET_BASE_PATH, name_hint=name_hint)
        # Rewrite project JSON
        if field_path == "brassFramedPreview":
            project["brassFramedPreview"] = new_url
        elif field_path.startswith("cogwheelSchema["):
            m = re.search(r"\[(\d+)\]", field_path)
            if m:
                i = int(m.group(1))
                arr = project.get("cogwheelSchema") or []
                if i < len(arr):
                    arr[i] = new_url
                else:
                    # pad if necessary
                    while len(arr) <= i:
                        arr.append("")
                    arr[i] = new_url
                project["cogwheelSchema"] = arr


def upsert_project(api_base: str, project: Dict[str, Any]) -> Dict[str, Any]:
    slug = ensure_slug(project)
    # Try POST first
    url_post = f"{api_base.rstrip('/')}/projects"
    resp = SESSION.post(url_post, data=json.dumps(project), timeout=DEFAULT_TIMEOUT)
    if resp.status_code in (200, 201):
        return resp.json() if resp.content else {"status": "created", "slug": slug}
    # If conflict or already exists, try PUT by slug
    if resp.status_code in (400, 409, 422):
        url_put = f"{api_base.rstrip('/')}/projects/{slug}"
        resp2 = SESSION.put(url_put, data=json.dumps(project), timeout=DEFAULT_TIMEOUT)
        resp2.raise_for_status()
        return resp2.json() if resp2.content else {"status": "updated", "slug": slug}
    resp.raise_for_status()
    return {"status": "unknown", "slug": slug}


def process_repo(entry: Dict[str, Any], api_base: str) -> Tuple[str, str]:
    repo = entry["repo"]
    branch = entry.get("branch", "main")
    path = entry.get("path", "portfolio.json")
    raw_url = make_raw_url(repo, branch, path)
    pj = fetch_portfolio_json(raw_url)
    ensure_slug(pj)
    maybe_localize_assets(pj)
    result = upsert_project(api_base, pj)
    return (pj["slug"], result.get("status", "ok"))


def main(argv: Optional[List[str]] = None) -> int:
    parser = argparse.ArgumentParser(description="Sync portfolio projects from repos")
    parser.add_argument("--config", default=DEFAULT_CONFIG, help="Path to repos.json")
    parser.add_argument("--api", default=DEFAULT_API_BASE, help="Backend API base URL")
    parser.add_argument("--download-assets", action="store_true", help="Download and localize assets")
    args = parser.parse_args(argv)

    if args.download_assets:
        global DOWNLOAD_ASSETS
        DOWNLOAD_ASSETS = True

    config_path = Path(args.config)
    if not config_path.exists():
        print(f"Config not found: {config_path}", file=sys.stderr)
        return 2

    try:
        repos = read_repos_config(config_path)
    except Exception as e:
        print(f"Failed to read config: {e}", file=sys.stderr)
        return 2

    successes: List[Tuple[str, str]] = []
    failures: List[Tuple[str, str]] = []

    for entry in repos:
        try:
            slug, status = process_repo(entry, args.api)
            print(f"[{status}] {slug}")
            successes.append((slug, status))
        except Exception as e:
            repo = entry.get("repo", "<unknown>")
            print(f"[error] {repo}: {e}", file=sys.stderr)
            failures.append((repo, str(e)))

    print(f"Done. {len(successes)} succeeded, {len(failures)} failed.")
    return 0 if not failures else 1


if __name__ == "__main__":
    raise SystemExit(main())