feat(analytics): add local validation analytics and CLI

pyproject.toml

@@ -16,6 +16,9 @@ dependencies = [
     "regex==2025.9.18"
 ]
 
+[project.scripts]
+postal-regex = "postal_regex.cli:main"
+
 [tool.setuptools]
 packages = ["postal_regex"]
 package-dir = {"" = "src"}
@@ -34,4 +37,4 @@ update_changelog_on_bump = true
 pandas = ["pandas>=2.0", "pyarrow>=11.0"]
 spark = ["pyspark>=4.0"]
 dask = ["dask[complete]>=2025.9.0"]
-dataframe = ["pandas>=2.0", "pyarrow>=11.0", "dask[complete]>=2025.9.0"]
+dataframe = ["pandas>=2.0", "pyarrow>=11.0", "dask[complete]>=2025.9.0"]

src/postal_regex/__init__.py

@@ -0,0 +1,52 @@
+# Core validation and lookup functions
+from .core import (
+    validate,
+    normalize,
+    get_entry,
+    get_country_regex,
+    get_supported_countries,
+)
+
+# Bulk processing utilities for DataFrames and iterables
+from .bulk import (
+    bulk_validate,
+    bulk_normalize,
+    validate_dataframe,
+    validate_spark_dataframe,
+    load_json,
+    load_pandas,
+    load_spark,
+    export_csv,
+    export_parquet,
+)
+
+# Local analytics and statistics functions
+from .analytics import (
+    record_validation,
+    show_stats,
+    reset_stats,
+)
+
+__all__ = [
+    # Core
+    "validate",
+    "normalize",
+    "get_entry",
+    "get_country_regex",
+    "get_supported_countries",
+    # Bulk
+    "bulk_validate",
+    "bulk_normalize",
+    "validate_dataframe",
+    "validate_spark_dataframe",
+    "load_json",
+    "load_pandas",
+    "load_spark",
+    "export_csv",
+    "export_parquet",
+    # Analytics
+    "record_validation",
+    "show_stats",
+    "reset_stats",
+    "get_stats"
+]

src/postal_regex/analytics.py

@@ -0,0 +1,114 @@
+import json
+from pathlib import Path
+
+STATS_FILE = Path.home() / ".postalregex_stats.json"
+
+def _load_stats():
+    """Load statistics from the local JSON file."""
+    if not STATS_FILE.exists():
+        return {}
+    with open(STATS_FILE, "r") as f:
+        try:
+            return json.load(f)
+        except json.JSONDecodeError:
+            return {}
+
+def _save_stats(stats):
+    """Save statistics to the local JSON file."""
+    with open(STATS_FILE, "w") as f:
+        json.dump(stats, f, indent=2)
+
+def record_validation(country_code, is_valid):
+    """
+    Record a validation attempt for a given country.
+
+    Args:
+        country_code (str): The ISO 3166-1 alpha-2 country code.
+        is_valid (bool): True if the validation was successful, False otherwise.
+    """
+    stats = _load_stats()
+
+    # Ensure the country entry exists
+    if country_code not in stats:
+        stats[country_code] = {"valid": 0, "invalid": 0}
+
+    # Increment the appropriate counter
+    if is_valid:
+        stats[country_code]["valid"] += 1
+    else:
+        stats[country_code]["invalid"] += 1
+    _save_stats(stats)
+
+def reset_stats():
+    """Clear all recorded statistics."""
+    if STATS_FILE.exists():
+        STATS_FILE.unlink()
+        print("Local validation statistics have been reset.")
+    else:
+        print("No statistics file found to reset.")
+
+def get_stats() -> dict:
+    """
+    Loads and returns the validation statistics from the stats file.
+    This function only retrieves data and does not print anything.
+
+    Returns:
+        dict: A dictionary containing the validation stats, or an empty dict if none exist.
+    """
+    if not STATS_FILE.exists():
+        return {}
+    try:
+        with open(STATS_FILE, "r") as f:
+            return json.load(f)
+    except (json.JSONDecodeError, FileNotFoundError):
+        return {}
+
+def show_stats():
+    """
+    Loads and prints a formatted dashboard of the validation statistics.
+    This function handles all presentation logic.
+    """
+    stats = get_stats() # This is the main change: call the new data function
+    if not stats:
+        print("No validation statistics recorded yet.")
+        return
+
+    # Prepare data for the table
+    table_data = []
+    for country, counts in stats.items():
+        valid = counts.get("valid", 0)
+        invalid = counts.get("invalid", 0)
+        total = valid + invalid
+        table_data.append([country, valid, invalid, total])
+
+    # Sort by total validations
+    table_data.sort(key=lambda row: row[3], reverse=True)
+
+    # --- Print Table ---
+    print("\nPostal Code Validation Stats (Local Project)")
+    header = ["Country", "Valid", "Invalid", "Total"]
+    col_widths = [len(h) for h in header]
+    for row in table_data:
+        for i, cell in enumerate(row):
+            col_widths[i] = max(col_widths[i], len(str(cell)))
+
+    header_line = " | ".join(h.ljust(w) for h, w in zip(header, col_widths))
+    separator = "-+-".join("-" * w for w in col_widths)
+    print(header_line)
+    print(separator)
+
+    for row in table_data:
+        row_line = " | ".join(str(cell).ljust(w) for cell, w in zip(row, col_widths))
+        print(row_line)
+
+    # --- Print Bar Chart ---
+    print("\nValidation Success Rate")
+    for country, valid, invalid, total in table_data:
+        if total == 0:
+            success_rate = 0
+        else:
+            success_rate = (valid / total) * 100
+        bar_length = 40
+        filled_length = int(bar_length * success_rate / 100)
+        bar = "█" * filled_length + " " * (bar_length - filled_length)
+        print(f"{country.ljust(5)} |{bar}| {success_rate:.0f}% valid")

