http_cache.py

frombase64importb64decode
importos
importre
fromdataclassesimportdataclass
fromtimeimportsleep
fromtimeimporttime
fromtypingimportOptional
importjson

importcertifi
importrequests
importtenacity

fromappimportlogger
fromtenacityimportretry, stop_after_attempt, wait_exponential, \
retry_if_result
importrequests.exceptions
fromutilimportelapsed
fromzyte_sessionimportget_matching_policies

MAX_PAYLOAD_SIZE_BYTES=1000*1000*10# 10mb

os.environ['NO_PROXY'] ='impactstory.crawlera.com'


@dataclass
classResponseObject:
content: bytes
headers: dict
status_code: int
url: str
cookies: Optional[str] =None

def__post_init__(self):
self.headers= {header['name']: header['value'] forheaderin
self.headers}

deftext_small(self):
returnself.content

deftext_big(self):
returnself.content

defcontent_big(self):
returnself.content

def__enter__(self):
returnself

def__exit__(self, exc_type, exc_value, traceback):
pass

defraise_for_status(self):
ifself.status_code>=400:
raiserequests.HTTPError(
f'Bad status code for URL {self.url}: {self.status_code}')


def_create_cert_bundle():
crt_file='data/custom-certs.crt'

withopen(crt_file, 'w') ascombined_certs:
forsourcein [certifi.where(), 'data/crawlera-ca.crt']:
withopen(source, 'r') ass:
forlineins:
combined_certs.write(line)

returncrt_file


_cert_bundle=_create_cert_bundle()


defis_response_too_large(r):
ifnot"Content-Length"inr.headers:
# logger.info(u"can't tell if page is too large, no Content-Length header {}".format(r.url))
returnFalse

content_length=r.headers["Content-Length"]
# if is bigger than 25 MB, don't keep it don't parse it, act like we couldn't get it
# if doing 100 in parallel, this would be 100MB, which fits within 512MB dyno limit
ifint(content_length) >= (25*1000*1000):
logger.info("Content Too Large on GET on {url}".format(url=r.url))
returnTrue
returnFalse


defget_session_id():
# set up proxy
session_id=None

whilenotsession_id:
crawlera_username=os.getenv("CRAWLERA_KEY")
r=requests.post("http://impactstory.crawlera.com:8010/sessions",
auth=(crawlera_username, 'DUMMY'),
proxies={'http': None, 'https': None})
ifr.status_code==200:
session_id=r.headers["X-Crawlera-Session"]
else:
# bad call. sleep and try again.
sleep(1)

returnsession_id


defchooser_redirect(r):
"""
 Check if the response contains a "Chooser" title and extract redirect links.

 This function works with ResponseObject instances returned by Zyte API.
 """
# Get the content as text
try:
content=r.text_small() ifhasattr(r, 'text_small') elser.content

# If content is bytes, decode it
ifisinstance(content, bytes):
content=content.decode('utf-8', 'ignore')

# Look for the Chooser title
if'<title>Chooser</title>'incontent:
# Find resource links
links=re.findall(
r'<div class="resource-line">.*?<a\s+href="(.*?)".*?</div>',
content, re.DOTALL
 )

iflinks:
logger.info(f'Found chooser redirect to: {links[0]}')
returnlinks[0]
exceptExceptionase:
logger.error(f"Error in chooser_redirect: {str(e)}")

returnNone


defrequest_ua_headers():
return {
'User-Agent': 'Unpaywall (http://unpaywall.org/; mailto:team@impactstory.org)',
'From': 'team@impactstory.org',
 }


defbefore_retry(retry_state):
redirected_url=retry_state.outcome.result().url
logger.info(f"retrying due to {retry_state.outcome.result().status_code}")
retry_state.kwargs['redirected_url'] =redirected_url
retry_state.kwargs['attempt_n'] =retry_state.attempt_number


defis_retry_status(response):
returnresponse.status_codein {429, 500, 502, 503, 504, 520, 403}


@retry(stop=stop_after_attempt(2),
wait=wait_exponential(multiplier=1, min=4, max=10),
retry=retry_if_result(is_retry_status),
before_sleep=before_retry)
defcall_requests_get(url=None, headers=None, read_timeout=300, connect_timeout=300,
stream=False, publisher=None, session_id=None, ask_slowly=False,
verify=False, cookies=None, redirected_url=None, attempt_n=0):
ifredirected_url:
url=redirected_url

# default parameters setup
# default parameters setup
default_params= {
"url": url,
"httpResponseBody": True,
"httpResponseHeaders": True,
 }

# get policies if they exist
matching_policies=get_matching_policies(url)
ifmatching_policies:
policy=matching_policies[min(attempt_n, len(matching_policies) -1)]

ifpolicy.profile=='api'andpolicy.paramsisnotNone:
logger.info(f'Using Zyte API profile for url: {url}')
zyte_params=policy.params
else:
logger.info(f'Using Zyte profile with proxy mode for url: {url}')
zyte_params=default_params
else:
logger.info(f'No matching policies - using default Zyte profile for url: {url}')
zyte_params=default_params

ifzyte_paramsisNone:
logger.warning(f"zyte_params was None for url: {url}, using default params")
zyte_params=default_params.copy()

# set URL in parameters
zyte_params["url"] =url

