Spaces:
Running
Running
| """ | |
| A robust, resilient analyzer using the Google Gemini Pro API. | |
| This module prompts Gemini for a simple key-value format and then constructs | |
| the final JSON object in Python, making it resilient to LLM syntax errors. | |
| """ | |
| import os | |
| import logging | |
| import httpx | |
| from typing import Optional, TypedDict, List, Union | |
| logger = logging.getLogger(__name__) | |
| class AnalysisResult(TypedDict): | |
| sentiment: str | |
| sentiment_score: float | |
| reason: str | |
| entities: List[str] | |
| topic: str | |
| impact: str | |
| summary: str | |
| error: Optional[str] | |
| class GeminiAnalyzer: | |
| API_URL = "https://generativelanguage.googleapis.com/v1beta/models/gemini-1.5-pro-latest:generateContent" | |
| def __init__(self, client: httpx.AsyncClient, api_key: Optional[str] = None): | |
| self.client = client | |
| self.api_key = api_key or os.getenv("GEMINI_API_KEY") | |
| if not self.api_key: | |
| raise ValueError("GEMINI_API_KEY is not set.") | |
| self.params = {"key": self.api_key} | |
| self.headers = {"Content-Type": "application/json"} | |
| def _build_prompt(self, text: str) -> dict: | |
| """Creates a prompt asking for a simple, parsable text format.""" | |
| return { | |
| "contents": [{ | |
| "parts": [{ | |
| "text": f""" | |
| Analyze the following financial text from the cryptocurrency world. | |
| Respond using a simple key::value format, with each key-value pair on a new line. Do NOT use JSON. | |
| KEYS: | |
| sentiment:: [POSITIVE, NEGATIVE, or NEUTRAL] | |
| sentiment_score:: [A float between -1.0 and 1.0] | |
| reason:: [A brief, one-sentence explanation for the sentiment.] | |
| entities:: [A comma-separated list of cryptocurrencies mentioned, e.g., Bitcoin, ETH] | |
| topic:: [One of: Regulation, Partnership, Technical Update, Market Hype, Security, General News] | |
| impact:: [One of: LOW, MEDIUM, HIGH] | |
| summary:: [A concise, one-sentence summary of the text.] | |
| TEXT TO ANALYZE: "{text}" | |
| """ | |
| }] | |
| }] | |
| } | |
| def _parse_structured_text(self, text: str) -> AnalysisResult: | |
| """Parses the key::value text response from Gemini into a structured dict.""" | |
| data = {} | |
| for line in text.splitlines(): | |
| if '::' in line: | |
| key, value = line.split('::', 1) | |
| data[key.strip()] = value.strip() | |
| # Build the final, validated object | |
| return { | |
| "sentiment": data.get("sentiment", "NEUTRAL").upper(), | |
| "sentiment_score": float(data.get("sentiment_score", 0.0)), | |
| "reason": data.get("reason", "N/A"), | |
| "entities": [e.strip() for e in data.get("entities", "").split(',') if e.strip()], | |
| "topic": data.get("topic", "General News"), | |
| "impact": data.get("impact", "LOW").upper(), | |
| "summary": data.get("summary", "Summary not available."), | |
| "error": None | |
| } | |
| async def analyze_text(self, text: str) -> AnalysisResult: | |
| """Sends text to Gemini and returns a structured analysis.""" | |
| prompt = self._build_prompt(text) | |
| try: | |
| response = await self.client.post(self.API_URL, headers=self.headers, params=self.params, json=prompt, timeout=60.0) | |
| response.raise_for_status() | |
| full_response = response.json() | |
| response_text = full_response["candidates"][0]["content"]["parts"][0]["text"] | |
| # Use our new, robust parser | |
| return self._parse_structured_text(response_text) | |
| except Exception as e: | |
| logger.error(f"❌ Gemini API or Parsing Error: {e}") | |
| return { | |
| "sentiment": "ERROR", "sentiment_score": 0.0, "reason": str(e), | |
| "entities": [], "topic": "Unknown", "impact": "Unknown", | |
| "summary": "Failed to perform analysis.", "error": str(e) | |
| } |