diff --git a/.github/dependabot.yaml b/.github/dependabot.yaml index f461b93..422c06f 100644 --- a/.github/dependabot.yaml +++ b/.github/dependabot.yaml @@ -4,7 +4,16 @@ updates: directory: / schedule: interval: daily + groups: + github_actions: + patterns: + - "*" + - package-ecosystem: pip directory: / schedule: interval: daily + groups: + python: + patterns: + - "*" diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 5fff20f..3ff0434 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -1,6 +1,6 @@ repos: - repo: https://github.com/astral-sh/ruff-pre-commit - rev: v0.3.3 + rev: v0.3.5 hooks: - id: ruff-format - id: ruff @@ -15,7 +15,7 @@ repos: - --number - --wrap=keep - repo: https://github.com/pre-commit/pre-commit-hooks - rev: v4.5.0 + rev: v4.6.0 hooks: - id: check-ast - id: check-builtin-literals @@ -30,9 +30,6 @@ repos: exclude_types: - json - xml - - id: fix-encoding-pragma - args: - - --remove - id: mixed-line-ending args: - --fix=auto diff --git a/perdoo/__main__.py b/perdoo/__main__.py index cc38951..e087cfc 100644 --- a/perdoo/__main__.py +++ b/perdoo/__main__.py @@ -9,23 +9,18 @@ from tempfile import TemporaryDirectory from pydantic import ValidationError +from rich.prompt import Prompt from perdoo import ARCHIVE_EXTENSIONS, IMAGE_EXTENSIONS, __version__, setup_logging from perdoo.archives import BaseArchive, CB7Archive, CBTArchive, CBZArchive, get_archive from perdoo.console import CONSOLE from perdoo.models import ComicInfo, Metadata, MetronInfo -from perdoo.models.metadata import Format, Meta, Tool +from perdoo.models._base import InfoModel +from perdoo.models.metadata import Format, Meta, Source, Tool +from perdoo.models.metron_info import InformationSource from perdoo.services import Comicvine, League, Marvel, Metron from perdoo.settings import OutputFormat, Settings -from perdoo.utils import ( - comic_to_metadata, - create_metadata, - list_files, - metadata_to_comic, - metadata_to_metron, - metron_to_metadata, - sanitize, -) +from perdoo.utils import Details, Identifications, get_metadata_id, list_files, sanitize LOGGER = logging.getLogger("perdoo") @@ -52,54 +47,120 @@ def convert_collection(path: Path, output: OutputFormat) -> None: archive_type.convert(old_archive=archive) -def read_archive(archive: BaseArchive) -> tuple[Metadata, MetronInfo, ComicInfo]: +def read_meta(archive: BaseArchive) -> tuple[Meta, Details]: filenames = archive.list_filenames() - metadata = None + + def read_meta_file(cls: type[InfoModel], filename: str) -> InfoModel | None: + if filename in filenames: + return cls.from_bytes(content=archive.read_file(filename=filename)) + return None + try: - if "/Metadata.xml" in filenames: - metadata = Metadata.from_bytes(content=archive.read_file(filename="/Metadata.xml")) - elif "Metadata.xml" in filenames: - metadata = Metadata.from_bytes(content=archive.read_file(filename="Metadata.xml")) + metadata = read_meta_file(cls=Metadata, filename="/Metadata.xml") or read_meta_file( + cls=Metadata, filename="Metadata.xml" + ) + if metadata: + meta = metadata.meta + details = Details( + series=Identifications( + search=metadata.issue.series.title, + comicvine=get_metadata_id( + resources=metadata.issue.series.resources, source=Source.COMICVINE + ), + league=get_metadata_id( + resources=metadata.issue.series.resources, + source=Source.LEAGUE_OF_COMIC_GEEKS, + ), + marvel=get_metadata_id( + resources=metadata.issue.series.resources, source=Source.MARVEL + ), + metron=get_metadata_id( + resources=metadata.issue.series.resources, source=Source.METRON + ), + ), + issue=Identifications( + search=metadata.issue.number, + comicvine=get_metadata_id( + resources=metadata.issue.resources, source=Source.COMICVINE + ), + league=get_metadata_id( + resources=metadata.issue.resources, source=Source.LEAGUE_OF_COMIC_GEEKS + ), + marvel=get_metadata_id( + resources=metadata.issue.resources, source=Source.MARVEL + ), + metron=get_metadata_id( + resources=metadata.issue.resources, source=Source.METRON + ), + ), + ) + return meta, details except ValidationError: LOGGER.error("%s contains an invalid Metadata file", archive.path.name) # noqa: TRY400 - metron_info = None try: - if "/MetronInfo.xml" in filenames: - metron_info = MetronInfo.from_bytes( - content=archive.read_file(filename="/MetronInfo.xml") - ) - elif "MetronInfo.xml" in filenames: - metron_info = MetronInfo.from_bytes( - content=archive.read_file(filename="MetronInfo.xml") + metron_info = read_meta_file(cls=MetronInfo, filename="/MetronInfo.xml") or read_meta_file( + cls=MetronInfo, filename="MetronInfo.xml" + ) + if metron_info: + details = Details( + series=Identifications( + search=metron_info.series.name, + comicvine=metron_info.series.id + if metron_info.id and metron_info.id.source == InformationSource.COMIC_VINE + else None, + league=metron_info.series.id + if metron_info.id + and metron_info.id.source == InformationSource.LEAGUE_OF_COMIC_GEEKS + else None, + marvel=metron_info.series.id + if metron_info.id and metron_info.id.source == InformationSource.MARVEL + else None, + metron=metron_info.series.id + if metron_info.id and metron_info.id.source == InformationSource.METRON + else None, + ), + issue=Identifications( + search=metron_info.number, + comicvine=metron_info.id.value + if metron_info.id and metron_info.id.source == InformationSource.COMIC_VINE + else None, + league=metron_info.id.value + if metron_info.id + and metron_info.id.source == InformationSource.LEAGUE_OF_COMIC_GEEKS + else None, + marvel=metron_info.id.value + if metron_info.id and metron_info.id.source == InformationSource.MARVEL + else None, + metron=metron_info.id.value + if metron_info.id and metron_info.id.source == InformationSource.METRON + else None, + ), ) + return Meta(date_=date.today(), tool=Tool(value="MetronInfo")), details except ValidationError: LOGGER.error("%s contains an invalid MetronInfo file", archive.path.name) # noqa: TRY400 - comic_info = None try: - if "/ComicInfo.xml" in filenames: - comic_info = ComicInfo.from_bytes(content=archive.read_file(filename="/ComicInfo.xml")) - elif "ComicInfo.xml" in filenames: - comic_info = ComicInfo.from_bytes(content=archive.read_file(filename="ComicInfo.xml")) + comic_info = read_meta_file(cls=ComicInfo, filename="/ComicInfo.xml") or read_meta_file( + cls=ComicInfo, filename="ComicInfo.xml" + ) + if comic_info: + details = Details( + series=Identifications(search=comic_info.series), + issue=Identifications(search=comic_info.number), + ) + return Meta(date_=date.today(), tool=Tool(value="ComicInfo")), details except ValidationError: LOGGER.error("%s contains an invalid ComicInfo file", archive.path.name) # noqa: TRY400 - if not metadata: - if metron_info: - metadata = metron_to_metadata(metron_info=metron_info) - elif comic_info: - metadata = comic_to_metadata(comic_info=comic_info) - else: - metadata = create_metadata(archive=archive) - if not metron_info: - metron_info = metadata_to_metron(metadata=metadata) - if not comic_info: - comic_info = metadata_to_comic(metadata=metadata) - return metadata, metron_info, comic_info + return Meta(date_=date.today(), tool=Tool(value="Manual")), Details( + series=Identifications(search=Prompt.ask("Series title", console=CONSOLE)), + issue=Identifications(), + ) def fetch_from_services( - settings: Settings, metainfo: tuple[Metadata, MetronInfo, ComicInfo] -) -> None: + settings: Settings, details: Details +) -> tuple[Metadata | None, MetronInfo | None, ComicInfo | None]: marvel = None if settings.marvel and settings.marvel.public_key and settings.marvel.private_key: marvel = Marvel(settings=settings.marvel) @@ -118,13 +179,13 @@ def fetch_from_services( league = League(settings.league_of_comic_geeks) if not marvel and not metron and not comicvine and not league: LOGGER.warning("No external services configured") - return + return None, None, None - success = any( - service and service.fetch(*metainfo) for service in (marvel, metron, comicvine, league) - ) - if not success: - LOGGER.warning("Unable to fetch information fron any service") + for service in (marvel, metron, comicvine, league): + metadata, metron_info, comic_info = service.fetch(details=details) + if metadata and metron_info and comic_info: + return metadata, metron_info, comic_info + return None, None, None def generate_filename(root: Path, extension: str, metadata: Metadata) -> Path: @@ -169,7 +230,7 @@ def rename_images(folder: Path, filename: str) -> None: new_filename = f"{filename}_{str(index).zfill(pad_count)}{img_file.suffix}" if img_file.name != new_filename: LOGGER.info("Renamed %s to %s", img_file.name, new_filename) - img_file.rename(folder / f"{filename}-{str(index).zfill(pad_count)}{img_file.suffix}") + shutil.move(img_file, folder / new_filename) def process_pages( @@ -210,16 +271,16 @@ def start(settings: Settings, force: bool = False) -> None: convert_collection(path=settings.collection_folder, output=settings.output.format) for file in list_files(settings.collection_folder, f".{settings.output.format}"): archive = get_archive(path=file) - metadata, metron_info, comic_info = read_archive(archive=archive) + CONSOLE.rule(file.stem) + LOGGER.info("Processing %s", file.stem) + meta, details = read_meta(archive=archive) if not force: - difference = abs(date.today() - metadata.meta.date_) - if metadata.meta.tool == Tool() and difference.days < 28: + difference = abs(date.today() - meta.date_) + if meta.tool == Tool() and difference.days < 28: continue - CONSOLE.rule(file.stem) - LOGGER.info("Processing %s", file.name) - fetch_from_services(settings=settings, metainfo=(metadata, metron_info, comic_info)) + metadata, metron_info, comic_info = fetch_from_services(settings=settings, details=details) new_file = generate_filename( root=settings.collection_folder, extension=settings.output.format.value, @@ -229,6 +290,7 @@ def start(settings: Settings, force: bool = False) -> None: temp_folder = Path(temp_str) if not archive.extract_files(destination=temp_folder): return + LOGGER.info("Processing %s pages", file.stem) process_pages( folder=temp_folder, metadata=metadata, diff --git a/perdoo/console.py b/perdoo/console.py index e59a879..40f251c 100644 --- a/perdoo/console.py +++ b/perdoo/console.py @@ -1,14 +1,13 @@ from __future__ import annotations -__all__ = ["CONSOLE", "create_menu", "DatePrompt"] +__all__ = ["CONSOLE", "create_menu"] import logging -from datetime import date, datetime from rich import box from rich.console import Console from rich.panel import Panel -from rich.prompt import IntPrompt, InvalidResponse, PromptBase +from rich.prompt import IntPrompt from rich.theme import Theme LOGGER = logging.getLogger(__name__) @@ -70,17 +69,3 @@ def create_menu( options=options, title=title, subtitle=subtitle, prompt=prompt, default=default ) return selected - - -class DatePrompt(PromptBase[date]): - response_type = date - validate_error_message = ( - "[prompt.invalid]Please enter a valid date in the format of 'yyyy-mm-dd'" - ) - prompt_suffix = "[yyyy-mm-dd]: " - - def process_response(self: DatePrompt, value: str) -> date: - try: - return datetime.strptime(value.strip(), "%Y-%m-%d").date() - except ValueError as err: - raise InvalidResponse(self.validate_error_message) from err diff --git a/perdoo/models/comic_info.py b/perdoo/models/comic_info.py index 4eae107..465da9c 100644 --- a/perdoo/models/comic_info.py +++ b/perdoo/models/comic_info.py @@ -10,7 +10,7 @@ import xmltodict from natsort import humansorted, ns from PIL import Image -from pydantic import Field, HttpUrl +from pydantic import Field, HttpUrl, NonNegativeFloat from perdoo.models._base import InfoModel, PascalModel @@ -37,7 +37,7 @@ def load(value: str) -> YesNo: for entry in YesNo: if entry.value.replace(" ", "").casefold() == value.replace(" ", "").casefold(): return entry - raise ValueError(f"'{value}' isnt a valid comic_info.YesNo") + raise ValueError(f"`{value}` isn't a valid comic_info.YesNo") def __lt__(self: YesNo, other) -> int: # noqa: ANN001 if not isinstance(other, type(self)): @@ -59,7 +59,7 @@ def load(value: str) -> Manga: for entry in Manga: if entry.value.replace(" ", "").casefold() == value.replace(" ", "").casefold(): return entry - raise ValueError(f"'{value}' isnt a valid comic_info.Manga") + raise ValueError(f"`{value}` isn't a valid comic_info.Manga") def __lt__(self: Manga, other) -> int: # noqa: ANN001 if not isinstance(other, type(self)): @@ -92,7 +92,7 @@ def load(value: str) -> AgeRating: for entry in AgeRating: if entry.value.replace(" ", "").casefold() == value.replace(" ", "").casefold(): return entry - raise ValueError(f"'{value}' isnt a valid comic_info.AgeRating") + raise ValueError(f"`{value}` isn't a valid comic_info.AgeRating") def __lt__(self: AgeRating, other) -> int: # noqa: ANN001 if not isinstance(other, type(self)): @@ -121,7 +121,7 @@ def load(value: str) -> PageType: for entry in PageType: if entry.value.replace(" ", "").casefold() == value.replace(" ", "").casefold(): return entry - raise ValueError(f"'{value}' isnt a valid comic_info.PageType") + raise ValueError(f"`{value}` isn't a valid comic_info.PageType") def __lt__(self: PageType, other) -> int: # noqa: ANN001 if not isinstance(other, type(self)): @@ -215,14 +215,14 @@ class ComicInfo(PascalModel, InfoModel): series_group: str | None = None age_rating: AgeRating = AgeRating.UNKNOWN pages: list[Page] = Field(default_factory=list) - community_rating: float | None = Field(default=None, ge=0, le=5) + community_rating: NonNegativeFloat | None = Field(default=None, le=5) main_character_or_team: str | None = None review: str | None = None list_fields: ClassVar[dict[str, str]] = {"Pages": "Page"} def __init__(self: ComicInfo, **data: Any): - self.unwrap_list(mappings=ComicInfo.list_fields, content=data) + self.unwrap_list(mappings=self.list_fields, content=data) super().__init__(**data) @property diff --git a/perdoo/models/metadata.py b/perdoo/models/metadata.py index 3d9402a..53d04a4 100644 --- a/perdoo/models/metadata.py +++ b/perdoo/models/metadata.py @@ -23,7 +23,7 @@ import xmltodict from PIL import Image -from pydantic import Field +from pydantic import Field, PositiveInt from perdoo import __version__ from perdoo.models._base import InfoModel, PascalModel @@ -41,7 +41,7 @@ def load(value: str) -> Source: for entry in Source: if entry.value.replace(" ", "").casefold() == value.replace(" ", "").casefold(): return entry - raise ValueError(f"'{value}' isnt a valid metadata.Source") + raise ValueError(f"`{value}` isn't a valid metadata.Source") def __lt__(self: Source, other) -> int: # noqa: ANN001 if not isinstance(other, type(self)): @@ -77,7 +77,7 @@ class TitledResource(PascalModel): list_fields: ClassVar[dict[str, str]] = {"Resources": "Resource"} def __init__(self: TitledResource, **data: Any): - self.unwrap_list(mappings=TitledResource.list_fields, content=data) + self.unwrap_list(mappings=self.list_fields, content=data) super().__init__(**data) def __lt__(self: TitledResource, other) -> int: # noqa: ANN001 @@ -101,13 +101,13 @@ class Credit(PascalModel): list_fields: ClassVar[dict[str, str]] = {"Roles": "Role"} def __init__(self: Credit, **data: Any): - self.unwrap_list(mappings=Credit.list_fields, content=data) + self.unwrap_list(mappings=self.list_fields, content=data) super().__init__(**data) def __lt__(self: Credit, other) -> int: # noqa: ANN001 if not isinstance(other, type(self)): raise NotImplementedError - return self.creator < other.crestor + return self.creator < other.creator def __eq__(self: Credit, other) -> bool: # noqa: ANN001 if not isinstance(other, type(self)): @@ -131,7 +131,7 @@ def load(value: str) -> Format: for entry in Format: if entry.value.replace(" ", "").casefold() == value.replace(" ", "").casefold(): return entry - raise ValueError(f"'{value}' isnt a valid metadata.Format") + raise ValueError(f"`{value}` isn't a valid metadata.Format") def __lt__(self: Format, other) -> int: # noqa: ANN001 if not isinstance(other, type(self)): @@ -146,18 +146,18 @@ class Series(TitledResource): genres: list[TitledResource] = Field(default_factory=list) publisher: TitledResource start_year: int | None = None - volume: int = Field(default=1) + volume: PositiveInt = Field(default=1) list_fields: ClassVar[dict[str, str]] = {**TitledResource.list_fields, "Genres": "Genre"} def __init__(self: Series, **data: Any): - self.unwrap_list(mappings=Series.list_fields, content=data) + self.unwrap_list(mappings=self.list_fields, content=data) super().__init__(**data) def __lt__(self: Series, other) -> int: # noqa: ANN001 if not isinstance(other, type(self)): raise NotImplementedError - if self.publisher != other.publusher: + if self.publisher != other.publisher: return self.publisher < other.publisher if self.title.casefold() != other.title.casefold(): return self.title.casefold() < other.title.casefold() @@ -188,7 +188,7 @@ class Issue(PascalModel): language: str = Field(alias="@language", default="en") locations: list[TitledResource] = Field(default_factory=list) number: str | None = None - page_count: int + page_count: int = 0 resources: list[Resource] = Field(default_factory=list) series: Series store_date: date | None = None @@ -211,7 +211,7 @@ class Issue(PascalModel): } def __init__(self: Issue, **data: Any): - self.unwrap_list(mappings=Issue.list_fields, content=data) + self.unwrap_list(mappings=self.list_fields, content=data) super().__init__(**data) def __lt__(self: Issue, other) -> int: # noqa: ANN001 @@ -263,7 +263,7 @@ def load(value: str) -> PageType: for entry in PageType: if entry.value.replace(" ", "").casefold() == value.replace(" ", "").casefold(): return entry - raise ValueError(f"'{value}' isnt a valid metadata.PageType") + raise ValueError(f"`{value}` isn't a valid metadata.PageType") def __lt__(self: PageType, other) -> int: # noqa: ANN001 if not isinstance(other, type(self)): @@ -328,7 +328,7 @@ class Metadata(PascalModel, InfoModel): list_fields: ClassVar[dict[str, str]] = {**Issue.list_fields, "Pages": "Page"} def __init__(self: Metadata, **data: Any): - self.unwrap_list(mappings=Metadata.list_fields, content=data) + self.unwrap_list(mappings=self.list_fields, content=data) super().__init__(**data) def __lt__(self: Metadata, other) -> int: # noqa: ANN001 diff --git a/perdoo/models/metron_info.py b/perdoo/models/metron_info.py index 22e0704..bf64493 100644 --- a/perdoo/models/metron_info.py +++ b/perdoo/models/metron_info.py @@ -1,23 +1,23 @@ from __future__ import annotations __all__ = [ - "InformationSource", - "Source", - "Resource", + "GTIN", + "AgeRating", + "Arc", + "Credit", "Format", - "Series", - "Price", "Genre", "GenreResource", - "Arc", - "GTIN", - "AgeRating", + "InformationSource", + "MetronInfo", + "Price", + "Resource", "Role", "RoleResource", - "Credit", - "PageType", - "Page", - "MetronInfo", + "Series", + "Source", + "Sources", + "Universe", ] from datetime import date @@ -27,7 +27,7 @@ import xmltodict from PIL import Image -from pydantic import Field, HttpUrl +from pydantic import Field, HttpUrl, PositiveInt from perdoo.models._base import InfoModel, PascalModel @@ -44,7 +44,7 @@ def load(value: str) -> InformationSource: for entry in InformationSource: if entry.value.replace(" ", "").casefold() == value.replace(" ", "").casefold(): return entry - raise ValueError(f"'{value}' isnt a valid metron_info.InformationSource") + raise ValueError(f"`{value}` isn't a valid metron_info.InformationSource") def __lt__(self: InformationSource, other) -> int: # noqa: ANN001 if not isinstance(other, type(self)): @@ -57,11 +57,29 @@ def __str__(self: InformationSource) -> str: class Source(PascalModel): source: InformationSource = Field(alias="@source") - value: int = Field(alias="#text", gt=0) + value: PositiveInt = Field(alias="#text") + + def __lt__(self: Source, other) -> int: # noqa: ANN001 + if not isinstance(other, type(self)): + raise NotImplementedError + return self.source < other.source + + def __eq__(self: Source, other) -> bool: # noqa: ANN001 + if not isinstance(other, type(self)): + raise NotImplementedError + return self.source == other.source + + def __hash__(self: Source) -> int: + return hash((type(self), self.source)) + + +class Sources(PascalModel): + primary: Source + alternative: list[Source] = Field(default_factory=list) class Resource(PascalModel): - id: int | None = Field(alias="@id", default=None, gt=0) + id: PositiveInt | None = Field(alias="@id", default=None) value: str = Field(alias="#text") def __lt__(self: Resource, other) -> int: # noqa: ANN001 @@ -81,10 +99,11 @@ def __hash__(self: Resource) -> int: class Format(Enum): ANNUAL = "Annual" GRAPHIC_NOVEL = "Graphic Novel" - LIMITED = "Limited" + LIMITED_SERIES = "Limited Series" ONE_SHOT = "One-Shot" SERIES = "Series" TRADE_PAPERBACK = "Trade Paperback" + HARDCOVER = "Hardcover" @staticmethod def load(value: str) -> Format: @@ -92,14 +111,10 @@ def load(value: str) -> Format: if entry.value.replace(" ", "").casefold() == value.replace(" ", "").casefold(): return entry # region Manual matches - if value.casefold() == "Limited Series".casefold(): - return Format.LIMITED - if value.casefold() == "Cancelled Series".casefold(): - return Format.SERIES - if value.casefold() == "Hard Cover".casefold(): + if value.casefold() in ["Cancelled Series".casefold(), "Ongoing Series".casefold()]: return Format.SERIES # endregion - raise ValueError(f"'{value}' isnt a valid metron_info.Format") + raise ValueError(f"`{value}` isn't a valid metron_info.Format") def __lt__(self: Format, other) -> int: # noqa: ANN001 if not isinstance(other, type(self)): @@ -111,7 +126,7 @@ def __str__(self: Format) -> str: class Series(PascalModel): - id: int | None = Field(alias="@id", default=None, gt=0) + id: PositiveInt | None = Field(alias="@id", default=None) lang: str = Field(alias="@lang", default="en") name: str sort_name: str | None = None @@ -159,7 +174,7 @@ def load(value: str) -> Genre: for entry in Genre: if entry.value.replace(" ", "").casefold() == value.replace(" ", "").casefold(): return entry - raise ValueError(f"'{value}' isnt a valid metron_info.Genre") + raise ValueError(f"`{value}` isn't a valid metron_info.Genre") def __lt__(self: Genre, other) -> int: # noqa: ANN001 if not isinstance(other, type(self)): @@ -171,7 +186,7 @@ def __str__(self: Genre) -> str: class GenreResource(PascalModel): - id: int | None = Field(alias="@id", default=None, gt=0) + id: PositiveInt | None = Field(alias="@id", default=None) value: Genre = Field(alias="#text") def __lt__(self: GenreResource, other) -> int: # noqa: ANN001 @@ -189,9 +204,9 @@ def __hash__(self: GenreResource) -> int: class Arc(PascalModel): - id: int | None = Field(alias="@id", default=None, gt=0) + id: PositiveInt | None = Field(alias="@id", default=None) name: str - number: int | None = Field(default=None, gt=0) + number: PositiveInt | None = None def __lt__(self: Arc, other) -> int: # noqa: ANN001 if not isinstance(other, type(self)): @@ -207,6 +222,25 @@ def __hash__(self: Arc) -> int: return hash((type(self), self.name)) +class Universe(PascalModel): + id: int | None = Field(alias="@id", default=None, gt=0) + name: str + designation: str | None = None + + def __lt__(self: Universe, other) -> int: # noqa: ANN001 + if not isinstance(other, type(self)): + raise NotImplementedError + return self.name < other.name + + def __eq__(self: Universe, other) -> bool: # noqa: ANN001 + if not isinstance(other, type(self)): + raise NotImplementedError + return self.name == other.name + + def __hash__(self: Universe) -> int: + return hash((type(self), self.name)) + + class GTIN(PascalModel): isbn: str | None = Field(alias="ISBN", default=None) upc: str | None = Field(alias="UPC", default=None) @@ -224,7 +258,7 @@ def load(value: str) -> AgeRating: for entry in AgeRating: if entry.value.replace(" ", "").casefold() == value.replace(" ", "").casefold(): return entry - raise ValueError(f"'{value}' isnt a valid metron_info.AgeRating") + raise ValueError(f"`{value}` isn't a valid metron_info.AgeRating") def __lt__(self: AgeRating, other) -> int: # noqa: ANN001 if not isinstance(other, type(self)): @@ -284,7 +318,7 @@ def load(value: str) -> Role: for entry in Role: if entry.value.replace(" ", "").casefold() == value.replace(" ", "").casefold(): return entry - raise ValueError(f"'{value}' isnt a valid metron_info.Role") + raise ValueError(f"`{value}` isn't a valid metron_info.Role") def __lt__(self: Role, other) -> int: # noqa: ANN001 if not isinstance(other, type(self)): @@ -296,7 +330,7 @@ def __str__(self: Role) -> str: class RoleResource(PascalModel): - id: int | None = Field(alias="@id", default=None, gt=0) + id: PositiveInt | None = Field(alias="@id", default=None) value: Role = Field(alias="#text") def __lt__(self: RoleResource, other) -> int: # noqa: ANN001 @@ -321,8 +355,8 @@ class Credit(PascalModel): text_fields: ClassVar[list[str]] = ["Creator", "Roles"] def __init__(self: Credit, **data: Any): - self.unwrap_list(mappings=Credit.list_fields, content=data) - self.to_xml_text(mappings=Credit.text_fields, content=data) + self.unwrap_list(mappings=self.list_fields, content=data) + self.to_xml_text(mappings=self.text_fields, content=data) super().__init__(**data) def __lt__(self: Credit, other) -> int: # noqa: ANN001 @@ -357,7 +391,7 @@ def load(value: str) -> PageType: for entry in PageType: if entry.value.replace(" ", "").casefold() == value.replace(" ", "").casefold(): return entry - raise ValueError(f"'{value}' isnt a valid metron_info.PageType") + raise ValueError(f"`{value}` isn't a valid metron_info.PageType") def __lt__(self: PageType, other) -> int: # noqa: ANN001 if not isinstance(other, type(self)): @@ -414,7 +448,7 @@ def from_path(file: Path, index: int, is_final_page: bool, page: Page | None) -> class MetronInfo(PascalModel, InfoModel): - id: Source | None = Field(alias="ID", default=None) + id: Sources | None = Field(alias="ID", default=None) publisher: Resource series: Series collection_title: str | None = None @@ -423,16 +457,16 @@ class MetronInfo(PascalModel, InfoModel): summary: str | None = None notes: str | None = None prices: list[Price] = Field(default_factory=list) - cover_date: date + cover_date: date | None = None store_date: date | None = None page_count: int = 0 - genres: list[GenreResource] = Field(default_list=list) + genres: list[GenreResource] = Field(default_factory=list) tags: list[Resource] = Field(default_factory=list) arcs: list[Arc] = Field(default_factory=list) characters: list[Resource] = Field(default_factory=list) teams: list[Resource] = Field(default_factory=list) + universes: list[Universe] = Field(default_factory=list) locations: list[Resource] = Field(default_factory=list) - black_and_white: bool = False gtin: GTIN | None = Field(alias="GTIN", default=None) age_rating: AgeRating = Field(default=AgeRating.UNKNOWN) reprints: list[Resource] = Field(default_factory=list) @@ -449,6 +483,7 @@ class MetronInfo(PascalModel, InfoModel): "Arcs": "Arc", "Characters": "Character", "Teams": "Team", + "Universes": "Universe", "Locations": "Location", "Reprints": "Reprint", "Credits": "Credit", @@ -468,8 +503,8 @@ class MetronInfo(PascalModel, InfoModel): ] def __init__(self: MetronInfo, **data: Any): - self.unwrap_list(mappings=MetronInfo.list_fields, content=data) - self.to_xml_text(mappings=MetronInfo.text_fields, content=data) + self.unwrap_list(mappings=self.list_fields, content=data) + self.to_xml_text(mappings=self.text_fields, content=data) super().__init__(**data) @classmethod diff --git a/perdoo/services/_base.py b/perdoo/services/_base.py index 1f572cb..70280d9 100644 --- a/perdoo/services/_base.py +++ b/perdoo/services/_base.py @@ -6,64 +6,35 @@ from typing import Generic, TypeVar from perdoo.models import ComicInfo, Metadata, MetronInfo +from perdoo.utils import Details -P = TypeVar("P") S = TypeVar("S") C = TypeVar("C") -class BaseService(Generic[P, S, C]): +class BaseService(Generic[S, C]): @abstractmethod - def _search_publishers(self: BaseService, title: str | None) -> int | None: ... + def _get_series_id(self: BaseService, title: str) -> int | None: ... @abstractmethod - def _get_publisher_id( - self: BaseService, metadata: Metadata, metron_info: MetronInfo - ) -> int | None: ... + def fetch_series(self: BaseService, details: Details) -> S | None: ... @abstractmethod - def fetch_publisher( - self: BaseService, metadata: Metadata, metron_info: MetronInfo, comic_info: ComicInfo - ) -> P | None: ... + def _get_issue_id(self: BaseService, series_id: int, number: str | None) -> int | None: ... @abstractmethod - def _search_series(self: BaseService, publisher_id: int, title: str | None) -> int | None: ... + def fetch_issue(self: BaseService, series_id: int, details: Details) -> C | None: ... @abstractmethod - def _get_series_id( - self: BaseService, publisher_id: int, metadata: Metadata, metron_info: MetronInfo - ) -> int | None: ... + def _process_metadata(self: BaseService, series: S, issue: C) -> Metadata | None: ... @abstractmethod - def fetch_series( - self: BaseService, - metadata: Metadata, - metron_info: MetronInfo, - comic_info: ComicInfo, - publisher_id: int, - ) -> S | None: ... + def _process_metron_info(self: BaseService, series: S, issue: C) -> MetronInfo | None: ... @abstractmethod - def _search_issues(self: BaseService, series_id: int, number: str | None) -> int | None: ... - - @abstractmethod - def _get_issue_id( - self: BaseService, series_id: int, metadata: Metadata, metron_info: MetronInfo - ) -> int | None: ... - - @abstractmethod - def fetch_issue( - self: BaseService, - metadata: Metadata, - metron_info: MetronInfo, - comic_info: ComicInfo, - series_id: int, - ) -> C | None: ... + def _process_comic_info(self: BaseService, series: S, issue: C) -> ComicInfo | None: ... @abstractmethod def fetch( - self: BaseService, - metadata: Metadata | None, - metron_info: MetronInfo | None, - comic_info: ComicInfo | None, - ) -> bool: ... + self: BaseService, details: Details + ) -> tuple[Metadata | None, MetronInfo | None, ComicInfo | None]: ... diff --git a/perdoo/services/comicvine.py b/perdoo/services/comicvine.py index 01863d7..00c72fa 100644 --- a/perdoo/services/comicvine.py +++ b/perdoo/services/comicvine.py @@ -10,221 +10,45 @@ from simyan.comicvine import Comicvine as Simyan from simyan.exceptions import ServiceError from simyan.schemas.issue import Issue -from simyan.schemas.publisher import Publisher from simyan.schemas.volume import Volume from simyan.sqlite_cache import SQLiteCache from perdoo import get_cache_dir -from perdoo.console import CONSOLE, DatePrompt, create_menu +from perdoo.console import CONSOLE, create_menu from perdoo.models import ComicInfo, Metadata, MetronInfo from perdoo.models.metadata import Source from perdoo.models.metron_info import InformationSource from perdoo.services._base import BaseService from perdoo.settings import Comicvine as ComicvineSettings +from perdoo.utils import Details LOGGER = logging.getLogger(__name__) -def add_publisher_to_metadata(publisher: Publisher, metadata: Metadata) -> None: - from perdoo.models.metadata import Resource - - resources = set(metadata.issue.series.publisher.resources) - resources.add(Resource(source=Source.COMICVINE, value=publisher.id)) - metadata.issue.series.publisher.resources = list(resources) - metadata.issue.series.publisher.title = publisher.name - - -def add_publisher_to_metron_info(publisher: Publisher, metron_info: MetronInfo) -> None: - if not metron_info.id or metron_info.id.source == InformationSource.COMIC_VINE: - metron_info.publisher.id = publisher.id - metron_info.publisher.value = publisher.name - - -def add_publisher_to_comic_info(publisher: Publisher, comic_info: ComicInfo) -> None: - comic_info.publisher = publisher.name - - -def add_series_to_metadata(series: Volume, metadata: Metadata) -> None: - from perdoo.models.metadata import Resource - - resources = set(metadata.issue.series.resources) - resources.add(Resource(source=Source.COMICVINE, value=series.id)) - metadata.issue.series.resources = list(resources) - metadata.issue.series.start_year = series.start_year - metadata.issue.series.title = series.name - - -def add_series_to_metron_info(series: Volume, metron_info: MetronInfo) -> None: - if not metron_info.id or metron_info.id.source == InformationSource.COMIC_VINE: - metron_info.series.id = series.id - metron_info.series.name = series.name - - -def add_series_to_comic_info(series: Volume, comic_info: ComicInfo) -> None: - comic_info.series = series.name - - -def add_issue_to_metadata(issue: Issue, metadata: Metadata) -> None: - from perdoo.models.metadata import Credit, Resource, StoryArc, TitledResource - - resources = set(metadata.issue.resources) - resources.add(Resource(source=Source.COMICVINE, value=issue.id)) - metadata.issue.resources = list(resources) - metadata.issue.characters = [ - TitledResource(title=x.name, resources=[Resource(source=Source.COMICVINE, value=x.id)]) - for x in issue.characters - ] - metadata.issue.cover_date = issue.cover_date - metadata.issue.credits = [ - Credit( - creator=TitledResource( - title=x.name, resources=[Resource(source=Source.COMICVINE, value=x.id)] - ), - roles=[TitledResource(title=r.strip()) for r in re.split(r"[~\r\n,]+", x.roles)], - ) - for x in issue.creators - ] - metadata.issue.locations = [ - TitledResource(title=x.name, resources=[Resource(source=Source.COMICVINE, value=x.id)]) - for x in issue.locations - ] - metadata.issue.number = issue.number - metadata.issue.store_date = issue.store_date - metadata.issue.story_arcs = [ - StoryArc(title=x.name, resources=[Resource(source=Source.COMICVINE, value=x.id)]) - for x in issue.story_arcs - ] - metadata.issue.summary = issue.summary - metadata.issue.teams = [ - TitledResource(title=x.name, resources=[Resource(source=Source.COMICVINE, value=x.id)]) - for x in issue.teams - ] - metadata.issue.title = issue.name - - -def add_issue_to_metron_info(issue: Issue, metron_info: MetronInfo) -> None: - from perdoo.models.metron_info import Arc, Credit, Resource, Role, RoleResource, Source - - metron_info.arcs = [Arc(id=x.id, name=x.name) for x in issue.story_arcs] - metron_info.characters = [Resource(id=x.id, value=x.name) for x in issue.characters] - metron_info.cover_date = ( - issue.cover_date or DatePrompt.ask("Cover Date", default=date.today(), console=CONSOLE), - ) - credits_ = [] - for x in issue.creators: - roles = [] - for r in re.split(r"[~\r\n,]+", x.roles): - try: - roles.append(RoleResource(value=Role.load(value=r.strip()))) - except ValueError: # noqa: PERF203 - roles.append(RoleResource(value=Role.OTHER)) - credits_.append(Credit(creator=Resource(id=x.id, value=x.name), roles=roles)) - metron_info.credits = credits_ - if not metron_info.id or metron_info.id.source == InformationSource.COMIC_VINE: - metron_info.id = Source(source=InformationSource.COMIC_VINE, value=issue.id) - metron_info.locations = [Resource(id=x.id, value=x.name) for x in issue.locations] - metron_info.number = issue.number - metron_info.store_date = issue.store_date - metron_info.summary = issue.summary - metron_info.teams = [Resource(id=x.id, value=x.name) for x in issue.teams] - metron_info.collection_title = issue.name - metron_info.url = issue.site_url - - -def add_issue_to_comic_info(issue: Issue, comic_info: ComicInfo) -> None: - comic_info.character_list = [x.name for x in issue.characters] - comic_info.credits = { - x.name: [r.strip() for r in re.split(r"[~\r\n,]+", x.roles)] for x in issue.creators - } - comic_info.cover_date = issue.cover_date - comic_info.location_list = [x.name for x in issue.locations] - comic_info.number = issue.number - comic_info.story_arc_list = [x.name for x in issue.story_arcs] - comic_info.summary = issue.summary - comic_info.team_list = [x.name for x in issue.teams] - comic_info.title = issue.name - comic_info.web = issue.site_url - - -class Comicvine(BaseService[Publisher, Volume, Issue]): +class Comicvine(BaseService[Volume, Issue]): def __init__(self: Comicvine, settings: ComicvineSettings): cache = SQLiteCache(path=get_cache_dir() / "simyan.sqlite", expiry=14) self.session = Simyan(api_key=settings.api_key, cache=cache) - def _search_publishers(self: Comicvine, title: str | None) -> int | None: - title = title or Prompt.ask("Publisher title", console=CONSOLE) - try: - options = sorted( - self.session.list_publishers({"filter": f"name:{title}"}), key=lambda x: x.name - ) - if not options: - LOGGER.warning("Unable to find any publishers with the title: '%s'", title) - index = create_menu( - options=[f"{x.id} | {x.name}" for x in options], - title="Comicvine Publisher", - default="None of the Above", - ) - if index != 0: - return options[index - 1].id - if not Confirm.ask("Try Again", console=CONSOLE): - return None - return self._search_publishers(title=None) - except ServiceError: - LOGGER.exception("") - return None - - def _get_publisher_id( - self: Comicvine, metadata: Metadata, metron_info: MetronInfo - ) -> int | None: - publisher_id = next( - ( - x.value - for x in metadata.issue.series.publisher.resources - if x.source == Source.COMICVINE - ), - None, - ) or ( - metron_info.publisher.id - if metron_info.id and metron_info.id.source == InformationSource.COMIC_VINE - else None - ) - return publisher_id or self._search_publishers(title=metadata.issue.series.publisher.title) - - def fetch_publisher( - self: Comicvine, metadata: Metadata, metron_info: MetronInfo, comic_info: ComicInfo - ) -> Publisher | None: - publisher_id = self._get_publisher_id(metadata=metadata, metron_info=metron_info) - if not publisher_id: - return None - try: - publisher = self.session.get_publisher(publisher_id=publisher_id) - add_publisher_to_metadata(publisher=publisher, metadata=metadata) - add_publisher_to_metron_info(publisher=publisher, metron_info=metron_info) - add_publisher_to_comic_info(publisher=publisher, comic_info=comic_info) - return publisher - except ServiceError: - LOGGER.exception("") - return None - - def _search_series(self: Comicvine, publisher_id: int, title: str | None) -> int | None: + def _get_series_id(self: Comicvine, title: str | None) -> int | None: title = title or Prompt.ask("Series title", console=CONSOLE) try: options = sorted( - [ - x - for x in self.session.list_volumes({"filter": f"name:{title}"}) - if x.publisher.id == publisher_id - ], - key=lambda x: (x.name, x.start_year), + self.session.list_volumes({"filter": f"name:{title}"}), + key=lambda x: ( + x.publisher.name if x.publisher and x.publisher.name else "", + x.name, + x.start_year or 0, + ), ) if not options: - LOGGER.warning( - "Unable to find any Series with a PublisherId: %s and the title: '%s'", - publisher_id, - title, - ) + LOGGER.warning("Unable to find any Series with the title: '%s'", title) index = create_menu( - options=[f"{x.id} | {x.name} ({x.start_year})" for x in options], + options=[ + f"{x.id} | {x.publisher.name if x.publisher and x.publisher.name else ''}" + f" | {x.name} ({x.start_year})" + for x in options + ], title="Comicvine Series", default="None of the Above", ) @@ -232,48 +56,24 @@ def _search_series(self: Comicvine, publisher_id: int, title: str | None) -> int return options[index - 1].id if not Confirm.ask("Try Again", console=CONSOLE): return None - return self._search_series(publisher_id=publisher_id, title=None) + return self._get_series_id(title=None) except ServiceError: LOGGER.exception("") return None - def _get_series_id( - self: Comicvine, publisher_id: int, metadata: Metadata, metron_info: MetronInfo - ) -> int | None: - series_id = next( - (x.value for x in metadata.issue.series.resources if x.source == Source.COMICVINE), None - ) or ( - metron_info.series.id - if metron_info.id and metron_info.id.source == InformationSource.COMIC_VINE - else None - ) - return series_id or self._search_series( - publisher_id=publisher_id, title=metadata.issue.series.title - ) - - def fetch_series( - self: Comicvine, - metadata: Metadata, - metron_info: MetronInfo, - comic_info: ComicInfo, - publisher_id: int, - ) -> Volume | None: - series_id = self._get_series_id( - publisher_id=publisher_id, metadata=metadata, metron_info=metron_info - ) + def fetch_series(self: Comicvine, details: Details) -> Volume | None: + series_id = details.series.comicvine or self._get_series_id(title=details.series.search) if not series_id: return None try: series = self.session.get_volume(volume_id=series_id) - add_series_to_metadata(series=series, metadata=metadata) - add_series_to_metron_info(series=series, metron_info=metron_info) - add_series_to_comic_info(series=series, comic_info=comic_info) + details.series.comicvine = series_id return series except ServiceError: LOGGER.exception("") return None - def _search_issues(self: Comicvine, series_id: int, number: str | None) -> int | None: + def _get_issue_id(self: Comicvine, series_id: int, number: str | None) -> int | None: try: options = sorted( self.session.list_issues( @@ -298,65 +98,178 @@ def _search_issues(self: Comicvine, series_id: int, number: str | None) -> int | return options[index - 1].id if number: LOGGER.info("Searching again without the issue number") - return self._search_issues(series_id=series_id, number=None) + return self._get_issue_id(series_id=series_id, number=None) return None except ServiceError: LOGGER.exception("") return None - def _get_issue_id( - self: Comicvine, series_id: int, metadata: Metadata, metron_info: MetronInfo - ) -> int | None: - issue_id = next( - (x.value for x in metadata.issue.resources if x.source == Source.COMICVINE), None - ) or ( - metron_info.id.value - if metron_info.id and metron_info.id.source == InformationSource.COMIC_VINE - else None - ) - return issue_id or self._search_issues(series_id=series_id, number=metadata.issue.number) - - def fetch_issue( - self: Comicvine, - metadata: Metadata, - metron_info: MetronInfo, - comic_info: ComicInfo, - series_id: int, - ) -> Issue | None: - issue_id = self._get_issue_id( - series_id=series_id, metadata=metadata, metron_info=metron_info + def fetch_issue(self: Comicvine, series_id: int, details: Details) -> Issue | None: + issue_id = details.issue.comicvine or self._get_issue_id( + series_id=series_id, number=details.issue.search ) if not issue_id: return None try: issue = self.session.get_issue(issue_id=issue_id) - add_issue_to_metadata(issue=issue, metadata=metadata) - add_issue_to_metron_info(issue=issue, metron_info=metron_info) - add_issue_to_comic_info(issue=issue, comic_info=comic_info) + details.issue.comicvine = issue_id return issue except ServiceError: LOGGER.exception("") return None - def fetch( - self: Comicvine, metadata: Metadata, metron_info: MetronInfo, comic_info: ComicInfo - ) -> bool: - publisher = self.fetch_publisher( - metadata=metadata, metron_info=metron_info, comic_info=comic_info + def _process_metadata(self: Comicvine, series: Volume, issue: Issue) -> Metadata | None: + from perdoo.models.metadata import ( + Credit, + Issue, + Meta, + Resource, + Series, + StoryArc, + TitledResource, ) - if not publisher: - return False - series = self.fetch_series( - metadata=metadata, - metron_info=metron_info, - comic_info=comic_info, - publisher_id=publisher.id, + + return Metadata( + issue=Issue( + characters=[ + TitledResource( + resources=[Resource(source=Source.COMICVINE, value=x.id)], title=x.name + ) + for x in issue.characters + ], + cover_date=issue.cover_date, + credits=[ + Credit( + creator=TitledResource( + resources=[Resource(source=Source.COMICVINE, value=x.id)], title=x.name + ), + roles=[ + TitledResource(title=r.strip()) for r in re.split(r"[~\r\n,]+", x.roles) + ], + ) + for x in issue.creators + ], + locations=[ + TitledResource( + resources=[Resource(source=Source.COMICVINE, value=x.id)], title=x.name + ) + for x in issue.locations + ], + number=issue.number, + resources=[Resource(source=Source.COMICVINE, value=issue.id)], + series=Series( + publisher=TitledResource( + resources=[Resource(source=Source.COMICVINE, value=series.publisher.id)], + title=series.publisher.name, + ), + resources=[Resource(source=Source.COMICVINE, value=series.id)], + start_year=series.start_year, + title=series.name, + ), + store_date=issue.store_date, + story_arcs=[ + StoryArc( + resources=[Resource(source=Source.COMICVINE, value=x.id)], title=x.name + ) + for x in issue.story_arcs + ], + summary=issue.summary, + teams=[ + TitledResource( + resources=[Resource(source=Source.COMICVINE, value=x.id)], title=x.name + ) + for x in issue.teams + ], + title=issue.name, + ), + meta=Meta(date_=date.today()), ) - if not series: - return False - issue = self.fetch_issue( - metadata=metadata, metron_info=metron_info, comic_info=comic_info, series_id=series.id + + def _process_metron_info(self: BaseService, series: Volume, issue: Issue) -> MetronInfo | None: + def load_role(value: str) -> Role: + try: + return Role.load(value=value.strip()) + except ValueError: + return Role.OTHER + + from perdoo.models.metron_info import ( + Arc, + Credit, + Resource, + Role, + RoleResource, + Series, + Source, + Sources, + ) + + return MetronInfo( + id=Sources(primary=Source(source=InformationSource.COMIC_VINE, value=issue.id)), + publisher=Resource(id=series.publisher.id, value=series.publisher.name), + series=Series(id=series.id, name=series.name), + collection_title=issue.name, + number=issue.number, + summary=issue.summary, + cover_date=issue.cover_date, + store_date=issue.store_date, + arcs=[Arc(id=x.id, name=x.name) for x in issue.story_arcs], + characters=[Resource(id=x.id, value=x.name) for x in issue.characters], + teams=[Resource(id=x.id, value=x.name) for x in issue.teams], + locations=[Resource(id=x.id, value=x.name) for x in issue.locations], + url=issue.site_url, + credits=[ + Credit( + creator=Resource(id=x.id, value=x.name), + roles=[ + RoleResource(value=load_role(value=r)) + for r in re.split(r"[~\r\n,]+", x.roles) + ], + ) + for x in issue.creators + ], + ) + + def _process_comic_info(self: BaseService, series: Volume, issue: Issue) -> ComicInfo | None: + comic_info = ComicInfo( + title=issue.name, + series=series.name, + number=issue.number, + summary=issue.summary, + publisher=series.publisher.name if series.publisher else None, + web=issue.site_url, ) + + comic_info.cover_date = issue.cover_date + comic_info.credits = { + x.name: [r.strip() for r in re.split(r"[~\r\n,]+", x.roles)] for x in issue.creators + } + comic_info.character_list = [x.name for x in issue.characters] + comic_info.team_list = [x.name for x in issue.teams] + comic_info.location_list = [x.name for x in issue.locations] + comic_info.story_arc_list = [x.name for x in issue.story_arcs] + + return comic_info + + def fetch( + self: Comicvine, details: Details + ) -> tuple[Metadata | None, MetronInfo | None, ComicInfo | None]: + if not details.series.comicvine and details.issue.comicvine: + try: + temp = self.session.get_issue(issue_id=details.issue.comicvine) + details.series.comicvine = temp.volume.id + except ServiceError: + pass + + series = self.fetch_series(details=details) + if not series: + return None, None, None + + issue = self.fetch_issue(series_id=series.id, details=details) if not issue: - return False - return True + return None, None, None + + metadata = self._process_metadata(series=series, issue=issue) + metron_info = self._process_metron_info(series=series, issue=issue) + comic_info = self._process_comic_info(series=series, issue=issue) + + return metadata, metron_info, comic_info diff --git a/perdoo/services/league.py b/perdoo/services/league.py index 3b34d56..afb4397 100644 --- a/perdoo/services/league.py +++ b/perdoo/services/league.py @@ -5,19 +5,23 @@ import logging from himon.league_of_comic_geeks import LeagueofComicGeeks as Himon +from himon.schemas.comic import Comic +from himon.schemas.series import Series from himon.sqlite_cache import SQLiteCache from perdoo import get_cache_dir from perdoo.models import ComicInfo, Metadata, MetronInfo +from perdoo.services._base import BaseService from perdoo.settings import LeagueofComicGeeks as LeagueSettings +from perdoo.utils import Details LOGGER = logging.getLogger(__name__) -class League: +class League(BaseService[Series, Comic]): def __init__(self: League, settings: LeagueSettings): cache = SQLiteCache(path=get_cache_dir() / "himon.sqlite", expiry=14) - self.himon = Himon( + self.session = Himon( client_id=settings.client_id, client_secret=settings.client_secret, access_token=settings.access_token, @@ -25,12 +29,31 @@ def __init__(self: League, settings: LeagueSettings): ) if not settings.access_token: LOGGER.info("Generating new access token") - self.himon.access_token = settings.access_token = self.himon.generate_access_token() + self.session.access_token = settings.access_token = self.session.generate_access_token() + + def _get_series_id(self: League, title: str) -> int | None: + pass + + def fetch_series(self: League, details: Details) -> Series | None: + pass + + def _get_issue_id(self: League, series_id: int, number: str | None) -> int | None: + pass + + def fetch_issue(self: League, series_id: int, details: Details) -> Comic | None: + pass + + def _process_metadata(self: League, series: Series, issue: Comic) -> Metadata | None: + pass + + def _process_metron_info(self: League, series: Series, issue: Comic) -> MetronInfo | None: + pass + + def _process_comic_info(self: League, series: Series, issue: Comic) -> ComicInfo | None: + pass def fetch( self: League, - metadata: Metadata, # noqa: ARG002 - metron_info: MetronInfo, # noqa: ARG002 - comic_info: ComicInfo, # noqa: ARG002 - ) -> bool: - return False + details: Details, # noqa: ARG002 + ) -> tuple[Metadata | None, MetronInfo | None, ComicInfo | None]: + return None, None, None diff --git a/perdoo/services/marvel.py b/perdoo/services/marvel.py index e89c481..e679fe9 100644 --- a/perdoo/services/marvel.py +++ b/perdoo/services/marvel.py @@ -4,27 +4,50 @@ import logging +from esak.comic import Comic +from esak.series import Series from esak.session import Session as Esak from esak.sqlite_cache import SqliteCache from perdoo import get_cache_dir from perdoo.models import ComicInfo, Metadata, MetronInfo +from perdoo.services._base import BaseService from perdoo.settings import Marvel as MarvelSettings +from perdoo.utils import Details LOGGER = logging.getLogger(__name__) -class Marvel: +class Marvel(BaseService[Series, Comic]): def __init__(self: Marvel, settings: MarvelSettings): cache = SqliteCache(db_name=str(get_cache_dir() / "mokkari.sqlite"), expire=14) - self.esak = Esak( + self.session = Esak( public_key=settings.public_key, private_key=settings.private_key, cache=cache ) + def _get_series_id(self: Marvel, title: str) -> int | None: + pass + + def fetch_series(self: Marvel, details: Details) -> Series | None: + pass + + def _get_issue_id(self: Marvel, series_id: int, number: str | None) -> int | None: + pass + + def fetch_issue(self: Marvel, series_id: int, details: Details) -> Comic | None: + pass + + def _process_metadata(self: Marvel, series: Series, issue: Comic) -> Metadata | None: + pass + + def _process_metron_info(self: Marvel, series: Series, issue: Comic) -> MetronInfo | None: + pass + + def _process_comic_info(self: Marvel, series: Series, issue: Comic) -> ComicInfo | None: + pass + def fetch( self: Marvel, - metadata: Metadata, # noqa: ARG002 - metron_info: MetronInfo, # noqa: ARG002 - comic_info: ComicInfo, # noqa: ARG002 - ) -> bool: - return False + details: Details, # noqa: ARG002 + ) -> tuple[Metadata | None, MetronInfo | None, ComicInfo | None]: + return None, None, None diff --git a/perdoo/services/metron.py b/perdoo/services/metron.py index 0ad256c..b01d744 100644 --- a/perdoo/services/metron.py +++ b/perdoo/services/metron.py @@ -3,10 +3,10 @@ __all__ = ["Metron"] import logging +from datetime import date from mokkari.exceptions import ApiError from mokkari.schemas.issue import Issue -from mokkari.schemas.publisher import Publisher from mokkari.schemas.series import Series from mokkari.session import Session as Mokkari from mokkari.sqlite_cache import SqliteCache @@ -17,257 +17,21 @@ from perdoo.models import ComicInfo, Metadata, MetronInfo from perdoo.models.metadata import Source from perdoo.models.metron_info import InformationSource -from perdoo.services import BaseService +from perdoo.services._base import BaseService from perdoo.settings import Metron as MetronSettings +from perdoo.utils import Details LOGGER = logging.getLogger(__name__) -def add_publisher_to_metadata(publisher: Publisher, metadata: Metadata) -> None: - from perdoo.models.metadata import Resource - - resources = set(metadata.issue.series.publisher.resources) - if publisher.cv_id: - resources.add(Resource(source=Source.COMICVINE, value=publisher.cv_id)) - resources.add(Resource(source=Source.METRON, value=publisher.id)) - metadata.issue.series.publisher.resources = list(resources) - metadata.issue.series.publisher.title = publisher.name - - -def add_publisher_to_metron_info(publisher: Publisher, metron_info: MetronInfo) -> None: - if not metron_info.id or metron_info.id.source == InformationSource.METRON: - metron_info.publisher.id = publisher.id - if not metron_info.id or metron_info.id.source == InformationSource.COMIC_VINE: - metron_info.publisher.id = publisher.cv_id - metron_info.publisher.value = publisher.name - - -def add_publisher_to_comic_info(publisher: Publisher, comic_info: ComicInfo) -> None: - comic_info.publisher = publisher.name - - -def add_series_to_metadata(series: Series, metadata: Metadata) -> None: - from perdoo.models.metadata import Resource, TitledResource - - resources = set(metadata.issue.series.resources) - if series.cv_id: - resources.add(Resource(source=Source.COMICVINE, value=series.cv_id)) - resources.add(Resource(source=Source.METRON, value=series.id)) - metadata.issue.series.resources = list(resources) - metadata.issue.series.genres = [ - TitledResource(resources=[Resource(source=Source.METRON, value=x.id)], title=x.name) - for x in series.genres - ] - metadata.issue.series.start_year = series.year_began - metadata.issue.series.title = series.name - metadata.issue.series.volume = series.volume - - -def add_series_to_metron_info(series: Series, metron_info: MetronInfo) -> None: - from perdoo.models.metron_info import Format - - if not metron_info.id or metron_info.id.source == InformationSource.METRON: - metron_info.series.id = series.id - if not metron_info.id or metron_info.id.source == InformationSource.COMIC_VINE: - metron_info.series.id = series.cv_id - metron_info.series.name = series.name - metron_info.series.sort_name = series.sort_name - metron_info.series.volume = series.volume - metron_info.series.format = Format.load(value=series.series_type.name) - - -def add_series_to_comic_info(series: Series, comic_info: ComicInfo) -> None: - comic_info.series = series.name - comic_info.volume = series.volume - comic_info.genre_list = [x.name for x in series.genres] - comic_info.format = series.series_type.name - - -def add_issue_to_metadata(issue: Issue, metadata: Metadata) -> None: - from perdoo.models.metadata import Credit, Format, Resource, StoryArc, TitledResource - - resources = set(metadata.issue.resources) - if issue.cv_id: - resources.add(Resource(source=Source.COMICVINE, value=issue.cv_id)) - resources.add(Resource(source=Source.METRON, value=issue.id)) - metadata.issue.resources = list(resources) - metadata.issue.story_arcs = [ - StoryArc(resources=[Resource(source=Source.METRON, value=x.id)], title=x.name) - for x in issue.arcs - ] - metadata.issue.characters = [ - TitledResource(resources=[Resource(source=Source.METRON, value=x.id)], title=x.name) - for x in issue.characters - ] - metadata.issue.title = issue.collection_title if issue.collection_title else None - metadata.issue.cover_date = issue.cover_date - metadata.issue.credits = [ - Credit( - creator=TitledResource( - resources=[Resource(source=Source.METRON, value=x.id)], title=x.creator - ), - roles=[ - TitledResource(resources=[Resource(source=Source.METRON, value=r.id)], title=r.name) - for r in x.role - ], - ) - for x in issue.credits - ] - metadata.issue.summary = issue.desc - metadata.issue.number = issue.number - metadata.issue.page_count = issue.page_count or metadata.issue.page_count - try: - metadata.issue.format = Format.load(value=issue.series.series_type.name) - except ValueError: - metadata.issue.format = Format.COMIC - metadata.issue.store_date = issue.store_date - metadata.issue.teams = [ - TitledResource(resources=[Resource(source=Source.METRON, value=x.id)], title=x.name) - for x in issue.teams - ] - - -def add_issue_to_metron_info(issue: Issue, metron_info: MetronInfo) -> None: - from perdoo.models.metron_info import ( - GTIN, - AgeRating, - Arc, - Credit, - Price, - Resource, - Role, - RoleResource, - Source, - ) - - if not metron_info.id or metron_info.id.source == InformationSource.METRON: - metron_info.id = Source(source=InformationSource.METRON, value=issue.id) - if not metron_info.id or metron_info.id.source == InformationSource.COMIC_VINE: - metron_info.id = Source(source=InformationSource.COMIC_VINE, value=issue.cv_id) - metron_info.arcs = [Arc(id=x.id, name=x.name) for x in issue.arcs] - metron_info.characters = [Resource(id=x.id, value=x.name) for x in issue.characters] - metron_info.collection_title = issue.collection_title if issue.collection_title else None - metron_info.cover_date = issue.cover_date - credits_ = [] - for x in issue.credits: - roles = [] - for r in x.role: - try: - roles.append(RoleResource(id=r.id, value=Role.load(value=r.name))) - except ValueError: # noqa: PERF203 - roles.append(RoleResource(id=r.id, value=Role.OTHER)) - credits_.append(Credit(creator=Resource(id=x.id, value=x.creator), roles=roles)) - metron_info.credits = credits_ - metron_info.summary = issue.desc - metron_info.number = issue.number - metron_info.page_count = issue.page_count or metron_info.page_count - metron_info.prices = [Price(country="US", value=issue.price)] if issue.price else [] - metron_info.age_rating = AgeRating.load(value=issue.rating.name) - metron_info.reprints = [Resource(id=x.id, value=x.issue) for x in issue.reprints] - metron_info.url = str(issue.resource_url) - metron_info.store_date = issue.store_date - metron_info.stories = [Resource(value=x) for x in issue.story_titles] - metron_info.teams = [Resource(id=x.id, value=x.name) for x in issue.teams] - metron_info.gtin = GTIN(upc=issue.upc) if issue.upc else None - - -def add_issue_to_comic_info(issue: Issue, comic_info: ComicInfo) -> None: - comic_info.story_arc_list = [x.name for x in issue.arcs] - comic_info.character_list = [x.name for x in issue.characters] - comic_info.title = issue.collection_title - comic_info.cover_date = issue.cover_date - comic_info.credits = {x.creator: [r.name for r in x.role] for x in issue.credits} - comic_info.summary = issue.desc - comic_info.number = issue.number - comic_info.page_count = issue.page_count or comic_info.page_count - comic_info.web = issue.resource_url - comic_info.team_list = [x.name for x in issue.teams] - - -class Metron(BaseService[Publisher, Series, Issue]): +class Metron(BaseService[Series, Issue]): def __init__(self: Metron, settings: MetronSettings): cache = SqliteCache(db_name=str(get_cache_dir() / "mokkari.sqlite"), expire=14) self.session = Mokkari(username=settings.username, passwd=settings.password, cache=cache) - def _get_publisher(self: Metron, comicvine_id: int) -> int | None: - try: - publisher = self.session.publishers_list({"cv_id": comicvine_id}) - if publisher and len(publisher) >= 1: - return publisher[0].id - return None - except ApiError: - LOGGER.exception("") - return None - - def _search_publishers(self: Metron, title: str | None) -> int | None: - title = title or Prompt.ask("Publisher title", console=CONSOLE) - try: - options = sorted( - self.session.publishers_list(params={"name": title}), key=lambda x: x.name - ) - if not options: - LOGGER.warning("Unable to find any publishers with the title: '%s'", title) - index = create_menu( - options=[f"{x.id} | {x.name}" for x in options], - title="Metron Publisher", - default="None of the Above", - ) - if index != 0: - return options[index - 1].id - if not Confirm.ask("Try Again", console=CONSOLE): - return None - return self._search_publishers(title=None) - except ApiError: - LOGGER.exception("") - return None - - def _get_publisher_id(self: Metron, metadata: Metadata, metron_info: MetronInfo) -> int | None: - publisher_id = next( - ( - x.value - for x in metadata.issue.series.publisher.resources - if x.source == Source.METRON - ), - None, - ) or ( - metron_info.publisher.id - if metron_info.id and metron_info.id.source == InformationSource.METRON - else None - ) - if not publisher_id: - comicvine_id = next( - ( - x.value - for x in metadata.issue.series.publisher.resources - if x.source == Source.COMICVINE - ), - None, - ) or ( - metron_info.publisher.id - if metron_info.id and metron_info.id.source == InformationSource.COMIC_VINE - else None - ) - if comicvine_id: - publisher_id = self._get_publisher(comicvine_id=comicvine_id) - return publisher_id or self._search_publishers(title=metadata.issue.series.publisher.title) - - def fetch_publisher( - self: Metron, metadata: Metadata, metron_info: MetronInfo, comic_info: ComicInfo - ) -> Publisher | None: - publisher_id = self._get_publisher_id(metadata=metadata, metron_info=metron_info) - if not publisher_id: + def _get_series_via_comicvine(self: Metron, comicvine_id: int | None) -> int | None: + if not comicvine_id: return None - try: - publisher = self.session.publisher(_id=publisher_id) - add_publisher_to_metadata(publisher=publisher, metadata=metadata) - add_publisher_to_metron_info(publisher=publisher, metron_info=metron_info) - add_publisher_to_comic_info(publisher=publisher, comic_info=comic_info) - return publisher - except ApiError: - LOGGER.exception("") - return None - - def _get_series(self: Metron, comicvine_id: int) -> int | None: try: series = self.session.series_list({"cv_id": comicvine_id}) if series and len(series) >= 1: @@ -277,21 +41,21 @@ def _get_series(self: Metron, comicvine_id: int) -> int | None: LOGGER.exception("") return None - def _search_series(self: Metron, publisher_id: int, title: str | None) -> int | None: + def _get_series_id(self: Metron, title: str | None) -> int | None: title = title or Prompt.ask("Series title", console=CONSOLE) try: options = sorted( - self.session.series_list(params={"publisher_id": publisher_id, "name": title}), - key=lambda x: x.display_name, + self.session.series_list(params={"name": title}), key=lambda x: x.display_name ) if not options: - LOGGER.warning( - "Unable to find any Series with a PublisherId: %s and the title: '%s'", - publisher_id, - title, - ) + LOGGER.warning("Unable to find any Series with the title: '%s'", title) index = create_menu( - options=[f"{x.id} | {x.display_name}" for x in options], + options=[ + f"{x.id} | {x.display_name} v{x.volume}" + if x.volume > 1 + else f"{x.id} | {x.display_name}" + for x in options + ], title="Metron Series", default="None of the Above", ) @@ -299,74 +63,47 @@ def _search_series(self: Metron, publisher_id: int, title: str | None) -> int | return options[index - 1].id if not Confirm.ask("Try Again", console=CONSOLE): return None - return self._search_series(publisher_id=publisher_id, title=None) + return self._get_series_id(title=None) except ApiError: LOGGER.exception("") return None - def _get_series_id( - self: Metron, publisher_id: int, metadata: Metadata, metron_info: MetronInfo - ) -> int | None: - series_id = next( - (x.value for x in metadata.issue.series.resources if x.source == Source.METRON), None - ) or ( - metron_info.series.id - if metron_info.id and metron_info.id.source == InformationSource.METRON - else None - ) - if not series_id: - comicvine_id = next( - (x.value for x in metadata.issue.series.resources if x.source == Source.COMICVINE), - None, - ) or ( - metron_info.series.id - if metron_info.id and metron_info.id.source == InformationSource.COMIC_VINE - else None - ) - if comicvine_id: - series_id = self._get_series(comicvine_id=comicvine_id) - return series_id or self._search_series( - publisher_id=publisher_id, title=metadata.issue.series.title - ) - - def fetch_series( - self: Metron, - metadata: Metadata, - metron_info: MetronInfo, - comic_info: ComicInfo, - publisher_id: int, - ) -> Series | None: - series_id = self._get_series_id( - publisher_id=publisher_id, metadata=metadata, metron_info=metron_info + def fetch_series(self: Metron, details: Details) -> Series | None: + series_id = ( + details.series.metron + or self._get_series_via_comicvine(comicvine_id=details.series.comicvine) + or self._get_series_id(title=details.series.search) ) if not series_id: return None try: series = self.session.series(_id=series_id) - add_series_to_metadata(series=series, metadata=metadata) - add_series_to_metron_info(series=series, metron_info=metron_info) - add_series_to_comic_info(series=series, comic_info=comic_info) + details.series.metron = series_id return series except ApiError: LOGGER.exception("") return None - def _get_issue(self: Metron, comicvine_id: int) -> int | None: + def _get_issue_via_comicvine(self: Metron, comicvine_id: int | None) -> int | None: + if not comicvine_id: + return None try: - issue = self.session.issues_list({"cv_id": comicvine_id}) - if issue and len(issue) >= 1: - return issue[0].id + issues = self.session.issues_list({"cv_id": comicvine_id}) + if issues and len(issues) >= 1: + return issues[0].id return None except ApiError: LOGGER.exception("") return None - def _search_issues(self: Metron, series_id: int, number: str | None) -> int | None: + def _get_issue_id(self: Metron, series_id: int, number: str | None) -> int | None: try: options = sorted( - self.session.issues_list(params={"series_id": series_id, "number": number}) - if number - else self.session.issues_list(params={"series_id": series_id}), + self.session.issues_list( + params={"series_id": series_id, "number": number} + if number + else {"series_id": series_id} + ), key=lambda x: (x.number, x.issue_name), ) if not options: @@ -384,75 +121,226 @@ def _search_issues(self: Metron, series_id: int, number: str | None) -> int | No return options[index - 1].id if number: LOGGER.info("Searching again without the issue number") - return self._search_issues(series_id=series_id, number=None) + return self._get_issue_id(series_id=series_id, number=None) return None except ApiError: LOGGER.exception("") return None - def _get_issue_id( - self: Metron, series_id: int, metadata: Metadata, metron_info: MetronInfo - ) -> int | None: - issue_id = next( - (x.value for x in metadata.issue.resources if x.source == Source.METRON), None - ) or ( - metron_info.id.value - if metron_info.id and metron_info.id.source == InformationSource.METRON - else None - ) - if not issue_id: - comicvine_id = next( - (x.value for x in metadata.issue.resources if x.source == Source.COMICVINE), None - ) or ( - metron_info.id.value - if metron_info.id and metron_info.id.source == InformationSource.COMIC_VINE - else None - ) - if comicvine_id: - issue_id = self._get_issue(comicvine_id=comicvine_id) - return issue_id or self._search_issues(series_id=series_id, number=metadata.issue.number) - - def fetch_issue( - self: Metron, - metadata: Metadata, - metron_info: MetronInfo, - comic_info: ComicInfo, - series_id: int, - ) -> Issue | None: - issue_id = self._get_issue_id( - series_id=series_id, metadata=metadata, metron_info=metron_info + def fetch_issue(self: Metron, series_id: int, details: Details) -> Issue | None: + issue_id = ( + details.issue.metron + or self._get_issue_via_comicvine(comicvine_id=details.issue.comicvine) + or self._get_issue_id(series_id=series_id, number=details.issue.search) ) if not issue_id: return None try: issue = self.session.issue(_id=issue_id) - add_issue_to_metadata(issue=issue, metadata=metadata) - add_issue_to_metron_info(issue=issue, metron_info=metron_info) - add_issue_to_comic_info(issue=issue, comic_info=comic_info) + details.issue.metron = issue_id return issue except ApiError: LOGGER.exception("") return None - def fetch( - self: Metron, metadata: Metadata, metron_info: MetronInfo, comic_info: ComicInfo - ) -> bool: - publisher = self.fetch_publisher( - metadata=metadata, metron_info=metron_info, comic_info=comic_info + def _process_metadata(self: Metron, series: Series, issue: Issue) -> Metadata | None: + def load_format(value: str) -> Format: + try: + return Format.load(value=value.strip()) + except ValueError: + return Format.COMIC + + from perdoo.models.metadata import ( + Credit, + Format, + Issue, + Meta, + Resource, + Series, + StoryArc, + TitledResource, ) - if not publisher: - return False - series = self.fetch_series( - metadata=metadata, - metron_info=metron_info, - comic_info=comic_info, - publisher_id=publisher.id, + + resources = [Resource(source=Source.METRON, value=issue.id)] + if issue.cv_id: + resources.append(Resource(source=Source.COMICVINE, value=issue.cv_id)) + + series_resources = [Resource(source=Source.METRON, value=series.id)] + if series.cv_id: + resources.append(Resource(source=Source.COMICVINE, value=series.cv_id)) + + return Metadata( + issue=Issue( + characters=[ + TitledResource( + resources=[Resource(source=Source.METRON, value=x.id)], title=x.name + ) + for x in issue.characters + ], + cover_date=issue.cover_date, + credits=[ + Credit( + creator=TitledResource( + resources=[Resource(source=Source.METRON, value=x.id)], title=x.creator + ), + roles=[ + TitledResource( + resources=[Resource(source=Source.METRON, value=r.id)], title=r.name + ) + for r in x.role + ], + ) + for x in issue.credits + ], + format=load_format(value=issue.series.series_type.name), + number=issue.number, + page_count=issue.page_count or 0, + resources=resources, + series=Series( + genres=[ + TitledResource( + resources=[Resource(source=Source.METRON, value=x.id)], title=x.name + ) + for x in series.genres + ], + publisher=TitledResource( + resources=[Resource(source=Source.METRON, value=series.publisher.id)], + title=series.publisher.name, + ), + resources=series_resources, + start_year=series.year_began, + title=series.name, + volume=series.volume, + ), + store_date=issue.store_date, + story_arcs=[ + StoryArc(resources=[Resource(source=Source.METRON, value=x.id)], title=x.name) + for x in issue.arcs + ], + summary=issue.desc, + teams=[ + TitledResource( + resources=[Resource(source=Source.METRON, value=x.id)], title=x.name + ) + for x in issue.teams + ], + title=issue.collection_title or None, + ), + meta=Meta(date_=date.today()), ) - if not series: - return False - issue = self.fetch_issue( - metadata=metadata, metron_info=metron_info, comic_info=comic_info, series_id=series.id + + def _process_metron_info(self: Metron, series: Series, issue: Issue) -> MetronInfo | None: + def load_role(value: str) -> Role: + try: + return Role.load(value=value.strip()) + except ValueError: + return Role.OTHER + + from perdoo.models.metron_info import ( + GTIN, + AgeRating, + Arc, + Credit, + Format, + Genre, + GenreResource, + Price, + Resource, + Role, + RoleResource, + Series, + Source, + Sources, + Universe, + ) + + return MetronInfo( + id=Sources( + primary=Source(source=InformationSource.METRON, value=issue.id), + alternative=[Source(source=InformationSource.COMIC_VINE, value=issue.cv_id)] + if issue.cv_id + else [], + ), + publisher=Resource(id=series.publisher.id, value=series.publisher.name), + series=Series( + id=series.id, + name=series.name, + sort_name=series.sort_name, + volume=series.volume, + format=Format.load(value=series.series_type.name), + ), + collection_title=issue.collection_title or None, + number=issue.number, + stories=[Resource(value=x) for x in issue.story_titles], + summary=issue.desc, + prices=[Price(country="US", value=issue.price)] if issue.price else [], + cover_date=issue.cover_date, + store_date=issue.store_date, + page_count=issue.page_count or 0, + genres=[ + GenreResource(id=x.id, value=Genre.load(value=x.name)) for x in issue.series.genres + ], + arcs=[Arc(id=x.id, name=x.name) for x in issue.arcs], + characters=[Resource(id=x.id, value=x.name) for x in issue.characters], + teams=[Resource(id=x.id, value=x.name) for x in issue.teams], + universes=[Universe(id=x.id, name=x.name) for x in issue.universes], + gtin=GTIN(isbn=issue.isbn or None, upc=issue.upc or None) + if issue.isbn or issue.upc + else None, + age_rating=AgeRating.load(value=issue.rating.name), + reprints=[Resource(id=x.id, value=x.issue) for x in issue.reprints], + url=issue.resource_url, + credits=[ + Credit( + creator=Resource(id=x.id, value=x.creator), + roles=[RoleResource(id=r.id, value=load_role(value=r.name)) for r in x.role], + ) + for x in issue.credits + ], ) + + def _process_comic_info(self: Metron, series: Series, issue: Issue) -> ComicInfo | None: + comic_info = ComicInfo( + title=issue.collection_title, + series=series.name, + number=issue.number, + volume=series.volume, + summary=issue.desc, + publisher=series.publisher.name, + web=issue.resource_url, + page_count=issue.page_count or 0, + format=series.series_type.name, + ) + + comic_info.cover_date = issue.cover_date + comic_info.credits = {x.creator: [r.name for r in x.role] for x in issue.credits} + comic_info.genre_list = [x.name for x in series.genres] + comic_info.character_list = [x.name for x in issue.characters] + comic_info.team_list = [x.name for x in issue.teams] + comic_info.story_arc_list = [x.name for x in issue.arcs] + + return comic_info + + def fetch( + self: Metron, details: Details + ) -> tuple[Metadata | None, MetronInfo | None, ComicInfo | None]: + if not details.series.metron and details.issue.metron: + try: + temp = self.session.issue(_id=details.issue.metron) + details.series.metron = temp.series.id + except ApiError: + pass + + series = self.fetch_series(details=details) + if not series: + return None, None, None + + issue = self.fetch_issue(series_id=series.id, details=details) if not issue: - return False - return True + return None, None, None + + metadata = self._process_metadata(series=series, issue=issue) + metron_info = self._process_metron_info(series=series, issue=issue) + comic_info = self._process_comic_info(series=series, issue=issue) + + return metadata, metron_info, comic_info diff --git a/perdoo/settings.py b/perdoo/settings.py index 5f5ffe3..2dd361f 100644 --- a/perdoo/settings.py +++ b/perdoo/settings.py @@ -66,7 +66,7 @@ def load(value: str) -> OutputFormat: for entry in OutputFormat: if entry.value.casefold() == value.casefold(): return entry - raise ValueError(f"'{value}' isnt a valid OutputFormat") + raise ValueError(f"`{value}` isn't a valid OutputFormat") def __lt__(self: OutputFormat, other) -> int: # noqa: ANN001 if not isinstance(other, type(self)): diff --git a/perdoo/utils.py b/perdoo/utils.py index f258458..f4b2c2d 100644 --- a/perdoo/utils.py +++ b/perdoo/utils.py @@ -1,31 +1,34 @@ from __future__ import annotations -__all__ = [ - "list_files", - "sanitize", - "metron_to_metadata", - "comic_to_metadata", - "create_metadata", - "metadata_to_metron", - "metadata_to_comic", -] +__all__ = ["list_files", "sanitize", "Details", "Identifications", "get_metadata_id"] import logging import re -from datetime import date +from dataclasses import dataclass from pathlib import Path from natsort import humansorted, ns -from rich.prompt import Prompt -from perdoo import IMAGE_EXTENSIONS -from perdoo.archives import BaseArchive -from perdoo.console import CONSOLE, DatePrompt -from perdoo.models import ComicInfo, Metadata, MetronInfo +from perdoo.models.metadata import Resource, Source LOGGER = logging.getLogger(__name__) +@dataclass +class Identifications: + search: str | None = None + comicvine: int | None = None + league: int | None = None + marvel: int | None = None + metron: int | None = None + + +@dataclass +class Details: + series: Identifications | None + issue: Identifications | None + + def list_files(path: Path, *extensions: str) -> list[Path]: files = [] for file in path.iterdir(): @@ -47,457 +50,5 @@ def sanitize(value: str | None) -> str | None: return value.replace(" ", "-") -def metron_to_metadata(metron_info: MetronInfo) -> Metadata: - LOGGER.info("Generating Metadata from MetronInfo details") - from perdoo.models.metadata import ( - Credit, - Format, - Issue, - Meta, - Page, - PageType, - Resource, - Series, - Source, - StoryArc, - TitledResource, - Tool, - ) - - try: - source = Source.load(value=metron_info.id.source.value) if metron_info.id else None - except ValueError as err: - LOGGER.warning(err) - source = None - try: - format_ = Format.load(value=metron_info.series.format.value) - except ValueError as err: - LOGGER.warning(err) - format_ = Format.COMIC - - pages = [] - for x in metron_info.pages: - try: - type_ = PageType.load(value=x.type.value) - except ValueError as err: - LOGGER.warning(err) - type_ = PageType.OTHER - pages.append( - Page( - double_page=x.double_page, - filename="", - size=x.imageSize or 0, - height=x.imageHeight or 0, - width=x.imageWidth or 0, - index=x.image, - type_=type_, - ) - ) - - return Metadata( - issue=Issue( - characters=[ - TitledResource( - resources=[Resource(source=source, value=x.id)] if source and x.id else [], - title=x.value, - ) - for x in metron_info.characters - ], - cover_date=metron_info.cover_date, - credits=[ - Credit( - creator=TitledResource( - resources=[Resource(source=source, value=x.creator.id)] - if source and x.creator.id - else [], - title=x.creator.value, - ), - roles=[ - TitledResource( - resources=[Resource(source=source, value=y.id)] - if source and y.id - else [], - title=y.value.value, - ) - for y in x.roles - ], - ) - for x in metron_info.credits - ], - format_=format_, - language=metron_info.series.lang, - locations=[ - TitledResource( - resources=[Resource(source=source, value=x.id)] if source and x.id else [], - title=x.value, - ) - for x in metron_info.locations - ], - number=metron_info.number, - page_count=metron_info.page_count, - resources=[Resource(source=source, value=metron_info.id.value)] if source else [], - series=Series( - genres=[ - TitledResource( - resources=[Resource(source=source, value=x.id)] if source and x.id else [], - title=x.value.value, - ) - for x in metron_info.genres - ], - publisher=TitledResource( - resources=[Resource(source=source, value=metron_info.publisher.id)] - if source and metron_info.publisher.id - else [], - title=metron_info.publisher.value, - ), - resources=[Resource(source=source, value=metron_info.series.id)] - if source and metron_info.series.id - else [], - title=metron_info.series.name, - volume=metron_info.series.volume or 1, - ), - store_date=metron_info.store_date, - story_arcs=[ - StoryArc( - number=x.number, - resources=[Resource(source=source, value=x.id)] if source and x.id else [], - title=x.name, - ) - for x in metron_info.arcs - ], - summary=metron_info.summary, - teams=[ - TitledResource( - resources=[Resource(source=source, value=x.id)] if source and x.id else [], - title=x.value, - ) - for x in metron_info.teams - ], - title=metron_info.title, - ), - meta=Meta(date_=date.today(), tool=Tool(value="MetronInfo")), - notes=metron_info.notes, - pages=pages, - ) - - -def comic_to_metadata(comic_info: ComicInfo) -> Metadata: - LOGGER.info("Generating Metadata from ComicInfo details") - from perdoo.models.metadata import ( - Credit, - Format, - Issue, - Meta, - Page, - PageType, - Series, - StoryArc, - TitledResource, - Tool, - ) - - try: - format_ = Format.load(value=comic_info.format) if comic_info.format else Format.COMIC - except ValueError as err: - LOGGER.warning(err) - format_ = Format.COMIC - - pages = [] - for x in comic_info.pages: - try: - type_ = PageType.load(value=x.type.value) - except ValueError as err: - LOGGER.warning(err) - type_ = PageType.OTHER - pages.append( - Page( - double_page=x.double_page, - filename="", - size=x.image_size or 0, - height=x.image_height or 0, - width=x.image_width or 0, - index=x.image, - type_=type_, - ) - ) - - return Metadata( - issue=Issue( - characters=[TitledResource(title=x) for x in comic_info.character_list], - cover_date=comic_info.cover_date, - credits=[ - Credit( - creator=TitledResource(title=creator), - roles=[TitledResource(title=x) for x in roles], - ) - for creator, roles in comic_info.credits.items() - ], - format_=format_, - language=comic_info.language_iso, - locations=[TitledResource(title=x) for x in comic_info.location_list], - number=comic_info.number or Prompt.ask("Issue number", console=CONSOLE), - page_count=comic_info.page_count, - series=Series( - genres=[TitledResource(title=x) for x in comic_info.genre_list], - publisher=TitledResource( - title=comic_info.publisher or Prompt.ask("Publisher title", console=CONSOLE) - ), - start_year=comic_info.volume - if comic_info.volume and comic_info.volume >= 1900 - else None, - title=comic_info.series or Prompt.ask("Series title", console=CONSOLE), - volume=comic_info.volume if comic_info.volume and comic_info.volume < 1900 else 1, - ), - story_arcs=[StoryArc(title=x) for x in comic_info.story_arc_list], - summary=comic_info.summary, - teams=[TitledResource(title=x) for x in comic_info.team_list], - title=comic_info.title, - ), - meta=Meta(date_=date.today(), tool=Tool(value="ComicInfo")), - notes=comic_info.notes, - pages=pages, - ) - - -def create_metadata(archive: BaseArchive) -> Metadata: - LOGGER.info("Manually generating Metadata details") - from perdoo.models.metadata import Issue, Meta, Series, TitledResource, Tool - - return Metadata( - issue=Issue( - series=Series( - publisher=TitledResource(title=Prompt.ask("Publisher title", console=CONSOLE)), - title=Prompt.ask("Series title", console=CONSOLE), - ), - number=Prompt.ask("Issue number", console=CONSOLE), - page_count=len( - [x for x in archive.list_filenames() if Path(x).suffix in IMAGE_EXTENSIONS] - ), - ), - meta=Meta(date_=date.today(), tool=Tool(value="Manual")), - ) - - -def metadata_to_metron(metadata: Metadata) -> MetronInfo: - LOGGER.info("Generating MetronInfo from Metadata details") - from perdoo.models.metadata import Resource as MetadataResource, Source as MetadataSource - from perdoo.models.metron_info import ( - Arc, - Credit, - Format, - Genre, - GenreResource, - InformationSource, - Page, - PageType, - Resource, - Role, - RoleResource, - Series, - Source, - ) - - def get_primary_source( - resources: list[MetadataResource], resolution_order: list[MetadataSource] - ) -> Resource | None: - source_list = [x.source for x in resources] - for entry in resolution_order: - if entry in source_list: - index = source_list.index(entry) - return resources[index] - return None - - primary_source = get_primary_source( - resources=metadata.issue.resources, - resolution_order=[ - MetadataSource.MARVEL, - MetadataSource.METRON, - MetadataSource.GRAND_COMICS_DATABASE, - MetadataSource.COMICVINE, - MetadataSource.LEAGUE_OF_COMIC_GEEKS, - ], - ) - - try: - format_ = Format.load(value=metadata.issue.format.value) - except ValueError as err: - LOGGER.warning(err) - format_ = None - - genres = [] - for genre in metadata.issue.series.genres: - try: - value = Genre.load(value=genre.title) - except ValueError as err: - LOGGER.warning(err) - continue - genres.append( - GenreResource( - id=next((x.value for x in genre.resources if x.source == primary_source), None), - value=value, - ) - ) - credits_ = [] - for credit in metadata.issue.credits: - roles = [] - for role in credit.roles: - try: - value = Role.load(value=role.title) - except ValueError as err: - LOGGER.warning(err) - value = Role.OTHER - roles.append( - RoleResource( - id=next((x.value for x in role.resources if x.source == primary_source), None), - value=value, - ) - ) - credits_.append( - Credit( - creator=Resource( - id=next( - (x.value for x in credit.creator.resources if x.source == primary_source), - None, - ), - value=credit.creator.title, - ), - roles=roles, - ) - ) - pages = [] - for page in metadata.pages: - try: - type_ = PageType.load(value=page.type.value) - except ValueError as err: - LOGGER.warning(err) - type_ = PageType.OTHER - pages.append( - Page( - image=page.index, - type_=type_, - double_page=page.double_page, - image_size=page.size, - image_height=page.height, - image_width=page.width, - ) - ) - - return MetronInfo( - id=Source( - source=InformationSource.load(value=primary_source.value), - value=next( - (x.value for x in metadata.issue.resources if x.source == primary_source), None - ), - ) - if primary_source - else None, - publisher=Resource( - id=next( - ( - x.value - for x in metadata.issue.series.publisher.resources - if x.source == primary_source - ), - None, - ), - value=metadata.issue.series.publisher.title, - ), - series=Series( - id=next( - (x.value for x in metadata.issue.series.resources if x.source == primary_source), - None, - ), - lang=metadata.issue.language, - name=metadata.issue.series.title, - sort_name=metadata.issue.series.title, - volume=metadata.issue.series.volume, - format_=format_, - ), - collection_title=metadata.issue.title, - number=metadata.issue.number, - summary=metadata.issue.summary, - notes=metadata.notes, - cover_date=metadata.issue.cover_date - or DatePrompt.ask("Cover date", default=date.today(), console=CONSOLE), - store_date=metadata.issue.store_date, - page_count=metadata.issue.page_count, - genres=genres, - arcs=[ - Arc( - id=next((x.value for x in arc.resources if x.source == primary_source), None), - name=arc.title, - number=arc.number, - ) - for arc in metadata.issue.story_arcs - ], - characters=[ - Resource( - id=next((x.value for x in character.resources if x.source == primary_source), None), - value=character.title, - ) - for character in metadata.issue.characters - ], - teams=[ - Resource( - id=next((x.value for x in team.resources if x.source == primary_source), None), - value=team.title, - ) - for team in metadata.issue.teams - ], - locations=[ - Resource( - id=next((x.value for x in location.resources if x.source == primary_source), None), - value=location.title, - ) - for location in metadata.issue.locations - ], - credits=credits_, - pages=pages, - ) - - -def metadata_to_comic(metadata: Metadata) -> ComicInfo: - LOGGER.info("Generating ComicInfo from Metadata details") - from perdoo.models.comic_info import Page, PageType - - pages = [] - for page in metadata.pages: - try: - type_ = PageType.load(value=page.type.value) - except ValueError as err: - LOGGER.warning(err) - type_ = PageType.OTHER - pages.append( - Page( - image=page.index, - type_=type_, - double_page=page.double_page, - image_size=page.size, - image_height=page.height, - image_width=page.width, - ) - ) - - output = ComicInfo( - title=metadata.issue.title, - series=metadata.issue.series.title, - number=metadata.issue.number, - volume=metadata.issue.series.volume, - summary=metadata.issue.summary, - notes=metadata.notes, - publisher=metadata.issue.series.publisher.title, - page_count=metadata.issue.page_count, - language_iso=metadata.issue.language, - format_=metadata.issue.format.value, - pages=pages, - ) - output.cover_date = metadata.issue.cover_date or metadata.issue.store_date - output.credits = { - creator.title: [x.title for x in roles] for creator, roles in metadata.issue.credits - } - output.genre_list = [x.title for x in metadata.issue.series.genres] - output.character_list = [x.title for x in metadata.issue.characters] - output.team_list = [x.title for x in metadata.issue.teams] - output.location_list = [x.title for x in metadata.issue.locations] - output.story_arc_list = [x.title for x in metadata.issue.story_arcs] - return output +def get_metadata_id(resources: list[Resource], source: Source) -> int | None: + return next((x.value for x in resources if x.source == source), None) diff --git a/pyproject.toml b/pyproject.toml index 7b0d605..f233247 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -27,12 +27,12 @@ dependencies = [ "esak >= 1.3.2", "eval-type-backport >= 0.1.3 ; python_version < \"3.10\"", "himon >= 0.5.0", - "mokkari >= 3.0.0 ; python_version >= \"3.10\"", + "mokkari >= 3.1.0 ; python_version >= \"3.10\"", "mokkari@git+https://github.com/Buried-In-Code/mokkari ; python_version < \"3.10\"", "natsort >= 8.4.0", - "pillow >= 10.2.0", + "pillow >= 10.3.0", "pydantic >= 2.6.4", - "rarfile >= 4.1", + "rarfile >= 4.2", "rich >= 13.7.1", "simyan >= 1.2.1", "tomli >= 2.0.1 ; python_version < \"3.11\"",