Blame for text_tools.py

import json
import logging
import re
import time
import unicodedata
from collections import Counter
from datetime import datetime, timezone
from difflib import get_close_matches
from functools import wraps


logger = logging.getLogger(__name__)

POSITIVE_WORDS = {
    "alegre", "amor", "bueno", "buen", "bonito", "brillante", "calma", "feliz",
    "genial", "grande", "hermoso", "ideal", "increible", "maravilloso", "mejor",
    "optimista", "paz", "positivo", "sano", "satisfecho", "triunfo", "victoria",
}
NEGATIVE_WORDS = {
    "odio", "malo", "mal", "triste", "angustia", "asco", "depresion", "dolor",
    "enfermo", "fracaso", "furia", "horrible", "miedo", "negativo", "pesimo",
    "rabia", "sufrimiento", "terrible", "vacio", "desamor", "desesperacion",
}
LANGUAGE_STOPWORDS = {
    "es": {
        "el", "la", "los", "las", "de", "del", "y", "que", "en", "un", "una", "por",
        "con", "para", "no", "se", "es", "al", "lo", "como", "mas", "pero",
        "sus", "ya", "o", "si", "entre", "cuando", "muy", "sin", "sobre", "me", "mi",
    },
    "en": {
        "the", "and", "to", "of", "a", "in", "is", "it", "you", "that", "he", "was",
        "for", "on", "are", "with", "as", "i", "his", "they", "be", "at", "one", "have",
    },
    "fr": {
        "le", "la", "les", "de", "des", "et", "un", "une", "en", "que", "est", "pour",
        "dans", "pas", "qui", "du", "au", "sur", "ce", "il", "elle", "nous", "vous",
    },
    "pt": {
        "o", "a", "os", "as", "de", "do", "da", "e", "em", "que", "um", "uma", "para",
        "com", "nao", "por", "como", "mais", "mas", "se", "dos", "das", "no",
    },
    "it": {
        "il", "la", "lo", "gli", "le", "di", "e", "che", "un", "una", "per", "con",
        "non", "come", "piu", "su", "del", "della", "nel", "nei", "da", "al",
    },
}

SPANISH_COMMON_WORDS = {
    "agua", "ahora", "aire", "alguien", "algo", "amigo", "amor", "antes", "aqui",
    "aun", "bajo", "bastante", "bien", "cabe", "cada", "camino", "casa", "cerca",
    "ciudad", "claro", "comer", "cosa", "cuerpo", "dia", "diez", "donde", "dueno",
    "ejemplo", "entonces", "escribir", "escuela", "este", "familia", "feliz", "fuerza",
    "gente", "grande", "hablar", "hacer", "hombre", "hoy", "idea", "jugar", "lugar",
    "mano", "mar", "mientras", "mundo", "nada", "niño", "noches", "nombre", "noche",
    "octubre", "ojo", "pais", "palabra", "paz", "persona", "problema", "puede", "pueblo",
    "querer", "razon", "salir", "saber", "sala", "sierra", "soria", "tarde", "tiempo",
    "tierra", "trabajo", "todo", "trigo", "ultramar", "vida", "viaje", "viajero", "voz",
}
ENGLISH_COMMON_WORDS = {
    "about", "after", "again", "air", "all", "also", "am", "an", "and", "any", "are",
    "back", "be", "because", "before", "big", "book", "but", "by", "can", "come",
    "day", "do", "down", "each", "end", "even", "family", "for", "from", "give",
    "good", "great", "had", "has", "have", "he", "her", "here", "his", "home", "house",
    "how", "i", "if", "in", "into", "it", "its", "just", "know", "like", "little",
    "look", "make", "man", "many", "me", "more", "most", "my", "new", "no", "not",
    "now", "one", "or", "our", "out", "people", "place", "said", "see", "she", "so",
    "some", "time", "to", "two", "up", "us", "very", "was", "we", "well", "what",
    "when", "where", "which", "who", "why", "will", "with", "work", "world", "you",
}


def _summarize_text(text: str) -> dict:
    words = text.split()
    return {
        "chars": len(text),
        "words": len(words),
        "lines": text.count("\n") + 1 if text else 0,
    }


def _normalize_word(word: str) -> str:
    word = word.lower().strip()
    word = re.sub(r"^[^\wáéíóúüñ]+|[^\wáéíóúüñ]+$", "", word, flags=re.UNICODE)
    normalized = unicodedata.normalize("NFD", word)
    return "".join(char for char in normalized if unicodedata.category(char) != "Mn")


