fastbox.py

"""FastBox v1.0: a lightweight single-file FastAPI LAN sharing app.

FastBox keeps uploads, chat, boards, media browsing, search, moderation tools,
and operational pages in one readable module so it stays easy to inspect,
modify, back up, and deploy on a small local network.
"""

from __future__ import annotations

import html
import json
import base64
import hashlib
import hmac
import logging
import mimetypes
import os
import re
import sqlite3
import uuid
from contextlib import closing
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
from urllib.parse import quote

import uvicorn
from fastapi import FastAPI, File, Form, HTTPException, Request, UploadFile, WebSocket, WebSocketDisconnect, status
from fastapi.responses import FileResponse, HTMLResponse, JSONResponse, RedirectResponse, Response
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates

APP_NAME = "FastBox"
APP_VERSION = "1.0.0"
HOST = "0.0.0.0"
PORT = 8000
BASE_DIR = Path(__file__).resolve().parent
STORAGE_DIR = BASE_DIR / "storage"
UPLOADS_DIR = STORAGE_DIR / "uploads"
ATTACHMENTS_DIR = STORAGE_DIR / "attachments"
STATIC_DIR = BASE_DIR / "static"
TEMPLATES_DIR = BASE_DIR / "templates"
DATABASE_PATH = BASE_DIR / "fastbox.db"
MAX_UPLOAD_BYTES = 25 * 1024 * 1024
MAX_ATTACHMENT_BYTES = 5 * 1024 * 1024
UPLOAD_CHUNK_BYTES = 1024 * 1024
MAX_CHAT_MESSAGE_CHARS = 1000
MAX_NICKNAME_CHARS = 24
MAX_THREAD_TITLE_CHARS = 120
MAX_POST_BODY_CHARS = 5000
RECENT_CHAT_LIMIT = 50
RECENT_BOARD_LIMIT = 20
SAFE_FILENAME_PATTERN = re.compile(r"[^A-Za-z0-9._ -]+")
ALLOWED_ATTACHMENT_TYPES = {"image/png", "image/jpeg", "image/gif", "image/webp"}
IMAGE_MIME_TYPES = {"image/png", "image/jpeg", "image/gif", "image/webp", "image/svg+xml"}
VIDEO_MIME_TYPES = {"video/mp4", "video/webm", "video/ogg"}
AUDIO_MIME_TYPES = {"audio/mpeg", "audio/mp3", "audio/wav", "audio/ogg", "audio/webm", "audio/mp4"}
PDF_MIME_TYPES = {"application/pdf"}
MEDIA_FILTERS = {"all", "image", "video", "audio", "pdf", "other"}
SEARCH_FILTERS = {"all", "uploads", "chat", "threads", "replies"}
MAX_SEARCH_QUERY_CHARS = 100
DEFAULT_ACCENT_COLOR = os.environ.get("FASTBOX_ACCENT_COLOR", "#32cd32")
ACCENT_COLOR_PATTERN = re.compile(r"^#[0-9A-Fa-f]{6}$")
ADMIN_PASSWORD = os.environ.get("FASTBOX_ADMIN_PASSWORD", "fastbox")
ADMIN_SESSION_COOKIE = "fastbox_admin_session"
LOG_LEVEL = os.environ.get("FASTBOX_LOG_LEVEL", "INFO").upper()
logging.basicConfig(level=getattr(logging, LOG_LEVEL, logging.INFO), format="%(asctime)s %(levelname)s %(name)s: %(message)s")
logger = logging.getLogger("fastbox")

