Update app.py

app.py

@@ -1,529 +1,162 @@
-import os
-from dataclasses import dataclass  # ✅ Fixes the error
-import re
+import requests
 import json
-import logging
-import asyncio
+import io
 import time
-import base64
-from pathlib import Path  # ✅ Fixes the error
-from typing import Dict, List, Tuple, Any, Optional, TypeVar
-from datetime import datetime
 from decimal import Decimal
-
-import aiohttp
-import openai
+from typing import Dict, Any, List, Tuple, Optional
 import gradio as gr
-from tenacity import retry, stop_after_attempt, wait_exponential
-from pathlib import Path  # ✅ Fix 'Path' not defined error
-from dataclasses import dataclass  # ✅ Fix 'dataclass' not defined error
-import os
-import json
-import logging
-import asyncio
-from typing import List, Dict, Tuple, Any, Optional
-
-# ✅ Fix 'Config' self-referencing error by using a string in type hint
-@dataclass
-class Config:
-    """Application configuration settings."""
-    
-    SYSTEM_PROMPT: str = "You are an AI-powered blockchain assistant."
-    OPENAI_MODEL: str = "gpt-4o-mini"
-    MAX_TOKENS: int = 4000
-    TEMPERATURE: float = 0.7
-
-    @classmethod
-    def load(cls, config_path: str | Path) -> "Config":  # ✅ Fix self-referencing error
-        """Load configuration from a JSON file."""
-        try:
-            with open(config_path) as f:
-                config_data = json.load(f)
-                return cls(**config_data)
-        except Exception as e:
-            logging.error(f"Error loading config: {e}")
-            return cls()
-
-# Type variables
-T = TypeVar('T')
-ChatHistory = List[Tuple[str, str]]
+from PIL import Image
 
-# Configure logging
-logging.basicConfig(
-    level=logging.INFO,
-    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
-    handlers=[
-        logging.FileHandler("blockchain_analyzer.log"),
-        logging.StreamHandler()
-    ]
-)
-logger = logging.getLogger(__name__)
+# API Keys (Replace with your own)
+ETHERSCAN_API_KEY = "YOUR_ETHERSCAN_API_KEY"
+OPENSEA_API_KEY = "YOUR_OPENSEA_API_KEY"
+OPENAI_API_KEY = "YOUR_OPENAI_API_KEY"
 
-# Custom Exceptions
-class APIError(Exception):
-    """Raised when an API request fails."""
-    pass
+# API URLs
+ETHERSCAN_BASE_URL = "https://api.etherscan.io/api"
+OPENSEA_BASE_URL = "https://api.opensea.io/api/v2/chain/ethereum/contract"
+OPENSEA_COLLECTION_STATS = "https://api.opensea.io/api/v1/collection"
 
-class ValidationError(Exception):
-    """Raised when input validation fails."""
-    pass
+HEADERS = {"X-API-KEY": OPENSEA_API_KEY} if OPENSEA_API_KEY else {}
 
-@dataclass
-class Config:
-    """Application configuration settings."""
-    SYSTEM_PROMPT: str = """
-    You are LOSS DOG 🐕 (Learning & Observing Smart Systems Digital Output Generator), 
-    an adorable blockchain-sniffing puppy!
-    Your personality:
-    - Friendly and enthusiastic
-    - Explain findings in fun, simple ways
-    Instructions:
-    - Use Alchemy API to fetch wallet data (NFTs and token holdings)
-    - Reference exact numbers and collections when discussing assets
-    - Highlight interesting findings
+### 🏦 WALLET ANALYZER ###
+def fetch_wallet_data(address: str) -> Dict[str, Any]:
     """
-    ALCHEMY_BASE_URL: str = "https://eth-mainnet.g.alchemy.com/v2"
-    ETHEREUM_ADDRESS_REGEX: str = r"0x[a-fA-F0-9]{40}"
-    RATE_LIMIT_DELAY: float = 0.2
-    MAX_RETRIES: int = 3
-    OPENAI_MODEL: str = "gpt-4o-mini"
-    MAX_TOKENS: int = 4000
-    TEMPERATURE: float = 0.7
-    HISTORY_LIMIT: int = 5
+    Fetch wallet data including ETH balance, tokens, and NFTs.
 
