"""Handles collecting data from source talkers. """ # Copyright 2012-2014 Anthony Beville # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. from __future__ import annotations import logging from typing import Callable from comicapi.genericmetadata import GenericMetadata from comictalker.resulttypes import ComicIssue, ComicVolume from comictalker.talkerbase import SourceStaticOptions, TalkerError logger = logging.getLogger(__name__) # To signal image loaded etc. def list_fetch_complete(url_list: list[str]) -> None: ... def url_fetch_complete(image_url: str, thumb_url: str | None) -> None: ... class ComicTalker: alt_url_list_fetch_complete = list_fetch_complete url_fetch_complete = url_fetch_complete def __init__(self, source_name) -> None: # ID of the source to use e.g. comicvine self.source = source_name # Retrieve the available sources modules self.sources = self.get_talkers() # Set the active talker self.talker = self.get_active_talker() self.static_options = self.get_static_options() def get_active_talker(self): # This should always work because it will have errored at get_talkers if there are none if not self.sources[self.source] is None: return self.sources[self.source] @staticmethod def get_talkers(): # Hardcode import for now. Placed here to prevent circular import import comictalker.talkers.comicvine return {"comicvine": comictalker.talkers.comicvine.ComicVineTalker()} # For issueidentifier def set_log_func(self, log_func: Callable[[str], None]) -> None: self.talker.log_func = log_func def get_static_options(self) -> SourceStaticOptions: return self.talker.source_details.static_options def check_api_key(self, key: str, url: str, source_id: str): for source in self.sources.values(): if source.source_details.id == source_id: return source.check_api_key(key, url) # Return false as back up or error? return False # Master function to search for series/volumes def search_for_series( self, series_name: str, callback: Callable[[int, int], None] | None = None, refresh_cache: bool = False, literal: bool = False, ) -> list[ComicVolume]: try: series_result = self.talker.search_for_series(series_name, callback, refresh_cache, literal) return series_result except NotImplementedError: logger.warning(f"{self.talker.source_details.name} has not implemented: 'search_for_series'") raise TalkerError( self.talker.source_details.name, 4, "The source has not implemented: 'search_for_series'", ) # Master function to fetch series data (i.e for sources without issue level details) def fetch_volume_data(self, series_id: int) -> GenericMetadata: try: series_result = self.talker.fetch_volume_data(series_id) return series_result except NotImplementedError: logger.warning(f"{self.talker.source_details.name} has not implemented: 'fetch_volume_data'") raise TalkerError( self.talker.source_details.name, 4, "The source has not implemented: 'fetch_volume_data'", ) # Master function to get issues in a series/volume def fetch_issues_by_volume(self, series_id: int) -> list[ComicIssue]: try: issues_result = self.talker.fetch_issues_by_volume(series_id) return issues_result except NotImplementedError: logger.warning(f"{self.talker.source_details.name} has not implemented: 'fetch_issues_by_volume'") raise TalkerError( self.talker.source_details.name, 4, "The source has not implemented: 'fetch_issues_by_volume'", ) # Master function to get issue information def fetch_issue_data(self, series_id: int, issue_number: str) -> GenericMetadata: try: issue_result = self.talker.fetch_issue_data(series_id, issue_number) return issue_result except NotImplementedError: logger.warning(f"{self.talker.source_details.name} has not implemented: 'fetch_issue_data'") raise TalkerError( self.talker.source_details.name, 4, "The source has not implemented: 'fetch_issue_data'", ) # For CLI def fetch_issue_data_by_issue_id(self, issue_id: int) -> GenericMetadata: try: issue_result = self.talker.fetch_issue_data_by_issue_id(issue_id) return issue_result except NotImplementedError: logger.warning(f"{self.talker.source_details.name} has not implemented: 'fetch_issue_data_by_issue_id'") raise TalkerError( self.talker.source_details.name, 4, "The source has not implemented: 'fetch_issue_data_by_issue_id'", ) # For issueidentifer def fetch_alternate_cover_urls(self, issue_id: int, issue_page_url: str) -> list[str]: try: alt_covers = self.talker.fetch_alternate_cover_urls(issue_id, issue_page_url) return alt_covers except NotImplementedError: logger.warning(f"{self.talker.source_details.name} has not implemented: 'fetch_alternate_cover_urls'") raise TalkerError( self.talker.source_details.name, 4, "The source has not implemented: 'fetch_alternate_cover_urls'", ) # For issueidentifier def fetch_issues_by_volume_issue_num_and_year( self, volume_id_list: list[int], issue_number: str, year: str | int | None ) -> list[ComicIssue]: try: issue_results = self.talker.fetch_issues_by_volume_issue_num_and_year(volume_id_list, issue_number, year) return issue_results except NotImplementedError: logger.warning( f"{self.talker.source_details.name} has not implemented: 'fetch_issues_by_volume_issue_num_and_year'" ) raise TalkerError( self.talker.source_details.name, 4, "The source has not implemented: 'fetch_issues_by_volume_issue_num_and_year'", ) def fetch_issue_cover_urls(self, issue_id: int) -> tuple[str | None, str | None]: try: cover_urls = self.talker.fetch_issue_cover_urls(issue_id) return cover_urls except NotImplementedError: logger.warning(f"{self.talker.source_details.name} has not implemented: 'fetch_issue_cover_urls'") raise TalkerError( self.talker.source_details.name, 4, "The source has not implemented: 'fetch_issue_cover_urls'", ) # Master function to get issue cover. Used by coverimagewidget def async_fetch_issue_cover_urls(self, issue_id: int) -> None: try: # TODO: Figure out async image_url, thumb_url = self.fetch_issue_cover_urls(issue_id) ComicTalker.url_fetch_complete(image_url or "", thumb_url) logger.info("Should be downloading image: %s thumb: %s", image_url, thumb_url) return # Should be all that's needed? CV functions will trigger everything. self.talker.async_fetch_issue_cover_urls(issue_id) except NotImplementedError: logger.warning(f"{self.talker.source_details.name} has not implemented: 'async_fetch_issue_cover_urls'") # Used by coverimagewidget.start_alt_cover_search def fetch_issue_page_url(self, issue_id: int) -> str | None: try: page_url = self.talker.fetch_issue_page_url(issue_id) return page_url except NotImplementedError: logger.warning(f"{self.talker.source_details.name} has not implemented: 'fetch_issue_page_url'") return None def async_fetch_alternate_cover_urls(self, issue_id: int, issue_page_url: str) -> None: try: # TODO: Figure out async url_list = self.fetch_alternate_cover_urls(issue_id, issue_page_url) ComicTalker.alt_url_list_fetch_complete(url_list) logger.info("Should be downloading alt image list: %s", url_list) return self.talker.async_fetch_alternate_cover_urls(issue_id, issue_page_url) except NotImplementedError: logger.warning(f"{self.talker.source_details.name} has not implemented: 'async_fetch_alternate_cover_urls'")