diff --git a/docker-compose.yml b/docker-compose.yml index cfcb6e1..6fce82a 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/Dockerfile b/request-processor/Dockerfile index ad26beb..ce6a88f 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 1950506..8212741 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 ec87361..c5bd183 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/requirements/test_requirements.txt b/request-processor/requirements/test_requirements.txt index 100a3cc..654ac6e 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/duplicates.py b/request-processor/src/application/core/duplicates.py new file mode 100644 index 0000000..05ebce0 --- /dev/null +++ b/request-processor/src/application/core/duplicates.py @@ -0,0 +1,369 @@ +import csv +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 = os.getenv( + "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", + "organisation_entity", + "organisation-entity", + "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) + conn = None + try: + import spatialite + from digital_land.expectations.operations.dataset import ( + duplicate_geometry_check, + ) + + 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: + pass + + +def _build_candidate( + *, + match: dict, + match_type: str, + spatial_field: str, + old_entity: dict, + new_entity: dict, + dataset: str, + old_organisation_entity: str = "", + new_organisation_entity: 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 old_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 new_organisation_entity + or match.get("organisation_entity_b", "") + ), + "evidence": ", ".join(evidence), + "name_similarity": similarity, + } + + +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, + specification, + transformed_csv_path: str, + redirect_lookups: dict | None = None, + 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 + ) + for row in provision_rows: + row.setdefault("organisation_entity", organisation_entity) + + 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 + redirect_lookups = redirect_lookups or {} + + 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: + 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")), + _normalise_entity_id(new_entity.get("entity")), + ) + if key in seen or key[0] == key[1]: + continue + seen.add(key) + 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 4eddcc9..7087f89 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__) @@ -254,6 +256,107 @@ 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, + 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, + ) + except Exception as err: + logger.exception("Duplicate analysis failed for dataset %s: %s", dataset, err) + return [] + + def fetch_add_data_response( dataset, organisation_provider, @@ -303,84 +406,50 @@ 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 ) + duplicate_candidates = _find_duplicate_candidates( + dataset, + specification, + output_path, + pipeline.redirect_lookups(), + organisations[0], + organisation, + ) if issues_log: issues_log.add_severity_column(severity_mapping=specification.issue_type) @@ -391,6 +460,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/application/core/test_pipeline.py b/request-processor/tests/unit/src/application/core/test_pipeline.py index 60f2832..669af4a 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 = [ @@ -614,6 +640,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: {}) 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 0000000..7535dc0 --- /dev/null +++ b/request-processor/tests/unit/src/test_duplicates.py @@ -0,0 +1,374 @@ +import csv +import importlib +import sqlite3 +import sys +import types + +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: + def __init__(self): + self.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): + 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 { + "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), + 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, + ) + + 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]["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( + tmp_path, monkeypatch +): + transformed_path = tmp_path / "transformed.csv" + _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 { + "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"] == "318" + + +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_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") + == "100%" + ) + + +def test_run_duplicate_check_commits_before_spatialite_metadata(monkeypatch): + 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))", + }, + ] + + 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 + 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 == []