mirror of
https://github.com/lrnselfreliance/wrolpi.git
synced 2026-10-26 19:01:39 -04:00
Converting Tag methods to use tag id or name.
This commit is contained in:
Roland
parent
2ad363281f
commit
c839acb270
22 changed files with 379 additions and 163 deletions
28
main.py
28
main.py
|
|
@ -244,16 +244,24 @@ async def start_single_tasks(app: Sanic):
|
|||
download_manager.retry_downloads()
|
||||
|
||||
# Hotspot/throttle are not supported in Docker containers.
|
||||
if not DOCKERIZED and get_wrolpi_config().hotspot_on_startup:
|
||||
try:
|
||||
admin.enable_hotspot()
|
||||
except Exception as e:
|
||||
logger.error('Failed to enable hotspot', exc_info=e)
|
||||
if not DOCKERIZED and get_wrolpi_config().throttle_on_startup:
|
||||
try:
|
||||
admin.throttle_cpu_on()
|
||||
except Exception as e:
|
||||
logger.error('Failed to throttle CPU', exc_info=e)
|
||||
if not DOCKERIZED:
|
||||
if get_wrolpi_config().hotspot_on_startup:
|
||||
logger.info('Starting hotspot...')
|
||||
try:
|
||||
admin.enable_hotspot()
|
||||
except Exception as e:
|
||||
logger.error('Failed to enable hotspot', exc_info=e)
|
||||
else:
|
||||
logger.info('Hotspot on startup is disabled.')
|
||||
|
||||
if get_wrolpi_config().throttle_on_startup:
|
||||
logger.info('Throttling CPU...')
|
||||
try:
|
||||
admin.throttle_cpu_on()
|
||||
except Exception as e:
|
||||
logger.error('Failed to throttle CPU', exc_info=e)
|
||||
else:
|
||||
logger.info('CPU throttle on startup is disabled')
|
||||
|
||||
# Run the startup functions
|
||||
for func in BEFORE_STARTUP_FUNCTIONS:
|
||||
|
|
|
|||
|
|
@ -27,7 +27,7 @@ PRETTY_NAME = 'Archive'
|
|||
|
||||
logger = logger.getChild(__name__)
|
||||
|
||||
__all__ = ['ArchiveDownloader', 'archive_downloader']
|
||||
__all__ = ['ArchiveDownloader', 'archive_downloader', 'model_archive']
|
||||
|
||||
|
||||
class ArchiveDownloader(Downloader, ABC):
|
||||
|
|
|
|||
|
|
@ -1,7 +1,7 @@
|
|||
import json
|
||||
import pathlib
|
||||
import re
|
||||
from typing import Iterable, List, Optional, Union
|
||||
from typing import Iterable, List, Optional
|
||||
|
||||
import pytz
|
||||
from sqlalchemy import Column, Integer, String, ForeignKey, BigInteger
|
||||
|
|
@ -59,6 +59,20 @@ class Archive(Base, ModelHelper):
|
|||
raise UnknownArchive(f'Cannot find Archive with id {id_}')
|
||||
return archive
|
||||
|
||||
@staticmethod
|
||||
@optional_session
|
||||
def get_by_path(path: pathlib.Path | str, session: Session = None) -> Optional['Archive']:
|
||||
archive = session.query(Archive).join(FileGroup).filter(FileGroup.primary_path == str(path)).one_or_none()
|
||||
return archive
|
||||
|
||||
@staticmethod
|
||||
@optional_session
|
||||
def find_by_path(path: pathlib.Path | str, session: Session = None) -> Optional['Archive']:
|
||||
archive = Archive.get_by_path(path, session)
|
||||
if archive:
|
||||
return archive
|
||||
raise UnknownArchive(f'Cannot find Archive with path: {path}')
|
||||
|
||||
def my_paths(self, *mimetypes: str) -> List[pathlib.Path]:
|
||||
return self.file_group.my_paths(*mimetypes)
|
||||
|
||||
|
|
@ -285,9 +299,8 @@ class Archive(Base, ModelHelper):
|
|||
archive = model_archive(file_group, session)
|
||||
return archive
|
||||
|
||||
def add_tag(self, tag_or_tag_name: Union[Tag, str]) -> TagFile:
|
||||
tag = Tag.get_by_name(tag_or_tag_name) if isinstance(tag_or_tag_name, str) else tag_or_tag_name
|
||||
return self.file_group.add_tag(tag)
|
||||
def add_tag(self, tag_id_or_name: int | str) -> TagFile:
|
||||
return self.file_group.add_tag(tag_id_or_name)
|
||||
|
||||
|
||||
class Domain(Base, ModelHelper):
|
||||
|
|
|
|||
|
|
@ -16,6 +16,7 @@ from modules.videos.models import Channel, Video
|
|||
from wrolpi.api_utils import api_app
|
||||
from wrolpi.downloader import DownloadFrequency, DownloadManager
|
||||
from wrolpi.files.models import FileGroup
|
||||
from wrolpi.tags import Tag
|
||||
from wrolpi.vars import PROJECT_DIR
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -1,7 +1,7 @@
|
|||
import json
|
||||
import pathlib
|
||||
from pathlib import Path
|
||||
from typing import Optional, Dict, List, Union
|
||||
from typing import Optional, Dict, List
|
||||
|
||||
from sqlalchemy import Column, Integer, String, Boolean, JSON, Date, ForeignKey, BigInteger
|
||||
from sqlalchemy.orm import relationship, Session, deferred
|
||||
|
|
@ -224,7 +224,7 @@ class Video(ModelHelper, Base):
|
|||
try:
|
||||
validate_video(self, self.channel.generate_posters if self.channel else False)
|
||||
except Exception as e:
|
||||
logger.warning(f'Failed to validate video {self}', exc_info=e)
|
||||
logger.error(f'Failed to validate video {self}', exc_info=e)
|
||||
if PYTEST:
|
||||
raise
|
||||
|
||||
|
|
@ -339,9 +339,10 @@ class Video(ModelHelper, Base):
|
|||
raise UnknownVideo(f'Cannot find Video with id {id_}')
|
||||
return video
|
||||
|
||||
def add_tag(self, tag_id: int, session: Session = None) -> TagFile:
|
||||
def add_tag(self, tag_id_or_name: int, session: Session = None) -> TagFile:
|
||||
session = session or Session.object_session(self)
|
||||
return self.file_group.add_tag(tag_id, session=session)
|
||||
tag = Tag.find_by_name(tag_id_or_name) if isinstance(tag_id_or_name, str) else Tag.find_by_id(tag_id_or_name)
|
||||
return self.file_group.add_tag(tag.id, session=session)
|
||||
|
||||
async def get_ffprobe_json(self) -> dict:
|
||||
"""Return the ffprobe json object if previously stored.
|
||||
|
|
|
|||
|
|
@ -225,13 +225,13 @@ def test_search_videos(test_session, video_factory, assert_video_search, simple_
|
|||
vid3: Video = video_factory(upload_date='2022-09-18', with_video_file=True, title='vid3')
|
||||
tag1, tag2 = tag_factory(), tag_factory()
|
||||
|
||||
vid1.add_tag(tag1)
|
||||
vid1.add_tag(tag1.id)
|
||||
|
||||
# vid2 has two tags
|
||||
vid2.add_tag(tag1)
|
||||
vid2.add_tag(tag2)
|
||||
vid2.add_tag(tag1.id)
|
||||
vid2.add_tag(tag2.id)
|
||||
|
||||
vid3.add_tag(tag2)
|
||||
vid3.add_tag(tag2.id)
|
||||
|
||||
test_session.commit()
|
||||
assert test_session.query(Video).count() == 3
|
||||
|
|
|
|||
|
|
@ -45,11 +45,12 @@ def test_delete_video(test_session, video_factory):
|
|||
assert test_session.query(FileGroup).count() == 0, 'Video files were not deleted.'
|
||||
|
||||
|
||||
def test_delete_video_with_tag(test_session, video_factory, tag_factory):
|
||||
@pytest.mark.asyncio
|
||||
async def test_delete_video_with_tag(test_async_client, test_session, video_factory, tag_factory):
|
||||
"""You cannot delete a video with a tag."""
|
||||
video = video_factory(with_video_file=True, with_poster_ext='png')
|
||||
tag = tag_factory()
|
||||
video.add_tag(tag)
|
||||
video.add_tag(tag.name)
|
||||
test_session.commit()
|
||||
|
||||
with pytest.raises(FileGroupIsTagged):
|
||||
|
|
|
|||
|
|
@ -151,7 +151,7 @@ def headline_zim(search_str: str, zim_id: int, tag_names: List[str] = None, offs
|
|||
# Remove duplicate entries (disambiguation or redirect).
|
||||
results = get_unique_paths(*results, *search_results)
|
||||
results = results[:limit]
|
||||
entries = [zim.get_entry(i) for i in results]
|
||||
entries = [zim.find_entry(i) for i in results]
|
||||
estimate = zim.estimate(search_str)
|
||||
else:
|
||||
raise RuntimeError('Must provide search_str or tag_names.')
|
||||
|
|
@ -238,7 +238,7 @@ def get_zim(zim_id: int, session: Session = None) -> Zim:
|
|||
def get_entry(path: str, zim_id: int) -> Entry:
|
||||
with get_db_session() as session:
|
||||
zim = get_zim(zim_id, session=session)
|
||||
entry = zim.get_entry(path, throw=True)
|
||||
entry = zim.find_entry(path)
|
||||
return entry
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -163,61 +163,54 @@ class Zim(Base):
|
|||
results = list(suggestions.getResults(offset, limit))
|
||||
return results
|
||||
|
||||
def get_entry(self, path: str, throw: bool = True) -> Optional[Entry]:
|
||||
def get_entry(self, path: str) -> Optional[Entry]:
|
||||
"""Search this Zim file for the provided Entry at the `path`."""
|
||||
zim = self.get_zim()
|
||||
try:
|
||||
entry = zim.get_entry_by_path(path)
|
||||
except KeyError:
|
||||
if throw:
|
||||
raise UnknownZimEntry(f'Cannot find entry at {path=}')
|
||||
else:
|
||||
return None
|
||||
return None
|
||||
return entry
|
||||
|
||||
def get_entry_html(self, path: str) -> Optional[str]:
|
||||
"""Search this Zim file for the provided Entry at `path`, read its contents and return the HTML."""
|
||||
content = bytes(self.get_entry(path).get_item().content).decode('UTF-8')
|
||||
return content
|
||||
def find_entry(self, path: str) -> Entry:
|
||||
"""Search this Zim file for the provided Entry at the `path`.
|
||||
|
||||
def get_entry_text(self, path: str) -> Optional[str]:
|
||||
"""Search this Zim file for the provided Entry at `path`, read its contents and return the text in the HTML."""
|
||||
html = self.get_entry_html(path)
|
||||
content = extract_html_text(html)
|
||||
return content
|
||||
@raise UnknownZimEntry: if an entry at path does not exist."""
|
||||
if entry := self.get_entry(path):
|
||||
return entry
|
||||
|
||||
def tag_entry(self, tag_name: str, zim_entry: str) -> 'TagZimEntry':
|
||||
raise UnknownZimEntry(f'Cannot find entry at {path=}')
|
||||
|
||||
def tag_entry(self, tag_id_or_name: int | str, zim_entry: str) -> 'TagZimEntry':
|
||||
"""Create a TagZimEntry for this Zim at the provided entry (path) for the provided Tag."""
|
||||
session = Session.object_session(self)
|
||||
tag = Tag.get_by_name(tag_name, session=session)
|
||||
|
||||
# Ensure the entry exists.
|
||||
self.find_entry(zim_entry)
|
||||
|
||||
tag = Tag.get_by_name(tag_id_or_name, session) if isinstance(tag_id_or_name, str) \
|
||||
else Tag.get_by_id(tag_id_or_name, session)
|
||||
tag_zim_entry = TagZimEntry(tag=tag, zim=self, zim_entry=zim_entry)
|
||||
session.add(tag_zim_entry)
|
||||
tags.schedule_save()
|
||||
tags.save_tags_config()
|
||||
return tag_zim_entry
|
||||
|
||||
def untag_entry(self, tag_name: str, zim_entry: str):
|
||||
def untag_entry(self, tag_id_or_name: int | str, zim_entry: str):
|
||||
"""Removes any TagZimEntry of the Zim entry at the provided path for the provided Tag.
|
||||
|
||||
@raise UnknownZimTagEntry: No entry exists to remove."""
|
||||
session = Session.object_session(self)
|
||||
tag = Tag.get_by_name(tag_name, session=session)
|
||||
tag_zim_entry = session.query(TagZimEntry).filter_by(
|
||||
tag_id=tag.id,
|
||||
zim_id=self.id,
|
||||
zim_entry=zim_entry,
|
||||
).one_or_none()
|
||||
tag = Tag.get_by_name(tag_id_or_name, session) if isinstance(tag_id_or_name, str) \
|
||||
else Tag.get_by_id(tag_id_or_name, session)
|
||||
tag_zim_entry = TagZimEntry.get_by_primary_keys(tag.id, self.id, zim_entry, session)
|
||||
if tag_zim_entry:
|
||||
session.delete(tag_zim_entry)
|
||||
else:
|
||||
raise UnknownZimTagEntry(f'Could not find at {repr(str(zim_entry))}')
|
||||
tags.schedule_save()
|
||||
|
||||
@optional_session
|
||||
def entries_with_tags(self, tag_names: List[str], offset: int = 0, limit: int = 10, session: Session = None) \
|
||||
-> List[Entry]:
|
||||
"""Return Zim Entries tagged with the provided names."""
|
||||
tags_sub_select, params = tag_names_to_zim_sub_select(tag_names, zim_id=self.id)
|
||||
tags.save_tags_config()
|
||||
|
||||
@staticmethod
|
||||
def _entries_with_tags_process(limit: int, offset: int, params: dict, session: Session, tags_sub_select: str):
|
||||
with get_db_curs() as curs:
|
||||
curs.execute(tags_sub_select, params)
|
||||
zim_ids_entries: List[Tuple[int, int]] = list(curs.fetchall())
|
||||
|
|
@ -231,11 +224,31 @@ class Zim(Base):
|
|||
|
||||
entries = list()
|
||||
for tag_zim_entry, zim in tag_zim_entries:
|
||||
entry = zim.get_entry(tag_zim_entry.zim_entry)
|
||||
entry = zim.find_entry(tag_zim_entry.zim_entry)
|
||||
entries.append(entry)
|
||||
|
||||
return entries
|
||||
|
||||
@optional_session
|
||||
def entries_with_tags(*args, session: Session = None, **kwargs) -> List[Entry]:
|
||||
"""Return Zim Entries tagged with the provided names."""
|
||||
limit = int(kwargs.pop('limit', 10))
|
||||
offset = int(kwargs.pop('offset', 0))
|
||||
args = list(args)
|
||||
|
||||
zim: Zim | None = args.pop(0) if isinstance(args[0], Zim) else None
|
||||
zim_id = zim.id if zim else None
|
||||
tag_names: List[str] = args.pop(0)
|
||||
|
||||
if limit < 1:
|
||||
raise RuntimeError('Limit must be greater than zero')
|
||||
if args:
|
||||
raise RuntimeError('Extra unknown arguments')
|
||||
|
||||
tags_sub_select, params = tag_names_to_zim_sub_select(tag_names, zim_id=zim_id)
|
||||
entries = Zim._entries_with_tags_process(limit, offset, params, session, tags_sub_select)
|
||||
return entries
|
||||
|
||||
@functools.cached_property
|
||||
def all_entries(self) -> Set[EntrySummary]:
|
||||
"""
|
||||
|
|
@ -258,11 +271,6 @@ class Zim(Base):
|
|||
entries.add(summary)
|
||||
return entries
|
||||
|
||||
def parse_name(self):
|
||||
from modules.zim import lib
|
||||
name, date = lib.parse_name(self.path)
|
||||
return name, date
|
||||
|
||||
|
||||
class Zims:
|
||||
"""Convenience class which searches all known Zim files."""
|
||||
|
|
@ -317,7 +325,13 @@ class TagZimEntry(Base):
|
|||
zim = self.zim_id
|
||||
if self.zim:
|
||||
zim = get_relative_to_media_directory(self.zim.path)
|
||||
return f'<TagZimEntry {tag=} {zim=} zim_entry={self.zim_entry}>'
|
||||
return f'<TagZimEntry {tag=} zim={str(zim)} zim_entry={self.zim_entry}>'
|
||||
|
||||
@staticmethod
|
||||
def get_by_primary_keys(tag_id: int, zim_id: int, zim_entry: str, session: Session):
|
||||
return session.query(TagZimEntry) \
|
||||
.filter_by(tag_id=tag_id, zim_id=zim_id, zim_entry=zim_entry) \
|
||||
.one_or_none()
|
||||
|
||||
|
||||
class ZimSubscription(Base):
|
||||
|
|
|
|||
|
|
@ -5,7 +5,7 @@ import pytest
|
|||
from libzim import Entry
|
||||
|
||||
from modules.zim import lib
|
||||
from modules.zim.errors import UnknownZim
|
||||
from modules.zim.errors import UnknownZim, UnknownZimEntry
|
||||
from modules.zim.models import TagZimEntry, ZimSubscription
|
||||
from wrolpi import tags, flags
|
||||
from wrolpi.common import DownloadFileInfo, get_wrolpi_config
|
||||
|
|
@ -49,6 +49,13 @@ async def test_zim_get_entry(test_async_client, test_session, zim_path_factory):
|
|||
assert entry.title == 'Homepage'
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_zim_bad_entry(test_async_client, test_session, test_zim, tag_factory):
|
||||
tag1 = tag_factory()
|
||||
with pytest.raises(UnknownZimEntry):
|
||||
test_zim.tag_entry(tag1.name, 'path does not exist')
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_zim_get_entries_tags(test_async_client, test_session, test_zim, tag_factory):
|
||||
tag1, tag2 = tag_factory('tag1'), tag_factory('tag2')
|
||||
|
|
@ -61,6 +68,37 @@ async def test_zim_get_entries_tags(test_async_client, test_session, test_zim, t
|
|||
== dict(one=['tag1', 'tag2'], two=['tag1'], home=[])
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_tag_zim_entry_model(test_async_client, test_session, zim_factory, tag_factory):
|
||||
tze = zim_factory('zim1').tag_entry(tag_factory().name, 'one')
|
||||
# Test TagZimEntry.__repr__
|
||||
assert repr(tze) == "<TagZimEntry tag='one' zim=zim1 zim_entry=one>"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_zim_entries_with_tags(test_async_client, test_session, zim_factory, tag_factory):
|
||||
"""Can fetch Zim Entries that have Tags. Can filter only those entries that are tagged on a specific Zim."""
|
||||
tag1, tag2 = tag_factory(), tag_factory()
|
||||
zim1, zim2 = zim_factory('zim1'), zim_factory('zim2')
|
||||
# Tag Zim1 with both tags, but Zim2 shares one of the tags and is returned when no Zim is passed.
|
||||
zim1.tag_entry(tag1.name, 'one')
|
||||
zim1.tag_entry(tag2.name, 'two')
|
||||
zim2.tag_entry(tag2.name, 'home')
|
||||
test_session.commit()
|
||||
|
||||
# Get entries from zim1 only.
|
||||
entries = zim1.entries_with_tags([tag1.name, ])
|
||||
assert len(entries) == 1
|
||||
assert entries[0].path == 'one'
|
||||
entries = zim1.entries_with_tags([tag2.name, ])
|
||||
assert len(entries) == 1
|
||||
assert entries[0].path == 'two'
|
||||
|
||||
# Get entries from any Zim.
|
||||
entries = lib.Zim.entries_with_tags([tag2.name, ])
|
||||
assert {i.path for i in entries} == {'two', 'home'}, 'Tag2 was applied to `two` and `home`'
|
||||
|
||||
|
||||
def test_get_unique_paths():
|
||||
paths = ['A/Prickly_Pear', 'A/Prickly_pear', 'A/Prickly-pear', 'A/Arborescent_Prickly_Pear',
|
||||
'A/Arborescent_prickly_pear', 'A/Beavertail_prickly_pear', 'A/Brittle_prickly-pear',
|
||||
|
|
|
|||
|
|
@ -1844,15 +1844,21 @@ def get_title_from_html(html: str, url: str = None) -> str:
|
|||
logger_.debug(f'Unable to extract title {url}')
|
||||
|
||||
|
||||
def can_connect_to_server(hostname: str) -> bool:
|
||||
"""Return True only if `hostname` responds on port 80."""
|
||||
def can_connect_to_server(hostname: str, port: int = 80) -> bool:
|
||||
"""Return True only if `hostname` responds."""
|
||||
# Thanks https://stackoverflow.com/questions/20913411/test-if-an-internet-connection-is-present-in-python
|
||||
try:
|
||||
# connect to the host -- tells us if the host is actually
|
||||
# reachable
|
||||
conn = socket.create_connection((hostname, 53))
|
||||
conn = socket.create_connection((hostname, port))
|
||||
conn.close()
|
||||
return True
|
||||
except Exception:
|
||||
pass # We ignore any errors, returning False
|
||||
return False
|
||||
|
||||
|
||||
def is_valid_hex_color(hex_color: str) -> bool:
|
||||
# Regular expression to match valid HTML hex color codes
|
||||
hex_color_regex = re.compile(r'^#([A-F0-9]{6}|[A-F0-9]{3})$', re.IGNORECASE)
|
||||
return bool(hex_color_regex.match(hex_color))
|
||||
|
|
|
|||
|
|
@ -1,14 +1,13 @@
|
|||
import pathlib
|
||||
import traceback
|
||||
from abc import ABC
|
||||
from typing import List
|
||||
|
||||
from wrolpi.common import get_media_directory, logger, background_task, get_download_info, \
|
||||
from wrolpi.common import get_media_directory, logger, get_download_info, \
|
||||
download_file, trim_file_name
|
||||
from wrolpi.db import get_db_session
|
||||
from wrolpi.downloader import Downloader, Download, DownloadResult
|
||||
from wrolpi.errors import UnrecoverableDownloadError
|
||||
from wrolpi.files.models import FileGroup
|
||||
from wrolpi.files.lib import upsert_file
|
||||
from wrolpi.vars import PYTEST
|
||||
|
||||
__all__ = ['FileDownloader', 'file_downloader']
|
||||
|
|
@ -52,10 +51,9 @@ class FileDownloader(Downloader, ABC):
|
|||
|
||||
try:
|
||||
await download_file(download.url, output_path, info)
|
||||
background_task(save_and_tag(output_path, download.settings.get('tag_names')))
|
||||
|
||||
with get_db_session(commit=True) as session:
|
||||
fg = FileGroup.from_paths(session, output_path)
|
||||
fg = await upsert_file(output_path, session, download.settings.get('tag_names'))
|
||||
location = fg.location
|
||||
|
||||
return DownloadResult(
|
||||
|
|
@ -74,22 +72,4 @@ class FileDownloader(Downloader, ABC):
|
|||
)
|
||||
|
||||
|
||||
async def save_and_tag(file: pathlib.Path, tag_names: List[str] = None):
|
||||
"""Stores the file as a FileGroup. Applies any provided Tags to the FileGroup."""
|
||||
from wrolpi.tags import Tag
|
||||
|
||||
tag_names = tag_names or list()
|
||||
|
||||
with get_db_session(commit=True) as session:
|
||||
file_group = FileGroup.get_by_path(file)
|
||||
if not file_group:
|
||||
file_group = FileGroup.from_paths(session, file)
|
||||
session.commit()
|
||||
|
||||
if tag_names:
|
||||
tags = session.query(Tag).filter(Tag.name.in_(tag_names))
|
||||
for tag in tags:
|
||||
file_group.add_tag(tag)
|
||||
|
||||
|
||||
file_downloader = FileDownloader()
|
||||
|
|
|
|||
|
|
@ -5,7 +5,7 @@ import pathlib
|
|||
from typing import List, Optional, Tuple
|
||||
|
||||
from sqlalchemy import Column, Integer, BigInteger, ForeignKey, or_, and_
|
||||
from sqlalchemy.orm import relationship
|
||||
from sqlalchemy.orm import relationship, Session
|
||||
|
||||
from wrolpi.common import ModelHelper, Base, register_modeler, get_html_soup
|
||||
from wrolpi.db import get_db_session
|
||||
|
|
@ -23,13 +23,16 @@ except ImportError:
|
|||
logger = logging.getLogger(__name__)
|
||||
|
||||
__all__ = [
|
||||
'EBOOK_MIMETYPES',
|
||||
'EBook',
|
||||
'EBookData',
|
||||
'EPUB_MIMETYPE',
|
||||
'MOBI_MIMETYPE',
|
||||
'ebook_modeler',
|
||||
'extract_ebook_cover',
|
||||
'extract_ebook_data',
|
||||
'ebook_modeler',
|
||||
'MOBI_MIMETYPE',
|
||||
'EPUB_MIMETYPE'
|
||||
'mimetype_is_ebook',
|
||||
'model_ebook',
|
||||
]
|
||||
|
||||
EBOOK_SUFFIXES = ('.epub', '.mobi')
|
||||
|
|
@ -211,11 +214,14 @@ class EBook(ModelHelper, Base):
|
|||
session.delete(file_group)
|
||||
|
||||
|
||||
def _model_ebook(ebook: EBook) -> EBook:
|
||||
"""Creates an EBook model based off a File. Searches for it's cover in the provided `files`."""
|
||||
# Multiple formats may share this cover.
|
||||
epub_file = next(iter(ebook.file_group.my_files('application/epub')), None)
|
||||
mobi_file = next(iter(ebook.file_group.my_files('application/x-mobipocket-ebook')), None)
|
||||
def model_ebook(file_group: FileGroup, session: Session) -> EBook:
|
||||
"""Creates an EBook model based off a File. Searches for its cover in the provided `files`."""
|
||||
ebook = EBook(file_group=file_group)
|
||||
session.add(ebook)
|
||||
|
||||
# Multiple formats may be available.
|
||||
epub_file = next(iter(ebook.file_group.my_epub_files()), None)
|
||||
mobi_file = next(iter(ebook.file_group.my_mobi_files()), None)
|
||||
# epub is preferred because it can be indexed.
|
||||
if epub_file:
|
||||
ebook_file = epub_file['path']
|
||||
|
|
@ -290,11 +296,10 @@ async def ebook_modeler():
|
|||
for file_group, ebook in file_groups:
|
||||
processed += 1
|
||||
try:
|
||||
ebook = ebook or EBook(file_group=file_group)
|
||||
if PYTEST:
|
||||
# Handle caching issue during testing.
|
||||
session.expire(file_group)
|
||||
_model_ebook(ebook)
|
||||
ebook = ebook or model_ebook(file_group, session)
|
||||
session.add(ebook)
|
||||
file_group.model = EBook.__tablename__
|
||||
file_group.indexed = True
|
||||
|
|
|
|||
|
|
@ -48,7 +48,7 @@ logger = logger.getChild(__name__)
|
|||
|
||||
__all__ = ['list_directories_contents', 'delete', 'split_path_stem_and_suffix', 'refresh_files', 'search_files',
|
||||
'get_mimetype', 'split_file_name_words', 'get_primary_file', 'get_file_statistics',
|
||||
'search_file_suggestion_count', 'glob_shared_stem',
|
||||
'search_file_suggestion_count', 'glob_shared_stem', 'upsert_file',
|
||||
'move', 'rename', 'delete_directory', 'handle_file_group_search_results']
|
||||
|
||||
|
||||
|
|
@ -1018,7 +1018,7 @@ async def _get_tag_and_file_group(file_group_id: int, file_group_primary_path: s
|
|||
async def add_file_group_tag(file_group_id: int, file_group_primary_path: str, tag_name: str, tag_id: int,
|
||||
session: Session = None) -> TagFile:
|
||||
file_group, tag = await _get_tag_and_file_group(file_group_id, file_group_primary_path, tag_name, tag_id, session)
|
||||
tag_file = file_group.add_tag(tag, session)
|
||||
tag_file = file_group.add_tag(tag.id, session)
|
||||
return tag_file
|
||||
|
||||
|
||||
|
|
@ -1026,7 +1026,7 @@ async def add_file_group_tag(file_group_id: int, file_group_primary_path: str, t
|
|||
async def remove_file_group_tag(file_group_id: int, file_group_primary_path: str, tag_name: str, tag_id: int,
|
||||
session: Session = None):
|
||||
file_group, tag = await _get_tag_and_file_group(file_group_id, file_group_primary_path, tag_name, tag_id, session)
|
||||
file_group.untag(tag, session)
|
||||
file_group.untag(tag.id, session)
|
||||
|
||||
|
||||
@dataclasses.dataclass
|
||||
|
|
@ -1398,13 +1398,19 @@ async def upsert_file(file: pathlib.Path | str, session: Session = None, tag_nam
|
|||
# Re-index the contents of the file.
|
||||
session.add(file_group)
|
||||
session.flush([file_group, ])
|
||||
file_group.do_index()
|
||||
file_group.indexed = True
|
||||
|
||||
try:
|
||||
file_group.do_model(session)
|
||||
except Exception as e:
|
||||
logger.error(f'Failed to model FileGroup: {file_group}', exc_info=e)
|
||||
if PYTEST:
|
||||
raise
|
||||
|
||||
for tag_name in (tag_names or []):
|
||||
if tag_name not in file_group.tag_names:
|
||||
tag = Tag.get_by_name(tag_name)
|
||||
file_group.add_tag(tag.id)
|
||||
|
||||
session.commit()
|
||||
else:
|
||||
logger.warning('upsert_file called, but all files are in ignored directories!')
|
||||
|
|
|
|||
|
|
@ -14,7 +14,7 @@ from wrolpi.common import Base, ModelHelper, tsvector, logger, recursive_map, ge
|
|||
get_relative_to_media_directory
|
||||
from wrolpi.dates import TZDateTime, now, from_timestamp, strptime_ms, strftime
|
||||
from wrolpi.db import optional_session
|
||||
from wrolpi.errors import FileGroupIsTagged
|
||||
from wrolpi.errors import FileGroupIsTagged, UnknownFile
|
||||
from wrolpi.files import indexers
|
||||
from wrolpi.media_path import MediaPathType
|
||||
from wrolpi.tags import Tag, TagFile
|
||||
|
|
@ -144,15 +144,12 @@ class FileGroup(ModelHelper, Base):
|
|||
self.viewed = now()
|
||||
|
||||
@optional_session
|
||||
def add_tag(self, tag_id: int, session: Session = None) -> TagFile:
|
||||
return Tag.tag_file_group(self.id, tag_id, session)
|
||||
def add_tag(self, tag_id_or_name: int | str, session: Session = None) -> TagFile:
|
||||
return Tag.tag_file_group(self.id, tag_id_or_name, session)
|
||||
|
||||
@optional_session
|
||||
def untag(self, tag_id: int, session: Session = None):
|
||||
if not isinstance(tag_id, int):
|
||||
raise RuntimeError('tag_id must be an integer')
|
||||
|
||||
Tag.untag_file_group(self.id, tag_id, session)
|
||||
def untag(self, tag_id_or_name: int | str, session: Session = None):
|
||||
Tag.untag_file_group(self.id, tag_id_or_name, session)
|
||||
|
||||
def append_files(self, *paths: pathlib.Path):
|
||||
"""Add all `paths` to this FileGroup.files."""
|
||||
|
|
@ -209,8 +206,11 @@ class FileGroup(ModelHelper, Base):
|
|||
subtitle_paths = [i['path'] for i in self.my_subtitle_files()]
|
||||
return [i for i in text_files if i['path'] not in subtitle_paths]
|
||||
|
||||
def my_ebook_files(self) -> List[dict]:
|
||||
return self.my_files('application/epub', 'application/x-mobipocket-ebook')
|
||||
def my_epub_files(self) -> List[dict]:
|
||||
return self.my_files('application/epub')
|
||||
|
||||
def my_mobi_files(self) -> List[dict]:
|
||||
return self.my_files('application/x-mobipocket-ebook')
|
||||
|
||||
def delete(self):
|
||||
"""Delete this FileGroup record, and all of its files.
|
||||
|
|
@ -320,7 +320,7 @@ class FileGroup(ModelHelper, Base):
|
|||
for tag_file in file_group.tag_files:
|
||||
if tag_file.tag.name not in self.tag_names:
|
||||
# Preserve the created at.
|
||||
self.add_tag(tag_file.tag).created_at = tag_file.created_at
|
||||
self.add_tag(tag_file.tag.id).created_at = tag_file.created_at
|
||||
session.delete(file_group)
|
||||
|
||||
self.files = collected_files
|
||||
|
|
@ -360,6 +360,50 @@ class FileGroup(ModelHelper, Base):
|
|||
query = urllib.parse.urlencode(dict(folders=str(parent), preview=str(preview)))
|
||||
return f'/files?{query}'
|
||||
|
||||
def get_model_class(self) -> Optional[Type[Base]]:
|
||||
from modules.videos.models import Video
|
||||
from modules.archive.lib import is_singlefile_file
|
||||
from modules.archive.models import Archive
|
||||
from wrolpi.files.ebooks import EBook
|
||||
from wrolpi.files.ebooks import mimetype_is_ebook
|
||||
|
||||
if self.mimetype.startswith('video/'):
|
||||
return Video
|
||||
elif self.mimetype.startswith('text') and is_singlefile_file(self.primary_path):
|
||||
return Archive
|
||||
elif mimetype_is_ebook(self.mimetype):
|
||||
return EBook
|
||||
|
||||
def get_model(self, session: Session) -> Optional[Base]:
|
||||
if model_class := self.get_model_class():
|
||||
return model_class.get_by_path(self.primary_path, session)
|
||||
|
||||
def do_model(self, session: Session) -> Base:
|
||||
"""Create the Model record (Video/Archive/etc.) for this FileGroup, if any."""
|
||||
if model := self.get_model(session):
|
||||
return model
|
||||
|
||||
from modules.archive.lib import is_singlefile_file
|
||||
from modules.archive import model_archive
|
||||
from modules.videos.models import Video
|
||||
from wrolpi.files.ebooks import model_ebook, mimetype_is_ebook
|
||||
|
||||
if self.mimetype.startswith('video/'):
|
||||
video = Video(file_group_id=self.id, file_group=self)
|
||||
video.validate()
|
||||
self.indexed = True
|
||||
return video
|
||||
elif self.mimetype.startswith('text') and is_singlefile_file(self.primary_path):
|
||||
archive = model_archive(self, session)
|
||||
self.indexed = True
|
||||
return archive
|
||||
elif mimetype_is_ebook(self.mimetype):
|
||||
ebook = model_ebook(self, session)
|
||||
self.indexed = True
|
||||
return ebook
|
||||
|
||||
raise UnknownFile(f'Cannot model file with mimetype {self.mimetype}')
|
||||
|
||||
|
||||
class Directory(ModelHelper, Base):
|
||||
"""A representation of a file directory in the media directory."""
|
||||
|
|
|
|||
|
|
@ -201,8 +201,8 @@ async def test_files_search_any_tag(test_async_client, test_session, make_files_
|
|||
assert bar.primary_path.name == 'bar.txt' \
|
||||
and foo.primary_path.name == 'foo.txt' \
|
||||
and foobar.primary_path.name == 'foo bar.txt'
|
||||
foo.add_tag(one)
|
||||
foobar.add_tag(two)
|
||||
foo.add_tag(one.id)
|
||||
foobar.add_tag(two.id)
|
||||
test_session.commit()
|
||||
|
||||
# Only `foo` is tagged with `one`
|
||||
|
|
@ -457,10 +457,8 @@ def test_post_upload_directory(test_session, test_client, test_directory, make_f
|
|||
dict(name='chunk', value='foo', filename='chunk')
|
||||
]
|
||||
body = make_multipart_form(forms)
|
||||
request, response = test_client.post('/api/files/upload', content=body,
|
||||
headers={
|
||||
'Content-Type': 'multipart/form-data; name=upload; filename="file.txt";'
|
||||
' boundary=-----sanic'})
|
||||
headers = {'Content-Type': 'multipart/form-data; boundary=-----------------------------sanic'}
|
||||
request, response = test_client.post('/api/files/upload', content=body, headers=headers)
|
||||
assert response.status_code == HTTPStatus.CREATED
|
||||
|
||||
assert (test_directory / 'uploads/foo/bar.txt').is_file()
|
||||
|
|
@ -516,6 +514,7 @@ def test_post_upload(test_session, test_client, test_directory, make_files_struc
|
|||
assert hashlib.md5(output.read_bytes()).hexdigest() == '2738c53bd7c01b01d408da11a55bfa36'
|
||||
|
||||
file_group: FileGroup = test_session.query(FileGroup).one()
|
||||
assert file_group.mimetype == 'video/mp4'
|
||||
assert set(file_group.tag_names) == {tag1.name, tag2.name}, 'Two tags should be applied.'
|
||||
assert file_group.indexed, 'File should be indexed after upload.'
|
||||
assert file_group.model, 'File should be modeled'
|
||||
|
|
|
|||
|
|
@ -11,7 +11,7 @@ from wrolpi.test.common import assert_dict_contains
|
|||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_index(test_session, test_directory, example_epub, example_mobi):
|
||||
async def test_index(test_async_client, test_session, test_directory, example_epub, example_mobi):
|
||||
"""Ebooks can be indexed.
|
||||
|
||||
Covers can be discovered."""
|
||||
|
|
@ -135,7 +135,7 @@ async def test_discover_calibre_cover(test_session, test_directory, example_epub
|
|||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_move_ebook(test_session, test_directory, example_epub, image_file, tag_factory):
|
||||
async def test_move_ebook(test_async_client, test_session, test_directory, example_epub, image_file, tag_factory):
|
||||
"""An ebook is re-indexed when moved."""
|
||||
tag = tag_factory()
|
||||
shutil.move(image_file, example_epub.with_suffix('.jpg'))
|
||||
|
|
@ -148,7 +148,7 @@ async def test_move_ebook(test_session, test_directory, example_epub, image_file
|
|||
test_directory / 'ebook example.epub',
|
||||
test_directory / 'ebook example.jpg',
|
||||
]
|
||||
ebook.file_group.add_tag(tag)
|
||||
ebook.file_group.add_tag(tag.id)
|
||||
test_session.commit()
|
||||
|
||||
new_directory = test_directory / 'new'
|
||||
|
|
|
|||
|
|
@ -82,7 +82,7 @@ async def test_delete_file_names(test_async_client, test_session, make_files_str
|
|||
foo_fg = FileGroup.from_paths(test_session, foo)
|
||||
foo1_fg = FileGroup.from_paths(test_session, foo1)
|
||||
tag = tag_factory()
|
||||
foo1_fg.add_tag(tag)
|
||||
foo1_fg.add_tag(tag.id)
|
||||
test_session.commit()
|
||||
assert foo.is_file()
|
||||
assert foo1.is_file()
|
||||
|
|
@ -110,7 +110,7 @@ async def test_delete_tagged(test_session, make_files_structure, tag_factory, vi
|
|||
await lib.refresh_files()
|
||||
# Both files end up in a group.
|
||||
bar = test_session.query(FileGroup).one()
|
||||
bar.add_tag(tag)
|
||||
bar.add_tag(tag.id)
|
||||
test_session.commit()
|
||||
|
||||
# Neither file can be deleted.
|
||||
|
|
@ -367,7 +367,7 @@ async def test_file_group_tag(test_session, make_files_structure, test_directory
|
|||
one = tag_factory()
|
||||
|
||||
foo: FileGroup = test_session.query(FileGroup).one()
|
||||
foo.add_tag(one)
|
||||
foo.add_tag(one.id)
|
||||
test_session.commit()
|
||||
|
||||
tag_file2: TagFile = test_session.query(TagFile).one()
|
||||
|
|
@ -735,8 +735,8 @@ def test_file_group_merge(test_async_client, test_session, test_directory, make_
|
|||
srt_group = FileGroup.from_paths(test_session, srt)
|
||||
test_session.add_all([vid_group, srt_group])
|
||||
test_session.flush([vid_group, srt_group])
|
||||
vid_tag_file = vid_group.add_tag(one, test_session)
|
||||
srt_tag_file = srt_group.add_tag(two, test_session)
|
||||
vid_tag_file = vid_group.add_tag(one.name, test_session)
|
||||
srt_tag_file = srt_group.add_tag(two.name, test_session)
|
||||
test_session.flush([vid_tag_file, srt_tag_file])
|
||||
tag_file_created_at = vid_tag_file.created_at
|
||||
srt_file_created_at = srt_tag_file.created_at
|
||||
|
|
@ -936,7 +936,7 @@ async def test_move_tagged(test_async_client, test_session, test_directory, make
|
|||
})
|
||||
await lib.refresh_files()
|
||||
bar_file_group, foo_file_group = test_session.query(FileGroup).order_by(FileGroup.primary_path)
|
||||
bar_file_group.add_tag(tag)
|
||||
bar_file_group.add_tag(tag.id)
|
||||
test_session.commit()
|
||||
|
||||
qux = test_directory / 'qux'
|
||||
|
|
|
|||
|
|
@ -9,7 +9,7 @@ from sqlalchemy.orm import relationship, Session
|
|||
|
||||
from wrolpi import dates
|
||||
from wrolpi.common import ModelHelper, Base, logger, ConfigFile, get_media_directory, background_task, run_after, \
|
||||
register_refresh_cleanup, get_relative_to_media_directory
|
||||
register_refresh_cleanup, get_relative_to_media_directory, is_valid_hex_color
|
||||
from wrolpi.dates import TZDateTime
|
||||
from wrolpi.db import optional_session, get_db_curs
|
||||
from wrolpi.errors import UnknownTag, UsedTag, InvalidTag, FileGroupAlreadyTagged
|
||||
|
|
@ -32,6 +32,13 @@ class TagFile(ModelHelper, Base):
|
|||
def get_by_primary_keys(file_group_id: int, tag_id: int, session: Session = None) -> Optional['TagFile']:
|
||||
return session.query(TagFile).filter_by(tag_id=tag_id, file_group_id=file_group_id).one_or_none()
|
||||
|
||||
@staticmethod
|
||||
@optional_session
|
||||
def get_by_tag_name(file_group_id: int, tag_name: str, session: Session = None) -> Optional['TagFile']:
|
||||
return session.query(TagFile).join(Tag) \
|
||||
.filter(TagFile.file_group_id == file_group_id, Tag.name == tag_name) \
|
||||
.one_or_none()
|
||||
|
||||
|
||||
class Tag(ModelHelper, Base):
|
||||
__tablename__ = 'tag'
|
||||
|
|
@ -56,38 +63,50 @@ class Tag(ModelHelper, Base):
|
|||
|
||||
@staticmethod
|
||||
@optional_session
|
||||
def tag_file_group(file_group_id, tag_id, session: Session = None) -> TagFile:
|
||||
def get_tag_file_group(file_group_id: int, tag_id_or_name: int | str, session: Session = None) -> Optional[TagFile]:
|
||||
if isinstance(tag_id_or_name, int):
|
||||
return TagFile.get_by_primary_keys(file_group_id, tag_id_or_name, session)
|
||||
|
||||
return TagFile.get_by_tag_name(file_group_id, tag_id_or_name, session)
|
||||
|
||||
@staticmethod
|
||||
@optional_session
|
||||
def tag_file_group(file_group_id: int, tag_id_or_name: int | str, session: Session = None) -> TagFile:
|
||||
"""Add a TagFile for the provided FileGroup and this Tag.
|
||||
|
||||
@warning: Commits the session to keep the config in sync."""
|
||||
existing = TagFile.get_by_primary_keys(file_group_id, tag_id, session)
|
||||
existing = Tag.get_tag_file_group(file_group_id, tag_id_or_name, session)
|
||||
|
||||
if existing:
|
||||
raise FileGroupAlreadyTagged('Tag already used')
|
||||
tag_file = TagFile(file_group_id=file_group_id, tag_id=tag_id)
|
||||
|
||||
tag = Tag.get_by_name(tag_id_or_name, session) if isinstance(tag_id_or_name, str) \
|
||||
else Tag.get_by_id(tag_id_or_name, session)
|
||||
tag_file = TagFile(file_group_id=file_group_id, tag_id=tag.id)
|
||||
session.add(tag_file)
|
||||
session.flush([tag_file])
|
||||
session.commit()
|
||||
logger.info(f'Tagged FileGroup {file_group_id} with Tag {tag_id}')
|
||||
logger.info(f'Tagged FileGroup {file_group_id} with Tag {tag_id_or_name}')
|
||||
|
||||
# Save changes to config.
|
||||
schedule_save(session)
|
||||
save_tags_config(session)
|
||||
return tag_file
|
||||
|
||||
@staticmethod
|
||||
@optional_session
|
||||
def untag_file_group(file_group_id: int, tag_id: int, session: Session = None):
|
||||
def untag_file_group(file_group_id: int, tag_id_or_name: int | str, session: Session = None):
|
||||
"""Remove the record of a Tag applied to the FileGroup.
|
||||
|
||||
@warning: Commits the session to keep config in sync."""
|
||||
tag_file = TagFile.get_by_primary_keys(file_group_id, tag_id, session)
|
||||
tag_file = Tag.get_tag_file_group(file_group_id, tag_id_or_name, session)
|
||||
if tag_file:
|
||||
session.delete(tag_file)
|
||||
session.commit()
|
||||
|
||||
# Save changes to config.
|
||||
schedule_save(session)
|
||||
save_tags_config(session)
|
||||
else:
|
||||
logger.warning(f'Could not find TagFile for FileGroup.id={file_group_id}/Tag.id={tag_id}')
|
||||
logger.warning(f'Could not find TagFile for FileGroup.id={file_group_id}/Tag.id={tag_id_or_name}')
|
||||
|
||||
@staticmethod
|
||||
@optional_session
|
||||
|
|
@ -97,10 +116,26 @@ class Tag(ModelHelper, Base):
|
|||
|
||||
@staticmethod
|
||||
@optional_session
|
||||
def get_by_id(id_: int, session: Session = None) -> Optional[Base]:
|
||||
def find_by_name(name: str, session: Session) -> 'Tag':
|
||||
if tag := Tag.get_by_name(name, session):
|
||||
return tag
|
||||
|
||||
raise UnknownTag(f'Unknown Tag with name={name}')
|
||||
|
||||
@staticmethod
|
||||
@optional_session
|
||||
def get_by_id(id_: int, session: Session = None) -> Optional['Tag']:
|
||||
tag = session.query(Tag).filter_by(id=id_).one_or_none()
|
||||
return tag
|
||||
|
||||
@staticmethod
|
||||
@optional_session
|
||||
def find_by_id(id_: int, session: Session = None) -> 'Tag':
|
||||
if tag := Tag.get_by_id(id_, session):
|
||||
return tag
|
||||
|
||||
raise UnknownTag(f'Unknown Tag with id={id_}')
|
||||
|
||||
def has_relations(self) -> bool:
|
||||
"""Returns True if this Tag has been used with any FileGroups or Zim Entries."""
|
||||
return bool(self.tag_files or self.tag_zim_entries)
|
||||
|
|
@ -215,7 +250,7 @@ def test_tags_config():
|
|||
|
||||
|
||||
@optional_session
|
||||
def schedule_save(session: Session = None):
|
||||
def save_tags_config(session: Session = None):
|
||||
"""Schedule a background task to save all TagFiles to the config file. If testing, save synchronously."""
|
||||
if PYTEST:
|
||||
get_tags_config().save_tags(session)
|
||||
|
|
@ -241,10 +276,12 @@ def get_tags() -> List[dict]:
|
|||
|
||||
|
||||
@optional_session
|
||||
@run_after(schedule_save)
|
||||
@run_after(save_tags_config)
|
||||
def upsert_tag(name: str, color: str, tag_id: int = None, session: Session = None) -> Tag:
|
||||
if ',' in name:
|
||||
raise InvalidTag('Tag name cannot have comma')
|
||||
if not is_valid_hex_color(color):
|
||||
raise InvalidTag('Tag color is invalid')
|
||||
|
||||
if tag_id:
|
||||
tag = session.query(Tag).filter_by(id=tag_id).one_or_none()
|
||||
|
|
@ -264,7 +301,7 @@ def upsert_tag(name: str, color: str, tag_id: int = None, session: Session = Non
|
|||
session.rollback()
|
||||
raise InvalidTag(f'Name already taken') from e
|
||||
|
||||
schedule_save()
|
||||
save_tags_config()
|
||||
|
||||
return tag
|
||||
|
||||
|
|
@ -283,7 +320,7 @@ def delete_tag(tag_id: int, session: Session = None):
|
|||
session.delete(tag)
|
||||
session.commit()
|
||||
|
||||
schedule_save()
|
||||
save_tags_config()
|
||||
|
||||
|
||||
@register_refresh_cleanup
|
||||
|
|
@ -360,7 +397,7 @@ def import_tags_config(session: Session = None):
|
|||
tag_file: TagFile = tag_files.get((tag.id, file_group.id))
|
||||
if not tag_file:
|
||||
# This FileGroup has not been tagged with the Tag, add it.
|
||||
tag_file = tag.tag_file_group(file_group, session)
|
||||
tag_file = Tag.tag_file_group(file_group.id, tag.id, session)
|
||||
tag_file.created_at = dates.strptime_ms(created_at) if created_at else dates.now()
|
||||
need_commit = True
|
||||
elif not file_group:
|
||||
|
|
|
|||
|
|
@ -850,6 +850,15 @@ async def test_aiohttp(simple_web_server):
|
|||
assert status == 200
|
||||
|
||||
|
||||
def test_can_connect_to_server(simple_web_server):
|
||||
httpd, server_url = simple_web_server
|
||||
_, host, port = server_url.split(':')
|
||||
host = host[2:]
|
||||
port = int(port[:-1])
|
||||
|
||||
assert common.can_connect_to_server(host, port)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_log_level(test_async_client, test_config):
|
||||
"""User can change API log level."""
|
||||
|
|
@ -880,3 +889,15 @@ async def test_log_level(test_async_client, test_config):
|
|||
@skip_circleci
|
||||
def test_html_screenshot(singlefile_contents_factory):
|
||||
assert common.html_screenshot(singlefile_contents_factory())
|
||||
|
||||
|
||||
@pytest.mark.parametrize('color,expected', [
|
||||
('#ffffff', True),
|
||||
('#FFFFFF', True),
|
||||
('#fff', True),
|
||||
('#FFf', True),
|
||||
('#asdf', False),
|
||||
('#ffff', False),
|
||||
])
|
||||
def test_is_valid_hex_color(color, expected):
|
||||
assert common.is_valid_hex_color(color) == expected
|
||||
|
|
|
|||
|
|
@ -40,7 +40,8 @@ async def test_events_api_feed(test_session, test_async_client, example_pdf):
|
|||
assert_dict_contains(event, expected)
|
||||
|
||||
|
||||
def test_events_history_limit():
|
||||
@pytest.mark.asyncio
|
||||
async def test_events_history_limit(test_async_client):
|
||||
"""EVENTS_HISTORY has a limited size."""
|
||||
for i in range(HISTORY_SIZE + 5):
|
||||
Events.send_ready()
|
||||
|
|
|
|||
|
|
@ -5,7 +5,7 @@ import pytest
|
|||
import yaml
|
||||
|
||||
from wrolpi import tags
|
||||
from wrolpi.errors import FileGroupIsTagged
|
||||
from wrolpi.errors import FileGroupIsTagged, InvalidTag, UnknownTag, FileGroupAlreadyTagged
|
||||
from wrolpi.files import lib as files_lib
|
||||
from wrolpi.files.models import FileGroup
|
||||
from wrolpi.tags import TagFile, Tag
|
||||
|
|
@ -25,12 +25,36 @@ async def test_tags_file_group_json(test_async_client, test_session, make_files_
|
|||
file_group.add_tag(tag_two.id)
|
||||
assert file_group.__json__()['tags'] == ['one', 'two']
|
||||
|
||||
with pytest.raises(FileGroupAlreadyTagged):
|
||||
# Cannot tag a FileGroup with the same Tag twice.
|
||||
file_group.add_tag(tag_two.id)
|
||||
|
||||
file_group.untag(tag_one.id)
|
||||
assert file_group.__json__()['tags'] == ['two']
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_tags_file_group(test_async_client, test_session, make_files_structure, tag_factory, video_bytes, image_bytes_factory):
|
||||
async def test_tags_model(test_async_client, test_session, make_files_structure, tag_factory, example_pdf):
|
||||
"""Can get Tag using class methods."""
|
||||
tag1 = tag_factory()
|
||||
assert tag1.__json__()
|
||||
|
||||
assert Tag.find_by_id(tag1.id)
|
||||
assert Tag.find_by_name(tag1.name)
|
||||
|
||||
assert Tag.get_by_id(123) is None, '.get should return None with bad id'
|
||||
assert Tag.get_by_name('bad name') is None, '.get should return None with bad name'
|
||||
|
||||
with pytest.raises(UnknownTag):
|
||||
Tag.find_by_id(123)
|
||||
|
||||
with pytest.raises(UnknownTag):
|
||||
Tag.find_by_name('bad name')
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_tags_file_group(test_async_client, test_session, make_files_structure, tag_factory, video_bytes,
|
||||
image_bytes_factory):
|
||||
"""A FileGroup can be tagged with multiple Tags."""
|
||||
make_files_structure({
|
||||
'video.mp4': video_bytes, 'video.png': image_bytes_factory(),
|
||||
|
|
@ -64,7 +88,8 @@ async def test_tags_file_group(test_async_client, test_session, make_files_struc
|
|||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_tags_config_(test_async_client, test_session, test_directory, tag_factory, example_pdf, video_file, test_tags_config):
|
||||
async def test_tags_config_file(test_async_client, test_session, test_directory, tag_factory, example_pdf, video_file,
|
||||
test_tags_config):
|
||||
"""Test that the config is updated when a FileGroup is tagged."""
|
||||
await files_lib.refresh_files()
|
||||
pdf: FileGroup = FileGroup.get_by_path(example_pdf, test_session)
|
||||
|
|
@ -73,7 +98,7 @@ async def test_tags_config_(test_async_client, test_session, test_directory, tag
|
|||
tag2 = tag_factory()
|
||||
test_session.commit()
|
||||
|
||||
tags.schedule_save(test_session)
|
||||
tags.save_tags_config(test_session)
|
||||
assert tag1.name in test_tags_config.read_text()
|
||||
assert tag2.name in test_tags_config.read_text()
|
||||
|
||||
|
|
@ -134,7 +159,7 @@ async def test_tags_crud(test_async_client, test_session, example_pdf, assert_ta
|
|||
|
||||
# Apply the tag to the PDF.
|
||||
tag = test_session.query(tags.Tag).one()
|
||||
pdf.add_tag(tag, test_session)
|
||||
pdf.add_tag(tag.name, test_session)
|
||||
test_session.commit()
|
||||
|
||||
# The tag can be updated.
|
||||
|
|
@ -156,7 +181,7 @@ async def test_tags_crud(test_async_client, test_session, example_pdf, assert_ta
|
|||
request, response = await test_async_client.delete('/api/tag/1')
|
||||
assert response.status_code == HTTPStatus.BAD_REQUEST
|
||||
|
||||
pdf.untag(tag, test_session)
|
||||
pdf.untag(tag.id, test_session)
|
||||
test_session.commit()
|
||||
|
||||
# Can delete unused Tag.
|
||||
|
|
@ -174,7 +199,7 @@ async def test_delete_tagged_file(test_session, example_pdf, tag_factory):
|
|||
tag = tag_factory()
|
||||
|
||||
pdf: FileGroup = test_session.query(FileGroup).one()
|
||||
pdf.add_tag(tag)
|
||||
pdf.add_tag(tag.id)
|
||||
test_session.commit()
|
||||
|
||||
with pytest.raises(FileGroupIsTagged):
|
||||
|
|
@ -189,8 +214,9 @@ def test_import_empty_tags_config(test_session, test_tags_config):
|
|||
tags.import_tags_config()
|
||||
|
||||
|
||||
@pytest.mark.asyncio()
|
||||
async def test_import_tags_config(test_session, test_directory, test_tags_config, example_singlefile):
|
||||
@pytest.mark.asyncio
|
||||
async def test_import_tags_config(test_async_client, test_session, test_directory, test_tags_config,
|
||||
example_singlefile):
|
||||
with test_tags_config.open('wt') as fh:
|
||||
data = dict(
|
||||
# Tag names can contain Unicode characters.
|
||||
|
|
@ -255,10 +281,10 @@ async def test_import_tags_delete_missing(test_session, test_directory, test_tag
|
|||
foo, = make_files_structure({'foo': 'text'})
|
||||
foo = FileGroup.from_paths(test_session, foo)
|
||||
test_session.flush([foo, ])
|
||||
foo.add_tag(tag1)
|
||||
foo.add_tag(tag1.id)
|
||||
# use tag4 for Zim entry.
|
||||
await zim_lib.add_tag(tag4.name, test_zim.id, 'home')
|
||||
tags.schedule_save()
|
||||
tags.save_tags_config()
|
||||
|
||||
# All tags were saved.
|
||||
assert tag1.name in test_tags_config.read_text()
|
||||
|
|
@ -287,8 +313,23 @@ async def test_import_tags_delete_missing(test_session, test_directory, test_tag
|
|||
}
|
||||
|
||||
# DB is written to config with only those Tags left.
|
||||
tags.schedule_save()
|
||||
tags.save_tags_config()
|
||||
# Tags that are used are restored.
|
||||
assert len(config.tag_files) == 1
|
||||
assert len(config.tag_zims) == 1
|
||||
assert len(config.tags) == 3
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_invalid_tag(test_async_client, test_session, test_directory):
|
||||
with pytest.raises(InvalidTag):
|
||||
tags.upsert_tag('cannot use comma: ,', '#fff')
|
||||
|
||||
with pytest.raises(InvalidTag):
|
||||
tags.upsert_tag('invalid color', 'foo')
|
||||
|
||||
with pytest.raises(UnknownTag):
|
||||
tags.upsert_tag('Tag ID does not exist', '#fff', 1)
|
||||
|
||||
with pytest.raises(UnknownTag):
|
||||
tags.delete_tag(1)
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue