devx/src/devx/ci/sync_wiki.py

#!/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 logging
import os
from pathlib import Path

import click
from dotenv import load_dotenv  # pyright: ignore[reportMissingImports,reportUnknownVariableType]
from tenacity import (
    before_sleep_log,
    retry,
    retry_if_exception_type,
    stop_after_attempt,
    wait_exponential,
)

from devx.api_clients import GiteaClient
from devx.config import GITEA_API_URL, REPO_NAME, REPO_OWNER
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, encoding="utf-8") 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, encoding="utf-8") 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}.

    Raises :class:`APIError` if the wiki API is unavailable — the caller
    is responsible for retrying or handling the failure.
    """
    pages = client._request("GET", "/wiki/pages").json()
    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 _list_wiki_pages_with_retry(client: GiteaClient) -> dict[str, str]:
    """List wiki pages with tenacity retry on APIError.

    The Gitea API can be briefly unavailable right after a batch of wiki
    page updates.  Uses the same tenacity pattern as ``api_clients`` for
    exponential backoff.
    """
    _logger = logging.getLogger("sync_wiki")

    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=2, min=2, max=8),
        retry=retry_if_exception_type(APIError),
        before_sleep=before_sleep_log(_logger, logging.WARNING),
        reraise=True,
    )
    def _do_list() -> dict[str, str]:
        return list_wiki_pages(client)

    return _do_list()


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).
    If the wiki API is temporarily unavailable (all retry attempts
    fail), returns an empty list with a warning — the sync itself
    already succeeded, so a transient API outage should not fail the job.
    """
    failures: list[str] = []

    try:
        existing_pages = _list_wiki_pages_with_retry(client)
    except APIError:
        click.echo(
            _(
                "WARNING: Could not fetch wiki page list after retries. "
                "The sync itself succeeded ({count} pages updated), but the "
                "integrity check could not verify them due to a transient API issue.",
                count=len(synced),
            )
        )
        return []

    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", "") or REPO_OWNER
        repo_name = os.environ.get("DEVX_REPO_NAME", "") or REPO_NAME
    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)))

    try:
        existing_pages = list_wiki_pages(client)
    except APIError:
        existing_pages = {}
    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
            try:
                existing_pages = _list_wiki_pages_with_retry(client)
            except APIError:
                click.echo(
                    _(
                        "WARNING: Could not re-fetch wiki page list for verification. "
                        "Skipping content verification due to transient API issue."
                    )
                )
                return
            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()