def _normalize_words_from_source(words: set[str]) -> set[str]:
    return {_normalize_word(word) for word in words}


SPELLING_LEXICONS = {
    "es": (
        _normalize_words_from_source(LANGUAGE_STOPWORDS["es"])
        | _normalize_words_from_source(SPANISH_COMMON_WORDS)
        | _normalize_words_from_source(POSITIVE_WORDS)
        | _normalize_words_from_source(NEGATIVE_WORDS)
    ),
    "en": (
        _normalize_words_from_source(LANGUAGE_STOPWORDS["en"])
        | _normalize_words_from_source(ENGLISH_COMMON_WORDS)
        | _normalize_words_from_source(POSITIVE_WORDS)
        | _normalize_words_from_source(NEGATIVE_WORDS)
    ),
}


def _utc_now_iso() -> str:
    return datetime.now(timezone.utc).isoformat()


def _json_log(level: int, payload: dict) -> None:
    logger.log(level, json.dumps(payload, ensure_ascii=False, separators=(",", ":")))


def _log_tool(name: str):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            metadata = {}
            if args and isinstance(args[0], str):
                metadata.update(_summarize_text(args[0]))
            if "count" in kwargs:
                metadata["count"] = kwargs["count"]

            start = time.perf_counter()
            try:
                result = func(*args, **kwargs)
                duration_ms = round((time.perf_counter() - start) * 1000, 2)
                _json_log(logging.INFO, {
                    "timestamp": _utc_now_iso(),
                    "tool": name,
                    "event": "completed",
                    "status": "success",
                    "duration_ms": duration_ms,
                    **metadata,
                    "result_chars": len(result) if isinstance(result, str) else None,
                })
                return result
            except Exception:
                duration_ms = round((time.perf_counter() - start) * 1000, 2)
                _json_log(logging.ERROR, {
                    "timestamp": _utc_now_iso(),
                    "tool": name,
                    "event": "completed",
                    "status": "error",
                    "duration_ms": duration_ms,
                    **metadata,
                })
                raise

        return wrapper

    return decorator


def _split_sentences(text: str) -> list[str]:
    cleaned = text.strip()
    if not cleaned:
        return []
    sentences = re.split(r"(?<=[.!?])\s+", cleaned)
    return [sentence.strip() for sentence in sentences if sentence.strip()]


def _extract_tokens(text: str) -> list[str]:
    return [token for token in re.findall(r"[A-Za-zÁÉÍÓÚÜÑáéíóúüñ]+", text)]


def _language_info(text: str) -> dict:
    tokens = [_normalize_word(token) for token in _extract_tokens(text)]
    tokens = [token for token in tokens if token]
    token_set = set(tokens)

    scores = {}
    evidence = {}
    accent_signals = {
        "es": {"ñ", "á", "é", "í", "ó", "ú", "¿", "¡"},
        "fr": {"à", "â", "ç", "é", "è", "ê", "ë", "î", "ï", "ô", "ù", "û", "ü"},
        "pt": {"ã", "ç", "á", "à", "â", "é", "ê", "í", "ó", "ô", "õ", "ú"},
        "it": {"à", "è", "é", "ì", "ò", "ù"},
    }

    lowered_text = text.lower()
    for language, stopwords in LANGUAGE_STOPWORDS.items():
        stopword_hits = sum(1 for token in token_set if token in stopwords)
        accent_hits = sum(1 for char in lowered_text if char in accent_signals.get(language, set()))
        scores[language] = stopword_hits + (accent_hits * 0.5)
        evidence[language] = {
            "stopword_hits": stopword_hits,
            "accent_hits": accent_hits,
        }

    best_language = max(scores, key=scores.get) if scores else "unknown"
    best_score = scores.get(best_language, 0)
    second_best = sorted(scores.values(), reverse=True)[1] if len(scores) > 1 else 0
    confidence = round(((best_score - second_best) / max(best_score, 1)) * 100, 1) if best_score else 0.0

    if best_score == 0:
        detected = "unknown"
        confidence = 0.0
    else:
        detected = best_language

    return {
        "language": detected,
        "confidence": confidence,
        "scores": scores,
        "evidence": evidence,
        "token_count": len(tokens),
        "tokens": tokens,
    }


def _sentence_stopwords(language: str) -> set[str]:
    return LANGUAGE_STOPWORDS.get(language, LANGUAGE_STOPWORDS["es"])