FAVICON_BYTES = base64.b64decode(
    """AAABAAYAEBAAAAAAIACaAgAAZgAAACAgAAAAACAABgUAAAADAAAwMAAAAAAgAFAHAAAGCAAAQEAAAAAAIAB5CQAAVg8AAICAAAAAACAAixAAAM8YAAAAAAAAAAAgACUGAABaKQAAiVBORw0KGgoAAAANSUhEUgAAABAAAAAQCAYAAAAf8/9hAAACYUlEQVR4nHWTz0tUURTHP+fe+57zZkYnraB01EqLqEWbIBByV7tWQQSt2rVz0yYQXEcQEUERRIs29UcU1CKqhYhEKmqWJWpUVuP8eu/dd1pMWpheuNzN93zP936/50hp76D6WpXSiZNE5TLqFUTY9qgiVmisrPBj/DU2FyHFwn7tOX+JYnc/tflpJMxtXwyQKZo2iQ4MUl/7yqfHD5GeoXO6b/gsH+/eRFyIr1Z26A4m14Zpi/DVdfqujPBt4g0uKvdTm59FnCMq97FraLiFzrT1IqhmmNBQefuO+sIcIkJ1dpqopw+nWYoEeTRJCLp2c2RsFARMG/DHCvUQdsLM6D0WJ8eRMIdYh/oUt4kCNPMkPz2oUluYQ+MYjKA+w7VbGkuLSBCAZq0CEdzWr4qzZM2YmWsjNJeXkDAE1da1FlsoklZ+beK3EEjLrCDg2O0HaBJjIkt1Zp75savg3H8J/6dgQ1phsNxqmgdtQiX2kFmMEcKdCRQEsjhhZuQq8ddVTFtIvF7nTG+RjryjnsDT2crOBOo9mqasT72lufwZgpDQGo4e7+LJQo0Lh4q8CgyZKgZwqP5VbixBpyVrRNhCAZMvbJoYCHQXLKFpjYiIgCpOrEPTFAkCkh9rvL9+C/WedL2CWIuokig8W43pzDlefEmoe8h5D9bhGitLdJwaJmvGND4t8uHODQBssR0xhg2FEz8Vr2CtJaj9In9wgLXpSaRY7Nbei5fJlbqozU79XSbv0X/cMa1wyNKE3MHDJEmTxUf3kdKeAfWNOl2nThOVe1C/bbD/+ASN1VW+v3yOCUN+A7grBFCpTLmYAAAAAElFTkSuQmCCiVBORw0KGgoAAAANSUhEUgAAACAAAAAgCAYAAABzenr0AAAEzUlEQVR4nK2XW2xUVRSGv73PpZ1LOx0KKVSLhVIwlkI0xqjEGJ+MlyAxKvHyqk+a+KRGY6ImxoTEaASigoE3TcRETB8MGhNCHwgCwQRNsUih2Cu2hXamZ27n7OXDGaZDZ+h9JfvlnD2zvv3v8++1tkqsbhOKoSwLk8/jp6aQwGdFQ1s48Tp0bS0SBKXHdnnywuR13MYmmp/dTWxDO6CKYzkhoCAz2M+1X34mc/UyTsMqxIQQKrG6TZRlUbg+zrpndrPlrffIDY3gXepDREApEJknyfwRWd9CtG0DfQe+pP/Q/hKEfXPl63bupuPDT7nw9uuMdx8vS7rc5EUFlaK+czsdew+hHZe+rz/DTa5C1SdbxYrGePjoCS5+9A4Tx3/FbVpbBrDULZgFrjSFiTFim7bQefA7fn/lKbwrfWh/apKmx3eSGxpgovs33KZ1iO8jQRB+LGKQRQ7EIEZK/yFBgPgFnMbVpHrOc+P0SZp3vYifnsIWExDb2M70pd4iuJlZgzGYbGZpAiiNrqm59ZkxKNthureHSPtm0HrGBYiUZFdKEWQzxDu20/rGm0jOgNILSywGVaPJDY5yac8HVd7LLR+1XTkDUArxfZyGBlY9cj/GAxaYHwM6At7FawuaXh2gDMJPG4xnQBcJggBBUCiqOsQIKq/wU1PLBChCKK1D9YsAVp2eezeKChTGkisAUCX+O3acIDUJlk2FAgKIoF1FbnRs5QCEmdPgyr49eL0X0JEIGFM5VwxIqJwVi60MQHlox0G5Lsp1SwAKhYigbBvtuCGyUDrvlw1QfhZuevdjgoyH0rq0A2IC7LjF+ImTDBz+CjseR6qos2SA8kjcu7XCkhKAk4Ts0HXEL4QFbIGxaAA/XQjlLtNFggCURZDxFpV8SQB2wgltWPZliu/gJMGKxVBIVYbb1dTFu+CL/eRGR9CuUzpSxYTWy/YP4LsR/HyARm66EqWgRldXZgnnQBfTVWxoAVPK5tVta3jyrijZXPiupkZzZiTLnvNT1NuVEIvfgroETkOyAsBW4GcCtjbYdA9mOHLFw9Hw4Joanm6JUDCCqrIRiwYQM1PjywFEAcaQD2DIC+hP+zha0RIN8Hy5bVszP4AJGwwWZmtQ4BUM2bwhqxXpwtw/nBtAa3REI1JWgOaxmaMUL7THuTNuY2nF3Q0Ocaea+PMBaI3xPKYv9mMyJjz5ACnkq0IYgTpH8f3laR5YW0vUDr36T8rn6HiOqK0wVSjs0qpu+oWwFbNqI6R7/uSPl56rZCurAyUAQqudHc/TPZqjvKWttRQxu6jCrFw2SpMdvErisSdKKyeYv4hUCwGidphs9nMjoLRGfJ9IaxvZkSEwBu3U1TN6rIto22bqO++jMD6GchyUZS1piLYws4ZoC+U4+FOTRO5oIbnjUYa7fsCKxrB1bQTv38v0Hficjr3fcP61l0n1/IWy7RW5EQFhexf4RJpb2HbwW4a7jnDj3CncZCO2BD5uspH+Q/vQjk3nwcNMnj5Luvfvom7LvRsCIkRbN9Cw4yGGu37iwifv4yaSSBCEd0MApS3yNyao23IPzbueJ7J+IwtvhecOpSA7MsBw149MnjuFnUjOXAFmX88Dz8NPp0Dftu9dPABhwbKiUex43S3X8/8Bf0lAQ9CsYm8AAAAASUVORK5CYIKJUE5HDQoaCgAAAA1JSERSAAAAMAAAADAIBgAAAFcC+YcAAAcXSURBVHic7ZpdbFxHFcd/Z+be9frurtdx1q7Ih1MpwtCkUaACUaRAQmgoIEERqA8gEEgVDYqg9IFKSAj1Az+UiISKtyCBlAJVi8QDTSmEIoJEgiAtolUb4uLWiRIrBDtO1vbe9X7cmeHhbhyvvWuv43UsJP7S1Wr3zt79/8+cmXPmnJVsbqujCURrEAFrmw1ZXYgCHM5acI1peo2/p3DOEU1NYqMI5TUctroQsNUIUQovnUE8D2fMgmELmInWRGEBnJD7yD569+wjM7AjfkDTuWo/BEd4fpiJU39i7PcvUp3K43d1LxAhc11ItKY6lSe74/3c8ej3yO16L+Vxy8y5c9hKGZSC1RYhgHOIKJL9W+jclGTq7EWGBh9j7A8v4HdlY5eaL+A6+dyufbzv6Z9TuTzOyOEfkD99iupkHoyJ18MtEoDErpO+cye3H3iInl3bee3h73LhZ0fwu9fNzoRkc1udKEVUDOnatpMP/voFpl87wxsHvko1fxWdziBaIyKrzr1OA4AxmGIIohh4/Ek2fflT/P0rD3Hp+efws7E7xTMggimGfOCXLxFsyPHK5z6DKxVRQRArvZXOP1+M1rjIYMICO44cJXXHNk7uu5uoGCKehxKlMYVpenfvI7drOyOHnyLKT6BSKVwUrSl5ILaypxFPM3L4IMnb0my8/0tE4TSiNAol2KhKbs+9VMaq5E//GS+dicnPh0i8kEXFr+24rj8LWVSEClIU336TyVfP0/vRT6I8H5zFw1qU55Me2EZ4/iJRPo8KgoaWd1HUWFgbIL6PKNX8vgi2UiYcep3uD+/FS2ewUXQjDojn4SrleIsSWSDAWYvXlSWRW48zLGaw5SHecKiMj2PCQm02mo112HIZUbUMgbmBzDlQjVmJ1kTTU/R94tMMPPEo1byJ04x28LcGL6M5+61HuHL8GF52YbCqJ1PPcRG5/xv4v4C1RtvSTGfNTaUZzpp4U7jJeNM2AV5GIzexrp3VeBkQP3FTIlYuwDnEE0aP/oLypVGUn8Ath4izSEIRDp9FJZN1mWYraM8MKBg79ium/vEyOpXCmWWQqGVuOgiQRMeyZ6F9LtS9jkSuDx0ES1rRWQfOzvus+bFx0d9d9jeaIJrMU5kYR8+kmgsQAetQySSqY/nWboSVCxDBGej/2sNEVycQz2++BpxFB4qx3xzn2skTsbutsGDQnhmwkNv7oSV3IWfA74HwX+eZ+OPv0On0in+6fS5UWNqHnTEgGlsuL8hpbhZtE7BYKlw3Tuu2kYdbHImdWVnUbYT2R2JH07OCizR+D7M7kBJwzca61jKTNgay41SujCG+39TCzjp0IBT++TokOggrNt6x5g0XAU8JfgtVnPakEloYPXokjsTBYnEAsEAySSaTYmOg0LKQggHGZwxXyxZvCRFtdKEs/rqexQUAWiBfiti93ue5e/qYLJq6g2DVQXdKM3j6Gt9/Y4reDkW0iIK2LuJ4kZpFBTgB6xxJLYwXIna/+B9E4slRAldLluc/1kc6oWglpVqDsvMNGAdXSmZWgBYYL1mqxrVcM1hTAQL4SuoEJGrvW8WaCoCY9FwBWi2vYrOmAirGcXnGQk2AJ1AqGWYi1/LpdHkCnKstUnPDTM6Bna0nLwtbenye2ZOb3YUECCPH3RuSvHKl0tJUtC7AOcT38bIaZ+ecfx1Igrha1oIIB3QoYWQ64idDBbrm7KEO6PSEZ0eK/GWsTMoTltqIWhLgrEUHKSZf/itnDjyCrbq6hEwUzIxeQPyOJc/D1kFSC8OTEV8/OYGZE4ilJkOJkPKEQAt2CZvUCxDVOFOsWb906SLhyDBSVzuN3UcFAaK9lhI1B/gKejubtKzig9tC8ter43UClMKaiOL5t+nbe29cWjdRrR825wnOIX4CvyPZmNQyz7QOiJr5R4PHOOcQ7RFsfTelS6OYsIDuDFA4EFFMnDpB56ZO0ne+B1MMGxdvry/iBteqNkKUwpXLdGzcTPdd72Li1AlMuQxao5yJ8DIZxl46xtTZy9x+4BsgChdV21aBXil5EUU0PUX/AwewVhh99qdx9cOYuDYqyiMqhgwNfoeeXdsZePxJTCGcrdeL1mtz1SxfGR+jf/836X/wPoYPHaIw8hY62Rm7dV2bdTJP/xf3s/NHTzD2279x7ocHKb71JrZSufW9MhHE80hu2MzmB/bT/+BnGT58lKHBb6ODzGxdqb7RrRTVyWu8477Ps+2xQTpu62Lq1REKQ2fizkgbz7KLwdW4pLYOkL1rO9bC8KGDnPvxU+jOoK6DJPP/7HF9JhLr+9h0/xfou+fjBFveCepWZx2O0r8vMHHqBBefeZriuWG8bHft1g3KCwRALMJWKkRhAeV5eOlMWysJrcKEIaZcQgcBOtnZsPXUUEB8R+L0wFnsKnUml0K8kHUtxjQOGv8F5s41K3WlNrUAAAAASUVORK5CYIKJUE5HDQoaCgAAAA1JSERSAAAAQAAAAEAIBgAAAKppcd4AAAlASURBVHic7Zt7jFxVHcc/v3PunTvPnaXststugMC2YEHSWhqqMSEagoE0KSCJEKx/CEZC5B8DJpgahRisBIJmTVQ0ouKjaFVADJSm1iAQUZEI4SWUtnSXpfvoPmdnZ+beOcc/7uyw233N7M5MdwPf5Pxxd++9c873/t7ndyTd0mmpBiJV3d5w2OqW41RykygFIlhjsYEPVPcjDYMFcVxEK7BgTXHRRxYmQARRimJ2gmI+j/Y8ImtaVq4UCPgjw/hjEyjt4KRSAFhj5n1kXgJEaUwhj5+dIH3Rx2jf8VlSF2wi0bkR5borUwjEkn3nEJm3Xqdv3+MMPvs0CDiJJLY4tzTIXDZAtEOQGSPa1kHnV26j/Zrr8NoEk4d8XxGTzyGi6r6eamCxiCi8dVF0EooZ6D/wNG//4H6GX/wHbvq0OUmYRYBoB3/kBM0Xf4LNXT8ldeE6Jt4cYWD/Uww/d5BcTw/FbAaUrtrg1BvKcfDO6KBp81Zar9xOess5+MNFXvvmLo79+idETm+dRcIMAkRpgolx0pu2suWBh4id2cx7e/dxtOtecr09iNYoNwJKEerACrMF1mL8AjYIcJIpOnbexFm33IJy4JU7vk73ww/ipptnkPA+ASJY3ydyegvbfrePxPoWen7xJw7dvQvleahoDIzBrrCvfjJEJFxLsYg/dIL2G25k/Td2IUr4zxd3MvDMAdxkuuwhyoosShFkM3TeegepC1p4b+9+Dt29C51MoSIeNghCa2rtih7WmPALixBZu47ePT/nyH33o2Ow4Wt34cQSWBMwJb0hAUoRZCdIX7SF9qt3MPHWCEe77kF5HiKyoBtZsbAWGwRETm+hd8+DDB74N2u2ddK2/Vr8sTFEa6BEgIjC5HO07/gc0bYIA/v3k+vtQUVjq3Px01FSh77H/4gNDB3Xfh4di8EMFbAG7UVJXbCJYg6GnzsYMrTaF08YBKlYnNEX/0XuXZ9k53l4reswfgFEcBDB+AUia1pIdm6g0J8n19ONciOLGzylQqPTaFQY5ob3WpTjEIyMkD3yFqd9/ELiZ59LrrcHx4tOiwRFEDeCyRuCiUzJ1S0AEcxkFpPLN9YbWhDHQSeTlcchIphCnmAig7iCaKf87MxQ2Npy/L/g+5SiODlJ8yWfpPmSiynmTEMiQ2stKiLkunvpf+JRlOtW7pan1mWZQVxF2eBcLzO5kIBzbv8C/hCIXtKbqoI14KRg6O+v0/fY7yESWXY0ujQCoKwC/lARf7hYdiv1hDUG6yuC8bGaZaRLJwBCI6g1omkIAYggWiGqdr+1slK6U4APCTjVEzjVWJ4NqATW1iyDtMZgDTWtQ9SdAOUJoqUmc7ZG4SRAedHlv6yEuhFgrUV5wtGu7zP+6kvoWLwmiZVoCEbHkYhXE8mqnwRYiyhh/JWXGHrmrzipptpkljZ0uSoarYkq1F8F4nGcVBonmaxZam2hZplq3QkQwhJVeVSBcgWqjqgvAQLB+Bj+0CDWL1QnARZ0IoG4bl1JqBsBIoL1oe2a60hv3YaKeBUvxFqLcoX+J/5CrvsdVKSC2sQSUT8JKBGw7prtiENVO0nWgE7A2Msvk337TZRXOXnVogEqUCxNvnL9t8Zg8grr+3Xfh6y/EVxK5lbK+hqxCfuBzwU+8ATUPxkyBlvlPmI9kp750IBIUIX1wqq8gEInwurv6g2ErEW0MP7K/whGR6oqmVljUFGFPzoSlsCFhZW1JGBFU33fRn2zwYhwtOsehp45iFtlMmQtaM9DvCiDkwFmkZUJkHIFV0lVJDSgHhDFSSTR8UTVyVDRGOJauHVTmrgjKJHZKiGCsRYjsPdwlkNjPlEtixI2hcZUhIwpj0ohgLEQ1cKuzWlwBHw7jy0ViCteGCjw6nCBmK7c4NafgGXCNxbfWB54aYw7/ztKa1QRlHjUAqO+4dNnxHjo8rXki9UbzBVPAICrBN/ASN4Q0TKDgJG8IRtYHCVL2qJcFQQAKAmJcKeVFXTpb84yIuZVQwCU9jV539WdfL0UfBgKn+oJVAohVIOpwRzXS8GqIMBYmCxaJguGUc0MIzhWMGR8S9EuTRWWR8A0H39y7l6rhMZRggJu3pjisvYYkWkNqgIEFpojCq0hsgSFXjoB1qIiHk5KYX01q0GiFgnN1AKfeneSNVGNd1J3riWMj4bzhueOZRnKGxxpRChsLeK6THYfZejpFwgyZlZbzckJTbUkWEIRz/iG6/82WBLxud8R2gfB00LcqTwMhvkIWGSy1hh0Ikn/E49w/LG981aurCVss/WiS67qCpBwZdrVAvNazA7MMYcZXWJTQxx38ZlZi7gRnIi38G012MGp5ovOByl1swAzOuAU1iKOgz8yTPadw3jrPLz2Dkwhv3gP4EmJzlzjlEMEEwQ4TWni55xPYaBArucYUtprCFtllSaYyJA59AY6IaQ3b8UGwco9GlMFRASTz5E4fyPRjgTZnl5yfb3hqRdjplplQRzN8ScfpZiB1iuvwkmmyl3XqxpKYf0CrVdcjdus6Nv3CP7YCKJDNVcQtp26ySZOPHuQ/oPPk95yJh07v4Q/dKIx3V91gnJd/BODrLn0ctp2XM74GyP07P0VTjw5+7wAACIc6voO/lCRs275Mu033EhhoP99A7IapKHUESqOQ2FwgKYt29hw57dxmlwO//B75Hq7ww6TkkeYeWRGa/zRYc68/iY++t3dmAAO33cv7+35JbYYoGJxxHHCjU9W3IEZrLXYYnioy/o+ay69jA137ibR2cyRHz/Ma3fdho7FZ7jDOQ5NaQonBjhr581cuHs3OgqDB56n7/FHGXvxn/ijo2UPsWJOzwilKrTGSaVJnL+RtVdsZ+2Oq3Cb4MgDe3j1W1/FiSdDKV6IAJiShBFaP/UZzrt9F6dt+wg2gFzvJNnDb1OspJu80bAW0Q7xc9cTbU/jNEPmjUEO/6iLY7/5GToam7V4mIcACEkIxkfRsSRt26+m49rrSa7fiLc2gbgsrwpRR+QHfSa7j9D35J/p+cNvw3MBTenwn3OI7LwEQEiCDQL88XF0LIbX2kr87PXls7krDaJg8t1j5I4fxx8bxYknUJ4376lRWISA8k1alzq1fUwh35A9u6XAAsqNoCKR8pwXm2tF2eAUg+K6Yey/0sz/NEwteqGvPh3/B2BJaZAM25tVAAAAAElFTkSuQmCCiVBORw0KGgoAAAANSUhEUgAAAIAAAACACAYAAADDPmHLAAAQUklEQVR4nO2de2xk1X3HP79zzr13xvZ4vLt+dfGG5iEVQZrApqIVaXgL2BKlCxRB2oQE1CZNCFUJlP4RFBoFtREqDXk0QLWsoqCqTUVTStMFBIuA7IZseJiWlEjpYwlvr23ssT32zH2c0z/ueHl4vWvPjsczO/cjzR/e9dxzfX/f8zu/c87vd64U+9/ryOhY1HrfQMb6kgmgwzFr3oLImjdxTOPWdoReEwGIUqAUOLBhdS2a6AwciNGI8cA5nE0aLojGCUAEUQobRSTzZZJKFdGK3PBmRCmySHOVOIdoIZouEU5OIEZjuguI56XPM0ka0szRC0AE0RpbqRBX5vF6+9h09gX0bT0Zl3gcd9HHUYEHtgF320E4a9F5xdRTo0w/8wQow2v3/QOVsXFsdQGvtwgiRy0EOZppYNrjQ+K5WfIjx7Pl8is57uKP4w/04/WCSyCeg6z714GAs6DzoHNgI4im5ig99wz7d9zO5J7HcdbiFQo45+oeGuoTQC2wS8plgsFhtlz+aUYu+xT5kSJJGWzksNV07FdBgBjJRFAnSTWExIJWKN9H5wEHE3se5YWddzK593GU8VBBUJc3WLUARCmcTYjLZYa3fYyTvnoLwUAfcRmShRARQeU8dDeIgbgEUWkWUZpMBatEFMFQHtFgKxCXHS6OEG3wCgoUvL7rIf7rxmuJZqcxXd2rFsGqBCBKYeMIZy0fvO0uhs47F1uFpJLelOkRlA8LL5UY+9d7EB0z8x8/Y2b0SVS+C2wWCKwIAaxDfJ+hj16MszGF929l01kfRnkQz4KtxojW6C4hmp3huRu+yNj9P8AfGMJF0cqbWqkAUuPHuDjmlO/czdC2MwgnLDiHCjTKg8kf7aX09D4mHrqfyosv4JxDBQEqyK35fPaYxDmS+TLOWlQQsPH0s+j70Fb6z7uEYLhIUna4JEHnDZAwevUfMfbAvasSwcoEoBRu0fi3383Q+WcQjsegBN2lSWZn+O+bb2L8oQdwcYzpKaD8oKZki7M2WxCqE1E6DQgTS1Kew8UR/sAg7/vSzfSfeybJvMXFDuUrxDhGP/eHjD14L37/EC4+sgiOLABJw1Ebx5zy7e8xdEHN+AL+JsP4wz/mF1/+M5LZGVS+C9G1OWrW4xvO4gKbrVaw1Sqbzj6PE752K1iFDe3bRHDg4R/ibdiIi+PDXnMFewFCXC7zwdu+m7r9txh/Yvdefn7950gWyujubnA2bTAz/prgbPp8xXiYQi/jD/yQn19/LaJTD2BDi4vhlDt2sPG0M4mmpxCtD3vNwwtAa5LyLMPbtjN83lmEE6nbXzT+89d9HtEGHeQatjKVsQJcOvb7A4NM7L6f56/701QEOY2tWsQo3nfNDZientQuhxl+lxeACC4M8QeHOemrXycJ016tuxTjD+/h+euuRrRBGZMZf51wUYTfP8DE7l08f/21uGgB8YR4OmLgrN/g3Z/54hG9wLICEK2J52bYcvlV+AM57EKMCjTxbJVffPnP0y97XhrgZawbLooIBoY4sOsHvHz33+NvSo0dTVlGLv0D8luOTxfllvEChxaACLZSSZd3L7uCpCyIVihP+J+bbySeLaFzmdtvFWwUEQxt5pXv3cnkY6OYokdSiciP9DFy2aeJ52aW9QKHFIAoRVKZZ+TyK8mPFLGVEN2jmXz8J0w89O/o/OpXnDLWEOcQEeK5WV6663ZcFKOMIZ6DLZd9iq7j37OsF1gqABFsFGN6+xi5+BPE5fTflA+lp36MjSNEZ4lErYazFtPdw9zzz1EdL6MCja1E5N9VZOiC7cSlQ8cChxZAZZ6Np/42fn8PLkpQgcfCSyUmHr4f09Ob9f5WxDnEGKKZKcbu+z4q5wCHrcKm087A9G1Ip+jv8AJLBCBKYasL9G09Fa/oYcMI0w1j993Dwov7UUGQzfNbFGctOsgz9i/fpzpWRgU+dgE2bP1NTG/fITvuoWMAnU7tXLL4czoryAzfBoikewd+Lt18FbDVCqLNIe2n3vllW60SDG/muIt+n3gOlO8RlUJmnn0aFeSyaV+LI1oTz5aYefYpVA5sNSS3uYfN2y8nLs8uiQOWmQamyQeLSYnRTMjM6FPZrl6r4xzieUSlaUqjT6ICDm7EpfZcarvlcwIXf9mBKEHl843P8O20HcJmdB7nEKVR+fzb82+WaXvlSaFr4PqPtFN1rCFKQJoxhXYrttfaF4YshwimUCBNGugABGwY4sKwpTxf8wUgcjBp5AM778EUunCxa6mH0mhcHONvNPzyzu/yyztvw9/Y3zLeb909gCnkcDHHtCNwscP0CuK33hrK+gkAamlmriM8gItNmujfYqyrABCpfTimBXDw72xBsl2dDicTQIeTCaDDyQTQ4WQC6HAyAXQ4mQA6nEwAHU4mgA4nE0CHkwmgw1nfvYAG4KxNkx+Elj2BxiUxLqYl8ynbWwAuLVbVeYWzrbujbGODPwC6qyvbDm4Ui+foTe97mul9e9FdXS3ZwwBYvNefPo3K5VvqPttWAAcf6r69/O8tf0EwMIRdwZEo64EgOOvQ+Ty6xQ7Lal8BQG0I6MIfGMTb2I9LWiPNagm1Ao2D5yW1EO0tAN48NiUNtFpUAC1MNg3scDIBdDiZADqcTAAdTtsHgcBbsoubuBTUYgs69dL+AqgdTrn4aRbNq/NbW9pbAA6UH2B6i5hCb/ME0KJ1fvXQtgIQY4hmYfj3PsHQxy5tmiFauc6vHtpWAEDNA3hIzmteky1c51cP7S0AAJcapWnNtXCdXz20vwCgueNwC9f51UP7h7EZR0UmgA4nE0CHkwmgw8kE0OFkAuhwMgF0OJkAOpxMAB1OJoAOp+2XgptdGtbKZV710N4CWIfSsFYu86qHthXAupWGtXCZVz20rQDWqzRsuTKvRnufZvmW9hUArE9p2DJlXrF787+OxniL31eAasKY1t4CoDVKwwQoeo3KE0hlFCaOqnVrHte0vQDWEwEiB0Vf8fjvDNHjKdxRBoaJBZNXfOvZEl8ZLTGY18R27QaETAANYNED5H0BJ/X7bgcusYivyGlpShyQCaBBxC6dFToL5dCy2kjAAYESlIBxsIad/m1kAmggooRymHDGrjFKkcPIymRgFBxYsNx0cpFrTi429ayjTAANxgGlyDEVWnyBlawSeEqYCi2VpPkLS5kA1gAj4Akr9wC132/GtG9J281v8tjHveOz0t9fD7LdwA4nE0CHkwmgw8kE0OFkAuhwMgF0OJkAOpxMAB1OthC0BhgFRskq9gIEr7YR1GwyATQYR7qxMxVavFUIYH4+oRy5pr/4IhNAI3GOQAk3nVykkrgV92glMBc5zt6cw0YWZZo3MmcCaBCJhThxaCVc88Hi6rJEF5MJY8dCmOBpyfIB2gkB+vIKfFXfzs6i2zdCXmvo0nR7zRHBugpgMZHTxW7VCZWtUKGzmL1bTRx/NVrCr6VxHU0sZ51DGcWPX6vQbQS7xsUn6ycAEbyNRbxecBGrfmqtUqGjBKrW8ZVnS42L3Rx0GaGnCV6g+QJwaW+3YZUX7/hblO+lR+6tttu0UIWOAIN5RSPLQ6xzx+4QIErhwpAX77gtTaOu47m12ouYYgst++LCw7DOQ0B//Z2mhV/E1E4cWgDOLX2oazDOtuxbvo4F3movt3ywvHTFwTlE69StLl5DBPH9Y6IcuiMQVbNX7UdNzZ5L7fd2ATiXHsNemmLqqSdQObDVkGAwz+BHLyaen0O0bsafkFEPIthqldzwZoYuvIi4nJ6mHk2HNXsuDZaXeADxPKLpN5h65ieoHLgkQbTgbLLugVbGChCFiyJEe+BAtCaeiZh++glUkFviBZYZAgyiNC62iNbYKvT++lZUkMsCrhZGlCJZKNN9womYXg+XJOnbTaIKKsivYAgg7fGmu8Cr9/0T0VSC+D5J2bLpzNPYePo5JHOz2TDQojhrUUHAyJWfR+d8bBRjCo7XH7yPyqsvooKlL7k4tAfwPKpjr1J6bhSdd9g4QnmO4od+Kz2FQ2V5JC2HCC6OCfoHKZz0fpJ5UMbDVoQ39v0IZbyVeQBIXYmtVtm/45vgBNGGeFYYOG9behRLZSETQYuxGLwPXHgJXlFjwxDdo5jcu48Du3ehC73pkPAODmlFlyR4vX1M7nmEyT0/xRQUNowIhjbwvi/9JbZaRY6ht2a0O2IMcWmajR85hy1XXkFcJu2gDvbv/BbiWNZey3fj2irb/p3fQJSkAca8pf/c09l09jbCyXHEa97LmjIOjSiFDUNEG9712S9guvPYaozXZ5jc+ySTe3ZjCsVD9n44jABckqALvUzufZTXdz2C7lK4xJIsJJzwtVvoP/dCwolMBOuJKIWtnY904tfvYMOpJxOVYpSniGfn+b87/5oj7U8ffiB3DmU8fnbjF4hnq+i8SbdureLEW2+j/5xthBPjKN8/fCsZDUeMSY0fx5x46+30n3Ma0XSMs5Zg2LB/x12M774fU+xbtvfDSgQQBCSzM/znDVcDFuUrbGhxiaQiOHcbCy+/BIpsetgMRBDPI54p4aKYE/8mNX44GYODYNDntXsf44Wd38QfGMId4exEKfa/94gL/KI14eQ4Qxds55Tv7MDFgg0tKqdxUcIrd+/g5e/dRTw3g+npQbR5c8Eo2z84OkQO7s8sHokXlabY+JGzOf6zn6Xv1FOJplPj+wOGsQceY/TqTyKL+wFHWLhbkQAgXSIOx8cYOn87p9y+AxeDrVrEaPx+YfKxfbx01w7mfv4z4ukpVC6fqlVrxJgjjkUZb+HgiZMuDfCUIpkvI36OYGATgxdewshVf4zpFqLpBGcTgkE/Nf7nP4kYgzJmRau2KxYALIrgAEPn/y6n3HEXygjhVAQCXtHDRVAdf4MD//bPvP6Df8TZmHh2jrg0lS4tt2HCxPpQ6/VKEfzKZmwU0vNrJ7HlyqvoOfEDeH3dxGWw1QjlaYJhxWv3PsroFz6FaL1i48MqBQAgxiOammDTaWfz3mv+hMGzPkw4BUklTJUXKHQOqmNllA8zzz5HafRJdL79D1ZuHgLOojyfwY9ejGiF6S2gcpDMgw0jRGn8DYpoxrJ/xzd4YeftuDBE+f6qnvOqBQC1VafpNzA9Rd79mWsYufQK8iN9xHOQVCLAoQMf50DnQNWWoLMRYBWkToBkjtoBkg4bRYgxeIX0ePyJvY+y/847OLB7F8HAEIvCWVUz9QgA0sDQJQnR9BT5Lcczctmn2XLZFXS9q4+kCslCGr8k1TCtmsisv3ocqCBATPrwTAFsJTX8Czv/jsk9j4GzeMU+bFTfSel1CyD9ttS2iyvEs7N0/ep7GL5gOxtPO50NW08mqQr5zT3pZDMb/leNaIimIJqZwUUxrz/4fd74yT4OPPIgOPAKBaCWq1FvG0clgINXEURpbFglKk3hFTfgFXsR7bN5++Vp6ncmgNVRS3ufemqUqaefQAc5Fl55GeV5mEIh3f07zALPSmmMAA5eLd05dHGU3pxzxHOzWQBQJ86BDoJ0Sl1blMO6o+rx76SxaeHOHVx5EpNe2u8faGgTHYe1B4+gX4v3IaxdXcAa3nRG48iyOjqcTAAdzv8D1qJ45/FMiF4AAAAASUVORK5CYIKJUE5HDQoaCgAAAA1JSERSAAABAAAAAQAIBgAAAFxyqGYAAAXsSURBVHic7d07blRZFIbR45ZzEiLUGUyhZFKkGkQPkUFYIqXkKRCingadULLVTeP7OK9b/1oxKh0k76923VuPuzdv3/8oQKQ/Rh8AGEcAIJgAQDABgGACAMEEAIIJAAQTAAgmABBMACCYAECw+9EH2OvT12+jj0C4Lx8/jD7CZndH+jCQYecojhKF6QNg6Dm6mWMwbQAMPrdmxhBMFwCDz62bKQTTBMDgk2aGEAwPgMEn3cgQDH0fgOGHsXMwZAOo/R++nE9VHw+Wenh8qvp4vbeB7gHYO/yGndntjULPCHQNwJ7hN/gczZ4Q9IpAtwBsGX5Dz63YEoMeEegSgLXDb/C5VWtD0DoCze8CGH54tvbvu/UdgqYbwJrDG3zSrNkGWm0CzTYAww+/t+bvvtUm0CQAhh+WGR2B6gEw/LDOyAgMeyuw4Ydno+ahagCW1snww38tnYuaW0C1ABh+2K93BLq+BDD88Lqec1IlAEtqZPhhuSXzUmML8LsAEGx3ADz7Qxs9toDmG4Dhh+1az8+uAPhKLxhvzxw23QA8+8N+LedocwA8+8M8ts6juwAQrFkArP9QT6t52hQA6z/MZ8tcNtkAPPtDfS3myjUACCYAEEwAINjqALgACPNaO5/VNwAXAKGd2vPlJQAEEwAIJgAQTAAgmABAMAGAYAIAwQQAgt2PPsAoa36bnWPyprTX2QAgmABAMAGAYAIAwQQAggkABBMACCYAEEwAIJgAQDABgGACAMEEAIIJAAQTAAgmABBMACCYAEAwAYBgAgDBBACCCQAEEwAIJgAQTAAgmABAMAGAYAIAwQQAggkABBMACCYAEEwAIJgAQLD70QeglMv5NPoIhLIBQDABgGACAMEEAIIJAAQTAAgmABBMACCYAEAwAYBgAgDBBACCCQAEEwAIJgAQTAAgmABAMAGAYAIAwQQAgvlS0Ak8PD6NPkJXvgR1HjYACCYAEEwAIJgAQDABgGACAMEEAIIJAAQTAAgmABBMACCYAEAwAYBgAgDBBACCCQAEEwAIJgAQTAAgmABAMAGAYAIAwQQAggkABBMACCYAEEwAIJgAQDABgGACAMEEAIIJAAS7H30ASrmcT6OPQCgbAAQTAAgmABBMACCYAEAwAYBgbgMylb//+nP0EV717vP30UeoxgYAwQQAggkABBMACCYAEMxdAA5hxJX3I9yR2MsGAMEEAIIJAAQTAAgmABBMACCYAEAwAYBgAgDBBACCCQAEEwAIJgAQTAAgmABAMAGAYAIAwQQAggkABBMACCYAEEwAIJgAQDABgGACAMEEAIIJAAQTAAjmx0E5hIQf6hzBBgDBBACCCQAEEwAIJgAQzF0ApvLu8/fRR4hiA4BgAgDBBACCxV4DuJxPo48Aw9kAIJgAQDABgGACAMEEAIIJAAQTAAgmABCsegAeHp9qPyTwU+35Wh2ALx8/VD0AUM/a+fQSAIIJAAQTAAjWJAAuBEJ9LeZqUwBcCIT5bJnLZi8BbAFQT6t5cg0Agm0OgJcBMI+t89h0A/AyAPZrOUe7AmALgPH2zGHzawC2ANiu9fzsDsCS+ogArLdkbvZu4e4CQLAqAbAFQF09nv1L6bwBiAC8ruecVAvA0hqJAPy/pfNR6w5c1Q1ABGC73sNfysCLgCIAz0bNQ/UArKmTCMC6Oaj95rsmG4AIwDIjh7+Uhi8BRAB+b/Twl1LK3Zu37380eeSfPn39turfX86nRieBOax9wmv5mZvmFwHXHt42wC2bafhL6bABXK3dBEqxDXA7tjyx9fi0bbcAlLItAldiwNHs2WZ7fdS+awBK2ReBUoSA+e19Gdvzeza6B6CU/RH4N1FglNrXrHp/yc6QAFzVDgEc1ahv1xr6fQC+UgzGzsHQDeAl2wBpZngCnCYAV0LArZth8K+mC8CVEHBrZhr8q2kDcCUEHN2Mg381fQBeEgOOYuahf+lQAfgVUWC0owz7rxw+AMB2fhcAggkABBMACCYAEEwAIJgAQDABgGACAMEEAIIJAAT7B8BPSdJpe/bZAAAAAElFTkSuQmCC"""
)

