diff --git a/simple/stats/data.py b/simple/stats/data.py index 22c026366..e9c5524e7 100644 --- a/simple/stats/data.py +++ b/simple/stats/data.py @@ -28,6 +28,7 @@ from stats import schema_constants as sc from stats.util import base64_decode_and_gunzip_json from stats.util import gzip_and_base64_encode_json +from stats.util import has_namespace_prefix from stats.util import is_uri_or_namespace _PREDICATE_TYPE_OF = "typeOf" @@ -181,6 +182,7 @@ def triples(self) -> list[Triple]: class Entity: entity_dcid: str entity_type: str + provenance_ids: set[str] = field(default_factory=set) def triples(self) -> list[Triple]: # Currently only 1 triple is generated but could be more in the future (e.g. name) @@ -196,6 +198,7 @@ class Provenance: name: str url: str = "" properties: dict[str, str] = field(default_factory=dict) + provenance_id: str = "" def triples(self) -> list[Triple]: triples: list[Triple] = [] @@ -227,6 +230,7 @@ class Source: url: str = "" domain: str = field(init=False) properties: dict[str, str] = field(default_factory=dict) + provenance_id: str = "" def __post_init__(self): self.domain = urlparse(self.url).netloc @@ -331,6 +335,7 @@ def _get_flattened_dataclass_field_names(cls) -> list[str]: class Property: dcid: str name: str + provenance_ids: set[str] = field(default_factory=set) def triples(self) -> list[Triple]: return [ @@ -546,7 +551,10 @@ def add_triple(self, triple: Triple) -> Self: def to_mcf(self) -> str: parts: list[str] = [] - parts.append(f"Node: dcid:{self.id}") + if has_namespace_prefix(self.id): + parts.append(f"Node: {self.id}") + else: + parts.append(f"Node: dcid:{self.id}") parts.extend([f"{p}: {v}" for p, v in self.properties.items()]) return "\n".join(parts) diff --git a/simple/stats/db.py b/simple/stats/db.py index d5812c397..18e0d0d55 100644 --- a/simple/stats/db.py +++ b/simple/stats/db.py @@ -230,7 +230,10 @@ class Db: def maybe_clear_before_import(self): pass - def insert_triples(self, triples: list[Triple], input_file: File = None): + def insert_triples(self, + triples: list[Triple], + input_file: File = None, + provenance_dir: str = None): pass def insert_observations(self, observations_df: pd.DataFrame, @@ -281,7 +284,10 @@ def __init__(self, db_params: dict) -> None: # dcid to node dict self.nodes: dict[str, McfNode] = {} - def insert_triples(self, triples: list[Triple], input_file: File = None): + def insert_triples(self, + triples: list[Triple], + input_file: File = None, + provenance_dir: str = None): for triple in triples: self._add_triple(triple) @@ -351,7 +357,10 @@ def __init__(self, config: dict) -> None: def maybe_clear_before_import(self): self.engine.clear_tables_and_indexes() - def insert_triples(self, triples: list[Triple], input_file: File = None): + def insert_triples(self, + triples: list[Triple], + input_file: File = None, + provenance_dir: str = None): logging.info("Writing %s triples to [%s]", len(triples), self.engine) if triples: self.engine.executemany(_INSERT_TRIPLES_STATEMENT, @@ -425,7 +434,10 @@ def maybe_clear_before_import(self): # Not applicable for Data Commons Platform. pass - def insert_triples(self, triples: list[Triple], input_file: File = None): + def insert_triples(self, + triples: list[Triple], + input_file: File = None, + provenance_dir: str = None): """ Convert triples to a jsonld graph and writes the graph to the Data Commons Platform instance. """ diff --git a/simple/stats/events_importer.py b/simple/stats/events_importer.py index 2b9cf51b9..f36a3ebb5 100644 --- a/simple/stats/events_importer.py +++ b/simple/stats/events_importer.py @@ -20,6 +20,7 @@ from stats.data import AggregationConfig from stats.data import Event from stats.data import filter_invalid_observation_values +from stats.data import strip_namespace from stats.data import strip_namespace_series from stats.data import TimePeriod from stats.data import Triple @@ -27,6 +28,8 @@ from stats.importer import Importer from stats.nodes import Nodes from stats.reporter import FileImportReporter +from stats.util import get_namespace_prefix_and_suffix +from stats.util import has_namespace_prefix from util.filesystem import File from util import dc_client as dc @@ -237,9 +240,9 @@ def _resolve_entities(self) -> None: pre_resolved_entities = {} def remove_pre_resolved(entity: str) -> bool: - if entity.startswith(constants.DCID_OVERRIDE_PREFIX): - pre_resolved_entities[entity] = entity[ - len(constants.DCID_OVERRIDE_PREFIX):].strip() + if has_namespace_prefix(entity): + prefix, suffix = get_namespace_prefix_and_suffix(entity) + pre_resolved_entities[entity] = suffix.strip() return False return True @@ -277,6 +280,19 @@ def remove_pre_resolved(entity: str) -> bool: unresolved=unresolved_list, ) + prov_id = getattr(self, "provenance", "") + for dcid in df[constants.COLUMN_DCID].dropna().unique(): + clean_dcid = strip_namespace(dcid) + if self.nodes.has_entity(clean_dcid): + if prov_id: + self.nodes.entities[clean_dcid].provenance_ids.add(prov_id) + elif prov_id: + self.nodes.entity_with_type( + clean_dcid, + self.entity_type or "Thing", + provenance_id=prov_id, + ) + def _resolve(self, entities: list[str]) -> dict[str, str]: lower_case_entity_name = self.entity_column_name.lower() diff --git a/simple/stats/jsonld_stream_db.py b/simple/stats/jsonld_stream_db.py index 591e19a8b..973bcfa8e 100644 --- a/simple/stats/jsonld_stream_db.py +++ b/simple/stats/jsonld_stream_db.py @@ -102,21 +102,13 @@ def _parse_numeric(val): return str(val) -def _write_observation_shard(chunk_or_args, +def _write_observation_shard(chunk, shard_index: Optional[int] = None, jsonld_dir_path: Optional[str] = None, ns_map: Optional[dict[str, str]] = None, prov_urls: Optional[dict[str, str]] = None, track_hash_fn: Optional[Callable] = None, file_name: str = ""): - if isinstance(chunk_or_args, tuple): - chunk = chunk_or_args[0] - shard_index = chunk_or_args[1] - jsonld_dir_path = chunk_or_args[2] - ns_map = chunk_or_args[3] - prov_urls = chunk_or_args[4] - else: - chunk = chunk_or_args graph_list = [] chunk_hashes = [] @@ -207,6 +199,9 @@ def _write_observation_shard(chunk_or_args, logging.info(f"Saved JSON-LD shard to {shard_name}") +# TODO(gmechali): When parallelizing node shard exports (MCF and Events CSV files), +# include the sanitized file_name in the shard name (e.g., node-{sanitized_stem}-{shard_index:05d}.jsonld) +# to prevent shard filename collisions across parallel workers, similar to observation shards. def _write_node_shard(args): # TODO(gmechali): Get rid of this and keep only the "fast" mode. fast_export = os.getenv("FAST_NODE_EXPORT", @@ -268,6 +263,8 @@ def _write_node_shard_fast(args): graph_list = sorted(list(subjects.values()), key=lambda x: x["@id"]) compacted_jsonld = {"@context": ns_map, "@graph": graph_list} + # TODO(gmechali): When parallelizing node shard exports, include file_name in shard_name + # (e.g., f"node-{sanitized_stem}-{shard_index:05d}.jsonld") to prevent collisions across workers. shard_name = f"node-{shard_index:05d}.jsonld" with create_store(jsonld_dir_path) as store: output_dir = store.as_dir() @@ -465,38 +462,43 @@ def _write_triples_to_disk(self, triples: list[Triple], import_name: str): (chunk, self.node_shard_index, import_temp_dir, self.ns_map)) self.node_shard_index += 1 - def insert_triples(self, triples: list[Triple], input_file: File = None): + def insert_triples(self, + triples: list[Triple], + input_file: File = None, + provenance_dir: str = None): if not triples: return - if not input_file: + if not provenance_dir and input_file: + provenance_dir = self.config.import_name(input_file) + + if not provenance_dir: with self.lock: self._triples["_global"].extend(triples) return - import_name = self.config.import_name(input_file) - self._init_import_export_dir(import_name) - self._write_triples_to_disk(triples, import_name) + self._init_import_export_dir(provenance_dir) + self._write_triples_to_disk(triples, provenance_dir) def commit(self): pass def commit_and_close(self): - # Add global triples to every processed import's triples global_triples = self._triples.pop("_global", []) if not self._processed_imports: self._processed_imports.add(self.import_name) - # Write global triples as node shards to the local temp directory for each import - for import_name in self._processed_imports: - import_temp_dir = os.path.join(self.temp_local_dir, import_name) + # Write any remaining untagged triples once to the primary import directory (no duplication across provenances) + if global_triples: + target_import = self.import_name if self.import_name in self._processed_imports else next( + iter(self._processed_imports)) + import_temp_dir = os.path.join(self.temp_local_dir, target_import) os.makedirs(import_temp_dir, exist_ok=True) - if global_triples: - for i in range(0, len(global_triples), _CHUNK_SIZE): - chunk = global_triples[i:i + _CHUNK_SIZE] - _write_node_shard( - (chunk, self.node_shard_index, import_temp_dir, self.ns_map)) - self.node_shard_index += 1 + for i in range(0, len(global_triples), _CHUNK_SIZE): + chunk = global_triples[i:i + _CHUNK_SIZE] + _write_node_shard( + (chunk, self.node_shard_index, import_temp_dir, self.ns_map)) + self.node_shard_index += 1 has_local_files = any(os.scandir(self.temp_local_dir)) if os.path.exists( self.temp_local_dir) else False @@ -591,7 +593,10 @@ def _upload_shards_local(self, temp_local_dir: str, files: list[str]): parent_dirs = set(os.path.dirname(f) for f in files if os.path.dirname(f)) for d in sorted(parent_dirs): - target_store.open_dir(d) + try: + target_store.open_dir(d) + except Exception: + pass def _copy_single(rel_path: str): local_file_path = os.path.join(temp_local_dir, rel_path) diff --git a/simple/stats/mcf_importer.py b/simple/stats/mcf_importer.py index 0987d319a..5780fe6b6 100644 --- a/simple/stats/mcf_importer.py +++ b/simple/stats/mcf_importer.py @@ -105,7 +105,10 @@ def do_import(self) -> None: # Register all collected metadata nodes at the end if all_metadata_triples: - _register_metadata_nodes(all_metadata_triples, self.nodes) + prov_id = getattr(self, "provenance", "") + _register_metadata_nodes(all_metadata_triples, + self.nodes, + provenance_id=prov_id) self.reporter.report_success() except Exception as e: @@ -161,7 +164,9 @@ def _to_triple(parser_triple: list[str], local2dcid: dict[str, str]) -> Triple: return Triple(resolved_subject, predicate, object_value=value) -def _register_metadata_nodes(triples: list[Triple], nodes: Nodes) -> None: +def _register_metadata_nodes(triples: list[Triple], + nodes: Nodes, + provenance_id: str = "") -> None: """Extracts and registers Provenance and Source nodes from parsed MCF triples. This helper is robust against: @@ -191,8 +196,8 @@ def _register_metadata_nodes(triples: list[Triple], nodes: Nodes) -> None: val = triple.object_value or triple.object_id or "" subject_properties[sub_id]["name"] = _clean_literal(val) elif pred == "source": - # Map source to the source ID - subject_properties[sub_id]["source"] = triple.object_id + val = triple.object_id or triple.object_value or "" + subject_properties[sub_id]["source"] = _clean_literal(val) for sub_id, props in subject_properties.items(): node_type = props.get("typeOf") @@ -204,7 +209,8 @@ def _register_metadata_nodes(triples: list[Triple], nodes: Nodes) -> None: elif node_type == "Source": nodes.register_source(id=sub_id, name=props.get("name", ""), - url=props.get("url", "")) + url=props.get("url", ""), + provenance_id=provenance_id) def _clean_literal(val: str) -> str: diff --git a/simple/stats/nodes.py b/simple/stats/nodes.py index 6f596b68a..0daceb9a4 100644 --- a/simple/stats/nodes.py +++ b/simple/stats/nodes.py @@ -12,6 +12,7 @@ # See the License for the specific language governing permissions and # limitations under the License. +from collections import defaultdict from functools import wraps import logging import re @@ -157,7 +158,8 @@ def register_provenance(self, source_id=clean_source_id, name=name or clean_id, url=url, - properties=properties or {}) + properties=properties or {}, + provenance_id=clean_id) self.provenances[clean_id] = prov if name: self.provenances[name] = prov @@ -172,6 +174,8 @@ def register_provenance(self, prov.source_id = clean_source_id if properties: prov.properties.update(properties) + if not prov.provenance_id: + prov.provenance_id = clean_id return prov @thread_safe @@ -179,7 +183,8 @@ def register_source(self, id: str, name: str = "", url: str = "", - properties: dict[str, str] = None) -> Source: + properties: dict[str, str] = None, + provenance_id: str = "") -> Source: self.has_custom_mcf_nodes = True clean_id = _clean_metadata_id(id) @@ -188,7 +193,8 @@ def register_source(self, src = Source(id=clean_id, name=name or clean_id, url=url, - properties=properties or {}) + properties=properties or {}, + provenance_id=provenance_id or "") self.sources[clean_id] = src if name: self.sources[name] = src @@ -201,6 +207,8 @@ def register_source(self, src.url = url if properties: src.properties.update(properties) + if provenance_id and not src.provenance_id: + src.provenance_id = provenance_id return src @thread_safe @@ -238,10 +246,14 @@ def variable(self, sv_column_name: str, input_file: File) -> StatVar: self.provenance(input_file)) @thread_safe - def property(self, property_column_name: str) -> Property: + def property(self, + property_column_name: str, + provenance_id: str = "") -> Property: if not property_column_name in self.properties: self.properties[property_column_name] = Property( self._property_id(property_column_name), property_column_name) + if provenance_id: + self.properties[property_column_name].provenance_ids.add(provenance_id) return self.properties[property_column_name] @@ -357,9 +369,14 @@ def _default_custom_group(self) -> StatVarGroup: return self.groups[_DEFAULT_CUSTOM_GROUP_PATH] @thread_safe - def entity_with_type(self, entity_dcid: str, entity_type: str): + def entity_with_type(self, + entity_dcid: str, + entity_type: str, + provenance_id: str = ""): if entity_dcid not in self.entities: self.entities[entity_dcid] = Entity(entity_dcid, entity_type) + if provenance_id: + self.entities[entity_dcid].provenance_ids.add(provenance_id) @thread_safe def has_entity(self, entity_dcid: str) -> bool: @@ -377,18 +394,27 @@ def get_provenance_urls(self) -> dict[str, str]: return prov_urls @thread_safe - def entities_with_type(self, entity_dcids: list[str], entity_type: str): + def entities_with_type(self, + entity_dcids: list[str], + entity_type: str, + provenance_id: str = ""): for entity_dcid in entity_dcids: - self.entity_with_type(entity_dcid, entity_type) + self.entity_with_type(entity_dcid, + entity_type, + provenance_id=provenance_id) @thread_safe - def entities_with_types(self, dcid2type: dict[str, str]): + def entities_with_types(self, + dcid2type: dict[str, str], + provenance_id: str = ""): """ Adds each dcid2type mapping to the list of entities with their types. The full list will be inserted into the DB in the final stages of the import. """ for entity_dcid, entity_type in dcid2type.items(): - self.entity_with_type(entity_dcid, entity_type) + self.entity_with_type(entity_dcid, + entity_type, + provenance_id=provenance_id) @thread_safe def triples(self, triples_file: File | None = None) -> list[Triple]: @@ -420,6 +446,49 @@ def triples(self, triples_file: File | None = None) -> list[Triple]: return triples + @thread_safe + def triples_by_provenance_dir(self) -> dict[str, list[Triple]]: + result: dict[str, list[Triple]] = defaultdict(list) + for source in self.sources.values(): + if self.has_custom_mcf_nodes and source.id == _DEFAULT_SOURCE.id and _DEFAULT_SOURCE.id not in self._used_source_ids: + continue + raw = getattr(source, "provenance_id", "") or "_global" + imp = strip_namespace(raw) if raw != "_global" else "_global" + result[imp].extend(source.triples()) + for provenance in self.provenances.values(): + if self.has_custom_mcf_nodes and provenance.id == _DEFAULT_PROVENANCE.id and _DEFAULT_PROVENANCE.id not in self._used_provenance_ids: + continue + raw = getattr(provenance, "provenance_id", + "") or provenance.id or "_global" + imp = strip_namespace(raw) if raw != "_global" else "_global" + result[imp].extend(provenance.triples()) + for group in self.groups.values(): + result["_global"].extend(group.triples()) + for collection in ( + self.variables, + self.event_types, + self.entity_types, + self.properties, + self.entities, + ): + for node in collection.values(): + p_ids = getattr(node, "provenance_ids", None) + target_dirs = ({strip_namespace(p) for p in p_ids} + if p_ids else {"_global"}) + for dir_name in target_dirs: + result[dir_name].extend(node.triples()) + deduped_result = {} + for k, triples in result.items(): + seen = set() + deduped = [] + for t in triples: + key = (t.subject_id, t.predicate, t.object_id, t.object_value) + if key not in seen: + seen.add(key) + deduped.append(t) + deduped_result[k] = deduped + return deduped_result + def _clean_metadata_id(id: str) -> str: """Only strips standard Data Commons system prefixes, preserving custom namespaces.""" diff --git a/simple/stats/observations_importer.py b/simple/stats/observations_importer.py index f635aa812..7079e9d18 100644 --- a/simple/stats/observations_importer.py +++ b/simple/stats/observations_importer.py @@ -24,6 +24,8 @@ from stats.importer import Importer from stats.nodes import Nodes from stats.reporter import FileImportReporter +from stats.util import get_namespace_prefix_and_suffix +from stats.util import has_namespace_prefix from util.filesystem import File from util import dc_client as dc @@ -147,6 +149,12 @@ def _add_entity_nodes(self) -> None: dcid for dcid in entity_dcids if not self.nodes.has_entity(dcid) ] + prov_id = getattr(self, "provenance", "") + if prov_id: + for dcid in entity_dcids: + if self.nodes.has_entity(dcid): + self.nodes.entities[dcid].provenance_ids.add(prov_id) + logging.info("Found %s total entities, of which %s are already imported.", len(entity_dcids), len(entity_dcids) - len(new_entity_dcids)) @@ -162,11 +170,13 @@ def _add_entity_nodes(self) -> None: if dcid2type: logging.info("Importing %s of %s entities.", len(dcid2type), len(new_entity_dcids)) - self.nodes.entities_with_types(dcid2type) + self.nodes.entities_with_types(dcid2type, provenance_id=prov_id) elif self.entity_type: logging.info("Importing %s entities with type %s.", len(new_entity_dcids), self.entity_type) - self.nodes.entities_with_type(new_entity_dcids, self.entity_type) + self.nodes.entities_with_type(new_entity_dcids, + self.entity_type, + provenance_id=prov_id) def _resolve_entities(self) -> None: df = self.df @@ -176,9 +186,9 @@ def _resolve_entities(self) -> None: pre_resolved_entities = {} def remove_pre_resolved(entity: str) -> bool: - if entity.startswith(constants.DCID_OVERRIDE_PREFIX): - pre_resolved_entities[entity] = entity[ - len(constants.DCID_OVERRIDE_PREFIX):].strip() + if has_namespace_prefix(entity): + prefix, suffix = get_namespace_prefix_and_suffix(entity) + pre_resolved_entities[entity] = suffix.strip() return False return True diff --git a/simple/stats/runner.py b/simple/stats/runner.py index 39f2053ba..64648d52e 100644 --- a/simple/stats/runner.py +++ b/simple/stats/runner.py @@ -177,6 +177,7 @@ class ImportProcResult: provenances: dict groups: dict properties: dict + processed_imports: set def _run_single_csv_import_proc( @@ -219,7 +220,8 @@ def _run_single_csv_import_proc( db.commit_and_close() resolved_entities = { - e.entity_dcid: e.entity_type for e in nodes.entities.values() + e.entity_dcid: (e.entity_type, getattr(e, "provenance_ids", set())) + for e in nodes.entities.values() } event_types = dict(nodes.event_types) entity_types = dict(nodes.entity_types) @@ -241,6 +243,7 @@ def _run_single_csv_import_proc( provenances=provenances, groups=groups, properties=properties, + processed_imports=set(db._processed_imports), ) @@ -1038,7 +1041,16 @@ def _run_all_data_imports(self): res = future.result() self._log_file_progress("Imported file", res.file_rel_path) if res.resolved_entities: - self.nodes.entities_with_types(res.resolved_entities) + for dcid, val in res.resolved_entities.items(): + if isinstance(val, tuple): + t, p_ids = val + else: + t, p_ids = val, set() + if isinstance(p_ids, set): + for p_id in (p_ids or [""]): + self.nodes.entity_with_type(dcid, t, provenance_id=p_id) + else: + self.nodes.entity_with_type(dcid, t, provenance_id=p_ids) if res.event_types: self.nodes.event_types.update(res.event_types) if res.entity_types: @@ -1049,7 +1061,9 @@ def _run_all_data_imports(self): for src in res.sources.values(): self.nodes.register_source(id=src.id, name=src.name, - url=src.url) + url=src.url, + provenance_id=getattr( + src, "provenance_id", "")) if res.provenances: for prov in res.provenances.values(): self.nodes.register_provenance(id=prov.id, @@ -1062,7 +1076,14 @@ def _run_all_data_imports(self): for svg in res.groups.values(): self.nodes.ids_to_groups[svg.id] = svg if res.properties: - self.nodes.properties.update(res.properties) + for prop_name, prop_obj in res.properties.items(): + if prop_name not in self.nodes.properties: + self.nodes.properties[prop_name] = prop_obj + else: + self.nodes.properties[prop_name].provenance_ids.update( + getattr(prop_obj, "provenance_ids", set())) + if res.processed_imports: + self.db._processed_imports.update(res.processed_imports) if res.obs_collision_count and hasattr(self.db, "obs_collision_count"): self.db.obs_collision_count += res.obs_collision_count @@ -1128,9 +1149,12 @@ def _run_imports_and_export_jsonld(self): # Run data imports (CSV and MCF) self._run_all_data_imports() - # Generate triples from nodes and write directly - triples = self.nodes.triples() - self.db.insert_triples(triples) + # Generate triples from nodes grouped by provenance directory and write directly + for prov_dir, triples in self.nodes.triples_by_provenance_dir().items(): + if prov_dir == "_global": + self.db.insert_triples(triples) + else: + self.db.insert_triples(triples, provenance_dir=prov_dir) # Perform strict metadata validation before committing and closing MetadataValidator(self.config, self.db).validate() @@ -1139,12 +1163,12 @@ def _run_imports_and_export_jsonld(self): output_path = self.db.jsonld_dir.full_path() import_name = self.db.import_name if os.getenv("WORKFLOW_EXECUTION_ID") and output_path.startswith("gs://"): - processed_imports = list(self.db._processed_imports) + processed_imports = sorted(list(self.db._processed_imports)) if not processed_imports: processed_imports = [import_name] import_list = [] for imp in processed_imports: - gcs_pattern = f"{output_path.rstrip('/')}/{imp}/*.jsonld" + gcs_pattern = f"{output_path.rstrip('/')}/{imp}/**/*.jsonld" import_list.append({"importName": imp, "graphPath": gcs_pattern}) self.trigger_workflow_info = import_list else: diff --git a/simple/stats/util.py b/simple/stats/util.py index a5f3c0b48..6b74423c4 100644 --- a/simple/stats/util.py +++ b/simple/stats/util.py @@ -55,6 +55,33 @@ def base64_decode_and_gunzip_json(encoded_data: str) -> dict: return json.loads(base64_decode_and_gunzip(encoded_data)) +_STANDARD_PREFIXES = {"http", "https", "dcid", "schema", "dcs"} +_NAMESPACE_PREFIX_RE = re.compile(r"^[a-zA-Z0-9_\-]+:") + + +def has_namespace_prefix(val: str) -> bool: + """Returns True if string starts with a namespace prefix (e.g. 'dcid:', 'undata:', 'acme:').""" + if not isinstance(val, str) or not val: + return False + return bool(_NAMESPACE_PREFIX_RE.match(val.strip())) + + +def get_namespace_prefix_and_suffix(val: str) -> tuple[str, str]: + """Extracts (prefix, suffix) from a namespaced string (e.g. 'undata:place/1' -> ('undata', 'place/1')).""" + if isinstance(val, str) and ":" in val and " " not in val.split(":", 1)[0]: + parts = val.split(":", 1) + return parts[0], parts[1] + return "", val or "" + + +def is_custom_namespace_prefix(prefix: str) -> bool: + """Returns True if the prefix represents a custom instance namespace (not standard 'dcid', 'http', 'schema').""" + if not prefix: + return False + p_lower = prefix.lower() + return p_lower not in _STANDARD_PREFIXES + + def is_uri_or_namespace(val: str) -> bool: """Returns True if the value is a full URL, standard DCID, or valid custom namespace.""" if not isinstance(val, str): @@ -65,7 +92,6 @@ def is_uri_or_namespace(val: str) -> bool: return True if ":" in val and " " not in val: prefix = val.split(":", 1)[0] - # A valid namespace prefix must be purely alphanumeric (e.g. 'custom', 'un', 'myorg') return prefix.isalnum() return False diff --git a/simple/stats/validation.py b/simple/stats/validation.py index e93c22b4c..61a35a264 100644 --- a/simple/stats/validation.py +++ b/simple/stats/validation.py @@ -15,8 +15,10 @@ import logging from stats.config import Config +from stats.data import strip_namespace from stats.data import ValidationErrorType from stats.db import Db +from stats.util import has_namespace_prefix from stats.util import is_uri_or_namespace @@ -102,7 +104,7 @@ def _collect_defined_nodes(self) -> tuple[set[str], dict[str, str]]: for triple in all_triples: sub = self._clean_dcid(triple.subject_id) - pred = triple.predicate + pred = strip_namespace(triple.predicate) if pred == "typeOf": obj = triple.object_id or "" @@ -110,7 +112,7 @@ def _collect_defined_nodes(self) -> tuple[set[str], dict[str, str]]: defined_provenances.add(sub) elif pred == "source": - obj = triple.object_id or "" + obj = triple.object_id or triple.object_value or "" if obj: provenance_to_source[sub] = self._clean_dcid(obj) @@ -151,11 +153,9 @@ def _validate_source_links(self, defined_provs: set[str], raise ex def _clean_dcid(self, val: str) -> str: - """Normalizes a DCID value by ensuring it starts with 'dcid:' and has no prefix namespaces.""" - if val.startswith(("http://", "https://")): - return val - if val.startswith("dcid:"): + """Normalizes a DCID value by ensuring it starts with a namespace prefix.""" + if not val: + return "" + if has_namespace_prefix(val): return val - if ":" in val: - return f"dcid:{val.split(':', 1)[1]}" return f"dcid:{val}" diff --git a/simple/stats/variable_per_row_importer.py b/simple/stats/variable_per_row_importer.py index b68c32aee..d01f86028 100644 --- a/simple/stats/variable_per_row_importer.py +++ b/simple/stats/variable_per_row_importer.py @@ -337,9 +337,16 @@ def _add_entity_nodes(self) -> None: new_entity_dcids = [ strip_namespace(dcid) for dcid in self.entity_dcids - if not self.nodes.has_entity(dcid) + if not self.nodes.has_entity(strip_namespace(dcid)) ] + prov_id = getattr(self, "provenance", "") + if prov_id: + for dcid in self.entity_dcids: + clean_dcid = strip_namespace(dcid) + if self.nodes.has_entity(clean_dcid): + self.nodes.entities[clean_dcid].provenance_ids.add(prov_id) + logging.info("Found %s total entities, of which %s are already imported.", len(self.entity_dcids), len(self.entity_dcids) - len(new_entity_dcids)) @@ -357,4 +364,4 @@ def _add_entity_nodes(self) -> None: if dcid2type: logging.info("Importing %s of %s entities.", len(dcid2type), len(new_entity_dcids)) - self.nodes.entities_with_types(dcid2type) + self.nodes.entities_with_types(dcid2type, provenance_id=prov_id) diff --git a/simple/tests/stats/nodes_test.py b/simple/tests/stats/nodes_test.py index 88fcba108..013258808 100644 --- a/simple/tests/stats/nodes_test.py +++ b/simple/tests/stats/nodes_test.py @@ -407,3 +407,14 @@ def test_default_custom_root_group_name_override(self): ] self.assertTrue(default_groups, "Default custom root SVG should exist") self.assertEqual(default_groups[0].name, "ONE Data") + + def test_triples_by_provenance_dir_and_deduplication(self): + nodes = Nodes(Config(CONFIG_DATA)) + var = nodes.variable("Variable 1", self.a) + # Add duplicate provenance IDs to verify deduplication + var.provenance_ids = ["prov/1", "prov/1"] + shards = nodes.triples_by_provenance_dir() + self.assertIn("prov/1", shards) + triples = shards["prov/1"] + db_tuples = [t.db_tuple() for t in triples] + self.assertEqual(len(db_tuples), len(set(db_tuples))) diff --git a/simple/tests/stats/observations_importer_test.py b/simple/tests/stats/observations_importer_test.py index 037af4539..548168937 100644 --- a/simple/tests/stats/observations_importer_test.py +++ b/simple/tests/stats/observations_importer_test.py @@ -17,6 +17,7 @@ import shutil import tempfile import unittest +from unittest import mock from unittest.mock import MagicMock import pandas as pd @@ -105,3 +106,43 @@ def test_countryalpha3codes(self): def test_obs_props(self): _test_import(self, "obs_props") + + def test_custom_namespace_pre_resolution(self): + config = Config({ + "inputFiles": [{ + "pattern": "data.csv", + "provenance": "undata:provenance/WHO" + }] + }) + nodes = Nodes(config) + mock_input_file = MagicMock() + mock_input_file.path = "data.csv" + importer = ObservationsImporter( + input_file=mock_input_file, + db=MagicMock(), + debug_resolve_file=MagicMock(), + reporter=MagicMock(), + nodes=nodes, + ) + importer.entity_type = "State" + importer.entity_column_name = "entity" + importer.df = pd.DataFrame( + {"dcid": ["undata:place/custom_1", "dcid:geoId/06", "California"]}) + with mock.patch.object( + dc_client, + "resolve_entities", + return_value={"California": "geoId/06"}, + ) as mock_resolve, mock.patch.object(dc_client, + "get_property_of_entities", + return_value={}): + importer._resolve_entities() + + mock_resolve.assert_called_once_with( + entities=["California"], + entity_type="State", + property_name="description", + ) + self.assertEqual( + importer.df["dcid"].tolist(), + ["place/custom_1", "geoId/06", "geoId/06"], + ) diff --git a/simple/tests/stats/test_data/events_importer/expected/countryalpha3codes.triples.db.csv b/simple/tests/stats/test_data/events_importer/expected/countryalpha3codes.triples.db.csv index 3b1640a50..9692c7c2f 100644 --- a/simple/tests/stats/test_data/events_importer/expected/countryalpha3codes.triples.db.csv +++ b/simple/tests/stats/test_data/events_importer/expected/countryalpha3codes.triples.db.csv @@ -94,3 +94,6 @@ PRIMARY_DESCRIPTION,typeOf,Property, PRIMARY_DESCRIPTION,name,,PRIMARY DESCRIPTION SECONDARY_DESCRIPTION,typeOf,Property, SECONDARY_DESCRIPTION,name,,SECONDARY DESCRIPTION +country/USA,typeOf,Country, +country/BRA,typeOf,Country, +country/CHN,typeOf,Country, diff --git a/simple/tests/stats/test_data/events_importer/expected/idcolumns.triples.db.csv b/simple/tests/stats/test_data/events_importer/expected/idcolumns.triples.db.csv index bd28712a1..e87b0bcb4 100644 --- a/simple/tests/stats/test_data/events_importer/expected/idcolumns.triples.db.csv +++ b/simple/tests/stats/test_data/events_importer/expected/idcolumns.triples.db.csv @@ -93,3 +93,6 @@ PRIMARY_DESCRIPTION,typeOf,Property, PRIMARY_DESCRIPTION,name,,PRIMARY DESCRIPTION SECONDARY_DESCRIPTION,typeOf,Property, SECONDARY_DESCRIPTION,name,,SECONDARY DESCRIPTION +country/USA,typeOf,Country, +country/BRA,typeOf,Country, +country/CHN,typeOf,Country, diff --git a/simple/util/filesystem.py b/simple/util/filesystem.py index 0ea1d0a23..27db18ab5 100644 --- a/simple/util/filesystem.py +++ b/simple/util/filesystem.py @@ -201,10 +201,8 @@ def __init__(self, store: "Store", path: str, create_if_missing: bool): super().__init__(store, path) if not self.fs().exists(self.path): if create_if_missing: - # Make parent dir if needed. parent_dir_path = fspath.dirname(path) - if not self.fs().isdir(parent_dir_path): - self.fs().makedirs(parent_dir_path) + self.fs().makedirs(parent_dir_path, recreate=True) # Make empty file. self.fs().touch(path) else: @@ -232,10 +230,7 @@ def read_string_io(self) -> io.StringIO: return io.StringIO(self.read()) def open_stream(self, max_retries: int = 3): - """Returns an open text stream for streaming line-by-line without downloading full contents upfront. - - Includes retries for transient network drops. - """ + """Returns an open text stream for streaming line-by-line with retries for transient network drops.""" for attempt in range(max_retries): try: return self.fs().open(self.path, "r")