"""Shared time/filter parsing for `hermes sessions prune` / `archive`.

Turns user-friendly CLI values into the epoch bounds and filter kwargs
consumed by ``SessionDB.prune_sessions`` / ``archive_sessions`` /
``list_prune_candidates``.

Two value shapes are accepted anywhere a point in time is expected:

* Durations (relative to now): ``5h``, ``30m``, ``2d``, ``1w`` — and, for
  backward compatibility with the original ``--older-than N`` flag, a bare
  integer which means **days**.
* Absolute timestamps: ``2026-07-05``, ``2026-07-05 14:30``,
  ``2026-07-05T14:30:00`` (any ISO-8601 form ``datetime.fromisoformat``
  understands; naive values are interpreted in local time).
"""

from __future__ import annotations

import re
import time
from datetime import datetime, timezone
from typing import Any, Dict, Optional

_DURATION_RE = re.compile(
    r"^(\d+(?:\.\d+)?)\s*"
    r"(s|sec|secs|second|seconds|"
    r"m|min|mins|minute|minutes|"
    r"h|hr|hrs|hour|hours|"
    r"d|day|days|"
    r"w|wk|wks|week|weeks)$"
)

_UNIT_SECONDS = {"s": 1, "m": 60, "h": 3600, "d": 86400, "w": 604800}


def parse_duration_seconds(value: str) -> Optional[float]:
    """Parse ``5h`` / ``30m`` / ``2d`` / ``1w`` / ``90`` (bare = days) into
    seconds. Returns None when the value doesn't look like a duration."""
    s = str(value).strip().lower()
    if not s:
        return None
    if re.fullmatch(r"\d+(?:\.\d+)?", s):
        # Bare number = days (backward compatible with --older-than 90)
        return float(s) * 86400
    m = _DURATION_RE.match(s)
    if not m:
        return None
    return float(m.group(1)) * _UNIT_SECONDS[m.group(2)[0]]


def parse_point_in_time(value: str, flag: str) -> float:
    """Parse a CLI time value into an epoch timestamp.

    Durations are interpreted as "that long ago" (``5h`` → now − 5 hours).
    Absolute ISO timestamps are returned as-is (naive = local time).
    Raises ``ValueError`` with a user-facing message on unparseable input.
    """
    s = str(value).strip()
    dur = parse_duration_seconds(s)
    if dur is not None:
        return time.time() - dur
    try:
        dt = datetime.fromisoformat(s)
    except ValueError:
        raise ValueError(
            f"Invalid value for {flag}: '{value}'. Use a duration like '5h', "
            f"'30m', '2d', '1w', a bare number of days, or an ISO timestamp "
            f"like '2026-07-05' or '2026-07-05 14:30'."
        ) from None
    if dt.tzinfo is None:
        return dt.timestamp()
    return dt.astimezone(timezone.utc).timestamp()


def format_epoch(ts: Optional[float]) -> str:
    """Render an epoch timestamp as a short local-time string."""
    if ts is None:
        return "-"
    return datetime.fromtimestamp(ts).strftime("%Y-%m-%d %H:%M")


