Public Access
329 lines
12 KiB
Python
329 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""Sync documentation from /docs/ to the Gitea wiki via API.
|
|
|
|
Reads markdown files from the ``docs/`` directory, uses ``mapping.json`` to
|
|
map file paths to wiki page titles, and creates/updates wiki pages via the
|
|
Gitea API. Pages that exist in the wiki but not in the mapping are left
|
|
untouched (not deleted).
|
|
|
|
Gitea 1.26 wiki API endpoints (all use content_base64, NOT content):
|
|
- Create: POST /repos/{owner}/{repo}/wiki/new {title, content_base64, message}
|
|
- Update: PATCH /repos/{owner}/{repo}/wiki/page/{sub_url} {title, content_base64, message}
|
|
- List: GET /repos/{owner}/{repo}/wiki/pages → [{title, sub_url, ...}]
|
|
- Fetch: GET /repos/{owner}/{repo}/wiki/page/{sub_url} → {title, content_base64, ...}
|
|
- Delete: DELETE /repos/{owner}/{repo}/wiki/page/{sub_url}
|
|
|
|
Usage:
|
|
CI_GITEA_TOKEN=<token> python3 -m devx.ci.sync_wiki [--dry-run] [--repo owner/repo]
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
|
|
import click
|
|
from dotenv import load_dotenv # pyright: ignore[reportMissingImports,reportUnknownVariableType]
|
|
|
|
from devx.api_clients import GiteaClient
|
|
from devx.config import GITEA_API_URL
|
|
from devx.exceptions import APIError
|
|
from devx.i18n import _
|
|
|
|
load_dotenv()
|
|
|
|
# DOCS_DIR is the repo's docs/ directory. When devx is installed as a
|
|
# package (e.g., in .venv/lib/python3.12/site-packages/devx/), the
|
|
# __file__-relative path would point inside the venv, not the repo.
|
|
# Use DEVX_DOCS_DIR env var if set, otherwise fall back to ./docs
|
|
# (relative to the current working directory, which is the repo root
|
|
# in CI and local development).
|
|
DOCS_DIR = Path(os.environ.get("DEVX_DOCS_DIR", "docs"))
|
|
MAPPING_FILE = DOCS_DIR / "mapping.json"
|
|
|
|
|
|
def load_mapping() -> dict[str, str]:
|
|
"""Load the file-to-wiki-page mapping from mapping.json.
|
|
|
|
Validates that the mapping is a dict of string-to-string pairs.
|
|
"""
|
|
with open(MAPPING_FILE) as f:
|
|
data = json.load(f)
|
|
if not isinstance(data, dict):
|
|
raise click.ClickException(
|
|
_("mapping.json must be a dict of file-path -> page-title, got {type}", type=type(data).__name__)
|
|
)
|
|
for k, v in data.items():
|
|
if not isinstance(k, str) or not isinstance(v, str):
|
|
raise click.ClickException(_("mapping.json keys and values must be strings, got {k}={v}", k=k, v=v))
|
|
return data
|
|
|
|
|
|
def read_doc_content(file_path: str) -> str:
|
|
"""Read markdown content from a docs file."""
|
|
full_path = DOCS_DIR / file_path
|
|
with open(full_path) as f:
|
|
return f.read()
|
|
|
|
|
|
def encode_content(content: str) -> str:
|
|
"""Encode content as base64 for the Gitea wiki API.
|
|
|
|
The Gitea wiki API requires content_base64, not plain content.
|
|
Sending plain content silently fails (pages are created/updated
|
|
but with empty content).
|
|
"""
|
|
return base64.b64encode(content.encode("utf-8")).decode("ascii")
|
|
|
|
|
|
def decode_content(content_b64: str) -> str:
|
|
"""Decode base64 content from the Gitea wiki API."""
|
|
if not content_b64:
|
|
return ""
|
|
return base64.b64decode(content_b64).decode("utf-8")
|
|
|
|
|
|
def list_wiki_pages(client: GiteaClient) -> dict[str, str]:
|
|
"""List existing wiki pages, returning {title: sub_url}."""
|
|
try:
|
|
pages = client._request("GET", "/wiki/pages").json()
|
|
except APIError:
|
|
return {}
|
|
return {page.get("title", ""): page.get("sub_url", page.get("title", "")) for page in pages}
|
|
|
|
|
|
def fetch_page_content(client: GiteaClient, sub_url: str) -> str:
|
|
"""Fetch a wiki page's content by sub_url, decoded from base64."""
|
|
try:
|
|
page = client._request("GET", f"/wiki/page/{sub_url}").json()
|
|
return decode_content(page.get("content_base64", ""))
|
|
except APIError:
|
|
return ""
|
|
|
|
|
|
def sync_page(
|
|
client: GiteaClient,
|
|
page_title: str,
|
|
content: str,
|
|
existing_pages: dict[str, str],
|
|
dry_run: bool,
|
|
) -> str:
|
|
"""Create or update a single wiki page.
|
|
|
|
Returns "created", "updated", or "skipped" (if dry-run).
|
|
"""
|
|
if dry_run:
|
|
click.echo(_("[dry-run] Would sync page: {title} ({chars} chars)", title=page_title, chars=len(content)))
|
|
return "skipped"
|
|
|
|
content_b64 = encode_content(content)
|
|
|
|
if page_title in existing_pages:
|
|
# Update existing page via PATCH
|
|
sub_url = existing_pages[page_title]
|
|
client._request(
|
|
"PATCH",
|
|
f"/wiki/page/{sub_url}",
|
|
json={
|
|
"title": page_title,
|
|
"content_base64": content_b64,
|
|
"message": f"Sync from docs/ — update {page_title}",
|
|
},
|
|
)
|
|
return "updated"
|
|
|
|
# Create new page via POST /wiki/new
|
|
client._request(
|
|
"POST",
|
|
"/wiki/new",
|
|
json={
|
|
"title": page_title,
|
|
"content_base64": content_b64,
|
|
"message": f"Sync from docs/ — create {page_title}",
|
|
},
|
|
)
|
|
return "created"
|
|
|
|
|
|
def verify_wiki_page(
|
|
client: GiteaClient, page_title: str, expected_content: str, existing_pages: dict[str, str]
|
|
) -> bool:
|
|
"""Verify that a wiki page has non-empty content matching the docs.
|
|
|
|
Returns True if the page content matches, False otherwise.
|
|
"""
|
|
if page_title not in existing_pages:
|
|
return False
|
|
sub_url = existing_pages[page_title]
|
|
actual = fetch_page_content(client, sub_url)
|
|
return actual.strip() == expected_content.strip()
|
|
|
|
|
|
def verify_wiki_integrity(
|
|
client: GiteaClient,
|
|
mapping: dict[str, str],
|
|
synced: dict[str, str],
|
|
) -> list[str]:
|
|
"""Comprehensive wiki verification.
|
|
|
|
Checks:
|
|
1. Every mapped page exists in the wiki
|
|
2. Every mapped page has non-empty content
|
|
3. Every mapped page's content matches the docs
|
|
4. No stale pages exist in the wiki (pages not in mapping)
|
|
5. Page count matches
|
|
|
|
Returns a list of failure messages (empty if all checks pass).
|
|
"""
|
|
failures: list[str] = []
|
|
existing_pages = list_wiki_pages(client)
|
|
expected_titles = set(mapping.values())
|
|
|
|
# Check 1: Page count
|
|
if len(existing_pages) != len(expected_titles):
|
|
failures.append(f"Page count mismatch: wiki has {len(existing_pages)}, mapping has {len(expected_titles)}")
|
|
|
|
# Check 2: Missing pages (in mapping but not in wiki)
|
|
missing = expected_titles - set(existing_pages.keys())
|
|
for title in sorted(missing):
|
|
failures.append(f"Missing page: {title}")
|
|
|
|
# Check 3: Stale pages (in wiki but not in mapping)
|
|
stale = set(existing_pages.keys()) - expected_titles
|
|
for title in sorted(stale):
|
|
failures.append(f"Stale page (not in mapping): {title}")
|
|
|
|
# Check 4: Content verification
|
|
for page_title, expected_content in sorted(synced.items()):
|
|
ok = verify_wiki_page(client, page_title, expected_content, existing_pages)
|
|
if not ok:
|
|
sub_url = existing_pages.get(page_title, "?")
|
|
actual = fetch_page_content(client, sub_url)
|
|
if not actual.strip():
|
|
failures.append(f"Empty content: {page_title}")
|
|
else:
|
|
failures.append(f"Content mismatch: {page_title}")
|
|
|
|
return failures
|
|
|
|
|
|
@click.command()
|
|
@click.option("--dry-run", is_flag=True, default=False, help="Show what would happen without making changes.")
|
|
@click.option("--repo", default=None, help="Repository in owner/name format (auto-detected if omitted).")
|
|
@click.option(
|
|
"--verify",
|
|
is_flag=True,
|
|
default=False,
|
|
help="After syncing, verify each page has non-empty content. Exit 1 if any page is empty or mismatched.",
|
|
)
|
|
@click.option(
|
|
"--strict",
|
|
is_flag=True,
|
|
default=False,
|
|
help="Full integrity check: verify page count, missing pages, stale pages, and content. Implies --verify.",
|
|
)
|
|
def main(dry_run: bool, repo: str | None, verify: bool, strict: bool) -> None:
|
|
token = os.environ.get("CI_GITEA_TOKEN", "")
|
|
if not token:
|
|
raise click.ClickException(_("ERROR: CI_GITEA_TOKEN is not set."))
|
|
|
|
if repo is None:
|
|
owner = os.environ.get("DEVX_REPO_OWNER", "oblachno-oss")
|
|
repo_name = os.environ.get("DEVX_REPO_NAME", "devx")
|
|
else:
|
|
owner, repo_name = repo.split("/")
|
|
|
|
if not MAPPING_FILE.exists():
|
|
raise click.ClickException(_("ERROR: mapping.json not found at {path}", path=MAPPING_FILE))
|
|
|
|
mapping = load_mapping()
|
|
client = GiteaClient(GITEA_API_URL, token, owner, repo_name)
|
|
|
|
click.echo(_("Syncing {count} documentation pages to wiki...", count=len(mapping)))
|
|
|
|
existing_pages = list_wiki_pages(client)
|
|
if existing_pages:
|
|
click.echo(_("Found {count} existing wiki pages.", count=len(existing_pages)))
|
|
|
|
created = 0
|
|
updated = 0
|
|
skipped = 0
|
|
synced: dict[str, str] = {} # title -> content, for verification
|
|
|
|
for file_path, page_title in sorted(mapping.items()):
|
|
try:
|
|
content = read_doc_content(file_path)
|
|
except FileNotFoundError:
|
|
raise click.ClickException(
|
|
_("Mapped file {file} not found. Update mapping.json or create the file.", file=file_path)
|
|
) from None
|
|
|
|
if not content.strip():
|
|
raise click.ClickException(
|
|
_("Mapped file {file} is empty. Update the content or remove from mapping.json.", file=file_path)
|
|
) from None
|
|
|
|
result = sync_page(client, page_title, content, existing_pages, dry_run)
|
|
if result == "created":
|
|
created += 1
|
|
click.echo(_(" Created: {title}", title=page_title))
|
|
elif result == "updated":
|
|
updated += 1
|
|
click.echo(_(" Updated: {title}", title=page_title))
|
|
else:
|
|
skipped += 1
|
|
|
|
synced[page_title] = content
|
|
|
|
click.echo(
|
|
_(
|
|
"\nDone! Created: {created}, Updated: {updated}, Skipped: {skipped}",
|
|
created=created,
|
|
updated=updated,
|
|
skipped=skipped,
|
|
)
|
|
)
|
|
|
|
# --strict implies --verify
|
|
do_verify = verify or strict
|
|
|
|
if do_verify and not dry_run:
|
|
if strict:
|
|
click.echo(_("\nRunning full wiki integrity check..."))
|
|
failures = verify_wiki_integrity(client, mapping, synced)
|
|
if failures:
|
|
click.echo(_("\nIntegrity check FAILED ({count} issues):", count=len(failures)))
|
|
for f in failures:
|
|
click.echo(f" - {f}")
|
|
raise click.ClickException(_("Wiki integrity check failed — {count} issue(s)", count=len(failures)))
|
|
click.echo(_("\nIntegrity check passed — all {count} pages verified.", count=len(synced)))
|
|
else:
|
|
click.echo(_("\nVerifying wiki pages have content..."))
|
|
# Re-fetch the page list to get updated sub_urls
|
|
existing_pages = list_wiki_pages(client)
|
|
failures = 0
|
|
for page_title, expected_content in sorted(synced.items()):
|
|
ok = verify_wiki_page(client, page_title, expected_content, existing_pages)
|
|
if ok:
|
|
click.echo(_(" OK: {title} ({chars} chars)", title=page_title, chars=len(expected_content)))
|
|
else:
|
|
click.echo(_(" FAIL: {title} — content mismatch or empty!", title=page_title))
|
|
failures += 1
|
|
if failures > 0:
|
|
click.echo(
|
|
_(
|
|
"\nVerification FAILED: {failures} page(s) have empty or mismatched content!",
|
|
failures=failures,
|
|
)
|
|
)
|
|
raise click.ClickException(
|
|
_("Wiki verification failed — {failures} page(s) empty or mismatched", failures=failures)
|
|
)
|
|
click.echo(_("\nVerification passed — all wiki pages have correct content."))
|
|
|
|
|
|
if __name__ == "__main__": # pragma: no cover
|
|
main()
|