src/postal_regex/cli.py

@@ -0,0 +1,30 @@
+import argparse
+from . import analytics
+
+def main():
+    """Main function for the command-line interface."""
+    parser = argparse.ArgumentParser(
+        description="Postal Regex command-line tools."
+    )
+    subparsers = parser.add_subparsers(dest="command", help="Available commands")
+
+    # The 'stats' command
+    stats_parser = subparsers.add_parser(
+        "stats", help="Display local validation statistics."
+    )
+    stats_parser.add_argument(
+        "--reset", action="store_true", help="Reset all recorded statistics."
+    )
+
+    args = parser.parse_args()
+
+    if args.command == "stats":
+        if args.reset:
+            analytics.reset_stats()
+        else:
+            analytics.show_stats()
+    else:
+        parser.print_help()
+
+if __name__ == "__main__":
+    main()

src/postal_regex/core.py

@@ -3,6 +3,7 @@
 from functools import lru_cache
 from dataclasses import dataclass
 from importlib.resources import files
+from . import analytics
 
 # ---------------------------
 # Data Loading
@@ -67,11 +68,13 @@ def validate(country_identifier: str, postal_code: str, timeout: float = 0.1) ->
     Validate a postal code against the regex pattern for a given country.
     Timeout (default 100ms) prevents ReDoS hangs.
     """
-    entry = get_entry(country_identifier)
     try:
-        return bool(entry.regex.fullmatch(postal_code, timeout=timeout))
-    except regex.TimeoutError:
-        return False  # treat timeout as invalid
+        entry = get_entry(country_identifier)
+        is_valid = bool(entry.regex.fullmatch(postal_code, timeout=timeout))
+    except (regex.TimeoutError, ValueError):
+        is_valid = False
+
+    return is_valid
 
 
 def get_supported_countries():

tests/test_analytics.py

@@ -0,0 +1,83 @@
+import json
+from pathlib import Path
+import sys
+
+# Add src folder to sys.path
+PROJECT_ROOT = Path(__file__).resolve().parent.parent
+SRC_DIR = PROJECT_ROOT / "src"
+sys.path.insert(0, str(SRC_DIR))
+
+from postal_regex import analytics
+
+def test_record_validation_and_load_stats(monkeypatch):
+    """
+    Test that recording validations correctly creates and updates the stats file.
+    """
+    temp_stats_file = Path.home() / ".postalregex_stats_temp.json"
+    monkeypatch.setattr(analytics, 'STATS_FILE', temp_stats_file)
+
+    # Ensure the file doesn't exist initially
+    if temp_stats_file.exists():
+        temp_stats_file.unlink()
+
+    analytics.record_validation("US", is_valid=True)
+
+    with open(temp_stats_file, "r") as f:
+        stats = json.load(f)
+
+    assert stats["US"]["valid"] == 1
+    assert stats["US"]["invalid"] == 0
+
+    analytics.record_validation("US", is_valid=False)
+
+    with open(temp_stats_file, "r") as f:
+        stats = json.load(f)
+
+    assert stats["US"]["valid"] == 1
+    assert stats["US"]["invalid"] == 1
+
+    analytics.record_validation("CA", is_valid=True)
+
+    with open(temp_stats_file, "r") as f:
+        stats = json.load(f)
+
+    assert stats["CA"]["valid"] == 1
+    assert "invalid" in stats["CA"] # an 'invalid' key should be created
+
+    temp_stats_file.unlink()
+
+def test_get_stats(monkeypatch):
+    """
+    Test that get_stats correctly reads and returns data.
+    """
+    temp_stats_file = Path.home() / ".postalregex_stats_temp.json"
+    monkeypatch.setattr(analytics, 'STATS_FILE', temp_stats_file)
+
+    # 1. Test when file doesn't exist
+    if temp_stats_file.exists():
+        temp_stats_file.unlink()
+    assert analytics.get_stats() == {}
+
+    dummy_data = {"US": {"valid": 5, "invalid": 1}}
+    with open(temp_stats_file, "w") as f:
+        json.dump(dummy_data, f)
+
+    assert analytics.get_stats() == dummy_data
+
+    temp_stats_file.unlink()
+
+def test_reset_stats(monkeypatch):
+    """
+    Test that reset_stats correctly deletes the statistics file.
+    """
+    temp_stats_file = Path.home() / ".postalregex_stats_temp.json"
+    monkeypatch.setattr(analytics, 'STATS_FILE', temp_stats_file)
+
+    with open(temp_stats_file, "w") as f:
+        json.dump({"US": {"valid": 1, "invalid": 0}}, f)
+
+    assert temp_stats_file.exists()
+
+    analytics.reset_stats()
+
+    assert not temp_stats_file.exists()