def ensure_directories() -> None:
    """Create runtime directories expected by the app."""
    STORAGE_DIR.mkdir(exist_ok=True)
    UPLOADS_DIR.mkdir(parents=True, exist_ok=True)
    ATTACHMENTS_DIR.mkdir(parents=True, exist_ok=True)
    STATIC_DIR.mkdir(exist_ok=True)
    TEMPLATES_DIR.mkdir(exist_ok=True)
    favicon_path = STATIC_DIR / "favicon.ico"
    if not favicon_path.exists():
        favicon_path.write_bytes(FAVICON_BYTES)


def get_connection() -> sqlite3.Connection:
    """Return a SQLite connection with row dictionaries enabled."""
    connection = sqlite3.connect(DATABASE_PATH)
    connection.row_factory = sqlite3.Row
    return connection


def init_db() -> None:
    """Create the SQLite database and metadata tables."""
    ensure_directories()
    with closing(get_connection()) as connection:
        connection.execute("""CREATE TABLE IF NOT EXISTS app_metadata (key TEXT PRIMARY KEY, value TEXT NOT NULL)""")
        connection.execute("""CREATE TABLE IF NOT EXISTS settings (key TEXT PRIMARY KEY, value TEXT NOT NULL)""")
        connection.execute(
            """
            CREATE TABLE IF NOT EXISTS uploads (
                id TEXT PRIMARY KEY,
                original_filename TEXT NOT NULL,
                stored_filename TEXT NOT NULL UNIQUE,
                content_type TEXT,
                size_bytes INTEGER NOT NULL,
                uploaded_at TEXT NOT NULL
            )
            """
        )
        connection.execute(
            """
            CREATE TABLE IF NOT EXISTS chat_messages (
                id TEXT PRIMARY KEY,
                nickname TEXT NOT NULL,
                message TEXT NOT NULL,
                kind TEXT NOT NULL DEFAULT 'message',
                created_at TEXT NOT NULL
            )
            """
        )
        connection.execute(
            """
            CREATE TABLE IF NOT EXISTS board_threads (
                id TEXT PRIMARY KEY,
                title TEXT NOT NULL,
                nickname TEXT NOT NULL,
                body_markdown TEXT NOT NULL,
                body_html TEXT NOT NULL,
                attachment_filename TEXT,
                attachment_original_filename TEXT,
                attachment_content_type TEXT,
                created_at TEXT NOT NULL
            )
            """
        )
        connection.execute(
            """
            CREATE TABLE IF NOT EXISTS board_replies (
                id TEXT PRIMARY KEY,
                thread_id TEXT NOT NULL,
                nickname TEXT NOT NULL,
                body_markdown TEXT NOT NULL,
                body_html TEXT NOT NULL,
                attachment_filename TEXT,
                attachment_original_filename TEXT,
                attachment_content_type TEXT,
                created_at TEXT NOT NULL,
                FOREIGN KEY(thread_id) REFERENCES board_threads(id)
            )
            """
        )
        connection.execute(
            """
            CREATE VIRTUAL TABLE IF NOT EXISTS search_index USING fts5(
                source_type UNINDEXED,
                source_id UNINDEXED,
                title,
                body,
                nickname UNINDEXED,
                created_at UNINDEXED
            )
            """
        )
        connection.execute("INSERT OR REPLACE INTO app_metadata (key, value) VALUES (?, ?)", ("version", APP_VERSION))
        connection.execute("INSERT OR IGNORE INTO settings (key, value) VALUES (?, ?)", ("accent_color", validate_accent_color(DEFAULT_ACCENT_COLOR)))
        connection.commit()


def validate_accent_color(value: str | None) -> str:
    """Return a safe six-digit hex accent color for CSS variables."""
    color = (value or "").strip()
    if not ACCENT_COLOR_PATTERN.fullmatch(color):
        raise ValueError("Accent color must be a six-digit hex value like #32cd32.")
    return color.lower()


def get_setting(key: str, default: str = "") -> str:
    """Read a small persisted application setting from SQLite."""
    try:
        with closing(get_connection()) as connection:
            row = connection.execute("SELECT value FROM settings WHERE key = ?", (key,)).fetchone()
    except sqlite3.OperationalError:
        return default
    return str(row["value"]) if row else default


def set_setting(key: str, value: str) -> None:
    """Persist a small application setting."""
    with closing(get_connection()) as connection:
        connection.execute("INSERT OR REPLACE INTO settings (key, value) VALUES (?, ?)", (key, value))
        connection.commit()


def get_accent_color() -> str:
    """Return the configured accent color, falling back safely if needed."""
    try:
        return validate_accent_color(get_setting("accent_color", DEFAULT_ACCENT_COLOR))
    except ValueError:
        return validate_accent_color(DEFAULT_ACCENT_COLOR)


def set_accent_color(value: str) -> str:
    """Validate and persist the UI accent color."""
    color = validate_accent_color(value)
    set_setting("accent_color", color)
    logger.info("Updated accent color to %s", color)
    return color


def validate_megabytes(value: str | int | None, *, label: str) -> int:
    """Return a safe positive megabyte value for runtime size settings."""
    try:
        megabytes = int(str(value or "").strip())
    except ValueError as exc:
        raise ValueError(f"{label} must be a whole number of megabytes.") from exc
    if megabytes < 1 or megabytes > 10240:
        raise ValueError(f"{label} must be between 1 and 10240 MB.")
    return megabytes


