feat(dsi): add search and accurate thumbnail helpers

Add native search forwarding, page handling, posting limits and cached previews using the saved thumbnail frame.

reximemo.py

@@ -1,9 +1,12 @@
 from __future__ import annotations
 
 import html
+import hashlib
 import io
 import math
+from collections import OrderedDict
 from pathlib import Path
+from threading import RLock
 from urllib.parse import urlencode
 
 from PIL import Image
@@ -15,6 +18,8 @@ from reximemo_image_formats import encode_image
 
 ROOT_HOST = "flipnote.hatena.com"
 ROOT = Path(__file__).resolve().parent
+_THUMBNAILS = OrderedDict()
+_THUMBNAIL_LOCK = RLock()
 
 
 def region_from_request(request):
@@ -37,6 +42,23 @@ def current_account(request):
     return Database.account_for_ip(client_ip(request))
 
 
+def posting_account(request):
+    from server_config import load_config
+    account = current_account(request)
+    if account is None and not load_config()["guest_posting"]:
+        return None, dialog(request, "Sign in from Account before posting on this server.")
+    return account, None
+
+
+def read_ppm_upload(request):
+    from server_config import load_config
+    maximum = int(load_config()["max_upload_bytes"])
+    data = request.content.read(maximum + 1)
+    if len(data) > maximum:
+        raise ValueError("This Flipnote is larger than the server's upload limit.")
+    return data
+
+
 def current_creator(request, *, fsid="", display_name=""):
     account = current_account(request)
     return Database.get_or_create_creator(
@@ -51,6 +73,15 @@ def keyboard_value(request):
     return request_headers(request).get("x-email-addr", "").strip()
 
 
+def search_keyboard_response(request, kind="flipnotes"):
+    query = keyboard_value(request)[:100].strip()
+    if not query:
+        return dialog(request, "Enter a title or creator name to search for.")
+    scope = "creators" if kind == "creators" else "flipnotes"
+    forward(request, dsi_url(request, "search.uls?") + urlencode({"q": query, "kind": scope}))
+    return b"OK"
+
+
 def dialog(request, message, status=200):
     request.setResponseCode(int(status))
     request.setHeader(b"X-DSi-Dialog-Type", b"1")
@@ -62,9 +93,19 @@ def forward(request, target):
     request.setHeader(b"X-DSi-Forwarder", str(target).encode("ascii", "ignore"))
 
 
+def dsi_text_html(value, preserve_newlines=False):
+    """Escape DSi text nodes without unsupported quote entities."""
+    text = "" if value is None else str(value)
+    text = text.replace("\x00", "")
+    text = text.replace("&", "&#38;").replace("<", "&#60;").replace(">", "&#62;")
+    if preserve_newlines:
+        text = text.replace("\r\n", "\n").replace("\r", "\n").replace("\n", "<br>")
+    return text
+
+
 def html_page(request, title, body, extra_head=""):
     request.setHeader(b"content-type", b"text/html; charset=utf-8")
-    page = f"""<!doctype html><html><head><title>{html.escape(str(title))}</title>
+    page = f"""<!doctype html><html><head><title>{dsi_text_html(title)}</title>
 <meta name="uppertitle" content="{html.escape(str(title), quote=True)}">
 <link rel="stylesheet" type="text/css" href="http://{ROOT_HOST}/css/ds/common.css">
 <link rel="stylesheet" type="text/css" href="http://{ROOT_HOST}/css/ds/basic.css">
@@ -81,6 +122,26 @@ def query_int(request, name, default=1, minimum=1, maximum=1000000):
     return min(maximum, max(minimum, n))
 
 
+def page_number(request, total, per_page=50, name="page"):
+    return min(query_int(request, name), max(1, int(math.ceil(total / float(per_page)))))
+
+
+def pagination_html(request, base, page, total, per_page=30, params=None, name="page"):
+    pages = max(1, int(math.ceil(total / float(per_page))))
+    if pages == 1:
+        return ""
+    params = dict(params or {})
+    links = []
+    if page > 1:
+        query = dict(params, **{name: page - 1})
+        links.append('<a href="%s">Previous</a>' % html.escape(dsi_url(request, base) + "?" + urlencode(query), quote=True))
+    links.append("Page %d of %d" % (page, pages))
+    if page < pages:
+        query = dict(params, **{name: page + 1})
+        links.append('<a href="%s">Next</a>' % html.escape(dsi_url(request, base) + "?" + urlencode(query), quote=True))
+    return '<div align="center">' + " | ".join(links) + "</div>"
+
+
 def make_menu(layout=(4,), title=None, subtitle_top="", subtitle_bottom="", left="", right="", buttons=None, dropdowns=None, corner=None):
     ugo = UGO()
     ugo.Loaded = True
@@ -151,22 +212,44 @@ def pagination_buttons(request, base, page, total, per_page=50, params=None):
 
 
 def ppm_first_frame_image(raw):
+    return ppm_frame_image(raw, 0)
+
+
+def ppm_frame_image(raw, frame_index):
     ppm = PPM().Read(raw, ReadFrames=True, ReadSound=False)
     if not ppm:
         raise ValueError("Invalid PPM")
-    frame = ppm.GetFrame(0)
+    frame_index = int(frame_index)
+    if frame_index < 0 or frame_index >= int(ppm.FrameCount):
+        frame_index = 0
+    frame = ppm.GetFrame(frame_index)
     if frame is False or frame is None:
-        raise ValueError("Missing first frame")
+        raise ValueError("Missing frame")
     data = frame.tobytes(order="F")
     return Image.frombytes("RGBA", (len(frame), len(frame[0])), data)
 
 
-def ppm_thumbnail_png(raw, size=(256, 192)):
-    image = ppm_first_frame_image(raw).convert("RGB")
+def ppm_thumbnail_png(raw, size=(256, 192), *, first_frame=False):
+    raw = bytes(raw)
+    size = tuple(map(int, size))
+    key = (hashlib.sha256(raw).digest(), size, bool(first_frame))
+    with _THUMBNAIL_LOCK:
+        cached = _THUMBNAILS.get(key)
+        if cached is not None:
+            _THUMBNAILS.move_to_end(key)
+            return cached
+    frame_index = 0 if first_frame else int.from_bytes(raw[0x12:0x14], "little")
+    image = ppm_frame_image(raw, frame_index).convert("RGB")
     image.thumbnail(size, Image.Resampling.NEAREST)
     out = io.BytesIO()
     image.save(out, "PNG", optimize=True)
-    return out.getvalue()
+    result = out.getvalue()
+    with _THUMBNAIL_LOCK:
+        _THUMBNAILS[key] = result
+        _THUMBNAILS.move_to_end(key)
+        while len(_THUMBNAILS) > 64:
+            _THUMBNAILS.popitem(last=False)
+    return result
 
 
 def comment_npf(comment_id, width=128, height=64):