endpoint.py

importdatetime
importjson
importos
fromrandomimportrandom
fromtimeimportsleep
fromtimeimporttime

importrequests
importshortuuid
fromsickleimportSickle, oaiexceptions
fromsickle.iteratorimportOAIItemIterator
fromsickle.modelsimportResumptionToken
fromsickle.oaiexceptionsimportNoRecordsMatch, BadArgument
fromsickle.responseimportOAIResponse
fromsqlalchemyimportor_

importpmh_record
fromappimportdb
fromappimportlogger
fromhttp_cacheimportrequest_ua_headers
fromrepositoryimportRepository
fromutilimportelapsed
fromutilimportsafe_commit


deflookup_endpoint_by_pmh_url(pmh_url_query=None):
endpoints=Endpoint.query.filter(Endpoint.pmh_url.ilike("%{}%".format(pmh_url_query))).all()
returnendpoints


classEndpoint(db.Model):
id=db.Column(db.Text, primary_key=True)
id_old=db.Column(db.Text)
repo_unique_id=db.Column(db.Text, db.ForeignKey(Repository.id))
pmh_url=db.Column(db.Text)
pmh_set=db.Column(db.Text)
last_harvest_started=db.Column(db.DateTime)
last_harvest_finished=db.Column(db.DateTime)
most_recent_year_harvested=db.Column(db.DateTime)
earliest_timestamp=db.Column(db.DateTime)
email=db.Column(db.Text) # to help us figure out what kind of repo it is
error=db.Column(db.Text)
repo_request_id=db.Column(db.Text)
harvest_identify_response=db.Column(db.Text)
harvest_test_recent_dates=db.Column(db.Text)
sample_pmh_record=db.Column(db.Text)
contacted=db.Column(db.DateTime)
contacted_text=db.Column(db.Text)
policy_promises_no_submitted=db.Column(db.Boolean)
policy_promises_no_submitted_evidence=db.Column(db.Text)
ready_to_run=db.Column(db.Boolean)
metadata_prefix=db.Column(db.Text)
retry_interval=db.Column(db.Interval)
retry_at=db.Column(db.DateTime)

meta=db.relationship(
'Repository',
lazy='subquery',
cascade="all",
backref=db.backref("endpoints", lazy="subquery")
 )

def__init__(self, **kwargs):
super(self.__class__, self).__init__(**kwargs)
ifnotself.id:
self.id=shortuuid.uuid()[0:20].lower()
ifnotself.metadata_prefix:
self.metadata_prefix='oai_dc'

@property
defrepo(self):
returnself.meta

defrun_diagnostics(self):
response=test_harvest_url(self.pmh_url, self.pmh_set)
self.harvest_identify_response=response["harvest_identify_response"]
# self.harvest_test_initial_dates = response["harvest_test_initial_dates"]
self.harvest_test_recent_dates=response["harvest_test_recent_dates"]
self.sample_pmh_record=response["sample_pmh_record"]

defharvest(self):
ifnotself.harvest_identify_responseornotself.harvest_test_recent_dates:
self.set_identify_and_initial_query()

today=datetime.datetime.utcnow().date()
tomorrow=today+datetime.timedelta(days=1)
yesterday=today-datetime.timedelta(days=1)

first= (self.most_recent_year_harvestedordatetime.datetime(2000, 1, 1)).date()
first=min(first, yesterday)

ifself.id_oldin ['citeseerx.ist.psu.edu/oai2', 'europepmc.org/oai.cgi']:
first_plus_delta=first+datetime.timedelta(days=1)
elifself.id_oldin ['www.ncbi.nlm.nih.gov/pmc/oai/oai.cgi', 'www.ncbi.nlm.nih.gov/pmc/oai/oai.cgi2']:
first_plus_delta=first+datetime.timedelta(days=7)
elifself.id=='4bd6f8f5107c0df6f48':
first_plus_delta=first+datetime.timedelta(days=1)
elifself.id=='0d27b133730393e00e1':
first_plus_delta=first+datetime.timedelta(days=1)
elifself.id=='jmpfmmfru5pzhy4lbrdm':
first_plus_delta=first+datetime.timedelta(days=1)
elifself.id_oldin ['export.arxiv.org/oai2']:
first_plus_delta=first+datetime.timedelta(days=1)
elif'osti.gov/oai'inself.pmh_url:
first_plus_delta=first+datetime.timedelta(days=1)
elif'share.osf.io'inself.pmh_url:
first_plus_delta=first+datetime.timedelta(days=1)
else:
first_plus_delta=first+datetime.timedelta(days=7)

