import queue from colorama import Fore, Style from langchain_core.callbacks import AsyncCallbackHandler from langchain_core.outputs.llm_result import LLMResult from typing import List from langchain_core.messages import BaseMessage TEAM_LOGO_BASE_URL = "https://huggingface.co/spaces/yamilsteven/ifx-assets/resolve/main/assets/team_logos/" image_base = """

{player_name} - #{player_number} (DL)

Ht: {height} | Wt: {weight} lbs

Team: {team_name}

Experience: 6 Years

Instagram Profile

""" game_card_html = """

{game_title}

${team_home} Logo$

{team_home}

{team1_score}

${team_away} Logo$

{team_away}

{team2_score}

Goal Highlights:

{highlights}

""" team_info_card_html = """

${team_display_name} Logo$

{team_display_name}

{city_display}

{team_page_cta_html}

""" team_image_map = { 'everglade-fc': 'Everglade_FC', 'fraser-valley-united': 'Fraser_Valley_United', 'tierra-alta-fc': 'Tierra_Alta_FC', 'yucatan-force': 'Yucatan_Force', } class GradioEventHandler(AsyncCallbackHandler): """ Example async event handler: prints streaming tokens and tool results. Replace with websocket or other side effects as needed. """ def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.queue = queue.Queue() def info_box(self, message: str): self.queue.put( { "type": "info", "message": message, } ) def ots_box(self, message: str): self.queue.put( { "type": "ots", "message": message, } ) async def on_chat_model_start(self, *args, **kwargs): pass # self.info_box('[CHAT START]') # self.ots_box(""" #

# """) async def on_llm_new_token(self, token: str, **kwargs): if token: self.queue.put(token) async def on_llm_end(self, result: LLMResult, *args, **kwargs): pass # if self.is_chat_stream_end(result): # self.queue.put(None) async def on_tool_end(self, output: any, **kwargs): print(f"\n{Fore.CYAN}[TOOL END] {output}{Style.RESET_ALL}") # Check for Semi-Final Game Card first, based on game_name if output and isinstance(output, list) and len(output) > 0 and \ isinstance(output[0], object) and hasattr(output[0], 'metadata') and \ isinstance(output[0].metadata.get('game_name'), str) and \ "semi-final" in output[0].metadata.get('game_name', '').lower(): # This is largely the same logic as the generic game card below game_teams_set = set() game_events_details = [] # Ensure raw_game_name uses the game_name from metadata if available raw_game_name = output[0].metadata.get('game_name', "Unknown Semi-Final") for i, doc in enumerate(output): # Iterate through all docs for game details if hasattr(doc, 'metadata') and doc.metadata.get('team'): game_teams_set.add(doc.metadata.get('team')) # game_name is taken from output[0] already if hasattr(doc, 'metadata') and doc.metadata.get('event') == 'Goal': event_desc = doc.metadata.get('description', 'Goal scored') event_minute = doc.metadata.get('minute', '') event_team = doc.metadata.get('team', '') game_events_details.append(f"({event_minute}') {event_team}: {event_desc}") if len(game_teams_set) >= 1: team_list = list(game_teams_set) team1_name_str = team_list[0] team2_name_str = team_list[1] if len(team_list) > 1 else "(opponent not specified)" score_team1 = 0 score_team2 = 0 for doc in output: if hasattr(doc, 'metadata') and doc.metadata.get('event') == 'Goal': scoring_team = doc.metadata.get('team') if scoring_team == team1_name_str: score_team1 += 1 elif scoring_team == team2_name_str: score_team2 += 1 game_title_str = raw_game_name.replace('_', ' ').title() highlights_html_str = "

{highlight}

" if not game_events_details: highlights_html_str = "

No goal highlights available.

{highlight}

" if not game_events_details: highlights_html_str = "

No goal highlights available.

