Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test():
assert (
betterbib.tools._translate_title(
"The Magnus expansion and some of its applications"
)
== "The {Magnus} expansion and some of its applications"
)
assert (
betterbib.tools._translate_title("VODE: a variable-coefficient ODE solver")
== "{VODE:} {A} variable-coefficient {ODE} solver"
)
assert (
betterbib.tools._translate_title(
"Peano's kernel theorem for vector-valued functions"
)
== "{Peano's} kernel theorem for vector-valued functions"
)
executor.submit(source.find_unique, entry): (bib_id, tools.decode(entry))
for bib_id, entry in d.items()
}
for future in tqdm(
concurrent.futures.as_completed(responses), total=len(responses)
):
bib_id, entry = responses[future]
try:
data = future.result()
except (errors.NotFoundError, errors.UniqueError):
pass
except errors.HttpError as e:
print(e.args[0])
else:
num_success += 1
d[bib_id] = tools.update(entry, data)
print("\n\nTotal number of entries: {}".format(len(d)))
print("Found: {}".format(num_success))
return d
def sync(d, source, long_journal_name, max_workers):
if source == "crossref":
source = crossref.Crossref(long_journal_name)
else:
assert source == "dblp", "Illegal source."
source = dblp.Dblp()
print()
num_success = 0
with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
responses = {
executor.submit(source.find_unique, entry): (bib_id, tools.decode(entry))
for bib_id, entry in d.items()
}
for future in tqdm(
concurrent.futures.as_completed(responses), total=len(responses)
):
bib_id, entry = responses[future]
try:
data = future.result()
except (errors.NotFoundError, errors.UniqueError):
pass
except errors.HttpError as e:
print(e.args[0])
else:
num_success += 1
d[bib_id] = tools.update(entry, data)