last=min(first_plus_delta, tomorrow)

# now do the harvesting
self.call_pmh_endpoint(first=first, last=last)

# if success, update so we start at next point next time
base_retry_interval=datetime.timedelta(minutes=5)
ifself.error:
logger.info("error so not saving finished info: {}".format(self.error))
retry_interval=self.retry_intervalorbase_retry_interval
self.retry_at=datetime.datetime.utcnow() +retry_interval
self.retry_interval=retry_interval*2
self.last_harvest_started=None
else:
logger.info("success! saving info")
self.last_harvest_finished=datetime.datetime.utcnow().isoformat()
self.most_recent_year_harvested=min(yesterday, last)
self.last_harvest_started=None
self.retry_at=None
self.retry_interval=base_retry_interval

defget_pmh_record(self, record_id):
my_sickle=_get_my_sickle(self.pmh_url)
pmh_input_record=my_sickle.GetRecord(identifier=record_id, metadataPrefix=self.metadata_prefix)
my_pmh_record=pmh_record.PmhRecord()
my_pmh_record.populate(self.id, pmh_input_record, metadata_prefix=self.metadata_prefix)
my_pmh_record.repo_id=self.id_old# delete once endpoint_id is populated
returnmy_pmh_record

defset_identify_and_initial_query(self):
ifnotself.pmh_url:
self.harvest_identify_response="error, no pmh_url given"
return

my_sickle=None
try:
# set timeout quick... if it can't do this quickly, won't be good for harvesting
logger.debug("getting my_sickle for {}".format(self))
my_sickle=_get_my_sickle(self.pmh_url, timeout=10)
my_sickle.Identify()
self.harvest_identify_response="SUCCESS!"

exceptExceptionase:
logger.exception("in set_identify_and_initial_query")
self.error="error in calling identify: {} {}".format(e.__class__.__name__, str(e))
ifmy_sickle:
self.error+=" calling {}".format(my_sickle.get_http_response_url())

self.harvest_identify_response=self.error

self.sample_pmh_record=None

try:
sample_pmh_record=self.get_recent_pmh_record()
ifsample_pmh_record:
self.harvest_test_recent_dates="SUCCESS!"
self.sample_pmh_record=json.dumps(sample_pmh_record.metadata)
else:
self.harvest_test_recent_dates="error, no pmh_input_records returned"
exceptExceptionase:
self.error="error in get_recent_pmh_record: {} {}".format(e.__class__.__name__, str(e))
self.harvest_test_recent_dates=self.error

defget_recent_pmh_record(self, use_date_default_format=True):
last=datetime.datetime.utcnow()
first=last-datetime.timedelta(days=30)

args= {'metadataPrefix': self.metadata_prefix}

my_sickle=_get_my_sickle(self.pmh_url)
logger.info("connected to sickle with {}".format(self.pmh_url))

args['from'] =first.isoformat()[0:10]
ifnotuse_date_default_format:
args['from'] +="T00:00:00Z"

args["until"] =last.isoformat()[0:10]
ifnotuse_date_default_format:
args['until'] +="T00:00:00Z"

ifself.pmh_set:
args["set"] =self.pmh_set

