fix(pipeline): count hot-list and domain reporters on one UTC day

The Worker's reporter buckets are HMACs of the UTC day, so one reporter
whose reports straddle midnight shows up as two, or four with a Wi-Fi and a
mobile network. The hot list counted identities over a rolling 24 hours
(three needed) and the domain feed over its whole queue (two needed), so
two people either side of midnight could put a number on every phone's hot
list. Both now count reporters on the busiest UTC day, the same rule the
community merge uses, and the /48 device cap still applies within the day.

A domain report without a parseable reported_at no longer counts, since its
bucket can't be placed on a day. The Worker always stamps one.

Author: SysAdminDoc

CHANGELOG.md

@@ -43,6 +43,10 @@ All notable changes to CallShield will be documented in this file.
   version, Android version, phone and Protection test result.
 - Database v49 re-reads FCC's newest 500,000 complaints and adds 200 new FTC
   ones, for 51,362 numbers.
+- The hot list and the spam domain feed count reporters on one UTC day. The
+  report Worker's reporter ID changes at midnight, so two people reporting
+  either side of it counted as four and could put a number on every phone's
+  hot list.
 - Calls from outside North America no longer get a US or Caribbean place name
   in the caller-ID popup. A +65 call used to read as Birmingham, AL, and Number
   details offered to block that US area code. On a phone outside North America,

README.md

@@ -334,8 +334,8 @@ scrape Nomorobo's restricted carrier feed.
 ### Hot List (checked every 30 minutes, regenerated by hand)
 | File | Contents |
 |------|----------|
-| `hot_numbers.json` | Top 500 trending numbers (last 24h) |
-| `hot_ranges.json` | NPA-NXX prefixes where at least 4 trending numbers drew reports from at least 6 reporters between them |
+| `hot_numbers.json` | Top 500 trending numbers (last 24h, with three reporters on the same UTC day) |
+| `hot_ranges.json` | NPA-NXX prefixes where at least 4 trending numbers drew reports from at least 6 reporters between them on one UTC day |
 | `spam_domains.json` | Phishing/spam domains from community SMS reports |
 
 ### Real-Time Lookup (overlay only)

scripts/extract_spam_domains.py

@@ -30,7 +30,13 @@
     parse_cleared_feeds,
     report_queue_digest,
 )
-from report_dedup import validated_report_id, validated_reporter_bucket
+from report_dedup import (
+    busiest_day_reporters,
+    parse_reported_at,
+    reporter_day_key,
+    validated_report_id,
+    validated_reporter_bucket,
+)
 
 DATA_DIR = Path(os.environ.get("CALLSHIELD_DATA_DIR", Path(__file__).parent.parent / "data"))
 REPORTS_DIR = Path(os.environ.get("CALLSHIELD_REPORTS_DIR", DATA_DIR / "reports"))
@@ -167,8 +173,10 @@ def main(argv: list[str] | None = None) -> int:
     domain_numbers: dict[str, set[str]] = {}
     # Reporters count per /48 group here, not per /64 device as on the hot list:
     # with MIN_REPORTERS at 2, the hot list's two-devices-per-/48 cap would let
-    # one /48 meet the quorum alone.
-    domain_reporters: dict[str, set[str]] = {}
+    # one /48 meet the quorum alone. Each is kept with its UTC day, because the
+    # bucket changes at midnight and only one day's buckets are known to be
+    # different people.
+    domain_reporters: dict[str, set[tuple[str, str]]] = {}
     # A report the app resent under the same id (from another network, so from
     # another reporter bucket) must not count as a second reporter.
     seen_report_ids: set[str] = set()
@@ -188,8 +196,11 @@ def main(argv: list[str] | None = None) -> int:
                     continue
 
                 number = validated_report_number(report.get("number", ""))
-                reporter_bucket = validated_reporter_bucket(report.get("reporter_bucket"))
-                if not number or not reporter_bucket:
+                reporter_day = reporter_day_key(
+                    validated_reporter_bucket(report.get("reporter_bucket")),
+                    parse_reported_at(report.get("reported_at")),
+                )
+                if not number or reporter_day is None:
                     continue
                 report_id = validated_report_id(report.get("report_id"))
                 if report_id:
@@ -200,7 +211,7 @@ def main(argv: list[str] | None = None) -> int:
                 reports_scanned += 1
                 for domain in domains:
                     domain_numbers.setdefault(domain, set()).add(number)
-                    domain_reporters.setdefault(domain, set()).add(reporter_bucket)
+                    domain_reporters.setdefault(domain, set()).add((reporter_day[1], reporter_day[0]))
 
             except Exception as e:
                 print(f"  Skipping {report_file.name}: {e}")
@@ -212,7 +223,7 @@ def main(argv: list[str] | None = None) -> int:
     candidates = [
         domain
         for domain, count in domain_counts.most_common()
-        if count >= MIN_REPORTS and len(domain_reporters.get(domain, set())) >= MIN_REPORTERS
+        if count >= MIN_REPORTS and busiest_day_reporters(domain_reporters.get(domain, set()), count=len) >= MIN_REPORTERS
     ]
     try:
         approved = approved_domains()
