from __future__ import annotations import html from urllib.parse import parse_qs from twisted.web import resource from DB import Database from hatena import client_ip, request_args from reximemo import ppm_thumbnail_png from server_config import load_config from web_ui import page_number, pagination, protect_forms, valid_csrf from web_features import WebFeatures, staff def _fields(request): if request.args and any(key in request.args for key in (b"csrf", "csrf")): return {(key.decode("utf-8", "replace") if isinstance(key, bytes) else key): (values[-1].decode("utf-8", "replace") if isinstance(values[-1], bytes) else values[-1]) for key, values in request.args.items() if values and key not in {b"logo", "logo"}} try: raw = request.content.read(131073).decode("utf-8", "replace") except Exception: raw = "" return {k: v[-1] if v else "" for k, v in parse_qs(raw, keep_blank_values=True).items()} def _redirect(request, path): request.setResponseCode(303) request.setHeader(b"Location", str(path).encode("utf-8")) return b"" def _fmt_time(ts): import datetime try: return datetime.datetime.fromtimestamp(int(ts)).strftime("%Y-%m-%d %H:%M") except Exception: return "" class WebRoot(WebFeatures, resource.Resource): isLeaf = True def __init__(self, docs_enabled=False): super().__init__() self.docs_enabled = bool(docs_enabled) def _account(self, request): return Database.account_for_ip(client_ip(request)) def _page(self, request, title, body, *, account=None, active=""): config = load_config() account = account if account is not None else self._account(request) browse_class = " active" if active == "browse" else "" home_class = " active" if active == "home" else "" account_class = " active" if active == "account" else "" admin_class = " active" if active == "admin" else "" about_class = " active" if active == "about" else "" docs_class = " active" if active == "docs" else "" nav_html = ( f'Home' f'Browse' 'Channels Search' f'About' ) if self.docs_enabled: nav_html += f'Docs' if account: actions = f'{html.escape(account["username"])}' if int(account.get("is_admin") or 0): actions += f'Admin' else: actions = 'Sign in' if config["registration_enabled"]: actions += 'Create account' site_name = html.escape(config["site_name"]) full_title = site_name if title == "Home" else f"{html.escape(title)} · {site_name}" docs_footer = 'Docs' if self.docs_enabled else "" page = f""" {full_title}

{body}

""" request.setHeader(b"content-type", b"text/html; charset=utf-8") request.setHeader(b"Cache-Control", b"private, no-store") request.setHeader(b"X-Content-Type-Options", b"nosniff") return protect_forms(request, page).encode("utf-8") def _flipnote_card(self, fn): return ( '

' '

%(title)s' '%(creator_name)s' '

%(stars)s ★%(views)s views

' '

' ) % { "public": html.escape(fn["public_id"], quote=True), "creator": html.escape(fn["creator_public_id"], quote=True), "creator_name": html.escape(fn["creator_name"]), "title": html.escape(fn.get("title") or "Untitled Flipnote"), "stars": int(fn.get("stars") or 0), "views": int(fn.get("views") or 0), "hidden_class": " hidden-content" if fn.get("hidden") else "", } def render_GET(self, request): path = request.path.decode("utf-8", "replace") account = self._account(request) include_hidden = staff(account) extra = self._feature_get(request, path, account) if extra is not None: return extra if path == "/branding/logo.png": logo = Database.db_path.parent / "site-logo.png" if not logo.is_file(): request.setResponseCode(404) return b"" request.setHeader(b"Content-Type", b"image/png") request.setHeader(b"Cache-Control", b"no-cache") return logo.read_bytes() if path == "/": rows = Database.list_flipnotes(limit=12, include_hidden=include_hidden) config = load_config() body = ( '

' f'{html.escape(config["tagline"])}' f'{html.escape(config["site_name"])}

{html.escape(config["welcome_message"])}

' '

Browse Flipnotes

' '

Recent Flipnotes

View all

' '

' + "".join(self._flipnote_card(r) for r in rows) + "

" ) if not rows: body += '

No Flipnotes have been posted yet.

' return self._page(request, "Home", body, account=account, active="home") if path == "/about": body = ( '

About

Open-source projects and community work used by RexiMemo OSE.

' '

Open-source credits

' '

RexiMemo OSE builds on earlier Flipnote preservation and replacement-server work. The projects below are included directly or used by the server.

' '

pbsds / hatena-serverAGPL-3.0

The original open-source Flipnote Hatena replacement server and the main foundation for the older public server code.

' '

HatenatoolsAGPL-3.0 · pbsds and contributors

Utilities for PPM/TMB, UGO, NTFT and related Nintendo DSi and Flipnote formats. The bundled copy has been ported to Python 3.

' '

flipnote.jsMIT · James Daniel

Browser-side PPM decoding, animation rendering and audio playback on Flipnote pages.

' '

PythonPSF License

The language and runtime used by RexiMemo OSE.

' '

TwistedMIT

The networking and HTTP framework used by the server.

' '

NumPyBSD-3-Clause

Used by the legacy Flipnote media and format tools.

' '

PillowMIT-CMU

Used for image decoding and thumbnail generation.

' '

Acknowledgements

' "

Credit also goes to the Flipnote preservation and reverse-engineering community, including the people who documented Flipnote formats, DSi image formats, UGO menus, audio codecs and the original service's network behaviour.

" '

The Hatenatools sources included in this repository also credit Steven, Remark, JSAfive, Austin Burk, Midmad and WDLmaster for format research and sample files.

' '

RexiMemo is an independent project and is not affiliated with Nintendo or Hatena. Third-party software and names remain the property of their respective owners.

' ) return self._page(request, "About", body, account=account, active="about") if path == "/docs" and self.docs_enabled: from docs_page import render_docs return self._page(request, "Documentation", render_docs(), account=account, active="docs") if path.startswith("/thumb/") and path.endswith(".png"): public_id = path[len("/thumb/"):-4] fn = Database.get_flipnote_by_public_id(public_id, include_hidden=include_hidden) if not fn: request.setResponseCode(404); return b"" try: first = request_args(request).get("first", [""])[0] == "1" data = ppm_thumbnail_png(Database.GetFlipnotePPM(fn["creator_fsid"], fn["filename"], include_hidden=include_hidden), (320, 240), first_frame=first) except Exception: request.setResponseCode(500); return b"" request.setHeader(b"content-type", b"image/png") request.setHeader(b"Cache-Control", b"private, no-store") return data if path.startswith("/comment-thumb/") and path.endswith(".png"): raw_id = path[len("/comment-thumb/"):-4] try: comment_id = int(raw_id) except ValueError: request.setResponseCode(404); return b"" raw = Database.comment_ppm(comment_id, include_hidden=include_hidden) if raw is None: request.setResponseCode(404); return b"" try: data = ppm_thumbnail_png(raw, (256, 192)) except Exception: request.setResponseCode(500); return b"" request.setHeader(b"content-type", b"image/png") request.setHeader(b"Cache-Control", b"private, no-store") return data if path.startswith("/media/") and path.endswith(".ppm"): public_id = path[len("/media/"):-4] fn = Database.get_flipnote_by_public_id(public_id, include_hidden=include_hidden) if not fn: request.setResponseCode(404); return b"" request.setHeader(b"content-type", b"application/octet-stream") request.setHeader(b"content-disposition", b"inline") request.setHeader(b"Cache-Control", b"private, no-store") return Database.GetFlipnotePPM(fn["creator_fsid"], fn["filename"], include_hidden=include_hidden) if path.startswith("/flipnote/") and path.endswith(".ppm"): public_id = path[len("/flipnote/"):-4] fn = Database.get_flipnote_by_public_id(public_id, include_hidden=include_hidden) if not fn: request.setResponseCode(404); return b"" Database.AddDownload(fn["creator_fsid"], fn["filename"]) request.setHeader(b"content-type", b"application/octet-stream") request.setHeader(b"content-disposition", (f'attachment; filename="{fn["filename"]}.ppm"').encode("ascii", "ignore")) request.setHeader(b"Cache-Control", b"private, no-store") return Database.GetFlipnotePPM(fn["creator_fsid"], fn["filename"], include_hidden=include_hidden) if path.startswith("/watch/"): public_id = path[len("/watch/"):].strip("/") fn = Database.get_flipnote_by_public_id(public_id, include_hidden=include_hidden) if not fn: request.setResponseCode(404) return self._page(request, "Not found", '

Flipnote not found

', account=account) Database.AddView(fn["creator_fsid"], fn["filename"]) fn = Database.get_flipnote_by_public_id(public_id, include_hidden=include_hidden) comment_total = Database.count_comments(fn["id"], include_hidden=include_hidden) comment_page = page_number(request_args(request).get("page", ["1"])[0], comment_total, 25) comments = Database.list_comments(fn["id"], limit=25, offset=(comment_page-1)*25, include_hidden=include_hidden) channel = Database.get_channel(fn["channel_id"]) if fn.get("channel_id") else None room = Database.get_creator_room(fn["creator_room_id"]) related = [r for r in Database.list_creator_flipnotes(fn["creator_room_id"], limit=8, include_hidden=include_hidden) if int(r["id"]) != int(fn["id"])][:4] try: ppm_size = Database.FlipnotePath(fn["creator_fsid"], fn["filename"]).stat().st_size size_text = "%.1f KiB" % (ppm_size / 1024.0) except OSError: size_text = "Unknown" rendered_comments = [] for c in comments: if c["type"] == "memo": content = ( '' '

' 'Mini Flipnote comment' % (int(c["id"]), int(c["id"])) ) else: content = '

' % html.escape(c.get("text_content") or "") if c.get("hidden"): content = '

Hidden by moderation. Only administrators can see this comment.

' + content if staff(account) or Database.owns_comment(account["id"] if account else None, client_ip(request), c): content += f'' rendered_comments.append( '

%s%s

' % ( html.escape(c["creator_public_id"], quote=True), html.escape(c["creator_name"]), _fmt_time(c["posted_at"]), content ) ) comment_form = ( '' ) % html.escape(public_id, quote=True) if not account and not load_config()["guest_posting"]: comment_form = '

' comment_form += pagination(comment_page, comment_total, 25, "/watch/" + public_id) related_html = "".join(self._flipnote_card(r) for r in related) body = ( '

Flipnote

%(title)s

' '

by %(creator_name)s · %(posted)s

' 'Creator\'s Room

' '

Flipnote playerPPM · flipnote.js

' '

' '' '

Loading Flipnote…

' '

' '' '' '0:00 / 0:00' '' '

' '' '

' '

Comments%(comment_count)s

' '

%(form)s

%(comments)s

' '%(related_section)s' '' '' ) % { "public": html.escape(fn["public_id"], quote=True), "title": html.escape(fn.get("title") or "Untitled Flipnote"), "title_attr": html.escape(fn.get("title") or "Untitled Flipnote", quote=True), "creator": html.escape(fn["creator_public_id"], quote=True), "creator_name": html.escape(fn["creator_name"]), "identity": "Account Creator's Room" if room and room.get("account_id") else "Guest/IP Creator's Room", "stars": int(fn.get("stars") or 0), "views": int(fn.get("views") or 0), "downloads": int(fn.get("downloads") or 0), "comment_count": comment_total, "posted": _fmt_time(fn.get("posted_at")), "channel": html.escape((channel or {}).get("title") or "General"), "fsid": html.escape(fn.get("creator_fsid") or ""), "filename": html.escape(fn.get("filename") or ""), "size": html.escape(size_text), "form": comment_form, "comments": "".join(rendered_comments) or '

No comments yet.

', "related_section": ( '

More from %s

Open Creator\'s Room

' % (html.escape(fn["creator_name"]), html.escape(fn["creator_public_id"], quote=True), related_html) ) if related_html else "", } if fn.get("description"): description = '

Description

' + html.escape(fn["description"]) + '

' body = body.replace('

Manage Flipnote' if fn.get("hidden"): controls = '

This Flipnote is hidden. Only administrators can see it.

' + controls if controls: body = '

' + controls + '

' + body return self._page(request, fn.get("title") or "Flipnote", body, account=account) if path == "/login": body = ( '

' '

OSE login state is tied to your IP address.

' '

Create an account

' ) return self._page(request, "Sign in", body, account=account) if path == "/register": if not load_config()["registration_enabled"]: return self._error(request, "New account registration is currently closed.", 403) body = ( '

Create account

' '' '

Already have an account?

' ) return self._page(request, "Create account", body, account=account) if path == "/logout": Database.logout_ip(client_ip(request)) return _redirect(request, "/") if path == "/account": if not account: return _redirect(request, "/login") room = Database.room_for_account(account["id"]) room_link = 'Open Creator\'s Room' % html.escape(room["public_id"], quote=True) if room else "Creator's Room will be created after your first Flipnote post." body = ( '

%s

Account

' '

Account details

' '

Login IP: %s

Sign out this IP

' ) % (html.escape(account["username"]), room_link, html.escape(client_ip(request))) return self._page(request, "Account", body, account=account, active="account") request.setResponseCode(404) return self._page(request, "Not found", '

Not found

', account=account) def render_POST(self, request): path = request.path.decode("utf-8", "replace") fields = _fields(request) ip = client_ip(request) account = self._account(request) if not valid_csrf(request, fields): return self._error(request, "That form expired. Reload the page and try again.", 403) extra = self._feature_post(request, path, fields, account) if extra is not None: return extra if path == "/login": found = Database.verify_account(fields.get("username", ""), fields.get("password", "")) if not found: request.setResponseCode(403) return self._page(request, "Sign in", '

Sign in failed

Try again

', account=None) Database.login_ip(ip, found["id"]) return _redirect(request, "/account") if path == "/register": if not load_config()["registration_enabled"]: return self._error(request, "New account registration is currently closed.", 403) try: aid = Database.create_account(fields.get("username", ""), fields.get("password", "")) Database.login_ip(ip, aid) except ValueError as exc: request.setResponseCode(400) return self._page(request, "Create account", '

Could not create account

' % html.escape(str(exc)), account=None) return _redirect(request, "/account") if path.startswith("/watch/") and path.endswith("/comment"): if not account and not load_config()["guest_posting"]: return self._error(request, "Sign in to leave a comment.", 403) public_id = path[len("/watch/"):-len("/comment")].strip("/") fn = Database.get_flipnote_by_public_id(public_id) if not fn: request.setResponseCode(404); return b"" try: Database.add_text_comment(fn["id"], fields.get("text", ""), ip_address=ip, account_id=(account["id"] if account else None)) except ValueError: pass return _redirect(request, "/watch/" + public_id) if path.startswith("/admin/"): if not account or not int(account.get("is_admin") or 0): request.setResponseCode(403); return b"" if path == "/admin/ban": try: Database.ban_ip(fields.get("ip", ""), fields.get("reason", "")) except ValueError: pass elif path == "/admin/unban": Database.unban_ip(fields.get("ip", "")) elif path == "/admin/delete": try: Database.delete_flipnote(int(fields.get("id", "0"))) except ValueError: pass return _redirect(request, "/admin") request.setResponseCode(404) return b""