wrolpi/modules/archive/__init__.py
lrnselfreliance dab333124e
Upgrade to SQLAlchemy 2.0 (synchronous) (#803)
SQLAlchemy 1.3 has been end-of-life since 2021, and alembic was held at 1.10
because newer releases need SQLAlchemy 2.  This is a mechanical port; the code
stays synchronous.

- SQLAlchemy[asyncio]==2.0.54, alembic==1.20.0.
- Model annotations use Mapped[...] (2.0 rejects bare annotations on mapped
  attributes); declarative_base moves to sqlalchemy.orm.
- Plain SQL strings go through text() or exec_driver_sql; schema_ddl.execute_sql
  accepts a SQLAlchemy Connection or a raw sqlite3 connection, so rebuild_fts
  works from the baseline migration again.
- Session.get replaces Query.get; session.in_transaction() replaces
  session.transaction; cache_ok on MediaPathType and FancyJSON.
- The tag-directory sync query joins through TagFile's relationships instead of
  the legacy two-entity outerjoin.
- Section-hint rows use attribute access (2.0 rows are tuples).
- Migrations run on an AUTOCOMMIT connection, matching 1.3, where each
  statement committed on its own.  2.0 autobegins, so the unique-stem
  migration's read snapshot could not become a writer after its helpers wrote
  on another connection ("database is locked").  The begin listener emits no
  BEGIN for AUTOCOMMIT connections.
- Newer alembic compares column types: WrolpiKV now describes the TEXT table
  its migration created.
- The test database installs its DDL inside engine.begin(); 2.0 connections
  no longer autocommit, so the FTS tables were rolled back.
- Dependabot now holds SQLAlchemy at 2.0 instead of 1.3.

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-09 15:08:40 -06:00

431 lines
19 KiB
Python

import json
import os
import pathlib
import tempfile
from abc import ABC
from dataclasses import dataclass
from typing import Callable, List, Optional, Tuple
from sqlalchemy import text
from sqlalchemy.orm import Session
from wrolpi.cmd import SINGLE_FILE_BIN, CHROMIUM, FIREFOX, DENO_BIN, SINGLE_FILE_DENO_SCRIPT
from wrolpi.collections import Collection
from wrolpi.common import logger, register_modeler, register_refresh_cleanup, limit_concurrent, split_lines_by_length, \
slow_logger, get_title_from_html, strip_surrogates, TRACE_LEVEL
from wrolpi.db import get_db_session
from wrolpi.downloader import Downloader, Download, DownloadContext, DownloadResult
from wrolpi.errors import UnrecoverableDownloadError
from wrolpi.files.modeler import run_modeler_loop, skip_clause, SkipModeler
from wrolpi.files.models import FileGroup
from wrolpi.vars import PYTEST, DOCKERIZED, DOWNLOAD_USER_AGENT, PROJECT_DIR
from . import lib
from .api import archive_bp # noqa
from .errors import InvalidArchive
from .lib import is_singlefile_file, request_archive, SINGLEFILE_HEADER, get_url_from_singlefile, \
SinglefileArtifacts, WrittenArchiveFiles, extract_singlefile_artifacts, register_archive, write_archive_files
from .models import Archive
PRETTY_NAME = 'Archive'
logger = logger.getChild(__name__)
__all__ = ['ArchiveDownloader', 'archive_downloader', 'model_archive']
@dataclass
class PreparedArchive:
"""Plan produced by ArchiveDownloader.prepare_download.
Holds everything execute_download needs (URL, destination, settings) so the I/O
phase has no DB session.
"""
url: str
destination: Optional[pathlib.Path]
settings: dict
@dataclass
class ExecutedArchive:
"""Output of ArchiveDownloader.execute_download: files are on disk and ready to register."""
written: WrittenArchiveFiles
class ArchiveDownloader(Downloader, ABC):
name = 'archive'
pretty_name = 'Archive'
# A page that has not finished archiving after this long is hung; the config-level
# download_timeout still takes precedence when set.
timeout = 10 * 60
def __repr__(self):
return f'<ArchiveDownloader>'
def already_downloaded(self, session: Session, *urls: List[str]) -> List:
file_groups = list(session.query(FileGroup).filter(FileGroup.url.in_(urls), FileGroup.model == 'archive'))
return file_groups
def prepare_download(self, session: Session, download: Download) -> PreparedArchive:
"""Resolve destination directory. Caller commits (or rolls back on raise)."""
if (download.attempts or 0) > 3:
raise UnrecoverableDownloadError(f'Max download attempts reached for {download.url}')
settings = download.settings or dict()
# Destination from settings (e.g. set by the RSS sub-downloader) wins over collection lookup.
destination = pathlib.Path(settings['destination']) if settings.get('destination') else None
if not destination:
collection = lib.get_or_create_domain_collection(session, download.url)
if collection:
destination = collection.get_or_set_directory(session)
return PreparedArchive(url=download.url, destination=destination, settings=settings)
async def execute_download(self, prepared: PreparedArchive, ctx: DownloadContext,
download: Download = None) -> ExecutedArchive:
"""Run the archive container or local singlefile + readability/screenshot extraction,
then write the artifacts to disk. No DB."""
compress = bool(prepared.settings.get('compress_singlefile'))
if DOCKERIZED or PYTEST:
singlefile, readability, screenshot = await request_archive(prepared.url, compress=compress)
artifacts = SinglefileArtifacts(url=prepared.url, singlefile=singlefile,
readability=readability, screenshot=screenshot)
else:
singlefile_bytes, page_html = await self.do_singlefile(prepared, ctx, download=download)
# Readability/screenshot/title come from the uncompressed page; only the stored
# singlefile may be compressed.
artifacts = await extract_singlefile_artifacts(page_html)
artifacts.singlefile = singlefile_bytes
written = write_archive_files(artifacts.url, artifacts.singlefile, artifacts.readability,
artifacts.screenshot, destination=prepared.destination)
return ExecutedArchive(written=written)
def finalize_download(self, session: Session, download: Download,
executed: ExecutedArchive) -> DownloadResult:
"""Register the Archive record, apply tags, persist parent_download_url. Caller commits."""
archive = register_archive(session, executed.written)
if tag_names := download.tag_names:
for name in tag_names:
archive.add_tag(session, name)
settings = download.settings or dict()
if parent_download_url := settings.get('parent_download_url'):
try:
archive.file_group.update_wrolpi_json({'parent_download_url': parent_download_url})
except ValueError as e:
logger.warning(f'Failed to update wrolpi json for {archive}: {e}')
logger.info(f'Successfully downloaded Archive {download.url} {archive}')
return DownloadResult(success=True, location=archive.location)
async def do_singlefile(self, prepared: PreparedArchive, ctx: DownloadContext,
download: Download = None) -> Tuple[bytes, bytes]:
"""Create a Singlefile from the archive's URL.
Returns (singlefile, page_html): the singlefile to store, and the uncompressed page used
for readability/screenshot/title. The page is always captured uncompressed; when
compress_singlefile is set, a second single-file run converts the captured file into a
compressed (SingleFileZ) self-extracting file (single-file preserves the original URL
header when re-saving its own pages). If the conversion fails, the uncompressed
singlefile is stored instead.
download is the original Download row when called via the manager dispatch; tests
may omit it and a stub is built so process_runner has a url for log messages.
"""
# Get browser settings from config
config = lib.get_archive_downloader_config()
# Determine browser executable
browser_executable = config.browser_executable
if browser_executable:
# Use configured browser path
browser = browser_executable
else:
# Auto-detect: prefer Chromium, fallback to Firefox
browser = CHROMIUM or FIREFOX
if not browser:
raise RuntimeError('No browser found. Install Chromium or Firefox, or configure browser_executable.')
# Get browser args and user agent from config
browser_args = config.browser_args or '["--no-sandbox"]'
user_agent = config.user_agent or DOWNLOAD_USER_AGENT
env = None
browser_wrapper = PROJECT_DIR / 'scripts/singlefile_browser_wrapper.sh'
if DENO_BIN and SINGLE_FILE_DENO_SCRIPT and browser_wrapper.is_file():
# single-file-cli is designed for Deno. Under node 18 (Debian 12/RPi OS) it cannot
# reach the browser CDP on IPv6-enabled hosts ("localhost" resolves to ::1 first with
# no IPv4 fallback) and it never exits after a successful save.
runtime = (str(DENO_BIN), 'run', '--node-modules-dir=manual',
'--allow-read', '--allow-write', '--allow-net', '--allow-env', '--allow-run',
str(SINGLE_FILE_DENO_SCRIPT))
# single-file pipes the browser's stderr but never reads it; under Deno >= 2.3 that
# kills the browser during startup. The wrapper discards the browser's stderr.
env = {**os.environ, 'WROLPI_BROWSER': str(browser)}
browser = browser_wrapper
else:
logger.warning('deno not found, running single-file under node. Upgrade this WROLPi to install deno.')
runtime = (str(SINGLE_FILE_BIN),)
# node 18 has no IPv4 fallback when "localhost" resolves to ::1 first; without this
# single-file cannot reach the browser CDP and fails with "fetch failed".
env = {**os.environ, 'NODE_OPTIONS': '--enable-network-family-autoselection'}
cwd = pathlib.Path('/home/wrolpi')
cwd = cwd if cwd.is_dir() else None
download = download or Download(url=prepared.url)
async def run_single_file(input_url: str, output_path: pathlib.Path, compress: bool):
cmd = ('/usr/bin/nice', '-n15', # Nice above map importing, but very low priority.
*runtime,
'--browser-executable-path', str(browser),
'--browser-args', browser_args,
'--user-agent', user_agent,
'--load-deferred-images-dispatch-scroll-event',
# Create a compressed (SingleFileZ) self-extracting HTML file.
*(('--compress-content',) if compress else ()),
input_url,
str(output_path),
)
# cancel_wrapper reads cancellation from ctx; the Download is forwarded so
# the real download id is used for unkill cleanup and log attribution.
# start_new_session reaps the browser single-file leaves behind (it does not close it).
result = await self.process_runner(download, cmd, cwd, env=env, start_new_session=True, ctx=ctx)
stderr = result.stderr.decode()
log_output = stderr or result.stdout.decode(errors='replace') or 'No stderr or stdout!'
output = output_path.read_bytes() if output_path.is_file() else b''
return result.return_code, output, log_output
# single-file writes the result to a file; Deno >= 2.3 truncates large --dump-content
# output at exit, which corrupted compressed (ZIP) singlefiles.
with tempfile.TemporaryDirectory(prefix='wrolpi-singlefile-') as tmp_dir:
page_path = pathlib.Path(tmp_dir) / 'page.html'
return_code, page_html, log_output = await run_single_file(prepared.url, page_path, compress=False)
if return_code != 0:
e = ChildProcessError(log_output[:1000])
raise RuntimeError(f'singlefile exited with {return_code}') from e
if not page_html:
e = ChildProcessError(log_output[:1000])
raise RuntimeError(f'Singlefile created was empty: {prepared.url}') from e
if SINGLEFILE_HEADER.encode() not in page_html:
e = ChildProcessError(log_output[:1000])
raise RuntimeError(f'Singlefile created was invalid: {prepared.url}') from e
singlefile = page_html
if prepared.settings.get('compress_singlefile'):
# Convert the captured page into a compressed (SingleFileZ) self-extracting file.
# This is a local (file://) operation; the page is not fetched again.
compressed_path = pathlib.Path(tmp_dir) / 'compressed.html'
return_code, compressed, log_output = await run_single_file(
f'file://{page_path}', compressed_path, compress=True)
if return_code == 0 and lib.is_compressed_singlefile(compressed):
singlefile = compressed
else:
logger.error(f'Failed to compress singlefile ({return_code=}), '
f'storing it uncompressed: {prepared.url}\n{log_output[:1000]}')
return singlefile, page_html
archive_downloader = ArchiveDownloader()
def model_archive(session: Session, file_group: FileGroup) -> Archive:
"""
Models an Archive from a given FileGroup.
This function takes in a FileGroup and attempts to create an Archive object.
It does this by checking for the presence of HTML files within the FileGroup,
determining if any of these are SingleFiles, and then extracting relevant data
(title, contents) from either the JSON or text readability files.
Args:
session: Database session.
file_group: The FileGroup to model as an Archive.
Returns:
An Archive object representing the modeled archive.
Raises:
InvalidArchive: If no HTML SingleFile is found in the FileGroup, or if
any other error occurs during modeling.
"""
file_group_id = file_group.id
if not file_group_id:
session.flush([file_group])
file_group_id = file_group.id
# All Archives have an HTML Singlefile. Singlefile may be text HTML, or bytes HTML.
html_paths = file_group.my_html_paths()
if not html_paths:
logger.error('Query returned a group without an HTML file!')
raise InvalidArchive('FileGroup does not contain any html files')
# All Archives have a Singlefile.
singlefile_path = None
for file in html_paths:
try:
if is_singlefile_file(file):
singlefile_path = file
break
except Exception as e:
if PYTEST:
raise
logger.debug(f'Cannot check is_singlefile_file of {repr(file)}', exc_info=e)
if singlefile_path is None:
logger.debug(f'No Archive singlefile found in {file_group}')
raise InvalidArchive('FileGroup does not contain a singlefile')
readability_json_path = None
json_files = file_group.my_json_files()
for json_file in json_files:
path = json_file['path']
if path.name.endswith('.readability.json'):
readability_json_path = path
break
readability_txt_path = None
text_files = file_group.my_text_files()
for text_file in text_files:
path = text_file['path']
if path.name.endswith('.readability.txt'):
readability_txt_path = path
break
try:
file_group.model = 'archive'
file_group.primary_path = singlefile_path
title = None
if readability_json_path:
title = get_title(readability_json_path)
if not title:
title = get_title_from_html(lib.read_singlefile_html(singlefile_path))
# Scraped titles can contain unpaired surrogates which sqlite3 refuses to store; a poisoned
# title fails the modeler's batch commit (it killed archive modeling on 10.0.0.9).
title = strip_surrogates(title)
contents = None
if readability_txt_path:
contents = strip_surrogates(get_article(readability_txt_path))
# Check if an Archive already exists for this FileGroup
archive = session.query(Archive).filter_by(file_group_id=file_group_id).one_or_none()
if not archive:
# Create new Archive if it doesn't exist
archive = Archive(file_group_id=file_group_id, file_group=file_group)
# Set collection_id BEFORE adding to session to avoid autoflush constraint violation
from modules.archive.lib import get_or_create_domain_collection
if file_group.url:
# URL is already set on the FileGroup
collection = get_or_create_domain_collection(session, file_group.url)
archive.collection_id = collection.id if collection else None
else:
# No URL - try to extract from singlefile
try:
url = get_url_from_singlefile(singlefile_path.read_bytes())
file_group.url = url
collection = get_or_create_domain_collection(session, url)
archive.collection_id = collection.id if collection else None
except (RuntimeError, ValueError) as e:
# Could not extract URL from singlefile - archive will have no collection
logger.debug(f'Could not extract URL from singlefile: {e}')
archive.collection_id = None
session.add(archive)
archive.validate()
archive.flush()
# Note: We intentionally do NOT clear collection.directory when archive is outside it.
# Archives are associated with domain collections based on URL, not file location.
# The collection's directory should remain intact for other archives.
file_group.title = file_group.a_text = title or archive.file_group.title
file_group.d_text = contents
archive.rebuild_data()
return archive
except Exception as e:
logger.error('Failed to model Archive', exc_info=e)
raise InvalidArchive(f'Failed to model Archive {file_group}') from e
ARCHIVE_PROCESSING_LIMIT = 20
def _select_archive_ids(session, skip_ids: set) -> list[int]:
query = session.query(FileGroup.id).filter(
# Archive shares text/html with generic HTML. Unlike video/doc/zim, do NOT
# select Archive.id IS NULL: that would re-probe every leftover HTML file
# every refresh. Non-SingleFile HTML stays indexed=0 so apply_indexers
# can claim it.
FileGroup.indexed != True,
FileGroup.mimetype == 'text/html',
)
query = skip_clause(query, FileGroup.id, skip_ids)
return [row[0] for row in query.limit(ARCHIVE_PROCESSING_LIMIT).all()]
def _apply_archive(session: Session, file_group: FileGroup, prepared):
with slow_logger(1, f'Modeling archive took %(elapsed)s seconds: {file_group}', logger__=logger):
archive = session.query(Archive).filter_by(file_group_id=file_group.id).one_or_none()
if archive:
archive.validate()
return
try:
model_archive(session, file_group)
except InvalidArchive as e:
raise SkipModeler from e
@register_modeler
async def archive_modeler(progress_callback: Callable[[int], None] = None):
"""Searches DB for FileGroups that contain an HTML file. If the HTML file is a SingleFile, we model it as an
Archive."""
await run_modeler_loop(
name='archive_modeler',
batch_size=ARCHIVE_PROCESSING_LIMIT,
select_ids=_select_archive_ids,
apply=_apply_archive,
txn='batch',
progress_callback=progress_callback,
)
@register_refresh_cleanup
@limit_concurrent(1)
def archive_cleanup():
with get_db_session(commit=True) as session:
# Remove any domain Collections without any Archives.
# Get all collection_ids that have archives
collection_ids = [i[0] for i in session.execute(text('SELECT DISTINCT collection_id FROM archive')) if i[0]]
# Find domain collections that have no archives
for collection in session.query(Collection).filter_by(kind='domain'):
if collection.id not in collection_ids:
logger.info(f'Deleting empty domain collection: {collection.name}')
session.delete(collection)
def get_title(path):
with path.open('rt') as fh:
try:
return json.load(fh)['title']
except Exception:
if not PYTEST:
logger.warning(f'Archive readability json file exists, but cannot get title. {path}')
return None
def get_article(path):
if path.is_file():
text = path.read_text()
text = split_lines_by_length(text)
return text