Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
#### same as before, but with one ref
import copy
from wikidataintegrator import wdi_fastrun, wdi_core
from wikidataintegrator.ref_handlers import update_retrieved_if_new as custom_ref_handler
import pprint
class frc_fake_query_data_paper1(wdi_fastrun.FastRunContainer):
def __init__(self, *args, **kwargs):
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'}
def init_fastrun(self):
for c in WDItemEngine.fast_run_store:
if (c.base_filter == self.fast_run_base_filter) and (c.use_refs == self.fast_run_use_refs) and \
(c.sparql_endpoint_url == self.sparql_endpoint_url):
self.fast_run_container = c
self.fast_run_container.ref_handler = self.ref_handler
if not self.fast_run_container:
self.fast_run_container = FastRunContainer(base_filter=self.fast_run_base_filter,
base_data_type=WDBaseDataType, engine=self.__class__,
sparql_endpoint_url=self.sparql_endpoint_url,
mediawiki_api_url=self.mediawiki_api_url,
use_refs=self.fast_run_use_refs,
ref_handler=self.ref_handler)
WDItemEngine.fast_run_store.append(self.fast_run_container)
self.require_write = self.fast_run_container.write_required(self.data, append_props=self.append_value,
cqid=self.wd_item_id)
# set item id based on fast run data
if not self.require_write and not self.wd_item_id:
self.wd_item_id = self.fast_run_container.current_qid
def init_fastrun(self):
for c in self.fast_run_store:
if c.base_filter == self.fast_run_base_filter:
self.fast_run_container = c
if not self.fast_run_container:
self.fast_run_container = FastRunContainer(base_filter=self.fast_run_base_filter)
self.require_write = self.fast_run_container.check_data(self.data, append_props=self.append_value,
cqid=self.wd_item_id)
self.fast_run_store.append(self.fast_run_container)
# set item id based on fast run data
if not self.require_write and not self.wd_item_id:
self.wd_item_id = self.fast_run_container.current_qid
for x in data:
if x.value and x.data_type:
data_props.add(x.get_prop_nr())
write_required = False
match_sets = []
for date in data:
# skip to next if statement has no value or no data type defined, e.g. for deletion objects
current_value = date.get_value()
if not current_value and not date.data_type:
del_props.add(date.get_prop_nr())
continue
prop_nr = date.get_prop_nr()
if prop_nr not in self.prop_dt_map:
self.prop_dt_map.update({prop_nr: FastRunContainer.get_prop_datatype(prop_nr=prop_nr)})
self.__query_data(prop_nr=prop_nr)
# more sophisticated data types like dates and globe coordinates need special treatment here
if self.prop_dt_map[prop_nr] == 'time':
current_value = current_value[0]
elif self.prop_dt_map[prop_nr] == 'wikibase-item':
current_value = 'Q{}'.format(current_value)
elif self.prop_dt_map[prop_nr] == 'globe-coordinate':
write_required = True # temporary workaround for handling globe coordinates
if not __debug__:
print(current_value)
try:
if not __debug__:
print(current_value)
temp_set = set(self.rev_lookup[current_value])