chanbans/chanbans/http.py

from collections import defaultdict
import functools
import json
from typing import Any, Mapping, MutableMapping, Optional, Sequence
import urllib.parse

from aiohttp import web
from jinja2 import Environment, PackageLoader, select_autoescape

from . import config
from .db import get_db, search_db


# 2023-07-24 - eating stirfry tonight. It's pretty bad.
# 2023-07-31 - In a meeting. Hope I don't get caught


def route_url(url: str, args: Optional[Mapping[str, Any]] = None):
    if args is None:
        args = dict()
    assert url == "" or url[0] == "/", "URL must be blank or start with a slash"
    if url in ("", "/"):
        base = config.HTTP_ROOT
    else:
        base = f"{config.HTTP_ROOT}{url}"

    if args:
        return base + "?" + urllib.parse.urlencode(args)
    else:
        return base


def static_url(resource: str):
    return f"{config.STATIC_ROOT}/{resource}"


_env = Environment(
    loader=PackageLoader("chanbans"),
    autoescape=select_autoescape(),
)
_env.globals.update(
    {
        "route_url": route_url,
        "static_url": static_url,
    }
)

TemplateContext = MutableMapping[str, Any]


class HtmlResponse(web.Response):
    def __init__(self, *args, **kwargs):
        headers = kwargs.get(
            "headers",
            {
                "Content-Type": "text/html",
            },
        )
        kwargs["headers"] = headers
        super().__init__(*args, **kwargs)


class TemplateView(web.View):
    template_path: str

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        assert not hasattr(self, "_env")

        global _env
        self._env = _env

        self.template = self.env.get_template(self.template_path)

    @property
    def env(self) -> Environment:
        return self._env

    def get_context(self) -> TemplateContext:
        "Gets the context for this template"
        return {
            "config": config,
        }

    def template_render(self) -> str:
        "Renders the template on this object."
        context = self.get_context()
        return self.template.render(**context)


################################################################################
# Template views
################################################################################


def template_view_factory(path: str) -> type:
    class FactoryView(TemplateView):
        template_path = path

        async def get(self):
            html = self.template_render()
            return HtmlResponse(text=html)

    return FactoryView


class IndexView(TemplateView):
    template_path = "index.html"

    @functools.cache
    def get_context(self) -> TemplateContext:
        ctx = super().get_context()
        query = self.get_search_query()
        ctx["posts"] = search_db(**query)
        ctx["query"] = query
        ctx["next_page"] = route_url("/", {**query, "offset": query["offset"] + 100})
        ctx["prev_page"] = route_url("/", {**query, "offset": query["offset"] - 100})
        return ctx

    # Query parameters:
    # ?board=pol
    # &reason=evasion
    # &name=Anonymous
    # &trip=asdfbgh
    # &com=comment%20search
    # &sub=subject%20search
    # &time_before=123456
    # &time_after=123456
    # &md5=md5_sum
    # &offset=0

    def get_search_query(self):
        allowed_keys = (
            "board",
            "reason",
            "name",
            "trip",
            "com",
            "sub",
            "time_before",
            "time_after",
            "md5",
            "offset",
        )
        query = {
            key: value
            for key, value in self.request.query.items()
            if key in allowed_keys and key in self.request.query
        }
        if "offset" in query:
            try:
                query["offset"] = int(query["offset"])
            except ValueError:
                query["offset"] = 0
        else:
            query["offset"] = 0

        if "time_before" in query:
            try:
                query["time_before"] = int(query["time_before"])
            except ValueError:
                query.pop("time_before")
        if "time_after" in query:
            try:
                query["time_after"] = int(query["time_after"])
            except ValueError:
                query.pop("time_after")
        return query

    async def get(self):
        html = self.template_render()
        return HtmlResponse(text=html)


################################################################################
# Routes
################################################################################


app = web.Application()
app.add_routes(
    [
        web.get(config.HTTP_ROOT, IndexView),
        web.get(f"{config.HTTP_ROOT}/faq", template_view_factory("faq.html")),
        web.get(f"{config.HTTP_ROOT}/news", template_view_factory("news.html")),
    ]
)

if config.STATIC_HANDLER == "local":
    app.add_routes(
        [
            web.static(
                config.STATIC_ROOT,
                config.STATIC_LOCAL_PATH,
                follow_symlinks=config.STATIC_LOCAL_FOLLOW_SYMLINKS,
            )
        ]
    )


def run_app():
    web.run_app(app)