Update app/app.py

app/app.py

@@ -1,172 +1,122 @@
-"""
-CryptoSentinel AI — High-performance FastAPI application.
+# In app/main.py (showing key changes and additions)
+import os
+from typing import Optional, Union # Make sure these are imported
+from app.gemini_analyzer import GeminiAnalyzer
+from newsapi import NewsApiClient # New import
 
-This is the main entry point that orchestrates the entire application.
-- Integrates the asynchronous PriceFetcher for live market data.
-- Integrates the asynchronous SentimentAnalyzer for real-time analysis.
-- Serves the interactive frontend and provides all necessary API endpoints.
-"""
-import asyncio
-import json
-from contextlib import asynccontextmanager
-
-import httpx
-from fastapi import FastAPI, Request, BackgroundTasks
-from fastapi.responses import HTMLResponse, StreamingResponse
-from fastapi.templating import Jinja2Templates
-from pydantic import BaseModel, constr
-
-# Import our modular, asynchronous service classes using relative paths
-from .price_fetcher import PriceFetcher
-from .sentiment import SentimentAnalyzer
-
-
-# --- Pydantic Model for API Input Validation ---
-
-class SentimentRequest(BaseModel):
-    """Ensures the text for sentiment analysis is a non-empty string."""
-    text: constr(strip_whitespace=True, min_length=1)
-
-# --- Application Lifespan for Resource Management ---
+# ... (keep existing imports and Pydantic model) ...
 
 @asynccontextmanager
 async def lifespan(app: FastAPI):
-    """
-    Manages application startup and shutdown events using the modern
-    lifespan context manager.
-    """
-    # On startup:
+    """Manages application startup and shutdown events."""
     async with httpx.AsyncClient() as client:
-        # Instantiate and store our services in the application state.
-        # This makes them accessible in any request handler via `request.app.state`.
+        # --- MODIFIED ---
         app.state.price_fetcher = PriceFetcher(client=client, coins=["bitcoin", "ethereum", "dogecoin"])
-        app.state.sentiment_analyzer = SentimentAnalyzer(client=client)
+        # Use the new Gemini Analyzer
+        app.state.gemini_analyzer = GeminiAnalyzer(client=client) 
+        # For the news feed
+        app.state.news_api = NewsApiClient(api_key=os.getenv("NEWS_API_KEY"))
+        app.state.news_queue: asyncio.Queue = asyncio.Queue()
+        app.state.sentiment_queue: asyncio.Queue = asyncio.Queue() # New queue for manual analysis
         app.state.request_counter = 0
 
-        # Create a cancellable background task for continuous price updates.
-        price_update_task = asyncio.create_task(
-            run_periodic_updates(app.state.price_fetcher, interval_seconds=10)
-        )
-        
-        print("🚀 CryptoSentinel AI started successfully.")
-        yield  # The application is now running and ready to accept requests.
+        # --- KEPT THE SAME ---
+        price_task = asyncio.create_task(run_periodic_updates(app.state.price_fetcher, 30))
+        # --- NEW BACKGROUND TASK ---
+        news_task = asyncio.create_task(run_periodic_news_analysis(app, interval_seconds=900)) # every 15 mins
+
+        print("🚀 CryptoSentinel Pro started successfully.")
+        yield
         
-        # On shutdown:
         print("⏳ Shutting down background tasks...")
-        price_update_task.cancel()
-        try:
-            await price_update_task
-        except asyncio.CancelledError:
-            print("Price update task cancelled successfully.")
-        print("✅ Shutdown complete.")
+        price_task.cancel()
+        news_task.cancel()
+        # ... (rest of shutdown) ...
 
-async def run_periodic_updates(fetcher: PriceFetcher, interval_seconds: int):
-    """A robust asyncio background task that periodically updates prices."""
+# --- NEW BACKGROUND FUNCTION ---
+async def run_periodic_news_analysis(app: FastAPI, interval_seconds: int):
+    """Fetches, analyzes, and queues top crypto news periodically."""
     while True:
-        await fetcher.update_prices_async()
+        print("📰 Fetching latest crypto news...")
+        try:
+            top_headlines = app.state.news_api.get_everything(
+                q='bitcoin OR ethereum OR crypto',
+                language='en',
+                sort_by='publishedAt',
+                page_size=5
+            )
+            analyzer: GeminiAnalyzer = app.state.gemini_analyzer
+            for article in top_headlines.get('articles', []):
+                title = article.get('title')
+                if title and "[Removed]" not in title:
+                    # Run full analysis on each headline
+                    analysis = await analyzer.analyze_text(title)
+                    # Add article URL to the payload
+                    analysis['url'] = article.get('url')
+                    await app.state.news_queue.put(analysis)
+        except Exception as e:
+            print(f"❌ Error fetching or analyzing news: {e}")
+            
         await asyncio.sleep(interval_seconds)
 
-# --- FastAPI App Initialization ---
-
-app = FastAPI(title="CryptoSentinel AI", lifespan=lifespan)
-
-# ====================================================================
-#                       FIX APPLIED HERE
-# ====================================================================
-# The 'templates' directory is at the root of the project, not inside 'app'.
-# This path is relative to the directory where the run command is executed.
-templates = Jinja2Templates(directory="templates")
-# ====================================================================
-
-
-# --- API Endpoints ---
-
-@app.get("/", response_class=HTMLResponse)
-async def serve_dashboard(request: Request):
-    """Serves the main interactive dashboard from `index.html`."""
-    return templates.TemplateResponse("index.html", {"request": request})
-
-@app.get("/api/prices", response_class=HTMLResponse)
-async def get_prices_fragment(request: Request):
-    """Returns an HTML fragment with the latest cached crypto prices for HTMX."""
-    price_fetcher: PriceFetcher = request.app.state.price_fetcher
-    prices = price_fetcher.get_current_prices()
-
-    html_fragment = ""
-    for coin, price in prices.items():
-        # Format the price nicely, handling the initial '--' state
-        price_str = f"${price:,.2f}" if isinstance(price, (int, float)) else price
-        html_fragment += f"<div><strong>{coin.capitalize()}:</strong> {price_str}</div>"
-    
-    return HTMLResponse(content=html_fragment)
+# ... (keep / and /api/prices endpoints) ...
 
+# --- MODIFIED SENTIMENT ENDPOINT ---
 @app.post("/api/sentiment")
-async def analyze_sentiment(
-    payload: SentimentRequest,
-    request: Request,
-    background_tasks: BackgroundTasks
-):
-    """
-    Validates and queues a text for sentiment analysis. The heavy lifting is
-    done in the background to ensure the API responds instantly.
-    """
-    analyzer: SentimentAnalyzer = request.app.state.sentiment_analyzer
-    request.app.state.request_counter += 1
-    request_id = request.app.state.request_counter
-
-    # The actual API call to Hugging Face will run after this response is sent.
-    background_tasks.add_task(analyzer.compute_and_publish, payload.text, request_id)
+async def analyze_sentiment(payload: SentimentRequest, request: Request, background_tasks: BackgroundTasks):
+    """Queues text for a full Gemini-powered analysis."""
+    analyzer: GeminiAnalyzer = request.app.state.gemini_analyzer
     
-    return HTMLResponse(content="<small>Queued for analysis...</small>")
+    async def task_wrapper():
+        analysis = await analyzer.analyze_text(payload.text)
+        await request.app.state.sentiment_queue.put(analysis)
+    
+    background_tasks.add_task(task_wrapper)
+    return HTMLResponse(content="<small>Queued for deep analysis...</small>")
 
+# --- NEW SENTIMENT STREAM ---
 @app.get("/api/sentiment/stream")
 async def sentiment_stream(request: Request):