def bytes_to_megabytes(size_bytes: int) -> int:
    """Convert a byte limit to whole megabytes for configuration forms."""
    return max(1, int(size_bytes) // (1024 * 1024))


def megabytes_to_bytes(megabytes: int) -> int:
    """Convert whole megabytes to bytes."""
    return int(megabytes) * 1024 * 1024


def get_size_limit(setting_key: str, default_bytes: int) -> int:
    """Read a persisted megabyte limit, falling back to the current constant."""
    raw_value = get_setting(setting_key, "")
    if not raw_value:
        return int(default_bytes)
    try:
        return megabytes_to_bytes(validate_megabytes(raw_value, label=setting_key))
    except ValueError:
        return int(default_bytes)


def get_max_upload_bytes() -> int:
    """Return the active maximum upload size in bytes."""
    return get_size_limit("max_upload_mb", MAX_UPLOAD_BYTES)


def get_max_attachment_bytes() -> int:
    """Return the active maximum board attachment size in bytes."""
    return get_size_limit("max_attachment_mb", MAX_ATTACHMENT_BYTES)


def set_runtime_configuration(*, max_upload_mb: str | int, max_attachment_mb: str | int, accent_color: str) -> dict[str, Any]:
    """Validate and persist user-editable runtime settings."""
    upload_mb = validate_megabytes(max_upload_mb, label="Max upload")
    attachment_mb = validate_megabytes(max_attachment_mb, label="Max attachment")
    color = validate_accent_color(accent_color)
    set_setting("max_upload_mb", str(upload_mb))
    set_setting("max_attachment_mb", str(attachment_mb))
    set_setting("accent_color", color)
    logger.info("Updated configuration: max_upload=%s MB max_attachment=%s MB accent=%s", upload_mb, attachment_mb, color)
    return {"max_upload_mb": upload_mb, "max_attachment_mb": attachment_mb, "accent_color": color}


def accent_soft_rgba(color: str, alpha: float = 0.18) -> str:
    """Return a soft RGBA background color derived from a hex accent."""
    safe = validate_accent_color(color).lstrip("#")
    red, green, blue = (int(safe[index:index + 2], 16) for index in (0, 2, 4))
    return f"rgba({red}, {green}, {blue}, {alpha})"


def now_iso() -> str:
    return datetime.now(timezone.utc).isoformat(timespec="seconds")


def generate_upload_id() -> str:
    return str(uuid.uuid4())


def generate_chat_id() -> str:
    return str(uuid.uuid4())


def generate_thread_id() -> str:
    return str(uuid.uuid4())


def generate_reply_id() -> str:
    return str(uuid.uuid4())


def sanitize_filename(filename: str | None) -> str:
    """Return a safe display filename while preventing path traversal."""
    if not filename:
        raise ValueError("A filename is required.")
    name = Path(filename).name.strip().replace("\\", "")
    name = SAFE_FILENAME_PATTERN.sub("_", name)
    name = re.sub(r"\s+", " ", name).strip(" .")
    if not name or name in {".", ".."}:
        raise ValueError("Invalid filename.")
    return name[:255]


def sanitize_text(value: str | None, max_length: int) -> str:
    value = re.sub(r"\s+", " ", value or "").strip()
    if not value:
        raise ValueError("Text cannot be empty.")
    return html.escape(value[:max_length], quote=True)


def validate_nickname(nickname: str | None) -> str:
    if nickname is None:
        raise ValueError("Nickname is required.")
    return sanitize_text(nickname, MAX_NICKNAME_CHARS)


def sanitize_chat_message(message: str | None) -> str:
    if message is None:
        raise ValueError("Message is required.")
    return sanitize_text(message, MAX_CHAT_MESSAGE_CHARS)


def validate_thread_title(title: str | None) -> str:
    return sanitize_text(title, MAX_THREAD_TITLE_CHARS)


def validate_post_body(body: str | None) -> str:
    value = (body or "").strip()
    if not value:
        raise ValueError("Post body is required.")
    return value[:MAX_POST_BODY_CHARS]


def rate_limit_allows_message(nickname: str) -> bool:
    """Placeholder hook for future per-client or per-nickname rate limiting."""
    return True


def format_file_size(size_bytes: int) -> str:
    value = float(size_bytes)
    for unit in ("B", "KB", "MB", "GB"):
        if value < 1024 or unit == "GB":
            return f"{value:.1f} {unit}" if unit != "B" else f"{int(value)} B"
        value /= 1024
    return f"{size_bytes} B"  # pragma: no cover - loop always returns for known units


def render_markdown(markdown_text: str) -> str:
    """Render a small safe Markdown subset, escaping HTML first to prevent XSS."""
    raw = re.sub(r"<script.*?>.*?</script>", "", markdown_text[:MAX_POST_BODY_CHARS], flags=re.IGNORECASE | re.DOTALL)
    raw = re.sub(r"\son\w+\s*=\s*([^\s>]+|[\'\"].*?[\'\"])", "", raw, flags=re.IGNORECASE | re.DOTALL)
    raw = re.sub(r"javascript:", "", raw, flags=re.IGNORECASE)
    safe = html.escape(raw, quote=True)
    safe = re.sub(r"`([^`]+)`", r"<code>\1</code>", safe)
    safe = re.sub(r"\*\*([^*]+)\*\*", r"<strong>\1</strong>", safe)
    safe = re.sub(r"\*([^*]+)\*", r"<em>\1</em>", safe)
    safe = re.sub(r"^### (.+)$", r"<h3>\1</h3>", safe, flags=re.MULTILINE)
    safe = re.sub(r"^## (.+)$", r"<h2>\1</h2>", safe, flags=re.MULTILINE)
    safe = re.sub(r"^# (.+)$", r"<h1>\1</h1>", safe, flags=re.MULTILINE)
    blocks = [block.strip().replace("\n", "<br>") for block in safe.split("\n\n") if block.strip()]
    return "".join(block if block.startswith("<h") else f"<p>{block}</p>" for block in blocks)


def sanitize_rendered_html(rendered: str) -> str:
    """Defense-in-depth sanitizer for rendered Markdown HTML."""
    rendered = re.sub(r"<script.*?>.*?</script>", "", rendered, flags=re.IGNORECASE | re.DOTALL)
    rendered = re.sub(r"\son\w+\s*=\s*(['\"]).*?\1", "", rendered, flags=re.IGNORECASE | re.DOTALL)
    rendered = re.sub(r"javascript:", "", rendered, flags=re.IGNORECASE)
    return rendered


def render_safe_markdown(markdown_text: str) -> str:
    return sanitize_rendered_html(render_markdown(markdown_text))


def normalize_search_query(query: str | None) -> str:
    """Return a safe FTS5 query string using simple token prefix matching."""
    cleaned = re.sub(r"[^\w\s.-]+", " ", (query or "")[:MAX_SEARCH_QUERY_CHARS], flags=re.UNICODE)
    tokens = [token.strip(".-_") for token in cleaned.split() if token.strip(".-_")]
    return " ".join(f'"{token}"*' for token in tokens[:8])


def highlight_text(text: str | None, query: str | None) -> str:
    """Escape text and wrap query token matches in <mark>."""
    escaped = html.escape(text or "", quote=True)
    tokens = [re.escape(token.strip(".-_")) for token in re.sub(r"[^\w\s.-]+", " ", query or "", flags=re.UNICODE).split() if token.strip(".-_")]
    if not tokens:
        return escaped
    pattern = re.compile("(" + "|".join(tokens[:8]) + ")", re.IGNORECASE)
    return pattern.sub(r"<mark>\1</mark>", escaped)


def index_search_document(*, source_type: str, source_id: str, title: str, body: str = "", nickname: str = "", created_at: str = "") -> None:
    """Insert or replace a document in the unified SQLite FTS5 search index."""
    with closing(get_connection()) as connection:
        connection.execute("DELETE FROM search_index WHERE source_type = ? AND source_id = ?", (source_type, source_id))
        connection.execute(
            "INSERT INTO search_index (source_type, source_id, title, body, nickname, created_at) VALUES (?, ?, ?, ?, ?, ?)",
            (source_type, source_id, title, body, nickname, created_at),
        )
        connection.commit()


def search_content(query: str | None, filter_type: str = "all", limit: int = 50) -> list[dict[str, Any]]:
    """Search uploads, chat, threads, and replies through a single FTS5 index."""
    if filter_type not in SEARCH_FILTERS:
        raise ValueError("Unsupported search filter.")
    fts_query = normalize_search_query(query)
    if not fts_query:
        return []
    params: list[Any] = [fts_query]
    sql = """
        SELECT source_type, source_id, title, body, nickname, created_at, bm25(search_index) AS rank
        FROM search_index
        WHERE search_index MATCH ?
    """
    if filter_type != "all":
        sql += " AND source_type = ?"
        params.append(filter_type.rstrip("s") if filter_type != "uploads" else "upload")
        if filter_type == "threads":
            params[-1] = "thread"
        elif filter_type == "replies":
            params[-1] = "reply"
    sql += " ORDER BY rank LIMIT ?"
    params.append(limit)
    with closing(get_connection()) as connection:
        rows = connection.execute(sql, tuple(params)).fetchall()
    results = []
    for row in rows:
        item = dict(row)
        item["title_html"] = highlight_text(item.get("title"), query)
        item["body_html"] = highlight_text((item.get("body") or "")[:280], query)
        item["url"] = search_result_url(item)
        results.append(item)
    return results


def search_result_url(item: dict[str, Any]) -> str:
    if item["source_type"] == "upload":
        return f"/download/{item['source_id']}"
    if item["source_type"] == "chat":
        return "/chat"
    if item["source_type"] == "thread":
        return f"/thread/{item['source_id']}"
    if item["source_type"] == "reply":
        with closing(get_connection()) as connection:
            row = connection.execute("SELECT thread_id FROM board_replies WHERE id = ?", (item["source_id"],)).fetchone()
        return f"/thread/{row['thread_id']}" if row else "/board"
    return "/"


def insert_upload_metadata(*, upload_id: str, original_filename: str, stored_filename: str, content_type: str | None, size_bytes: int, uploaded_at: str) -> None:
    with closing(get_connection()) as connection:
        connection.execute(
            """INSERT INTO uploads (id, original_filename, stored_filename, content_type, size_bytes, uploaded_at) VALUES (?, ?, ?, ?, ?, ?)""",
            (upload_id, original_filename, stored_filename, content_type, size_bytes, uploaded_at),
        )
        connection.execute("DELETE FROM search_index WHERE source_type = ? AND source_id = ?", ("upload", upload_id))
        connection.execute(
            "INSERT INTO search_index (source_type, source_id, title, body, nickname, created_at) VALUES (?, ?, ?, ?, ?, ?)",
            ("upload", upload_id, original_filename, f"{original_filename} {content_type or ''}", "", uploaded_at),
        )
        connection.commit()


def list_uploads(limit: int | None = None) -> list[dict[str, Any]]:
    query = "SELECT * FROM uploads ORDER BY uploaded_at DESC"
    params: tuple[Any, ...] = ()
    if limit is not None:
        query += " LIMIT ?"
        params = (limit,)
    with closing(get_connection()) as connection:
        rows = connection.execute(query, params).fetchall()
    uploads = [dict(row) for row in rows]
    for item in uploads:
        item["display_size"] = format_file_size(int(item["size_bytes"]))
    return uploads


def get_upload(upload_id: str) -> dict[str, Any] | None:
    with closing(get_connection()) as connection:
        row = connection.execute("SELECT * FROM uploads WHERE id = ?", (upload_id,)).fetchone()
    return dict(row) if row else None


def insert_chat_message(*, nickname: str, message: str, kind: str = "message", message_id: str | None = None, created_at: str | None = None) -> dict[str, Any]:
    if kind not in {"message", "system"}:
        raise ValueError("Invalid chat message kind.")
    item = {"id": message_id or generate_chat_id(), "nickname": validate_nickname(nickname), "message": sanitize_chat_message(message), "kind": kind, "created_at": created_at or now_iso()}
    with closing(get_connection()) as connection:
        connection.execute("INSERT INTO chat_messages (id, nickname, message, kind, created_at) VALUES (?, ?, ?, ?, ?)", (item["id"], item["nickname"], item["message"], item["kind"], item["created_at"]))
        connection.execute("DELETE FROM search_index WHERE source_type = ? AND source_id = ?", ("chat", item["id"]))
        connection.execute(
            "INSERT INTO search_index (source_type, source_id, title, body, nickname, created_at) VALUES (?, ?, ?, ?, ?, ?)",
            ("chat", item["id"], f"Chat from {item['nickname']}", item["message"], item["nickname"], item["created_at"]),
        )
        connection.commit()
    return item


def list_chat_messages(limit: int = RECENT_CHAT_LIMIT) -> list[dict[str, Any]]:
    with closing(get_connection()) as connection:
        rows = connection.execute("SELECT * FROM (SELECT * FROM chat_messages ORDER BY created_at DESC, rowid DESC LIMIT ?) ORDER BY created_at ASC, rowid ASC", (limit,)).fetchall()
    return [dict(row) for row in rows]


def get_chat_message(message_id: str) -> dict[str, Any] | None:
    """Return a single chat message by ID for moderation review/deletion."""
    with closing(get_connection()) as connection:
        row = connection.execute("SELECT * FROM chat_messages WHERE id = ?", (message_id,)).fetchone()
    return dict(row) if row else None


def delete_chat_message(message_id: str) -> bool:
    """Delete a persisted chat message and its search index row."""
    if get_chat_message(message_id) is None:
        return False
    with closing(get_connection()) as connection:
        connection.execute("DELETE FROM chat_messages WHERE id = ?", (message_id,))
        connection.execute("DELETE FROM search_index WHERE source_type = ? AND source_id = ?", ("chat", message_id))
        connection.commit()
    logger.info("Deleted chat message %s", message_id)
    return True


class ConnectionManager:
    def __init__(self) -> None:
        self.active_connections: list[WebSocket] = []

    async def connect(self, websocket: WebSocket) -> None:
        await websocket.accept()
        self.active_connections.append(websocket)

    def disconnect(self, websocket: WebSocket) -> None:
        if websocket in self.active_connections:
            self.active_connections.remove(websocket)

    async def send_personal_json(self, websocket: WebSocket, payload: dict[str, Any]) -> None:
        await websocket.send_json(payload)

    async def broadcast(self, payload: dict[str, Any]) -> None:
        stale: list[WebSocket] = []
        for connection in list(self.active_connections):
            try:
                await connection.send_json(payload)
            except RuntimeError:
                stale.append(connection)
        for connection in stale:
            self.disconnect(connection)


manager = ConnectionManager()


def validate_upload(file: UploadFile) -> str:
    try:
        return sanitize_filename(file.filename)
    except ValueError as exc:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)) from exc


