import gradio as gr import os import requests import re import traceback from huggingface_hub import InferenceClient # --- CONFIGURATION --- TOKEN = os.getenv("HF_TOKEN") if not TOKEN: raise ValueError("HF_TOKEN environment variable is missing. Please add it in Settings.") MODEL_ID = "Qwen/Qwen2.5-Coder-32B-Instruct" client = InferenceClient(MODEL_ID, token=TOKEN) # --- FPL THEME (Official Colors) --- fpl_theme = gr.themes.Base( primary_hue="green", secondary_hue="indigo", neutral_hue="slate", ).set( body_background_fill="#37003c", # Official FPL Purple block_background_fill="#47004d", # Lighter Purple for bubbles body_text_color="#ffffff", # White text block_label_text_color="#00ff85", # Neon Green labels block_title_text_color="#00ff85", # Neon Green titles button_primary_background_fill="#00ff85", # Neon Green Button button_primary_text_color="#37003c", # Purple Text on Button border_color_primary="#00ff85", # Green Borders ) # --- API TOOLS --- BASE_URL = "https://fantasy.premierleague.com/api" HEADERS = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36" } DATA_CACHE = {"bootstrap": None} def fetch_bootstrap(): """Fetches data and caches it. Retries if empty.""" if DATA_CACHE["bootstrap"] is None: try: r = requests.get(f"{BASE_URL}/bootstrap-static/", headers=HEADERS, timeout=10) if r.status_code == 200: DATA_CACHE["bootstrap"] = r.json() else: print(f"❌ API connection failed. Status: {r.status_code}") return None except Exception as e: print(f"❌ Network error: {e}") return None return DATA_CACHE.get("bootstrap") def get_team_name(team_id): data = fetch_bootstrap() if not data: return "UNK" for t in data['teams']: if t['id'] == team_id: return t['short_name'] return "UNK" def get_player_id(name_query): data = fetch_bootstrap() if not data: return None name_query = str(name_query).lower().strip() for p in data['elements']: full_name = f"{p['first_name']} {p['second_name']}".lower() if name_query in full_name or name_query in p['web_name'].lower(): return p return None def get_player_report(name_query): player = get_player_id(name_query) if not player: return f"❌ Player '{name_query}' not found in 2024/25 Database." try: r = requests.get(f"{BASE_URL}/element-summary/{player['id']}/", headers=HEADERS, timeout=10) if r.status_code != 200: return "⚠️ API Error: Cannot fetch history." data = r.json() history = data.get('history', []) last_3 = history[-3:] if len(history) >= 3 else history recent_pts = sum(m['total_points'] for m in last_3) fixtures = data.get('fixtures', []) next_3 = fixtures[:3] fix_str = "" for f in next_3: is_home = f['is_home'] opp_id = f['team_a'] if is_home else f['team_h'] opp = get_team_name(opp_id) diff = f['difficulty'] fix_str += f"{opp} (Dif:{diff}), " return ( f"### 📊 {player['web_name']}\n" f"**Price:** £{player['now_cost']/10}m | **Form:** {player['form']}\n" f"**Next 3:** {fix_str.strip(', ')}\n" f"**Last 3 Pts:** {recent_pts}" ) except: return "⚠️ Error processing stats." def compare_players(name1, name2): p1 = get_player_id(name1) p2 = get_player_id(name2) if not p1 or not p2: return "❌ Player not found." s1 = (float(p1['form']) * 0.6) + (float(p1['points_per_game']) * 0.4) s2 = (float(p2['form']) * 0.6) + (float(p2['points_per_game']) * 0.4) return f"### ⚔️ Compare\n**{p1['web_name']}** ({s1:.1f}) vs **{p2['web_name']}** ({s2:.1f})\nPick: **{p1['web_name'] if s1>s2 else p2['web_name']}**" def find_hidden_gems(position="MID", max_price=15.0): data = fetch_bootstrap() if not data: return "⚠️ API Error: Cannot load database." pos_map = { "GK": 1, "GKP": 1, "GOALKEEPER": 1, "DEF": 2, "DEFENDER": 2, "DF": 2, "CB": 2, "MID": 3, "MIDFIELDER": 3, "MF": 3, "WINGER": 3, "FWD": 4, "FORWARD": 4, "STRIKER": 4, "ATTACKER": 4, "ATT": 4 } pos_key = str(position).upper().strip() pos_id = pos_map.get(pos_key, 3) try: limit = float(max_price) if limit > 200: limit = limit / 1000000 if limit > 100000 else limit / 10 except: limit = 15.0 gems = [] for p in data['elements']: cost = p['now_cost'] / 10 # Logic: Correct Pos + Under Budget + Active if p['element_type'] == pos_id and cost <= limit: if float(p['form']) > 0.0 or float(p['selected_by_percent']) > 0.5: gems.append(p) # Sort by Form gems = sorted(gems, key=lambda x: float(x['form']), reverse=True)[:5] if not gems: return f"No gems found for {pos_key} under £{limit}m." res = f"### 💎 Gems ({pos_key} < £{limit}m)\n" for g in gems: r = requests.get(f"{BASE_URL}/element-summary/{g['id']}/", headers=HEADERS, timeout=5) next_fix = "UNK" if r.status_code == 200: fixtures = r.json().get('fixtures', []) if fixtures: f = fixtures[0] opp = get_team_name(f['team_a'] if f['is_home'] else f['team_h']) next_fix = f"{opp} (Dif:{f['difficulty']})" res += f"- **{g['web_name']}** (£{g['now_cost']/10}m) Form: {g['form']} | Next: {next_fix}\n" return res # --- PRO-LEVEL SYSTEM PROMPT --- SYSTEM_PROMPT = """ You are 'FPL Scout', an elite Fantasy Premier League tactical assistant. You have access to a LIVE database of the current 2024/25 season via tools. ### 🧠 KNOWLEDGE BASE (OFFICIAL RULES - USE THIS): - **Wildcard:** Allows unlimited PERMANENT transfers. Best used when your team has multiple injuries or poor form. - **Free Hit:** Allows unlimited transfers for ONE Gameweek only. Squad reverts to previous state. Use in Blank/Double Gameweeks. - **Bench Boost:** Points from bench players count for one GW. Use when your bench has strong fixtures. - **Triple Captain:** Captain gets 3x points. Best used on Double Gameweeks. ### 🛑 ANTI-HALLUCINATION PROTOCOLS: 1. **IGNORE INTERNAL DATA:** Your training data is from the past. Do NOT rely on it for player teams, prices, or form. Jordan Henderson is NOT at Liverpool. 2. **TOOL DEPENDENCY:** If a user asks for "Best Midfielder", you CANNOT answer until you run `[TOOL:find_hidden_gems:...]`. 3. **STOP SEQUENCE:** After outputting a `[TOOL:...]` tag, STOP generating text immediately. Wait for the data. ### 🛠️ AVAILABLE TOOLS: 1. [TOOL:get_player_report:Name] - Use for: Specific stats, form, price, and **Upcoming Fixtures**. - Example: "Should I keep Salah?" -> [TOOL:get_player_report:Salah] 2. [TOOL:compare_players:Name1,Name2] - Use for: Captaincy choices, "Start vs Bench", or Transfer dilemmas. - Example: "Saka or Palmer?" -> [TOOL:compare_players:Saka,Palmer] 3. [TOOL:find_hidden_gems:Position,MaxPrice] - Use for: Differentials, replacements, budget enablers. - Defaults: If user doesn't specify price, use 15.0. - Example: "Cheap defender?" -> [TOOL:find_hidden_gems:DEF,4.5] ### 📝 RESPONSE GUIDELINES (After Tool Result): - **Fixtures are King:** Analyze the 'Next 3' fixtures in the tool output. Green/Easy fixtures = BUY. - **Format:** Use **Bold** for names/prices. Use lists. - **Failure Handling:** If the tool returns "No gems found" or "API Error", simply say: "I couldn't find any players matching those criteria in the live database." DO NOT INVENT A LIST. """ def chat_logic(message, history): messages = [{"role": "system", "content": SYSTEM_PROMPT}] for h in history: if isinstance(h, dict): messages.append(h) elif isinstance(h, list) and len(h) >= 2: messages.append({"role": "user", "content": str(h[0])}) messages.append({"role": "assistant", "content": str(h[1])}) messages.append({"role": "user", "content": message}) try: response = client.chat_completion(messages, max_tokens=1000, temperature=0.1).choices[0].message.content # Lazy Parser tool_tag = None if "[TOOL:" in response: tool_tag = response.split("[TOOL:")[1].split("]")[0] elif "get_player_report(" in response: arg = response.split("get_player_report(")[1].split(")")[0] tool_tag = f"get_player_report:{arg}" elif "find_hidden_gems(" in response: args = response.split("find_hidden_gems(")[1].split(")")[0] tool_tag = f"find_hidden_gems:{args}" if tool_tag: try: # Anti-Hallucination Truncation if "[TOOL:" in response: clean_resp = response.split("]")[0] + "]" else: clean_resp = response if ":" in tool_tag: func_name, args_raw = tool_tag.split(":", 1) args = [a.strip() for a in re.split(r'[,;|]', args_raw) if a.strip()] else: if "," in tool_tag: func_name = tool_tag.split("(")[0].strip() if "(" in tool_tag else tool_tag.split(",")[0] func_name, args = tool_tag.split(":")[0], [] else: func_name, args = tool_tag.strip(), [] tool_result = "Unknown Tool" if "get_player_report" in func_name: arg1 = args[0] if len(args) > 0 else "Haaland" tool_result = get_player_report(arg1) elif "compare_players" in func_name: arg1 = args[0] if len(args) > 0 else "Salah" arg2 = args[1] if len(args) > 1 else "Haaland" tool_result = compare_players(arg1, arg2) elif "find_hidden_gems" in func_name: pos = args[0] if len(args) > 0 else "MID" try: p_str = args[1] if len(args) > 1 else "15.0" clean_p = re.sub(r'[^\d.]', '', p_str) price = float(clean_p) except: price = 15.0 tool_result = find_hidden_gems(pos, price) messages.append({"role": "assistant", "content": clean_resp}) messages.append({"role": "user", "content": f"Tool Result:\n{tool_result}\n\nUsing this data, answer the user."}) stream = client.chat_completion(messages, max_tokens=1024, stream=True) full = "" for chunk in stream: if chunk.choices and len(chunk.choices) > 0: delta = chunk.choices[0].delta if delta.content: full += delta.content yield full except Exception as e: traceback.print_exc() yield f"⚠️ Tool Error: {e}" else: yield response except Exception as e: yield f"⚠️ API Error: {str(e)}" # --- CONNECTION CHECK --- print("--- FPL AGENT STARTUP DIAGNOSTICS ---") check = fetch_bootstrap() if check: print(f"✅ Connection Established! Loaded {len(check['elements'])} players.") else: print("❌ Connection FAILED. Check internet or User-Agent.") print("---------------------------------------") # --- LAUNCH --- with gr.Blocks(title="⚽️ FPL Scout Agent") as demo: gr.Markdown("# 🦁 FPL Scout Agent (Official)") gr.ChatInterface( fn=chat_logic, description="Your AI Assistant for Fantasy Premier League. Ask about Price, Form, Fixtures, and Differentials." ) if __name__ == "__main__": print("🚀 Launching...") demo.launch(theme=fpl_theme)