logger.info("calling ListIdentifiers with {} {}".format(self.pmh_url, args))
try:
pmh_identifiers=my_sickle.ListIdentifiers(ignore_deleted=True, **args)
pmh_identifier=self.safe_get_next_record(pmh_identifiers)
ifpmh_identifier:
returnmy_sickle.GetRecord(identifier=pmh_identifier.identifier, metadataPrefix=self.metadata_prefix)
else:
returnNone
exceptNoRecordsMatch:
logger.info("no records with {} {}".format(self.pmh_url, args))
returnNone
exceptBadArgument:
ifuse_date_default_format:
returnself.get_recent_pmh_record(use_date_default_format=False)

defget_pmh_input_record(self, first, last, use_date_default_format=True):
args= {'metadataPrefix': self.metadata_prefix}
pmh_records= []
self.error=None

my_sickle=_get_my_sickle(self.pmh_url)
logger.info("connected to sickle with {}".format(self.pmh_url))

args['from'] =first.isoformat()[0:10]
ifnotuse_date_default_format:
args['from'] +="T00:00:00Z"

iflast:
args["until"] =last.isoformat()[0:10]
ifnotuse_date_default_format:
args['until'] +="T00:00:00Z"

ifself.pmh_set:
args["set"] =self.pmh_set

logger.info("calling ListRecords with {} {}".format(self.pmh_url, args))
try:
try:
pmh_records=my_sickle.ListRecords(ignore_deleted=True, **args)
pmh_input_record=self.safe_get_next_record(pmh_records)
exceptNoRecordsMatch:
logger.info("no records with {} {}".format(self.pmh_url, args))
pmh_input_record=None
exceptBadArgumentase:
ifuse_date_default_format:
returnself.get_pmh_input_record(first, last, use_date_default_format=False)
else:
raisee
exceptExceptionase:
logger.exception("error with {} {}".format(self.pmh_url, args))
pmh_input_record=None
self.error="error in get_pmh_input_record: {} {}".format(e.__class__.__name__, str(e))
ifmy_sickle:
self.error+=" calling {}".format(my_sickle.get_http_response_url())

returnpmh_input_record, pmh_records, self.error

defcall_pmh_endpoint(self,
first=None,
last=None,
chunk_size=50,
scrape=False):

start_time=time()
records_to_save= []
num_records_updated=0
loop_counter=0
self.error=None

 (pmh_input_record, pmh_records, error) =self.get_pmh_input_record(first, last)

iferror:
self.error="error in get_pmh_input_record: {}".format(error)
return

whilepmh_input_record:
loop_counter+=1
# create the record
my_pmh_record=pmh_record.PmhRecord()

# set its vars
my_pmh_record.repo_id=self.id_old# delete once endpoint_ids are all populated
my_pmh_record.rand=random()
my_pmh_record.populate(self.id, pmh_input_record, metadata_prefix=self.metadata_prefix)

ifis_complete(my_pmh_record):
my_pages=my_pmh_record.mint_pages(reset_scrape_date=True)
my_pmh_record.pages=my_pages
ifscrape:
formy_pageinmy_pages:
my_page.scrape_if_matches_pub()
records_to_save.append(my_pmh_record)
my_pmh_record.delete_old_record()
db.session.merge(my_pmh_record)
db.session.flush()
my_pmh_record.enqueue_representative_page()
else:
logger.info("pmh record is not complete")
# print my_pmh_record
pass

iflen(records_to_save) >=chunk_size:
num_records_updated+=len(records_to_save)
safe_commit(db)
records_to_save= []

ifloop_counter%100==0:
logger.info("iterated through 100 more items, loop_counter={} for {}".format(loop_counter, self.id))

pmh_input_record=self.safe_get_next_record(pmh_records)

# make sure to get the last ones
ifrecords_to_save:
num_records_updated+=len(records_to_save)
last_record=records_to_save[-1]
logger.info("saving {} last ones, last record saved: {} for {}, loop_counter={}".format(
len(records_to_save), last_record.id, self.id, loop_counter))
safe_commit(db)
else:
logger.info("finished loop, but no records to save, loop_counter={}".format(loop_counter))

logger.info("updated {} PMH records for endpoint_id={}, took {} seconds".format(
num_records_updated, self.id, elapsed(start_time, 2)))