-    @classmethod
-    def load(cls, config_path: str | Path) -> Config:
-        """Load configuration from a JSON file."""
-        try:
-            with open(config_path) as f:
-                config_data = json.load(f)
-                return cls(**config_data)
-        except Exception as e:
-            logger.error(f"Error loading config: {e}")
-            return cls()
+    Args:
+        address (str): Ethereum wallet address.
 
-class AlchemyAPI:
-    """Handles interactions with the Alchemy API."""
+    Returns:
+        Dict[str, Any]: Wallet details including ETH balance, tokens, and NFTs.
+    """
+    def get_eth_balance():
+        params = {"module": "account", "action": "balance", "address": address, "tag": "latest", "apikey": ETHERSCAN_API_KEY}
+        response = requests.get(ETHERSCAN_BASE_URL, params=params)
+        data = response.json()
+        return Decimal(data["result"]) / Decimal("1000000000000000000") if "result" in data else 0
+
+    def get_tokens():
+        params = {"module": "account", "action": "tokentx", "address": address, "sort": "desc", "apikey": ETHERSCAN_API_KEY}
+        response = requests.get(ETHERSCAN_BASE_URL, params=params)
+        data = response.json()
+
+        tokens = {}
+        for tx in data.get("result", []):
+            contract = tx["contractAddress"]
+            if contract not in tokens:
+                tokens[contract] = {"name": tx["tokenName"], "symbol": tx["tokenSymbol"], "balance": Decimal(0)}
+            
+            amount = Decimal(tx["value"]) / Decimal(10 ** int(tx["tokenDecimal"]))
+            tokens[contract]["balance"] += amount if tx["to"].lower() == address.lower() else -amount
 
-    def __init__(self, api_key: str):
-        """Initialize Alchemy API client."""
-        self.api_key = api_key
-        self.base_url = f"{Config.ALCHEMY_BASE_URL}/{api_key}"
-        self.session: Optional[aiohttp.ClientSession] = None
+        return [{"name": v["name"], "symbol": v["symbol"], "balance": float(v["balance"])} for v in tokens.values() if v["balance"] > 0]
 
-    async def __aenter__(self) -> "AlchemyAPI":
-        """Create an aiohttp session on context manager enter."""
-        self.session = aiohttp.ClientSession()
-        return self
+    def get_nfts():
+        params = {"module": "account", "action": "tokennfttx", "address": address, "sort": "desc", "apikey": ETHERSCAN_API_KEY}
+        response = requests.get(ETHERSCAN_BASE_URL, params=params)
+        data = response.json()
 
-    async def __aexit__(self, exc_type, exc_val, exc_tb) -> None:
-        """Close aiohttp session on context manager exit."""
-        if self.session:
-            await self.session.close()
-            self.session = None
+        nfts = {}
+        for tx in data.get("result", []):
+            contract = tx["contractAddress"]
+            token_id = tx["tokenID"]
+            collection = tx.get("tokenName", "Unknown Collection")
 
-    async def _make_request(self, endpoint: str, params: Dict[str, Any]) -> Dict[str, Any]:
-        """Make an API request to Alchemy."""
-        if not self.session:
-            raise APIError("No active session. Use context manager.")
-        
-        try:
-            url = f"{self.base_url}/{endpoint}"
-            async with self.session.get(url, params=params) as response:
-                if response.status != 200:
-                    raise APIError(f"API request failed with status {response.status}")
-                return await response.json()
-        except aiohttp.ClientError as e:
-            raise APIError(f"Network error: {str(e)}")
+            if contract not in nfts:
+                nfts[contract] = {"collection": collection, "token_ids": set()}
+            
+            nfts[contract]["token_ids"].add(token_id)
 
-    async def get_wallet_nfts(self, address: str) -> Dict[str, Any]:
-        """Retrieve NFT holdings for a given wallet."""
-        try:
-            params = {"owner": address, "withMetadata": "true"}
-            data = await self._make_request("getNFTs", params)
-            return self._process_nfts(data)
-        except Exception as e:
-            raise APIError(f"Error getting wallet NFTs: {str(e)}")
+        return [{"contract": k, "collection": v["collection"], "token_ids": list(v["token_ids"])} for k, v in nfts.items()]
 
-    def _process_nfts(self, data: Dict[str, Any]) -> Dict[str, Any]:
-        """Process NFT data from Alchemy."""
-        nfts = data.get("ownedNfts", [])
-        collections = {}
+    return {
+        "eth_balance": float(get_eth_balance()),
+        "tokens": get_tokens(),
+        "nfts": get_nfts()
+    }
 
-        for nft in nfts:
-            contract = nft["contract"]["address"]
-            token_id = nft["tokenId"]
-            name = nft.get("title", "Unknown NFT")
-            image_url = nft.get("media", [{}])[0].get("gateway", "")
+### 🎨 NFT FETCHING ###
+def fetch_nft_metadata(contract: str, token_id: str) -> Optional[Dict[str, Any]]:
+    """
+    Fetch NFT metadata from OpenSea API.
 
