mirror of
https://github.com/lrnselfreliance/wrolpi.git
synced 2026-10-26 19:01:39 -04:00
Migrating FileDownloader.
This commit is contained in:
parent
7b6fc60de5
commit
a2be7a3d8b
5 changed files with 170 additions and 27 deletions
|
|
@ -76,7 +76,8 @@ class ArchiveDownloader(Downloader, ABC):
|
|||
|
||||
return PreparedArchive(url=download.url, destination=destination, settings=settings)
|
||||
|
||||
async def execute_download(self, prepared: PreparedArchive, ctx: DownloadContext) -> ExecutedArchive:
|
||||
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."""
|
||||
if DOCKERIZED or PYTEST:
|
||||
|
|
|
|||
|
|
@ -436,9 +436,15 @@ class Downloader:
|
|||
"""
|
||||
return None
|
||||
|
||||
async def execute_download(self, prepared: Any, ctx: 'DownloadContext') -> Any:
|
||||
async def execute_download(self, prepared: Any, ctx: 'DownloadContext',
|
||||
download: Download = None) -> Any:
|
||||
"""Async I/O-bound work. No DB session — does network and subprocess work,
|
||||
returns plain data for finalize_download to persist."""
|
||||
returns plain data for finalize_download to persist.
|
||||
|
||||
download is passed through by the manager so subclasses that drive parent
|
||||
helpers (download_file, process_runner) can attribute logs/progress to the
|
||||
right download. It is optional so unit tests can construct a stub or omit it.
|
||||
"""
|
||||
raise NotImplementedError()
|
||||
|
||||
def finalize_download(self, session: Session, download: Download, executed: Any) -> DownloadResult:
|
||||
|
|
@ -571,7 +577,8 @@ class Downloader:
|
|||
|
||||
async def download_file(self, download: Download, url: str, destination: pathlib.Path,
|
||||
check_for_meta4: bool = True, concurrent: int = 3,
|
||||
meta4_xml: bytes = None) -> pathlib.Path:
|
||||
meta4_xml: bytes = None,
|
||||
ctx: 'DownloadContext' = None) -> pathlib.Path:
|
||||
from wrolpi.files.lib import glob_shared_stem
|
||||
|
||||
if not ARIA2C_PATH:
|
||||
|
|
@ -614,7 +621,8 @@ class Downloader:
|
|||
on_stdout = make_progress_callback(download.id, parse_aria2c_progress)
|
||||
|
||||
try:
|
||||
result = await self.process_runner(download, cmd, destination, stdout_callback=on_stdout)
|
||||
result = await self.process_runner(download, cmd, destination,
|
||||
stdout_callback=on_stdout, ctx=ctx)
|
||||
finally:
|
||||
clear_download_progress(download.id)
|
||||
stderr = result.stderr.decode() if getattr(result, 'stderr', None) else ''
|
||||
|
|
@ -1539,7 +1547,7 @@ async def signal_download_download(download_id: int, download_url: str):
|
|||
|
||||
if prepared is not None:
|
||||
# New phase-split path: caller (us) owns transaction boundaries.
|
||||
coro = downloader.execute_download(prepared, ctx)
|
||||
coro = downloader.execute_download(prepared, ctx, download=download)
|
||||
executed = await downloader.cancel_wrapper(coro, download, ctx=ctx)
|
||||
if isinstance(executed, DownloadResult):
|
||||
# cancel_wrapper short-circuited with a cancel result.
|
||||
|
|
|
|||
|
|
@ -1,19 +1,44 @@
|
|||
import pathlib
|
||||
import traceback
|
||||
from abc import ABC
|
||||
from dataclasses import dataclass
|
||||
from typing import Optional
|
||||
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from wrolpi.common import get_media_directory, logger
|
||||
from wrolpi.downloader import Downloader, Download, DownloadResult
|
||||
from wrolpi.downloader import Downloader, Download, DownloadContext, DownloadResult
|
||||
from wrolpi.errors import UnrecoverableDownloadError, IgnoredDirectoryError
|
||||
from wrolpi.files.lib import upsert_file, get_file_location_href
|
||||
from wrolpi.log_levels import TRACE_LEVEL
|
||||
from wrolpi.vars import PYTEST
|
||||
|
||||
__all__ = ['FileDownloader', 'file_downloader']
|
||||
__all__ = ['FileDownloader', 'PreparedFile', 'ExecutedFile', 'file_downloader']
|
||||
|
||||
logger = logger.getChild(__name__)
|
||||
|
||||
|
||||
@dataclass
|
||||
class PreparedFile:
|
||||
"""Plan produced by FileDownloader.prepare_download. Destination is absolute, has been
|
||||
bounded to the media directory, and exists on disk."""
|
||||
url: str
|
||||
destination: pathlib.Path
|
||||
|
||||
|
||||
@dataclass
|
||||
class ExecutedFile:
|
||||
"""Output of FileDownloader.execute_download.
|
||||
|
||||
`location` is set when the FileGroup was registered (or when the file landed in an
|
||||
ignored directory and we still want to expose a /files?... href). `error` carries
|
||||
the formatted traceback when the download or upsert failed in a recoverable way.
|
||||
"""
|
||||
output_path: pathlib.Path
|
||||
location: Optional[str] = None
|
||||
error: Optional[str] = None
|
||||
|
||||
|
||||
class FileDownloader(Downloader, ABC):
|
||||
"""A Downloader which can download an arbitrary HTTP file.
|
||||
|
||||
|
|
@ -26,8 +51,11 @@ class FileDownloader(Downloader, ABC):
|
|||
def __repr__(self):
|
||||
return '<FileDownloader>'
|
||||
|
||||
async def do_download(self, download: Download) -> DownloadResult:
|
||||
if download.attempts > 3:
|
||||
def prepare_download(self, session: Session, download: Download) -> PreparedFile:
|
||||
"""Validate attempt cap, resolve destination, ensure it sits inside the media
|
||||
directory, and create it if missing. session is unused — FileDownloader has no
|
||||
DB work in its prep or finalize phases."""
|
||||
if (download.attempts or 0) > 3:
|
||||
raise UnrecoverableDownloadError(f'Max download attempts reached for {download.url}')
|
||||
|
||||
settings = download.settings or dict()
|
||||
|
|
@ -47,31 +75,46 @@ class FileDownloader(Downloader, ABC):
|
|||
raise UnrecoverableDownloadError(f'Cannot download outside media directory!')
|
||||
destination.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
return PreparedFile(url=download.url, destination=destination)
|
||||
|
||||
async def execute_download(self, prepared: PreparedFile, ctx: DownloadContext,
|
||||
download: Download = None) -> ExecutedFile:
|
||||
"""Drive aria2c via the parent's download_file helper, threading ctx through so
|
||||
cancellation is observed by process_runner. Then register the resulting file via
|
||||
upsert_file.
|
||||
|
||||
Note: upsert_file (wrolpi/files/lib.py:1752) opens its own DB session because
|
||||
non-downloader callers (api.py:508, modules/zim/downloader.py:110) also use it.
|
||||
Splitting upsert_file is a separate future PR; for now finalize_download just
|
||||
translates ExecutedFile into a DownloadResult.
|
||||
"""
|
||||
# download_file reads .id (for progress) and .url (for log messages). Production
|
||||
# gets the real Download from the dispatch; tests may bypass and construct a stub.
|
||||
download = download if download is not None else Download(url=prepared.url)
|
||||
|
||||
output_path = None
|
||||
try:
|
||||
output_path = await self.download_file(download, download.url, destination)
|
||||
output_path = await self.download_file(download, prepared.url, prepared.destination, ctx=ctx)
|
||||
fg = await upsert_file(output_path, tag_names=download.tag_names)
|
||||
location = fg.location
|
||||
return DownloadResult(
|
||||
success=True,
|
||||
location=location,
|
||||
)
|
||||
return ExecutedFile(output_path=output_path, location=fg.location)
|
||||
except IgnoredDirectoryError:
|
||||
# `upsert_file` says this is an ignored file, that's fine.
|
||||
return DownloadResult(
|
||||
success=True,
|
||||
location=get_file_location_href(output_path)
|
||||
return ExecutedFile(
|
||||
output_path=output_path,
|
||||
location=get_file_location_href(output_path),
|
||||
)
|
||||
except Exception as e:
|
||||
logger.error(f'Failed to download {repr(str(download.url))}', exc_info=e)
|
||||
logger.error(f'Failed to download {repr(str(prepared.url))}', exc_info=e)
|
||||
if PYTEST:
|
||||
raise
|
||||
return ExecutedFile(output_path=output_path, error=str(traceback.format_exc()))
|
||||
|
||||
error = str(traceback.format_exc())
|
||||
return DownloadResult(
|
||||
success=False,
|
||||
error=error,
|
||||
)
|
||||
def finalize_download(self, session: Session, download: Download,
|
||||
executed: ExecutedFile) -> DownloadResult:
|
||||
"""Translate the ExecutedFile into a DownloadResult. No DB work; session is unused."""
|
||||
if executed.error:
|
||||
return DownloadResult(success=False, error=executed.error)
|
||||
return DownloadResult(success=True, location=executed.location)
|
||||
|
||||
|
||||
file_downloader = FileDownloader()
|
||||
|
|
|
|||
|
|
@ -2,10 +2,13 @@ import hashlib
|
|||
from http import HTTPStatus
|
||||
from unittest import mock
|
||||
|
||||
import pytest
|
||||
|
||||
from wrolpi import common
|
||||
from wrolpi.common import DownloadFileInfo
|
||||
from wrolpi.downloader import Download
|
||||
from wrolpi.files.downloader import FileDownloader
|
||||
from wrolpi.errors import UnrecoverableDownloadError
|
||||
from wrolpi.files.downloader import FileDownloader, PreparedFile, file_downloader
|
||||
from wrolpi.test.common import skip_circleci
|
||||
|
||||
|
||||
|
|
@ -190,3 +193,90 @@ async def test_file_downloader_meta4_no_output_flag(test_directory):
|
|||
assert '-o' not in captured_cmd
|
||||
# But -M should be present.
|
||||
assert '-M' in captured_cmd
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Phase-split unit tests for prepare_download.
|
||||
#
|
||||
# These exercise validation invariants in isolation — no async_client, no
|
||||
# test_download_manager, no aria2c. They prove FileDownloader's prep phase
|
||||
# rejects bad inputs cleanly without needing the Sanic stack.
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_prepare_max_attempts_raises(test_session, test_directory):
|
||||
"""prepare_download enforces the attempt cap before any I/O."""
|
||||
download = Download(
|
||||
url='https://example.com/file.zip',
|
||||
downloader='file',
|
||||
attempts=4,
|
||||
destination=str(test_directory / 'downloads'),
|
||||
)
|
||||
|
||||
with pytest.raises(UnrecoverableDownloadError):
|
||||
file_downloader.prepare_download(test_session, download)
|
||||
|
||||
|
||||
def test_prepare_requires_destination(test_session):
|
||||
"""No destination on the column or in settings → unrecoverable."""
|
||||
download = Download(
|
||||
url='https://example.com/file.zip',
|
||||
downloader='file',
|
||||
settings={},
|
||||
)
|
||||
|
||||
with pytest.raises(UnrecoverableDownloadError, match='without a destination'):
|
||||
file_downloader.prepare_download(test_session, download)
|
||||
|
||||
|
||||
def test_prepare_rejects_destination_outside_media_dir(test_session, test_directory, tmp_path):
|
||||
"""Destinations outside the media directory are rejected. test_directory sets up
|
||||
TEST_MEDIA_DIRECTORY; tmp_path lives outside it as a sibling temp dir."""
|
||||
outside = tmp_path / 'somewhere_else'
|
||||
outside.mkdir()
|
||||
# Sanity-check that tmp_path is genuinely outside the test media directory.
|
||||
assert not str(outside.resolve()).startswith(str(test_directory.resolve()))
|
||||
|
||||
download = Download(
|
||||
url='https://example.com/file.zip',
|
||||
downloader='file',
|
||||
destination=str(outside),
|
||||
)
|
||||
|
||||
with pytest.raises(UnrecoverableDownloadError, match='outside media directory'):
|
||||
file_downloader.prepare_download(test_session, download)
|
||||
|
||||
|
||||
def test_prepare_creates_destination_within_media_dir(test_session, test_directory):
|
||||
"""A destination inside the media dir that doesn't exist yet is created."""
|
||||
dest = test_directory / 'new_downloads_subdir'
|
||||
assert not dest.exists()
|
||||
|
||||
download = Download(
|
||||
url='https://example.com/file.zip',
|
||||
downloader='file',
|
||||
destination=str(dest),
|
||||
)
|
||||
|
||||
prepared = file_downloader.prepare_download(test_session, download)
|
||||
|
||||
assert isinstance(prepared, PreparedFile)
|
||||
assert prepared.url == 'https://example.com/file.zip'
|
||||
assert prepared.destination == dest.resolve()
|
||||
assert dest.is_dir()
|
||||
|
||||
|
||||
def test_prepare_uses_settings_destination_fallback(test_session, test_directory):
|
||||
"""When download.destination is empty, settings['destination'] is used."""
|
||||
dest = test_directory / 'from_settings'
|
||||
download = Download(
|
||||
url='https://example.com/file.zip',
|
||||
downloader='file',
|
||||
destination=None,
|
||||
settings={'destination': str(dest)},
|
||||
)
|
||||
|
||||
prepared = file_downloader.prepare_download(test_session, download)
|
||||
|
||||
assert prepared.destination == dest.resolve()
|
||||
assert dest.is_dir()
|
||||
|
|
|
|||
|
|
@ -100,7 +100,8 @@ class ScrapeHTMLDownloader(Downloader):
|
|||
destination=destination,
|
||||
)
|
||||
|
||||
async def execute_download(self, prepared: PreparedScrape, ctx: DownloadContext) -> ExecutedScrape:
|
||||
async def execute_download(self, prepared: PreparedScrape, ctx: DownloadContext,
|
||||
download: Download = None) -> ExecutedScrape:
|
||||
"""Crawl HTML pages up to depth, collecting links that match the configured suffix."""
|
||||
urls = [prepared.url]
|
||||
download_urls: List[str] = []
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue