diff --git a/.env.example b/.env.example index beb2092..dfabb94 100644 --- a/.env.example +++ b/.env.example @@ -6,6 +6,7 @@ PGPORT=5432 PGUSER=example_user PGPASSWORD=example_password PGDATABASE=example_db +WIKIDATABASE=example_wiki_db # 额外示例变量： APP_ENV=development @@ -14,4 +15,10 @@ APP_SECRET_KEY=change-me WORKBOOK_PATH=KSP Engine Tweak Chart.xlsx SQLALCHEMY_ECHO=false +# 登录到wiki的账号信息 +wiki_useremail=example@email.com +wiki_username=example_user +wiki_password=example_password + + # 说明：此文件为示例，不包含真实凭据。将真实凭据写入本地的 .env 文件并确保其被 .gitignore 忽略。 diff --git a/.github/skills/wiki-backend-update/SKILL.md b/.github/skills/wiki-backend-update/SKILL.md new file mode 100644 index 0000000..977f708 --- /dev/null +++ b/.github/skills/wiki-backend-update/SKILL.md @@ -0,0 +1,80 @@ +--- +name: wiki-backend-update +description: "Use when: updating Wiki.js pages from local Markdown/HTML, writing Wiki.js pages through backend/database workflows, fixing stale Wiki.js frontend cache after direct DB edits, refreshing pages.update/pages.render/pages.flushCache, or publishing KSP wiki articles. Requires scripts/wiki_update_page.py and final frontend verification." +argument-hint: "path= content-file= title=" +--- + +# Wiki Backend Update + +## 使用场景 + +当任务涉及 Wiki.js 页面后台更新、直接写库后的页面刷新、把本地 Markdown/HTML 发布到 Wiki.js、或处理“编辑/历史记录已更新但前台仍显示旧内容”的问题时，必须使用此流程。 + +## 必走流程 + +1. 先确认本地内容文件是最终版本，并运行适合该文件的基本检查。 +2. 使用 `.env` 读取 Wiki.js PostgreSQL 与登录信息；不要打印密码、JWT 或完整账号信息。 +3. 不要只改 `pages.content/render/toc` 就结束。直接 DB 更新后必须走官方 GraphQL 刷新链路。 +4. 确认登录账号所在组的命中 Page Rule 包含 `write:pages` 和 `manage:pages`。组级权限有这两个值但 Page Rule 没有时，`pages.update` 仍会返回 `PageUpdateForbidden`。 +5. 调用官方 `pages.update` 时必须传 `tags`，即使页面没有标签也传 `tags: []`，否则 Wiki.js resolver 可能报 `Cannot read properties of undefined (reading 'map')`。 +6. `pages.render` 与 `pages.flushCache` 需要 `manage:system`。使用前必须 ask user；若获准，临时授予，完成后立即移除。 +7. 前台验证必须检查普通页面 URL 已包含新内容且不包含旧占位文本。若工具返回疑似缓存结果，用终端无缓存抓取复核。 +8. 上传图片等资产后，必须直接验证资源 URL 返回 200 和正确 Content-Type；不要只依赖 `assets` / `assetData` 数据库记录。 +9. 清理所有 `_tmp_` 临时脚本或输出文件。 +10. 任务收尾必须使用 ask user，确认页面状态或下一步。 + +## 固化脚本 + +使用仓库脚本 [scripts/wiki_update_page.py](../../../scripts/wiki_update_page.py)。它会： + +- 读取 `.env` 中的 `PGHOST`、`PGUSER`、`PGPASSWORD`、`WIKIDATABASE`/`PGDATABASE` 和 Wiki 登录信息。 +- 解析目标页面并保留现有 tags。 +- 检查命中 Page Rule 是否具备 `write:pages` / `manage:pages`。 +- 通过官方 GraphQL 执行 `pages.update`、`pages.render`、`pages.flushCache`。 +- 对 HTML 页面使用 `--extract-style-to-script-css`，把本地 `<style>` 发布到 Wiki.js 页面 CSS 字段，避免 Wiki.js 过滤正文中的样式。 +- 如果 HTML 页面包含交互脚本，使用 `--extract-script-to-script-js`，把第一个 `<script>` 发布到 Wiki.js 页面 JavaScript 字段，避免内联脚本留在正文中。 +- 可在用户授权后临时添加 `manage:system`，并在 `finally` 中移除。 +- 抓取前台页面并验证 `--must-contain` / `--must-not-contain`。 + +上传 Wiki.js 资产使用 [scripts/wiki_upload_assets.py](../../../scripts/wiki_upload_assets.py)。它会通过 `/u` 上传端点写入既有资源文件夹，并验证 `assets` / `assetData` 记录。上传后仍需单独抓取资源 URL；当前实例存在部分 DB-backed assets 直连 500 的情况，页面交付前必须确认图片在前台实际可见。若资源端未修好，可临时使用压缩图片 `data:` URI，并保留 `data-wiki-asset` 标记以便之后切回正式资源 URL。 + +## 示例命令 + +```powershell +python scripts\wiki_update_page.py ` + --path home/Space_Shuttles/Vulture_Shuttle/html_test ` + --content-file vulture_shuttle_html_test.html ` + --title "HTML version" ` + --editor code ` + --extract-style-to-script-css ` + --extract-script-to-script-js ` + --temporary-manage-system ` + --must-contain "秃鹫航天飞机" ` + --must-not-contain "# Title" ` + --must-not-contain "Some text here" +``` + +如果命中 Page Rule 缺少写权限，先 ask user，再使用： + +```powershell +python scripts\wiki_update_page.py ` + --path home/Space_Shuttles/Vulture_Shuttle/html_test ` + --content-file vulture_shuttle_html_test.html ` + --title "HTML version" ` + --editor code ` + --extract-style-to-script-css ` + --extract-script-to-script-js ` + --patch-page-rule ` + --temporary-manage-system ` + --must-contain "秃鹫航天飞机" ` + --must-not-contain "# Title" ` + --must-not-contain "Some text here" +``` + +## 失败判断 + +- `PageUpdateForbidden`: 命中 Page Rule 仍缺少 `write:pages` 或路径规则没匹配。 +- `Cannot read properties of undefined (reading 'map')`: `pages.update` 没有传 `tags`。 +- `pages.render` 或 `pages.flushCache` 为 `Forbidden`: 缺少 `manage:system`，或权限缓存未刷新。 +- 上传资产后数据库记录存在但 URL 返回 500：资源服务/cache 层异常，DB 验证不够；先直接验证图片 URL，必要时临时使用 `data:` URI 保证页面可见。 +- 编辑页/历史记录正确但前台旧内容：必须执行 render/flushCache，必要时重启 Wiki.js 容器。 \ No newline at end of file diff --git a/README.md b/README.md index cfd26e1..65084a9 100644 --- a/README.md +++ b/README.md @@ -97,6 +97,32 @@ docker compose up --build - 根目录已经补上 .gitignore，避免后续再次提交 .env - 如果 .env 已经在 Git 索引里，.gitignore 不会自动取消跟踪，需要你本地手动把它移出索引后再提交 +## Wiki.js 资产与页面发布记录 + +Wiki.js 页面发布使用 `scripts/wiki_update_page.py`，图片上传使用 `scripts/wiki_upload_assets.py`。发布 HTML 页面时，页面内 `<style>` 应通过 `--extract-style-to-script-css` 写入 Wiki.js 页面 CSS；页面内 `<script>` 应通过 `--extract-script-to-script-js` 写入页面 JS。Wiki.js 会把 `scriptJs` 原样注入页面 body，因此发布脚本会保留完整 `<script>...</script>` 包裹。 + +曾出现所有图片 URL 返回 `500 Internal Server Error`，但数据库 `assets` / `assetData` 中记录和二进制数据都存在的情况。根因是 Wiki.js 的 storage target `disk` 被启用，但 `config.path` 为空；Wiki.js 的 Local File System storage 要求绝对路径，空路径会导致资产服务读取/缓存文件路径异常。修复方式是通过 Wiki.js GraphQL `storage.updateTargets` 将 `disk` 的路径设为绝对目录 `/wiki/data/storage`，随后执行 `storage.executeAction(targetKey: "disk", handler: "dump")`，并在操作前后临时授予/移除 `manage:system`。 + +验证图片时不要只用 `HEAD` 或只查数据库。应实际 `GET` 资产 URL，并确认 HTTP 200、`Content-Type` 为 `image/*`。上传脚本现在默认会做真实 GET 验证，例如： + +```bash +python scripts/wiki_upload_assets.py --source-dir data/KSP/wiki-upload-echo-current --file echo_mk2_cross_section.png --folder-slug echo --skip-existing +``` + +如果 Echo / Enterprise 图片误传到 `/vulture`，先 dry-run 查看目标，再通过官方 `assets.deleteAsset` 删除错放资源。清理脚本只匹配 `/vulture` 下文件名以 `echo_` 或 `enterprise_` 开头的资产，并在操作后移除临时系统权限： + +```bash +py scripts/wiki_cleanup_misplaced_assets.py +py scripts/wiki_cleanup_misplaced_assets.py --delete +py scripts/wiki_verify_asset_cleanup.py +``` + +六个航天飞机页面的图片点击放大由 `scripts/wiki_apply_image_lightbox.py` 统一注入，发布后可用以下脚本做浏览器级验证： + +```bash +py scripts/wiki_verify_lightbox.py +``` + ## 下一步建议 1. 把 Communication、Tank、Vehicle Cost 的页面也补成 CRUD diff --git a/scripts/wiki_apply_image_lightbox.py b/scripts/wiki_apply_image_lightbox.py new file mode 100644 index 0000000..a1aa0d6 --- /dev/null +++ b/scripts/wiki_apply_image_lightbox.py @@ -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()) \ No newline at end of file diff --git a/scripts/wiki_cleanup_misplaced_assets.py b/scripts/wiki_cleanup_misplaced_assets.py new file mode 100644 index 0000000..566fd73 --- /dev/null +++ b/scripts/wiki_cleanup_misplaced_assets.py @@ -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) \ No newline at end of file diff --git a/scripts/wiki_embed_image_fallbacks.py b/scripts/wiki_embed_image_fallbacks.py new file mode 100644 index 0000000..76b66bd --- /dev/null +++ b/scripts/wiki_embed_image_fallbacks.py @@ -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()) diff --git a/scripts/wiki_update_page.py b/scripts/wiki_update_page.py new file mode 100644 index 0000000..e1b087b --- /dev/null +++ b/scripts/wiki_update_page.py @@ -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) \ No newline at end of file diff --git a/scripts/wiki_upload_assets.py b/scripts/wiki_upload_assets.py new file mode 100644 index 0000000..c90aaa2 --- /dev/null +++ b/scripts/wiki_upload_assets.py @@ -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) \ No newline at end of file diff --git a/scripts/wiki_verify_asset_cleanup.py b/scripts/wiki_verify_asset_cleanup.py new file mode 100644 index 0000000..222aa4c --- /dev/null +++ b/scripts/wiki_verify_asset_cleanup.py @@ -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()) \ No newline at end of file diff --git a/scripts/wiki_verify_lightbox.py b/scripts/wiki_verify_lightbox.py new file mode 100644 index 0000000..4a2c702 --- /dev/null +++ b/scripts/wiki_verify_lightbox.py @@ -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()) \ No newline at end of file