def _spell_lexicon(language: str) -> set[str]:
    if language in SPELLING_LEXICONS:
        return SPELLING_LEXICONS[language]
    return SPELLING_LEXICONS["es"] | SPELLING_LEXICONS["en"]


def _clean_result_sentences(sentences: list[str]) -> list[str]:
    return [sentence.strip() for sentence in sentences if sentence.strip()]


@_log_tool("analyze_text")
def analyze_text(text: str) -> str:
    words = text.split()
    chars = len(text)
    chars_no_spaces = len(text.replace(" ", ""))
    sentences = text.count(".") + text.count("!") + text.count("?")

    avg_word_length = round(chars_no_spaces / len(words), 2) if words else 0
    avg_sentence_length = round(len(words) / max(sentences, 1), 2)

    return json.dumps({
        "total_characters": chars,
        "characters_without_spaces": chars_no_spaces,
        "total_words": len(words),
        "total_sentences": max(sentences, 1),
        "average_word_length": avg_word_length,
        "average_sentence_length": avg_sentence_length,
    }, indent=2, ensure_ascii=False)


@_log_tool("extract_keywords")
def extract_keywords(text: str, count: int = 5) -> str:
    stopwords = {
        "the", "a", "an", "and", "or", "but", "in", "on", "at", "to", "for",
        "of", "with", "is", "are", "was", "were", "be", "been", "by", "from",
    }

    words = text.lower().split()
    filtered = [w.strip(".,!?;:") for w in words if w.lower() not in stopwords]

    word_freq = Counter(filtered)
    top_words = word_freq.most_common(count)

    return json.dumps({
        "keywords": [{"word": w, "frequency": f} for w, f in top_words]
    }, indent=2, ensure_ascii=False)


@_log_tool("check_reading_level")
def check_reading_level(text: str) -> str:
    sentences = max(text.count(".") + text.count("!") + text.count("?"), 1)
    words = len(text.split())
    vowels = "aeiou"
    syllables = sum(1 for c in text.lower() if c in vowels)

    if words == 0:
        return json.dumps({"error": "No text to analyze"}, ensure_ascii=False)

    grade = max(0, (0.39 * (words / sentences)) + (11.8 * (syllables / words)) - 15.59)

    if grade < 6:
        level = "Elementary School"
    elif grade < 9:
        level = "Middle School"
    elif grade < 13:
        level = "High School"
    else:
        level = "College/Academic"

    return json.dumps({
        "grade_level": round(grade, 1),
        "reading_level": level,
    }, indent=2, ensure_ascii=False)


@_log_tool("analyze_sentiment")
def analyze_sentiment(text: str) -> str:
    tokens = [_normalize_word(token) for token in _extract_tokens(text)]
    tokens = [token for token in tokens if token]

    positive_hits = sum(1 for token in tokens if token in POSITIVE_WORDS)
    negative_hits = sum(1 for token in tokens if token in NEGATIVE_WORDS)
    total_hits = positive_hits + negative_hits

    if total_hits == 0:
        label = "neutral"
        score = 0
    else:
        score = positive_hits - negative_hits
        if score > 0:
            label = "positive"
        elif score < 0:
            label = "negative"
        else:
            label = "neutral"

    confidence = round((abs(score) / total_hits) * 100, 1) if total_hits else 0.0

    return json.dumps({
        "sentiment": label,
        "score": score,
        "confidence": confidence,
        "positive_hits": positive_hits,
        "negative_hits": negative_hits,
        "total_scored_tokens": total_hits,
    }, indent=2, ensure_ascii=False)


@_log_tool("detect_language")
def detect_language(text: str) -> str:
    info = _language_info(text)
    return json.dumps({
        "language": info["language"],
        "confidence": info["confidence"],
        "scores": info["scores"],
        "evidence": info["evidence"],
        "token_count": info["token_count"],
    }, indent=2, ensure_ascii=False)


