tinyweb/handlers/data.py

import json
import threading

from db import get_db, return_db, get_setting, set_setting, index_url
from templates import esc
from ._helpers import _respond, _json_response, _redirect, _csrf_field

MAX_EXPORT = 10000
_reindex_thread = None


def handle_export(query=None):
    try:
        batch = int((query or {}).get("batch", ["0"])[0])
    except (TypeError, ValueError):
        batch = 0
    db = get_db()
    try:
        rows = db.execute(
            "SELECT url, title, note FROM pages ORDER BY id LIMIT ? OFFSET ?",
            (MAX_EXPORT, batch * MAX_EXPORT),
        ).fetchall()
    finally:
        return_db(db)
    data = [{"url": r["url"], "title": r["title"], "note": r["note"]} for r in rows]
    return _json_response(data, headers={"Content-Disposition": "attachment; filename=tinyweb-export.json"})


def handle_import_form(msg=""):
    return _respond(
        f"<h1>import</h1>"
        f"<p>Paste the contents of a tinyweb export file (JSON).</p>"
        f'<form method="post" action="/import">'
        f'{_csrf_field()}'
        f'<textarea name="data" rows="12" cols="60" placeholder=\'[{{"url": "...", "note": "..."}}]\'></textarea><br><br>'
        f'<button type="submit">import</button>'
        f"</form>"
        f"<p>{msg}</p>"
        f'<a href="/pages">back</a>'
    )


def handle_import_submit(body):
    raw = body.get("data", [""])[0].strip()
    if not raw:
        return handle_import_form("Paste JSON data.")
    try:
        data = json.loads(raw)
    except json.JSONDecodeError:
        return handle_import_form("Invalid JSON.")
    if not isinstance(data, list):
        return handle_import_form("Expected a JSON array.")

    MAX_IMPORT = 100
    if len(data) > MAX_IMPORT:
        return handle_import_form(f"Too many entries. Maximum is {MAX_IMPORT}.")

    imported = 0
    errors = 0
    for entry in data:
        url = entry.get("url", "").strip()
        note = entry.get("note", "").strip()
        if not url:
            continue
        try:
            index_url(url, note)
            imported += 1
        except Exception:
            errors += 1

    return handle_import_form(f"Imported {imported} page(s). {errors} error(s).")


def handle_reindex_form():
    if get_setting("semantic_search", "0") != "1":
        return _respond(
            f"<h2>semantic search index</h2>"
            f"<p>Semantic search is disabled. Enable it in <a href=\"/style\">settings</a> to use embeddings.</p>"
            f'<p><a href="/">back to search</a></p>'
        )
    db = get_db()
    try:
        total_pages = db.execute("SELECT count(*) FROM pages").fetchone()[0]
        pages_with_chunks = db.execute(
            "SELECT count(DISTINCT page_id) FROM chunks WHERE page_id IS NOT NULL"
        ).fetchone()[0]
    finally:
        return_db(db)
    progress = get_setting("reindex_progress", "")
    status_html = ""
    if progress:
        status_html = f'<p class="meta">Reindex in progress: {esc(progress)}</p>'
    elif _reindex_thread and _reindex_thread.is_alive():
        status_html = '<p class="meta">Reindex running...</p>'
    return _respond(
        f"<h2>semantic search index</h2>"
        f"<p>{pages_with_chunks} of {total_pages} pages have embeddings.</p>"
        f'{status_html}'
        f'<form method="post" action="/reindex">'
        f'{_csrf_field()}'
        f'<button type="submit">reindex all pages</button>'
        f'</form>'
        f'<p><a href="/">back to search</a></p>'
    )


def handle_reindex_submit(body):
    global _reindex_thread
    if _reindex_thread and _reindex_thread.is_alive():
        return handle_reindex_form()

    def _run():
        try:
            from embeddings import reindex_all
            def progress(current, total):
                set_setting("reindex_progress", f"{current}/{total}")
            reindex_all(progress_callback=progress)
        except Exception:
            pass
        finally:
            set_setting("reindex_progress", "")

    _reindex_thread = threading.Thread(target=_run, daemon=True)
    _reindex_thread.start()
    return _redirect("/reindex")