mirror of
https://github.com/lrnselfreliance/wrolpi.git
synced 2026-10-26 02:48:38 -04:00
files: shared modeler batch loop for video, doc, zim, archive (#722)
* files: shared modeler batch loop for video, doc, zim, archive Paging, skip-ids, progress, and cancel yields live in wrolpi.files.modeler.run_modeler_loop. Each module still owns select / prepare / apply and its transaction style: video probes outside the write lock, docs use one write per file with lock retries, zim and archive use one write per page. Archive HTML that is not a SingleFile raises SkipModeler so apply_indexers can claim it. * files: quieter batch-modeler errors; drop unused imports Break out of a batch page on InvalidRequestError after a flush failure so the rest of the page does not each log the same poisoned session. Re-raise PYTEST apply failures without a misleading commit-failure message. Remove asyncio/List/OperationalError/PYTEST/Iterable leftovers from docs, archive, and zim.
This commit is contained in:
lrnselfreliance
Roland
parent
2d0f433450
commit
92fa3f785f
9 changed files with 443 additions and 279 deletions
|
|
@ -1,13 +1,11 @@
|
|||
import asyncio
|
||||
import json
|
||||
import os
|
||||
import pathlib
|
||||
import tempfile
|
||||
from abc import ABC
|
||||
from dataclasses import dataclass
|
||||
from typing import Callable, List, Optional, Tuple, Iterable
|
||||
from typing import Callable, List, Optional, Tuple
|
||||
|
||||
from sqlalchemy import not_
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from wrolpi.cmd import SINGLE_FILE_BIN, CHROMIUM, FIREFOX, DENO_BIN, SINGLE_FILE_DENO_SCRIPT
|
||||
|
|
@ -17,6 +15,7 @@ from wrolpi.common import logger, register_modeler, register_refresh_cleanup, li
|
|||
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
|
||||
|
|
@ -358,85 +357,46 @@ def model_archive(session: Session, file_group: FileGroup) -> Archive:
|
|||
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."""
|
||||
invalid_archives = set()
|
||||
total_processed = 0
|
||||
|
||||
while True:
|
||||
with get_db_session(commit=True) as session:
|
||||
results = session.query(FileGroup, Archive) \
|
||||
.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',
|
||||
).filter(not_(FileGroup.id.in_(list(invalid_archives)))) \
|
||||
.outerjoin(Archive, Archive.file_group_id == FileGroup.id) \
|
||||
.limit(20)
|
||||
results: Iterable[Tuple[FileGroup, Archive]]
|
||||
results_list = list(results)
|
||||
|
||||
processed = 0
|
||||
for processed, (file_group, archive) in enumerate(results_list):
|
||||
with slow_logger(1, f'Modeling archive took %(elapsed)s seconds: {file_group}',
|
||||
logger__=logger):
|
||||
if archive:
|
||||
try:
|
||||
archive_id = archive.id
|
||||
archive.validate()
|
||||
# Successfully validated, mark as indexed
|
||||
file_group.indexed = True
|
||||
except Exception as e:
|
||||
logger.error(f'Unable to validate Archive {archive_id}')
|
||||
# Don't mark as indexed - will retry later
|
||||
if PYTEST:
|
||||
raise
|
||||
else:
|
||||
try:
|
||||
model_archive(session, file_group)
|
||||
# Successfully modeled, mark as indexed
|
||||
file_group.indexed = True
|
||||
except InvalidArchive:
|
||||
# It was not a real Archive. Many HTML files will not be an Archive.
|
||||
file_group.indexed = False
|
||||
invalid_archives.add(file_group.id)
|
||||
except Exception as e:
|
||||
# Some other error occurred during modeling - don't mark as indexed so we can retry
|
||||
logger.error(f'Failed to model Archive for FileGroup {file_group.id}: {e}')
|
||||
if PYTEST:
|
||||
raise
|
||||
|
||||
try:
|
||||
session.commit()
|
||||
except Exception as e:
|
||||
# One poisoned row must not kill the modeler: skip this batch for the rest of this
|
||||
# run and keep modeling. The batch stays unindexed, so the next refresh retries it.
|
||||
session.rollback()
|
||||
logger.error(f'archive_modeler failed to commit batch, skipping {len(results_list)} FileGroups',
|
||||
exc_info=e)
|
||||
invalid_archives.update(fg.id for fg, _ in results_list)
|
||||
if PYTEST:
|
||||
raise
|
||||
|
||||
# Report batch progress
|
||||
batch_count = len(results_list)
|
||||
total_processed += batch_count
|
||||
if progress_callback and batch_count > 0:
|
||||
progress_callback(total_processed)
|
||||
|
||||
if processed < 19:
|
||||
# Did not reach limit (enumerate is 0-indexed, so 19 = 20 items), do not query again.
|
||||
if __debug__ and logger.isEnabledFor(TRACE_LEVEL):
|
||||
logger.trace(f'archive_modeler: DONE (processed {processed + 1} files)')
|
||||
break
|
||||
|
||||
# Sleep to catch cancel.
|
||||
await asyncio.sleep(0)
|
||||
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
|
||||
|
|
|
|||
|
|
@ -1,17 +1,16 @@
|
|||
import asyncio
|
||||
import logging
|
||||
from typing import Callable, List
|
||||
from typing import Callable
|
||||
|
||||
from sqlalchemy import or_
|
||||
from sqlalchemy.exc import OperationalError
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from wrolpi.common import register_modeler, register_refresh_cleanup, \
|
||||
truncate_object_bytes, split_lines_by_length
|
||||
from wrolpi.db import get_db_session
|
||||
from wrolpi.files.indexers import Indexer
|
||||
from wrolpi.files.modeler import run_modeler_loop, skip_clause
|
||||
from wrolpi.files.models import FileGroup
|
||||
from wrolpi.vars import PYTEST, FILE_MAX_TEXT_SIZE
|
||||
from wrolpi.vars import FILE_MAX_TEXT_SIZE
|
||||
from .extractors import extract_metadata
|
||||
from .lib import get_or_create_subject_collection, get_or_create_author_collection, is_valid_author, split_authors, \
|
||||
normalize_author, normalize_subject, split_subjects, is_valid_subject, discover_calibre_cover
|
||||
|
|
@ -24,6 +23,33 @@ DOC_PROCESSING_LIMIT = 10
|
|||
DOC_LOCK_RETRIES = 3
|
||||
|
||||
|
||||
def _select_doc_ids(session, skip_ids: set) -> list[int]:
|
||||
query = session.query(FileGroup.id) \
|
||||
.outerjoin(Doc, Doc.file_group_id == FileGroup.id) \
|
||||
.filter(
|
||||
or_(
|
||||
*[FileGroup.mimetype == mt for mt in DOC_MIMETYPES],
|
||||
*[FileGroup.primary_path.like(f'%{suffix}') for suffix in COMIC_BOOK_SUFFIXES],
|
||||
),
|
||||
# Model any doc that has no Doc row yet (even if `apply_indexers` already set
|
||||
# `indexed=True` before this modeler ran), or any doc explicitly flagged for
|
||||
# re-indexing. Gating solely on `indexed == False` left already-indexed docs
|
||||
# permanently unmodeled and invisible in /api/docs.
|
||||
or_(
|
||||
Doc.id.is_(None),
|
||||
FileGroup.indexed != True,
|
||||
),
|
||||
)
|
||||
query = skip_clause(query, FileGroup.id, skip_ids)
|
||||
return [row[0] for row in query.limit(DOC_PROCESSING_LIMIT).all()]
|
||||
|
||||
|
||||
def _apply_doc(session: Session, file_group: FileGroup, prepared):
|
||||
doc = _model_doc(file_group, session)
|
||||
session.add(doc)
|
||||
file_group.model = Doc.__tablename__
|
||||
|
||||
|
||||
@register_modeler
|
||||
async def doc_modeler(progress_callback: Callable[[int], None] = None):
|
||||
"""Searches for doc files (epub, mobi, pdf, docx, doc, odt, cbz, cbr) and models them.
|
||||
|
|
@ -36,78 +62,15 @@ async def doc_modeler(progress_callback: Callable[[int], None] = None):
|
|||
modeler run. Per-doc transactions keep the write window short and isolate failures so one
|
||||
locked/broken doc cannot poison the rest of the run.
|
||||
"""
|
||||
total_processed = 0
|
||||
# Ids of docs that failed to model this run. A failed doc keeps matching the `Doc.id IS NULL`
|
||||
# gate below, so without excluding it the loop would re-select it forever, hang, and starve
|
||||
# later docs. Successfully-modeled docs are excluded automatically (they gain a Doc row), so
|
||||
# only failures need tracking -- keeping this set (and the `notin_` clause) small. Failed docs
|
||||
# are retried on the next refresh.
|
||||
failed_ids: set = set()
|
||||
while True:
|
||||
# Short read-only transaction to choose the next batch of ids. Do NOT hold it open across
|
||||
# the slow per-doc modeling below.
|
||||
with get_db_session() as session:
|
||||
query = session.query(FileGroup.id) \
|
||||
.outerjoin(Doc, Doc.file_group_id == FileGroup.id) \
|
||||
.filter(
|
||||
or_(
|
||||
*[FileGroup.mimetype == mt for mt in DOC_MIMETYPES],
|
||||
*[FileGroup.primary_path.like(f'%{suffix}') for suffix in COMIC_BOOK_SUFFIXES],
|
||||
),
|
||||
# Model any doc that has no Doc row yet (even if `apply_indexers` already set
|
||||
# `indexed=True` before this modeler ran), or any doc explicitly flagged for
|
||||
# re-indexing. Gating solely on `indexed == False` left already-indexed docs
|
||||
# permanently unmodeled and invisible in /api/docs.
|
||||
or_(
|
||||
Doc.id.is_(None),
|
||||
FileGroup.indexed != True,
|
||||
),
|
||||
)
|
||||
if failed_ids:
|
||||
query = query.filter(FileGroup.id.notin_(failed_ids))
|
||||
fg_ids: List[int] = [row[0] for row in query.limit(DOC_PROCESSING_LIMIT).all()]
|
||||
|
||||
if not fg_ids:
|
||||
break
|
||||
|
||||
for fg_id in fg_ids:
|
||||
for attempt in range(DOC_LOCK_RETRIES):
|
||||
try:
|
||||
# One short write transaction per doc.
|
||||
with get_db_session(commit=True) as session:
|
||||
file_group = session.query(FileGroup).get(fg_id)
|
||||
if file_group is None:
|
||||
# Deleted between discovery and now; nothing to model.
|
||||
break
|
||||
doc = _model_doc(file_group, session)
|
||||
session.add(doc)
|
||||
file_group.model = Doc.__tablename__
|
||||
file_group.indexed = True
|
||||
break
|
||||
except OperationalError as e:
|
||||
# Another connection committed during our transaction. Retrying with a fresh
|
||||
# transaction gets a fresh snapshot and usually succeeds.
|
||||
if 'locked' in str(e).lower() and attempt < DOC_LOCK_RETRIES - 1:
|
||||
await asyncio.sleep(0.2 * (attempt + 1))
|
||||
continue
|
||||
logger.error(f'Failed to model doc file_group_id={fg_id}: database error', exc_info=e)
|
||||
failed_ids.add(fg_id)
|
||||
break
|
||||
except Exception as e:
|
||||
# Never reference the ORM object here: after a failed flush the session is
|
||||
# rolled back, so touching a lazy/expired attribute (e.g. in its repr) raises a
|
||||
# second exception that would abort the whole modeler. Log the plain id only.
|
||||
logger.error(f'Failed to model doc file_group_id={fg_id}', exc_info=e)
|
||||
failed_ids.add(fg_id)
|
||||
if PYTEST:
|
||||
raise
|
||||
break
|
||||
# Yield so a cancel/other tasks can run between docs.
|
||||
await asyncio.sleep(0)
|
||||
|
||||
total_processed += len(fg_ids)
|
||||
if progress_callback:
|
||||
progress_callback(total_processed)
|
||||
await run_modeler_loop(
|
||||
name='doc_modeler',
|
||||
batch_size=DOC_PROCESSING_LIMIT,
|
||||
select_ids=_select_doc_ids,
|
||||
apply=_apply_doc,
|
||||
txn='per_item',
|
||||
lock_retries=DOC_LOCK_RETRIES,
|
||||
progress_callback=progress_callback,
|
||||
)
|
||||
|
||||
|
||||
def _model_doc(file_group: FileGroup, session: Session) -> Doc:
|
||||
|
|
|
|||
|
|
@ -167,7 +167,7 @@ async def test_doc_modeler_terminates_on_persistent_failure(async_client, test_s
|
|||
# PYTEST re-raises inside the loop's per-item handler; disable that so we exercise the failure
|
||||
# path and the loop's termination guard, not the re-raise.
|
||||
with patch('modules.docs._model_doc', side_effect=RuntimeError('boom')), \
|
||||
patch('modules.docs.PYTEST', False):
|
||||
patch('wrolpi.files.modeler.PYTEST', False):
|
||||
await asyncio.wait_for(doc_modeler(), timeout=10)
|
||||
|
||||
# No Doc rows were created (every attempt failed), but the loop still terminated.
|
||||
|
|
|
|||
|
|
@ -10,6 +10,7 @@ from modules.videos.common import get_or_create_ffprobe_json
|
|||
from modules.videos.models import Video, AUDIO_PLAYLIST_MIMETYPES
|
||||
from wrolpi.common import logger, limit_concurrent, register_modeler, register_refresh_cleanup
|
||||
from wrolpi.db import get_db_curs, get_db_session
|
||||
from wrolpi.files.modeler import run_modeler_loop, skip_clause
|
||||
from wrolpi.files.models import FileGroup
|
||||
from wrolpi.vars import PYTEST
|
||||
from .downloader import video_downloader # Import downloaders so they are registered.
|
||||
|
|
@ -43,76 +44,55 @@ def _model_video(session: Session, file_group: FileGroup, video: Video | None, p
|
|||
return video
|
||||
|
||||
|
||||
def _select_video_ids(session, skip_ids: set) -> list[int]:
|
||||
query = session.query(FileGroup.id) \
|
||||
.outerjoin(Video, Video.file_group_id == FileGroup.id) \
|
||||
.filter(
|
||||
or_(FileGroup.mimetype.like('video/%'), FileGroup.mimetype.like('audio/%')),
|
||||
FileGroup.mimetype.notin_(AUDIO_PLAYLIST_MIMETYPES),
|
||||
# Exclusive-mimetype modeler: claim unmodeled rows even if apply_indexers
|
||||
# already set indexed=True, and re-model when files changed (indexed != True).
|
||||
or_(Video.id.is_(None), FileGroup.indexed != True),
|
||||
)
|
||||
query = skip_clause(query, FileGroup.id, skip_ids)
|
||||
return [row[0] for row in query.limit(VIDEO_PROCESSING_LIMIT).all()]
|
||||
|
||||
|
||||
async def _prepare_videos(ids: list[int]) -> dict:
|
||||
"""ffprobe with no write transaction open. A subprocess per video can run for seconds."""
|
||||
with get_db_session() as session:
|
||||
rows = session.query(FileGroup.id, FileGroup.primary_path).filter(FileGroup.id.in_(ids)).all()
|
||||
probed = dict()
|
||||
for file_group_id, primary_path in rows:
|
||||
try:
|
||||
probed[file_group_id] = await get_or_create_ffprobe_json(pathlib.Path(str(primary_path)))
|
||||
except Exception as e:
|
||||
if PYTEST:
|
||||
raise
|
||||
logger.error(f'Unable to ffprobe Video: {primary_path}', exc_info=e)
|
||||
await asyncio.sleep(0)
|
||||
return probed
|
||||
|
||||
|
||||
def _apply_video(session: Session, file_group: FileGroup, prepared: dict):
|
||||
video = session.query(Video).filter_by(file_group_id=file_group.id).one_or_none()
|
||||
_model_video(session, file_group, video, prepared or {})
|
||||
|
||||
|
||||
@register_modeler
|
||||
async def video_modeler(progress_callback: Callable[[int], None] = None):
|
||||
total_processed = 0
|
||||
# Ids that failed to model this run. A failure leaves no Video row, so the `Video.id IS
|
||||
# NULL` gate below would re-select it forever; failures are retried on the next refresh.
|
||||
failed_ids: set = set()
|
||||
while True:
|
||||
# Read the batch; nothing is claimed yet, so the write lock stays free while ffprobe runs.
|
||||
with get_db_session() as session:
|
||||
query = session.query(FileGroup.id, FileGroup.primary_path) \
|
||||
.outerjoin(Video, Video.file_group_id == FileGroup.id) \
|
||||
.filter(
|
||||
or_(FileGroup.mimetype.like('video/%'), FileGroup.mimetype.like('audio/%')),
|
||||
FileGroup.mimetype.notin_(AUDIO_PLAYLIST_MIMETYPES),
|
||||
# Exclusive-mimetype modeler: claim unmodeled rows even if apply_indexers
|
||||
# already set indexed=True, and re-model when files changed (indexed != True).
|
||||
or_(Video.id.is_(None), FileGroup.indexed != True),
|
||||
)
|
||||
if failed_ids:
|
||||
query = query.filter(FileGroup.id.notin_(failed_ids))
|
||||
batch: List[Tuple[int, pathlib.Path]] = list(query.limit(VIDEO_PROCESSING_LIMIT).all())
|
||||
|
||||
if not batch:
|
||||
break
|
||||
|
||||
# ffprobe each file with no transaction open. This is a subprocess per video and can run
|
||||
# for seconds; inside the write transaction below it would hold the write lock for the whole
|
||||
# batch, and every other writer on the box would wait out `busy_timeout` (30s).
|
||||
probed = dict()
|
||||
for file_group_id, primary_path in batch:
|
||||
try:
|
||||
probed[file_group_id] = await get_or_create_ffprobe_json(pathlib.Path(str(primary_path)))
|
||||
except Exception as e:
|
||||
if PYTEST:
|
||||
raise
|
||||
logger.error(f'Unable to ffprobe Video: {primary_path}', exc_info=e)
|
||||
# Sleep to catch cancel.
|
||||
await asyncio.sleep(0)
|
||||
|
||||
with get_db_session(commit=True) as session:
|
||||
file_groups: List[Tuple[FileGroup, Video]] = list(
|
||||
session.query(FileGroup, Video)
|
||||
.filter(FileGroup.id.in_([i for i, _ in batch]))
|
||||
.outerjoin(Video, Video.file_group_id == FileGroup.id))
|
||||
|
||||
for file_group, video in file_groups:
|
||||
try:
|
||||
_model_video(session, file_group, video, probed)
|
||||
except Exception as e:
|
||||
# Before the re-raise, so no failure path can re-select this id forever.
|
||||
failed_ids.add(file_group.id)
|
||||
if PYTEST:
|
||||
raise
|
||||
logger.error(f'Unable to model Video: {file_group.primary_path}', exc_info=e)
|
||||
|
||||
file_group.indexed = True
|
||||
|
||||
# Report batch progress
|
||||
total_processed += len(batch)
|
||||
if progress_callback:
|
||||
progress_callback(total_processed)
|
||||
|
||||
logger.debug(f'Modeled {len(batch)} videos')
|
||||
|
||||
if len(batch) < VIDEO_PROCESSING_LIMIT:
|
||||
# Did not reach limit, do not query again.
|
||||
break
|
||||
|
||||
# Sleep to catch cancel.
|
||||
await asyncio.sleep(0)
|
||||
await run_modeler_loop(
|
||||
name='video_modeler',
|
||||
batch_size=VIDEO_PROCESSING_LIMIT,
|
||||
select_ids=_select_video_ids,
|
||||
prepare=_prepare_videos,
|
||||
apply=_apply_video,
|
||||
txn='batch',
|
||||
# A failure leaves no Video row, so Video.id IS NULL would re-select it. Mark indexed
|
||||
# anyway; the skip set covers this run, the next refresh retries via Video.id IS NULL.
|
||||
mark_indexed_on_failure=True,
|
||||
progress_callback=progress_callback,
|
||||
)
|
||||
|
||||
|
||||
# Rows written per transaction when claiming Videos for their Channels. Small enough that the
|
||||
|
|
|
|||
|
|
@ -163,7 +163,7 @@ async def test_video_modeler_terminates_on_persistent_failure(async_client, test
|
|||
# loop's per-item handler; disable that to exercise the failure path and the termination
|
||||
# guard, not the re-raise.
|
||||
with patch('modules.videos._model_video', side_effect=RuntimeError('boom')), \
|
||||
patch('modules.videos.PYTEST', False):
|
||||
patch('wrolpi.files.modeler.PYTEST', False):
|
||||
await asyncio.wait_for(video_modeler(), timeout=10)
|
||||
|
||||
assert test_session.query(Video).count() == 0, 'every attempt failed, yet the loop terminated'
|
||||
|
|
|
|||
|
|
@ -1,4 +1,3 @@
|
|||
import asyncio
|
||||
import functools
|
||||
import json
|
||||
import pathlib
|
||||
|
|
@ -26,6 +25,7 @@ from wrolpi.events import Events
|
|||
from wrolpi.db import get_db_session, get_db_curs
|
||||
from wrolpi.downloader import DownloadFrequency
|
||||
from wrolpi.files.lib import split_file_name_words
|
||||
from wrolpi.files.modeler import run_modeler_loop, skip_clause
|
||||
from wrolpi.files.worker import file_worker
|
||||
from wrolpi.files.models import FileGroup
|
||||
from wrolpi.vars import PYTEST, DOCKERIZED
|
||||
|
|
@ -73,58 +73,43 @@ def model_zim(file_group: FileGroup, session: Session) -> Zim:
|
|||
return zim
|
||||
|
||||
|
||||
ZIM_PROCESSING_LIMIT = 10
|
||||
|
||||
|
||||
def _select_zim_ids(session, skip_ids: set) -> list[int]:
|
||||
query = session.query(FileGroup.id) \
|
||||
.outerjoin(Zim, Zim.file_group_id == FileGroup.id) \
|
||||
.filter(FileGroup.primary_path.ilike('%.zim')) \
|
||||
.filter(
|
||||
# Exclusive-mimetype modeler: claim unmodeled rows even if apply_indexers
|
||||
# already set indexed=True, and re-model when files changed.
|
||||
or_(Zim.id.is_(None), FileGroup.indexed != True),
|
||||
)
|
||||
query = skip_clause(query, FileGroup.id, skip_ids)
|
||||
return [row[0] for row in query.limit(ZIM_PROCESSING_LIMIT).all()]
|
||||
|
||||
|
||||
def _apply_zim(session: Session, file_group: FileGroup, prepared):
|
||||
zim = session.query(Zim).filter_by(file_group_id=file_group.id).one_or_none()
|
||||
if not zim:
|
||||
zim = model_zim(file_group, session)
|
||||
zim.flush()
|
||||
else:
|
||||
file_group.title = file_group.primary_path.name
|
||||
file_group.a_text = split_file_name_words(file_group.primary_path.name)
|
||||
file_group.model = 'zim'
|
||||
|
||||
|
||||
@register_modeler
|
||||
async def zim_modeler(progress_callback: Callable[[int], None] = None):
|
||||
total_processed = 0
|
||||
failed_ids: set = set()
|
||||
while True:
|
||||
with get_db_session(commit=True) as session:
|
||||
query = session.query(FileGroup, Zim) \
|
||||
.outerjoin(Zim, Zim.file_group_id == FileGroup.id) \
|
||||
.filter(FileGroup.primary_path.ilike('%.zim')) \
|
||||
.filter(
|
||||
# Exclusive-mimetype modeler: claim unmodeled rows even if apply_indexers
|
||||
# already set indexed=True, and re-model when files changed.
|
||||
or_(Zim.id.is_(None), FileGroup.indexed != True),
|
||||
)
|
||||
if failed_ids:
|
||||
query = query.filter(FileGroup.id.notin_(failed_ids))
|
||||
file_groups: List[Tuple[FileGroup, Zim]] = list(query.limit(10))
|
||||
|
||||
processed = 0
|
||||
for file_group, zim in file_groups:
|
||||
processed += 1
|
||||
|
||||
zim_id = None
|
||||
try:
|
||||
if not zim:
|
||||
zim = model_zim(file_group, session)
|
||||
zim.flush()
|
||||
else:
|
||||
zim_id = zim.id
|
||||
file_group.title = file_group.primary_path.name
|
||||
file_group.a_text = split_file_name_words(file_group.primary_path.name)
|
||||
file_group.indexed = True
|
||||
file_group.model = 'zim'
|
||||
except Exception as e:
|
||||
failed_ids.add(file_group.id)
|
||||
if PYTEST:
|
||||
raise
|
||||
logger.error(f'Unable to model Zim {zim_id=} {file_group.primary_path=}', exc_info=e)
|
||||
|
||||
logger.debug(f'Modeled {processed} zim files')
|
||||
|
||||
# Report batch progress
|
||||
total_processed += len(file_groups)
|
||||
if progress_callback and len(file_groups) > 0:
|
||||
progress_callback(total_processed)
|
||||
|
||||
if processed < 10:
|
||||
# Did not reach limit, do not query again.
|
||||
break
|
||||
|
||||
# Sleep to catch cancel.
|
||||
await asyncio.sleep(0)
|
||||
await run_modeler_loop(
|
||||
name='zim_modeler',
|
||||
batch_size=ZIM_PROCESSING_LIMIT,
|
||||
select_ids=_select_zim_ids,
|
||||
apply=_apply_zim,
|
||||
txn='batch',
|
||||
progress_callback=progress_callback,
|
||||
)
|
||||
|
||||
|
||||
def get_all_entries_tags():
|
||||
|
|
|
|||
|
|
@ -8,14 +8,11 @@ import shutil
|
|||
|
||||
import pytest
|
||||
|
||||
from modules.zim.lib import zim_modeler
|
||||
from modules.zim.lib import zim_modeler, ZIM_PROCESSING_LIMIT
|
||||
from modules.zim.models import Zim
|
||||
from wrolpi.files.models import FileGroup
|
||||
from wrolpi.vars import PROJECT_DIR
|
||||
|
||||
# The zim_modeler uses a hardcoded limit of 10
|
||||
ZIM_PROCESSING_LIMIT = 10
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_zim_modeler_processes_more_than_batch_limit(async_client, test_session, test_directory):
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue