637 lines
22 KiB
Python
637 lines
22 KiB
Python
#!/usr/bin/env python3
|
|
"""Mirror project documentation files into Gitea repository wikis."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import base64
|
|
import dataclasses
|
|
import hashlib
|
|
import pathlib
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import urllib.parse
|
|
from typing import Any
|
|
|
|
from gitea_common import GiteaClient, GiteaError, infer_target, load_dotenv, repo_path, require_token
|
|
|
|
|
|
GIT_ROOT = pathlib.Path("/mnt/DATA/git")
|
|
PRODUCTS_ROOT = pathlib.Path("/mnt/DATA/Nextcloud/ADD ideas UG/Products")
|
|
MANAGED_MARKER = "<!-- codex-wiki-sync:"
|
|
TEXT_SUFFIXES = {".md", ".txt", ".csv", ".toml", ".json", ".yaml", ".yml", ".rst"}
|
|
SENSITIVE_NAME_RE = re.compile(r"(password|passwd|secret|credential|api[_-]?keys?|token|private[_-]?key)", re.IGNORECASE)
|
|
EXCLUDED_PARTS = {
|
|
".git",
|
|
".gitea",
|
|
".venv",
|
|
"node_modules",
|
|
"__pycache__",
|
|
"dist",
|
|
"build",
|
|
".cache",
|
|
".module-test-build",
|
|
}
|
|
EXCLUDED_NAMES = {
|
|
"package-lock.json",
|
|
"pnpm-lock.yaml",
|
|
"yarn.lock",
|
|
"uv.lock",
|
|
}
|
|
|
|
|
|
@dataclasses.dataclass(frozen=True)
|
|
class RepoInfo:
|
|
root: pathlib.Path
|
|
owner: str
|
|
repo: str
|
|
|
|
@property
|
|
def key(self) -> str:
|
|
return self.root.name
|
|
|
|
|
|
@dataclasses.dataclass(frozen=True)
|
|
class WikiSource:
|
|
repo_key: str
|
|
path: pathlib.Path
|
|
origin: str
|
|
page_title: str
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--env-file", type=pathlib.Path, default=pathlib.Path("/home/zemion/.config/gitea/gitea.env"))
|
|
parser.add_argument("--git-root", type=pathlib.Path, default=GIT_ROOT)
|
|
parser.add_argument("--products-root", type=pathlib.Path, default=PRODUCTS_ROOT)
|
|
parser.add_argument("--repo", action="append", help="limit to one or more local repository directory names")
|
|
parser.add_argument("--page", action="append", help="limit to one or more generated wiki page names")
|
|
parser.add_argument("--overwrite-unmanaged", action="store_true", help="update existing wiki pages not previously managed by this script")
|
|
parser.add_argument("--transport", choices=("git", "api"), default="git", help="sync through the wiki git repository or the Gitea REST API")
|
|
parser.add_argument("--wiki-cache-dir", type=pathlib.Path, default=pathlib.Path("/tmp/codex-gitea-wiki-sync"), help="local cache for wiki git checkouts")
|
|
parser.add_argument("--commit-message", default="Sync wiki from project files", help="commit message used by git transport")
|
|
parser.add_argument("--prune-managed", action="store_true", help="delete managed wiki pages whose source files are no longer discovered")
|
|
parser.add_argument("--apply", action="store_true")
|
|
args = parser.parse_args()
|
|
|
|
try:
|
|
load_dotenv(args.env_file)
|
|
repos = discover_hosted_repos(args.git_root)
|
|
if args.repo:
|
|
requested = set(args.repo)
|
|
repos = {key: repo for key, repo in repos.items() if key in requested}
|
|
missing = requested - set(repos)
|
|
if missing:
|
|
raise GiteaError(f"requested repos are not hosted on git.add-ideas.de locally: {', '.join(sorted(missing))}")
|
|
|
|
sources = discover_sources(repos, args.products_root)
|
|
if args.page:
|
|
requested_pages = set(args.page)
|
|
sources = [source for source in sources if source.page_title in requested_pages]
|
|
missing_pages = requested_pages - {source.page_title for source in sources}
|
|
if missing_pages:
|
|
raise GiteaError(f"requested pages were not discovered: {', '.join(sorted(missing_pages))}")
|
|
print(f"Hosted repositories: {len(repos)}")
|
|
print(f"Wiki source files: {len(sources)}")
|
|
for repo_key, count in grouped_counts(sources).items():
|
|
print(f" {repo_key}: {count}")
|
|
|
|
if not args.apply:
|
|
preview_sources(sources)
|
|
print("Dry run only. Re-run with --apply to write wiki pages.")
|
|
return 0
|
|
|
|
token = require_token() if args.transport == "api" else ""
|
|
failures = 0
|
|
for repo_key, repo_sources in group_sources(sources).items():
|
|
repo = repos[repo_key]
|
|
try:
|
|
if args.transport == "git":
|
|
sync_repo_wiki_git(
|
|
repo,
|
|
repo_sources,
|
|
cache_dir=args.wiki_cache_dir,
|
|
overwrite_unmanaged=args.overwrite_unmanaged,
|
|
commit_message=args.commit_message,
|
|
write_index=not bool(args.page),
|
|
prune_managed=args.prune_managed and not bool(args.page),
|
|
)
|
|
else:
|
|
sync_repo_wiki(
|
|
repo,
|
|
repo_sources,
|
|
token,
|
|
overwrite_unmanaged=args.overwrite_unmanaged,
|
|
write_index=not bool(args.page),
|
|
prune_managed=args.prune_managed and not bool(args.page),
|
|
)
|
|
except GiteaError as exc:
|
|
failures += 1
|
|
print(f"error syncing wiki for {repo_key}: {exc}", file=sys.stderr)
|
|
return 1 if failures else 0
|
|
except GiteaError as exc:
|
|
print(f"error: {exc}", file=sys.stderr)
|
|
return 1
|
|
|
|
|
|
def discover_hosted_repos(git_root: pathlib.Path) -> dict[str, RepoInfo]:
|
|
repos: dict[str, RepoInfo] = {}
|
|
for gitdir in sorted(git_root.glob("*/.git")):
|
|
root = gitdir.parent
|
|
result = subprocess.run(
|
|
["git", "-C", str(root), "remote", "get-url", "origin"],
|
|
check=False,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.DEVNULL,
|
|
text=True,
|
|
)
|
|
if "git.add-ideas.de" not in result.stdout:
|
|
continue
|
|
target = infer_target(root)
|
|
repos[root.name] = RepoInfo(root=root, owner=target.owner, repo=target.repo)
|
|
return repos
|
|
|
|
|
|
def discover_sources(repos: dict[str, RepoInfo], products_root: pathlib.Path) -> list[WikiSource]:
|
|
sources: list[WikiSource] = []
|
|
for repo in repos.values():
|
|
for path in repo.root.rglob("*"):
|
|
if is_repo_doc(repo.root, path):
|
|
rel = path.relative_to(repo.root)
|
|
sources.append(
|
|
WikiSource(
|
|
repo_key=repo.key,
|
|
path=path,
|
|
origin="repository",
|
|
page_title=title_for_path("Repo", rel),
|
|
)
|
|
)
|
|
|
|
product_map = {
|
|
"co2api": "emission-api-lib",
|
|
"govoplan": "govoplan-core",
|
|
"meubility": "meubility-workbench",
|
|
"mousehold": "mousehold",
|
|
"prvnncr": "prvnncr-server",
|
|
}
|
|
if products_root.exists():
|
|
for product_dir in sorted(path for path in products_root.iterdir() if path.is_dir()):
|
|
repo_key = product_map.get(product_dir.name)
|
|
if not repo_key or repo_key not in repos:
|
|
continue
|
|
for path in product_dir.rglob("*"):
|
|
if is_product_doc(path):
|
|
rel = path.relative_to(product_dir)
|
|
sources.append(
|
|
WikiSource(
|
|
repo_key=repo_key,
|
|
path=path,
|
|
origin=f"product:{product_dir.name}",
|
|
page_title=title_for_path(f"Product {product_dir.name}", rel),
|
|
)
|
|
)
|
|
|
|
unique: dict[tuple[str, str], WikiSource] = {}
|
|
for source in sources:
|
|
unique[(source.repo_key, source.page_title)] = source
|
|
return sorted(unique.values(), key=lambda item: (item.repo_key, item.page_title))
|
|
|
|
|
|
def is_repo_doc(repo_root_path: pathlib.Path, path: pathlib.Path) -> bool:
|
|
if not path.is_file() or not is_text_file(path):
|
|
return False
|
|
parts = set(path.relative_to(repo_root_path).parts)
|
|
if parts & EXCLUDED_PARTS:
|
|
return False
|
|
if path.name in EXCLUDED_NAMES:
|
|
return False
|
|
if SENSITIVE_NAME_RE.search(path.name):
|
|
return False
|
|
rel = path.relative_to(repo_root_path)
|
|
if len(rel.parts) == 1 and path.name.lower().startswith(("readme", "license", "changelog", "contributing", "security")):
|
|
return True
|
|
if "generated" in rel.parts:
|
|
return False
|
|
if rel.parts[0] in {"docs", "doc", "codex"} and path.suffix.lower() in {".md", ".txt", ".csv"}:
|
|
return True
|
|
if re.search(r"(backlog|todo|roadmap|plan|architecture|workflow|manifest)", path.name, re.IGNORECASE):
|
|
return True
|
|
return False
|
|
|
|
|
|
def is_product_doc(path: pathlib.Path) -> bool:
|
|
if not path.is_file() or not is_text_file(path):
|
|
return False
|
|
if path.name in EXCLUDED_NAMES:
|
|
return False
|
|
if SENSITIVE_NAME_RE.search(path.name):
|
|
return False
|
|
if any(part in {"python_backup", "bruno", "fluege"} for part in path.parts):
|
|
return False
|
|
if re.search(
|
|
r"(readme|todo|roadmap|plan|concept|continuation|copyright|notes|whitepaper|pitch|request_response)",
|
|
path.name,
|
|
re.IGNORECASE,
|
|
) and path.suffix.lower() in {".md", ".txt"}:
|
|
return True
|
|
return False
|
|
|
|
|
|
def is_text_file(path: pathlib.Path) -> bool:
|
|
if path.suffix.lower() not in TEXT_SUFFIXES:
|
|
return False
|
|
try:
|
|
chunk = path.read_bytes()[:4096]
|
|
except OSError:
|
|
return False
|
|
return b"\x00" not in chunk
|
|
|
|
|
|
def title_for_path(prefix: str, rel: pathlib.Path) -> str:
|
|
stem = rel.with_suffix("").as_posix()
|
|
stem = re.sub(r"[^A-Za-z0-9]+", "-", stem).strip("-")
|
|
return f"{prefix}-{stem}"[:180]
|
|
|
|
|
|
def group_sources(sources: list[WikiSource]) -> dict[str, list[WikiSource]]:
|
|
grouped: dict[str, list[WikiSource]] = {}
|
|
for source in sources:
|
|
grouped.setdefault(source.repo_key, []).append(source)
|
|
return dict(sorted(grouped.items()))
|
|
|
|
|
|
def grouped_counts(sources: list[WikiSource]) -> dict[str, int]:
|
|
return {repo_key: len(items) for repo_key, items in group_sources(sources).items()}
|
|
|
|
|
|
def preview_sources(sources: list[WikiSource]) -> None:
|
|
for repo_key, repo_sources in group_sources(sources).items():
|
|
print(f"{repo_key}:")
|
|
for source in repo_sources[:80]:
|
|
print(f" {source.page_title} <- {source.path}")
|
|
if len(repo_sources) > 80:
|
|
print(f" ... {len(repo_sources) - 80} more")
|
|
|
|
|
|
def sync_repo_wiki(
|
|
repo: RepoInfo,
|
|
sources: list[WikiSource],
|
|
token: str,
|
|
*,
|
|
overwrite_unmanaged: bool,
|
|
write_index: bool = True,
|
|
prune_managed: bool = False,
|
|
) -> None:
|
|
target = infer_target(repo.root)
|
|
client = GiteaClient(target, token)
|
|
existing_pages = list_existing_wiki_pages(client, target.owner, target.repo)
|
|
existing_page_names = {
|
|
str(page.get("title")): str(page.get("sub_url") or page.get("title"))
|
|
for page in existing_pages
|
|
if page.get("title")
|
|
}
|
|
index_lines = [
|
|
f"# {target.repo} Project Wiki Index",
|
|
"",
|
|
f"{MANAGED_MARKER}index -->",
|
|
"",
|
|
"This page is generated from repository and product-directory project files.",
|
|
"",
|
|
]
|
|
|
|
for source in sources:
|
|
content = render_wiki_content(source)
|
|
changed = put_wiki_page(
|
|
client,
|
|
target.owner,
|
|
target.repo,
|
|
source.page_title,
|
|
content,
|
|
existing_page_names,
|
|
overwrite_unmanaged=overwrite_unmanaged,
|
|
)
|
|
print(f"{'updated' if changed else 'unchanged'} {target.owner}/{target.repo} wiki:{source.page_title}")
|
|
index_lines.append(f"- [{source.page_title}]({source.page_title}) - `{source.path}`")
|
|
|
|
if write_index:
|
|
put_wiki_page(
|
|
client,
|
|
target.owner,
|
|
target.repo,
|
|
"Codex-Project-Index",
|
|
"\n".join(index_lines) + "\n",
|
|
existing_page_names,
|
|
overwrite_unmanaged=True,
|
|
)
|
|
else:
|
|
print(f"skipped {target.owner}/{target.repo} wiki:Codex-Project-Index during page-limited sync")
|
|
if prune_managed and write_index:
|
|
prune_managed_wiki_pages_api(client, target.owner, target.repo, existing_page_names, sources)
|
|
|
|
|
|
def sync_repo_wiki_git(
|
|
repo: RepoInfo,
|
|
sources: list[WikiSource],
|
|
*,
|
|
cache_dir: pathlib.Path,
|
|
overwrite_unmanaged: bool,
|
|
commit_message: str,
|
|
write_index: bool = True,
|
|
prune_managed: bool = False,
|
|
) -> None:
|
|
target = infer_target(repo.root)
|
|
wiki_root = prepare_wiki_checkout(repo, cache_dir=cache_dir)
|
|
index_lines = [
|
|
f"# {target.repo} Project Wiki Index",
|
|
"",
|
|
f"{MANAGED_MARKER}index -->",
|
|
"",
|
|
"This page is generated from repository and product-directory project files.",
|
|
"",
|
|
]
|
|
|
|
for source in sources:
|
|
content = render_wiki_content(source)
|
|
status = write_wiki_page_file(
|
|
wiki_root,
|
|
source.page_title,
|
|
content,
|
|
overwrite_unmanaged=overwrite_unmanaged,
|
|
)
|
|
print(f"{status} {target.owner}/{target.repo} wiki:{source.page_title}")
|
|
index_lines.append(f"- [{source.page_title}]({source.page_title}) - `{source.path}`")
|
|
|
|
if write_index:
|
|
write_wiki_page_file(
|
|
wiki_root,
|
|
"Codex-Project-Index",
|
|
"\n".join(index_lines) + "\n",
|
|
overwrite_unmanaged=True,
|
|
)
|
|
else:
|
|
print(f"skipped {target.owner}/{target.repo} wiki:Codex-Project-Index during page-limited sync")
|
|
if prune_managed and write_index:
|
|
prune_managed_wiki_pages_git(wiki_root, sources)
|
|
if not git_has_changes(wiki_root):
|
|
print(f"unchanged {target.owner}/{target.repo} wiki repository")
|
|
return
|
|
run_git(wiki_root, "add", "-A")
|
|
if not git_has_staged_changes(wiki_root):
|
|
print(f"unchanged {target.owner}/{target.repo} wiki repository")
|
|
return
|
|
ensure_git_identity(wiki_root)
|
|
run_git(wiki_root, "commit", "-m", commit_message)
|
|
run_git(wiki_root, "push")
|
|
print(f"pushed {target.owner}/{target.repo} wiki repository")
|
|
|
|
|
|
def prepare_wiki_checkout(repo: RepoInfo, *, cache_dir: pathlib.Path) -> pathlib.Path:
|
|
remote = wiki_remote_for_repo(repo.root)
|
|
cache_dir.mkdir(parents=True, exist_ok=True)
|
|
checkout = cache_dir / f"{repo.owner}-{repo.repo}.wiki"
|
|
if (checkout / ".git").exists():
|
|
run_git(checkout, "remote", "set-url", "origin", remote)
|
|
run_git(checkout, "fetch", "origin")
|
|
branch = current_branch(checkout)
|
|
if branch:
|
|
run_git(checkout, "reset", "--hard", f"origin/{branch}")
|
|
else:
|
|
run_git(checkout, "reset", "--hard")
|
|
run_git(checkout, "clean", "-fd")
|
|
return checkout
|
|
if checkout.exists():
|
|
shutil.rmtree(checkout)
|
|
result = subprocess.run(
|
|
["git", "clone", remote, str(checkout)],
|
|
check=False,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
text=True,
|
|
)
|
|
if result.returncode == 0:
|
|
return checkout
|
|
checkout.mkdir(parents=True, exist_ok=True)
|
|
run_git(checkout, "init")
|
|
run_git(checkout, "remote", "add", "origin", remote)
|
|
run_git(checkout, "checkout", "-b", "master")
|
|
return checkout
|
|
|
|
|
|
def wiki_remote_for_repo(repo_root_path: pathlib.Path) -> str:
|
|
result = subprocess.run(
|
|
["git", "-C", str(repo_root_path), "remote", "get-url", "origin"],
|
|
check=False,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
text=True,
|
|
)
|
|
if result.returncode != 0 or not result.stdout.strip():
|
|
raise GiteaError(f"Could not read origin remote for {repo_root_path}")
|
|
remote = result.stdout.strip()
|
|
if remote.endswith(".git"):
|
|
return f"{remote[:-4]}.wiki.git"
|
|
return f"{remote}.wiki.git"
|
|
|
|
|
|
def write_wiki_page_file(wiki_root: pathlib.Path, title: str, content: str, *, overwrite_unmanaged: bool) -> str:
|
|
path = wiki_root / f"{wiki_file_stem(title)}.md"
|
|
if path.exists():
|
|
current = read_text(path)
|
|
if current == content:
|
|
return "unchanged"
|
|
if MANAGED_MARKER not in current and not overwrite_unmanaged:
|
|
return "skipped unmanaged"
|
|
path.write_text(content, encoding="utf-8")
|
|
return "updated"
|
|
|
|
|
|
def prune_managed_wiki_pages_git(wiki_root: pathlib.Path, sources: list[WikiSource]) -> None:
|
|
desired = {f"{wiki_file_stem(source.page_title)}.md" for source in sources}
|
|
desired.add(f"{wiki_file_stem('Codex-Project-Index')}.md")
|
|
for path in sorted(wiki_root.glob("*.md")):
|
|
if path.name in desired:
|
|
continue
|
|
current = read_text(path)
|
|
if MANAGED_MARKER not in current:
|
|
continue
|
|
path.unlink()
|
|
print(f"deleted stale managed wiki page {path.name}")
|
|
|
|
|
|
def prune_managed_wiki_pages_api(
|
|
client: GiteaClient,
|
|
owner: str,
|
|
repo: str,
|
|
existing_page_names: dict[str, str],
|
|
sources: list[WikiSource],
|
|
) -> None:
|
|
desired = {source.page_title for source in sources}
|
|
desired.add("Codex-Project-Index")
|
|
for title, page_name in sorted(existing_page_names.items()):
|
|
if title in desired:
|
|
continue
|
|
current = client.request_json("GET", repo_path(owner, repo, f"/wiki/page/{quote_wiki_page_name(page_name)}"))
|
|
if MANAGED_MARKER not in decode_wiki_content(current):
|
|
continue
|
|
client.request_json("DELETE", repo_path(owner, repo, f"/wiki/page/{quote_wiki_page_name(page_name)}"))
|
|
print(f"deleted stale managed wiki page {owner}/{repo}:{title}")
|
|
|
|
|
|
def wiki_file_stem(title: str) -> str:
|
|
return re.sub(r"[/\\]+", "-", title).strip() or "Home"
|
|
|
|
|
|
def git_has_changes(root: pathlib.Path) -> bool:
|
|
result = subprocess.run(
|
|
["git", "-C", str(root), "status", "--porcelain"],
|
|
check=False,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
text=True,
|
|
)
|
|
if result.returncode != 0:
|
|
raise GiteaError(f"git status failed in {root}: {result.stderr.strip()}")
|
|
return bool(result.stdout.strip())
|
|
|
|
|
|
def git_has_staged_changes(root: pathlib.Path) -> bool:
|
|
result = subprocess.run(
|
|
["git", "-C", str(root), "diff", "--cached", "--quiet"],
|
|
check=False,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
text=True,
|
|
)
|
|
if result.returncode == 0:
|
|
return False
|
|
if result.returncode == 1:
|
|
return True
|
|
raise GiteaError(f"git diff --cached failed in {root}: {result.stderr.strip()}")
|
|
|
|
|
|
def current_branch(root: pathlib.Path) -> str:
|
|
result = subprocess.run(
|
|
["git", "-C", str(root), "rev-parse", "--abbrev-ref", "HEAD"],
|
|
check=False,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
text=True,
|
|
)
|
|
branch = result.stdout.strip()
|
|
if result.returncode != 0 or branch == "HEAD":
|
|
return ""
|
|
return branch
|
|
|
|
|
|
def ensure_git_identity(root: pathlib.Path) -> None:
|
|
if not git_config_value(root, "user.email"):
|
|
run_git(root, "config", "user.email", "codex@govoplan.local")
|
|
if not git_config_value(root, "user.name"):
|
|
run_git(root, "config", "user.name", "Codex")
|
|
|
|
|
|
def git_config_value(root: pathlib.Path, key: str) -> str:
|
|
result = subprocess.run(
|
|
["git", "-C", str(root), "config", "--get", key],
|
|
check=False,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
text=True,
|
|
)
|
|
return result.stdout.strip() if result.returncode == 0 else ""
|
|
|
|
|
|
def run_git(root: pathlib.Path, *args: str) -> None:
|
|
result = subprocess.run(
|
|
["git", "-C", str(root), *args],
|
|
check=False,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
text=True,
|
|
)
|
|
if result.returncode != 0:
|
|
command = "git -C " + str(root) + " " + " ".join(args)
|
|
raise GiteaError(f"{command} failed: {result.stderr.strip() or result.stdout.strip()}")
|
|
|
|
|
|
def list_existing_wiki_pages(client: GiteaClient, owner: str, repo: str) -> list[dict[str, Any]]:
|
|
try:
|
|
return client.paginate(repo_path(owner, repo, "/wiki/pages"), limit=50)
|
|
except GiteaError as exc:
|
|
if "HTTP 404" in str(exc) and "/wiki/pages" in str(exc):
|
|
return []
|
|
raise
|
|
|
|
|
|
def render_wiki_content(source: WikiSource) -> str:
|
|
raw = read_text(source.path)
|
|
fingerprint = hashlib.sha256(raw.encode("utf-8")).hexdigest()[:24]
|
|
header = [
|
|
f"{MANAGED_MARKER}{fingerprint} -->",
|
|
"",
|
|
f"> Mirrored from `{source.path}`.",
|
|
f"> Origin: `{source.origin}`.",
|
|
"> Active tasks and changing state belong in Gitea issues; this wiki page is durable project context.",
|
|
"",
|
|
"---",
|
|
"",
|
|
]
|
|
if source.path.suffix.lower() == ".csv":
|
|
return "\n".join(header + ["```csv", raw.rstrip(), "```", ""])
|
|
return "\n".join(header) + raw.rstrip() + "\n"
|
|
|
|
|
|
def put_wiki_page(
|
|
client: GiteaClient,
|
|
owner: str,
|
|
repo: str,
|
|
title: str,
|
|
content: str,
|
|
existing_page_names: dict[str, str],
|
|
*,
|
|
overwrite_unmanaged: bool,
|
|
) -> bool:
|
|
body = {
|
|
"title": title,
|
|
"content_base64": base64.b64encode(content.encode("utf-8")).decode("ascii"),
|
|
"message": f"Sync {title} from project files",
|
|
}
|
|
if title in existing_page_names:
|
|
page_name = existing_page_names[title]
|
|
current = client.request_json("GET", repo_path(owner, repo, f"/wiki/page/{quote_wiki_page_name(page_name)}"))
|
|
current_content = decode_wiki_content(current)
|
|
if current_content == content:
|
|
return False
|
|
if MANAGED_MARKER not in current_content and not overwrite_unmanaged:
|
|
print(f"skip unmanaged existing wiki page {owner}/{repo}:{title}")
|
|
return False
|
|
client.request_json("PATCH", repo_path(owner, repo, f"/wiki/page/{quote_wiki_page_name(page_name)}"), body=body)
|
|
return True
|
|
client.request_json("POST", repo_path(owner, repo, "/wiki/new"), body=body)
|
|
existing_page_names[title] = title
|
|
return True
|
|
|
|
|
|
def quote_wiki_page_name(value: str) -> str:
|
|
return urllib.parse.quote(value, safe="+")
|
|
|
|
|
|
def decode_wiki_content(page: dict[str, Any]) -> str:
|
|
encoded = str(page.get("content_base64") or "")
|
|
if not encoded:
|
|
return ""
|
|
return base64.b64decode(encoded).decode("utf-8")
|
|
|
|
|
|
def read_text(path: pathlib.Path) -> str:
|
|
try:
|
|
return path.read_text(encoding="utf-8")
|
|
except UnicodeDecodeError:
|
|
return path.read_text(encoding="latin-1")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|