def normalize(information):
return {"textual content": (information.get("textual content") or "").strip().decrease()}
def tokenize(information):
textual content = information.get("textual content", "")
cleaned = "".be part of(c if (c.isalnum() or c.isspace()) else " " for c in textual content)
tokens = [t for t in cleaned.split() if t]
return {"tokens": tokens, "depend": len(tokens)}
def sentiment(information):
toks = information.get("tokens", [])
pos = sum(t in POSITIVE for t in toks)
neg = sum(t in NEGATIVE for t in toks)
rating = pos - neg
label = "optimistic" if rating > 0 else "detrimental" if rating < 0 else "impartial"
return {"label": label, "rating": rating, "pos": pos, "neg": neg}
def key phrases(information):
toks = information.get("tokens", [])
cease = {"the","a","an","is","it","to","of","and","in","for","on","how"}
freq = Counter(t for t in toks if t not in cease and len(t) > 2)
return {"key phrases": freq.most_common(information.get("top_n", 5))}
def analyze(information):
norm = employee.set off({"function_id": "textual content::normalize", "payload": {"textual content": information.get("textual content","")}})
toks = employee.set off({"function_id": "textual content::tokenize", "payload": norm})
despatched = employee.set off({"function_id": "textual content::sentiment", "payload": toks})
keys = employee.set off({"function_id": "textual content::key phrases", "payload": {**toks, "top_n": information.get("top_n", 5)}})
with _LOCK:
_STATE["docs_analyzed"] += 1
for ok, c in keys["keywords"]:
_STATE["keyword_totals"][k] += c
n = _STATE["docs_analyzed"]
return {"tokens": toks["count"], "sentiment": despatched, "key phrases": keys["keywords"], "docs_analyzed": n}
def report(information):
with _LOCK:
return {"docs_analyzed": _STATE["docs_analyzed"],
"heartbeats": _STATE["heartbeats"],
"top_keywords_all_docs": _STATE["keyword_totals"].most_common(5)}
def http_analyze(information):
physique = information.get("physique") or {}
consequence = employee.set off({"function_id": "pipeline::analyze", "payload": physique})
return {"status_code": 200, "physique": consequence, "headers": {"Content material-Sort": "utility/json"}}
def heartbeat(information):
with _LOCK:
_STATE["heartbeats"] += 1
return {"okay": True}
for fid, fn in [
("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),
]:
employee.register_function(fid, fn)