async def save_file_to_directory(file: UploadFile, directory: Path, max_bytes: int, allowed_types: set[str] | None = None) -> dict[str, Any] | None:
    original_filename = sanitize_filename(file.filename)
    if allowed_types and file.content_type not in allowed_types:
        raise HTTPException(status_code=400, detail="Only image attachments are allowed.")
    file_id = generate_upload_id()
    suffix = Path(original_filename).suffix[:32]
    stored_filename = f"{file_id}{suffix}"
    destination = (directory / stored_filename).resolve()
    if directory.resolve() not in destination.parents:
        raise HTTPException(status_code=400, detail="Invalid storage path.")  # pragma: no cover - resolved path defense
    size_bytes = 0
    try:
        with destination.open("wb") as output:
            while chunk := await file.read(UPLOAD_CHUNK_BYTES):
                size_bytes += len(chunk)
                if size_bytes > max_bytes:
                    output.close()
                    destination.unlink(missing_ok=True)
                    raise HTTPException(status_code=413, detail=f"File exceeds maximum size of {format_file_size(max_bytes)}.")
                output.write(chunk)
    finally:
        await file.close()
    if size_bytes == 0:
        destination.unlink(missing_ok=True)
        raise HTTPException(status_code=400, detail="Uploaded file is empty.")
    return {"stored_filename": stored_filename, "original_filename": original_filename, "content_type": file.content_type, "size_bytes": size_bytes}


async def save_upload(file: UploadFile) -> dict[str, Any]:
    original_filename = validate_upload(file)
    file.filename = original_filename
    saved = await save_file_to_directory(file, UPLOADS_DIR, get_max_upload_bytes())
    assert saved is not None
    upload_id = Path(saved["stored_filename"]).stem
    uploaded_at = now_iso()
    insert_upload_metadata(upload_id=upload_id, original_filename=original_filename, stored_filename=saved["stored_filename"], content_type=saved["content_type"], size_bytes=saved["size_bytes"], uploaded_at=uploaded_at)
    return {"id": upload_id, **saved, "uploaded_at": uploaded_at}


async def save_optional_attachment(file: UploadFile | None) -> dict[str, Any] | None:
    if file is None or not file.filename:
        return None
    return await save_file_to_directory(file, ATTACHMENTS_DIR, get_max_attachment_bytes(), ALLOWED_ATTACHMENT_TYPES)


def create_thread(*, title: str, nickname: str, body: str, attachment: dict[str, Any] | None = None, thread_id: str | None = None, created_at: str | None = None) -> dict[str, Any]:
    item = {
        "id": thread_id or generate_thread_id(),
        "title": validate_thread_title(title),
        "nickname": validate_nickname(nickname),
        "body_markdown": validate_post_body(body),
        "created_at": created_at or now_iso(),
    }
    item["body_html"] = render_safe_markdown(item["body_markdown"])
    attachment = attachment or {}
    item.update({"attachment_filename": attachment.get("stored_filename"), "attachment_original_filename": attachment.get("original_filename"), "attachment_content_type": attachment.get("content_type")})
    with closing(get_connection()) as connection:
        connection.execute(
            """INSERT INTO board_threads (id, title, nickname, body_markdown, body_html, attachment_filename, attachment_original_filename, attachment_content_type, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""",
            (item["id"], item["title"], item["nickname"], item["body_markdown"], item["body_html"], item["attachment_filename"], item["attachment_original_filename"], item["attachment_content_type"], item["created_at"]),
        )
        connection.execute("DELETE FROM search_index WHERE source_type = ? AND source_id = ?", ("thread", item["id"]))
        connection.execute(
            "INSERT INTO search_index (source_type, source_id, title, body, nickname, created_at) VALUES (?, ?, ?, ?, ?, ?)",
            ("thread", item["id"], item["title"], item["body_markdown"], item["nickname"], item["created_at"]),
        )
        connection.commit()
    return item


def create_reply(*, thread_id: str, nickname: str, body: str, attachment: dict[str, Any] | None = None, reply_id: str | None = None, created_at: str | None = None) -> dict[str, Any]:
    if get_thread(thread_id) is None:
        raise ValueError("Thread not found.")
    item = {"id": reply_id or generate_reply_id(), "thread_id": thread_id, "nickname": validate_nickname(nickname), "body_markdown": validate_post_body(body), "created_at": created_at or now_iso()}
    item["body_html"] = render_safe_markdown(item["body_markdown"])
    attachment = attachment or {}
    item.update({"attachment_filename": attachment.get("stored_filename"), "attachment_original_filename": attachment.get("original_filename"), "attachment_content_type": attachment.get("content_type")})
    with closing(get_connection()) as connection:
        connection.execute(
            """INSERT INTO board_replies (id, thread_id, nickname, body_markdown, body_html, attachment_filename, attachment_original_filename, attachment_content_type, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""",
            (item["id"], item["thread_id"], item["nickname"], item["body_markdown"], item["body_html"], item["attachment_filename"], item["attachment_original_filename"], item["attachment_content_type"], item["created_at"]),
        )
        connection.execute("DELETE FROM search_index WHERE source_type = ? AND source_id = ?", ("reply", item["id"]))
        connection.execute(
            "INSERT INTO search_index (source_type, source_id, title, body, nickname, created_at) VALUES (?, ?, ?, ?, ?, ?)",
            ("reply", item["id"], "Board reply", item["body_markdown"], item["nickname"], item["created_at"]),
        )
        connection.commit()
    return item


def list_threads(limit: int | None = None) -> list[dict[str, Any]]:
    query = """
        SELECT t.*, COUNT(r.id) AS reply_count
        FROM board_threads t
        LEFT JOIN board_replies r ON r.thread_id = t.id
        GROUP BY t.id
        ORDER BY t.created_at DESC, t.rowid DESC
    """
    params: tuple[Any, ...] = ()
    if limit is not None:
        query += " LIMIT ?"
        params = (limit,)
    with closing(get_connection()) as connection:
        rows = connection.execute(query, params).fetchall()
    return [dict(row) for row in rows]


def get_thread(thread_id: str) -> dict[str, Any] | None:
    with closing(get_connection()) as connection:
        row = connection.execute("SELECT * FROM board_threads WHERE id = ?", (thread_id,)).fetchone()
    return dict(row) if row else None


def list_replies(thread_id: str) -> list[dict[str, Any]]:
    with closing(get_connection()) as connection:
        rows = connection.execute("SELECT * FROM board_replies WHERE thread_id = ? ORDER BY created_at ASC, rowid ASC", (thread_id,)).fetchall()
    return [dict(row) for row in rows]


def list_board_attachments(limit: int | None = 25) -> list[dict[str, Any]]:
    """Return recent message-board attachments for moderation review/deletion."""
    query = """
        SELECT attachment_filename, attachment_original_filename, attachment_content_type,
               id AS parent_id, 'thread' AS parent_type, title AS parent_title, created_at
        FROM board_threads
        WHERE attachment_filename IS NOT NULL AND attachment_filename != ''
        UNION ALL
        SELECT r.attachment_filename, r.attachment_original_filename, r.attachment_content_type,
               r.id AS parent_id, 'reply' AS parent_type, t.title AS parent_title, r.created_at
        FROM board_replies r
        LEFT JOIN board_threads t ON t.id = r.thread_id
        WHERE r.attachment_filename IS NOT NULL AND r.attachment_filename != ''
        ORDER BY created_at DESC
    """
    params: tuple[Any, ...] = ()
    if limit is not None:
        query += " LIMIT ?"
        params = (limit,)
    with closing(get_connection()) as connection:
        rows = connection.execute(query, params).fetchall()
    return [dict(row) for row in rows]


def get_board_attachment(stored_filename: str) -> dict[str, Any] | None:
    """Return a board attachment by stored filename."""
    safe_name = sanitize_filename(stored_filename)
    matches = [item for item in list_board_attachments(limit=None) if item["attachment_filename"] == safe_name]
    return matches[0] if matches else None


def delete_board_attachment(stored_filename: str) -> bool:
    """Delete a board attachment file and detach it from threads/replies."""
    try:
        safe_name = sanitize_filename(stored_filename)
    except ValueError:
        return False
    if get_board_attachment(safe_name) is None:
        return False
    path = (ATTACHMENTS_DIR / safe_name).resolve()
    if ATTACHMENTS_DIR.resolve() in path.parents:
        path.unlink(missing_ok=True)
    with closing(get_connection()) as connection:
        connection.execute(
            """UPDATE board_threads
               SET attachment_filename = NULL, attachment_original_filename = NULL, attachment_content_type = NULL
               WHERE attachment_filename = ?""",
            (safe_name,),
        )
        connection.execute(
            """UPDATE board_replies
               SET attachment_filename = NULL, attachment_original_filename = NULL, attachment_content_type = NULL
               WHERE attachment_filename = ?""",
            (safe_name,),
        )
        connection.commit()
    logger.info("Deleted board attachment %s", safe_name)
    return True


def app_config() -> dict[str, Any]:
    """Return safe runtime configuration values for the configuration page."""
    return {
        "app_name": APP_NAME,
        "version": APP_VERSION,
        "host": HOST,
        "port": PORT,
        "database": str(DATABASE_PATH),
        "storage": str(STORAGE_DIR),
        "max_upload": format_file_size(get_max_upload_bytes()),
        "max_attachment": format_file_size(get_max_attachment_bytes()),
        "moderation_password_configured": bool(ADMIN_PASSWORD),
        "log_level": LOG_LEVEL,
        "accent_color": get_accent_color(),
    }


def directory_size(path: Path) -> int:
    """Return the total size of files below a directory."""
    if not path.exists():
        return 0
    return sum(item.stat().st_size for item in path.rglob("*") if item.is_file())


def storage_statistics() -> dict[str, Any]:
    """Return counts and filesystem sizes for the storage statistics page."""
    uploads = list_uploads()
    threads = list_threads()
    with closing(get_connection()) as connection:
        chat_count = connection.execute("SELECT COUNT(*) FROM chat_messages").fetchone()[0]
        reply_count = connection.execute("SELECT COUNT(*) FROM board_replies").fetchone()[0]
    upload_bytes = sum(int(item["size_bytes"]) for item in uploads)
    stats = {
        "upload_count": len(uploads),
        "upload_bytes": upload_bytes,
        "upload_size": format_file_size(upload_bytes),
        "chat_count": chat_count,
        "thread_count": len(threads),
        "reply_count": reply_count,
        "storage_bytes": directory_size(STORAGE_DIR),
        "storage_size": format_file_size(directory_size(STORAGE_DIR)),
        "database_size": format_file_size(DATABASE_PATH.stat().st_size if DATABASE_PATH.exists() else 0),
    }
    return stats


def delete_upload(upload_id: str) -> bool:
    """Delete upload metadata, search index row, and stored file."""
    item = get_upload(upload_id)
    if item is None:
        return False
    path = (UPLOADS_DIR / item["stored_filename"]).resolve()
    if UPLOADS_DIR.resolve() in path.parents:
        path.unlink(missing_ok=True)
    with closing(get_connection()) as connection:
        connection.execute("DELETE FROM uploads WHERE id = ?", (upload_id,))
        connection.execute("DELETE FROM search_index WHERE source_type = ? AND source_id = ?", ("upload", upload_id))
        connection.commit()
    logger.info("Deleted upload %s", upload_id)
    return True


def delete_thread(thread_id: str) -> bool:
    """Delete a board thread, its replies, attachments, and search index rows."""
    thread = get_thread(thread_id)
    if thread is None:
        return False
    replies = list_replies(thread_id)
    attachment_names = [thread.get("attachment_filename")] + [reply.get("attachment_filename") for reply in replies]
    for attachment_name in attachment_names:
        if attachment_name:
            path = (ATTACHMENTS_DIR / attachment_name).resolve()
            if ATTACHMENTS_DIR.resolve() in path.parents:
                path.unlink(missing_ok=True)
    with closing(get_connection()) as connection:
        for reply in replies:
            connection.execute("DELETE FROM search_index WHERE source_type = ? AND source_id = ?", ("reply", reply["id"]))
        connection.execute("DELETE FROM board_replies WHERE thread_id = ?", (thread_id,))
        connection.execute("DELETE FROM board_threads WHERE id = ?", (thread_id,))
        connection.execute("DELETE FROM search_index WHERE source_type = ? AND source_id = ?", ("thread", thread_id))
        connection.commit()
    logger.info("Deleted thread %s", thread_id)
    return True


def moderation_password_valid(password: str | None) -> bool:
    """Check the simple single-password moderation guard."""
    return bool(ADMIN_PASSWORD) and password == ADMIN_PASSWORD


def require_moderation(password: str | None) -> None:
    """Raise a 403 if the submitted moderation password is invalid."""
    if not moderation_password_valid(password):
        raise HTTPException(status_code=403, detail="Invalid moderation password.")




def admin_session_token() -> str:
    """Return the deterministic cookie token for the configured admin password."""
    seed = f"{APP_NAME}:{ADMIN_PASSWORD}".encode("utf-8")
    return hashlib.sha256(seed).hexdigest()


def portal_authenticated(request: Request) -> bool:
    """Return True when the request has a valid admin session cookie."""
    cookie_value = request.cookies.get(ADMIN_SESSION_COOKIE, "")
    return bool(ADMIN_PASSWORD) and hmac.compare_digest(cookie_value, admin_session_token())


def portal_redirect() -> RedirectResponse:
    """Redirect unauthenticated users to the admin login."""
    return RedirectResponse(url="/admin", status_code=status.HTTP_303_SEE_OTHER)

def config_page_html(message: str = "") -> str:
    """Render runtime configuration with editable values inside the settings table."""
    config = app_config()
    note = f"<p class='muted'>{html.escape(message)}</p>" if message else ""
    accent = html.escape(get_accent_color(), quote=True)
    upload_mb = html.escape(str(bytes_to_megabytes(get_max_upload_bytes())), quote=True)
    attachment_mb = html.escape(str(bytes_to_megabytes(get_max_attachment_bytes())), quote=True)
    editable_rows = {
        "max_upload": (
            f"<input name='max_upload_mb' type='number' min='1' max='10240' step='1' "
            f"value='{upload_mb}' aria-label='Maximum upload size in megabytes' required> MB"
        ),
        "max_attachment": (
            f"<input name='max_attachment_mb' type='number' min='1' max='10240' step='1' "
            f"value='{attachment_mb}' aria-label='Maximum board attachment size in megabytes' required> MB"
        ),
        "accent_color": (
            f"<input name='accent_color' type='color' value='{accent}' aria-label='Accent color'> "
            f"<code>{accent}</code>"
        ),
    }
    rows = "".join(
        f"<tr><th>{html.escape(str(key))}</th><td>{editable_rows.get(key, html.escape(str(value)))}</td></tr>"
        for key, value in config.items()
    )
    return f"""
    <h1>Configuration</h1>{note}<p class='muted'>Runtime settings stay intentionally small. Editable values are saved in SQLite.</p>
    <form action='/config/settings' method='post' class='config-form'>
      <table>{rows}</table>
      <div class='inline-form config-actions'>
        <p class='muted'>Authenticated through the admin.</p>
        <button type='submit'>Save configuration</button>
      </div>
    </form>
    """


def stats_page_html() -> str:
    rows = "".join(f"<tr><th>{html.escape(str(key))}</th><td>{html.escape(str(value))}</td></tr>" for key, value in storage_statistics().items())
    return f"<h1>Storage statistics</h1><p class='muted'>Use these values for housekeeping and backup checks.</p><table>{rows}</table>"


def moderation_delete_item(item_id: str) -> str | None:
    """Delete an moderation item by ID without requiring the UI to choose a type."""
    if delete_upload(item_id):
        return "upload"
    if delete_thread(item_id):
        return "thread"
    if delete_chat_message(item_id):
        return "chat message"
    if delete_board_attachment(item_id):
        return "board attachment"
    return None


