diff --git a/web_features.py b/web_features.py new file mode 100644 index 0000000..d0d2b84 --- /dev/null +++ b/web_features.py @@ -0,0 +1,210 @@ +"""Public browsing, creator tools and the compact administration portal.""" +from __future__ import annotations + +from html import escape as h +from urllib.parse import urlencode +import io +import os + +from PIL import Image, ImageOps +from DB import Database +from hatena import client_ip, request_args +from server_config import load_config, save_config +from web_ui import page_number, pagination + + +def staff(account): + return bool(account and account.get("is_admin")) + + +def channel_options(selected=1, *, excluding=None): + return ''.join(f'' + for row in Database.list_channels() if row["id"] != excluding) + + +class WebFeatures: + def _error(self, request, message, status=400): + request.setResponseCode(status) + return self._page(request, "Could not save", '

Could not save

' + h(str(message)) + '

Home

') + + def _feature_get(self, request, path, account): + args = request_args(request) + value = lambda key, default="": (args.get(key) or [default])[0] + include_hidden = staff(account) + if path in {"/browse", "/search"}: + query = value("q").strip()[:100] + sort = value("sort", "new") + sort = sort if sort in {"new", "popular", "top"} else "new" + scope = "creators" if path == "/search" and value("type") == "creators" else "flipnotes" + try: + channel = int(value("channel", "0")) or None + except ValueError: + channel = None + per_page = 24 + total = Database.count_creators(query) if scope == "creators" else Database.count_flipnotes(q=query, channel_id=channel, include_hidden=include_hidden) + page = page_number(value("page", "1"), total, per_page) + if scope == "creators": + rows = Database.search_creators(query, page=page, per_page=per_page) + cards = '
' + ''.join(f'{h(row["display_name"])}Creator’s Room' for row in rows) + '
' + else: + rows = Database.list_flipnotes(q=query, sort=sort, channel_id=channel, page=page, per_page=per_page, include_hidden=include_hidden) + cards = '
' + ''.join(self._flipnote_card(row) for row in rows) + '
' + title = "Search" if path == "/search" else "Browse Flipnotes" + scope_field = ('') if path == '/search' else '' + form = f'''
+{scope_field} + +
''' + body = f'

{title}

{form}

{total} {scope}

' + body += cards if rows else '

Nothing matched this search.

' + body += pagination(page, total, per_page, path, q=query, sort=sort, channel=channel or 0, type=scope) + return self._page(request, title, body, account=account, active="browse") + if path == "/channels": + cards = ''.join(f'{h(row["title"])}

{h(row["description"])}

{Database.count_flipnotes(channel_id=row["id"])} Flipnotes' for row in Database.list_channels()) + return self._page(request, "Channels", '

Channels

Browse Flipnotes by subject.

' + cards + '
', account=account) + if path.startswith("/creator/"): + room = Database.get_creator_room_by_public_id(path[len("/creator/"):].strip("/")) + if not room: + return self._error(request, "Creator not found.", 404) + total = Database.count_flipnotes(room_id=room["id"], include_hidden=include_hidden) + page = page_number(value("page", "1"), total, 24) + rows = Database.list_flipnotes(room_id=room["id"], page=page, per_page=24, include_hidden=include_hidden) + body = f'

{h(room["display_name"])}

{total} Flipnotes

' + ''.join(self._flipnote_card(row) for row in rows) + '
' + body += pagination(page, total, 24, path) + return self._page(request, room["display_name"], body, account=account) + if path.startswith("/watch/") and path.endswith("/manage"): + note_id = path[len("/watch/"):-len("/manage")].strip("/") + note = Database.get_flipnote_by_public_id(note_id, include_hidden=include_hidden) + if not note: + return self._error(request, "Flipnote not found.", 404) + if not staff(account) and not Database.owns_flipnote(account["id"] if account else None, client_ip(request), note): + return self._error(request, "Only the creator can change this Flipnote.", 403) + body = f'''

Manage Flipnote

Back to Flipnote
+
+
+
Delete this Flipnote

This removes the Flipnote from this server’s web and DSi listings.

''' + return self._page(request, "Manage Flipnote", body, account=account) + if path == "/admin": + if not staff(account): + return self._error(request, "Administrator access is required.", 403) + return self._admin_page(request, account, args) + return None + + def _admin_page(self, request, account, args): + value = lambda key, default="": (args.get(key) or [default])[0] + tabs = {"flipnotes": "Flipnotes", "comments": "Comments", "channels": "Channels", "server": "Server settings", "bans": "IP bans"} + tab = value("tab", "flipnotes") + tab = tab if tab in tabs else "flipnotes" + body = '

Admin

Manage this server and its content.

' + if tab in {"flipnotes", "comments"}: + query = value("q").strip()[:100] + total = Database.count_moderation_comments(query) if tab == "comments" else Database.count_flipnotes(q=query, include_hidden=True) + page = page_number(value("page", "1"), total, 25) + rows = Database.list_moderation_comments(query, page=page, per_page=25) if tab == "comments" else Database.list_flipnotes(q=query, page=page, per_page=25, include_hidden=True) + body += f'
' + for row in rows: + hidden = bool(row["hidden"]) + title = row.get("text_content") if tab == "comments" and row["type"] == "text" else "Mini Flipnote comment" if tab == "comments" else row["title"] + public_id = row.get("flipnote_public_id") if tab == "comments" else row["public_id"] + body += f'''''' + body += pagination(page, total, 25, "/admin", tab=tab, q=query) + if not rows: + body += '

No matching content.

' + elif tab == "channels": + body += '
New channel
' + for channel in Database.list_channels(): + body += f'''
''' + if channel["id"] != 1: + body += f'
Remove channel
' + body += '
' + elif tab == "server": + config = load_config() + body += '
Server appearance
' + for key,label,limit in (("site_name","Server name",60),("tagline","Tagline",100),("welcome_message","Welcome message",500)): + body += f'' + body += f'' + for key,label in (("registration_enabled","Allow new accounts"),("guest_posting","Allow guest posts and comments")): + body += f'' + body += '
Server logo
Up to 2 MB. Converted to PNG and fitted within 1024 × 512 pixels.
' + else: + body += '
Block an IP
' + for ban in Database.list_bans(): + body += f'
{h(ban["ip"])}

{h(ban["reason"])}

' + return self._page(request, "Admin", body, account=account, active="admin") + + def _feature_post(self, request, path, fields, account): + def redirect(target): + request.setResponseCode(303) + request.setHeader(b"Location", target.encode()) + return b"" + try: + if path.startswith('/watch/') and path.rsplit('/',1)[-1] in {'manage','delete'}: + public_id, action = path[len('/watch/'):].rsplit('/',1) + note = Database.get_flipnote_by_public_id(public_id, include_hidden=staff(account)) + if not note: + return self._error(request, "Flipnote not found.", 404) + if not staff(account) and not Database.owns_flipnote(account['id'] if account else None, client_ip(request), note): + return self._error(request, "Only the creator can change this Flipnote.", 403) + if action == 'delete': + Database.delete_flipnote(note['id']) + return redirect('/creator/' + note['creator_public_id']) + Database.update_flipnote(note['id'], fields.get('title',''), fields.get('description',''), int(fields.get('channel_id','1'))) + return redirect('/watch/' + public_id) + if path.startswith('/comments/') and path.endswith('/delete'): + comment = Database.get_comment(int(path.split('/')[2]), include_hidden=staff(account)) + if not comment: + return self._error(request, "Comment not found.", 404) + if not staff(account) and not Database.owns_comment(account['id'] if account else None, client_ip(request), comment): + return self._error(request, "Only the commenter can remove this comment.", 403) + Database.delete_comment(comment['id']) + parent = Database.get_flipnote_by_id(comment['flipnote_id'], include_hidden=staff(account)) + return redirect('/watch/' + parent['public_id'] if parent else '/') + if not path.startswith('/admin/'): + return None + if not staff(account): + return self._error(request, "Administrator access is required.", 403) + if path in {'/admin/flipnote-visibility','/admin/comment-visibility'}: + if fields.get('hidden') not in {'0','1'}: + raise ValueError('Choose whether to hide or restore the content.') + setter = Database.set_comment_hidden if path == '/admin/comment-visibility' else Database.set_flipnote_hidden + setter(int(fields.get('id','0')), fields['hidden']=='1', fields.get('reason','')) + return redirect('/admin?tab=' + ('comments' if path == '/admin/comment-visibility' else 'flipnotes')) + if path == '/admin/channel-create': + Database.create_channel(fields.get('title',''), fields.get('description','')) + elif path == '/admin/channel-update': + Database.update_channel(int(fields.get('id','0')), fields.get('title',''), fields.get('description','')) + elif path == '/admin/channel-delete': + Database.delete_channel(int(fields.get('id','0')), int(fields.get('target_channel_id','1'))) + elif path == '/admin/server': + changes = {key: fields.get(key,'') for key in ('site_name','tagline','welcome_message','accent_colour')} + changes.update(registration_enabled=fields.get('registration_enabled')=='1', guest_posting=fields.get('guest_posting')=='1', max_upload_bytes=int(fields.get('max_upload_kib','2048'))*1024) + save_config(changes) + return redirect('/admin?tab=server') + elif path == '/admin/logo': + values = request.args.get(b'logo', []) + raw = values[0] if values else b'' + if not raw or len(raw)>2*1024*1024: + raise ValueError('Choose an image up to 2 MB.') + with Image.open(io.BytesIO(raw)) as original: + if original.width*original.height>16_000_000: + raise ValueError('This image is too large.') + image = ImageOps.exif_transpose(original).convert('RGBA') + image.thumbnail((1024,512), Image.Resampling.LANCZOS) + target = Database.db_path.parent/'site-logo.png' + temporary = target.with_suffix('.tmp') + image.save(temporary, format='PNG') + os.replace(temporary,target) + save_config({'logo_url':'/branding/logo.png'}) + return redirect('/admin?tab=server') + elif path == '/admin/logo-reset': + save_config({'logo_url':'/static/RexiMemo_OSE_Logo.png'}) + return redirect('/admin?tab=server') + else: + return None + return redirect('/admin?tab=channels') + except (ValueError, TypeError, OSError, Image.DecompressionBombError) as exc: + return self._error(request, str(exc))