tai/tai_pkg/text.py

"""Pure string, time, token, record, and tag helpers."""

import hashlib
import json
import os
import re
from datetime import datetime, timedelta, timezone

from tai_pkg.consts import Consts

__all__ = ["Text"]


class Text:
    """Text."""

    @staticmethod
    # Python 3.6 compatibility notes:
    # - subprocess 'text' and 'capture_output' kwargs were added in 3.7, so this
    #   module only uses 'universal_newlines' plus explicit PIPEs (works on 3.6+).
    # - datetime.fromisoformat was added in 3.7, so all parsing goes through
    #   _fromisoformat below (native when available, strptime fallback on 3.6).


    def _fromisoformat(text):
        """datetime.fromisoformat backport for Python 3.6 (handles Z and +HH:MM)."""
        native = getattr(datetime, "fromisoformat", None)
        if native is not None:
            raw = str(text or "").strip()
            if raw[-1:] in ("Z", "z"):
                raw = raw[:-1] + "+00:00"
            return native(raw)
        raw = str(text or "").strip()
        if not raw:
            raise ValueError("empty datetime")
        if raw[-1:] in ("Z", "z"):
            raw = raw[:-1] + "+00:00"
        tzinfo = None
        core = raw
        match = re.match(r"^(\d{4}-\d{2}-\d{2}(?:[T ]\d{2}:\d{2}(?::\d{2}(?:\.\d{1,6})?)?)?)([+-]\d{2}:?\d{2}(?::\d{2})?|[+-]\d{4}|[+-]\d{2})$", raw)
        if match is not None and match.group(2):
            core, tzpart = match.group(1), match.group(2)
            sign = 1 if tzpart[0] == "+" else -1
            digits = tzpart[1:].replace(":", "")
            hours = int(digits[:2])
            minutes = int(digits[2:4]) if len(digits) >= 4 else 0
            tzinfo = timezone(sign * timedelta(hours=hours, minutes=minutes))
        for fmt in ("%Y-%m-%dT%H:%M:%S.%f", "%Y-%m-%dT%H:%M:%S", "%Y-%m-%dT%H:%M",
                    "%Y-%m-%d %H:%M:%S.%f", "%Y-%m-%d %H:%M:%S", "%Y-%m-%d %H:%M",
                    "%Y-%m-%d"):
            try:
                parsed = datetime.strptime(core, fmt)
            except ValueError:
                continue
            return parsed.replace(tzinfo=tzinfo)
        raise ValueError("invalid datetime, use ISO like 2026-10-08T09:00")

    @staticmethod
    def curl_hint(command):
        if re.search(r"\bcurl\b|\bwget\b", command or "") and "http" in (command or "").lower():
            return "\n" + Consts.CURL_HINT
        return ""

    @staticmethod
    def now_iso():
        return datetime.now(timezone.utc).isoformat(timespec="seconds")

    @staticmethod
    def short_error(exc):
        text = str(exc) or type(exc).__name__
        return text[:300]

    @staticmethod
    def truncate(text, head=1500, tail=500):
        if len(text) <= head + tail + 50:
            return text
        skipped = len(text) - head - tail
        return text[:head] + "\n...[%d chars truncated]...\n" % skipped + text[-tail:]

    @staticmethod
    def capped_text(text):
        if text is None:
            return None, None
        if len(text) <= Consts.AUDIT_MAX_CHARS:
            return text, len(text.encode("utf-8"))
        return text[:Consts.AUDIT_MAX_CHARS] + "\n...[truncated, full size %d bytes]..." % len(text.encode("utf-8")), len(text.encode("utf-8"))

    @staticmethod
    def trunc_marker(true_bytes):
        return "\n...[truncated, full size %d bytes]..." % true_bytes

    @staticmethod
    def read_capped(path, cap=Consts.AUDIT_MAX_CHARS):
        try:
            size = os.path.getsize(path)
        except OSError:
            return None, None
        try:
            with open(path, "r", encoding="utf-8", errors="replace") as handle:
                text = handle.read(cap + 1)
        except OSError:
            return None, None
        if len(text) > cap:
            return text[:cap], size
        return text, None

    @staticmethod
    def chunk_lines(text, max_lines=Consts.CHUNK_MAX_LINES, overlap_lines=Consts.CHUNK_OVERLAP_LINES, max_chars=Consts.CHUNK_MAX_CHARS):
        """Generic RAG chunker: bounded windows of whole lines with small overlap, never splitting mid-line.

    Used for any content that gets indexed for retrieval (terminal captures, long records, ...),
    so there is one chunking rule in the codebase instead of one per content type.
    """
        lines = text.splitlines()
        n = len(lines)
        if n == 0:
            return []
        chunks = []
        start = 0
        while start < n:
            end = start
            size = 0
            count = 0
            while end < n and count < max_lines and size < max_chars:
                size += len(lines[end]) + 1
                count += 1
                end += 1
            piece = "\n".join(lines[start:end]).strip()
            if piece:
                chunks.append(piece)
            if end >= n:
                break
            start = max(start + 1, end - overlap_lines)
        return chunks

    @staticmethod
    def collapse_repeated_lines(text, threshold=3):
        """Collapses runs of 3+ identical lines (progress bars, retry spam) to one line + a count note."""
        lines = text.splitlines()
        out = []
        i = 0
        n = len(lines)
        while i < n:
            j = i
            while j < n and lines[j] == lines[i]:
                j += 1
            run = j - i
            if run >= threshold:
                out.append(lines[i])
                out.append("... (repeated %d more times) ..." % (run - 1))
            else:
                out.extend(lines[i:j])
            i = j
        return "\n".join(out)

    @staticmethod
    def estimate_tokens(text):
        return max(1, len(text or "") // 4)

    @staticmethod
    def as_int(value, default=0):
        try:
            return int(value)
        except (TypeError, ValueError):
            return default

    @staticmethod
    def tool_args_text(raw_arguments):
        if not raw_arguments:
            return "{}"
        if isinstance(raw_arguments, str):
            return raw_arguments
        try:
            return json.dumps(raw_arguments, default=str)
        except (TypeError, ValueError):
            return str(raw_arguments)

    @staticmethod
    def action_signature(name, raw_arguments):
        try:
            flat = json.dumps(json.loads(Text.tool_args_text(raw_arguments)), sort_keys=True, default=str)
        except ValueError:
            flat = str(raw_arguments or "")
        return "%s %s" % (name, flat[:500])

    @staticmethod
    def result_digest(result):
        return hashlib.sha1((result or "").encode("utf-8", "replace")).hexdigest()

    @staticmethod
    def human_size(num):
        num = max(0, int(num))
        if num < 1024:
            return "%d B" % num
        if num < 1048576:
            return "%.1f KB" % (num / 1024)
        return "%.1f MB" % (num / 1048576)

    @staticmethod
    def is_free_model(model):
        model = (model or "").lower()
        return model.endswith(":free") or model.endswith("-free")

    @staticmethod
    def secret_env_name(name):
        return "TAI_SECRET_" + re.sub(r"[^A-Z0-9_]", "_", name.upper())

    @staticmethod
    def normalize_tag(raw):
        return re.sub(r"[^a-z0-9]+", "-", str(raw or "").strip().lower()).strip("-")[:32]

    @staticmethod
    def normalize_tags(items):
        seen = []
        for item in items:
            cleaned = Text.normalize_tag(item)
            if cleaned and cleaned not in seen:
                seen.append(cleaned)
        return seen[:20]

    @staticmethod
    def singular_noun(word):
        if word in Consts.SINGULAR_KEEP or len(word) <= 3:
            return word
        if word.endswith("ies"):
            return word[:-3] + "y"
        if word.endswith(("ses", "xes", "zes", "ches", "shes")):
            return word[:-2]
        if word.endswith("s") and not word.endswith(("ss", "us", "is", "os")):
            return word[:-1]
        return word

    @staticmethod
    def tag_variants(tag):
        found = {tag, Text.singular_noun(tag)}
        if not tag.endswith("s"):
            found.add(tag + "s")
        return sorted(found)

    @staticmethod
    def parse_node_id(raw):
        match = re.fullmatch(r"(mem|secret|sched):(.+)", str(raw or "").strip())
        if not match:
            return None
        kind, key = match.group(1), match.group(2)
        if kind == "mem" and re.fullmatch(r"[0-9a-f]{16}", key):
            return kind, key
        if kind == "secret" and Consts.SECRET_RE.match(key):
            return kind, key
        if kind == "sched" and key.isdigit():
            return kind, key
        return None