Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
super(frc_fake_query_data_paper1, self).__init__(*args, **kwargs)
self.prop_data['Q15397819'] = {'P698': {
'fake statement id': {
'qual': set(),
'ref': {
'ref1': {
('P248', 'Q5412157'), # stated in Europe PubMed Central
('P813', '+2017-01-01T00:00:00Z'),
('P698', '99999999999')},
},
'v': '99999999999'}}}
self.rev_lookup = {'99999999999': {'Q15397819'}}
self.prop_dt_map = {'P527': 'wikibase-item', 'P248': 'wikibase-item', 'P698': 'external-id', 'P813': 'time'}
class fake_itemengine1(wdi_core.WDItemEngine):
def get_wd_entity(self):
# https://www.wikidata.org/w/api.php?action=wbgetclaims&entity=Q15397819&property=P698&format=json
claims = {'claims': {
'P698': [{'id': 'Q15397819$9460c2a2-4d42-adec-e841-9d5bbdc6695a',
'mainsnak': {'datatype': 'external-id',
'datavalue': {'type': 'string', 'value': '99999999999'},
'property': 'P698',
'snaktype': 'value'},
'rank': 'normal',
'references': [{'hash': '9537cf2da990a2455ab924d027a0a1e5890bde8a',
'snaks': {'P248': [{'datatype': 'wikibase-item',
'datavalue': {'type': 'wikibase-entityid',
'value': {
'entity-type': 'item',
'id': 'Q5412157',
'numeric-id': 5412157}},
def guess_equivalent_property_pid(self):
# get the equivalent property PID without knowing the PID for equivalent property!!!
query = '''SELECT * WHERE {
?item ?prop .
?item ?prop .
}'''
pid = wdi_core.WDItemEngine.execute_sparql_query(query, endpoint=self.sparql_endpoint_url)
pid = pid['results']['bindings'][0]['prop']['value']
pid = pid.split("/")[-1]
return pid
def test_query():
login_instance = wdi_login.WDLogin(user=APP.config.WIKIDATA_USER_NAME, pwd=APP.config.WIKIDATA_PASSWORD)
results = wdi_core.WDItemEngine.execute_sparql_query(QUERIES['list-formats'].replace('\n', ''))
return results
def qid_to_basic_details(qid):
"""Input item qid and returns a tuple: (qid, label, description) using WikiDataIntegrator"""
item = wdi_core.WDItemEngine(wd_item_id=qid)
label = item.get_label(lang=LANG) if item.get_label(lang=LANG) != '' else item.get_label(lang=FALLBACK_LANG)
return {
"id": item.wd_item_id,
"label": label.replace("'", "'"),
"description": (item.get_description(lang=LANG) or item.get_description(lang=FALLBACK_LANG)).replace("'","'"),
"aliases": item.get_aliases(lang=LANG) or item.get_aliases(lang=FALLBACK_LANG)
}
query = '''
#Tool: wdi_core fastrun
SELECT ?p ?label WHERE {{
{0}
OPTIONAL {{
?p {1} ?label FILTER (lang(?label) = "{2}") .
}}
}}
'''.format(self.base_filter_string, lang_data_type_dict[lang_data_type], lang)
if not __debug__:
print(query)
return wdi_core.WDItemEngine.execute_sparql_query(query=query, prefix=prefix)
def get_last_modified_header(entity="http://www.wikidata.org", endpoint='https://query.wikidata.org/sparql'):
# this will work on wikidata or any particular entity
query = "select ?d where {{<{}> schema:dateModified ?d}}".format(entity)
results = wdi_core.WDItemEngine.execute_sparql_query(query, endpoint=endpoint)['results']['bindings']
results = [{k: v['value'] for k, v in x.items()} for x in results]
t = results[0]['d']
try:
# wikidata format
dt = datetime.datetime.strptime(t, '%Y-%m-%dT%H:%M:%SZ')
except ValueError:
# wikibase format
dt = datetime.datetime.strptime(t, '%Y-%m-%dT%H:%M:%S.%fZ')
return dt
modifies s in place
:param s: a WDBaseDataType statement
:param mrt: one of {'close', 'broad', 'exact', 'related', 'narrow'}
:return: s
"""
valid_mrts_abv = self.ABV_MRT.keys()
valid_mrts_uri = self.ABV_MRT.values()
if mrt in valid_mrts_abv:
mrt_uri = self.ABV_MRT[mrt]
elif mrt in valid_mrts_uri:
mrt_uri = mrt
else:
raise ValueError("mrt must be one of {}, found {}".format(valid_mrts_abv, mrt))
mrt_qid = self.mrt_qids[mrt_uri]
q = wdi_core.WDItemID(mrt_qid, self.mrt_pid, is_qualifier=True)
s.qualifiers.append(q)
return s
modifies s in place
:param s: a WDBaseDataType statement
:param mrt: one of {'close', 'broad', 'exact', 'related', 'narrow'}
:return: s
"""
valid_mrts_abv = self.ABV_MRT.keys()
valid_mrts_uri = self.ABV_MRT.values()
if mrt in valid_mrts_abv:
mrt_uri = self.ABV_MRT[mrt]
elif mrt in valid_mrts_uri:
mrt_uri = mrt
else:
raise ValueError("mrt must be one of {}, found {}".format(valid_mrts_abv, mrt))
mrt_qid = self.mrt_qids[mrt_uri]
q = wdi_core.WDItemID(mrt_qid, self.mrt_pid, is_qualifier=True)
s.qualifiers.append(q)
return s