Merge pull request #3342 from mtmail/tyops

[nominatim.git] / nominatim / api / v1 / helpers.py

diff --git a/nominatim/api/v1/helpers.py b/nominatim/api/v1/helpers.py
index ed66be8cfc5e08a95472a41e26900e1f1402c092..ffd06a6a35e3439d4b47ae92a4bcd4d321506521 100644 (file)
--- a/nominatim/api/v1/helpers.py
+++ b/nominatim/api/v1/helpers.py
@@ -37,7 +37,7 @@ def zoom_to_rank(zoom: int) -> int:
     return REVERSE_MAX_RANKS[max(0, min(18, zoom))]
 
 
-FEATURE_TYPE_TO_RANK: Dict[Optional[str], Any] = {
+FEATURE_TYPE_TO_RANK: Dict[Optional[str], Tuple[int, int]] = {
     'country': (4, 4),
     'state': (8, 8),
     'city': (14, 16),
@@ -52,8 +52,8 @@ def feature_type_to_rank(feature_type: Optional[str]) -> Tuple[int, int]:
     return FEATURE_TYPE_TO_RANK.get(feature_type, (0, 30))
 
 
-#pylint: disable=too-many-arguments
-def extend_query_parts(queryparts: dict[str, Any], details: dict[str, Any],
+#pylint: disable=too-many-arguments,too-many-branches
+def extend_query_parts(queryparts: Dict[str, Any], details: Dict[str, Any],
                        feature_type: Optional[str],
                        namedetails: bool, extratags: bool,
                        excluded: Iterable[str]) -> None:
@@ -62,9 +62,14 @@ def extend_query_parts(queryparts: dict[str, Any], details: dict[str, Any],
     """
     parsed = SearchDetails.from_kwargs(details)
     if parsed.geometry_output != GeometryFormat.NONE:
-        for flag in parsed.geometry_output:
-            assert flag.name
-            queryparts[f'polygon_{flag.name.lower()}'] = '1'
+        if GeometryFormat.GEOJSON in parsed.geometry_output:
+            queryparts['polygon_geojson'] = '1'
+        if GeometryFormat.KML in parsed.geometry_output:
+            queryparts['polygon_kml'] = '1'
+        if GeometryFormat.SVG in parsed.geometry_output:
+            queryparts['polygon_svg'] = '1'
+        if GeometryFormat.TEXT in parsed.geometry_output:
+            queryparts['polygon_text'] = '1'
     if parsed.address_details:
         queryparts['addressdetails'] = '1'
     if namedetails:
@@ -78,7 +83,7 @@ def extend_query_parts(queryparts: dict[str, Any], details: dict[str, Any],
     if parsed.countries:
         queryparts['countrycodes'] = ','.join(parsed.countries)
     queryparts['exclude_place_ids'] = \
-        ','.join(chain(excluded, map(str, parsed.excluded)))
+        ','.join(chain(excluded, map(str, (e for e in parsed.excluded if e > 0))))
     if parsed.viewbox:
         queryparts['viewbox'] = ','.join(f"{c:.7g}" for c in parsed.viewbox.coords)
     if parsed.bounded_viewbox:
@@ -103,15 +108,18 @@ def deduplicate_results(results: SearchResults, max_results: int) -> SearchResul
             assert result.names and 'ref' in result.names
             if any(_is_postcode_relation_for(r, result.names['ref']) for r in results):
                 continue
-        classification = (result.osm_object[0] if result.osm_object else None,
-                          result.category,
-                          result.display_name,
-                          result.rank_address)
-        if result.osm_object not in osm_ids_done \
-           and classification not in classification_done:
+        if result.source_table == SourceTable.PLACEX:
+            classification = (result.osm_object[0] if result.osm_object else None,
+                              result.category,
+                              result.display_name,
+                              result.rank_address)
+            if result.osm_object not in osm_ids_done \
+               and classification not in classification_done:
+                deduped.append(result)
+            osm_ids_done.add(result.osm_object)
+            classification_done.add(classification)
+        else:
             deduped.append(result)
-        osm_ids_done.add(result.osm_object)
-        classification_done.add(classification)
         if len(deduped) >= max_results:
             break
 
@@ -131,10 +139,10 @@ def _deg(axis:str) -> str:
     return f"(?P<{axis}_deg>\\d+\\.\\d+)°?"
 
 def _deg_min(axis: str) -> str:
-    return f"(?P<{axis}_deg>\\d+)[°\\s]+(?P<{axis}_min>[\\d.]+)?[′']*"
+    return f"(?P<{axis}_deg>\\d+)[°\\s]+(?P<{axis}_min>[\\d.]+)[′']*"
 
 def _deg_min_sec(axis: str) -> str:
-    return f"(?P<{axis}_deg>\\d+)[°\\s]+(?P<{axis}_min>\\d+)[′'\\s]+(?P<{axis}_sec>[\\d.]+)?[\"″]*"
+    return f"(?P<{axis}_deg>\\d+)[°\\s]+(?P<{axis}_min>\\d+)[′'\\s]+(?P<{axis}_sec>[\\d.]+)[\"″]*"
 
 COORD_REGEX = [re.compile(r'(?:(?P<pre>.*?)\s+)??' + r + r'(?:\s+(?P<post>.*))?') for r in (
     r"(?P<ns>[NS])\s*" + _deg('lat') + r"[\s,]+" + r"(?P<ew>[EW])\s*" + _deg('lon'),
@@ -147,7 +155,7 @@ COORD_REGEX = [re.compile(r'(?:(?P<pre>.*?)\s+)??' + r + r'(?:\s+(?P<post>.*))?'
 )]
 
 def extract_coords_from_query(query: str) -> Tuple[str, Optional[float], Optional[float]]:
-    """ Look for something that is formated like a coordinate at the
+    """ Look for something that is formatted like a coordinate at the
         beginning or end of the query. If found, extract the coordinate and
         return the remaining query (or the empty string if the query
         consisted of nothing but a coordinate).