def build_prune_filters(args: Any) -> Dict[str, Any]:
    """Translate argparse Namespace flags into SessionDB filter kwargs.

    Understands: ``--older-than``, ``--newer-than``, ``--before``,
    ``--after``, ``--source``, ``--title``, ``--end-reason``, ``--cwd``,
    ``--min-messages``, ``--max-messages``, ``--archived``/``--no-archived``.

    ``--older-than`` / ``--newer-than`` bound last activity, while
    ``--before`` / ``--after`` explicitly bound session start time. Last
    activity is the latest message timestamp, falling back to ``started_at``
    for empty sessions.

    Raises ``ValueError`` on unparseable values or an empty/inverted window.
    """
    last_active_before: Optional[float] = None
    last_active_after: Optional[float] = None
    started_before: Optional[float] = None
    started_after: Optional[float] = None

    def _tighter(current: Optional[float], new: float, upper: bool) -> float:
        if current is None:
            return new
        return min(current, new) if upper else max(current, new)

    older_than = getattr(args, "older_than", None)
    if older_than is not None:
        last_active_before = _tighter(
            last_active_before,
            parse_point_in_time(older_than, "--older-than"),
            True,
        )
    newer_than = getattr(args, "newer_than", None)
    if newer_than is not None:
        last_active_after = _tighter(
            last_active_after,
            parse_point_in_time(newer_than, "--newer-than"),
            False,
        )
    before = getattr(args, "before", None)
    if before is not None:
        started_before = _tighter(
            started_before, parse_point_in_time(before, "--before"), True
        )
    after = getattr(args, "after", None)
    if after is not None:
        started_after = _tighter(
            started_after, parse_point_in_time(after, "--after"), False
        )

    if (
        started_before is not None
        and started_after is not None
        and started_after >= started_before
    ):
        raise ValueError(
            "Empty start-time window: the --after bound "
            f"({format_epoch(started_after)}) is not earlier than the "
            f"--before bound ({format_epoch(started_before)})."
        )
    if (
        last_active_before is not None
        and last_active_after is not None
        and last_active_after >= last_active_before
    ):
        raise ValueError(
            "Empty activity window: the --newer-than bound "
            f"({format_epoch(last_active_after)}) is not earlier than the "
            f"--older-than bound ({format_epoch(last_active_before)})."
        )

    filters: Dict[str, Any] = {
        # older_than_days=None: the epoch bounds above are the whole story.
        # Without this, prune_sessions' default 90-day cutoff would silently
        # cap an --after/--newer-than-only window.
        "older_than_days": None,
        "last_active_before": last_active_before,
        "last_active_after": last_active_after,
        "started_before": started_before,
        "started_after": started_after,
        "source": getattr(args, "source", None),
        "title_like": getattr(args, "title", None),
        "end_reason": getattr(args, "end_reason", None),
        "cwd_prefix": getattr(args, "cwd", None),
        "min_messages": getattr(args, "min_messages", None),
        "max_messages": getattr(args, "max_messages", None),
        "model_like": getattr(args, "model", None),
        "provider": getattr(args, "provider", None),
        "user_id": getattr(args, "user", None),
        "chat_id": getattr(args, "chat_id", None),
        "chat_type": getattr(args, "chat_type", None),
        "branch_like": getattr(args, "branch", None),
        "min_tokens": getattr(args, "min_tokens", None),
        "max_tokens": getattr(args, "max_tokens", None),
        "min_cost": getattr(args, "min_cost", None),
        "max_cost": getattr(args, "max_cost", None),
        "min_tool_calls": getattr(args, "min_tool_calls", None),
        "max_tool_calls": getattr(args, "max_tool_calls", None),
    }
    return filters


def describe_filters(filters: Dict[str, Any]) -> str:
    """Human-readable summary of active filters for confirmation prompts."""
    parts = []
    if filters.get("last_active_before") is not None:
        parts.append(
            f"last active before {format_epoch(filters['last_active_before'])}"
        )
    if filters.get("last_active_after") is not None:
        parts.append(
            f"last active after {format_epoch(filters['last_active_after'])}"
        )
    if filters.get("started_before") is not None:
        parts.append(f"started before {format_epoch(filters['started_before'])}")
    if filters.get("started_after") is not None:
        parts.append(f"started after {format_epoch(filters['started_after'])}")
    if filters.get("source"):
        parts.append(f"source '{filters['source']}'")
    if filters.get("title_like"):
        parts.append(f"title contains '{filters['title_like']}'")
    if filters.get("end_reason"):
        parts.append(f"end reason '{filters['end_reason']}'")
    if filters.get("cwd_prefix"):
        parts.append(f"cwd under '{filters['cwd_prefix']}'")
    if filters.get("min_messages") is not None:
        parts.append(f">= {filters['min_messages']} messages")
    if filters.get("max_messages") is not None:
        parts.append(f"<= {filters['max_messages']} messages")
    if filters.get("model_like"):
        parts.append(f"model contains '{filters['model_like']}'")
    if filters.get("provider"):
        parts.append(f"provider '{filters['provider']}'")
    if filters.get("user_id"):
        parts.append(f"user '{filters['user_id']}'")
    if filters.get("chat_id"):
        parts.append(f"chat '{filters['chat_id']}'")
    if filters.get("chat_type"):
        parts.append(f"chat type '{filters['chat_type']}'")
    if filters.get("branch_like"):
        parts.append(f"git branch contains '{filters['branch_like']}'")
    if filters.get("min_tokens") is not None:
        parts.append(f">= {filters['min_tokens']} tokens")
    if filters.get("max_tokens") is not None:
        parts.append(f"<= {filters['max_tokens']} tokens")
    if filters.get("min_cost") is not None:
        parts.append(f">= ${filters['min_cost']}")
    if filters.get("max_cost") is not None:
        parts.append(f"<= ${filters['max_cost']}")
    if filters.get("min_tool_calls") is not None:
        parts.append(f">= {filters['min_tool_calls']} tool calls")
    if filters.get("max_tool_calls") is not None:
        parts.append(f"<= {filters['max_tool_calls']} tool calls")
    return ", ".join(parts) if parts else "no filters (all ended sessions)"
