manga_py/base_classes/base.py

   1 from logging import warning
   2 from os import path
   3 from typing import Optional, List, Union
   4 from requests import Response
   5 import re
   6
   7 from lxml.html import HtmlElement
   8
   9 from manga_py.http import Http
  10 from manga_py.http.flare_solver import Http as FS_Http
  11 from .params import ProviderParams
  12
  13
  14 CF_PROXY_RE = re.compile(r'(https?://[^/]+)')
  15
  16
  17 class Base(ProviderParams):
  18     _storage = None
  19     _params = None
  20     _image_params = None
  21     _http_kwargs = None
  22     __http = None
  23     __arguments = None
  24     _use_flare_solver = False
  25     __flare_solver_http = None
  26     _flare_solver_url = None
  27     chapter_id = 0
  28     quiet = False
  29     original_url = None
  30
  31     def __init__(self):
  32
  33         self._storage = {
  34             'cookies': {},
  35             'main_content': None,
  36             'chapters': [],
  37             'current_chapter': 0,
  38             'proxies': {},
  39             'domain_uri': None,
  40         }
  41         self._params = {
  42             'destination': 'Manga',
  43             'cf-protect': False,
  44             'user_agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:98.0) Gecko/20100101 Firefox/98.0',
  45         }
  46         self._image_params = {
  47             'crop': (0, 0, 0, 0),
  48             # 'crop': (left, upper, right, lower)
  49             'auto_crop': False,
  50             # 'auto_crop': True,
  51         }
  52         self._http_kwargs = {}
  53
  54     def _archive_type(self) -> str:
  55         arc_type = 'zip'
  56         if self._params['cbz']:
  57             arc_type = 'cbz'
  58         return arc_type
  59
  60     def get_url(self):
  61         return self._params['url']
  62
  63     def _build_http_params(self, params):
  64         if params is None:
  65             params = {}
  66         params.setdefault('allow_webp', not self._params.get('no_webp', False))
  67         params.setdefault('referer', self._storage.get('referer', self.domain))
  68         params.setdefault('user_agent', self._get_user_agent())
  69         params.setdefault('proxies', self._storage.get('proxies', None))
  70         params.setdefault('cookies', self._storage.get('cookies', None))
  71         params.setdefault('kwargs', self._http_kwargs)
  72         return params
  73
  74     def normalize_uri(self, uri, referer=None):
  75         return self.http_normal().normalize_uri(uri=uri, referer=referer)
  76
  77     def http(self, new=False, params=None) -> Union[FS_Http, Http]:
  78         if self._use_flare_solver:
  79             return self.flare_solver_http(new, params)
  80         else:
  81             return self.http_normal(new, params)
  82
  83     def flare_solver_http(self, new=False, params=None) -> FS_Http:
  84         allow_webp = True == (params or {}).get('no_webp', False)
  85         headers = {}
  86         if allow_webp:
  87             headers['Accept'] = Http.webp_header
  88         if self.__flare_solver_http is None:
  89             self.__flare_solver_http = FS_Http(self._flare_solver_url, self._get_user_agent())
  90             self.__flare_solver_http.create_session()
  91         if new:
  92             http = FS_Http(self._flare_solver_url, self._get_user_agent())
  93             http.create_session()
  94             return http
  95         return self.__flare_solver_http
  96
  97     def http_normal(self, new=False, params=None) -> Http:
  98         http_params = self._build_http_params(params)
  99         if new:
 100             http = Http(**http_params)
 101             return http
 102
 103         if self.__http is None:
 104             self.__http = Http(**http_params)
 105
 106         return self.__http
 107
 108     def http_get(self, url: str, headers: dict = None, cookies: dict = None):
 109         http = self.http()
 110         with http.get(url=url, headers=headers, cookies=cookies) as resp:
 111             if type(http) == Http:
 112                 return resp.text
 113             else:
 114                 content = resp.json().get('solution', {}).get('response', b'')
 115                 try:
 116                     return content.decode()
 117                 except AttributeError:
 118                     return content
 119
 120     def http_post(self, url: str, headers: dict = None, cookies: dict = None, data=()):
 121         http = self.http()
 122         with http.post(url=url, headers=headers, cookies=cookies, data=data) as resp:
 123             if type(http) == Http:
 124                 return resp.text
 125             else:
 126                 return resp.json().get('solution', {}).get('response', b'').decode()
 127
 128     def _get_user_agent(self):
 129         return self._params.get('user_agent', None)
 130
 131     @classmethod
 132     def __normalize_chapters(cls, n, element):
 133         if isinstance(element, HtmlElement):
 134             return n(element.get('href'))
 135         if isinstance(element, str):
 136             return n(element)
 137         return element
 138
 139     def _prepare_chapters(self, chapters):
 140         n = self.normalize_uri
 141         items = []
 142         if chapters and len(chapters):
 143             for i in chapters:
 144                 url = self.__normalize_chapters(n, i)
 145                 items.append(url)
 146         else:
 147             warning('Chapters list empty. Check %s' % self.get_url())
 148         return items
 149
 150     def book_meta(self) -> dict:
 151         return {}
 152
 153     def _image_name(self, idx, filename):
 154         fn, extension = path.splitext(filename)
 155         _path = '{:0>3}_{}'.format(idx, fn)
 156         if self._params['rename_pages']:
 157             _path = '{:0>3}'.format(idx)
 158         return _path + extension
 159
 160     def chapter_for_json(self) -> str:
 161         return self.chapter
 162
 163     def put_info_json(self, meta):
 164         # manga_name, url, directory
 165         pass
 166
 167     def _fill_arguments(self, arguments: List[str]):
 168         know_args = [
 169             'login',
 170             'password',
 171             'language',
 172             'translator',
 173         ]
 174
 175         if self.__arguments is None:
 176             self.__arguments = {}
 177
 178         for arg in arguments:
 179             key, value = arg.split('=', 1)  # type: str, str
 180             if key in know_args:
 181                 self.__arguments[key] = value
 182
 183     def arg(self, key: str) -> Optional[str]:
 184         if self.__arguments is None:
 185             return None
 186         return self.__arguments.get(key)
 187
 188     def allow_auto_change_url(self):
 189         return True
 190
 191     def cookies(self, response: Response) -> dict:
 192         if self._use_flare_solver:
 193             return response.json().get('solution', {}).get('cookies')
 194         return response.cookies.__dict__
 195
 196     @property
 197     def cf_proxy(self) -> Optional[str]:
 198         cf = self._params.get('cf_proxy')
 199         if cf is not None:
 200             cf = CF_PROXY_RE.search(cf)
 201         return cf.group(1) if cf else None
 202
 203     def __del__(self):
 204         if self.__flare_solver_http is not None:
 205             self.flare_solver_http().destroy_session()