lang_signals.py

"""
Multilingual trigger detection for "recency" and "how-to" intents.

- Reads a JSON file with language-specific keyword lists (default: languages.json next to this file).
- Exposes:
    recency_signals(text: str) -> bool
    howto_signals(text: str) -> bool

You can override the path via env var LANGUAGE_FILE.

Design notes:
- We lowercase input and phrases and match by substring (robust across scripts).
- We also treat any 4-digit year (20xx) as a weak recency cue.
- The loader caches the parsed sets and auto-reloads if the file's mtime changes
  (checked at most every 30s to keep overhead minimal).
"""

from __future__ import annotations

import json
import os
import time
import re
from typing import Dict, List, Set, Tuple

_LANG_FILE_DEFAULT = os.getenv("LANGUAGE_FILE", os.path.join(os.path.dirname(__file__), "languages.json"))

_YEAR_RE = re.compile(r"\b20\d{2}\b", flags=re.IGNORECASE)

# Runtime cache
_recency_set: Set[str] = set()
_howto_set: Set[str] = set()
_lang_path: str = _LANG_FILE_DEFAULT
_last_mtime: float = 0.0
_last_check: float = 0.0
_CHECK_INTERVAL_SECS = 30.0  # how often we stat() the file


def _normalize(items: List[str]) -> Set[str]:
    out: Set[str] = set()
    for s in items:
        s = (s or "").strip().lower()
        if not s:
            continue
        # extremely short tokens provoke false positives; skip if < 2 chars
        if len(s) < 2:
            continue
        out.add(s)
    return out


def _load_file(path: str) -> Tuple[Set[str], Set[str]]:
    try:
        with open(path, "r", encoding="utf-8") as f:
            data = json.load(f)
        rec_map: Dict[str, List[str]] = data.get("recency", {})
        how_map: Dict[str, List[str]] = data.get("howto", {})
        rec_all: List[str] = []
        how_all: List[str] = []
        for arr in rec_map.values():
            rec_all.extend(arr or [])
        for arr in how_map.values():
            how_all.extend(arr or [])
        return _normalize(rec_all), _normalize(how_all)
    except Exception:
        # conservative English-only fallback (won't crash app)
        rec_fb = [
            "today", "yesterday", "tomorrow", "current", "currently", "right now",
            "this year", "this month", "this week",
            "breaking", "latest", "news", "headline", "headlines",
            "update", "release", "version", "changelog",
            "price", "prices", "cost", "costs",
            "law", "laws", "regulation", "regulations",
            "schedule", "fixtures", "timetable", "calendar",
        ]
        how_fb = [
            "how to", "step by step", "step-by-step", "guide", "tutorial",
            "docs", "documentation", "example", "examples",
            "quickstart", "getting started", "manual", "installation guide", "api reference",
        ]
        return _normalize(rec_fb), _normalize(how_fb)


def _ensure_loaded() -> None:
    global _recency_set, _howto_set, _last_mtime, _last_check, _lang_path
    now = time.time()
    if now - _last_check < _CHECK_INTERVAL_SECS and _recency_set and _howto_set:
        return
    _last_check = now
    path = _lang_path
    try:
        mtime = os.path.getmtime(path)
    except Exception:
        mtime = 0.0
    if mtime == 0.0 or mtime != _last_mtime or not _recency_set or not _howto_set:
        _recency_set, _howto_set = _load_file(path)
        _last_mtime = mtime


def recency_signals(text: str) -> bool:
    """
    True if the text contains recency/time/news style triggers in any supported language,
    or references a 20xx year (weak recency cue).
    """
    _ensure_loaded()
    q = (text or "").lower()
    if _YEAR_RE.search(q):
        return True
    return any(p in q for p in _recency_set)


def howto_signals(text: str) -> bool:
    """True if the text contains how-to/tutorial/guide style triggers in any supported language."""
    _ensure_loaded()
    q = (text or "").lower()
    return any(p in q for p in _howto_set)