def moderation_upload_preview_html(item: dict[str, Any]) -> str:
    """Render a compact preview for an upload shown on the moderation page."""
    media = build_media_item(item)
    name = html.escape(item["original_filename"])
    kind = media["media_kind"]
    src = html.escape(media["view_url"], quote=True)
    if kind == "image":
        return f'<img class="moderation-preview" src="{src}" alt="Preview of {name}">'
    if kind == "video":
        return f'<video class="moderation-preview" src="{src}" muted preload="metadata" aria-label="Video preview for {name}"></video>'
    if kind == "audio":
        return "<div class='moderation-preview moderation-preview-icon' aria-label='Audio file preview'>Audio</div>"
    if kind == "pdf":
        return "<div class='moderation-preview moderation-preview-icon' aria-label='PDF file preview'>PDF</div>"
    return "<div class='moderation-preview moderation-preview-icon' aria-label='File preview'>File</div>"


def moderation_attachment_preview_html(item: dict[str, Any]) -> str:
    """Render a compact inline preview for a board attachment."""
    filename = html.escape(item["attachment_filename"], quote=True)
    alt = html.escape(item.get("attachment_original_filename") or "Board attachment")
    return f'<img class="moderation-preview" src="/attachment/{filename}" alt="Preview of {alt}">'


def moderation_delete_checkbox_html(item_id: str, label: str) -> str:
    """Render a selectable moderation deletion checkbox for a batch delete row."""
    safe_id = html.escape(item_id, quote=True)
    safe_label = html.escape(label, quote=True)
    return f"<input type='checkbox' name='item_ids' value='{safe_id}' aria-label='Select {safe_label} for deletion'>"


def moderation_item_id_lists_html() -> str:
    """Render recent moderation-manageable items with previews and batch checkboxes."""
    uploads = list_uploads(limit=25)
    threads = list_threads(limit=25)
    chat_messages = list_chat_messages(limit=25)
    attachments = list_board_attachments(limit=25)

    if uploads:
        upload_rows = "".join(
            "<tr>"
            f"<td>{moderation_upload_preview_html(item)}</td>"
            f"<td><code>{html.escape(item['id'])}</code><br><span class='muted'>{html.escape(item['original_filename'])}</span></td>"
            f"<td>{html.escape(item['display_size'])}</td>"
            f"<td>{moderation_delete_checkbox_html(item['id'], item['original_filename'])}</td>"
            "</tr>"
            for item in uploads
        )
    else:
        upload_rows = "<tr><td colspan='4' class='muted'>No uploads yet.</td></tr>"

    if threads:
        thread_rows = "".join(
            "<tr>"
            "<td><div class='moderation-preview moderation-preview-icon' aria-label='Thread preview'>Thread</div></td>"
            f"<td><code>{html.escape(item['id'])}</code><br><a href='/thread/{html.escape(item['id'], quote=True)}'>{html.escape(item['title'])}</a></td>"
            f"<td>{html.escape(str(item['reply_count']))} replies</td>"
            f"<td>{moderation_delete_checkbox_html(item['id'], item['title'])}</td>"
            "</tr>"
            for item in threads
        )
    else:
        thread_rows = "<tr><td colspan='4' class='muted'>No threads yet.</td></tr>"

    if chat_messages:
        chat_rows = "".join(
            "<tr>"
            "<td><div class='moderation-preview moderation-preview-icon' aria-label='Chat message preview'>Chat</div></td>"
            f"<td><code>{html.escape(item['id'])}</code><br><span class='muted'>{html.escape(item['nickname'])}</span>: {html.escape(item['message'])}</td>"
            f"<td>{html.escape(item['created_at'])}</td>"
            f"<td>{moderation_delete_checkbox_html(item['id'], 'chat message')}</td>"
            "</tr>"
            for item in chat_messages
        )
    else:
        chat_rows = "<tr><td colspan='4' class='muted'>No chat messages yet.</td></tr>"

    if attachments:
        attachment_rows = "".join(
            "<tr>"
            f"<td>{moderation_attachment_preview_html(item)}</td>"
            f"<td><code>{html.escape(item['attachment_filename'])}</code><br><span class='muted'>{html.escape(item.get('attachment_original_filename') or 'attachment')}</span></td>"
            f"<td>{html.escape(item['parent_type'])}: {html.escape(item.get('parent_title') or item['parent_id'])}</td>"
            f"<td>{moderation_delete_checkbox_html(item['attachment_filename'], item.get('attachment_original_filename') or 'board attachment')}</td>"
            "</tr>"
            for item in attachments
        )
    else:
        attachment_rows = "<tr><td colspan='4' class='muted'>No board attachments yet.</td></tr>"

    return f"""
    <section aria-labelledby='moderation-ids-heading'>
      <h2 id='moderation-ids-heading'>Recent item IDs</h2>
      <p class='muted'>Select uploads, threads, chat messages, or board attachments, enter the moderation password, then delete them in one batch. Showing the 25 most recent items in each category.</p>
      <form action='/moderation/delete' method='post' class='moderation-batch-delete-form'>
        <div class='moderation-batch-controls'>
          <p class='muted'>Admin session active. You may also submit the moderation password with this form. Select items below, then delete them in one batch.</p>
          <label class='visually-hidden'>Moderation password <input id='moderation-delete-password' name='password' type='password' autocomplete='current-password'></label><button type='submit' class='danger-button'>Delete selected items</button>
        </div>
        <h3>Uploads</h3>
        <table aria-label='Recent upload IDs'><thead><tr><th>Preview</th><th>ID / filename</th><th>Size</th><th>Select</th></tr></thead><tbody>{upload_rows}</tbody></table>
        <h3>Threads</h3>
        <table aria-label='Recent thread IDs'><thead><tr><th>Preview</th><th>ID / title</th><th>Replies</th><th>Select</th></tr></thead><tbody>{thread_rows}</tbody></table>
        <h3>Chat messages</h3>
        <table aria-label='Recent chat message IDs'><thead><tr><th>Preview</th><th>ID / message</th><th>Created</th><th>Select</th></tr></thead><tbody>{chat_rows}</tbody></table>
        <h3>Board attachments</h3>
        <table aria-label='Recent board attachment IDs'><thead><tr><th>Preview</th><th>ID / filename</th><th>Attached to</th><th>Select</th></tr></thead><tbody>{attachment_rows}</tbody></table>
      </form>
    </section>
    """

def moderation_page_html(message: str = "") -> str:
    note = f"<p class='muted'>{html.escape(message)}</p>" if message else ""
    return f"""
    <h1>Moderation tools</h1>{note}<p class='muted'>Single-password local moderation. Change FASTBOX_ADMIN_PASSWORD before deployment.</p>
    {moderation_item_id_lists_html()}
    <p><a href='/stats'>Storage statistics</a> - <a href='/config'>Configuration</a></p>
    """


def portal_login_page_html(message: str = "") -> str:
    """Render the admin login form."""
    note = f"<p class='muted'>{html.escape(message)}</p>" if message else ""
    return f"""
    <h1>Admin</h1>{note}
    <p class='muted'>Sign in once to access operational pages for this FastBox instance.</p>
    <form action='/admin/login' method='post' class='inline-form'>
      <label>Admin password <input name='password' type='password' autocomplete='current-password' required autofocus></label>
      <button type='submit'>Log in</button>
    </form>
    """


def portal_page_html() -> str:
    """Render the authenticated admin link hub."""
    return """
    <h1>Admin</h1>
    <p class='muted'>Operational pages are available after login. Use these tools for housekeeping, configuration, moderation, and service checks.</p>
    <section class='media-grid' aria-label='Admin links'>
      <article class='media-card'><h2><a href='/stats'>Stats</a></h2><p>Review storage usage, database size, upload counts, and attachment totals.</p></article>
      <article class='media-card'><h2><a href='/config'>Config</a></h2><p>Review runtime configuration and change the UI accent color.</p></article>
      <article class='media-card'><h2><a href='/moderation'>Moderation</a></h2><p>Select and delete uploads, board threads, chat messages, and board attachments.</p></article>
      <article class='media-card'><h2><a href='/health'>Health</a></h2><p>Check the lightweight JSON health endpoint.</p></article>
    </section>
    <form action='/admin/logout' method='post'><button type='submit'>Log out</button></form>
    """

def error_page_html(status_code: int, detail: str) -> str:
    return f"<h1>{status_code}</h1><p>{html.escape(detail)}</p><p><a href='/'>Return home</a></p>"

def render_page(title: str, body: str) -> HTMLResponse:
    accent = get_accent_color()
    accent_soft = accent_soft_rgba(accent)
    return HTMLResponse(f"""
        <!doctype html><html lang="en"><head><meta charset="utf-8"><meta name="viewport" content="width=device-width, initial-scale=1">
        <title>{title} - FastBox</title><link rel="icon" href="/favicon.ico">
        <style>
        :root{{--accent:{accent};--accent-strong:{accent};--accent-soft:{accent_soft};--bg:#111;--surface:#171717;--surface-2:#101010;--text:#e8ffe8;--muted:#aaa;--border:#333;--mark:#e6ff57}}
        body{{font-family:system-ui,sans-serif;margin:1rem;line-height:1.5;background:var(--bg);color:var(--text)}} main{{max-width:960px;margin:0 auto;background:var(--surface);padding:1rem;border:1px solid var(--accent);box-shadow:0 0 18px var(--accent-soft);border-radius:12px}} a{{color:var(--accent)}}a:hover{{color:var(--accent-strong);text-decoration-thickness:2px}} nav{{display:flex;flex-wrap:wrap;gap:.35rem 1rem;margin-bottom:1rem}} nav a{{margin-right:0}}.dropzone{{border:2px dashed var(--accent);border-radius:12px;padding:2rem;text-align:center;cursor:pointer;transition:background .15s,border-color .15s}}.dropzone.dragover,.dropzone.highlight{{background:var(--accent-soft);border-color:var(--accent-strong)}}.dropzone-status{{margin:.4rem 0 0}}table{{border-collapse:collapse;width:100%;margin-top:1rem}}th,td{{border-bottom:1px solid var(--border);padding:.6rem;text-align:left}}input,textarea,button,select{{font:inherit;padding:.55rem;margin:.25rem 0;background:#080808;color:var(--text);border:1px solid var(--accent);border-radius:6px}}input[type=color]{{min-height:2.75rem;padding:.2rem}}textarea{{width:100%;min-height:8rem}}button{{cursor:pointer}}button:focus,a:focus,input:focus,textarea:focus,select:focus,.dropzone:focus{{outline:3px solid var(--accent-strong);outline-offset:2px}}.visually-hidden{{position:absolute!important;width:1px;height:1px;padding:0;margin:-1px;overflow:hidden;clip:rect(0,0,0,0);white-space:nowrap;border:0}}.muted{{color:var(--muted)}}.chat-box,.post{{border:1px solid var(--border);border-radius:10px;overflow-wrap:anywhere;padding:1rem;background:var(--surface-2);margin:.75rem 0}}.chat-box{{height:55vh;min-height:280px;overflow-y:auto}}.message{{margin-bottom:.7rem}}.message.system{{color:var(--muted);font-style:italic}}time{{color:var(--muted);font-size:.85rem}}.chat-controls{{display:grid;grid-template-columns:1fr;gap:.5rem;margin-top:.75rem}}.thread-form{{display:grid;gap:.4rem}}.inline-form{{display:flex;flex-wrap:wrap;align-items:end;gap:.75rem}}.inline-form label{{display:grid;gap:.2rem}}.attachment,.media-preview{{max-width:100%;border:1px solid var(--border);border-radius:8px;margin-top:.5rem}}.media-grid{{display:grid;grid-template-columns:repeat(auto-fit,minmax(220px,1fr));gap:1rem}}.moderation-preview{{width:84px;height:64px;object-fit:cover;border:1px solid var(--border);border-radius:8px;background:#080808;display:flex;align-items:center;justify-content:center;font-size:.8rem}}.moderation-preview-icon{{color:var(--accent)}}.moderation-batch-delete-form{{margin:0}}.moderation-batch-controls{{display:flex;flex-wrap:wrap;align-items:end;gap:.75rem;margin:1rem 0;padding:1rem;border:1px solid var(--border);border-radius:10px;background:var(--surface-2)}}.danger-button{{border-color:#ff5c7a;color:#ffb3c0}}.danger-button:hover{{background:rgba(255,92,122,.12)}}.media-card{{border:1px solid var(--border);border-radius:10px;background:var(--surface-2);padding:1rem;overflow-wrap:anywhere}}.filter-bar a{{margin-right:.75rem}}mark{{background:var(--mark);color:#111;padding:0 .15rem}}.search-result{{border:1px solid var(--border);border-radius:10px;background:var(--surface-2);padding:1rem;margin:.75rem 0}}@media(max-width:520px){{body{{margin:.5rem}}main{{padding:.75rem}}.dropzone{{padding:1.25rem}}table{{font-size:.9rem}}}}@media(min-width:680px){{.chat-controls{{grid-template-columns:11rem 1fr auto}}}}
        </style></head><body><main><nav><a href="/">Home</a><a href="/files">Files</a><a href="/chat">Chat</a><a href="/board">Board</a><a href="/search">Search</a></nav>{body}</main></body></html>
    """.strip())


def upload_form_html() -> str:
    """Render the mobile-friendly upload target.

    The file input is visually hidden but remains in the DOM for browser file
    picker support. Users can drag and drop, double-click/double-tap the upload
    region, or focus the region and press Enter/Space. Selection uploads the file
    asynchronously so the user remains on the home page.
    """
    return """
    <section class="dropzone" id="dropzone" role="button" tabindex="0" aria-label="Upload a file by dropping it here or double-clicking to choose a file"><h2>Upload a file</h2><p class="muted">Drag and drop a file here, or double-click / double-tap this area to choose a file. Upload starts automatically after selection and keeps you on this page.</p><form id="upload-form" action="/upload" method="post" enctype="multipart/form-data"><input id="file-input" class="visually-hidden" name="file" type="file" required aria-label="Choose file to upload"></form><p id="upload-status" class="muted dropzone-status" aria-live="polite"></p></section>
    <script>
    const dropzone=document.getElementById('dropzone');
    const input=document.getElementById('file-input');
    const statusEl=document.getElementById('upload-status');
    async function refreshRecentUploads(){
        const response=await fetch('/files?partial=1');
        if(!response.ok){throw new Error('Failed to refresh uploads')}
        const html=await response.text();
        const recent=document.getElementById('recent-uploads');
        if(recent){recent.innerHTML=html}
    }
    async function uploadSelectedFile(file){
        if(!file){statusEl.textContent='No file was dropped.';return}
        const body=new FormData();
        body.append('file',file);
        statusEl.textContent='Uploading...';
        try{
            const response=await fetch('/upload',{method:'POST',body});
            const result=await response.json();
            if(!response.ok){throw new Error(result.detail||'Upload failed')}
            statusEl.textContent='Uploaded: '+result.original_filename;
            await refreshRecentUploads();
            input.value='';
        }catch(error){statusEl.textContent=error.message||'Upload failed'}
    }
    function submitIfFileSelected(){if(input.files&&input.files.length){uploadSelectedFile(input.files[0])}}
    function openFilePicker(){input.value='';input.click()}
    function preventDefaults(e){e.preventDefault();e.stopPropagation()}
    function highlight(){dropzone.classList.add('dragover')}
    function unhighlight(){dropzone.classList.remove('dragover')}
    function handleDrop(e){const files=e.dataTransfer&&e.dataTransfer.files;e.preventDefault();e.stopPropagation();unhighlight();if(files&&files.length){uploadSelectedFile(files[0])}else{statusEl.textContent='No file was dropped.'}}
    ['dragenter','dragover','dragleave','drop'].forEach(name=>dropzone.addEventListener(name,preventDefaults,false));
    ['dragenter','dragover'].forEach(name=>dropzone.addEventListener(name,highlight,false));
    ['dragleave','drop'].forEach(name=>dropzone.addEventListener(name,unhighlight,false));
    dropzone.addEventListener('drop',handleDrop,false);
    dropzone.addEventListener('dblclick',e=>{e.preventDefault();openFilePicker()});
    dropzone.addEventListener('keydown',e=>{if(e.key==='Enter'||e.key===' '){e.preventDefault();openFilePicker()}});
    input.addEventListener('change',submitIfFileSelected);
    </SCRIPT>
    """