@@ -229,7 +240,7 @@ def main(argv: list[str] | None = None) -> int:
             {
                 "domain": domain,
                 "distinct_numbers": domain_counts[domain],
-                "distinct_reporters": len(domain_reporters[domain]),
+                "distinct_reporters": busiest_day_reporters(domain_reporters[domain], count=len),
             }
             for domain in candidates
             if domain not in approved

scripts/generate_hot_list.py

@@ -30,7 +30,7 @@
 from report_dedup import (
     BURST_DUPLICATE_SECONDS,
     MAX_DEVICES_PER_GROUP,
-    capped_reporter_count,
+    busiest_day_reporters,
     find_burst_duplicates,
     find_resent_reports,
     parse_reported_at,
@@ -187,6 +187,7 @@ def main(argv: list[str] | None = None) -> int:
             collapsed_group += 1
             continue
         seen_reporter_days.add(identity_key)
+        dated_identity = (day_key[1], identity)
         entry = velocity.get(number)
         if entry is None:
             velocity[number] = {
@@ -199,13 +200,13 @@ def main(argv: list[str] | None = None) -> int:
                 "first_seen": reported_at_str,
                 "last_seen": reported_at_str,
                 "description": "Trending community report",
-                "_reporters": {identity},
+                "_reporters": {dated_identity},
                 "_devices_by_group": {group: {device}},
                 "_times": [reported_at],
             }
         else:
             entry["reports"] += 1
-            entry["_reporters"].add(identity)
+            entry["_reporters"].add(dated_identity)
             entry["_devices_by_group"].setdefault(group, set()).add(device)
             entry["_times"].append(reported_at)
             if reported_at_str > entry["last_seen"]:
@@ -263,7 +264,7 @@ def main(argv: list[str] | None = None) -> int:
         if not times:
             continue
         span_minutes = int((max(times) - min(times)).total_seconds() // 60)
-        entry["distinct_reporters"] = capped_reporter_count(reporters)
+        entry["distinct_reporters"] = busiest_day_reporters(reporters)
         entry["report_span_minutes"] = span_minutes
         if (
             entry["reports"] >= MIN_REPORTS_HOT
@@ -299,7 +300,7 @@ def main(argv: list[str] | None = None) -> int:
     for entry in hot_internal:
         if (
             entry["reports"] < CAMPAIGN_REPORTS_PER_NUMBER
-            or capped_reporter_count(entry["_reporters"]) < CAMPAIGN_REPORTERS_PER_NUMBER
+            or busiest_day_reporters(entry["_reporters"]) < CAMPAIGN_REPORTERS_PER_NUMBER
             or entry["report_span_minutes"] < CAMPAIGN_MIN_SPAN_MINUTES
         ):
             continue
@@ -310,7 +311,7 @@ def main(argv: list[str] | None = None) -> int:
     hot_ranges = []
     for npanxx, entries in sorted(npanxx_entries.items(), key=lambda item: -len(item[1])):
         # The cap applies across the range too, so one /48 can't fill the union.
-        union_reporters = capped_reporter_count(set().union(*(entry["_reporters"] for entry in entries)))
+        union_reporters = busiest_day_reporters(set().union(*(entry["_reporters"] for entry in entries)))
         if len(entries) < CAMPAIGN_THRESHOLD or union_reporters < CAMPAIGN_MIN_UNION_REPORTERS:
             continue
         hot_ranges.append(

scripts/report_dedup.py

@@ -64,6 +64,20 @@ def capped_reporter_count(identities) -> int:
     return sum(min(len(devices), MAX_DEVICES_PER_GROUP) for devices in devices_by_group.values())
 
 
+def busiest_day_reporters(dated_identities, count=capped_reporter_count) -> int:
+    """Reporters shown to be different people: the count on the busiest UTC day.
+
+    The Worker derives every bucket from the UTC day, so one reporter's reports
+    either side of midnight carry different buckets. Only identities seen on the
+    same day are known to be different reporters. `dated_identities` holds
+    `(day, identity)` pairs and `count` counts one day's identities.
+    """
+    by_day: dict[str, set] = {}
+    for day, identity in dated_identities:
+        by_day.setdefault(day, set()).add(identity)
+    return max((count(identities) for identities in by_day.values()), default=0)
+
+
 def validated_report_id(value: object) -> str:
     """Return the id the app gave a report, lowercased, or an empty string."""
     if not isinstance(value, str):

scripts/test_report_dedup.py

@@ -7,6 +7,7 @@
 from report_dedup import (
     BURST_DUPLICATE_SECONDS,
     MAX_DEVICES_PER_GROUP,
+    busiest_day_reporters,
     capped_reporter_count,
     find_burst_duplicates,
     find_resent_reports,
@@ -174,6 +175,18 @@ def main() -> None:
     for minimum in (MIN_REPORTERS_HOT, CAMPAIGN_REPORTERS_PER_NUMBER, CAMPAIGN_MIN_UNION_REPORTERS):
         assert minimum > MAX_DEVICES_PER_GROUP, minimum
 
+    # ── buckets rotate at UTC midnight, so only one day's count as distinct ──
+    before, after = "2026-07-29", "2026-07-30"
+    one_reporter_either_side = [(before, (group, device)), (after, ("00000000000a0009", "00000000000d0009"))]
+    assert busiest_day_reporters(one_reporter_either_side) == 1
+    three_on_one_day = [(after, (f"00000000000a000{n}", f"00000000000d000{n}")) for n in range(1, 4)]
+    # The day after midnight holds the three plus the second of those buckets.
+    assert busiest_day_reporters(three_on_one_day + one_reporter_either_side) == 4
+    # The /48 cap still applies within the day.
+    assert busiest_day_reporters([(after, identity) for identity in rotating]) == MAX_DEVICES_PER_GROUP
+    assert busiest_day_reporters([(before, "a"), (after, "b"), (after, "c")], count=len) == 2
+    assert busiest_day_reporters([]) == 0
+
     print("report_dedup tests passed")
 
 

scripts/test_report_pipeline.py

@@ -795,6 +795,42 @@ def assert_reporters_count_per_device_with_group_cap(data_dir: Path) -> None:
         raise AssertionError(f"one /48 rotating /64s must count as two reporters at most: {hot[delegated_number]}")
 
 
+def assert_reporters_count_on_one_utc_day(data_dir: Path) -> None:
+    """The Worker's buckets change at UTC midnight, so one reporter either side of it counts once."""
+    previous_day = (datetime.fromisoformat(BASE_DAY) - timedelta(days=1)).date().isoformat()
+    straddle = [f"{previous_day}T22:00:00+00:00", f"{previous_day}T23:30:00+00:00",
+                f"{BASE_DAY}T01:00:00+00:00", f"{BASE_DAY}T03:00:00+00:00"]
+    two_either_side, third_same_day = "+12122340301", "+12122340302"
+    for number, times in ((two_either_side, straddle), (third_same_day, straddle + [f"{BASE_DAY}T05:00:00+00:00"])):
+        for index, reported_at in enumerate(times, start=1):
+            write_report(data_dir, f"{number[1:]}-{index}.json", number, f"00000000000b{index:04x}", reported_at,
+                         device=f"00000000000c{index:04x}")
+    # A domain seen by one bucket before midnight and another after, and one seen
+    # by two buckets on the same day.
+    for index, (domain, reported_at, bucket) in enumerate([
+        ("straddle.example", f"{previous_day}T23:50:00+00:00", BUCKETS[0]),
+        ("straddle.example", f"{previous_day}T23:55:00+00:00", BUCKETS[0]),
+        ("straddle.example", f"{BASE_DAY}T00:10:00+00:00", BUCKETS[1]),
+        ("sameday.example", f"{BASE_DAY}T01:00:00+00:00", BUCKETS[0]),
+        ("sameday.example", f"{BASE_DAY}T01:30:00+00:00", BUCKETS[0]),
+        ("sameday.example", f"{BASE_DAY}T02:00:00+00:00", BUCKETS[1]),
+    ]):
+        write_report(data_dir, f"domain-{index}.json", f"+1212234032{index}", bucket, reported_at,
+                     report_type="sms_spam", domains=[domain])
+
+    run_script("generate_hot_list.py", data_dir, ["--allow-collapse", "--cleared", "numbers,ranges"])
+    run_script("extract_spam_domains.py", data_dir, ["--allow-collapse", "--cleared", "domains"])
+    hot = {entry["number"]: entry for entry in json.loads((data_dir / "hot_numbers.json").read_text(encoding="utf-8"))["numbers"]}
+    if two_either_side in hot:
+        raise AssertionError(f"two reporters either side of midnight were counted as four: {hot[two_either_side]}")
+    if hot.get(third_same_day, {}).get("distinct_reporters") != 3:
+        raise AssertionError(f"three reporters on one day should trend: {hot.get(third_same_day)}")
+    review = json.loads((data_dir / "spam_domains_review.json").read_text(encoding="utf-8"))
+    reporters = {candidate["domain"]: candidate["distinct_reporters"] for candidate in review["candidates"]}
+    if "straddle.example" in reporters or reporters.get("sameday.example") != 2:
+        raise AssertionError(f"domain reporters should be counted on one UTC day: {reporters}")
+
+
 def run_drain(data_dir: Path) -> None:
     # Each drain here leaves the derived feeds empty, which their collapse
     # guards refuse to publish without being told, and an empty feed has to be
@@ -990,6 +1026,9 @@ def main() -> None:
     with tempfile.TemporaryDirectory() as tmp:
         assert_reporters_count_per_device_with_group_cap(Path(tmp) / "data")
 
+    with tempfile.TemporaryDirectory() as tmp:
+        assert_reporters_count_on_one_utc_day(Path(tmp) / "data")
+
 
 if __name__ == "__main__":
     main()