-    """
-    Establishes a Server-Sent Events (SSE) connection. It efficiently pushes
-    new sentiment results as HTML fragments to the client as they become available.
-    """
-    analyzer: SentimentAnalyzer = request.app.state.sentiment_analyzer
-
+    """SSE stream for results from manual sentiment analysis requests."""
+    queue: asyncio.Queue = request.app.state.sentiment_queue
     async def event_generator():
-        # Clear the initial "waiting..." message on the client.
-        # hx-swap-oob="innerHTML" swaps this div out-of-band without affecting the target.
-        yield f"event: sentiment_update\ndata: <div id='sentiment-results' hx-swap-oob='innerHTML'></div>\n\n"
-        
-        # Listen for new results from the analyzer's internal queue.
-        async for result_payload in analyzer.stream_results():
-            try:
-                result = result_payload['result']
-                label = str(result.get('label', 'NEUTRAL')).lower()
-                score = result.get('score', 0.0) * 100
-                text = result_payload['text']
-                
-                # Dynamically build the HTML fragment to be sent to the client.
-                html_fragment = f"""
-                <div>
-                    <blockquote>{text}</blockquote>
-                    <p>
-                        <strong>Result:</strong> 
-                        <span class="sentiment-{label}">{label.upper()}</span> 
-                        (Confidence: {score:.1f}%)
-                    </p>
-                </div>
-                """
-                # First, process the string to remove newlines.
-                data_payload = html_fragment.replace('\n', '')
-                
-                # Then, use the clean variable in the f-string to build the message.
-                sse_message = f"event: sentiment_update\ndata: {data_payload}\n\n"
-                
-                yield sse_message
-
-            except (KeyError, TypeError):
-                continue # Ignore malformed payloads
+        while True:
+            payload = await queue.get()
+            html = render_analysis_card(payload)
+            yield f"event: sentiment_update\ndata: {html.replace('\n', '')}\n\n"
+    return StreamingResponse(event_generator(), media_type="text/event-stream")
 
+# --- NEW NEWS STREAM ---
+@app.get("/api/news/stream")
+async def news_stream(request: Request):
+    """SSE stream for the automated Market Intelligence Feed."""
+    queue: asyncio.Queue = request.app.state.news_queue
+    async def event_generator():
+        while True:
+            payload = await queue.get()
+            html = render_analysis_card(payload, is_news=True)
+            yield f"event: news_update\ndata: {html.replace('\n', '')}\n\n"
     return StreamingResponse(event_generator(), media_type="text/event-stream")
 
-# This block is now mostly for IDEs, the primary run method is the uvicorn command.
-if __name__ == "__main__":
-    # Note: Running this file directly (`python app/main.py`) will fail due to relative imports.
-    # Use the command: `uvicorn app.main:app --reload` from the project root.
-    print("To run this application, use the command from the root directory:")
-    print("uvicorn app.main:app --host 0.0.0.0 --port 7860 --reload")
+# --- NEW HELPER FUNCTION ---
+def render_analysis_card(payload: dict, is_news: bool = False) -> str:
+    """Renders a dictionary of analysis into an HTML card."""
+    s = payload
+    text_to_show = s.get('summary', 'N/A')
+    if is_news:
+        url = s.get('url', '#')
+        text_to_show = f'<a href="{url}" target="_blank">{s.get("summary", "N/A")}</a>'
+
+    return f"""
+    <div class="card impact-{s.get('impact', 'low').lower()}">
+        <blockquote>{text_to_show}</blockquote>
+        <div class="grid">
+            <div><strong>Sentiment:</strong> <span class="sentiment-{s.get('sentiment', 'neutral').lower()}">{s.get('sentiment')} ({s.get('sentiment_score', 0):.2f})</span></div>
+            <div><strong>Impact:</strong> {s.get('impact')}</div>
+        </div>
+        <div class="grid">
+            <div><strong>Topic:</strong> {s.get('topic')}</div>
+            <div><strong>Entities:</strong> {', '.join(s.get('entities', []))}</div>
+        </div>
+    </div>
+    """