-            if contract not in collections:
-                collections[contract] = {
-                    "count": 0,
-                    "nfts": []
-                }
+    Args:
+        contract (str): NFT contract address.
+        token_id (str): NFT Token ID.
 
-            collections[contract]["count"] += 1
-            collections[contract]["nfts"].append({
-                "token_id": token_id,
-                "name": name,
-                "image_url": image_url
-            })
+    Returns:
+        Dict[str, Any]: Metadata with name and image URL.
+    """
+    url = f"{OPENSEA_BASE_URL}/{contract}/nfts/{token_id}"
+    
+    try:
+        response = requests.get(url, headers=HEADERS)
+        response.raise_for_status()
+        data = response.json()
+        nft_metadata = data.get("nft", {})
 
         return {
-            "collections": collections,
-            "total_nfts": sum(col["count"] for col in collections.values())
+            "name": nft_metadata.get("name", f"NFT #{token_id}"),
+            "image_url": nft_metadata.get("image_url", ""),
         }
+    
+    except requests.exceptions.RequestException as e:
+        return {"error": f"Error fetching NFT: {str(e)}"}
 
-    async def get_wallet_tokens(self, address: str) -> Dict[str, Any]:
-        """Retrieve token holdings for a given wallet."""
-        try:
-            params = {"owner": address}
-            data = await self._make_request("getTokenBalances", params)
-            return self._process_tokens(data)
-        except Exception as e:
-            raise APIError(f"Error getting wallet tokens: {str(e)}")
-
-    def _process_tokens(self, data: Dict[str, Any]) -> Dict[str, Any]:
-        """Process token data from Alchemy."""
-        tokens = data.get("tokenBalances", [])
-        processed_tokens = []
-
-        for token in tokens:
-            contract = token["contractAddress"]
-            balance = Decimal(token["tokenBalance"]) / Decimal(10**18)
-
-            processed_tokens.append({
-                "contract": contract,
-                "balance": float(balance)
-            })
+### 🌐 GRADIO UI ###
+with gr.Blocks() as app:
+    gr.Markdown("# 🏦 Blockchain Wallet & NFT Analyzer")
 
-        return {"tokens": processed_tokens, "total_tokens": len(processed_tokens)}
+    # Wallet Analysis Section
+    with gr.Row():
+        wallet_input = gr.Textbox(label="Enter Ethereum Address", placeholder="0x...")
+        analyze_btn = gr.Button("Analyze Wallet")
+    wallet_output = gr.JSON(label="Wallet Data")
 
-    async def get_wallet_summary(self, address: str) -> Dict[str, Any]:
-        """Retrieve a complete wallet summary including NFTs and tokens."""
-        async with asyncio.TaskGroup() as group:
-            nft_task = group.create_task(self.get_wallet_nfts(address))
-            token_task = group.create_task(self.get_wallet_tokens(address))
+    analyze_btn.click(fn=fetch_wallet_data, inputs=[wallet_input], outputs=[wallet_output])
 
-        return {
-            "nfts": nft_task.result(),
-            "tokens": token_task.result()
+    # NFT Viewer Section
+    gr.Markdown("## 🎨 Ethereum NFT Viewer")
+    
+    with gr.Row():
+        contract_address = gr.Textbox(label="NFT Contract Address")
+        token_id = gr.Textbox(label="Token ID")
+        fetch_btn = gr.Button("Fetch NFT")
+
+    nft_name = gr.Textbox(label="NFT Name", interactive=False)
+    nft_image = gr.Image(label="NFT Image")
+
+    def display_nft(contract: str, token: str):
+        metadata = fetch_nft_metadata(contract, token)
+        if "error" in metadata:
+            return metadata["error"], None
+        return metadata["name"], metadata["image_url"]
+
+    fetch_btn.click(fn=display_nft, inputs=[contract_address, token_id], outputs=[nft_name, nft_image])
+
+    # Chatbot
+    gr.Markdown("## 💬 Ask About Blockchain Data")
+    chatbot = gr.Chatbot(label="Chat with Blockchain AI")
+    chat_input = gr.Textbox(label="Ask a question...")
+    chat_btn = gr.Button("Ask")
+
+    def chat_response(message: str):
+        headers = {"Authorization": f"Bearer {OPENAI_API_KEY}"}
+        data = {
+            "model": "gpt-4o-mini",
+            "messages": [{"role": "user", "content": message}],
+            "temperature": 0.7
         }
-class ChatInterface:
-    """Handles chat interaction using OpenAI API with blockchain data integration."""
-
-    def __init__(self, openai_key: str, alchemy_key: str):
-        """
-        Initialize chat interface with API keys.
-        
-        Args:
-            openai_key: OpenAI API key
-            alchemy_key: Alchemy API key
-        """
-        self.openai_key = openai_key
-        self.alchemy_key = alchemy_key
-        self.context: Dict[str, Any] = {}
-        openai.api_key = openai_key
-
-    @staticmethod
-    async def validate_api_keys(openai_key: str, alchemy_key: str) -> Tuple[bool, str]:
-        """
-        Validate OpenAI and Alchemy API keys.
-        
-        Args:
-            openai_key: OpenAI API key
-            alchemy_key: Alchemy API key
-            
-        Returns:
-            Tuple[bool, str]: (is_valid, message)
-        """
-        try:
-            # Validate OpenAI API Key (Sync Call)
-            client = openai.OpenAI(api_key=openai_key)
-            client.chat.completions.create(
-                model=Config.OPENAI_MODEL,
-                messages=[{"role": "user", "content": "test"}],
-                max_tokens=1
-            )
-
-            # Validate Alchemy API Key (Async Call)
-            async with AlchemyAPI(alchemy_key) as client:
-                await client.get_wallet_nfts("0x0000000000000000000000000000000000000000")
-
-            return True, "✅ API keys validated successfully!"
-
-        except Exception as e:
-            logger.error(f"Validation error: {str(e)}")
-            return False, f"❌ API key validation failed: {str(e)}"
-
-    def _format_context_message(self) -> str:
-        """
-        Format wallet data as context message for the AI.
-        
-        Returns:
-            str: Formatted context message
-        """
-        if not self.context:
-            return "No wallet data available."
-
-        context_msg = ["Current Wallet Data:\n"]
-
-        for addr, data in self.context.items():
-            context_msg.append(f"\n🔹 Wallet {addr[:8]}...{addr[-6:]}:")
-
-            if data["tokens"]["total_tokens"]:
-                context_msg.append(f"🪙 Tokens: {data['tokens']['total_tokens']}")
-                for token in data["tokens"]["tokens"][:5]:
-                    context_msg.append(f"  - {token['contract']} : {token['balance']}")
 
-            if data["nfts"]["total_nfts"]:
-                context_msg.append(f"🎨 NFTs: {data['nfts']['total_nfts']}")
-                for collection, info in list(data["nfts"]["collections"].items())[:3]:
-                    context_msg.append(f"  - {collection}: {info['count']} NFTs")
-
-        return "\n".join(context_msg)
-
-    async def fetch_wallet_data(self, address: str) -> Dict[str, Any]:
-        """
-        Fetch wallet data from Alchemy API.
-        
-        Args:
-            address: Ethereum wallet address
-            
-        Returns:
-            Dict[str, Any]: Wallet data summary
-        """
-        async with AlchemyAPI(self.alchemy_key) as client:
-            return await client.get_wallet_summary(address)
-
-    async def process_message(
-        self,
-        message: str,
-        history: Optional[ChatHistory] = None
-    ) -> Tuple[ChatHistory, Dict[str, Any], str]:
-        """
-        Process user message and generate response.
-        
-        Args:
-            message: User message
-            history: Chat history
-            
-        Returns:
-            Tuple containing:
-            - Updated chat history
-            - Updated wallet context
-            - Error message (if any)
-        """
-        if not message.strip():
-            return history or [], self.context, ""
-
-        history = history or []
-        match = re.search(Config.ETHEREUM_ADDRESS_REGEX, message)
-
-        if match:
-            try:
-                address = match.group(0)
-                wallet_data = await self.fetch_wallet_data(address)
-                self.context[address] = wallet_data
-
-                bot_message = f"✅ Wallet data for {address[:8]}...{address[-6:]} fetched successfully!"
-                history.append((message, bot_message))
-                return history, self.context, ""
-
-            except Exception as e:
-                logger.error(f"Error analyzing wallet: {e}")
-                error_message = f"Error analyzing wallet: {str(e)}"
-                history.append((message, error_message))
-                return history, self.context, ""
-
-        # Generate AI response
         try:
-            context_msg = self._format_context_message()
-            chat_history = []
-
-            for user_msg, assistant_msg in history[-Config.HISTORY_LIMIT:]:
-                chat_history.extend([
-                    {"role": "user", "content": user_msg},
-                    {"role": "assistant", "content": assistant_msg}
-                ])
-
-            client = openai.OpenAI(api_key=self.openai_key)
-            response = client.chat.completions.create(
-                model=Config.OPENAI_MODEL,
-                messages=[
-                    {"role": "system", "content": Config.SYSTEM_PROMPT},
-                    {"role": "system", "content": context_msg},
-                    *chat_history,
-                    {"role": "user", "content": message}
-                ],
-                temperature=Config.TEMPERATURE,
-                max_tokens=Config.MAX_TOKENS
-            )
-
-            bot_message = response.choices[0].message.content
-            history.append((message, bot_message))
-            return history, self.context, ""
-
-        except Exception as e:
-            logger.error(f"Error generating response: {e}")
-            error_message = f"Failed to generate response: {str(e)}"
-            history.append((message, error_message))
-            return history, self.context, ""
-
-    def clear_context(self) -> Tuple[Dict[str, Any], List[Tuple[str, str]]]:
-        """
-        Clear the wallet context and chat history.
+            response = requests.post("https://api.openai.com/v1/chat/completions", headers=headers, json=data)
+            return response.json()["choices"][0]["message"]["content"]
         
-        Returns:
-            Tuple containing empty context and history
-        """
-        self.context = {}
-        return {}, []
-import gradio as gr
-
-class GradioInterface:
-    """Handles Gradio web interface setup and interactions."""
-
-    def __init__(self):
-        """Initialize Gradio interface."""
-        self.chat_interface: Optional[ChatInterface] = None
-        self.demo = self._create_interface()
-
-    def _create_interface(self) -> gr.Blocks:
-        """Create and configure Gradio interface."""
-        with gr.Blocks(theme=gr.themes.Soft()) as demo:
-            gr.Markdown("""
-            # 🛠 Alchemy Blockchain Chatbot
-            - Enter your API keys to start
-            - Input an Ethereum wallet address
-            - Get NFT and token data from Alchemy
-            - Chat with AI about your holdings!
-            """)
-
-            # API Keys Section
-            with gr.Row():
-                openai_key = gr.Textbox(
-                    label="OpenAI API Key",
-                    type="password",
-                    placeholder="Enter your OpenAI API key..."
-                )
-                alchemy_key = gr.Textbox(
-                    label="Alchemy API Key",
-                    type="password",
-                    placeholder="Enter your Alchemy API key..."
-                )
-
-            validation_status = gr.Textbox(
-                label="Validation Status",
-                interactive=False
-            )
-            validate_btn = gr.Button("Validate API Keys", variant="primary")
-
-            # Main Interface
-            with gr.Row():
-                # Chat Area
-                chatbot = gr.Chatbot(label="Chat History", height=500, value=[])
-                with gr.Row():
-                    msg_input = gr.Textbox(
-                        label="Message",
-                        placeholder="Enter wallet address or ask about holdings..."
-                    )
-                    send_btn = gr.Button("Send", variant="primary")
-
-                # Right Sidebar (Wallet Data)
-                with gr.Column(scale=1):
-                    gr.Markdown("### Wallet Data")
-                    wallet_context = gr.JSON(label="Wallet Data", show_label=True, value={})
-                    clear_btn = gr.Button("Clear All Data", variant="secondary")
-
-            # Initial States
-            msg_input.interactive = False
-            send_btn.interactive = False
-
-            async def validate_keys(openai_k: str, alchemy_k: str) -> Tuple[str, gr.update, gr.update]:
-                """Validate API keys and initialize chat interface."""
-                try:
-                    is_valid, message = await ChatInterface.validate_api_keys(openai_k, alchemy_k)
-                    if is_valid:
-                        self.chat_interface = ChatInterface(openai_k, alchemy_k)
-                        return (
-                            "✅ API keys validated!",
-                            gr.update(interactive=True),
-                            gr.update(interactive=True)
-                        )
-                    return (f"❌ Validation failed: {message}", gr.update(interactive=False), gr.update(interactive=False))
-                except Exception as e:
-                    return (f"❌ Error: {str(e)}", gr.update(interactive=False), gr.update(interactive=False))
-
-            async def handle_message(
-                message: str,
-                chat_history: List[Tuple[str, str]],
-                wallet_context_data: Dict[str, Any]
-            ) -> Tuple[str, List[Tuple[str, str]], Dict[str, Any]]:
-                """Handle user messages."""
-                if not self.chat_interface:
-                    return "", [], {}
-
-                try:
-                    history, new_context, _ = await self.chat_interface.process_message(
-                        message, chat_history
-                    )
-                    return "", history, new_context
-                except Exception as e:
-                    chat_history.append((message, f"Error: {str(e)}"))
-                    return "", chat_history, wallet_context_data
-
-            def clear_all_data() -> Tuple[Dict[str, Any], List[Tuple[str, str]]]:
-                """Clear chat history and wallet data."""
-                if self.chat_interface:
-                    return self.chat_interface.clear_context()
-                return {}, []
-
-            # Connect Event Handlers
-            validate_btn.click(
-                fn=validate_keys,
-                inputs=[openai_key, alchemy_key],
-                outputs=[validation_status, msg_input, send_btn]
-            )
-
-            clear_btn.click(
-                fn=clear_all_data,
-                inputs=[],
-                outputs=[wallet_context, chatbot]
-            )
-
-            msg_input.submit(
-                fn=handle_message,
-                inputs=[msg_input, chatbot, wallet_context],
-                outputs=[msg_input, chatbot, wallet_context]
-            )
-
-            send_btn.click(
-                fn=handle_message,
-                inputs=[msg_input, chatbot, wallet_context],
-                outputs=[msg_input, chatbot, wallet_context]
-            )
-
-        return demo
-
-    def launch(self, **kwargs):
-        """Launch the Gradio interface."""
-        self.demo.queue()
-        self.demo.launch(**kwargs)
-
-
-def main():
-    """Main entry point for the application."""
-    try:
-        interface = GradioInterface()
-        interface.launch(
-            server_name="0.0.0.0",
-            server_port=7860
-        )
-    except Exception as e:
-        logger.error(f"Application startup failed: {e}")
-        raise
+        except requests.exceptions.RequestException as e:
+            return f"Error: {str(e)}"
 
+    chat_btn.click(fn=chat_response, inputs=[chat_input], outputs=[chatbot])
 
 if __name__ == "__main__":
-    main()
+    app.launch()