from __future__ import annotations import json import re from typing import Any def chunk_text_for_stream(text: str, chunk_size: int = 36) -> list[str]: if not text: return [] chunks: list[str] = [] for idx in range(0, len(text), chunk_size): chunks.append(text[idx : idx + chunk_size]) return chunks def format_sse(event: str, data: dict[str, Any]) -> str: return f"event: {event}\ndata: {json.dumps(data, ensure_ascii=False)}\n\n" def extract_json_object(raw_text: str) -> dict[str, Any] | None: stripped = raw_text.strip() if not stripped: return None try: loaded = json.loads(stripped) if isinstance(loaded, dict): return loaded except json.JSONDecodeError: pass fenced = re.findall(r"```json\s*(\{.*?\})\s*```", stripped, flags=re.DOTALL) for block in fenced: try: loaded = json.loads(block) if isinstance(loaded, dict): return loaded except json.JSONDecodeError: continue brace_match = re.search(r"(\{.*\})", stripped, flags=re.DOTALL) if brace_match: candidate = brace_match.group(1) try: loaded = json.loads(candidate) if isinstance(loaded, dict): return loaded except json.JSONDecodeError: return None return None # Intent phrases — any one of these anywhere in the message indicates the user wants # the system to surface people. Designed to catch natural phrasing. _CANDIDATE_SEARCH_INTENTS = ( "give me", "show me", "find me", "fetch me", "get me", "send me", "bring me", "i need", "i want", "i'm looking", "im looking", "looking for", "looking to hire", "want to hire", "need to hire", "who has", "who knows", "who can", "anyone with", "any candidate", "any developer", "any engineer", "any profile", "search ", "find ", "fetch ", "rank ", "shortlist", "list ", "show ", "match ", "filter ", "top ", "best ", "candidate for", "candidates for", "candidate with", "candidates with", "candidate having", "candidates having", "candidate who", "candidates who", "engineer for", "engineers for", "developer for", "developers for", ) _CANDIDATE_ROLE_WORDS = ( "engineer", "engineers", "developer", "developers", "scientist", "scientists", "analyst", "analysts", "designer", "designers", "manager", "managers", "architect", "architects", "intern", "interns", "lead", "leads", "consultant", "consultants", "specialist", "specialists", "candidate", "candidates", "resume", "resumes", "profile", "profiles", "people", "talent", "experience", # "candidate for AI experience" "expertise", "skill", "skills", "background", ) _CANDIDATE_SEARCH_NEGATIVE_PHRASES = ( "this pdf", "the pdf", "this resume", "the resume", "this file", "the file", "uploaded pdf", "uploaded resume", "uploaded file", "this candidate", "the candidate", "this person", "the person", "this applicant", "the applicant", "this profile", "the profile", "their resume", "their profile", "their experience", "his resume", "her resume", "his profile", "her profile", "about him", "about her", ) def looks_like_candidate_search(query: str) -> bool: """Heuristic: does the user's message ask the system to find people? Accepts natural phrasing — "give me candidate for AI experience", "I need a frontend developer", "anyone with NLP background", "looking for top python engineers in Bangalore", etc. Requires both an *intent phrase* (give me / show me / find / I need / looking for / etc.) and a *role/skill noun* (engineer / developer / candidate / experience / skill / etc.). Filters out chat actions and messages about a specific uploaded resume so we don't hijack other flows. """ lowered = (query or "").strip().lower() if not lowered: return False if lowered.startswith(("open_profile", "shortlist:", "details:", "publish_job_posting", "edit_job_posting")): return False if any(neg in lowered for neg in _CANDIDATE_SEARCH_NEGATIVE_PHRASES): return False has_intent = any(intent in lowered for intent in _CANDIDATE_SEARCH_INTENTS) has_role = any(role in lowered for role in _CANDIDATE_ROLE_WORDS) return has_intent and has_role def parse_action_command(query: str) -> tuple[str, str] | None: """Detect chat messages that look like card action commands such as ``open_profile:``.""" if not query: return None cleaned = query.strip() if ":" not in cleaned: return None head, _, tail = cleaned.partition(":") head = head.strip().lower() tail = tail.strip() if not tail or " " in head: return None if head in {"open_profile", "details", "shortlist", "publish_job_posting", "edit_job_posting"}: return head, tail return None def looks_like_c1_response(text: str) -> bool: lowered = text.lower() has_content_root = "" in lowered if not has_content_root: return False return ( " str: """Wrap plain markdown in the schema-stable C1 envelope. Thesys's renders reliably without their fine-tuned model — anything richer requires the API. Wrapping any agent text in this envelope means every response renders through the SDK, giving uniform typography/styling. """ text = (text or "").strip() if not text: return "" if looks_like_c1_response(text): return text safe = ( text.replace("&", "&") .replace("<", "<") .replace(">", ">") ) return f"{safe}" def build_c1_response(text: str, payload: dict[str, Any] | None = None) -> str: clean_text = (text or "").strip() if clean_text and looks_like_c1_response(clean_text): return clean_text if payload: payload_c1 = payload.get("c1_response") or payload.get("c1Response") if isinstance(payload_c1, str): payload_c1 = payload_c1.strip() if payload_c1 and looks_like_c1_response(payload_c1): return payload_c1 return ""