diff --git a/digital_land/check.py b/digital_land/check.py index 9f043e7a..8b64119b 100644 --- a/digital_land/check.py +++ b/digital_land/check.py @@ -1,6 +1,7 @@ import duckdb import logging import dask.dataframe as dd +import pandas as pd # TODO This might need to move into expectations as it is a form of data checking @@ -22,10 +23,11 @@ def duplicate_reference_check(issues=None, csv_path=None): "field", "value", "entry_date", + "entity", COUNT(*) AS count, STRING_AGG("entry_number"::TEXT, ',') AS entry_numbers FROM filtered_table - GROUP BY "field", "value", "entry_date" + GROUP BY "field", "value", "entry_date", "entity" HAVING COUNT(*) > 1; """ @@ -34,11 +36,14 @@ def duplicate_reference_check(issues=None, csv_path=None): if len(count_table) >= 1: duplicate_references = count_table[count_table["count"] > 1] for idx, row in duplicate_references.iterrows(): + entity = row["entity"] + entity = int(entity) if pd.notna(entity) else None for entry_number in row["entry_numbers"].split(","): issues.log_issue( "reference", "reference values are not unique", row["value"], + entity=entity, entry_number=int(entry_number), line_number=int(entry_number) + 1, # TODO Check this makes sense in all cases diff --git a/digital_land/phase/lookup.py b/digital_land/phase/lookup.py index c5dfb712..6f4161b0 100644 --- a/digital_land/phase/lookup.py +++ b/digital_land/phase/lookup.py @@ -84,14 +84,27 @@ def check_associated_organisation(self, entity): return "" return entity - def get_entity(self, block): + def get_entity(self, block, organisation=None): + """Look up the entity for a block's prefix and reference. + + Args: + organisation (str, optional): Organisation to resolve the entity against. + Defaults to None, meaning use the organisation already on the row. + EntityLookupPhase passes this explicitly when fanning out a + multi-organisation resource, where the row's organisation is blank + because the resource-level default is only applied when a resource + has exactly one organisation. + + Returns: + str: The entity number, or "" if nothing matched. + """ row = block["row"] prefix = row.get("prefix", "") reference = row.get("reference", "") # Eventually won't be needed but leave in for now - organisation = row.get("organisation", "").replace( - "local-authority-eng", "local-authority" - ) + if organisation is None: + organisation = row.get("organisation", "") + organisation = organisation.replace("local-authority-eng", "local-authority") entry_number = block["entry-number"] entity = ( @@ -143,6 +156,32 @@ def redirect_entity(self, entity): return "" return entity + def _log_unknown_entity(self, reference, curie, line_number): + """Raise an unknown-entity issue, distinguishing a missing reference.""" + if not self.issues: + return + if not reference: + self.issues.log_issue( + "entity", + "unknown entity - missing reference", + curie, + line_number=line_number, + ) + else: + self.issues.log_issue( + "entity", + "unknown entity", + curie, + line_number=line_number, + ) + if self.operational_issues: + self.operational_issues.log_issue( + "entity", + "unknown entity", + curie, + line_number=line_number, + ) + def process(self, stream): for block in stream: row = block["row"] @@ -187,15 +226,145 @@ def process(self, stream): class EntityLookupPhase(LookupPhase): + """ + lookup the entity for an entry's reference + + A resource is identified by the hash of its contents, so when several + organisations publish identical data the collection merges them into one + resource carrying every organisation's code (e.g. a joint local plan, or a + brownfield register published by two authorities). Those rows arrive here + with a blank organisation, because the resource-level default is only + applied when the resource has exactly one organisation. + + When given more than one organisation this phase resolves the entity once + per organisation and emits a row per distinct entity. With one organisation + (or none) it behaves exactly as it always has. + """ + entity_field = "entity" + def __init__( + self, + lookups={}, + redirect_lookups={}, + issue_log=None, + operational_issue_log=None, + entity_range=[], + lookup_rules=None, + providers=None, + ): + """ + Args: + providers (list, optional): Organisation codes that provided the + resource. More than one means the resource is shared, and the entity + is resolved once per organisation. Defaults to None (no fan-out). + """ + + super().__init__( + lookups=lookups, + redirect_lookups=redirect_lookups, + issue_log=issue_log, + operational_issue_log=operational_issue_log, + entity_range=entity_range, + lookup_rules=lookup_rules, + ) + self.providers = providers or [] + def process(self, stream): - for block in super().process(stream): + # With a single provider (or none) the row already carries the + # organisation, so the inherited behaviour is used unchanged. Only a + # resource shared by several organisations (e.g. a joint local plan) + # needs the entity resolving once per organisation. + if len(self.providers) <= 1: + for block in super().process(stream): + if self.issues: + self.issues.record_entity_map( + block["entry-number"], block["row"]["entity"] + ) + yield block + return + + for block in stream: + row = block["row"] + + # Nothing to look up if there is no prefix, or an entity has + # already been assigned upstream. + if not row.get("prefix", "") or row.get(self.entity_field, ""): + if self.issues: + self.issues.record_entity_map( + block["entry-number"], row.get(self.entity_field, "") + ) + yield block + continue + + yield from self._get_provider_entity_numbers(block) + + def _get_provider_entity_numbers(self, block): + """Resolve the entity for a row that has no organisation of its own. + + The reference is looked up once per provider, and distinct entities + produce distinct rows, each stamped with the organisation it was looked + up with; if two providers resolve to the same entity only one row is + emitted. A provider that resolves nothing simply produces no row, so an + entry may yield fewer rows than there are providers — including none. + + Some collections do map a source column to the organisation (e.g. + conservation-area's `borough`), and that organisation need not be one of + the providers — so when the row knows, it is resolved against that alone. + + An unknown-entity issue is raised only when NO organisation resolves an + entity. A resource shared by several organisations often holds one + organisation's rows alongside another's, so a reference missing for the + organisation that does not own it is normal, not a fault. Issues also + carry no organisation of their own and are fanned out to every + organisation on the resource when they become tasks, so a per-organisation + miss would tell both authorities their records were missing while the + entity sat correctly assigned. + + The organisation is set on the block as well as the row because + FactLookupPhase reads it after the pivot, once row-level fields are gone. + """ + + row = block["row"] + prefix = row.get("prefix", "") + reference = row.get("reference", "") + curie = f"{prefix}:{reference}" + line_number = block["line-number"] + + # Some collections map a source column to the organisation (e.g. + # conservation-area's `borough`), so a row may already say which + # authority it belongs to — and it need not be one of the providers. + # Trust the row when it knows; only try every provider when it doesn't. + organisations = ( + [row["organisation"]] if row.get("organisation", "") else self.providers + ) + + emitted_entities = set() + for organisation in organisations: + entity = self.get_entity(block, organisation=organisation) + + if not entity: + # this organisation does not own the reference: no row for it + continue + + entity = self.redirect_entity(entity) + if not entity or entity in emitted_entities: + continue + emitted_entities.add(entity) + + new_block = dict(block) + new_block["row"] = dict(row) + new_block["row"][self.entity_field] = entity + new_block["row"]["organisation"] = organisation + new_block["organisation"] = organisation if self.issues: - self.issues.record_entity_map( - block["entry-number"], block["row"]["entity"] - ) - yield block + self.issues.record_entity_map(block["entry-number"], entity) + yield new_block + + # no organisation resolved an entity: raise a single unknown-entity + # issue, as the inherited single-organisation path does + if not emitted_entities: + self._log_unknown_entity(reference, curie, line_number) class FactLookupPhase(LookupPhase): diff --git a/digital_land/pipeline/main.py b/digital_land/pipeline/main.py index c259ce24..829333e2 100644 --- a/digital_land/pipeline/main.py +++ b/digital_land/pipeline/main.py @@ -605,7 +605,9 @@ def transform( resource (str): Resource file identifier (hash), TBD can be removed. valid_category_values (dict): Dictionary of valid category values per field from the API/specification. endpoints (list, optional): List of endpoint hashes/identifiers for this resource. Defaults to None. - organisations (list, optional): List of organisation codes/identifiers associated with the resource. Defaults to None, Note if one passed, org will become default value + organisations (list, optional): List of organisation codes/identifiers associated with the resource. Defaults to None. + If one is passed it becomes the default organisation for every row; if more than one, no default is applied and + EntityLookupPhase resolves the entity once per organisation. entry_date (str, optional): Default entry-date value to apply to all records. Defaults to "". converted_path (str, optional): Path to save converted (pre-normalised) resource. Defaults to None. harmonised_output_path (str, optional): Path to save the harmonised/intermediate output. Defaults to None. @@ -713,6 +715,7 @@ def transform( operational_issue_log=self.operational_issue_log, entity_range=[entity_range_min, entity_range_max], lookup_rules=lookup_rules, + providers=organisations, ), ] ) diff --git a/tests/unit/phase/test_lookup.py b/tests/unit/phase/test_lookup.py index 7fe963fa..6d766a51 100644 --- a/tests/unit/phase/test_lookup.py +++ b/tests/unit/phase/test_lookup.py @@ -315,6 +315,187 @@ def test_entity_lookup_phase_blank(self): assert len(output) == 0 + def test_multi_org_fans_out_to_distinct_entities(self): + # One joint-plan row, two orgs, two different entities -> two rows out, + # each stamped with the org it was looked up with (row + block level). + input_stream = [ + { + "row": { + "prefix": "local-plan", + "reference": "owen", + "organisation": "", + }, + "entry-number": 1, + "line-number": 2, + } + ] + lookups = { + ",local-plan,owen,local-authorityoe": "101", + ",local-plan,owen,local-authoritytb": "102", + } + phase = EntityLookupPhase( + lookups=lookups, + providers=["local-authority:OE", "local-authority:TB"], + ) + output = [block for block in phase.process(input_stream)] + + assert len(output) == 2 + entity_to_org = {b["row"]["entity"]: b["row"]["organisation"] for b in output} + assert entity_to_org == { + "101": "local-authority:OE", + "102": "local-authority:TB", + } + # org also stamped at block level (FactLookupPhase reads this post-pivot) + assert {b["organisation"] for b in output} == { + "local-authority:OE", + "local-authority:TB", + } + + def test_multi_org_same_entity_produces_single_row(self): + # Both orgs resolve to the same entity -> one row, no dedupe-after needed. + input_stream = [ + { + "row": { + "prefix": "local-plan", + "reference": "owen", + "organisation": "", + }, + "entry-number": 1, + "line-number": 2, + } + ] + lookups = { + ",local-plan,owen,local-authorityoe": "101", + ",local-plan,owen,local-authoritytb": "101", + } + phase = EntityLookupPhase( + lookups=lookups, + providers=["local-authority:OE", "local-authority:TB"], + ) + output = [block for block in phase.process(input_stream)] + + assert len(output) == 1 + assert output[0]["row"]["entity"] == "101" + # first org to resolve the entity wins + assert output[0]["row"]["organisation"] == "local-authority:OE" + + def test_multi_org_one_missing_emits_found_without_issue(self): + # OE resolves, TB has no lookup -> one row for OE and NO issue. A shared + # resource routinely holds one organisation's rows alongside another's, + # so a reference missing for the organisation that does not own it is + # normal. Issues carry no organisation and are fanned out to every + # organisation on the resource as tasks, so raising one here would tell + # both authorities their records were missing. + input_stream = [ + { + "row": { + "prefix": "local-plan", + "reference": "owen", + "organisation": "", + }, + "entry-number": 1, + "line-number": 2, + } + ] + lookups = {",local-plan,owen,local-authorityoe": "101"} + issues = IssueLog() + phase = EntityLookupPhase( + lookups=lookups, + issue_log=issues, + providers=["local-authority:OE", "local-authority:TB"], + ) + output = [block for block in phase.process(input_stream)] + + assert len(output) == 1 + assert output[0]["row"]["entity"] == "101" + assert output[0]["row"]["organisation"] == "local-authority:OE" + assert issues.rows == [] + + def test_multi_org_neither_found_emits_nothing_and_raises_one_issue(self): + input_stream = [ + { + "row": { + "prefix": "local-plan", + "reference": "owen", + "organisation": "", + }, + "entry-number": 1, + "line-number": 2, + } + ] + lookups = {} + issues = IssueLog() + phase = EntityLookupPhase( + lookups=lookups, + issue_log=issues, + providers=["local-authority:OE", "local-authority:TB"], + ) + output = [block for block in phase.process(input_stream)] + + assert output == [] + # nothing resolved at all, so one issue for the entry -- not one per + # organisation, which would fan out into a task for each of them + assert [i["issue-type"] for i in issues.rows] == ["unknown entity"] + + def test_multi_org_row_with_own_organisation_is_not_fanned_out(self): + # Some collections map a source column to the organisation (e.g. + # conservation-area's `borough`), and it need not be one of the + # providers. The row knows which authority it belongs to, so it is + # resolved against that alone and the providers are never tried. + input_stream = [ + { + "row": { + "prefix": "local-plan", + "reference": "owen", + "organisation": "local-authority:XY", + }, + "entry-number": 1, + "line-number": 2, + } + ] + lookups = { + ",local-plan,owen,local-authorityoe": "101", + ",local-plan,owen,local-authoritytb": "102", + ",local-plan,owen,local-authorityxy": "103", + } + issues = IssueLog() + phase = EntityLookupPhase( + lookups=lookups, + issue_log=issues, + providers=["local-authority:OE", "local-authority:TB"], + ) + output = [block for block in phase.process(input_stream)] + + # one row for the row's own org, not one per provider + assert len(output) == 1 + assert output[0]["row"]["entity"] == "103" + assert output[0]["row"]["organisation"] == "local-authority:XY" + # the providers were never looked up, so nothing "missed" + assert issues.rows == [] + + def test_single_org_uses_unchanged_path(self): + # len(providers) <= 1 must not fan out; org read from the row as before. + input_stream = [ + { + "row": { + "prefix": "local-plan", + "reference": "owen", + "organisation": "local-authority:OE", + }, + "entry-number": 1, + "line-number": 2, + } + ] + lookups = {",local-plan,owen,local-authorityoe": "101"} + phase = EntityLookupPhase( + lookups=lookups, + providers=["local-authority:OE"], + ) + output = [block for block in phase.process(input_stream)] + + assert len(output) == 1 + assert output[0]["row"]["entity"] == "101" + class TestFactLookupPhase: def test_missing_associated_entity_issue_raised( diff --git a/tests/unit/test_check.py b/tests/unit/test_check.py index 383d0d75..3bf91c81 100644 --- a/tests/unit/test_check.py +++ b/tests/unit/test_check.py @@ -109,6 +109,25 @@ def test_duplicate_reference_check(tmp_path): }, ], ), + ( + "different_entity.csv", + [ + { + "entity": 7010002598, + "entry-date": "2024-07-19", + "entry-number": 1, + "field": "reference", + "value": "ref1", + }, + { + "entity": 7010002599, + "entry-date": "2024-07-19", + "entry-number": 1, + "field": "reference", + "value": "ref1", + }, + ], + ), ( "no_references.csv", [