def uploads_table_html(uploads: list[dict[str, Any]]) -> str:
    if not uploads:
        return '<p class="muted">No files have been uploaded yet.</p>'
    rows = [f'<tr><td><a href="/download/{item["id"]}">{html.escape(item["original_filename"])}</a></td><td>{item["display_size"]}</td><td>{item["uploaded_at"]}</td></tr>' for item in uploads]
    return "<table><thead><tr><th>Name</th><th>Size</th><th>Uploaded</th></tr></thead><tbody>" + "".join(rows) + "</tbody></table>"



def detect_mime_type(filename: str, stored_path: Path | None = None, content_type: str | None = None) -> str:
    """Return a browser-facing MIME type from metadata or filename."""
    guessed, _ = mimetypes.guess_type(filename)
    if guessed:
        return guessed
    if content_type and content_type not in {"", "application/octet-stream"}:
        return content_type
    if stored_path:
        guessed, _ = mimetypes.guess_type(str(stored_path))
        if guessed:
            return guessed
    return "application/octet-stream"


def classify_media_type(mime_type: str | None) -> str:
    mime_type = (mime_type or "application/octet-stream").lower()
    if mime_type in IMAGE_MIME_TYPES or mime_type.startswith("image/"):
        return "image"
    if mime_type in VIDEO_MIME_TYPES or mime_type.startswith("video/"):
        return "video"
    if mime_type in AUDIO_MIME_TYPES or mime_type.startswith("audio/"):
        return "audio"
    if mime_type in PDF_MIME_TYPES:
        return "pdf"
    return "other"


def should_generate_thumbnail(media_kind: str, mime_type: str | None, size_bytes: int) -> bool:
    """Phase 4 keeps thumbnails lightweight: image files use themselves as previews."""
    return media_kind == "image" and int(size_bytes) <= get_max_upload_bytes() and (mime_type or "").lower() != "image/svg+xml"


def build_media_item(upload: dict[str, Any]) -> dict[str, Any]:
    path = (UPLOADS_DIR / upload["stored_filename"]).resolve()
    mime_type = detect_mime_type(upload["original_filename"], path, upload.get("content_type"))
    media_kind = classify_media_type(mime_type)
    item = dict(upload)
    item.setdefault("display_size", format_file_size(int(item["size_bytes"])))
    item.update({
        "mime_type": mime_type,
        "media_kind": media_kind,
        "thumbnail_url": f"/view/{upload['id']}" if should_generate_thumbnail(media_kind, mime_type, int(upload["size_bytes"])) else None,
        "detail_url": f"/files/{upload['id']}",
        "view_url": f"/view/{upload['id']}",
        "download_url": f"/download/{upload['id']}",
    })
    return item


def list_media(filter_kind: str = "all") -> list[dict[str, Any]]:
    if filter_kind not in MEDIA_FILTERS:
        raise ValueError("Unsupported media filter.")
    items = [build_media_item(item) for item in list_uploads()]
    if filter_kind == "all":
        return items
    return [item for item in items if item["media_kind"] == filter_kind]


def render_media_preview(item: dict[str, Any], *, detail: bool = False) -> str:
    src = item.get("view_url") or f"/view/{item['id']}"
    name = html.escape(item["original_filename"])
    mime = html.escape(item.get("mime_type") or "application/octet-stream")
    kind = item.get("media_kind", "other")
    if kind == "image":
        return f'<img class="media-preview" src="{src}" alt="{name}">'
    if kind == "video":
        return f'<video class="media-preview" src="{src}" controls preload="metadata">Your browser cannot play this video. <a href="{src}">Download it</a>.</video>'
    if kind == "audio":
        return f'<audio class="media-preview" src="{src}" controls preload="metadata">Your browser cannot play this audio. <a href="{src}">Download it</a>.</audio>'
    if kind == "pdf":
        height = "620" if detail else "320"
        return f'<iframe class="media-preview" src="{src}" title="{name}" width="100%" height="{height}"></iframe>'
    return f'<p class="muted">No browser-native preview is available for {mime}. <a href="{src}">Download the file</a>.</p>'


def media_filter_html(active: str = "all") -> str:
    """Render media-type filters for the gallery view on the unified files page."""
    labels = [("all", "All"), ("image", "Images"), ("video", "Videos"), ("audio", "Audio"), ("pdf", "PDFs"), ("other", "Other")]
    return '<p class="filter-bar">' + "".join(f'<a href="/files?view=gallery&type={key}">{"[" if key == active else ""}{label}{"]" if key == active else ""}</a>' for key, label in labels) + '</p>'


def files_view_toggle_html(active_view: str, active_type: str = "all") -> str:
    """Render links that switch between list and gallery views without changing pages."""
    list_marker = "[List]" if active_view == "list" else "List"
    gallery_marker = "[Gallery]" if active_view == "gallery" else "Gallery"
    return (
        '<p class="filter-bar" aria-label="Files view selector">'
        f'<a href="/files?view=list">{list_marker}</a>'
        f'<a href="/files?view=gallery&type={html.escape(active_type)}">{gallery_marker}</a>'
        '</p>'
    )


def media_grid_html(items: list[dict[str, Any]]) -> str:
    if not items:
        return '<p class="muted">No media files match this filter yet.</p>'
    cards = []
    for item in items:
        title = html.escape(item["original_filename"])
        cards.append(f'<article class="media-card"><h3><a href="{item["detail_url"]}">{title}</a></h3>{render_media_preview(item)}<p class="muted">{item["media_kind"]} - {item["display_size"]} - {item["uploaded_at"]}</p><p><a href="{item["download_url"]}">Download</a></p></article>')
    return '<section class="media-grid">' + "".join(cards) + '</section>'


def media_detail_html(item: dict[str, Any]) -> str:
    title = html.escape(item["original_filename"])
    return f'<h1>{title}</h1>{render_media_preview(item, detail=True)}<p class="muted">Type: {item["media_kind"]} - MIME: {html.escape(item["mime_type"])} - Size: {item["display_size"]} - Uploaded: {item["uploaded_at"]}</p><p><a href="{item["download_url"]}">Download original</a> - <a href="/files?view=gallery">Back to gallery</a></p>'


def search_filter_html(active: str = "all", query: str = "") -> str:
    labels = [("all", "All"), ("uploads", "Uploads"), ("chat", "Chat"), ("threads", "Threads"), ("replies", "Replies")]
    q = quote(query)
    return '<p class="filter-bar">' + "".join(f'<a href="/search?q={q}&type={key}">{"[" if key == active else ""}{label}{"]" if key == active else ""}</a>' for key, label in labels) + '</p>'


def search_page_html(query: str = "", filter_type: str = "all") -> str:
    safe_query = html.escape(query or "", quote=True)
    form = f'<h1>Search FastBox</h1><form action="/search" method="get"><input name="q" value="{safe_query}" maxlength="{MAX_SEARCH_QUERY_CHARS}" placeholder="Search files, chat, and board posts"><button type="submit">Search</button></form>'
    if not query:
        return form + '<p class="muted">Enter a term to search uploaded filenames, chat messages, board threads, and replies.</p>'
    results = search_content(query, filter_type)
    if not results:
        return form + search_filter_html(filter_type, query) + '<p class="muted">No results found.</p>'
    cards = []
    for item in results:
        source = html.escape(item["source_type"])
        nickname = f' by {html.escape(item["nickname"])}' if item.get("nickname") else ""
        cards.append(f'<article class="search-result"><h2><a href="{item["url"]}">{item["title_html"]}</a></h2><p class="muted">{source}{nickname} - {html.escape(item.get("created_at") or "")}</p><p>{item["body_html"]}</p></article>')
    return form + search_filter_html(filter_type, query) + f'<p class="muted">{len(results)} result(s)</p>' + "".join(cards)

def chat_history_html(messages: list[dict[str, Any]]) -> str:
    if not messages:
        return '<p class="muted" id="empty-chat">No messages yet. Say hello.</p>'
    parts = []
    for item in messages:
        if item["kind"] == "system":
            parts.append(f'<div class="message system"><time>{item["created_at"]}</time> {item["message"]}</div>')
        else:
            parts.append(f'<div class="message"><strong>{item["nickname"]}</strong>: {item["message"]} <time>{item["created_at"]}</time></div>')
    return "".join(parts)


def chat_page_html() -> str:
    return f"""
    <h1>Global chat</h1><p class="muted">Anonymous real-time messages are visible to everyone connected to this FastBox.</p><section id="messages" class="chat-box" aria-live="polite">{chat_history_html(list_chat_messages())}</section>
    <form id="chat-form" class="chat-controls"><input id="nickname" name="nickname" maxlength="{MAX_NICKNAME_CHARS}" placeholder="Nickname" value="Anon" required><input id="message" name="message" maxlength="{MAX_CHAT_MESSAGE_CHARS}" placeholder="Type a message" required autocomplete="off"><button type="submit">Send</button></form><p id="chat-status" class="muted">Connecting...</p>
    <script>const messages=document.getElementById('messages'),form=document.getElementById('chat-form'),nickname=document.getElementById('nickname'),input=document.getElementById('message'),statusLine=document.getElementById('chat-status');let socket;function appendMessage(item){{const empty=document.getElementById('empty-chat');if(empty)empty.remove();const div=document.createElement('div');div.className=item.kind==='system'?'message system':'message';div.innerHTML=item.kind==='system'?`<time>${{item.created_at||''}}</time> ${{item.message}}`:`<strong>${{item.nickname}}</strong>: ${{item.message}} <time>${{item.created_at||''}}</time>`;messages.appendChild(div);messages.scrollTop=messages.scrollHeight}}function connect(){{const protocol=location.protocol==='https:'?'wss':'ws';socket=new WebSocket(`${{protocol}}://${{location.host}}/ws/chat`);socket.onopen=()=>{{statusLine.textContent='Connected';socket.send(JSON.stringify({{type:'join',nickname:nickname.value}}))}};socket.onmessage=e=>appendMessage(JSON.parse(e.data));socket.onclose=()=>{{statusLine.textContent='Disconnected. Reconnecting...';setTimeout(connect,1000)}}}}form.addEventListener('submit',e=>{{e.preventDefault();if(!socket||socket.readyState!==WebSocket.OPEN)return;socket.send(JSON.stringify({{type:'message',nickname:nickname.value,message:input.value}}));input.value=''}});connect();</script>
    """


def attachment_html(item: dict[str, Any]) -> str:
    if not item.get("attachment_filename"):
        return ""
    src = f"/attachment/{html.escape(item['attachment_filename'])}"
    alt = html.escape(item.get("attachment_original_filename") or "attachment")
    return f'<p><img class="attachment" src="{src}" alt="{alt}"></p>'


def board_dropzone_script() -> str:
    """Return the board attachment drop-zone script.

    This follows the vanilla drag-and-drop pattern documented by Smashing
    Magazine: prevent defaults on drag events, highlight the drop area, read
    dropped files directly from ``event.dataTransfer.files``, and send them in
    ``FormData``. The file input remains as the accessible fallback for picker
    selection.
    """
    return """
    <SCRIPT>
    document.querySelectorAll('.board-upload-form').forEach(form=>{
        const dropzone=form.querySelector('.board-dropzone');
        const input=form.querySelector('.board-file-input');
        const status=form.querySelector('.dropzone-status');
        let droppedFile=null;
        function preventDefaults(e){e.preventDefault();e.stopPropagation()}
        function highlight(){dropzone.classList.add('dragover')}
        function unhighlight(){dropzone.classList.remove('dragover')}
        function chooseFile(file){
            droppedFile=file||null;
            status.textContent=droppedFile?'Selected: '+droppedFile.name:'No image selected.';
            if(droppedFile){input.value=''}
        }
        function openFilePicker(){droppedFile=null;input.value='';input.click()}
        function handleDrop(e){
            const files=e.dataTransfer&&e.dataTransfer.files;
            e.preventDefault();e.stopPropagation();unhighlight();
            if(files&&files.length){chooseFile(files[0])}else{status.textContent='No image was dropped.'}
        }
        ['dragenter','dragover','dragleave','drop'].forEach(name=>dropzone.addEventListener(name,preventDefaults,false));
        ['dragenter','dragover'].forEach(name=>dropzone.addEventListener(name,highlight,false));
        ['dragleave','drop'].forEach(name=>dropzone.addEventListener(name,unhighlight,false));
        dropzone.addEventListener('drop',handleDrop,false);
        dropzone.addEventListener('dblclick',e=>{e.preventDefault();openFilePicker()});
        dropzone.addEventListener('keydown',e=>{if(e.key==='Enter'||e.key===' '){e.preventDefault();openFilePicker()}});
        input.addEventListener('change',()=>{if(input.files&&input.files.length){droppedFile=null;status.textContent='Selected: '+input.files[0].name}else{status.textContent='No image selected.'}});
        form.addEventListener('submit',async e=>{
            if(!droppedFile){return}
            e.preventDefault();
            const body=new FormData(form);
            body.set('attachment',droppedFile,droppedFile.name);
            status.textContent='Uploading attachment...';
            try{
                const response=await fetch(form.action,{method:'POST',body,redirect:'follow'});
                if(!response.ok){status.textContent='Attachment upload failed.';return}
                window.location.href=response.url;
            }catch(error){status.textContent='Attachment upload failed.'}
        });
    });
    </SCRIPT>
    """


def thread_form_html() -> str:
    return f"""
    <h2>Create thread</h2><form class="thread-form board-upload-form" action="/thread" method="post" enctype="multipart/form-data"><input name="nickname" maxlength="{MAX_NICKNAME_CHARS}" placeholder="Nickname" value="Anon" required><input name="title" maxlength="{MAX_THREAD_TITLE_CHARS}" placeholder="Thread title" required><textarea name="body" maxlength="{MAX_POST_BODY_CHARS}" placeholder="Markdown supported: **bold**, *italic*, `code`" required></textarea><section class="dropzone board-dropzone" role="button" tabindex="0" aria-label="Attach an image by dropping it here or double-clicking to choose an image"><p><strong>Optional image attachment</strong></p><p class="muted">Drag and drop an image here, or double-click / double-tap to choose one.</p><input class="visually-hidden board-file-input" type="file" name="attachment" accept="image/png,image/jpeg,image/gif,image/webp"><p class="muted dropzone-status" aria-live="polite">No image selected.</p></section><button type="submit">Post thread</button></form>{board_dropzone_script()}
    """


