Basic CI

.github/workflows/ci.yaml

@@ -0,0 +1,24 @@
+name: CI
+on: [push, pull_request]
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+
+      - name: Install uv
+        uses: astral-sh/setup-uv@v6
+
+      - name: "Set up Python"
+        uses: actions/setup-python@v5
+        with:
+          python-version-file: "pyproject.toml"
+
+      - name: Install dependencies
+        run: uv sync --locked --dev
+
+      - name: Run Ruff
+        run: uv run ruff check --output-format=github .
+
+      - name: Run mypy
+        run: uv run mypy .

.pre-commit-config.yaml

@@ -0,0 +1,25 @@
+repos:
+  - repo: https://github.com/astral-sh/uv-pre-commit
+    # uv version.
+    rev: 0.7.20
+    hooks:
+      - id: uv-lock
+      - id: uv-sync
+  - repo: https://github.com/astral-sh/ruff-pre-commit
+    # Ruff version.
+    rev: v0.12.3
+    hooks:
+      # Run the linter.
+      - id: ruff-check
+        args: [--fix]
+      # Run the formatter.
+      - id: ruff-format
+  - repo: local
+    hooks:
+      - id: mypy
+        name: mypy
+        language: python
+        entry: "uv run --active mypy"
+        types: [python]
+        require_serial: true
+        verbose: true

pyproject.toml

@@ -27,4 +27,10 @@ funding = "https://opencollective.com/lockdown-systems"
 icewatch = "icewatch:main"
 
 [dependency-groups]
-dev = ["ruff>=0.12.3"]
+dev = [
+    "mypy>=1.16.1",
+    "pandas-stubs>=2.3.0.250703",
+    "pre-commit>=4.2.0",
+    "ruff>=0.12.3",
+    "types-requests>=2.32.4.20250611",
+]

src/icewatch/geocode_facilities.py

@@ -2,75 +2,102 @@
 """
 Geocode facilities from a JSON file using OpenStreetMap Nominatim, with caching.
 """
-import os
-import sys
+
+import argparse
 import json
+import os
 import time
-import argparse
-from pathlib import Path
 from datetime import datetime
+from pathlib import Path
+
 import requests
 
 CACHE_FILENAME = "geocode_cache.json"
 NOMINATIM_URL = "https://nominatim.openstreetmap.org/search"
 USER_AGENT = "icewatch/1.0 (collective@lockdown.systems)"
 
 
-def load_json(path):
-    with open(path, 'r', encoding='utf-8') as f:
+def load_json(path: Path | str) -> dict:
+    with open(path, "r", encoding="utf-8") as f:
         return json.load(f)
 
-def save_json(data, path):
-    with open(path, 'w', encoding='utf-8') as f:
+
+def save_json(data: dict, path: Path | str) -> None:
+    with open(path, "w", encoding="utf-8") as f:
         json.dump(data, f, indent=2, ensure_ascii=False)
 
-def load_cache(cache_path):
+
+def load_cache(cache_path: Path | str) -> dict:
     if os.path.exists(cache_path):
-        with open(cache_path, 'r', encoding='utf-8') as f:
+        with open(cache_path, "r", encoding="utf-8") as f:
             return json.load(f)
     return {}
 
-def save_cache(cache, cache_path):
-    with open(cache_path, 'w', encoding='utf-8') as f:
+
+def save_cache(cache: dict, cache_path: Path | str) -> None:
+    with open(cache_path, "w", encoding="utf-8") as f:
         json.dump(cache, f, indent=2, ensure_ascii=False)
 
-def build_address(facility):
-    parts = [facility.get('Address', ''), facility.get('City', ''), facility.get('State', ''), str(facility.get('Zip', ''))]
-    return ', '.join([str(p).strip() for p in parts if p and str(p).strip()])
 
-def geocode_address(address, session=None):
+def build_address(facility: dict) -> str:
+    parts = [
+        facility.get("Address", ""),
+        facility.get("City", ""),
+        facility.get("State", ""),
+        str(facility.get("Zip", "")),
+    ]
+    return ", ".join([str(p).strip() for p in parts if p and str(p).strip()])
+
+
+def geocode_address(
+    address: str, session: requests.Session | None = None
+) -> dict | None:
     params = {
-        'q': address,
-        'format': 'json',
-        'limit': 1,
+        "q": address,
+        "format": "json",
+        "limit": "1",  # set as string to match request param type
     }
-    headers = {'User-Agent': USER_AGENT}
+    headers = {"User-Agent": USER_AGENT}
     s = session or requests.Session()
     response = s.get(NOMINATIM_URL, params=params, headers=headers, timeout=15)
     response.raise_for_status()
     results = response.json()
     if results:
-        return {
-            'lat': float(results[0]['lat']),
-            'lon': float(results[0]['lon'])
-        }
+        return {"lat": float(results[0]["lat"]), "lon": float(results[0]["lon"])}
     return None
 
+
 def main():
