rt/wrolpi
1
0
Fork 0
mirror of https://github.com/lrnselfreliance/wrolpi.git synced 2026-10-26 02:48:38 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions 5

domains.yaml imports by diffing (#871)

Browse source 
* domains.yaml imports by diffing

domains.yaml declares one section, its domain Collections keyed by name,
comparing description, directory, Tag and file format.  An unchanged
import writes nothing.

A removed domain without Archives is deleted; one with Archives is reset
to its default settings (decision 7c: a refresh of its Archives would
re-create it), and the config is saved with it.  An invalid entry (name
or directory) is skipped, counted, and its domain kept.  Once tags.yaml
is imported, a domain's Tag that no longer exists is dropped and the
config saved without it (`unknown_tag_names`).

Linking Downloads to domains reads first (the domains once, not one query
per RSS Download) and opens a short write only when there is something to
link; the config is then saved with the links.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* test_bookmarks_config_import starts the app like its siblings

It imports a config without `async_client`, so it fails (no
shared_ctx.configs_imported) whenever it is the first test on its
worker.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* domains.yaml: converge without a save; no links while entries are skipped

A save after the import is refused while an entry is skipped, so the
import must converge without one:
- a removed domain with Archives that is already reset is left as it is
  (prepare reads `import_db_rows`); before, every import reset it again;
- after prepare, the framework drops changed rows that now match the
  database (a dropped unknown Tag), so they are not written again.

Linking Downloads changes domains.yaml after its version is recorded;
while entries are skipped the save is refused, the database stays newer,
and the next startup would save over the skipped entries.  Linking now
waits for an import without skipped entries.

An entry that is not a mapping is skipped and counted, instead of failing
the whole file.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
lrnselfreliance 2026-10-25 11:41:46 -06:00 • committed by Roland
parent 5cdb4aaac0
commit 57c2a29022
No known key found for this signature in database
GPG key ID: 1BA620CE6C9A4680
5 changed files with 407 additions and 142 deletions
Download patch file Download diff file Expand all files Collapse all files

297
modules/archive/lib.py
View file

@ -12,7 +12,7 @@ import zipfile
from dataclasses import dataclass, field
from datetime import datetime
from json import JSONDecodeError
from typing import Optional, Tuple, List, Union, Dict
from typing import Optional, Tuple, List, Union, Dict, Set
from urllib.parse import urlparse
import pytz
@ -27,7 +27,8 @@ from wrolpi.collections import Collection
from wrolpi.common import get_media_directory, get_relative_to_media_directory, logger, extract_domain, \
escape_file_name, aiohttp_post, format_html_string, split_lines_by_length, get_html_soup, get_title_from_html, \
get_wrolpi_config, html_screenshot, html_file_screenshot, ConfigFile, trim_file_name, chunks
from wrolpi.config_version import IMPORT, SAVE
from wrolpi.collections.models import validate_collection_directory
from wrolpi.config_import import ConfigSection, media_relative
from wrolpi.dates import now, Seconds
from wrolpi.db import get_db_session, get_db_curs, session_curs
from wrolpi.errors import UnknownArchive, InvalidOrderBy, InvalidDatetime
@ -36,7 +37,7 @@ from wrolpi.files.lib import handle_file_group_search_results, cached_search_tot
search_filter_cache_key, count_file_groups, get_kv_value, set_kv_value
from wrolpi.files.models import FileGroup
from wrolpi.switches import register_switch_handler, ActivateSwitchMethod
from wrolpi.tags import tag_append_sub_select_where
from wrolpi.tags import Tag, tag_append_sub_select_where, unknown_tag_names
from wrolpi.vars import PYTEST, DOCKERIZED
logger = logger.getChild(__name__)
@ -46,6 +47,81 @@ __all__ = ['DomainsConfig', 'domains_config', 'get_domains_config', 'save_domain
ARCHIVE_SERVICE = 'http://archive:8080'
def _domain_row(description, directory, tag_name, file_format) -> dict:
return dict(
description=description or None,
directory=media_relative(directory) if directory else None,
tag_name=tag_name or None,
file_format=file_format or None,
)
class DomainsSection(ConfigSection):
"""domains.yaml's domain Collections, keyed by name. (Their Downloads are linked after the import, see
`link_domain_and_downloads`.)"""
name = 'domains'
def file_rows(self, config: dict) -> Tuple[Dict[str, dict], Set[str]]:
rows, invalid = dict(), set()
for position, data in enumerate(config.get('collections') or []):
name = data.get('name') if isinstance(data, dict) else None
if not name:
logger.error(f'Domain collection at index {position} has no name, skipping: {data!r}')
# Counted as skipped; never a domain's name.
invalid.add(f'\0{position}')
continue
try:
if not Collection.is_valid_domain_name(name):
raise ValueError(f"Invalid domain name {name!r} (must contain at least one '.' and not start/end"
f" with '.')")
if name in rows:
raise ValueError(f'Another entry has the same domain: {name!r}')
directory = validate_collection_directory(data['directory']) if data.get('directory') else None
rows[name] = _domain_row(data.get('description'), directory, data.get('tag_name'),
data.get('file_format'))
except Exception as e:
logger.error(f'Skipping invalid domain collection at index {position}', exc_info=e)
invalid.add(name)
return rows, invalid
def db_rows(self, session: Session) -> Dict[str, dict]:
query = session.query(Collection.name, Collection.description, Collection.directory, Tag.name,
Collection.file_format) \
.outerjoin(Tag, Tag.id == Collection.tag_id) \
.filter(Collection.kind == 'domain')
return {name: _domain_row(*values) for name, *values in query}
def upsert(self, session: Session, rows: Dict[str, dict]):
existing = {i.name: i for i in
session.query(Collection).filter(Collection.kind == 'domain', Collection.name.in_(list(rows)))}
tags = {i.name: i for i in
session.query(Tag).filter(Tag.name.in_(list({i['tag_name'] for i in rows.values()} - {None})))}
for name, row in rows.items():
if not (collection := existing.get(name)):
logger.info(f'Creating domain collection {name!r}')
collection = Collection(name=name, kind='domain')
session.add(collection)
collection.description = row['description']
collection.directory = get_media_directory() / row['directory'] if row['directory'] else None
collection.tag = tags.get(row['tag_name'])
collection.file_format = row['file_format']
session.flush()
def delete(self, session: Session, keys: Set[str]):
"""A domain with Archives is reset to its default settings (a refresh of its Archives would re-create it,
and deleting it would delete them); any other is deleted."""
with_archives = {i for i, in session.query(Archive.collection_id).distinct()}
for collection in session.query(Collection).filter(Collection.kind == 'domain',
Collection.name.in_(list(keys))):
if collection.id in with_archives:
logger.warning(f'Resetting domain collection {collection.name!r} which is not in the config, because'
f' it has archives')
collection.description = collection.directory = collection.tag = collection.file_format = None
else:
logger.info(f'Deleting domain collection {collection.name!r} (no longer in config)')
session.delete(collection)
@dataclass
class DomainsConfigValidator:
"""Validator for domains config file."""
@ -73,6 +149,7 @@ class DomainsConfig(ConfigFile):
Items are managed dynamically when Archives are indexed.
"""
file_name = 'domains.yaml'
sections = (DomainsSection(),)
validator = DomainsConfigValidator
default_config = dict(
version=0,
@ -92,99 +169,38 @@ class DomainsConfig(ConfigFile):
"""Get list of collection configs."""
return self._config.get('collections', [])
def prepare(self, diffs: dict):
"""A Tag that no longer exists (it was removed) is dropped, and the config is saved without it. A removed
domain with Archives which is already reset (see `DomainsSection.delete`) is left as it is; one which is not
is reset, and the config is saved with it."""
diff = diffs['domains']
if diff.removed:
with get_db_session() as session:
with_archives = {name for name, in session.query(Collection.name).join(
Archive, Archive.collection_id == Collection.id).filter(
Collection.kind == 'domain', Collection.name.in_(list(diff.removed))).distinct()}
reset = _domain_row(None, None, None, None)
for name in with_archives:
if self.import_db_rows['domains'][name] == reset:
diff.removed.discard(name)
else:
self._config_outdated = True
rows = [*diff.added.values(), *diff.changed.values()]
unknown = unknown_tag_names({i['tag_name'] for i in rows if i['tag_name']})
for row in rows:
if row['tag_name'] in unknown:
logger.warning(f'Removing unknown Tag {row["tag_name"]!r} from a domain')
row['tag_name'] = None
self._config_outdated = True
def import_config(self, file: pathlib.Path = None, send_events=False):
"""Import domain collections from config file into database using batch operations."""
file = file or self.get_file()
file_str = str(get_relative_to_media_directory(file))
self._config_outdated = False
super().import_config(file, send_events)
# If config file doesn't exist, mark import as successful (nothing to import)
if not file.is_file():
logger.info('No domains config file, skipping import')
self.successful_import = True
return
ConfigFile.import_config(self, file, send_events)
decision = self.import_decision()
if decision == SAVE:
self.save_database_instead()
return
collections_data = self._config.get('collections', [])
# Empty collections list = never delete DB records
if not collections_data:
logger.info(f'No domain collections in config, preserving existing DB domain collections')
self.successful_import = True
return
logger.info(f'Importing {len(collections_data)} domain collections from {file_str}')
try:
with get_db_session(commit=True) as session:
# Validate and prepare data first
valid_data_list = []
for idx, collection_data in enumerate(collections_data):
try:
name = collection_data.get('name')
if not name:
logger.error(f'Domain collection at index {idx} has no name, skipping')
self.import_skipped += 1
continue
# Validate domain name format before batching
if not Collection.is_valid_domain_name(name):
logger.error(
f"Domain collection at index {idx} has invalid name '{name}' "
f"(must contain at least one '.' and not start/end with '.'), skipping"
)
self.import_skipped += 1
continue
# Ensure kind is 'domain'
collection_data = collection_data.copy()
collection_data['kind'] = 'domain'
# Note: Collections can now be tagged even without a directory
# Tags enable UI search/filtering for directory-less collections
valid_data_list.append(collection_data)
except Exception as e:
logger.error(f'Failed to validate domain collection at index {idx}', exc_info=e)
self.import_skipped += 1
continue
# Batch create/update collections
imported_domains = set()
if valid_data_list:
collections = Collection.batch_from_config(session, valid_data_list)
imported_domains = {c.name for c in collections}
# Delete domain collections that are no longer in config. A domain with Archives is kept: Archives
# create their domain (e.g. one downloaded before this import), and deleting it would delete them.
with_archives = {i for i, in session.query(Archive.collection_id).distinct()}
all_domain_collections = session.query(Collection).filter_by(kind='domain').all()
for collection in all_domain_collections if decision == IMPORT else []:
if collection.name in imported_domains:
continue
if collection.id in with_archives:
logger.warning(f'Keeping domain collection {repr(collection.name)} which is not in config,'
f' because it has archives')
continue
logger.info(f'Deleting domain collection {repr(collection.name)} (no longer in config)')
session.delete(collection)
logger.info(f'Successfully imported {len(imported_domains)} domain collections from {file_str}')
self.record_import_version()
self.successful_import = True
except Exception as e:
self.successful_import = False
message = f'Failed to import {file_str} config!'
logger.error(message, exc_info=e)
if send_events:
Events.send_config_import_failed(message)
raise
def after_import(self, diffs: dict):
if self._config_outdated:
self.background_dump.activate_switch()
def dump_config(self, file: pathlib.Path = None, send_events=False, overwrite=False):
"""Dump all domain collections from database to config file."""
@ -295,13 +311,22 @@ save_domains_config: ActivateSwitchMethod
def import_domains_config():
"""Import domain collections from config file into database."""
"""Import domain collections from config file into database, then link their Downloads (a short write, only
when there is something to link)."""
logger.info('Importing domains config')
domains_config.import_config()
if domains_config.import_skipped:
# Linking changes domains.yaml, which cannot be saved while it has skipped entries: the database would be
# newer than the file, and the next startup would save over them.
logger.warning('Not linking Downloads to domains while domains.yaml has skipped entries')
return
# Link downloads to domain collections after import
with get_db_session() as session:
link_domain_and_downloads(session)
links = domain_download_links(session)
if links:
with get_db_session(commit=True) as session:
_link_downloads(session, links)
save_domains_config.activate_switch()
logger.info('Importing domains config completed')
@ -978,60 +1003,52 @@ def get_or_create_domain_collection(session: Session, url, directory: pathlib.Pa
return collection
def link_domain_and_downloads(session: Session):
"""Associate any Download related to a Domain Collection.
def domain_download_links(session: Session) -> Dict[int, int]:
"""The Downloads to link to a Domain Collection: {Download id: Collection id}.
Downloads are linked to Collections (via collection_id).
This function finds Domain Collections and links their Downloads.
Matching criteria:
- Recurring downloads whose destination is within the domain collection's directory (including subdirectories)
Matching criteria (only recurring Downloads without a Collection):
- a destination within the domain collection's directory (including subdirectories)
- RSS downloads with sub_downloader='archive' (matched by URL domain)
"""
# Local import to avoid circular import: downloader -> archive -> downloader
from wrolpi.downloader import Download
# Only Downloads with a frequency can be a Collection Download.
downloads = list(session.query(Download).filter(Download.frequency.isnot(None)).all())
domains = session.query(Collection.id, Collection.name, Collection.directory) \
.filter(Collection.kind == 'domain').all()
by_name = {name: id_ for id_, name, _ in domains}
# Ensure each directory ends with / for proper prefix matching.
directories = [(id_, str(directory), str(directory).rstrip('/') + '/')
for id_, _, directory in domains if directory]
# Get domain collections that have a directory
domain_collections = session.query(Collection).filter(
Collection.kind == 'domain',
Collection.directory.isnot(None)
).all()
links = dict()
downloads = session.query(Download.id, Download.url, Download.downloader, Download.sub_downloader,
Download.settings) \
.filter(Download.frequency.isnot(None), Download.collection_id.is_(None))
for id_, url, downloader, sub_downloader, settings in downloads:
if destination := (settings or dict()).get('destination'):
for collection_id, directory, prefix in directories:
if destination == directory or destination.startswith(prefix):
links[id_] = collection_id
break
if id_ not in links and downloader == 'rss' and sub_downloader == 'archive':
if collection_id := by_name.get(extract_domain(url)):
links[id_] = collection_id
return links
need_commit = False
# Match downloads by destination directory (including subdirectories)
downloads_with_destination = [d for d in downloads if (d.settings or {}).get('destination')]
for collection in domain_collections:
# Ensure directory ends with / for proper prefix matching
directory = str(collection.directory)
directory_prefix = directory if directory.endswith('/') else directory + '/'
for download in downloads_with_destination:
dest = download.settings['destination']
# Match if destination equals directory or is a subdirectory
if not download.collection_id and (dest == directory or dest.startswith(directory_prefix)):
download.collection_id = collection.id
need_commit = True
def _link_downloads(session: Session, links: Dict[int, int]):
# Local import to avoid circular import: downloader -> archive -> downloader
from wrolpi.downloader import Download
for ids in chunks(list(links), 500):
for download in session.query(Download).filter(Download.id.in_(ids)):
download.collection_id = links[download.id]
session.flush()
# Match RSS downloads with archive sub_downloader by URL domain
rss_archive_downloads = [
d for d in downloads
if d.downloader == 'rss' and d.sub_downloader == 'archive' and not d.collection_id
]
for download in rss_archive_downloads:
# Extract domain from the RSS URL and find matching domain collection
domain_name = extract_domain(download.url)
collection = session.query(Collection).filter_by(
name=domain_name,
kind='domain'
).one_or_none()
if collection:
download.collection_id = collection.id
need_commit = True
if need_commit:
def link_domain_and_downloads(session: Session):
"""Associate any Download related to a Domain Collection (see `domain_download_links`)."""
if links := domain_download_links(session):
_link_downloads(session, links)
session.commit()

233
modules/archive/test/test_domains_config_import.py Normal file
View file

@ -0,0 +1,233 @@
"""domains.yaml imports by diffing its domain Collections against the database, and writing only the differences
(see wrolpi.config_import). A domain removed from the file is deleted, unless it has Archives: then it is reset to
its default settings (a refresh of its Archives would re-create it)."""
import pytest
import yaml
from modules.archive.lib import get_domains_config, import_domains_config
from modules.archive.models import Archive
from wrolpi.collections import Collection
from wrolpi.config_version import get_config_version
from wrolpi.downloader import Download
from wrolpi.test.test_config_import import edit_config, statements, writes
@pytest.fixture
async def saved_domains(test_session, test_directory, async_client, archive_factory, tag_factory, await_switches):
"""`example.com` (tagged, in a custom directory), `other.org`, and `archived.net` (which has an Archive and a
description), saved to the config (which records its version)."""
tag = await tag_factory('news')
archive_factory(domain='archived.net', url='https://archived.net/a')
test_session.commit()
archived = test_session.query(Collection).filter_by(name='archived.net', kind='domain').one()
archived.description = 'Has archives'
test_session.add_all([
Collection(name='example.com', kind='domain', description='Example', tag_id=tag.id,
directory=test_directory / 'archive/news/example.com'),
Collection(name='other.org', kind='domain'),
])
test_session.commit()
get_domains_config().dump_config()
await await_switches()
assert get_config_version(test_session, 'domains.yaml') is not None
return get_domains_config().get_file()
def domains(test_session) -> dict:
test_session.expire_all()
return {i.name: i for i in test_session.query(Collection).filter_by(kind='domain')}
def edit_domain(path, name, **values):
def change(data):
next(i for i in data['collections'] if i['name'] == name).update(values)
edit_config(path, change)
@pytest.mark.asyncio
async def test_unchanged_import_writes_nothing(test_session, saved_domains):
"""Importing a config the database already matches opens no write transaction, nor does linking Downloads when
there are none to link."""
config = get_domains_config()
with statements() as collected:
import_domains_config()
assert config.successful_import is True
assert writes(collected) == []
@pytest.mark.asyncio
async def test_import_writes_only_the_differences(test_session, test_directory, saved_domains):
"""A changed domain is updated, a removed one (without Archives) is deleted, an added one is created; the next
import writes nothing."""
def change(data):
data['collections'] = [i for i in data['collections'] if i['name'] != 'other.org']
data['collections'].append(dict(name='new.com', kind='domain', description='New'))
example = next(i for i in data['collections'] if i['name'] == 'example.com')
example.update(description='Changed', tag_name=None, directory='archive/example.com')
edit_config(saved_domains, change)
config = get_domains_config()
config.import_config()
assert config.successful_import is True
found = domains(test_session)
assert set(found) == {'example.com', 'archived.net', 'new.com'}
example = found['example.com']
assert (example.description, example.tag_id, example.directory) == \
('Changed', None, test_directory / 'archive/example.com')
assert found['new.com'].description == 'New'
with statements() as collected:
config.import_config()
assert writes(collected) == []
@pytest.mark.asyncio
async def test_removed_domain_with_archives_is_reset(test_session, saved_domains, await_switches):
"""A domain with Archives that is removed from the config keeps its Archives, and is reset to its default
settings; the config is saved with it, so the next import writes nothing."""
edit_config(saved_domains, lambda data: data.update(
collections=[i for i in data['collections'] if i['name'] != 'archived.net']))
config = get_domains_config()
config.import_config()
await await_switches()
archived = domains(test_session)['archived.net']
assert archived.description is None
assert test_session.query(Archive).filter_by(collection_id=archived.id).count() == 1
assert 'archived.net' in saved_domains.read_text()
with statements() as collected:
config.import_config()
assert writes(collected) == []
@pytest.mark.asyncio
async def test_invalid_entries_are_kept(test_session, saved_domains):
"""An entry with an invalid name or directory is skipped (counted), and its domain is not deleted."""
edit_domain(saved_domains, 'other.org', directory='../outside')
edit_config(saved_domains, lambda data: data['collections'].append(dict(name='nodot', kind='domain')))
config = get_domains_config()
config.import_config()
assert config.import_skipped == 2
assert set(domains(test_session)) == {'example.com', 'other.org', 'archived.net'}
@pytest.mark.asyncio
async def test_unknown_tag_is_dropped(test_session, saved_domains, test_tags_config, await_switches):
"""Once tags.yaml is imported, a domain's Tag that no longer exists is dropped and the config is saved without
it: the next import writes nothing."""
from wrolpi.tags import get_tags_config
get_tags_config().dump_config()
get_tags_config().import_config()
edit_domain(saved_domains, 'other.org', tag_name='gone')
config = get_domains_config()
config.import_config()
await await_switches()
assert domains(test_session)['other.org'].tag_id is None
assert 'gone' not in saved_domains.read_text()
with statements() as collected:
config.import_config()
assert writes(collected) == []
@pytest.mark.asyncio
async def test_downloads_are_linked(test_session, saved_domains, await_switches):
"""After the import, an RSS archive Download of a domain is linked to it (the domains are read once, not once
per Download), and the config is saved with it."""
test_session.add_all([
Download(url='https://other.org/feed.xml', downloader='rss', sub_downloader='archive', frequency=86400),
Download(url='https://example.com/feed.xml', downloader='rss', sub_downloader='archive', frequency=86400),
])
test_session.commit()
with statements() as collected:
import_domains_config()
await await_switches()
found = domains(test_session)
assert {i.url: i.collection_id for i in test_session.query(Download)} == {
'https://other.org/feed.xml': found['other.org'].id,
'https://example.com/feed.xml': found['example.com'].id,
}
assert len([i for i in collected if 'FROM collection' in i and 'WHERE collection.name' in i]) == 0
assert 'https://other.org/feed.xml' in saved_domains.read_text()
def add_invalid_entry(path):
edit_config(path, lambda data: data['collections'].append(dict(name='nodot', kind='domain')))
@pytest.mark.asyncio
async def test_reset_with_a_skipped_entry_converges(test_session, saved_domains, await_switches):
"""A skipped entry stops the config from being saved; a domain reset by this import is not reset again by the
next one (which writes nothing)."""
edit_config(saved_domains, lambda data: data.update(
collections=[i for i in data['collections'] if i['name'] != 'archived.net']))
add_invalid_entry(saved_domains)
config = get_domains_config()
config.import_config()
await await_switches()
assert domains(test_session)['archived.net'].description is None
with statements() as collected:
config.import_config()
assert config.import_skipped == 1
assert writes(collected) == []
@pytest.mark.asyncio
async def test_unknown_tag_with_a_skipped_entry_converges(test_session, saved_domains, test_tags_config,
await_switches):
"""A dropped Tag the config cannot be saved without (an entry was skipped) is not dropped again."""
from wrolpi.tags import get_tags_config
get_tags_config().dump_config()
get_tags_config().import_config()
edit_domain(saved_domains, 'other.org', tag_name='gone')
add_invalid_entry(saved_domains)
config = get_domains_config()
config.import_config()
await await_switches()
with statements() as collected:
config.import_config()
assert writes(collected) == []
@pytest.mark.asyncio
async def test_no_links_while_entries_are_skipped(test_session, saved_domains, await_switches):
"""Linking changes domains.yaml; while an entry is skipped the config cannot be saved, so Downloads are not
linked (the database would be newer than the file, and the next startup would save over the skipped entry)."""
test_session.add(Download(url='https://other.org/feed.xml', downloader='rss', sub_downloader='archive',
frequency=86400))
test_session.commit()
add_invalid_entry(saved_domains)
import_domains_config()
await await_switches()
test_session.expire_all()
assert test_session.query(Download).one().collection_id is None
assert get_config_version(test_session, 'domains.yaml') == yaml.safe_load(saved_domains.read_text())['version']
assert 'nodot' in saved_domains.read_text()
@pytest.mark.asyncio
async def test_entry_that_is_not_a_mapping_is_skipped(test_session, saved_domains):
"""An entry that is not a mapping is skipped (counted); the rest of the file is imported."""
edit_config(saved_domains, lambda data: data['collections'].extend(['example.net', None]))
edit_domain(saved_domains, 'other.org', description='Changed')
config = get_domains_config()
config.import_config()
assert config.successful_import is True
assert config.import_skipped == 2
assert domains(test_session)['other.org'].description == 'Changed'