""" agents/graph.py โ€” Post processing pipeline for feelin' """ from __future__ import annotations from typing import TypedDict from agents.tools import detect_bragging, classify_emotion, PERSONAS from data.db import insert_post, flag_brag, update_post_scores CAT_EMOJIS = { "hilarious": "๐Ÿ˜‚", "tragic": "๐Ÿ˜ข", "unhinged": "๐Ÿ˜ค", "awkward": "๐Ÿ˜ฌ", "chaotic": "๐Ÿคฏ", } RANK_MSGS = { 1: "๐Ÿฅ‡ **#1 all time.** Frame this.", 2: "๐Ÿฅˆ **#2.** Somebody above you had a worse day.", 3: "๐Ÿฅ‰ **#3.** Bronze is still a medal in suffering.", 4: "**Top 5.** This is the content we came for.", 5: "**Top 5.** Solid.", 10: "**Top 10.** You're in the conversation.", } class FeelState(TypedDict): post_text: str post_id: str is_brag: bool brag_roast: str scores: dict top_category: str top_score: float reaction: str response_message: str response_type: str # "success" | "brag" | "error" def run_post_pipeline(text: str) -> dict: """Full pipeline: submit โ†’ detect โ†’ classify โ†’ respond.""" state: FeelState = { "post_text": text, "post_id": "", "is_brag": False, "brag_roast": "", "scores": {}, "top_category": "", "top_score": 0.0, "reaction": "", "response_message": "", "response_type": "success", } # Node 1: write to DB state["post_id"] = insert_post(text) # Node 2: brag detection brag_result = detect_bragging(text) state["is_brag"] = brag_result.get("is_brag", False) state["brag_roast"] = brag_result.get("roast", "") confidence = brag_result.get("confidence", 0) if state["is_brag"] and confidence > 0.6: flag_brag(state["post_id"], state["brag_roast"]) roast = state["brag_roast"] or "LinkedIn called. They want their energy back." state["response_type"] = "brag" state["response_message"] = _compose_brag_rejection(roast) return state # Node 3: emotion classification emotion_result = classify_emotion(text) state["scores"] = emotion_result.get("scores", {}) state["top_category"] = emotion_result.get("top_category", "chaotic") state["top_score"] = emotion_result.get("top_score", 5.0) state["reaction"] = emotion_result.get("reaction", "") # Node 4: update DB scores + get rank update_post_scores( state["post_id"], state["scores"], state["top_category"], state["top_score"], state["reaction"], ) rank = _compute_rank(state["post_id"], state["top_category"]) # Node 5: compose response state["response_message"] = _compose_success_response(state, rank) return state def _compute_rank(post_id: str, category: str) -> int: """Returns 1-based rank in the given category.""" from data.db import get_leaderboard posts = get_leaderboard(category, limit=20) for i, p in enumerate(posts): if p["id"] == post_id: return i + 1 return len(posts) + 1 def _compose_brag_rejection(roast: str) -> str: return f"""## ๐Ÿšฉ Brag Detected Our AI smelled the LinkedIn energy from miles away. > *{roast}* This is a **no-brag zone**. Come back when something has gone terribly wrong. --- *Try again with a story about a meeting that could've been an email.*""" def _compose_success_response(state: FeelState, rank: int) -> str: cat = state["top_category"] score = state["top_score"] emoji = CAT_EMOJIS.get(cat, "๐Ÿคฏ") persona = PERSONAS.get(cat, {}) scores = state["scores"] # Score bar (visual) bar_filled = int(score) bar = "โ–ˆ" * bar_filled + "โ–‘" * (10 - bar_filled) # Rank message rank_msg = "" for threshold, msg in RANK_MSGS.items(): if rank <= threshold: rank_msg = msg break if not rank_msg: rank_msg = f"Rank #{rank} โ€” every confession counts." # Score breakdown breakdown = " ยท ".join( [f"{CAT_EMOJIS[k]} {v:.1f}" for k, v in sorted(scores.items(), key=lambda x: -x[1])] ) if scores else "" # Podcast mention podcast_tease = "" if score >= 7.0 and persona.get("name"): podcast_tease = f"\n\n๐ŸŽ™๏ธ *{persona['name']} might read this on **{persona['show']}**.*" reaction_line = f"\n\n> *\"{state['reaction']}\"*" if state["reaction"] else "" return f"""## {emoji} {cat.upper()} ยท {score:.1f}/10 `{bar}` **{score:.1f}** {rank_msg}{reaction_line} --- **Score breakdown:** {breakdown}{podcast_tease}"""