From 7e5c71a5c4d96ece35c002fad49cf031e58ba9ff Mon Sep 17 00:00:00 2001 From: Gibah Joseph Date: Mon, 6 Jul 2026 12:48:03 +0100 Subject: [PATCH 01/12] feat: Implement duplicate entity redirection logic and add corresponding tests --- request-processor/Dockerfile | 2 +- .../requirements/requirements.in | 1 + .../requirements/requirements.txt | 2 + .../src/application/core/duplicates.py | 332 ++++++++++++++++++ .../src/application/core/pipeline.py | 16 + .../tests/unit/src/test_duplicates.py | 248 +++++++++++++ 6 files changed, 600 insertions(+), 1 deletion(-) create mode 100644 request-processor/src/application/core/duplicates.py create mode 100644 request-processor/tests/unit/src/test_duplicates.py diff --git a/request-processor/Dockerfile b/request-processor/Dockerfile index ad26bebc..ce6a88fc 100644 --- a/request-processor/Dockerfile +++ b/request-processor/Dockerfile @@ -7,7 +7,7 @@ RUN pip install --no-cache-dir cython RUN apt update && apt install -y --no-install-recommends \ gcc g++ make git libc-dev curl proj-bin libproj-dev procps\ - gdal-bin gdal-data awscli postgresql-client jq && \ + gdal-bin gdal-data awscli postgresql-client jq libsqlite3-mod-spatialite && \ rm -rf /var/lib/apt/lists/* diff --git a/request-processor/requirements/requirements.in b/request-processor/requirements/requirements.in index 19505063..82127415 100644 --- a/request-processor/requirements/requirements.in +++ b/request-processor/requirements/requirements.in @@ -7,6 +7,7 @@ cryptography==3.4.8 #Needed to control version of transient dependency used by m cchardet==2.1.7 httpx pre-commit +rapidfuzz==3.9.7 -e git+https://github.com/digital-land/pipeline.git#egg=digital-land shortuuid==1.0.13 sentry-sdk[celery]==2.44.0 diff --git a/request-processor/requirements/requirements.txt b/request-processor/requirements/requirements.txt index ec87361b..c5bd1831 100644 --- a/request-processor/requirements/requirements.txt +++ b/request-processor/requirements/requirements.txt @@ -241,6 +241,8 @@ pyyaml==6.0.2 # digital-land # pre-commit # responses +rapidfuzz==3.9.7 + # via -r requirements/requirements.in rdflib==7.1.4 # via sparqlwrapper requests==2.32.3 diff --git a/request-processor/src/application/core/duplicates.py b/request-processor/src/application/core/duplicates.py new file mode 100644 index 00000000..d555a007 --- /dev/null +++ b/request-processor/src/application/core/duplicates.py @@ -0,0 +1,332 @@ +import csv +import json +import os +import sqlite3 +import tempfile + +import requests +from rapidfuzz import fuzz + +from application.logging.logger import get_logger + +logger = get_logger(__name__) + +DATASETTE_BASE_URL = "https://datasette.planning.data.gov.uk" +REDIRECT_NOTE = "Redirect duplicate entity selected in Assign Entities" + + +def _normalise_entity_id(raw) -> str: + if raw is None or raw == "": + return "" + try: + return str(int(float(str(raw)))) + except (ValueError, TypeError): + return str(raw) + + +def _name_similarity(existing_name: str, new_name: str) -> str: + if not existing_name or not new_name: + return "" + similarity = fuzz.partial_ratio(existing_name.lower(), new_name.lower()) + return f"{round(similarity)}%" + + +def _read_provision_entities(transformed_csv_path: str) -> list: + if not os.path.exists(transformed_csv_path): + return [] + + entity_fields = { + "reference", + "name", + "geometry", + "point", + "organisation", + "entry-date", + "end-date", + } + entities = {} + with open(transformed_csv_path, "r", encoding="utf-8") as f: + for row in csv.DictReader(f): + entity = _normalise_entity_id(row.get("entity")) + field = row.get("field", "") + if not entity or field not in entity_fields: + continue + + entity_row = entities.setdefault(entity, {"entity": entity}) + entity_row[field] = row.get("value", "") + + return list(entities.values()) + + +def _organisation_row_for_provider(organisation_index, organisation_provider: str) -> dict: + if not organisation_index or not organisation_provider: + return {} + + try: + organisation = organisation_index.lookup(organisation_provider) + if not organisation: + return {} + return organisation_index.get(organisation) or {} + except (AttributeError, KeyError): + return {} + + +def _resolve_organisation_entity(organisation_index, organisation_provider: str) -> str: + return str( + _organisation_row_for_provider(organisation_index, organisation_provider).get( + "entity", "" + ) + ) + + +def _organisation_identifier_for_entity(organisation_index, organisation_entity: str) -> str: + if not organisation_index or not organisation_entity: + return "" + + try: + for row in organisation_index.organisation.values(): + if str(row.get("entity", "")) == str(organisation_entity): + return row.get("organisation") or row.get("reference", "") + except AttributeError: + return "" + + return "" + + +def _fetch_platform_entities(dataset: str, organisation_entity: str) -> list: + if not organisation_entity: + return [] + + url = f"{DATASETTE_BASE_URL.rstrip('/')}/{dataset}/entity.json" + response = requests.get( + url, + params={ + "_shape": "array", + "_size": "max", + "organisation_entity__exact": organisation_entity, + }, + timeout=120, + ) + response.raise_for_status() + data = response.json() + if isinstance(data, list): + return data + return data.get("rows") or [] + + +def _create_entity_table(conn: sqlite3.Connection, rows: list): + conn.execute( + """ + CREATE TABLE entity ( + entity INTEGER, + reference TEXT, + organisation_entity INTEGER, + geometry TEXT, + point TEXT, + name TEXT + ); + """ + ) + conn.executemany( + """ + INSERT INTO entity ( + entity, reference, organisation_entity, geometry, point, name + ) VALUES (?, ?, ?, ?, ?, ?); + """, + [ + ( + _normalise_entity_id(row.get("entity")), + row.get("reference", ""), + row.get("organisation_entity") + or row.get("organisation-entity") + or None, + row.get("geometry", ""), + row.get("point", ""), + row.get("name", ""), + ) + for row in rows + if _normalise_entity_id(row.get("entity")) + ], + ) + + +def _run_duplicate_check(rows: list, spatial_field: str) -> dict: + if not any(row.get(spatial_field) for row in rows): + return {"complete_matches": [], "single_matches": []} + + fd, path = tempfile.mkstemp(suffix=".sqlite3") + os.close(fd) + try: + import spatialite + from digital_land.expectations.operations.dataset import ( + duplicate_geometry_check, + ) + + with spatialite.connect(path) as conn: + _create_entity_table(conn, rows) + conn.commit() + _, _, details = duplicate_geometry_check(conn, spatial_field) + return { + "complete_matches": details.get("complete_matches", []), + "single_matches": details.get("single_matches", []), + } + finally: + try: + os.unlink(path) + except OSError: + pass + + +def _build_candidate( + *, + match: dict, + match_type: str, + spatial_field: str, + old_entity: dict, + new_entity: dict, + dataset: str, + organisation_index=None, + organisation_provider: str = "", +) -> dict: + existing_name = str(old_entity.get("name", "") or "") + new_name = str(new_entity.get("name", "") or "") + new_organisation = ( + _organisation_row_for_provider(organisation_index, organisation_provider).get( + "organisation", "" + ) + or str(new_entity.get("organisation", "") or "") + ) + old_organisation_entity = str( + old_entity.get("organisation_entity", "") + or match.get("organisation_entity_a", "") + ) + old_organisation = _organisation_identifier_for_entity( + organisation_index, old_organisation_entity + ) + evidence = [ + "point exact match" if spatial_field == "point" else f"geometry {match_type}" + ] + similarity = _name_similarity(existing_name, new_name) + if similarity: + evidence.append(f"name similarity {similarity}") + + redirect = { + "old_entity": _normalise_entity_id(old_entity.get("entity")), + "entity": _normalise_entity_id(new_entity.get("entity")), + "dataset": dataset, + "old_reference": str(old_entity.get("reference", "") or ""), + "new_reference": str(new_entity.get("reference", "") or ""), + "match_type": match_type, + "notes": REDIRECT_NOTE, + } + + return { + **redirect, + "old_name": existing_name, + "new_name": new_name, + "old_entry_date": str( + old_entity.get("entry_date", "") or old_entity.get("entry-date", "") or "" + ), + "new_entry_date": str( + new_entity.get("entry_date", "") or new_entity.get("entry-date", "") or "" + ), + "old_end_date": str( + old_entity.get("end_date", "") or old_entity.get("end-date", "") or "" + ), + "new_end_date": str( + new_entity.get("end_date", "") or new_entity.get("end-date", "") or "" + ), + "old_organisation": old_organisation, + "new_organisation": new_organisation, + "old_organisation_entity": old_organisation_entity, + "new_organisation_entity": str( + new_entity.get("organisation_entity", "") + or match.get("organisation_entity_b", "") + ), + "evidence": ", ".join(evidence), + "name_similarity": similarity, + "form_value": json.dumps(redirect, separators=(",", ":")), + } + + +def find_duplicate_redirect_candidates( + *, + dataset: str, + specification, + transformed_csv_path: str, + organisation_provider: str = "", + organisation_index=None, + fetch_platform_entities=_fetch_platform_entities, +) -> list[dict]: + if dataset != "conservation-area": + return [] + + if specification.get_dataset_typology(dataset) != "geography": + return [] + + provision_rows = _read_provision_entities(transformed_csv_path) + if not provision_rows: + return [] + + organisation_entity = _resolve_organisation_entity( + organisation_index, organisation_provider + ) + platform_rows = fetch_platform_entities(dataset, organisation_entity) + if not platform_rows: + return [] + + provision_by_id = { + _normalise_entity_id(row.get("entity")): row for row in provision_rows + } + platform_by_id = { + _normalise_entity_id(row.get("entity")): row for row in platform_rows + } + combined_rows = platform_rows + provision_rows + + matches = [] + for spatial_field in ("geometry", "point"): + field_matches = _run_duplicate_check(combined_rows, spatial_field) + matches.extend( + (match, "complete_match", spatial_field) + for match in field_matches.get("complete_matches", []) + ) + matches.extend( + (match, "single_match", spatial_field) + for match in field_matches.get("single_matches", []) + ) + + candidates = [] + seen = set() + for match, match_type, spatial_field in matches: + entity_a = _normalise_entity_id(match.get("entity_a")) + entity_b = _normalise_entity_id(match.get("entity_b")) + if entity_a in provision_by_id and entity_b in platform_by_id: + new_entity = provision_by_id[entity_a] + old_entity = platform_by_id[entity_b] + elif entity_b in provision_by_id and entity_a in platform_by_id: + new_entity = provision_by_id[entity_b] + old_entity = platform_by_id[entity_a] + else: + continue + + key = ( + _normalise_entity_id(old_entity.get("entity")), + _normalise_entity_id(new_entity.get("entity")), + ) + if key in seen or key[0] == key[1]: + continue + seen.add(key) + candidates.append( + _build_candidate( + match=match, + match_type=match_type, + spatial_field=spatial_field, + old_entity=old_entity, + new_entity=new_entity, + dataset=dataset, + organisation_index=organisation_index, + organisation_provider=organisation_provider, + ) + ) + + return candidates diff --git a/request-processor/src/application/core/pipeline.py b/request-processor/src/application/core/pipeline.py index 4eddcc93..0fa5ff72 100644 --- a/request-processor/src/application/core/pipeline.py +++ b/request-processor/src/application/core/pipeline.py @@ -11,6 +11,8 @@ from digital_land.commands import get_resource_unidentified_lookups from pathlib import Path +from application.core.duplicates import find_duplicate_redirect_candidates + logger = get_logger(__name__) @@ -381,6 +383,19 @@ def fetch_add_data_response( existing_entities_breakdown = _get_existing_entities_breakdown( existing_entities ) + try: + duplicate_candidates = find_duplicate_redirect_candidates( + dataset=dataset, + specification=specification, + transformed_csv_path=output_path, + organisation_provider=organisations[0], + organisation_index=organisation, + ) + except Exception as err: + logger.exception( + "Duplicate analysis failed for dataset %s: %s", dataset, err + ) + duplicate_candidates = [] if issues_log: issues_log.add_severity_column(severity_mapping=specification.issue_type) @@ -391,6 +406,7 @@ def fetch_add_data_response( "new-entities": new_entities_breakdown, "existing-entities": existing_entities_breakdown, "entity-organisation": entity_org_mapping, + "duplicate-candidates": duplicate_candidates, "pipeline-issues": ( [dict(issue) for issue in issues_log.rows] if issues_log else [] ), diff --git a/request-processor/tests/unit/src/test_duplicates.py b/request-processor/tests/unit/src/test_duplicates.py new file mode 100644 index 00000000..8ec0d30e --- /dev/null +++ b/request-processor/tests/unit/src/test_duplicates.py @@ -0,0 +1,248 @@ +import csv + +from application.core import duplicates + + +class FakeSpecification: + def __init__(self, typology="geography"): + self.typology = typology + + def get_dataset_typology(self, dataset): + return self.typology + + +class FakeOrganisationIndex: + organisation = { + "local-authority:STH": { + "entity": "318", + "reference": "STH", + "organisation": "local-authority:STH", + } + } + + def lookup(self, organisation): + return organisation + + def get(self, organisation): + return self.organisation[organisation] + + +def _write_transformed_csv(path): + rows = [ + { + "entry-number": "1", + "entity": "200", + "field": "reference", + "value": "new-ref", + }, + {"entry-number": "1", "entity": "200", "field": "name", "value": "New name"}, + { + "entry-number": "1", + "entity": "200", + "field": "entry-date", + "value": "2026-01-01", + }, + { + "entry-number": "1", + "entity": "200", + "field": "end-date", + "value": "", + }, + { + "entry-number": "1", + "entity": "200", + "field": "geometry", + "value": "POLYGON ((0 0, 0 1, 1 1, 1 0, 0 0))", + }, + ] + with open(path, "w", newline="", encoding="utf-8") as f: + writer = csv.DictWriter( + f, fieldnames=["entry-number", "entity", "field", "value"] + ) + writer.writeheader() + writer.writerows(rows) + + +def test_duplicate_candidates_are_provision_entities_against_existing_platform( + tmp_path, monkeypatch +): + transformed_path = tmp_path / "transformed.csv" + _write_transformed_csv(transformed_path) + + def fake_run_duplicate_check(rows, spatial_field): + if spatial_field == "point": + return {"complete_matches": [], "single_matches": []} + return { + "complete_matches": [ + { + "entity_a": "100", + "organisation_entity_a": "318", + "entity_b": "200", + "organisation_entity_b": "", + }, + { + "entity_a": "101", + "organisation_entity_a": "11", + "entity_b": "102", + "organisation_entity_b": "12", + }, + ], + "single_matches": [], + } + + def fake_fetch_platform_entities(dataset, organisation_entity): + assert dataset == "conservation-area" + assert organisation_entity == "318" + return [ + { + "entity": "100", + "reference": "old-ref", + "name": "Old name", + "entry_date": "2020-01-01", + "end_date": "", + "geometry": "POLYGON ((0 0, 0 1, 1 1, 1 0, 0 0))", + "organisation_entity": "318", + }, + {"entity": "101", "reference": "existing-a"}, + {"entity": "102", "reference": "existing-b"}, + ] + + monkeypatch.setattr(duplicates, "_run_duplicate_check", fake_run_duplicate_check) + + candidates = duplicates.find_duplicate_redirect_candidates( + dataset="conservation-area", + specification=FakeSpecification(), + transformed_csv_path=str(transformed_path), + organisation_provider="local-authority:STH", + organisation_index=FakeOrganisationIndex(), + fetch_platform_entities=fake_fetch_platform_entities, + ) + + assert len(candidates) == 1 + assert candidates[0]["old_entity"] == "100" + assert candidates[0]["entity"] == "200" + assert candidates[0]["old_reference"] == "old-ref" + assert candidates[0]["new_reference"] == "new-ref" + assert candidates[0]["old_entry_date"] == "2020-01-01" + assert candidates[0]["new_entry_date"] == "2026-01-01" + assert candidates[0]["old_end_date"] == "" + assert candidates[0]["new_end_date"] == "" + assert candidates[0]["old_organisation"] == "local-authority:STH" + assert candidates[0]["new_organisation"] == "local-authority:STH" + assert candidates[0]["match_type"] == "complete_match" + + +def test_fetch_platform_entities_reads_datasette_without_pagination(monkeypatch): + calls = [] + + class FakeResponse: + def raise_for_status(self): + pass + + def json(self): + return [ + { + "entity": "100", + "reference": "old-ref", + "organisation_entity": "318", + } + ] + + def fake_get(url, params, timeout): + calls.append({"url": url, "params": params, "timeout": timeout}) + return FakeResponse() + + monkeypatch.setattr(duplicates.requests, "get", fake_get) + + rows = duplicates._fetch_platform_entities("conservation-area", "318") + + assert rows[0]["entity"] == "100" + assert calls == [ + { + "url": ( + "https://datasette.planning.data.gov.uk/" + "conservation-area/entity.json" + ), + "params": { + "_shape": "array", + "_size": "max", + "organisation_entity__exact": "318", + }, + "timeout": 120, + } + ] + + +def test_name_similarity_uses_partial_ratio_for_added_words(): + assert ( + duplicates._name_similarity( + "South Jesmond", "South Jesmond Conservation Area" + ) + == "100%" + ) + + +def test_run_duplicate_check_commits_before_spatialite_metadata(): + rows = [ + { + "entity": "100", + "reference": "old-ref", + "geometry": "POLYGON ((0 0, 0 1, 1 1, 1 0, 0 0))", + }, + { + "entity": "200", + "reference": "new-ref", + "geometry": "POLYGON ((0 0, 0 1, 1 1, 1 0, 0 0))", + }, + ] + + matches = duplicates._run_duplicate_check(rows, "geometry") + + assert matches["complete_matches"][0]["entity_a"] == 100 + assert matches["complete_matches"][0]["entity_b"] == 200 + + +def test_duplicate_candidates_skip_non_conservation_area(tmp_path, monkeypatch): + transformed_path = tmp_path / "transformed.csv" + _write_transformed_csv(transformed_path) + monkeypatch.setattr( + duplicates, + "_run_duplicate_check", + lambda rows, spatial_field: (_ for _ in ()).throw(AssertionError()), + ) + + candidates = duplicates.find_duplicate_redirect_candidates( + dataset="tree", + specification=FakeSpecification(), + transformed_csv_path=str(transformed_path), + organisation_provider="local-authority:STH", + organisation_index=FakeOrganisationIndex(), + fetch_platform_entities=lambda dataset, organisation_entity: (_ for _ in ()).throw( + AssertionError() + ), + ) + + assert candidates == [] + + +def test_duplicate_candidates_skip_non_geography(tmp_path, monkeypatch): + transformed_path = tmp_path / "transformed.csv" + _write_transformed_csv(transformed_path) + monkeypatch.setattr( + duplicates, + "_run_duplicate_check", + lambda rows, spatial_field: (_ for _ in ()).throw(AssertionError()), + ) + + candidates = duplicates.find_duplicate_redirect_candidates( + dataset="article-4-direction", + specification=FakeSpecification(typology="legal-instrument"), + transformed_csv_path=str(transformed_path), + organisation_provider="local-authority:STH", + organisation_index=FakeOrganisationIndex(), + fetch_platform_entities=lambda dataset, organisation_entity: [ + {"entity": "100"} + ], + ) + + assert candidates == [] From bb4ebc6088d87d83cd3fd95ca73fc6122bca56a4 Mon Sep 17 00:00:00 2001 From: Gibah Joseph Date: Mon, 6 Jul 2026 14:01:40 +0100 Subject: [PATCH 02/12] refactor: Improve code readability and organization in duplicates.py; add tests for organisation entity mapping --- .../src/application/core/duplicates.py | 39 +++++++++---- .../tests/unit/src/test_duplicates.py | 56 ++++++++++++++++++- 2 files changed, 80 insertions(+), 15 deletions(-) diff --git a/request-processor/src/application/core/duplicates.py b/request-processor/src/application/core/duplicates.py index d555a007..1a9c9800 100644 --- a/request-processor/src/application/core/duplicates.py +++ b/request-processor/src/application/core/duplicates.py @@ -1,5 +1,4 @@ import csv -import json import os import sqlite3 import tempfile @@ -58,7 +57,9 @@ def _read_provision_entities(transformed_csv_path: str) -> list: return list(entities.values()) -def _organisation_row_for_provider(organisation_index, organisation_provider: str) -> dict: +def _organisation_row_for_provider( + organisation_index, organisation_provider: str +) -> dict: if not organisation_index or not organisation_provider: return {} @@ -79,7 +80,9 @@ def _resolve_organisation_entity(organisation_index, organisation_provider: str) ) -def _organisation_identifier_for_entity(organisation_index, organisation_entity: str) -> str: +def _organisation_identifier_for_entity( + organisation_index, organisation_entity: str +) -> str: if not organisation_index or not organisation_entity: return "" @@ -156,21 +159,24 @@ def _run_duplicate_check(rows: list, spatial_field: str) -> dict: fd, path = tempfile.mkstemp(suffix=".sqlite3") os.close(fd) + conn = None try: import spatialite from digital_land.expectations.operations.dataset import ( duplicate_geometry_check, ) - with spatialite.connect(path) as conn: - _create_entity_table(conn, rows) - conn.commit() - _, _, details = duplicate_geometry_check(conn, spatial_field) - return { - "complete_matches": details.get("complete_matches", []), - "single_matches": details.get("single_matches", []), - } + conn = spatialite.connect(path) + _create_entity_table(conn, rows) + conn.commit() + _, _, details = duplicate_geometry_check(conn, spatial_field) + return { + "complete_matches": details.get("complete_matches", []), + "single_matches": details.get("single_matches", []), + } finally: + if conn is not None: + conn.close() try: os.unlink(path) except OSError: @@ -185,6 +191,8 @@ def _build_candidate( old_entity: dict, new_entity: dict, dataset: str, + old_organisation_entity: str = "", + new_organisation_entity: str = "", organisation_index=None, organisation_provider: str = "", ) -> dict: @@ -198,6 +206,7 @@ def _build_candidate( ) old_organisation_entity = str( old_entity.get("organisation_entity", "") + or old_organisation_entity or match.get("organisation_entity_a", "") ) old_organisation = _organisation_identifier_for_entity( @@ -241,11 +250,11 @@ def _build_candidate( "old_organisation_entity": old_organisation_entity, "new_organisation_entity": str( new_entity.get("organisation_entity", "") + or new_organisation_entity or match.get("organisation_entity_b", "") ), "evidence": ", ".join(evidence), "name_similarity": similarity, - "form_value": json.dumps(redirect, separators=(",", ":")), } @@ -303,9 +312,13 @@ def find_duplicate_redirect_candidates( if entity_a in provision_by_id and entity_b in platform_by_id: new_entity = provision_by_id[entity_a] old_entity = platform_by_id[entity_b] + old_organisation_entity = match.get("organisation_entity_b", "") + new_organisation_entity = match.get("organisation_entity_a", "") elif entity_b in provision_by_id and entity_a in platform_by_id: new_entity = provision_by_id[entity_b] old_entity = platform_by_id[entity_a] + old_organisation_entity = match.get("organisation_entity_a", "") + new_organisation_entity = match.get("organisation_entity_b", "") else: continue @@ -324,6 +337,8 @@ def find_duplicate_redirect_candidates( old_entity=old_entity, new_entity=new_entity, dataset=dataset, + old_organisation_entity=old_organisation_entity, + new_organisation_entity=new_organisation_entity, organisation_index=organisation_index, organisation_provider=organisation_provider, ) diff --git a/request-processor/tests/unit/src/test_duplicates.py b/request-processor/tests/unit/src/test_duplicates.py index 8ec0d30e..cf033775 100644 --- a/request-processor/tests/unit/src/test_duplicates.py +++ b/request-processor/tests/unit/src/test_duplicates.py @@ -129,9 +129,59 @@ def fake_fetch_platform_entities(dataset, organisation_entity): assert candidates[0]["new_end_date"] == "" assert candidates[0]["old_organisation"] == "local-authority:STH" assert candidates[0]["new_organisation"] == "local-authority:STH" + assert candidates[0]["old_organisation_entity"] == "318" + assert candidates[0]["new_organisation_entity"] == "" assert candidates[0]["match_type"] == "complete_match" +def test_duplicate_candidates_map_organisation_entities_when_new_entity_is_a( + tmp_path, monkeypatch +): + transformed_path = tmp_path / "transformed.csv" + _write_transformed_csv(transformed_path) + + def fake_run_duplicate_check(rows, spatial_field): + if spatial_field == "point": + return {"complete_matches": [], "single_matches": []} + return { + "complete_matches": [ + { + "entity_a": "200", + "organisation_entity_a": "", + "entity_b": "100", + "organisation_entity_b": "318", + }, + ], + "single_matches": [], + } + + def fake_fetch_platform_entities(dataset, organisation_entity): + return [ + { + "entity": "100", + "reference": "old-ref", + "name": "Old name", + "geometry": "POLYGON ((0 0, 0 1, 1 1, 1 0, 0 0))", + }, + ] + + monkeypatch.setattr(duplicates, "_run_duplicate_check", fake_run_duplicate_check) + + candidates = duplicates.find_duplicate_redirect_candidates( + dataset="conservation-area", + specification=FakeSpecification(), + transformed_csv_path=str(transformed_path), + organisation_provider="local-authority:STH", + organisation_index=FakeOrganisationIndex(), + fetch_platform_entities=fake_fetch_platform_entities, + ) + + assert candidates[0]["old_entity"] == "100" + assert candidates[0]["entity"] == "200" + assert candidates[0]["old_organisation_entity"] == "318" + assert candidates[0]["new_organisation_entity"] == "" + + def test_fetch_platform_entities_reads_datasette_without_pagination(monkeypatch): calls = [] @@ -217,9 +267,9 @@ def test_duplicate_candidates_skip_non_conservation_area(tmp_path, monkeypatch): transformed_csv_path=str(transformed_path), organisation_provider="local-authority:STH", organisation_index=FakeOrganisationIndex(), - fetch_platform_entities=lambda dataset, organisation_entity: (_ for _ in ()).throw( - AssertionError() - ), + fetch_platform_entities=lambda dataset, organisation_entity: ( + _ for _ in () + ).throw(AssertionError()), ) assert candidates == [] From 0059ab2819243fbaeddb285024e97f6bb2f02513 Mon Sep 17 00:00:00 2001 From: Gibah Joseph Date: Mon, 6 Jul 2026 14:13:52 +0100 Subject: [PATCH 03/12] refactor: Simplify organisation retrieval in _build_candidate and improve test readability for name similarity --- request-processor/src/application/core/duplicates.py | 9 +++------ request-processor/tests/unit/src/test_duplicates.py | 4 +--- 2 files changed, 4 insertions(+), 9 deletions(-) diff --git a/request-processor/src/application/core/duplicates.py b/request-processor/src/application/core/duplicates.py index 1a9c9800..c4f9fdab 100644 --- a/request-processor/src/application/core/duplicates.py +++ b/request-processor/src/application/core/duplicates.py @@ -198,12 +198,9 @@ def _build_candidate( ) -> dict: existing_name = str(old_entity.get("name", "") or "") new_name = str(new_entity.get("name", "") or "") - new_organisation = ( - _organisation_row_for_provider(organisation_index, organisation_provider).get( - "organisation", "" - ) - or str(new_entity.get("organisation", "") or "") - ) + new_organisation = _organisation_row_for_provider( + organisation_index, organisation_provider + ).get("organisation", "") or str(new_entity.get("organisation", "") or "") old_organisation_entity = str( old_entity.get("organisation_entity", "") or old_organisation_entity diff --git a/request-processor/tests/unit/src/test_duplicates.py b/request-processor/tests/unit/src/test_duplicates.py index cf033775..3ea42a3a 100644 --- a/request-processor/tests/unit/src/test_duplicates.py +++ b/request-processor/tests/unit/src/test_duplicates.py @@ -225,9 +225,7 @@ def fake_get(url, params, timeout): def test_name_similarity_uses_partial_ratio_for_added_words(): assert ( - duplicates._name_similarity( - "South Jesmond", "South Jesmond Conservation Area" - ) + duplicates._name_similarity("South Jesmond", "South Jesmond Conservation Area") == "100%" ) From 49cecf4b3415322a6d647c1260e35b92ef716840 Mon Sep 17 00:00:00 2001 From: Gibah Joseph Date: Tue, 7 Jul 2026 11:29:50 +0100 Subject: [PATCH 04/12] feat: Add organisation entity handling in duplicate candidate matching and update tests --- .../src/application/core/duplicates.py | 48 +++++++++++++------ .../tests/unit/src/test_duplicates.py | 23 +++++---- 2 files changed, 48 insertions(+), 23 deletions(-) diff --git a/request-processor/src/application/core/duplicates.py b/request-processor/src/application/core/duplicates.py index c4f9fdab..c2895edb 100644 --- a/request-processor/src/application/core/duplicates.py +++ b/request-processor/src/application/core/duplicates.py @@ -40,6 +40,8 @@ def _read_provision_entities(transformed_csv_path: str) -> list: "geometry", "point", "organisation", + "organisation_entity", + "organisation-entity", "entry-date", "end-date", } @@ -255,6 +257,28 @@ def _build_candidate( } +def _entities_for_match(match: dict, provision_by_id: dict, platform_by_id: dict): + entity_a = _normalise_entity_id(match.get("entity_a")) + entity_b = _normalise_entity_id(match.get("entity_b")) + if entity_a in provision_by_id and entity_b in platform_by_id: + return { + "new_entity": provision_by_id[entity_a], + "old_entity": platform_by_id[entity_b], + "old_organisation_entity": match.get("organisation_entity_b", ""), + "new_organisation_entity": match.get("organisation_entity_a", ""), + } + + if entity_b in provision_by_id and entity_a in platform_by_id: + return { + "new_entity": provision_by_id[entity_b], + "old_entity": platform_by_id[entity_a], + "old_organisation_entity": match.get("organisation_entity_a", ""), + "new_organisation_entity": match.get("organisation_entity_b", ""), + } + + return None + + def find_duplicate_redirect_candidates( *, dataset: str, @@ -277,6 +301,9 @@ def find_duplicate_redirect_candidates( organisation_entity = _resolve_organisation_entity( organisation_index, organisation_provider ) + for row in provision_rows: + row.setdefault("organisation_entity", organisation_entity) + platform_rows = fetch_platform_entities(dataset, organisation_entity) if not platform_rows: return [] @@ -304,20 +331,11 @@ def find_duplicate_redirect_candidates( candidates = [] seen = set() for match, match_type, spatial_field in matches: - entity_a = _normalise_entity_id(match.get("entity_a")) - entity_b = _normalise_entity_id(match.get("entity_b")) - if entity_a in provision_by_id and entity_b in platform_by_id: - new_entity = provision_by_id[entity_a] - old_entity = platform_by_id[entity_b] - old_organisation_entity = match.get("organisation_entity_b", "") - new_organisation_entity = match.get("organisation_entity_a", "") - elif entity_b in provision_by_id and entity_a in platform_by_id: - new_entity = provision_by_id[entity_b] - old_entity = platform_by_id[entity_a] - old_organisation_entity = match.get("organisation_entity_a", "") - new_organisation_entity = match.get("organisation_entity_b", "") - else: + matched_entities = _entities_for_match(match, provision_by_id, platform_by_id) + if not matched_entities: continue + new_entity = matched_entities["new_entity"] + old_entity = matched_entities["old_entity"] key = ( _normalise_entity_id(old_entity.get("entity")), @@ -334,8 +352,8 @@ def find_duplicate_redirect_candidates( old_entity=old_entity, new_entity=new_entity, dataset=dataset, - old_organisation_entity=old_organisation_entity, - new_organisation_entity=new_organisation_entity, + old_organisation_entity=matched_entities["old_organisation_entity"], + new_organisation_entity=matched_entities["new_organisation_entity"], organisation_index=organisation_index, organisation_provider=organisation_provider, ) diff --git a/request-processor/tests/unit/src/test_duplicates.py b/request-processor/tests/unit/src/test_duplicates.py index 3ea42a3a..f01ca1e7 100644 --- a/request-processor/tests/unit/src/test_duplicates.py +++ b/request-processor/tests/unit/src/test_duplicates.py @@ -12,13 +12,14 @@ def get_dataset_typology(self, dataset): class FakeOrganisationIndex: - organisation = { - "local-authority:STH": { - "entity": "318", - "reference": "STH", - "organisation": "local-authority:STH", + def __init__(self): + self.organisation = { + "local-authority:STH": { + "entity": "318", + "reference": "STH", + "organisation": "local-authority:STH", + } } - } def lookup(self, organisation): return organisation @@ -70,6 +71,9 @@ def test_duplicate_candidates_are_provision_entities_against_existing_platform( _write_transformed_csv(transformed_path) def fake_run_duplicate_check(rows, spatial_field): + provision_row = next(row for row in rows if row["entity"] == "200") + assert provision_row["organisation_entity"] == "318" + if spatial_field == "point": return {"complete_matches": [], "single_matches": []} return { @@ -130,7 +134,7 @@ def fake_fetch_platform_entities(dataset, organisation_entity): assert candidates[0]["old_organisation"] == "local-authority:STH" assert candidates[0]["new_organisation"] == "local-authority:STH" assert candidates[0]["old_organisation_entity"] == "318" - assert candidates[0]["new_organisation_entity"] == "" + assert candidates[0]["new_organisation_entity"] == "318" assert candidates[0]["match_type"] == "complete_match" @@ -141,6 +145,9 @@ def test_duplicate_candidates_map_organisation_entities_when_new_entity_is_a( _write_transformed_csv(transformed_path) def fake_run_duplicate_check(rows, spatial_field): + provision_row = next(row for row in rows if row["entity"] == "200") + assert provision_row["organisation_entity"] == "318" + if spatial_field == "point": return {"complete_matches": [], "single_matches": []} return { @@ -179,7 +186,7 @@ def fake_fetch_platform_entities(dataset, organisation_entity): assert candidates[0]["old_entity"] == "100" assert candidates[0]["entity"] == "200" assert candidates[0]["old_organisation_entity"] == "318" - assert candidates[0]["new_organisation_entity"] == "" + assert candidates[0]["new_organisation_entity"] == "318" def test_fetch_platform_entities_reads_datasette_without_pagination(monkeypatch): From 356853f46c3d7f51255ca5dbc37a454ee1096bbe Mon Sep 17 00:00:00 2001 From: Gibah Joseph Date: Tue, 7 Jul 2026 11:45:16 +0100 Subject: [PATCH 05/12] refactor: Implement transformation and processing logic for add data resources, including duplicate candidate analysis --- .../src/application/core/pipeline.py | 211 +++++++++++------- 1 file changed, 127 insertions(+), 84 deletions(-) diff --git a/request-processor/src/application/core/pipeline.py b/request-processor/src/application/core/pipeline.py index 0fa5ff72..b40babe6 100644 --- a/request-processor/src/application/core/pipeline.py +++ b/request-processor/src/application/core/pipeline.py @@ -256,6 +256,101 @@ def _assign_entries( return [] +def _transform_add_data_resource( + *, + pipeline, + resource_file_path, + output_path, + organisation, + organisations, + valid_category_values, + converted_path, + endpoints, + dataset, + pipeline_dir, + specification, + cache_dir, +): + issues_log = pipeline.transform( + input_path=resource_file_path, + output_path=output_path, + organisation=organisation, + organisations=organisations, + resource=resource_from_path(resource_file_path), + valid_category_values=valid_category_values, + disable_lookups=False, + endpoints=endpoints, + converted_path=converted_path, + ) + + existing_entities = _map_transformed_entities(output_path, pipeline_dir) + unknown_issue_types = { + "unknown entity", + "unknown entity - missing reference", + } + has_unknown = any( + row.get("issue-type") in unknown_issue_types + for row in issues_log.rows + if isinstance(row, dict) + ) + + if not has_unknown: + return pipeline, issues_log, existing_entities, [], [] + + new_lookups = _assign_entries( + resource_path=resource_file_path, + dataset=dataset, + organisation=organisations[0], + pipeline_dir=pipeline_dir, + specification=specification, + cache_dir=cache_dir, + endpoints=endpoints if endpoints else None, + ) + entity_org_mapping = _create_entity_organisation( + new_lookups, dataset, organisations[0] + ) + + # Reload pipeline to pick up newly saved lookups before rerunning transform. + pipeline = Pipeline(pipeline_dir, dataset, specification=specification) + issues_log = pipeline.transform( + input_path=resource_file_path, + output_path=output_path, + organisation=organisation, + organisations=organisations, + resource=resource_from_path(resource_file_path), + valid_category_values=valid_category_values, + disable_lookups=False, + endpoints=endpoints, + converted_path=converted_path, + ) + + return pipeline, issues_log, existing_entities, new_lookups, entity_org_mapping + + +def _process_add_data_resource(resource_file, **kwargs): + try: + return _transform_add_data_resource(**kwargs) + except Exception as err: + logger.exception(f"Error processing {resource_file}: {err}") + raise + + +def _find_duplicate_candidates( + dataset, specification, output_path, organisation_provider, organisation_index +): + try: + return find_duplicate_redirect_candidates( + dataset=dataset, + specification=specification, + transformed_csv_path=output_path, + organisation_provider=organisation_provider, + organisation_index=organisation_index, + ) + except Exception as err: + logger.exception("Duplicate analysis failed for dataset %s: %s", dataset, err) + return [] + + def fetch_add_data_response( dataset, organisation_provider, @@ -305,97 +400,45 @@ def fetch_add_data_response( logger.info( f"Processing file {idx + 1}/{len(files_in_resource)}: {resource_file}" ) - try: - # Try add data with pipeline transform to see if no entities found - issues_log = pipeline.transform( - input_path=resource_file_path, - output_path=output_path, - organisation=organisation, - organisations=organisations, - resource=resource_from_path(resource_file_path), - valid_category_values=valid_category_values, - disable_lookups=False, - endpoints=endpoints, - converted_path=converted_path, - ) - - existing_entities.extend( - _map_transformed_entities(output_path, pipeline_dir) - ) + ( + pipeline, + issues_log, + transformed_entities, + new_lookups, + new_entity_org_mapping, + ) = _process_add_data_resource( + resource_file, + pipeline=pipeline, + resource_file_path=resource_file_path, + output_path=output_path, + organisation=organisation, + organisations=organisations, + valid_category_values=valid_category_values, + converted_path=converted_path, + endpoints=endpoints, + dataset=dataset, + pipeline_dir=pipeline_dir, + specification=specification, + cache_dir=cache_dir, + ) - # Check if there are unknown entity issues in the log - unknown_issue_types = { - "unknown entity", - "unknown entity - missing reference", - } - has_unknown = any( - row.get("issue-type") in unknown_issue_types - for row in issues_log.rows - if isinstance(row, dict) + existing_entities.extend(transformed_entities) + if new_lookups: + logger.info( + f"Found {len(new_lookups)} unidentified lookups in {resource_file}" ) - - if has_unknown: - new_lookups = _assign_entries( - resource_path=resource_file_path, - dataset=dataset, - organisation=organisations[0], - pipeline_dir=pipeline_dir, - specification=specification, - cache_dir=cache_dir, - endpoints=endpoints if endpoints else None, - ) - logger.info( - f"Found {len(new_lookups)} unidentified lookups in {resource_file}" - ) - new_entities.extend(new_lookups) - - # Default create a entity-organisation mapping, front end can use the 'authoritative' flag - entity_org_mapping = _create_entity_organisation( - new_lookups, dataset, organisations[0] - ) - # TODO, save to pipeline as well for rerun? - - # Reload pipeline to pick up newly saved lookups - pipeline = Pipeline( - pipeline_dir, dataset, specification=specification - ) - - # Now re-run transform to check and return issue log - issues_log = pipeline.transform( - input_path=resource_file_path, - output_path=output_path, - organisation=organisation, - organisations=organisations, - resource=resource_from_path(resource_file_path), - valid_category_values=valid_category_values, - disable_lookups=False, - endpoints=endpoints, - converted_path=converted_path, - ) - else: - logger.info(f"No unidentified lookups found in {resource_file}") - - except Exception as err: - logger.exception(f"Error processing {resource_file}: {err}") - raise + new_entities.extend(new_lookups) + entity_org_mapping = new_entity_org_mapping + else: + logger.info(f"No unidentified lookups found in {resource_file}") new_entities_breakdown = _get_entities_breakdown(new_entities) existing_entities_breakdown = _get_existing_entities_breakdown( existing_entities ) - try: - duplicate_candidates = find_duplicate_redirect_candidates( - dataset=dataset, - specification=specification, - transformed_csv_path=output_path, - organisation_provider=organisations[0], - organisation_index=organisation, - ) - except Exception as err: - logger.exception( - "Duplicate analysis failed for dataset %s: %s", dataset, err - ) - duplicate_candidates = [] + duplicate_candidates = _find_duplicate_candidates( + dataset, specification, output_path, organisations[0], organisation + ) if issues_log: issues_log.add_severity_column(severity_mapping=specification.issue_type) From f73642b156178c6679c85aa5c69f0e491d8f8cc5 Mon Sep 17 00:00:00 2001 From: Gibah Joseph Date: Tue, 7 Jul 2026 12:10:44 +0100 Subject: [PATCH 06/12] fix: Add severity column handling in issue logs and refactor related code --- .../requirements/test_requirements.txt | 2 ++ request-processor/src/application/core/pipeline.py | 13 +++++++++---- 2 files changed, 11 insertions(+), 4 deletions(-) diff --git a/request-processor/requirements/test_requirements.txt b/request-processor/requirements/test_requirements.txt index 100a3cce..654ac6ed 100644 --- a/request-processor/requirements/test_requirements.txt +++ b/request-processor/requirements/test_requirements.txt @@ -417,6 +417,8 @@ pyyaml==6.0.2 # digital-land # pre-commit # responses +rapidfuzz==3.9.7 + # via -r requirements/requirements.txt rdflib==7.1.4 # via # -r requirements/requirements.txt diff --git a/request-processor/src/application/core/pipeline.py b/request-processor/src/application/core/pipeline.py index b40babe6..add9fb1a 100644 --- a/request-processor/src/application/core/pipeline.py +++ b/request-processor/src/application/core/pipeline.py @@ -67,8 +67,6 @@ def run_task_pipeline( dataset=dataset, organisation=organisation, issue_path=issue_path, - column_field_path=column_field_path, - mandatory_fields=mandatory_fields, ) if status == TaskPipelineStatus.FAILED: raise RuntimeError(f"TaskPipeline failed for dataset '{dataset}'") @@ -86,6 +84,13 @@ def run_task_pipeline( return task_log +def _add_severity_column(issue_log, specification): + severity_mapping_path = getattr(specification, "issue_type", None) or os.path.join( + specification.path, "issue-type.csv" + ) + issue_log.add_severity_column(severity_mapping_path) + + def fetch_response_data( dataset, organisation, @@ -166,7 +171,7 @@ def fetch_response_data( disable_lookups=True, ) # Issue log needs severity column added, so manually added and saved here - issue_log.add_severity_column(severity_mapping=specification.issue_type) + _add_severity_column(issue_log, specification) issue_log.save( os.path.join(issue_dir, dataset, request_id, resource + ".csv") ) @@ -441,7 +446,7 @@ def fetch_add_data_response( ) if issues_log: - issues_log.add_severity_column(severity_mapping=specification.issue_type) + _add_severity_column(issues_log, specification) pipeline_summary = { "new-in-resource": len(new_entities), From 749da0632ad1fa97eaa81a0b9f996bc25f1f5311 Mon Sep 17 00:00:00 2001 From: Gibah Joseph Date: Tue, 7 Jul 2026 12:18:49 +0100 Subject: [PATCH 07/12] test: Enhance duplicate check tests with SQLite mock and geometry validation --- .../tests/unit/src/test_duplicates.py | 45 ++++++++++++++++++- 1 file changed, 44 insertions(+), 1 deletion(-) diff --git a/request-processor/tests/unit/src/test_duplicates.py b/request-processor/tests/unit/src/test_duplicates.py index f01ca1e7..b907fa92 100644 --- a/request-processor/tests/unit/src/test_duplicates.py +++ b/request-processor/tests/unit/src/test_duplicates.py @@ -1,4 +1,7 @@ import csv +import sqlite3 +import sys +import types from application.core import duplicates @@ -237,7 +240,7 @@ def test_name_similarity_uses_partial_ratio_for_added_words(): ) -def test_run_duplicate_check_commits_before_spatialite_metadata(): +def test_run_duplicate_check_commits_before_spatialite_metadata(monkeypatch): rows = [ { "entity": "100", @@ -251,6 +254,46 @@ def test_run_duplicate_check_commits_before_spatialite_metadata(): }, ] + class FakeConnection: + def __init__(self, path): + self.conn = sqlite3.connect(path) + self.committed = False + + def execute(self, *args, **kwargs): + return self.conn.execute(*args, **kwargs) + + def executemany(self, *args, **kwargs): + return self.conn.executemany(*args, **kwargs) + + def commit(self): + self.committed = True + return self.conn.commit() + + def close(self): + return self.conn.close() + + def fake_duplicate_geometry_check(conn, spatial_field): + assert conn.committed + assert spatial_field == "geometry" + return ( + None, + None, + { + "complete_matches": [{"entity_a": 100, "entity_b": 200}], + "single_matches": [], + }, + ) + + monkeypatch.setitem( + sys.modules, + "spatialite", + types.SimpleNamespace(connect=lambda path: FakeConnection(path)), + ) + monkeypatch.setattr( + "digital_land.expectations.operations.dataset.duplicate_geometry_check", + fake_duplicate_geometry_check, + ) + matches = duplicates._run_duplicate_check(rows, "geometry") assert matches["complete_matches"][0]["entity_a"] == 100 From 6acd48cf8771605bf3f11ba1c9b62c512815f1cc Mon Sep 17 00:00:00 2001 From: Gibah Joseph Date: Tue, 7 Jul 2026 12:53:59 +0100 Subject: [PATCH 08/12] refactor: Replace custom severity column addition with direct method call in issue logs --- request-processor/src/application/core/pipeline.py | 11 ++--------- 1 file changed, 2 insertions(+), 9 deletions(-) diff --git a/request-processor/src/application/core/pipeline.py b/request-processor/src/application/core/pipeline.py index add9fb1a..c298fee7 100644 --- a/request-processor/src/application/core/pipeline.py +++ b/request-processor/src/application/core/pipeline.py @@ -84,13 +84,6 @@ def run_task_pipeline( return task_log -def _add_severity_column(issue_log, specification): - severity_mapping_path = getattr(specification, "issue_type", None) or os.path.join( - specification.path, "issue-type.csv" - ) - issue_log.add_severity_column(severity_mapping_path) - - def fetch_response_data( dataset, organisation, @@ -171,7 +164,7 @@ def fetch_response_data( disable_lookups=True, ) # Issue log needs severity column added, so manually added and saved here - _add_severity_column(issue_log, specification) + issue_log.add_severity_column(severity_mapping=specification.issue_type) issue_log.save( os.path.join(issue_dir, dataset, request_id, resource + ".csv") ) @@ -446,7 +439,7 @@ def fetch_add_data_response( ) if issues_log: - _add_severity_column(issues_log, specification) + issues_log.add_severity_column(severity_mapping=specification.issue_type) pipeline_summary = { "new-in-resource": len(new_entities), From c9699704a2d9dbe7f0ed97cfb7e3f5827407df20 Mon Sep 17 00:00:00 2001 From: Gibah Joseph Date: Tue, 7 Jul 2026 13:33:47 +0100 Subject: [PATCH 09/12] fix: Add column_field_path and mandatory_fields parameters to run_task_pipeline function --- request-processor/src/application/core/pipeline.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/request-processor/src/application/core/pipeline.py b/request-processor/src/application/core/pipeline.py index c298fee7..b40babe6 100644 --- a/request-processor/src/application/core/pipeline.py +++ b/request-processor/src/application/core/pipeline.py @@ -67,6 +67,8 @@ def run_task_pipeline( dataset=dataset, organisation=organisation, issue_path=issue_path, + column_field_path=column_field_path, + mandatory_fields=mandatory_fields, ) if status == TaskPipelineStatus.FAILED: raise RuntimeError(f"TaskPipeline failed for dataset '{dataset}'") From 3b72c00994bb1810ae4d40f1338333ea3768fa96 Mon Sep 17 00:00:00 2001 From: Gibah Joseph Date: Tue, 7 Jul 2026 14:17:48 +0100 Subject: [PATCH 10/12] test: Add test for run_task_pipeline to validate column_field_path and mandatory_fields parameters --- .../src/application/core/test_pipeline.py | 37 +++++++++++++++++++ 1 file changed, 37 insertions(+) diff --git a/request-processor/tests/unit/src/application/core/test_pipeline.py b/request-processor/tests/unit/src/application/core/test_pipeline.py index 60f2832c..a0f14eac 100644 --- a/request-processor/tests/unit/src/application/core/test_pipeline.py +++ b/request-processor/tests/unit/src/application/core/test_pipeline.py @@ -614,6 +614,43 @@ def test_run_task_pipeline_raises_on_failed_status(tmp_path, monkeypatch): ) +def test_run_task_pipeline_passes_column_field_path_and_mandatory_fields( + tmp_path, monkeypatch +): + from digital_land.pipeline.task import TaskPipelineStatus + + mock_pipeline = MagicMock() + mock_pipeline.run.return_value = TaskPipelineStatus.COMPLETE + monkeypatch.setattr( + "src.application.core.pipeline.TaskPipeline", lambda: mock_pipeline + ) + monkeypatch.setattr("src.application.core.pipeline.load_mappings", lambda: {}) + + task_log_path = str(tmp_path / "tasks.csv") + issue_path = str(tmp_path / "issues.csv") + column_field_path = str(tmp_path / "column-field.csv") + mandatory_fields = ["reference", "name"] + + result = run_task_pipeline( + task_log_path=task_log_path, + dataset="conservation-area", + organisation="local-authority:CTY", + issue_path=issue_path, + column_field_path=column_field_path, + mandatory_fields=mandatory_fields, + ) + + mock_pipeline.run.assert_called_once_with( + output_path=task_log_path, + dataset="conservation-area", + organisation="local-authority:CTY", + issue_path=issue_path, + column_field_path=column_field_path, + mandatory_fields=mandatory_fields, + ) + assert result == [] + + def test_run_task_pipeline_empty_issue_path_returns_empty(tmp_path, monkeypatch): task_log_path = str(tmp_path / "tasks.csv") monkeypatch.setattr("src.application.core.pipeline.load_mappings", lambda: {}) From 788380930f9a9e6882888fdaa3fdd54dab99b725 Mon Sep 17 00:00:00 2001 From: Gibah Joseph Date: Thu, 9 Jul 2026 09:50:17 +0100 Subject: [PATCH 11/12] fix: Add redirect lookups support to find_duplicate_redirect_candidates and related tests --- .../src/application/core/duplicates.py | 31 +++++++++++-------- .../src/application/core/pipeline.py | 15 +++++++-- .../src/application/core/test_pipeline.py | 26 ++++++++++++++++ .../tests/unit/src/test_duplicates.py | 11 +++++++ 4 files changed, 68 insertions(+), 15 deletions(-) diff --git a/request-processor/src/application/core/duplicates.py b/request-processor/src/application/core/duplicates.py index c2895edb..74f833eb 100644 --- a/request-processor/src/application/core/duplicates.py +++ b/request-processor/src/application/core/duplicates.py @@ -284,6 +284,7 @@ def find_duplicate_redirect_candidates( dataset: str, specification, transformed_csv_path: str, + redirect_lookups: dict | None = None, organisation_provider: str = "", organisation_index=None, fetch_platform_entities=_fetch_platform_entities, @@ -315,6 +316,7 @@ def find_duplicate_redirect_candidates( _normalise_entity_id(row.get("entity")): row for row in platform_rows } combined_rows = platform_rows + provision_rows + redirect_lookups = redirect_lookups or {} matches = [] for spatial_field in ("geometry", "point"): @@ -344,19 +346,22 @@ def find_duplicate_redirect_candidates( if key in seen or key[0] == key[1]: continue seen.add(key) - candidates.append( - _build_candidate( - match=match, - match_type=match_type, - spatial_field=spatial_field, - old_entity=old_entity, - new_entity=new_entity, - dataset=dataset, - old_organisation_entity=matched_entities["old_organisation_entity"], - new_organisation_entity=matched_entities["new_organisation_entity"], - organisation_index=organisation_index, - organisation_provider=organisation_provider, - ) + candidate = _build_candidate( + match=match, + match_type=match_type, + spatial_field=spatial_field, + old_entity=old_entity, + new_entity=new_entity, + dataset=dataset, + old_organisation_entity=matched_entities["old_organisation_entity"], + new_organisation_entity=matched_entities["new_organisation_entity"], + organisation_index=organisation_index, + organisation_provider=organisation_provider, + ) + existing_redirect = redirect_lookups.get(key[0]) + candidate["old_entity_redirects"] = ( + [{"old-entity": key[0], **existing_redirect}] if existing_redirect else [] ) + candidates.append(candidate) return candidates diff --git a/request-processor/src/application/core/pipeline.py b/request-processor/src/application/core/pipeline.py index b40babe6..7087f896 100644 --- a/request-processor/src/application/core/pipeline.py +++ b/request-processor/src/application/core/pipeline.py @@ -336,13 +336,19 @@ def _process_add_data_resource(resource_file, **kwargs): def _find_duplicate_candidates( - dataset, specification, output_path, organisation_provider, organisation_index + dataset, + specification, + output_path, + redirect_lookups, + organisation_provider, + organisation_index, ): try: return find_duplicate_redirect_candidates( dataset=dataset, specification=specification, transformed_csv_path=output_path, + redirect_lookups=redirect_lookups, organisation_provider=organisation_provider, organisation_index=organisation_index, ) @@ -437,7 +443,12 @@ def fetch_add_data_response( existing_entities ) duplicate_candidates = _find_duplicate_candidates( - dataset, specification, output_path, organisations[0], organisation + dataset, + specification, + output_path, + pipeline.redirect_lookups(), + organisations[0], + organisation, ) if issues_log: diff --git a/request-processor/tests/unit/src/application/core/test_pipeline.py b/request-processor/tests/unit/src/application/core/test_pipeline.py index a0f14eac..669af4a7 100644 --- a/request-processor/tests/unit/src/application/core/test_pipeline.py +++ b/request-processor/tests/unit/src/application/core/test_pipeline.py @@ -7,6 +7,7 @@ _get_entities_breakdown, _get_existing_entities_breakdown, _format_task_summary, + _find_duplicate_candidates, run_task_pipeline, ) @@ -412,6 +413,31 @@ def test_fetch_add_data_response_reraises_processing_error(monkeypatch, tmp_path ) +def test_find_duplicate_candidates_passes_redirect_lookups(monkeypatch, tmp_path): + calls = {} + + def fake_find_duplicate_redirect_candidates(**kwargs): + calls.update(kwargs) + return [{"old_entity": "100"}] + + monkeypatch.setattr( + "src.application.core.pipeline.find_duplicate_redirect_candidates", + fake_find_duplicate_redirect_candidates, + ) + + result = _find_duplicate_candidates( + dataset="conservation-area", + specification=MagicMock(), + output_path=str(tmp_path / "transformed.csv"), + redirect_lookups={"100": {"entity": "300", "status": "301"}}, + organisation_provider="local-authority:STH", + organisation_index=MagicMock(), + ) + + assert result == [{"old_entity": "100"}] + assert calls["redirect_lookups"] == {"100": {"entity": "300", "status": "301"}} + + def test_get_entities_breakdown_success(): """Test converting entities to breakdown format""" new_entities = [ diff --git a/request-processor/tests/unit/src/test_duplicates.py b/request-processor/tests/unit/src/test_duplicates.py index b907fa92..484f22ba 100644 --- a/request-processor/tests/unit/src/test_duplicates.py +++ b/request-processor/tests/unit/src/test_duplicates.py @@ -120,6 +120,10 @@ def fake_fetch_platform_entities(dataset, organisation_entity): dataset="conservation-area", specification=FakeSpecification(), transformed_csv_path=str(transformed_path), + redirect_lookups={ + "100": {"entity": "300", "status": "301"}, + "999": {"entity": "998", "status": "301"}, + }, organisation_provider="local-authority:STH", organisation_index=FakeOrganisationIndex(), fetch_platform_entities=fake_fetch_platform_entities, @@ -139,6 +143,13 @@ def fake_fetch_platform_entities(dataset, organisation_entity): assert candidates[0]["old_organisation_entity"] == "318" assert candidates[0]["new_organisation_entity"] == "318" assert candidates[0]["match_type"] == "complete_match" + assert candidates[0]["old_entity_redirects"] == [ + { + "old-entity": "100", + "status": "301", + "entity": "300", + } + ] def test_duplicate_candidates_map_organisation_entities_when_new_entity_is_a( From decc2fd52268191b7ca4af8dcf2e286f709be6fb Mon Sep 17 00:00:00 2001 From: Gibah Joseph Date: Mon, 13 Jul 2026 13:02:43 +0100 Subject: [PATCH 12/12] fix: Configure DATASETTE_BASE_URL via environment variable and add corresponding test --- docker-compose.yml | 1 + .../src/application/core/duplicates.py | 4 +++- .../tests/unit/src/test_duplicates.py | 17 +++++++++++++++++ 3 files changed, 21 insertions(+), 1 deletion(-) diff --git a/docker-compose.yml b/docker-compose.yml index cfcb6e1f..6fce82a4 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -67,6 +67,7 @@ services: CELERY_BROKER_REGION: eu-west-2 CELERY_BROKER_IS_SECURE: "false" DATABASE_URL: postgresql://postgres:password@request-db/request_database + DATASETTE_BASE_URL: ${DATASETTE_BASE_URL:-https://datasette.planning.data.gov.uk} REQUEST_FILES_BUCKET_NAME: dluhc-data-platform-request-files-local SENTRY_ENABLED: "false" SENTRY_DSN: https://secret.sentry.url/here diff --git a/request-processor/src/application/core/duplicates.py b/request-processor/src/application/core/duplicates.py index 74f833eb..05ebce08 100644 --- a/request-processor/src/application/core/duplicates.py +++ b/request-processor/src/application/core/duplicates.py @@ -10,7 +10,9 @@ logger = get_logger(__name__) -DATASETTE_BASE_URL = "https://datasette.planning.data.gov.uk" +DATASETTE_BASE_URL = os.getenv( + "DATASETTE_BASE_URL", "https://datasette.planning.data.gov.uk" +) REDIRECT_NOTE = "Redirect duplicate entity selected in Assign Entities" diff --git a/request-processor/tests/unit/src/test_duplicates.py b/request-processor/tests/unit/src/test_duplicates.py index 484f22ba..7535dc00 100644 --- a/request-processor/tests/unit/src/test_duplicates.py +++ b/request-processor/tests/unit/src/test_duplicates.py @@ -1,4 +1,5 @@ import csv +import importlib import sqlite3 import sys import types @@ -244,6 +245,22 @@ def fake_get(url, params, timeout): ] +def test_datasette_base_url_can_be_configured(monkeypatch): + monkeypatch.setenv( + "DATASETTE_BASE_URL", "https://datasette.staging.planning.data.gov.uk" + ) + reloaded = importlib.reload(duplicates) + + try: + assert ( + reloaded.DATASETTE_BASE_URL + == "https://datasette.staging.planning.data.gov.uk" + ) + finally: + monkeypatch.delenv("DATASETTE_BASE_URL") + importlib.reload(duplicates) + + def test_name_similarity_uses_partial_ratio_for_added_words(): assert ( duplicates._name_similarity("South Jesmond", "South Jesmond Conservation Area")