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)