Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 4 additions & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -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"}
Expand All @@ -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"]
53 changes: 53 additions & 0 deletions src/postal_regex/__init__.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,53 @@
# 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,
get_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"
]
114 changes: 114 additions & 0 deletions src/postal_regex/analytics.py
Original file line number Diff line number Diff line change
@@ -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")
30 changes: 30 additions & 0 deletions src/postal_regex/cli.py
Original file line number Diff line number Diff line change
@@ -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()
3 changes: 1 addition & 2 deletions src/postal_regex/core.py
Original file line number Diff line number Diff line change
Expand Up @@ -71,8 +71,7 @@ def validate(country_identifier: str, postal_code: str, timeout: float = 0.1) ->
try:
return bool(entry.regex.fullmatch(postal_code, timeout=timeout))
except regex.TimeoutError:
return False # treat timeout as invalid

return False

def get_supported_countries():
"""
Expand Down
83 changes: 83 additions & 0 deletions tests/test_analytics.py
Original file line number Diff line number Diff line change
@@ -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()