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

Merge pull request #132 from lrnselfreliance/feature/rss-downloader

Browse source 
Feature/rss downloader
This commit is contained in:
lrnselfreliance 2022-06-08 12:26:22 -06:00 • committed by GitHub
commit afa515a8d9
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
23 changed files with 1525 additions and 746 deletions
Download patch file Download diff file Expand all files Collapse all files

29
alembic/versions/467235201c70_download_sub_downloader.py Normal file
View file

@ -0,0 +1,29 @@
"""Download.sub_downloader.
Revision ID: 467235201c70
Revises: 1a70196c05bb
Create Date: 2022-05-20 11:51:58.064020
"""
import os
from alembic import op
from sqlalchemy.orm import Session
# revision identifiers, used by Alembic.
revision = '467235201c70'
down_revision = '1a70196c05bb'
branch_labels = None
depends_on = None
DOCKERIZED = True if os.environ.get('DOCKER', '').lower().startswith('t') else False
def upgrade():
session = Session(bind=op.get_bind())
session.execute('ALTER TABLE download ADD COLUMN IF NOT EXISTS sub_downloader TEXT')
def downgrade():
session = Session(bind=op.get_bind())
session.execute('ALTER TABLE download DROP COLUMN IF EXISTS sub_downloader')

7
app/src/App.js
View file

