2f26f2d836
- onboard-client.py: client replica scaffolding CLI - gws/: Google Workspace sync (Gmail, Calendar, Drive) - lifestream/: life event stream collector - muscriptor-mcp/: audio → MIDI MCP server - music-mcp/: music library MCP server - data_sharing/: consent-gated data sharing (Python + TS) - sync-mirrors.py: GitHub → Forgejo mirror engine - brain-to-gbrain.py, vault-sync.py, test-all.sh - shared/: TS data-sharing library + index - dirac: provider registry update - .gitignore: exclude Rust build artifacts Co-authored-by: Álvaro de Campos <campos@portugalfuturista.org>
788 lines
34 KiB
Python
788 lines
34 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Mirror pipeline: replicate upstream sources (GitHub, Jira, Confluence) into
|
|
self-hosted mirrors (Forgejo, Plane, Outline).
|
|
|
|
Single source of truth: .aurelio/mirrors/sync-mirrors.yaml
|
|
Generated dist: .aurelio/mirrors/dist/*.json (catalog for surfaces)
|
|
|
|
Three modes:
|
|
--write-in-place Regenerate dist/ mirrors from the YAML registry.
|
|
--check CI guard: exit 1 if dist/ is stale.
|
|
--sync [TARGET] Run live mirror sync (forgejo | plane | outline | all).
|
|
--emit-trajectory-rewards Append RL reward signals from last sync drift.
|
|
|
|
Auth: tokens are read from environment variables named in the YAML (token_env).
|
|
They are NEVER hardcoded. Fetch from Vaultwarden and export before --sync.
|
|
|
|
Usage:
|
|
export FORGEJO_MIRROR_TOKEN=...
|
|
export GITHUB_MIRROR_TOKEN=...
|
|
python3 scripts/sync-mirrors.py --sync forgejo
|
|
|
|
python3 scripts/sync-mirrors.py --write-in-place
|
|
python3 scripts/sync-mirrors.py --check
|
|
|
|
The --sync path is idempotent: re-running only touches what changed. Each
|
|
target's sync returns a summary dict; failures are logged but don't abort
|
|
siblings unless defaults.fail_fast is true.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import sys
|
|
import time
|
|
import urllib.error
|
|
import urllib.request
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
try:
|
|
import yaml
|
|
except ImportError:
|
|
print("PyYAML required: pip install pyyaml", file=sys.stderr)
|
|
sys.exit(2)
|
|
|
|
REPLICA_ROOT = Path(__file__).resolve().parent.parent
|
|
REGISTRY = REPLICA_ROOT / ".aurelio" / "mirrors" / "sync-mirrors.yaml"
|
|
DIST_DIR = REPLICA_ROOT / ".aurelio" / "mirrors" / "dist"
|
|
REWARD_PATH = REPLICA_ROOT / ".aurelio" / "brain" / "trajectory-rewards" / "mirror-sync.jsonl"
|
|
|
|
# ── YAML load ──────────────────────────────────────────────────────────────
|
|
|
|
|
|
def load_registry() -> dict[str, Any]:
|
|
with REGISTRY.open("r", encoding="utf-8") as f:
|
|
return yaml.safe_load(f)
|
|
|
|
|
|
def _env(name: str | None) -> str | None:
|
|
if not name:
|
|
return None
|
|
return os.environ.get(name)
|
|
|
|
|
|
# ── HTTP helper ────────────────────────────────────────────────────────────
|
|
|
|
|
|
def _request(
|
|
method: str,
|
|
url: str,
|
|
*,
|
|
token: str | None = None,
|
|
basic_auth: tuple[str, str] | None = None,
|
|
json_body: dict | None = None,
|
|
accept: str = "application/json",
|
|
timeout: int = 30,
|
|
user_agent: str = "replica-omnisciente-mirror/1.0",
|
|
) -> tuple[int, dict | str]:
|
|
"""Minimal urllib wrapper. Returns (status_code, parsed_json | raw_text)."""
|
|
headers = {"Accept": accept, "User-Agent": user_agent}
|
|
data = None
|
|
if json_body is not None:
|
|
data = json.dumps(json_body).encode()
|
|
headers["Content-Type"] = "application/json"
|
|
if token:
|
|
headers["Authorization"] = f"Bearer {token}"
|
|
if basic_auth:
|
|
import base64
|
|
cred = base64.b64encode(f"{basic_auth[0]}:{basic_auth[1]}".encode()).decode()
|
|
headers["Authorization"] = f"Basic {cred}"
|
|
req = urllib.request.Request(url, data=data, headers=headers, method=method)
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=timeout) as resp:
|
|
body = resp.read().decode("utf-8")
|
|
try:
|
|
return resp.status, json.loads(body)
|
|
except json.JSONDecodeError:
|
|
return resp.status, body
|
|
except urllib.error.HTTPError as e:
|
|
body = e.read().decode("utf-8", errors="replace")
|
|
try:
|
|
return e.code, json.loads(body)
|
|
except json.JSONDecodeError:
|
|
return e.code, body
|
|
except urllib.error.URLError as e:
|
|
return 0, str(e.reason)
|
|
|
|
|
|
# ════════════════════════════════════════════════════════════════════════════
|
|
# FORGEJO: GitHub → Forgejo code mirror
|
|
# ════════════════════════════════════════════════════════════════════════════
|
|
|
|
|
|
def _forgejo_headers(token: str) -> dict:
|
|
return {"Authorization": f"token {token}", "Content-Type": "application/json"}
|
|
|
|
|
|
def _github_list_repos(org: str | None, user: str | None, token: str, ua: str) -> list[dict]:
|
|
"""List repos from a GitHub org or user account."""
|
|
repos = []
|
|
base = "https://api.github.com"
|
|
if org:
|
|
url = f"{base}/orgs/{org}/repos?per_page=100&type=all"
|
|
elif user:
|
|
url = f"{base}/users/{user}/repos?per_page=100&type=all"
|
|
else:
|
|
return []
|
|
page = 1
|
|
while True:
|
|
paged_url = f"{url}&page={page}"
|
|
status, data = _request("GET", paged_url, token=token, user_agent=ua)
|
|
if status != 200 or not isinstance(data, list):
|
|
break
|
|
repos.extend(data)
|
|
if len(data) < 100:
|
|
break
|
|
page += 1
|
|
return repos
|
|
|
|
|
|
def sync_forgejo(reg: dict, target_cfg: dict, dry: bool = False) -> dict:
|
|
"""Mirror GitHub repos into Forgejo using the migration API."""
|
|
forgejo_url = target_cfg["endpoint"]
|
|
forgejo_token = _env(target_cfg.get("token_env", ""))
|
|
default_owner = target_cfg.get("default_owner", "mirror")
|
|
ua = reg.get("defaults", {}).get("user_agent", "replica-mirror/1.0")
|
|
summary: dict[str, Any] = {"target": "forgejo", "mirrored": [], "skipped": [], "errors": []}
|
|
|
|
if not forgejo_token and not dry:
|
|
summary["errors"].append(f"missing env {target_cfg.get('token_env')} — export it first")
|
|
return summary
|
|
|
|
for src in target_cfg.get("sources", []):
|
|
gh_token = _env(src.get("token_env", "")) or ""
|
|
include_forks = src.get("include_forks", False)
|
|
|
|
if dry and not gh_token:
|
|
# In dry-run without a token, report intent without hitting the API.
|
|
summary["mirrored"].append({
|
|
"source": src.get("id"),
|
|
"dry": True,
|
|
"note": f"would list {src.get('type')} and create mirrors",
|
|
})
|
|
continue
|
|
|
|
if src["type"] == "github-org":
|
|
repos = _github_list_repos(org=src["org"], user=None, token=gh_token, ua=ua)
|
|
elif src["type"] == "github-user":
|
|
repos = _github_list_repos(org=None, user=src["user"], token=gh_token, ua=ua)
|
|
# Apply include/exclude globs.
|
|
import fnmatch
|
|
inc = src.get("include_patterns", ["*"])
|
|
exc = src.get("exclude_patterns", [])
|
|
repos = [
|
|
r for r in repos
|
|
if any(fnmatch.fnmatch(r["name"], p) for p in inc)
|
|
and not any(fnmatch.fnmatch(r["name"], p) for p in exc)
|
|
]
|
|
else:
|
|
summary["errors"].append(f"unknown source type {src['type']} for forgejo")
|
|
continue
|
|
|
|
for repo in repos:
|
|
if repo.get("fork") and not include_forks:
|
|
summary["skipped"].append({"repo": repo["full_name"], "reason": "fork"})
|
|
continue
|
|
if repo.get("archived"):
|
|
summary["skipped"].append({"repo": repo["full_name"], "reason": "archived"})
|
|
continue
|
|
|
|
owner = default_owner
|
|
repo_name = repo["name"]
|
|
clone_url = repo["clone_url"] # HTTPS
|
|
if src.get("clone_transport") == "ssh":
|
|
clone_url = repo.get("ssh_url", clone_url)
|
|
|
|
# Check if mirror already exists in Forgejo.
|
|
check_url = f"{forgejo_url}/api/v1/repos/{owner}/{repo_name}"
|
|
if not dry and forgejo_token:
|
|
st, _ = _request("GET", check_url, token=forgejo_token, user_agent=ua)
|
|
if st == 200:
|
|
summary["skipped"].append({"repo": f"{owner}/{repo_name}", "reason": "exists"})
|
|
continue
|
|
|
|
# Create migration (mirror=true makes Forgejo poll upstream).
|
|
migrate_url = f"{forgejo_url}/api/v1/repos/migrate"
|
|
body = {
|
|
"clone_addr": clone_url,
|
|
"repo_owner": owner,
|
|
"repo_name": repo_name,
|
|
"service": "github",
|
|
"auth_token": gh_token or None,
|
|
"mirror": True,
|
|
"private": repo.get("private", False),
|
|
"description": repo.get("description", "")[:255],
|
|
"wiki": False,
|
|
"issues": False,
|
|
"labels": src.get("topic_labels", False),
|
|
"pull_requests": False,
|
|
"releases": True,
|
|
}
|
|
if dry:
|
|
summary["mirrored"].append({"repo": f"{owner}/{repo_name}", "dry": True})
|
|
continue
|
|
|
|
st, resp = _request("POST", migrate_url, token=forgejo_token, json_body=body, user_agent=ua, timeout=120)
|
|
if st in (200, 201):
|
|
summary["mirrored"].append({"repo": f"{owner}/{repo_name}", "id": resp.get("id") if isinstance(resp, dict) else None})
|
|
else:
|
|
summary["errors"].append({"repo": repo["full_name"], "status": st, "detail": str(resp)[:300]})
|
|
|
|
return summary
|
|
|
|
|
|
# ════════════════════════════════════════════════════════════════════════════
|
|
# PLANE: Jira → Plane issue mirror
|
|
# ════════════════════════════════════════════════════════════════════════════
|
|
|
|
|
|
def _jira_get(url: str, email: str, token: str, ua: str, params: str = "") -> dict:
|
|
full = f"{url}{params}"
|
|
st, data = _request("GET", full, basic_auth=(email, token), user_agent=ua)
|
|
return {"status": st, "data": data}
|
|
|
|
|
|
def _plane_headers(token: str) -> dict:
|
|
return {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}
|
|
|
|
|
|
def sync_plane(reg: dict, target_cfg: dict, dry: bool = False) -> dict:
|
|
"""Mirror Jira issues into Plane."""
|
|
plane_url = target_cfg["endpoint"].rstrip("/")
|
|
plane_token = _env(target_cfg.get("token_env", ""))
|
|
ws = target_cfg.get("workspace_slug", "savearth")
|
|
ua = reg.get("defaults", {}).get("user_agent", "replica-mirror/1.0")
|
|
summary: dict[str, Any] = {"target": "plane", "projects": [], "issues": {"synced": 0, "skipped": 0}, "errors": []}
|
|
|
|
if not plane_token and not dry:
|
|
summary["errors"].append(f"missing env {target_cfg.get('token_env')}")
|
|
return summary
|
|
|
|
for src in target_cfg.get("sources", []):
|
|
if src["type"] != "jira-cloud":
|
|
summary["errors"].append(f"unknown source type {src['type']} for plane")
|
|
continue
|
|
|
|
jira_url = src.get("endpoint")
|
|
if not jira_url:
|
|
summary["errors"].append({"source": src["id"], "reason": "Jira endpoint is null — set it in sync-mirrors.yaml"})
|
|
continue
|
|
|
|
jira_token = _env(src.get("token_env", "")) or ""
|
|
jira_email = _env(src.get("email_env", "")) or ""
|
|
if not dry and (not jira_token or not jira_email):
|
|
summary["errors"].append(f"missing Jira creds ({src.get('token_env')} / {src.get('email_env')})")
|
|
continue
|
|
|
|
# 1. List Jira projects.
|
|
projects_resp = _jira_get(jira_url, jira_email, jira_token, ua, "/rest/api/3/project")
|
|
if projects_resp["status"] != 200:
|
|
summary["errors"].append({"step": "jira projects", "status": projects_resp["status"], "detail": str(projects_resp["data"])[:300]})
|
|
continue
|
|
jira_projects = projects_resp["data"] or []
|
|
if src.get("projects"):
|
|
wanted = set(src["projects"])
|
|
jira_projects = [p for p in jira_projects if p.get("key") in wanted]
|
|
|
|
for jp in jira_projects:
|
|
pkey = jp["key"]
|
|
pname = jp.get("name", pkey)
|
|
|
|
# 2. Ensure Plane project exists.
|
|
if dry:
|
|
summary["projects"].append({"key": pkey, "name": pname, "dry": True})
|
|
continue
|
|
|
|
# List existing Plane projects to find by name.
|
|
st, existing = _request("GET", f"{plane_url}/api/v1/workspaces/{ws}/projects/", token=plane_token, user_agent=ua)
|
|
project_id = None
|
|
if st == 200 and isinstance(existing, list):
|
|
for ep in existing:
|
|
if ep.get("name", "").lower() == pname.lower() or ep.get("identifier", "").upper() == pkey:
|
|
project_id = ep.get("id")
|
|
break
|
|
if not project_id:
|
|
body = {"name": pname, "identifier": pkey[:8]}
|
|
st, resp = _request("POST", f"{plane_url}/api/v1/workspaces/{ws}/projects/", token=plane_token, json_body=body, user_agent=ua)
|
|
if st in (200, 201) and isinstance(resp, dict):
|
|
project_id = resp.get("id")
|
|
summary["projects"].append({"key": pkey, "id": project_id, "created": True})
|
|
else:
|
|
summary["errors"].append({"step": f"create plane project {pkey}", "status": st, "detail": str(resp)[:300]})
|
|
continue
|
|
else:
|
|
summary["projects"].append({"key": pkey, "id": project_id, "existed": True})
|
|
|
|
if not project_id:
|
|
continue
|
|
|
|
# 3. Fetch Jira issues for this project.
|
|
jql = f"project = {pkey} ORDER BY updated DESC"
|
|
issues_resp = _jira_get(jira_url, jira_email, jira_token, ua, f"/rest/api/3/search?jql={urllib.parse.quote(jql)}&maxResults=100")
|
|
if issues_resp["status"] != 200:
|
|
summary["errors"].append({"step": f"jira issues {pkey}", "status": issues_resp["status"]})
|
|
continue
|
|
jira_issues = (issues_resp["data"] or {}).get("issues", [])
|
|
|
|
# 4. List existing Plane issues to avoid duplicates (by external_id or name match).
|
|
st, plane_issues = _request("GET", f"{plane_url}/api/v1/workspaces/{ws}/projects/{project_id}/issues/", token=plane_token, user_agent=ua)
|
|
existing_names = set()
|
|
if st == 200 and isinstance(plane_issues, list):
|
|
for ei in plane_issues:
|
|
existing_names.add(ei.get("name", ""))
|
|
|
|
# 5. Fetch Plane states + priorities for mapping.
|
|
st, states = _request("GET", f"{plane_url}/api/v1/workspaces/{ws}/projects/{project_id}/states/", token=plane_token, user_agent=ua)
|
|
st, priorities = _request("GET", f"{plane_url}/api/v1/workspaces/{ws}/projects/{project_id}/priorities/", token=plane_token, user_agent=ua)
|
|
|
|
for ji in jira_issues:
|
|
fields = ji.get("fields", {})
|
|
name = fields.get("summary", f"{pkey}-{ji.get('key','')}")
|
|
if name in existing_names:
|
|
summary["issues"]["skipped"] += 1
|
|
continue
|
|
# Convert Jira description (ADF) to simple HTML/text for Plane.
|
|
desc = _adf_to_html(fields.get("description"))
|
|
body = {
|
|
"name": name,
|
|
"description_html": desc,
|
|
"priority": _map_priority(fields.get("priority", {}).get("name", "")),
|
|
}
|
|
if dry:
|
|
summary["issues"]["synced"] += 1
|
|
continue
|
|
st, resp = _request("POST", f"{plane_url}/api/v1/workspaces/{ws}/projects/{project_id}/issues/", token=plane_token, json_body=body, user_agent=ua)
|
|
if st in (200, 201):
|
|
summary["issues"]["synced"] += 1
|
|
else:
|
|
summary["errors"].append({"issue": ji.get("key"), "status": st, "detail": str(resp)[:200]})
|
|
|
|
return summary
|
|
|
|
|
|
def _adf_to_html(adf: Any) -> str:
|
|
"""Convert Atlassian Document Format (ADF) to minimal HTML for Plane."""
|
|
if adf is None:
|
|
return ""
|
|
if isinstance(adf, str):
|
|
return f"<p>{adf}</p>"
|
|
if not isinstance(adf, dict):
|
|
return ""
|
|
parts = []
|
|
for block in adf.get("content", []):
|
|
btype = block.get("type", "")
|
|
texts = []
|
|
for node in block.get("content", []):
|
|
if node.get("type") == "text":
|
|
texts.append(node.get("text", ""))
|
|
text = "".join(texts)
|
|
if btype == "heading":
|
|
level = block.get("attrs", {}).get("level", 2)
|
|
parts.append(f"<h{level}>{text}</h{level}>")
|
|
elif btype == "paragraph":
|
|
parts.append(f"<p>{text}</p>")
|
|
elif btype == "codeBlock":
|
|
parts.append(f"<pre><code>{text}</code></pre>")
|
|
elif btype == "bulletList":
|
|
parts.append(f"<ul><li>{text}</li></ul>")
|
|
else:
|
|
parts.append(f"<p>{text}</p>")
|
|
return "\n".join(parts)
|
|
|
|
|
|
def _map_priority(jira_prio: str) -> str:
|
|
"""Map Jira priority names to Plane urgency values."""
|
|
m = {
|
|
"Highest": "urgent",
|
|
"High": "high",
|
|
"Medium": "medium",
|
|
"Low": "low",
|
|
"Lowest": "low",
|
|
}
|
|
return m.get(jira_prio, "none")
|
|
|
|
|
|
# ════════════════════════════════════════════════════════════════════════════
|
|
# OUTLINE: Confluence → Outline doc mirror
|
|
# ════════════════════════════════════════════════════════════════════════════
|
|
|
|
|
|
def sync_outline(reg: dict, target_cfg: dict, dry: bool = False) -> dict:
|
|
"""Mirror Confluence pages into Outline documents."""
|
|
outline_url = target_cfg["endpoint"].rstrip("/")
|
|
outline_token = _env(target_cfg.get("token_env", ""))
|
|
collection_name = target_cfg.get("collection_name", "Mirror")
|
|
ua = reg.get("defaults", {}).get("user_agent", "replica-mirror/1.0")
|
|
summary: dict[str, Any] = {"target": "outline", "pages": {"synced": 0, "skipped": 0}, "collection": None, "errors": []}
|
|
|
|
if not outline_token and not dry:
|
|
summary["errors"].append(f"missing env {target_cfg.get('token_env')}")
|
|
return summary
|
|
|
|
for src in target_cfg.get("sources", []):
|
|
if src["type"] != "confluence-cloud":
|
|
summary["errors"].append(f"unknown source type {src['type']} for outline")
|
|
continue
|
|
|
|
cf_url = src.get("endpoint")
|
|
if not cf_url:
|
|
summary["errors"].append({"source": src["id"], "reason": "Confluence endpoint is null — set it in sync-mirrors.yaml"})
|
|
continue
|
|
|
|
cf_token = _env(src.get("token_env", "")) or ""
|
|
cf_email = _env(src.get("email_env", "")) or ""
|
|
if not dry and (not cf_token or not cf_email):
|
|
summary["errors"].append(f"missing Confluence creds ({src.get('token_env')} / {src.get('email_env')})")
|
|
continue
|
|
|
|
# 1. Ensure Outline collection exists.
|
|
collection_id = None
|
|
if not dry:
|
|
st, cols = _request("GET", f"{outline_url}/api/collections.list", token=outline_token, user_agent=ua)
|
|
if st == 200 and isinstance(cols, dict):
|
|
for c in cols.get("data", []):
|
|
if c.get("name", "").lower() == collection_name.lower():
|
|
collection_id = c.get("id")
|
|
break
|
|
if not collection_id:
|
|
body = {"name": collection_name}
|
|
st, resp = _request("POST", f"{outline_url}/api/collections.create", token=outline_token, json_body=body, user_agent=ua)
|
|
if st in (200, 201) and isinstance(resp, dict):
|
|
collection_id = (resp.get("data") or {}).get("id")
|
|
summary["collection"] = {"id": collection_id, "created": True}
|
|
else:
|
|
summary["errors"].append({"step": "create collection", "status": st, "detail": str(resp)[:300]})
|
|
continue
|
|
else:
|
|
summary["collection"] = {"id": collection_id, "existed": True}
|
|
else:
|
|
summary["collection"] = {"name": collection_name, "dry": True}
|
|
|
|
# 2. List Confluence spaces.
|
|
spaces_resp = _jira_get(cf_url, cf_email, cf_token, ua, "/wiki/api/v2/spaces?limit=100")
|
|
if spaces_resp["status"] != 200:
|
|
summary["errors"].append({"step": "confluence spaces", "status": spaces_resp["status"], "detail": str(spaces_resp["data"])[:300]})
|
|
continue
|
|
spaces = (spaces_resp["data"] or {}).get("results", [])
|
|
if src.get("spaces"):
|
|
wanted = set(src["spaces"])
|
|
spaces = [s for s in spaces if s.get("key") in wanted]
|
|
|
|
for space in spaces:
|
|
space_key = space["key"]
|
|
# 3. List pages in space.
|
|
pages_resp = _jira_get(cf_url, cf_email, cf_token, ua, f"/wiki/api/v2/spaces/{space_key}/pages?limit=100&body-format=storage")
|
|
if pages_resp["status"] != 200:
|
|
summary["errors"].append({"step": f"confluence pages {space_key}", "status": pages_resp["status"]})
|
|
continue
|
|
pages = (pages_resp["data"] or {}).get("results", [])
|
|
|
|
# 4. List existing Outline docs to dedupe by title.
|
|
existing_titles = set()
|
|
if not dry and collection_id:
|
|
st, docs = _request("GET", f"{outline_url}/api/documents.list", token=outline_token, user_agent=ua)
|
|
if st == 200 and isinstance(docs, dict):
|
|
for d in docs.get("data", []):
|
|
if (d.get("collectionId") or "") == collection_id:
|
|
existing_titles.add(d.get("title", "").lower())
|
|
|
|
for page in pages:
|
|
title = page.get("title", f"Untitled-{page.get('id')}")
|
|
if title.lower() in existing_titles:
|
|
summary["pages"]["skipped"] += 1
|
|
continue
|
|
# Convert Confluence storage format → Markdown.
|
|
raw_body = (page.get("body") or {})
|
|
storage_xml = raw_body.get("storage", {}).get("value", "") if isinstance(raw_body, dict) else str(raw_body)
|
|
md = _confluence_storage_to_md(storage_xml)
|
|
|
|
if dry:
|
|
summary["pages"]["synced"] += 1
|
|
continue
|
|
|
|
body = {
|
|
"title": f"[{space_key}] {title}",
|
|
"text": md,
|
|
"collectionId": collection_id,
|
|
"publish": True,
|
|
}
|
|
st, resp = _request("POST", f"{outline_url}/api/documents.create", token=outline_token, json_body=body, user_agent=ua)
|
|
if st in (200, 201):
|
|
summary["pages"]["synced"] += 1
|
|
else:
|
|
summary["errors"].append({"page": page.get("id"), "status": st, "detail": str(resp)[:200]})
|
|
|
|
return summary
|
|
|
|
|
|
def _confluence_storage_to_md(xml: str) -> str:
|
|
"""Minimal Confluence storage-format (XHTML) → Markdown conversion.
|
|
|
|
Handles the common constructs: headings, paragraphs, lists, code blocks,
|
|
links, bold/italic. Full Confluence XHTML is complex; this covers the 90%
|
|
case and leaves unknown tags as stripped text.
|
|
"""
|
|
import re
|
|
if not xml:
|
|
return ""
|
|
# Remove XML namespaces for simpler matching.
|
|
text = re.sub(r'xmlns[^"]*"[^"]*"', "", xml)
|
|
text = re.sub(r"<ac:structured-macro[^>]*>.*?</ac:structured-macro>", "[macro]", text, flags=re.DOTALL)
|
|
# Headings.
|
|
for i in range(6, 0, -1):
|
|
text = re.sub(rf"<h{i}[^>]*>(.*?)</h{i}>", lambda m, lvl=i: "#" * lvl + " " + m.group(1).strip(), text, flags=re.DOTALL)
|
|
# Code blocks: <pre><code>...</code></pre> → fenced, before inline <code>.
|
|
text = re.sub(r"<pre[^>]*>\s*<code[^>]*>(.*?)</code>\s*</pre>", lambda m: f"```\n{m.group(1).strip()}\n```", text, flags=re.DOTALL)
|
|
text = re.sub(r"<ac:plain-text-body[^>]*><!\[CDATA\[(.*?)\]\]></ac:plain-text-body>", lambda m: f"```\n{m.group(1)}\n```", text, flags=re.DOTALL)
|
|
text = re.sub(r"<code[^>]*>(.*?)</code>", lambda m: f"`{m.group(1).strip()}`", text, flags=re.DOTALL)
|
|
# Bold / italic.
|
|
text = re.sub(r"<b>(.*?)</b>", r"**\1**", text, flags=re.DOTALL)
|
|
text = re.sub(r"<strong>(.*?)</strong>", r"**\1**", text, flags=re.DOTALL)
|
|
text = re.sub(r"<i>(.*?)</i>", r"*\1*", text, flags=re.DOTALL)
|
|
text = re.sub(r"<em>(.*?)</em>", r"*\1*", text, flags=re.DOTALL)
|
|
# Links.
|
|
text = re.sub(r'<a[^>]*href="([^"]*)"[^>]*>(.*?)</a>', lambda m: f"[{m.group(2).strip()}]({m.group(1)})", text, flags=re.DOTALL)
|
|
# Lists.
|
|
text = re.sub(r"<li[^>]*>(.*?)</li>", lambda m: f"- {m.group(1).strip()}\n", text, flags=re.DOTALL)
|
|
text = re.sub(r"</?[ou]l[^>]*>", "", text)
|
|
# Paragraphs / line breaks.
|
|
text = re.sub(r"<p[^>]*>", "\n", text)
|
|
text = re.sub(r"</p>", "\n", text)
|
|
text = re.sub(r"<br\s*/?>", "\n", text)
|
|
# Strip remaining tags.
|
|
text = re.sub(r"<[^>]+>", "", text)
|
|
# Collapse excessive whitespace.
|
|
text = re.sub(r"\n{3,}", "\n\n", text).strip()
|
|
return text
|
|
|
|
|
|
# ════════════════════════════════════════════════════════════════════════════
|
|
# RL feedback: emit trajectory reward signals from sync drift
|
|
# ════════════════════════════════════════════════════════════════════════════
|
|
|
|
|
|
def emit_trajectory_rewards(reg: dict, sync_summaries: list[dict] | None = None) -> dict:
|
|
"""Append RL reward signals to the trajectory-rewards JSONL.
|
|
|
|
Called after each --sync run. Reads the sync summaries (what changed,
|
|
what drifted) and emits reward signals the RL pipeline consumes.
|
|
"""
|
|
rl_cfg = reg.get("rl_feedback", {})
|
|
if not rl_cfg.get("enabled", False):
|
|
return {"skipped": "rl_feedback disabled"}
|
|
output = REPLICA_ROOT / rl_cfg.get("output", str(REWARD_PATH))
|
|
output.parent.mkdir(parents=True, exist_ok=True)
|
|
signals = rl_cfg.get("signals", {})
|
|
|
|
events: list[dict] = []
|
|
ts = datetime.now(timezone.utc).isoformat()
|
|
|
|
if sync_summaries:
|
|
for s in sync_summaries:
|
|
target = s.get("target", "unknown")
|
|
# Positive: sync succeeded with no errors.
|
|
if not s.get("errors"):
|
|
events.append({
|
|
"type": "trajectory_reward",
|
|
"timestamp": ts,
|
|
"target": target,
|
|
"signal": "sync_hit",
|
|
"reward": signals.get("sync_hit", 0.1),
|
|
"detail": f"{target} sync clean",
|
|
})
|
|
else:
|
|
# Negative: errors mean drift / staleness.
|
|
events.append({
|
|
"type": "trajectory_reward",
|
|
"timestamp": ts,
|
|
"target": target,
|
|
"signal": "sync_miss",
|
|
"reward": signals.get("sync_miss", -0.2),
|
|
"detail": f"{target} sync had {len(s['errors'])} errors",
|
|
})
|
|
# If items were skipped because they already existed, that's neutral.
|
|
|
|
with output.open("a", encoding="utf-8") as f:
|
|
for ev in events:
|
|
f.write(json.dumps(ev, ensure_ascii=False) + "\n")
|
|
|
|
return {"emitted": len(events), "output": str(output)}
|
|
|
|
|
|
# ════════════════════════════════════════════════════════════════════════════
|
|
# Dist generation (--write-in-place / --check)
|
|
# ════════════════════════════════════════════════════════════════════════════
|
|
|
|
|
|
def generate_dist(reg: dict) -> dict[str, str]:
|
|
"""Produce the downstream catalog files from the YAML registry."""
|
|
targets_out = {}
|
|
for tname, tcfg in reg.get("targets", {}).items():
|
|
entry = {
|
|
"id": tname,
|
|
"kind": tcfg.get("kind"),
|
|
"label": tcfg.get("label", tname),
|
|
"endpoint": tcfg.get("endpoint"),
|
|
"token_env": tcfg.get("token_env"),
|
|
"sources": [],
|
|
}
|
|
for src in tcfg.get("sources", []):
|
|
entry["sources"].append({
|
|
"id": src.get("id"),
|
|
"type": src.get("type"),
|
|
"label": src.get("label"),
|
|
"status": src.get("status", "declared"),
|
|
"endpoint": src.get("endpoint"),
|
|
})
|
|
targets_out[tname] = entry
|
|
|
|
catalog = {
|
|
"version": reg.get("version", 1),
|
|
"defaults": reg.get("defaults", {}),
|
|
"targets": targets_out,
|
|
"rl_feedback": reg.get("rl_feedback", {}),
|
|
}
|
|
return {
|
|
"mirrors.catalog.json": json.dumps(catalog, indent=2) + "\n",
|
|
}
|
|
|
|
|
|
def write_in_place() -> None:
|
|
reg = load_registry()
|
|
DIST_DIR.mkdir(parents=True, exist_ok=True)
|
|
for name, content in generate_dist(reg).items():
|
|
(DIST_DIR / name).write_text(content, encoding="utf-8")
|
|
print(f"wrote {DIST_DIR / name}")
|
|
|
|
|
|
def check() -> int:
|
|
reg = load_registry()
|
|
stale = []
|
|
for name, want in generate_dist(reg).items():
|
|
p = DIST_DIR / name
|
|
if not p.exists() or p.read_text(encoding="utf-8") != want:
|
|
stale.append(name)
|
|
if stale:
|
|
print(f"stale mirror catalog (run --write-in-place): {', '.join(stale)}", file=sys.stderr)
|
|
return 1
|
|
print("mirror catalog up to date")
|
|
return 0
|
|
|
|
|
|
# ════════════════════════════════════════════════════════════════════════════
|
|
# CLI
|
|
# ════════════════════════════════════════════════════════════════════════════
|
|
|
|
|
|
SYNC_DISPATCH = {
|
|
"forgejo": sync_forgejo,
|
|
"plane": sync_plane,
|
|
"outline": sync_outline,
|
|
}
|
|
|
|
|
|
def run_sync(target: str | None, dry: bool, emit_rewards: bool) -> int:
|
|
reg = load_registry()
|
|
targets = reg.get("targets", {})
|
|
names = [target] if target else list(targets.keys())
|
|
fail_fast = reg.get("defaults", {}).get("fail_fast", False)
|
|
summaries: list[dict] = []
|
|
rc = 0
|
|
for name in names:
|
|
if name not in targets:
|
|
print(f"unknown target: {name} (known: {', '.join(targets)})", file=sys.stderr)
|
|
rc = 1
|
|
continue
|
|
fn = SYNC_DISPATCH.get(targets[name].get("kind"))
|
|
if not fn:
|
|
# Map kind → fn for aliases.
|
|
kind = targets[name].get("kind", "")
|
|
if "git" in kind:
|
|
fn = sync_forgejo
|
|
elif "issue" in kind:
|
|
fn = sync_plane
|
|
elif "doc" in kind:
|
|
fn = sync_outline
|
|
if not fn:
|
|
print(f"no sync handler for target {name} (kind={targets[name].get('kind')})", file=sys.stderr)
|
|
rc = 1
|
|
continue
|
|
print(f"[{name}] syncing (dry={dry})...")
|
|
try:
|
|
s = fn(reg, targets[name], dry=dry)
|
|
summaries.append(s)
|
|
_print_summary(s)
|
|
except Exception as exc:
|
|
summaries.append({"target": name, "errors": [str(exc)]})
|
|
print(f"[{name}] FAILED: {exc}", file=sys.stderr)
|
|
if fail_fast:
|
|
return 1
|
|
rc = 1
|
|
|
|
if emit_rewards:
|
|
r = emit_trajectory_rewards(reg, summaries)
|
|
print(f"[rl] reward signals: {r}")
|
|
|
|
# Persist last-sync summary for drift detection on next run.
|
|
state_dir = REPLICA_ROOT / ".aurelio" / "mirrors" / "state"
|
|
state_dir.mkdir(parents=True, exist_ok=True)
|
|
(state_dir / "last-sync.json").write_text(
|
|
json.dumps({"timestamp": datetime.now(timezone.utc).isoformat(), "summaries": summaries}, indent=2),
|
|
encoding="utf-8",
|
|
)
|
|
return rc
|
|
|
|
|
|
def _print_summary(s: dict) -> None:
|
|
target = s.get("target", "?")
|
|
errs = s.get("errors", [])
|
|
if target == "forgejo":
|
|
print(f" mirrored: {len(s.get('mirrored', []))} skipped: {len(s.get('skipped', []))} errors: {len(errs)}")
|
|
elif target == "plane":
|
|
print(f" projects: {len(s.get('projects', []))} issues synced: {s.get('issues', {}).get('synced', 0)} skipped: {s.get('issues', {}).get('skipped', 0)} errors: {len(errs)}")
|
|
elif target == "outline":
|
|
print(f" pages synced: {s.get('pages', {}).get('synced', 0)} skipped: {s.get('pages', {}).get('skipped', 0)} errors: {len(errs)}")
|
|
for e in errs[:5]:
|
|
print(f" ! {e}")
|
|
|
|
|
|
def main() -> int:
|
|
import urllib.parse # noqa: F401 — used in query quoting above
|
|
|
|
ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
|
|
mode = ap.add_mutually_exclusive_group()
|
|
mode.add_argument("--write-in-place", action="store_true", help="Regenerate dist/ from YAML registry.")
|
|
mode.add_argument("--check", action="store_true", help="CI guard: exit 1 if dist/ is stale.")
|
|
mode.add_argument("--sync", nargs="?", const="all", default=None, help="Run live mirror sync (target name or 'all').")
|
|
mode.add_argument("--emit-trajectory-rewards", action="store_true", help="Emit RL reward signals from last sync.")
|
|
ap.add_argument("--dry-run", action="store_true", help="Preview what would sync without making API calls.")
|
|
args = ap.parse_args()
|
|
|
|
if args.check:
|
|
return check()
|
|
if args.write_in_place:
|
|
write_in_place()
|
|
return 0
|
|
if args.emit_trajectory_rewards:
|
|
reg = load_registry()
|
|
state = REPLICA_ROOT / ".aurelio" / "mirrors" / "state" / "last-sync.json"
|
|
sums = []
|
|
if state.exists():
|
|
sums = json.loads(state.read_text()).get("summaries", [])
|
|
r = emit_trajectory_rewards(reg, sums)
|
|
print(json.dumps(r, indent=2))
|
|
return 0
|
|
if args.sync is not None:
|
|
target = None if args.sync == "all" else args.sync
|
|
return run_sync(target, dry=args.dry_run, emit_rewards=True)
|
|
|
|
ap.print_help()
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|