tinyweb/handlers/_helpers.py

import json
import re
import secrets
import threading

from db import get_db, return_db, get_setting, set_setting
from templates import wrap_page


_request_local = threading.local()


def _get_csrf_token():
    return getattr(_request_local, 'csrf_token', '')


def _csrf_field():
    return f'<input type="hidden" name="_csrf" value="{_get_csrf_token()}">'


def _check_csrf(body):
    token = body.get("_csrf", [""])[0]
    expected = _get_csrf_token()
    if not expected or not token:
        return False
    return secrets.compare_digest(token, expected)


_STOPWORDS = frozenset({
    "a", "an", "the", "and", "or", "but", "is", "are", "was", "were",
    "in", "on", "at", "to", "for", "of", "with", "by", "from", "as",
    "into", "about", "how", "what", "which", "who", "where", "when",
    "do", "does", "did", "be", "been", "being", "have", "has", "had",
    "it", "its", "this", "that", "not", "no", "so", "if", "can", "will",
    "my", "your", "i", "me", "we", "you", "he", "she", "they",
})


def _sanitize_fts_query(query):
    words = query.split()
    if not words:
        return '""'
    tokens = []
    last_idx = len(words) - 1
    for i, w in enumerate(words):
        cleaned = re.sub(r'["\'\(\)\*\+\-\^~:]', '', w).strip()
        if not cleaned:
            continue
        if cleaned.lower() in _STOPWORDS:
            continue
        if cleaned.upper() in ("AND", "OR", "NOT", "NEAR"):
            continue
        if i == last_idx:
            tokens.append(f"{cleaned}*")
        else:
            tokens.append(f'"{cleaned}"')
    return " ".join(tokens) if tokens else '""'


def _get_bookmark_token():
    token = get_setting("bookmark_token")
    if not token:
        token = secrets.token_hex(16)
        set_setting("bookmark_token", token)
    return token


def _respond(body_html, status=200, use_default=False, head_html=""):
    return {
        "status": status,
        "content_type": "text/html; charset=utf-8",
        "body": wrap_page(body_html, use_default=use_default, head_html=head_html),
        "headers": {},
    }


def _redirect(location):
    if not location.startswith("/") or location.startswith("//"):
        location = "/"
    return {
        "status": 302,
        "content_type": "text/html; charset=utf-8",
        "body": "",
        "headers": {"Location": location},
    }


def _json_response(data, status=200, headers=None):
    return {
        "status": status,
        "content_type": "application/json",
        "body": json.dumps(data, indent=2),
        "headers": headers or {},
    }


def _text_response(text, status=200, headers=None):
    return {
        "status": status,
        "content_type": "text/plain",
        "body": text,
        "headers": headers or {},
    }


def _error(status):
    return _respond(f"<h1>{status}</h1>", status)


PER_PAGE = 10
BROWSE_PER_PAGE = 50


def _paginate(query, key="p"):
    try:
        page = int(query.get(key, ["1"])[0])
    except (ValueError, IndexError):
        page = 1
    return max(1, page)


def _page_nav(page, total, base_url, per_page=None):
    per_page = per_page or PER_PAGE
    if total <= per_page:
        return ""
    total_pages = (total + per_page - 1) // per_page
    sep = "&" if "?" in base_url else "?"
    parts = []
    if page > 1:
        parts.append(f'<a href="{base_url}{sep}p={page - 1}">&laquo; prev</a>')
    parts.append(f"page {page} of {total_pages}")
    if page < total_pages:
        parts.append(f'<a href="{base_url}{sep}p={page + 1}">next &raquo;</a>')
    return f'<p class="pagination">{" | ".join(parts)}</p>'


def _get_page_tags(page_id, db=None):
    close = False
    if db is None:
        db = get_db()
        close = True
    rows = db.execute(
        "SELECT t.name FROM tags t JOIN page_tags pt ON t.id = pt.tag_id "
        "WHERE pt.page_id = ? ORDER BY t.name", (page_id,)
    ).fetchall()
    if close:
        return_db(db)
    return [r["name"] for r in rows]


def _set_page_tags(page_id, tag_string, db=None):
    close = False
    if db is None:
        db = get_db()
        close = True
    db.execute("DELETE FROM page_tags WHERE page_id = ?", (page_id,))
    for name in (t.strip().lower() for t in tag_string.split(",") if t.strip()):
        db.execute("INSERT OR IGNORE INTO tags (name) VALUES (?)", (name,))
        tag_id = db.execute("SELECT id FROM tags WHERE name = ?", (name,)).fetchone()["id"]
        db.execute("INSERT OR IGNORE INTO page_tags (page_id, tag_id) VALUES (?, ?)", (page_id, tag_id))
    if close:
        db.commit()
        return_db(db)


def _cleanup_orphaned_tags(db):
    db.execute("DELETE FROM tags WHERE id NOT IN (SELECT DISTINCT tag_id FROM page_tags)")