Spaces:
Sleeping
Sleeping
| """ | |
| pipeline/key_validator.py -- Validate API keys before running analysis. | |
| """ | |
| from __future__ import annotations | |
| import sys | |
| from pathlib import Path | |
| from typing import Tuple | |
| sys.path.insert(0, str(Path(__file__).parent.parent)) | |
| def validate_fmp(key: str) -> Tuple[bool, str]: | |
| """Test FMP key with a lightweight profile call.""" | |
| if not key or key.strip() in ("", "your_fmp_key_here"): | |
| return False, "No key entered." | |
| try: | |
| import requests | |
| r = requests.get( | |
| "https://financialmodelingprep.com/api/v3/profile/AAPL", | |
| params={"apikey": key.strip()}, | |
| timeout=10, | |
| ) | |
| data = r.json() | |
| if isinstance(data, list) and data and data[0].get("symbol"): | |
| return True, f"Valid [OK] (plan: {data[0].get('exchange','FMP')})" | |
| if isinstance(data, dict) and "Error Message" in data: | |
| return False, f"Invalid key: {data['Error Message']}" | |
| return False, "Unexpected response -- key may be invalid." | |
| except Exception as e: | |
| return False, f"Connection error: {e}" | |
| def validate_gemini(key: str) -> Tuple[bool, str]: | |
| """Test Gemini key with a minimal generate call.""" | |
| if not key or key.strip() in ("", "your_gemini_key_here"): | |
| return False, "No key entered." | |
| try: | |
| # Try new google.genai SDK first | |
| from google import genai | |
| client = genai.Client(api_key=key.strip()) | |
| resp = client.models.generate_content( | |
| model="gemini-2.0-flash", | |
| contents="Reply with the single word: OK", | |
| ) | |
| text = resp.text.strip() if resp.text else "" | |
| return True, f"Valid [OK] (response: '{text[:20]}')" | |
| except Exception as e1: | |
| # Fallback to old SDK | |
| try: | |
| import google.generativeai as genai_old | |
| genai_old.configure(api_key=key.strip()) | |
| model = genai_old.GenerativeModel("gemini-1.5-flash") | |
| resp = model.generate_content("Reply with the single word: OK") | |
| return True, f"Valid [OK] (response: '{resp.text.strip()[:20]}')" | |
| except Exception as e2: | |
| msg = str(e1) | |
| if "API_KEY_INVALID" in msg or "invalid" in msg.lower(): | |
| return False, "Invalid API key. Check and re-enter." | |
| if "quota" in msg.lower(): | |
| return True, "Key valid but quota exceeded -- will still work." | |
| return False, f"Error: {msg[:80]}" | |
| def validate_claude(key: str) -> Tuple[bool, str]: | |
| """Test Claude key with a minimal message.""" | |
| if not key or key.strip() in ("", "your_claude_key_here"): | |
| return False, "No key entered." | |
| try: | |
| import anthropic | |
| client = anthropic.Anthropic(api_key=key.strip()) | |
| msg = client.messages.create( | |
| model="claude-3-5-haiku-20241022", | |
| max_tokens=10, | |
| messages=[{"role": "user", "content": "Reply with the single word: OK"}], | |
| ) | |
| text = msg.content[0].text.strip() if msg.content else "" | |
| return True, f"Valid [OK] (response: '{text[:20]}')" | |
| except Exception as e: | |
| msg = str(e) | |
| if "authentication" in msg.lower() or "401" in msg or "invalid" in msg.lower(): | |
| return False, "Invalid API key. Check and re-enter." | |
| if "quota" in msg.lower() or "credit" in msg.lower(): | |
| return True, "Key valid but credit exhausted." | |
| return False, f"Error: {msg[:80]}" | |
| def validate_openai(key: str) -> Tuple[bool, str]: | |
| """Test OpenAI key with a minimal chat completion.""" | |
| if not key or key.strip() in ("", "your_openai_key_here"): | |
| return False, "No key entered." | |
| try: | |
| from openai import OpenAI | |
| client = OpenAI(api_key=key.strip()) | |
| resp = client.chat.completions.create( | |
| model="gpt-4o-mini", | |
| messages=[{"role": "user", "content": "Reply with the single word: OK"}], | |
| max_tokens=5, | |
| ) | |
| text = resp.choices[0].message.content.strip() if resp.choices else "" | |
| return True, f"Valid [OK] (response: '{text[:20]}')" | |
| except Exception as e: | |
| msg = str(e) | |
| if "401" in msg or "Incorrect API key" in msg or "invalid_api_key" in msg: | |
| return False, "Invalid API key. Check and re-enter." | |
| if "quota" in msg.lower() or "insufficient" in msg.lower(): | |
| return True, "Key valid but quota/credit exhausted." | |
| return False, f"Error: {msg[:80]}" | |
| def validate_groq(key: str) -> tuple: | |
| """Test Groq key with a minimal chat completion (free API).""" | |
| if not key or key.strip() in ("", "your_groq_key_here"): | |
| return False, "No key entered." | |
| try: | |
| from openai import OpenAI | |
| client = OpenAI( | |
| api_key=key.strip(), | |
| base_url="https://api.groq.com/openai/v1", | |
| ) | |
| resp = client.chat.completions.create( | |
| model="llama-3.3-70b-versatile", | |
| messages=[{"role": "user", "content": "Reply with the single word: OK"}], | |
| max_tokens=5, | |
| ) | |
| text = resp.choices[0].message.content.strip() if resp.choices else "" | |
| return True, f"Valid [OK] (Llama 3.3 70B, response: '{text[:20]}')" | |
| except Exception as e: | |
| msg = str(e) | |
| if "401" in msg or "invalid_api_key" in msg.lower() or "unauthorized" in msg.lower(): | |
| return False, "Invalid Groq API key. Check and re-enter." | |
| if "quota" in msg.lower() or "rate" in msg.lower(): | |
| return True, "Key valid but rate-limited -- try again shortly." | |
| return False, f"Error: {msg[:80]}" | |
| def validate_nvidia(key: str) -> tuple: | |
| """Test NVIDIA NIM key with a minimal chat completion.""" | |
| if not key or key.strip() in ("", "your_nvidia_key_here"): | |
| return False, "No key entered." | |
| try: | |
| from openai import OpenAI | |
| client = OpenAI( | |
| api_key=key.strip(), | |
| base_url="https://integrate.api.nvidia.com/v1", | |
| ) | |
| resp = client.chat.completions.create( | |
| model="meta/llama-3.1-405b-instruct", | |
| messages=[{"role": "user", "content": "Reply with the single word: OK"}], | |
| max_tokens=5, | |
| ) | |
| text = resp.choices[0].message.content.strip() if resp.choices else "" | |
| return True, f"Valid [OK] (NVIDIA NIM, response: '{text[:20]}')" | |
| except Exception as e: | |
| msg = str(e) | |
| if "401" in msg or "invalid_api_key" in msg.lower() or "unauthorized" in msg.lower(): | |
| return False, "Invalid NVIDIA API key. Check and re-enter." | |
| return False, f"Error: {msg[:80]}" | |
| def check_any_ai_model(gemini_key: str = "", claude_key: str = "", openai_key: str = "", groq_key: str = "", nvidia_key: str = "") -> tuple: | |
| """Quickly check if at least one AI model is usable.""" | |
| checks = [ | |
| ("nvidia", nvidia_key, validate_nvidia), | |
| ("groq", groq_key, validate_groq), | |
| ("gemini", gemini_key, validate_gemini), | |
| ("claude", claude_key, validate_claude), | |
| ("openai", openai_key, validate_openai), | |
| ] | |
| for name, key, fn in checks: | |
| if key and key.strip(): | |
| ok, msg = fn(key) | |
| if ok: | |
| return True, f"{name.capitalize()}: {msg}" | |
| return False, ( | |
| "No working AI model key found.\n\n" | |
| "Free options (no credit card needed):\n" | |
| " • NVIDIA NIM (Free credits) -> build.nvidia.com\n" | |
| " • Groq (Llama 3.3 70B) -> console.groq.com\n" | |
| " • Gemini -> aistudio.google.com\n\n" | |
| "Paid options:\n" | |
| " • Claude -> console.anthropic.com\n" | |
| " • ChatGPT -> platform.openai.com/api-keys\n\n" | |
| "OR uncheck all AI models to run data-only analysis." | |
| ) | |