428 lines
15 KiB
Python
Executable File
428 lines
15 KiB
Python
Executable File
"""Editable page, upload, and diagnostics operations."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import posixpath
|
|
import re
|
|
import struct
|
|
import sys
|
|
from datetime import datetime
|
|
from email.parser import BytesParser
|
|
from email.policy import default as email_default_policy
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from .config import (
|
|
ALLOWED_UPLOAD_EXTENSIONS,
|
|
BLOGS_DIR,
|
|
EXCLUDED_CONTENT_DIR_NAMES,
|
|
GENERATED_CONTENT_NAMES,
|
|
HZONE_ASSETS_DIR,
|
|
IMAGE_ASSETS_DIR,
|
|
LIMA_DIR,
|
|
MONTH_NAMES,
|
|
POSTS_DIR,
|
|
ROOT,
|
|
)
|
|
from .models import ContentPage
|
|
from .utils import html_escape, normalise_tags, org_date, parse_org_datetime, slugify
|
|
|
|
def safe_relative_path(path: str) -> Path:
|
|
rel = Path(path)
|
|
if rel.is_absolute() or ".." in rel.parts:
|
|
raise ValueError("Path must stay inside this repository.")
|
|
full = (ROOT / rel).resolve()
|
|
if not full.is_relative_to(ROOT):
|
|
raise ValueError("Path must stay inside this repository.")
|
|
if full.name in GENERATED_CONTENT_NAMES or "sync-conflict" in full.name:
|
|
raise ValueError("Generated and sync-conflict files are not editable here.")
|
|
rel_parts = full.relative_to(ROOT).parts
|
|
if any(part in EXCLUDED_CONTENT_DIR_NAMES for part in rel_parts):
|
|
raise ValueError("This path is outside the editable content folders.")
|
|
if full.suffix == ".org":
|
|
return full
|
|
if full.suffix == ".md" and full.is_relative_to(LIMA_DIR):
|
|
return full
|
|
raise ValueError("Only .org content files and .md files under lima can be edited here.")
|
|
|
|
|
|
def safe_target_path(path: str, slug: str, page_type: str) -> Path:
|
|
candidate = path.strip()
|
|
if not candidate:
|
|
raise ValueError("Path is required.")
|
|
default_ext = ".md" if page_type == "lima" else ".org"
|
|
if candidate.endswith("/"):
|
|
candidate = f"{candidate}{slug}{default_ext}"
|
|
elif not Path(candidate).suffix:
|
|
candidate = f"{candidate}{default_ext}"
|
|
return safe_relative_path(candidate)
|
|
|
|
|
|
def markdown_title(content: str, fallback: str) -> str:
|
|
for line in content.splitlines():
|
|
match = re.match(r"^#{1,6}\s+(.+?)\s*$", line)
|
|
if match:
|
|
return match.group(1).strip()
|
|
return fallback.replace("-", " ").replace("_", " ").title()
|
|
|
|
|
|
def read_markdown_page(path: Path) -> ContentPage:
|
|
content = path.read_text(encoding="utf-8")
|
|
rel = path.relative_to(ROOT).as_posix()
|
|
title = markdown_title(content, path.stem)
|
|
return ContentPage(
|
|
path=rel,
|
|
page_type="lima",
|
|
title=title,
|
|
slug=path.stem,
|
|
tags=[],
|
|
content=content,
|
|
date="",
|
|
comments=True,
|
|
options="",
|
|
format="markdown",
|
|
)
|
|
|
|
|
|
def read_page(path: Path) -> ContentPage:
|
|
if path.suffix == ".md" and path.is_relative_to(LIMA_DIR):
|
|
return read_markdown_page(path)
|
|
text = path.read_text(encoding="utf-8")
|
|
meta: dict[str, str] = {}
|
|
body_lines: list[str] = []
|
|
in_header = True
|
|
for line in text.splitlines():
|
|
if in_header and line.startswith("#+"):
|
|
key, _, value = line[2:].partition(":")
|
|
meta[key.strip().upper()] = value.strip()
|
|
else:
|
|
in_header = False
|
|
body_lines.append(line)
|
|
rel = path.relative_to(ROOT).as_posix()
|
|
if path.is_relative_to(BLOGS_DIR):
|
|
page_type = "blog"
|
|
elif path.is_relative_to(POSTS_DIR):
|
|
page_type = "post"
|
|
else:
|
|
page_type = "page"
|
|
slug = meta.get("SLUG") or path.stem
|
|
tags = normalise_tags(meta.get("FILETAGS", ""))
|
|
return ContentPage(
|
|
path=rel,
|
|
page_type=page_type,
|
|
title=meta.get("TITLE", path.stem),
|
|
slug=slug,
|
|
tags=tags,
|
|
content="\n".join(body_lines).lstrip("\n"),
|
|
date=meta.get("DATE", org_date(datetime.fromtimestamp(path.stat().st_mtime))),
|
|
comments=meta.get("COMMENTS", "t").lower() == "t",
|
|
options=meta.get("OPTIONS", "num:nil"),
|
|
format="org",
|
|
wip=meta.get("WIP"),
|
|
)
|
|
|
|
|
|
def page_to_dict(page: ContentPage) -> dict[str, Any]:
|
|
return {
|
|
"path": page.path,
|
|
"pageType": page.page_type,
|
|
"title": page.title,
|
|
"slug": page.slug,
|
|
"tags": page.tags,
|
|
"content": page.content,
|
|
"date": page.date,
|
|
"comments": page.comments,
|
|
"options": page.options,
|
|
"format": page.format,
|
|
"wip": page.wip or "",
|
|
}
|
|
|
|
|
|
def server_diagnostics() -> dict[str, Any]:
|
|
pages = list_pages()
|
|
try:
|
|
cwd = Path.cwd().as_posix()
|
|
except OSError as exc:
|
|
cwd = f"<unavailable: {exc}>"
|
|
return {
|
|
"root": ROOT.as_posix(),
|
|
"cwd": cwd,
|
|
"executable": sys.executable,
|
|
"pid": os.getpid(),
|
|
"pageCount": len(pages),
|
|
"firstPage": pages[0]["path"] if pages else "",
|
|
}
|
|
|
|
|
|
def list_pages() -> list[dict[str, Any]]:
|
|
pages = []
|
|
org_paths = []
|
|
if ROOT.exists():
|
|
try:
|
|
for path in ROOT.rglob("*.org"):
|
|
try:
|
|
rel_parts = path.relative_to(ROOT).parts
|
|
except ValueError:
|
|
continue
|
|
if any(part in EXCLUDED_CONTENT_DIR_NAMES for part in rel_parts):
|
|
continue
|
|
org_paths.append(path)
|
|
except OSError:
|
|
org_paths = []
|
|
try:
|
|
md_paths = list(LIMA_DIR.rglob("*.md")) if LIMA_DIR.exists() else []
|
|
except OSError:
|
|
md_paths = []
|
|
for path in sorted(org_paths + md_paths):
|
|
if path.name in GENERATED_CONTENT_NAMES or "sync-conflict" in path.name:
|
|
continue
|
|
try:
|
|
page = read_page(path)
|
|
parsed = parse_org_datetime(page.date)
|
|
timestamp = parsed.timestamp() if parsed else path.stat().st_mtime
|
|
except (OSError, UnicodeDecodeError, ValueError):
|
|
continue
|
|
pages.append(
|
|
{
|
|
"path": page.path,
|
|
"pageType": page.page_type,
|
|
"title": page.title,
|
|
"slug": page.slug,
|
|
"tags": page.tags,
|
|
"date": page.date,
|
|
"format": page.format,
|
|
"timestamp": timestamp,
|
|
}
|
|
)
|
|
return sorted(pages, key=lambda item: item["timestamp"], reverse=True)
|
|
|
|
|
|
def target_path(data: dict[str, Any], existing_path: str | None) -> Path:
|
|
if existing_path:
|
|
return safe_relative_path(existing_path)
|
|
title = str(data.get("title") or "").strip()
|
|
slug = slugify(str(data.get("slug") or title))
|
|
page_type = str(data.get("pageType") or "blog")
|
|
explicit_path = str(data.get("targetPath") or "").strip()
|
|
if explicit_path:
|
|
return safe_target_path(explicit_path, slug, page_type)
|
|
if page_type == "blog":
|
|
dt = parse_org_datetime(str(data.get("date") or "")) or datetime.now()
|
|
folder = BLOGS_DIR / str(dt.year) / f"{dt.month:02d}-{MONTH_NAMES[dt.month - 1]}"
|
|
return folder / f"{slug}.org"
|
|
if page_type == "post":
|
|
raw_section = str(data.get("section") or "").strip()
|
|
section = slugify(raw_section) if raw_section else ""
|
|
folder = POSTS_DIR / section if section else POSTS_DIR
|
|
return folder / f"{slug}.org"
|
|
if page_type == "lima":
|
|
return LIMA_DIR / f"{slug}.md"
|
|
if page_type == "page":
|
|
return ROOT / f"{slug}.org"
|
|
raise ValueError("pageType must be blog, post, page, or lima.")
|
|
|
|
|
|
def render_markdown(data: dict[str, Any]) -> str:
|
|
content = str(data.get("content") or "").replace("\r\n", "\n").strip()
|
|
title = str(data.get("title") or "").strip()
|
|
if not title:
|
|
raise ValueError("Title is required.")
|
|
content = re.sub(
|
|
r'<a\b[^>]*>\s*<img\b[^>]*\bsrc="([^"]+)"[^>]*\balt="([^"]*)"[^>]*>\s*</a>',
|
|
lambda match: f"})",
|
|
content,
|
|
flags=re.IGNORECASE,
|
|
)
|
|
if content:
|
|
if re.search(r"^#{1,6}[^\S\r\n]+.+?[^\S\r\n]*$", content, flags=re.MULTILINE):
|
|
content = re.sub(
|
|
r"^#{1,6}[^\S\r\n]+.+?[^\S\r\n]*$",
|
|
f"# {title}",
|
|
content,
|
|
count=1,
|
|
flags=re.MULTILINE,
|
|
)
|
|
else:
|
|
content = f"# {title}\n\n{content}"
|
|
return content + "\n"
|
|
return f"# {title}\n"
|
|
|
|
|
|
def render_org(data: dict[str, Any], previous: ContentPage | None) -> str:
|
|
title = str(data.get("title") or "").strip()
|
|
if not title:
|
|
raise ValueError("Title is required.")
|
|
slug = slugify(str(data.get("slug") or title))
|
|
tags = normalise_tags(data.get("tags", []))
|
|
content = str(data.get("content") or "").replace("\r\n", "\n").strip()
|
|
date = str(data.get("date") or "").strip()
|
|
if not parse_org_datetime(date):
|
|
date = previous.date if previous else org_date(datetime.now())
|
|
options = str(data.get("options") or (previous.options if previous else "num:nil")).strip()
|
|
comments = bool(data.get("comments", True))
|
|
lines = [
|
|
f"#+TITLE: {title}",
|
|
f"#+OPTIONS: {options}",
|
|
f"#+DATE: {date}",
|
|
f"#+filetags: {''.join(f':{tag}' for tag in tags)}:",
|
|
]
|
|
wip = str(data.get("wip") or (previous.wip if previous else "") or "").strip()
|
|
if wip:
|
|
lines.append(f"#+WIP: {wip}")
|
|
lines.extend(
|
|
[
|
|
f"#+COMMENTS: {'t' if comments else ''}",
|
|
f"#+SLUG: {slug}",
|
|
"",
|
|
content,
|
|
"",
|
|
]
|
|
)
|
|
return "\n".join(lines)
|
|
|
|
|
|
def save_page(data: dict[str, Any]) -> dict[str, Any]:
|
|
existing_path = data.get("path") or None
|
|
target = target_path(data, str(existing_path) if existing_path else None)
|
|
previous = read_page(target) if target.exists() else None
|
|
if not target.parent.exists():
|
|
target.parent.mkdir(parents=True)
|
|
if target.exists() and not existing_path:
|
|
raise ValueError(f"{target.relative_to(ROOT)} already exists.")
|
|
if target.suffix == ".md":
|
|
target.write_text(render_markdown(data), encoding="utf-8")
|
|
else:
|
|
target.write_text(render_org(data, previous), encoding="utf-8")
|
|
return page_to_dict(read_page(target))
|
|
|
|
|
|
def image_dimensions(payload: bytes, ext: str) -> tuple[int, int] | None:
|
|
if ext == ".png" and payload.startswith(b"\x89PNG\r\n\x1a\n") and len(payload) >= 24:
|
|
width, height = struct.unpack(">II", payload[16:24])
|
|
return width, height
|
|
if ext == ".gif" and payload[:6] in {b"GIF87a", b"GIF89a"} and len(payload) >= 10:
|
|
width, height = struct.unpack("<HH", payload[6:10])
|
|
return width, height
|
|
if ext in {".jpg", ".jpeg"} and payload.startswith(b"\xff\xd8"):
|
|
i = 2
|
|
while i + 9 < len(payload):
|
|
if payload[i] != 0xFF:
|
|
i += 1
|
|
continue
|
|
marker = payload[i + 1]
|
|
i += 2
|
|
if marker in {0xD8, 0xD9}:
|
|
continue
|
|
if i + 2 > len(payload):
|
|
break
|
|
size = int.from_bytes(payload[i:i + 2], "big")
|
|
if size < 2:
|
|
break
|
|
if marker in {0xC0, 0xC1, 0xC2, 0xC3, 0xC5, 0xC6, 0xC7, 0xC9, 0xCA, 0xCB, 0xCD, 0xCE, 0xCF}:
|
|
if i + 7 <= len(payload):
|
|
height = int.from_bytes(payload[i + 3:i + 5], "big")
|
|
width = int.from_bytes(payload[i + 5:i + 7], "big")
|
|
return width, height
|
|
break
|
|
i += size
|
|
return None
|
|
|
|
|
|
def relative_asset_path(page_path: str, asset_path: str) -> str:
|
|
page = safe_relative_path(page_path) if page_path else ROOT / "index.org"
|
|
page_rel = page.relative_to(ROOT).as_posix()
|
|
page_output_dir = posixpath.dirname(page_rel)
|
|
return posixpath.relpath(asset_path, page_output_dir or ".")
|
|
|
|
|
|
def gallery_image_html(filename: str, asset_path: str, page_path: str, payload: bytes, ext: str) -> str:
|
|
absolute_url = f"https://zainezq.com/{asset_path}"
|
|
relative_url = relative_asset_path(page_path, asset_path)
|
|
dims = image_dimensions(payload, ext)
|
|
width, height = dims if dims else (1920, 1080)
|
|
alt = html_escape(filename)
|
|
return (
|
|
f'<a href="{absolute_url}" data-img="{absolute_url}" data-alt="{alt}" '
|
|
f'data-width="{width}" data-height="{height}">'
|
|
f'<img src="{relative_url}" alt="{alt}" style="cursor: zoom-in;"></a>'
|
|
)
|
|
|
|
|
|
def attachment_image_dir(page_path: str) -> Path:
|
|
page = safe_relative_path(page_path) if page_path else ROOT / "index.org"
|
|
if page.suffix == ".md" and page.is_relative_to(LIMA_DIR):
|
|
return HZONE_ASSETS_DIR
|
|
if page.is_relative_to(POSTS_DIR):
|
|
rel = page.relative_to(POSTS_DIR)
|
|
section = rel.parts[0] if len(rel.parts) > 1 else "posts"
|
|
return IMAGE_ASSETS_DIR / slugify(section)
|
|
if page.is_relative_to(BLOGS_DIR):
|
|
return IMAGE_ASSETS_DIR / "blogs"
|
|
rel = page.relative_to(ROOT)
|
|
if len(rel.parts) > 1:
|
|
return IMAGE_ASSETS_DIR / slugify(rel.parts[0])
|
|
return IMAGE_ASSETS_DIR / "pages"
|
|
|
|
|
|
def save_upload(filename: str, payload: bytes, page_path: str = "") -> dict[str, str]:
|
|
original = Path(filename or "attachment").name
|
|
ext = Path(original).suffix.lower()
|
|
if ext not in ALLOWED_UPLOAD_EXTENSIONS:
|
|
raise ValueError("Only common image files can be uploaded.")
|
|
now = datetime.now()
|
|
target_dir = attachment_image_dir(page_path)
|
|
target_dir.mkdir(parents=True, exist_ok=True)
|
|
stem = slugify(Path(original).stem)
|
|
prefix = "" if re.match(r"^\d{4}-\d{2}-\d{2}-", stem) else f"{now.strftime('%Y-%m-%d')}-"
|
|
target = target_dir / f"{prefix}{stem}{ext}"
|
|
counter = 2
|
|
while target.exists():
|
|
target = target_dir / f"{prefix}{stem}-{counter}{ext}"
|
|
counter += 1
|
|
target.write_bytes(payload)
|
|
rel = target.relative_to(ROOT).as_posix()
|
|
absolute_url = f"https://zainezq.com/{rel}"
|
|
relative_url = relative_asset_path(page_path, rel)
|
|
is_markdown = page_path.endswith(".md")
|
|
insert_text = f"" if is_markdown else f"[[{relative_url}]]"
|
|
return {
|
|
"url": absolute_url,
|
|
"relativeUrl": relative_url,
|
|
"path": rel,
|
|
"markdown": insert_text,
|
|
"insertText": insert_text,
|
|
"filename": target.name,
|
|
}
|
|
|
|
|
|
|
|
|
|
def parse_upload_form(content_type: str, body: bytes) -> tuple[str, bytes, str]:
|
|
if not content_type.lower().startswith("multipart/form-data"):
|
|
raise ValueError("Uploads must use multipart/form-data.")
|
|
message = BytesParser(policy=email_default_policy).parsebytes(
|
|
f"Content-Type: {content_type}\r\nMIME-Version: 1.0\r\n\r\n".encode("utf-8") + body
|
|
)
|
|
if not message.is_multipart():
|
|
raise ValueError("Upload form is not multipart.")
|
|
|
|
filename = ""
|
|
payload = b""
|
|
page_path = ""
|
|
for part in message.iter_parts():
|
|
name = part.get_param("name", header="content-disposition")
|
|
if name == "attachment":
|
|
filename = part.get_filename() or ""
|
|
payload = part.get_payload(decode=True) or b""
|
|
elif name == "pagePath":
|
|
raw_value = part.get_payload(decode=True) or b""
|
|
page_path = raw_value.decode(part.get_content_charset() or "utf-8", errors="replace")
|
|
|
|
if not filename:
|
|
raise ValueError("No attachment was uploaded.")
|
|
if not payload:
|
|
raise ValueError("Attachment is empty.")
|
|
return filename, payload, page_path
|
|
|