@_log_tool("summarize_text")
def summarize_text(text: str) -> str:
    sentences = _clean_result_sentences(_split_sentences(text))
    if not sentences:
        return json.dumps({"error": "No text to summarize"}, ensure_ascii=False)

    language = _language_info(text)["language"]
    stopwords = _sentence_stopwords(language)
    tokens = [_normalize_word(token) for token in _extract_tokens(text)]
    tokens = [token for token in tokens if token and token not in stopwords]

    if len(sentences) <= 2:
        summary = " ".join(sentences)
        return json.dumps({
            "summary": summary,
            "method": "direct",
            "source_sentences": len(sentences),
            "selected_sentences": len(sentences),
        }, indent=2, ensure_ascii=False)

    frequency = Counter(tokens)
    sentence_scores = []
    for index, sentence in enumerate(sentences):
        sentence_tokens = [_normalize_word(token) for token in _extract_tokens(sentence)]
        sentence_tokens = [token for token in sentence_tokens if token and token not in stopwords]
        if not sentence_tokens:
            continue
        score = sum(frequency[token] for token in sentence_tokens) / (len(sentence_tokens) ** 0.5)
        sentence_scores.append((index, score, sentence))

    if not sentence_scores:
        summary_sentences = sentences[:2]
    else:
        max_sentences = 1 if len(sentences) < 4 else 2
        selected_indexes = sorted(
            index for index, _, _ in sorted(sentence_scores, key=lambda item: item[1], reverse=True)[:max_sentences]
        )
        summary_sentences = [sentences[index] for index in selected_indexes]

    summary = " ".join(summary_sentences)
    if len(summary) > 320:
        summary = summary[:317].rsplit(" ", 1)[0] + "..."

    return json.dumps({
        "summary": summary,
        "method": "extractive",
        "source_sentences": len(sentences),
        "selected_sentences": len(summary_sentences),
    }, indent=2, ensure_ascii=False)


@_log_tool("spell_check")
def spell_check(text: str) -> str:
    info = _language_info(text)
    language = info["language"] if info["language"] in {"es", "en"} else "es"
    lexicon = _spell_lexicon(language)

    raw_words = re.findall(r"[A-Za-zÁÉÍÓÚÜÑáéíóúüñ]+", text)
    misspellings = []

    for raw_word in raw_words:
        normalized = _normalize_word(raw_word)
        if not normalized:
            continue
        if normalized in lexicon:
            continue
        if raw_word[:1].isupper() and raw_word[1:].islower():
            continue
        if len(normalized) <= 2:
            continue

        suggestions = get_close_matches(normalized, list(lexicon), n=3, cutoff=0.84)
        misspellings.append({
            "word": raw_word,
            "normalized": normalized,
            "suggestions": suggestions,
        })

    unique_misspellings = []
    seen = set()
    for item in misspellings:
        key = item["normalized"]
        if key in seen:
            continue
        seen.add(key)
        unique_misspellings.append(item)

    return json.dumps({
        "language": language,
        "misspelled_words": unique_misspellings,
        "misspelled_count": len(unique_misspellings),
        "dictionary_mode": "heuristic",
        "note": "Heuristic spell checker; proper nouns and technical terms may require manual review.",
    }, indent=2, ensure_ascii=False)


@_log_tool("readability_advice")
def readability_advice(text: str) -> str:
    words = text.split()
    sentences = _clean_result_sentences(_split_sentences(text))
    sentence_count = max(len(sentences), 1)
    word_count = len(words)
    char_count = len(text)
    avg_sentence_length = round(word_count / sentence_count, 2)
    avg_word_length = round((len(text.replace(" ", "")) / word_count), 2) if word_count else 0
    long_words = sum(1 for word in words if len(_normalize_word(word)) >= 12)
    paragraphs = [paragraph for paragraph in text.split("\n\n") if paragraph.strip()]

    advice = []
    if avg_sentence_length > 24:
        advice.append("Divide las oraciones largas para reducir la carga cognitiva.")
    if avg_sentence_length > 35:
        advice.append("Hay oraciones muy extensas; considera fragmentarlas en dos o más frases.")
    if avg_word_length > 6:
        advice.append("Sustituye palabras largas o rebuscadas por términos más directos cuando sea posible.")
    if long_words / max(word_count, 1) > 0.2:
        advice.append("Hay muchos términos largos; mezcla vocabulario complejo con palabras más simples.")
    if word_count > 120 and len(paragraphs) <= 1:
        advice.append("Se beneficiaría de párrafos más cortos y espacios visuales entre ideas.")
    if sentence_count < 3 and word_count > 80:
        advice.append("Intenta repartir el contenido en más frases para mejorar el ritmo de lectura.")
    if not advice:
        advice.append("La claridad general es buena; solo revisa si puedes simplificar alguna frase puntual.")

    return json.dumps({
        "metrics": {
            "characters": char_count,
            "words": word_count,
            "sentences": sentence_count,
            "average_sentence_length": avg_sentence_length,
            "average_word_length": avg_word_length,
            "long_word_ratio": round(long_words / max(word_count, 1), 3),
        },
        "advice": advice,
    }, indent=2, ensure_ascii=False)