From 26987fc497fcaf9caeba3e8b5d1062f3b9828495 Mon Sep 17 00:00:00 2001 From: rekushi <1+rekushi@noreply.flipnote.world> Date: Thu, 29 Oct 2026 04:23:46 +0000 Subject: [PATCH] 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 | 95 +++++++++++++++++++++++++++++++++++++++++++++++++---- 1 file changed, 89 insertions(+), 6 deletions(-) diff --git a/reximemo.py b/reximemo.py index bff8709..8600553 100644 --- a/reximemo.py +++ b/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("&", "&").replace("<", "<").replace(">", ">") + if preserve_newlines: + text = text.replace("\r\n", "\n").replace("\r", "\n").replace("\n", "
") + return text + + def html_page(request, title, body, extra_head=""): request.setHeader(b"content-type", b"text/html; charset=utf-8") - page = f"""{html.escape(str(title))} + page = f"""{dsi_text_html(title)} @@ -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('Previous' % 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('Next' % html.escape(dsi_url(request, base) + "?" + urlencode(query), quote=True)) + return '
' + " | ".join(links) + "
" + + 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):