defsafe_get_next_record(self, current_record, tries=3):
self.error=None
try:
next_record=next(current_record)
except (requests.exceptions.HTTPError, requests.exceptions.SSLError) ase:
iftries>0:
logger.info("requests exception! trying again {}".format(e))
returnself.safe_get_next_record(current_record, tries-1)
else:
logger.info("requests exception! skipping {}".format(e))
self.error="requests error in safe_get_next_record; try again"
returnNone
except (KeyboardInterrupt, SystemExit):
# done
returnNone
exceptStopIteration:
logger.info("stop iteration! stopping")
returnNone
exceptNoRecordsMatch:
logger.info("no records! stopping")
returnNone
exceptAttributeError:
logger.info("pmh records likely missing metadata section, skipping.")
returnNone
exceptExceptionase:
logger.exception("misc exception!: {} skipping".format(e))
self.error="error in safe_get_next_record"
returnNone
returnnext_record

defget_num_pmh_records(self):
frompmh_recordimportPmhRecord
num=db.session.query(PmhRecord.id).filter(PmhRecord.endpoint_id==self.id).count()
returnnum

defget_num_pages(self):
frompageimportPageNew
num=db.session.query(PageNew.id).filter(PageNew.endpoint_id==self.id).count()
returnnum

defget_num_open_with_dois(self):
frompageimportPageNew
num=db.session.query(PageNew.id).\
distinct(PageNew.normalized_title).\
filter(PageNew.endpoint_id==self.id).\
filter(PageNew.num_pub_matches.isnot(None), PageNew.num_pub_matches>=1).\
filter(or_(PageNew.scrape_pdf_url.isnot(None), PageNew.scrape_metadata_url.isnot(None))).\
count()
returnnum

defget_num_title_matching_dois(self):
frompageimportPageNew
num=db.session.query(PageNew.id).\
distinct(PageNew.normalized_title).\
filter(PageNew.endpoint_id==self.id).\
filter(PageNew.num_pub_matches.isnot(None), PageNew.num_pub_matches>=1).\
count()
returnnum

defget_open_pages(self, limit=10):
frompageimportPageNew
pages=db.session.query(PageNew).\
distinct(PageNew.normalized_title).\
filter(PageNew.endpoint_id==self.id).\
filter(PageNew.num_pub_matches.isnot(None), PageNew.num_pub_matches>=1).\
filter(or_(PageNew.scrape_pdf_url.isnot(None), PageNew.scrape_metadata_url.isnot(None))).\
limit(limit).all()
return [(p.id, p.url, p.normalized_title, p.pub.url, p.pub.unpaywall_api_url, p.scrape_version) forpinpages]

defget_closed_pages(self, limit=10):
frompageimportPageNew
pages=db.session.query(PageNew).\
distinct(PageNew.normalized_title).\
filter(PageNew.endpoint_id==self.id).\
filter(PageNew.num_pub_matches.isnot(None), PageNew.num_pub_matches>=1).\
filter(PageNew.scrape_updated.isnot(None), PageNew.scrape_pdf_url.is_(None), PageNew.scrape_metadata_url .is_(None)).\
limit(limit).all()
return [(p.id, p.url, p.normalized_title, p.pub.url, p.pub.unpaywall_api_url, p.scrape_updated) forpinpages]

defget_num_pages_still_processing(self):
frompageimportPageNew
num=db.session.query(PageNew.id).filter(PageNew.endpoint_id==self.id, PageNew.num_pub_matches.is_(None)).count()
returnnum

def__repr__(self):
return"<Endpoint ( {} ) {}>".format(self.id, self.pmh_url)

defto_dict(self):
response= {
"_endpoint_id": self.id,
"_pmh_url": self.pmh_url,
"num_pmh_records": self.get_num_pmh_records(),
"num_pages": self.get_num_pages(),
"num_open_with_dois": self.get_num_open_with_dois(),
"num_title_matching_dois": self.get_num_title_matching_dois(),
"num_pages_still_processing": self.get_num_pages_still_processing(),
"pages_open": "{}/debug/repo/{}/examples/open".format("http://localhost:5000", self.repo_unique_id), # self.get_open_pages(),
"pages_closed": "{}/debug/repo/{}/examples/closed".format("http://localhost:5000", self.repo_unique_id), # self.get_closed_pages(),
"metadata": {}
 }

