dce16e313e
- _helpers.py: CSRF, FTS sanitizer, pagination, response helpers, tag helpers - search.py: BM25 + hybrid search, trusted/remote result rendering - pages.py: add/edit/delete/bulk/bookmark handlers - subscriptions.py: sync, share preview, API sites, subscription CRUD - customize.py: settings form, about page - tags.py: tag list and tag browse handlers - data.py: export, import, semantic reindex handlers - __init__.py: dispatch, re-exports, forum_plugin, _request_local All 58 external symbols re-exported. No changes to app.py, conftest.py, or any test file.
387 lines
15 KiB
Python
387 lines
15 KiB
Python
from pathlib import Path
|
|
import json
|
|
import secrets
|
|
from urllib.parse import unquote
|
|
|
|
from db import get_db, return_db, get_setting, set_setting, get_site_name, index_url, clean_url
|
|
from templates import esc
|
|
from ._helpers import (
|
|
_csrf_field, _respond, _redirect, _error,
|
|
_paginate, _page_nav, _get_page_tags, _set_page_tags, _cleanup_orphaned_tags,
|
|
_get_bookmark_token, _text_response,
|
|
BROWSE_PER_PAGE,
|
|
)
|
|
|
|
|
|
def handle_add_form(msg="", action_type="index", prefill_url=""):
|
|
if action_type == "subscribe":
|
|
return _respond(
|
|
f"<h1>subscribe</h1>"
|
|
f"<p>Subscribe to a friend's TinyWeb instance to sync their index</p>"
|
|
f'<form method="post" action="/subscriptions/add">'
|
|
f'{_csrf_field()}'
|
|
f'<input name="dest_hash" placeholder="destination hash (32 hex chars)" size="50"><br><br>'
|
|
f'<button type="submit">subscribe</button>'
|
|
f"</form>"
|
|
f"<p><small>or <a href=\"/add\">add a single site</a></small></p>"
|
|
f"<p>{msg}</p>"
|
|
f'<a href="/">back</a>'
|
|
)
|
|
url_value = f'value="{esc(prefill_url)}" ' if prefill_url else ""
|
|
return _respond(
|
|
f"<h1>add url</h1>"
|
|
f"<p>Add a site to your index</p>"
|
|
f'<form method="post" action="/add">'
|
|
f'{_csrf_field()}'
|
|
f'<input name="url" placeholder="https://example.com" size="50" {url_value}><br><br>'
|
|
f'<input name="note" placeholder="why are you saving this? (optional)" size="50"><br><br>'
|
|
f'<input name="tags" placeholder="tags (comma-separated, e.g. solarpunk, mesh)" size="50"><br>'
|
|
f'<small>tag: private to exclude from sharing</small><br><br>'
|
|
f'<button type="submit">index</button>'
|
|
f"</form>"
|
|
f"<p>{msg}</p>"
|
|
f'<a href="/">back</a>'
|
|
)
|
|
|
|
|
|
def handle_add_submit(body):
|
|
input_type = body.get("input_type", ["url"])[0]
|
|
url = body.get("url", [""])[0].strip()
|
|
reticulum_dest = body.get("reticulum_dest", [""])[0].strip().replace("<", "").replace(">", "")
|
|
note = body.get("note", [""])[0].strip()
|
|
tags = body.get("tags", [""])[0].strip()
|
|
|
|
if input_type == "url":
|
|
if not url:
|
|
return handle_add_form("URL is required.")
|
|
url = clean_url(url)
|
|
if not url.startswith(("http://", "https://")):
|
|
return handle_add_form("URL must start with http:// or https://")
|
|
else:
|
|
if not reticulum_dest:
|
|
return handle_add_form("Reticulum destination hash is required.")
|
|
if len(reticulum_dest) != 32 or not all(c in "0123456789abcdefABCDEF" for c in reticulum_dest):
|
|
return handle_add_form("Invalid reticulum destination hash. Must be 32 hex characters.")
|
|
url = f"reticulum:{reticulum_dest}"
|
|
|
|
try:
|
|
title = index_url(url, note, reticulum_dest if reticulum_dest else "")
|
|
if tags:
|
|
db = get_db()
|
|
try:
|
|
row = db.execute("SELECT id FROM pages WHERE url = ?", (url,)).fetchone()
|
|
if row:
|
|
_set_page_tags(row["id"], tags, db)
|
|
db.commit()
|
|
finally:
|
|
return_db(db)
|
|
|
|
return handle_add_form(f'Indexed: {esc(url)}')
|
|
|
|
except ValueError as e:
|
|
return handle_add_form(f"Error: {esc(str(e))}")
|
|
|
|
except Exception as e:
|
|
error_msg = str(e).lower()
|
|
if "block" in error_msg or "cloudflare" in error_msg or "403" in error_msg:
|
|
return _respond(
|
|
f"<h1>add url (manual entry)</h1>"
|
|
f"<p><strong>{esc(url)}</strong> blocks automated access. "
|
|
f"You can still save it manually:</p>"
|
|
f'<form method="post" action="/add/manual">'
|
|
f'{_csrf_field()}'
|
|
f'<input type="hidden" name="url" value="{esc(url)}">'
|
|
f'<input type="hidden" name="note" value="{esc(note)}">'
|
|
f'<input type="hidden" name="tags" value="{esc(tags)}">'
|
|
f'<label>Title:</label><br>'
|
|
f'<input name="manual_title" size="50" placeholder="page title" required><br><br>'
|
|
f'<label>Description:</label><br>'
|
|
f'<textarea name="manual_description" rows="4" cols="50" placeholder="what is this site about? (optional)"></textarea><br><br>'
|
|
f'<button type="submit">save manually</button>'
|
|
f"</form>"
|
|
f'<a href="/">back</a>'
|
|
)
|
|
return handle_add_form(f"Error: could not fetch or index that URL. {esc(str(e)[:100])}")
|
|
|
|
|
|
def handle_add_manual_submit(body):
|
|
url = clean_url(body.get("url", [""])[0].strip())
|
|
note = body.get("note", [""])[0].strip()
|
|
tags = body.get("tags", [""])[0].strip()
|
|
manual_title = body.get("manual_title", [""])[0].strip()
|
|
manual_desc = body.get("manual_description", [""])[0].strip()
|
|
|
|
if not url:
|
|
return handle_add_form("URL is required.")
|
|
|
|
if not manual_title:
|
|
return handle_add_form("Title is required for manual entry.")
|
|
|
|
db = get_db()
|
|
try:
|
|
now = __import__("datetime").datetime.now().strftime("%Y-%m-%dT%H:%M:%S")
|
|
|
|
db.execute(
|
|
"INSERT INTO pages (url, title, body, note, last_modified, summary) VALUES (?, ?, ?, ?, ?, ?) "
|
|
"ON CONFLICT(url) DO UPDATE SET title=excluded.title, body=excluded.body, "
|
|
"note=excluded.note, last_modified=excluded.last_modified, summary=excluded.summary",
|
|
(url, manual_title, manual_desc, note, now, manual_desc[:200]),
|
|
)
|
|
|
|
page_id = db.execute("SELECT id FROM pages WHERE url = ?", (url,)).fetchone()[0]
|
|
|
|
if tags:
|
|
_set_page_tags(page_id, tags, db)
|
|
|
|
db.commit()
|
|
|
|
if get_setting("semantic_search", "0") == "1":
|
|
try:
|
|
from embeddings import store_embeddings
|
|
store_embeddings(page_id, manual_title, manual_desc, db)
|
|
db.commit()
|
|
except Exception as e:
|
|
print(f"Error generating embeddings: {e}")
|
|
|
|
return handle_add_form(f'Added manually: <a href="{esc(url)}" rel="noreferrer noopener">{esc(manual_title)}</a>')
|
|
finally:
|
|
return_db(db)
|
|
|
|
|
|
def handle_pages(query=None):
|
|
msg = query.get("msg", [""])[0] if query else ""
|
|
msg_html = f'<p class="success">{esc(msg)}</p>' if msg else ""
|
|
page = _paginate(query or {})
|
|
offset = (page - 1) * BROWSE_PER_PAGE
|
|
db = get_db()
|
|
try:
|
|
total = db.execute("SELECT count(*) FROM pages").fetchone()[0]
|
|
rows = db.execute(
|
|
"SELECT id, url, title, note FROM pages ORDER BY id DESC LIMIT ? OFFSET ?",
|
|
(BROWSE_PER_PAGE, offset),
|
|
).fetchall()
|
|
items = ""
|
|
for r in rows:
|
|
note_html = f' — <em>{esc(r["note"])}</em>' if r["note"] else ""
|
|
tags = _get_page_tags(r["id"], db)
|
|
tags_html = ""
|
|
if tags:
|
|
tag_links = " ".join(f'<a href="/tags/{esc(t)}">[{esc(t)}]</a>' for t in tags)
|
|
tags_html = f' {tag_links}'
|
|
items += (
|
|
f'<li><label><input type="checkbox" name="ids" value="{r["id"]}"> '
|
|
f'{esc(r["title"])}</label>{note_html}{tags_html} '
|
|
f'<small>(<a href="{esc(r["url"])}" rel="noreferrer noopener">{esc(r["url"])}</a>)</small> '
|
|
f'<a href="/edit/{r["id"]}">edit</a> '
|
|
f'<a href="/delete/{r["id"]}">remove</a></li>'
|
|
)
|
|
finally:
|
|
return_db(db)
|
|
return _respond(
|
|
f"<h1>indexed pages ({total})</h1>"
|
|
f"{msg_html}"
|
|
f'<form method="post" action="/pages/bulk">'
|
|
f'{_csrf_field()}'
|
|
f'<p><label><input type="checkbox" id="select-all"> select all</label></p>'
|
|
f"<ul>{items}</ul>"
|
|
f'{_page_nav(page, total, "/pages", BROWSE_PER_PAGE)}'
|
|
f'<details><summary>bulk actions</summary>'
|
|
f'<p><button type="submit" name="action" value="delete" id="bulk-delete">delete selected</button></p>'
|
|
f'<p><input name="bulk_tags" placeholder="tags (comma-separated)" size="40"> '
|
|
f'<select name="tag_mode"><option value="add">add tags</option><option value="replace">replace tags</option></select> '
|
|
f'<button type="submit" name="action" value="retag">retag selected</button></p>'
|
|
f'</details>'
|
|
f'</form>'
|
|
f'<script>'
|
|
f'document.getElementById("select-all").addEventListener("change",function(){{'
|
|
f'document.querySelectorAll("input[name=ids]").forEach(function(c){{c.checked=this.checked}}.bind(this))'
|
|
f'}});'
|
|
f'document.getElementById("bulk-delete").addEventListener("click",function(e){{'
|
|
f'var n=document.querySelectorAll("input[name=ids]:checked").length;'
|
|
f'if(!n){{e.preventDefault();return}}'
|
|
f'if(!confirm("Delete "+n+" selected page"+(n===1?"":"s")+"?"))e.preventDefault()'
|
|
f'}});'
|
|
f'</script>'
|
|
f'<p><a href="/export">export</a> | <a href="/import">import</a></p>'
|
|
f'<a href="/">back</a>'
|
|
)
|
|
|
|
|
|
def _render_bulk_delete_confirm(page_ids):
|
|
db = get_db()
|
|
try:
|
|
placeholders = ",".join("?" * len(page_ids))
|
|
rows = db.execute(
|
|
f"SELECT id, url, title FROM pages WHERE id IN ({placeholders})",
|
|
page_ids,
|
|
).fetchall()
|
|
finally:
|
|
return_db(db)
|
|
if not rows:
|
|
return _redirect("/pages")
|
|
items = "".join(
|
|
f'<li><b>{esc(r["title"] or r["url"])}</b><br>'
|
|
f'<small>{esc(r["url"])}</small></li>'
|
|
for r in rows
|
|
)
|
|
hidden_ids = "".join(
|
|
f'<input type="hidden" name="ids" value="{int(r["id"])}">' for r in rows
|
|
)
|
|
n = len(rows)
|
|
return _respond(
|
|
f"<h1>confirm delete</h1>"
|
|
f"<p>Remove the following {n} page{'' if n == 1 else 's'}?</p>"
|
|
f"<ul>{items}</ul>"
|
|
f'<form method="post" action="/pages/bulk">'
|
|
f'{_csrf_field()}'
|
|
f'{hidden_ids}'
|
|
f'<input type="hidden" name="action" value="delete">'
|
|
f'<input type="hidden" name="confirmed" value="1">'
|
|
f'<button type="submit">yes, delete {n} page{"" if n == 1 else "s"}</button>'
|
|
f"</form>"
|
|
f' <a href="/pages">cancel</a>'
|
|
)
|
|
|
|
|
|
def handle_bulk_action(body):
|
|
ids = body.get("ids", [])
|
|
action = body.get("action", [""])[0]
|
|
if not ids:
|
|
return _redirect("/pages")
|
|
try:
|
|
page_ids = [int(i) for i in ids]
|
|
except ValueError:
|
|
return _error(400)
|
|
if action == "delete" and body.get("confirmed", [""])[0] != "1":
|
|
return _render_bulk_delete_confirm(page_ids)
|
|
db = get_db()
|
|
try:
|
|
if action == "delete":
|
|
for pid in page_ids:
|
|
db.execute("DELETE FROM page_tags WHERE page_id = ?", (pid,))
|
|
db.execute("DELETE FROM links WHERE page_id = ?", (pid,))
|
|
db.execute("DELETE FROM pages WHERE id = ?", (pid,))
|
|
_cleanup_orphaned_tags(db)
|
|
db.commit()
|
|
elif action == "retag":
|
|
bulk_tags = body.get("bulk_tags", [""])[0].strip()
|
|
tag_mode = body.get("tag_mode", ["add"])[0]
|
|
if bulk_tags:
|
|
for pid in page_ids:
|
|
if tag_mode == "add":
|
|
existing = _get_page_tags(pid, db)
|
|
new_tags = [t.strip().lower() for t in bulk_tags.split(",") if t.strip()]
|
|
merged = ", ".join(sorted(set(existing + new_tags)))
|
|
_set_page_tags(pid, merged, db)
|
|
else:
|
|
_set_page_tags(pid, bulk_tags, db)
|
|
_cleanup_orphaned_tags(db)
|
|
db.commit()
|
|
finally:
|
|
return_db(db)
|
|
return _redirect("/pages")
|
|
|
|
|
|
def handle_edit_form(page_id, msg=""):
|
|
db = get_db()
|
|
try:
|
|
row = db.execute("SELECT id, url, title, body, note, summary FROM pages WHERE id = ?", (page_id,)).fetchone()
|
|
if not row:
|
|
return _error(404)
|
|
tags = ", ".join(_get_page_tags(page_id, db))
|
|
finally:
|
|
return_db(db)
|
|
|
|
return _respond(
|
|
f"<h1>edit page</h1>"
|
|
f"<p><b>{esc(row['title'])}</b><br>"
|
|
f"<small>{esc(row['url'])}</small></p>"
|
|
f'<form method="post" action="/edit/{row["id"]}">'
|
|
f'{_csrf_field()}'
|
|
f'<label>Title:</label><br>'
|
|
f'<input name="title" value="{esc(row["title"])}" size="60"><br><br>'
|
|
f'<label>Summary (shown in search results):</label><br>'
|
|
f'<textarea name="summary" rows="3" cols="60">{esc(row["summary"] or "")}</textarea><br><br>'
|
|
f'<label>Note (why you saved this):</label><br>'
|
|
f'<input name="note" value="{esc(row["note"])}" size="50"><br><br>'
|
|
f'<label>Tags (comma-separated):</label><br>'
|
|
f'<input name="tags" value="{esc(tags)}" size="50"> '
|
|
f'<small>(tag: private to keep private)</small><br><br>'
|
|
f'<button type="submit">save</button>'
|
|
f"</form>"
|
|
f"<p>{msg}</p>"
|
|
f'<a href="/pages">back</a>'
|
|
)
|
|
|
|
|
|
def handle_edit_submit(page_id, body):
|
|
title = body.get("title", [""])[0].strip()
|
|
summary = body.get("summary", [""])[0].strip()
|
|
note = body.get("note", [""])[0].strip()
|
|
tags = body.get("tags", [""])[0].strip()
|
|
|
|
db = get_db()
|
|
try:
|
|
db.execute(
|
|
"UPDATE pages SET title = ?, summary = ?, note = ? WHERE id = ?",
|
|
(title, summary, note, page_id)
|
|
)
|
|
|
|
_set_page_tags(page_id, tags, db)
|
|
_cleanup_orphaned_tags(db)
|
|
|
|
db.commit()
|
|
|
|
finally:
|
|
return_db(db)
|
|
|
|
return _redirect("/pages")
|
|
|
|
|
|
def handle_delete_confirm(page_id):
|
|
db = get_db()
|
|
try:
|
|
row = db.execute("SELECT id, url, title FROM pages WHERE id = ?", (page_id,)).fetchone()
|
|
finally:
|
|
return_db(db)
|
|
if not row:
|
|
return _error(404)
|
|
return _respond(
|
|
f"<h1>confirm delete</h1>"
|
|
f"<p>Remove <b>{esc(row['title'])}</b><br>"
|
|
f"<small>{esc(row['url'])}</small></p>"
|
|
f'<form method="post" action="/delete/{row["id"]}">'
|
|
f'{_csrf_field()}'
|
|
f'<button type="submit">yes, delete</button>'
|
|
f"</form>"
|
|
f' <a href="/pages">cancel</a>'
|
|
)
|
|
|
|
|
|
def handle_delete(page_id):
|
|
db = get_db()
|
|
try:
|
|
db.execute("DELETE FROM page_tags WHERE page_id = ?", (page_id,))
|
|
db.execute("DELETE FROM links WHERE page_id = ?", (page_id,))
|
|
db.execute("DELETE FROM pages WHERE id = ?", (page_id,))
|
|
_cleanup_orphaned_tags(db)
|
|
db.commit()
|
|
finally:
|
|
return_db(db)
|
|
return _redirect("/pages")
|
|
|
|
|
|
def handle_bookmark(query):
|
|
token = query.get("token", [""])[0]
|
|
expected = _get_bookmark_token()
|
|
if not token or not secrets.compare_digest(token, expected):
|
|
return _text_response("error: invalid or missing token", status=403, headers={"Access-Control-Allow-Origin": "*"})
|
|
url = clean_url(query.get("url", [""])[0].strip())
|
|
if not url or not url.startswith(("http://", "https://")):
|
|
return _text_response("error: invalid url", headers={"Access-Control-Allow-Origin": "*"})
|
|
try:
|
|
title = index_url(url)
|
|
msg = f"ok: {title}"
|
|
except Exception as e:
|
|
msg = f"error: {e}"
|
|
return _text_response(msg, headers={"Access-Control-Allow-Origin": "*"})
|