| |
| """ |
| Crypto API Clean Client - Integration with HuggingFace Space |
| https://really-amin-crypto-api-clean-fixed.hf.space |
| |
| Provides access to 281+ cryptocurrency resources across 12 categories: |
| - RPC Nodes (24) |
| - Block Explorers (33) |
| - Market Data APIs (33) |
| - News APIs (17) |
| - Sentiment APIs (14) |
| - On-chain Analytics APIs (14) |
| - Whale Tracking APIs (10) |
| - HuggingFace Resources (9) |
| - Free HTTP Endpoints (13) |
| - Local Backend Routes (106) |
| - CORS Proxies (7) |
| - Community Sentiment APIs (1) |
| """ |
|
|
| import httpx |
| import asyncio |
| import logging |
| from typing import Dict, Any, List, Optional |
| from datetime import datetime, timezone |
| from functools import lru_cache |
|
|
| logger = logging.getLogger(__name__) |
|
|
| |
| CRYPTO_API_CLEAN_BASE_URL = "https://really-amin-crypto-api-clean-fixed.hf.space" |
|
|
| |
| _resources_cache: Dict[str, Any] = {} |
| _cache_timestamp: Optional[datetime] = None |
| CACHE_TTL_SECONDS = 300 |
|
|
|
|
| class CryptoAPICleanService: |
| """ |
| Service for accessing Crypto API Clean HuggingFace Space |
| Provides comprehensive cryptocurrency resource database access |
| """ |
| |
| def __init__(self, timeout: int = 15): |
| self.base_url = CRYPTO_API_CLEAN_BASE_URL |
| self.timeout = timeout |
| self._client: Optional[httpx.AsyncClient] = None |
| |
| async def _get_client(self) -> httpx.AsyncClient: |
| """Get or create async HTTP client""" |
| if self._client is None or self._client.is_closed: |
| self._client = httpx.AsyncClient(timeout=self.timeout) |
| return self._client |
| |
| async def close(self): |
| """Close the HTTP client""" |
| if self._client and not self._client.is_closed: |
| await self._client.aclose() |
| |
| async def _request(self, endpoint: str, params: Dict = None) -> Dict[str, Any]: |
| """ |
| Make async request to Crypto API Clean with proper error handling |
| |
| Returns standardized response format matching project patterns |
| """ |
| provider = "CryptoAPIClean" |
| start_time = datetime.now(timezone.utc) |
| |
| try: |
| client = await self._get_client() |
| url = f"{self.base_url}{endpoint}" |
| |
| response = await client.get(url, params=params) |
| response.raise_for_status() |
| |
| data = response.json() |
| response_time_ms = (datetime.now(timezone.utc) - start_time).total_seconds() * 1000 |
| |
| logger.info(f"✅ {provider} - {endpoint} - {response_time_ms:.0f}ms") |
| |
| return { |
| "provider": provider, |
| "endpoint": endpoint, |
| "data": data, |
| "timestamp": datetime.now(timezone.utc).isoformat(), |
| "response_time_ms": response_time_ms, |
| "success": True, |
| "error": None |
| } |
| |
| except httpx.HTTPStatusError as e: |
| logger.error(f"❌ {provider} - {endpoint} - HTTP {e.response.status_code}") |
| return { |
| "provider": provider, |
| "endpoint": endpoint, |
| "data": None, |
| "timestamp": datetime.now(timezone.utc).isoformat(), |
| "success": False, |
| "error": f"HTTP {e.response.status_code}", |
| "error_type": "http_error" |
| } |
| except httpx.TimeoutException: |
| logger.error(f"❌ {provider} - {endpoint} - Timeout") |
| return { |
| "provider": provider, |
| "endpoint": endpoint, |
| "data": None, |
| "timestamp": datetime.now(timezone.utc).isoformat(), |
| "success": False, |
| "error": "Request timeout", |
| "error_type": "timeout" |
| } |
| except Exception as e: |
| logger.error(f"❌ {provider} - {endpoint} - {str(e)}") |
| return { |
| "provider": provider, |
| "endpoint": endpoint, |
| "data": None, |
| "timestamp": datetime.now(timezone.utc).isoformat(), |
| "success": False, |
| "error": str(e), |
| "error_type": "exception" |
| } |
| |
| |
| |
| async def get_resources_stats(self) -> Dict[str, Any]: |
| """ |
| Get resources database statistics |
| Returns: total_resources, total_categories, category breakdown |
| """ |
| return await self._request("/api/resources/stats") |
| |
| async def get_all_resources(self) -> Dict[str, Any]: |
| """ |
| Get complete list of all 281+ resources |
| Returns: Full resource database |
| """ |
| return await self._request("/api/resources/list") |
| |
| async def get_categories(self) -> Dict[str, Any]: |
| """ |
| Get list of all 12 resource categories |
| Categories: rpc_nodes, block_explorers, market_data_apis, news_apis, |
| sentiment_apis, onchain_analytics_apis, whale_tracking_apis, |
| hf_resources, free_http_endpoints, cors_proxies, |
| community_sentiment_apis, local_backend_routes |
| """ |
| return await self._request("/api/categories") |
| |
| async def get_resources_by_category(self, category: str) -> Dict[str, Any]: |
| """ |
| Get resources for a specific category |
| |
| Args: |
| category: Category name (e.g., 'market_data_apis', 'sentiment_apis') |
| """ |
| return await self._request(f"/api/resources/category/{category}") |
| |
| |
| |
| async def health_check(self) -> Dict[str, Any]: |
| """Check API health status""" |
| return await self._request("/health") |
| |
| |
| |
| async def get_total_resources_count(self) -> int: |
| """ |
| Get total number of resources available |
| |
| Returns: |
| int: Total resource count |
| """ |
| result = await self.get_resources_stats() |
| if result["success"] and result["data"]: |
| return result["data"].get("total_resources", 0) |
| return 0 |
| |
| async def get_category_summary(self) -> Dict[str, int]: |
| """ |
| Get summary of resources per category |
| |
| Returns: |
| Dict mapping category names to resource counts |
| """ |
| result = await self.get_resources_stats() |
| if result["success"] and result["data"]: |
| return result["data"].get("categories", {}) |
| return {} |
| |
| async def get_market_data_providers(self) -> List[Dict[str, Any]]: |
| """ |
| Get all market data API providers (33 resources) |
| """ |
| result = await self.get_resources_by_category("market_data_apis") |
| if result["success"] and result["data"]: |
| return result["data"].get("resources", []) |
| return [] |
| |
| async def get_sentiment_providers(self) -> List[Dict[str, Any]]: |
| """ |
| Get all sentiment API providers (14 resources) |
| """ |
| result = await self.get_resources_by_category("sentiment_apis") |
| if result["success"] and result["data"]: |
| return result["data"].get("resources", []) |
| return [] |
| |
| async def get_rpc_nodes(self) -> List[Dict[str, Any]]: |
| """ |
| Get all RPC node providers (24 resources) |
| """ |
| result = await self.get_resources_by_category("rpc_nodes") |
| if result["success"] and result["data"]: |
| return result["data"].get("resources", []) |
| return [] |
| |
| async def get_block_explorers(self) -> List[Dict[str, Any]]: |
| """ |
| Get all block explorer APIs (33 resources) |
| """ |
| result = await self.get_resources_by_category("block_explorers") |
| if result["success"] and result["data"]: |
| return result["data"].get("resources", []) |
| return [] |
|
|
|
|
| |
|
|
| _service_instance: Optional[CryptoAPICleanService] = None |
|
|
|
|
| def get_crypto_api_clean_service() -> CryptoAPICleanService: |
| """Get singleton instance of Crypto API Clean Service""" |
| global _service_instance |
| if _service_instance is None: |
| _service_instance = CryptoAPICleanService() |
| return _service_instance |
|
|
|
|
| |
|
|
| async def fetch_crypto_api_clean_stats() -> Dict[str, Any]: |
| """Fetch resource statistics from Crypto API Clean""" |
| service = get_crypto_api_clean_service() |
| return await service.get_resources_stats() |
|
|
|
|
| async def fetch_crypto_api_clean_resources(category: Optional[str] = None) -> Dict[str, Any]: |
| """Fetch resources from Crypto API Clean""" |
| service = get_crypto_api_clean_service() |
| if category: |
| return await service.get_resources_by_category(category) |
| return await service.get_all_resources() |
|
|
|
|
| |
|
|
| if __name__ == "__main__": |
| async def main(): |
| service = get_crypto_api_clean_service() |
| |
| print("=" * 70) |
| print("Testing Crypto API Clean Service") |
| print("=" * 70) |
| |
| |
| print("\n1. Health Check:") |
| result = await service.health_check() |
| print(f" Success: {result['success']}") |
| |
| |
| print("\n2. Resource Statistics:") |
| result = await service.get_resources_stats() |
| if result['success']: |
| data = result['data'] |
| print(f" Total Resources: {data.get('total_resources')}") |
| print(f" Total Categories: {data.get('total_categories')}") |
| print(f" Categories:") |
| for cat, count in data.get('categories', {}).items(): |
| print(f" - {cat}: {count}") |
| |
| |
| print("\n3. Market Data Providers:") |
| providers = await service.get_market_data_providers() |
| print(f" Found {len(providers)} providers") |
| |
| |
| print("\n4. Sentiment Providers:") |
| providers = await service.get_sentiment_providers() |
| print(f" Found {len(providers)} providers") |
| |
| await service.close() |
| print("\n" + "=" * 70) |
| print("Tests completed!") |
| |
| asyncio.run(main()) |
|
|