ifself.meta:
response.update({
"metadata": {
"home_page": self.repo.home_page,
"institution_name": self.repo.institution_name,
"repository_name": self.repo.repository_name
 }
 })
returnresponse

defto_dict_status(self):
response= {
"results": {},
"metadata": {}
 }

forfieldin ["id", "repo_unique_id", "pmh_url", "email"]:
response[field] =getattr(self, field)

forfieldin ["harvest_identify_response", "harvest_test_recent_dates", "sample_pmh_record"]:
response["results"][field] =getattr(self, field)

ifself.meta:
forfieldin ["home_page", "institution_name", "repository_name"]:
response["metadata"][field] =getattr(self.meta, field)

returnresponse

defto_dict_repo_pulse(self):
return {
"metadata": {
"endpoint_id": self.id,
"repository_name": self.repo.repository_name,
"institution_name": self.repo.institution_name,
"pmh_url": self.pmh_url
 },
"status": {
"check0_identify_status": self.harvest_identify_response,
"check1_query_status": self.harvest_test_recent_dates,
"num_pmh_records": None,
"last_harvest": self.most_recent_year_harvested,
"num_pmh_records_matching_dois": None,
"num_pmh_records_matching_dois_with_fulltext": None
 },
"by_version_distinct_pmh_records_matching_dois": {}
 }


deftest_harvest_url(pmh_url, pmh_set=None):
response= {}
temp_endpoint=Endpoint()
temp_endpoint.pmh_url=pmh_url
temp_endpoint.pmh_set=pmh_set
temp_endpoint.set_identify_and_initial_query()
response["harvest_identify_response"] =temp_endpoint.harvest_identify_response
response["sample_pmh_record"] =temp_endpoint.sample_pmh_record
response["harvest_test_recent_dates"] =temp_endpoint.harvest_test_recent_dates

returnresponse


defis_complete(record):
ifnotrecord.pmh_id:
returnFalse
ifnotrecord.title:
returnFalse
ifnotrecord.urls:
returnFalse

ifrecord.oa=="0":
logger.info("record {} is closed access. skipping.".format(record["id"]))
returnFalse

returnTrue


classMyOAIItemIterator(OAIItemIterator):
def_get_resumption_token(self):
"""Extract and store the resumptionToken from the last response."""
resumption_token_element=self.oai_response.xml.find(
'.//'+self.sickle.oai_namespace+'resumptionToken')
ifresumption_token_elementisNone:
returnNone
token=resumption_token_element.text
cursor=resumption_token_element.attrib.get('cursor', None)
complete_list_size=resumption_token_element.attrib.get(
'completeListSize', None)
expiration_date=resumption_token_element.attrib.get(
'expirationDate', None)
resumption_token=ResumptionToken(
token=token, cursor=cursor,
complete_list_size=complete_list_size,
expiration_date=expiration_date
 )
returnresumption_token

defget_complete_list_size(self):
"""Extract and store the resumptionToken from the last response."""
resumption_token_element=self.oai_response.xml.find(
'.//'+self.sickle.oai_namespace+'resumptionToken')
ifresumption_token_elementisNone:
returnNone
complete_list_size=resumption_token_element.attrib.get(
'completeListSize', None)
ifcomplete_list_size:
returnint(complete_list_size)
returncomplete_list_size


classOSTIOAIItemIterator(MyOAIItemIterator):
def_next_response(self):
"""Get the next response from the OAI server.

 Copy-pasted from OAIItemIterator._next_response but adds metadataPrefix to params.
 """
params=self.params
ifself.resumption_token:
params= {
'resumptionToken': self.resumption_token.token,
'verb': self.verb,
'metadataPrefix': params.get('metadataPrefix')
 }