" team1_logo_filename = GradioEventHandler.get_team_logo_slug(team1_name_str) team2_logo_filename = GradioEventHandler.get_team_logo_slug(team2_name_str) team1_logo_url = f"{TEAM_LOGO_BASE_URL}{team1_logo_filename}" team2_logo_url = f"{TEAM_LOGO_BASE_URL}{team2_logo_filename}" default_logo_url = f"{TEAM_LOGO_BASE_URL}default.png" formatted_game_html = game_card_html.format( game_title=game_title_str, team_home=team1_name_str, team_away=team2_name_str, team1_score=score_team1, team2_score=score_team2, highlights=highlights_html_str, team1_logo_url=team1_logo_url, team2_logo_url=team2_logo_url, default_logo_url=default_logo_url ) self.ots_box(formatted_game_html) return else: print(f"\n{Fore.RED}[TOOL END - GAME CARD] Not enough team data found in events or output structure not as expected.{Style.RESET_ALL}") # Attempt to process Team Card if no Game Card was processed elif output and isinstance(output, list) and len(output) > 0 and isinstance(output[0], object) and hasattr(output[0], 'metadata') and output[0].metadata.get("show_team_card"): doc_for_team_card = output[0] team_name_from_meta = doc_for_team_card.metadata.get("team_name", "Unknown Team") city_raw = doc_for_team_card.metadata.get("city", "N/A") display_team_name = str(team_name_from_meta).replace('-', ' ').replace('_', ' ').title() city_display = city_raw.title() if city_raw != "N/A" else "Location N/A" logo_filename = GradioEventHandler.get_team_logo_slug(team_name_from_meta) team_specific_logo_url = f"{TEAM_LOGO_BASE_URL}{logo_filename}" current_default_logo_url = f"{TEAM_LOGO_BASE_URL}default.png" team_id_slug = doc_for_team_card.metadata.get("team_id", "") team_page_url = f"https://www.team.com/{team_id_slug}" if team_id_slug else "#" team_page_cta_html = f'''Visit Team Page''' formatted_team_card_html = team_info_card_html.format( team_logo_url=team_specific_logo_url, team_display_name=display_team_name, city_display=city_display, default_logo_url=current_default_logo_url, team_page_cta_html=team_page_cta_html ) self.ots_box(formatted_team_card_html) return # Attempt to process Player Card if no Game or Team Card was processed # This iterates through all docs in output, so it's more general elif output and isinstance(output, list): for doc in output: if hasattr(doc, 'metadata') and doc.metadata.get("show_profile_card"): raw_player_name = doc.metadata.get("name", "unknown-player") player_name = "Unknown Player" if raw_player_name != "unknown-player": parts = raw_player_name.split('-') if len(parts) == 2: first_name = parts[0].capitalize() last_name = parts[1].capitalize() player_name = f"{last_name} {first_name}" else: player_name = raw_player_name.replace('-', ' ').title() else: player_name = "Unknown Player" team_slug = doc.metadata.get("team", "unknown-team") formatted_team_name = team_slug.replace('-', ' ').title() if team_slug == "unknown-team": formatted_team_name = "Unknown Team" player_number = doc.metadata.get("number", "N/A") height = doc.metadata.get("height", "175") weight = doc.metadata.get("weight", "150") instagram_url = doc.metadata.get("instagram_url", raw_player_name) img = image_base.format( filename=self.get_image_filename(doc), player_name=player_name, team_name=formatted_team_name, player_number=player_number, height=height, weight=weight, instagram_url=instagram_url ) self.ots_box(img) return # Return after the first player card is found # Fallback if no specific card type was identified print(f"\n{Fore.YELLOW}[TOOL END] No specific card type processed for OTS.{Style.RESET_ALL}") async def on_tool_start(self, input: any, *args, **kwargs): self.info_box(input.get("name", "[TOOL START]")) async def on_workflow_end(self, state, *args, **kwargs): print(f"\n{Fore.CYAN}[WORKFLOW END]{Style.RESET_ALL}") self.queue.put(None) # for msg in state["messages"]: # print(f'{Fore.YELLOW}{msg.content}{Style.RESET_ALL}') @staticmethod def is_chat_stream_end(result: LLMResult) -> bool: try: content = result.generations[0][0].message.content return bool(content and content.strip()) except (IndexError, AttributeError): return False @staticmethod def get_image_filename(doc): return f'{team_image_map.get(doc.metadata.get("team"))}_{doc.metadata.get("number")}.png' @staticmethod def get_team_logo_slug(team_name: str) -> str: if not team_name or team_name == "(opponent not specified)": return "default.png" # Normalize accented for the Yucatán logo normalized_team_name = team_name.lower().replace('á', 'a') slug = normalized_team_name.replace(' ', '-') return slug + ".png"