-    parser = argparse.ArgumentParser(description="Geocode facilities JSON using OpenStreetMap Nominatim with caching.")
-    parser.add_argument('--input', required=True, help='Input facilities JSON file')
-    parser.add_argument('--output', help='Output JSON file (default: facilities_geocoded_TIMESTAMP.json in same dir)')
-    parser.add_argument('--cache', help='Geocode cache file (default: geocode_cache.json in same dir as input)')
-    parser.add_argument('--delay', type=float, default=2, help='Delay between API requests (seconds, default: 2)')
+    parser = argparse.ArgumentParser(
+        description="Geocode facilities JSON using OpenStreetMap Nominatim with caching."
+    )
+    parser.add_argument("--input", required=True, help="Input facilities JSON file")
+    parser.add_argument(
+        "--output",
+        help="Output JSON file (default: facilities_geocoded_TIMESTAMP.json in same dir)",
+    )
+    parser.add_argument(
+        "--cache",
+        help="Geocode cache file (default: geocode_cache.json in same dir as input)",
+    )
+    parser.add_argument(
+        "--delay",
+        type=float,
+        default=2,
+        help="Delay between API requests (seconds, default: 2)",
+    )
     args = parser.parse_args()
 
     input_path = Path(args.input)
-    output_path = args.output or str(input_path.parent / f"facilities_geocoded_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json")
+    output_path = args.output or str(
+        input_path.parent
+        / f"facilities_geocoded_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
+    )
     cache_path = args.cache or str(input_path.parent / CACHE_FILENAME)
 
     print(f"Loading facilities from: {input_path}")
     data = load_json(input_path)
-    facilities = data.get('facilities', [])
+    facilities = data.get("facilities", [])
 
     print(f"Loading geocode cache from: {cache_path}")
     cache = load_cache(cache_path)
@@ -80,15 +107,15 @@ def main():
     for i, facility in enumerate(facilities):
         address = build_address(facility)
         if not address:
-            print(f"[{i+1}/{len(facilities)}] No address for facility, skipping.")
-            facility['latitude'] = None
-            facility['longitude'] = None
+            print(f"[{i + 1}/{len(facilities)}] No address for facility, skipping.")
+            facility["latitude"] = None
+            facility["longitude"] = None
             continue
         if address in cache and cache[address] is not None:
             result = cache[address]
-            print(f"[{i+1}/{len(facilities)}] Cached: {address} -> {result}")
+            print(f"[{i + 1}/{len(facilities)}] Cached: {address} -> {result}")
         else:
-            print(f"[{i+1}/{len(facilities)}] Geocoding: {address}")
+            print(f"[{i + 1}/{len(facilities)}] Geocoding: {address}")
             try:
                 result = geocode_address(address, session=session)
                 time.sleep(args.delay)
@@ -102,11 +129,11 @@ def main():
                 # Remove failed/None result from cache if present
                 del cache[address]
         if result:
-            facility['latitude'] = result['lat']
-            facility['longitude'] = result['lon']
+            facility["latitude"] = result["lat"]
+            facility["longitude"] = result["lon"]
         else:
-            facility['latitude'] = None
-            facility['longitude'] = None
+            facility["latitude"] = None
+            facility["longitude"] = None
 
     print(f"Writing geocoded facilities to: {output_path}")
     save_json(data, output_path)
@@ -119,5 +146,6 @@ def main():
 
     print("Done.")
 
+
 if __name__ == "__main__":
-    main()
+    main()

src/icewatch/ice_detention_scraper.py

@@ -10,15 +10,17 @@
 Date: 2024
 """
 
-import requests
+import logging
 import os
-import sys
 import re
+import sys
 from datetime import datetime
 from pathlib import Path
-import logging
+from typing import TypedDict
 from urllib.parse import urljoin, urlparse
-from bs4 import BeautifulSoup
+
+import requests
+from bs4 import BeautifulSoup, Tag
 
 # Configure logging
 logging.basicConfig(
@@ -82,8 +84,8 @@ def extract_date_from_filename(url: str) -> str | None:
 
 
 def find_detention_stats_link(
-    base_url="https://www.ice.gov/detain/detention-management",
-):
+    base_url: str = "https://www.ice.gov/detain/detention-management",
+) -> str | None:
     """
     Scrape the ICE detention management page to find the latest statistics download link.
 
@@ -126,11 +128,17 @@ def find_detention_stats_link(
             "FY2025",
         ]
 
-        found_links = []
+        class RelevantLink(TypedDict):
+            url: str
+            text: str
+            relevance_score: int
+
+        found_links: list[RelevantLink] = []
 
         # Search for all links on the page
         for link in soup.find_all("a", href=True):
-            href = link.get("href", "").lower()
+            assert isinstance(link, Tag)  # this pleases mypy
+            href = str(link.get("href", "")).lower()
             text = link.get_text().lower()
 
             # Check if link text or href contains relevant keywords
@@ -140,7 +148,7 @@ def find_detention_stats_link(
             )
 
             if is_relevant:
-                full_url = urljoin(base_url, link["href"])
+                full_url = urljoin(base_url, str(link["href"]))
                 link_text = link.get_text().strip()
                 found_links.append(
                     {
@@ -179,7 +187,11 @@ def find_detention_stats_link(
         return None
 
 
-def download_ice_detention_stats(url=None, output_dir="data", auto_find_link=True):
+def download_ice_detention_stats(
+    url: str | None = None,
+    output_dir: str = "data",
+    auto_find_link: bool = True,
+) -> tuple[str | None, str | None]:
     """
     Download ICE detention statistics Excel file.
 
@@ -268,7 +280,7 @@ def download_ice_detention_stats(url=None, output_dir="data", auto_find_link=Tru
                         progress = (downloaded_size / file_size) * 100
                         logger.info(f"Download progress: {progress:.1f}%")
 
-        logger.info(f"Download completed successfully!")
+        logger.info("Download completed successfully!")
         logger.info(f"File saved to: {filepath}")
         logger.info(f"File size: {os.path.getsize(filepath) / 1024:.1f} KB")
         if source_date: