queue_green_oa_scrape.py

importargparse
importconcurrent.futures
importlogging
importos
importpickle
fromdatetimeimportdatetime, timedelta
frommultiprocessingimportcurrent_process
fromtimeimportsleep
fromtimeimporttime
fromurllib.parseimporturlparse

importredis
fromredisimportWatchError
fromsqlalchemyimportorm, text
fromsqlalchemy.excimportIntegrityError
fromsqlalchemy.ormimportmake_transient

fromappimportdb
fromappimportlogger
fromoa_pageimportpublisher_equivalent_endpoint_id
frompageimportPageNew
fromqueue_mainimportDbQueue
fromrecordthresher.recordimportRecordthresherParentRecord
fromrecordthresher.record_makerimportPmhRecordMaker
fromutilimportelapsed
fromutilimportsafe_commit

frompubimportPub# magic
importendpoint# magic
importpmh_record# magic

def_procs_per_worker():
returnint(os.getenv('GREEN_SCRAPE_PROCS_PER_WORKER', 10))


def_redis_max_connections():
return2


_redis_client=None
_redis_init=False


defget_redis_client():
global_redis_client, _redis_init

ifnot_redis_init:
try:
_redis_client=redis.from_url(os.environ.get("REDIS_URL"), max_connections=1)
exceptExceptionase:
logger.exception(f'failed creating redis client: {e}')

_redis_init=True

return_redis_client


defscrape_pages(pages):
forpageinpages:
make_transient(page)

# free up the connection while doing net IO
db.session.close()
db.engine.dispose()

withconcurrent.futures.ProcessPoolExecutor(max_workers=_procs_per_worker()) aspool:
map_results=pool.map(scrape_page, pages, chunksize=1)
scraped_pages= [pforpinmap_resultsifp]

logger.info('finished scraping all pages')

logger.info('preparing update records')
extant_page_ids= [
row[0] forrowin
db.session.query(PageNew.id).filter(PageNew.id.in_(
 [p.idforpinscraped_pages]
 )).all()
 ]

scraped_pages= [db.session.merge(p) forpinscraped_pagesifp.idinextant_page_ids]

forscraped_pageinscraped_pages:
scraped_page.save_first_version_availability()

returnscraped_pages


defscrape_page(page):
withconcurrent.futures.ProcessPoolExecutor(max_workers=1) aspool:
worker=current_process().name
try:
returnpool.submit(scrape_page_worker, page).result(timeout=300)
exceptconcurrent.futures.TimeoutErrorase:
logger.error(f'{worker} timed out')
forpid, processinpool._processes.items():
process.terminate()
pool.shutdown()
returnNone
except (KeyboardInterrupt, SystemExit):
pass
exceptExceptionase:
logger.exception(f'{worker} exception scraping page {page.id}')
returnNone


defscrape_page_worker(page):
worker=current_process().name
site_key_stem=redis_key(page, '')

logger.info('{} started scraping page {} {} {}'.format(worker, page.id, site_key_stem, page))

total_wait_seconds=0
wait_seconds=5
whiletotal_wait_seconds<60:
ifbegin_rate_limit(page):
page.scrape()
end_rate_limit(page)
logger.info('{} finished scraping page {} {} {}'.format(worker, page.id, site_key_stem, page))
returnpage
else:
logger.info('{} not ready to scrape page {} {} {}, waiting'.format(worker, page.id, site_key_stem, page))
sleep(wait_seconds)
total_wait_seconds+=wait_seconds

logger.info('{} done waiting to scrape page {} {} {}, giving up'.format(worker, page.id, site_key_stem, page))

returnNone


defunpickle(v):
returnpickle.loads(v) ifvelseNone


defredis_key(page, scrape_property):
domain=urlparse(page.url).netloc
return'green-scrape-p3:{}:{}:{}'.format(page.endpoint_id, domain, scrape_property)


defscrape_interval_seconds(page):
hostname=urlparse(page.url).hostname

one_sec_hosts= [
'citeseerx.ist.psu.edu',
'www.ncbi.nlm.nih.gov',
'pt.cision.com',
'doaj.org',
'hal.archives-ouvertes.fr',
'figshare.com',
'arxiv.org',
'europepmc.org',
'bibliotheques-specialisees.paris.fr',
'nbn-resolving.de',
'osti.gov',
'zenodo.org',
'kuleuven.be',
'edoc.hu-berlin.de',
'rug.nl',
 ]

forhostinone_sec_hosts:
ifhostnameandhostname.endswith(host):
return1

return10


