rekushi
3c70f364f8
Implement search, channels, creator controls, moderation tabs and server customisation.
211 lines
19 KiB
Python
211 lines
19 KiB
Python
"""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'<option value="{row["id"]}"{" selected" if row["id"] == selected else ""}>{h(row["title"])}</option>'
|
||
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", '<header class="page-head"><h1>Could not save</h1></header><p class="notice">' + h(str(message)) + '</p><p><a class="button" href="/">Home</a></p>')
|
||
|
||
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 = '<div class="creator-grid">' + ''.join(f'<a class="panel creator-result" href="/creator/{h(row["public_id"], quote=True)}"><strong>{h(row["display_name"])}</strong><small>Creator’s Room</small></a>' for row in rows) + '</div>'
|
||
else:
|
||
rows = Database.list_flipnotes(q=query, sort=sort, channel_id=channel, page=page, per_page=per_page, include_hidden=include_hidden)
|
||
cards = '<div class="flipnote-grid">' + ''.join(self._flipnote_card(row) for row in rows) + '</div>'
|
||
title = "Search" if path == "/search" else "Browse Flipnotes"
|
||
scope_field = ('<label>Show<select name="type"><option value="flipnotes">Flipnotes</option>'
|
||
+ ('<option value="creators" selected>Creators</option>' if scope == 'creators' else '<option value="creators">Creators</option>')
|
||
+ '</select></label>') if path == '/search' else ''
|
||
form = f'''<form method="get" class="browse-controls" action="{path}"><label>Search<input type="search" name="q" value="{h(query, quote=True)}" maxlength="100" placeholder="Flipnote title or creator"></label>
|
||
{scope_field}
|
||
<label>Channel<select name="channel"><option value="0">All channels</option>{channel_options(channel)}</select></label>
|
||
<label>Order<select name="sort">{''.join(f'<option value="{key}"{" selected" if sort == key else ""}>{label}</option>' for key,label in (('new','Newest'),('popular','Popular'),('top','Most stars')))}</select></label><button class="button primary">Search</button></form>'''
|
||
body = f'<header class="page-head"><h1>{title}</h1></header>{form}<p class="results-count">{total} {scope}</p>'
|
||
body += cards if rows else '<p class="empty-state">Nothing matched this search.</p>'
|
||
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'<a class="panel creator-result" href="/browse?channel={row["id"]}"><strong>{h(row["title"])}</strong><p>{h(row["description"])}</p><small>{Database.count_flipnotes(channel_id=row["id"])} Flipnotes</small></a>' for row in Database.list_channels())
|
||
return self._page(request, "Channels", '<header class="page-head"><h1>Channels</h1><p>Browse Flipnotes by subject.</p></header><div class="creator-grid">' + cards + '</div>', 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'<header class="page-head"><h1>{h(room["display_name"])}</h1><p>{total} Flipnotes</p></header><div class="flipnote-grid">' + ''.join(self._flipnote_card(row) for row in rows) + '</div>'
|
||
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'''<header class="page-head split"><h1>Manage Flipnote</h1><a class="button" href="/watch/{h(note_id)}">Back to Flipnote</a></header>
|
||
<section class="panel"><form method="post" action="{path}" class="stacked-form padded-form"><label>Title<input name="title" required maxlength="80" value="{h(note['title'], quote=True)}"></label>
|
||
<label>Description<textarea name="description" rows="5" maxlength="500">{h(note.get('description') or '')}</textarea></label><label>Channel<select name="channel_id">{channel_options(note['channel_id'])}</select></label><div><button class="button primary">Save changes</button></div></form></section>
|
||
<details class="panel delete-confirm"><summary>Delete this Flipnote</summary><p>This removes the Flipnote from this server’s web and DSi listings.</p><form method="post" action="/watch/{h(note_id)}/delete"><button class="button danger">Delete Flipnote</button></form></details>'''
|
||
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 = '<header class="page-head"><h1>Admin</h1><p>Manage this server and its content.</p></header><nav class="tabs admin-tabs">'
|
||
body += ''.join(f'<a class="{"active" if key == tab else ""}" href="/admin?tab={key}">{label}</a>' for key,label in tabs.items()) + '</nav>'
|
||
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'<form method="get" class="browse-controls"><input type="hidden" name="tab" value="{tab}"><label>Search<input type="search" name="q" value="{h(query, quote=True)}"></label><button class="button">Search</button></form>'
|
||
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'''<section class="panel moderation-row{' hidden-content' if hidden else ''}"><div><a href="/watch/{h(public_id, quote=True)}"><strong>{h(title or 'Untitled Flipnote')}</strong></a><p>{h(row['creator_name'])} · {'Hidden' if hidden else 'Visible'}</p><small>{h(row.get('moderation_reason') or '')}</small></div>
|
||
<form method="post" action="/admin/{'comment' if tab == 'comments' else 'flipnote'}-visibility"><input type="hidden" name="id" value="{row['id']}"><input type="hidden" name="hidden" value="{0 if hidden else 1}"><label class="sr-only" for="reason-{tab}-{row['id']}">Reason</label><input id="reason-{tab}-{row['id']}" name="reason" maxlength="200" placeholder="Reason (optional)"><button class="button {'subtle' if hidden else 'danger'}">{'Restore' if hidden else 'Hide'}</button></form></section>'''
|
||
body += pagination(page, total, 25, "/admin", tab=tab, q=query)
|
||
if not rows:
|
||
body += '<p class="empty-state">No matching content.</p>'
|
||
elif tab == "channels":
|
||
body += '<section class="panel"><div class="panel-title">New channel</div><form class="padded-form stacked-form" method="post" action="/admin/channel-create"><label>Name<input name="title" maxlength="40" required></label><label>Description<textarea name="description" maxlength="500" rows="2"></textarea></label><div><button class="button primary">Create channel</button></div></form></section>'
|
||
for channel in Database.list_channels():
|
||
body += f'''<section class="panel channel-edit"><form method="post" action="/admin/channel-update" class="padded-form stacked-form"><input type="hidden" name="id" value="{channel['id']}"><label>Name<input name="title" required maxlength="40" value="{h(channel['title'], quote=True)}"></label><label>Description<textarea name="description" maxlength="500" rows="2">{h(channel['description'])}</textarea></label><div><button class="button">Save channel</button></div></form>'''
|
||
if channel["id"] != 1:
|
||
body += f'<details class="delete-confirm"><summary>Remove channel</summary><form method="post" action="/admin/channel-delete" class="browse-controls"><input type="hidden" name="id" value="{channel["id"]}"><label>Move its Flipnotes to<select name="target_channel_id">{channel_options(excluding=channel["id"])}</select></label><button class="button danger">Remove channel</button></form></details>'
|
||
body += '</section>'
|
||
elif tab == "server":
|
||
config = load_config()
|
||
body += '<section class="panel"><div class="panel-title">Server appearance</div><form method="post" action="/admin/server" class="padded-form stacked-form">'
|
||
for key,label,limit in (("site_name","Server name",60),("tagline","Tagline",100),("welcome_message","Welcome message",500)):
|
||
body += f'<label>{label}<input name="{key}" maxlength="{limit}" value="{h(str(config[key]), quote=True)}"></label>'
|
||
body += f'<label>Accent colour<input type="color" name="accent_colour" value="{h(config["accent_colour"], quote=True)}"></label><label>Maximum PPM upload (KiB)<input type="number" name="max_upload_kib" min="64" max="16384" value="{config["max_upload_bytes"]//1024}"></label>'
|
||
for key,label in (("registration_enabled","Allow new accounts"),("guest_posting","Allow guest posts and comments")):
|
||
body += f'<label class="check-label"><input type="checkbox" name="{key}" value="1"{" checked" if config[key] else ""}>{label}</label>'
|
||
body += '<div><button class="button primary">Save settings</button></div></form></section><section class="panel section"><div class="panel-title">Server logo</div><form class="padded-form stacked-form" method="post" action="/admin/logo" enctype="multipart/form-data"><label>Image<input type="file" name="logo" accept="image/png,image/jpeg,image/webp" required></label><small>Up to 2 MB. Converted to PNG and fitted within 1024 × 512 pixels.</small><div><button class="button">Upload logo</button></div></form><form class="padded-form" method="post" action="/admin/logo-reset"><button class="button subtle">Use default logo</button></form></section>'
|
||
else:
|
||
body += '<section class="panel"><div class="panel-title">Block an IP</div><form class="browse-controls padded-form" action="/admin/ban" method="post"><label>IP address<input name="ip" required></label><label>Reason<input name="reason" maxlength="200"></label><button class="button danger">Block IP</button></form></section>'
|
||
for ban in Database.list_bans():
|
||
body += f'<section class="panel moderation-row"><div><strong>{h(ban["ip"])}</strong><p>{h(ban["reason"])}</p></div><form method="post" action="/admin/unban"><input type="hidden" name="ip" value="{h(ban["ip"], quote=True)}"><button class="button">Unblock</button></form></section>'
|
||
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))
|