diff --git a/backend/app/utils/extract_location_metadata.py b/backend/app/utils/extract_location_metadata.py index ccce4a8f6..1c738786b 100644 --- a/backend/app/utils/extract_location_metadata.py +++ b/backend/app/utils/extract_location_metadata.py @@ -10,7 +10,7 @@ import json from datetime import datetime -from typing import Optional, Tuple, Dict, Any +from typing import Optional, Tuple, Dict, Any, Iterable from app.logging.setup_logging import get_logger @@ -31,26 +31,87 @@ def __init__(self): """Initialize the metadata extractor.""" pass + @staticmethod + def _resolve_coordinate( + field: str, candidates: Iterable[Any], limit: float + ) -> Optional[float]: + """ + Return the first candidate that is usable as a coordinate. + + Candidates are supplied in order of preference and each one is converted + and range checked in turn. A candidate is skipped when it is absent + (None or a blank string), cannot be read as a number, or falls outside + the valid range, so a malformed or out-of-range value in a preferred + field does not mask a good value in a less preferred one. + + Zero is deliberately kept: it is a real location on the equator and the + prime meridian, but it is falsy in Python, so an `a or b` fallback chain + would silently discard it. + + Args: + field: Field name, used only for log messages + candidates: Candidate values, in order of preference + limit: Largest valid magnitude (90 for latitude, 180 for longitude) + + Returns: + The first usable coordinate, or None if no candidate qualifies + """ + for value in candidates: + if value is None: + continue + if isinstance(value, str) and not value.strip(): + continue + # bool is a subclass of int, so float(True) would otherwise pass as 1.0 + if isinstance(value, bool): + logger.warning(f"Ignoring boolean {field}: {value!r}") + continue + + try: + number = float(value) + except (ValueError, TypeError, OverflowError): + # OverflowError covers integers too large to become a float, + # which JSON metadata can carry with no size limit + logger.warning(f"Ignoring unreadable {field}: {value!r}") + continue + + # Also rejects NaN, which fails every comparison + if not -limit <= number <= limit: + logger.warning(f"Ignoring out-of-range {field}: {number}") + continue + + return number + + return None + def extract_gps_coordinates( self, metadata: Dict[str, Any] ) -> Tuple[Optional[float], Optional[float]]: """ Extract GPS coordinates from metadata dictionary. - Supports multiple metadata structures: + Supports multiple metadata structures, checked in order of preference: - Top-level: {"latitude": 28.6, "longitude": 77.2} - Nested EXIF: {"exif": {"gps": {"latitude": 28.6, "longitude": 77.2}}} - Alternative names: lat, lon, Latitude, Longitude + Latitude and longitude are resolved independently, and each falls + through to the next source when a value is missing, unreadable or out of + range, so a coordinate pair may be assembled from two different sources. + Args: metadata: Parsed metadata dictionary Returns: - Tuple of (latitude, longitude) or (None, None) if not found + Tuple of (latitude, longitude), or (None, None) when either half + cannot be resolved, since one coordinate on its own is not a location Validates: - Latitude: -90 to 90 - Longitude: -180 to 180 + + Note: + A coordinate of 0 is a real location, not a missing value, so it is + kept rather than falling through to the next source. """ latitude = None longitude = None @@ -59,43 +120,42 @@ def extract_gps_coordinates( if not isinstance(metadata, dict): return None, None - # Method 1: Direct top-level fields - lat = metadata.get("latitude") - lon = metadata.get("longitude") - - # Method 2: Check nested 'exif' -> 'gps' structure - if not lat or not lon: - exif = metadata.get("exif", {}) - if isinstance(exif, dict): - gps = exif.get("gps", {}) - if isinstance(gps, dict): - lat = lat or gps.get("latitude") - lon = lon or gps.get("longitude") - - # Method 3: Check alternative field names - if not lat or not lon: - lat = lat or metadata.get("lat") or metadata.get("Latitude") - lon = lon or metadata.get("lon") or metadata.get("Longitude") - - # Validate and convert coordinates - if lat is not None and lon is not None: - try: - lat = float(lat) - lon = float(lon) - - # Sanity check: valid coordinate ranges - if -90 <= lat <= 90 and -180 <= lon <= 180: - latitude = lat - longitude = lon - else: - logger.warning( - f"Invalid coordinate range: lat={lat}, lon={lon}" - ) - except (ValueError, TypeError) as e: - logger.warning(f"Could not convert coordinates to float: {e}") + # Nested 'exif' -> 'gps' structure, when the image has one + exif = metadata.get("exif") + exif = exif if isinstance(exif, dict) else {} + gps = exif.get("gps") + gps = gps if isinstance(gps, dict) else {} + + # Direct top-level fields, then nested EXIF GPS, then the + # alternative spellings some sources use. + latitude = self._resolve_coordinate( + "latitude", + ( + metadata.get("latitude"), + gps.get("latitude"), + metadata.get("lat"), + metadata.get("Latitude"), + ), + 90.0, + ) + longitude = self._resolve_coordinate( + "longitude", + ( + metadata.get("longitude"), + gps.get("longitude"), + metadata.get("lon"), + metadata.get("Longitude"), + ), + 180.0, + ) + + # A lone latitude or longitude is not a usable location + if latitude is None or longitude is None: + return None, None except Exception as e: logger.error(f"Unexpected error extracting GPS coordinates: {e}") + return None, None return latitude, longitude @@ -181,7 +241,7 @@ def extract_datetime(self, metadata: Dict[str, Any]) -> Optional[datetime]: try: captured_at = datetime.strptime(date_str, fmt) break - except (ValueError, TypeError): + except (ValueError, TypeError, OverflowError): continue if not captured_at: diff --git a/backend/app/utils/images.py b/backend/app/utils/images.py index dde078fff..919f5155a 100644 --- a/backend/app/utils/images.py +++ b/backend/app/utils/images.py @@ -275,7 +275,8 @@ def image_util_prepare_image_records( latitude, longitude, captured_at = extractor.extract_all(metadata_json) # Log GPS extraction results - if latitude and longitude: + # (0 is a valid coordinate, so test for None rather than truthiness) + if latitude is not None and longitude is not None: logger.info( f"GPS extracted for {os.path.basename(image_path)}: ({latitude}, {longitude})" ) diff --git a/backend/tests/test_extract_location_metadata.py b/backend/tests/test_extract_location_metadata.py new file mode 100644 index 000000000..e77ceb5e1 --- /dev/null +++ b/backend/tests/test_extract_location_metadata.py @@ -0,0 +1,408 @@ +import json + +import pytest + +from app.utils.extract_location_metadata import MetadataExtractor + + +@pytest.fixture +def extractor(): + """A MetadataExtractor instance.""" + return MetadataExtractor() + + +# ############################## +# Test Classes +# ############################## + + +LATITUDE_LIMIT = 90.0 +LONGITUDE_LIMIT = 180.0 + + +def resolve_latitude(*candidates): + """Resolve a latitude from candidates in order of preference.""" + return MetadataExtractor._resolve_coordinate("latitude", candidates, LATITUDE_LIMIT) + + +class TestResolveCoordinate: + """Test class for the per-candidate resolver behind the fallback chain.""" + + def test_zero_is_kept(self): + """0 and 0.0 are real coordinates, not missing values.""" + assert resolve_latitude(0) == 0.0 + assert resolve_latitude(0.0) == 0.0 + + def test_none_is_skipped(self): + """None falls through to the next candidate.""" + assert resolve_latitude(None, 28.6) == 28.6 + + @pytest.mark.parametrize("blank", ["", " ", "\t"]) + def test_blank_string_is_skipped(self, blank): + """Blank strings fall through to the next candidate.""" + assert resolve_latitude(blank, 28.6) == 28.6 + + def test_first_usable_candidate_wins(self): + """The earliest usable candidate takes precedence.""" + assert resolve_latitude(28.6, 45.0) == 28.6 + + def test_unreadable_candidate_is_skipped(self): + """A value that cannot become a float falls through.""" + assert resolve_latitude("not-a-number", 28.6) == 28.6 + + def test_out_of_range_candidate_is_skipped(self): + """A value outside the valid range falls through.""" + assert resolve_latitude(200, 28.6) == 28.6 + + def test_numeric_strings_are_converted(self): + """Coordinates arriving as strings are converted.""" + assert resolve_latitude("28.6") == 28.6 + + def test_limits_are_inclusive(self): + """The exact range boundaries are accepted.""" + assert resolve_latitude(90) == 90.0 + assert resolve_latitude(-90) == -90.0 + + @pytest.mark.parametrize("value", ["nan", "inf", "-inf", float("nan")]) + def test_non_finite_values_are_skipped(self, value): + """NaN and infinity are not usable coordinates.""" + assert resolve_latitude(value, 28.6) == 28.6 + + @pytest.mark.parametrize("value", [[1, 2], {"a": 1}, object()]) + def test_unconvertible_types_are_skipped(self, value): + """Values of the wrong type fall through instead of raising.""" + assert resolve_latitude(value, 28.6) == 28.6 + + @pytest.mark.parametrize("value", [True, False]) + def test_booleans_are_skipped(self, value): + """bool is a subclass of int, but it is not a coordinate.""" + assert resolve_latitude(value, 28.6) == 28.6 + + @pytest.mark.parametrize("sign", [1, -1]) + def test_oversized_integers_are_skipped(self, sign): + """An integer too large to become a float falls through, not raises.""" + oversized = sign * int("9" * 400) + assert resolve_latitude(oversized, 28.6) == 28.6 + + def test_oversized_integer_alone_returns_none(self): + """An oversized integer with no fallback resolves to None.""" + assert resolve_latitude(int("9" * 400)) is None + + def test_returns_none_when_no_candidate_qualifies(self): + """None is returned when every candidate is unusable.""" + assert resolve_latitude(None, "", "junk", 200) is None + assert resolve_latitude() is None + + def test_longitude_range_is_wider_than_latitude(self): + """120 is a valid longitude but not a valid latitude.""" + assert ( + MetadataExtractor._resolve_coordinate("longitude", (120,), LONGITUDE_LIMIT) + == 120.0 + ) + assert resolve_latitude(120) is None + + +class TestFallthroughToLowerPrioritySource: + """ + Test class for falling through unusable values in a preferred field. + + A malformed or out-of-range coordinate in a higher-priority field must not + mask a valid coordinate in a lower-priority one. + """ + + @pytest.mark.parametrize("bad", ["not-a-number", 200, -200, "", None, "nan"]) + def test_bad_top_level_latitude_falls_back_to_exif(self, extractor, bad): + """An unusable top-level latitude falls through to exif.gps.""" + metadata = { + "latitude": bad, + "longitude": 77.2, + "exif": {"gps": {"latitude": 28.6}}, + } + assert extractor.extract_gps_coordinates(metadata) == (28.6, 77.2) + + @pytest.mark.parametrize("bad", ["not-a-number", 400, -400, "", None]) + def test_bad_top_level_longitude_falls_back_to_exif(self, extractor, bad): + """An unusable top-level longitude falls through to exif.gps.""" + metadata = { + "latitude": 28.6, + "longitude": bad, + "exif": {"gps": {"longitude": 77.2}}, + } + assert extractor.extract_gps_coordinates(metadata) == (28.6, 77.2) + + def test_bad_top_level_falls_back_to_alias(self, extractor): + """An unusable top-level value falls through to the lat/lon aliases.""" + metadata = { + "latitude": "junk", + "longitude": 999, + "lat": 28.6, + "lon": 77.2, + } + assert extractor.extract_gps_coordinates(metadata) == (28.6, 77.2) + + def test_falls_through_two_bad_sources_to_the_third(self, extractor): + """Resolution continues past more than one unusable source.""" + metadata = { + "latitude": "junk", + "longitude": "junk", + "exif": {"gps": {"latitude": 200, "longitude": 400}}, + "lat": 28.6, + "lon": 77.2, + } + assert extractor.extract_gps_coordinates(metadata) == (28.6, 77.2) + + def test_falls_back_to_a_zero_coordinate(self, extractor): + """Falling through still preserves a valid 0 further down the chain.""" + metadata = { + "latitude": "junk", + "longitude": "junk", + "exif": {"gps": {"latitude": 0.0, "longitude": 0.0}}, + } + assert extractor.extract_gps_coordinates(metadata) == (0.0, 0.0) + + def test_each_coordinate_falls_through_independently(self, extractor): + """Latitude and longitude may end up resolved from different sources.""" + metadata = { + "latitude": 28.6, + "longitude": "junk", + "exif": {"gps": {"longitude": 77.2}}, + } + assert extractor.extract_gps_coordinates(metadata) == (28.6, 77.2) + + def test_no_usable_fallback_returns_none(self, extractor): + """(None, None) is returned when no source holds a usable value.""" + metadata = { + "latitude": "junk", + "longitude": 999, + "exif": {"gps": {"latitude": 200, "longitude": "also-junk"}}, + } + assert extractor.extract_gps_coordinates(metadata) == (None, None) + + def test_unusable_latitude_with_valid_longitude_returns_none(self, extractor): + """A resolvable longitude alone is not a location.""" + metadata = {"latitude": "junk", "longitude": 77.2} + assert extractor.extract_gps_coordinates(metadata) == (None, None) + + def test_oversized_integer_falls_back_to_exif(self, extractor): + """ + An integer too large to convert falls through to exif.gps. + + JSON puts no size limit on integers, so metadata can carry a value that + raises OverflowError rather than ValueError on conversion. + """ + metadata = { + "latitude": int("9" * 400), + "longitude": 77.2, + "exif": {"gps": {"latitude": 28.6}}, + } + assert extractor.extract_gps_coordinates(metadata) == (28.6, 77.2) + + def test_oversized_integer_survives_the_json_entry_point(self, extractor): + """The same value arriving as a JSON string is handled end to end.""" + metadata_json = ( + '{"latitude": ' + "9" * 400 + ', "longitude": 77.2,' + ' "exif": {"gps": {"latitude": 28.6}}}' + ) + latitude, longitude, _ = extractor.extract_all(metadata_json) + assert (latitude, longitude) == (28.6, 77.2) + + +class TestExtractGPSCoordinatesZeroValues: + """Test class for coordinates on the equator and the prime meridian.""" + + @pytest.mark.parametrize( + "metadata, expected", + [ + # Equator: latitude is exactly 0 + ({"latitude": 0.0, "longitude": 77.2}, (0.0, 77.2)), + # Prime meridian: longitude is exactly 0 + ({"latitude": 51.5, "longitude": 0.0}, (51.5, 0.0)), + # Null Island: both are 0 + ({"latitude": 0.0, "longitude": 0.0}, (0.0, 0.0)), + # Integer zeros + ({"latitude": 0, "longitude": 0}, (0.0, 0.0)), + # Zeros as strings + ({"latitude": "0.0", "longitude": "0.0"}, (0.0, 0.0)), + # Negative zero + ({"latitude": -0.0, "longitude": -0.0}, (0.0, 0.0)), + ], + ) + def test_zero_coordinates_are_preserved(self, extractor, metadata, expected): + """Top-level zero coordinates survive extraction instead of being dropped.""" + assert extractor.extract_gps_coordinates(metadata) == expected + + def test_zero_in_nested_exif_gps(self, extractor): + """Zero coordinates nested under exif.gps are preserved.""" + metadata = {"exif": {"gps": {"latitude": 0.0, "longitude": 0.0}}} + assert extractor.extract_gps_coordinates(metadata) == (0.0, 0.0) + + @pytest.mark.parametrize( + "metadata", + [ + {"lat": 0.0, "lon": 0.0}, + {"Latitude": 0.0, "Longitude": 0.0}, + ], + ) + def test_zero_in_alternative_field_names(self, extractor, metadata): + """Zero coordinates under the lat/lon aliases are preserved.""" + assert extractor.extract_gps_coordinates(metadata) == (0.0, 0.0) + + def test_zero_mixed_with_other_source(self, extractor): + """A zero from one source pairs correctly with a value from another.""" + metadata = {"latitude": 0.0, "exif": {"gps": {"longitude": 77.2}}} + assert extractor.extract_gps_coordinates(metadata) == (0.0, 77.2) + + +class TestExtractGPSCoordinatesPrecedence: + """Test class for which source wins when several supply the same field.""" + + def test_top_level_wins_over_nested(self, extractor): + """The top-level field takes precedence over exif.gps.""" + metadata = { + "latitude": 28.6, + "longitude": 77.2, + "exif": {"gps": {"latitude": 51.5, "longitude": -0.1}}, + } + assert extractor.extract_gps_coordinates(metadata) == (28.6, 77.2) + + def test_top_level_zero_wins_over_nested(self, extractor): + """A top-level 0 is not overridden by a non-zero nested value.""" + metadata = { + "latitude": 0.0, + "longitude": 0.0, + "exif": {"gps": {"latitude": 51.5, "longitude": -0.1}}, + } + assert extractor.extract_gps_coordinates(metadata) == (0.0, 0.0) + + def test_nested_wins_over_aliases(self, extractor): + """exif.gps takes precedence over the lat/lon aliases.""" + metadata = { + "exif": {"gps": {"latitude": 51.5, "longitude": -0.1}}, + "lat": 28.6, + "lon": 77.2, + } + assert extractor.extract_gps_coordinates(metadata) == (51.5, -0.1) + + def test_blank_string_falls_through_to_next_source(self, extractor): + """A blank top-level field is treated as absent, not as a bad value.""" + metadata = { + "latitude": "", + "longitude": " ", + "exif": {"gps": {"latitude": 28.6, "longitude": 77.2}}, + } + assert extractor.extract_gps_coordinates(metadata) == (28.6, 77.2) + + +class TestExtractGPSCoordinatesExisting: + """Test class guarding the behaviour that was already correct.""" + + def test_normal_coordinates(self, extractor): + """Ordinary non-zero coordinates are extracted.""" + metadata = {"latitude": 28.6139, "longitude": 77.2090} + assert extractor.extract_gps_coordinates(metadata) == (28.6139, 77.2090) + + def test_negative_coordinates(self, extractor): + """Southern and western coordinates are extracted.""" + metadata = {"latitude": -33.8688, "longitude": -151.2093} + assert extractor.extract_gps_coordinates(metadata) == (-33.8688, -151.2093) + + @pytest.mark.parametrize( + "metadata, expected", + [ + ({"latitude": 90, "longitude": 180}, (90.0, 180.0)), + ({"latitude": -90, "longitude": -180}, (-90.0, -180.0)), + ], + ) + def test_boundary_coordinates(self, extractor, metadata, expected): + """The extremes of the valid ranges are accepted.""" + assert extractor.extract_gps_coordinates(metadata) == expected + + @pytest.mark.parametrize( + "metadata", + [ + {}, + {"latitude": 28.6}, # longitude missing + {"longitude": 77.2}, # latitude missing + {"latitude": None, "longitude": None}, + {"other": "value"}, + ], + ) + def test_missing_coordinates_return_none(self, extractor, metadata): + """Absent or half-present coordinates yield (None, None).""" + assert extractor.extract_gps_coordinates(metadata) == (None, None) + + @pytest.mark.parametrize( + "metadata", + [ + {"latitude": 91, "longitude": 0}, + {"latitude": -91, "longitude": 0}, + {"latitude": 0, "longitude": 181}, + {"latitude": 0, "longitude": -181}, + ], + ) + def test_out_of_range_coordinates_rejected(self, extractor, metadata): + """Coordinates outside the valid ranges are still rejected.""" + assert extractor.extract_gps_coordinates(metadata) == (None, None) + + def test_unparseable_coordinates_return_none(self, extractor): + """Values that cannot become floats yield (None, None).""" + metadata = {"latitude": "not-a-number", "longitude": "also-not"} + assert extractor.extract_gps_coordinates(metadata) == (None, None) + + @pytest.mark.parametrize("metadata", [None, "string", 42, [1, 2]]) + def test_non_dict_metadata_returns_none(self, extractor, metadata): + """Non-dict metadata yields (None, None) rather than raising.""" + assert extractor.extract_gps_coordinates(metadata) == (None, None) + + @pytest.mark.parametrize( + "metadata", + [ + {"exif": "not-a-dict"}, + {"exif": {"gps": "not-a-dict"}}, + {"exif": None}, + {"exif": {"gps": None}}, + ], + ) + def test_malformed_exif_is_ignored(self, extractor, metadata): + """A malformed exif/gps section is skipped without raising.""" + assert extractor.extract_gps_coordinates(metadata) == (None, None) + + def test_malformed_exif_falls_back_to_top_level(self, extractor): + """A malformed exif section does not block the top-level fields.""" + metadata = {"latitude": 0.0, "longitude": 0.0, "exif": "not-a-dict"} + assert extractor.extract_gps_coordinates(metadata) == (0.0, 0.0) + + +class TestExtractAll: + """Test class for the JSON entry point used during image upload.""" + + def test_null_island_survives_json_round_trip(self, extractor): + """(0, 0) is preserved end to end from the metadata JSON string.""" + metadata_json = json.dumps({"latitude": 0.0, "longitude": 0.0}) + latitude, longitude, _ = extractor.extract_all(metadata_json) + assert (latitude, longitude) == (0.0, 0.0) + + def test_equator_with_datetime(self, extractor): + """Coordinates and datetime are extracted together.""" + metadata_json = json.dumps( + { + "latitude": 0.0, + "longitude": 32.5825, + "date_created": "2024-01-15 14:30:45", + } + ) + latitude, longitude, captured_at = extractor.extract_all(metadata_json) + assert (latitude, longitude) == (0.0, 32.5825) + assert captured_at is not None + assert captured_at.year == 2024 + + def test_bytes_metadata(self, extractor): + """Metadata supplied as bytes is decoded before parsing.""" + metadata_json = json.dumps({"latitude": 0.0, "longitude": 0.0}).encode("utf-8") + latitude, longitude, _ = extractor.extract_all(metadata_json) + assert (latitude, longitude) == (0.0, 0.0) + + @pytest.mark.parametrize("metadata_json", ["", "null", None, "{not json}"]) + def test_empty_or_invalid_json(self, extractor, metadata_json): + """Empty or unparseable metadata yields all-None without raising.""" + assert extractor.extract_all(metadata_json) == (None, None, None)