defbegin_rate_limit(page, interval_seconds=None):
redis_client=get_redis_client()

ifpage.endpoint_id==publisher_equivalent_endpoint_id:
returnTrue

interval_seconds=interval_secondsorscrape_interval_seconds(page)

started_key=redis_key(page, 'started')
finished_key=redis_key(page, 'finished')

withredis_client.pipeline() aspipe:
try:
pipe.watch(started_key)
pipe.watch(finished_key)

scrape_started=unpickle(pipe.get(started_key))

ifscrape_startedandscrape_started>=datetime.utcnow() -timedelta(seconds=interval_seconds):
returnFalse

pipe.multi()
pipe.set(started_key, pickle.dumps(datetime.utcnow()))
pipe.set(finished_key, pickle.dumps(None))
pipe.execute()
returnTrue
exceptWatchError:
returnFalse


defend_rate_limit(page):
redis_client=get_redis_client()
redis_client.set(redis_key(page, 'started'), pickle.dumps(None))
redis_client.set(redis_key(page, 'finished'), pickle.dumps(datetime.utcnow()))


defmerge_and_commit_objects(objects, retry=2):
try:
logger.info('starting merge')
merge_start_time=time()
 [db.session.merge(o) foroinobjects]
logger.info("merge took {} seconds".format(elapsed(merge_start_time, 2)))

logger.info('starting commit')
commit_start_time=time()
db.session.commit()
logger.info("commit took {} seconds".format(elapsed(commit_start_time, 2)))
exceptIntegrityErrorase:
logger.exception(f'integrity error merging objects: {e}')
db.session.rollback()

ifretry>0:
logger.info('retrying merge_and_commit_objects')
merge_and_commit_objects(objects, retry=retry-1)
else:
logger.error('giving up on merge_and_commit_objects')
except (KeyboardInterrupt, SystemExit):
raise
exceptExceptionase:
db.session.rollback()
logger.exception(f'error merging objects: {e}')


classDbQueueGreenOAScrape(DbQueue):
deftable_name(self, job_type):
return'page_green_scrape_queue'

defprocess_name(self, job_type):
return'run_green_oa_scrape'

defworker_run(self, **kwargs):
run_class=PageNew

single_id=kwargs.get("id", None)
chunk_size=kwargs.get("chunk", 100)
limit=kwargs.get("limit", None)
scrape_publisher=kwargs.get("scrape_publisher", False)

iflimitisNone:
limit=float("inf")

ifsingle_id:
page=run_class.query.filter(run_class.id==single_id).first()
page.scrape()
page.save_first_version_availability()
db.session.merge(page)

safe_commit(db) orlogger.info("COMMIT fail")

ifrecordthresher_record:=PmhRecordMaker.make_record(page.pmh_record):
db.session.merge(recordthresher_record)
secondary_records=PmhRecordMaker.make_secondary_repository_responses(recordthresher_record)
forsecondary_recordinsecondary_records:
db.session.merge(secondary_record)
db.session.merge(
RecordthresherParentRecord(
record_id=secondary_record.id,
parent_record_id=recordthresher_record.id
 )
 )

safe_commit(db) orlogger.info("COMMIT fail")
else:
index=0
num_updated=0
start_time=time()

whilenum_updated<limit:
new_loop_start_time=time()

objects=self.fetch_queue_chunk(chunk_size, scrape_publisher)

ifnotobjects:
logger.info('no queued pages ready. waiting...')
sleep(5)
continue

scraped_pages=scrape_pages(objects)

scraped_ids= [p.idforpinscraped_pages]
unscraped_ids= [obj.idforobjinobjectsifobj.idnotinscraped_ids]

logger.info('scraped {} pages and returned {} to the queue'.format(
len(scraped_ids), len(unscraped_ids)
 ))

scraped_batch_text='''
 update {queue_table}
 set finished = now(), started=null
 where id = any(:ids)'''.format(queue_table=self.table_name(None))

unscraped_batch_text='''
 update {queue_table}
 set started=null
 where id = any(:ids)'''.format(queue_table=self.table_name(None))

scraped_batch_command=text(scraped_batch_text).bindparams(
ids=scraped_ids)

unscraped_batch_command=text(unscraped_batch_text).bindparams(
ids=unscraped_ids)

db.session.execute(scraped_batch_command)
db.session.execute(unscraped_batch_command)

commit_start_time=time()
safe_commit(db) orlogger.info("COMMIT fail")
logger.info("commit took {} seconds".format(elapsed(commit_start_time, 2)))

logger.info('making recordthresher records')

