course-text-processor-mcp / text_tools.py
pvidal66
logs
68be937
Raw History Blame Contribute Delete
17.6 kB
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)