538 lines
22 KiB
Python
538 lines
22 KiB
Python
import argparse
|
|
import json
|
|
import os
|
|
import re
|
|
import sys
|
|
import time
|
|
from pathlib import Path
|
|
from urllib.error import HTTPError, URLError
|
|
from urllib.request import Request, urlopen
|
|
|
|
import psycopg
|
|
from psycopg.rows import dict_row
|
|
|
|
|
|
DEFAULT_WIKI_URL = "http://192.168.195.241:38353"
|
|
PAGE_RULE_ROLES = ["write:pages", "manage:pages"]
|
|
SYSTEM_PERMISSION = "manage:system"
|
|
|
|
|
|
class WikiUpdateError(RuntimeError):
|
|
pass
|
|
|
|
|
|
def load_env(path):
|
|
values = {}
|
|
if not path.exists():
|
|
return values
|
|
for raw_line in path.read_text(encoding="utf-8").splitlines():
|
|
line = raw_line.strip()
|
|
if not line or line.startswith("#") or "=" not in line:
|
|
continue
|
|
key, value = line.split("=", 1)
|
|
values[key.strip()] = value.strip().strip('"').strip("'")
|
|
return values
|
|
|
|
|
|
def normalize_json(value):
|
|
if value is None or isinstance(value, (dict, list)):
|
|
return value
|
|
if isinstance(value, str):
|
|
return json.loads(value)
|
|
return value
|
|
|
|
|
|
def normalize_wiki_path(path, locale):
|
|
normalized = path.strip().strip("/")
|
|
locale_prefix = f"{locale}/"
|
|
if normalized.startswith(locale_prefix):
|
|
normalized = normalized[len(locale_prefix) :]
|
|
return normalized
|
|
|
|
|
|
def rule_matches_path(rule, page_path):
|
|
path = str(rule.get("path") or "").strip("/")
|
|
if not path:
|
|
return True
|
|
return page_path == path or page_path.startswith(path.rstrip("/") + "/")
|
|
|
|
|
|
def graphql(wiki_url, query, variables=None, token=None):
|
|
body = json.dumps({"query": query, "variables": variables or {}}).encode("utf-8")
|
|
headers = {"Content-Type": "application/json", "Accept": "application/json"}
|
|
if token:
|
|
headers["Authorization"] = f"Bearer {token}"
|
|
request = Request(f"{wiki_url}/graphql", data=body, headers=headers, method="POST")
|
|
try:
|
|
with urlopen(request, timeout=30) as response:
|
|
return json.loads(response.read().decode("utf-8"))
|
|
except HTTPError as exc:
|
|
payload = exc.read().decode("utf-8", errors="replace")
|
|
return {"transportError": f"HTTP {exc.code}", "payload": payload}
|
|
except URLError as exc:
|
|
return {"transportError": str(exc)}
|
|
|
|
|
|
def graphql_has_errors(label, payload):
|
|
if payload.get("errors"):
|
|
print(f"{label}=errors")
|
|
print(json.dumps(payload["errors"], ensure_ascii=True))
|
|
return True
|
|
if payload.get("transportError"):
|
|
print(f"{label}=transport_error {payload['transportError']}")
|
|
if payload.get("payload"):
|
|
print(payload["payload"][:1000])
|
|
return True
|
|
print(f"{label}=ok")
|
|
return False
|
|
|
|
|
|
def response_result(payload, operation):
|
|
result = payload.get("data", {}).get("pages", {}).get(operation, {}).get("responseResult", {})
|
|
print(
|
|
"pages.{operation}.response=succeeded={succeeded} errorCode={errorCode} slug={slug} message={message}".format(
|
|
operation=operation,
|
|
succeeded=result.get("succeeded"),
|
|
errorCode=result.get("errorCode"),
|
|
slug=result.get("slug"),
|
|
message=result.get("message"),
|
|
)
|
|
)
|
|
return bool(result.get("succeeded"))
|
|
|
|
|
|
def get_json_cast(cur, column_name):
|
|
cur.execute(
|
|
"""
|
|
SELECT udt_name
|
|
FROM information_schema.columns
|
|
WHERE table_name = 'groups' AND column_name = %s
|
|
""",
|
|
(column_name,),
|
|
)
|
|
row = cur.fetchone()
|
|
if not row:
|
|
raise WikiUpdateError(f"groups.{column_name} was not found")
|
|
return "jsonb" if row["udt_name"] == "jsonb" else "json"
|
|
|
|
|
|
def connect_info(env):
|
|
dbname = env.get("WIKIDATABASE") or env.get("PGDATABASE")
|
|
required = {
|
|
"PGHOST": env.get("PGHOST"),
|
|
"PGUSER": env.get("PGUSER"),
|
|
"PGPASSWORD": env.get("PGPASSWORD"),
|
|
"WIKIDATABASE/PGDATABASE": dbname,
|
|
}
|
|
missing = [key for key, value in required.items() if not value]
|
|
if missing:
|
|
raise WikiUpdateError(f"Missing database environment values: {', '.join(missing)}")
|
|
return {
|
|
"host": env.get("PGHOST"),
|
|
"port": int(env.get("PGPORT", "5432")),
|
|
"dbname": dbname,
|
|
"user": env.get("PGUSER"),
|
|
"password": env.get("PGPASSWORD"),
|
|
}
|
|
|
|
|
|
def resolve_page(cur, page_id, page_path, locale, missing_ok=False):
|
|
if page_id is not None:
|
|
cur.execute(
|
|
"""
|
|
SELECT id, path, title, description, "isPublished", "isPrivate", "editorKey", "localeCode"
|
|
FROM pages
|
|
WHERE id = %s
|
|
""",
|
|
(page_id,),
|
|
)
|
|
else:
|
|
cur.execute(
|
|
"""
|
|
SELECT id, path, title, description, "isPublished", "isPrivate", "editorKey", "localeCode"
|
|
FROM pages
|
|
WHERE path = %s AND "localeCode" = %s
|
|
ORDER BY id
|
|
LIMIT 1
|
|
""",
|
|
(page_path, locale),
|
|
)
|
|
page = cur.fetchone()
|
|
if not page:
|
|
if missing_ok:
|
|
return None
|
|
raise WikiUpdateError(f"Wiki page was not found: locale={locale} path={page_path}")
|
|
return page
|
|
|
|
|
|
def fetch_tags(cur, page_id):
|
|
try:
|
|
cur.execute(
|
|
"""
|
|
SELECT t.tag
|
|
FROM tags t
|
|
JOIN "pageTags" pt ON pt."tagId" = t.id
|
|
WHERE pt."pageId" = %s
|
|
ORDER BY t.tag
|
|
""",
|
|
(page_id,),
|
|
)
|
|
return [row["tag"] for row in cur.fetchall()]
|
|
except Exception as exc:
|
|
print(f"tags_preserved=False reason={type(exc).__name__}")
|
|
return []
|
|
|
|
|
|
def find_auth_group(cur, group_name, identities):
|
|
cur.execute(
|
|
"""
|
|
SELECT g.id, g.name, g.permissions, g."pageRules"
|
|
FROM groups g
|
|
WHERE g.name = %s
|
|
AND EXISTS (
|
|
SELECT 1
|
|
FROM "userGroups" ug
|
|
JOIN users u ON u.id = ug."userId"
|
|
WHERE ug."groupId" = g.id
|
|
AND (lower(u.email) = ANY(%s) OR lower(u.name) = ANY(%s) OR lower(u."providerKey") = ANY(%s))
|
|
)
|
|
ORDER BY g.id
|
|
LIMIT 1
|
|
""",
|
|
(group_name, identities, identities, identities),
|
|
)
|
|
group = cur.fetchone()
|
|
if not group:
|
|
raise WikiUpdateError(f"Authenticated user is not in Wiki.js group: {group_name}")
|
|
return group
|
|
|
|
|
|
def ensure_page_rule(cur, group, page_path, patch_page_rule):
|
|
page_rules_cast = get_json_cast(cur, "pageRules")
|
|
rules = normalize_json(group["pageRules"]) or []
|
|
if not isinstance(rules, list):
|
|
raise WikiUpdateError("Group pageRules is not a JSON array")
|
|
|
|
relevant_rules = [rule for rule in rules if isinstance(rule, dict) and rule.get("deny") is not True and rule_matches_path(rule, page_path)]
|
|
if not relevant_rules:
|
|
raise WikiUpdateError(f"No allow Page Rule matches path: {page_path}")
|
|
|
|
missing_by_rule = []
|
|
changed = False
|
|
for rule in relevant_rules:
|
|
roles = rule.get("roles")
|
|
if not isinstance(roles, list):
|
|
roles = []
|
|
rule["roles"] = roles
|
|
missing = [role for role in PAGE_RULE_ROLES if role not in roles]
|
|
print(f"matched_page_rule_path={rule.get('path')} match={rule.get('match') or rule.get('matchType')} roles={roles}")
|
|
if missing:
|
|
missing_by_rule.append((rule, missing))
|
|
if patch_page_rule:
|
|
for role in missing:
|
|
roles.append(role)
|
|
changed = True
|
|
|
|
if missing_by_rule and not patch_page_rule:
|
|
missing_roles = sorted({role for _rule, missing in missing_by_rule for role in missing})
|
|
raise WikiUpdateError(
|
|
"Matching Page Rule is missing roles: "
|
|
+ ", ".join(missing_roles)
|
|
+ ". Add them in Wiki.js Groups > Page Rules or rerun with --patch-page-rule after user approval."
|
|
)
|
|
|
|
if changed:
|
|
cur.execute(
|
|
f"UPDATE groups SET \"pageRules\" = %s::{page_rules_cast} WHERE id = %s",
|
|
(json.dumps(rules), group["id"]),
|
|
)
|
|
print("page_rule_patched=True")
|
|
else:
|
|
print("page_rule_patched=False")
|
|
|
|
|
|
def grant_temporary_system_permission(cur, group):
|
|
permissions_cast = get_json_cast(cur, "permissions")
|
|
permissions = normalize_json(group["permissions"]) or []
|
|
if SYSTEM_PERMISSION in permissions:
|
|
print("temporary_manage_system_added=False reason=already_present")
|
|
return False
|
|
permissions.append(SYSTEM_PERMISSION)
|
|
cur.execute(
|
|
f"UPDATE groups SET permissions = %s::{permissions_cast} WHERE id = %s",
|
|
(json.dumps(permissions), group["id"]),
|
|
)
|
|
print("temporary_manage_system_added=True")
|
|
return True
|
|
|
|
|
|
def remove_temporary_system_permission(conninfo, group_id):
|
|
with psycopg.connect(**conninfo, row_factory=dict_row) as conn:
|
|
with conn.cursor() as cur:
|
|
permissions_cast = get_json_cast(cur, "permissions")
|
|
cur.execute("SELECT permissions FROM groups WHERE id = %s", (group_id,))
|
|
row = cur.fetchone()
|
|
permissions = normalize_json(row["permissions"]) or []
|
|
permissions = [permission for permission in permissions if permission != SYSTEM_PERMISSION]
|
|
cur.execute(
|
|
f"UPDATE groups SET permissions = %s::{permissions_cast} WHERE id = %s",
|
|
(json.dumps(permissions), group_id),
|
|
)
|
|
conn.commit()
|
|
print("temporary_manage_system_removed=True")
|
|
|
|
|
|
def login(wiki_url, identities, password):
|
|
login_query = """
|
|
mutation Login($username: String!, $password: String!, $strategy: String!) {
|
|
authentication {
|
|
login(username: $username, password: $password, strategy: $strategy) {
|
|
responseResult { succeeded errorCode slug message }
|
|
jwt
|
|
}
|
|
}
|
|
}
|
|
"""
|
|
for index, identity in enumerate(identities, start=1):
|
|
payload = graphql(wiki_url, login_query, {"username": identity, "password": password, "strategy": "local"})
|
|
login_data = payload.get("data", {}).get("authentication", {}).get("login") or {}
|
|
result = login_data.get("responseResult", {})
|
|
print(
|
|
"login[{index}]=succeeded={succeeded} errorCode={errorCode} slug={slug} message={message} token_received={token_received}".format(
|
|
index=index,
|
|
succeeded=result.get("succeeded"),
|
|
errorCode=result.get("errorCode"),
|
|
slug=result.get("slug"),
|
|
message=result.get("message"),
|
|
token_received=bool(login_data.get("jwt")),
|
|
)
|
|
)
|
|
if login_data.get("jwt"):
|
|
return login_data["jwt"]
|
|
raise WikiUpdateError("Wiki.js login failed for all configured identities")
|
|
|
|
|
|
def extract_style_to_script_css(content):
|
|
match = re.search(r"<style>\s*(.*?)\s*</style>", content, flags=re.IGNORECASE | re.DOTALL)
|
|
if not match:
|
|
print("script_css_extracted=False")
|
|
return content, ""
|
|
css = match.group(1).strip()
|
|
content_without_style = (content[: match.start()] + content[match.end() :]).lstrip()
|
|
print(f"script_css_extracted=True length={len(css)}")
|
|
return content_without_style, css
|
|
|
|
|
|
def extract_script_to_script_js(content):
|
|
match = re.search(r"<script>\s*(.*?)\s*</script>", content, flags=re.IGNORECASE | re.DOTALL)
|
|
if not match:
|
|
print("script_js_extracted=False")
|
|
return content, ""
|
|
script_js = match.group(0).strip()
|
|
content_without_script = (content[: match.start()] + content[match.end() :]).rstrip()
|
|
print(f"script_js_extracted=True length={len(script_js)}")
|
|
return content_without_script, script_js
|
|
|
|
|
|
def update_page(wiki_url, token, page, args, content, tags, script_css, script_js):
|
|
update_query = """
|
|
mutation UpdatePage($id: Int!, $content: String!, $editor: String!, $locale: String!, $path: String!, $title: String!, $description: String!, $isPublished: Boolean!, $isPrivate: Boolean!, $tags: [String]!, $scriptCss: String, $scriptJs: String) {
|
|
pages {
|
|
update(id: $id, content: $content, editor: $editor, locale: $locale, path: $path, title: $title, description: $description, isPublished: $isPublished, isPrivate: $isPrivate, tags: $tags, scriptCss: $scriptCss, scriptJs: $scriptJs) {
|
|
responseResult { succeeded errorCode slug message }
|
|
page { id path title updatedAt }
|
|
}
|
|
}
|
|
}
|
|
"""
|
|
payload = graphql(
|
|
wiki_url,
|
|
update_query,
|
|
{
|
|
"id": page["id"],
|
|
"content": content,
|
|
"editor": args.editor or page["editorKey"] or "markdown",
|
|
"locale": page["localeCode"],
|
|
"path": page["path"],
|
|
"title": args.title or page["title"],
|
|
"description": args.description if args.description is not None else (page["description"] or ""),
|
|
"isPublished": bool(page["isPublished"]),
|
|
"isPrivate": bool(page["isPrivate"]),
|
|
"tags": tags,
|
|
"scriptCss": script_css,
|
|
"scriptJs": script_js,
|
|
},
|
|
token,
|
|
)
|
|
if graphql_has_errors("pages.update", payload):
|
|
return False
|
|
return response_result(payload, "update")
|
|
|
|
|
|
def create_page(wiki_url, token, args, page_path, content, script_css, script_js):
|
|
create_query = """
|
|
mutation CreatePage($content: String!, $description: String!, $editor: String!, $isPublished: Boolean!, $isPrivate: Boolean!, $locale: String!, $path: String!, $publishEndDate: Date, $publishStartDate: Date, $scriptCss: String, $scriptJs: String, $tags: [String]!, $title: String!) {
|
|
pages {
|
|
create(content: $content, description: $description, editor: $editor, isPublished: $isPublished, isPrivate: $isPrivate, locale: $locale, path: $path, publishEndDate: $publishEndDate, publishStartDate: $publishStartDate, scriptCss: $scriptCss, scriptJs: $scriptJs, tags: $tags, title: $title) {
|
|
responseResult { succeeded errorCode slug message }
|
|
page { id path title updatedAt }
|
|
}
|
|
}
|
|
}
|
|
"""
|
|
title = args.title or page_path.rstrip("/").rsplit("/", 1)[-1].replace("_", " ")
|
|
payload = graphql(
|
|
wiki_url,
|
|
create_query,
|
|
{
|
|
"content": content,
|
|
"description": args.description or "",
|
|
"editor": args.editor or "markdown",
|
|
"isPublished": True,
|
|
"isPrivate": False,
|
|
"locale": args.locale,
|
|
"path": page_path,
|
|
"publishEndDate": None,
|
|
"publishStartDate": None,
|
|
"scriptCss": script_css,
|
|
"scriptJs": script_js,
|
|
"tags": [],
|
|
"title": title,
|
|
},
|
|
token,
|
|
)
|
|
if graphql_has_errors("pages.create", payload):
|
|
return None
|
|
if not response_result(payload, "create"):
|
|
return None
|
|
page = payload.get("data", {}).get("pages", {}).get("create", {}).get("page")
|
|
if not page or not page.get("id"):
|
|
print("pages.create.page_missing=True")
|
|
return None
|
|
page["localeCode"] = args.locale
|
|
print(f"created_page_id={page['id']} locale={args.locale} path={page['path']}")
|
|
return page
|
|
|
|
|
|
def render_page(wiki_url, token, page_id):
|
|
render_query = """
|
|
mutation RenderPage($id: Int!) { pages { render(id: $id) { responseResult { succeeded errorCode slug message } } } }
|
|
"""
|
|
payload = graphql(wiki_url, render_query, {"id": page_id}, token)
|
|
if graphql_has_errors("pages.render", payload):
|
|
return False
|
|
return response_result(payload, "render")
|
|
|
|
|
|
def flush_cache(wiki_url, token):
|
|
flush_query = """
|
|
mutation FlushCache { pages { flushCache { responseResult { succeeded errorCode slug message } } } }
|
|
"""
|
|
payload = graphql(wiki_url, flush_query, token=token)
|
|
if graphql_has_errors("pages.flushCache", payload):
|
|
return False
|
|
return response_result(payload, "flushCache")
|
|
|
|
|
|
def verify_frontend(wiki_url, locale, page_path, must_contain, must_not_contain):
|
|
url = f"{wiki_url}/{locale}/{page_path}"
|
|
ok = True
|
|
targets = [("plain", url), ("cache_bust", f"{url}?_={int(time.time())}")]
|
|
for label, target_url in targets:
|
|
request = Request(target_url, headers={"Cache-Control": "no-cache", "Pragma": "no-cache", "Accept": "text/html"})
|
|
with urlopen(request, timeout=30) as response:
|
|
text = response.read().decode("utf-8", errors="replace")
|
|
print(f"frontend_{label}_http_status={response.status}")
|
|
for needle in must_contain:
|
|
contains = needle in text
|
|
print(f"frontend_{label}_must_contain={json.dumps(needle, ensure_ascii=True)} result={contains}")
|
|
ok = ok and contains
|
|
for needle in must_not_contain:
|
|
contains = needle in text
|
|
print(f"frontend_{label}_must_not_contain={json.dumps(needle, ensure_ascii=True)} result={not contains}")
|
|
ok = ok and not contains
|
|
return ok
|
|
|
|
|
|
def parse_args():
|
|
parser = argparse.ArgumentParser(description="Update an existing Wiki.js page through the official API, then render, flush cache, and verify the frontend.")
|
|
parser.add_argument("--path", required=True, help="Wiki.js page path without locale, e.g. home/Space_Shuttles/Vulture_Shuttle/html_test")
|
|
parser.add_argument("--content-file", required=True, help="UTF-8 local content file to publish")
|
|
parser.add_argument("--locale", default="zh")
|
|
parser.add_argument("--page-id", type=int)
|
|
parser.add_argument("--title")
|
|
parser.add_argument("--description")
|
|
parser.add_argument("--editor", help="Wiki.js editor key; defaults to the existing page editorKey")
|
|
parser.add_argument("--wiki-url", default=os.environ.get("WIKI_URL", DEFAULT_WIKI_URL).rstrip("/"))
|
|
parser.add_argument("--env-file", default=".env")
|
|
parser.add_argument("--auth-group", default="automation")
|
|
parser.add_argument("--patch-page-rule", action="store_true", help="Patch missing write/manage Page Rule roles. Ask the user before using this.")
|
|
parser.add_argument("--temporary-manage-system", action="store_true", help="Temporarily grant manage:system for render/flush. Ask the user before using this.")
|
|
parser.add_argument("--extract-style-to-script-css", action="store_true", help="Extract the first <style> block from HTML content and publish it as Wiki.js page CSS.")
|
|
parser.add_argument("--extract-script-to-script-js", action="store_true", help="Extract the first <script> block from HTML content and publish it as Wiki.js page JavaScript.")
|
|
parser.add_argument("--create-if-missing", action="store_true", help="Create the page through Wiki.js pages.create if it does not already exist.")
|
|
parser.add_argument("--must-contain", action="append", default=[])
|
|
parser.add_argument("--must-not-contain", action="append", default=[])
|
|
return parser.parse_args()
|
|
|
|
|
|
def main():
|
|
args = parse_args()
|
|
repo_root = Path.cwd()
|
|
env = load_env((repo_root / args.env_file).resolve())
|
|
conninfo = connect_info(env)
|
|
page_path = normalize_wiki_path(args.path, args.locale)
|
|
content = (repo_root / args.content_file).read_text(encoding="utf-8")
|
|
script_css = ""
|
|
script_js = ""
|
|
if args.extract_style_to_script_css:
|
|
content, script_css = extract_style_to_script_css(content)
|
|
if args.extract_script_to_script_js:
|
|
content, script_js = extract_script_to_script_js(content)
|
|
identities = []
|
|
for candidate in (env.get("wiki_useremail"), env.get("wiki_username")):
|
|
if candidate and candidate not in identities:
|
|
identities.append(candidate)
|
|
if not identities or not env.get("wiki_password"):
|
|
raise WikiUpdateError("Missing wiki_useremail/wiki_username or wiki_password in .env")
|
|
|
|
temporary_group_id = None
|
|
try:
|
|
with psycopg.connect(**conninfo, row_factory=dict_row) as conn:
|
|
with conn.cursor() as cur:
|
|
page = resolve_page(cur, args.page_id, page_path, args.locale, args.create_if_missing)
|
|
if page:
|
|
tags = fetch_tags(cur, page["id"])
|
|
print(f"page_id={page['id']} locale={page['localeCode']} path={page['path']} preserved_tags={len(tags)}")
|
|
rule_path = page["path"]
|
|
else:
|
|
tags = []
|
|
rule_path = page_path
|
|
print(f"page_found=False locale={args.locale} path={page_path} create_if_missing=True")
|
|
group = find_auth_group(cur, args.auth_group, [identity.lower() for identity in identities])
|
|
ensure_page_rule(cur, group, rule_path, args.patch_page_rule)
|
|
if args.temporary_manage_system and grant_temporary_system_permission(cur, group):
|
|
temporary_group_id = group["id"]
|
|
conn.commit()
|
|
|
|
token = login(args.wiki_url, identities, env["wiki_password"])
|
|
if page:
|
|
page_ok = update_page(args.wiki_url, token, page, args, content, tags, script_css, script_js)
|
|
else:
|
|
page = create_page(args.wiki_url, token, args, page_path, content, script_css, script_js)
|
|
page_ok = bool(page)
|
|
render_ok = render_page(args.wiki_url, token, page["id"]) if page else False
|
|
flush_ok = flush_cache(args.wiki_url, token)
|
|
frontend_ok = verify_frontend(args.wiki_url, page.get("localeCode", args.locale), page["path"], args.must_contain, args.must_not_contain) if page else False
|
|
return 0 if page_ok and render_ok and flush_ok and frontend_ok else 1
|
|
finally:
|
|
if temporary_group_id is not None:
|
|
remove_temporary_system_permission(conninfo, temporary_group_id)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
sys.exit(main())
|
|
except WikiUpdateError as exc:
|
|
print(f"wiki_update_error={exc}")
|
|
sys.exit(2) |