Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
"""Tests for Crossref.journals"""
import os
import vcr
from nose.tools import *
from habanero import exceptions
from habanero import Crossref
cr = Crossref()
@vcr.use_cassette('test/vcr_cassettes/journals.yaml')
def test_journals():
"journals - basic test"
res = cr.journals(limit = 1)
assert dict == res.__class__
assert dict == res['message'].__class__
assert 1 == res['message']['items-per-page']
@vcr.use_cassette('test/vcr_cassettes/journals_query.yaml')
def test_journals_query():
"journals - param: query"
res = cr.journals(query = "ecology", limit = 2)
assert dict == res.__class__
assert 2 == res['message']['items-per-page']
assert 'journal-list' == res['message-type']
"""Tests for user agent strings via the ua_string setting"""
import os
import vcr
import yaml
from nose.tools import *
from habanero import Crossref
cr_with_ua = Crossref(ua_string = "foo bar")
cr_without_ua = Crossref()
cr_with_bad_ua = Crossref(ua_string = 5)
vcr_path = 'test/vcr_cassettes/setting_ua_string.yaml'
@vcr.use_cassette(vcr_path)
def test_ua_string():
"settings (ua_string) - with ua string, works"
res = cr_with_ua.works(ids = '10.1371/journal.pone.0033693')
x = open(vcr_path, "r").read()
xy = yaml.safe_load(x)
heads = xy['interactions'][0]['request']['headers']
assert 'foo bar' in heads['User-Agent'][0]
assert 'foo bar' in heads['X-USER-AGENT'][0]
vcr_noua_path = 'test/vcr_cassettes/setting_no_ua_string.yaml'
"""Tests for user agent strings via the ua_string setting"""
import os
import vcr
import yaml
from nose.tools import *
from habanero import Crossref
cr_with_ua = Crossref(ua_string = "foo bar")
cr_without_ua = Crossref()
cr_with_bad_ua = Crossref(ua_string = 5)
vcr_path = 'test/vcr_cassettes/setting_ua_string.yaml'
@vcr.use_cassette(vcr_path)
def test_ua_string():
"settings (ua_string) - with ua string, works"
res = cr_with_ua.works(ids = '10.1371/journal.pone.0033693')
x = open(vcr_path, "r").read()
xy = yaml.safe_load(x)
heads = xy['interactions'][0]['request']['headers']
assert 'foo bar' in heads['User-Agent'][0]
assert 'foo bar' in heads['X-USER-AGENT'][0]
vcr_noua_path = 'test/vcr_cassettes/setting_no_ua_string.yaml'
@vcr.use_cassette(vcr_noua_path)
if __name__=='__main__':
doi='10.1126/science.169.3946.635'
doi='10.1175/1520-0477(2001)082<1377:IOGPPT>2.3.CO;2k'
doi='10.1029/2002JD002499'
doi='10.1175/1525-7541(2003)004<1147:tvgpcp>2.0.co;2'
aa=cn.content_negotiation(ids = doi, format = "bibentry")
cr=Crossref(mailto='xugzhi1987@gmail.com')
works=cr.works(ids=doi)
pprint(works)
bb=crossRefToMetaDict(works['message'])
print(bb)
'''
eti=Etiquette('MeiTing-Trunk', 'v0.1alpha', 'not published yet',
'xugzhi1987@gmail.com')
print(str(eti))
works=Works(etiquette=eti)
#aa=works.doi('10.1590/0102-311x00133115')
aa=works.doi(doi)
pprint(aa)
bb=crossRefToMetaDict(aa)
def check_trialpubs_nctids(review_id, review_doi=None, sess_id=None):
"""
resolve the references of a review to PMIDs and NCTIDs
@param review_id: PubMed ID of review
@param review_doi: DOI of review
@param sess_id: session ID if transitting progress via websocket
@return: namedtuple with found PMIDs and NCTIDs
"""
if sess_id:
socketio = SocketIO(message_queue='amqp://localhost')
ec = Client(api_key=eutils_key)
cr = Crossref(mailto=config.MAIL_USERNAME)
if not review_doi:
while True:
try:
paset = ec.efetch(db='pubmed', id=review_id)
break
except (
eutils.exceptions.EutilsNCBIError, eutils.exceptions.EutilsRequestError, requests.exceptions.SSLError,
requests.exceptions.ConnectionError) as e:
print e
time.sleep(5)
pa = iter(paset).next()
if hasattr(pa, 'doi'):
review_doi = pa.doi
if not review_doi:
if sess_id:
socketio.emit('crossrefbot_update', {'msg': 'No trials found. Crossrefbot complete'}, room=sess_id)
def crossref_query(authors, title):
"""
Query Crossref database.
Args:
authors (list): a list of strings for up the first authors last names.
title (str): the title of the article.
filename (str): the original path of the file to link to.
Returns:
A tuple (bibtex, json, score) where the first element is the data in
bibtex format (returned as a record/dict), the second element is the
data returned in json format, and the third element is the score of the
match given by Crossref.
"""
cr = Crossref()
# works?query.title=An+Improved+Adaptive+Constraint+Aggregation+for+Integrated+Layout+and+Topology+Optimization&query.author=Gao+Zhu+Zhang+Zhou&sort=score&rows=1
# query = ['+' + name + '' for name in authors]
# query = 'query.title=' + urllib.parse.quote_plus(title) + '&query.author=' + urllib.parse.quote_plus(' '.join(authors)) + '&sort=score&rows=1'
# print(query)
if ''.join(authors):
args = dict(
query_bibliographic=urllib.parse.quote_plus(title),
query_author=urllib.parse.quote_plus(' '.join(authors))
)
else:
args = dict(
query=urllib.parse.quote_plus(title),
)
x = cr.works(sort='score', limit=1, **args)
# x = cr.works(query=query)
assert x['status'] == "ok"
def _get_crossref_works(**kwargs):
cr = habanero.Crossref()
return cr.works(**kwargs)
# Try to cache queries.
cache_path = Path(app.user_cache_dir) / 'crossref.json'
cache = {}
if cache_path.exists():
with cache_path.open() as f:
cache = json.load(f)
if text in cache:
meta = cache[text]
else:
# Look up the article metadata on CrossRef.
try:
crossref = habanero.Crossref(mailto='kale_kundert@hms.harvard.edu')
meta = crossref.works(ids=[text])['message']
# The reference list is big, and we don't care about it, so don't
# cache it.
if 'reference' in meta:
del meta['reference']
cache[text] = meta
cache_path.parent.mkdir(exist_ok=True)
with cache_path.open('w') as f:
json.dump(cache, f)
# Give a useful warning if the article can't be found.
except requests.exceptions.HTTPError as e:
warning = inliner.reporter.warning(f"No matches found for DOI: {text}", line=lineno)
p = nodes.paragraph(text, text)
def make_references(publications, output_dir):
"""
Create reference bib file
Args:
publications: the list of publications
output_dir: the output directory
Returns:
A list of reference identifiers
"""
log = Logger()
cr = Crossref()
lines = []
references = []
for i, publication in enumerate(publications):
log.notice(
f"Querying and formatting {i + 1} out of {len(publications)} publications"
)
link = publication[LINK]
title = publication[TITLE]
# Check if it is a DOI url
if link and "doi.org" in link:
doi = urlparse(link).path.strip("/")
# Extract the DOI using the title
else: