97 lines
3.2 KiB
Python
97 lines
3.2 KiB
Python
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()) |