Cómo construir un backend de Document Intelligence con iii usando trabajadores, funciones y activadores cron

def normalize(data): return {“text”: (data.get(“text”) o “”).strip().lower()} def tokenize(data): text = data.get(“text”, “”) clean = “”.join(c if (c.isalnum() or c.isspace()) else ” ” para c en texto) tokens = [t for t in cleaned.split() if t]
return {“tokens”: tokens, “count”: len(tokens)} def sentimiento(datos): toks = data.get(“tokens”, []) pos = suma(t en POSITIVO para t en toks) neg = suma(t en NEGATIVO para t en toks) puntuación = pos – neg etiqueta = “positivo” si puntuación > 0 más “negativo” si puntuación < 0 más "neutral" return {"label": etiqueta, "score": puntuación, "pos": pos, "neg": neg} def palabras clave(datos): toks = data.get("tokens", []) stop = {"the","a","an","es","it","de","y","in","for","on","how"} freq = Counter(t for t in toks if t not in stop and len return {"tokens": toks["count"]"sentimiento": enviado, "palabras clave": claves["keywords"]"docs_analyzed": n} def informe(datos): con _LOCK: return {"docs_analyzed": _STATE["docs_analyzed"]"latidos del corazón": _ESTADO["heartbeats"]"top_keywords_all_docs": _ESTADO["keyword_totals"].most_common(5)} def http_analyze(data): body = data.get("body") o {} resultado = trabajador.trigger({"function_id": "pipeline::analyze", "payload": body}) return {"status_code": 200, "body": result, "headers": {"Content-Type": "application/json"}} def heartbeat(data): con _LOCK: _STATE["heartbeats"] += 1 return {"ok": True} para fid, fn en [ ("text::normalize", normalize), ("text::tokenize", tokenize), ("text::sentiment", sentiment), ("text::keywords", keywords), ("pipeline::analyze", analyze), ("stats::report", report), ("http::analyze", http_analyze), ("cron::heartbeat", heartbeat), ]: trabajador.register_function(fid, fn)