479 lines
17 KiB
Python
479 lines
17 KiB
Python
import threading
|
|
import time
|
|
from datetime import datetime
|
|
|
|
from tinyweb.db import get_db, return_db, get_setting, set_setting, get_site_name, index_url, clean_url
|
|
from tinyweb.templates import esc
|
|
from tinyweb.rns_client import fetch_remote_sites
|
|
from ._helpers import (
|
|
_get_page_tags, _respond, _redirect, _json_response, _error,
|
|
_csrf_field,
|
|
)
|
|
|
|
_sync_threads = {}
|
|
_sync_starts = {}
|
|
_SYNC_TIMEOUT = 120
|
|
|
|
MAX_API_SITES = 5000
|
|
MAX_BROWSE = 5000
|
|
|
|
|
|
def _page_is_shared(tags, mode):
|
|
if "private" in tags:
|
|
return False
|
|
if mode == "require_public" and "public" not in tags:
|
|
return False
|
|
return True
|
|
|
|
|
|
def _shared_sites(db, since=""):
|
|
mode = get_setting("sharing_mode", "exclude_private")
|
|
if since:
|
|
rows = db.execute(
|
|
"SELECT id, url, title, note, last_modified FROM pages "
|
|
"WHERE last_modified > ? ORDER BY id DESC LIMIT ?",
|
|
(since, MAX_API_SITES),
|
|
).fetchall()
|
|
else:
|
|
rows = db.execute(
|
|
"SELECT id, url, title, note, last_modified FROM pages ORDER BY id DESC LIMIT ?",
|
|
(MAX_API_SITES,),
|
|
).fetchall()
|
|
sites = []
|
|
for r in rows:
|
|
tags = _get_page_tags(r["id"], db)
|
|
if not _page_is_shared(tags, mode):
|
|
continue
|
|
sites.append({
|
|
"url": r["url"], "title": r["title"], "note": r["note"],
|
|
"tags": tags, "last_modified": r["last_modified"] or "",
|
|
})
|
|
return sites
|
|
|
|
|
|
def _shared_all_urls(db):
|
|
mode = get_setting("sharing_mode", "exclude_private")
|
|
rows = db.execute(
|
|
"SELECT id, url FROM pages ORDER BY id DESC LIMIT ?", (MAX_API_SITES,)
|
|
).fetchall()
|
|
return [r["url"] for r in rows if _page_is_shared(_get_page_tags(r["id"], db), mode)]
|
|
|
|
|
|
def _count_shared_pages():
|
|
db = get_db()
|
|
try:
|
|
return len(_shared_all_urls(db))
|
|
finally:
|
|
return_db(db)
|
|
|
|
|
|
def handle_share_preview():
|
|
mode = get_setting("sharing_mode", "exclude_private")
|
|
mode_label = (
|
|
"only pages tagged <code>public</code>"
|
|
if mode == "require_public"
|
|
else "all pages except those tagged <code>private</code>"
|
|
)
|
|
sharing_on = get_setting("sharing_enabled", "0") == "1"
|
|
status = (
|
|
'<p>Sharing is <b>enabled</b>. Subscribers see the pages listed below.</p>'
|
|
if sharing_on else
|
|
'<p>Sharing is <b>disabled</b>. Nothing is actually being shared right now; '
|
|
'this is the list that would be exposed if you enabled it.</p>'
|
|
)
|
|
db = get_db()
|
|
try:
|
|
sites = _shared_sites(db)
|
|
finally:
|
|
return_db(db)
|
|
if not sites:
|
|
body = (
|
|
"<h1>sharing preview</h1>"
|
|
f"<p>Rule: {mode_label}.</p>"
|
|
f"{status}"
|
|
"<p><em>No pages match the current rule.</em></p>"
|
|
'<p><a href="/style">back to settings</a></p>'
|
|
)
|
|
return _respond(body)
|
|
rows = ""
|
|
for s in sites:
|
|
tags_html = ""
|
|
if s["tags"]:
|
|
tags_html = " " + " ".join(f"[{esc(t)}]" for t in s["tags"])
|
|
note_html = f' — <em>{esc(s["note"])}</em>' if s["note"] else ""
|
|
rows += (
|
|
f'<li>'
|
|
f'<a href="{esc(s["url"])}" rel="noreferrer noopener">{esc(s["title"] or s["url"])}</a>'
|
|
f'{note_html}{tags_html} '
|
|
f'<br><small>{esc(s["url"])}</small>'
|
|
f'</li>'
|
|
)
|
|
body = (
|
|
"<h1>sharing preview</h1>"
|
|
f"<p>Rule: {mode_label}.</p>"
|
|
f"{status}"
|
|
f"<p><b>{len(sites)}</b> page(s) visible to subscribers.</p>"
|
|
f"<ul>{rows}</ul>"
|
|
'<p><a href="/style">back to settings</a></p>'
|
|
)
|
|
return _respond(body)
|
|
|
|
|
|
def handle_api_sites(query=None):
|
|
if get_setting("sharing_enabled", "0") != "1":
|
|
return _json_response(
|
|
{"error": "sharing disabled"},
|
|
status=403,
|
|
headers={"Access-Control-Allow-Origin": "*"},
|
|
)
|
|
since = (query or {}).get("since", [""])[0].strip()
|
|
db = get_db()
|
|
try:
|
|
sites = _shared_sites(db, since=since)
|
|
all_urls = _shared_all_urls(db) if not since else None
|
|
finally:
|
|
return_db(db)
|
|
data = {"name": get_site_name(), "sites": sites}
|
|
if all_urls is not None:
|
|
data["all_urls"] = all_urls
|
|
return _json_response(data, headers={"Access-Control-Allow-Origin": "*"})
|
|
|
|
|
|
def handle_subscriptions(msg=""):
|
|
db = get_db()
|
|
try:
|
|
subs = db.execute("SELECT * FROM subscriptions ORDER BY id DESC").fetchall()
|
|
finally:
|
|
return_db(db)
|
|
now_t = time.time()
|
|
for sub_id, start_t in list(_sync_starts.items()):
|
|
if now_t - start_t > _SYNC_TIMEOUT:
|
|
set_setting(f"sync_status_{sub_id}", "error:Timed out")
|
|
_sync_threads.pop(sub_id, None)
|
|
_sync_starts.pop(sub_id, None)
|
|
|
|
cards = ""
|
|
for s in subs:
|
|
sub_id = s["id"]
|
|
auto_label = "on" if s["auto_sync"] else "off"
|
|
last = s["last_sync"] or "never"
|
|
sync_status = get_setting(f"sync_status_{sub_id}", "")
|
|
is_syncing = sub_id in _sync_threads and _sync_threads[sub_id].is_alive()
|
|
|
|
if is_syncing:
|
|
status_html = '<div style="margin-top:0.4rem;font-size:0.85rem;color:#2070c0">syncing...</div>'
|
|
elif sync_status.startswith("done:"):
|
|
count = sync_status[5:]
|
|
status_html = f'<div style="margin-top:0.4rem;font-size:0.85rem;color:#30a030">synced {esc(count)} site(s)</div>'
|
|
elif sync_status.startswith("error:"):
|
|
err_msg = sync_status[6:]
|
|
status_html = f'<div style="margin-top:0.4rem;font-size:0.85rem;color:#c03030">{esc(err_msg)}</div>'
|
|
else:
|
|
status_html = ""
|
|
|
|
if is_syncing:
|
|
sync_btn = '<button disabled>syncing...</button>'
|
|
else:
|
|
sync_btn = (
|
|
f'<form method="post" action="/subscriptions/sync/{sub_id}" style="display:inline-block;margin:0">'
|
|
f'{_csrf_field()}<button>sync now</button></form>'
|
|
)
|
|
|
|
cards += (
|
|
f'<div style="border:1px solid #ddd;border-radius:4px;padding:0.9rem 1rem;margin-bottom:0.75rem">'
|
|
f'<div style="margin-bottom:0.4rem"><b>{esc(s["name"] or "unknown")}</b></div>'
|
|
f'<div><small>{esc(s["dest_hash"])}</small></div>'
|
|
f'<div style="margin-top:0.4rem;font-size:0.85rem;color:#606060">last sync: {esc(last)}</div>'
|
|
f'{status_html}'
|
|
f'<div style="display:flex;gap:0.5rem;align-items:center;flex-wrap:wrap;margin-top:0.7rem">'
|
|
f'<a href="/subscriptions/browse/{sub_id}" style="display:inline-flex;align-items:center;padding:0.3em 0">browse</a>'
|
|
f'{sync_btn}'
|
|
f'<form method="post" action="/subscriptions/autosync/{sub_id}" style="display:inline-block;margin:0">'
|
|
f'{_csrf_field()}<button>auto-sync: {auto_label}</button></form>'
|
|
f'<form method="post" action="/subscriptions/delete/{sub_id}" style="display:inline-block;margin:0">'
|
|
f'{_csrf_field()}<button>remove</button></form>'
|
|
f'</div>'
|
|
f'</div>'
|
|
)
|
|
any_syncing = any(
|
|
s["id"] in _sync_threads and _sync_threads[s["id"]].is_alive()
|
|
for s in subs
|
|
)
|
|
head_html = '<meta http-equiv="refresh" content="3">' if any_syncing else ""
|
|
listing = ""
|
|
if subs:
|
|
syncall_btn = '<button disabled>syncing...</button>' if any_syncing else '<button>sync all</button>'
|
|
listing = (
|
|
f'{cards}'
|
|
f'<form method="post" action="/subscriptions/syncall">'
|
|
f'{_csrf_field()}{syncall_btn}</form>'
|
|
)
|
|
return _respond(
|
|
f"<h1>subscriptions</h1>"
|
|
f'<form method="post" action="/subscriptions/add">'
|
|
f'{_csrf_field()}'
|
|
f'<input name="dest_hash" placeholder="destination hash" size="40"> '
|
|
f'<button>subscribe</button>'
|
|
f'</form>'
|
|
f'<p><small>or <a href="/subscriptions/add">subscribe to an instance</a></small></p>'
|
|
f'<p>{msg}</p>'
|
|
f'<hr>{listing}'
|
|
f'<br><a href="/">back</a>',
|
|
head_html=head_html,
|
|
)
|
|
|
|
|
|
def handle_subscription_add(body):
|
|
dest_hash = body.get("dest_hash", [""])[0].strip().replace("<", "").replace(">", "")
|
|
if not dest_hash or len(dest_hash) != 32:
|
|
return handle_subscriptions("Enter a valid 32-character destination hash.")
|
|
try:
|
|
int(dest_hash, 16)
|
|
except ValueError:
|
|
return handle_subscriptions("Invalid destination hash (must be hex).")
|
|
try:
|
|
data = fetch_remote_sites(dest_hash)
|
|
name = data.get("name", "")
|
|
except PermissionError:
|
|
return handle_subscriptions("That instance has sharing disabled.")
|
|
except Exception:
|
|
return handle_subscriptions("Could not reach that instance.")
|
|
db = get_db()
|
|
try:
|
|
db.execute(
|
|
"INSERT INTO subscriptions (dest_hash, name) VALUES (?, ?) "
|
|
"ON CONFLICT(dest_hash) DO UPDATE SET name=excluded.name",
|
|
(dest_hash, name),
|
|
)
|
|
db.commit()
|
|
finally:
|
|
return_db(db)
|
|
return handle_subscriptions(f"Subscribed to {esc(name or dest_hash)}.")
|
|
|
|
|
|
def handle_subscription_browse(sub_id):
|
|
db = get_db()
|
|
try:
|
|
sub = db.execute("SELECT * FROM subscriptions WHERE id = ?", (sub_id,)).fetchone()
|
|
if not sub:
|
|
return _error(404)
|
|
local_urls = set(r["url"] for r in db.execute("SELECT url FROM pages LIMIT ?", (MAX_BROWSE,)).fetchall())
|
|
|
|
remote_rows = db.execute(
|
|
"SELECT url, title, note, tags FROM remote_pages WHERE subscription_id = ? LIMIT ?",
|
|
(sub_id, MAX_BROWSE),
|
|
).fetchall()
|
|
finally:
|
|
return_db(db)
|
|
|
|
if remote_rows:
|
|
sites = []
|
|
for r in remote_rows:
|
|
tags = [t for t in r["tags"].split(",") if t] if r["tags"] else []
|
|
sites.append({"url": r["url"], "title": r["title"], "note": r["note"], "tags": tags})
|
|
else:
|
|
try:
|
|
data = fetch_remote_sites(sub["dest_hash"])
|
|
sites = data.get("sites", [])
|
|
except PermissionError:
|
|
return handle_subscriptions("That instance has sharing disabled.")
|
|
except Exception:
|
|
return handle_subscriptions("Could not fetch sites from that instance.")
|
|
|
|
new_items = ""
|
|
existing_items = ""
|
|
new_count = 0
|
|
for s in sites:
|
|
if s["url"] in local_urls:
|
|
existing_items += (
|
|
f'<li style="opacity:0.5">{esc(s["title"])} '
|
|
f'<small>({esc(s["url"])})</small> — already indexed</li>'
|
|
)
|
|
else:
|
|
new_count += 1
|
|
note_html = f' — <em>{esc(s["note"])}</em>' if s.get("note") else ""
|
|
tags_html = ""
|
|
if s.get("tags"):
|
|
tags_html = " " + " ".join(f'[{esc(t)}]' for t in s["tags"])
|
|
new_items += (
|
|
f'<li><label><input type="checkbox" name="urls" value="{esc(s["url"])}">'
|
|
f' {esc(s["title"])}{note_html}{tags_html}'
|
|
f' <small>({esc(s["url"])})</small></label></li>'
|
|
)
|
|
|
|
buttons = ""
|
|
if new_count:
|
|
buttons = '<button>import selected</button> <button name="import_all" value="1">import all new</button>'
|
|
return _respond(
|
|
f'<h1>browsing: {esc(sub["name"] or sub["dest_hash"])}</h1>'
|
|
f'<p>{len(sites)} site(s) available, {new_count} new</p>'
|
|
f'<form method="post" action="/subscriptions/pick">'
|
|
f'{_csrf_field()}'
|
|
f'<input type="hidden" name="sub_id" value="{sub_id}">'
|
|
f'<ul>{new_items}</ul>'
|
|
f'{buttons}'
|
|
f'</form>'
|
|
f'<h3>already indexed</h3><ul>{existing_items}</ul>'
|
|
f'<a href="/subscriptions">back</a>'
|
|
)
|
|
|
|
|
|
def handle_subscription_pick(body):
|
|
sub_id = body.get("sub_id", [""])[0]
|
|
import_all = body.get("import_all", [""])[0]
|
|
|
|
db = get_db()
|
|
try:
|
|
remote_rows = db.execute(
|
|
"SELECT url, tags FROM remote_pages WHERE subscription_id = ?", (sub_id,)
|
|
).fetchall()
|
|
remote_tags = {r["url"]: r["tags"] for r in remote_rows}
|
|
|
|
if import_all:
|
|
local_urls = set(r["url"] for r in db.execute("SELECT url FROM pages LIMIT ?", (MAX_BROWSE,)).fetchall())
|
|
urls = [r["url"] for r in remote_rows if r["url"] not in local_urls]
|
|
else:
|
|
urls = body.get("urls", [])
|
|
finally:
|
|
return_db(db)
|
|
|
|
if not urls:
|
|
return handle_subscriptions("No sites selected.")
|
|
|
|
imported = 0
|
|
errors = 0
|
|
for url in urls:
|
|
try:
|
|
index_url(url)
|
|
tags_str = remote_tags.get(url, "")
|
|
if tags_str:
|
|
db = get_db()
|
|
try:
|
|
row = db.execute("SELECT id FROM pages WHERE url = ?", (url,)).fetchone()
|
|
if row:
|
|
_set_page_tags(row["id"], tags_str, db)
|
|
db.commit()
|
|
finally:
|
|
return_db(db)
|
|
imported += 1
|
|
except Exception:
|
|
errors += 1
|
|
return handle_subscriptions(f"Imported {imported} page(s). {errors} error(s).")
|
|
|
|
|
|
def _sync_subscription(sub_id):
|
|
db = None
|
|
try:
|
|
set_setting(f"sync_status_{sub_id}", "syncing")
|
|
db = get_db()
|
|
sub = db.execute("SELECT * FROM subscriptions WHERE id = ?", (sub_id,)).fetchone()
|
|
if not sub:
|
|
set_setting(f"sync_status_{sub_id}", "error:Subscription not found.")
|
|
return
|
|
since = sub["last_sync"].replace(" ", "T") if sub["last_sync"] else ""
|
|
try:
|
|
data = fetch_remote_sites(sub["dest_hash"], since=since)
|
|
sites = data.get("sites", [])
|
|
all_urls = data.get("all_urls")
|
|
remote_name = data.get("name", sub["name"])
|
|
except PermissionError:
|
|
set_setting(f"sync_status_{sub_id}", "error:That instance has sharing disabled.")
|
|
return
|
|
except Exception as e:
|
|
set_setting(f"sync_status_{sub_id}", f"error:Could not sync \u2014 {e}")
|
|
return
|
|
|
|
if all_urls is not None:
|
|
existing = db.execute(
|
|
"SELECT id, url FROM remote_pages WHERE subscription_id = ?", (sub_id,)
|
|
).fetchall()
|
|
remote_url_set = set(all_urls)
|
|
for row in existing:
|
|
if row["url"] not in remote_url_set:
|
|
db.execute("DELETE FROM remote_pages WHERE id = ?", (row["id"],))
|
|
|
|
synced = 0
|
|
for s in sites:
|
|
try:
|
|
tags_str = ",".join(s.get("tags", []))
|
|
db.execute(
|
|
"INSERT INTO remote_pages (subscription_id, url, title, note, tags) VALUES (?, ?, ?, ?, ?) "
|
|
"ON CONFLICT(subscription_id, url) DO UPDATE SET title=excluded.title, note=excluded.note, tags=excluded.tags",
|
|
(sub_id, s["url"], s["title"], s.get("note", ""), tags_str),
|
|
)
|
|
if get_setting("semantic_search", "0") == "1":
|
|
try:
|
|
from tinyweb.embeddings import store_remote_embeddings
|
|
rp_id = db.execute(
|
|
"SELECT id FROM remote_pages WHERE subscription_id = ? AND url = ?",
|
|
(sub_id, s["url"]),
|
|
).fetchone()["id"]
|
|
store_remote_embeddings(rp_id, s["title"], s.get("note", ""), db)
|
|
except Exception:
|
|
pass
|
|
synced += 1
|
|
except Exception:
|
|
pass
|
|
now = datetime.now().strftime("%Y-%m-%dT%H:%M:%S")
|
|
db.execute("UPDATE subscriptions SET last_sync = ?, name = ? WHERE id = ?", (now, remote_name, sub_id))
|
|
db.commit()
|
|
set_setting(f"sync_status_{sub_id}", f"done:{synced}")
|
|
except Exception as e:
|
|
set_setting(f"sync_status_{sub_id}", f"error:{e}")
|
|
finally:
|
|
if db:
|
|
return_db(db)
|
|
_sync_threads.pop(sub_id, None)
|
|
_sync_starts.pop(sub_id, None)
|
|
|
|
|
|
def handle_subscription_sync(sub_id):
|
|
if sub_id in _sync_threads and _sync_threads[sub_id].is_alive():
|
|
return _redirect("/subscriptions")
|
|
set_setting(f"sync_status_{sub_id}", "syncing")
|
|
_sync_starts[sub_id] = time.time()
|
|
t = threading.Thread(target=_sync_subscription, args=(sub_id,), daemon=True)
|
|
_sync_threads[sub_id] = t
|
|
t.start()
|
|
return _redirect("/subscriptions")
|
|
|
|
|
|
def handle_subscription_autosync(sub_id):
|
|
db = get_db()
|
|
try:
|
|
db.execute("UPDATE subscriptions SET auto_sync = 1 - auto_sync WHERE id = ?", (sub_id,))
|
|
db.commit()
|
|
finally:
|
|
return_db(db)
|
|
return _redirect("/subscriptions")
|
|
|
|
|
|
def handle_subscription_delete(sub_id):
|
|
db = get_db()
|
|
try:
|
|
db.execute("DELETE FROM remote_pages WHERE subscription_id = ?", (sub_id,))
|
|
db.execute("DELETE FROM subscriptions WHERE id = ?", (sub_id,))
|
|
db.commit()
|
|
finally:
|
|
return_db(db)
|
|
return _redirect("/subscriptions")
|
|
|
|
|
|
def handle_subscription_syncall():
|
|
db = get_db()
|
|
try:
|
|
subs = db.execute("SELECT * FROM subscriptions WHERE auto_sync = 1").fetchall()
|
|
finally:
|
|
return_db(db)
|
|
if not subs:
|
|
return handle_subscriptions("No subscriptions have auto-sync enabled.")
|
|
for sub in subs:
|
|
sub_id = sub["id"]
|
|
if sub_id in _sync_threads and _sync_threads[sub_id].is_alive():
|
|
continue
|
|
set_setting(f"sync_status_{sub_id}", "syncing")
|
|
_sync_starts[sub_id] = time.time()
|
|
t = threading.Thread(target=_sync_subscription, args=(sub_id,), daemon=True)
|
|
_sync_threads[sub_id] = t
|
|
t.start()
|
|
return _redirect("/subscriptions")
|