def board_list_html(threads: list[dict[str, Any]]) -> str:
    if not threads:
        return '<p class="muted">No threads yet. Start one.</p>'
    rows = [f'<tr><td><a href="/thread/{item["id"]}">{item["title"]}</a><br><span class="muted">by {item["nickname"]}</span></td><td>{item["reply_count"]}</td><td>{item["created_at"]}</td></tr>' for item in threads]
    return "<table><thead><tr><th>Thread</th><th>Replies</th><th>Created</th></tr></thead><tbody>" + "".join(rows) + "</tbody></table>"


def thread_page_html(thread: dict[str, Any], replies: list[dict[str, Any]]) -> str:
    reply_html = "".join(f'<article class="post"><p><strong>{reply["nickname"]}</strong> <time>{reply["created_at"]}</time></p>{reply["body_html"]}{attachment_html(reply)}</article>' for reply in replies) or '<p class="muted">No replies yet.</p>'
    return f"""
    <h1>{thread['title']}</h1><article class="post"><p><strong>{thread['nickname']}</strong> <time>{thread['created_at']}</time></p>{thread['body_html']}{attachment_html(thread)}</article><h2>Replies</h2>{reply_html}
    <h2>Reply</h2><form class="thread-form board-upload-form" action="/reply" method="post" enctype="multipart/form-data"><input type="hidden" name="thread_id" value="{thread['id']}"><input name="nickname" maxlength="{MAX_NICKNAME_CHARS}" placeholder="Nickname" value="Anon" required><textarea name="body" maxlength="{MAX_POST_BODY_CHARS}" placeholder="Write a reply" required></textarea><section class="dropzone board-dropzone" role="button" tabindex="0" aria-label="Attach an image by dropping it here or double-clicking to choose an image"><p><strong>Optional image attachment</strong></p><p class="muted">Drag and drop an image here, or double-click / double-tap to choose one.</p><input class="visually-hidden board-file-input" type="file" name="attachment" accept="image/png,image/jpeg,image/gif,image/webp"><p class="muted dropzone-status" aria-live="polite">No image selected.</p></section><button type="submit">Reply</button></form>{board_dropzone_script()}
    """


def create_app() -> FastAPI:
    ensure_directories()
    init_db()
    app = FastAPI(title=APP_NAME, version=APP_VERSION)
    app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static")
    templates = Jinja2Templates(directory=TEMPLATES_DIR)


    @app.exception_handler(HTTPException)
    async def http_exception_handler(request: Request, exc: HTTPException) -> HTMLResponse:
        logger.warning("HTTP %s at %s: %s", exc.status_code, request.url.path, exc.detail)
        response = render_page("Error", error_page_html(exc.status_code, str(exc.detail)))
        response.status_code = exc.status_code
        return response

    @app.exception_handler(Exception)
    async def unhandled_exception_handler(request: Request, exc: Exception) -> HTMLResponse:
        logger.exception("Unhandled error at %s", request.url.path)
        response = render_page("Error", error_page_html(500, "Unexpected server error."))
        response.status_code = 500
        return response

    @app.get("/", response_class=HTMLResponse)
    def homepage() -> HTMLResponse:
        return render_page("Home", f"<h1>FastBox</h1><p>A modern take on the classic PirateBox UI.</p>{upload_form_html()}<h2>Recent uploads</h2><div id='recent-uploads'>{uploads_table_html(list_uploads(limit=5))}</div><h2>Recent chat</h2><div class='chat-box' style='height:auto;min-height:0'>{chat_history_html(list_chat_messages(limit=5))}</div><h2>Recent threads</h2>{board_list_html(list_threads(limit=5))}")

    @app.get("/favicon.ico")
    def favicon() -> FileResponse:
        """Serve the FastBox favicon from static assets."""
        return FileResponse(STATIC_DIR / "favicon.ico", media_type="image/x-icon")

    @app.get("/files", response_class=HTMLResponse)
    def files(request: Request) -> HTMLResponse:
        """Show uploaded files as either a table or a media gallery."""
        if request.query_params.get("partial") == "1":
            return HTMLResponse(uploads_table_html(list_uploads(limit=5)))
        view = request.query_params.get("view", "gallery")
        if view not in {"list", "gallery"}:
            raise HTTPException(status_code=400, detail="Unsupported files view.")
        filter_kind = request.query_params.get("type", "all")
        if view == "gallery":
            try:
                items = list_media(filter_kind)
            except ValueError as exc:
                raise HTTPException(status_code=400, detail=str(exc)) from exc
            body = (
                "<h1>Files</h1>"
                "<p class='muted'>Switch between a classic file list and browser-native media previews for images, video, audio, and PDFs.</p>"
                f"{files_view_toggle_html('gallery', filter_kind)}"
                f"{media_filter_html(filter_kind)}"
                f"{media_grid_html(items)}"
            )
        else:
            body = (
                "<h1>Files</h1>"
                "<p class='muted'>All uploads are listed here. Use List view for a compact table of every upload.</p>"
                f"{files_view_toggle_html('list', filter_kind)}"
                f"{uploads_table_html(list_uploads())}"
            )
        return render_page("Files", body)


    @app.get("/files/{upload_id}", response_class=HTMLResponse)
    def file_detail(upload_id: str) -> HTMLResponse:
        item = get_upload(upload_id)
        if item is None:
            raise HTTPException(status_code=404, detail="File not found.")
        path = (UPLOADS_DIR / item["stored_filename"]).resolve()
        if UPLOADS_DIR.resolve() not in path.parents or not path.exists() or not path.is_file():
            raise HTTPException(status_code=404, detail="Stored file is missing.")
        return render_page(item["original_filename"], media_detail_html(build_media_item(item)))

    @app.post("/upload")
    async def upload(file: UploadFile = File(...)) -> JSONResponse:
        saved = await save_upload(file)
        logger.info("Uploaded %s", saved["id"])
        return JSONResponse({
            "id": saved["id"],
            "original_filename": saved["original_filename"],
            "display_size": format_file_size(saved["size_bytes"]),
            "uploaded_at": saved["uploaded_at"],
        })

    @app.get("/view/{upload_id}")
    def view_upload(upload_id: str) -> FileResponse:
        """Serve an uploaded file inline for browser-native previews.

        The /download route intentionally uses an attachment disposition. Media
        previews must not use that route, because browsers may immediately
        download preview sources when the media page renders.
        """
        item = get_upload(upload_id)
        if not item:
            raise HTTPException(status_code=404, detail="Upload not found.")
        path = (UPLOADS_DIR / item["stored_filename"]).resolve()
        if UPLOADS_DIR.resolve() not in path.parents or not path.exists() or not path.is_file():
            raise HTTPException(status_code=404, detail="Stored file is missing.")
        return FileResponse(
            path,
            media_type=item.get("content_type") or "application/octet-stream",
            filename=item["original_filename"],
            headers={"Content-Disposition": f"inline; filename*=UTF-8''{quote(item['original_filename'])}"},
        )

    @app.get("/download/{upload_id}")
    def download(upload_id: str) -> FileResponse:
        item = get_upload(upload_id)
        if item is None:
            raise HTTPException(status_code=404, detail="Upload not found.")
        path = (UPLOADS_DIR / item["stored_filename"]).resolve()
        if UPLOADS_DIR.resolve() not in path.parents or not path.exists() or not path.is_file():
            raise HTTPException(status_code=404, detail="Stored file is missing.")
        return FileResponse(path, media_type=item.get("content_type") or "application/octet-stream", filename=item["original_filename"], headers={"Content-Disposition": f"attachment; filename*=UTF-8''{quote(item['original_filename'])}"})

    @app.get("/attachment/{stored_filename}")
    def attachment(stored_filename: str) -> FileResponse:
        safe_name = sanitize_filename(stored_filename)
        path = (ATTACHMENTS_DIR / safe_name).resolve()
        if ATTACHMENTS_DIR.resolve() not in path.parents or not path.exists() or not path.is_file():
            raise HTTPException(status_code=404, detail="Attachment not found.")
        return FileResponse(path)

    @app.get("/chat", response_class=HTMLResponse)
    def chat() -> HTMLResponse:
        return render_page("Chat", chat_page_html())

    @app.websocket("/ws/chat")
    async def websocket_chat(websocket: WebSocket) -> None:
        await manager.connect(websocket)
        nickname = "Anon"
        try:
            while True:
                raw = await websocket.receive_text()
                try:
                    payload = json.loads(raw)
                    event_type = payload.get("type", "message")
                    nickname = validate_nickname(payload.get("nickname", nickname))
                    if not rate_limit_allows_message(nickname):
                        await manager.send_personal_json(websocket, {"kind": "system", "nickname": "System", "message": "Rate limit exceeded.", "created_at": now_iso()})
                        continue
                    if event_type == "join":
                        item = insert_chat_message(nickname="System", message=f"{nickname} joined the chat.", kind="system")
                    elif event_type == "message":
                        item = insert_chat_message(nickname=nickname, message=payload.get("message", ""), kind="message")
                    else:
                        raise ValueError("Unsupported chat event type.")
                except (json.JSONDecodeError, TypeError, ValueError) as exc:
                    await manager.send_personal_json(websocket, {"kind": "system", "nickname": "System", "message": html.escape(f"Invalid message: {exc}"), "created_at": now_iso()})
                    continue
                await manager.broadcast(item)
        except WebSocketDisconnect:
            manager.disconnect(websocket)
            try:
                await manager.broadcast(insert_chat_message(nickname="System", message=f"{nickname} left the chat.", kind="system"))
            except ValueError:  # pragma: no cover - defensive disconnect cleanup
                pass

    @app.get("/board", response_class=HTMLResponse)
    def board() -> HTMLResponse:
        return render_page("Board", f"<h1>Message board</h1><p class='muted'>Anonymous threaded posts with safe Markdown.</p>{thread_form_html()}<h2>Threads</h2>{board_list_html(list_threads())}")

    @app.get("/thread/{thread_id}", response_class=HTMLResponse)
    def thread(thread_id: str) -> HTMLResponse:
        item = get_thread(thread_id)
        if item is None:
            raise HTTPException(status_code=404, detail="Thread not found.")
        return render_page(item["title"], thread_page_html(item, list_replies(thread_id)))

    @app.post("/thread")
    async def post_thread(title: str = Form(""), nickname: str = Form("Anon"), body: str = Form(""), attachment: UploadFile | None = File(None)) -> RedirectResponse:
        try:
            saved_attachment = await save_optional_attachment(attachment)
            item = create_thread(title=title, nickname=nickname, body=body, attachment=saved_attachment)
            logger.info("Created thread %s", item["id"])
        except ValueError as exc:
            raise HTTPException(status_code=400, detail=str(exc)) from exc
        return RedirectResponse(url=f"/thread/{item['id']}", status_code=status.HTTP_303_SEE_OTHER)

    @app.post("/reply")
    async def post_reply(thread_id: str = Form(""), nickname: str = Form("Anon"), body: str = Form(""), attachment: UploadFile | None = File(None)) -> RedirectResponse:
        try:
            saved_attachment = await save_optional_attachment(attachment)
            reply = create_reply(thread_id=thread_id, nickname=nickname, body=body, attachment=saved_attachment)
            logger.info("Created reply %s", reply["id"])
        except ValueError as exc:
            raise HTTPException(status_code=400, detail=str(exc)) from exc
        return RedirectResponse(url=f"/thread/{thread_id}", status_code=status.HTTP_303_SEE_OTHER)

    @app.get("/search", response_class=HTMLResponse)
    def search(request: Request) -> HTMLResponse:
        query = request.query_params.get("q", "")
        filter_type = request.query_params.get("type", "all")
        try:
            body = search_page_html(query, filter_type)
        except ValueError as exc:
            raise HTTPException(status_code=400, detail=str(exc)) from exc
        return render_page("Search", body)



    @app.get("/admin", response_class=HTMLResponse)
    def admin(request: Request) -> HTMLResponse:
        if portal_authenticated(request):
            return render_page("Admin", portal_page_html())
        return render_page("Admin", portal_login_page_html())

    @app.post("/admin/login")
    def admin_login(password: str = Form("")):
        if not moderation_password_valid(password):
            return render_page("Admin", portal_login_page_html("Invalid admin password."))
        response = RedirectResponse(url="/admin", status_code=status.HTTP_303_SEE_OTHER)
        response.set_cookie(ADMIN_SESSION_COOKIE, admin_session_token(), httponly=True, samesite="lax")
        return response

    @app.post("/admin/logout")
    def admin_logout() -> RedirectResponse:
        response = RedirectResponse(url="/admin", status_code=status.HTTP_303_SEE_OTHER)
        response.delete_cookie(ADMIN_SESSION_COOKIE)
        return response

    @app.get("/config", response_class=HTMLResponse)
    def config(request: Request):
        if not portal_authenticated(request):
            return portal_redirect()
        return render_page("Configuration", config_page_html())

    @app.post("/config/settings", response_class=HTMLResponse)
    def update_configuration(request: Request, max_upload_mb: str = Form(""), max_attachment_mb: str = Form(""), accent_color: str = Form("")):
        if not portal_authenticated(request):
            raise HTTPException(status_code=403, detail="Admin login required.")
        try:
            settings = set_runtime_configuration(max_upload_mb=max_upload_mb, max_attachment_mb=max_attachment_mb, accent_color=accent_color)
        except ValueError as exc:
            raise HTTPException(status_code=400, detail=str(exc)) from exc
        return render_page("Configuration", config_page_html(f"Configuration updated: max upload {settings['max_upload_mb']} MB, max attachment {settings['max_attachment_mb']} MB, accent {settings['accent_color']}."))

    @app.post("/config/accent", response_class=HTMLResponse)
    def update_accent(request: Request, accent_color: str = Form("")):
        if not portal_authenticated(request):
            raise HTTPException(status_code=403, detail="Admin login required.")
        try:
            color = set_accent_color(accent_color)
        except ValueError as exc:
            raise HTTPException(status_code=400, detail=str(exc)) from exc
        return render_page("Configuration", config_page_html(f"Accent color updated to {color}."))

    @app.get("/stats", response_class=HTMLResponse)
    def stats(request: Request):
        return render_page("Storage statistics", stats_page_html())

    @app.get("/moderation", response_class=HTMLResponse)
    def moderation(request: Request):
        return render_page("Moderation", moderation_page_html())

    @app.post("/moderation/delete", response_class=HTMLResponse)
    async def moderation_delete(request: Request, password: str = Form(""), item_ids: list[str] = Form(default=[])):
        if not portal_authenticated(request):
            require_moderation(password)
        if not item_ids:
            raise HTTPException(status_code=400, detail="Select at least one item to delete.")

        deleted: list[str] = []
        missing: list[str] = []
        for item_id in item_ids:
            deleted_kind = moderation_delete_item(item_id)
            if deleted_kind is None:
                missing.append(item_id)
            else:
                deleted.append(f"{deleted_kind} {item_id}")

        if not deleted:
            raise HTTPException(status_code=404, detail="No selected moderation items were found.")

        message = f"Deleted {len(deleted)} item(s): " + ", ".join(deleted)
        if missing:
            message += f". Not found: {', '.join(missing)}"
        return render_page("Moderation", moderation_page_html(message))

    @app.get("/health")
    def health() -> dict[str, str]:
        return {"status": "ok", "app": APP_NAME, "version": APP_VERSION}

    app.state.templates = templates
    return app


app = create_app()


if __name__ == "__main__":  # pragma: no cover - manual startup path
    uvicorn.run("fastbox:app", host=HOST, port=PORT, reload=False)