recordthresher_records= [PmhRecordMaker.make_record(p.pmh_record) forpinscraped_pages]

distinct_records= {}
forrecordthresher_recordinrecordthresher_records:
ifrecordthresher_record:
distinct_records[recordthresher_record.id] =recordthresher_record

ifdistinct_records:
logger.info('saving recordthresher records')
merge_and_commit_objects(distinct_records.values())

logger.info('making secondary repository records')
secondary_records= {}
parent_relationships= {}

forrindistinct_records.values():
r_secondary_records=PmhRecordMaker.make_secondary_repository_responses(r)
forr_secondary_recordinr_secondary_records:
secondary_records[r_secondary_record.id] =r_secondary_record
parent_relationships[r_secondary_record.id] =RecordthresherParentRecord(
record_id=r_secondary_record.id,
parent_record_id=r.id
 )

ifsecondary_records:
logger.info('saving secondary repository records')
merge_and_commit_objects(secondary_records.values())

ifparent_relationships:
logger.info('saving parent_relationships')
merge_and_commit_objects(parent_relationships.values())

index+=1
num_updated+=chunk_size
self.print_update(new_loop_start_time, len(scraped_ids), limit, start_time, index)

deffetch_queue_chunk(self, chunk_size, scrape_publisher):
logger.info("looking for new jobs")

endpoint_filter="and qt.endpoint_id {} '{}'".format(
'='ifscrape_publisherelse'is distinct from',
publisher_equivalent_endpoint_id
 )

text_query_pattern="""
 with update_chunk as (
 select
 lru_by_endpoint.id
 from
 endpoint e
 cross join lateral (
 select qt.*
 from
 {queue_table} qt
 join page_new p using (id)
 where
 qt.endpoint_id = e.id
 and qt.started is null
 and e.green_scrape
 {endpoint_filter}
 limit {per_endpoint_limit}
 for update of qt skip locked
 ) lru_by_endpoint
 where
 finished is null
 order by lru_by_endpoint.rand
 limit {chunk_size}
 )
 update {queue_table} queue_rows_to_update
 set started=now()
 from update_chunk
 where update_chunk.id = queue_rows_to_update.id
 returning update_chunk.id;
 """

text_query=text_query_pattern.format(
chunk_size=chunk_size,
queue_table=self.table_name(None),
endpoint_filter=endpoint_filter,
per_endpoint_limit=chunk_sizeifscrape_publisherelse10
 )

job_time=time()
row_list=db.engine.execute(text(text_query).execution_options(autocommit=True)).fetchall()
object_ids= [row[0] forrowinrow_list]
logger.info("got {} ids, took {} seconds".format(len(object_ids), elapsed(job_time)))

job_time=time()
q=db.session.query(PageNew).options(
orm.undefer('*')
 ).filter(PageNew.id.in_(object_ids))

objects=q.all()
logger.info("got page_new objects in {} seconds".format(elapsed(job_time)))

returnobjects


if__name__=="__main__":
ifos.getenv('OADOI_LOG_SQL'):
logging.getLogger('sqlalchemy.engine').setLevel(logging.INFO)
db.session.configure()

parser=argparse.ArgumentParser(description="Run stuff.")
parser.add_argument('--id', nargs="?", type=str, help="id of the one thing you want to update (case sensitive)")
parser.add_argument('--doi', nargs="?", type=str, help="id of the one thing you want to update (case insensitive)")

parser.add_argument('--reset', default=False, action='store_true', help="do you want to just reset?")
parser.add_argument('--run', default=False, action='store_true', help="to run the queue")
parser.add_argument('--status', default=False, action='store_true', help="to logger.info(the status")
parser.add_argument('--dynos', default=None, type=int, help="scale to this many dynos")
parser.add_argument('--logs', default=False, action='store_true', help="logger.info(out logs")
parser.add_argument('--monitor', default=False, action='store_true', help="monitor till done, then turn off dynos")
parser.add_argument('--kick', default=False, action='store_true', help="put started but unfinished dois back to unstarted so they are retried")
parser.add_argument('--limit', "-l", nargs="?", type=int, help="how many jobs to do")
parser.add_argument('--chunk', "-ch", nargs="?", default=100, type=int, help="how many to take off db at once")

parser.add_argument('--scrape-publisher', default=False, action='store_true', help="scrape publisher-equivalent pages")

parsed_args=parser.parse_args()

job_type="normal"# should be an object attribute
my_queue=DbQueueGreenOAScrape()
my_queue.parsed_vars=vars(parsed_args)
my_queue.run_right_thing(parsed_args, job_type)