filter_users.py

"""User filtering CLI: filter by name, age, or email.

Usage examples:
    python users_filter.py --file users.json --by name  --query "Alice"
    python users_filter.py --file users.json --by age   --query ">=18"
    python users_filter.py --file users.json --by email --query "@example.com"

If --by / --query are omitted, the script will prompt interactively.
"""

from __future__ import annotations

import argparse
import json
import operator
from pathlib import Path
from typing import Any, Callable, List, Optional, TypedDict


# ----------------------------- Types & Schema -------------------------------

class User(TypedDict, total=False):
    """Minimal expected user schema in users.json."""
    name: str
    age: int
    email: str


# ----------------------------- I/O & Utilities ------------------------------

def read_users(file_path: str | Path) -> List[User]:
    """Load users from a JSON file; return [] on error or wrong shape."""
    try:
        p = Path(file_path)
        with p.open("r", encoding="utf-8") as f:
            data = json.load(f)
        return data if isinstance(data, list) else []
    except (OSError, json.JSONDecodeError) as exc:
        print(f"Error loading {file_path}: {exc}")
        return []


def print_users(users: List[User]) -> None:
    """Pretty-print users, one JSON object per line."""
    for user in users:
        print(json.dumps(user, ensure_ascii=False))


def _normalize_str(value: Any) -> Optional[str]:
    """Return a trimmed string or None if not a non-empty string."""
    if isinstance(value, str):
        s = value.strip()
        return s if s else None
    return None


def _to_int(value: Any) -> Optional[int]:
    """Coerce age-like values to int (int, integral float, or digit-string)."""
    if isinstance(value, int):
        return value
    if isinstance(value, float) and value.is_integer():
        return int(value)
    if isinstance(value, str) and value.strip().isdigit():
        return int(value.strip())
    return None


# ----------------------------- Age Filter Parser ----------------------------

def _parse_age_filter(expr: str) -> Callable[[Any], bool]:
    """Build a predicate from '30', '>=18', '<21', '20-29', '30+'."""
    s = (_normalize_str(expr) or "").replace(" ", "")

    # range: "a-b"
    if "-" in s and not s.startswith("-"):
        left, right = s.split("-", 1)
        try:
            a, b = int(left), int(right)
        except ValueError:
            return lambda _age: False

        def pred(age: Any) -> bool:
            n = _to_int(age)
            return n is not None and a <= n <= b

        return pred

    # "n+" shorthand
    if s.endswith("+"):
        try:
            n_min = int(s[:-1])
        except ValueError:
            return lambda _age: False

        def pred(age: Any) -> bool:
            n = _to_int(age)
            return n is not None and n >= n_min

        return pred

    # comparators
    for op_str, op_fn in ((">=", operator.ge), ("<=", operator.le),
                          (">", operator.gt), ("<", operator.lt)):
        if s.startswith(op_str):
            try:
                n_ref = int(s[len(op_str):])
            except ValueError:
                return lambda _age: False

            def pred(age: Any, n=n_ref, cmp=op_fn) -> bool:
                v = _to_int(age)
                return v is not None and cmp(v, n)

            return pred

    # exact int
    try:
        n_exact = int(s)
    except ValueError:
        return lambda _age: False

    def pred(age: Any, n=n_exact) -> bool:
        v = _to_int(age)
        return v is not None and v == n

    return pred


# ----------------------------- Filter Functions -----------------------------

def filter_users_by_name(users: List[User], name: str) -> List[User]:
    """Return users whose name matches exactly (case-insensitive)."""
    needle = _normalize_str(name)
    if not needle:
        return []
    needle_l = needle.lower()
    return [
        u for u in users
        if isinstance(u.get("name"), str)
        and u["name"].strip().lower() == needle_l
    ]


def filter_users_by_age(users: List[User], age_expression: str) -> List[User]:
    """Filter users by age using expressions like '30', '>=18', '<21', '20-29', '30+'."""
    pred = _parse_age_filter(age_expression)
    return [u for u in users if pred(u.get("age"))]


def filter_users_by_email(users: List[User], query: str) -> List[User]:
    """Filter by email (case-insensitive).

    Supported queries:
      - Exact address: "alice@example.com"
      - Domain only:   "@example.com" or "example.com"
    """
    q = _normalize_str(query)
    if not q:
        return []
    q_l = q.lower()

    results: List[User] = []
    for u in users:
        email = u.get("email")
        if not isinstance(email, str):
            continue
        e = email.strip().lower()
        if not e or "@" not in e:
            continue

        _local, _at, domain = e.partition("@")

        # Domain-only: "@example.com"
        if q_l.startswith("@"):
            dom_q = q_l[1:]
            if dom_q and domain == dom_q:
                results.append(u)
            continue

        # Domain-only: "example.com"
        if "@" not in q_l:
            if domain == q_l:
                results.append(u)
            continue

        # Exact address
        if e == q_l:
            results.append(u)

    return results


# --------------------------------- CLI --------------------------------------

def _build_arg_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        description="Filter users (from users.json) by name, age, or email."
    )
    parser.add_argument(
        "--file", "-f", default="users.json",
        help="Path to users JSON file (default: users.json)",
    )
    parser.add_argument(
        "--by", "-b", choices=("name", "age", "email"),
        help="Filter category to use.",
    )
    parser.add_argument(
        "--query", "-q",
        help="Filter value. Examples: 'Alice' | '>=18' | '20-29' | '@example.com'.",
    )
    parser.add_argument(
        "--interactive", "-i", action="store_true",
        help="Prompt for missing values (if --by/--query not provided).",
    )
    return parser


def _prompt_missing(by: str | None, query: str | None) -> tuple[str, str]:
    """Ask user for missing --by/--query interactively."""
    if not by:
        while True:
            by = input("Filter by ('name', 'age', 'email'): ").strip().lower()
            if by in {"name", "age", "email"}:
                break
            print("Please choose one of: name, age, email.")
    if not query:
        if by == "name":
            query = input("Enter name (exact match): ").strip()
        elif by == "age":
            query = input(
                "Enter age filter (e.g., '30', '>=18', '<21', '20-29', '30+'): "
            ).strip()
        else:
            query = input(
                "Enter email or domain (e.g., 'alice@example.com', "
                "'@example.com', 'example.com'): "
            ).strip()
    return by, query


def main() -> None:
    parser = _build_arg_parser()
    args = parser.parse_args()

    # Interactive fallback if requested or if required args are missing
    by = args.by
    query = args.query
    if args.interactive or not by or not query:
        by, query = _prompt_missing(by, query)

    users = read_users(args.file)
    if not users:
        return

    if by == "name":
        result = filter_users_by_name(users, query)
    elif by == "age":
        result = filter_users_by_age(users, query)
    else:
        result = filter_users_by_email(users, query)

    print_users(result)


if __name__ == "__main__":
    main()