# make the API call
zyte_api_response=call_with_zyte_api(url, zyte_params)
good_status_code=zyte_api_response.get('statusCode')
bad_status_code=zyte_api_response.get('status')

ifgood_status_codeisnotNoneandgood_status_code<400:
logger.info(f"zyte api good status code for {url}: {good_status_code}")

headers=zyte_api_response.get('httpResponseHeaders', [])

content=None
if'httpResponseBody'inzyte_api_response:
content=b64decode(zyte_api_response.get('httpResponseBody'))
elif'browserHtml'inzyte_api_response:
content=zyte_api_response.get('browserHtml').encode()
else:
content=b''

# create response object
r=ResponseObject(
content=content,
headers=headers,
status_code=good_status_code,
url=zyte_api_response.get('url', url),
 )

content_type=r.headers.get("Content-Type", "").lower()
is_pdf="application/pdf"incontent_type

# try to detect PDF signature in content
ifnotis_pdfandisinstance(r.content, bytes) andlen(r.content) >4:
is_pdf=r.content.startswith(b'%PDF-')

ifnotis_pdfandisinstance(r.content, bytes):
try:
r.content=r.content.decode('utf-8', 'ignore')
except (UnicodeDecodeError, AttributeError):
# Keep as binary if decoding fails
pass

# check for doi.org chooser redirects
redirect_url=chooser_redirect(r)
ifredirect_url:
logger.info(f"Following chooser redirect to {redirect_url}")

# recursively follow the redirect with a new request
returncall_requests_get(
url=redirect_url,
headers=headers,
read_timeout=read_timeout,
connect_timeout=connect_timeout,
stream=stream,
publisher=publisher,
session_id=session_id,
ask_slowly=ask_slowly,
verify=verify,
cookies=cookies,
attempt_n=attempt_n
 )

returnr
else:
# Create a response for error cases
r=ResponseObject(
content='',
headers=[],
status_code=bad_status_code,
url=url,
 )
logger.info(f"zyte api bad status code for {url}: {bad_status_code}")
returnr


defhttp_get(url,
headers=None,
read_timeout=60,
connect_timeout=60,
stream=False,
cache_enabled=False,
allow_redirects=True,
publisher=None,
session_id=None,
ask_slowly=False,
verify=False,
cookies=None):
headers=headersor {}

start_time=time()

# reset
os.environ["HTTP_PROXY"] =""

try:
logger.info("LIVE GET on {}".format(url))
exceptUnicodeDecodeError:
logger.info("LIVE GET on an url that throws UnicodeDecodeError")

try:
r=call_requests_get(url,
headers=headers,
read_timeout=read_timeout,
connect_timeout=connect_timeout,
stream=stream,
publisher=publisher,
session_id=session_id,
ask_slowly=ask_slowly,
verify=verify,
cookies=cookies)
excepttenacity.RetryError:
logger.info(f"tried too many times for {url}")
raise
logger.info("finished http_get for {} in {} seconds".format(url, elapsed(
start_time, 2)))
returnr


defcall_with_zyte_api(url, params=None):
zyte_api_url="https://api.zyte.com/v1/extract"
zyte_api_key=os.getenv("ZYTE_API_KEY")
default_params= {
"url": url,
"httpResponseHeaders": True,
"httpResponseBody": True,
"requestHeaders": {"referer": "https://www.google.com/"},
 }
ifnotparams:
params=default_params
params['url'] =url
os.environ["HTTP_PROXY"] =''
os.environ["HTTPS_PROXY"] =''

logger.info(f"calling zyte api for {url}")
if"wiley.com"inurl:
# get cookies
cookies_response=requests.post(zyte_api_url, auth=(zyte_api_key, ''),
json={
"url": url,
"browserHtml": True,
"javascript": True,
"experimental": {
"responseCookies": True
 }
 }, verify=False)
cookies_response=json.loads(cookies_response.text)
cookies=cookies_response.get("experimental", {}).get(
"responseCookies", {})

# use cookies to get valid response
ifcookies:
response=requests.post(zyte_api_url, auth=(zyte_api_key, ''),
json={
"url": url,
"httpResponseHeaders": True,
"httpResponseBody": True,
"experimental": {
"requestCookies": cookies
 }
 }, verify=False)
else:
response=requests.post(zyte_api_url, auth=(zyte_api_key, ''),
json={
"url": url,
"httpResponseHeaders": True,
"httpResponseBody": True,
"requestHeaders": {
"referer": "https://www.google.com/"},
 }, verify=False)
else:
response=requests.post(zyte_api_url, auth=(zyte_api_key, ''),
json=params, verify=False)
returnresponse.json()


defget_cookies_with_zyte_api(url):
zyte_api_url="https://api.zyte.com/v1/extract"
zyte_api_key=os.getenv("ZYTE_API_KEY")
cookies_response=requests.post(zyte_api_url, auth=(zyte_api_key, ''),
json={
"url": url,
"browserHtml": True,
"javascript": True,
"experimental": {
"responseCookies": True
 }
 })
cookies_response=json.loads(cookies_response.text)
cookies=cookies_response.get("experimental", {}).get("responseCookies",
 {})
returncookies


if__name__=='__main__':
r=http_get('https://doi.org/10.1002/jum.15761')
print(r.status_code)