Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def on_task_start(self, task, config):
try:
import cloudscraper
except ImportError as e:
log.debug('Error importing cloudscraper: %s' % e)
raise plugin.DependencyError(
'cfscraper', 'cloudscraper', 'cloudscraper module required. ImportError: %s' % e
)
class CFScrapeWrapper(Session, cloudscraper.CloudScraper):
"""
This class allows the FlexGet session to inherit from CloudScraper instead of the requests.Session directly.
"""
if config is True:
task.requests.headers = OrderedDict(
[
('User-Agent', task.requests.headers['User-Agent']),
('Accept', 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8'),
('Accept-Language', 'en-US,en;q=0.5'),
('Accept-Encoding', 'gzip, deflate'),
('Connection', 'close'),
('Upgrade-Insecure-Requests', '1'),
]
)
task.requests = CFScrapeWrapper.create_scraper(task.requests)
def _request(self, method, url, *args, **kwargs):
ourSuper = super(CloudScraper, self)
resp = ourSuper.request(method, url, *args, **kwargs)
if resp.headers.get('Content-Encoding') == 'br':
if self.allow_brotli and resp._content:
resp._content = brotli.decompress(resp.content)
else:
logging.warning('Brotli content detected, But option is disabled, we will not continue.')
return resp
# Debug request
if self.debug:
self.debugRequest(resp)
# Check if Cloudflare anti-bot is on
try:
if self.isChallengeRequest(resp):
kwargs['timeout'] = self.timeout
return super(TimeoutSession, self).request(method, url, *args, **kwargs)
class CertifiSession(TimeoutSession):
def __init__(self):
super(CertifiSession, self).__init__()
self.verify = pem_file
class NeedsCaptchaException(Exception):
pass
class CFSession(CloudScraper):
def __init__(self, *args, **kwargs):
super(CFSession, self).__init__(*args, **kwargs)
self.debug = os.environ.get("CF_DEBUG", False)
def _request(self, method, url, *args, **kwargs):
ourSuper = super(CloudScraper, self)
resp = ourSuper.request(method, url, *args, **kwargs)
if resp.headers.get('Content-Encoding') == 'br':
if self.allow_brotli and resp._content:
resp._content = brotli.decompress(resp.content)
else:
logging.warning('Brotli content detected, But option is disabled, we will not continue.')
return resp
# Debug request