main.py
Rama main · réplica de lectura del repositorio original.
from __future__ import annotations
import base64
import collections
import contextlib
import hashlib
import html
import io
import json
import logging
import math
import mimetypes
import os
import re
import secrets
import shutil
import sqlite3
import sys
import threading
import time
from contextlib import asynccontextmanager, contextmanager
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any, Callable, Generator, Iterable, Literal, Sequence, cast, overload
from urllib.parse import parse_qs, parse_qsl, quote, quote_plus, unquote, urlencode, urljoin, urlparse, urlsplit, urlunparse, urlunsplit
import feedparser
import httpx
def _parse_month_first_pubdate(date_string: str):
# Handles non-RFC-2822 dates like "May 11, 2026 19:15:50 +0000" (no day-of-week).
try:
dt = datetime.strptime(date_string.strip(), "%B %d, %Y %H:%M:%S %z")
return dt.utctimetuple()
except ValueError:
return None
feedparser.registerDateHandler(_parse_month_first_pubdate)
# The import block below was never isort-ordered; this suppresses I001 for the whole block.
from fastapi import FastAPI, Form, Query, Request # noqa: I001
from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse, Response, StreamingResponse
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates
from PIL import Image as _PILImage
from readability import Document
from starlette.concurrency import run_in_threadpool
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.middleware.gzip import GZipMiddleware
from starlette.middleware.sessions import SessionMiddleware
from services import entry_content, escalation_reprobe
from services import bluesky
from services import flaresolverr as flaresolverr_service
from services import page_fetch
from services import full_content_fetch, itad_waitlist
from services import page_topics
from services.errors import public_error
from services.redirects import local_url
from services import site_content_plugins
from services import publish_date as publish_date_service
from services import deviantart as deviantart_service
from services import devto as devto_service
from services import youtube_oauth as youtube_oauth_service
from services import pinterest_oauth as pinterest_oauth_service
from services import quire as quire_service
from services import reddit as reddit_service
from services import inoreader as inoreader_service
from services import passwords
from services import podcast_audio
from services import podcast_feed_discovery
from services import content_edits
from services import link_canonical
from services import saved_articles as saved_articles_service
from services import saved_autofile as saved_autofile_service # noqa: F401 -- re-exported: routes/saved.py does `from main import saved_autofile_service`
from services import archive_old_stars as archive_old_stars_service # noqa: F401 -- re-exported, same as above
from services import refetch_batch
from services import unstar_tagged as unstar_tagged_service
from services import scraper_service
from services import html_sanitize
from services import tenancy
from services import youtube_embeds
from services import url_guard
from services.webhooks import WEBHOOK_VALID_FORMATS
from services.users import UserStore
from services.email import send_article_email, send_digest_email
from services import feed_discovery
from services.feed_discovery import discover_feed_urls_ex # noqa: F401 -- re-exported: routes/feeds.py does `from main import discover_feed_urls_ex`
from services.feed_refresh import FeedRefreshService
from services.lead_images import LeadImageService, upgrade_image_size_param
from services.reader_api import ReaderApi
from services.starred_archive import StarredArchiveService
from services.youtube import YouTubeDurationService
from services.websub import WebSubService
from services.fever import FeverService
from services.greader import GReaderService
from services.miniflux import MinifluxService
from services.youtube_sync import sync_youtube_folder
from state import (
_AUTOFETCH_HOST_COOLDOWN_S,
_BROWSER_UA_CACHE_TTL,
_DA_SYNC_MAX_AUTO_RESUMES,
_FLARESOLVERR_FEEDS_CACHE_TTL,
_INSTANCE_SETTING_TTL_SECONDS,
_LAST_SEEN_THROTTLE_SECONDS,
_MEDIA_SCAN_TTL_ERROR,
_PROXY_DOWN_COOLDOWN_SECONDS,
_PROXY_FEEDS_CACHE_TTL,
_TAILSCALE_DOWN_COOLDOWN_SECONDS,
_TAILSCALE_FEEDS_CACHE_TTL,
_THUMB_FETCH_FAIL_CACHE, # noqa: F401 -- re-exported: routes/system.py's /thumb proxy does `main._THUMB_FETCH_FAIL_CACHE`
_THUMB_FETCH_FAIL_LOCK, # noqa: F401 -- re-exported, same as above
_THUMB_FETCH_FAIL_TTL, # noqa: F401 -- re-exported, same as above
_app_settings_cache,
_app_settings_cache_lock,
_autofetch_failed_hosts,
_autofetch_hosts_lock,
_browser_ua_cache,
_browser_ua_cache_at,
_browser_ua_cache_lock,
_da_sync_active,
_da_sync_lock,
_enhancement_inflight_feeds,
_enhancement_inflight_lock,
_feed_last_post_cache,
_feed_last_post_cache_lock,
_flaresolverr_feeds_cache,
_flaresolverr_feeds_cache_at,
_flaresolverr_feeds_cache_lock,
_home_request_semaphore, # noqa: F401 -- re-exported: routes/home.py's `home` route does `main._home_request_semaphore`
_instance_setting_cache,
_instance_setting_cache_lock,
_last_seen_touch,
_last_seen_touch_lock,
_login_failures, # noqa: F401 -- re-exported: routes/system.py's login rate-limit helpers do `main._login_failures`
_login_failures_lock, # noqa: F401 -- re-exported, same as above
_media_scan_in_progress,
_media_scan_lock,
_prefetch_header_log_lock,
_prefetch_header_log_remaining,
_proxy_down_lock,
_proxy_down_until,
_proxy_feeds_cache,
_proxy_feeds_cache_at,
_proxy_feeds_cache_lock,
_scheduler_state,
_scheduler_state_lock,
_tag_alias_cache,
_tag_alias_lock,
_tailscale_down_lock,
_tailscale_down_until,
_tailscale_feeds_cache,
_tailscale_feeds_cache_at,
_tailscale_feeds_cache_lock,
FEED_LAST_POST_TTL_SECONDS,
updating_feeds,
updating_feeds_lock,
_PerUserDict, # noqa: F401 -- re-exported: routes/integrations_youtube.py and tests still do `main._PerUserDict`
_ad_asset_hashes_cache,
_ad_asset_hashes_lock,
_autofetch_jobs,
_autofetch_jobs_lock,
_AUTOFETCH_JOB_STALE_S,
_has_manual_tags_cache,
_has_manual_tags_lock,
HAS_MANUAL_TAGS_CACHE_TTL_SECONDS,
_meta_structure_cache,
_meta_structure_lock,
_problematic_feeds_cache,
_problematic_feeds_cache_lock,
PROBLEMATIC_FEEDS_CACHE_TTL_SECONDS,
PROBLEMATIC_FEEDS_LIMIT,
_refetch_jobs,
_refetch_jobs_lock,
feed_title_map_cache,
feed_title_map_cache_lock,
FEED_TITLE_MAP_CACHE_TTL_SECONDS,
tag_counts_cache,
tag_counts_cache_lock,
TAG_COUNTS_CACHE_TTL_SECONDS,
unread_counts_cache,
unread_counts_cache_lock,
UNREAD_COUNTS_CACHE_TTL_SECONDS,
_bump_unread_counts_generation,
get_unread_counts_generation,
invalidate_unread_counts_cache,
try_start_unread_refresh,
clear_unread_refresh_inflight,
next_refresh_rotation_offset,
check_manual_refresh_cooldown,
unread_counts_compute_lock,
)
BASE_DIR = Path(__file__).resolve().parent
# Dedicated app logger. Previously this piggybacked on "uvicorn.error", but
# uvicorn owns that logger and (depending on startup/rollover timing) can leave
# it not propagating to our root file handler — silently dropping every app log
# line ([maintenance], [refresh], …) even though the code ran. "lectio" is our
# own namespace that nothing else reconfigures, so it always reaches root.
LOGGER = logging.getLogger("lectio")
LOGGER.setLevel(logging.INFO)
# Refresh and page-fetch threads keep this one process's GIL busy, and a request makes many short DB calls that each
# have to win it back (5ms per turn by default) -- measured 2026-10-02 as the same steps taking ~50ms at rest and 1-4s
# during a refresh burst. A shorter interval hands it back sooner; throughput cost is small.
try:
sys.setswitchinterval(float(os.getenv("LECTIO_GIL_SWITCH_INTERVAL", "0.001")))
except ValueError:
pass
class _ReaderNonFatalParseWarningFilter(logging.Filter):
"""Filter known non-fatal feed parsing warnings from `reader` logs."""
def filter(self, record: logging.LogRecord) -> bool:
message = record.getMessage()
if not message.startswith("parse "):
return True
suppressed_markers = (
"got NonXMLContentType(",
"got CharacterEncodingOverride(",
)
return not any(marker in message for marker in suppressed_markers)
class _ReaderUpdateFetchErrorFilter(logging.Filter):
"""Demote reader's per-feed fetch/parse failures from ERROR+traceback to a
one-line WARNING.
reader logs every failed ``update_feed`` at ERROR with a full stack trace via
the ``reader.update`` logger — but Lectio already catches each one, applies
backoff, records ``feed_failure_state.last_error``, and surfaces it in the
problematic-feeds panel. So a routine 429/403/timeout/dead-feed produced a
scary red traceback in the log (and the new Admin → Logs tab) for a condition
that's fully handled. Keep the one-line summary as a WARNING but drop the
redundant traceback; genuinely unexpected reader errors stay loud (ERROR)."""
# Signatures of expected, Lectio-handled fetch/parse failures.
_EXPECTED = (
"bad http status code", # any HTTP error status (429/403/404/410/5xx)
"too many requests",
"forbidden",
"not found",
"gone",
"service unavailable",
"timed out",
"timeout",
"connecttimeout",
"readtimeout",
"connectionerror",
"connection reset",
"connection aborted",
"name or service not known",
"temporary failure in name resolution",
"certificate",
"ssl",
"tlsv1",
"nonxmlcontenttype",
"no parser",
"document declared as",
"not well-formed",
)
def filter(self, record: logging.LogRecord) -> bool:
if record.levelno < logging.ERROR:
return True
message = record.getMessage().lower()
if "status=error" not in message and "updating feed" not in message:
return True
if any(marker in message for marker in self._EXPECTED):
record.levelno = logging.WARNING
record.levelname = "WARNING"
record.exc_info = None
record.exc_text = None
return True
def _configure_reader_logging() -> None:
if os.getenv("LECTIO_SUPPRESS_READER_PARSE_WARNINGS", "1") != "1":
return
logging.getLogger("reader").addFilter(_ReaderNonFatalParseWarningFilter())
logging.getLogger("reader.update").addFilter(_ReaderUpdateFetchErrorFilter())
# Query-string secrets that reach the log in request URLs: httpx logs every outgoing URL (the YouTube Data API takes its key as
# ``&key=``, signed image URLs carry a signature) and the access log records incoming ones (OAuth callbacks carry ``code``/``state``).
_LOG_SECRET_RE = re.compile(
r"([?&](?:key|api_key|access_token|refresh_token|client_secret|token|code|state|password|jwt|sig|signature"
r"|x-amz-signature|x-amz-security-token)=)[^&\s\"']+",
re.IGNORECASE,
)
class _RedactingFormatter(logging.Formatter):
"""Log formatter that masks URL query-string secrets before the line is written anywhere."""
def format(self, record: logging.LogRecord) -> str:
return _LOG_SECRET_RE.sub(r"\1<redacted>", super().format(record))
def _configure_persistent_logging() -> None:
"""Attach a rotating file handler when LECTIO_LOG_DIR is set.
Defaults: 5 MB per file, 5 backups (LECTIO_LOG_MAX_BYTES, LECTIO_LOG_BACKUPS).
This is the durable record; :func:`_configure_console_logging` mirrors the same
stream to the console. Without LECTIO_LOG_DIR only the console handler runs.
"""
log_dir_str = os.getenv("LECTIO_LOG_DIR", "").strip()
if not log_dir_str:
return
from logging.handlers import RotatingFileHandler
log_dir = Path(log_dir_str)
log_dir.mkdir(parents=True, exist_ok=True)
max_bytes = int(os.getenv("LECTIO_LOG_MAX_BYTES", str(5 * 1024 * 1024)))
backup_count = int(os.getenv("LECTIO_LOG_BACKUPS", "5"))
handler = RotatingFileHandler(
log_dir / "lectio.log",
maxBytes=max_bytes,
backupCount=backup_count,
encoding="utf-8",
)
handler.setFormatter(_RedactingFormatter("%(asctime)s %(levelname)s %(name)s: %(message)s"))
# Attach to the root logger so it captures everything (uvicorn, reader, app).
root = logging.getLogger()
if root.level > logging.INFO or root.level == logging.NOTSET:
root.setLevel(logging.INFO)
root.addHandler(handler)
def _configure_console_logging() -> None:
"""Mirror everything the app logs to the console, so `docker compose logs` shows it.
uvicorn installs handlers on its own loggers with propagate=False, so its startup lines appear on the
console while every app, reader and library log record propagates to the root logger — which, in the
container, owns only the rotating file handler under LECTIO_LOG_DIR. The result was that all app logging
was invisible to `docker compose logs` and readable only from inside the container. That cost hours during
the 2026-08-18 restart hunt: the scheduler watchdog logged two ERROR lines before every exit and the
container log showed nothing but a silent restart.
Writes to stderr, which is where uvicorn's own handlers write, so the interleaving stays sane. The check
below excludes FileHandler because RotatingFileHandler is itself a StreamHandler subclass.
"""
root = logging.getLogger()
for existing in root.handlers:
if isinstance(existing, logging.StreamHandler) and not isinstance(existing, logging.FileHandler):
if not isinstance(existing.formatter, _RedactingFormatter):
existing.setFormatter(_RedactingFormatter(existing.formatter._fmt if existing.formatter else None))
return
handler = logging.StreamHandler()
handler.setFormatter(_RedactingFormatter("%(asctime)s %(levelname)s %(name)s: %(message)s"))
# INFO by default, because the whole point is that app logs become visible. But loosening the root
# level is a decision about someone else's process, so LECTIO_LOG_LEVEL takes precedence when set —
# a deployment that wants a quieter console says so instead of being overruled.
wanted = os.getenv("LECTIO_LOG_LEVEL", "").strip().upper()
if wanted:
root.setLevel(getattr(logging, wanted, logging.INFO))
elif root.level > logging.INFO or root.level == logging.NOTSET:
root.setLevel(logging.INFO)
root.addHandler(handler)
def _configure_access_log_filter() -> None:
# Kept for backward-compat with previous wiring; the actual access log is
# now emitted by _AccessLogMiddleware (uvicorn's --no-access-log disables
# the built-in one whose filters we couldn't reliably hook).
pass
def _attach_pending_access_filter() -> None:
pass
_configure_reader_logging()
_configure_console_logging()
_configure_persistent_logging()
_configure_access_log_filter()
def _load_local_env(env_path: str | Path | None = None) -> None:
"""Load simple KEY=VALUE pairs from a .env file into os.environ if not already set.
This is intentionally minimal (no interpolation) and avoids adding a dependency.
"""
candidates = []
if env_path:
candidates.append(Path(env_path))
candidates.append(BASE_DIR / ".env")
candidates.append(Path.cwd() / ".env")
for p in candidates:
try:
if not p or not p.exists():
continue
with p.open("r", encoding="utf-8") as fh:
for raw in fh:
line = raw.strip()
if not line or line.startswith("#"):
continue
if "=" not in line:
continue
key, val = line.split("=", 1)
key = key.strip()
val = val.strip()
if (val.startswith('"') and val.endswith('"')) or (val.startswith("'") and val.endswith("'")):
val = val[1:-1]
# Do not overwrite existing environment vars
if key and os.getenv(key) is None:
os.environ[key] = val
break
except Exception:
continue
# Load local .env early so os.getenv() below can pick up values provided by the developer
_load_local_env()
# Default to BASE_DIR/data (matching the Docker ./data:/data mapping) so a local
# run without LECTIO_DATA_DIR set writes DBs and service dirs (scraped-feeds/,
# deviantart-feeds/) into ./data instead of polluting the repo root.
# expanduser so a tilde path (LECTIO_DATA_DIR=~/.lectio) resolves against $HOME
# rather than being treated as a literal "./~" segment.
_data_dir_env = os.getenv("LECTIO_DATA_DIR")
DATA_DIR = Path(os.path.expanduser(_data_dir_env) if _data_dir_env else BASE_DIR / "data").resolve()
DATA_DIR.mkdir(parents=True, exist_ok=True)
THUMB_DB_PATH = DATA_DIR / "lectio_thumb_cache.sqlite"
# Global, content-addressed (source-URL hash -> original image bytes) cache for the
# /api/img proxy. Like the thumb cache, it holds no per-user data and is NOT routed
# through tenancy. See ensure_img_cache_schema / api_img_proxy.
IMG_CACHE_DB_PATH = DATA_DIR / "lectio_img_cache.sqlite"
# Global, content-addressed (video_id -> duration): a fact about the video, not
# per-user, so it's shared — a user without a YouTube key still gets durations
# another user's key already fetched. Like the thumb cache, NOT routed through tenancy.
YT_DURATION_DB_PATH = DATA_DIR / "lectio_yt_durations.sqlite"
# Global account registry (NOT per-user, NOT routed through tenancy): one users
# table for the whole instance.
AUTH_DB_PATH = DATA_DIR / "lectio_auth.sqlite"
THUMB_CACHE_DIR = DATA_DIR / "thumb_cache" # legacy on-disk cache; entries migrate lazily on access
# Bind the tenancy resolver: per-user DBs live under DATA_DIR/users/{uid}/. The thumb
# cache (THUMB_DB_PATH) is intentionally NOT routed through tenancy — it is a
# content-addressed global cache shared across all users. See services/tenancy.py.
tenancy.configure(data_dir=DATA_DIR)
ROOT_FOLDER_NAME = "All Feeds"
_LECTIO_FOLDER_NAME = "_Lectio"
# Virtual, derived folder pinned to the bottom of the sidebar that holds every
# subscribed feed with no folder_feeds row (e.g. feeds imported via OPML/reader
# migration that never got a folder). It has no DB row; its id is a negative
# sentinel so it never collides with real (positive) folder ids, and its
# membership is computed per-render as (all reader feeds − foldered feeds).
UNCATEGORIZED_FOLDER_ID = -1
UNCATEGORIZED_FOLDER_NAME = "Uncategorized"
scraper_service.init(DATA_DIR)
deviantart_service.init(DATA_DIR)
devto_service.init(DATA_DIR)
def _da_seed_lead_image(feed_url: str, entry_id: str, image_url: str) -> None:
"""Sink for the DeviantArt service: store the API image as the entry's lead
image (DB + live cache) unless one is already cached (preserve customizations)."""
try:
if not lead_image_service.get_cached_entry_thumbnail(feed_url, entry_id, ""):
lead_image_service.store_entry_lead_image(feed_url, entry_id, image_url)
except Exception:
LOGGER.exception("[deviantart] failed to seed lead image for %s", entry_id)
deviantart_service.set_lead_image_sink(_da_seed_lead_image)
def _devto_seed_lead_image(feed_url: str, entry_id: str, image_url: str) -> None:
"""Sink for the dev.to service: store the API cover image as the entry's lead
image (DB + live cache) unless one is already cached (preserve customizations)."""
try:
if not lead_image_service.get_cached_entry_thumbnail(feed_url, entry_id, ""):
lead_image_service.store_entry_lead_image(feed_url, entry_id, image_url)
except Exception:
LOGGER.exception("[devto] failed to seed lead image for %s", entry_id)
devto_service.set_lead_image_sink(_devto_seed_lead_image)
DEFAULT_AUTO_REFRESH_MINUTES = 60
MIN_AUTO_REFRESH_MINUTES = 5
MANUAL_REFRESH_COOLDOWN_SECONDS = 60
FAILED_FEED_BACKOFF_BASE_SECONDS = 60
FAILED_FEED_BACKOFF_MAX_SECONDS = 60 * 60 * 24
AUTO_REFRESH_SETTING_KEY = "auto_refresh_minutes"
SORT_BY_SETTING_KEY = "sort_by"
SORT_DIR_SETTING_KEY = "sort_dir"
# Feeds and Saved remember their sort SEPARATELY. One shared pair meant picking
# an order in Saved silently changed Feeds, and vice versa — the two are
# different jobs (a publish-date backlog vs a to-do pile), so they want
# different orders. The unprefixed keys stay Feeds' so existing installs keep
# their remembered value.
SAVED_SORT_BY_SETTING_KEY = "saved_sort_by"
SAVED_SORT_DIR_SETTING_KEY = "saved_sort_dir"
def sort_setting_keys(star_only: bool) -> tuple[str, str]:
"""The (sort_by, sort_dir) setting keys for the active scope."""
if star_only:
return SAVED_SORT_BY_SETTING_KEY, SAVED_SORT_DIR_SETTING_KEY
return SORT_BY_SETTING_KEY, SORT_DIR_SETTING_KEY
GLOBAL_NOTE_SETTING_KEY = "global_note"
PROBLEMATIC_FEEDS_LAST_VIEWED_AT_SETTING_KEY = "problematic_feeds_last_viewed_at"
YOUTUBE_SYNC_LAST_AT_KEY = "youtube_sync_last_at"
YOUTUBE_SYNC_LAST_RESULT_KEY = "youtube_sync_last_result"
EMAIL_TO_SETTING_KEY = "email_to"
EMAIL_BCC_SETTING_KEY = "email_bcc_address"
PROFILE_NAME_SETTING_KEY = "profile_name"
PROFILE_EMAIL_SETTING_KEY = "profile_email"
SETTING_TZ_DISPLAY = "tz_display"
SETTING_PORTRAIT_IMG_MAX_WIDTH = "portrait_img_max_width"
# Per-user: Ctrl/Cmd-click or middle-click on a post title (opens the article in a new tab without opening it in the reader) marks the
# post read. Off ("0") by default: those clicks leave it unread.
SETTING_MARK_READ_ON_NEW_TAB = "mark_read_on_new_tab"
# Per-user: rewrite every third-party <img src> in the article pane body to
# /api/img?u=… instead of loading it directly. Off ("0") by default — raises
# /api/img traffic and cache size, and drops srcset (so high-DPI screens fetch
# the proxied single-resolution src). Read Mode always does this regardless of
# this setting (see proxy_all_body_images).
SETTING_PROXY_BODY_IMAGES = "proxy_body_images"
# Outbound-fetch SOCKS5/HTTP proxy (e.g. a gluetun VPN container). URL is
# admin-only (a non-admin choosing an arbitrary proxy target is SSRF-adjacent);
# mode is per-user-overridable — "" means inherit the instance default. See
# get_proxy_url / get_proxy_mode.
SETTING_PROXY_URL = "proxy_url"
SETTING_PROXY_MODE = "proxy_mode" # "off" / "as_needed" / "always"
# Last-resort outbound proxy (e.g. a Tailscale exit node reaching the real home
# IP) — a different trust tier from SETTING_PROXY_URL: real exposure cost, not a
# disposable datacenter NAT. Rides the SAME proxy_mode as the primary proxy
# rather than its own mode (Josh's own framing: mirror the existing pattern,
# not invent a new one) — its actual gate is simply being unconfigured by
# default, same as the primary proxy. Only ever reached via as_needed's
# per-feed escalation, one rung past the primary proxy, never via "always"
# (that would route every fetch through the home IP, which "last resort" rules
# out). URL is admin-only for the same SSRF-adjacent reason as SETTING_PROXY_URL.
SETTING_TAILSCALE_URL = "tailscale_url"
# FlareSolverr (real headless Chrome, purpose-built to solve Cloudflare/DDoS-
# Guard style challenges) — one rung past the primary proxy, before the
# last-resort backend above. Rides the same proxy_mode too. No token/auth (an
# internal-network-only container); zero personal exposure — always stacked
# with the primary proxy (SETTING_PROXY_URL) when one is configured, never
# with the last-resort one, since it doesn't need the home IP to work and
# spending it here would defeat "last resort". Gated narrower than the other
# two tiers: only a real challenge page (bot_challenge.FeedBlockedError, see
# FeedRefreshService._is_bot_challenge), not any refusal — a plain IP-block
# 403 wants a different exit IP, not a browser solving a challenge that was
# never served.
SETTING_FLARESOLVERR_URL = "flaresolverr_url"
SETTING_MAINTENANCE_HOUR = "maintenance_hour"
SETTING_IMG_CACHE_DAYS = "img_cache_days"
SETTING_IMG_CACHE_MAX_DIM = "img_cache_max_dim"
SETTING_IMG_TARGET_BYTES = "img_target_bytes"
SETTING_YT_API_KEY = "yt_api_key"
SETTING_YT_CHANNEL_ID = "yt_channel_id"
SETTING_YT_FOLDER_NAME = "yt_folder_name"
# Per-user: use the standard youtube.com embed host (account features — Share /
# Watch Later — work) instead of the privacy-enhanced youtube-nocookie.com. Off
# (privacy) by default; the standard host sets YouTube cookies. "1"/"0".
SETTING_YT_EMBED_ACCOUNT_FEATURES = "yt_embed_account_features"
# Global per-user toggle: auto-mark YouTube Shorts read across ALL YouTube feeds at
# refresh, regardless of the per-feed hide_shorts pref. Off ("0") by default.
SETTING_YT_HIDE_SHORTS_GLOBAL = "yt_hide_shorts_global"
# Global per-user toggle: don't show YouTube videos that haven't premiered yet
# (liveBroadcastContent == "upcoming") across ALL YouTube feeds, regardless of
# the per-feed hide_unpremiered pref. Off ("0") by default.
SETTING_YT_HIDE_UNPREMIERED_GLOBAL = "yt_hide_unpremiered_global"
# Global per-user toggle: auto-mark members-only videos read across ALL YouTube
# feeds at refresh, regardless of the per-feed hide_members_only pref. Off
# ("0") by default. Detection has a real per-video cost (a watch-page fetch —
# unlike hide_shorts/hide_unpremiered, membership status isn't in the Data API
# response already fetched for duration/premiere status), so it only runs for
# entries that have never been checked before (see youtube_video_duration's
# members_only column).
SETTING_YT_HIDE_MEMBERS_ONLY_GLOBAL = "yt_hide_members_only_global"
# Global per-user toggle: hide a webcomic strip a publisher has locked behind a
# supporter tier (cad-comic.com and the like) until its own stated unlock date,
# across ALL webcomic feeds, regardless of the per-feed hide_locked_comics pref.
# Not YouTube-specific — see LeadImageService.check_and_cache_webcomic_lock.
# Off ("0") by default.
SETTING_HIDE_LOCKED_COMICS_GLOBAL = "hide_locked_comics_global"
# Daily YouTube Data API quota cap (units). Google's default is 10,000/day; make it
# a setting in case a higher quota is granted.
SETTING_YT_QUOTA_CAP = "yt_quota_cap"
YT_QUOTA_DEFAULT_CAP = 10000
# Watermark for the auto-add-to-playlist rule's "what's new since I last
# checked" scan — see _run_youtube_playlist_rules_after_refresh.
SETTING_YT_PLAYLIST_AUTO_LAST_CHECK = "yt_playlist_auto_last_check"
# "On star, also send to…" — per-user destinations fired when an article is starred.
SETTING_STAR_SEND_INSTAPAPER = "star_send_instapaper" # "1"/"0"
SETTING_STAR_SEND_YT_PLAYLIST = "star_send_yt_playlist" # playlist id ("" = off)
SETTING_STAR_SEND_YT_PLAYLIST_TITLE = "star_send_yt_playlist_title"
SETTING_STAR_SEND_EMAIL = "star_send_email" # address ("" = off)
SETTING_RESEND_API_KEY = "resend_api_key"
SETTING_EMAIL_FROM = "email_from"
SETTING_INSTAPAPER_USERNAME = "instapaper_username"
SETTING_INSTAPAPER_PASSWORD = "instapaper_password"
SETTING_DEVIANTART_CLIENT_ID = "deviantart_client_id"
SETTING_DEVIANTART_CLIENT_SECRET = "deviantart_client_secret"
SETTING_DEVIANTART_ACCESS_TOKEN = "deviantart_access_token"
SETTING_DEVIANTART_REFRESH_TOKEN = "deviantart_refresh_token"
SETTING_DEVIANTART_TOKEN_EXPIRES_AT = "deviantart_token_expires_at"
SETTING_DEVIANTART_USERNAME = "deviantart_username"
SETTING_DEVIANTART_OAUTH_STATE = "deviantart_oauth_state"
SETTING_DEVIANTART_OAUTH_VERIFIER = "deviantart_oauth_verifier"
SETTING_DEVIANTART_FOLDER_NAME = "deviantart_folder_name"
SETTING_DEVIANTART_SYNC_STATUS = "deviantart_sync_status"
# Structured companion to the status string: JSON {failed:[{username,error}], unwatched:[username]}
# so the Settings UI can list the failed/unwatched artists as profile links instead
# of telling the user to "see logs" (which they have no access to).
SETTING_DEVIANTART_SYNC_DETAIL = "deviantart_sync_detail"
# Set "1" the moment a sync finds a NEWLY-unwatched (now auto-paused) artist;
# cleared only when the DeviantArt Integrations sub-tab is actually opened —
# drives the Integrations/DeviantArt tab dots, deliberately separate from
# whether the unwatched list itself is empty (that list stays populated for
# Josh to act on "at some point or not"; this flag is just "there's something
# new in it you haven't looked at yet").
SETTING_DEVIANTART_UNWATCHED_DIRTY = "deviantart_unwatched_dirty"
SETTING_YT_OAUTH_CLIENT_ID = "yt_oauth_client_id"
SETTING_YT_OAUTH_CLIENT_SECRET = "yt_oauth_client_secret"
SETTING_YT_OAUTH_ACCESS_TOKEN = "yt_oauth_access_token"
SETTING_YT_OAUTH_REFRESH_TOKEN = "yt_oauth_refresh_token"
SETTING_YT_OAUTH_TOKEN_EXPIRES_AT = "yt_oauth_token_expires_at"
SETTING_YT_OAUTH_STATE = "yt_oauth_state"
# Shared-instance YouTube OAuth creds stored in the admin's app_settings.
SETTING_SHARED_YT_OAUTH_CLIENT_ID = "shared_yt_oauth_client_id"
SETTING_SHARED_YT_OAUTH_CLIENT_SECRET = "shared_yt_oauth_client_secret"
# Pinterest outbound (save / pin an article to a board).
SETTING_PINTEREST_OAUTH_CLIENT_ID = "pinterest_oauth_client_id"
SETTING_PINTEREST_OAUTH_CLIENT_SECRET = "pinterest_oauth_client_secret"
SETTING_PINTEREST_OAUTH_ACCESS_TOKEN = "pinterest_oauth_access_token"
SETTING_PINTEREST_OAUTH_REFRESH_TOKEN = "pinterest_oauth_refresh_token"
SETTING_PINTEREST_OAUTH_TOKEN_EXPIRES_AT = "pinterest_oauth_token_expires_at"
SETTING_PINTEREST_OAUTH_STATE = "pinterest_oauth_state"
# Shared-instance Pinterest OAuth creds stored in the admin's app_settings.
SETTING_SHARED_PINTEREST_OAUTH_CLIENT_ID = "shared_pinterest_oauth_client_id"
SETTING_SHARED_PINTEREST_OAUTH_CLIENT_SECRET = "shared_pinterest_oauth_client_secret"
# Inoreader migration — OAuth tokens and import checkpoint.
SETTING_INOREADER_CLIENT_ID = "inoreader_client_id"
SETTING_INOREADER_CLIENT_SECRET = "inoreader_client_secret"
SETTING_INOREADER_ACCESS_TOKEN = "inoreader_access_token"
SETTING_INOREADER_REFRESH_TOKEN = "inoreader_refresh_token"
SETTING_INOREADER_TOKEN_EXPIRES_AT = "inoreader_token_expires_at"
SETTING_INOREADER_OAUTH_STATE = "inoreader_oauth_state"
SETTING_INOREADER_IMPORT_STATE = "inoreader_import_state"
SETTING_INOREADER_EXPORT_DIR = "inoreader_export_dir" # server-side path to JSON export files
# Miniflux migration (FROM a Miniflux instance — distinct from the Miniflux API compat server).
SETTING_MINIFLUX_IMPORT_URL = "miniflux_import_url"
SETTING_MINIFLUX_IMPORT_TOKEN = "miniflux_import_token"
SETTING_MINIFLUX_IMPORT_STATE = "miniflux_import_state"
# FreshRSS migration.
SETTING_FRESHRSS_URL = "freshrss_url"
SETTING_FRESHRSS_USERNAME = "freshrss_username"
SETTING_FRESHRSS_PASSWORD = "freshrss_password"
SETTING_FRESHRSS_IMPORT_STATE = "freshrss_import_state"
# tt-rss migration.
SETTING_TTRSS_URL = "ttrss_url"
SETTING_TTRSS_USERNAME = "ttrss_username"
SETTING_TTRSS_PASSWORD = "ttrss_password"
SETTING_TTRSS_IMPORT_STATE = "ttrss_import_state"
# Reddit OAuth integration.
SETTING_REDDIT_CLIENT_ID = "reddit_client_id"
SETTING_REDDIT_CLIENT_SECRET = "reddit_client_secret"
SETTING_REDDIT_ACCESS_TOKEN = "reddit_oauth_access_token"
SETTING_REDDIT_REFRESH_TOKEN = "reddit_oauth_refresh_token"
SETTING_REDDIT_TOKEN_EXPIRES_AT = "reddit_oauth_token_expires_at"
SETTING_REDDIT_OAUTH_STATE = "reddit_oauth_state"
SETTING_REDDIT_USERNAME = "reddit_username"
SETTING_SHARED_REDDIT_CLIENT_ID = "shared_reddit_client_id"
SETTING_SHARED_REDDIT_CLIENT_SECRET = "shared_reddit_client_secret"
SETTING_STAR_SEND_REDDIT_SUBREDDIT = "star_send_reddit_subreddit"
# Instance tuning settings (admin-only, stored in admin's app_settings).
SETTING_FETCH_HISTORY_MAX_AGE_DAYS = "fetch_history_max_age_days"
# Auto-read age (days; 0 = off) at instance and account level. Each level below can only shorten it: folder, then feed.
SETTING_AUTO_READ_DAYS_INSTANCE = "auto_read_days_instance"
SETTING_AUTO_READ_DAYS_USER = "auto_read_days_user"
SETTING_TOMBSTONE_SWEEP_DAYS = "tombstone_sweep_days"
SETTING_LOGIN_MAX_FAILURES = "login_max_failures"
SETTING_LOGIN_WINDOW_SECONDS = "login_window_seconds"
SETTING_DEFAULT_AUTO_REFRESH_MINUTES = "default_auto_refresh_minutes"
# Quire outbound (turn an article into a task in a chosen Quire project).
SETTING_QUIRE_CLIENT_ID = "quire_client_id"
SETTING_QUIRE_CLIENT_SECRET = "quire_client_secret"
SETTING_QUIRE_ACCESS_TOKEN = "quire_access_token"
SETTING_QUIRE_REFRESH_TOKEN = "quire_refresh_token"
SETTING_QUIRE_TOKEN_EXPIRES_AT = "quire_token_expires_at"
SETTING_QUIRE_OAUTH_STATE = "quire_oauth_state"
SETTING_QUIRE_USERNAME = "quire_username"
SETTING_QUIRE_PROJECT_OID = "quire_project_oid" # default destination project ("" = none)
SETTING_QUIRE_PROJECT_NAME = "quire_project_name" # cached display name of that project
SETTING_STAR_SEND_QUIRE = "star_send_quire" # "1"/"0"
# Quire rate limits are per-organization, per-minute and per-hour (Free: 50/min, 200/hr).
# These are Lectio's own sliding-window caps used to drive the usage meter + back off.
SETTING_QUIRE_RATE_CAP_MIN = "quire_rate_cap_min"
SETTING_QUIRE_RATE_CAP_HOUR = "quire_rate_cap_hour"
SETTING_QUIRE_PLAN = "quire_plan" # detected plan name of the destination project's org (display only)
QUIRE_RATE_DEFAULT_CAP_MIN = 50
QUIRE_RATE_DEFAULT_CAP_HOUR = 200
AUTO_REFRESH_OPTION_MINUTES = (0, 5, 15, 30, 60, 360, 720)
SCHEDULER_POLL_SECONDS = 30
DEFAULT_SORT_BY = "post"
DEFAULT_SORT_DIR = "asc"
CHUNK_SIZE = 20
# feed_fetch_history retention configured via the Administration panel.
READABILITY_USER_AGENT = "Lectio/0.1 (+https://localhost)"
# Honest identifier for outbound fetches — names the app and links to the repo,
# the good-citizen behavior some hosts (e.g. rachelbythebay.com) explicitly
# reward. This is the DEFAULT for podcast discovery/borrow fetches.
LECTIO_HONEST_USER_AGENT = "Lectio/0.1 (+https://github.com/joshg253/Lectio)"
# Some podcast-host feeds (Buzzsprout, Libsyn, …) sit behind Cloudflare, which
# 403s any non-browser UA. We escalate to this browser UA ONLY after the honest
# request is actually refused (see _polite_safe_get) — never preemptively, so we
# don't spoof hosts that are happy to serve Lectio.
PODCAST_FETCH_USER_AGENT = "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0 Safari/537.36"
# In-memory cache of domains known to have Cross-Origin-Resource-Policy restrictions.
# Values: True = same-site/same-origin (proxy needed), False = no restriction.
_CORP_DOMAIN_CACHE: dict[str, bool] = {}
# Hosts that serve a "this image was hotlinked" placeholder when the request
# carries a foreign Referer (or no same-origin signal). referrerpolicy="no-referrer"
# defeats this for *fresh* image loads, but a browser that cached the placeholder
# under the (unchanged) image URL keeps serving it. Routing these hosts' images
# through the /api/img proxy gives the browser a new same-origin URL it hasn't
# cached, and the server fetch carries no Referer — so the real image loads and
# stays correct. Add a registrable domain (matches it and any subdomain).
# wixmp.com serves DeviantArt images behind short-lived signed (?token=) URLs;
# proxying caches the bytes server-side so the article image survives token expiry.
# private-user-images.githubusercontent.com serves GitHub release/issue screenshots
# behind short-lived (~5 min) JWT-signed URLs — same problem, same fix.
# fabiensanglard.net is the *inverse* hotlink case: it 403s an image fetched with
# no Referer (e.g. its .webp files) but serves it with a same-origin Referer. The
# browser can't send fabiensanglard's own origin as Referer, so a direct <img>
# load fails (its .webp break in reader/web view while its .jpg loads); routing
# through /api/img lets the proxy's same-origin-Referer retry (api_img_proxy)
# fetch the real bytes. See ARCHITECTURE "Same-origin Referer".
_HOTLINK_IMG_HOSTS: frozenset[str] = frozenset(
{"nanolx.org", "wixmp.com", "private-user-images.githubusercontent.com", "fabiensanglard.net"}
)
def _is_hotlink_img_host(netloc: str) -> bool:
host = (netloc or "").split("@")[-1].split(":")[0].lower()
return any(host == h or host.endswith("." + h) for h in _HOTLINK_IMG_HOSTS)
MANUAL_TAG_KEY_PREFIX = "lectio.manual_tag."
MAX_MANUAL_TAGS = 12
# Hard cap on how many of an entry's feed tags are carried to the page. RPS
# ships 28 on a single post, so this is generous on purpose — the cost of one
# extra chip is that it is ignored, and the cost of a missing one is that the
# tag cannot be filtered on at all.
MAX_FEED_TAG_SUGGESTIONS = 40
# How many chips the row shows before the "+N more" expander. Eight fits the
# header without wrapping; the rest are one click away rather than absent.
FEED_TAG_CHIPS_COLLAPSED = 8
TAG_VALUE_PATTERN = re.compile(r"^[A-Za-z0-9_.#+][A-Za-z0-9_.#+-]{0,31}$")
def _static_asset_version() -> str:
"""Cache-buster for every ?v=-versioned static asset.
Hashes ALL css/js under static/ rather than a hand-kept list. The list went
stale exactly as you would expect: offline-probe.js and js/cleanup.js are
both served with ?v={{ static_asset_version }} and neither was in it, so
editing them left the version unchanged and browsers — and the service
worker, which caches /static by design — kept serving the previous file.
Deploying a fix and watching the old copy run is a bad way to spend an hour.
`.webmanifest` is in the set for the same reason: it is served with the same
?v= and the same immutable, year-long Cache-Control, so a manifest edit that
did not move the digest would be pinned in the browser indefinitely.
Sorted so the digest is stable across filesystems.
"""
try:
_static = Path(__file__).parent / "static"
paths = sorted(p for p in _static.rglob("*") if p.is_file() and p.suffix in (".css", ".js", ".webmanifest"))
combined = b"".join(p.read_bytes() for p in paths)
return hashlib.md5(combined).hexdigest()[:10]
except Exception:
return "dev"
STATIC_ASSET_VERSION = os.getenv("LECTIO_ASSET_VERSION") or _static_asset_version()
REFRESH_DEBUG_ENABLED = os.getenv("LECTIO_REFRESH_DEBUG", "0") == "1"
# Gates the slow-SQL timing/progress-step instrumentation (Plan.md Tier 1
# refresh-contention item) on both the meta and reader DB connections. Off by
# default: a sqlite3 progress handler fires every 1000 VM instructions on
# every statement, and the timing wrapper adds a perf_counter() pair per call
# -- real, if small, overhead on every single query, all the time, not worth
# paying outside an active investigation. Flip on when chasing a live stall.
LECTIO_PERF_DEBUG = os.getenv("LECTIO_PERF_DEBUG", "0") == "1"
def _env_int(name: str, default: int) -> int:
try:
return int(os.getenv(name, str(default)))
except ValueError:
return default
# /api/img proxy cache tuning (instance-level; admins can override in the
# Administration page, which takes precedence over these env fallbacks).
# LECTIO_IMG_CACHE_DAYS last-accessed TTL in days; 0 = keep forever (default 90)
# LECTIO_IMG_CACHE_MAX_DIM longest-side px to downscale stored images to (default 3840)
_ENV_IMG_CACHE_DAYS = _env_int("LECTIO_IMG_CACHE_DAYS", 90)
_ENV_IMG_CACHE_MAX_DIM = _env_int("LECTIO_IMG_CACHE_MAX_DIM", 3840)
# Byte budget for a cached image. The dimension cap above is not a size cap — a
# 4K RGBA PNG sits exactly at MAX_DIM and still weighs ~12 MB, which is then
# served whole on every view. Over this, the image is re-encoded to WebP
# (lossless first, so screenshots and line art keep their edges). 0 disables.
_ENV_IMG_TARGET_BYTES = _env_int("LECTIO_IMG_TARGET_BYTES", 1_500_000)
# Max width (px) for portrait (taller-than-wide) article images; 0 = disabled.
_ENV_PORTRAIT_IMG_MAX_WIDTH = _env_int("LECTIO_PORTRAIT_IMG_MAX_WIDTH", 650)
# Outbound-fetch proxy endpoint (e.g. socks5h://gluetun:1080). "" = none configured.
_ENV_PROXY_URL = os.getenv("LECTIO_PROXY_URL", "").strip()
# Last-resort outbound proxy (e.g. socks5h://tailscale:1080). "" = none configured.
_ENV_TAILSCALE_URL = os.getenv("LECTIO_TAILSCALE_URL", "").strip()
# FlareSolverr endpoint (e.g. http://flaresolverr:8191/v1). "" = none configured.
_ENV_FLARESOLVERR_URL = os.getenv("LECTIO_FLARESOLVERR_URL", "").strip()
# --- Scheduler resilience (see ARCHITECTURE "Refresh scheduler") ---
# Feed fetches are strictly sequential, so one host that accepts a connection and
# then goes quiet delays every feed behind it. These bound that.
# LECTIO_FEED_CONNECT_TIMEOUT seconds to establish a feed connection
# LECTIO_FEED_READ_TIMEOUT seconds to wait for any single read from a feed host
FEED_CONNECT_TIMEOUT_SECONDS = float(_env_int("LECTIO_FEED_CONNECT_TIMEOUT", 10))
FEED_READ_TIMEOUT_SECONDS = float(_env_int("LECTIO_FEED_READ_TIMEOUT", 30))
# A read deadline is per-socket-read, not total: a host trickling one byte at a
# time never trips it. The watchdog is the backstop — it watches for PROGRESS,
# not elapsed time, because a full-library pass legitimately runs for an hour.
# LECTIO_SCHEDULER_STALL_SECONDS no progress for this long => log loudly
# LECTIO_SCHEDULER_STALL_RESTART_SECONDS no progress for this long => exit so the
# container restarts; 0 disables
SCHEDULER_STALL_SECONDS = _env_int("LECTIO_SCHEDULER_STALL_SECONDS", 600)
SCHEDULER_STALL_RESTART_SECONDS = _env_int("LECTIO_SCHEDULER_STALL_RESTART_SECONDS", 1800)
SCHEDULER_WATCHDOG_POLL_SECONDS = 60
DEBUG_MODE = os.getenv("LECTIO_DEBUG", "0") == "1"
# Public base URL of this instance (e.g. https://lectio.example.com).
# Required for WebSub: hubs need a reachable callback URL. Leave blank to disable WebSub.
LECTIO_PUBLIC_URL = os.getenv("LECTIO_PUBLIC_URL", "").strip().rstrip("/")
_FEVER_PASSWORD = os.getenv("LECTIO_FEVER_PASSWORD", "").strip()
# --- Email (Resend) config — env vars are fallbacks; DB settings take precedence at runtime ---
_ENV_RESEND_API_KEY = os.getenv("RESEND_API_KEY", "").strip()
_ENV_RESEND_FROM = os.getenv("LECTIO_EMAIL_FROM", "").strip()
# Seed value for the bootstrap admin's default recipient (then per-user).
_ENV_RESEND_TO = os.getenv("LECTIO_EMAIL_TO", "").strip()
def get_resend_api_key() -> str:
# Instance-shared: the Resend account/key is owned at the instance level
# (one verified domain), so the env value is the shared default for everyone.
return get_runtime_setting(SETTING_RESEND_API_KEY, _ENV_RESEND_API_KEY)
def get_resend_from() -> str:
# Instance sending identity (admin-managed, like the Resend key): env value is
# the shared default for everyone; an admin override lives in their settings.
return get_runtime_setting(SETTING_EMAIL_FROM, _ENV_RESEND_FROM)
def is_email_configured() -> bool:
return bool(get_resend_api_key() and get_resend_from())
def is_instapaper_configured() -> bool:
return bool(get_runtime_setting(SETTING_INSTAPAPER_USERNAME) and get_runtime_setting(SETTING_INSTAPAPER_PASSWORD))
# --- YouTube subscription sync config — env vars are fallbacks; DB settings take precedence ---
_ENV_YT_API_KEY = os.getenv("YOUTUBE_API_KEY", "").strip()
_ENV_YT_CHANNEL_ID = os.getenv("YOUTUBE_CHANNEL_ID", "").strip()
_ENV_YT_FOLDER_NAME = os.getenv("YOUTUBE_FOLDER_NAME", "").strip() or "YouTube Subscriptions"
# DeviantArt API creds — per-user DB settings take precedence; env is the fallback.
_ENV_DEVIANTART_CLIENT_ID = os.getenv("DEVIANTART_CLIENT_ID", "").strip()
_ENV_DEVIANTART_CLIENT_SECRET = os.getenv("DEVIANTART_CLIENT_SECRET", "").strip()
# YouTube OAuth client creds are app-level (one registered Google app), not
# per-user — only the resulting tokens are per-user. So these are read straight
# from env for every user (sharing the app, never the tokens).
_ENV_YT_OAUTH_CLIENT_ID = os.getenv("YOUTUBE_OAUTH_CLIENT_ID", "").strip()
_ENV_YT_OAUTH_CLIENT_SECRET = os.getenv("YOUTUBE_OAUTH_CLIENT_SECRET", "").strip()
# Pinterest OAuth client creds are app-level (one registered Pinterest app), same
# as YouTube above — only the resulting tokens are per-user.
_ENV_PINTEREST_OAUTH_CLIENT_ID = os.getenv("PINTEREST_OAUTH_CLIENT_ID", "").strip()
_ENV_PINTEREST_OAUTH_CLIENT_SECRET = os.getenv("PINTEREST_OAUTH_CLIENT_SECRET", "").strip()
# Quire API creds — per-user DB settings take precedence; env is the fallback
# (same pattern as DeviantArt). Only the resulting tokens are ever per-user.
_ENV_QUIRE_CLIENT_ID = os.getenv("QUIRE_CLIENT_ID", "").strip()
_ENV_QUIRE_CLIENT_SECRET = os.getenv("QUIRE_CLIENT_SECRET", "").strip()
# Inoreader migration OAuth creds — env is the fallback; DB settings take precedence.
_ENV_INOREADER_CLIENT_ID = os.getenv("INOREADER_CLIENT_ID", "").strip()
_ENV_INOREADER_CLIENT_SECRET = os.getenv("INOREADER_CLIENT_SECRET", "").strip()
_yt_sync_hour_raw = os.getenv("YOUTUBE_SYNC_HOUR", "").strip()
_ENV_MAINTENANCE_HOUR: int | None = None
if _yt_sync_hour_raw:
try:
_h = int(_yt_sync_hour_raw)
if 0 <= _h <= 23:
_ENV_MAINTENANCE_HOUR = _h
except ValueError:
pass
def get_yt_api_key() -> str:
return get_runtime_setting(SETTING_YT_API_KEY, _ENV_YT_API_KEY)
def get_yt_channel_id() -> str:
return get_runtime_setting(SETTING_YT_CHANNEL_ID, _ENV_YT_CHANNEL_ID)
def get_yt_folder_name() -> str:
return get_runtime_setting(SETTING_YT_FOLDER_NAME, _ENV_YT_FOLDER_NAME) or "YouTube Subscriptions"
def youtube_embed_account_features_enabled() -> bool:
"""Per-user: True if YouTube embeds should use the standard youtube.com host
(Share / Watch Later) instead of the privacy-enhanced -nocookie host.
Privacy-enhanced is the default."""
return get_runtime_setting(SETTING_YT_EMBED_ACCOUNT_FEATURES, "0") == "1"
def youtube_hide_shorts_global() -> bool:
"""Per-user: auto-mark Shorts read on ALL YouTube feeds at refresh (overrides the
per-feed pref). Off by default."""
return get_runtime_setting(SETTING_YT_HIDE_SHORTS_GLOBAL, "0") == "1"
def youtube_hide_unpremiered_global() -> bool:
"""Per-user: hide not-yet-premiered videos on ALL YouTube feeds (overrides the
per-feed pref). Off by default."""
return get_runtime_setting(SETTING_YT_HIDE_UNPREMIERED_GLOBAL, "0") == "1"
def youtube_hide_members_only_global() -> bool:
"""Per-user: auto-mark members-only videos read on ALL YouTube feeds at
refresh (overrides the per-feed pref). Off by default."""
return get_runtime_setting(SETTING_YT_HIDE_MEMBERS_ONLY_GLOBAL, "0") == "1"
def hide_locked_comics_global() -> bool:
"""Per-user: hide a supporter-locked webcomic strip on ALL webcomic feeds
until its own stated unlock date (overrides the per-feed pref). Off by
default."""
return get_runtime_setting(SETTING_HIDE_LOCKED_COMICS_GLOBAL, "0") == "1"
def _pacific_today() -> str:
"""Today's date (YYYY-MM-DD) in US/Pacific — the timezone YouTube resets quota on."""
import datetime as _dt
try:
from zoneinfo import ZoneInfo
return _dt.datetime.now(ZoneInfo("America/Los_Angeles")).strftime("%Y-%m-%d")
except Exception:
# Fallback: fixed -08:00 offset (good enough for a day-bucket if tzdata is absent).
return (_dt.datetime.now(_dt.timezone.utc) - _dt.timedelta(hours=8)).strftime("%Y-%m-%d")
def youtube_quota_cap() -> int:
try:
return max(1, int(get_runtime_setting(SETTING_YT_QUOTA_CAP, str(YT_QUOTA_DEFAULT_CAP)) or YT_QUOTA_DEFAULT_CAP))
except TypeError, ValueError:
return YT_QUOTA_DEFAULT_CAP
def record_yt_quota_spend(units: int) -> None:
"""Add ``units`` to the current user's Pacific-day YouTube quota tally. Called by
the YT services after each billed API call (sink wired in at startup)."""
if units <= 0:
return
try:
with get_meta_connection() as conn:
conn.execute(
"INSERT INTO yt_quota_spend (day, units) VALUES (?, ?) ON CONFLICT(day) DO UPDATE SET units = units + excluded.units",
(_pacific_today(), int(units)),
)
except Exception:
LOGGER.debug("[yt-quota] failed to record %d units", units, exc_info=True)
def mark_yt_quota_exhausted() -> None:
"""Snap today's tally to the cap after an actual quotaExceeded response."""
try:
cap = youtube_quota_cap()
with get_meta_connection() as conn:
conn.execute(
"INSERT INTO yt_quota_spend (day, units) VALUES (?, ?) ON CONFLICT(day) DO UPDATE SET units = MAX(units, excluded.units)",
(_pacific_today(), cap),
)
except Exception:
LOGGER.debug("[yt-quota] failed to mark exhausted", exc_info=True)
def get_yt_quota_spent_today() -> int:
try:
with get_meta_connection() as conn:
row = conn.execute("SELECT units FROM yt_quota_spend WHERE day = ?", (_pacific_today(),)).fetchone()
return int(row["units"]) if row else 0
except Exception:
return 0
def get_yt_quota_status() -> dict:
"""Quota meter payload for the UI: spent / cap / remaining + a low/exhausted flag."""
cap = youtube_quota_cap()
spent = get_yt_quota_spent_today()
remaining = max(0, cap - spent)
state = "exhausted" if remaining <= 0 else ("low" if remaining < 500 else "ok")
return {"spent": spent, "cap": cap, "remaining": remaining, "state": state}
def youtube_embed_host() -> str:
"""The YouTube embed host for the current user's privacy/features preference."""
return "www.youtube.com" if youtube_embed_account_features_enabled() else "www.youtube-nocookie.com"
def invalidate_instance_setting_cache() -> None:
with _instance_setting_cache_lock:
_instance_setting_cache.clear()
def _get_admin_instance_setting(key: str) -> str:
"""First enabled admin's stored value for *key* ("" if none), from any context.
Reads via get_setting (DB-backed) rather than the in-memory cache alone, so
it works in background threads before the admin's cache has ever loaded
(e.g. the 3am maintenance check right after a container restart).
"""
now = time.monotonic()
with _instance_setting_cache_lock:
hit = _instance_setting_cache.get(key)
if hit and now - hit[0] < _INSTANCE_SETTING_TTL_SECONDS:
return hit[1]
val = ""
# Never crash the caller (login lockout checks run pre-auth; a
# half-provisioned admin dir must degrade to the env fallback).
try:
for u in user_store.list_users():
if not u.get("is_admin") or u.get("disabled"):
continue
with tenancy.user_context(u["user_id"]):
with get_meta_connection() as conn:
v = get_setting(conn, key)
if v:
val = v
break
except Exception:
LOGGER.debug("admin instance-setting lookup failed for %r", key, exc_info=True)
with _instance_setting_cache_lock:
_instance_setting_cache[key] = (now, val)
return val
def get_instance_setting(key: str, env_fallback: str = "") -> str:
"""Resolve an instance-level (admin-managed) setting from any context.
Resolution: current context's own setting (when a user is bound) → first
enabled admin's setting → env fallback. The admin tier is what makes
Administration → Instance Config authoritative for background threads,
pre-auth checks, and other users.
"""
val = get_cached_setting(key)
if val:
return val
return _get_admin_instance_setting(key) or env_fallback
def _get_shared_credential(key: str) -> str:
"""Read a shared-instance credential from the first admin user's settings.
Shared-instance OAuth creds are stored in the admin's own app_settings
under a namespaced key (e.g. shared_yt_oauth_client_id) and are used as a
middle tier: per-user setting → shared-instance → env var.
"""
return _get_admin_instance_setting(key)
def get_fetch_history_max_age_days() -> int:
return int(get_instance_setting(SETTING_FETCH_HISTORY_MAX_AGE_DAYS) or 30)
# Fixed per-feed row ceiling for feed_fetch_history — a backstop against a
# pathologically-retried feed, not a user-facing setting. Normal cadence (even
# hourly for the full max-age window) stays far under this.
FETCH_HISTORY_ROW_CEILING = 5000
def get_tombstone_sweep_days() -> int:
"""Age (days) after which deleted-entry tombstones are swept by nightly
maintenance; 0 disables sweeping. By then the entry has long left the
publisher's feed window, so a refresh can no longer resurrect it."""
return int(get_instance_setting(SETTING_TOMBSTONE_SWEEP_DAYS) or 90)
def get_login_max_failures() -> int:
return int(get_instance_setting(SETTING_LOGIN_MAX_FAILURES) or 5)
def get_login_window_seconds() -> int:
return int(get_instance_setting(SETTING_LOGIN_WINDOW_SECONDS) or 300)
def get_instance_default_auto_refresh() -> int:
raw = int(get_instance_setting(SETTING_DEFAULT_AUTO_REFRESH_MINUTES) or DEFAULT_AUTO_REFRESH_MINUTES)
return 0 if raw <= 0 else max(raw, MIN_AUTO_REFRESH_MINUTES)
def get_youtube_oauth_credentials() -> tuple[str, str]:
"""YouTube OAuth client (client_id, client_secret).
Resolution: per-user setting → shared-instance (admin's settings) → env var.
"""
cid = (
get_runtime_setting(SETTING_YT_OAUTH_CLIENT_ID)
or _get_shared_credential(SETTING_SHARED_YT_OAUTH_CLIENT_ID)
or _ENV_YT_OAUTH_CLIENT_ID
)
secret = (
get_runtime_setting(SETTING_YT_OAUTH_CLIENT_SECRET)
or _get_shared_credential(SETTING_SHARED_YT_OAUTH_CLIENT_SECRET)
or _ENV_YT_OAUTH_CLIENT_SECRET
)
return cid, secret
def youtube_oauth_connected() -> bool:
"""True if the current user has connected their YouTube account."""
with get_meta_connection() as conn:
return bool(get_setting(conn, SETTING_YT_OAUTH_REFRESH_TOKEN))
def get_youtube_oauth_token() -> str:
"""Return a valid user access token for playlist writes, refreshing if needed.
Empty string if the user hasn't connected their YouTube account or the
refresh failed (revoked / testing-mode 7-day expiry) — caller must prompt a
reconnect.
"""
with get_meta_connection() as conn:
access = get_setting(conn, SETTING_YT_OAUTH_ACCESS_TOKEN) or ""
refresh = get_setting(conn, SETTING_YT_OAUTH_REFRESH_TOKEN) or ""
try:
expires_at = float(get_setting(conn, SETTING_YT_OAUTH_TOKEN_EXPIRES_AT) or 0)
except TypeError, ValueError:
expires_at = 0.0
if not refresh:
return ""
if access and time.time() < expires_at - 60:
return access
cid, secret = get_youtube_oauth_credentials()
try:
data = youtube_oauth_service.refresh_access_token(cid, secret, refresh)
except Exception as exc: # noqa: BLE001
LOGGER.warning("[youtube-oauth] token refresh failed; reconnect required: %s", exc)
return ""
with get_meta_connection() as conn:
set_setting(conn, SETTING_YT_OAUTH_ACCESS_TOKEN, data["access_token"])
if data.get("refresh_token"):
set_setting(conn, SETTING_YT_OAUTH_REFRESH_TOKEN, data["refresh_token"])
set_setting(conn, SETTING_YT_OAUTH_TOKEN_EXPIRES_AT, str(time.time() + float(data.get("expires_in", 3600))))
return data["access_token"]
def get_pinterest_oauth_credentials() -> tuple[str, str]:
"""Pinterest OAuth client (client_id, client_secret).
Resolution: per-user setting → shared-instance (admin's settings) → env var.
"""
cid = (
get_runtime_setting(SETTING_PINTEREST_OAUTH_CLIENT_ID)
or _get_shared_credential(SETTING_SHARED_PINTEREST_OAUTH_CLIENT_ID)
or _ENV_PINTEREST_OAUTH_CLIENT_ID
)
secret = (
get_runtime_setting(SETTING_PINTEREST_OAUTH_CLIENT_SECRET)
or _get_shared_credential(SETTING_SHARED_PINTEREST_OAUTH_CLIENT_SECRET)
or _ENV_PINTEREST_OAUTH_CLIENT_SECRET
)
return cid, secret
def pinterest_oauth_connected() -> bool:
"""True if the current user has connected their Pinterest account."""
with get_meta_connection() as conn:
return bool(get_setting(conn, SETTING_PINTEREST_OAUTH_REFRESH_TOKEN))
def get_pinterest_oauth_token() -> str:
"""Return a valid user access token for pin writes, refreshing if needed.
Empty string if the user hasn't connected Pinterest or the refresh failed
(revoked / expired) — caller must prompt a reconnect."""
with get_meta_connection() as conn:
access = get_setting(conn, SETTING_PINTEREST_OAUTH_ACCESS_TOKEN) or ""
refresh = get_setting(conn, SETTING_PINTEREST_OAUTH_REFRESH_TOKEN) or ""
try:
expires_at = float(get_setting(conn, SETTING_PINTEREST_OAUTH_TOKEN_EXPIRES_AT) or 0)
except TypeError, ValueError:
expires_at = 0.0
if not refresh:
return ""
if access and time.time() < expires_at - 60:
return access
cid, secret = get_pinterest_oauth_credentials()
try:
data = pinterest_oauth_service.refresh_access_token(cid, secret, refresh)
except Exception as exc: # noqa: BLE001
LOGGER.warning("[pinterest-oauth] token refresh failed; reconnect required: %s", exc)
return ""
with get_meta_connection() as conn:
set_setting(conn, SETTING_PINTEREST_OAUTH_ACCESS_TOKEN, data["access_token"])
if data.get("refresh_token"):
set_setting(conn, SETTING_PINTEREST_OAUTH_REFRESH_TOKEN, data["refresh_token"])
set_setting(conn, SETTING_PINTEREST_OAUTH_TOKEN_EXPIRES_AT, str(time.time() + float(data.get("expires_in", 3600))))
return data["access_token"]
def get_reddit_credentials() -> tuple[str, str]:
"""Reddit OAuth client (client_id, client_secret).
Resolution: per-user setting → shared-instance (admin's settings) → empty.
No env-var fallback — credentials are DB-only for this integration.
"""
cid = get_runtime_setting(SETTING_REDDIT_CLIENT_ID) or _get_shared_credential(SETTING_SHARED_REDDIT_CLIENT_ID) or ""
secret = get_runtime_setting(SETTING_REDDIT_CLIENT_SECRET) or _get_shared_credential(SETTING_SHARED_REDDIT_CLIENT_SECRET) or ""
return cid, secret
def reddit_connected() -> bool:
"""True if the current user has connected their Reddit account."""
with get_meta_connection() as conn:
return bool(get_setting(conn, SETTING_REDDIT_REFRESH_TOKEN))
def get_reddit_user_token() -> str:
"""Return a valid Reddit access token, refreshing if needed.
Empty string if not connected or refresh failed — caller must prompt reconnect.
"""
with get_meta_connection() as conn:
access = get_setting(conn, SETTING_REDDIT_ACCESS_TOKEN) or ""
refresh = get_setting(conn, SETTING_REDDIT_REFRESH_TOKEN) or ""
try:
expires_at = float(get_setting(conn, SETTING_REDDIT_TOKEN_EXPIRES_AT) or 0)
except TypeError, ValueError:
expires_at = 0.0
if not refresh:
return ""
if access and time.time() < expires_at - 60:
return access
cid, secret = get_reddit_credentials()
if not cid or not secret:
return ""
try:
data = reddit_service.refresh_access_token(cid, secret, refresh)
except Exception as exc: # noqa: BLE001
LOGGER.warning("[reddit] token refresh failed; reconnect required: %s", exc)
return ""
with get_meta_connection() as conn:
set_setting(conn, SETTING_REDDIT_ACCESS_TOKEN, data["access_token"])
if data.get("refresh_token"):
set_setting(conn, SETTING_REDDIT_REFRESH_TOKEN, data["refresh_token"])
set_setting(conn, SETTING_REDDIT_TOKEN_EXPIRES_AT, str(time.time() + float(data.get("expires_in", 3600))))
return data["access_token"]
def _reddit_redirect_uri(request: Request) -> str:
base = os.getenv("LECTIO_PUBLIC_URL", "").rstrip("/")
if not base:
base = str(request.base_url).rstrip("/")
return f"{base}/integrations/reddit/oauth/callback"
def get_inoreader_credentials() -> tuple[str, str]:
"""Inoreader OAuth credentials (client_id, client_secret).
DB settings take precedence over env vars so any user can override them
without restarting the container.
"""
cid = get_runtime_setting(SETTING_INOREADER_CLIENT_ID, _ENV_INOREADER_CLIENT_ID)
secret = get_runtime_setting(SETTING_INOREADER_CLIENT_SECRET, _ENV_INOREADER_CLIENT_SECRET)
return cid, secret
def inoreader_connected() -> bool:
with get_meta_connection() as conn:
return bool(get_setting(conn, SETTING_INOREADER_REFRESH_TOKEN))
def get_inoreader_token() -> str:
"""Return a valid Inoreader access token, refreshing it if expired."""
with get_meta_connection() as conn:
access = get_setting(conn, SETTING_INOREADER_ACCESS_TOKEN) or ""
refresh = get_setting(conn, SETTING_INOREADER_REFRESH_TOKEN) or ""
expires_at = float(get_setting(conn, SETTING_INOREADER_TOKEN_EXPIRES_AT) or 0)
if not refresh:
return ""
if access and time.time() < expires_at - 60:
return access
cid, secret = get_inoreader_credentials()
try:
data = inoreader_service.refresh_access_token(cid, secret, refresh)
except Exception as exc: # noqa: BLE001
LOGGER.warning("[inoreader] token refresh failed: %s", exc)
if "invalid_grant" in str(exc):
# Inoreader has permanently rejected this refresh token (not a transient
# network/outage failure) — clear it so inoreader_connected() stops
# reporting stale "Connected." and the UI offers Connect again.
with get_meta_connection() as conn:
set_setting(conn, SETTING_INOREADER_ACCESS_TOKEN, "")
set_setting(conn, SETTING_INOREADER_REFRESH_TOKEN, "")
set_setting(conn, SETTING_INOREADER_TOKEN_EXPIRES_AT, "")
return ""
with get_meta_connection() as conn:
set_setting(conn, SETTING_INOREADER_ACCESS_TOKEN, data["access_token"])
if data.get("refresh_token"):
set_setting(conn, SETTING_INOREADER_REFRESH_TOKEN, data["refresh_token"])
set_setting(conn, SETTING_INOREADER_TOKEN_EXPIRES_AT, str(time.time() + float(data.get("expires_in", 3600))))
return data["access_token"]
def get_deviantart_credentials() -> tuple[str, str]:
"""Per-user DeviantArt API credentials (client_id, client_secret).
Stored per-user in app-settings; env vars are instance-wide fallbacks.
"""
cid = get_runtime_setting(SETTING_DEVIANTART_CLIENT_ID, _ENV_DEVIANTART_CLIENT_ID)
secret = get_runtime_setting(SETTING_DEVIANTART_CLIENT_SECRET, _ENV_DEVIANTART_CLIENT_SECRET)
return cid, secret
def get_deviantart_user_token() -> str:
"""Return a valid user access token for watch-list ops, refreshing if expired.
Empty string if the user hasn't connected their DeviantArt account.
"""
with get_meta_connection() as conn:
access = get_setting(conn, SETTING_DEVIANTART_ACCESS_TOKEN) or ""
refresh = get_setting(conn, SETTING_DEVIANTART_REFRESH_TOKEN) or ""
try:
expires_at = float(get_setting(conn, SETTING_DEVIANTART_TOKEN_EXPIRES_AT) or 0)
except TypeError, ValueError:
# A corrupt/manually-edited value must not break every token read;
# treat it as expired so we fall through to a forced refresh.
expires_at = 0.0
if not access:
return ""
if time.time() < expires_at - 60:
return access
if not refresh:
return access # no refresh token; try the (possibly stale) access token
cid, secret = get_deviantart_credentials()
try:
data = deviantart_service.refresh_access_token(cid, secret, refresh)
except Exception as exc: # noqa: BLE001
# Refresh failed (expired/rotated refresh token, or the client_secret no
# longer matches the app) — the session is dead; signal "reconnect needed"
# rather than handing back a stale token that 401s deep in a sync.
LOGGER.warning("[deviantart] token refresh failed; reconnect required: %s", exc)
return ""
with get_meta_connection() as conn:
set_setting(conn, SETTING_DEVIANTART_ACCESS_TOKEN, data["access_token"])
if data.get("refresh_token"):
set_setting(conn, SETTING_DEVIANTART_REFRESH_TOKEN, data["refresh_token"])
set_setting(conn, SETTING_DEVIANTART_TOKEN_EXPIRES_AT, str(time.time() + float(data.get("expires_in", 3600))))
return data["access_token"]
def get_quire_credentials() -> tuple[str, str]:
"""Per-user Quire API credentials (client_id, client_secret).
Stored per-user in app-settings; env vars are instance-wide fallbacks.
"""
cid = get_runtime_setting(SETTING_QUIRE_CLIENT_ID, _ENV_QUIRE_CLIENT_ID)
secret = get_runtime_setting(SETTING_QUIRE_CLIENT_SECRET, _ENV_QUIRE_CLIENT_SECRET)
return cid, secret
def is_quire_connected() -> bool:
"""True once the user has completed the Quire OAuth flow (has a token)."""
return bool(get_runtime_setting(SETTING_QUIRE_ACCESS_TOKEN))
def quire_project_oid() -> str:
"""The chosen default destination project's OID ("" if none picked yet)."""
return (get_runtime_setting(SETTING_QUIRE_PROJECT_OID) or "").strip()
def quire_project_name() -> str:
"""Display name of the chosen default destination project ("" if none)."""
return (get_runtime_setting(SETTING_QUIRE_PROJECT_NAME) or "").strip()
def is_quire_configured() -> bool:
"""True when Quire is connected AND a default destination project is set —
the precondition for the entry button, On-Star, and Automation rules."""
return bool(is_quire_connected() and quire_project_oid())
def get_quire_user_token() -> str:
"""Return a valid Quire access token, refreshing if expired. Empty string if the
user hasn't connected their Quire account."""
with get_meta_connection() as conn:
access = get_setting(conn, SETTING_QUIRE_ACCESS_TOKEN) or ""
refresh = get_setting(conn, SETTING_QUIRE_REFRESH_TOKEN) or ""
try:
expires_at = float(get_setting(conn, SETTING_QUIRE_TOKEN_EXPIRES_AT) or 0)
except TypeError, ValueError:
expires_at = 0.0
if not access:
return ""
if time.time() < expires_at - 60:
return access
if not refresh:
return access
cid, secret = get_quire_credentials()
try:
data = quire_service.refresh_access_token(cid, secret, refresh)
except Exception as exc: # noqa: BLE001
LOGGER.warning("[quire] token refresh failed; reconnect required: %s", exc)
return ""
with get_meta_connection() as conn:
set_setting(conn, SETTING_QUIRE_ACCESS_TOKEN, data["access_token"])
if data.get("refresh_token"):
set_setting(conn, SETTING_QUIRE_REFRESH_TOKEN, data["refresh_token"])
set_setting(conn, SETTING_QUIRE_TOKEN_EXPIRES_AT, str(time.time() + float(data.get("expires_in", 3600))))
return data["access_token"]
def quire_rate_caps() -> tuple[int, int]:
"""(per-minute, per-hour) caps Lectio meters its Quire calls against."""
def _cap(key: str, default: int) -> int:
try:
return max(1, int(get_runtime_setting(key, str(default)) or default))
except TypeError, ValueError:
return default
return (_cap(SETTING_QUIRE_RATE_CAP_MIN, QUIRE_RATE_DEFAULT_CAP_MIN), _cap(SETTING_QUIRE_RATE_CAP_HOUR, QUIRE_RATE_DEFAULT_CAP_HOUR))
def record_quire_call(calls: int = 1) -> None:
"""Log ``calls`` Quire API calls into the sliding-window meter (wired as the
service usage sink at startup). Best-effort; prunes rows older than an hour."""
if calls <= 0:
return
try:
now = int(time.time())
with get_meta_connection() as conn:
conn.executemany(
"INSERT INTO quire_call_log (ts) VALUES (?)",
[(now,) for _ in range(int(calls))],
)
conn.execute("DELETE FROM quire_call_log WHERE ts < ?", (now - 3600,))
except Exception:
LOGGER.debug("[quire] failed to record %d calls", calls, exc_info=True)
def get_quire_usage_status() -> dict:
"""Usage-meter payload: calls in the last minute/hour vs caps, plus a state flag."""
cap_min, cap_hour = quire_rate_caps()
try:
now = int(time.time())
with get_meta_connection() as conn:
minute_used = conn.execute("SELECT COUNT(*) AS n FROM quire_call_log WHERE ts >= ?", (now - 60,)).fetchone()["n"]
hour_used = conn.execute("SELECT COUNT(*) AS n FROM quire_call_log WHERE ts >= ?", (now - 3600,)).fetchone()["n"]
except Exception:
minute_used = hour_used = 0
minute_used, hour_used = int(minute_used), int(hour_used)
blocked = minute_used >= cap_min or hour_used >= cap_hour
low = (minute_used >= cap_min * 0.8) or (hour_used >= cap_hour * 0.8)
state = "blocked" if blocked else ("low" if low else "ok")
return {"minute_used": minute_used, "minute_cap": cap_min, "hour_used": hour_used, "hour_cap": cap_hour, "state": state}
def detect_quire_plan_and_caps() -> str:
"""Detect the destination project's organization plan and align the rate-meter
caps to it (Free 50/200, Professional 300/1250, Premium 1000/5000). Stores the
plan name for display. Best-effort; returns the detected plan ("" on failure).
Quire rate-limits per organization, so the cap that matters is the plan of the
org owning the chosen project."""
project_oid = quire_project_oid()
if not project_oid:
return ""
token = get_quire_user_token()
if not token:
return ""
try:
plan = quire_service.get_project_plan(token, project_oid)
except Exception as exc: # noqa: BLE001
LOGGER.warning("[quire] plan detection failed: %s", exc)
return ""
caps = quire_service.PLAN_RATE_CAPS.get(plan.strip().lower())
with get_meta_connection() as conn:
set_setting(conn, SETTING_QUIRE_PLAN, plan)
if caps:
# Enterprise (and any unknown plan) keeps the existing/default caps since
# its quota scales with member count and isn't a fixed pair.
set_setting(conn, SETTING_QUIRE_RATE_CAP_MIN, str(caps[0]))
set_setting(conn, SETTING_QUIRE_RATE_CAP_HOUR, str(caps[1]))
return plan
def _deviantart_folder_name() -> str:
return get_runtime_setting(SETTING_DEVIANTART_FOLDER_NAME) or "DeviantArt"
def _apply_deviantart_image_strategy(conn: sqlite3.Connection, file_url: str) -> None:
"""Pin a DeviantArt feed to the 'inline' image strategy.
DA feeds embed the authoritative full-size image inline (a stable, non-expiring
wixmp URL), so deriving both the article lead image and the list thumbnail from
that content — statelessly, every render — avoids source-page scraping and the
lead-image cache getting clobbered to 'no image'. Locked (manual) so auto-detect
can't change it.
"""
try:
lead_image_service.store_feed_strategy(file_url, "inline", manual=True)
except Exception:
LOGGER.exception("[deviantart] failed to pin inline strategy for %s", file_url)
try:
conn.execute("INSERT INTO feed_display_prefs (feed_url) VALUES (?) ON CONFLICT(feed_url) DO NOTHING", (file_url,))
conn.execute("UPDATE feed_display_prefs SET thumb_strategy = 'inline' WHERE feed_url = ?", (file_url,))
except Exception:
LOGGER.exception("[deviantart] failed to set inline thumb_strategy for %s", file_url)
def _is_da_deactivated_error(exc: Exception) -> bool:
"""True when a DeviantArt add failed because the artist deactivated their
account — DA returns HTTP 400 with ``"Account is inactive."``. Distinct from
a 404 (deleted/renamed), so we can park it for a daily reactivation re-check
instead of retrying every sync."""
return "account is inactive" in str(exc).lower()
def _humanize_da_add_error(exc: Exception) -> str:
"""Short, user-facing reason a watch-list add failed, for the Settings list.
DeviantArt gallery fetches raise ``RuntimeError('gallery fetch failed for X:
HTTP 404: ...')``; surface the status where we can, else a trimmed message."""
detail = " ".join(str(exc).split())
for needle, label in (
("Account is inactive", "deactivated"),
("404", "not found"),
("403", "access denied"),
("500", "DeviantArt error"),
):
if needle in detail:
return label
return detail[:120] or "unknown error"
def _load_da_sync_detail() -> dict:
"""Parse the structured sync-detail setting, or an empty shell on miss/garbage."""
raw = get_runtime_setting(SETTING_DEVIANTART_SYNC_DETAIL)
if raw:
try:
data = json.loads(raw)
if isinstance(data, dict):
return data
except ValueError, TypeError:
pass
return {"failed": [], "unwatched": []}
def _da_deactivated_usernames(conn: sqlite3.Connection) -> set[str]:
"""Lower-cased usernames of watched artists whose DA account is deactivated."""
return {str(r["username"]).lower() for r in conn.execute("SELECT username FROM deviantart_deactivated").fetchall()}
def _da_deactivated_list() -> list[dict]:
"""Parked deactivated artists for the Settings UI (opens its own connection)."""
try:
with get_meta_connection() as conn:
return [
{"username": str(r["username"])}
for r in conn.execute("SELECT username FROM deviantart_deactivated ORDER BY username").fetchall()
]
except Exception: # noqa: BLE001 — table missing mid-migration, etc.
return []
def _record_da_deactivated(conn: sqlite3.Connection, username: str) -> None:
"""Mark an artist deactivated (idempotent; preserves first_seen_at)."""
now = datetime.now(timezone.utc).isoformat()
conn.execute(
"INSERT INTO deviantart_deactivated (username, first_seen_at, last_checked_at)"
" VALUES (?, ?, ?)"
" ON CONFLICT(username) DO UPDATE SET last_checked_at = excluded.last_checked_at",
(username, now, now),
)
_DA_SYNC_RESUME_FALLBACK_S = 900.0 # no Retry-After header → conservative delay
def _schedule_da_sync_resume(uid: str, delay_s: float, next_round: int) -> None:
def _resume() -> None:
try:
result = sync_deviantart_watchlist(auto_resume_round=next_round)
if result.get("error"):
with get_meta_connection() as conn:
set_setting(conn, SETTING_DEVIANTART_SYNC_STATUS, f"Sync error: {result['error']}")
except Exception as exc:
LOGGER.exception("[deviantart] watchlist auto-resume failed")
with get_meta_connection() as conn:
set_setting(conn, SETTING_DEVIANTART_SYNC_STATUS, f"Sync failed: {_humanize_da_add_error(exc)}. Click Sync to retry.")
timer = threading.Timer(delay_s, _run_in_user_context, args=(uid, _resume))
timer.daemon = True # lost on restart; the daily maintenance sync catches up
timer.start()
def sync_deviantart_watchlist(auto_resume_round: int = 0) -> dict:
"""Add a gallery feed for every artist the user Watches (add-only), and
pause (disable_feed) any subscribed artist newly dropped from the watch
list — see the reconcile step in _sync_deviantart_watchlist_locked.
Skips artists already subscribed (anywhere) so it won't duplicate existing
DeviantArt feeds. When DeviantArt's rate cap interrupts the run, schedules a
background continuation (honoring Retry-After) instead of waiting for a
manual re-click. Returns a result summary dict.
"""
token = get_deviantart_user_token()
if not token:
return {"added": 0, "total": 0, "error": "DeviantArt account not connected."}
uid = tenancy.current_user_id()
with _da_sync_lock:
if uid in _da_sync_active:
LOGGER.info("[deviantart] watchlist sync already running for %s — skipped", uid)
return {"added": 0, "total": 0, "skipped": True}
_da_sync_active.add(uid)
try:
return _sync_deviantart_watchlist_locked(token, uid, auto_resume_round)
finally:
with _da_sync_lock:
_da_sync_active.discard(uid)
def _sync_deviantart_watchlist_locked(token: str, uid: str, auto_resume_round: int) -> dict:
cid, secret = get_deviantart_credentials()
try:
username = get_runtime_setting(SETTING_DEVIANTART_USERNAME) or deviantart_service.whoami(token)
watching = deviantart_service.list_watching(token, username)
except deviantart_service.DeviantArtRateLimited as exc:
return {"added": 0, "total": 0, "error": f"DeviantArt rate limit hit — try again in a few minutes. ({exc})"}
except Exception as exc: # noqa: BLE001
return {"added": 0, "total": 0, "error": f"Could not read your watch list: {exc}"}
folder_name = _deviantart_folder_name()
with get_meta_connection() as conn:
# Exclude the combined "deviantsyouwatch" Watch feed (source='watch'): its
# username is a synthetic browse endpoint, not a real artist, so it must
# not count as a subscribed gallery for add-dedup or the unwatched reconcile
# (otherwise it's perpetually reported as "subscribed but no longer watched").
existing = {
str(r["username"]).lower()
for r in conn.execute("SELECT username FROM deviantart_feeds WHERE COALESCE(source, 'gallery') != 'watch'").fetchall()
}
# Known-deactivated artists can never be added; skip them here so the sync
# stops re-probing (and re-failing) them every run. Daily maintenance
# re-checks them for reactivation.
deactivated = _da_deactivated_usernames(conn)
folder_id = _get_or_create_folder_by_name(conn, folder_name)
# The Watch feed IS the watch list — one feed carrying every watched
# artist's posts. While it exists, a per-artist gallery feed is a
# duplicate of what it already delivers.
has_watch_feed = bool(conn.execute("SELECT 1 FROM deviantart_feeds WHERE COALESCE(source, 'gallery') = 'watch' LIMIT 1").fetchone())
to_add = [a for a in watching if a.lower() not in existing and a.lower() not in deactivated]
if has_watch_feed and to_add:
# This sync and the gallery consolidation were undoing each other. The
# consolidation folds a gallery feed into the Watch feed and then DELETES
# its deviantart_feeds row (it must — otherwise the refresh loop
# regenerates the subscription it just merged). That row is this sync's
# only record of "already subscribed", so every consolidated artist came
# back looking new and was re-created the next night: 537 gallery feeds
# returned over two runs after the 2026-08-12 consolidation, filed back
# into the folder, carrying entries the Watch feed had already delivered.
LOGGER.info(
"[deviantart] watchlist sync: Watch feed present — skipping %d per-artist gallery feed(s); the Watch feed already covers them",
len(to_add),
)
to_add = []
LOGGER.info("[deviantart] watchlist sync: %d watched, %d to add into %r", len(watching), len(to_add), folder_name)
added = 0
failed = 0
deactivated_this_run = 0
failed_artists: list[dict[str, str]] = []
rate_limited = False
retry_after_s: float | None = None
for i, artist in enumerate(to_add, 1):
if i > 1:
time.sleep(0.4) # pace requests to stay under DeviantArt's rate limit
try:
with get_meta_connection() as conn:
with get_reader() as reader:
_fid, file_url = deviantart_service.create_deviantart_feed(
conn, reader, artist, cid, secret, access_token=token, limit=24
)
conn.execute(
"INSERT OR IGNORE INTO folder_feeds (folder_id, feed_url) VALUES (?, ?)",
(folder_id, file_url),
)
added += 1
except deviantart_service.DeviantArtRateLimited as exc:
# Quota exhausted — stop now (the rest are re-runnable; sync is add-only
# and dedupes, so the auto-resume continuation picks up where this left off).
rate_limited = True
retry_after_s = exc.retry_after
LOGGER.info("[deviantart] watchlist sync paused at %d/%d (added=%d) — rate limited", i, len(to_add), added)
break
except Exception as exc: # noqa: BLE001
if _is_da_deactivated_error(exc):
# Deactivated account — park it (not a hard failure); daily
# maintenance re-checks for reactivation.
deactivated_this_run += 1
with get_meta_connection() as conn:
_record_da_deactivated(conn, artist)
LOGGER.info("[deviantart] watched artist %s is deactivated — parked for re-check", artist)
else:
failed += 1
failed_artists.append({"username": artist, "error": _humanize_da_add_error(exc)})
LOGGER.warning("[deviantart] watchlist add failed for %s: %s", artist, exc)
if i % 25 == 0:
LOGGER.info("[deviantart] watchlist sync progress: %d/%d (added=%d failed=%d)", i, len(to_add), added, failed)
with get_meta_connection() as conn:
set_setting(conn, SETTING_DEVIANTART_SYNC_STATUS, f"Syncing {i}/{len(to_add)} (added {added})")
invalidate_meta_structure_cache()
invalidate_unread_counts_cache()
invalidate_meta_structure_cache()
invalidate_unread_counts_cache()
# Reconcile: surface subscribed DA artists no longer on the watch list.
# Unsubscribing stays a manual decision (curated feeds may outlive a
# Watch), but leaving them actively refreshing was surprising once
# noticed — so a NEWLY-unwatched artist's feed(s) get paused (disable_feed,
# same as Settings' own pause/disable — no data lost, nothing unsubscribed)
# once, the first run that sees them drop off the watch list. Not reapplied
# on later runs: if Josh re-enables one by hand (decided to keep it live
# despite the unwatch), a later sync must not fight that choice by
# disabling it again every time it still shows up as unwatched.
watching_lower = {a.lower() for a in watching}
unwatched = sorted(u for u in existing if u not in watching_lower) if watching else []
prior_unwatched_lower = {str(u.get("username") or "").strip().lower() for u in _load_da_sync_detail().get("unwatched", [])}
newly_unwatched = [u for u in unwatched if u not in prior_unwatched_lower]
if newly_unwatched:
with get_meta_connection() as conn:
placeholders = ",".join("?" for _ in newly_unwatched)
rows = conn.execute(
f"SELECT id, username FROM deviantart_feeds WHERE COALESCE(source, 'gallery') != 'watch'"
f" AND lower(username) IN ({placeholders})",
newly_unwatched,
).fetchall()
for row in rows:
disable_feed(deviantart_service.feed_file_url(str(row["id"])), reason="unwatched on DeviantArt")
LOGGER.info(
"[deviantart] %d newly-unwatched artist(s) paused (updates disabled, not unsubscribed): %s",
len(rows),
", ".join(sorted(str(r["username"]) for r in rows)),
)
with get_meta_connection() as conn:
set_setting(conn, SETTING_DEVIANTART_UNWATCHED_DIRTY, "1")
if unwatched:
LOGGER.info("[deviantart] %d subscribed artist(s) no longer watched: %s", len(unwatched), ", ".join(unwatched))
# Persist the failed/unwatched artists (structured) so the Settings UI can
# list them as profile links. A fresh sync (round 0) starts clean; auto-resume
# continuation rounds accumulate failures from earlier rounds (each round
# processes a different slice of to_add). unwatched is recomputed from full
# data every round, so it's overwritten rather than merged.
prior_failed: list[dict[str, str]] = []
if auto_resume_round > 0:
prior_failed = _load_da_sync_detail().get("failed", [])
all_failed = prior_failed + failed_artists
with get_meta_connection() as conn:
set_setting(
conn, SETTING_DEVIANTART_SYNC_DETAIL, json.dumps({"failed": all_failed, "unwatched": [{"username": u} for u in unwatched]})
)
remaining = len(to_add) - added - failed
if rate_limited:
if auto_resume_round < _DA_SYNC_MAX_AUTO_RESUMES:
delay_s = max(60.0, retry_after_s or _DA_SYNC_RESUME_FALLBACK_S)
_schedule_da_sync_resume(uid, delay_s, auto_resume_round + 1)
final = (
f"Rate limited — added {added} so far, ~{remaining} left. "
f"Auto-resuming in ~{int(delay_s // 60) or 1} min "
f"(round {auto_resume_round + 1}/{_DA_SYNC_MAX_AUTO_RESUMES})."
)
else:
final = f"Rate limited — added {added} so far, ~{remaining} left. Auto-resume limit reached; click Sync to continue."
else:
# to_add now excludes both already-subscribed and known-deactivated, so
# derive each count from the watch list directly rather than the residual.
already = sum(1 for a in watching if a.lower() in existing)
deactivated_total = sum(1 for a in watching if a.lower() in deactivated) + deactivated_this_run
parts = [f"Added {added} of {len(watching)} watched"]
if already:
parts.append(f"{already} already subscribed")
if all_failed:
parts.append(f"{len(all_failed)} failed")
if deactivated_total:
parts.append(f"{deactivated_total} deactivated")
if unwatched:
parts.append(f"{len(unwatched)} subscribed but no longer watched")
final = ", ".join(parts)
with get_meta_connection() as conn:
set_setting(conn, SETTING_DEVIANTART_SYNC_STATUS, final)
return {
"added": added,
"failed": failed,
"deactivated": deactivated_this_run,
"total": len(watching),
"folder": folder_name,
"rate_limited": rate_limited,
"unwatched": unwatched,
"failed_artists": all_failed,
}
def _deviantart_recheck_deactivated(max_checks: int = 40) -> int:
"""Re-probe parked deactivated artists; subscribe any that reactivated.
Runs in daily maintenance. For each parked artist, a lightweight gallery
fetch tells us the current state: success → reactivated, so create the feed
and drop it from the deactivated table; still inactive → bump last_checked_at;
rate-limited → stop early (re-check resumes next maintenance run). Capped per
run to stay polite against the API quota. Returns the count reactivated."""
token = get_deviantart_user_token()
if not token:
return 0
cid, secret = get_deviantart_credentials()
with get_meta_connection() as conn:
rows = conn.execute(
"SELECT username FROM deviantart_deactivated ORDER BY last_checked_at ASC LIMIT ?",
(max_checks,),
).fetchall()
if not rows:
return 0
folder_id = None
reactivated = 0
now = datetime.now(timezone.utc).isoformat()
for row in rows:
artist = str(row["username"])
try:
with get_meta_connection() as conn:
with get_reader() as reader:
if folder_id is None:
folder_id = _get_or_create_folder_by_name(conn, _deviantart_folder_name())
_fid, file_url = deviantart_service.create_deviantart_feed(
conn, reader, artist, cid, secret, access_token=token, limit=24
)
conn.execute(
"INSERT OR IGNORE INTO folder_feeds (folder_id, feed_url) VALUES (?, ?)",
(folder_id, file_url),
)
conn.execute("DELETE FROM deviantart_deactivated WHERE username = ?", (artist,))
reactivated += 1
LOGGER.info("[deviantart] deactivated artist %s reactivated — subscribed", artist)
except deviantart_service.DeviantArtRateLimited:
LOGGER.info("[deviantart] deactivated re-check hit rate limit; stopping this run")
break
except Exception as exc: # noqa: BLE001
if _is_da_deactivated_error(exc):
with get_meta_connection() as conn:
conn.execute(
"UPDATE deviantart_deactivated SET last_checked_at = ? WHERE username = ?",
(now, artist),
)
else:
# A different error (deleted/renamed) — leave it parked; not worth
# promoting to a hard failure from a background re-check.
LOGGER.info("[deviantart] deactivated re-check for %s errored (non-inactive): %s", artist, exc)
if reactivated:
invalidate_meta_structure_cache()
invalidate_unread_counts_cache()
return reactivated
def push_galleries_to_deviantart_watchlist() -> dict:
"""Watch (on DeviantArt) every artist you have a gallery feed for. Add-only."""
token = get_deviantart_user_token()
if not token:
return {"watched": 0, "total": 0, "error": "DeviantArt account not connected."}
with get_meta_connection() as conn:
usernames = [str(r["username"]) for r in conn.execute("SELECT DISTINCT username FROM deviantart_feeds").fetchall()]
watched = 0
errors = 0
for name in usernames:
ok, _msg = deviantart_service.watch_user(token, name)
if ok:
watched += 1
else:
errors += 1
return {"watched": watched, "total": len(usernames), "errors": errors}
def get_maintenance_hour() -> int | None:
val = get_instance_setting(SETTING_MAINTENANCE_HOUR, "")
if val:
try:
h = int(val)
if 0 <= h <= 23:
return h
except ValueError:
pass
return _ENV_MAINTENANCE_HOUR
_busy_work_lock = threading.Lock()
_busy_work: dict[str, dict[str, int]] = {}
@contextlib.contextmanager
def busy_work(label: str):
"""Mark the current tenancy user as running long background work (also usable as a decorator).
Admin "Full vacuum" refuses to touch a user's databases while any of these is open; see user_busy_reasons."""
uid = tenancy.current_user_id()
with _busy_work_lock:
counts = _busy_work.setdefault(uid, {})
counts[label] = counts.get(label, 0) + 1
try:
yield
finally:
with _busy_work_lock:
counts = _busy_work.get(uid, {})
counts[label] = counts.get(label, 1) - 1
if counts[label] <= 0:
counts.pop(label, None)
if not counts:
_busy_work.pop(uid, None)
def user_busy_reasons(uid: str, *, include_activity: bool = True, ignore: tuple[str, ...] = ()) -> list[str]:
"""Why a full vacuum of *uid*'s databases should wait, as short phrases; empty means it is safe to go ahead.
Background work is read from the busy_work marks plus the existing per-user job registries (those need the user's own tenancy
context). include_activity=False skips the "recently active" check, for an admin vacuuming their own account. Web, Google
Reader-style and session requests touch last_seen_at; Fever and Miniflux API clients do not, so they can read as idle."""
reasons: list[str] = []
if include_activity:
row = user_store.get_by_id(uid)
last_seen = float(row["last_seen_at"] or 0) if row else 0.0
# last_seen_at is written at most once per throttle window, so a user active a second ago can read as up to a window old.
if time.time() - last_seen < 2 * _LAST_SEEN_THROTTLE_SECONDS:
reasons.append("the user was active in the last 10 minutes")
with _busy_work_lock:
reasons += [f"{label} is running" for label in _busy_work.get(uid, {}) if label not in ignore]
with tenancy.user_context(uid):
with _da_sync_lock:
if uid in _da_sync_active:
reasons.append("a DeviantArt sync is running")
if uid in _escalation_reprobe_running:
reasons.append("an escalation reprobe is running")
for name, jobs in (
("a bulk re-fetch", _refetch_jobs),
("a folder refresh", _folder_refresh_jobs),
):
job = jobs.get("job")
if isinstance(job, dict) and job.get("running"):
reasons.append(f"{name} is running")
with _autofetch_jobs_lock:
# Keyed by (feed_url, entry_id), one job per post, so look through the values rather than for a single "job".
if any(isinstance(j, dict) and j.get("running") for j in _autofetch_jobs.values()):
reasons.append("an auto-fetch is running")
return reasons
def truncate_wal(conn: sqlite3.Connection, label: str, *, attempts: int = 10, delay: float = 1.0) -> bool:
"""Fold the WAL back and truncate it, retrying while live readers block it.
``PRAGMA wal_checkpoint(TRUNCATE)`` does not raise when a reader holds the WAL: it returns ``busy=1`` and leaves the file alone.
After a VACUUM (which writes the whole rebuilt DB through the WAL) that left the WAL at ~DB size, so the total came out larger
than before. Pooled connections only read briefly, so a short retry usually gets through; if not, say so instead of staying silent.
"""
# A busy checkpoint waits out the connection's whole busy timeout (30 s on the admin path) before it reports busy, and holds off other
# writers while it waits; probe quickly instead so the retries take attempts * (0.25 s + delay) and nobody else is locked out for long.
original_timeout = conn.execute("PRAGMA busy_timeout").fetchone()[0]
conn.execute("PRAGMA busy_timeout=250")
try:
for attempt in range(attempts):
busy = conn.execute("PRAGMA wal_checkpoint(TRUNCATE)").fetchone()[0]
if not busy:
return True
if attempt + 1 < attempts:
time.sleep(delay)
finally:
conn.execute(f"PRAGMA busy_timeout={int(original_timeout)}")
LOGGER.warning("[maintenance] %s: WAL not truncated (readers still active); it shrinks at the next checkpoint", label)
return False
def get_portrait_img_max_width() -> int:
"""Max width (px) for portrait (taller-than-wide) images in article bodies;
landscape images stay full container width. 0 = disabled. Per-user DB
override takes precedence over the env fallback."""
with get_meta_connection() as conn:
val = get_setting(conn, SETTING_PORTRAIT_IMG_MAX_WIDTH)
if val:
try:
w = int(val)
if w >= 0:
return w
except ValueError:
pass
return _ENV_PORTRAIT_IMG_MAX_WIDTH
def mark_read_on_new_tab_enabled() -> bool:
"""Per-user: opening a post's article in a new tab from its list title also marks it read. Off by default."""
return get_runtime_setting(SETTING_MARK_READ_ON_NEW_TAB, "0") == "1"
def proxy_body_images_enabled() -> bool:
"""Per-user: preemptively proxy every article-pane body image through
/api/img instead of loading third-party URLs directly. Off by default."""
return get_runtime_setting(SETTING_PROXY_BODY_IMAGES, "0") == "1"
_PROXY_MODES = ("off", "as_needed", "always")
def get_proxy_url() -> str:
"""Outbound-fetch proxy endpoint (e.g. gluetun's socks5h://gluetun:1080).
Admin-managed; env default, DB override takes precedence."""
return get_instance_setting(SETTING_PROXY_URL, _ENV_PROXY_URL)
def get_proxy_mode() -> str:
"""off / as_needed / always for outbound feed fetches. Per-user DB setting
overrides the instance (first-admin) default; unset resolves to "off" —
the proxy is opt-in everywhere."""
val = get_instance_setting(SETTING_PROXY_MODE, "off")
return val if val in _PROXY_MODES else "off"
def get_tailscale_url() -> str:
"""Last-resort outbound proxy endpoint (e.g. Tailscale's
socks5h://tailscale:1080). Admin-managed; env default, DB override takes
precedence. Empty means not configured, which is itself the opt-in gate —
as_needed escalation never reaches this tier without one."""
return get_instance_setting(SETTING_TAILSCALE_URL, _ENV_TAILSCALE_URL)
def get_flaresolverr_url() -> str:
"""FlareSolverr endpoint (e.g. http://flaresolverr:8191/v1). Admin-managed;
env default, DB override takes precedence. Empty means not configured,
same opt-in gate as the other two backends."""
return get_instance_setting(SETTING_FLARESOLVERR_URL, _ENV_FLARESOLVERR_URL)
def get_img_cache_days() -> int:
"""Last-accessed TTL (days) for the /api/img cache. 0 = keep forever.
DB override (admin-managed) takes precedence over the env fallback."""
val = get_instance_setting(SETTING_IMG_CACHE_DAYS, "")
if val:
try:
d = int(val)
if d >= 0:
return d
except ValueError:
pass
return _ENV_IMG_CACHE_DAYS
def get_img_cache_max_dim() -> int:
"""Longest-side px to downscale cached images to. 0 = store originals as-is."""
val = get_instance_setting(SETTING_IMG_CACHE_MAX_DIM, "")
if val:
try:
d = int(val)
if d >= 0:
return d
except ValueError:
pass
return _ENV_IMG_CACHE_MAX_DIM
def get_img_target_bytes() -> int:
"""Byte budget for a cached image, above which it is re-encoded to WebP.
Separate from the max dimension because the two answer different questions:
an image can sit exactly at the dimension cap and still weigh 12 MB (a 4K
RGBA PNG does), and it is the bytes that make it slow to load.
0 = leave every image at whatever size it arrives.
"""
val = get_instance_setting(SETTING_IMG_TARGET_BYTES, "")
if val:
try:
n = int(val)
if n >= 0:
return n
except ValueError:
pass
return _ENV_IMG_TARGET_BYTES
# --- Auth config ---
# When set, skip the login form and auto-authenticate as the admin on every request.
# Intended for local/private-network installs that don't need a password prompt.
AUTO_LOGIN = os.getenv("LECTIO_AUTO_LOGIN", "0") == "1"
# Password hashing scheme for stored credentials. scrypt and pbkdf2_sha256 are stdlib;
# argon2 needs the optional argon2-cffi package.
PASSWORD_HASH_SCHEME = os.getenv("LECTIO_PASSWORD_HASH_SCHEME", passwords.DEFAULT_SCHEME).strip().lower()
if PASSWORD_HASH_SCHEME not in passwords.available_schemes():
LOGGER.warning(
"LECTIO_PASSWORD_HASH_SCHEME=%r not available (have: %s); using %s.",
PASSWORD_HASH_SCHEME,
", ".join(passwords.available_schemes()),
passwords.DEFAULT_SCHEME,
)
PASSWORD_HASH_SCHEME = passwords.DEFAULT_SCHEME
# Bootstrap admin account, seeded on first startup when the users table is empty.
# CHANGE the default password before exposing the instance.
_DEFAULT_ADMIN_PASSWORD = "ChangeA$ap"
BOOTSTRAP_ADMIN_USERNAME = os.getenv("LECTIO_ADMIN_USERNAME", "admin")
BOOTSTRAP_ADMIN_PASSWORD = os.getenv("LECTIO_ADMIN_PASSWORD", _DEFAULT_ADMIN_PASSWORD)
AUTH_ENABLED = True # kept as a module-level bool so tests can monkeypatch it
user_store: UserStore = UserStore(AUTH_DB_PATH)
SESSION_SECRET_KEY = os.getenv("LECTIO_SECRET_KEY") or secrets.token_hex(32)
if AUTH_ENABLED and not os.getenv("LECTIO_SECRET_KEY"):
LOGGER.warning(
"LECTIO_SECRET_KEY is not set — using a random key. Sessions will not survive server restarts. Set a stable key in your .env."
)
SESSION_MAX_AGE_SECONDS = 365 * 24 * 3600 # 1 year; baked into SessionMiddleware at startup
# Set LECTIO_HTTPS_ONLY=1 when running behind a TLS-terminating reverse proxy.
_HTTPS_ONLY = os.getenv("LECTIO_HTTPS_ONLY", "0") == "1"
# Proxies trusted for X-Forwarded-* headers. "*" (default) trusts any upstream —
# fine when the app port is only reachable via your reverse proxy. Set a
# comma-separated allowlist (e.g. the proxy's IP) for a stricter posture.
_TRUSTED_PROXIES = os.getenv("LECTIO_TRUSTED_PROXIES", "*").strip()
# Have the app emit baseline security headers itself. Default off because a
# reverse proxy (e.g. the bundled Traefik config) usually sets them; enable when
# fronting Lectio with a proxy that does not (Caddy without a header block, a
# bare/no-proxy setup, etc.). Keeps the headers from depending on the proxy.
_SECURITY_HEADERS_ENABLED = os.getenv("LECTIO_SECURITY_HEADERS", "0") == "1"
# Paths that are always public (no login required)
_AUTH_EXEMPT_PREFIXES = (
"/login",
"/static",
"/healthz",
"/api/img",
"/api/favicon",
"/api/save",
"/api/bookmarklet/save",
"/dev/feeds/",
"/fever",
"/greader/",
"/v1/",
"/websub/",
"/sw.js",
)
def _note_scheduler_progress(stage: str) -> None:
"""Record that the scheduled refresh advanced, and what it is doing now."""
with _scheduler_state_lock:
_scheduler_state["last_progress_at"] = time.monotonic()
_scheduler_state["stage"] = stage
def _scheduler_stall_seconds() -> float | None:
"""Seconds since the in-flight pass last advanced, or None when idle.
None while no pass is running: between passes there is nothing to stall.
"""
with _scheduler_state_lock:
started = _scheduler_state.get("pass_started_at")
last = _scheduler_state.get("last_progress_at")
if started is None or last is None:
return None
return time.monotonic() - float(last) # ty: ignore[invalid-argument-type]
def invalidate_meta_structure_cache() -> None:
with _meta_structure_lock:
_meta_structure_cache.clear()
# Refresh-contention investigation (Plan.md Tier 1): this cache is designed
# to invalidate only on explicit user mutations (subscribe/unsubscribe,
# folder changes, feed disable/enable) -- a 2026-09-03 live capture showed
# it going cold roughly every 30-45s during ordinary browsing (no such
# action taken), each miss costing 5+ real seconds of get_meta_structure_snapshot's
# 5-query rebuild under refresh contention. There are 37 call sites; rather
# than audit them all, log the caller so the next occurrence says which one
# actually fired during a routine refresh pass.
try:
caller = sys._getframe(1)
LOGGER.info(
"[perf] invalidate_meta_structure_cache from %s:%d (%s)", caller.f_code.co_filename, caller.f_lineno, caller.f_code.co_name
)
except Exception:
pass
def invalidate_problematic_feeds_cache() -> None:
with _problematic_feeds_cache_lock:
_problematic_feeds_cache.clear()
def get_problematic_feeds_cached(conn: sqlite3.Connection, limit: int = PROBLEMATIC_FEEDS_LIMIT) -> list[dict]:
"""TTL-cached problematic-feeds list (see cache comment above)."""
now_pf = time.time()
with _problematic_feeds_cache_lock:
cached_pf = _problematic_feeds_cache.get(limit)
if cached_pf and now_pf - cached_pf[0] < PROBLEMATIC_FEEDS_CACHE_TTL_SECONDS:
return [dict(r) for r in cached_pf[1]]
problematic_feeds = feed_refresh_service.get_problematic_feeds(conn, limit=limit)
with _problematic_feeds_cache_lock:
_problematic_feeds_cache[limit] = (time.time(), [dict(r) for r in problematic_feeds])
return [dict(r) for r in problematic_feeds]
def is_async_action_request(request: Request, expected_header: str | None = None) -> bool:
requested_with = (request.headers.get("X-Requested-With") or "").strip()
if not requested_with:
return False
if expected_header is None:
return requested_with.startswith("lectio-")
return requested_with == expected_header
@asynccontextmanager
async def lifespan(app: FastAPI):
# Startup
_attach_pending_access_filter()
ensure_thumb_schema()
ensure_img_cache_schema()
ensure_yt_duration_schema()
ensure_websub_schema()
bootstrap_admin()
# Bring every existing user's meta/starred schema up to current. Per-user DBs
# are otherwise schema-init'd only at provision time, so any table added
# after a user was provisioned (e.g. feed_fetch_history) is missing from
# their DB and surfaces as a "no such table" 500 (Feed Properties, etc.).
# ensure_*_schema are idempotent, so this is a cheap no-op once migrated.
def _ensure_user_schema() -> None:
ensure_meta_schema()
ensure_starred_archive_schema()
ensure_reader_indexes()
migrate_spaced_manual_tags()
_for_each_background_user("per-user schema migration", _ensure_user_schema)
# Kill switch for heavy startup work. Set LECTIO_DISABLE_STARTUP_BACKFILL=1
# to skip the scheduled-refresh loop, the FTS index build, and the
# lead-image / YouTube / archive / read-history backfills. Useful after an
# OPML import of hundreds of feeds (boot calm, let pages render, then unset
# and restart). The test suite also sets it so these background DB writers
# don't race a test's own DB ops on the same temp DB ("database is locked").
backfill_disabled = os.getenv("LECTIO_DISABLE_STARTUP_BACKFILL", "0") == "1"
if backfill_disabled:
LOGGER.warning("LECTIO_DISABLE_STARTUP_BACKFILL=1 — skipping FTS build, scheduled refresh, and backfill threads")
# reader's FTS index is no longer built or maintained: both search surfaces
# resolve in SQL (_search_entry_keys_in_sql, _filter_star_keys_by_search)
# because search_entries builds a highlighted snippet per result, which
# measured at ~95% of a 10-20s search. Keeping the index cost 1.3ms per new
# entry on every refresh plus a file about the size of the reader DB itself
# (564MB against 743MB on the live library) — all of it for an index no
# query touched. scripts/drop_search_index.py reclaims it on an existing
# install; a fresh one never creates it.
if LECTIO_PUBLIC_URL:
_migrate_websub_to_shared()
def _prepare_user_dbs() -> None:
# Checkpoint both WAL files so the first request does not have to rebuild
# the WAL index from a large file left over from the previous run. Use
# direct connections so we're not racing the reader.
for _ckpt_path in (tenancy.reader_db_path(), tenancy.meta_db_path()):
try:
with contextlib.closing(sqlite3.connect(str(_ckpt_path), timeout=5)) as _ckpt_conn:
_ckpt_conn.execute("PRAGMA wal_checkpoint(TRUNCATE)")
except Exception:
pass
with get_meta_connection() as conn:
purge_lower_level_folders(conn)
with _app_settings_cache_lock:
_app_settings_cache[tenancy.current_user_id()] = _load_app_settings_cache(conn)
with get_reader(): # creates the reader DB if missing
pass
_for_each_background_user("per-user DB prep", _prepare_user_dbs)
app.state.last_scheduled_refresh_started_at = time.monotonic()
# Warm YouTube duration in-memory cache from DB so first renders are instant.
youtube_duration_service.warm_cache_from_db()
# Warm lead image cache from DB so thumbnails are available on first render.
# Must run per-user: lead images live in each tenant's own meta DB, and the
# render path (get_cached_entry_thumbnail) only consults the in-memory cache
# with no per-user DB fallback, so every user's cache must be warmed or their
# thumbnails stay blank until the rate-limited background backfill catches up.
_for_each_background_user("lead-image cache warm", lead_image_service.warm_cache_from_db)
# Warm the per-user unread-counts cache so the first home load after a restart
# doesn't pay the synchronous reader-DB scan under startup contention (cold
# cache + concurrent first requests + a refresh kicking off was producing
# multi-second — occasionally ~20s — home loads right after a redeploy).
def _warm_unread_counts() -> None:
try:
counts = _compute_unread_counts_by_feed()
with unread_counts_cache_lock:
unread_counts_cache["unread_counts"] = (time.time(), counts)
except Exception:
LOGGER.debug("unread-counts cache warm failed", exc_info=True)
_for_each_background_user("unread-counts cache warm", _warm_unread_counts)
# Ensure existing scraped file:// feeds have their entries imported into the
# reader DB. Feeds may be in backoff from stale "no retriever" errors that
# pre-date the feed_root='' fix — clear that state and force a reader sync.
def _sync_scraped_feeds() -> None:
try:
with get_meta_connection() as _sc:
_sf_rows = _sc.execute("SELECT id, mode FROM scraped_feeds").fetchall()
for _sf_row in _sf_rows:
_sf_id = str(_sf_row["id"])
_sf_url = scraper_service.feed_file_url(_sf_id)
try:
with get_meta_connection() as _sc:
_sc.execute("DELETE FROM feed_failure_state WHERE feed_url = ?", (_sf_url,))
# link_list feeds seed initial links as hidden so only new links
# appear as entries. If ALL entries are still hidden (nothing has
# appeared yet), unhide them so the feed isn't permanently empty.
if str(_sf_row["mode"]) == "link_list":
_visible = _sc.execute(
"SELECT count(*) FROM scraped_entries WHERE scraped_feed_id=? AND NOT hidden",
(_sf_id,),
).fetchone()[0]
if _visible == 0:
_sc.execute(
"UPDATE scraped_entries SET hidden=0 WHERE scraped_feed_id=?",
(_sf_id,),
)
scraper_service._write_feed_file(_sc, _sf_id)
with get_reader() as _r:
_r.update_feed(_sf_url)
except Exception:
pass
except Exception:
LOGGER.exception("[scraper] startup scraped-feed sync failed")
if not backfill_disabled:
threading.Thread(
target=lambda: _for_each_background_user("scraped-feed sync", _sync_scraped_feeds),
daemon=True,
name="sync-scraped-feeds",
).start()
# Auto-taggers and the dedup cleanup write per-user tag/strategy state, so
# run them once per background user (each under its own tenancy context).
def _startup_auto_tag_and_dedup() -> None:
# Artwork tagger runs first so webcomic tagger won't clobber ArtStation
# feeds that live in folders whose name also contains "comic".
_auto_tag_artwork_feeds()
# Auto-tag feeds in "comic*" folders with strategy='webcomic'.
_auto_tag_webcomic_feeds()
# GitHub release feeds get og_scrape + no list thumbnail.
_auto_tag_github_release_feeds()
# Retroactive dedup: suppress duplicate unread entries (same slug or
# title+date) that accumulated before _suppress_guid_churn could catch them.
try:
with get_reader() as reader:
with get_meta_connection() as conn:
cleaned = _cleanup_intra_feed_slug_dupes(reader, conn)
if cleaned:
LOGGER.info("[guid-churn-cleanup] suppressed %d pre-existing duplicate entries", cleaned)
except Exception:
LOGGER.exception("[guid-churn-cleanup] startup cleanup failed")
_for_each_background_user("auto-tag/dedup", _startup_auto_tag_and_dedup)
stop_event = threading.Event()
if not backfill_disabled:
thread = threading.Thread(
target=scheduled_refresh_loop,
args=(stop_event,),
daemon=True,
)
app.state.refresh_stop_event = stop_event
app.state.refresh_thread = thread
thread.start()
# Shares the refresh loop's stop_event: the watchdog exists only to watch
# that loop, so the two start and stop together.
watchdog = threading.Thread(
target=scheduler_watchdog_loop,
args=(stop_event,),
daemon=True,
)
app.state.scheduler_watchdog_thread = watchdog
watchdog.start()
else:
app.state.refresh_stop_event = stop_event
app.state.refresh_thread = None
app.state.scheduler_watchdog_thread = None
# Backfill durations for any existing YouTube entries not yet stored.
def _backfill() -> None:
with get_meta_connection() as conn:
rows = conn.execute(
"SELECT DISTINCT feed_url FROM folder_feeds WHERE feed_url LIKE '%youtube.com/feeds/videos.xml%'"
).fetchall()
for row in rows:
youtube_duration_service.fetch_and_store_durations_for_feed(str(row["feed_url"]))
if not backfill_disabled:
threading.Thread(
target=lambda: _for_each_background_user("youtube-duration backfill", _backfill),
daemon=True,
).start()
# Backfill lead images for all feeds whose entries haven't been checked yet.
# `force_retry_negative=False` keeps this incremental: only feeds that have
# never been checked do work, so a startup right after an OPML import of
# hundreds of feeds doesn't flood the meta DB with writes and starve the
# request path.
def _backfill_lead_images() -> None:
with get_meta_connection() as conn:
rows = conn.execute("SELECT DISTINCT feed_url FROM folder_feeds").fetchall()
for row in rows:
lead_image_service.fetch_and_store_lead_images_for_feed(str(row["feed_url"]), force_retry_negative=False)
if not backfill_disabled:
threading.Thread(
target=lambda: _for_each_background_user("lead-image backfill", _backfill_lead_images),
daemon=True,
).start()
# Start the starred archive worker, then backfill pending rows for any
# saved entries that don't yet have a complete archive (covers the
# initial rollout and any re-stars after maintenance pruning). Also
# one-shot metadata backfill for archive rows that completed before
# title/link/etc columns were added to the schema.
if not backfill_disabled:
starred_archive_service.start_worker()
def _archive_backfill_task() -> None:
starred_archive_service.backfill_saved_entries_from_archive()
starred_archive_service.backfill_missing_archives()
starred_archive_service.backfill_metadata_for_complete_rows()
if not backfill_disabled:
threading.Thread(
target=lambda: _for_each_background_user("starred-archive backfill", _archive_backfill_task),
daemon=True,
name="starred-archive-backfill",
).start()
# One-time backfill: populate read_history from entry_read_state if history is empty.
def _backfill_read_history() -> None:
try:
with get_meta_connection() as conn:
count = conn.execute("SELECT COUNT(*) FROM read_history").fetchone()[0]
if count > 0:
return
rows = conn.execute(
"SELECT feed_url, entry_id, read_at FROM entry_read_state ORDER BY read_at DESC LIMIT ?",
(READ_HISTORY_CAP,),
).fetchall()
if not rows:
return
needed = {(str(r["feed_url"]), str(r["entry_id"])): str(r["read_at"]) for r in rows}
to_insert = []
with get_reader() as reader:
for (feed_url, entry_id), read_at in needed.items():
entry = reader.get_entry((feed_url, entry_id), None)
if entry is None:
continue
feed = reader.get_feed(feed_url, None)
to_insert.append(
(
feed_url,
entry_id,
str(getattr(entry, "title", None) or ""),
str(getattr(entry, "link", None) or ""),
str(getattr(feed, "resolved_title", None) or getattr(feed, "title", None) or ""),
read_at,
)
)
if not to_insert:
return
with get_meta_connection() as conn:
conn.executemany(
"INSERT INTO read_history (feed_url, entry_id, title, link, feed_title, read_at)"
" VALUES (?, ?, ?, ?, ?, ?)"
" ON CONFLICT(feed_url, entry_id) DO NOTHING",
to_insert,
)
LOGGER.info("[read_history] backfilled %d entries from entry_read_state", len(to_insert))
except Exception:
LOGGER.exception("[read_history] backfill error")
if not backfill_disabled:
threading.Thread(
target=lambda: _for_each_background_user("read-history backfill", _backfill_read_history),
daemon=True,
name="read-history-backfill",
).start()
# Daily Maintenance loop — runs once per day at the configured maintenance hour.
# Gated by the same kill switch as the other daemons: the loop is a persistent
# thread that ticks every 30s doing per-user DB writes (email-batch flush, and
# a full maintenance run when due). Under the test suite (which sets
# LECTIO_DISABLE_STARTUP_BACKFILL=1) it would otherwise keep running for the
# rest of the pytest process once any TestClient test triggered the lifespan,
# racing later tests' fresh-DB setup as an intermittent "database is locked".
# Tests that need maintenance behavior call _run_daily_maintenance() directly.
maint_stop_event = threading.Event()
app.state.maint_stop_event = maint_stop_event
if not backfill_disabled:
threading.Thread(
target=_daily_maintenance_loop,
args=(maint_stop_event,),
daemon=True,
name="daily-maintenance",
).start()
# In debug mode, auto-subscribe dev feeds to the _Lectio folder.
if DEBUG_MODE:
_dev_base = "http://127.0.0.1:8000"
_dev_urls = [
f"{_dev_base}/dev/feeds/email-match.xml",
f"{_dev_base}/dev/feeds/email-match.atom",
f"{_dev_base}/dev/feeds/email-match.json",
f"{_dev_base}/dev/feeds/email-skip.xml",
f"{_dev_base}/dev/feeds/email-skip.atom",
f"{_dev_base}/dev/feeds/email-skip.json",
]
with get_meta_connection() as conn:
_lectio_folder_id = _get_lectio_folder_id(conn)
if _lectio_folder_id:
_dev_newly_added: list[str] = []
for _dev_url in _dev_urls:
try:
with get_meta_connection() as _c:
_already = _c.execute(
"SELECT 1 FROM folder_feeds WHERE folder_id = ? AND feed_url = ?",
(_lectio_folder_id, _dev_url),
).fetchone()
if not _already:
add_feed_to_folder(_dev_url, _lectio_folder_id)
_dev_newly_added.append(_dev_url)
except Exception:
LOGGER.exception("[dev-feeds] failed to subscribe %s", _dev_url)
if _dev_newly_added:
feed_refresh_service.update_feeds(_dev_newly_added)
try:
yield
finally:
stop_event = getattr(app.state, "refresh_stop_event", None)
thread = getattr(app.state, "refresh_thread", None)
if stop_event and thread:
stop_event.set()
# A scheduled refresh in flight can take tens of seconds (one HTTP
# timeout per feed). Wait long enough to let it finish cleanly so
# we don't leave the meta DB in a partial-write state.
shutdown_timeout = float(os.getenv("LECTIO_SHUTDOWN_TIMEOUT_SECONDS", "30"))
thread.join(timeout=shutdown_timeout)
if thread.is_alive():
LOGGER.warning(
"[shutdown] refresh worker did not finish within %.0fs; abandoning (daemon thread will be killed by interpreter exit)",
shutdown_timeout,
)
# Stop the starred archive worker — short timeout, since the only
# work in flight is HTTP fetches with 15s ceilings.
try:
starred_archive_service.stop_worker(timeout=5.0)
except Exception as exc: # noqa: BLE001
LOGGER.warning("[shutdown] starred archive worker stop failed: %s", exc)
yt_stop = getattr(app.state, "yt_stop_event", None)
if yt_stop:
yt_stop.set()
app = FastAPI(title="Lectio", lifespan=lifespan)
def _session_logged_in(request: Request) -> bool:
"""Whether the session is fully authenticated (requires a valid user_id)."""
if not request.session.get("authenticated"):
return False
uid = request.session.get("user_id")
return bool(uid and tenancy.is_valid_user_id(uid))
class _AuthMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next):
if not AUTH_ENABLED:
return await call_next(request)
path = request.url.path
if any(path.startswith(p) for p in _AUTH_EXEMPT_PREFIXES):
return await call_next(request)
if _session_logged_in(request):
return await call_next(request)
if AUTO_LOGIN:
# Auto-authenticate as the first non-disabled admin user without
# showing the login form. Intended for local/private-network installs.
for _u in user_store.list_users():
if _u.get("is_admin") and not _u.get("disabled"):
request.session["authenticated"] = True
request.session["user_id"] = _u["user_id"]
break
if _session_logged_in(request):
return await call_next(request)
# Store a same-origin relative path (path + query) so _safe_next accepts
# it and the user lands back on the deep link after logging in — e.g. a
# /?subscribe=<feed> quick-subscription link. An absolute URL would be
# rejected by _safe_next as an open-redirect and drop the user on "/".
next_url = request.url.path
if request.url.query:
next_url += "?" + request.url.query
return RedirectResponse(url=f"/login?next={quote_plus(next_url)}", status_code=303)
class _SecurityHeadersMiddleware:
"""Emit baseline security response headers from the app (opt-in via
LECTIO_SECURITY_HEADERS=1), so they hold regardless of the reverse proxy.
Mirrors the headers the bundled Traefik config sets. Existing headers are not
overwritten, so a proxy that already sets them wins and there are no
duplicates. HSTS is only added when LECTIO_HTTPS_ONLY=1."""
def __init__(self, app, *, hsts: bool):
self.app = app
self._headers: list[tuple[bytes, bytes]] = [
(b"x-content-type-options", b"nosniff"),
(b"x-frame-options", b"DENY"),
(b"referrer-policy", b"no-referrer-when-downgrade"),
]
if hsts:
self._headers.append((b"strict-transport-security", b"max-age=31536000; includeSubDomains"))
async def __call__(self, scope, receive, send):
if scope.get("type") != "http":
return await self.app(scope, receive, send)
async def send_wrapper(message):
if message["type"] == "http.response.start":
headers = message.setdefault("headers", [])
present = {k.lower() for k, _ in headers}
for key, val in self._headers:
if key not in present:
headers.append((key, val))
await send(message)
await self.app(scope, receive, send_wrapper)
def _touch_user_last_seen(uid: str) -> None:
"""Record per-user activity time, throttled (this runs on every request)."""
now = time.time()
with _last_seen_touch_lock:
if now - _last_seen_touch.get(uid, 0.0) < _LAST_SEEN_THROTTLE_SECONDS:
return
_last_seen_touch[uid] = now
try:
user_store.touch_last_seen(uid, now)
except Exception:
LOGGER.debug("touch_last_seen failed", exc_info=True)
class _TenancyMiddleware:
"""Bind the request's user into the tenancy context.
Pure-ASGI and registered innermost (downstream of the auth gate), so the
binding wraps the route handler. Sync handlers run via anyio's threadpool,
which copies the current contextvars into the worker thread, so a value set
here is visible to get_reader() / get_meta_connection() deep in the call
stack.
Requests without an authenticated session user or API token (static assets,
the login page, other unauthenticated hits) run unbound: anything they touch
must be global, and a per-user DB access raises TenancyUnboundError.
"""
def __init__(self, app):
self.app = app
@staticmethod
def _greader_user_from_scope(scope) -> str | None:
"""Resolve a /greader/ request's bearer token to a user, using only the
Authorization header / query string (no body read needed)."""
token = ""
for key, val in scope.get("headers", []):
if key == b"authorization":
auth = val.decode("latin-1")
if auth.startswith("GoogleLogin auth="):
token = auth[17:].strip()
elif auth.startswith("Bearer "):
token = auth[7:].strip()
break
if not token:
qs = parse_qs(scope.get("query_string", b"").decode("latin-1"))
token = (qs.get("token") or [""])[0]
return user_store.resolve_greader_token(token)
async def __call__(self, scope, receive, send):
if scope.get("type") != "http":
return await self.app(scope, receive, send)
path = scope.get("path", "")
uid: str | None = None
if path.startswith("/greader/"):
# API requests carry their identity in a bearer token, not a session.
uid = self._greader_user_from_scope(scope)
else:
session = scope.get("session") or {}
sid = session.get("user_id")
if sid and session.get("authenticated") and tenancy.is_valid_user_id(sid):
uid = sid
if uid and tenancy.is_valid_user_id(uid):
_touch_user_last_seen(uid)
token = tenancy.set_current_user(uid)
try:
return await self.app(scope, receive, send)
finally:
tenancy.reset_current_user(token)
return await self.app(scope, receive, send)
# Paths exempt from CSRF validation. /login is the auth gate itself (rate-
# limited separately). /static and /healthz are GET-only anyway, but listing
# explicitly documents intent.
_CSRF_EXEMPT_PREFIXES = (
"/login",
"/static",
"/healthz",
"/api/img",
"/api/favicon",
"/api/save",
"/api/bookmarklet/save",
"/fever",
"/greader/",
"/v1/",
"/websub/",
)
_CSRF_SAFE_METHODS = frozenset({"GET", "HEAD", "OPTIONS"})
_CSRF_SESSION_KEY = "csrf_token"
_CSRF_HEADER_NAME = "x-csrf-token"
_CSRF_FORM_FIELD = "_csrf"
def _ensure_csrf_token(session: dict) -> str:
"""Return the session's CSRF token, generating one on first use.
Token is per-session (rotated whenever the session itself rotates, e.g.
on logout/login). Stored in the session cookie which is HttpOnly + signed
so attackers on other origins can neither read nor forge it.
"""
token = session.get(_CSRF_SESSION_KEY)
if not token:
token = secrets.token_urlsafe(32)
session[_CSRF_SESSION_KEY] = token
return token
class _AccessLogMiddleware:
"""Replaces uvicorn's built-in access log so we can suppress the prefetch
204/503s without wrestling with uvicorn's logger config (filters added to
uvicorn.access weren't taking effect in this deployment).
Format mirrors uvicorn's: ``CLIENT - "METHOD PATH HTTP/x.y" STATUS REASON``.
"""
_SUPPRESS_STATUSES = (204, 503)
_STATUS_REASONS = {
200: "OK",
204: "No Content",
301: "Moved Permanently",
302: "Found",
303: "See Other",
304: "Not Modified",
307: "Temporary Redirect",
308: "Permanent Redirect",
400: "Bad Request",
401: "Unauthorized",
403: "Forbidden",
404: "Not Found",
405: "Method Not Allowed",
409: "Conflict",
413: "Payload Too Large",
429: "Too Many Requests",
500: "Internal Server Error",
502: "Bad Gateway",
503: "Service Unavailable",
}
def __init__(self, app):
self.app = app
async def __call__(self, scope, receive, send):
if scope["type"] != "http":
await self.app(scope, receive, send)
return
_t0 = time.monotonic()
status_holder = [0]
async def send_wrapper(message):
if message.get("type") == "http.response.start":
status_holder[0] = int(message.get("status", 0))
await send(message)
try:
await self.app(scope, receive, send_wrapper)
finally:
status = status_holder[0]
elapsed_ms = int((time.monotonic() - _t0) * 1000)
client = scope.get("client") or ("?", 0)
client_addr = f"{client[0]}:{client[1]}"
method = scope.get("method", "?")
raw_path = scope.get("raw_path", b"") or scope.get("path", "").encode()
qs = scope.get("query_string", b"") or b""
request_path = raw_path.decode("latin-1", errors="replace")
if qs:
request_path = f"{request_path}?{qs.decode('latin-1', errors='replace')}"
http_version = scope.get("http_version", "1.1")
request_line = f"{method} {request_path} HTTP/{http_version}"
# Suppression: prefetch 204/503 with list_feed_url= are pure noise.
if not (status in self._SUPPRESS_STATUSES and "list_feed_url=" in request_path):
reason = self._STATUS_REASONS.get(status, "")
duration = f" {elapsed_ms}ms" if elapsed_ms >= 200 else ""
LOGGER.info('%s - "%s" %d %s%s', client_addr, request_line, status, reason, duration)
class _RejectPrefetchMiddleware:
"""Two-layer defense against browser link-prefetch floods.
1. **Drop prefetches outright.** Browsers tag them with one of several
headers (Sec-Purpose, Purpose, X-Moz, Sec-Fetch-Dest=empty for nav-style
prefetch). Match generously — false positives just make a prefetch miss,
which is fine.
2. **Concurrency cap on home() requests.** Backstop in case detection
misses a browser variant: hold a semaphore so only N home renders run
at once. Excess returns 503, which tells the prefetcher to stop. /healthz
and /static bypass both layers so probes and asset prefetches still work.
"""
PASSTHROUGH_PREFIXES = ("/healthz", "/static", "/login", "/api/img", "/api/favicon")
def __init__(self, app):
self.app = app
async def __call__(self, scope, receive, send):
if scope["type"] != "http":
await self.app(scope, receive, send)
return
path = scope.get("path", "")
if any(path.startswith(p) for p in self.PASSTHROUGH_PREFIXES):
await self.app(scope, receive, send)
return
headers = scope.get("headers") or []
is_prefetch = False
is_spa_fetch = False
sec_fetch_dest = b""
sec_fetch_mode = b""
x_requested_with = b""
user_agent = b""
referer = b""
for header_name, header_value in headers:
name = header_name.decode("latin-1").lower()
if name == "sec-purpose":
if b"prefetch" in header_value.lower() or b"prerender" in header_value.lower():
is_prefetch = True
elif name == "purpose":
low = header_value.lower()
if b"prefetch" in low or b"preview" in low or b"prerender" in low:
is_prefetch = True
elif name == "x-moz":
if b"prefetch" in header_value.lower() or b"prerender" in header_value.lower():
is_prefetch = True
elif name == "sec-fetch-dest":
sec_fetch_dest = header_value
elif name == "sec-fetch-mode":
sec_fetch_mode = header_value
elif name == "x-requested-with":
x_requested_with = header_value
if b"lectio" in header_value.lower():
is_spa_fetch = True
elif name == "user-agent":
user_agent = header_value
elif name == "referer":
referer = header_value
# Heuristic: browser-side prefetch via fetch() (Chrome NoState Prefetch
# and similar) doesn't always set Sec-Purpose, but it has a unique
# fingerprint — Sec-Fetch-Dest=empty + Sec-Fetch-Mode=cors on a path
# that isn't an actual API endpoint, with no X-Requested-With from our
# SPA. Real SPA fetches always send X-Requested-With=lectio-...; real
# top-level navigations send Sec-Fetch-Dest=document Sec-Fetch-Mode=navigate.
if not is_prefetch and not is_spa_fetch and sec_fetch_dest == b"empty" and sec_fetch_mode == b"cors" and path == "/":
is_prefetch = True
# Diagnostic: dump the first few suspect-looking requests so we can
# see what's actually firing them. Capped to avoid log spam.
with _prefetch_header_log_lock:
remaining = _prefetch_header_log_remaining[0]
if remaining > 0:
with _prefetch_header_log_lock:
if _prefetch_header_log_remaining[0] > 0:
_prefetch_header_log_remaining[0] -= 1
LOGGER.info(
"[prefetch-diag] path=%s qs=%r dest=%r mode=%r xrw=%r referer=%r ua=%r",
path,
scope.get("query_string", b"")[:200],
sec_fetch_dest,
sec_fetch_mode,
x_requested_with,
referer[:120],
user_agent[:120],
)
if is_prefetch:
response = Response(status_code=204, headers={"Cache-Control": "no-store"})
await response(scope, receive, send)
return
if path == "/" or path.startswith("/?"):
await self.app(scope, receive, send)
return
await self.app(scope, receive, send)
class _CSRFMiddleware:
"""Pure-ASGI CSRF protection.
On unsafe methods (POST/PUT/PATCH/DELETE) requires a token that matches
the session's stored token. Token is supplied either via:
- X-CSRF-Token header (used by SPA fetch handlers), or
- `_csrf` form field (for plain HTML form submits / multipart uploads).
Read-only methods are allowed through unchanged. The body is buffered and
re-streamed to downstream handlers via a replay receive callable so route
handlers can still parse it normally.
"""
def __init__(self, app):
self.app = app
async def __call__(self, scope, receive, send):
if scope["type"] != "http":
await self.app(scope, receive, send)
return
# Ensure every request has a CSRF token in its session so templates
# can render it and SPA fetches can read it from a meta tag. The
# token is created lazily on first request after a fresh session.
session = scope.get("session")
if session is not None and not session.get(_CSRF_SESSION_KEY):
session[_CSRF_SESSION_KEY] = secrets.token_urlsafe(32)
method = scope.get("method", "GET").upper()
path = scope.get("path", "")
if method in _CSRF_SAFE_METHODS or any(path.startswith(p) for p in _CSRF_EXEMPT_PREFIXES):
await self.app(scope, receive, send)
return
# Buffer the request body so downstream handlers can still read it.
body_chunks: list[bytes] = []
more_body = True
while more_body:
message = await receive()
if message["type"] != "http.request":
# Unexpected (e.g. http.disconnect); pass through unchanged.
await self.app(scope, receive, send)
return
body_chunks.append(message.get("body", b""))
more_body = bool(message.get("more_body", False))
body = b"".join(body_chunks)
# Token from header (preferred for SPA fetches).
submitted: str | None = None
for header_name, header_value in scope.get("headers") or []:
if header_name.decode("latin-1").lower() == _CSRF_HEADER_NAME:
submitted = header_value.decode("latin-1")
break
# Fall back to a `_csrf` field in form-encoded bodies.
if not submitted and body:
content_type = b""
for header_name, header_value in scope.get("headers") or []:
if header_name.decode("latin-1").lower() == "content-type":
content_type = header_value
break
ctype = content_type.decode("latin-1", errors="ignore").lower()
if ctype.startswith("application/x-www-form-urlencoded"):
from urllib.parse import parse_qs
try:
form = parse_qs(body.decode("utf-8", errors="ignore"))
values = form.get(_CSRF_FORM_FIELD, [])
if values:
submitted = values[0]
except Exception:
submitted = None
elif ctype.startswith("multipart/form-data"):
# Cheap multipart probe: scan the raw body for the field name.
# Avoids pulling in a streaming parser for the one OPML form.
marker = b'name="' + _CSRF_FORM_FIELD.encode() + b'"'
idx = body.find(marker)
if idx >= 0:
# The value sits two CRLFs after the disposition line.
sep = b"\r\n\r\n"
val_start = body.find(sep, idx)
if val_start >= 0:
val_start += len(sep)
val_end = body.find(b"\r\n", val_start)
if val_end >= 0:
try:
submitted = body[val_start:val_end].decode("utf-8", errors="ignore")
except Exception:
submitted = None
expected = (scope.get("session") or {}).get(_CSRF_SESSION_KEY)
if not expected or not submitted or not secrets.compare_digest(str(submitted), str(expected)):
response = JSONResponse(
{"detail": "CSRF token missing or invalid."},
status_code=403,
)
await response(scope, receive, send)
return
# Replay the buffered body to downstream handlers.
replayed = False
async def replay_receive():
nonlocal replayed
if not replayed:
replayed = True
return {"type": "http.request", "body": body, "more_body": False}
return {"type": "http.disconnect"}
await self.app(scope, replay_receive, send)
# Middleware order matters. Starlette's add_middleware does insert(0) and the
# stack-builder reverses, so LAST add_middleware = OUTERMOST = runs FIRST
# inbound. We need this inbound chain:
# _RejectPrefetchMiddleware (drop browser speculation-rules prefetches)
# ProxyHeaders (rewrite client IP / scheme from X-Forwarded-*)
# → SessionMiddleware (parse signed session cookie → scope["session"])
# → _CSRFMiddleware (validate POST tokens; needs scope["session"])
# → _AuthMiddleware (gate on session["authenticated"]; needs scope["session"])
# → app
# Therefore add Auth FIRST (innermost) and outer middlewares LAST.
# _TenancyMiddleware is added before Auth so it is innermost of all — it binds
# the per-user context right around the endpoint, after the auth gate has run.
app.add_middleware(_TenancyMiddleware)
app.add_middleware(_AuthMiddleware)
app.add_middleware(_CSRFMiddleware)
app.add_middleware(
SessionMiddleware,
secret_key=SESSION_SECRET_KEY,
max_age=SESSION_MAX_AGE_SECONDS,
https_only=_HTTPS_ONLY,
same_site="lax",
)
# When running behind Traefik/nginx/Caddy, trust forwarded headers so
# uvicorn sees the real scheme and client IP.
if _HTTPS_ONLY:
from uvicorn.middleware.proxy_headers import ProxyHeadersMiddleware
_trusted_hosts = "*" if _TRUSTED_PROXIES == "*" else [h.strip() for h in _TRUSTED_PROXIES.split(",") if h.strip()]
app.add_middleware(ProxyHeadersMiddleware, trusted_hosts=_trusted_hosts)
# Drop prefetch traffic before any other middleware does work.
app.add_middleware(_RejectPrefetchMiddleware)
# App-emitted security headers (opt-in), outermost so they apply to every
# response regardless of which inner middleware produced it.
if _SECURITY_HEADERS_ENABLED:
app.add_middleware(_SecurityHeadersMiddleware, hsts=_HTTPS_ONLY)
# Compress responses before they reach Traefik so Traefik's own compress
# middleware sees Content-Encoding: gzip and passes through without buffering.
# Buffering the full 500KB page before sending a single byte was the main
# culprit for slow first-byte times on variable connections.
app.add_middleware(GZipMiddleware, minimum_size=1000)
# Access log replaces uvicorn's built-in (disabled via --no-access-log) so we
# can suppress prefetch 204/503 noise. Add LAST so it sees the final response.
app.add_middleware(_AccessLogMiddleware)
class _CachedStaticFiles(StaticFiles):
"""StaticFiles that sends a long-lived Cache-Control header.
Safe because every <link>/<script> URL includes a `?v={STATIC_ASSET_VERSION}`
cache-buster — bump the version to invalidate. Vendored third-party assets
under static/vendor/<lib>-<version>/ (e.g. KaTeX) are the exception: they
carry no `?v=` query param at all, since a vendored release is immutable —
a version bump there means a new directory name, not a hash recompute, so
the long-lived cache header is still safe without the query-string scheme.
"""
async def get_response(self, path: str, scope):
response = await super().get_response(path, scope)
if response.status_code == 200:
response.headers["Cache-Control"] = "public, max-age=31536000, immutable"
return response
# Python's mimetypes table predates .webmanifest, so StaticFiles would serve the
# PWA manifest as octet-stream and Chrome would refuse to install the app.
mimetypes.add_type("application/manifest+json", ".webmanifest")
app.mount("/static", _CachedStaticFiles(directory=str(BASE_DIR / "static")), name="static")
templates = Jinja2Templates(directory=str(BASE_DIR / "templates"))
templates.env.filters["urlencode"] = lambda value: quote_plus(str(value))
def _csrf_token_for(request: Request) -> str:
"""Read the request's CSRF token; ensure one exists if the session lacks it."""
return _ensure_csrf_token(request.session)
def _csrf_input(request: Request) -> str:
"""Render a hidden CSRF input for inclusion in <form> blocks."""
token = _csrf_token_for(request)
return f'<input type="hidden" name="{_CSRF_FORM_FIELD}" value="{token}" />'
# Expose helpers as Jinja globals so templates can call csrf_input(request).
# ty's strict view of globals' value-type doesn't accept arbitrary callables;
# Jinja itself does. Suppress the false positives.
templates.env.globals["csrf_input"] = _csrf_input # ty: ignore[invalid-assignment]
templates.env.globals["csrf_token_for"] = _csrf_token_for # ty: ignore[invalid-assignment]
@dataclass
class FolderOption:
id: int
name: str
path: str
depth: int
@dataclass
class FeedInFolder:
url: str
title: str
icon_url: str | None
unread_count: int
has_error: bool = False
disabled: bool = False
_meta_conn_local = threading.local()
# Refresh-contention investigation (Plan.md Tier 1) -- see
# services.reader_api's identical _TimedConnection for the full rationale.
# Kept as a separate copy rather than importing reader_api's: it wall-clock-
# times every statement issued over the META connection specifically, so a
# slow meta_block gets its own "db=meta" tag distinguishing it from the
# reader-DB one.
_SLOW_SQL_MS = 250
# "Read Above/Below" lead: elapsed_ms alone can't distinguish a genuinely slow
# query from one that spent nearly all its wall time in SQLite's busy-handler
# retry loop -- both look identical from outside. The progress handler fires
# every _PROGRESS_HANDLER_N VM instructions *during execution*, not during the
# busy-wait, so near-zero steps on a "slow" query is direct proof of lock-wait.
# Read-only: the handler always returns 0, never aborting a statement.
_PROGRESS_HANDLER_N = 1000
class _TimedMetaCursor(sqlite3.Cursor):
def execute(self, sql, parameters=()):
conn = cast("_TimedMetaConnection", self.connection)
conn._lectio_progress_steps = 0
self._lectio_sql = sql
start = time.perf_counter()
try:
return super().execute(sql, parameters)
finally:
elapsed_ms = (time.perf_counter() - start) * 1000
if elapsed_ms > _SLOW_SQL_MS:
LOGGER.info(
"[perf] slow_sql db=meta elapsed_ms=%d progress_steps=%d sql=%s",
int(elapsed_ms),
conn._lectio_progress_steps,
" ".join(str(sql).split())[:200],
)
def executemany(self, sql, parameters):
conn = cast("_TimedMetaConnection", self.connection)
conn._lectio_progress_steps = 0
start = time.perf_counter()
try:
return super().executemany(sql, parameters)
finally:
elapsed_ms = (time.perf_counter() - start) * 1000
if elapsed_ms > _SLOW_SQL_MS:
LOGGER.info(
"[perf] slow_sql db=meta elapsed_ms=%d progress_steps=%d executemany sql=%s",
int(elapsed_ms),
conn._lectio_progress_steps,
" ".join(str(sql).split())[:200],
)
def fetchall(self):
# `execute()` above only times statement *preparation*; a plain SELECT
# doesn't eagerly scan, so the real cost of `conn.execute(...).fetchall()`
# (the dominant read pattern in this codebase) happened here, previously
# invisible to slow-SQL logging entirely -- found 2026-09-03 chasing a
# live stall that traced back to exactly this shape (Plan.md Tier 1).
# Separate before/after progress-step snapshot so this doesn't double
# count whatever execute() already attributed to itself.
conn = cast("_TimedMetaConnection", self.connection)
steps_before = conn._lectio_progress_steps
start = time.perf_counter()
try:
return super().fetchall()
finally:
elapsed_ms = (time.perf_counter() - start) * 1000
if elapsed_ms > _SLOW_SQL_MS:
sql = getattr(self, "_lectio_sql", "?")
LOGGER.info(
"[perf] slow_sql db=meta elapsed_ms=%d progress_steps=%d fetchall sql=%s",
int(elapsed_ms),
conn._lectio_progress_steps - steps_before,
" ".join(str(sql).split())[:200],
)
class _TimedMetaConnection(sqlite3.Connection):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self._lectio_progress_steps = 0
self.set_progress_handler(self._lectio_on_progress, _PROGRESS_HANDLER_N)
def _lectio_on_progress(self) -> int:
self._lectio_progress_steps += 1
return 0 # never abort the statement
def cursor(self, factory=None):
return super().cursor(factory or _TimedMetaCursor)
def execute(self, sql, parameters=()):
# Delegate to self.cursor() rather than the sqlite3.Connection.execute()
# shortcut directly: the shortcut returns a bare sqlite3.Cursor (verified
# -- it does not route through the overridden cursor() factory at all),
# so `conn.execute(...).fetchall()` -- the dominant read pattern in this
# module -- was silently bypassing _TimedMetaCursor's fetchall() override
# above. Going through self.cursor().execute(...) guarantees callers
# always get a _TimedMetaCursor back.
return self.cursor().execute(sql, parameters)
def executemany(self, sql, parameters):
# Batched-write flush points (lead images, alt/title, entry_read_state)
# use conn.executemany directly -- routed through self.cursor() for the
# same reason execute() is above.
return self.cursor().executemany(sql, parameters)
def get_meta_connection() -> sqlite3.Connection:
"""Per-(thread, user) persistent SQLite connection to the meta DB.
The target DB is resolved through the tenancy seam from the current user
(defaults to the single legacy user). Each worker thread keeps one
connection per user_id it has served.
Opening a fresh connection per request was a measured bottleneck: simple
SELECTs that take 5ms standalone took 2-3s under 8-way concurrency because
every request paid for file open, schema load, mmap setup, and PRAGMA
application. Each uvicorn worker thread now opens once and reuses, so the
cost is paid 8-40 times total instead of N-times-per-request.
Note: sqlite3.Connection's context-manager protocol commits the transaction
on success but does NOT close the connection — the existing
`with get_meta_connection() as conn:` pattern continues to work and is
transaction-scoped per call site, while the underlying connection persists.
"""
uid = tenancy.current_user_id()
pool = getattr(_meta_conn_local, "pool", None)
if pool is None:
pool = {}
_meta_conn_local.pool = pool
conn = pool.get(uid)
if conn is not None:
return conn
_meta_factory = _TimedMetaConnection if LECTIO_PERF_DEBUG else sqlite3.Connection
conn = sqlite3.connect(str(tenancy.meta_db_path(uid)), timeout=10.0, factory=_meta_factory)
conn.row_factory = sqlite3.Row
# WAL + busy_timeout so overlapping writers (e.g. background refresh writing
# folder_feeds while a request persists a setting) wait briefly instead of
# immediately failing with "database is locked".
conn.execute("PRAGMA journal_mode=WAL")
conn.execute("PRAGMA busy_timeout=10000")
# Keep WAL small: checkpoint at 200 pages (~800KB) rather than the default
# 1000 pages so the file never balloons to tens of MB between restarts.
# Raised to 1000 and reverted same day, 2026-09-03 (Plan.md Tier 1, Read
# Above/Below lead) -- see services/reader_api.py's _LectioReaderStorage
# docstring for the full writeup: the experiment showed no benefit (a live
# capture right after deploy was worse, not better) and a follow-up
# capture found near-zero real work across many different queries in both
# DBs, arguing against checkpoint frequency as the mechanism.
conn.execute("PRAGMA wal_autocheckpoint=200")
# Whenever a checkpoint restarts the WAL, cut the file back to this size instead of leaving a vacuum-sized one behind.
conn.execute("PRAGMA journal_size_limit=67108864")
# synchronous=NORMAL 2026-09-03 (Plan.md Tier 1, Read Above/Below lead),
# replacing SQLite's compiled-in FULL default -- see
# services/reader_api.py's _LectioReaderStorage docstring for the full
# writeup; kept in sync here since both connections share the theory.
conn.execute("PRAGMA synchronous=NORMAL")
pool[uid] = conn
return conn
_websub_conn_local = threading.local()
WEBSUB_DB_PATH = DATA_DIR / "lectio_websub.sqlite"
def get_websub_connection() -> sqlite3.Connection:
"""Per-thread connection to the shared (non-per-user) WebSub subscription DB."""
pool = getattr(_websub_conn_local, "pool", None)
if pool is not None:
return pool
conn = sqlite3.connect(str(WEBSUB_DB_PATH), timeout=10.0)
conn.row_factory = sqlite3.Row
conn.execute("PRAGMA journal_mode=WAL")
conn.execute("PRAGMA busy_timeout=10000")
conn.execute("PRAGMA wal_autocheckpoint=200")
_websub_conn_local.pool = conn
return conn
def get_thumb_connection() -> sqlite3.Connection:
conn = sqlite3.connect(str(THUMB_DB_PATH))
conn.row_factory = sqlite3.Row
return conn
def ensure_websub_schema() -> None:
"""Create the shared (non-per-user) WebSub tables if they don't exist."""
with get_websub_connection() as conn:
conn.execute(
"""
CREATE TABLE IF NOT EXISTS websub_subscriptions (
feed_url TEXT PRIMARY KEY,
hub_url TEXT,
secret TEXT,
subscribed_at REAL,
verified INTEGER DEFAULT 0,
expires_at REAL,
lease_seconds INTEGER,
hub_tried_at REAL
)
"""
)
conn.execute(
"""
CREATE TABLE IF NOT EXISTS websub_subscribers (
feed_url TEXT NOT NULL,
user_id TEXT NOT NULL,
PRIMARY KEY (feed_url, user_id)
)
"""
)
def _migrate_websub_to_shared() -> None:
"""One-time migration: copy per-user websub_subscriptions rows to the shared DB.
Idempotent (INSERT OR IGNORE) so safe to re-run on every startup.
After this runs, the service only writes to the shared DB."""
for uid in _background_user_ids():
with tenancy.user_context(uid):
try:
conn = get_meta_connection()
rows = conn.execute(
"SELECT feed_url, hub_url, secret, subscribed_at, verified, "
"expires_at, lease_seconds, hub_tried_at "
"FROM websub_subscriptions"
).fetchall()
except Exception:
continue # table may not exist for new users
if not rows:
continue
with get_websub_connection() as wconn:
for r in rows:
wconn.execute(
"INSERT OR IGNORE INTO websub_subscriptions "
"(feed_url, hub_url, secret, subscribed_at, verified, "
"expires_at, lease_seconds, hub_tried_at) "
"VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
(
r["feed_url"],
r["hub_url"],
r["secret"],
r["subscribed_at"],
r["verified"],
r["expires_at"],
r["lease_seconds"],
r["hub_tried_at"],
),
)
wconn.execute(
"INSERT OR IGNORE INTO websub_subscribers (feed_url, user_id) VALUES (?, ?)",
(r["feed_url"], uid),
)
def ensure_thumb_schema() -> None:
with get_thumb_connection() as conn:
# WAL allows concurrent readers; thumb content is regeneratable so
# synchronous=NORMAL is the right tradeoff (durability not critical).
conn.execute("PRAGMA journal_mode=WAL")
conn.execute("PRAGMA synchronous=NORMAL")
conn.execute(
"""
CREATE TABLE IF NOT EXISTS thumb_cache (
cache_key TEXT PRIMARY KEY,
jpeg BLOB NOT NULL,
created_at REAL NOT NULL
)
"""
)
def get_img_cache_connection() -> sqlite3.Connection:
"""Connection to the global (shared) /api/img proxy cache."""
conn = sqlite3.connect(str(IMG_CACHE_DB_PATH))
conn.row_factory = sqlite3.Row
return conn
def ensure_img_cache_schema() -> None:
with get_img_cache_connection() as conn:
# WAL for concurrent readers; cached bytes are regeneratable so
# synchronous=NORMAL is the right durability/throughput tradeoff.
conn.execute("PRAGMA journal_mode=WAL")
conn.execute("PRAGMA synchronous=NORMAL")
conn.execute(
"""
CREATE TABLE IF NOT EXISTS img_cache (
cache_key TEXT PRIMARY KEY,
content_type TEXT NOT NULL,
body BLOB NOT NULL,
size INTEGER NOT NULL,
created_at REAL NOT NULL,
last_accessed REAL NOT NULL
)
"""
)
# Eviction scans by last_accessed; index keeps the daily sweep cheap.
conn.execute("CREATE INDEX IF NOT EXISTS idx_img_cache_last_accessed ON img_cache(last_accessed)")
def get_yt_duration_connection() -> sqlite3.Connection:
"""Connection to the global (shared) YouTube-duration cache."""
conn = sqlite3.connect(str(YT_DURATION_DB_PATH))
conn.row_factory = sqlite3.Row
return conn
def ensure_yt_duration_schema() -> None:
with get_yt_duration_connection() as conn:
conn.execute("PRAGMA journal_mode=WAL")
conn.execute("PRAGMA synchronous=NORMAL")
conn.execute(
"""
CREATE TABLE IF NOT EXISTS youtube_video_duration (
video_id TEXT PRIMARY KEY,
duration_seconds INTEGER,
duration_display TEXT,
fetched_at TEXT NOT NULL DEFAULT (datetime('now'))
)
"""
)
# Premiere/live status, fetched alongside duration from the same
# videos.list call (snippet.liveBroadcastContent, liveStreamingDetails.
# scheduledStartTime) — used to detect videos that haven't aired yet.
try:
conn.execute("ALTER TABLE youtube_video_duration ADD COLUMN live_broadcast_content TEXT")
except Exception:
pass
try:
conn.execute("ALTER TABLE youtube_video_duration ADD COLUMN scheduled_start_time TEXT")
except Exception:
pass
# Members-only status: NOT in the videos.list response the other columns
# ride along on for free — the Data API has no field for it, and the
# "members" endpoint only works for channels you own. Detected instead
# by fetching the video's own watch page and checking for its
# BADGE_STYLE_TYPE_MEMBERS_ONLY badge. NULL = never checked, so a check
# only costs a fetch once per video, ever (see fetch_and_cache_members_only).
try:
conn.execute("ALTER TABLE youtube_video_duration ADD COLUMN members_only INTEGER")
except Exception:
pass
def get_starred_archive_connection() -> sqlite3.Connection:
# Per-user DB (resolved via tenancy); returns a fresh connection per call.
conn = sqlite3.connect(str(tenancy.starred_archive_db_path()))
conn.row_factory = sqlite3.Row
return conn
@contextmanager
def archive_conn() -> Generator[sqlite3.Connection]:
"""An archive connection that commits (or rolls back) *and* closes.
``with sqlite3_connection:`` only wraps a transaction — it leaves the
handle open, and this factory returns a fresh one per call. Use this
wherever the archive DB is touched.
"""
conn = get_starred_archive_connection()
try:
with conn:
yield conn
finally:
conn.close()
def ensure_starred_archive_schema() -> None:
with archive_conn() as conn:
# Archive content is irreplaceable if the source goes down — leave
# synchronous at the default FULL for durability. WAL still helps with
# concurrent reads from the render path.
conn.execute("PRAGMA journal_mode=WAL")
conn.execute(
"""
CREATE TABLE IF NOT EXISTS archived_entry (
feed_url TEXT NOT NULL,
entry_id TEXT NOT NULL,
status TEXT NOT NULL,
starred_at REAL NOT NULL,
archived_at REAL,
error TEXT,
source_html_zlib BLOB,
readability_html_zlib BLOB,
content_html_zlib BLOB,
title TEXT,
link TEXT,
feed_title TEXT,
author TEXT,
published_at REAL,
received_at REAL,
PRIMARY KEY (feed_url, entry_id)
)
"""
)
# Migration: add columns to pre-existing tables created before the
# orphan-save support landed. PRAGMA table_info is the standard way
# to detect what's already there.
existing_cols = {row["name"] for row in conn.execute("PRAGMA table_info(archived_entry)").fetchall()}
for col_name, col_decl in (
("content_html_zlib", "BLOB"),
("title", "TEXT"),
("link", "TEXT"),
("feed_title", "TEXT"),
("author", "TEXT"),
("published_at", "REAL"),
("received_at", "REAL"),
# Maintained, not computed live -- written once at archive-completion
# time (which capture AND re-fetch both funnel through), not derived
# from LENGTH()+join on every render. NULL until the first successful
# archive. See Plan.md "Saved: see and sort by item size".
("content_size_bytes", "INTEGER"),
# How the source-page fetch went at capture time: ok, no_link, http_<code>, timeout, tls, connect, blocked,
# error. NULL for rows captured before this existed. A failed fetch still archives as 'complete' (the feed's
# own content may be all there is), so this is the only record of WHY a complete archive can be empty.
("source_fetch_status", "TEXT"),
# Nightly retry bookkeeping for transient fetch failures (see StarredArchiveService.requeue_transient_failures).
("capture_retries", "INTEGER NOT NULL DEFAULT 0"),
("capture_retried_at", "REAL"),
("capture_retry_pending", "INTEGER NOT NULL DEFAULT 0"),
):
if col_name not in existing_cols:
conn.execute(f"ALTER TABLE archived_entry ADD COLUMN {col_name} {col_decl}")
conn.execute("CREATE INDEX IF NOT EXISTS idx_archived_entry_status ON archived_entry (status)")
# Covers get_capture_failed_keys, so it never reads the capture blobs.
conn.execute(
"CREATE INDEX IF NOT EXISTS idx_archived_entry_capture"
" ON archived_entry (source_fetch_status, status, content_size_bytes, feed_url, entry_id)"
)
conn.execute(
"""
CREATE TABLE IF NOT EXISTS archived_asset (
asset_hash TEXT PRIMARY KEY,
data BLOB NOT NULL,
content_type TEXT NOT NULL,
width INTEGER,
height INTEGER,
byte_size INTEGER NOT NULL,
created_at REAL NOT NULL
)
"""
)
conn.execute(
"""
CREATE TABLE IF NOT EXISTS archived_asset_link (
feed_url TEXT NOT NULL,
entry_id TEXT NOT NULL,
source_url TEXT NOT NULL,
asset_hash TEXT NOT NULL,
PRIMARY KEY (feed_url, entry_id, source_url)
)
"""
)
conn.execute("CREATE INDEX IF NOT EXISTS idx_archived_asset_link_hash ON archived_asset_link (asset_hash)")
def ensure_reader_indexes() -> None:
"""Add Lectio-owned performance indexes to the current tenant's reader DB.
reader manages the `entries` table; we add a partial index it doesn't ship.
The per-feed unread-count query (`WHERE read=0 GROUP BY feed`) otherwise scans
the whole entries table (75k+ rows at scale) on each cache refresh, which under
refresh write-contention stretched to several seconds and made the UI sluggish.
A partial covering index over only the unread rows makes it a ~1k-row scan.
Idempotent and safe (an extra index reader never drops)."""
try:
conn = sqlite3.connect(str(tenancy.reader_db_path()), timeout=5)
try:
has_entries = conn.execute("SELECT 1 FROM sqlite_master WHERE type='table' AND name='entries'").fetchone()
if has_entries:
conn.execute("CREATE INDEX IF NOT EXISTS entries_unread_by_feed ON entries(feed) WHERE read=0")
conn.commit()
finally:
conn.close()
except Exception:
LOGGER.debug("ensure_reader_indexes failed", exc_info=True)
def provision_user_storage(user_id: str) -> None:
"""Create and schema-init a user's isolated databases (keyed by stable
user_id).
Routes through the tenancy seam: under ``user_context`` the schema-init
helpers and reader open resolve to this user's own DB files. Idempotent —
safe to call on an already-provisioned user.
"""
tenancy.ensure_user_data_dir(user_id)
with tenancy.user_context(user_id):
ensure_meta_schema()
ensure_starred_archive_schema()
# reader builds its own schema on first client(); opening provisions it.
with get_reader():
pass
ensure_reader_indexes()
def delete_user_storage(user_id: str) -> None:
"""Recursively remove a user's isolated data directory
(``DATA_DIR/users/<user_id>/`` and all its DBs).
Content-addressed caches (thumbnails, image proxy, lead-image/strategy
results) are global and hold no per-user data, so they are intentionally
left untouched.
"""
data_dir = tenancy.user_data_dir(user_id)
if data_dir.exists():
shutil.rmtree(data_dir, ignore_errors=True)
def _seed_admin_integrations_from_env(admin_id: str) -> None:
"""Copy env-provided per-user integration defaults into the admin's settings,
so a fresh multi-user instance has the admin configured (env is "used for
bootstrap then ignored" for these per-user values). Only seeds keys not
already set; the instance-shared Resend API key is NOT seeded here (it stays
env-read for everyone via get_resend_api_key)."""
seeds = [
(SETTING_EMAIL_FROM, _ENV_RESEND_FROM),
(EMAIL_TO_SETTING_KEY, _ENV_RESEND_TO),
# YouTube is per-user; the env vars seed the first (admin) user only.
(SETTING_YT_API_KEY, _ENV_YT_API_KEY),
(SETTING_YT_CHANNEL_ID, _ENV_YT_CHANNEL_ID),
(SETTING_YT_FOLDER_NAME, _ENV_YT_FOLDER_NAME),
]
with tenancy.user_context(admin_id):
with get_meta_connection() as conn:
for key, val in seeds:
if val and not get_setting(conn, key):
set_setting(conn, key, val)
def bootstrap_admin() -> None:
"""Seed the bootstrap admin on first startup when no users exist.
No-op once any user exists. Provisions the new admin's isolated storage
and warns loudly if the default password is still in use.
"""
if user_store.count() > 0:
return
username = BOOTSTRAP_ADMIN_USERNAME
if not tenancy.is_valid_user_id(username):
LOGGER.error(
"Cannot bootstrap admin: LECTIO_ADMIN_USERNAME=%r is not a valid username (must match {A-Za-z0-9_-}, 1-64 chars).",
username,
)
return
try:
admin_id = user_store.create(username, BOOTSTRAP_ADMIN_PASSWORD, is_admin=True, scheme=PASSWORD_HASH_SCHEME)
provision_user_storage(admin_id)
_seed_admin_integrations_from_env(admin_id)
except Exception:
LOGGER.exception("failed to bootstrap admin user %r", username)
return
if BOOTSTRAP_ADMIN_PASSWORD == _DEFAULT_ADMIN_PASSWORD:
LOGGER.warning(
"Bootstrapped admin %r with the DEFAULT password. Set LECTIO_ADMIN_PASSWORD "
"(and change the account password) before exposing this instance.",
username,
)
else:
LOGGER.info("Bootstrapped admin user %r.", username)
def _rebuild_email_batch_queue(conn: sqlite3.Connection) -> None:
"""Swap email_batch_queue to the wider UNIQUE key, keeping rows and ids (the caller owns the transaction)."""
conn.execute("ALTER TABLE email_batch_queue RENAME TO email_batch_queue_old")
conn.execute(
"""
CREATE TABLE email_batch_queue (
id INTEGER PRIMARY KEY AUTOINCREMENT,
rule_scope TEXT NOT NULL,
rule_scope_id TEXT NOT NULL,
rule_keyword TEXT NOT NULL,
queued_at TEXT NOT NULL,
feed_url TEXT NOT NULL,
entry_id TEXT NOT NULL,
title TEXT,
link TEXT,
feed_title TEXT,
excerpt TEXT,
email_to TEXT NOT NULL,
cc_me INTEGER NOT NULL DEFAULT 0,
rule_uid TEXT NOT NULL DEFAULT '',
UNIQUE(rule_scope, rule_scope_id, rule_keyword, rule_uid, feed_url, entry_id)
)
"""
)
conn.execute(
"INSERT INTO email_batch_queue (id, rule_scope, rule_scope_id, rule_keyword, rule_uid, queued_at, feed_url, entry_id,"
" title, link, feed_title, excerpt, email_to, cc_me)"
" SELECT id, rule_scope, rule_scope_id, rule_keyword, rule_uid, queued_at, feed_url, entry_id,"
" title, link, feed_title, excerpt, email_to, cc_me FROM email_batch_queue_old ORDER BY id"
)
conn.execute("DROP TABLE email_batch_queue_old")
def ensure_meta_schema() -> None:
with get_meta_connection() as conn:
conn.execute(
"""
CREATE TABLE IF NOT EXISTS folders (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
parent_id INTEGER,
UNIQUE(name, parent_id),
FOREIGN KEY(parent_id) REFERENCES folders(id)
)
"""
)
conn.execute(
"""
CREATE TABLE IF NOT EXISTS folder_feeds (
folder_id INTEGER NOT NULL,
feed_url TEXT NOT NULL,
PRIMARY KEY(folder_id, feed_url),
FOREIGN KEY(folder_id) REFERENCES folders(id)
)
"""
)
conn.execute(
"""
CREATE TABLE IF NOT EXISTS app_settings (
key TEXT PRIMARY KEY,
value TEXT NOT NULL
)
"""
)
conn.execute(
"""
CREATE TABLE IF NOT EXISTS saved_entries (
feed_url TEXT NOT NULL,
entry_id TEXT NOT NULL,
saved_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY(feed_url, entry_id)
)
"""
)
# A short-lived undo for /entries/saved's unstar (POST /entries/saved,
# saved=0) — the same shape as entry_unread_batch, one row per entry
# keyed by a shared timestamp token that IS the undo token, but this
# one also keeps the original saved_at so a restored star lands back
# in its actual star-order position instead of jumping to "just
# starred." Raised 2026-08-23: repeat-pressing the star-toggle key by
# accident unstarred ~16 articles with no way to identify which ones
# afterward. Deliberately scoped to this one route -- the bulk/
# administrative unstar paths (Archive, dedup/merge, unsubscribe) go
# through apply_star_state directly and don't write this table, since
# an undo toast doesn't make sense for a deliberate bulk action.
conn.execute(
"""
CREATE TABLE IF NOT EXISTS entry_unstar_batch (
feed_url TEXT NOT NULL,
entry_id TEXT NOT NULL,
unstarred_at TEXT NOT NULL,
saved_at TEXT,
PRIMARY KEY (feed_url, entry_id)
)
"""
)
# Mirrors entry_read_state in the other direction: a bulk mark-as-UNREAD
# stamps its whole batch with one timestamp so the toast can put exactly
# that batch back. entry_read_state cannot serve — its rows mean "read
# at", and this batch has just stopped being read (they are deleted).
conn.execute(
"""
CREATE TABLE IF NOT EXISTS entry_unread_batch (
feed_url TEXT NOT NULL,
entry_id TEXT NOT NULL,
unread_at TEXT NOT NULL,
PRIMARY KEY (feed_url, entry_id)
)
"""
)
# Feed-tag suggestion chips the user has dismissed, per (feed, tag).
#
# Manual rather than heuristic on purpose: two automatic rules were tried
# and both hid tags that were wanted. "VinylDeals" (a place) is noise while
# "Lessons" (a kind of content) is exactly the right filing tag, and nothing
# in the feed metadata distinguishes them — see get_feed_tag_suggestions.
# Scoped per feed because a tag useless on one feed can matter on another.
conn.execute(
"""
CREATE TABLE IF NOT EXISTS suppressed_feed_tags (
feed_url TEXT NOT NULL,
tag TEXT NOT NULL,
suppressed_at REAL NOT NULL,
PRIMARY KEY (feed_url, tag)
)
"""
)
# The opposite scope from suppressed_feed_tags above: a tag value that
# should never render as a suggestion chip on ANY feed (e.g. "comments"),
# not one dismissal per feed it happens to show up on. Filters the chip
# out of the suggestion UI itself -- explicitly not a rule that acts on
# entries carrying the tag, which is what tag_filter rules are for.
# Editable from Settings -> Tags.
conn.execute(
"""
CREATE TABLE IF NOT EXISTS suppressed_feed_tags_global (
tag TEXT PRIMARY KEY,
suppressed_at REAL NOT NULL
)
"""
)
# Extension chips dismissed from the "This feed links:" suggestion row. The scanner cannot tell a
# file type from a TLD lookalike — ".il" is Israel, but ".zip" and ".mov" are both real TLDs AND real
# file types, so no fixed blocklist gets this right. Let the user say so instead. Per feed, matching
# suppressed_feed_tags: a type that is noise on one feed can be the point on another.
conn.execute(
"""
CREATE TABLE IF NOT EXISTS suppressed_feed_attachment_exts (
feed_url TEXT NOT NULL,
ext TEXT NOT NULL,
suppressed_at REAL NOT NULL,
PRIMARY KEY (feed_url, ext)
)
"""
)
# Tag aliases: one spelling of a tag folded into another. Publishers
# disagree about the same subject — the live library carries cpp (204)
# beside c++ (181), csharp (149) beside c# (1), dotnet (108) beside
# .net (3) — so filtering on one silently misses the other half.
# Applied in normalize_tag_value, which every tag path already goes
# through, so one row covers manual tags, captured feed tags, filter
# rules and imports alike.
conn.execute(
"""
CREATE TABLE IF NOT EXISTS tag_aliases (
alias TEXT PRIMARY KEY,
canonical TEXT NOT NULL,
created_at REAL NOT NULL
)
"""
)
# Duplicate-scan groups the user has explicitly said aren't dupes
# (reported 2026-08-10: "Need a way to mark as Not Dupes, don't
# suggest"). Keyed by the exact set of feed URLs shown in the group
# at dismiss time -- if any of those feeds later drops out of the
# scan (unsubscribed, url changed), the dismissal naturally stops
# matching rather than silently hiding some other, unrelated group.
conn.execute(
"""
CREATE TABLE IF NOT EXISTS dedup_dismissed (
dismiss_key TEXT PRIMARY KEY,
dismissed_at TEXT NOT NULL
)
"""
)
# The feed-level equivalent of dedup_dismissed: a genuine user-initiated
# unsubscribe, so an Inoreader resync's subscriptions phase (which blindly
# re-adds anything Ino still lists as subscribed but reader is missing)
# can tell "deliberately removed" apart from "gone for some other reason"
# instead of resurrecting it. Written only from the actual unsubscribe
# paths -- not dedup/merge/format-upgrade (curation already migrates
# there) and not the YouTube-sync auto-removal callback (that one should
# stay free to re-add a channel the user re-subscribes to on YouTube).
conn.execute(
"""
CREATE TABLE IF NOT EXISTS declined_feeds (
feed_url TEXT PRIMARY KEY,
declined_at TEXT NOT NULL
)
"""
)
# The read-later "done" axis, deliberately its own table rather than a
# column on saved_entries. A saved_entries row *is* the star, and the
# star is the TODO axis: "I still have to decide what to do with this."
# Done is independent — Archive removes the star but the item stays
# archived, so the state cannot live on a row that archiving deletes.
# NOT the offline "starred archive" (archived_entry), which is content.
conn.execute(
"""
CREATE TABLE IF NOT EXISTS archived_entries (
feed_url TEXT NOT NULL,
entry_id TEXT NOT NULL,
archived_at TIMESTAMP NOT NULL,
PRIMARY KEY(feed_url, entry_id)
)
"""
)
conn.execute(
"""
CREATE TABLE IF NOT EXISTS deleted_entries (
feed_url TEXT NOT NULL,
entry_id TEXT NOT NULL,
deleted_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY(feed_url, entry_id)
)
"""
)
# Manual tags for orphan-archive entries — entries whose feed is no
# longer in `reader` (see _build_orphan_entry_detail), so there is no
# reader resource for reader's own entry_tags to attach to. Star
# already works for orphans (saved_entries is meta-DB, no reader
# dependency); this gives tags the same independence rather than
# silently no-oping when reader.get_entry() misses.
conn.execute(
"""
CREATE TABLE IF NOT EXISTS orphan_entry_tags (
feed_url TEXT NOT NULL,
entry_id TEXT NOT NULL,
tag TEXT NOT NULL,
PRIMARY KEY(feed_url, entry_id, tag)
)
"""
)
conn.execute(
"""
CREATE TABLE IF NOT EXISTS entry_date_overrides (
feed_url TEXT NOT NULL,
entry_id TEXT NOT NULL,
published TEXT NOT NULL,
source TEXT NOT NULL DEFAULT 'manual',
PRIMARY KEY(feed_url, entry_id)
)
"""
)
try:
# Existing rows predate this column and cannot be told apart after
# the fact, so they default to 'manual' — the safe direction, since
# scripts/backfill_url_inferred_dates.py's --refresh only recomputes
# rows marked 'inferred' (its own writes) and must never touch a
# real correction made through /entries/set-date.
conn.execute("ALTER TABLE entry_date_overrides ADD COLUMN source TEXT NOT NULL DEFAULT 'manual'")
except Exception:
pass # column already exists
conn.execute(
"""
CREATE TABLE IF NOT EXISTS entry_title_overrides (
feed_url TEXT NOT NULL,
entry_id TEXT NOT NULL,
title TEXT NOT NULL,
PRIMARY KEY(feed_url, entry_id)
)
"""
)
conn.execute(
"""
CREATE TABLE IF NOT EXISTS entry_link_overrides (
feed_url TEXT NOT NULL,
entry_id TEXT NOT NULL,
link TEXT NOT NULL,
PRIMARY KEY(feed_url, entry_id)
)
"""
)
# Re-fetched content for a STARRED feed entry (whose feed content lacked
# images or was truncated). Pinned here and reapplied after refresh, so
# the feed re-serving its own thinner content can't clobber the fuller
# copy the user deliberately pulled. Mirrors entry_link/title/date
# overrides. Content is reader's JSON content shape.
conn.execute(
"""
CREATE TABLE IF NOT EXISTS entry_content_overrides (
feed_url TEXT NOT NULL,
entry_id TEXT NOT NULL,
content TEXT NOT NULL,
PRIMARY KEY(feed_url, entry_id)
)
"""
)
# Aardvark-style per-entry cleanup: the pristine body (reader's JSON
# content shape) plus the ops that were replayed over it. The original
# is what "Revert cleanup" restores; the ops are the record of *what*
# was removed, which a later per-feed rule gets promoted from. Only the
# first edit writes original_content — repeated cleanups of the same
# entry must keep reverting to the true original, not to the last edit.
conn.execute(
"""
CREATE TABLE IF NOT EXISTS entry_content_edits (
feed_url TEXT NOT NULL,
entry_id TEXT NOT NULL,
original_content TEXT NOT NULL,
ops TEXT NOT NULL,
edited_at TEXT NOT NULL,
PRIMARY KEY(feed_url, entry_id)
)
"""
)
conn.execute(
"""
CREATE TABLE IF NOT EXISTS entry_read_state (
feed_url TEXT NOT NULL,
entry_id TEXT NOT NULL,
read_at TEXT NOT NULL,
PRIMARY KEY(feed_url, entry_id)
)
"""
)
# "Fix URLs" automation: per-feed host aliases. When an author moves
# domains without updating their feed's <guid>/<link> (they keep emitting
# the old host), every entry lands with an old-domain id/link. A rule
# here rewrites from_host -> to_host at ingest so entries arrive with the
# current domain. Per-feed by design: an author's dead domains are only
# rewritten inside their own feed, never a legit link elsewhere.
conn.execute(
"""
CREATE TABLE IF NOT EXISTS feed_url_rewrites (
feed_url TEXT NOT NULL,
from_host TEXT NOT NULL,
to_host TEXT NOT NULL,
PRIMARY KEY(feed_url, from_host)
)
"""
)
conn.execute(
"""
CREATE TABLE IF NOT EXISTS read_history (
id INTEGER PRIMARY KEY AUTOINCREMENT,
feed_url TEXT NOT NULL,
entry_id TEXT NOT NULL,
title TEXT,
link TEXT,
feed_title TEXT,
read_at TEXT NOT NULL,
UNIQUE(feed_url, entry_id)
)
"""
)
conn.execute("CREATE INDEX IF NOT EXISTS idx_read_history_read_at ON read_history(read_at DESC)")
conn.execute(
"""
CREATE TABLE IF NOT EXISTS youtube_video_duration (
video_id TEXT PRIMARY KEY,
duration_seconds INTEGER,
duration_display TEXT,
fetched_at TEXT NOT NULL DEFAULT (datetime('now'))
)
"""
)
conn.execute(
"""
CREATE TABLE IF NOT EXISTS feed_failure_state (
feed_url TEXT PRIMARY KEY,
consecutive_failures INTEGER NOT NULL DEFAULT 0,
next_retry_at REAL,
last_error TEXT,
last_failure_at REAL,
last_success_at REAL
)
"""
)
try:
conn.execute("ALTER TABLE feed_failure_state ADD COLUMN acknowledged_at REAL")
except Exception:
pass
try:
conn.execute("ALTER TABLE folders ADD COLUMN cadence_minutes INTEGER DEFAULT NULL")
except Exception:
pass
# Per-folder retention: delete read posts N days after they were read
# (NULL/0 = keep forever). Applied by the nightly maintenance prune.
try:
conn.execute("ALTER TABLE folders ADD COLUMN retention_days INTEGER DEFAULT NULL")
except Exception:
pass
# The entry ids each feed served in its most recent parse. Tomb
# sweeping refuses to drop tombstones still present here: a quiet
# YouTube channel serves the same 15 videos for years, and sweeping
# their tombstones would resurrect all of them on the channel's next
# upload (the reparse re-ingests the whole window).
conn.execute(
"""
CREATE TABLE IF NOT EXISTS feed_seen_window (
feed_url TEXT NOT NULL,
entry_id TEXT NOT NULL,
PRIMARY KEY (feed_url, entry_id)
)
"""
)
# Tombstone age for the nightly sweep. Pre-existing rows (and rows
# written by old code during a rolling deploy) are backfilled to "now"
# so they age out one sweep-window from the upgrade, not immediately.
try:
conn.execute("ALTER TABLE deleted_entries ADD COLUMN created_at TIMESTAMP DEFAULT NULL")
except Exception:
pass
try:
conn.execute(
"UPDATE deleted_entries SET created_at = ? WHERE created_at IS NULL",
(datetime.now().isoformat(),),
)
except Exception:
pass
# Read Mode "Archive" state: originally a per-saved-item flag (NULL =
# inbox). Superseded by the archived_entries table above — see its
# comment for why the done-axis cannot live on the star row. The column
# is still created (old rows carry it, and it is the migration source)
# but nothing reads it after the backfill below.
try:
conn.execute("ALTER TABLE saved_entries ADD COLUMN archived_at TIMESTAMP DEFAULT NULL")
except Exception:
pass
# One-time lift of the legacy column into the new table — and it must be
# ONE-TIME, not merely idempotent. The first version relied on INSERT OR
# IGNORE and left the column populated, so every boot re-lifted it:
# un-archiving deleted the archived_entries row, the old column survived,
# and the next restart brought the item back. Reported as "I've
# unarchived them 3 times now and they keep coming back" — a resurrection
# bug that is invisible until something restarts the app.
#
# Clearing the source in the same transaction is what makes it one-time,
# and it also removes the second source of truth. A marker flag would
# leave a populated column that still looks authoritative to the next
# person reading the schema.
try:
conn.execute(
"INSERT OR IGNORE INTO archived_entries (feed_url, entry_id, archived_at) "
"SELECT feed_url, entry_id, archived_at FROM saved_entries "
"WHERE archived_at IS NOT NULL"
)
conn.execute("UPDATE saved_entries SET archived_at = NULL WHERE archived_at IS NOT NULL")
conn.commit()
except Exception:
pass
conn.execute(
"""
CREATE TABLE IF NOT EXISTS entry_lead_images (
feed_url TEXT NOT NULL,
entry_id TEXT NOT NULL,
image_url TEXT,
fetched_at REAL NOT NULL,
PRIMARY KEY (feed_url, entry_id)
)
"""
)
conn.execute(
"""
CREATE TABLE IF NOT EXISTS entry_feed_tags (
feed_url TEXT NOT NULL,
entry_id TEXT NOT NULL,
tag TEXT NOT NULL,
first_seen_at REAL NOT NULL,
PRIMARY KEY (feed_url, entry_id, tag)
)
"""
)
# Where a tag came from: 'feed' (ingest) or 'page' (page topics, services/page_topics.py). The feed's per-entry replace only
# touches its own rows, so page topics survive the feed re-delivering an entry on every refresh.
if "source" not in {r[1] for r in conn.execute("PRAGMA table_info(entry_feed_tags)").fetchall()}:
conn.execute("ALTER TABLE entry_feed_tags ADD COLUMN source TEXT NOT NULL DEFAULT 'feed'")
conn.execute(
"""
CREATE TABLE IF NOT EXISTS domain_failure_state (
domain TEXT PRIMARY KEY,
consecutive_failures INTEGER NOT NULL DEFAULT 0,
next_retry_at REAL,
last_failure_at REAL
)
"""
)
conn.execute(
"""
CREATE TABLE IF NOT EXISTS feed_lead_image_strategy (
feed_url TEXT PRIMARY KEY,
strategy TEXT NOT NULL DEFAULT 'unknown',
detected_at REAL NOT NULL,
manual INTEGER NOT NULL DEFAULT 0
)
"""
)
try:
conn.execute("ALTER TABLE feed_lead_image_strategy ADD COLUMN manual INTEGER NOT NULL DEFAULT 0")
except Exception:
pass # column already exists
try:
conn.execute("ALTER TABLE entry_lead_images ADD COLUMN image_alt TEXT")
except Exception:
pass # column already exists
try:
conn.execute("ALTER TABLE entry_lead_images ADD COLUMN image_title TEXT")
except Exception:
pass
try:
conn.execute("ALTER TABLE entry_lead_images ADD COLUMN thumb_crop TEXT")
except Exception:
pass
# A webcomic page like cad-comic.com's can mark a strip as locked to
# paying supporters until a stated date — captured from the SAME
# source-page fetch already made for the lead image (see
# LeadImageService.check_and_cache_webcomic_lock), so hide_locked_comics
# costs nothing extra to detect. NULL/absent means "not locked" or
# "never checked"; a real value in the past means it has since unlocked,
# so no periodic re-check job is needed — the date does the expiring.
try:
conn.execute("ALTER TABLE entry_lead_images ADD COLUMN locked_until REAL")
except Exception:
pass
# The locked-comics lookup asks for every non-NULL row library-wide; a handful, but a full scan of ~150k rows (~55ms) without this.
conn.execute(
"CREATE INDEX IF NOT EXISTS idx_entry_lead_images_locked ON entry_lead_images(feed_url) WHERE locked_until IS NOT NULL"
)
try:
conn.execute("ALTER TABLE feed_strategy_cache ADD COLUMN image_alt TEXT")
except Exception:
pass
try:
conn.execute("ALTER TABLE feed_strategy_cache ADD COLUMN image_title TEXT")
except Exception:
pass
try:
conn.execute("ALTER TABLE feed_display_prefs ADD COLUMN caption_source TEXT")
except Exception:
pass
# "<param>=<value>", e.g. "w=1600". See upgrade_image_size_param.
try:
conn.execute("ALTER TABLE feed_display_prefs ADD COLUMN image_size_rule TEXT")
except Exception:
pass
conn.execute(
"""
CREATE TABLE IF NOT EXISTS disabled_feeds (
feed_url TEXT PRIMARY KEY,
disabled_at TEXT NOT NULL DEFAULT (datetime('now'))
)
"""
)
# Why a feed is disabled: '' = the user paused it; anything else is the system's own reason ("410 Gone", ...). Rows that
# predate the column are the user's, which is what the default says.
try:
conn.execute("ALTER TABLE disabled_feeds ADD COLUMN reason TEXT NOT NULL DEFAULT ''")
except Exception:
pass # column already exists
# Feeds unsubscribed-but-retained for their curation (starred/tagged
# entries). Kept feeds are hidden from the feed tree / All Feeds /
# Uncategorized and have updates disabled, but their reader feed and
# entries survive so the Saved/Kept view can still browse them per feed.
conn.execute(
"""
CREATE TABLE IF NOT EXISTS kept_feeds (
feed_url TEXT PRIMARY KEY,
kept_at TEXT NOT NULL DEFAULT (datetime('now'))
)
"""
)
# Hosts the user has marked as never having a feed, so the saved-article
# auto-filer stops proposing them. Their saves are genuine one-off
# read-later captures — a cheat sheet, a one-time tutorial — not unfiled
# feed articles, and without this they reappear in every scan pass and
# never resolve.
conn.execute(
"""
CREATE TABLE IF NOT EXISTS autofile_non_feed_hosts (
host TEXT PRIMARY KEY,
marked_at TEXT NOT NULL DEFAULT (datetime('now'))
)
"""
)
# Subscriptions that aren't really feeds — a single article URL that got
# added as one, a scraped page. They stay subscribed (their entries are
# real saved reading) but must never be offered as a destination for
# filing, or the site's whole backlog lands in a one-article stub.
conn.execute(
"""
CREATE TABLE IF NOT EXISTS non_feed_subscriptions (
feed_url TEXT PRIMARY KEY,
marked_at TEXT NOT NULL DEFAULT (datetime('now'))
)
"""
)
# Dead feeds the user has triaged as "needs a replacement": updates are
# disabled (we stop hitting them) and they move out of the Failing list
# into a Needs-replacement worklist, but the feed + its entries/curation
# are kept so a Change-URL swap can reattach a working source later.
conn.execute(
"""
CREATE TABLE IF NOT EXISTS feeds_needing_replacement (
feed_url TEXT PRIMARY KEY,
flagged_at TEXT NOT NULL DEFAULT (datetime('now'))
)
"""
)
conn.execute(
"""
CREATE TABLE IF NOT EXISTS browser_ua_feeds (
feed_url TEXT PRIMARY KEY,
flagged_at TEXT NOT NULL DEFAULT (datetime('now')),
reason TEXT NOT NULL DEFAULT ''
)
"""
)
# as_needed proxy escalation: a feed lands here only after browser-UA
# was already in play (flagged this cycle or earlier) and the fetch
# still failed — see _flag_proxy_feed_on_still_blocked. Mode-gated:
# only as_needed users ever get a feed flagged here (off never wants
# it, always doesn't need per-feed tracking).
conn.execute(
"""
CREATE TABLE IF NOT EXISTS proxy_feeds (
feed_url TEXT PRIMARY KEY,
flagged_at TEXT NOT NULL DEFAULT (datetime('now')),
reason TEXT NOT NULL DEFAULT ''
)
"""
)
# Last-resort escalation, one rung past proxy_feeds: a feed lands here
# only after the primary proxy was already in play and still failed —
# see _flag_tailscale_feed_on_still_blocked. Same mode-gating as
# proxy_feeds, plus requires SETTING_TAILSCALE_URL configured.
conn.execute(
"""
CREATE TABLE IF NOT EXISTS tailscale_feeds (
feed_url TEXT PRIMARY KEY,
flagged_at TEXT NOT NULL DEFAULT (datetime('now')),
reason TEXT NOT NULL DEFAULT ''
)
"""
)
# FlareSolverr escalation, between proxy_feeds and tailscale_feeds: a
# feed lands here only after the primary proxy was already in play and
# the fetch is STILL specifically a bot-challenge (not any refusal) —
# see _flag_flaresolverr_feed_on_still_blocked.
conn.execute(
"""
CREATE TABLE IF NOT EXISTS flaresolverr_feeds (
feed_url TEXT PRIMARY KEY,
flagged_at TEXT NOT NULL DEFAULT (datetime('now')),
reason TEXT NOT NULL DEFAULT ''
)
"""
)
conn.execute(
"""
CREATE TABLE IF NOT EXISTS feed_display_prefs (
feed_url TEXT PRIMARY KEY,
show_lead_image_in_article INTEGER NOT NULL DEFAULT 1,
show_lead_image_as_thumb INTEGER NOT NULL DEFAULT 1,
show_image_caption INTEGER NOT NULL DEFAULT -1,
caption_source TEXT,
inject_source_images INTEGER NOT NULL DEFAULT 0
)
"""
)
conn.execute(
"""
CREATE TABLE IF NOT EXISTS feed_strategy_cache (
feed_url TEXT NOT NULL,
strategy TEXT NOT NULL,
image_url TEXT,
fetched_at REAL NOT NULL,
error TEXT,
image_alt TEXT,
image_title TEXT,
PRIMARY KEY (feed_url, strategy)
)
"""
)
# Per-refresh fetch history — one row per non-skipped refresh attempt, so
# Feed Properties can show why a feed is stale or flagged problematic.
# Bounded by a per-feed cap + age prune in daily maintenance.
conn.execute(
"""
CREATE TABLE IF NOT EXISTS feed_fetch_history (
id INTEGER PRIMARY KEY AUTOINCREMENT,
feed_url TEXT NOT NULL,
fetched_at REAL NOT NULL,
status TEXT NOT NULL,
http_status INTEGER,
new_entries INTEGER,
duration_ms INTEGER,
error TEXT
)
"""
)
conn.execute("CREATE INDEX IF NOT EXISTS feed_fetch_history_by_feed ON feed_fetch_history (feed_url, fetched_at DESC)")
# Recovered media:content podcast audio (reader drops media:content, so we
# re-parse the raw feed on demand). entry_media_audio holds per-entry audio
# URLs found; feed_media_scan tracks when a feed was last scanned (and
# whether any audio was found) so we don't re-fetch non-podcast feeds.
conn.execute(
"""
CREATE TABLE IF NOT EXISTS entry_media_audio (
feed_url TEXT NOT NULL,
entry_id TEXT NOT NULL,
audio_url TEXT NOT NULL,
PRIMARY KEY (feed_url, entry_id)
)
"""
)
# Recovered YouTube embeds (feedparser strips the <iframe>, so a WP
# YouTube block reaches us as an empty figure). video_ids is a
# space-separated list in document order; '' means a YouTube embed was
# present but no id was recoverable (a cached negative). Populated by the
# same raw-feed re-parse as entry_media_audio.
conn.execute(
"""
CREATE TABLE IF NOT EXISTS entry_media_video (
feed_url TEXT NOT NULL,
entry_id TEXT NOT NULL,
video_ids TEXT NOT NULL DEFAULT '',
PRIMARY KEY (feed_url, entry_id)
)
"""
)
conn.execute(
"""
CREATE TABLE IF NOT EXISTS feed_media_scan (
feed_url TEXT PRIMARY KEY,
scanned_at REAL NOT NULL,
found INTEGER NOT NULL DEFAULT 0,
suggested_audio_feed TEXT NOT NULL DEFAULT '',
ok INTEGER NOT NULL DEFAULT 1
)
"""
)
# Added after feed_media_scan shipped: the podcast-host feed discovered
# for an audio-less feed (so we can suggest subscribing to the audio feed).
try:
conn.execute("ALTER TABLE feed_media_scan ADD COLUMN suggested_audio_feed TEXT NOT NULL DEFAULT ''")
except Exception:
pass
# Added later: distinguish a scan that completed cleanly with no audio
# (back off a long time) from one that errored mid-way — e.g. a host feed
# we discovered but couldn't fetch (Cloudflare 403) — so transient
# failures retry soon instead of inheriting the 7-day "empty" backoff.
try:
conn.execute("ALTER TABLE feed_media_scan ADD COLUMN ok INTEGER NOT NULL DEFAULT 1")
except Exception:
pass
conn.execute(
"""
CREATE TABLE IF NOT EXISTS highlight_keywords (
scope TEXT NOT NULL,
scope_id TEXT NOT NULL,
keyword TEXT NOT NULL,
color TEXT NOT NULL DEFAULT 'yellow',
PRIMARY KEY (scope, scope_id, keyword)
)
"""
)
try:
conn.execute("ALTER TABLE highlight_keywords ADD COLUMN is_regex INTEGER NOT NULL DEFAULT 0")
except Exception:
pass
try:
conn.execute("ALTER TABLE highlight_keywords ADD COLUMN enabled INTEGER NOT NULL DEFAULT 1")
except Exception:
pass
try:
conn.execute("ALTER TABLE highlight_keywords ADD COLUMN type TEXT NOT NULL DEFAULT 'highlight'")
except Exception:
pass
try:
conn.execute("ALTER TABLE highlight_keywords ADD COLUMN search_in TEXT NOT NULL DEFAULT 'title'")
except Exception:
pass
try:
conn.execute("ALTER TABLE highlight_keywords ADD COLUMN delivery TEXT NOT NULL DEFAULT 'immediately'")
except Exception:
pass
try:
conn.execute("ALTER TABLE highlight_keywords ADD COLUMN email_to TEXT NOT NULL DEFAULT ''")
except Exception:
pass
try:
conn.execute("ALTER TABLE highlight_keywords ADD COLUMN batch_time TEXT NOT NULL DEFAULT ''")
except Exception:
pass
try:
conn.execute("ALTER TABLE highlight_keywords ADD COLUMN batch_count INTEGER NOT NULL DEFAULT 0")
except Exception:
pass
try:
conn.execute("ALTER TABLE highlight_keywords ADD COLUMN cc_me INTEGER NOT NULL DEFAULT 0")
except Exception:
pass
try:
conn.execute("ALTER TABLE highlight_keywords ADD COLUMN dedup_window_hours INTEGER NOT NULL DEFAULT 24")
except Exception:
pass
try:
conn.execute("ALTER TABLE highlight_keywords ADD COLUMN exclude_scope_ids TEXT NOT NULL DEFAULT ''")
except Exception:
pass
try:
conn.execute("ALTER TABLE highlight_keywords ADD COLUMN dedup_fuzzy_pct INTEGER NOT NULL DEFAULT 80")
except Exception:
pass
try:
conn.execute("ALTER TABLE highlight_keywords ADD COLUMN dedup_min_title_words INTEGER NOT NULL DEFAULT 5")
except Exception:
pass
# The column shipped pre-filled with 4, so no rule holding 4 had it chosen
# by anyone. Raise those once — 5 drops two same-feed false collisions and
# loses no real duplicate. Flagged so a deliberate 4 set later survives.
try:
if get_setting(conn, "dedup_min_title_words_bumped_to_5") != "1":
conn.execute(
"UPDATE highlight_keywords SET dedup_min_title_words = 5 WHERE type = 'deduplicate' AND dedup_min_title_words = 4"
)
set_setting(conn, "dedup_min_title_words_bumped_to_5", "1")
except Exception:
pass
try:
conn.execute("ALTER TABLE highlight_keywords ADD COLUMN sort_order INTEGER NOT NULL DEFAULT 0")
conn.execute("UPDATE highlight_keywords SET sort_order = rowid WHERE sort_order = 0")
except Exception:
pass
try:
conn.execute("ALTER TABLE highlight_keywords ADD COLUMN webhook_url TEXT NOT NULL DEFAULT ''")
except Exception:
pass
try:
conn.execute("ALTER TABLE highlight_keywords ADD COLUMN webhook_format TEXT NOT NULL DEFAULT 'generic'")
except Exception:
pass
try:
conn.execute("ALTER TABLE highlight_keywords ADD COLUMN webhook_batch INTEGER NOT NULL DEFAULT 0")
except Exception:
pass
# youtube_playlist rule type: target playlist + behavior toggles.
try:
conn.execute("ALTER TABLE highlight_keywords ADD COLUMN yt_playlist_id TEXT NOT NULL DEFAULT ''")
except Exception:
pass
try:
conn.execute("ALTER TABLE highlight_keywords ADD COLUMN yt_playlist_title TEXT NOT NULL DEFAULT ''")
except Exception:
pass
try:
conn.execute("ALTER TABLE highlight_keywords ADD COLUMN yt_include_shorts INTEGER NOT NULL DEFAULT 0")
except Exception:
pass
try:
conn.execute("ALTER TABLE highlight_keywords ADD COLUMN yt_mark_read INTEGER NOT NULL DEFAULT 1")
except Exception:
pass
# Duration filter for youtube_playlist (minutes; 0 = no limit).
try:
conn.execute("ALTER TABLE highlight_keywords ADD COLUMN yt_min_minutes INTEGER NOT NULL DEFAULT 0")
except Exception:
pass
try:
conn.execute("ALTER TABLE highlight_keywords ADD COLUMN yt_max_minutes INTEGER NOT NULL DEFAULT 0")
except Exception:
pass
# A rule's identity is (scope, scope_id, keyword) everywhere in this file --
# fine for same-request lookups, but any table that needs to recognize "the
# same rule" *across time* breaks the moment scope/scope_id/keyword changes
# (editing a rule's feed list, say): edit_highlight_route does remove-old +
# insert-new on an identity change, so anything keyed on the old text is
# silently orphaned. rule_uid is a stable id that survives such edits --
# add_highlight_keyword carries the prior row's rule_uid forward instead of
# minting a new one. Existing rows get one backfilled below (once; empty
# rule_uid afterward means a row created before this migration ran and
# somehow missed the backfill, not "no rule_uid assigned yet").
try:
conn.execute("ALTER TABLE highlight_keywords ADD COLUMN rule_uid TEXT NOT NULL DEFAULT ''")
except Exception:
pass
for row in conn.execute("SELECT rowid FROM highlight_keywords WHERE rule_uid = ''").fetchall():
conn.execute(
"UPDATE highlight_keywords SET rule_uid = ? WHERE rowid = ?",
(secrets.token_hex(16), row["rowid"]),
)
# Optional display name, purely cosmetic (never matched against, never
# part of the (scope, scope_id, keyword) identity) -- for a MAR rule
# whose keyword has grown into a long regex, the rules list can show
# this instead, with the pattern still one hover away as a tooltip.
try:
conn.execute("ALTER TABLE highlight_keywords ADD COLUMN label TEXT NOT NULL DEFAULT ''")
except Exception:
pass
# Dedup guard for the youtube_playlist rule: playlistItems.insert is not
# idempotent, so record each (rule, entry) we've added to avoid re-adding the
# same video on a later refresh (the cutoff window alone can re-match).
# Keyed on (scope, scope_id, keyword) for historical reasons; that text
# identity breaks on a scope-changing edit (see rule_uid above), so new
# rows also carry rule_uid and get deduped through the partial unique index
# below instead once it's populated. Old rows keep rule_uid = '' rather
# than get backfilled: there's no reliable way to map them back to a
# *current* rule now that their scope text may no longer match anything.
conn.execute(
"""
CREATE TABLE IF NOT EXISTS youtube_playlist_added (
scope TEXT NOT NULL,
scope_id TEXT NOT NULL DEFAULT '',
keyword TEXT NOT NULL DEFAULT '',
entry_id TEXT NOT NULL,
video_id TEXT NOT NULL,
added_at TEXT NOT NULL,
PRIMARY KEY (scope, scope_id, keyword, entry_id, video_id)
)
"""
)
try:
conn.execute("ALTER TABLE youtube_playlist_added ADD COLUMN rule_uid TEXT NOT NULL DEFAULT ''")
except Exception:
pass
conn.execute(
"CREATE UNIQUE INDEX IF NOT EXISTS idx_ytpl_added_rule_uid"
" ON youtube_playlist_added (rule_uid, entry_id, video_id) WHERE rule_uid != ''"
)
# Per-user YouTube Data API quota spend, keyed by the Pacific calendar date
# (Google resets quota at midnight Pacific). The API exposes no remaining-quota
# read, so we estimate by summing each call's documented unit cost.
conn.execute(
"""
CREATE TABLE IF NOT EXISTS yt_quota_spend (
day TEXT PRIMARY KEY,
units INTEGER NOT NULL DEFAULT 0
)
"""
)
# Quire API call log — one row per billed call, used for the sliding-window
# rate meter (Quire limits per org by minute + hour, with no remaining-quota
# read). Rows older than an hour are pruned on each insert.
conn.execute(
"""
CREATE TABLE IF NOT EXISTS quire_call_log (
id INTEGER PRIMARY KEY AUTOINCREMENT,
ts INTEGER NOT NULL
)
"""
)
conn.execute("CREATE INDEX IF NOT EXISTS idx_quire_call_log_ts ON quire_call_log (ts)")
conn.execute(
"""
CREATE TABLE IF NOT EXISTS rule_run_log (
id INTEGER PRIMARY KEY AUTOINCREMENT,
run_at TEXT NOT NULL,
rule_type TEXT NOT NULL,
scope TEXT NOT NULL,
scope_id TEXT NOT NULL DEFAULT '',
keyword TEXT NOT NULL DEFAULT '',
entries_affected INTEGER NOT NULL DEFAULT 0,
trigger TEXT NOT NULL DEFAULT 'manual'
)
"""
)
conn.execute(
"""
CREATE TABLE IF NOT EXISTS rule_run_log_entries (
log_id INTEGER NOT NULL,
feed_url TEXT NOT NULL,
entry_id TEXT NOT NULL,
title TEXT,
link TEXT,
feed_title TEXT,
role TEXT NOT NULL DEFAULT 'marked'
)
"""
)
conn.execute("CREATE INDEX IF NOT EXISTS rule_run_log_entries_log_id ON rule_run_log_entries (log_id)")
try:
# 'marked' = entry was marked read; 'kept' = the surviving copy of a
# dedup group, logged so run history shows what each duplicate matched.
conn.execute("ALTER TABLE rule_run_log_entries ADD COLUMN role TEXT NOT NULL DEFAULT 'marked'")
except sqlite3.OperationalError as exc:
if "duplicate column" not in str(exc).lower():
raise
try:
# The kept copy a marked duplicate matched (its link), so dedup run
# history can pair each marked entry with its keeper for verification
# instead of listing all marked then all kept. NULL for non-dedup runs
# and for kept rows (which anchor their own group by their own link).
conn.execute("ALTER TABLE rule_run_log_entries ADD COLUMN matched_link TEXT")
except sqlite3.OperationalError as exc:
if "duplicate column" not in str(exc).lower():
raise
conn.execute(
"""
CREATE TABLE IF NOT EXISTS dedup_false_matches (
keep_link TEXT NOT NULL,
mark_link TEXT NOT NULL,
added_at TEXT NOT NULL,
PRIMARY KEY (keep_link, mark_link)
)
"""
)
conn.execute(
"""
CREATE TABLE IF NOT EXISTS email_contacts (
id INTEGER PRIMARY KEY AUTOINCREMENT,
label TEXT NOT NULL,
address TEXT NOT NULL UNIQUE
)
"""
)
conn.execute(
"""
CREATE TABLE IF NOT EXISTS email_batch_queue (
id INTEGER PRIMARY KEY AUTOINCREMENT,
rule_scope TEXT NOT NULL,
rule_scope_id TEXT NOT NULL,
rule_keyword TEXT NOT NULL,
queued_at TEXT NOT NULL,
feed_url TEXT NOT NULL,
entry_id TEXT NOT NULL,
title TEXT,
link TEXT,
feed_title TEXT,
excerpt TEXT,
email_to TEXT NOT NULL,
cc_me INTEGER NOT NULL DEFAULT 0,
rule_uid TEXT NOT NULL DEFAULT '',
UNIQUE(rule_scope, rule_scope_id, rule_keyword, rule_uid, feed_url, entry_id)
)
"""
)
# Same scope-text-identity fragility rule_uid was added for above
# (highlight_keywords / youtube_playlist_added): editing a batch email
# rule's scope/keyword while entries are queued splits its backlog
# across two text identities instead of following the same rule. New
# rows also carry rule_uid; old rows keep '' for the same reason
# youtube_playlist_added's do -- there is no reliable way to map them
# back to a *current* rule once their scope text may no longer match
# anything. Flush paths prefer rule_uid when present and fall back to
# the text tuple otherwise, same shape as the dedup index below it.
try:
conn.execute("ALTER TABLE email_batch_queue ADD COLUMN rule_uid TEXT NOT NULL DEFAULT ''")
except Exception:
pass
# The original UNIQUE key omitted feed_url and rule_uid, so two rules sharing scope and keyword text (different
# recipients), or one entry_id in two feeds, collided and the second INSERT OR IGNORE was dropped. A UNIQUE
# constraint can't be altered in place: rebuild the table when the stored DDL still has the old key.
_bq_ddl = conn.execute("SELECT sql FROM sqlite_master WHERE type='table' AND name='email_batch_queue'").fetchone()
if _bq_ddl and "rule_keyword, entry_id)" in (_bq_ddl[0] or ""):
# Python's sqlite3 doesn't open a transaction before DDL, so make one explicit: a failure mid-rebuild must roll back
# the rename and create too, or the queued rows are stranded in the _old table and detection never fires again.
if conn.in_transaction:
conn.commit()
conn.execute("BEGIN IMMEDIATE")
try:
_rebuild_email_batch_queue(conn)
except BaseException:
conn.rollback()
raise
conn.commit()
# Per-rule, per-entry claim for the webhook / Quire / immediate-email dispatchers: they select entries "added in the last
# 15 minutes", so two passes for one feed inside that window would otherwise fire a match twice. A row is inserted before
# the send and deleted if the send fails (batch-mode email claims with the queue insert, never released);
# rows past the window are pruned nightly.
conn.execute(
"""
CREATE TABLE IF NOT EXISTS automation_dispatched (
rule_key TEXT NOT NULL,
feed_url TEXT NOT NULL,
entry_id TEXT NOT NULL,
dispatched_at TEXT NOT NULL,
PRIMARY KEY (rule_key, feed_url, entry_id)
)
"""
)
conn.execute(
"""
CREATE TABLE IF NOT EXISTS scraped_feeds (
id TEXT PRIMARY KEY,
source_url TEXT NOT NULL,
mode TEXT NOT NULL,
selector TEXT,
feed_title TEXT NOT NULL,
created_at TEXT NOT NULL,
last_scraped_at TEXT,
last_content_hash TEXT
)
"""
)
# link_list only: a CSS selector applied to each NEW entry's own page
# (not the listing page `selector` already scopes) to fill its body.
# Readability/full-page guessing both fail on a chrome-heavy listing
# site the same way -- confirmed live 2026-09-19 on texasbluesalley.com,
# where the real content is a small, easily-selected region but the
# site's own nav/header markup is large enough to win readability's
# size-based scoring. See docs/architecture/feeds.md.
try:
conn.execute("ALTER TABLE scraped_feeds ADD COLUMN content_selector TEXT")
except Exception:
pass
conn.execute(
"""
CREATE TABLE IF NOT EXISTS scraped_entries (
id TEXT PRIMARY KEY,
scraped_feed_id TEXT NOT NULL REFERENCES scraped_feeds(id),
entry_url TEXT,
title TEXT NOT NULL,
content TEXT,
published_at TEXT NOT NULL,
hidden INTEGER NOT NULL DEFAULT 0,
UNIQUE(scraped_feed_id, entry_url)
)
"""
)
conn.execute(
"""
CREATE TABLE IF NOT EXISTS deviantart_feeds (
id TEXT PRIMARY KEY,
username TEXT NOT NULL,
feed_title TEXT NOT NULL,
created_at TEXT NOT NULL,
last_synced_at TEXT
)
"""
)
conn.execute(
"""
CREATE TABLE IF NOT EXISTS deviantart_entries (
id TEXT PRIMARY KEY,
deviantart_feed_id TEXT NOT NULL REFERENCES deviantart_feeds(id),
deviationid TEXT NOT NULL,
title TEXT NOT NULL,
entry_url TEXT,
content TEXT,
published_at TEXT NOT NULL,
UNIQUE(deviantart_feed_id, deviationid)
)
"""
)
# Watched artists whose DeviantArt account is deactivated ("Account is
# inactive."). Parked here so the watch-list sync stops retrying them on
# every run (they can never be added), and daily maintenance re-checks
# for reactivation.
conn.execute(
"""
CREATE TABLE IF NOT EXISTS deviantart_deactivated (
username TEXT PRIMARY KEY,
first_seen_at TEXT NOT NULL,
last_checked_at TEXT NOT NULL
)
"""
)
conn.execute(
"""
CREATE TABLE IF NOT EXISTS devto_feeds (
id TEXT PRIMARY KEY,
feed_title TEXT NOT NULL,
tag TEXT,
top_days INTEGER,
english_only INTEGER NOT NULL DEFAULT 1,
min_reactions INTEGER,
tags_exclude TEXT,
created_at TEXT NOT NULL,
last_synced_at TEXT
)
"""
)
conn.execute(
"""
CREATE TABLE IF NOT EXISTS devto_entries (
id TEXT PRIMARY KEY,
devto_feed_id TEXT NOT NULL REFERENCES devto_feeds(id),
article_id TEXT NOT NULL,
title TEXT NOT NULL,
entry_url TEXT,
content TEXT,
published_at TEXT NOT NULL,
UNIQUE(devto_feed_id, article_id)
)
"""
)
# Comma-joined tag slugs; the synthetic RSS is regenerated from these
# rows, so tags must persist here to come out as <category> elements.
try:
conn.execute("ALTER TABLE devto_entries ADD COLUMN tags TEXT NOT NULL DEFAULT ''")
except Exception:
pass
try:
conn.execute("ALTER TABLE deviantart_entries ADD COLUMN tags TEXT NOT NULL DEFAULT ''")
except Exception:
pass
# When the /deviation/metadata lookup ran for this entry (tags aren't in
# browse/gallery responses). Set even when the result is zero tags, so
# untagged deviations aren't re-looked-up on every refresh.
try:
conn.execute("ALTER TABLE deviantart_entries ADD COLUMN tags_fetched_at TEXT")
except Exception:
pass
try:
conn.execute("ALTER TABLE deviantart_entries ADD COLUMN author TEXT NOT NULL DEFAULT ''")
except Exception:
pass
# Set even when the metadata lookup came back with no author (mirrors
# tags_fetched_at above), so an entry the API never attributes isn't
# re-looked-up on every refresh forever.
try:
conn.execute("ALTER TABLE deviantart_entries ADD COLUMN author_fetched_at TEXT")
except Exception:
pass
try:
conn.execute("ALTER TABLE deviantart_feeds ADD COLUMN source TEXT NOT NULL DEFAULT 'gallery'")
except Exception:
pass
try:
conn.execute("ALTER TABLE feed_display_prefs ADD COLUMN hide_shorts INTEGER NOT NULL DEFAULT 0")
except Exception:
pass
try:
conn.execute("ALTER TABLE feed_display_prefs ADD COLUMN hide_unpremiered INTEGER NOT NULL DEFAULT 0")
except Exception:
pass
# Members-only videos: detected via the "Members only" badge on the
# video's own watch page (see youtube_video_duration.members_only) --
# unlike hide_shorts/hide_unpremiered this isn't in the Data API
# response already fetched for duration, so it costs a real per-video
# page fetch and is opt-in like hide_paywalled rather than free like
# its YouTube siblings.
try:
conn.execute("ALTER TABLE feed_display_prefs ADD COLUMN hide_members_only INTEGER NOT NULL DEFAULT 0")
except Exception:
pass
# Locked webcomic strips: a publisher (cad-comic.com and the like) can
# gate a strip behind a supporter tier until a stated date. Not
# YouTube-specific — see LeadImageService.check_and_cache_webcomic_lock.
try:
conn.execute("ALTER TABLE feed_display_prefs ADD COLUMN hide_locked_comics INTEGER NOT NULL DEFAULT 0")
except Exception:
pass
# Paywalled/subscriber-only posts: a Substack paid post ships a body that
# is nothing but a "Read more" link back to itself, so it can be spotted
# without any explicit marker (Substack provides none).
try:
conn.execute("ALTER TABLE feed_display_prefs ADD COLUMN hide_paywalled INTEGER NOT NULL DEFAULT 0")
except Exception:
pass
# Auto-mark posts read once older than N days (0 = no limit of its own); folders hold the same as NULL/N.
try:
conn.execute("ALTER TABLE feed_display_prefs ADD COLUMN auto_read_days INTEGER NOT NULL DEFAULT 0")
except Exception:
pass
try:
conn.execute("ALTER TABLE folders ADD COLUMN auto_read_days INTEGER DEFAULT NULL")
except Exception:
pass
# Tags this feed's posts usually want, pinned to the front of the
# suggestion chips. A feed with a stable subject (a guitar blog: #guitar,
# #bass) publishes no tags saying so, leaving the same words to be typed
# on every post. Stored space-separated, normalized on write.
try:
conn.execute("ALTER TABLE feed_display_prefs ADD COLUMN suggested_tags TEXT")
except Exception:
pass
# File extensions to keep alongside a post when it is saved — guitar-pro
# links .gp tabs and .pdf lyric sheets that vanish with the post. Stored
# space-separated and normalized; empty/absent means the feature is off
# for this feed.
try:
conn.execute("ALTER TABLE feed_display_prefs ADD COLUMN attachment_exts TEXT")
except Exception:
pass
try:
conn.execute("ALTER TABLE feed_display_prefs ADD COLUMN inject_source_images INTEGER NOT NULL DEFAULT 0")
except Exception:
pass
try:
conn.execute("ALTER TABLE feed_display_prefs ADD COLUMN feed_thumbnail_url TEXT")
except Exception:
pass
try:
conn.execute("ALTER TABLE feed_display_prefs ADD COLUMN thumb_crop TEXT NOT NULL DEFAULT 'cover'")
except Exception:
pass
# Migrate legacy 'left' value to 'cover-left'
conn.execute("UPDATE feed_display_prefs SET thumb_crop = 'cover-left' WHERE thumb_crop = 'left'")
# Normalize any unknown values to 'cover'
conn.execute(
"UPDATE feed_display_prefs SET thumb_crop = 'cover' WHERE thumb_crop NOT IN "
"('cover','cover-top-left','cover-top','cover-top-right','cover-left','cover-right',"
"'cover-bottom-left','cover-bottom','cover-bottom-right','contain','smart')"
)
try:
conn.execute("ALTER TABLE feed_display_prefs ADD COLUMN thumb_strategy TEXT")
except Exception:
pass
try:
# Per-feed SmartCrop min_scale (NULL = default 0.9); applies when thumb_crop='smart'.
conn.execute("ALTER TABLE feed_display_prefs ADD COLUMN smart_min_scale REAL")
except Exception:
pass
try:
# Per-feed Fill zoom multiplier (NULL = default 1.0); applies to cover crop modes.
conn.execute("ALTER TABLE feed_display_prefs ADD COLUMN fill_zoom REAL")
except Exception:
pass
try:
# Off by default: a bare $...$ pair is common in ordinary text
# ("between $50 and $100") and would render as broken math on any
# feed that isn't actually LaTeX-flavored -- see renderMathInEntryPane.
conn.execute("ALTER TABLE feed_display_prefs ADD COLUMN katex_dollar_math INTEGER NOT NULL DEFAULT 0")
except Exception:
pass
full_content_fetch.ensure_schema(conn)
page_topics.ensure_schema(conn)
conn.execute(
"""
CREATE TABLE IF NOT EXISTS websub_subscriptions (
feed_url TEXT PRIMARY KEY,
hub_url TEXT,
secret TEXT,
lease_seconds INTEGER DEFAULT 0,
subscribed_at REAL DEFAULT 0,
expires_at REAL DEFAULT 0,
verified INTEGER DEFAULT 0,
hub_tried_at REAL DEFAULT 0
)
"""
)
conn.execute(
"""
CREATE TABLE IF NOT EXISTS fever_feed_map (
id INTEGER PRIMARY KEY AUTOINCREMENT,
feed_url TEXT UNIQUE NOT NULL
)
"""
)
conn.execute(
"""
CREATE TABLE IF NOT EXISTS fever_group_map (
id INTEGER PRIMARY KEY AUTOINCREMENT,
title TEXT UNIQUE NOT NULL
)
"""
)
conn.execute(
"""
CREATE TABLE IF NOT EXISTS fever_entry_map (
id INTEGER PRIMARY KEY AUTOINCREMENT,
feed_url TEXT NOT NULL,
entry_id TEXT NOT NULL,
UNIQUE(feed_url, entry_id)
)
"""
)
conn.execute("CREATE INDEX IF NOT EXISTS idx_fever_entry_map_feed ON fever_entry_map(feed_url)")
conn.execute(
"""
CREATE TABLE IF NOT EXISTS miniflux_feed_map (
id INTEGER PRIMARY KEY AUTOINCREMENT,
feed_url TEXT UNIQUE NOT NULL
)
"""
)
conn.execute(
"""
CREATE TABLE IF NOT EXISTS miniflux_entry_map (
id INTEGER PRIMARY KEY AUTOINCREMENT,
feed_url TEXT NOT NULL,
entry_id TEXT NOT NULL,
UNIQUE(feed_url, entry_id)
)
"""
)
conn.execute("CREATE INDEX IF NOT EXISTS idx_miniflux_entry_map_feed ON miniflux_entry_map(feed_url)")
conn.execute(
"""
CREATE TABLE IF NOT EXISTS greader_tokens (
token TEXT PRIMARY KEY,
expires_at REAL NOT NULL
)
"""
)
root = conn.execute(
"SELECT id FROM folders WHERE name = ? AND parent_id IS NULL",
(ROOT_FOLDER_NAME,),
).fetchone()
if not root:
conn.execute(
"INSERT INTO folders (name, parent_id) VALUES (?, NULL)",
(ROOT_FOLDER_NAME,),
)
if DEBUG_MODE:
# Ensure the _Lectio system folder exists (used for dev feeds).
root_id = conn.execute("SELECT id FROM folders WHERE name = ? AND parent_id IS NULL", (ROOT_FOLDER_NAME,)).fetchone()
if root_id:
conn.execute(
"INSERT OR IGNORE INTO folders (name, parent_id) VALUES (?, ?)",
(_LECTIO_FOLDER_NAME, int(root_id["id"])),
)
conn.execute(
"INSERT OR IGNORE INTO app_settings (key, value) VALUES (?, ?)",
(AUTO_REFRESH_SETTING_KEY, str(get_instance_default_auto_refresh())),
)
conn.execute(
"INSERT OR IGNORE INTO app_settings (key, value) VALUES (?, ?)",
(SORT_BY_SETTING_KEY, DEFAULT_SORT_BY),
)
conn.execute(
"INSERT OR IGNORE INTO app_settings (key, value) VALUES (?, ?)",
(SORT_DIR_SETTING_KEY, DEFAULT_SORT_DIR),
)
conn.execute(
"INSERT OR IGNORE INTO app_settings (key, value) VALUES (?, ?)",
(GLOBAL_NOTE_SETTING_KEY, ""),
)
pass # email_to seeding removed; Contacts tab manages recipients
def _load_app_settings_cache(conn: sqlite3.Connection) -> dict[str, str]:
rows = conn.execute("SELECT key, value FROM app_settings").fetchall()
return {str(r["key"]): str(r["value"]) for r in rows}
def get_setting(conn: sqlite3.Connection, key: str) -> str | None:
"""Read a setting from the current user's in-memory cache. Loaded once per
user on first access, kept consistent through set_setting writes."""
uid = tenancy.current_user_id()
with _app_settings_cache_lock:
cache = _app_settings_cache.get(uid)
if cache is None:
cache = _load_app_settings_cache(conn)
_app_settings_cache[uid] = cache
return cache.get(key)
def set_setting(conn: sqlite3.Connection, key: str, value: str) -> None:
"""Idempotent setting write. Updates the current user's in-memory cache and
persists only when the value actually changed."""
uid = tenancy.current_user_id()
with _app_settings_cache_lock:
cache = _app_settings_cache.get(uid)
if cache is None:
cache = _load_app_settings_cache(conn)
_app_settings_cache[uid] = cache
if cache.get(key) == value:
return
cache[key] = value
conn.execute(
"INSERT OR REPLACE INTO app_settings (key, value) VALUES (?, ?)",
(key, value),
)
def get_cached_setting(key: str) -> str | None:
"""Read the current user's cache without a DB connection. None if unloaded or no user is bound
(pre-auth pages, bare background threads), so callers fall through to their env/admin fallback."""
uid = tenancy.bound_user_id()
if uid is None:
return None
with _app_settings_cache_lock:
cache = _app_settings_cache.get(uid)
if cache is None:
return None
return cache.get(key)
def get_runtime_setting(key: str, env_fallback: str = "") -> str:
"""DB setting takes precedence; env_fallback (typically os.getenv) is the fallback."""
val = get_cached_setting(key)
if val is not None:
return val
return env_fallback
def delete_setting(conn: sqlite3.Connection, key: str) -> None:
uid = tenancy.current_user_id()
with _app_settings_cache_lock:
cache = _app_settings_cache.get(uid)
if cache is not None:
cache.pop(key, None)
conn.execute("DELETE FROM app_settings WHERE key = ?", (key,))
_DISPLAY_PREF_KEYS = frozenset(
{
"show_lead_image_in_article",
"show_lead_image_as_thumb",
"show_image_caption",
"hide_shorts",
"hide_unpremiered",
"hide_members_only",
"hide_paywalled",
"auto_read_days",
"hide_locked_comics",
"inject_source_images",
"katex_dollar_math",
"fetch_full_content",
"capture_page_topics",
}
)
# Pre-built UPDATE statements (one per column) so conn.execute() never receives an f-string.
_DISPLAY_PREF_COLS: dict[str, str] = {k: k for k in _DISPLAY_PREF_KEYS}
_DISPLAY_PREF_SQLS: dict[str, str] = {k: f"UPDATE feed_display_prefs SET {k} = ? WHERE feed_url = ?" for k in _DISPLAY_PREF_KEYS}
_DISPLAY_PREF_DEFAULTS: dict = {
"show_lead_image_in_article": 1,
"show_lead_image_as_thumb": 1,
"show_image_caption": -1,
"hide_shorts": 0,
"hide_unpremiered": 0,
"hide_members_only": 0,
"hide_paywalled": 0,
"auto_read_days": 0,
"hide_locked_comics": 0,
"inject_source_images": 0,
"katex_dollar_math": 0,
"fetch_full_content": full_content_fetch.FEED_INHERIT,
"capture_page_topics": full_content_fetch.FEED_INHERIT,
"feed_thumbnail_url": None,
"thumb_crop": "cover",
"thumb_strategy": None,
"smart_min_scale": None,
"fill_zoom": None,
}
_VALID_THUMB_CROPS = frozenset(
{
"cover",
"cover-top-left",
"cover-top",
"cover-top-right",
"cover-left",
"cover-right",
"cover-bottom-left",
"cover-bottom",
"cover-bottom-right",
"contain",
"smart",
}
)
_VALID_THUMB_STRATEGIES = frozenset({"inline", "media_rss"})
# Caption text that is purely a date (e.g. "June 12, 2026" or "6/12/2026") — strip it.
_DATE_ONLY_CAP_RE = re.compile(
r"^\s*(?:\w+ \d{1,2},?\s*\d{4}|\d{1,2}[/.\-]\d{1,2}[/.\-]\d{2,4}|\d{4}[/.\-]\d{1,2}[/.\-]\d{1,2})\s*$",
re.IGNORECASE,
)
def get_feed_display_prefs(conn: sqlite3.Connection, feed_url: str) -> dict:
row = conn.execute("SELECT * FROM feed_display_prefs WHERE feed_url = ?", (feed_url,)).fetchone()
if row:
return dict(row)
return {"feed_url": feed_url, **_DISPLAY_PREF_DEFAULTS}
def get_all_feed_display_prefs(conn: sqlite3.Connection) -> dict[str, dict]:
rows = conn.execute("SELECT * FROM feed_display_prefs").fetchall()
return {row["feed_url"]: dict(row) for row in rows}
def upsert_feed_display_pref(conn: sqlite3.Connection, feed_url: str, key: str, value: int) -> None:
col = _DISPLAY_PREF_COLS.get(key)
if col is None:
raise ValueError(f"Unknown display pref key: {key}")
conn.execute(
"INSERT INTO feed_display_prefs (feed_url) VALUES (?) ON CONFLICT(feed_url) DO NOTHING",
(feed_url,),
)
conn.execute(_DISPLAY_PREF_SQLS[col], (value, feed_url))
def upsert_feed_thumbnail_url(conn: sqlite3.Connection, feed_url: str, thumbnail_url: str | None) -> None:
conn.execute(
"INSERT INTO feed_display_prefs (feed_url) VALUES (?) ON CONFLICT(feed_url) DO NOTHING",
(feed_url,),
)
conn.execute(
"UPDATE feed_display_prefs SET feed_thumbnail_url = ? WHERE feed_url = ?",
(thumbnail_url or None, feed_url),
)
def upsert_feed_smart_min_scale(conn: sqlite3.Connection, feed_url: str, min_scale: float | None) -> None:
value = min(1.0, max(0.5, min_scale)) if min_scale is not None else None
conn.execute(
"INSERT INTO feed_display_prefs (feed_url) VALUES (?) ON CONFLICT(feed_url) DO NOTHING",
(feed_url,),
)
conn.execute(
"UPDATE feed_display_prefs SET smart_min_scale = ? WHERE feed_url = ?",
(value, feed_url),
)
def upsert_feed_fill_zoom(conn: sqlite3.Connection, feed_url: str, zoom: float | None) -> None:
value = min(2.0, max(0.5, zoom)) if zoom is not None else None
conn.execute(
"INSERT INTO feed_display_prefs (feed_url) VALUES (?) ON CONFLICT(feed_url) DO NOTHING",
(feed_url,),
)
conn.execute(
"UPDATE feed_display_prefs SET fill_zoom = ? WHERE feed_url = ?",
(value, feed_url),
)
def upsert_feed_thumb_strategy(conn: sqlite3.Connection, feed_url: str, strategy: str | None) -> None:
s = strategy if strategy in _VALID_THUMB_STRATEGIES else None
conn.execute(
"INSERT INTO feed_display_prefs (feed_url) VALUES (?) ON CONFLICT(feed_url) DO NOTHING",
(feed_url,),
)
conn.execute(
"UPDATE feed_display_prefs SET thumb_strategy = ? WHERE feed_url = ?",
(s, feed_url),
)
_HIGHLIGHT_VALID_COLORS = frozenset({"yellow", "green", "blue", "pink", "orange"})
_HIGHLIGHT_VALID_SCOPES = frozenset({"global", "folder", "feed", "feeds", "folders"})
_HIGHLIGHT_VALID_TYPES = {
"highlight",
"mark_as_read",
"email_article",
"deduplicate",
"webhook",
"youtube_playlist",
"instapaper",
"quire",
"tag_filter",
"save_article",
}
_HIGHLIGHT_VALID_SEARCH_IN = {"title", "body", "both"}
_HIGHLIGHT_VALID_DELIVERY = {"immediately", "batch"}
_DEDUP_VALID_MATCH_METHODS = {"slug", "title", "both", "fuzzy", "safe"}
_DEDUP_FUZZY_PCT_DEFAULT = 80
# Dedup matching lives in services/dedup.py; these names are re-exported because route modules, automation_rules and tests import
# them from here.
from services.dedup import ( # noqa: E402
_DEDUP_MIN_TITLE_WORDS,
_safe_dedup_entry_slug,
entry_url_slug,
normalize_entry_title_for_dedupe,
)
from services.dedup import _cleanup_intra_feed_slug_dupes as _cleanup_intra_feed_dupes # noqa: E402
def _clamp_min_title_words(words: int | None) -> int:
"""Title-length floor, clamped. Under 3 words a title is a category label."""
try:
v = int(words if words is not None else _DEDUP_MIN_TITLE_WORDS)
except TypeError, ValueError:
v = _DEDUP_MIN_TITLE_WORDS
return max(3, min(10, v))
def _clamp_fuzzy_pct(pct: int | None) -> int:
"""Percent knob, clamped. Below 50% a fuzzy title match catches near-anything."""
try:
v = int(pct if pct is not None else _DEDUP_FUZZY_PCT_DEFAULT)
except TypeError, ValueError:
v = _DEDUP_FUZZY_PCT_DEFAULT
return max(50, min(100, v))
def _dedup_fuzzy_threshold(pct: int | None) -> float:
"""Percent knob -> similarity ratio."""
return _clamp_fuzzy_pct(pct) / 100.0
def get_highlight_keywords(conn: sqlite3.Connection) -> list[dict]:
rows = conn.execute(
"SELECT scope, scope_id, keyword, color, is_regex, enabled, type, search_in, delivery,"
" email_to, batch_time, batch_count, cc_me, dedup_window_hours, dedup_fuzzy_pct,"
" dedup_min_title_words,"
" exclude_scope_ids, sort_order,"
" webhook_url, webhook_format, webhook_batch,"
" yt_playlist_id, yt_playlist_title, yt_include_shorts, yt_mark_read,"
" yt_min_minutes, yt_max_minutes, rule_uid, label"
" FROM highlight_keywords ORDER BY sort_order ASC, rowid ASC"
).fetchall()
return [dict(r) for r in rows]
# Types whose `keyword` column is a plain search term (comma-list = OR of
# terms, already how a single rule's multi-keyword list works) rather than
# something else entirely -- deduplicate's is a match-method enum,
# youtube_playlist/instapaper/quire/save_article's is optional and its
# semantics for the merge/action types aren't confirmed. Scoped to exactly
# the two types the 2026-08-19 measurement found real mergeable groups in --
# see Plan.md "Utilities: find rules that could be one rule". tag_filter is
# mergeable too, but by spec union (_union_tag_filter_specs), not a keyword join.
_MERGEABLE_RULE_TYPES = frozenset({"highlight", "mark_as_read", "tag_filter"})
# Fields (besides scope/scope_id/type/search_in/is_regex/keyword) that change
# a rule's *behavior*, not just its keyword list. Two rules that differ here
# are NOT silently collapsed into one — see find_mergeable_rule_groups.
_MERGE_IDENTITY_FIELDS = ("color", "delivery", "email_to", "batch_time", "batch_count", "cc_me")
def find_mergeable_rule_groups(conn: sqlite3.Connection) -> tuple[list[dict], list[dict]]:
"""Groups of 2+ rules that share type/scope/scope_id/search_in/is_regex
AND every other behavior-affecting field but differ only in keyword --
collapsible into one rule with a joined keyword list. A keyword was one
term until comma lists landed, so these accumulated from a time when one
rule per term was the only way to add a second.
Two rules can share type/scope/search_in/is_regex and still carry a
*different* color/delivery/email setting and mean different things, so
each identity key is further split into settings-consistent subgroups
(2026-08-31) -- a folder with 2 orange rules, 2 blue rules and 1 green
rule offers the orange pair and the blue pair as two independent
mergeable groups, rather than reporting the whole five as one
"these differ, edit to match" blob the way a single identity-wide check
used to. The lone green rule has no partner either way and is silently
excluded, same as a genuinely solo identity group always has been.
``mismatched`` now means the narrower thing that name always implied: 2+
*leftover* singletons under the same identity that still disagree with
each other once every already-agreeing pair has been pulled out.
"""
rows = conn.execute(
"SELECT scope, scope_id, keyword, color, is_regex, enabled, type, search_in, delivery,"
" email_to, batch_time, batch_count, cc_me, sort_order"
" FROM highlight_keywords WHERE type IN ('highlight', 'mark_as_read')"
# Disabled tag filters are chip-tuning drafts (see toggle_feed_tag_filter); merging one would arm it.
" OR (type = 'tag_filter' AND enabled = 1)"
" ORDER BY sort_order ASC, rowid ASC"
).fetchall()
by_identity: dict[tuple, list[dict]] = {}
for r in rows:
key = (r["type"], r["scope"], r["scope_id"], r["search_in"], bool(r["is_regex"]))
by_identity.setdefault(key, []).append(dict(r))
groups = []
mismatched = []
for key, group_rows in by_identity.items():
if len(group_rows) < 2:
continue
rule_type, scope, scope_id, search_in, is_regex = key
by_settings: dict[tuple, list[dict]] = {}
for r in group_rows:
by_settings.setdefault(tuple(r[f] for f in _MERGE_IDENTITY_FIELDS), []).append(r)
leftover: list[dict] = []
for settings_rows in by_settings.values():
if len(settings_rows) >= 2:
group = {
"type": rule_type,
"scope": scope,
"scope_id": scope_id,
"search_in": search_in,
"is_regex": is_regex,
"rules": settings_rows,
}
if rule_type == "tag_filter":
group["behavior_change"] = _tag_filter_merge_changes_behavior([r["keyword"] for r in settings_rows])
groups.append(group)
else:
leftover.extend(settings_rows)
if len(leftover) >= 2:
mismatched.append(
{
"type": rule_type,
"scope": scope,
"scope_id": scope_id,
"search_in": search_in,
"is_regex": is_regex,
"rules": leftover,
}
)
return groups, mismatched
def merge_highlight_rule_group(
conn: sqlite3.Connection,
rule_type: str,
scope: str,
scope_id: str,
search_in: str,
is_regex: bool,
color: str,
delivery: str,
email_to: str,
batch_time: str,
batch_count: int,
cc_me: bool,
) -> dict | None:
"""Collapse every rule sharing (type, scope, scope_id, search_in, is_regex)
AND the given color/delivery/email/batch/cc_me settings into one, joining
their keywords (comma list, or regex alternation when is_regex).
Re-derives the current matching rows from these criteria rather than
trusting a caller-supplied row list, so a stale preview can't merge the
wrong rows.
The settings are part of the match itself now (2026-08-31), not a
same-group precondition checked after the fact -- an identity key can
hold several settings-distinct subgroups (see find_mergeable_rule_groups),
and a merge must act on exactly the subgroup that was offered, not
whichever settings happen to be most common among rows sharing identity.
Refuses (returns None) when fewer than 2 rows match.
"""
if rule_type not in _MERGEABLE_RULE_TYPES:
return None
# Same draft exclusion as find_mergeable_rule_groups: a disabled tag filter is never swept into a merge.
enabled_clause = " AND enabled = 1" if rule_type == "tag_filter" else ""
rows = conn.execute(
"SELECT rowid, * FROM highlight_keywords"
" WHERE type = ? AND scope = ? AND scope_id = ? AND search_in = ? AND is_regex = ?"
" AND color = ? AND delivery = ? AND email_to = ? AND batch_time = ? AND batch_count = ? AND cc_me = ?"
+ enabled_clause
+ " ORDER BY sort_order ASC, rowid ASC",
(
rule_type,
scope,
scope_id,
search_in,
1 if is_regex else 0,
color,
delivery,
email_to,
batch_time,
batch_count,
1 if cc_me else 0,
),
).fetchall()
if len(rows) < 2:
return None
rows = [dict(r) for r in rows]
template = rows[0]
seen_kw: list[str] = []
for r in rows:
kw = str(r["keyword"] or "").strip()
if kw and kw not in seen_kw:
seen_kw.append(kw)
if rule_type == "tag_filter":
merged_keyword = _union_tag_filter_specs(seen_kw)
elif is_regex:
merged_keyword = "|".join(f"({kw})" for kw in seen_kw)
else:
# Flatten to the individual-keyword level too, so merging a rule that
# is already itself a comma list doesn't produce a repeated term.
flat: list[str] = []
seen_lower: set[str] = set()
for kw in seen_kw:
for part in kw.split(","):
part = part.strip()
if part and part.lower() not in seen_lower:
seen_lower.add(part.lower())
flat.append(part)
merged_keyword = ", ".join(flat)
min_sort_order = min(int(r["sort_order"] or 0) for r in rows)
conn.execute(
"DELETE FROM highlight_keywords"
" WHERE type = ? AND scope = ? AND scope_id = ? AND search_in = ? AND is_regex = ?"
" AND color = ? AND delivery = ? AND email_to = ? AND batch_time = ? AND batch_count = ? AND cc_me = ?" + enabled_clause,
(
rule_type,
scope,
scope_id,
search_in,
1 if is_regex else 0,
color,
delivery,
email_to,
batch_time,
batch_count,
1 if cc_me else 0,
),
)
add_highlight_keyword(
conn,
scope,
scope_id,
merged_keyword,
template["color"],
is_regex,
rule_type,
search_in,
template["delivery"],
template["email_to"],
template["batch_time"],
template["batch_count"],
bool(template["cc_me"]),
template["enabled"],
template["dedup_window_hours"],
template["exclude_scope_ids"],
template["dedup_fuzzy_pct"],
template["dedup_min_title_words"],
template["webhook_url"],
template["webhook_format"],
bool(template["webhook_batch"]),
template["yt_playlist_id"],
template["yt_playlist_title"],
bool(template["yt_include_shorts"]),
bool(template["yt_mark_read"]),
template["yt_min_minutes"],
template["yt_max_minutes"],
label=str(template.get("label") or ""),
)
conn.execute(
"UPDATE highlight_keywords SET sort_order = ?"
" WHERE type = ? AND scope = ? AND scope_id = ? AND search_in = ? AND is_regex = ? AND keyword = ?",
(min_sort_order, rule_type, scope, scope_id, search_in, 1 if is_regex else 0, merged_keyword),
)
return {
"type": rule_type,
"scope": scope,
"scope_id": scope_id,
"search_in": search_in,
"is_regex": is_regex,
"keyword": merged_keyword,
"merged_count": len(rows),
}
def find_regex_convertible_rule_groups(conn: sqlite3.Connection) -> list[dict]:
"""Groups of 2+ rules sharing (type, scope, scope_id, search_in) that span
BOTH a regex and a plain-text rule -- a same-is_regex group is already
handled by find_mergeable_rule_groups, so this only surfaces the boundary
case that one can't see. A plain-text keyword is always representable as
an escaped regex (re.escape), so nothing here is actually blocked by
is_regex disagreeing -- only by shown deliberately as its own suggestion
tier since it changes a rule's is_regex, not just its keyword. Raised
2026-08-31: "Lowe's" (plain) sat next to a regex rule on the same folder
with nothing offering to combine them.
Same behavior-identity guard as find_mergeable_rule_groups -- a group
whose rules carry different color/delivery/email settings is excluded
entirely, not silently merged with one side's settings winning."""
rows = conn.execute(
"SELECT scope, scope_id, keyword, color, is_regex, enabled, type, search_in, delivery,"
" email_to, batch_time, batch_count, cc_me, sort_order"
" FROM highlight_keywords WHERE type IN ('highlight', 'mark_as_read')"
# Disabled tag filters are chip-tuning drafts (see toggle_feed_tag_filter); merging one would arm it.
" OR (type = 'tag_filter' AND enabled = 1)"
" ORDER BY sort_order ASC, rowid ASC"
).fetchall()
by_identity: dict[tuple, list[dict]] = {}
for r in rows:
key = (r["type"], r["scope"], r["scope_id"], r["search_in"])
by_identity.setdefault(key, []).append(dict(r))
groups = []
for key, group_rows in by_identity.items():
if len(group_rows) < 2:
continue
if len({bool(r["is_regex"]) for r in group_rows}) < 2:
continue # all-regex or all-plain -- find_mergeable_rule_groups' territory
if len({tuple(r[f] for f in _MERGE_IDENTITY_FIELDS) for r in group_rows}) > 1:
continue
rule_type, scope, scope_id, search_in = key
groups.append(
{
"type": rule_type,
"scope": scope,
"scope_id": scope_id,
"search_in": search_in,
"rules": group_rows,
}
)
return groups
def merge_regex_convertible_rule_group(
conn: sqlite3.Connection,
rule_type: str,
scope: str,
scope_id: str,
search_in: str,
) -> dict | None:
"""Like merge_highlight_rule_group, but across the is_regex boundary: a
plain-text keyword is escaped (re.escape) before joining, so the result
matches exactly the same literal text the plain rule did -- see
find_regex_convertible_rule_groups. Re-derives the current matching rows
rather than trusting a caller-supplied list, same reasoning as the
regex-only merge. Refuses (returns None) when fewer than 2 rows match, the
group turns out to be single-is_regex after all (stale suggestion), or
they carry different color/delivery/email settings."""
if rule_type not in _MERGEABLE_RULE_TYPES:
return None
rows = conn.execute(
"SELECT rowid, * FROM highlight_keywords"
" WHERE type = ? AND scope = ? AND scope_id = ? AND search_in = ?"
" ORDER BY sort_order ASC, rowid ASC",
(rule_type, scope, scope_id, search_in),
).fetchall()
if len(rows) < 2:
return None
rows = [dict(r) for r in rows]
if len({bool(r["is_regex"]) for r in rows}) < 2:
return None
if len({tuple(r[f] for f in _MERGE_IDENTITY_FIELDS) for r in rows}) > 1:
return None
template = rows[0]
seen_kw: list[str] = []
for r in rows:
kw = str(r["keyword"] or "").strip()
if not kw:
continue
# A plain keyword's commas are OR-term separators (split_keyword_terms),
# not literal text -- re.escape(kw) on the whole string turned "Pixel
# Watch, Ryobi, Google Pixel" into one dead literal requiring all three
# names adjacent with commas, instead of three alternatives. Escape each
# term individually, then alternate them, before this whole row becomes
# one outer (...) group below.
pattern = kw if r["is_regex"] else "|".join(re.escape(term) for term in split_keyword_terms(kw))
if pattern not in seen_kw:
seen_kw.append(pattern)
merged_keyword = "|".join(f"({kw})" for kw in seen_kw)
min_sort_order = min(int(r["sort_order"] or 0) for r in rows)
conn.execute(
"DELETE FROM highlight_keywords WHERE type = ? AND scope = ? AND scope_id = ? AND search_in = ?",
(rule_type, scope, scope_id, search_in),
)
add_highlight_keyword(
conn,
scope,
scope_id,
merged_keyword,
template["color"],
True,
rule_type,
search_in,
template["delivery"],
template["email_to"],
template["batch_time"],
template["batch_count"],
bool(template["cc_me"]),
template["enabled"],
template["dedup_window_hours"],
template["exclude_scope_ids"],
template["dedup_fuzzy_pct"],
template["dedup_min_title_words"],
template["webhook_url"],
template["webhook_format"],
bool(template["webhook_batch"]),
template["yt_playlist_id"],
template["yt_playlist_title"],
bool(template["yt_include_shorts"]),
bool(template["yt_mark_read"]),
template["yt_min_minutes"],
template["yt_max_minutes"],
label=str(template.get("label") or ""),
)
conn.execute(
"UPDATE highlight_keywords SET sort_order = ?"
" WHERE type = ? AND scope = ? AND scope_id = ? AND search_in = ? AND is_regex = 1 AND keyword = ?",
(min_sort_order, rule_type, scope, scope_id, search_in, merged_keyword),
)
return {
"type": rule_type,
"scope": scope,
"scope_id": scope_id,
"search_in": search_in,
"is_regex": True,
"keyword": merged_keyword,
"merged_count": len(rows),
}
def _split_plain_keywords(raw: str) -> set[str]:
return {kw.strip().lower() for kw in (raw or "").split(",") if kw.strip()}
def find_redundant_feed_rules(conn: sqlite3.Connection) -> list[dict]:
"""Plain keyword feed rules covered by a folder rule, plus tag_filter feed rules that do nothing (see the two helpers)."""
return _find_redundant_keyword_feed_rules(conn) + _find_redundant_tag_filter_feed_rules(conn)
def _find_redundant_tag_filter_feed_rules(conn: sqlite3.Connection) -> list[dict]:
"""Enabled tag_filter feed rules that change nothing: ``no_op`` when the spec has no ``-``/``++`` tag at all (a ``+`` only
rescues from its own rule's drops, never another rule's), ``covered`` when every ``-`` tag is already dropped by an enabled
folder tag_filter rule on a folder holding the feed that rescues no more than the feed rule does. ``++`` feed rules are never
flagged: a whitelist's cut isn't reproduced by anything else."""
feed_rows = conn.execute(
"SELECT scope_id AS feed_url, keyword FROM highlight_keywords WHERE type = 'tag_filter' AND scope = 'feed' AND enabled = 1"
).fetchall()
if not feed_rows:
return []
folder_specs: dict[int, list[tuple[set[str], set[str]]]] = {} # folder_id -> [(exclude, saving)]
for r in conn.execute(
"SELECT scope_id, keyword FROM highlight_keywords WHERE type = 'tag_filter' AND scope = 'folder' AND enabled = 1"
).fetchall():
try:
fid = int(r["scope_id"])
except TypeError, ValueError:
continue
req, good, exc = parse_tag_filter_spec(str(r["keyword"] or ""))
folder_specs.setdefault(fid, []).append((exc, req | good))
redundant = []
for r in feed_rows:
feed_url = str(r["feed_url"])
req, good, exc = parse_tag_filter_spec(str(r["keyword"] or ""))
if req:
continue
base = {"feed_url": feed_url, "keyword": r["keyword"], "type": "tag_filter", "search_in": "title"}
if not exc:
redundant.append({**base, "reason": "no_op", "covering_folder_id": None})
continue
feed_saving = good
for (fid,) in conn.execute("SELECT folder_id FROM folder_feeds WHERE feed_url = ?", (feed_url,)).fetchall():
covered = set()
for f_exc, f_saving in folder_specs.get(int(fid), []):
if f_saving <= feed_saving:
covered |= f_exc
if exc <= covered:
redundant.append({**base, "reason": "covered", "covering_folder_id": int(fid)})
break
return redundant
def _find_redundant_keyword_feed_rules(conn: sqlite3.Connection) -> list[dict]:
"""Feed-scoped rules already fully covered by a same-type, same-search_in
folder rule on a folder the feed belongs to -- the folder rule already
catches everything the feed rule would, so the feed rule does nothing.
Plain (non-regex) rules only: whether one regex's language is a subset of
another's isn't decidable this way, so a regex rule is never flagged.
"""
feed_rows = conn.execute(
"SELECT scope_id AS feed_url, keyword, type, search_in FROM highlight_keywords"
" WHERE scope = 'feed' AND is_regex = 0 AND type IN ('highlight', 'mark_as_read')"
).fetchall()
if not feed_rows:
return []
folder_rows = conn.execute(
"SELECT scope_id AS folder_id, keyword, type, search_in FROM highlight_keywords"
" WHERE scope = 'folder' AND is_regex = 0 AND type IN ('highlight', 'mark_as_read')"
).fetchall()
if not folder_rows:
return []
folder_keywords: dict[tuple, set[str]] = {}
for r in folder_rows:
key = (r["type"], r["search_in"], str(r["folder_id"]))
folder_keywords.setdefault(key, set()).update(_split_plain_keywords(r["keyword"]))
feed_urls = {str(r["feed_url"]) for r in feed_rows}
placeholders = ",".join("?" for _ in feed_urls)
folder_membership = conn.execute(
f"SELECT feed_url, folder_id FROM folder_feeds WHERE feed_url IN ({placeholders})",
list(feed_urls),
).fetchall()
feed_folder_ids: dict[str, set[int]] = {}
for r in folder_membership:
feed_folder_ids.setdefault(str(r["feed_url"]), set()).add(int(r["folder_id"]))
redundant = []
for r in feed_rows:
feed_url = str(r["feed_url"])
feed_kws = _split_plain_keywords(r["keyword"])
if not feed_kws:
continue
for folder_id in feed_folder_ids.get(feed_url, set()):
covering = folder_keywords.get((r["type"], r["search_in"], str(folder_id)))
if covering and feed_kws.issubset(covering):
redundant.append(
{
"feed_url": feed_url,
"keyword": r["keyword"],
"type": r["type"],
"search_in": r["search_in"],
"covering_folder_id": folder_id,
}
)
break
return redundant
# On a cross-rule sign conflict the union keeps the strongest: a drop outranks a whitelist, which outranks a rescue.
_TAG_FILTER_SIGN_RANK = {"-": 2, "++": 1, "+": 0}
def _union_tag_filter_specs(specs: list[str]) -> str:
"""One tag_filter spec holding every signed tag in *specs*, first-seen order, deduped by tag."""
by_tag: dict[str, str] = {}
for spec in specs:
for token in spec.split(","):
token = token.strip()
sign = next((p for p in ("++", "+", "-") if token.startswith(p)), "")
tag = normalize_tag_value(token[len(sign) :])
sign = sign or "+" # bare = good, as in parse_tag_filter_spec
if tag and (tag not in by_tag or _TAG_FILTER_SIGN_RANK[sign] > _TAG_FILTER_SIGN_RANK[by_tag[tag]]):
by_tag[tag] = sign
return ", ".join(f"{sign}{tag}" for tag, sign in by_tag.items())
def _tag_filter_merge_changes_behavior(specs: list[str]) -> bool:
"""Whether merging these tag_filter rules into one changes what they mark read.
A ``+``/``++`` tag only acts on its own rule's ``-`` tags, so separate rules are exactly their union only when they all share the
same ``+`` and ``++`` sets (e.g. all pure ``-`` rules). Otherwise a merge lets one rule's rescue reach another's drops."""
parsed = [parse_tag_filter_spec(s) for s in specs]
return any((req, good) != (parsed[0][0], parsed[0][1]) for req, good, _exc in parsed[1:])
def _merge_tag_filter_specs(conn: sqlite3.Connection, feed_url: str, incoming: str) -> str | None:
"""Fold *incoming* into the feed's existing tag_filter spec, and drop the
now-redundant rows. Returns the merged spec, or None when there is nothing to
merge with.
Signed tokens are deduped by TAG, with the incoming sign winning: re-adding
``+rust`` where ``-rust`` was set is a correction, not a contradiction to
preserve.
"""
rows = conn.execute(
"SELECT rowid, keyword FROM highlight_keywords WHERE type = 'tag_filter' AND scope = 'feed' AND scope_id = ? ORDER BY rowid",
(feed_url,),
).fetchall()
if not rows:
return None
by_tag: dict[str, str] = {} # tag -> sign, insertion-ordered
for source in [str(r["keyword"] or "") for r in rows] + [incoming]:
for token in source.split(","):
token = token.strip()
if not token:
continue
sign, tag = "", token
for prefix in ("++", "+", "-"):
if token.startswith(prefix):
sign, tag = prefix, token[len(prefix) :]
break
tag = normalize_tag_value(tag)
if tag:
by_tag[tag] = sign
# Delete ALL of them, not just the extras: the caller's INSERT OR REPLACE keys
# on the keyword, which the merge has just changed, so leaving the first row
# behind would leave a rival rather than replace it.
conn.executemany(
"DELETE FROM highlight_keywords WHERE rowid = ?",
[(r["rowid"],) for r in rows],
)
if len(rows) > 1:
LOGGER.info("[tag-filter] merged %d duplicate rule(s) for %s", len(rows) - 1, feed_url)
return ", ".join(f"{sign}{tag}" for tag, sign in by_tag.items())
def add_highlight_keyword(
conn: sqlite3.Connection,
scope: str,
scope_id: str,
keyword: str,
color: str,
is_regex: bool = False,
rule_type: str = "highlight",
search_in: str = "title",
delivery: str = "immediately",
email_to: str = "",
batch_time: str = "",
batch_count: int = 0,
cc_me: bool = False,
enabled: int = 0,
dedup_window_hours: int = 168,
exclude_scope_ids: str = "",
dedup_fuzzy_pct: int = _DEDUP_FUZZY_PCT_DEFAULT,
dedup_min_title_words: int = _DEDUP_MIN_TITLE_WORDS,
webhook_url: str = "",
webhook_format: str = "generic",
webhook_batch: bool = False,
yt_playlist_id: str = "",
yt_playlist_title: str = "",
yt_include_shorts: bool = False,
yt_mark_read: bool = True,
yt_min_minutes: int = 0,
yt_max_minutes: int = 0,
rule_uid: str = "",
label: str = "",
) -> None:
if scope not in _HIGHLIGHT_VALID_SCOPES:
raise ValueError(f"Invalid scope: {scope}")
if color not in _HIGHLIGHT_VALID_COLORS:
color = "yellow"
if rule_type not in _HIGHLIGHT_VALID_TYPES:
rule_type = "highlight"
if search_in not in _HIGHLIGHT_VALID_SEARCH_IN:
search_in = "title"
if delivery not in _HIGHLIGHT_VALID_DELIVERY:
delivery = "immediately"
if webhook_format not in WEBHOOK_VALID_FORMATS:
webhook_format = "generic"
# A feed can only have ONE tag_filter rule: get_feed_tag_filter_rule fetches a
# single row, so a second one is edited by nothing while still executing — two
# rules on the same dev.to feed ("-rust, -powerbi" and "-rust, -hindi,
# -javascript, -powerbi, -aws") were both running, invisibly. Merge into the
# existing rule rather than adding a rival.
if rule_type == "tag_filter" and scope == "feed":
merged = _merge_tag_filter_specs(conn, scope_id, keyword)
if merged is not None:
keyword = merged
conn.execute(
"INSERT OR REPLACE INTO highlight_keywords"
" (scope, scope_id, keyword, color, is_regex, enabled, type, search_in, delivery,"
" email_to, batch_time, batch_count, cc_me, dedup_window_hours, exclude_scope_ids,"
" dedup_fuzzy_pct, dedup_min_title_words,"
" webhook_url, webhook_format, webhook_batch,"
" yt_playlist_id, yt_playlist_title, yt_include_shorts, yt_mark_read,"
" yt_min_minutes, yt_max_minutes, rule_uid, label)"
" VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
(
scope,
scope_id,
keyword.strip(),
color,
1 if is_regex else 0,
1 if enabled else 0,
rule_type,
search_in,
delivery,
email_to.strip(),
batch_time.strip(),
max(0, int(batch_count or 0)),
1 if cc_me else 0,
max(1, int(dedup_window_hours or 168)),
exclude_scope_ids.strip(),
_clamp_fuzzy_pct(dedup_fuzzy_pct),
_clamp_min_title_words(dedup_min_title_words),
webhook_url.strip(),
webhook_format,
1 if webhook_batch else 0,
yt_playlist_id.strip(),
yt_playlist_title.strip(),
1 if yt_include_shorts else 0,
1 if yt_mark_read else 0,
max(0, int(yt_min_minutes or 0)),
max(0, int(yt_max_minutes or 0)),
rule_uid.strip() or secrets.token_hex(16),
label.strip(),
),
)
def get_email_contacts(conn: sqlite3.Connection) -> list[dict]:
rows = conn.execute("SELECT id, label, address FROM email_contacts ORDER BY label").fetchall()
return [dict(r) for r in rows]
def remove_highlight_keyword(conn: sqlite3.Connection, scope: str, scope_id: str, keyword: str) -> None:
conn.execute(
"DELETE FROM highlight_keywords WHERE scope = ? AND scope_id = ? AND keyword = ?",
(scope, scope_id, keyword.strip()),
)
_TRIVIAL_CAPTIONS = frozenset(
{
"responsive image",
"image",
"photo",
"picture",
"img",
"thumbnail",
"banner",
"featured image",
"header image",
}
)
_FILENAME_CAPTION_RE = re.compile(
r"(?:\.(jpe?g|png|gif|webp|avif|svg|bmp|tiff?)$|^(?:DSC|IMG|dsc|img)[_-]?\d{3,})",
re.IGNORECASE,
)
def should_show_caption(caption: str | None, *, entry_title: str | None, content_html: str | None, pref: int) -> bool:
"""Return whether the image caption should be rendered.
pref: -1 = auto-suppress heuristics, 0 = never, 1 = always.
"""
if not caption:
return False
if pref == 0:
return False
if pref == 1:
return True
# Auto heuristics
stripped = caption.strip()
if stripped.lower() in _TRIVIAL_CAPTIONS:
return False
if _FILENAME_CAPTION_RE.search(stripped):
return False
if entry_title and stripped.lower() == entry_title.strip().lower():
return False
if content_html and stripped in content_html:
# Cheap check above can false-positive on attribute text: a webcomic's own
# <img alt="hovertext" title="hovertext"> makes the hovertext a raw substring
# of content_html even though it is invisible without a hover — not "already
# shown" the way a rendered <figcaption> is. Confirm against visible text only.
from bs4 import BeautifulSoup
if stripped in BeautifulSoup(content_html, "html.parser").get_text():
return False
return True
def normalize_auto_refresh_minutes(value: int) -> int:
if value <= 0:
return 0
return max(value, MIN_AUTO_REFRESH_MINUTES)
def parse_epoch_setting(value: str | None) -> float | None:
if value is None:
return None
try:
parsed = float(value)
except TypeError, ValueError:
return None
if parsed <= 0:
return None
return parsed
def get_auto_refresh_minutes(conn: sqlite3.Connection) -> int:
raw_value = get_setting(conn, AUTO_REFRESH_SETTING_KEY)
if raw_value is None:
return get_instance_default_auto_refresh()
try:
return normalize_auto_refresh_minutes(int(raw_value))
except ValueError:
return get_instance_default_auto_refresh()
def purge_lower_level_folders(conn: sqlite3.Connection) -> None:
root_id = get_root_folder_id(conn)
rows = conn.execute(
"""
WITH RECURSIVE tree(id, parent_id, depth, top_level_id) AS (
SELECT id, parent_id, 0 AS depth, id AS top_level_id
FROM folders
WHERE id = ?
UNION ALL
SELECT f.id,
f.parent_id,
t.depth + 1,
CASE WHEN t.depth = 0 THEN f.id ELSE t.top_level_id END AS top_level_id
FROM folders f
JOIN tree t ON f.parent_id = t.id
)
SELECT id, depth, top_level_id
FROM tree
WHERE depth >= 2
""",
(root_id,),
).fetchall()
nested_folder_ids: list[int] = []
for row in rows:
folder_id = int(row["id"])
top_level_id = int(row["top_level_id"])
nested_folder_ids.append(folder_id)
conn.execute(
"""
INSERT OR IGNORE INTO folder_feeds (folder_id, feed_url)
SELECT ?, feed_url
FROM folder_feeds
WHERE folder_id = ?
""",
(top_level_id, folder_id),
)
if not nested_folder_ids:
return
placeholders = ",".join("?" for _ in nested_folder_ids)
conn.execute(
f"DELETE FROM folder_feeds WHERE folder_id IN ({placeholders})",
nested_folder_ids,
)
conn.execute(
f"DELETE FROM folders WHERE id IN ({placeholders})",
nested_folder_ids,
)
def get_root_folder_id(conn: sqlite3.Connection) -> int:
row = conn.execute(
"SELECT id FROM folders WHERE name = ? AND parent_id IS NULL",
(ROOT_FOLDER_NAME,),
).fetchone()
if not row:
raise RuntimeError("Root folder is missing.")
return int(row["id"])
def _get_or_create_folder_by_name(conn: sqlite3.Connection, name: str) -> int:
"""Return the id of a top-level folder with `name`, creating it under root if absent."""
row = conn.execute("SELECT id FROM folders WHERE name = ? LIMIT 1", (name,)).fetchone()
if row:
return int(row["id"])
root_id = get_root_folder_id(conn)
conn.execute("INSERT INTO folders (name, parent_id) VALUES (?, ?)", (name, root_id))
return int(conn.execute("SELECT last_insert_rowid()").fetchone()[0])
def _get_lectio_folder_id(conn: sqlite3.Connection) -> int | None:
root_id = get_root_folder_id(conn)
row = conn.execute(
"SELECT id FROM folders WHERE name = ? AND parent_id = ?",
(_LECTIO_FOLDER_NAME, root_id),
).fetchone()
return int(row["id"]) if row else None
def get_descendant_folder_ids(conn: sqlite3.Connection, folder_id: int) -> list[int]:
rows = conn.execute(
"""
WITH RECURSIVE descendants(id) AS (
SELECT id FROM folders WHERE id = ?
UNION ALL
SELECT f.id
FROM folders f
JOIN descendants d ON f.parent_id = d.id
)
SELECT id FROM descendants
""",
(folder_id,),
).fetchall()
return [int(row["id"]) for row in rows]
def get_folder_rows(conn: sqlite3.Connection) -> list[sqlite3.Row]:
debug_clause = "" if DEBUG_MODE else " AND f.name != :exclude"
params: dict = {"root": ROOT_FOLDER_NAME}
if not DEBUG_MODE:
params["exclude"] = _LECTIO_FOLDER_NAME
return conn.execute(
f"""
SELECT
f.id,
f.name,
f.cadence_minutes,
CASE WHEN f.parent_id IS NULL THEN 0 ELSE 1 END AS depth,
CASE WHEN f.parent_id IS NULL THEN f.name ELSE root.name || ' / ' || f.name END AS path,
(
SELECT COUNT(*)
FROM folder_feeds ff
WHERE ff.folder_id = f.id
) AS feed_count
FROM folders f
LEFT JOIN folders root ON f.parent_id = root.id
WHERE (f.parent_id IS NULL OR f.parent_id = (
SELECT id FROM folders WHERE name = :root AND parent_id IS NULL
)){debug_clause}
ORDER BY path
""",
params,
).fetchall()
def get_folder_options(conn: sqlite3.Connection) -> list[FolderOption]:
debug_clause = "" if DEBUG_MODE else " AND f.name != :exclude"
params: dict = {"root": ROOT_FOLDER_NAME}
if not DEBUG_MODE:
params["exclude"] = _LECTIO_FOLDER_NAME
rows = conn.execute(
f"""
SELECT
f.id,
f.name,
CASE WHEN f.parent_id IS NULL THEN 0 ELSE 1 END AS depth,
CASE WHEN f.parent_id IS NULL THEN f.name ELSE root.name || ' / ' || f.name END AS path
FROM folders f
LEFT JOIN folders root ON f.parent_id = root.id
WHERE (f.parent_id IS NULL OR f.parent_id = (
SELECT id FROM folders WHERE name = :root AND parent_id IS NULL
)){debug_clause}
ORDER BY path
""",
params,
).fetchall()
return [FolderOption(id=int(r["id"]), name=r["name"], path=r["path"], depth=int(r["depth"])) for r in rows]
def get_direct_feed_urls_by_folder(conn: sqlite3.Connection) -> dict[int, list[str]]:
rows = conn.execute("SELECT folder_id, feed_url FROM folder_feeds ORDER BY feed_url").fetchall()
by_folder: dict[int, list[str]] = {}
for row in rows:
folder_id = int(row["folder_id"])
by_folder.setdefault(folder_id, []).append(str(row["feed_url"]))
return by_folder
def get_folder_feed_urls(conn: sqlite3.Connection, folder_id: int) -> set[str]:
# The virtual "Uncategorized" folder has no folder_feeds rows; its members are
# every reader feed not in any folder. Resolving it here lets all folder
# actions (mark-read, refresh, …) operate on it uniformly. The Saved
# Articles virtual feed is excluded here too — it must never be
# actionable (mark-read/refresh/moved) as if it were an ordinary
# unfoldered feed; the Saved sidebar's own reachability need is served
# elsewhere (_home_inner's snapshot dict), not through this resolver.
if folder_id == UNCATEGORIZED_FOLDER_ID:
return get_all_reader_feed_urls() - get_all_feed_urls(conn) - {saved_articles_service.SAVED_FEED_URL}
# "All Feeds" (root) covers every reader feed, including orphans that live in
# no folder. The home view widens root the same way for display and counts, so
# folder actions (mark-read/older, refresh) must match — otherwise entries in
# uncategorized feeds show under All Feeds but can't be marked from there.
# The Saved Articles virtual feed (lectio:saved) is excluded here too: unlike
# Uncategorized (which keeps it reachable for the Saved sidebar's own
# Uncategorized grouping — see the home route), root has no such need, and
# it must never show as a subscription — or be actionable — in Feeds mode.
if folder_id == get_root_folder_id(conn):
return get_all_reader_feed_urls() - {saved_articles_service.SAVED_FEED_URL}
descendant_ids = get_descendant_folder_ids(conn, folder_id)
placeholders = ",".join("?" for _ in descendant_ids)
rows = conn.execute(
f"SELECT DISTINCT feed_url FROM folder_feeds WHERE folder_id IN ({placeholders})",
descendant_ids,
).fetchall()
return {str(r["feed_url"]) for r in rows}
def get_all_feed_urls(conn: sqlite3.Connection) -> set[str]:
rows = conn.execute("SELECT DISTINCT feed_url FROM folder_feeds").fetchall()
return {str(r["feed_url"]) for r in rows}
def get_kept_feed_urls(conn: sqlite3.Connection | None = None) -> set[str]:
"""Feeds unsubscribed-but-retained for their curation (see kept_feeds table).
They still exist in reader (so their starred/tagged entries and tags survive)
but are hidden from the feed tree, All Feeds, Uncategorized, and counts — only
the Saved/Kept view surfaces them, grouped under their original feed name.
"""
if conn is not None:
rows = conn.execute("SELECT feed_url FROM kept_feeds").fetchall()
return {str(r["feed_url"]) for r in rows}
with get_meta_connection() as own:
rows = own.execute("SELECT feed_url FROM kept_feeds").fetchall()
return {str(r["feed_url"]) for r in rows}
def get_feeds_needing_replacement(conn: sqlite3.Connection | None = None) -> set[str]:
"""Feeds the user triaged as dead/needs-replacement (disabled + worklisted)."""
if conn is not None:
return {str(r["feed_url"]) for r in conn.execute("SELECT feed_url FROM feeds_needing_replacement")}
with get_meta_connection() as own:
return {str(r["feed_url"]) for r in own.execute("SELECT feed_url FROM feeds_needing_replacement")}
def unsubscribed_feed_urls_among(feed_urls: Iterable[str | None]) -> set[str]:
"""Which of these feeds you are no longer subscribed to.
Two different states read as "unsubscribed" to a reader and both belong here:
a **kept feed** still exists in reader but is hidden from the tree, and an
**orphan archive**'s feed is gone from reader entirely. What they have in
common is the only thing that matters at the point of use — there is no
subscription behind the name, so clicking through to the feed will not show
you more of it.
Deliberately not "has no folder row": a feed in no folder is still
subscribed, it just lives under the virtual Uncategorized folder, and marking
those would be wrong.
Scoped to the feeds actually on screen so the template carries a handful of
URLs rather than the whole library.
"""
wanted = {str(u) for u in feed_urls if u}
if not wanted:
return set()
return wanted - get_all_reader_feed_urls(include_kept=False)
def get_all_reader_feed_urls(include_kept: bool = False) -> set[str]:
"""Return every feed URL the reader is subscribed to.
Unlike get_all_feed_urls (which reads folder_feeds), this covers feeds that
were never assigned to a folder — the source set for the virtual
"Uncategorized" folder. Read straight from the reader DB to stay O(feeds)
without materializing full feed objects.
Kept feeds (unsubscribed-but-retained for curation) are excluded by default
so they don't resurface in the tree / All Feeds / Uncategorized / counts. The
Saved/Kept view passes include_kept=True to browse them.
Goes through get_reader()'s pooled connection rather than a fresh
sqlite3.connect() per call -- the same anti-pattern fixed 2026-09-03 for
list_entries_for_feeds' sort paths (Plan.md Tier 1). This function is called
unconditionally on every home-route render (nested inside the
"structure_snapshot" perf tick, which is what was actually timing this, not
the cached snapshot lookup its name suggests) plus a dozen other call
sites, so a raw connection here paid file-open/schema-load cost, with a
shorter busy_timeout (5s vs 10s) than the pooled connection, on every
single request -- and was invisible to slow-SQL logging since a bare
sqlite3.connect() bypasses _TimedConnection entirely.
Logged with its own progress_steps when slow, same reasoning as
get_tagged_entry_keys: the `{... for r in db.execute(...)}` comprehension
fetches lazily, so a plain execute()-level timer would only ever see the
prepare cost, not the real scan -- confirmed live 2026-09-03 when this
function measured 3166ms with nothing logged to explain it.
"""
with get_reader() as reader:
db = reader._storage.get_db()
_start = time.perf_counter()
urls = {str(r[0]) for r in db.execute("SELECT url FROM feeds")}
_elapsed_ms = int((time.perf_counter() - _start) * 1000)
if _elapsed_ms > 200:
LOGGER.info(
"[perf] get_all_reader_feed_urls=%dms rows=%d progress_steps=%d",
_elapsed_ms,
len(urls),
getattr(db, "_lectio_progress_steps", 0),
)
if not include_kept:
urls -= get_kept_feed_urls()
return urls
# A "feeds" rule scope targets an explicit set of feeds (not a whole folder). Its
# scope_id is the feed URLs joined by newline (URLs may contain commas, so newline
# is the safe separator). "folders" is the same idea one level up: several whole
# folders in one rule instead of duplicating the rule per folder (requested by
# Josh 2026-08-31) — scope_id is the folder ids joined by newline. These helpers
# centralize parse + scope→feed-set + the per-feed "is this feed in scope" check
# used by the after-refresh runners.
_FEEDS_SCOPE_SEP = "\n"
_FOLDERS_SCOPE_SEP = "\n"
def parse_feeds_scope_id(scope_id: str) -> list[str]:
return [u.strip() for u in (scope_id or "").split(_FEEDS_SCOPE_SEP) if u.strip()]
def parse_folders_scope_id(scope_id: str) -> list[int]:
return [int(s) for s in (scope_id or "").split(_FOLDERS_SCOPE_SEP) if s.strip().isdigit()]
def rule_scope_folder_ids(scope: str, scope_id: str) -> set[int]:
"""Folder id(s) a rule's scope references, for "folder" and "folders" alike —
the single choke point the after-refresh runners' folder-prewarm passes use so
a multi-folder rule's folders are fetched same as a single-folder rule's."""
if scope == "folder":
return {int(scope_id)} if str(scope_id or "").isdigit() else set()
if scope == "folders":
return set(parse_folders_scope_id(scope_id))
return set()
def rule_scope_folder_feed_set(scope: str, scope_id: str, folder_feed_map: dict[int, set[str]]) -> set[str] | None:
"""Prewarmed feed set for a folder-based rule scope (single folder or several),
for feed_in_rule_scope's per-feed test. None for non-folder scopes."""
ids = rule_scope_folder_ids(scope, scope_id)
if not ids:
return None
feed_urls: set[str] = set()
for fid in ids:
feed_urls |= folder_feed_map.get(fid, set())
return feed_urls
def resolve_rule_feed_urls(conn: sqlite3.Connection, scope: str, scope_id: str) -> set[str] | None:
"""Return the set of feed URLs a rule applies to, or None for global (all feeds)."""
if scope == "global":
return None
if scope == "folder":
try:
return get_folder_feed_urls(conn, int(scope_id))
except ValueError, TypeError:
return set()
if scope == "folders":
feed_urls: set[str] = set()
for fid in parse_folders_scope_id(scope_id):
feed_urls |= get_folder_feed_urls(conn, fid)
return feed_urls
if scope == "feed":
return {scope_id} if scope_id else set()
if scope == "feeds":
return set(parse_feeds_scope_id(scope_id))
return set()
def feed_in_rule_scope(scope: str, scope_id: str, feed_url: str, folder_feed_urls: set[str] | None) -> bool:
"""Per-feed scope test for the after-refresh runners. ``folder_feed_urls`` is the
prefetched feed set for a folder-scoped rule (ignored for other scopes)."""
if scope == "global":
return True
if scope in ("folder", "folders"):
return feed_url in (folder_feed_urls or set())
if scope == "feed":
return scope_id == feed_url
if scope == "feeds":
return feed_url in parse_feeds_scope_id(scope_id)
return False
def get_disabled_feed_urls(conn: sqlite3.Connection) -> set[str]:
rows = conn.execute("SELECT feed_url FROM disabled_feeds").fetchall()
return {str(r["feed_url"]) for r in rows}
def get_feed_url_rewrites(feed_url: str) -> list[tuple[str, str]]:
"""[(from_host, to_host)] host-alias rules for *feed_url*, for the ingest
rewrite hook. Opens its own connection: called from the sanitizing parser,
which runs in the caller's tenancy context but has no meta connection."""
try:
conn = sqlite3.connect(str(tenancy.meta_db_path()), timeout=5.0)
try:
return [
(str(f), str(t))
for f, t in conn.execute(
"SELECT from_host, to_host FROM feed_url_rewrites WHERE feed_url = ?",
(feed_url,),
)
]
finally:
conn.close()
except Exception: # noqa: BLE001 — a rewrite lookup must never break a parse
return []
def get_dedupe_host_aliases() -> dict[str, str]:
"""{from_host: to_host} across every feed's URL-rewrite rules, for the saved
dedupe scan.
A feed_url_rewrites rule is the user declaring "this author moved domains" —
ingest already stamps new entries with the current host, but articles saved
before the move keep the dead host, so an old sadh.life/post/x and a new
tush.ar/post/x are the same article the host-scoped dedupe key can't pair.
Folding the migrated host onto its target makes them one key. Because the
map is user-declared (not a heuristic), it can't false-match the way a
cross-host slug tier would."""
try:
conn = sqlite3.connect(str(tenancy.meta_db_path()), timeout=5.0)
try:
aliases: dict[str, str] = {}
for f, t in conn.execute("SELECT DISTINCT from_host, to_host FROM feed_url_rewrites"):
fh = str(f).lower().removeprefix("www.")
th = str(t).lower().removeprefix("www.")
if fh and th and fh != th:
aliases[fh] = th
# Follow chained migrations (a -> b, b -> c) to the final host.
for src in list(aliases):
seen = {src}
dst = aliases[src]
while dst in aliases and dst not in seen:
seen.add(dst)
dst = aliases[dst]
aliases[src] = dst
return aliases
finally:
conn.close()
except Exception: # noqa: BLE001 — dedupe must degrade to host-scoped, not break
return {}
def get_browser_ua_feed_urls(conn: sqlite3.Connection) -> set[str]:
"""Feeds whose fetch should use a browser identity (UA + headers).
A feed lands here only after an honest-UA fetch was *refused* (HTTP
403/415/429/503 or a hang) — see `_maybe_flag_browser_ua`. The request hook in
services.reader_api consults this set so the next fetch escalates. This is a
good-citizen escalation (only on refusal, never preemptive); it does not evade
IP-level blocks."""
rows = conn.execute("SELECT feed_url FROM browser_ua_feeds").fetchall()
return {str(r["feed_url"]) for r in rows}
def _flag_escalation_tier(conn: sqlite3.Connection, table: str, feed_url: str, reason: str) -> bool:
"""INSERT-or-assert one escalation flag. Returns True if newly flagged.
A flag that is already there is re-asserted rather than ignored: `flagged_at` moves to now (a refusal seen while the re-probe is in
flight must not be cleared by it), and a manual flag replaces an automatic one's reason so the re-probe leaves it alone."""
cur = conn.execute(f"INSERT OR IGNORE INTO {table} (feed_url, reason) VALUES (?, ?)", (feed_url, reason[:200]))
if cur.rowcount > 0:
return True
if reason == "manual":
conn.execute(f"UPDATE {table} SET flagged_at = datetime('now'), reason = 'manual' WHERE feed_url = ?", (feed_url,))
else:
conn.execute(f"UPDATE {table} SET flagged_at = datetime('now') WHERE feed_url = ?", (feed_url,))
return False
def flag_browser_ua_feed(conn: sqlite3.Connection, feed_url: str, reason: str = "") -> bool:
"""Mark a feed for browser-identity fetches. Returns True if newly flagged."""
feed_url = feed_url.strip()
if not feed_url:
return False
return _flag_escalation_tier(conn, "browser_ua_feeds", feed_url, reason)
def unflag_browser_ua_feed(conn: sqlite3.Connection, feed_url: str) -> None:
conn.execute("DELETE FROM browser_ua_feeds WHERE feed_url = ?", (feed_url.strip(),))
def get_proxy_feed_urls(conn: sqlite3.Connection) -> set[str]:
"""Feeds whose fetch should route through the outbound proxy in as_needed mode.
A feed lands here only after browser-UA was already in play (flagged this
cycle or earlier) and the fetch still failed — see
`_flag_proxy_feed_on_still_blocked`. The request hook in services.reader_api
consults this set so the next fetch escalates. Irrelevant in off/always
mode — those never consult it."""
rows = conn.execute("SELECT feed_url FROM proxy_feeds").fetchall()
return {str(r["feed_url"]) for r in rows}
def flag_proxy_feed(conn: sqlite3.Connection, feed_url: str, reason: str = "") -> bool:
"""Mark a feed for as-needed proxy escalation. Returns True if newly flagged."""
feed_url = feed_url.strip()
if not feed_url:
return False
return _flag_escalation_tier(conn, "proxy_feeds", feed_url, reason)
def unflag_proxy_feed(conn: sqlite3.Connection, feed_url: str) -> None:
conn.execute("DELETE FROM proxy_feeds WHERE feed_url = ?", (feed_url.strip(),))
def get_tailscale_feed_urls(conn: sqlite3.Connection) -> set[str]:
"""Feeds whose fetch should route through the last-resort proxy in as_needed
mode.
A feed lands here only after the primary proxy was already in play (flagged
this cycle or earlier) and the fetch still failed — see
`_flag_tailscale_feed_on_still_blocked`. Irrelevant in off/always mode, same
as proxy_feeds."""
rows = conn.execute("SELECT feed_url FROM tailscale_feeds").fetchall()
return {str(r["feed_url"]) for r in rows}
def flag_tailscale_feed(conn: sqlite3.Connection, feed_url: str, reason: str = "") -> bool:
"""Mark a feed for last-resort proxy escalation. Returns True if newly flagged."""
feed_url = feed_url.strip()
if not feed_url:
return False
return _flag_escalation_tier(conn, "tailscale_feeds", feed_url, reason)
def unflag_tailscale_feed(conn: sqlite3.Connection, feed_url: str) -> None:
conn.execute("DELETE FROM tailscale_feeds WHERE feed_url = ?", (feed_url.strip(),))
def get_flaresolverr_feed_urls(conn: sqlite3.Connection) -> set[str]:
"""Feeds whose fetch should route through FlareSolverr in as_needed mode.
A feed lands here only after the primary proxy was already in play and the
fetch is STILL specifically a bot-challenge — see
`_flag_flaresolverr_feed_on_still_blocked`. Irrelevant in off/always mode."""
rows = conn.execute("SELECT feed_url FROM flaresolverr_feeds").fetchall()
return {str(r["feed_url"]) for r in rows}
def flag_flaresolverr_feed(conn: sqlite3.Connection, feed_url: str, reason: str = "") -> bool:
"""Mark a feed for FlareSolverr escalation. Returns True if newly flagged."""
feed_url = feed_url.strip()
if not feed_url:
return False
return _flag_escalation_tier(conn, "flaresolverr_feeds", feed_url, reason)
def unflag_flaresolverr_feed(conn: sqlite3.Connection, feed_url: str) -> None:
conn.execute("DELETE FROM flaresolverr_feeds WHERE feed_url = ?", (feed_url.strip(),))
def _disabled_label(reason: str, disabled_at: str) -> str:
""" "manual, Oct 12, 2026 3:00am" or "automatic: 410 Gone, Oct 12, 2026 3:00am" for the Inactive Feeds list.
disabled_at is SQLite's UTC datetime('now'); it is shown in the server's local time like every other timestamp in the UI."""
who = f"automatic: {reason}" if reason else "manual"
try:
when = format_datetime_for_ui(datetime.strptime(disabled_at, "%Y-%m-%d %H:%M:%S").replace(tzinfo=timezone.utc))
except ValueError:
when = None
return f"{who}, {when}" if when else who
def disable_feed(feed_url: str, reason: str = "", *, claim: bool = True) -> None:
"""Disable a feed (no fetching) without unsubscribing. Pause and Disable are the
same state: this sets both Lectio's disabled_feeds row AND reader's
updates_enabled flag, so the Settings tree and Feed Properties agree and the
scheduler skips it regardless of which one is consulted.
*reason* is empty when the user paused the feed and a short label when the system did ("410 Gone"). A system reason never
overwrites an existing row, and a user's own single-feed pause claims a system-disabled feed (so it stops reading as automatic).
Bulk and triage actions pass claim=False: they add rows but leave an existing automatic reason standing."""
feed_url = feed_url.strip()
if not feed_url:
return
reason = reason.strip()[:200]
with get_meta_connection() as conn:
conn.execute(
"INSERT OR IGNORE INTO disabled_feeds (feed_url, reason) VALUES (?, ?)",
(feed_url, reason),
)
if not reason and claim:
conn.execute("UPDATE disabled_feeds SET reason = '' WHERE feed_url = ?", (feed_url,))
try:
with get_reader() as reader:
reader.disable_feed_updates(feed_url)
except Exception:
LOGGER.debug("disable_feed: reader.disable_feed_updates failed for %s", feed_url, exc_info=True)
invalidate_meta_structure_cache()
def enable_feed(feed_url: str) -> None:
"""Re-enable a disabled/paused feed: clears both flags and resets the backoff so
it's checked on the next cycle (the inverse of disable_feed)."""
feed_url = feed_url.strip()
if not feed_url:
return
with get_meta_connection() as conn:
conn.execute("DELETE FROM disabled_feeds WHERE feed_url = ?", (feed_url,))
# A 410 auto-disable leaves a terminal failure record; re-enabling is the user's call to try again, so drop it rather
# than show the feed as permanently gone.
conn.execute("DELETE FROM feed_failure_state WHERE feed_url = ? AND last_error LIKE '410 Gone%'", (feed_url,))
# Clear the backoff so the feed is checked next cycle, not after the old
# retry window.
conn.execute(
"UPDATE feed_failure_state SET next_retry_at = NULL WHERE feed_url = ?",
(feed_url,),
)
try:
with get_reader() as reader:
reader.enable_feed_updates(feed_url)
except Exception:
LOGGER.debug("enable_feed: reader.enable_feed_updates failed for %s", feed_url, exc_info=True)
invalidate_problematic_feeds_cache()
invalidate_meta_structure_cache()
def get_meta_structure_snapshot(conn: sqlite3.Connection) -> dict[str, object]:
"""Cached bundle of folder/feed structure queries.
These tables change only on explicit user mutations, so once computed the
snapshot stays valid until something invalidates it. Collapses ~5 SQL calls
per home render to a dict lookup, which is what we actually need under
concurrent navigation: tiny SELECTs that are fine standalone but
catastrophic when 8 worker threads each issue 5 of them per page render.
Also builds folder_feed_urls_by_id (root → all feeds; non-root → that
folder's feeds), exploiting Lectio's flat one-level-below-root hierarchy
so we