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 def _fields(request): try: raw = request.content.read().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(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=""): 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' 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 Create account' full_title = "RexiMemo OSE" if title == "Home" else f"{html.escape(title)} · RexiMemo OSE" docs_footer = 'Docs' if self.docs_enabled else "" page = f""" {full_title}

{body}

""" request.setHeader(b"content-type", b"text/html; charset=utf-8") return 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), } def render_GET(self, request): path = request.path.decode("utf-8", "replace") account = self._account(request) if path == "/": rows = Database.list_flipnotes(limit=12) body = ( '

' 'Open Source Edition' 'RexiMemo!

A small Flipnote Hatena-compatible server.

' '

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 == "/browse": sort = request_args(request).get("sort", ["new"])[0] if sort not in {"new", "popular", "top"}: sort = "new" rows = Database.list_flipnotes(sort=sort, limit=100) body = ( '

Browse

Flipnotes posted to this server.

' '

Newest Popular Stars

' '

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

" ) if not rows: body += '

No Flipnotes found.

' return self._page(request, "Browse", body, account=account, active="browse") if path.startswith("/thumb/") and path.endswith(".png"): public_id = path[len("/thumb/"):-4] fn = Database.get_flipnote_by_public_id(public_id) if not fn: request.setResponseCode(404); return b"" try: data = ppm_thumbnail_png(Database.GetFlipnotePPM(fn["creator_fsid"], fn["filename"]), (320, 240)) except Exception: request.setResponseCode(500); return b"" request.setHeader(b"content-type", b"image/png") request.setHeader(b"Cache-Control", b"public, max-age=3600") 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) 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"public, max-age=3600") 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) 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"public, max-age=3600") return Database.GetFlipnotePPM(fn["creator_fsid"], fn["filename"]) if path.startswith("/flipnote/") and path.endswith(".ppm"): public_id = path[len("/flipnote/"):-4] fn = Database.get_flipnote_by_public_id(public_id) 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")) return Database.GetFlipnotePPM(fn["creator_fsid"], fn["filename"]) if path.startswith("/watch/"): public_id = path[len("/watch/"):].strip("/") fn = Database.get_flipnote_by_public_id(public_id) 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) comments = Database.list_comments(fn["id"], limit=100) 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) 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 "") 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) 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": len(comments), "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 "", } return self._page(request, fn.get("title") or "Flipnote", body, account=account) if path.startswith("/creator/"): public_id = path[len("/creator/"):].strip("/") room = Database.get_creator_room_by_public_id(public_id) if not room: request.setResponseCode(404) return self._page(request, "Not found", '

Creator not found

', account=account) notes = Database.list_creator_flipnotes(room["id"], limit=100) owner = "Account" if room.get("account_id") else "Guest/IP creator" body = ( '

%s

%s · Room created when this creator first posted a Flipnote.

' '

Posted Flipnotes

' '

' ) % (html.escape(room["display_name"]), owner, len(notes), "".join(self._flipnote_card(r) for r in notes)) return self._page(request, room["display_name"], 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": 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") if path == "/admin": if not account or not int(account.get("is_admin") or 0): request.setResponseCode(403) return self._page(request, "Admin", '

Admin access required

', account=account) bans = Database.list_bans() notes = Database.list_flipnotes(limit=100) ban_rows = "".join('%s%s' % (html.escape(b["ip"]), html.escape(b.get("reason") or ""), html.escape(b["ip"], quote=True)) for b in bans) note_rows = "".join('%s%s' % (html.escape(n["title"]), html.escape(n["creator_name"]), int(n["id"])) for n in notes) body = ( '

Admin

IP bans and Flipnote deletion.

' '

Ban IP

' '

Blocked IPs

IP	Reason

' '

Flipnotes

Title	Creator

' ) % (ban_rows, note_rows) return self._page(request, "Admin", body, account=account, active="admin") 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 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": 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"): 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""