feat: add wiki publishing tooling
This commit is contained in:
@@ -0,0 +1,168 @@
|
||||
import re
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
HTML_FILES = [
|
||||
"data/wiki/vulture_shuttle_zh.html",
|
||||
"data/wiki/vulture_shuttle_en.html",
|
||||
"data/wiki/echo_shuttle_zh.html",
|
||||
"data/wiki/echo_shuttle_en.html",
|
||||
"data/wiki/enterprise_shuttle_zh.html",
|
||||
"data/wiki/enterprise_shuttle_en.html",
|
||||
]
|
||||
|
||||
STYLE_RE = re.compile(r"</style>", re.IGNORECASE)
|
||||
SCRIPT_RE = re.compile(r"<script>.*?</script>", re.IGNORECASE | re.DOTALL)
|
||||
|
||||
LIGHTBOX_CSS = """
|
||||
.wiki-article img[data-wiki-asset], .vulture-article img[data-wiki-asset] { cursor: zoom-in; }
|
||||
.wiki-image-lightbox { position: fixed; inset: 0; display: flex; align-items: center; justify-content: center; padding: 24px; background: rgba(32,33,34,0.84); z-index: 10000; box-sizing: border-box; }
|
||||
.wiki-image-lightbox[hidden] { display: none; }
|
||||
.wiki-image-lightbox-panel { position: relative; width: min(1120px, 96vw); max-height: 92vh; overflow: auto; margin: 0; padding: 10px; border: 1px solid #a2a9b1; background: #fff; box-shadow: 0 12px 38px rgba(0,0,0,0.35); box-sizing: border-box; }
|
||||
.wiki-image-lightbox-panel img { display: block; width: 100%; height: auto; max-height: 82vh; object-fit: contain; background: #000; }
|
||||
.wiki-image-lightbox-close { position: sticky; top: 0; float: right; width: 30px; height: 30px; margin: -2px -2px 8px 10px; border: 1px solid #a2a9b1; background: #eaecf0; color: #202122; font-weight: bold; line-height: 26px; cursor: pointer; }
|
||||
.wiki-image-lightbox-close:hover, .wiki-image-lightbox-close:focus { background: #fff; }
|
||||
body.wiki-image-lightbox-open { overflow: hidden; }
|
||||
@media (max-width: 820px) { .wiki-image-lightbox { padding: 12px; } .wiki-image-lightbox-panel { width: 98vw; max-height: 94vh; } }
|
||||
""".strip()
|
||||
|
||||
LIGHTBOX_JS = """
|
||||
(function () {
|
||||
function ready(fn) {
|
||||
if (document.readyState === 'loading') {
|
||||
document.addEventListener('DOMContentLoaded', fn, { once: true });
|
||||
} else {
|
||||
fn();
|
||||
}
|
||||
}
|
||||
|
||||
function captionFor(image) {
|
||||
var container = image.closest('figure, td, tr, .infobox-image');
|
||||
if (!container) {
|
||||
return image.alt || '';
|
||||
}
|
||||
var caption = container.querySelector('.thumbcaption');
|
||||
return caption ? caption.textContent.trim() : (image.alt || '');
|
||||
}
|
||||
|
||||
function install(article) {
|
||||
if (!article || article.dataset.lightboxReady === '1') {
|
||||
return;
|
||||
}
|
||||
article.dataset.lightboxReady = '1';
|
||||
|
||||
var overlay = document.createElement('div');
|
||||
overlay.className = 'wiki-image-lightbox';
|
||||
overlay.hidden = true;
|
||||
overlay.innerHTML = '<figure class="wiki-image-lightbox-panel" role="dialog" aria-modal="true"><button class="wiki-image-lightbox-close" type="button" aria-label="Close image preview">x</button><img alt=""><figcaption class="thumbcaption"></figcaption></figure>';
|
||||
document.body.appendChild(overlay);
|
||||
|
||||
var previewImage = overlay.querySelector('img');
|
||||
var previewCaption = overlay.querySelector('figcaption');
|
||||
var closeButton = overlay.querySelector('.wiki-image-lightbox-close');
|
||||
|
||||
function closePreview() {
|
||||
overlay.hidden = true;
|
||||
previewImage.removeAttribute('src');
|
||||
previewCaption.textContent = '';
|
||||
document.body.classList.remove('wiki-image-lightbox-open');
|
||||
}
|
||||
|
||||
function openPreview(image, event) {
|
||||
event.preventDefault();
|
||||
event.stopPropagation();
|
||||
if (event.stopImmediatePropagation) {
|
||||
event.stopImmediatePropagation();
|
||||
}
|
||||
previewImage.src = image.currentSrc || image.src;
|
||||
previewImage.alt = image.alt || '';
|
||||
previewCaption.textContent = captionFor(image);
|
||||
overlay.hidden = false;
|
||||
document.body.classList.add('wiki-image-lightbox-open');
|
||||
closeButton.focus();
|
||||
}
|
||||
|
||||
Array.prototype.forEach.call(article.querySelectorAll('img[data-wiki-asset]'), function (image) {
|
||||
if (image.dataset.lightboxBound === '1') {
|
||||
return;
|
||||
}
|
||||
image.dataset.lightboxBound = '1';
|
||||
image.addEventListener('click', function (event) {
|
||||
openPreview(image, event);
|
||||
}, true);
|
||||
});
|
||||
|
||||
article.addEventListener('click', function (event) {
|
||||
var image = event.target.closest && event.target.closest('img[data-wiki-asset]');
|
||||
if (!image || !article.contains(image) || image.closest('.wiki-image-lightbox')) {
|
||||
return;
|
||||
}
|
||||
openPreview(image, event);
|
||||
});
|
||||
|
||||
closeButton.addEventListener('click', closePreview);
|
||||
overlay.addEventListener('click', function (event) {
|
||||
if (event.target === overlay) {
|
||||
closePreview();
|
||||
}
|
||||
});
|
||||
document.addEventListener('keydown', function (event) {
|
||||
if (event.key === 'Escape' && !overlay.hidden) {
|
||||
closePreview();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
function installAll() {
|
||||
Array.prototype.forEach.call(document.querySelectorAll('.wiki-article, .vulture-article'), install);
|
||||
}
|
||||
|
||||
ready(function () {
|
||||
installAll();
|
||||
if (window.MutationObserver) {
|
||||
var observer = new MutationObserver(installAll);
|
||||
observer.observe(document.body, { childList: true, subtree: true });
|
||||
}
|
||||
window.setTimeout(installAll, 500);
|
||||
window.setTimeout(installAll, 1500);
|
||||
window.setTimeout(installAll, 3000);
|
||||
});
|
||||
}());
|
||||
""".strip()
|
||||
|
||||
|
||||
def ensure_css(content):
|
||||
if ".wiki-image-lightbox" in content:
|
||||
return content
|
||||
if not STYLE_RE.search(content):
|
||||
raise RuntimeError("No style block found")
|
||||
return STYLE_RE.sub("\n" + LIGHTBOX_CSS + "\n</style>", content, count=1)
|
||||
|
||||
|
||||
def ensure_script(content):
|
||||
script = "<script>\n" + LIGHTBOX_JS + "\n</script>"
|
||||
if SCRIPT_RE.search(content):
|
||||
return SCRIPT_RE.sub(script, content, count=1)
|
||||
return content.rstrip() + "\n" + script + "\n"
|
||||
|
||||
|
||||
def main():
|
||||
repo_root = Path.cwd()
|
||||
changed = 0
|
||||
for relative in HTML_FILES:
|
||||
path = repo_root / relative
|
||||
original = path.read_text(encoding="utf-8")
|
||||
updated = ensure_script(ensure_css(original))
|
||||
if updated != original:
|
||||
path.write_text(updated, encoding="utf-8", newline="\n")
|
||||
changed += 1
|
||||
print(f"lightbox_updated file={relative}")
|
||||
else:
|
||||
print(f"lightbox_unchanged file={relative}")
|
||||
print(f"lightbox_summary html_changed={changed}")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
@@ -0,0 +1,193 @@
|
||||
import argparse
|
||||
import json
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from urllib.error import HTTPError, URLError
|
||||
from urllib.request import Request, urlopen
|
||||
|
||||
import psycopg
|
||||
from psycopg.rows import dict_row
|
||||
|
||||
from wiki_update_page import DEFAULT_WIKI_URL, WikiUpdateError, connect_info, find_auth_group, load_env, login
|
||||
|
||||
|
||||
TEMP_PERMISSIONS = ["manage:system", "manage:assets", "read:assets", "write:assets"]
|
||||
|
||||
|
||||
def normalize_json(value):
|
||||
if value is None or isinstance(value, (dict, list)):
|
||||
return value
|
||||
return json.loads(value)
|
||||
|
||||
|
||||
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 grant_temporary_permissions(cur, group):
|
||||
permissions_cast = get_json_cast(cur, "permissions")
|
||||
permissions = normalize_json(group["permissions"]) or []
|
||||
added = []
|
||||
for permission in TEMP_PERMISSIONS:
|
||||
if permission not in permissions:
|
||||
permissions.append(permission)
|
||||
added.append(permission)
|
||||
if added:
|
||||
cur.execute(
|
||||
f"UPDATE groups SET permissions = %s::{permissions_cast} WHERE id = %s",
|
||||
(json.dumps(permissions), group["id"]),
|
||||
)
|
||||
print(f"temporary_permissions_added={','.join(added) if added else 'none'}")
|
||||
return added
|
||||
|
||||
|
||||
def remove_temporary_permissions(conninfo, group_id, added):
|
||||
if not added:
|
||||
return
|
||||
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 not in added]
|
||||
cur.execute(
|
||||
f"UPDATE groups SET permissions = %s::{permissions_cast} WHERE id = %s",
|
||||
(json.dumps(permissions), group_id),
|
||||
)
|
||||
conn.commit()
|
||||
print(f"temporary_permissions_removed={','.join(added)}")
|
||||
|
||||
|
||||
def misplaced_assets(cur):
|
||||
cur.execute(
|
||||
"""
|
||||
SELECT a.id, a.filename, a.hash, a."fileSize", f.slug AS folder_slug
|
||||
FROM assets a
|
||||
JOIN "assetFolders" f ON f.id = a."folderId"
|
||||
WHERE f.slug = 'vulture'
|
||||
AND (left(a.filename, 5) = 'echo_' OR left(a.filename, 11) = 'enterprise_')
|
||||
ORDER BY a.filename, a.id
|
||||
"""
|
||||
)
|
||||
return cur.fetchall()
|
||||
|
||||
|
||||
def graphql(wiki_url, query, variables, token):
|
||||
body = json.dumps({"query": query, "variables": variables}).encode("utf-8")
|
||||
request = Request(
|
||||
f"{wiki_url}/graphql",
|
||||
data=body,
|
||||
headers={"Content-Type": "application/json", "Accept": "application/json", "Authorization": f"Bearer {token}"},
|
||||
method="POST",
|
||||
)
|
||||
try:
|
||||
with urlopen(request, timeout=120) as response:
|
||||
return json.loads(response.read().decode("utf-8"))
|
||||
except HTTPError as exc:
|
||||
return {"transportError": f"HTTP {exc.code}", "payload": exc.read().decode("utf-8", errors="replace")}
|
||||
except URLError as exc:
|
||||
return {"transportError": str(exc)}
|
||||
|
||||
|
||||
def delete_asset(wiki_url, token, asset_id):
|
||||
mutation = """
|
||||
mutation DeleteAsset($id: Int!) {
|
||||
assets {
|
||||
deleteAsset(id: $id) {
|
||||
responseResult { succeeded errorCode slug message }
|
||||
}
|
||||
}
|
||||
}
|
||||
"""
|
||||
payload = graphql(wiki_url, mutation, {"id": asset_id}, token)
|
||||
if payload.get("errors"):
|
||||
print(f"asset_delete id={asset_id} graphql_errors={json.dumps(payload['errors'], ensure_ascii=True)}")
|
||||
return False
|
||||
if payload.get("transportError"):
|
||||
print(f"asset_delete id={asset_id} transport_error={payload['transportError']} payload={payload.get('payload', '')[:500]}")
|
||||
return False
|
||||
result = payload.get("data", {}).get("assets", {}).get("deleteAsset", {}).get("responseResult", {})
|
||||
print(
|
||||
"asset_delete id={id} succeeded={succeeded} errorCode={errorCode} slug={slug} message={message}".format(
|
||||
id=asset_id,
|
||||
succeeded=result.get("succeeded"),
|
||||
errorCode=result.get("errorCode"),
|
||||
slug=result.get("slug"),
|
||||
message=result.get("message"),
|
||||
)
|
||||
)
|
||||
return bool(result.get("succeeded"))
|
||||
|
||||
|
||||
def parse_args():
|
||||
parser = argparse.ArgumentParser(description="Delete Echo/Enterprise assets that were mistakenly uploaded under the Vulture folder.")
|
||||
parser.add_argument("--env-file", default=".env")
|
||||
parser.add_argument("--wiki-url", default=DEFAULT_WIKI_URL)
|
||||
parser.add_argument("--auth-group", default="automation")
|
||||
parser.add_argument("--delete", action="store_true", help="Actually delete matching assets. Without this flag, only lists matches.")
|
||||
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)
|
||||
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
|
||||
added_permissions = []
|
||||
try:
|
||||
with psycopg.connect(**conninfo, row_factory=dict_row) as conn:
|
||||
with conn.cursor() as cur:
|
||||
assets = misplaced_assets(cur)
|
||||
print(f"misplaced_asset_count={len(assets)}")
|
||||
for asset in assets:
|
||||
print(f"misplaced_asset id={asset['id']} folder=/{asset['folder_slug']} filename={asset['filename']} size={asset['fileSize']}")
|
||||
if not args.delete or not assets:
|
||||
return 0
|
||||
group = find_auth_group(cur, args.auth_group, [identity.lower() for identity in identities])
|
||||
added_permissions = grant_temporary_permissions(cur, group)
|
||||
temporary_group_id = group["id"]
|
||||
conn.commit()
|
||||
|
||||
token = login(args.wiki_url.rstrip("/"), identities, env["wiki_password"])
|
||||
ok = True
|
||||
for asset in assets:
|
||||
ok = delete_asset(args.wiki_url.rstrip("/"), token, asset["id"]) and ok
|
||||
if not ok:
|
||||
return 1
|
||||
|
||||
with psycopg.connect(**conninfo, row_factory=dict_row) as conn:
|
||||
with conn.cursor() as cur:
|
||||
remaining = misplaced_assets(cur)
|
||||
print(f"misplaced_asset_remaining={len(remaining)}")
|
||||
return 0 if not remaining else 1
|
||||
finally:
|
||||
if temporary_group_id is not None:
|
||||
remove_temporary_permissions(conninfo, temporary_group_id, added_permissions)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
try:
|
||||
sys.exit(main())
|
||||
except WikiUpdateError as exc:
|
||||
print(f"wiki_cleanup_error={exc}")
|
||||
sys.exit(2)
|
||||
@@ -0,0 +1,149 @@
|
||||
import argparse
|
||||
import base64
|
||||
import re
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
from PIL import Image, ImageOps
|
||||
|
||||
|
||||
ASSET_SOURCES = {
|
||||
"/vulture/vulture_shuttle_full_stack_original.png": "data/KSP/wiki-upload/vulture_shuttle_full_stack_at_launch_pad.jpg",
|
||||
"/vulture/vulture_shuttle_ignition_original.png": "data/KSP/wiki-upload/vulture_shuttle_ignition.jpg",
|
||||
"/vulture/vulture_ascent_original.png": "data/KSP/wiki-upload/vulture_ascent.jpg",
|
||||
"/vulture/vulture_booster_recovery_original.png": "data/KSP/wiki-upload/vulture_booster_recovery.jpg",
|
||||
"/vulture/vulture_booster_separation_original.png": "data/KSP/wiki-upload/vulture_booster_separation.jpg",
|
||||
"/echo/echo_takeoff_side_view.png": "data/KSP/wiki-upload-echo-current/echo_takeoff_side_view.png",
|
||||
"/echo/echo_mk2_cross_section.png": "data/KSP/wiki-upload-echo-current/echo_mk2_cross_section.png",
|
||||
"/echo/echo_vtvl.png": "data/KSP/wiki-upload-echo-current/echo_vtvl.png",
|
||||
"/echo/echo_approach_guanghan_station.png": "data/KSP/wiki-upload-echo-current/echo_approach_guanghan_station.png",
|
||||
"/echo/echo_high_energy_return_braking.png": "data/KSP/wiki-upload-echo-current/echo_high_energy_return_braking.png",
|
||||
"/enterprise/enterprise_takeoff_side_view.png": "data/KSP/wiki-upload-enterprise-current/enterprise_takeoff_side_view.png",
|
||||
"/enterprise/enterprise_payload_bay_view.png": "data/KSP/wiki-upload-enterprise-current/enterprise_payload_bay_view.png",
|
||||
"/enterprise/enterprise_dorsal_airframe_view.png": "data/KSP/wiki-upload-enterprise-current/enterprise_dorsal_airframe_view.png",
|
||||
}
|
||||
|
||||
HTML_FILES = [
|
||||
"data/wiki/vulture_shuttle_zh.html",
|
||||
"data/wiki/vulture_shuttle_en.html",
|
||||
"data/wiki/echo_shuttle_zh.html",
|
||||
"data/wiki/echo_shuttle_en.html",
|
||||
"data/wiki/enterprise_shuttle_zh.html",
|
||||
"data/wiki/enterprise_shuttle_en.html",
|
||||
]
|
||||
|
||||
IMG_RE = re.compile(r"<img\b[^>]*>", re.IGNORECASE)
|
||||
SRC_RE = re.compile(r'src="[^"]*"', re.IGNORECASE)
|
||||
ASSET_RE = re.compile(r'data-wiki-asset="([^"]+)"', re.IGNORECASE)
|
||||
|
||||
|
||||
def jpeg_name(asset_path):
|
||||
return asset_path.strip("/").replace("/", "__").rsplit(".", 1)[0] + ".jpg"
|
||||
|
||||
|
||||
def write_compressed_jpeg(repo_root, source, target, max_dimension, quality):
|
||||
image = Image.open(repo_root / source)
|
||||
image = ImageOps.exif_transpose(image)
|
||||
image.thumbnail((max_dimension, max_dimension), Image.Resampling.LANCZOS)
|
||||
if image.mode in ("RGBA", "LA"):
|
||||
background = Image.new("RGB", image.size, (255, 255, 255))
|
||||
background.paste(image, mask=image.getchannel("A"))
|
||||
image = background
|
||||
elif image.mode != "RGB":
|
||||
image = image.convert("RGB")
|
||||
target.parent.mkdir(parents=True, exist_ok=True)
|
||||
image.save(target, format="JPEG", quality=quality, optimize=True, progressive=True)
|
||||
|
||||
|
||||
def build_data_uris(repo_root, embed_dir, max_dimension, quality):
|
||||
data_uris = {}
|
||||
for asset_path, source in ASSET_SOURCES.items():
|
||||
source_path = repo_root / source
|
||||
if not source_path.exists():
|
||||
raise FileNotFoundError(f"Missing source image for {asset_path}: {source}")
|
||||
target = repo_root / embed_dir / jpeg_name(asset_path)
|
||||
write_compressed_jpeg(repo_root, source, target, max_dimension, quality)
|
||||
encoded = base64.b64encode(target.read_bytes()).decode("ascii")
|
||||
data_uris[asset_path] = f"data:image/jpeg;base64,{encoded}"
|
||||
print(f"embedded_source asset={asset_path} file={target.as_posix()} bytes={target.stat().st_size}")
|
||||
return data_uris
|
||||
|
||||
|
||||
def rewrite_img_tag(match, data_uris):
|
||||
tag = match.group(0)
|
||||
asset_match = ASSET_RE.search(tag)
|
||||
if not asset_match:
|
||||
return tag
|
||||
asset_path = asset_match.group(1)
|
||||
data_uri = data_uris.get(asset_path)
|
||||
if not data_uri:
|
||||
return tag
|
||||
if not SRC_RE.search(tag):
|
||||
return tag
|
||||
return SRC_RE.sub(f'src="{data_uri}"', tag, count=1)
|
||||
|
||||
|
||||
def restore_img_tag(match):
|
||||
tag = match.group(0)
|
||||
asset_match = ASSET_RE.search(tag)
|
||||
if not asset_match or not SRC_RE.search(tag):
|
||||
return tag
|
||||
return SRC_RE.sub(f'src="{asset_match.group(1)}"', tag, count=1)
|
||||
|
||||
|
||||
def rewrite_html(repo_root, html_files, data_uris):
|
||||
changed = 0
|
||||
for relative in html_files:
|
||||
path = repo_root / relative
|
||||
original = path.read_text(encoding="utf-8")
|
||||
updated = IMG_RE.sub(lambda match: rewrite_img_tag(match, data_uris), original)
|
||||
if updated != original:
|
||||
path.write_text(updated, encoding="utf-8", newline="\n")
|
||||
changed += 1
|
||||
count = updated.count('src="data:image/jpeg;base64,')
|
||||
print(f"html_updated file={relative} data_uri_count={count}")
|
||||
else:
|
||||
print(f"html_unchanged file={relative}")
|
||||
return changed
|
||||
|
||||
|
||||
def restore_html(repo_root, html_files):
|
||||
changed = 0
|
||||
for relative in html_files:
|
||||
path = repo_root / relative
|
||||
original = path.read_text(encoding="utf-8")
|
||||
updated = IMG_RE.sub(restore_img_tag, original)
|
||||
if updated != original:
|
||||
path.write_text(updated, encoding="utf-8", newline="\n")
|
||||
changed += 1
|
||||
count = updated.count('src="data:image/jpeg;base64,')
|
||||
print(f"html_restored file={relative} remaining_data_uri_count={count}")
|
||||
else:
|
||||
print(f"html_unchanged file={relative}")
|
||||
return changed
|
||||
|
||||
|
||||
def parse_args():
|
||||
parser = argparse.ArgumentParser(description="Embed compressed image fallbacks in Wiki.js HTML pages.")
|
||||
parser.add_argument("--embed-dir", default="data/KSP/wiki-embed-current")
|
||||
parser.add_argument("--max-dimension", type=int, default=1600)
|
||||
parser.add_argument("--quality", type=int, default=82)
|
||||
parser.add_argument("--restore-asset-src", action="store_true", help="Restore image src attributes from data-wiki-asset paths")
|
||||
return parser.parse_args()
|
||||
|
||||
|
||||
def main():
|
||||
args = parse_args()
|
||||
repo_root = Path.cwd()
|
||||
if args.restore_asset_src:
|
||||
changed = restore_html(repo_root, HTML_FILES)
|
||||
print(f"restore_summary html_changed={changed}")
|
||||
return 0
|
||||
data_uris = build_data_uris(repo_root, Path(args.embed_dir), args.max_dimension, args.quality)
|
||||
changed = rewrite_html(repo_root, HTML_FILES, data_uris)
|
||||
print(f"embed_summary assets={len(data_uris)} html_changed={changed}")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
@@ -0,0 +1,538 @@
|
||||
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)
|
||||
@@ -0,0 +1,236 @@
|
||||
import argparse
|
||||
import html
|
||||
import json
|
||||
import mimetypes
|
||||
import os
|
||||
import re
|
||||
import sys
|
||||
import uuid
|
||||
from pathlib import Path
|
||||
from urllib.error import HTTPError, URLError
|
||||
from urllib.request import Request, urlopen
|
||||
|
||||
import psycopg
|
||||
from psycopg.rows import dict_row
|
||||
|
||||
from wiki_update_page import DEFAULT_WIKI_URL, WikiUpdateError, connect_info, load_env, login
|
||||
|
||||
|
||||
def sanitize_upload_name(name):
|
||||
sanitized = re.sub(r"[\s,;#]+", "_", name.lower())
|
||||
sanitized = re.sub(r"[^a-z0-9._-]", "", sanitized)
|
||||
return sanitized.strip("._")
|
||||
|
||||
|
||||
def resolve_folder(cur, folder_slug, parent_id):
|
||||
cur.execute(
|
||||
"""
|
||||
SELECT id, name, slug, "parentId"
|
||||
FROM "assetFolders"
|
||||
WHERE slug = %s AND "parentId" IS NOT DISTINCT FROM %s
|
||||
ORDER BY id
|
||||
LIMIT 1
|
||||
""",
|
||||
(folder_slug, parent_id),
|
||||
)
|
||||
folder = cur.fetchone()
|
||||
if not folder:
|
||||
raise WikiUpdateError(f"Asset folder was not found: slug={folder_slug} parentId={parent_id}")
|
||||
return folder
|
||||
|
||||
|
||||
def existing_assets(cur, folder_id):
|
||||
cur.execute(
|
||||
"""
|
||||
SELECT id, filename, mime, "fileSize"
|
||||
FROM assets
|
||||
WHERE "folderId" IS NOT DISTINCT FROM %s
|
||||
ORDER BY filename
|
||||
""",
|
||||
(folder_id,),
|
||||
)
|
||||
return {row["filename"]: row for row in cur.fetchall()}
|
||||
|
||||
|
||||
def verify_asset(cur, folder_id, filename):
|
||||
cur.execute(
|
||||
"""
|
||||
SELECT id, filename, mime, "fileSize"
|
||||
FROM assets
|
||||
WHERE "folderId" IS NOT DISTINCT FROM %s AND filename = %s
|
||||
ORDER BY id DESC
|
||||
LIMIT 1
|
||||
""",
|
||||
(folder_id, filename),
|
||||
)
|
||||
return cur.fetchone()
|
||||
|
||||
|
||||
def verify_served_asset(wiki_url, folder_slug, filename):
|
||||
url = f"{wiki_url}/{folder_slug}/{filename}"
|
||||
request = Request(url, headers={"Accept": "image/*,*/*", "Cache-Control": "no-cache"})
|
||||
try:
|
||||
with urlopen(request, timeout=30) as response:
|
||||
content_type = response.headers.get("Content-Type", "")
|
||||
first_bytes = response.read(16)
|
||||
return response.status, content_type, len(first_bytes), ""
|
||||
except HTTPError as exc:
|
||||
preview = exc.read(200).decode("utf-8", errors="replace")
|
||||
return exc.code, exc.headers.get("Content-Type", ""), 0, response_preview(preview)
|
||||
except URLError as exc:
|
||||
return 0, "", 0, str(exc)
|
||||
|
||||
|
||||
def multipart_body(folder_id, file_path, upload_name):
|
||||
boundary = f"----wikijs-upload-{uuid.uuid4().hex}"
|
||||
mime = mimetypes.guess_type(upload_name)[0] or "application/octet-stream"
|
||||
chunks = []
|
||||
chunks.append(
|
||||
f"--{boundary}\r\n"
|
||||
"Content-Disposition: form-data; name=\"mediaUpload\"\r\n\r\n"
|
||||
f"{json.dumps({'folderId': folder_id})}\r\n".encode("utf-8")
|
||||
)
|
||||
chunks.append(
|
||||
f"--{boundary}\r\n"
|
||||
f"Content-Disposition: form-data; name=\"mediaUpload\"; filename=\"{upload_name}\"\r\n"
|
||||
f"Content-Type: {mime}\r\n\r\n".encode("utf-8")
|
||||
)
|
||||
chunks.append(file_path.read_bytes())
|
||||
chunks.append(f"\r\n--{boundary}--\r\n".encode("utf-8"))
|
||||
return boundary, b"".join(chunks)
|
||||
|
||||
|
||||
def upload_file(wiki_url, token, folder_id, file_path, upload_name):
|
||||
boundary, body = multipart_body(folder_id, file_path, upload_name)
|
||||
request = Request(
|
||||
f"{wiki_url}/u",
|
||||
data=body,
|
||||
headers={
|
||||
"Authorization": f"Bearer {token}",
|
||||
"Content-Type": f"multipart/form-data; boundary={boundary}",
|
||||
"Accept": "application/json, text/plain, */*",
|
||||
},
|
||||
method="POST",
|
||||
)
|
||||
try:
|
||||
with urlopen(request, timeout=120) as response:
|
||||
text = response.read().decode("utf-8", errors="replace")
|
||||
return response.status, text
|
||||
except HTTPError as exc:
|
||||
text = exc.read().decode("utf-8", errors="replace")
|
||||
return exc.code, text
|
||||
except URLError as exc:
|
||||
raise WikiUpdateError(f"Upload transport failed for {file_path.name}: {exc}") from exc
|
||||
|
||||
|
||||
def response_preview(text):
|
||||
plain = html.unescape(re.sub(r"<[^>]+>", " ", text))
|
||||
plain = re.sub(r"\s+", " ", plain).strip()
|
||||
return plain[:1600] or text[:1600].replace("\r", " ").replace("\n", " ")
|
||||
|
||||
|
||||
def collect_files(source_dir, names):
|
||||
if names:
|
||||
files = [source_dir / name for name in names]
|
||||
else:
|
||||
files = sorted(path for path in source_dir.iterdir() if path.is_file())
|
||||
missing = [str(path) for path in files if not path.exists()]
|
||||
if missing:
|
||||
raise WikiUpdateError("Missing local asset files: " + ", ".join(missing))
|
||||
return files
|
||||
|
||||
|
||||
def parse_args():
|
||||
parser = argparse.ArgumentParser(description="Upload files to an existing Wiki.js asset folder through the /u endpoint.")
|
||||
parser.add_argument("--source-dir", required=True, help="Local directory containing files to upload")
|
||||
parser.add_argument("--file", action="append", default=[], help="Specific file name inside --source-dir. Repeat to upload multiple files.")
|
||||
parser.add_argument("--upload-name", help="Uploaded filename to use when exactly one file is selected")
|
||||
parser.add_argument("--folder-slug", required=True, help="Existing Wiki.js asset folder slug, e.g. vulture")
|
||||
parser.add_argument("--parent-id", type=int, default=None)
|
||||
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("--skip-existing", action="store_true", help="Skip files whose sanitized names already exist in the target folder")
|
||||
parser.add_argument("--skip-http-verify", action="store_true", help="Skip the frontend GET check for uploaded asset URLs")
|
||||
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)
|
||||
source_dir = (repo_root / args.source_dir).resolve()
|
||||
if not source_dir.exists():
|
||||
raise WikiUpdateError(f"Source directory was not found: {source_dir}")
|
||||
files = collect_files(source_dir, args.file)
|
||||
if args.upload_name and len(files) != 1:
|
||||
raise WikiUpdateError("--upload-name can only be used with exactly one selected file")
|
||||
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")
|
||||
|
||||
token = login(args.wiki_url, identities, env["wiki_password"])
|
||||
with psycopg.connect(**conninfo, row_factory=dict_row) as conn:
|
||||
with conn.cursor() as cur:
|
||||
folder = resolve_folder(cur, args.folder_slug, args.parent_id)
|
||||
before = existing_assets(cur, folder["id"])
|
||||
print(f"folder_id={folder['id']} slug={folder['slug']} existing_assets={len(before)}")
|
||||
|
||||
uploaded = 0
|
||||
skipped = 0
|
||||
for file_path in files:
|
||||
upload_name = sanitize_upload_name(args.upload_name or file_path.name)
|
||||
if not upload_name:
|
||||
raise WikiUpdateError(f"Sanitized filename is empty: {file_path.name}")
|
||||
if upload_name in before:
|
||||
if args.skip_existing:
|
||||
print(f"asset_skip_existing source={file_path.name} upload_name={upload_name}")
|
||||
skipped += 1
|
||||
continue
|
||||
raise WikiUpdateError(f"Asset already exists in folder {folder['slug']}: {upload_name}")
|
||||
status, text = upload_file(args.wiki_url, token, folder["id"], file_path, upload_name)
|
||||
print(f"asset_upload source={file_path.name} upload_name={upload_name} status={status} response={response_preview(text)}")
|
||||
if status < 200 or status >= 300 or text.strip().lower() != "ok":
|
||||
raise WikiUpdateError(f"Upload failed for {file_path.name}: HTTP {status}")
|
||||
uploaded += 1
|
||||
|
||||
verified = []
|
||||
for file_path in files:
|
||||
upload_name = sanitize_upload_name(args.upload_name or file_path.name)
|
||||
row = verify_asset(cur, folder["id"], upload_name)
|
||||
if row:
|
||||
verified.append(upload_name)
|
||||
public_url = f"/{folder['slug']}/{row['filename']}"
|
||||
print(f"asset_db_verified filename={row['filename']} size={row['fileSize']} url={public_url}")
|
||||
if not args.skip_http_verify:
|
||||
status, content_type, byte_count, error_preview = verify_served_asset(
|
||||
args.wiki_url,
|
||||
folder["slug"],
|
||||
row["filename"],
|
||||
)
|
||||
print(
|
||||
"asset_http_verified filename={filename} status={status} content_type={content_type} bytes_read={bytes_read}".format(
|
||||
filename=row["filename"],
|
||||
status=status,
|
||||
content_type=content_type,
|
||||
bytes_read=byte_count,
|
||||
)
|
||||
)
|
||||
if status != 200 or not content_type.startswith("image/") or byte_count == 0:
|
||||
detail = f" response={error_preview}" if error_preview else ""
|
||||
raise WikiUpdateError(f"Asset URL is not serving an image: {public_url} HTTP {status}{detail}")
|
||||
elif not args.skip_existing:
|
||||
raise WikiUpdateError(f"Uploaded asset not found in database: {upload_name}")
|
||||
print(f"asset_upload_summary uploaded={uploaded} skipped={skipped} verified={len(verified)}")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
try:
|
||||
sys.exit(main())
|
||||
except WikiUpdateError as exc:
|
||||
print(f"wiki_asset_upload_error={exc}")
|
||||
sys.exit(2)
|
||||
@@ -0,0 +1,97 @@
|
||||
import json
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from urllib.error import HTTPError, URLError
|
||||
from urllib.request import Request, urlopen
|
||||
|
||||
import psycopg
|
||||
from psycopg.rows import dict_row
|
||||
|
||||
from wiki_update_page import DEFAULT_WIKI_URL, connect_info, load_env
|
||||
|
||||
|
||||
EXPECTED_REMOVED = [
|
||||
"/vulture/echo_mk2_cross_section.png",
|
||||
"/vulture/enterprise_takeoff_side_view.png",
|
||||
]
|
||||
|
||||
EXPECTED_PRESENT = [
|
||||
"/echo/echo_mk2_cross_section.png",
|
||||
"/enterprise/enterprise_takeoff_side_view.png",
|
||||
"/vulture/vulture_shuttle_full_stack_original.png",
|
||||
]
|
||||
|
||||
|
||||
def normalize_json(value):
|
||||
if value is None or isinstance(value, (dict, list)):
|
||||
return value
|
||||
return json.loads(value)
|
||||
|
||||
|
||||
def verify_database(conninfo):
|
||||
with psycopg.connect(**conninfo, row_factory=dict_row) as conn:
|
||||
with conn.cursor() as cur:
|
||||
cur.execute(
|
||||
"""
|
||||
SELECT count(*) AS count
|
||||
FROM assets a
|
||||
JOIN "assetFolders" f ON f.id = a."folderId"
|
||||
WHERE f.slug = 'vulture'
|
||||
AND (left(a.filename, 5) = 'echo_' OR left(a.filename, 11) = 'enterprise_')
|
||||
"""
|
||||
)
|
||||
misplaced_count = cur.fetchone()["count"]
|
||||
print(f"vulture_misplaced_asset_count={misplaced_count}")
|
||||
|
||||
cur.execute("SELECT permissions FROM groups WHERE lower(name) = 'automation' LIMIT 1")
|
||||
group = cur.fetchone()
|
||||
permissions = normalize_json(group["permissions"]) if group else []
|
||||
has_manage_system = "manage:system" in (permissions or [])
|
||||
print(f"automation_has_manage_system={has_manage_system}")
|
||||
|
||||
return misplaced_count == 0 and not has_manage_system
|
||||
|
||||
|
||||
def request_image(base_url, path):
|
||||
try:
|
||||
request = Request(base_url + path, headers={"User-Agent": "ksp-wiki-asset-cleanup-check"})
|
||||
with urlopen(request, timeout=30) as response:
|
||||
content_type = response.headers.get("Content-Type", "")
|
||||
body_ok = bool(response.read(1))
|
||||
return response.status, content_type, response.status == 200 and content_type.lower().startswith("image/") and body_ok
|
||||
except HTTPError as exc:
|
||||
return exc.code, exc.headers.get("Content-Type", ""), False
|
||||
except URLError as exc:
|
||||
return f"urlerror:{exc.reason}", "", False
|
||||
|
||||
|
||||
def verify_http(wiki_url):
|
||||
ok = True
|
||||
base_url = wiki_url.rstrip("/")
|
||||
for path in EXPECTED_REMOVED:
|
||||
status, content_type, is_image = request_image(base_url, path)
|
||||
path_ok = not is_image
|
||||
ok = ok and path_ok
|
||||
print(f"asset_get path={path} status={status} content_type={content_type} ok={path_ok}")
|
||||
for path in EXPECTED_PRESENT:
|
||||
status, content_type, is_image = request_image(base_url, path)
|
||||
ok = ok and is_image
|
||||
print(f"asset_get path={path} status={status} content_type={content_type} ok={is_image}")
|
||||
return ok
|
||||
|
||||
|
||||
def main():
|
||||
repo_root = Path.cwd()
|
||||
env = load_env((repo_root / ".env").resolve())
|
||||
conninfo = connect_info(env)
|
||||
db_ok = verify_database(conninfo)
|
||||
http_ok = verify_http(DEFAULT_WIKI_URL)
|
||||
if db_ok and http_ok:
|
||||
print("asset_cleanup_validation=ok")
|
||||
return 0
|
||||
print("asset_cleanup_validation=failed")
|
||||
return 1
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
@@ -0,0 +1,64 @@
|
||||
import sys
|
||||
|
||||
from playwright.sync_api import sync_playwright
|
||||
|
||||
|
||||
BASE_URL = "http://192.168.195.241:38353"
|
||||
PAGES = [
|
||||
("zh Vulture", f"{BASE_URL}/zh/home/Space_Shuttles/Vulture_Shuttle"),
|
||||
("en Vulture", f"{BASE_URL}/en/home/Space_Shuttles/Vulture_Shuttle"),
|
||||
("zh Echo", f"{BASE_URL}/zh/home/Space_Shuttles/Echo_Shuttle"),
|
||||
("en Echo", f"{BASE_URL}/en/home/Space_Shuttles/Echo_Shuttle"),
|
||||
("zh Enterprise", f"{BASE_URL}/zh/home/Space_Shuttles/Enterprise_Shuttle"),
|
||||
("en Enterprise", f"{BASE_URL}/en/home/Space_Shuttles/Enterprise_Shuttle"),
|
||||
]
|
||||
|
||||
|
||||
def main():
|
||||
errors = []
|
||||
print("lightbox_validation_start", flush=True)
|
||||
with sync_playwright() as playwright:
|
||||
print("launching_chromium", flush=True)
|
||||
browser = playwright.chromium.launch(headless=True, timeout=30000)
|
||||
for label, url in PAGES:
|
||||
print(f"checking={label}", flush=True)
|
||||
page = browser.new_page(viewport={"width": 1366, "height": 900})
|
||||
page.goto(url, wait_until="commit", timeout=30000)
|
||||
page.wait_for_timeout(1500)
|
||||
article = page.locator(".wiki-article, .vulture-article").first
|
||||
image = page.locator(".wiki-article img[data-wiki-asset], .vulture-article img[data-wiki-asset]").first
|
||||
image.wait_for(state="visible", timeout=20000)
|
||||
ready = article.get_attribute("data-lightbox-ready")
|
||||
print(f"{label}: ready={ready}", flush=True)
|
||||
if ready != "1":
|
||||
errors.append(f"{label}: lightbox script not installed, ready={ready!r}")
|
||||
page.close()
|
||||
continue
|
||||
image.click(timeout=10000)
|
||||
overlay = page.locator(".wiki-image-lightbox:not([hidden])")
|
||||
try:
|
||||
overlay.wait_for(state="visible", timeout=5000)
|
||||
preview_src = overlay.locator("img").first.get_attribute("src") or ""
|
||||
caption = overlay.locator("figcaption").first.inner_text(timeout=5000).strip()
|
||||
if not preview_src:
|
||||
errors.append(f"{label}: empty preview src")
|
||||
if not caption:
|
||||
errors.append(f"{label}: empty caption")
|
||||
print(f"{label}: lightbox_open=True caption_len={len(caption)} preview_src={preview_src[:96]}", flush=True)
|
||||
page.keyboard.press("Escape")
|
||||
page.wait_for_function("!document.querySelector('.wiki-image-lightbox:not([hidden])')", timeout=5000)
|
||||
except Exception as exc:
|
||||
errors.append(f"{label}: {type(exc).__name__}: {exc}")
|
||||
page.close()
|
||||
browser.close()
|
||||
if errors:
|
||||
print("ERRORS:")
|
||||
for error in errors:
|
||||
print(error)
|
||||
return 1
|
||||
print("lightbox_click_validation=ok")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user