fix(domains): GTFS stop_sequence repeats instead of row order; strict ICPN format

src/freshdata/domains/media/validator.py

@@ -39,6 +39,7 @@
 _BUNDLED_DIR = _PACK_DIR.parent / "bundled"
 _NONDIGIT = re.compile(r"\D")
 _ICPN_LENGTHS = (12, 13)
+_ICPN_FORMAT = re.compile(r"\d[\d -]*\d")
 
 
 def _sort_key(value: Any) -> tuple[int, Any]:
@@ -92,10 +93,17 @@ def is_valid_eidr(value: Any) -> bool:
 
 
 def is_valid_icpn(value: Any) -> bool:
-    """True if *value* is a 12-digit UPC or 13-digit EAN with a valid GS1 mod-10 digit."""
+    """True if *value* is a 12-digit UPC or 13-digit EAN with a valid GS1 mod-10 digit.
+
+    Only digits, optionally grouped by spaces or hyphens, are accepted; text that
+    merely contains a barcode (``"tel: 036000291452"``) is rejected.
+    """
     if value is None:
         return False
-    digits = _NONDIGIT.sub("", str(value))
+    text = str(value).strip()
+    if _ICPN_FORMAT.fullmatch(text) is None:
+        return False
+    digits = _NONDIGIT.sub("", text)
     if len(digits) not in _ICPN_LENGTHS:
         return False
     body, check = digits[:-1], int(digits[-1])

src/freshdata/domains/transport/rules.yaml

@@ -139,7 +139,7 @@ rules:
     repair: flag_only
 
   - id: GTFS-ST004
-    name: stop_sequence increases within each trip
+    name: stop_sequence is not repeated within a trip (row order is not required)
     layer: business
     severity: error
     fields: [trip_id, stop_sequence]

src/freshdata/domains/transport/validator.py

@@ -105,7 +105,8 @@ def register_extensions(self) -> None:
         self.register_check("route_type_valid", self._check_route_type)
         self.register_check("gtfs_time", self._check_gtfs_time)
         self.register_check("departure_ge_arrival", self._check_departure_ge_arrival)
-        self.register_check("monotonic_sequence", self._check_monotonic_sequence)
+        # The ``monotonic_sequence`` func key is kept so existing rule files still resolve.
+        self.register_check("monotonic_sequence", self._check_stop_sequence_unique)
         self.register_check("cross_file_reference", self._check_cross_file_reference)
 
     def _run_rule(self, df: pd.DataFrame, mapping: ColumnMapping, rule: Rule) -> RuleResult:
@@ -156,22 +157,23 @@ def _check_departure_ge_arrival(
         bad = both & (departure < arrival)
         return df.index[bad].tolist()
 
-    def _check_monotonic_sequence(
+    def _check_stop_sequence_unique(
         self, df: pd.DataFrame, mapping: ColumnMapping, rule: Rule
     ) -> list[Any]:
+        """Flag rows repeating an earlier ``stop_sequence`` value within the same trip.
+
+        GTFS requires ``stop_sequence`` to increase along a trip, but not that the
+        rows of stop_times.txt be listed in that order. Ordering the trip by
+        ``stop_sequence`` therefore always yields an increasing sequence unless a
+        value is repeated, so only repeats are violations. The first occurrence of
+        each value (in file order) is kept; later ones are flagged.
+        """
         trip = df[mapping.actual("trip_id")]
         seq = pd.to_numeric(df[mapping.actual("stop_sequence")], errors="coerce")
         work = pd.DataFrame({"_trip": trip, "_seq": seq}, index=df.index)
-        bad: list[Any] = []
-        for _, group in work.groupby("_trip", sort=False):
-            prev: float | None = None
-            for idx, value in group["_seq"].items():
-                if pd.isna(value):
-                    continue
-                if prev is not None and value <= prev:
-                    bad.append(idx)
-                prev = value
-        return bad
+        work = work[work["_trip"].notna() & work["_seq"].notna()]
+        repeated = work.duplicated(subset=["_trip", "_seq"], keep="first")
+        return work.index[repeated].tolist()
 
     def _check_cross_file_reference(
         self, df: pd.DataFrame, mapping: ColumnMapping, rule: Rule

tests/test_domain_rules_transport_icpn.py

@@ -0,0 +1,126 @@
+"""Regression tests for GTFS-ST004 stop_sequence handling (#319) and strict ICPN format (#321)."""
+
+from __future__ import annotations
+
+import warnings
+
+import pandas as pd
+import pytest
+
+import freshdata as fd
+from freshdata.domains import run_domain
+from freshdata.domains.media.validator import is_valid_icpn
+
+
+def _stop_times(trip_ids, stop_sequence):
+    n = len(trip_ids)
+    times = [f"08:{i:02d}:00" for i in range(n)]
+    return pd.DataFrame(
+        {
+            "trip_id": trip_ids,
+            "arrival_time": times,
+            "departure_time": times,
+            "stop_id": [f"s{i}" for i in range(n)],
+            "stop_sequence": stop_sequence,
+        }
+    )
+
+
+def _violations(df: pd.DataFrame) -> dict:
+    with warnings.catch_warnings():
+        warnings.simplefilter("ignore")
+        _, outcome = run_domain(df, "transport", gtfs_file="stop_times")
+    return {r.rule_id: r.violation_rows for r in outcome.report.results if r.violated}
+
+
+# -- #319: GTFS-ST004 -------------------------------------------------------------
+
+
+def test_issue_319_repro_unsorted_rows_are_valid():
+    st = pd.DataFrame(
+        {
+            "trip_id": ["T", "T", "T"],
+            "arrival_time": ["08:00:00", "08:10:00", "08:05:00"],
+            "departure_time": ["08:00:00", "08:10:00", "08:05:00"],
+            "stop_id": ["a", "c", "b"],
+            "stop_sequence": [1, 3, 2],
+        }
+    )
+    assert "GTFS-ST004" not in _violations(st)
+    assert "GTFS-ST004" not in _violations(st.sort_values("stop_sequence"))
+
+
+def test_unsorted_interleaved_trips_with_gaps_pass():
+    st = _stop_times(["A", "B", "A", "B", "A"], [30, 2, 10, 1, 20])
+    assert "GTFS-ST004" not in _violations(st)
+
+
+def test_duplicated_stop_sequence_is_flagged():
+    st = _stop_times(["T", "T", "T", "U"], [1, 2, 1, 1])
+    # Row 2 repeats T's sequence 1; trip U's 1 is independent.
+    assert _violations(st).get("GTFS-ST004") == [2]
+
+
+def test_duplicate_detected_across_numeric_spellings():
+    st = _stop_times(["T", "T"], ["1", 1.0])
+    assert _violations(st).get("GTFS-ST004") == [1]
+
+
+def test_missing_trip_or_sequence_not_treated_as_duplicate():
+    st = _stop_times([None, None, "T", "T"], [1, 1, None, None])
+    assert "GTFS-ST004" not in _violations(st)
+
+
+def test_duplicate_flagged_via_clean_report():
+    st = _stop_times(["T", "T", "T"], [3, 1, 3])
+    _, rep = fd.clean(
+        st, domain="transport", gtfs_file="stop_times", return_report=True, verbose=False
+    )
+    st004 = [
+        f
+        for f in rep.domain_findings
+        if f["rule_id"] == "GTFS-ST004" and f["status"] == "violated"
+    ]
+    assert len(st004) == 1 and st004[0]["n_violations"] == 1
+
+
+# -- #321: is_valid_icpn ----------------------------------------------------------
+
+
+def test_issue_321_repro():
+    vals = ["tel: 036000291452", "call 0360-0029-1452 now", "036000291452"]
+    assert [is_valid_icpn(v) for v in vals] == [False, False, True]
+
+
+@pytest.mark.parametrize(
+    "value",
+    [
+        "036000291452",  # UPC-A
+        "4006381333931",  # EAN-13
+        "0-36000-29145-2",  # hyphen-grouped UPC
+        "0 36000 29145 2",  # space-grouped UPC
+        "400 6381 333931",  # space-grouped EAN
+        "  036000291452  ",  # surrounding whitespace
+    ],
+)
+def test_formatted_valid_icpn_accepted(value):
+    assert is_valid_icpn(value)
+
+
+@pytest.mark.parametrize(
+    "value",
+    [
+        "tel: 036000291452",
+        "call 0360-0029-1452 now",
+        "UPC036000291452",
+        "036000291452x",
+        "-036000291452",
+        "036000291452-",
+        "0360.0029.1452",
+        "0360_0029_1452",
+        "036000291452\n4006381333931",
+        "",
+    ],
+)
+def test_icpn_with_extra_text_rejected(value):
+    assert not is_valid_icpn(value)