"""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", '
', 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'''
{h(row['creator_name'])} · {'Hidden' if hidden else 'Visible'}
{h(row.get('moderation_reason') or '')}
'''
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
Server logo
'
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))