feat(titles): canonical incident names per adapter

disaster_report/_search_keys.py

@@ -48,7 +48,7 @@ def derive_search_keys(
     if len(report.places) == 1 and report.places[0].country_code:
         place = report.places[0]
         country = country_name(place.country_code)
-        token = _place_token(place)
+        token = place_token(place)
         place_str = f"{token}, {country}" if token else country
         strict = " ".join(
             p for p in (incident_type, place_str, month_year_label, disease) if p
@@ -109,7 +109,7 @@ def _shared_continent_tokens(regions: list[str]) -> str:
     return " ".join(winners)
 
 
-def _place_token(place: ReportPlace) -> str:
+def place_token(place: ReportPlace) -> str:
     locality = place.locality
     if not locality:
         return ""

disaster_report/_search_keys.pyi

@@ -1,4 +1,4 @@
-from disaster_report.models import SourceReport
+from disaster_report.models import ReportPlace, SourceReport
 
 def disease_from_title(title: str) -> str: ...
 def derive_search_keys(
@@ -7,3 +7,4 @@ def derive_search_keys(
     disease: str = ...,
 ) -> tuple[str, str]: ...
 def derive_repoll_keys(report: SourceReport) -> list[str]: ...
+def place_token(place: ReportPlace) -> str: ...

disaster_report/_title_format.py

@@ -0,0 +1,96 @@
+from __future__ import annotations
+
+import unicodedata
+
+from disaster_report._country_names import country_name
+from disaster_report._search_keys import place_token
+from disaster_report.models import ReportPlace
+
+_ADMIN_SUFFIXES: tuple[str, ...] = (
+    " Zhuangzu Zizhiqu",
+    " Autonomous Region",
+    " Special Administrative Region",
+    " SAR",
+    " Sheng",
+    " Province",
+    " State",
+    " Prefecture",
+    " Department",
+    " Region",
+    " District",
+    " Territory",
+)
+
+_COUNTRY_ALIASES: dict[str, str] = {
+    "democratic republic of the congo": "DR Congo",
+    "united kingdom": "UK",
+    "united states of america": "United States",
+    "republic of korea": "South Korea",
+    "democratic people's republic of korea": "North Korea",
+    "russian federation": "Russia",
+    "islamic republic of iran": "Iran",
+    "syrian arab republic": "Syria",
+    "czech republic": "Czechia",
+    "united republic of tanzania": "Tanzania",
+    "republic of north macedonia": "North Macedonia",
+    "bolivarian republic of venezuela": "Venezuela",
+    "brunei darussalam": "Brunei",
+    "lao people's democratic republic": "Laos",
+    "socialist republic of viet nam": "Vietnam",
+    "republic of moldova": "Moldova",
+    "state of palestine": "Palestine",
+}
+
+
+def format_title(*parts: str) -> str:
+    return " ".join(p.strip() for p in parts if p and p.strip())
+
+
+def normalise_subdivision(name: str) -> str:
+    if not name:
+        return ""
+    stripped = _strip_accents(name)
+    lowered = stripped.lower()
+    for suffix in _ADMIN_SUFFIXES:
+        if lowered.endswith(suffix.lower()):
+            stripped = stripped[: -len(suffix)].strip()
+            break
+    return stripped.strip()
+
+
+def normalise_country(name: str) -> str:
+    if not name:
+        return ""
+    return _COUNTRY_ALIASES.get(name.lower().strip(), name.strip())
+
+
+def smallest_place(places: list[ReportPlace]) -> tuple[str, str]:
+    if not places:
+        return ("", "")
+    place = places[0]
+    country = country_name(place.country_code)
+    locality_token = place_token(place)
+    if locality_token:
+        return (locality_token, country)
+    if place.subdivision:
+        return (normalise_subdivision(place.subdivision), country)
+    return (country, country)
+
+
+def format_place(smallest: str, country: str) -> str:
+    smallest = (smallest or "").strip()
+    country = (country or "").strip()
+    if not smallest and not country:
+        return ""
+    if not country:
+        return smallest
+    if smallest and smallest != country:
+        return f"{smallest}, {country}"
+    return country
+
+
+def _strip_accents(s: str) -> str:
+    nfkd = unicodedata.normalize("NFKD", s)
+    return "".join(c for c in nfkd if not unicodedata.combining(c))
+
+

disaster_report/_title_format.pyi

@@ -0,0 +1,7 @@
+from disaster_report.models import ReportPlace
+
+def format_title(*parts: str) -> str: ...
+def normalise_subdivision(name: str) -> str: ...
+def normalise_country(name: str) -> str: ...
+def smallest_place(places: list[ReportPlace]) -> tuple[str, str]: ...
+def format_place(smallest: str, country: str) -> str: ...

disaster_report/sources/gdacs.py

@@ -13,6 +13,7 @@
 
 from disaster_report._country_names import country_name
 from disaster_report._search_keys import derive_search_keys
+from disaster_report._title_format import format_place, format_title, smallest_place
 from disaster_report.models import ReportPlace, SourceReport
 from disaster_report.sources.errors import SourceFetchError
 
@@ -22,6 +23,11 @@
 _GEO = "{http://www.w3.org/2003/01/geo/wgs84_pos#}"
 _DC = "{http://purl.org/dc/elements/1.1/}"
 _EVENTID_RE = re.compile(r"[?&]eventid=(\d+)", re.IGNORECASE)
+_MAGNITUDE_RE = re.compile(r"Magnitude\s+(\d+(?:\.\d+)?)", re.IGNORECASE)
+_STORM_NAME_RE = re.compile(
+    r"(?:Tropical Cyclone|Typhoon|Hurricane|Cyclone)\s+([A-Z][A-Z]*-?\d*)",
+)
+_VOLCANO_NAME_RE = re.compile(r"^Eruption\s+(.+?)\s*$", re.IGNORECASE)
 _TYPES: dict[str, str] = {
     "TC": "Tropical Cyclone",
     "EQ": "Earthquake",
@@ -82,7 +88,6 @@ def derive_keys(self, report: SourceReport) -> tuple[str, str]:
 
 def _item_to_report(item: Any) -> SourceReport:
     raw_fields = _build_raw_fields(item)
-    title = str(raw_fields.get("title") or "")
     link = str(raw_fields.get("link") or "")
     eventtype = str(raw_fields.get("eventtype") or "")
     fromdate = str(raw_fields.get("fromdate") or "")
@@ -91,17 +96,81 @@ def _item_to_report(item: Any) -> SourceReport:
     lat = raw_fields.get("geo_lat")
     lon = raw_fields.get("geo_long")
     places = _extract_places(iso3, country_text, lat, lon)
+    incident_type = _TYPES.get(eventtype, eventtype)
+    report_date = _to_iso_date(fromdate)
     return SourceReport(
         source="GDACS",
         source_id=_extract_event_id(link),
-        incident_type=_TYPES.get(eventtype, eventtype),
-        name=title,
+        incident_type=incident_type,
+        name=_extract_canonical_name(raw_fields, places, report_date, incident_type),
         places=places,
-        report_date=_to_iso_date(fromdate),
+        report_date=report_date,
         raw_fields=raw_fields,
     )
 
 
+def _extract_canonical_name(
+    raw_fields: dict[str, object],
+    places: list[ReportPlace],
+    report_date: str,
+    incident_type: str,
+) -> str:
+    identifier = _extract_gdacs_identifier(incident_type, raw_fields)
+    smallest, country = smallest_place(places)
+    if incident_type == "Volcano" and identifier:
+        return format_title(incident_type, format_place(identifier, country), report_date)
+    return format_title(
+        incident_type,
+        identifier,
+        format_place(smallest, country),
+        report_date,
+    )
+
+
+def _extract_gdacs_identifier(
+    incident_type: str,
+    raw_fields: dict[str, object],
+) -> str:
+    if incident_type == "Earthquake":
+        return _extract_magnitude(raw_fields)
+    if incident_type == "Tropical Cyclone":
+        return _extract_storm_name(raw_fields)
+    if incident_type == "Volcano":
+        return _extract_volcano_name(raw_fields)
+    return ""
+
+
+def _extract_magnitude(raw_fields: dict[str, object]) -> str:
+    severity = raw_fields.get("severity")
+    if isinstance(severity, int | float) and not isinstance(severity, bool):
+        return f"M{float(severity):.1f}"
+    severitytext = str(raw_fields.get("severitytext") or "")
+    match = _MAGNITUDE_RE.search(severitytext)
+    if not match and isinstance(severity, str):
+        match = _MAGNITUDE_RE.search(severity)
+    if match:
+        try:
+            return f"M{float(match.group(1)):.1f}"
+        except ValueError:
+            return ""
+    return ""
+
+
+def _extract_storm_name(raw_fields: dict[str, object]) -> str:
+    eventname = str(raw_fields.get("eventname") or "").strip()
+    if eventname:
+        return eventname
+    title = str(raw_fields.get("title") or "")
+    match = _STORM_NAME_RE.search(title)
+    return match.group(1).strip() if match else ""
+
+
+def _extract_volcano_name(raw_fields: dict[str, object]) -> str:
+    title = str(raw_fields.get("title") or "").strip()
+    match = _VOLCANO_NAME_RE.match(title)
+    return match.group(1).strip() if match else ""
+
+
 def _build_raw_fields(item: Any) -> dict[str, object]:
 
     raw: dict[str, object] = {}

disaster_report/sources/gdacs.pyi

@@ -13,6 +13,12 @@ class GDACSAdapter:
     def derive_keys(self, report: SourceReport) -> tuple[str, str]: ...
 
 def _item_to_report(item: object) -> SourceReport: ...
+def _extract_canonical_name(
+    raw_fields: dict[str, object],
+    places: list[ReportPlace],
+    report_date: str,
+    incident_type: str,
+) -> str: ...
 def _extract_places(
     iso3: str, country_text: str, lat: object, lon: object
 ) -> list[ReportPlace]: ...

disaster_report/sources/usgs.py

@@ -13,6 +13,7 @@
 from disaster_report._country_names import country_name
 from disaster_report._regions import subregion_for_country
 from disaster_report._search_keys import derive_search_keys
+from disaster_report._title_format import format_place, format_title, smallest_place
 from disaster_report.models import ReportPlace, SourceReport
 from disaster_report.sources.errors import SourceFetchError
 
@@ -79,30 +80,44 @@ def _feature_to_report(
     feature_id = feature_dict.get("id")
     code = properties.get("code")
     source_id = str(feature_id or code or "")
-    name = str(properties.get("title") or "")
     place = str(properties.get("place") or "")
     places = _extract_places(lat, lon, place)
+    raw_fields = {
+        **properties,
+        "geometry": {
+            "type": geometry.get("type"),
+            "coordinates": coordinates,
+        }
+        if coordinates
+        else {},
+        "depth": depth,
+        "place": place,
+    }
     return SourceReport(
         source="USGS",
         source_id=source_id,
         incident_type="Earthquake",
-        name=name,
+        name=_extract_canonical_name(raw_fields, places, _to_iso_date(properties.get("time")), "Earthquake"),
         places=places,
         report_date=_to_iso_date(properties.get("time")),
-        raw_fields={
-            **properties,
-            "geometry": {
-                "type": geometry.get("type"),
-                "coordinates": coordinates,
-            }
-            if coordinates
-            else {},
-            "depth": depth,
-            "place": place,
-        },
+        raw_fields=raw_fields,
     )
 
 
+def _extract_canonical_name(
+    raw_fields: dict[str, object],
+    places: list[ReportPlace],
+    report_date: str,
+    incident_type: str,
+) -> str:
+    mag = raw_fields.get("mag")
+    mag_str = ""
+    if isinstance(mag, int | float) and not isinstance(mag, bool):
+        mag_str = f"M{float(mag):.1f}"
+    smallest, country = smallest_place(places)
+    return format_title(incident_type, mag_str, format_place(smallest, country), report_date)
+
+
 def _extract_places(
     lat: Any,
     lon: Any,

disaster_report/sources/usgs.pyi

@@ -1,4 +1,4 @@
-from disaster_report.models import SourceReport
+from disaster_report.models import ReportPlace, SourceReport
 import logging
 
 logger: logging.Logger
@@ -11,3 +11,10 @@ class USGSAdapter:
     def fetch(self) -> list[SourceReport]: ...
     def should_monitor(self, report: SourceReport) -> bool: ...
     def derive_keys(self, report: SourceReport) -> tuple[str, str]: ...
+
+def _extract_canonical_name(
+    raw_fields: dict[str, object],
+    places: list[ReportPlace],
+    report_date: str,
+    incident_type: str,
+) -> str: ...