cortexm/util.py

Rama main · réplica de lectura del repositorio original.

"""Shared utilities: time, ids, normalization, string similarity."""

from __future__ import annotations

import re
import unicodedata
import uuid
from datetime import datetime, timedelta, timezone

WORDS = re.compile(r"[a-z0-9']+")


def h64(feature: str, seed: int = 0) -> int:
    """Deterministic 64-bit hash of a string feature (stable across runs)."""
    import hashlib
    return int.from_bytes(
        hashlib.blake2b(feature.encode("utf-8"), digest_size=8,
                        key=(seed & 0xFFFFFFFFFFFFFFFF).to_bytes(8, "little")).digest(),
        "little")


def utc_now() -> datetime:
    return datetime.now(timezone.utc)


def iso(dt: datetime) -> str:
    return dt.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")


def parse_ts(value: str | datetime | None) -> datetime | None:
    if value is None:
        return None
    if isinstance(value, datetime):
        return value if value.tzinfo else value.replace(tzinfo=timezone.utc)
    v = value.strip().replace("Z", "+00:00")
    try:
        dt = datetime.fromisoformat(v)
    except ValueError:
        for fmt in ("%Y-%m-%d", "%Y/%m/%d", "%Y-%m-%d %H:%M:%S"):
            try:
                dt = datetime.strptime(value.strip(), fmt)
                break
            except ValueError:
                continue
        else:
            return None
    return dt if dt.tzinfo else dt.replace(tzinfo=timezone.utc)


def new_id() -> str:
    return uuid.uuid4().hex


def normalize(text: str) -> str:
    """Lowercase, strip punctuation/articles, collapse whitespace."""
    text = unicodedata.normalize("NFKD", text)
    text = text.lower()
    text = re.sub(r"[^a-z0-9\s']+", " ", text)
    text = re.sub(r"\b(the|a|an|is|are|was|were)\b", " ", text)
    return re.sub(r"\s+", " ", text).strip()


def words(text: str) -> list[str]:
    return WORDS.findall(text.lower())


def token_estimate(text: str) -> int:
    """Cheap token estimator (~1.3 tokens/word). Deterministic."""
    n = len(text.split())
    return max(1, round(n * 1.3))


def levenshtein(a: str, b: str, cutoff: float = 1.0) -> int:
    """Levenshtein distance with early exit when distance exceeds cutoff*maxlen."""
    la, lb = len(a), len(b)
    if la == 0:
        return lb
    if lb == 0:
        return la
    maxd = int(max(la, lb) * cutoff) + 1
    prev = list(range(lb + 1))
    for i in range(1, la + 1):
        cur = [i] + [0] * lb
        best = i
        for j in range(1, lb + 1):
            cost = 0 if a[i - 1] == b[j - 1] else 1
            cur[j] = min(prev[j] + 1, cur[j - 1] + 1, prev[j - 1] + cost)
            best = min(best, cur[j])
        if best > maxd:
            return maxd
        prev = cur
    return prev[lb]


def similarity(a: str, b: str) -> float:
    """Combined token-Jaccard + Levenshtein string similarity in [0, 1].

    Fast paths: jaccard decides when clearly similar or clearly
    different; Levenshtein (with early-exit cutoff) only runs in the
    ambiguous band. Keeps conflict analysis linear-ish at scale.
    """
    a_n, b_n = normalize(a), normalize(b)
    if not a_n or not b_n:
        return 0.0
    if a_n == b_n:
        return 1.0
    wa, wb = set(words(a_n)), set(words(b_n))
    union = wa | wb
    jac = len(wa & wb) / len(union) if union else 0.0
    if jac >= 0.92:
        return jac
    if jac < 0.30:
        return jac
    d = levenshtein(a_n, b_n, cutoff=0.75)
    lev = 1.0 - d / max(len(a_n), len(b_n))
    return max(jac, lev)


def month_number(name: str) -> int | None:
    m = name.strip()[:3].lower()
    table = {
        "jan": 1, "feb": 2, "mar": 3, "apr": 4, "may": 5, "jun": 6,
        "jul": 7, "aug": 8, "sep": 9, "oct": 10, "nov": 11, "dec": 12,
    }
    return table.get(m)


def month_name(num: int) -> str:
    return ["January", "February", "March", "April", "May", "June", "July",
            "August", "September", "October", "November", "December"][num - 1]


def days_in(year: int, month: int) -> int:
    if month == 12:
        return 31
    return (datetime(year, month + 1, 1) - timedelta(days=1)).day


def fmt_date(dt: datetime) -> str:
    return dt.strftime("%Y-%m-%d")


def fmt_month(dt: datetime) -> str:
    return dt.strftime("%Y-%m")


def clamp(x: float, lo: float, hi: float) -> float:
    return max(lo, min(hi, x))