@ -11,7 +11,7 @@ import 'react-semantic-toasts/styles/react-semantic-alert.css';
import {AppsRoute} from "./components/Apps";
import {InventoryRoute} from "./components/Inventory";
import {ArchiveRoute, ArchivesList} from "./components/Archive";
import {Saver} from "./components/Upload";
import {Downloads, Saver} from "./components/Upload";
import {MoreButton, PageContainer, SearchInput} from "./components/Common";
import {FilesRoute} from "./components/Files";
import {useSearch, useUp, useVersion} from "./hooks/customHooks";
@ -23,10 +23,7 @@ function Welcome() {
useUp();
let body = (
<>
<Header as='h2'>Save your media</Header>
<Saver/>
</>
<Downloads/>
);
if (searchStr) {

7
app/src/api.js
View file

@ -445,8 +445,11 @@ export async function getArchive(archiveId) {
}
}
export async function postDownload(urls, downloader) {
let body = {urls: urls, downloader: downloader};
export async function postDownload(urls, downloader, frequency, sub_downloader) {
let body = {urls: urls, downloader: downloader, frequency: frequency || null};
if (sub_downloader) {
body['sub_downloader'] = sub_downloader;
}
let response = await apiPost(`${API_URI}/download`, body);
return response;
}

33
app/src/components/Admin.js
View file

@ -3,7 +3,8 @@ import {
Button,
Checkbox,
Confirm,
Container, Dimmer,
Container,
Dimmer,
Divider,
Form,
Header,
@ -337,20 +338,46 @@ function ClearFailedDownloads({callback}) {
}
class DownloadRow extends React.Component {
constructor(props) {
super(props);
this.state = {
errorModalOpen: false,
}
}
render() {
let {url, frequency, last_successful_download, status, location, next_download} = this.props;
let {url, frequency, last_successful_download, status, location, next_download, error} = this.props;
const {errorModalOpen} = this.state;
let positive = false;
if (status === 'pending') {
positive = true;
}
const errorModal = <Modal
closeIcon
onClose={() => this.setState({errorModalOpen: false})}
onOpen={() => this.setState({errorModalOpen: true})}
open={errorModalOpen}
trigger={<Button icon='exclamation circle' color='red'/>}
>
<Modal.Header>Download Error</Modal.Header>
<Modal.Content>
<pre style={{overflowX: 'scroll'}}>{error}</pre>
</Modal.Content>
<Modal.Actions>
<Button onClick={() => this.setState({errorModalOpen: false})}>Close</Button>
</Modal.Actions>
</Modal>;
return (
<Table.Row positive={positive}>
<Table.Cell><a href={url} target='_blank'>{textEllipsis(url, 50)}</a></Table.Cell>
<Table.Cell>{secondsToFrequency(frequency)}</Table.Cell>
<Table.Cell>{last_successful_download ? secondsToDate(last_successful_download) : null}</Table.Cell>
<Table.Cell>{secondsToDate(next_download)}</Table.Cell>
<Table.Cell>{location && <a href={location}>View</a>}</Table.Cell>
<Table.Cell>{error && errorModal}{location && <a href={location}>View</a>}</Table.Cell>
</Table.Row>
);
}

16
app/src/components/Common.js
View file

@ -221,9 +221,21 @@ export const frequencyOptions = [
{key: '90days', text: '90 Days', value: 7776000},
];
export const rssFrequencyOptions = [
{key: 'once', text: 'Once', value: 0},
{key: 'hourly', text: 'Hourly', value: 3600},
{key: '3hours', text: '3 hours', value: 10800},
{key: '12hours', text: '12 hours', value: 43200},
{key: 'daily', text: 'Daily', value: 86400},
{key: 'weekly', text: 'Weekly', value: 604800},
{key: 'biweekly', text: 'Biweekly', value: 1209600},
{key: '30days', text: '30 Days', value: 2592000},
{key: '90days', text: '90 Days', value: 7776000},
];
export function secondsToFrequency(seconds) {
for (let i = 0; i < Object.keys(frequencyOptions).length; i++) {
let d = frequencyOptions[i];
for (let i = 0; i < Object.keys(rssFrequencyOptions).length; i++) {
let d = rssFrequencyOptions[i];
if (d.value === seconds) {
return d.text;
}

287
app/src/components/Upload.js
View file

@ -1,8 +1,9 @@
import React from "react";
import {Button, Dropdown, Form, Header, Segment, TextArea} from "semantic-ui-react";
import React, {useState} from "react";
import {Button, Form, Header, Loader, TextArea} from "semantic-ui-react";
import {getDownloaders, postDownload} from "../api";
import {WROLModeMessage} from "./Common";
import {Link} from "react-router-dom";
import {frequencyOptions, rssFrequencyOptions, WROLModeMessage} from "./Common";
import Icon from "semantic-ui-react/dist/commonjs/elements/Icon";
import Message from "semantic-ui-react/dist/commonjs/collections/Message";
const validUrl = /^(http|https):\/\/[^ "]+$/;
@ -13,23 +14,10 @@ class Downloader extends React.Component {
urls: '',
valid: true,
pending: false,
downloaders: [],
downloader: null,
downloader: props.downloader,
};
}
async componentDidMount() {
await this.fetchDownloaders();
}
fetchDownloaders = async () => {
let {downloaders} = await getDownloaders();
downloaders = downloaders.map((i) => {
return {key: i.name, text: i.pretty_name || i.name, value: i.name}
})
this.setState({downloaders});
}
submitDownload = async () => {
let {urls, downloader} = this.state;
if (urls) {
@ -64,46 +52,265 @@ class Downloader extends React.Component {
}
render() {
let textareaPlaceholder = 'Enter one URL per line';
let disabled = !this.state.urls || !this.state.valid || this.state.pending || !this.state.downloader;
const {header} = this.props;
return (
<Form onSubmit={this.submitDownload}>
<WROLModeMessage content='Downloading is disabled while WROL Mode is enabled'/>
<Header as='h4'>Enter the URLs you wish to save offline.</Header>
<Header as='h3'>{header}</Header>
<Form.Field>
<TextArea
placeholder={textareaPlaceholder}
name='urls'
onChange={this.handleInputChange}
value={this.state.urls}
error={!this.state.valid}
/>
</Form.Field>
<Form.Field>
<Dropdown selection
name='downloader'
options={this.state.downloaders}
placeholder='Select a downloader'
<TextArea required
placeholder={'Enter one URL per line'}
name='urls'
onChange={this.handleInputChange}
value={this.state.urls}
/>
</Form.Field>
<Button content='Cancel' onClick={this.props.clearSelected}/>
<Button primary style={{marginTop: '1em'}} disabled={disabled}>Download</Button>
<Link to={'/admin'}>View Downloads</Link>
</Form>
)
}
}
export class Saver extends React.Component {
class ChannelDownload extends React.Component {
constructor(props) {
super(props);
this.state = {
url: '',
frequency: 604800,
pending: false,
disabled: false,
ready: false,
error: null,
success: null,
};
this.freqOptions = [
{key: 'once', text: 'Once', value: 0},
...frequencyOptions,
];
this.handleUrlChange = this.handleUrlChange.bind(this);
this.handleSubmit = this.handleSubmit.bind(this);
this.handleFrequencyChange = this.handleFrequencyChange.bind(this);
}
handleUrlChange = (e, {value}) => {
e.preventDefault();
let ready = value && validUrl.test(value);
this.setState({url: value, ready, success: false});
}
handleFrequencyChange(e, {value}) {
this.setState({frequency: value, success: false});
}
handleSubmit = async (e) => {
e.preventDefault();
this.setState({disabled: true, pending: true, success: null, error: null});
const {url, frequency} = this.state;
if (!url) {
this.setState({error: 'URL is required'});
return;
}
let response = await postDownload(url, 'video_channel', frequency);
if (response.status === 204) {
this.setState({pending: false, disabled: false, success: true, url: '', ready: false});
} else {
const content = await response.json();
let error = content.message || null;
this.setState({pending: false, disabled: false, success: false, error});
}
}
render() {
const {ready, disabled, url, error, frequency, pending, success} = this.state;
const buttonDisabled = !ready || disabled;
const onceMessage = (<Message>
<Message.Header>Download Once</Message.Header>
<Message.Content>You have selected a frequency of Once, this is useful when you want to download
all videos in a Playlist, and when you do not want to download any videos added to the playlist
in the future.</Message.Content>
</Message>);
return (
<>
<Segment>
<Downloader/>
</Segment>
</>
<Form>
<Header as='h4'><Icon name='video'/> Channel / Playlist</Header>
<Form.Input
required
label='URL'
placeholder='https://example.com/channel/videos'
value={url}
error={error}
onChange={this.handleUrlChange}
/>
<Form.Dropdown
required
selection
label='Download Frequency'
name='download_frequency'
placeholder='Frequency'
options={this.freqOptions}
value={frequency}
selected={frequency}
onChange={this.handleFrequencyChange}
/>
{frequency === 0 && onceMessage}
<Button content='Cancel' onClick={this.props.clearSelected}/>
<Button
color='blue'
content='Download'
onClick={this.handleSubmit}
disabled={buttonDisabled}
/>
<Loader active={pending}/>
{success && <Icon name='check'/>}
</Form>
)
}
}
class RSSDownload extends ChannelDownload {
constructor(props) {
super(props);
this.state = {
url: '',
frequency: 604800,
pending: false,
disabled: false,
ready: false,
error: null,
success: null,
downloaders: [],
downloader: null,
sub_downloader: null,
};
this.freqOptions = rssFrequencyOptions;
}
async componentDidMount() {
await this.fetchDownloaders();
}
fetchDownloaders = async () => {
let {downloaders} = await getDownloaders();
downloaders = downloaders.map((i) => {
return {key: i.name, text: i.pretty_name || i.name, value: i.name}
})
this.setState({downloaders});
}
handleInputChange = async (event, {name, value}) => {
this.setState({[name]: value});
}
handleSubmit = async (e) => {
e.preventDefault();
this.setState({disabled: true, pending: true, success: null, error: null});
const {url, frequency, sub_downloader} = this.state;
if (!url) {
this.setState({error: 'URL is required'});
return;
}
let response = await postDownload(url, 'rss', frequency, sub_downloader);
if (response.status === 204) {
this.setState({pending: false, disabled: false, success: true, url: '', ready: false});
} else {
const content = await response.json();
let error = content.message || null;
this.setState({pending: false, disabled: false, success: false, error});
}
}
render() {
const {ready, disabled, url, error, frequency, pending, success} = this.state;
const buttonDisabled = !ready || disabled;
const onceMessage = (<Message>
<Message.Header>Download Once</Message.Header>
<Message.Content>You have selected a frequency of Once, this is useful when you want to download
all videos in a Playlist, but you do not want to download any videos added to the playlist
in the future.</Message.Content>
</Message>);
return (
<Form>
<Header as='h4'><Icon name='rss'/> RSS Feed</Header>
<Form.Input
required
label='URL'
placeholder='https://example.com/feed'
value={url}
error={error}
onChange={this.handleUrlChange}
/>
<Form.Dropdown
required
selection
label='Download Frequency'
name='download_frequency'
placeholder='Frequency'
options={this.freqOptions}
value={frequency}
selected={frequency}
onChange={this.handleFrequencyChange}
/>
{frequency === 0 && onceMessage}
<Form.Dropdown selection required
name='sub_downloader'
label='Downloader'
options={this.state.downloaders}
placeholder='Select a downloader'
onChange={this.handleInputChange}
/>
<Button content='Cancel' onClick={this.props.clearSelected}/>
<Button
color='blue'
content='Download'
onClick={this.handleSubmit}
disabled={buttonDisabled}
/>
<Loader active={pending}/>
{success && <Icon name='check'/>}
</Form>
)
}
}
export function Downloads() {
const [downloader, setDownloader] = useState();
let body = (<>
<Button content='Videos' onClick={() => setDownloader('video')}/>
<Button content='Archive' onClick={() => setDownloader('archive')}/>
<Button content='Channel/Playlist' onClick={() => setDownloader('video_channel')}/>
<Button content='RSS Feed' onClick={() => setDownloader('rss')}/>
</>);
function clearSelected() {
setDownloader(null);
body = null;
}
const downloaders = {
'archive': <Downloader clearSelected={clearSelected} header={<><Icon name='file alternate'/> Archive</>}
downloader='archive'/>,
'video': <Downloader clearSelected={clearSelected} header={<><Icon name='video'/> Videos</>}
downloader='video'/>,
'video_channel': <ChannelDownload clearSelected={clearSelected}/>,
'rss': <RSSDownload clearSelected={clearSelected} header={<><Icon name='rss square'/> RSS</>}/>,
};
body = downloader in downloaders ? downloaders[downloader] : body;
return (
<>
<Header as='h2'>Download</Header>
{body}
</>
)
}

11
modules/archive/__init__.py
View file

@ -1,11 +1,15 @@
from abc import ABC
from typing import Tuple
from typing import Tuple, Optional
from sqlalchemy.orm import Session
from wrolpi.common import logger
from wrolpi.db import optional_session
from wrolpi.downloader import Downloader, Download, DownloadResult
from wrolpi.errors import UnrecoverableDownloadError
from . import lib
from .api import bp # noqa
from .models import Archive
PRETTY_NAME = 'Archive'
@ -33,6 +37,11 @@ class ArchiveDownloader(Downloader, ABC):
archive = lib.do_archive(download.url)
return DownloadResult(success=True, location=f'/archive/{archive.id}')
@optional_session
def already_downloaded(self, url: str, session: Session = None) -> Optional[Archive]:
archive = session.query(Archive).filter_by(url=url).one_or_none()
return archive
# Archive downloader is the last downloader which should be used.
archive_downloader = ArchiveDownloader(priority=100)

908
modules/map/leaflet.css
View file

File diff suppressed because it is too large Load diff

1
modules/videos/channel/lib.py
View file

@ -100,6 +100,7 @@ def get_channel(session: Session, *, channel_id: int = None, source_id: str = No
raise UnknownChannel(f'No channel matches {channel_id=} {source_id=} {url=} {directory=}')
logger.debug(f'Found {channel=} using {channel_id=} {source_id=} {url=} {directory=}')
session.refresh(channel)
if return_dict:
statistics = channel.get_statistics()
channel = channel.dict()

7
modules/videos/channel/test/test_api.py
View file

@ -337,9 +337,7 @@ def test_channel_empty_url_doesnt_conflict(test_client, test_session, test_direc
def test_download_channel_no_refresh(test_session, download_channel, video_download_manager):
"""
A Channel cannot be downloaded until it has been refreshed.
"""
"""A Channel cannot be downloaded until it has been refreshed."""
def check_refreshed(expected: bool):
channel = test_session.query(Channel).one()
@ -349,7 +347,8 @@ def test_download_channel_no_refresh(test_session, download_channel, video_downl
test_session.commit()
with mock.patch('modules.videos.downloader.YDL.extract_info') as mock_extract_info:
mock_extract_info.return_value = {'entries': [], 'url': 'foo'}
mock_extract_info.return_value = {'entries': [], 'url': 'foo', 'uploader': 'some uploader',
'channel_id': 'the id', 'id': 'the id'}
video_download_manager.do_downloads_sync()
check_refreshed(True)

181
modules/videos/downloader.py
View file

@ -16,12 +16,13 @@ from wrolpi.cmd import which
from wrolpi.common import logger, extract_domain, get_media_directory
from wrolpi.dates import now
from wrolpi.db import get_db_session, get_db_curs
from wrolpi.db import optional_session
from wrolpi.downloader import Downloader, Download, DOWNLOAD_MANAGER_CONFIG, DownloadResult
from wrolpi.errors import UnknownChannel, ChannelURLEmpty, UnrecoverableDownloadError
from wrolpi.vars import PYTEST
from .channel.lib import create_channel, get_channel
from .common import apply_info_json, get_no_channel_directory, get_videos_directory
from .lib import upsert_video, refresh_channel_videos, get_downloader_config, get_channel_source_id
from .lib import upsert_video, refresh_channel_videos, get_downloader_config
from .models import Video, Channel
from .schema import ChannelPostRequest
from .video_url_resolver import video_url_resolver
@ -39,7 +40,13 @@ ChannelIEs = {
}
PREFERRED_VIDEO_EXTENSION = 'mp4'
PREFERRED_VIDEO_FORMAT = 'best[height=720],22,720p,mp4-480p,mp4-360p,mp4-240p,18,bestvideo+bestaudio'
PREFERRED_VIDEO_FORMAT = ','.join([
'res:720', # Use the best 720p available first
'136+140', # 130=720p video-only, 140= high quality audio only
'22', # 720p video with audio
'18', # 360p video with audio
'bestvideo+bestaudio', # Download the highest resolution as a last resort (1080p is quite large).
])
def extract_info(url: str, ydl: YoutubeDL = YDL, process=False) -> dict:
@ -53,7 +60,7 @@ def prepare_filename(entry: dict, ydl: YoutubeDL = YDL) -> str:
class ChannelDownloader(Downloader, ABC):
"""Handles downloading of all Videos in a Channel."""
"""Handles downloading of videos in a Channel or Playlist."""
name = 'video_channel'
pretty_name = 'Video Channel'
listable = False
@ -69,41 +76,84 @@ class ChannelDownloader(Downloader, ABC):
logger.debug(f'{cls.__name__} not suitable for {url}')
return False, None
@staticmethod
def is_a_playlist(info: dict):
# A playlist will have an id different from its channel.
return info['id'] != info['channel_id']
def do_download(self, download: Download) -> DownloadResult:
"""Update a Channel's catalog, then schedule downloads of every missing video."""
with get_db_session() as session:
channel = session.query(Channel).filter_by(url=download.url).one_or_none()
info = extract_info(download.url, process=False)
if not channel:
# Couldn't get channel by URL, this is probably a playlist. Find the channel for the playlist.
channel_source_id = get_channel_source_id(download.url)
channel = get_channel(source_id=channel_source_id, return_dict=False)
channel_id = channel.id
name = info['uploader']
channel_source_id = info['channel_id']
channel = get_or_create_channel(channel_source_id, download.url, name)
channel.dict() # get all attributes while we have the session.
# Resolve all entries.
info['entries'] = list(info['entries'])
if self.is_a_playlist(info):
result = self.upsert_playlist_videos(download, channel)
else:
result = self.upsert_channel_videos(download, info, channel)
if PYTEST:
self.manager.do_downloads_sync()
else:
self.manager.start_downloads()
return result
@staticmethod
def upsert_channel_videos(download: Download, info: dict, channel: Channel) -> DownloadResult:
"""Get a list of all videos in a Channel, schedule downloads for any missing videos."""
try:
update_channel(channel)
update_channel_catalog(channel, info)
domain = extract_domain(download.url)
missing_videos = find_all_missing_videos(channel.id)
with get_db_session(commit=True) as session:
for video_id, source_id, missing_video in missing_videos:
url = video_url_resolver(domain, missing_video)
if url in DOWNLOAD_MANAGER_CONFIG.skip_urls:
# Don't try to download skipped videos.
continue
# Schedule any missing videos for download.
download = self.manager.get_or_create_download(url, session)
download.downloader = 'video'
if PYTEST:
self.manager.do_downloads_sync()
else:
self.manager.start_downloads()
return DownloadResult(success=True, location=f'/videos/channel/{channel_id}/video')
downloads = []
for video_id, source_id, missing_video in missing_videos:
url = missing_video.get('webpage_url') or video_url_resolver(domain, missing_video)
downloads.append(url)
return DownloadResult(success=True, location=f'/videos/channel/{channel.id}/video', downloads=downloads)
except Exception:
logger.warning(f'Failed to update catalog of channel {download.url}')
location = f'/videos/channel/{channel_id}/video' if channel_id else None
location = f'/videos/channel/{channel.id}/video' if channel.id else None
return DownloadResult(success=False, location=location, error=str(traceback.format_exc()))
@staticmethod
def upsert_playlist_videos(download: Download, channel: Channel) -> DownloadResult:
"""Get a list of all videos in a playlist, schedule downloads for any missing videos."""
try:
# Playlists require more information to download.
info = extract_info(download.url, process=True)
# Get all Video.source_id of this Channel.
with get_db_session() as session:
existing_ids = [i for (i,) in session.query(Video.source_id).filter(Video.channel_id == channel.id)]
# 'id' is our 'source_id'
if any(i['id'] not in existing_ids for i in info['entries']):
# At least one video in the playlist is not in the Channel catalog, update the catalog.
channel_info = extract_info(channel.url, process=True)
update_channel_catalog(channel, channel_info)
with get_db_session() as session:
channel_video_urls = [i for (i,) in session.query(Video.url).filter(Video.channel_id == channel.id,
Video.url != None)]
skip_urls = set(DOWNLOAD_MANAGER_CONFIG.skip_urls)
# All videos in the channel catalog that have not been downloaded (most will not be in the playlist).
missing_source_ids = {source_id for _, source_id, _ in _find_all_missing_videos(channel.id)}
downloads = [i['webpage_url'] for i in info['entries'] if
i['id'] in missing_source_ids and i['webpage_url'] not in skip_urls]
# A video may be in a playlist, but not published in the channel catalog.
downloads.extend(i['webpage_url'] for i in info['entries'] if i['webpage_url'] not in channel_video_urls)
return DownloadResult(success=True, location=f'/videos/channel/{channel.id}/video', downloads=downloads)
except Exception:
logger.warning(f'Failed to update catalog of playlist {download.url}')
location = f'/videos/channel/{channel.id}/video' if channel.id else None
return DownloadResult(success=False, location=location, error=str(traceback.format_exc()))
@ -125,11 +175,14 @@ class VideoDownloader(Downloader, ABC):
def __repr__(self):
return f'<VideoDownloader>'
@optional_session
def already_downloaded(self, url: str, session: Session = None) -> Optional[Video]:
video = session.query(Video).filter_by(url=url).one_or_none()
return video
@classmethod
def valid_url(cls, url) -> Tuple[bool, Optional[dict]]:
"""
Match against all Youtube-DL Info Extractors, except those that match a Channel.
"""
"""Match against all Youtube-DL Info Extractors, except those that match a Channel."""
for ie in YDL._ies.values():
if ie.suitable(url) and not ChannelDownloader.valid_url(url)[0]:
try:
@ -307,25 +360,22 @@ def get_or_create_channel(source_id: str = None, url: str = None, name: str = No
return channel
def update_channel(channel: Channel):
def update_channel_catalog(channel: Channel, info: dict):
"""
Connect to the Channel's host website and pull a catalog of all videos. Insert any new videos into the DB.
It is expected that any missing videos will be downloaded later.
"""
logger.info(f'Downloading video list for {channel.name} at {channel.url} This may take several minutes.')
info = extract_info(channel.url)
if 'url' in info:
url = info['url']
info = extract_info(url)
# Resolve all entries to dictionaries.
entries = info['entries'] = list(info['entries'])
entries = info['entries']
# yt-dlp may hand back a list of URLs, lets use the "Uploads" URL, if available.
try:
entries[0]['id']
except Exception:
logger.warning('yt-dlp did not return a list of URLs')
for entry in entries:
if entry['title'] == 'Uploads':
logger.info('Youtube-DL gave back a list of URLs, found the "Uploads" URL and using it.')
@ -376,19 +426,9 @@ def update_channel(channel: Channel):
apply_info_json(channel_id)
def _find_all_missing_videos(channel_id: id = None) -> List[Tuple]:
"""Get all Video entries which don't have the required media files (i.e. hasn't been downloaded).
Restrict to a single channel if `channel_id` is provided.
"""
def _find_all_missing_videos(channel_id: id) -> List[Tuple]:
"""Get all Video entries which don't have the required media files (i.e. hasn't been downloaded)."""
with get_db_curs() as curs:
# Get all channels by default.
where = ''
params = ()
if channel_id:
where = 'AND channel_id = %s'
params = (channel_id,)
query = f'''
SELECT
video.id, video.source_id, video.channel_id
@ -399,11 +439,11 @@ def _find_all_missing_videos(channel_id: id = None) -> List[Tuple]:
channel.url IS NOT NULL
AND channel.url != ''
AND video.source_id IS NOT NULL
{where}
AND channel_id = %s
AND video.channel_id IS NOT NULL
AND (video_path IS NULL OR video_path = '' OR poster_path IS NULL OR poster_path = '')
'''
logger.debug(query)
params = (channel_id,)
curs.execute(query, params)
missing_videos = list(curs.fetchall())
return missing_videos
@ -415,50 +455,33 @@ def find_all_missing_videos(channel_id: int = None) -> Tuple[dict, dict]:
Yields a Channel Dict object, our Video id, and the "entry" of the video from the channel's info_json['entries'].
"""
with get_db_session() as session:
if channel_id:
channel = get_channel(channel_id=channel_id, return_dict=False)
if not channel.url:
raise ChannelURLEmpty('No URL for this channel')
channels = [channel, ]
else:
channels = session.query(Channel).filter(Channel.info_json != None).all() # noqa
channel: Channel = get_channel(channel_id=channel_id, return_dict=False)
if not channel.url:
raise ChannelURLEmpty('No URL for this channel')
# Get all channels while in the db context.
channels = list(channels)
# Check that the channel has some videos. We can't be sure what is missing if we don't know what we have.
if channel.refreshed is False:
# Refresh this channel's videos.
refresh_channel_videos(channel)
# Check that each channel has some videos. We can't be sure what is missing if we don't know what we have.
for channel in channels:
if channel.refreshed is False:
# Refresh this channel's videos.
refresh_channel_videos(channel)
channels = {i.id: i for i in channels}
match_regexen = {i: re.compile(j.match_regex) for i, j in channels.items() if j.match_regex}
match_regex = re.compile(channel.match_regex) if channel.match_regex else None
# Convert the channel video entries into a form that allows them to be quickly retrieved without searching through
# the entire entries list.
channels_entries = {}
for channel_id_, channel in channels.items():
channels_entries[channel_id_] = {i['id']: i for i in channel.info_json['entries']}
channel_entries = {i['id']: i for i in channel.info_json['entries']}
# Yield all videos not skipped.
missing_videos = _find_all_missing_videos(channel_id)
for video_id, source_id, channel_id in missing_videos:
channel = channels[channel_id]
if channel.skip_download_videos and source_id in channel.skip_download_videos:
# This video has been marked to skip.
continue
try:
missing_video = channels_entries[channel_id][source_id]
missing_video = channel_entries[source_id]
except KeyError:
logger.warning(f'Video {channel.name} / {source_id} is not in {channel.name} info_json')
continue
match_regex: re.compile = match_regexen.get(channel_id)
if not match_regex or (match_regex and missing_video['title'] and match_regex.match(missing_video['title'])):
# No title match regex, or the title matches the regex.
yield video_id, source_id, missing_video

21
modules/videos/lib.py
View file

@ -153,14 +153,7 @@ def validate_videos():
with get_db_session(commit=True) as session:
videos = session.query(Video).filter(Video.id.in_(chunk)).all()
for video in videos:
try:
channel_generate_poster = channel_generate_posters.get(video.channel_id)
validate_video(video, channel_generate_poster)
# All data about the Video has been found, we should not attempt to validate it again.
video.validated = True
except Exception as e:
# This video failed to validate, continue validation for the rest of the videos.
logger.warning(f'Failed to validate {video}', exc_info=e)
video.validate()
def validate_video(video: Video, channel_generate_poster: bool):
@ -605,7 +598,7 @@ YDL.add_default_info_extractors()
def get_channel_source_id(url: str) -> str:
channel_info = YDL.extract_info(url, download=False, process=False)
return channel_info.get('uploader_id') or channel_info['id']
return channel_info.get('uploader_id') or channel_info['channel_id']
async def get_statistics():
@ -747,15 +740,7 @@ def upsert_video(session: Session, video_path: pathlib.Path, channel: Channel =
session.flush()
session.refresh(video)
try:
# Fill in any missing data. Generate poster if enabled and necessary.
validate_video(video, channel.generate_posters if channel else False)
# All data about the Video has been found, we should not attempt to validate it again.
video.validated = True
except Exception as e:
# Could not validate, this could be an issue with a file. This should not prevent the video from being
# inserted.
logger.warning(f'Failed to validate {video}', exc_info=e)
video.validate()
session.flush()

37
modules/videos/models.py
View file

@ -55,8 +55,8 @@ class Video(ModelHelper, Base):
video_path = self.video_path
if self.video_path and isinstance(self.video_path, MediaPath):
video_path = self.video_path.path.relative_to(get_media_directory())
return f'<Video id={self.id} title={self.title} path={video_path} channel={self.channel_id} ' \
f'source_id={self.source_id}>'
return f'<Video id={self.id} title={repr(self.title)} path={video_path} channel={self.channel_id} ' \
f'source_id={repr(self.source_id)}>'
def dict(self) -> dict:
d = super().dict()
@ -255,6 +255,29 @@ class Video(ModelHelper, Base):
)
return d
def validate(self):
"""Perform a validation of this video and it's files. Mark this video as validated if no errors occur."""
if not self.video_path:
# Can't validate if there is no video file.
return False
from .lib import validate_video
try:
validate_video(self, self.channel.generate_posters if self.channel else False)
self.validated = True
except Exception as e:
logger.warning(f'Failed to validate video {self}', exc_info=e)
return self.validated
def check_for_corruption(self):
"""Uses ffprobe to check if this video is corrupt."""
if not self.video_path:
return
from .lib import check_for_video_corruption
return check_for_video_corruption(self.video_path.path)
class Channel(ModelHelper, Base):
__tablename__ = 'channel'
@ -279,6 +302,11 @@ class Channel(ModelHelper, Base):
def __repr__(self):
return f'<Channel id={self.id}, name={repr(self.name)}>'
def __eq__(self, other):
if isinstance(other, Channel):
return self.id == other.id
return False
def add_video_to_skip_list(self, source_id: str):
if not source_id:
raise UnknownVideo(f'Cannot skip video with empty source id: {source_id}')
@ -317,12 +345,13 @@ class Channel(ModelHelper, Base):
download.frequency = self.download_frequency
download.url = self.url
# Keep next_download if available.
download.next_download = download.next_download or download_manager.get_next_download(download, session)
download.next_download = download.next_download or download_manager.calculate_next_download(download,
session)
elif not download and self.download_frequency and self.url:
download = Download(frequency=self.download_frequency, url=self.url, downloader='video_channel')
session.add(download)
session.flush()
download.next_download = download_manager.get_next_download(download, session)
download.next_download = download_manager.calculate_next_download(download, session)
session.flush()
def config_view(self) -> dict:

111
modules/videos/test/test_downloader.py
View file

@ -9,10 +9,10 @@ from yt_dlp.utils import UnsupportedError
from modules.videos.channel.lib import download_channel
from modules.videos.downloader import find_all_missing_videos, VideoDownloader, \
ChannelDownloader, get_or_create_channel
ChannelDownloader, get_or_create_channel, channel_downloader
from modules.videos.models import Channel, Video
from wrolpi.db import get_db_context
from wrolpi.downloader import DownloadManager, Download
from wrolpi.downloader import DownloadManager, Download, DownloadResult
from wrolpi.errors import InvalidDownload
from wrolpi.test.common import TestAPI
from wrolpi.test.test_downloader import HTTPDownloader
@ -20,7 +20,7 @@ from wrolpi.vars import PROJECT_DIR
def test_find_all_missing_videos(test_session, channel_factory, video_factory):
channel1, channel2 = channel_factory(url='some url'), channel_factory(url='some other url')
channel1 = channel_factory(url='some url')
channel1.info_json = {'entries': [{'id': 'foo', 'view_count': 0}]}
test_session.commit()
@ -29,14 +29,14 @@ def test_find_all_missing_videos(test_session, channel_factory, video_factory):
video_factory()
# No missing videos
assert [] == list(find_all_missing_videos())
assert [] == list(find_all_missing_videos(channel1.id))
# Create a video that has no video file.
video = Video(title='needs to be downloaded', channel_id=channel1.id, source_id='foo')
test_session.add(video)
test_session.commit()
missing = list(find_all_missing_videos())
missing = list(find_all_missing_videos(channel1.id))
assert len(missing) == 1
# The video returned is the one we faked.
@ -73,6 +73,7 @@ example_video_json = {
example_channel_json = {
'_type': 'playlist',
'channel_id': 'some id',
'entries': [
{'_type': 'url',
'description': None,
@ -82,7 +83,8 @@ example_channel_json = {
'title': 'video 1 title',
'uploader': None,
'url': 'video_1_url',
'view_count': 58504},
'view_count': 58504,
'webpage_url': 'https://youtube.com/watch?v=video_1_url'},
{'_type': 'url',
'description': None,
'duration': None,
@ -91,12 +93,14 @@ example_channel_json = {
'title': 'video 2 title',
'uploader': None,
'url': 'video_2_url',
'view_count': 1413},
'view_count': 1413,
'webpage_url': 'https://youtube.com/watch?v=video_2_url'},
],
'extractor': 'youtube:tab',
'extractor_key': 'YoutubeTab',
'id': 'some id',
'title': 'channel title',
'uploader': 'the uploader',
'webpage_url': 'channel url',
'webpage_url_basename': 'videos',
}
@ -173,7 +177,8 @@ def test_video_download_no_channel(test_session, video_download_manager, video_f
def test_download_channel(test_session, simple_channel, video_download_manager, video_file,
mock_video_extract_info, mock_video_prepare_filename, mock_video_process_runner):
mock_video_extract_info, mock_video_prepare_filename,
mock_video_process_runner):
"""Downloading (updating the catalog of) a Channel creates download records for all of it's missing videos."""
url = 'https://www.youtube.com/c/LearningSelfReliance/videos'
@ -385,3 +390,93 @@ def test_download_result(test_session, test_directory, video_download_manager, m
download: Download = test_session.query(Download).one()
assert download.url == 'https://example.com'
assert download.location == '/videos/channel/1/video/1'
example_playlist_json = {
'_type': 'playlist',
'availability': None,
'channel': 'the channel name',
'channel_follower_count': None,
'channel_id': 'the channel id',
'channel_url': 'channel url',
'description': '',
'entries': [
{'_type': 'url',
'description': None,
'duration': None,
'id': 'video 2 id',
'ie_key': 'Youtube',
'title': 'video 2 title',
'uploader': None,
'url': 'video_2_url',
'view_count': 1413,
'webpage_url': 'https://youtube.com/watch?v=video_2_url'},
{'_type': 'url',
'description': None,
'duration': None,
'id': 'video 1 id',
'ie_key': 'Youtube',
'title': 'video 1 title',
'uploader': None,
'url': 'video_1_url',
'view_count': 58504,
'webpage_url': 'https://youtube.com/watch?v=video_1_url'},
{'_type': 'url',
'description': None,
'duration': None,
'id': 'video 3 id',
'ie_key': 'Youtube',
'title': 'video 3 title',
'uploader': None,
'url': 'video_3_url',
'view_count': 58504,
'webpage_url': 'https://youtube.com/watch?v=video_3_url'},
],
'extractor': 'youtube:tab',
'extractor_key': 'YoutubeTab',
'id': 'the playlist id',
'modified_date': '20220426',
'original_url': 'original url',
'playlist_count': 10,
'requested_entries': None,
'tags': [],
'title': 'some title',
'uploader': 'Playlist Uploader',
'uploader_id': 'uploader id',
'uploader_url': 'uploader url',
'view_count': 22298,
'webpage_url': 'webpage url',
'webpage_url_basename': 'playlist',
'webpage_url_domain': 'youtube.com',
}
def test_download_playlist(test_session, test_directory, mock_video_extract_info, video_download_manager):
"""
All videos in a playlist can be downloaded for it's Channel.
"""
download = Download(url='playlist url')
test_session.add(download)
channel = get_or_create_channel(example_playlist_json['channel_id'], download.url, example_channel_json['uploader'])
test_session.add(Video(url='https://youtube.com/watch?v=video_1_url',
source_id='video 1 id', channel_id=channel.id, video_path='video exists',
poster_path='poster exists'))
test_session.commit()
mock_video_extract_info.side_effect = [
example_playlist_json, # Playlist info is fetched first.
example_playlist_json, # Playlist info is fetched again for more details.
example_channel_json, # Video 3 causes channel catalog to be fetched.
example_video_json,
example_video_json,
example_video_json,
]
with mock.patch('modules.videos.downloader.VideoDownloader.do_download') as mock_video_do_download:
mock_video_do_download.return_value = DownloadResult(success=True) # Don't download the videos.
result = channel_downloader.do_download(download)
assert result.success is True, 'Download was not successful'
assert set(result.downloads) == {
'https://youtube.com/watch?v=video_2_url',
'https://youtube.com/watch?v=video_3_url',
}

1
requirements.txt
View file

@ -1,6 +1,7 @@
alembic==1.7.5
beautifulsoup4==4.10.0
cachetools==5.0.0
feedparser==6.0.9
mock==4.0.3
Pillow==9.0.1
Pint==0.18

13
wrolpi/conftest.py
View file

@ -15,7 +15,7 @@ from sqlalchemy.orm import Session, sessionmaker
from wrolpi.common import set_test_media_directory, Base, set_test_config
from wrolpi.dates import set_test_now
from wrolpi.db import postgres_engine, get_db_args
from wrolpi.downloader import DownloadManager, DownloadResult
from wrolpi.downloader import DownloadManager, DownloadResult, set_test_download_manager_config
from wrolpi.root_api import BLUEPRINTS, api_app
@ -112,7 +112,16 @@ def test_client() -> SanicTestClient:
@pytest.fixture
def test_download_manager():
def test_download_manager_config(test_directory):
(test_directory / 'config').mkdir(exist_ok=True)
config_path = test_directory / 'config/download_manager.yaml'
set_test_download_manager_config(True)
yield config_path
set_test_download_manager_config(False)
@pytest.fixture
def test_download_manager(test_download_manager_config):
manager = DownloadManager()
return manager

201
wrolpi/downloader.py
View file

@ -3,13 +3,17 @@ import multiprocessing
import pathlib
import subprocess
import traceback
from abc import ABC
from dataclasses import dataclass, field
from datetime import timedelta, datetime
from enum import Enum
from functools import partial
from operator import attrgetter
from typing import List, Optional, Tuple, Dict
from typing import List, Dict
from typing import Tuple, Optional
import feedparser
from feedparser import FeedParserDict
from sqlalchemy import Column, Integer, String, Text
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.orm import Session
@ -28,11 +32,13 @@ DEFAULT_RETRY_FREQUENCY = timedelta(hours=1)
class DownloadFrequency(int, Enum):
hourly = 3600
daily = 86400
weekly = 604800
biweekly = 1209600
days30 = 2592000
days90 = 7776000
hours3 = hourly * 3
hours12 = hourly * 12
daily = hourly * 24
weekly = daily * 7
biweekly = weekly * 2
days30 = daily * 30
days90 = daily * 90
@dataclass
@ -45,22 +51,21 @@ class DownloadResult:
class Download(ModelHelper, Base):
"""
Model that is used to schedule downloads.
"""
"""Model that is used to schedule downloads."""
__tablename__ = 'download'
id = Column(Integer, primary_key=True)
url = Column(String, nullable=False)
attempts = Column(Integer, default=0)
downloader = Column(Text)
error = Column(Text)
frequency = Column(Integer)
info_json = Column(JSONB)
last_successful_download = Column(TZDateTime)
location = Column(Text)
next_download = Column(TZDateTime)
status = Column(String, default='new')
info_json = Column(JSONB)
downloader = Column(Text)
location = Column(Text)
error = Column(Text)
sub_downloader = Column(Text)
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
@ -70,8 +75,10 @@ class Download(ModelHelper, Base):
def __repr__(self):
if self.next_download or self.frequency:
return f'<Download id={self.id} status={self.status} url={repr(self.url)} ' \
f'next_download={self.next_download} frequency={self.frequency} attempts={self.attempts}>'
return f'<Download id={self.id} status={self.status} url={repr(self.url)} attempts={self.attempts}>'
f'next_download={self.next_download} frequency={self.frequency} attempts={self.attempts} ' \
f'error={self.error}>'
return f'<Download id={self.id} status={self.status} url={repr(self.url)} attempts={self.attempts} ' \
f'error={self.error}>'
def __json__(self):
d = dict(
@ -148,12 +155,16 @@ class Downloader:
return dict(name=self.name, pretty_name=self.pretty_name)
@classmethod
def valid_url(cls, url) -> Tuple[bool, Optional[dict]]:
def valid_url(cls, url: str) -> Tuple[bool, Optional[dict]]:
raise NotImplementedError()
def do_download(self, download: Download) -> DownloadResult:
raise NotImplementedError()
@optional_session
def already_downloaded(self, url: str, session: Session = None):
raise NotImplementedError()
@property
def manager(self):
if self._manager is None:
@ -200,7 +211,7 @@ class Downloader:
logger.warning(f'Killing download {pid=}')
proc.kill()
proc.poll()
raise UnrecoverableDownloadError(f'Download {pid=} was killed!')
raise ChildProcessError(f'Download {pid=} was killed!')
finally:
self._kill.clear()
@ -217,9 +228,10 @@ class DownloadManager:
priority_sorter = partial(sorted, key=attrgetter('priority'))
def __init__(self):
self.instances = tuple()
self.instances: Tuple[Downloader] = tuple()
self._instances = dict()
self.disabled = multiprocessing.Event()
self.stopped: bool = False
def register_downloader(self, instance: Downloader):
if not isinstance(instance, Downloader):
@ -270,6 +282,9 @@ class DownloadManager:
download = self.get_download(session, url=url)
if not download:
if url in DOWNLOAD_MANAGER_CONFIG.skip_urls:
raise InvalidDownload(
f'Refusing to download {url} because it is in the download_manager.yaml skip list')
download = Download(url=url, status='new')
session.add(download)
session.flush()
@ -277,39 +292,45 @@ class DownloadManager:
return download
def create_download(self, url: str, session, downloader: str = None, skip_download: bool = False,
reset_attempts: bool = False) -> Download:
reset_attempts: bool = False, sub_downloader: str = None) -> Download:
"""Schedule a URL for download. If the URL failed previously, it may be retried."""
downloads = self.create_downloads([url], session, downloader, skip_download=skip_download,
reset_attempts=reset_attempts)
reset_attempts=reset_attempts, sub_downloader=sub_downloader)
return downloads[0]
def create_downloads(self, urls: List[str], session: Session = None, downloader: str = None,
skip_download: bool = False, reset_attempts: bool = False) -> List[Download]:
def create_downloads(self, urls: List[str], session: Session = None, downloader_name: str = None,
skip_download: bool = False, reset_attempts: bool = False, sub_downloader: str = None) \
-> List[Download]:
"""Schedule all URLs for download. If one cannot be downloaded, none will be added."""
if not session:
_, session = get_db_context()
downloads = []
forced_downloader = self.get_downloader_by_name(downloader) if downloader else None
if downloader and not forced_downloader:
downloader = self.get_downloader_by_name(downloader_name) if downloader_name else None
if downloader_name and not downloader:
# Could not find the downloader
raise InvalidDownload(f'Unknown downloader {downloader}')
raise InvalidDownload(f'Unknown downloader {downloader_name}')
with session.transaction:
for url in urls:
if not url:
raise ValueError('Download must have a URL')
if url in DOWNLOAD_MANAGER_CONFIG.skip_urls:
logger.warning(f'Skipping {url} because it is in download_manager.yaml skip list.')
continue
info_json = None
downloader = forced_downloader
if not forced_downloader:
if not downloader:
# User has requested automatic downloader selection, try and find it.
downloader, info_json = self.get_downloader(url)
downloader_, info_json = self.get_downloader(url)
else:
downloader_ = downloader
download = self.get_or_create_download(url, session)
# Download may have failed, try again.
download.renew(reset_attempts=reset_attempts)
download.downloader = downloader.name
download.info_json = info_json
download.downloader = downloader_.name
download.info_json = info_json or download.info_json
download.sub_downloader = sub_downloader
downloads.append(download)
if skip_download is True:
@ -338,6 +359,8 @@ class DownloadManager:
for download in downloads:
if self.disabled.is_set():
raise InvalidDownload('DownloadManager is disabled')
if self.stopped:
return
download_count += 1
download_id = download.id
@ -382,7 +405,11 @@ class DownloadManager:
download.location = result.location or download.location or None
# Clear any old errors if the download succeeded.
download.error = result.error if result.error else None
download.next_download = self.get_next_download(download, session)
download.next_download = self.calculate_next_download(download, session)
if result.downloads:
logger.info(f'Adding {len(result.downloads)} downloads from result of {download.url}')
self.create_downloads(result.downloads, session_, downloader_name=download.sub_downloader)
if try_again is False:
download.fail()
@ -420,10 +447,12 @@ class DownloadManager:
with get_db_curs(commit=True) as curs:
curs.execute("UPDATE download SET status='new' WHERE status='pending' OR status='deferred'")
def recurring_download(self, url: str, frequency: int, skip_download: bool = False) -> Download:
def recurring_download(self, url: str, frequency: int, skip_download: bool = False, downloader: str = None,
sub_downloader: str = None, reset_attempts: bool = False) -> Download:
"""Schedule a recurring download."""
with get_db_session(commit=True) as session:
download = self.create_download(url, session=session, skip_download=True)
download = self.create_download(url, session=session, skip_download=True, downloader=downloader,
sub_downloader=sub_downloader, reset_attempts=reset_attempts)
download.frequency = frequency
if skip_download is False and PYTEST:
@ -476,12 +505,14 @@ class DownloadManager:
renewed = True
if renewed:
logger.debug('Renewed recurring downloads')
session.commit()
self.start_downloads()
else:
logger.debug('No recurring downloads to renew')
def get_downloads(self, session: Session) -> List[Download]:
downloads = session.query(Download).all()
downloads = list(downloads)
downloads = list(session.query(Download).all())
for download in downloads:
download.manager = self
return downloads
@ -519,16 +550,19 @@ class DownloadManager:
downloader.kill()
download.fail()
def stop(self):
"""Stop all Downloaders, defer all pending downloads."""
self.stopped = True
for downloader in self.instances:
downloader.kill()
for download in self.get_pending_downloads():
download.defer()
def kill(self):
"""Kill all downloads. Do not start new downloads."""
self.disabled.set()
try:
with get_db_session(commit=True) as session:
downloads = self.get_pending_downloads(session)
for download in downloads:
downloader = download.get_downloader()
downloader.kill()
download.defer()
self.stop()
except Exception as e:
logger.critical(f'Failed to kill downloads!', exc_info=e)
@ -552,10 +586,8 @@ class DownloadManager:
session.delete(download)
def list_downloaders(self) -> List[Downloader]:
"""
Return a list of the Downloaders available on this Download Manager.
"""
return list(filter(lambda i: i.listable, self.instances))
"""Return a list of the Downloaders available on this Download Manager."""
return [i for i in self.instances if i.listable]
# Downloads should be sorted by their status in a particular order.
_status_order = '''CASE
@ -579,7 +611,8 @@ class DownloadManager:
next_download,
status,
url,
location
location,
error
FROM download
WHERE frequency IS NOT NULL
ORDER BY
@ -622,10 +655,13 @@ class DownloadManager:
@optional_session
def get_pending_downloads(self, session: Session) -> List[Download]:
return session.query(Download).filter_by(status='pending').all()
downloads = session.query(Download).filter_by(status='pending').all()
for download in downloads:
download.manager = self
return downloads
@optional_session
def get_next_download(self, download: Download, session: Session) -> Optional[datetime]:
def calculate_next_download(self, download: Download, session: Session) -> Optional[datetime]:
"""
If the download is "deferred", download soon. But, slowly increase the time between attempts.
@ -716,3 +752,72 @@ class DownloadMangerConfig(ConfigFile):
DOWNLOAD_MANAGER_CONFIG: DownloadMangerConfig = DownloadMangerConfig()
def set_test_download_manager_config(enabled: bool):
global DOWNLOAD_MANAGER_CONFIG
DOWNLOAD_MANAGER_CONFIG = None
if enabled:
DOWNLOAD_MANAGER_CONFIG = DownloadMangerConfig()
def parse_feed(url: str) -> FeedParserDict:
"""Calls `feedparser.parse`, used for testing."""
return feedparser.parse(url)
class RSSDownloader(Downloader, ABC):
"""Downloads an RSS feed and creates new downloads for every unique link in the feed."""
name = 'rss'
pretty_name = 'RSS'
listable = False
@classmethod
def valid_url(cls, url: str) -> Tuple[bool, Optional[dict]]:
"""Attempts to parse an RSS Feed. If it succeeds, it returns a FeedParserDict as a dictionary."""
feed = parse_feed(url)
feed = dict(feed)
if feed['bozo']:
# Feed did not parse.
return False, {}
return True, dict(feed=feed)
def do_download(self, download: Download) -> DownloadResult:
if isinstance(download.info_json, dict) and download.info_json.get('feed'):
feed = download.info_json['feed']
else:
# self.valid_url was not called by the manager, do it here.
feed: FeedParserDict = parse_feed(download.url)
if feed['bozo'] and not self.acceptable_bozo_errors(feed):
# Feed did not parse
return DownloadResult(success=False, error='Failed to parse RSS feed')
if not isinstance(feed, dict) or not feed.get('entries'):
# RSS parsed but does not have entries.
return DownloadResult(success=False, error='RSS feed did not have any entries')
# Only download URLs that have not yet been downloaded.
urls = []
sub_downloader = self.manager.get_downloader_by_name(download.sub_downloader)
for idx, entry in enumerate(feed['entries']):
if url := entry.get('link'):
if not sub_downloader.already_downloaded(url):
urls.append(url)
else:
logger.warning(f'RSS entry {idx} did not have a link!')
session = Session.object_session(download)
self.manager.create_downloads(urls, session=session, downloader_name=download.sub_downloader)
return DownloadResult(success=True)
@staticmethod
def acceptable_bozo_errors(feed):
"""
Feedparser can report some errors, some we can ignore.
"""
if 'CharacterEncodingOverride' in str(feed['bozo_exception']):
return True
return False
rss_downloader = RSSDownloader(30)

17
wrolpi/root_api.py
View file

@ -13,6 +13,7 @@ from sanic import Sanic, response, Blueprint, __version__ as sanic_version
from sanic.blueprint_group import BlueprintGroup
from sanic.request import Request
from sanic.response import HTTPResponse
from sanic.signals import Event
from sanic_ext import validate
from sanic_ext.extensions.openapi import openapi
@ -25,7 +26,7 @@ from wrolpi.downloader import download_manager
from wrolpi.errors import WROLModeEnabled, InvalidTimezone, API_ERRORS, APIError, ValidationError, HotspotError
from wrolpi.media_path import MediaPath
from wrolpi.schema import RegexRequest, RegexResponse, SettingsRequest, SettingsResponse, DownloadRequest, EchoResponse
from wrolpi.vars import DOCKERIZED
from wrolpi.vars import DOCKERIZED, PYTEST
from wrolpi.version import __version__
logger = logger.getChild(__name__)
@ -203,7 +204,12 @@ async def post_download(_: Request, body: DownloadRequest):
downloader = body.downloader
if not downloader or downloader in ('auto', 'None', 'null'):
downloader = None
download_manager.create_downloads(urls, downloader=downloader, reset_attempts=True)
if body.frequency:
download_manager.recurring_download(urls[0], body.frequency, downloader=downloader,
sub_downloader=body.sub_downloader, reset_attempts=True)
else:
download_manager.create_downloads(urls, downloader_name=downloader, sub_downloader=body.sub_downloader,
reset_attempts=True)
return response.empty()
@ -358,3 +364,10 @@ def json_error_handler(request, exception: Exception):
api_app.error_handler.add(APIError, json_error_handler)
@api_app.signal(Event.SERVER_SHUTDOWN_BEFORE)
def handle_server_shutdown(*args, **kwargs):
"""Stop downloads when server is shutting down."""
if not PYTEST:
download_manager.stop()

2
wrolpi/schema.py
View file

@ -49,6 +49,8 @@ class EchoResponse:
class DownloadRequest:
urls: str
downloader: Optional[str] = None
frequency: Optional[int] = None
sub_downloader: Optional[str] = None
@dataclass

263
wrolpi/test/test_downloader.py
View file

@ -6,18 +6,15 @@ from unittest.mock import MagicMock
import pytest
from wrolpi import downloader
from wrolpi.dates import local_timezone, Seconds
from wrolpi.db import get_db_context
from wrolpi.downloader import Downloader, Download, DownloadFrequency, DownloadResult
from wrolpi.errors import UnrecoverableDownloadError, InvalidDownload, WROLModeEnabled
from wrolpi.test.common import wrap_test_db, TestAPI
from wrolpi.test.common import assert_dict_contains
class PermissiveDownloader(Downloader):
"""
A testing Downloader which always says it's valid.
"""
"""A testing Downloader which always says it's valid."""
name = 'permissive'
def __repr__(self):
@ -28,9 +25,7 @@ class PermissiveDownloader(Downloader):
class HTTPDownloader(Downloader):
"""
A testing Downloader which says its valid when a URL starts with http/https
"""
"""A testing Downloader which says its valid when a URL starts with http/https"""
name = 'http'
def __repr__(self):
@ -40,124 +35,100 @@ class HTTPDownloader(Downloader):
return url.startswith('https://') or url.startswith('http://'), None
class TestDownloader(TestAPI):
def setUp(self) -> None:
super().setUp()
self.mgr = downloader.DownloadManager()
PermissiveDownloader.do_download = MagicMock()
HTTPDownloader.do_download = MagicMock()
def test_register_downloader(test_session, test_download_manager):
"""Downloaders can be registered and have specific priorities."""
assert test_download_manager.instances == tuple()
def test_register_downloader(self):
"""
Downloaders can be registered and have specific priorities.
"""
self.assertEqual(self.mgr.instances, tuple())
http_downloader = HTTPDownloader()
test_download_manager.register_downloader(http_downloader)
assert test_download_manager.instances == (http_downloader,)
http_downloader = HTTPDownloader()
self.mgr.register_downloader(http_downloader)
self.assertEqual(self.mgr.instances, (http_downloader,))
# PermissiveDownloader is first priority.
permissive_downloader = PermissiveDownloader(priority=0)
test_download_manager.register_downloader(permissive_downloader)
assert test_download_manager.instances == (permissive_downloader, http_downloader)
# PermissiveDownloader is first priority.
with pytest.raises(ValueError):
test_download_manager.register_downloader(http_downloader)
with pytest.raises(ValueError):
test_download_manager.register_downloader(permissive_downloader)
def test_do_downloads(test_session, test_download_manager):
_, session = get_db_context()
http_downloader = HTTPDownloader()
http_downloader.do_download = MagicMock()
http_downloader.do_download.return_value = DownloadResult(success=True)
test_download_manager.register_downloader(http_downloader)
permissive_downloader = PermissiveDownloader(priority=100)
permissive_downloader.do_download = MagicMock()
permissive_downloader.do_download.return_value = DownloadResult(success=False)
test_download_manager.register_downloader(permissive_downloader)
# https is handled by the HTTP Downloader.
test_download_manager.create_download('https://example.com', session)
http_downloader.do_download.assert_called_once()
permissive_downloader.do_download.assert_not_called()
assert test_download_manager.get_download(session, 'https://example.com') is not None
http_downloader.do_download.reset_mock()
# try the permissive download, which returns a failure.
test_download_manager.create_download('foo', session)
http_downloader.do_download.assert_not_called()
permissive_downloader.do_download.assert_called_once()
download = test_download_manager.get_download(session, 'foo')
assert download.attempts == 1
# try again
test_download_manager.create_download('foo', session)
test_download_manager._do_downloads(session)
download = test_download_manager.get_download(session, 'foo')
assert download.attempts == 2
# finally success
permissive_downloader.do_download.return_value = DownloadResult(success=True)
test_download_manager.create_download('foo', session)
test_download_manager._do_downloads(session)
download = test_download_manager.get_download(session, 'foo')
assert download.status == 'complete'
# No downloads left.
assert list(test_download_manager.get_new_downloads(session)) == []
def test_delete_old_once_downloads(test_session, test_download_manager):
"""Once-downloads over a month old should be deleted."""
with mock.patch('wrolpi.downloader.now') as mock_now:
mock_now.return_value = local_timezone(datetime(2020, 6, 5, 0, 0))
permissive_downloader = PermissiveDownloader(priority=0)
self.mgr.register_downloader(permissive_downloader)
self.assertEqual(self.mgr.instances, (permissive_downloader, http_downloader))
self.assertRaises(ValueError, self.mgr.register_downloader, http_downloader)
self.assertRaises(ValueError, self.mgr.register_downloader, permissive_downloader)
@wrap_test_db
def test_ensure_download(self):
_, session = get_db_context()
self.assertEqual(len(self.mgr.get_downloads(session)), 0)
http_downloader = HTTPDownloader()
http_downloader.do_download.return_value = DownloadResult(success=True)
self.mgr.register_downloader(http_downloader)
self.mgr.create_download('https://example.com', session)
download = self.mgr.get_download(session, 'https://example.com')
self.assertEqual(download.url, 'https://example.com')
self.assertEqual(len(self.mgr.get_downloads(session)), 1)
self.assertIsNotNone(self.mgr.get_download(session, 'https://example.com'))
self.assertIsNone(self.mgr.get_download(session, 'https://example.com/not downloading'))
@wrap_test_db
def test_do_downloads(self):
_, session = get_db_context()
http_downloader = HTTPDownloader()
http_downloader.do_download.return_value = DownloadResult(success=True)
self.mgr.register_downloader(http_downloader)
permissive_downloader = PermissiveDownloader(priority=100)
permissive_downloader.do_download.return_value = DownloadResult(success=False)
self.mgr.register_downloader(permissive_downloader)
# https is handled by the HTTP Downloader.
self.mgr.create_download('https://example.com', session)
http_downloader.do_download.assert_called_once()
permissive_downloader.do_download.assert_not_called()
self.assertIsNotNone(self.mgr.get_download(session, 'https://example.com'))
http_downloader.do_download.reset_mock()
# try the permissive download, which returns a failure.
self.mgr.create_download('foo', session)
http_downloader.do_download.assert_not_called()
permissive_downloader.do_download.assert_called_once()
download = self.mgr.get_download(session, 'foo')
self.assertEqual(download.attempts, 1)
# try again
self.mgr.create_download('foo', session)
self.mgr._do_downloads(session)
download = self.mgr.get_download(session, 'foo')
self.assertEqual(download.attempts, 2)
# finally success
permissive_downloader.do_download.return_value = DownloadResult(success=True)
self.mgr.create_download('foo', session)
self.mgr._do_downloads(session)
download = self.mgr.get_download(session, 'foo')
self.assertEqual(download.status, 'complete')
# No downloads left.
self.assertEqual(list(self.mgr.get_new_downloads(session)), [])
@wrap_test_db
@mock.patch('wrolpi.downloader.now', lambda: local_timezone(datetime(2020, 6, 5, 0, 0)))
def test_delete_old_once_downloads(self):
"""
Once-downloads over a month old should be deleted.
"""
permissive_downloader = PermissiveDownloader(priority=0)
self.mgr.register_downloader(permissive_downloader)
test_download_manager.register_downloader(permissive_downloader)
_, session = get_db_context()
# Recurring downloads should not be deleted.
d1 = self.mgr.create_download('https://example.com/1', session, skip_download=True)
d2 = self.mgr.create_download('https://example.com/2', session, skip_download=True)
d1 = test_download_manager.create_download('https://example.com/1', session, skip_download=True)
d2 = test_download_manager.create_download('https://example.com/2', session, skip_download=True)
d1.frequency = 1
d2.frequency = 1
d2.started()
# Should be deleted.
d3 = self.mgr.create_download('https://example.com/3', session, skip_download=True)
d4 = self.mgr.create_download('https://example.com/4', session, skip_download=True)
d3 = test_download_manager.create_download('https://example.com/3', session, skip_download=True)
d4 = test_download_manager.create_download('https://example.com/4', session, skip_download=True)
d3.complete()
d4.complete()
d3.last_successful_download = local_timezone(datetime(2020, 1, 1, 0, 0, 0))
d4.last_successful_download = local_timezone(datetime(2020, 5, 1, 0, 0, 0))
# Not a month old.
d5 = self.mgr.create_download('https://example.com/5', session, skip_download=True)
d5 = test_download_manager.create_download('https://example.com/5', session, skip_download=True)
d5.last_successful_download = local_timezone(datetime(2020, 6, 1, 0, 0, 0))
# An old, but pending download should not be deleted.
d6 = self.mgr.create_download('https://example.com/6', session, skip_download=True)
d6 = test_download_manager.create_download('https://example.com/6', session, skip_download=True)
d6.last_successful_download = local_timezone(datetime(2020, 4, 1, 0, 0, 0))
d6.started()
self.mgr.delete_old_once_downloads()
test_download_manager.delete_old_once_downloads()
# Two old downloads are deleted.
downloads = session.query(Download).order_by(Download.id).all()
@ -168,28 +139,42 @@ class TestDownloader(TestAPI):
dict(url='https://example.com/6', attempts=1, status='pending'),
]
for download, expected in zip_longest(downloads, expected):
self.assertDictContains(download.dict(), expected)
assert_dict_contains(download.dict(), expected)
@wrap_test_db
def test_create_downloads(self):
"""
Multiple downloads can be scheduled using DownloadManager.create_downloads.
"""
http_downloader = HTTPDownloader(priority=0)
self.mgr.register_downloader(http_downloader)
_, session = get_db_context()
def test_ensure_download(test_session, test_download_manager):
assert len(test_download_manager.get_downloads(test_session)) == 0
# Both URLs are valid and are scheduled.
self.mgr.create_downloads(['https://example.com/1', 'https://example.com/2'], skip_download=True)
downloads = self.mgr.get_downloads(session)
self.assertEqual({i.url for i in downloads}, {'https://example.com/1', 'https://example.com/2'})
http_downloader = HTTPDownloader()
http_downloader.do_download = MagicMock()
http_downloader.do_download.return_value = DownloadResult(success=True)
test_download_manager.register_downloader(http_downloader)
# One URL is bad, neither should be scheduled.
self.assertRaises(InvalidDownload, self.mgr.create_downloads, ['https://example.com/3', 'bad url should fail'],
skip_download=True)
downloads = self.mgr.get_downloads(session)
self.assertEqual({i.url for i in downloads}, {'https://example.com/1', 'https://example.com/2'})
test_download_manager.create_download('https://example.com', test_session)
download = test_download_manager.get_download(test_session, 'https://example.com')
assert download.url == 'https://example.com'
assert len(test_download_manager.get_downloads(test_session)) == 1
assert test_download_manager.get_download(test_session, 'https://example.com') is not None
assert test_download_manager.get_download(test_session, 'https://example.com/not downloading') is None
def test_create_downloads(test_session, test_download_manager):
"""Multiple downloads can be scheduled using DownloadManager.create_downloads."""
http_downloader = HTTPDownloader(priority=0)
test_download_manager.register_downloader(http_downloader)
# Both URLs are valid and are scheduled.
test_download_manager.create_downloads(['https://example.com/1', 'https://example.com/2'], skip_download=True)
downloads = test_download_manager.get_downloads(test_session)
assert {i.url for i in downloads} == {'https://example.com/1', 'https://example.com/2'}
# One URL is bad, neither should be scheduled.
with pytest.raises(InvalidDownload):
test_download_manager.create_downloads(['https://example.com/3', 'bad url should fail'], skip_download=True)
downloads = test_download_manager.get_downloads(test_session)
assert {i.url for i in downloads} == {'https://example.com/1', 'https://example.com/2'}
def test_downloader_must_have_name():
@ -215,7 +200,7 @@ async def test_download_wrol_mode(test_session, test_download_manager):
await test_download_manager.do_downloads_sync()
def test_download_get_downloader(test_session, test_download_manager):
def test_download_get_downloader(test_session, test_download_manager,):
"""
A Download can find it's Downloader.
"""
@ -232,7 +217,7 @@ def test_download_get_downloader(test_session, test_download_manager):
assert download2.get_downloader() is None
def test_get_next_download(test_session, test_download_manager, fake_now):
def test_calculate_next_download(test_session, test_download_manager, fake_now):
fake_now(datetime(2000, 1, 1))
download = Download()
download.frequency = DownloadFrequency.weekly
@ -250,7 +235,7 @@ def test_get_next_download(test_session, test_download_manager, fake_now):
]
for attempts, expected in attempts_expected:
download.attempts = attempts
result = test_download_manager.get_next_download(download)
result = test_download_manager.calculate_next_download(download)
assert result == expected, f'{attempts} != {result}'
d1 = Download(url='https://example.com/1', frequency=DownloadFrequency.weekly)
@ -259,9 +244,9 @@ def test_get_next_download(test_session, test_download_manager, fake_now):
test_session.add_all([d1, d2, d3])
test_session.commit()
# Downloads are spread out over the next week.
assert test_download_manager.get_next_download(d1) == local_timezone(datetime(2000, 1, 8))
assert test_download_manager.get_next_download(d2) == local_timezone(datetime(2000, 1, 11, 12))
assert test_download_manager.get_next_download(d3) == local_timezone(datetime(2000, 1, 9, 18))
assert test_download_manager.calculate_next_download(d1) == local_timezone(datetime(2000, 1, 8))
assert test_download_manager.calculate_next_download(d2) == local_timezone(datetime(2000, 1, 11, 12))
assert test_download_manager.calculate_next_download(d3) == local_timezone(datetime(2000, 1, 9, 18))
def test_recurring_downloads(test_session, test_download_manager, fake_now):
@ -359,3 +344,23 @@ def test_max_attempts(test_session, test_download_manager):
download = session.query(Download).one()
assert download.attempts == 3
assert download.status == 'failed'
def test_skip_urls(test_session, test_download_manager):
"""The DownloadManager will not create downloads for URLs in it's skip list."""
_, session = get_db_context()
from wrolpi.downloader import DOWNLOAD_MANAGER_CONFIG
DOWNLOAD_MANAGER_CONFIG.skip_urls = ['https://example.com/skipme']
http_downloader = HTTPDownloader()
http_downloader.do_download = MagicMock()
http_downloader.do_download.return_value = DownloadResult(success=True)
test_download_manager.register_downloader(http_downloader)
test_download_manager.create_downloads([
'https://example.com/1',
'https://example.com/skipme',
'https://example.com/2',
], downloader_name=HTTPDownloader.name)
downloads = test_download_manager.get_downloads(test_session)
assert {i.url for i in downloads} == {'https://example.com/1', 'https://example.com/2'}

2
wrolpi/test/test_root_api.py
View file

@ -229,7 +229,7 @@ def test_throttle_toggle(test_session, test_client, test_config):
assert response.json['throttle_status'] == 'ondemand'
def test_clear_downloads(test_session, test_client, test_config):
def test_clear_downloads(test_session, test_client, test_config, test_download_manager_config):
from wrolpi.downloader import DOWNLOAD_MANAGER_CONFIG
with get_db_session(commit=True) as session:

114
wrolpi/test/test_rss.py Normal file
View file

@ -0,0 +1,114 @@
import json
from http import HTTPStatus
from itertools import zip_longest
from typing import List
import mock
from wrolpi.db import optional_session
from wrolpi.downloader import Download, DownloadResult, RSSDownloader
from wrolpi.test.test_downloader import HTTPDownloader
class RSSHTTPDownloader(HTTPDownloader):
def do_download(self, download: Download) -> DownloadResult:
return DownloadResult(success=True)
@optional_session
def already_downloaded(self, url: str, session=None):
download = session.query(Download).filter_by(url=url).one_or_none()
return download
def test_rss_download_invalid(test_session, test_download_manager):
"""Feeds are not downloaded when FeedParser reports an issue."""
rss_downloader = RSSDownloader()
test_download_manager.register_downloader(rss_downloader)
def test_rss_download(test_session, test_download_manager):
"""An RSS Downloader will create new Downloads for every link in the feed."""
rss_downloader = RSSDownloader()
test_download_manager.register_downloader(rss_downloader)
http_downloader = RSSHTTPDownloader()
test_download_manager.register_downloader(http_downloader)
def check_downloads(expected: List[dict]):
downloads = test_session.query(Download).order_by(Download.id).all()
for download, expected in zip_longest(downloads, expected):
assert download.id == expected['id'] \
and download.status == expected['status'] \
and download.url == expected['url'] \
and download.attempts == expected['attempts'], f'{download} != {expected}'
with mock.patch('wrolpi.downloader.parse_feed') as mock_parse_feed:
mock_parse_feed.return_value = dict(
bozo=0,
entries=[
dict(link='https://example.com/a'),
dict(link='https://example.com/b'),
dict(link='https://example.com/c'),
]
)
test_download_manager.create_download('https://example.com/feed', test_session, sub_downloader='http')
# Feed download is complete.
check_downloads([
dict(id=1, status='complete', url='https://example.com/feed', attempts=1),
dict(id=2, status='complete', url='https://example.com/a', attempts=1),
dict(id=3, status='complete', url='https://example.com/b', attempts=1),
dict(id=4, status='complete', url='https://example.com/c', attempts=1),
])
with mock.patch('wrolpi.downloader.parse_feed') as mock_parse_feed:
mock_parse_feed.return_value = dict(
bozo=0,
entries=[
dict(link='https://example.com/a'),
dict(link='https://example.com/d'),
]
)
test_download_manager.create_download('https://example.com/feed', test_session, sub_downloader='http')
# Only the new URLs are Archived.
check_downloads([
dict(id=1, status='complete', url='https://example.com/feed', attempts=2),
dict(id=2, status='complete', url='https://example.com/a', attempts=1),
dict(id=3, status='complete', url='https://example.com/b', attempts=1),
dict(id=4, status='complete', url='https://example.com/c', attempts=1),
dict(id=5, status='complete', url='https://example.com/d', attempts=1),
])
def test_post_rss_download(test_session, test_client, test_download_manager):
"""An RSS can be downloaded in the UI"""
rss_downloader = RSSDownloader()
test_download_manager.register_downloader(rss_downloader)
http_downloader = RSSHTTPDownloader()
test_download_manager.register_downloader(http_downloader)
with mock.patch('wrolpi.downloader.parse_feed') as mock_parse_feed:
mock_parse_feed.return_value = dict(bozo=0, entries=[])
content = dict(urls='https://example.com/feed', downloader='rss', frequency=100)
request, response = test_client.post('/api/download', content=json.dumps(content))
assert response.status == HTTPStatus.NO_CONTENT
# Download was attempted.
(download,) = test_download_manager.get_downloads(test_session)
assert download.url == 'https://example.com/feed'
assert download.frequency == 100
assert download.downloader == 'rss'
assert download.attempts == 1
def test_rss_no_entries(test_session, test_download_manager):
"""An RSS feed with no entries is handled."""
rss_downloader = RSSDownloader()
test_download_manager.register_downloader(rss_downloader)
with mock.patch('wrolpi.downloader.parse_feed') as mock_parse_feed:
mock_parse_feed.return_value = dict(bozo=0, ) # missing `entries`
test_download_manager.create_download('https://example.com/feed', test_session, sub_downloader='http')
(download,) = test_download_manager.get_downloads(test_session)
assert 'entries' in download.error

2
wrolpi/version.py
View file

@ -1,4 +1,4 @@
__version__ = '0.4.9-beta'
__version__ = '0.5.0-beta'
import subprocess