self.oai_response=self.sickle.harvest(**params)
error=self.oai_response.xml.find(
'.//'+self.sickle.oai_namespace+'error')
iferrorisnotNone:
code=error.attrib.get('code', 'UNKNOWN')
description=error.textor''
try:
raisegetattr(
oaiexceptions, code[0].upper() +code[1:])(description)
exceptAttributeError:
raiseoaiexceptions.OAIError(description)
self.resumption_token=self._get_resumption_token()
self._items=self.oai_response.xml.iterfind(
'.//'+self.sickle.oai_namespace+self.element)


def_get_my_sickle(repo_pmh_url, timeout=120):
ifnotrepo_pmh_url:
returnNone

proxy_url=None

ifany(fragmentinrepo_pmh_urlforfragmentin [
"citeseerx",
"oai/61RMIT_INST", # RMIT University Library, y8aqthhhndmaebyspgmf
"oai/44SUR_INST", # Surrey, not added yet
 ]):
proxy_url=os.getenv("STATIC_IP_PROXY")
elifany(fragmentinrepo_pmh_urlforfragmentin ["pure.coventry.ac.uk"]):
proxy_url=os.getenv("VERY_STATIC_IP_PROXY")

ifproxy_url:
proxies= {"https": proxy_url, "http": proxy_url}
else:
proxies= {}

iterator=OSTIOAIItemIteratorif'osti.gov/oai'inrepo_pmh_urlelseMyOAIItemIterator
sickle=EuropePMCSickleif'europepmc.org'inrepo_pmh_urlelseMySickle
my_sickle=sickle(repo_pmh_url, proxies=proxies, timeout=timeout, iterator=iterator)
returnmy_sickle


# subclass so we can customize the number of retry seconds
classMySickle(Sickle):
RETRY_SECONDS=120

def__init__(self, *args, **kwargs):
self.http_response_url=None
super(MySickle, self).__init__(*args, **kwargs)

defget_http_response_url(self):
ifhasattr(self, "http_response_url"):
returnself.http_response_url
returnNone

def_massage_http_response(self, http_response):
returnhttp_response

defharvest(self, **kwargs): # pragma: no cover
"""Make HTTP requests to the OAI server.
 :param kwargs: OAI HTTP parameters.
 :rtype: :class:`sickle.OAIResponse`
 """
start_time=time()
verify= (
notself.endpoint.startswith('https://rcin.org.pl')
andnotself.endpoint.startswith('https://archive.nyu.edu')
 )
headers=request_ua_headers()
if"japanlinkcenter.org"inself.endpoint:
logger.info("using JapanLinkCenter API key")
headers["X-JaLC-Api-Key"] =f"Unpaywall00@{os.getenv('JALC_API_KEY')}"

for_inrange(self.max_retries):
ifself.http_method=='GET':
payload_str="&".join("{}={}".format(k, v) fork, vinlist(kwargs.items()))
url_without_encoding="{}?{}".format(self.endpoint, payload_str)
http_response=requests.get(url_without_encoding, headers=headers, verify=verify,
**self.request_args)

self.http_response_url=http_response.url
else:
http_response=requests.post(self.endpoint, headers=headers, data=kwargs,
**self.request_args)
self.http_response_url=http_response.url
ifhttp_response.status_code==503:
retry_after=int(http_response.headers.get('Retry-After', self.RETRY_SECONDS))
logger.info("HTTP 503! Retrying after %d seconds..."%retry_after)
sleep(retry_after)
else:
logger.info("took {} seconds to call pmh url: {}".format(elapsed(start_time), http_response.url))

http_response=self._massage_http_response(http_response)

http_response.raise_for_status()
ifself.encoding:
http_response.encoding=self.encoding
returnOAIResponse(http_response, params=kwargs)


classEuropePMCSickle(MySickle):
def_massage_http_response(self, http_response):
# server returns a 404 with NoRecordsMatch responses
# treat this as a successful http request and handle the OAI-PMH error further up the stack
ifhttp_response.status_code==404:
http_response.status_code=200

returnhttp_response