236 lines
9.8 KiB
Python
236 lines
9.8 KiB
Python
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) |