# encoding:utf-8 # --------------- # functions are placed here to remove cyclic import issues from placement in helpers # import ast import codecs import datetime import getpass import hashlib import io import logging import os import re import shutil import socket import stat import subprocess import sys import tempfile import threading import time import traceback import unicodedata from exceptions_helper import ex, ConnectionSkipException from json_helper import json_loads from cachecontrol import CacheControl, caches from lib.dateutil.parser import parser # from lib.tmdbsimple.configuration import Configuration # from lib.tmdbsimple.genres import Genres from cfscrape import CloudflareScraper from send2trash import send2trash # noinspection PyPep8Naming from encodingKludge import SYS_ENCODING import requests from _23 import decode_bytes, html_unescape, list_range, \ Popen, scandir, urlparse, urlsplit, urlunparse from six import integer_types, iteritems, iterkeys, itervalues, moves, PY2, string_types, text_type ACCEPT_ENCODING = "gzip,deflate" try: try: import brotlicffi as _unused_module_brotli # noqa: F401 except ImportError: import brotli as _unused_module_brotli # noqa: F401 except ImportError: pass else: ACCEPT_ENCODING += ",br" import zipfile # py7z hardwired removed, see comment below py7zr = None tz_p = parser() # noinspection PyUnreachableCode if False: from _23 import DirEntry from lxml_etree import etree try: # py7z hardwired removed because Python 3.9 interpreter crashes with a process kill signal 9 when memory is # low/exhausted during a native 7z compress action on Linux. Therefore, the native functions cannot be trusted. # `import` moved to this non-runtime scope to preserve code resolution in case reinstated at a later PY release # noinspection PyUnresolvedReferences,PyPackageRequirements import py7zr except ImportError: py7zr = None # sickgear is strictly used here for resolution, this is only possible because # this section is not used at runtime which would create circular reference issues # noinspection PyPep8Naming from sickgear import db, notifiers as NOTIFIERS # noinspection PyUnresolvedReferences from typing import Any, AnyStr, Dict, Generator, NoReturn, integer_types, Iterable, Iterator, List, Optional, \ Tuple, Type, Union html_convert_fractions = {0: '', 25: '¼', 50: '½', 75: '¾', 100: 1} PROG_DIR = os.path.join(os.path.dirname(os.path.normpath(os.path.abspath(__file__))), '..') # Mapping error status codes to official W3C names http_error_code = { 300: 'Multiple Choices', 301: 'Moved Permanently', 302: 'Found', 303: 'See Other', 304: 'Not Modified', 305: 'Use Proxy', 307: 'Temporary Redirect', 308: 'Permanent Redirect', 400: 'Bad Request', 401: 'Unauthorized', 402: 'Payment Required', 403: 'Forbidden', 404: 'Not Found', 405: 'Method Not Allowed', 406: 'Not Acceptable', 407: 'Proxy Authentication Required', 408: 'Request Timeout', 409: 'Conflict', 410: 'Gone', 411: 'Length Required', 412: 'Precondition Failed', 413: 'Request Entity Too Large', 414: 'Request-URI Too Long', 415: 'Unsupported Media Type', 416: 'Requested Range Not Satisfiable', 417: 'Expectation Failed', 429: 'Too Many Requests', 431: 'Request Header Fields Too Large', 444: 'No Response', 451: 'Unavailable For Legal Reasons', 500: 'Internal Server Error', 501: 'Not Implemented', 502: 'Bad Gateway', 503: 'Service Unavailable', 504: 'Gateway Timeout', 505: 'HTTP Version Not Supported', 511: 'Network Authentication Required'} logger = logging.getLogger('sg.helper') logger.addHandler(logging.NullHandler()) USER_AGENT = '' CACHE_DIR = None DATA_DIR = None PROXY_SETTING = None TRASH_REMOVE_SHOW = False REMOVE_FILENAME_CHARS = None MEMCACHE = {} FLARESOLVERR_HOST = None # noinspection PyRedeclaration db = None # noinspection PyRedeclaration NOTIFIERS = None class ConnectionFailTypes(object): http = 1 connection = 2 connection_timeout = 3 timeout = 4 other = 5 limit = 6 nodata = 7 names = {http: 'http', timeout: 'timeout', connection: 'connection', connection_timeout: 'connection_timeout', nodata: 'nodata', other: 'other', limit: 'limit'} def __init__(self): pass class ConnectionFail(object): def __init__(self, fail_type=ConnectionFailTypes.other, code=None, fail_time=None): self.code = code self.fail_type = fail_type self.fail_time = (datetime.datetime.now(), fail_time)[isinstance(fail_time, datetime.datetime)] class ConnectionFailDict(object): def __init__(self): self.domain_list = {} # type: Dict[AnyStr, ConnectionFailList] self.lock = threading.RLock() self.load_from_db() def load_from_db(self): if None is not db: with self.lock: my_db = db.DBConnection('cache.db') if my_db.has_table('connection_fails'): domains = my_db.select('SELECT DISTINCT domain_url from connection_fails') for domain in domains: self.domain_list[domain['domain_url']] = ConnectionFailList(domain['domain_url']) @staticmethod def get_domain(url): # type: (AnyStr) -> Optional[AnyStr] try: return urlsplit(url).hostname.lower() except (BaseException, Exception): pass def add_failure(self, url, fail_type): # type: (AnyStr, ConnectionFail) -> None host = self.get_domain(url) if None is not host: with self.lock: self.domain_list.setdefault(host, ConnectionFailList(host)).add_fail(fail_type) def inc_failure_count(self, url, # type: AnyStr *args, **kwargs): host = self.get_domain(url) if None is not host: with self.lock: if host in self.domain_list: domain = self.domain_list[host] fail_type = ('fail_type' in kwargs and kwargs['fail_type'].fail_type) or \ (isinstance(args, tuple) and isinstance(args[0], ConnectionFail) and args[0].fail_type) # noinspection PyProtectedMember if not isinstance(domain.failure_time, datetime.datetime) or \ fail_type != domain._last_fail_type or \ domain.fail_newest_delta() > datetime.timedelta(seconds=3): domain.failure_count += 1 domain.failure_time = datetime.datetime.now() domain._last_fail_type = fail_type domain.add_fail(*args, **kwargs) else: logger.debug('%s: Not logging same failure within 3 seconds' % url) def should_skip(self, url, log_warning=True, use_tmr_limit=True): # type: (AnyStr, bool, bool) -> bool host = self.get_domain(url) if None is not host: with self.lock: if host in self.domain_list: return self.domain_list[host].should_skip(log_warning=log_warning, use_tmr_limit=use_tmr_limit) return False DOMAIN_FAILURES = ConnectionFailDict() sp = 8 trakt_fail_times = {(i * sp) + m: s for m in range(1, 1 + sp) for i, s in enumerate([(0, 5), (0, 15), (0, 30), (1, 0), (2, 0)])} trakt_fail_times.update({i: s for i, s in enumerate([(3, 0), (6, 0), (12, 0), (24, 0)], len(trakt_fail_times))}) domain_fail_times = {'api.trakt.tv': trakt_fail_times} default_fail_times = {1: (0, 15), 2: (0, 30), 3: (1, 0), 4: (2, 0), 5: (3, 0), 6: (6, 0), 7: (12, 0), 8: (24, 0)} class ConnectionFailList(object): def __init__(self, url): # type: (AnyStr) -> None self.url = url self._fails = [] # type: List[ConnectionFail] self.lock = threading.Lock() self.clear_old() self.load_list() self.last_save = datetime.datetime.now() # type: datetime.datetime self._failure_count = 0 # type: int self._failure_time = None # type: Optional[datetime.datetime] self._tmr_limit_count = 0 # type: int self._tmr_limit_time = None # type: Optional[datetime.datetime] self._tmr_limit_wait = None # type: Optional[datetime.timedelta] self._last_fail_type = None # type: Optional[ConnectionFail] self.has_limit = False # type: bool self.fail_times = domain_fail_times.get(url, default_fail_times) # type: Dict[integer_types, Tuple[int, int]] self._load_fail_values() self.dirty = False # type: bool @property def failure_time(self): # type: (...) -> Union[None, datetime.datetime] return self._failure_time @failure_time.setter def failure_time(self, value): if None is value or isinstance(value, datetime.datetime): changed_val = self._failure_time != value self._failure_time = value if changed_val: # noinspection PyCallByClass,PyTypeChecker self._save_fail_value('failure_time', (_totimestamp(value), value)[None is value]) @property def tmr_limit_count(self): # type: (...) -> int return self._tmr_limit_count @tmr_limit_count.setter def tmr_limit_count(self, value): changed_val = self._tmr_limit_count != value self._tmr_limit_count = value if changed_val: self._save_fail_value('tmr_limit_count', value) def tmr_limit_update(self, period, unit, desc): # type: (Optional[AnyStr], Optional[AnyStr], AnyStr) -> None self.tmr_limit_time = datetime.datetime.now() self.tmr_limit_count += 1 limit_set = False if None not in (period, unit): limit_set = True if unit in ('s', 'sec', 'secs', 'seconds', 'second'): self.tmr_limit_wait = datetime.timedelta(seconds=try_int(period)) elif unit in ('m', 'min', 'mins', 'minutes', 'minute'): self.tmr_limit_wait = datetime.timedelta(minutes=try_int(period)) elif unit in ('h', 'hr', 'hrs', 'hours', 'hour'): self.tmr_limit_wait = datetime.timedelta(hours=try_int(period)) elif unit in ('d', 'days', 'day'): self.tmr_limit_wait = datetime.timedelta(days=try_int(period)) else: limit_set = False if not limit_set: time_index = self.fail_time_index(base_limit=0) self.tmr_limit_wait = self.wait_time(time_index) logger.warning('Request limit reached. Waiting for %s until next retry. Message: %s' % (self.tmr_limit_wait, desc or 'none found')) @property def tmr_limit_time(self): # type: (...) -> Union[None, datetime.datetime] return self._tmr_limit_time @tmr_limit_time.setter def tmr_limit_time(self, value): if None is value or isinstance(value, datetime.datetime): changed_val = self._tmr_limit_time != value self._tmr_limit_time = value if changed_val: # noinspection PyCallByClass,PyTypeChecker self._save_fail_value('tmr_limit_time', (_totimestamp(value), value)[None is value]) @property def last_fail(self): # type: (...) -> Optional[int] try: return sorted(self.fails, key=lambda x: x.fail_time, reverse=True)[0].fail_type except (BaseException, Exception): pass @property def failure_count(self): # type: (...) -> int return self._failure_count @failure_count.setter def failure_count(self, value): changed_val = self._failure_count != value self._failure_count = value if changed_val: self._save_fail_value('failure_count', value) def is_waiting(self): # type: (...) -> bool return self.fail_newest_delta() < self.wait_time() @property def max_index(self): # type: (...) -> int return len(self.fail_times) @property def tmr_limit_wait(self): # type: (...) -> Optional[datetime.timedelta] return self._tmr_limit_wait @tmr_limit_wait.setter def tmr_limit_wait(self, value): if isinstance(getattr(self, 'fails', None), ConnectionFailList) and isinstance(value, datetime.timedelta): self.add_fail(ConnectionFail(fail_type=ConnectionFailTypes.limit)) changed_val = self._tmr_limit_wait != value self._tmr_limit_wait = value if changed_val: if None is value: self._save_fail_value('tmr_limit_wait', value) elif isinstance(value, datetime.timedelta): self._save_fail_value('tmr_limit_wait', value.total_seconds()) def fail_time_index(self, base_limit=2): # type: (int) -> int i = max(self.failure_count - base_limit, 1) if i not in self.fail_times: i = list(self.fail_times)[-1] return (i, self.max_index)[i >= self.max_index] def valid_tmr_time(self): # type: (...) -> bool return isinstance(self.tmr_limit_wait, datetime.timedelta) and \ isinstance(self.tmr_limit_time, datetime.datetime) def wait_time(self, time_index=None): # type: (Optional[int]) -> datetime.timedelta """ Return a suitable wait time, selected by parameter, or based on the current failure count :param time_index: A key value index into the fail_times dict, or selects using failure count if None :return: Time """ if None is time_index: time_index = self.fail_time_index() return datetime.timedelta(hours=self.fail_times[time_index][0], minutes=self.fail_times[time_index][1]) def fail_newest_delta(self): # type: (...) -> datetime.timedelta """ Return how long since most recent failure :return: Period since most recent failure on record """ try: return datetime.datetime.now() - self.failure_time except (BaseException, Exception): return datetime.timedelta(days=1000) @property def get_next_try_time(self): # type: (...) -> datetime.timedelta n = None h = datetime.timedelta(seconds=0) f = datetime.timedelta(seconds=0) if self.valid_tmr_time(): h = self.tmr_limit_time + self.tmr_limit_wait - datetime.datetime.now() if 3 <= self.failure_count and isinstance(self.failure_time, datetime.datetime) and self.is_waiting(): h = self.failure_time + self.wait_time() - datetime.datetime.now() if datetime.timedelta(seconds=0) < max((h, f)): n = max((h, f)) return n def retry_next(self): if self.valid_tmr_time(): self.tmr_limit_time = datetime.datetime.now() - self.tmr_limit_wait if 3 <= self.failure_count and isinstance(self.failure_time, datetime.datetime) and self.is_waiting(): self.failure_time = datetime.datetime.now() - self.wait_time() @staticmethod def fmt_delta(delta): # type: (Union[datetime.datetime, datetime.timedelta]) -> AnyStr return str(delta).rsplit('.')[0] def should_skip(self, log_warning=True, use_tmr_limit=True): # type: (bool, bool) -> bool """ Determine if a subsequent server request should be skipped. The result of this logic is based on most recent server connection activity including, exhausted request limits, and counting connect failures to determine a "cool down" period before recommending reconnection attempts; by returning False. :param log_warning: Output to log if True (default) otherwise set False for no output. :param use_tmr_limit: Setting this to False will ignore a tmr limit being reached and will instead return False. :return: True for any known issue that would prevent a subsequent server connection, otherwise False. """ if self.valid_tmr_time(): time_left = self.tmr_limit_time + self.tmr_limit_wait - datetime.datetime.now() if time_left > datetime.timedelta(seconds=0): if log_warning: logger.warning('%sToo many requests reached at %s, waiting for %s' % ( self.url, self.fmt_delta(self.tmr_limit_time), self.fmt_delta(time_left))) return use_tmr_limit else: self.tmr_limit_time = None self.tmr_limit_wait = None if 3 <= self.failure_count: if None is self.failure_time: self.failure_time = datetime.datetime.now() if self.is_waiting(): if log_warning: time_left = self.wait_time() - self.fail_newest_delta() logger.warning('Failed %s times, skipping domain %s for %s, ' 'last failure at %s with fail type: %s' % (self.failure_count, self.url, self.fmt_delta(time_left), self.fmt_delta(self.failure_time), ConnectionFailTypes.names.get( self.last_fail, ConnectionFailTypes.names[ConnectionFailTypes.other]))) return True return False @property def fails(self): # type: (...) -> List return self._fails @property def fails_sorted(self): # type: (...) -> List fail_dict = {} b_d = {'count': 0} for e in self._fails: fail_date = e.fail_time.date() fail_hour = e.fail_time.time().hour date_time = datetime.datetime.combine(fail_date, datetime.time(hour=fail_hour)) if ConnectionFailTypes.names[e.fail_type] not in fail_dict.get(date_time, {}): default = {'date': str(fail_date), 'date_time': date_time, 'timestamp': try_int(_totimestamp(e.fail_time)), 'multirow': False} for et in itervalues(ConnectionFailTypes.names): default[et] = b_d.copy() fail_dict.setdefault(date_time, default)[ConnectionFailTypes.names[e.fail_type]]['count'] = 1 else: fail_dict[date_time][ConnectionFailTypes.names[e.fail_type]]['count'] += 1 if ConnectionFailTypes.http == e.fail_type: if e.code in fail_dict[date_time].get(ConnectionFailTypes.names[e.fail_type], {'code': {}}).get('code', {}): fail_dict[date_time][ConnectionFailTypes.names[e.fail_type]]['code'][e.code] += 1 else: fail_dict[date_time][ConnectionFailTypes.names[e.fail_type]].setdefault('code', {})[e.code] = 1 row_count = {} for (k, v) in iteritems(fail_dict): row_count.setdefault(v.get('date'), 0) if v.get('date') in row_count: row_count[v.get('date')] += 1 for (k, v) in iteritems(fail_dict): if 1 < row_count.get(v.get('date')): fail_dict[k]['multirow'] = True fail_list = sorted([fail_dict[k] for k in iterkeys(fail_dict)], key=lambda y: y.get('date_time'), reverse=True) totals = {} for fail_date in set([fail.get('date') for fail in fail_list]): daytotals = {} for et in itervalues(ConnectionFailTypes.names): daytotals.update({et: sum([x.get(et).get('count') for x in fail_list if fail_date == x.get('date')])}) totals.update({fail_date: daytotals}) for (fail_date, total) in iteritems(totals): for i, item in enumerate(fail_list): if fail_date == item.get('date'): if item.get('multirow'): fail_list[i:i] = [item.copy()] for et in itervalues(ConnectionFailTypes.names): fail_list[i][et] = {'count': total[et]} if et == ConnectionFailTypes.names[ConnectionFailTypes.http]: fail_list[i][et]['code'] = {} break return fail_list def add_fail(self, fail # type: ConnectionFail ): if isinstance(fail, ConnectionFail): with self.lock: self.dirty = True self._fails.append(fail) logger.debug('Adding fail.%s for %s' % (ConnectionFailTypes.names.get( fail.fail_type, ConnectionFailTypes.names[ConnectionFailTypes.other]), self.url)) self.save_list() def _load_fail_values(self): if None is not DATA_DIR: my_db = db.DBConnection('cache.db') if my_db.has_table('connection_fails_count'): r = my_db.select('SELECT * FROM connection_fails_count WHERE domain_url = ?', [self.url]) if r: self._failure_count = try_int(r[0]['failure_count'], 0) if r[0]['failure_time']: self._failure_time = datetime.datetime.fromtimestamp(r[0]['failure_time']) else: self._failure_time = None self._tmr_limit_count = try_int(r[0]['tmr_limit_count'], 0) if r[0]['tmr_limit_time']: self._tmr_limit_time = datetime.datetime.fromtimestamp(r[0]['tmr_limit_time']) else: self._tmr_limit_time = None if r[0]['tmr_limit_wait']: self._tmr_limit_wait = datetime.timedelta(seconds=try_int(r[0]['tmr_limit_wait'], 0)) else: self._tmr_limit_wait = None self._last_fail_type = self.last_fail def _save_fail_value(self, field, value): my_db = db.DBConnection('cache.db') if my_db.has_table('connection_fails_count'): r = my_db.action('UPDATE connection_fails_count SET %s = ? WHERE domain_url = ?' % field, [value, self.url]) if 0 == r.rowcount: my_db.action('REPLACE INTO connection_fails_count (domain_url, %s) VALUES (?,?)' % field, [self.url, value]) def save_list(self): if self.dirty: self.clear_old() if None is not db: with self.lock: try: my_db = db.DBConnection('cache.db') cl = [] for f in self._fails: cl.append(['INSERT OR IGNORE INTO connection_fails (domain_url, fail_type, fail_code, ' 'fail_time) ' 'VALUES (?,?,?,?)', [self.url, f.fail_type, f.code, _totimestamp(f.fail_time)]]) self.dirty = False if cl: my_db.mass_action(cl) except (BaseException, Exception): pass self.last_save = datetime.datetime.now() def load_list(self): if None is not db: with self.lock: try: my_db = db.DBConnection('cache.db') if my_db.has_table('connection_fails'): results = my_db.select('SELECT * FROM connection_fails WHERE domain_url = ?', [self.url]) self._fails = [] for r in results: try: self._fails.append(ConnectionFail( fail_type=try_int(r['fail_type']), code=try_int(r['fail_code']), fail_time=datetime.datetime.fromtimestamp(try_int(r['fail_time'])))) except (BaseException, Exception): continue except (BaseException, Exception): pass def clear_old(self): if None is not db: with self.lock: try: my_db = db.DBConnection('cache.db') if my_db.has_table('connection_fails'): # noinspection PyCallByClass,PyTypeChecker time_limit = _totimestamp(datetime.datetime.now() - datetime.timedelta(days=28)) my_db.action('DELETE FROM connection_fails WHERE fail_time < ?', [time_limit]) except (BaseException, Exception): pass def _totimestamp(dt=None): # type: (Optional[datetime.datetime]) -> integer_types """ This function should only be used in this module due to its 1970s+ limitation as that's all we need here and sgdatatime can't be used at this module level """ try: if PY2: import time return int(time.mktime(dt.timetuple())) return int(datetime.datetime.timestamp(dt)) except (BaseException, Exception): return 0 def _log_failure_url(url, post_data=None, post_json=None): # type: (AnyStr, Optional[AnyStr], Optional[AnyStr]) -> None if DOMAIN_FAILURES.should_skip(url, log_warning=False): post = [] if post_data: post += [' .. Post params: [%s]' % '&'.join([post_data])] if post_json: post += [' .. Json params: [%s]' % '&'.join([post_json])] logger.warning('Failure URL: %s%s' % (url, ''.join(post))) # try to convert to int, if the value is not already int def try_ord(c): # type: (Union[int, chr]) -> int if isinstance(c, int): return c return ord(c) # try to convert to int, if it fails the default will be returned def try_int(s, s_default=0): try: return int(s) except (BaseException, Exception): return s_default def try_date(s, s_default=None): # type: (AnyStr, Any) -> Optional[AnyStr] """ Convert string to a standard UTC date string :param s: :param s_default: :return: """ try: parse = tz_p.parse(clean_data(s)) return '%04d-%02d-%02d' % (parse.year, parse.month, parse.day) except(BaseException, Exception): return s_default def _maybe_request_url(e, def_url=''): return hasattr(e, 'request') and hasattr(e.request, 'url') and ' ' + e.request.url or def_url def clean_data(data): # type: (Any) -> Any """Cleans up strings, lists, dicts returned Issues corrected: - Replaces & with & - Replace multiple spaces with one space - Trailing whitespace - Decode html entities :param data: data :type data: List or Dict or AnyStr :return: :rtype: List or Dict or AnyStr """ if isinstance(data, list): return [clean_data(d) for d in data] if isinstance(data, set): return set(clean_data(d) for d in data) if isinstance(data, dict): return {k: clean_data(v) for k, v in iteritems(data)} if isinstance(data, string_types): return unicodedata.normalize('NFKD', re.sub(r' {2,}', ' ', html_unescape(data).strip().replace('&', '&'))) return data def get_system_temp_dir(): """ :return: Returns the [system temp dir]/tvdb_api-u501 (or tvdb_api-myuser) :rtype: AnyStr """ if hasattr(os, 'getuid'): uid = 'u%d' % (os.getuid()) else: # For Windows try: uid = getpass.getuser() except ImportError: return os.path.join(tempfile.gettempdir(), 'SickGear') return os.path.join(tempfile.gettempdir(), 'SickGear-%s' % uid) def proxy_setting(setting, request_url, force=False): """ Returns a list of a) proxy_setting address value or a PAC is fetched and parsed if proxy_setting starts with "PAC:" (case-insensitive) b) True/False if "PAC" is found in the proxy_setting. The PAC data parser is crude, javascript is not eval'd. The first "PROXY URL" found is extracted with a list of "url_a_part.url_remaining", "url_b_part.url_remaining", "url_n_part.url_remaining" and so on. Also, PAC data items are escaped for matching therefore regular expression items will not match a request_url. If force is True or request_url contains a PAC parsed data item then the PAC proxy address is returned else False. None is returned in the event of an error fetching PAC data. """ # check for "PAC" usage match = re.search(r'^\s*PAC:\s*(.*)', setting, re.I) if not match: return setting, False pac_url = match.group(1) # prevent a recursive test with existing proxy setting when fetching PAC url global PROXY_SETTING proxy_setting_backup = PROXY_SETTING PROXY_SETTING = '' resp = '' try: resp = get_url(pac_url) except (BaseException, Exception): pass PROXY_SETTING = proxy_setting_backup if not resp: return None, False proxy_address = None request_url_match = False parsed_url = urlparse(request_url) netloc = parsed_url.netloc for pac_data in re.finditer(r"""[^'"]*['"]([^.]+\.[^'"]*)['"]""", resp, re.I): data = re.search(r"""PROXY\s+([^'"]+)""", pac_data.group(1), re.I) if data: if force: return data.group(1), True # noinspection PyUnresolvedReferences proxy_address = (proxy_address, data.group(1))[None is proxy_address] elif re.search(re.escape(pac_data.group(1)), netloc, re.I): request_url_match = True if None is not proxy_address: break if None is proxy_address: return None, True return (False, proxy_address)[request_url_match], True def get_url(url, # type: AnyStr post_data=None, # type: Optional params=None, # type: Optional headers=None, # type: Optional[Dict] timeout=30, # type: int session=None, # type: Optional[requests.Session] parse_json=False, # type: bool memcache_cookies=None, # type: dict raise_status_code=False, # type: bool raise_exceptions=False, # type: bool as_binary=False, # type: bool encoding=None, # type: Optional[AnyStr] failure_monitor=True, # type: bool use_tmr_limit=True, # type: bool raise_skip_exception=False, # type: bool exclude_client_http_codes=True, # type: bool exclude_http_codes=(404, 429), # type: Tuple[integer_types] exclude_no_data=True, # type: bool use_method=None, # type: Optional[AnyStr] return_response=False, # type: bool **kwargs): # type: (...) -> Optional[Union[AnyStr, bool, bytes, Dict, Tuple[Union[Dict, List], requests.Session], requests.Response]] """ Return data from a URI with a possible check for authentication prior to the data fetch. Raised errors and no data in responses are tracked for making future logic decisions. # param url_solver=sickgear.FLARESOLVERR_HOST must be passed if url is behind CF for use in cf_scrape/__init__.py Returned data is either: 1) a byte-string retrieved from the URL provider. 2) a boolean if successfully used kwargs 'savefile' set to file pathname. 3) JSON dict if parse_json is True, and `Requests::session` when kwargs 'resp_sess' True. 4) `Requests::response`, and `Requests::session` when kwargs 'resp_sess' is True. :param url: address to request fetch data from :param post_data: if this or `post_json` is set, then request POST method is used to send this data :param params: :param headers: headers to add :param timeout: timeout :param session: optional session object :param parse_json: return JSON Dict :param memcache_cookies: memory persistent store for cookies :param raise_status_code: raise exception for status codes :param raise_exceptions: raise exceptions :param as_binary: return bytes instead of text :param encoding: overwrite encoding return header if as_binary is False :param failure_monitor: if True, will enable failure monitor for this request :param use_tmr_limit: an API limit can be +ve before a fetch, but unwanted, set False to short should_skip :param raise_skip_exception: if True, will raise ConnectionSkipException if this request should be skipped :param exclude_client_http_codes: if True, exclude client http codes 4XX from failure monitor :param exclude_http_codes: http codes to exclude from failure monitor, default: (404, 429) :param exclude_no_data: exclude no data as failure :param use_method: force any supported method by Session(): get, put, post, delete :param return_response: return response object :param kwargs: keyword params to passthru to Requests :return: None or data fetched from address """ domain = None if failure_monitor: domain = DOMAIN_FAILURES.get_domain(url) if domain not in DOMAIN_FAILURES.domain_list: DOMAIN_FAILURES.domain_list[domain] = ConnectionFailList(domain) if DOMAIN_FAILURES.should_skip(url, use_tmr_limit=use_tmr_limit): if raise_skip_exception: raise ConnectionSkipException return response_attr = ('text', 'content')[as_binary] # selectively mute some errors mute = list(filter(lambda x: kwargs.pop(x, False), [ 'mute_connect_err', 'mute_read_timeout', 'mute_connect_timeout', 'mute_http_error'])) # reuse or instantiate request session resp_sess = kwargs.pop('resp_sess', None) if None is session: session = CloudflareScraper.create_scraper() session.headers.update({'User-Agent': USER_AGENT}) proxy_browser = kwargs.get('proxy_browser') if isinstance(memcache_cookies, dict): parsed_url = urlparse(url) domain = parsed_url.netloc if domain in memcache_cookies: session.cookies.update(memcache_cookies[domain]) # download and save file or simply fetch url savename = kwargs.pop('savename', None) if savename: # session streaming session.stream = True if not kwargs.pop('nocache', False): cache_dir = CACHE_DIR or get_system_temp_dir() session = CacheControl(sess=session, cache=caches.FileCache(os.path.join(cache_dir, 'sessions'))) provider = kwargs.pop('provider', None) # handle legacy uses of `json` param if kwargs.get('json'): parse_json = kwargs.pop('json') post_json = kwargs.pop('post_json', None) # session main headers req_headers = {'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8', 'Accept-Encoding': ACCEPT_ENCODING} if headers: req_headers.update(headers) if hasattr(session, 'reserved') and 'headers' in session.reserved: req_headers.update(session.reserved['headers'] or {}) session.headers.update(req_headers) # session parameters session.params = params # session ssl verify session.verify = False # don't trust os environments (auth, proxies, ...) session.trust_env = False result = response = raised = connection_fail_params = log_failure_url = None try: # sanitise url parsed = list(urlparse(url)) parsed[2] = re.sub('/{2,}', '/', parsed[2]) # replace two or more / with one url = urlunparse(parsed) # session proxies if PROXY_SETTING: (proxy_address, pac_found) = proxy_setting(PROXY_SETTING, url) msg = '%sproxy for url: %s' % (('', 'PAC parsed ')[pac_found], url) if None is proxy_address: logger.debug('Proxy error, aborted the request using %s' % msg) return elif proxy_address: logger.debug('Using %s' % msg) session.proxies = {'http': proxy_address, 'https': proxy_address} if None is not use_method: method = getattr(session, use_method.strip().lower()) elif post_data or post_json: # decide if to post data or send a get request to server if True is post_data: post_data = None if post_data: kwargs.setdefault('data', post_data) if post_json: kwargs.setdefault('json', post_json) method = session.post else: method = session.get for r in range(0, 5): response = method(url, timeout=timeout, **kwargs) if not savename and response.ok and not response.content: if 'url=' in response.headers.get('Refresh', '').lower(): url = response.headers.get('Refresh').lower().split('url=')[1].strip('/') if not url.startswith('http'): parsed[2] = '/%s' % url url = urlunparse(parsed) response = session.get(url, timeout=timeout, **kwargs) elif 'github' in url: time.sleep(2) continue break # if encoding is not in header try to use best guess # ignore downloads with savename if not savename and not as_binary: if encoding: response.encoding = encoding elif not response.encoding or 'charset' not in response.headers.get('Content-Type', ''): response.encoding = response.apparent_encoding # noinspection PyProtectedMember if provider and provider._has_signature(response.text): result = getattr(response, response_attr) else: if raise_status_code: response.raise_for_status() if not response.ok: http_err_text = 'CloudFlare Ray ID' in response.text and \ 'CloudFlare reports, "Website is offline"; ' or '' if response.status_code in http_error_code: http_err_text += http_error_code[response.status_code] elif response.status_code in range(520, 527): http_err_text += 'Origin server connection failure' else: http_err_text = 'Custom HTTP error code' if 'mute_http_error' not in mute: logger.debug(f'Response not ok. {response.status_code}: {http_err_text} from requested url' f' {url}') except requests.exceptions.HTTPError as e: raised = e is_client_error = 400 <= e.response.status_code < 500 if failure_monitor and e.response.status_code not in exclude_http_codes and \ not (exclude_client_http_codes and is_client_error): connection_fail_params = dict(fail_type=ConnectionFailTypes.http, code=e.response.status_code) if not raise_status_code: logger.warning(f'HTTP error {e.errno} while loading URL{_maybe_request_url(e)}') except requests.exceptions.ConnectionError as e: raised = e if 'mute_connect_err' not in mute: logger.warning(f"Connection error msg:{ex(e)} while loading URL{_maybe_request_url(e)}") if failure_monitor: connection_fail_params = dict(fail_type=ConnectionFailTypes.connection) except requests.exceptions.ReadTimeout as e: raised = e if 'mute_read_timeout' not in mute: logger.warning(f'Read timed out msg:{ex(e)} while loading URL{_maybe_request_url(e)}') if failure_monitor: connection_fail_params = dict(fail_type=ConnectionFailTypes.timeout) except (requests.exceptions.Timeout, socket.timeout) as e: raised = e if 'mute_connect_timeout' not in mute: logger.warning(f'Connection timed out msg:{ex(e)} while loading URL {_maybe_request_url(e, url)}') if failure_monitor: connection_fail_params = dict(fail_type=ConnectionFailTypes.connection_timeout) except (BaseException, Exception) as e: raised = e logger.warning(('Exception caught while loading URL {0}\r\nDetail... %s\r\n{1}' % ex(e), 'Unknown exception while loading URL {0}\r\nDetail... {1}')[not ex(e)] .format(url, traceback.format_exc())) if failure_monitor: connection_fail_params = dict(fail_type=ConnectionFailTypes.other) log_failure_url = True finally: if None is not connection_fail_params: DOMAIN_FAILURES.inc_failure_count(url, ConnectionFail(**connection_fail_params)) save_failure(url, domain, log_failure_url, post_data, post_json) if isinstance(raised, Exception): if raise_exceptions or raise_status_code: try: if not hasattr(raised, 'text') and hasattr(response, 'text'): raised.text = response.text except (BaseException, Exception): pass raise raised return if return_response: result = response elif None is result and None is not response and response.ok: if isinstance(memcache_cookies, dict): parsed_url = urlparse(url) domain = parsed_url.netloc memcache_cookies[domain] = session.cookies.copy() if parse_json or proxy_browser: try: data_json = json_loads(requests_response=response) if proxy_browser: result = ({}, data_json.get('solution', {}).get('response', {}))[isinstance(data_json, dict)] else: result = ({}, data_json)[isinstance(data_json, (dict, list))] if resp_sess: result = result, session except (TypeError, Exception) as e: raised = e logger.warning(f'{("Proxy browser", "JSON")[parse_json]} data issue from URL {url}\r\n' f'Detail... {ex(e)}') elif savename: try: write_file(savename, response, raw=True, raise_exceptions=raise_exceptions) result = True except (BaseException, Exception) as e: raised = e else: result = getattr(response, response_attr) if resp_sess: result = result, session if raise_exceptions and isinstance(raised, Exception): raise raised if failure_monitor: if return_response or (result and not isinstance(result, tuple) or isinstance(result, tuple) and result[0]): domain = DOMAIN_FAILURES.get_domain(url) if 0 != DOMAIN_FAILURES.domain_list[domain].failure_count: logger.info('Unblocking: %s' % domain) DOMAIN_FAILURES.domain_list[domain].failure_count = 0 DOMAIN_FAILURES.domain_list[domain].failure_time = None save_failure(url, domain, False, post_data, post_json) elif not exclude_no_data: DOMAIN_FAILURES.inc_failure_count(url, ConnectionFail(fail_type=ConnectionFailTypes.nodata)) save_failure(url, domain, True, post_data, post_json) return result def save_failure(url, domain, log_failure_url, post_data, post_json): DOMAIN_FAILURES.domain_list[domain].save_list() if log_failure_url: _log_failure_url(url, post_data, post_json) def scantree(path, # type: AnyStr exclude=None, # type: Optional[AnyStr, List[AnyStr]] exclude_dirs=None, # type: Optional[AnyStr, List[AnyStr]] include=None, # type: Optional[AnyStr, List[AnyStr]] follow_symlinks=False, # type: bool filter_kind=None, # type: Optional[bool] recurse=True, # type: bool exclude_folders_with_files=None, # type: Optional[List[AnyStr]] internal_call=False, # type: bool rc_exc=None, # type: List rc_exc_dir=None, # type: List rc_inc=None, # type: List has_exclude=False, # type: bool has_exclude_dirs=False, # type: bool has_include=False # type: bool ): # type: (...) -> Generator[DirEntry, None, None] """Yield DirEntry objects for given path. Returns without yield if path fails sanity check :param path: Path to scan, sanity check is_dir and exists :param exclude: Escaped regex string(s) to exclude (files and directories) :param exclude_dirs: Escaped regex string(s) to exclude (directories only) :param include: Escaped regex string(s) to include :param follow_symlinks: Follow symlinks :param filter_kind: None to yield everything, True yields directories, False yields files :param recurse: Recursively scan the tree :param exclude_folders_with_files: exclude folder that contain the listed file(s) :param internal_call: internal use :param rc_exc: internal use :param rc_exc_dir: internal use :param rc_inc: internal use :param has_exclude: internal use :param has_exclude_dirs: internal use :param has_include: internal_use """ if isinstance(path, string_types) and path and os.path.isdir(path): if not internal_call: rc_exc, rc_exc_dir, rc_inc = [re.compile(rx % '|'.join( [x for x in (param, ([param], [])[None is param])[not isinstance(param, list)]])) for rx, param in ((r'(?i)^(?:(?!%s).)*$', exclude), (r'(?i)^(?:(?!%s).)*$', exclude_dirs), (r'(?i)%s', include))] has_exclude, has_exclude_dirs, has_include = bool(exclude), bool(exclude_dirs), bool(include) for entry in scandir(path): is_dir = entry.is_dir(follow_symlinks=follow_symlinks) is_file = entry.is_file(follow_symlinks=follow_symlinks) no_filter = any([None is filter_kind, filter_kind and is_dir, not filter_kind and is_file]) if ((not has_exclude or rc_exc.search(entry.name)) and (not has_exclude_dirs or not is_dir or rc_exc_dir.search(entry.name)) and (not has_include or rc_inc.search(entry.name)) and (no_filter or (not filter_kind and is_dir and recurse))): if is_dir and exclude_folders_with_files and any(os.path.isfile(os.path.join(entry.path, e_f)) for e_f in exclude_folders_with_files): logger.debug(f'Ignoring Folder: "{entry.path}", because it contains a exclude file' f' "{", ".join(exclude_folders_with_files)}"') continue if recurse and is_dir: for subentry in scantree( path=entry.path, exclude=exclude, exclude_dirs=exclude_dirs, include=include, follow_symlinks=follow_symlinks, filter_kind=filter_kind, recurse=recurse, exclude_folders_with_files=exclude_folders_with_files, internal_call=True, rc_exc=rc_exc, rc_exc_dir=rc_exc_dir, rc_inc=rc_inc, has_exclude=has_exclude, has_exclude_dirs=has_exclude_dirs, has_include=has_include): yield subentry if no_filter: yield entry def copy_file(src_file, dest_file): if os.name.startswith('posix'): subprocess.call(['cp', src_file, dest_file]) else: shutil.copyfile(src_file, dest_file) try: shutil.copymode(src_file, dest_file) except OSError: pass def move_file(src_file, dest_file, raise_exceptions=False): try: shutil.move(src_file, dest_file) fix_set_group_id(dest_file) except OSError: copy_file(src_file, dest_file) if os.path.exists(dest_file): fix_set_group_id(dest_file) os.unlink(src_file) elif raise_exceptions: raise OSError('Destination file could not be created: %s' % dest_file) def fix_set_group_id(child_path): """ :param child_path: path :type child_path: AnyStr :return: :rtype: None """ if os.name in ('nt', 'ce'): return parent_path = os.path.dirname(child_path) parent_stat = os.stat(parent_path) parent_mode = stat.S_IMODE(parent_stat[stat.ST_MODE]) if parent_mode & stat.S_ISGID: parent_gid = parent_stat[stat.ST_GID] child_stat = os.stat(child_path) child_gid = child_stat[stat.ST_GID] if child_gid == parent_gid: return child_path_owner = child_stat.st_uid user_id = os.geteuid() # only available on UNIX if 0 != user_id and user_id != child_path_owner: logger.debug(f'Not running as root or owner of {child_path}, not trying to set the set-group-id') return try: os.chown(child_path, -1, parent_gid) # only available on UNIX logger.debug(f'Respecting the set-group-ID bit on the parent directory for {child_path}') except OSError: logger.error(f'Failed to respect the set-group-id bit on the parent directory for {child_path}' f' (setting group id {parent_gid:d})') def remove_file_perm(filepath, log_err=True): # type: (AnyStr, Optional[bool]) -> Optional[bool] """ Remove file :param filepath: Path and file name :param log_err: False to suppress log msgs :return True if filepath does not exist else None if no removal """ if not os.path.exists(filepath): return True for t in list_range(10): # total seconds to wait 0 - 9 = 45s over 10 iterations try: os.remove(filepath) except OSError as e: if getattr(e, 'winerror', 0) not in (5, 32): # 5=access denied (e.g. av), 32=another process has lock if log_err: logger.warning('Unable to delete %s: %r / %s' % (filepath, e, ex(e))) return except (BaseException, Exception): pass time.sleep(t) if not os.path.exists(filepath): return True if log_err: logger.warning('Unable to delete %s' % filepath) def remove_file(filepath, tree=False, prefix_failure='', log_level=logging.INFO): """ Remove file based on setting for trash v permanent delete :param filepath: Path and file name :type filepath: String :param tree: Remove file tree :type tree: Bool :param prefix_failure: Text to prepend to error log, e.g. show id :type prefix_failure: String :param log_level: Log level to use for error :type log_level: Int :return: Type of removal ('Deleted' or 'Trashed') if filepath does not exist or None if no removal occurred :rtype: String or None """ result = None if filepath: for t in list_range(10): # total seconds to wait 0 - 9 = 45s over 10 iterations try: result = 'Deleted' if TRASH_REMOVE_SHOW: result = 'Trashed' send2trash(filepath) elif tree: shutil.rmtree(filepath) else: os.remove(filepath) except OSError as e: if getattr(e, 'winerror', 0) not in (5, 32): # 5=access denied (e.g. av), 32=another process has lock logger.log(level=log_level, msg=f'{prefix_failure}Unable to {("delete", "trash")[TRASH_REMOVE_SHOW]}' f' {("file", "dir")[tree]} {filepath}: {ex(e)}') break time.sleep(t) if not os.path.exists(filepath): break return (None, result)[filepath and not os.path.exists(filepath)] def touch_file(name, atime=None, dir_name=None): # type: (AnyStr, int, AnyStr) -> bool """ create an empty named file and set access time of file :param name: filename to touch with time :param atime: access time as epoch :param dir_name: create empty file and directory if file doesn't exist :return: success """ if None is not dir_name: name = os.path.join(dir_name, name) if make_path(dir_name): if not os.path.exists(name): with io.open(name, 'w') as fh: fh.flush() if None is atime: return True if None is not atime: try: with open(name, 'a'): os.utime(name, (atime, atime)) return True except (BaseException, Exception): logger.debug('File air date stamping not available on your OS') return False def make_path(name, syno=False): # type: (AnyStr, bool) -> bool """ Create an entire directory path and assign each subdir the parent permission :param name: directory to create :param syno: whether to trigger a syno library update for path :return: success or dir exists """ if not os.path.isdir(name): # Windows, create all missing folders if os.name in ('nt', 'ce'): try: logger.debug(f"Path {name} doesn't exist, creating it") os.makedirs(name) except (OSError, IOError) as e: logger.error(f'Failed creating {name} : {ex(e)}') return False # not Windows, create all missing folders and set permissions else: sofar = '' folder_list = name.split(os.path.sep) # look through each sub folder and make sure they all exist for cur_folder in folder_list: sofar += cur_folder + os.path.sep # if it exists then just keep walking down the line if os.path.isdir(sofar): continue try: logger.debug(f"Path {sofar} doesn't exist, creating it") os.mkdir(sofar) # use normpath to remove end separator, otherwise checks permissions against itself chmod_as_parent(os.path.normpath(sofar)) if syno: # do the library update for synoindex NOTIFIERS.NotifierFactory().get('SYNOINDEX').addFolder(sofar) except (OSError, IOError) as e: logger.error(f'Failed creating {sofar} : {ex(e)}') return False return True def chmod_as_parent(child_path): """ :param child_path: path :type child_path: AnyStr :return: :rtype: None """ if os.name in ('nt', 'ce'): return parent_path = os.path.dirname(child_path) if not parent_path: logger.debug(f'No parent path provided in {child_path}, unable to get permissions from it') return parent_path_stat = os.stat(parent_path) parent_mode = stat.S_IMODE(parent_path_stat[stat.ST_MODE]) child_path_stat = os.stat(child_path) child_path_mode = stat.S_IMODE(child_path_stat[stat.ST_MODE]) if os.path.isfile(child_path): child_mode = file_bit_filter(parent_mode) else: child_mode = parent_mode if child_path_mode == child_mode: return child_path_owner = child_path_stat.st_uid user_id = os.geteuid() # only available on UNIX if 0 != user_id and user_id != child_path_owner: logger.debug(f'Not running as root or owner of {child_path}, not trying to set permissions') return try: os.chmod(child_path, child_mode) logger.debug(f'Setting permissions for {child_path} to {child_mode:o} as parent directory has {parent_mode:o}') except OSError: logger.error(f'Failed to set permission for {child_path} to {child_mode:o}') def file_bit_filter(mode): for bit in [stat.S_IXUSR, stat.S_IXGRP, stat.S_IXOTH, stat.S_ISUID, stat.S_ISGID]: if mode & bit: mode -= bit return mode def write_file(filepath, # type: AnyStr data, # type: Union[AnyStr, etree.Element, requests.Response] raw=False, # type: bool xmltree=False, # type: bool xml_header=False, # type: bool utf8=False, # type: bool raise_exceptions=False # type: bool ): # type: (...) -> bool """ :param filepath: filepath :param data: data to write :param raw: write binary or text :param xmltree: use xmel tree :param xml_header: prepend xmel header :param utf8: use UTF8 :param raise_exceptions: raise excepitons :return: succuess """ result = False if make_path(os.path.dirname(filepath)): try: if raw: empty_file = True with io.FileIO(filepath, 'wb') as fh: for chunk in data.iter_content(chunk_size=1024): if chunk: empty_file = False fh.write(chunk) fh.flush() os.fsync(fh.fileno()) if empty_file: remove_file_perm(filepath, log_err=False) return result else: w_mode = 'w' if utf8: w_mode = 'a' with io.FileIO(filepath, 'wb') as fh: fh.write(codecs.BOM_UTF8) if xmltree: with io.FileIO(filepath, w_mode) as fh: params = {} if utf8: params = dict(encoding='utf-8') if xml_header: params.update(dict(xml_declaration=True)) data.write(fh, **params) else: if isinstance(data, text_type): with io.open(filepath, w_mode, encoding='utf-8') as fh: fh.write(data) else: with io.FileIO(filepath, w_mode) as fh: fh.write(data) chmod_as_parent(filepath) result = True except (EnvironmentError, IOError) as e: logger.error('Unable to write file %s : %s' % (filepath, ex(e))) if raise_exceptions: raise e return result def sanitize_filename(name): """ :param name: filename :type name: AnyStr :return: sanitized filename :rtype: AnyStr """ # remove bad chars from the filename name = re.sub(r'[\\/*]', '-', name) name = re.sub(r'[:"<>|?]', '', name) # remove leading/trailing periods and spaces name = name.strip(' .') for char in REMOVE_FILENAME_CHARS or []: name = name.replace(char, '') return name def replace_extension(filename, new_ext): # type: (AnyStr, AnyStr) -> AnyStr """ :param filename: filename :param new_ext: new extension :return: filename with new extension """ sep_file = filename.rpartition('.') if '' == sep_file[0]: return filename return '%s.%s' % (sep_file[0], new_ext) def long_path(path): # type: (AnyStr) -> AnyStr """add long path prefix for Windows""" if 'nt' == os.name and 260 < len(path) and not path.startswith('\\\\?\\') and os.path.isabs(path): return '\\\\?\\' + path return path def compress_file(target, filename, prefer_7z=True, remove_source=True): # type: (AnyStr, AnyStr, bool, bool) -> bool """ compress given file to zip or 7z archive :param target: file to compress with full path :param filename: filename inside the archive :param prefer_7z: prefer 7z over zip compression if available :param remove_source: remove source file after successful creation of archive :return: success of compression """ try: if prefer_7z and None is not py7zr: z_name = '%s.7z' % target.rpartition('.')[0] # noinspection PyUnresolvedReferences with py7zr.SevenZipFile(z_name, 'w') as z_file: z_file.write(target, filename) else: zip_name = '%s.zip' % target.rpartition('.')[0] with zipfile.ZipFile(zip_name, 'w', zipfile.ZIP_DEFLATED) as zip_fh: zip_fh.write(target, filename) except (BaseException, Exception) as e: logger.error('error compressing %s' % target) logger.debug('traceback: %s' % ex(e)) return False if remove_source: remove_file_perm(target) return True def cmdline_runner(cmd, shell=False, suppress_stderr=False, env=None): # type: (Union[AnyStr, List[AnyStr]], bool, bool, Dict) -> Tuple[AnyStr, Optional[AnyStr], int] """ Execute a child program in a new process. Can raise an exception to be caught in callee :param cmd: A string, or a sequence of program arguments :param shell: If true, the command will be executed through the shell. :param suppress_stderr: Suppress stderr output if True :param env: added env vars """ # noinspection PyUnresolvedReferences kw = dict(cwd=PROG_DIR, shell=shell, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=(open(os.devnull, 'w') if PY2 else subprocess.DEVNULL, subprocess.STDOUT)[not suppress_stderr]) if isinstance(env, dict): kw.update(env=dict(os.environ, **env)) kw.update(dict(encoding=SYS_ENCODING, text=True, bufsize=0)) if 'win32' == sys.platform: kw['creationflags'] = 0x08000000 # CREATE_NO_WINDOW (needed for py2exe) with Popen(cmd, **kw) as p: out, err = p.communicate() if out: out = out.strip() return out, err, p.returncode def md5_for_text(text): """ :param text: test :type text: AnyStr :return: :rtype: AnyStr or None """ result = None try: md5 = hashlib.md5() md5.update(decode_bytes(str(text))) raw_md5 = md5.hexdigest() result = raw_md5[17:] + raw_md5[9:17] + raw_md5[0:9] except (BaseException, Exception): pass return result def maybe_plural(subject=1): """ returns 's' or '' depending on numeric subject or length of subject :param subject: number or list or dict :type subject: int or list or dict :return: returns s or '' :rtype: AnyStr """ number = subject if not isinstance(subject, (list, dict)) else len(subject) return ('s', '')[1 == number] def time_to_int(dt): # type: (Union[datetime.time, None]) -> Optional[integer_types] """ converts datetime.time to integer (hour + minute only) :param dt: datetime.time obj :return: integer of hour + min """ if None is dt: return None try: return dt.hour * 100 + dt.minute except (BaseException, Exception): return 0 def int_to_time(d_int): # type: (Union[integer_types, None]) -> Optional[datetime.time] """ convert integer from dt_to_int back to datetime.time """ if None is d_int: return None if isinstance(d_int, integer_types): try: return datetime.time(*divmod(d_int, 100)) except (BaseException, Exception): pass return datetime.time(hour=0, minute=0) def indent_xml(elem, level=0): """ Does our pretty printing, makes Matt very happy """ i = '\n' + level * ' ' if len(elem): if not elem.text or not ('%s' % elem.text).strip(): elem.text = i + ' ' if not elem.tail or not elem.tail.strip(): elem.tail = i for elem in elem: indent_xml(elem, level + 1) if not elem.tail or not elem.tail.strip(): elem.tail = i else: # Strip out the newlines from text if elem.text: elem.text = ('%s' % elem.text).replace('\n', ' ') if level and (not elem.tail or not elem.tail.strip()): elem.tail = i def ast_eval(value, default=None): # type: (AnyStr, Any) -> Any """Convert string typed value into actual Python type and value :param value: string value to convert :param default: value to return if it cannot convert :return: converted type and value or default """ if not isinstance(value, string_types): return default if 'OrderedDict()' == value: value = dict() elif 'OrderedDict([(' == value[0:14]: try: list_of_tuples = ast.literal_eval(value[12:-1]) value = dict() for cur_tuple in list_of_tuples: value[cur_tuple[0]] = cur_tuple[1] except (BaseException, Exception): value = default elif '{' == value[0:1] and '}' == value[-1]: # this way avoids index out of range with (value = '' and [-1]) try: value = ast.literal_eval(value) except (BaseException, Exception): value = default else: value = default return value def download_file(url, filename, session=None, **kwargs): """ download given url to given filename :param url: url to download :type url: AnyStr :param filename: filename to save the data to :type filename: AnyStr :param session: optional requests session object :type session: requests.Session or None :param kwargs: :return: success of download :rtype: bool """ MEMCACHE.setdefault('cookies', {}) if None is get_url(url, session=session, savename=filename, url_solver=FLARESOLVERR_HOST, memcache_cookies=MEMCACHE['cookies'], **kwargs): remove_file_perm(filename) return False return True def calc_age(birthday, deathday=None, date=None): # type: (datetime.date, datetime.date, Optional[datetime.date]) -> Optional[int] """ returns age based on current date or given date :param birthday: birthdate :param deathday: deathdate :param date: """ if isinstance(birthday, datetime.date): today = (datetime.date.today(), date)[isinstance(date, datetime.date)] today = (today, deathday)[isinstance(deathday, datetime.date) and today > deathday] try: b_d = birthday.replace(year=today.year) # raised when birthdate is February 29 # and the current year is not a leap year except ValueError: b_d = birthday.replace(year=today.year, month=birthday.month + 1, day=1) if b_d > today: return today.year - birthday.year - 1 else: return today.year - birthday.year def convert_to_inch_faction_html(height): # type: (float) -> AnyStr """ returns html string in foot and inches including fractions :param height: height in cm """ total_inches = round(height / float(2.54), 2) foot, inches = divmod(total_inches, 12) _, fraction = '{0:.2f}'.format(total_inches).split('.') fraction = int(fraction) # fix rounding errors fraction = next((html_convert_fractions.get(fraction + round_error) or html_convert_fractions.get(fraction - round_error) for round_error in moves.xrange(0, 25) if fraction + round_error in html_convert_fractions or fraction - round_error in html_convert_fractions), '') if 1 == fraction: inches += 1 fraction = '' if 12 <= inches: foot += 1 inches = 0 inches = str(inches).split('.')[0] return '%s\' %s%s%s' % (int(foot), (inches, '')['0' == inches], fraction, ('', '"')['0' != inches or '' != fraction]) def spoken_height(height): # type: (float) -> AnyStr """ return text for spoken words of height :param height: height in cm """ return convert_to_inch_faction_html(height).replace('\'', ' foot').replace('"', '') def iterate_chunk(lst, n): # type: (List, integer_types) -> Iterator """ Yield successive n-sized chunks from lst. :param lst: list to split :param n: length of chunks """ for i in moves.range(0, len(lst), n): yield lst[i:i + n] def is_virtualenv(): # type: (...) -> bool """ :return: True if virtualenv Python environment is detected """ """Get base/real prefix, or `sys.prefix` if there is none.""" get_base_prefix_compat = getattr(sys, 'base_prefix', None) or getattr(sys, 'real_prefix', None) or sys.prefix return get_base_prefix_compat != sys.prefix def enforce_type(value, allowed_types, default): # type: (Any, Union[Type, Tuple[Type]], Any) -> Any """ enforces that value is given type(s) :param value: value to check :param allowed_types: type or tuple of types allowed :param default: value to return if other type """ if not isinstance(value, allowed_types): return default return value