export_1c_help/export1c_help/wiki.py

"""Push generated wiki tree into a Gitea wiki git repository."""

from __future__ import annotations

import json
import os
import shutil
import subprocess
import tempfile
from pathlib import Path

from export1c_help.fscompat import copy_file, ensure_dir, remove_file

# Always owned by export_1c_help (not user articles).
_MANAGED_FIXED = frozenset({"Home.md", "History.md", "_Sidebar.md", "manifest.json"})
_IMAGE_SUFFIXES = frozenset({".png", ".jpg", ".jpeg", ".gif", ".bmp", ".webp"})


class WikiPushError(RuntimeError):
    pass


def _clean_arg(value: str | Path) -> str:
    """Windows CreateProcess rejects argv/cwd strings that contain NUL."""
    s = os.fspath(value)
    if "\0" in s:
        s = s.replace("\0", "")
    return s


def _run(cmd: list[str], *, cwd: Path) -> str:
    clean_cmd = [_clean_arg(a) for a in cmd]
    clean_cwd = _clean_arg(cwd.resolve())
    proc = subprocess.run(
        clean_cmd,
        cwd=clean_cwd,
        capture_output=True,
        text=True,
        encoding="utf-8",
        errors="replace",
        check=False,
    )
    if proc.returncode != 0:
        raise WikiPushError(
            f"$ {' '.join(clean_cmd)}\n{proc.stdout}\n{proc.stderr}".strip()
        )
    return proc.stdout


def _git_commit(work_dir: Path, message: str) -> None:
    """Commit via ``-F`` so Windows never gets NUL/encoding issues in ``-m``."""
    text = _clean_arg(message).strip() or "export_1c_help: sync help"
    if not text.endswith("\n"):
        text += "\n"
    with tempfile.NamedTemporaryFile(
        mode="w",
        encoding="utf-8",
        newline="\n",
        suffix=".txt",
        delete=False,
    ) as fh:
        fh.write(text)
        msg_path = fh.name
    try:
        _run(["git", "commit", "-F", msg_path], cwd=work_dir)
    finally:
        try:
            Path(msg_path).unlink()
        except OSError:
            pass


def prepare_wiki_clone(
    wiki_url: str,
    work_dir: Path,
    *,
    branch: str = "main",
) -> Path:
    """Clone or reset wiki working tree."""
    work_dir = work_dir.resolve()
    if work_dir.exists() and (work_dir / ".git").exists():
        def _norm_git_url(u: str) -> str:
            return u.strip().rstrip("/")

        expected = _norm_git_url(wiki_url)
        try:
            actual = _run(
                ["git", "remote", "get-url", "origin"],
                cwd=work_dir,
            ).strip()
        except WikiPushError:
            actual = ""

        if _norm_git_url(actual) != expected:
            # Existing clone points to another wiki repo; don't reuse it.
            shutil.rmtree(work_dir)
        else:
            _run(["git", "fetch", "origin"], cwd=work_dir)
            _run(["git", "checkout", branch], cwd=work_dir)
            _run(["git", "reset", "--hard", f"origin/{branch}"], cwd=work_dir)
            return work_dir

    if work_dir.exists():
        shutil.rmtree(work_dir)
    ensure_dir(work_dir.parent)
    _run(
        [
            "git",
            "clone",
            "--branch",
            branch,
            "--single-branch",
            wiki_url,
            str(work_dir),
        ],
        cwd=work_dir.parent,
    )
    return work_dir


def managed_filenames_from_manifest(manifest_path: Path) -> set[str]:
    """Filenames previously written by export_1c_help."""
    names = set(_MANAGED_FIXED)
    if not manifest_path.is_file():
        return names
    try:
        data = json.loads(manifest_path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError):
        return names
    for key in ("pages", "deleted_pages"):
        for item in data.get(key) or []:
            if isinstance(item, dict) and item.get("filename"):
                names.add(str(item["filename"]))
    return names


def push_wiki(
    content_dir: Path,
    wiki_url: str,
    *,
    work_dir: Path,
    message: str,
    branch: str = "main",
    wipe_unmanaged: bool = False,
    already_prepared: bool = False,
) -> None:
    """
    Clone/fetch wiki, update only export-managed pages, commit, push.

    User-written wiki pages (not listed in previous manifest.json) are kept
    unless wipe_unmanaged=True.
    """
    content_dir = content_dir.resolve()
    work_dir = work_dir.resolve()

    if not already_prepared:
        prepare_wiki_clone(wiki_url, work_dir, branch=branch)

    new_files = {
        p.name
        for p in content_dir.iterdir()
        if p.is_file() and p.name != ".git"
    }
    prev_managed = managed_filenames_from_manifest(work_dir / "manifest.json")

    if wipe_unmanaged:
        for path in work_dir.iterdir():
            if path.name == ".git":
                continue
            if not path.is_file():
                continue
            suf = path.suffix.lower()
            if suf in {".md", ".json"} or suf in _IMAGE_SUFFIXES:
                remove_file(path)
    else:
        # Drop help pages that disappeared from the configuration export.
        for name in sorted(prev_managed - new_files):
            path = work_dir / name
            if path.is_file():
                remove_file(path)

    for src in content_dir.iterdir():
        if src.name == ".git":
            continue
        # -o pointed at parent of the clone: do not copy the clone into itself
        try:
            if src.resolve() == work_dir.resolve():
                continue
            src.resolve().relative_to(work_dir.resolve())
            continue  # src is inside work_dir
        except ValueError:
            pass
        dst = work_dir / src.name
        if src.is_dir():
            if dst.exists():
                shutil.rmtree(dst)
            shutil.copytree(src, dst)
        else:
            copy_file(src, dst)

    _run(["git", "add", "-A"], cwd=work_dir)
    status = _run(["git", "status", "--porcelain"], cwd=work_dir)
    if not status.strip():
        return

    _git_commit(work_dir, message)
    _run(["git", "push", "origin", f"HEAD:{branch}"], cwd=work_dir)