#!/usr/bin/env python3 """ Crypto Resources API - Hugging Face Space سرور API با رابط کاربری وب و WebSocket """ from fastapi import FastAPI, WebSocket, WebSocketDisconnect from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import JSONResponse, HTMLResponse from fastapi.staticfiles import StaticFiles from datetime import datetime from pathlib import Path import json import asyncio from typing import List, Dict, Any, Set import logging # Setup logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # Load resources def load_resources(): """بارگذاری منابع از فایل JSON""" resources_file = Path("api-resources/crypto_resources_unified_2025-11-11.json") if not resources_file.exists(): logger.warning(f"Resources file not found: {resources_file}") return {} try: with open(resources_file, 'r', encoding='utf-8') as f: data = json.load(f) logger.info(f"✅ Loaded resources from {resources_file}") return data.get('registry', {}) except Exception as e: logger.error(f"Error loading resources: {e}") return {} # Create FastAPI app app = FastAPI( title="Crypto Resources API", description="API جامع برای دسترسی به منابع داده کریپتوکارنسی", version="2.0.0", docs_url="/docs", redoc_url="/redoc" ) # CORS middleware app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # Load resources RESOURCES = load_resources() # WebSocket connection manager class ConnectionManager: def __init__(self): self.active_connections: Set[WebSocket] = set() async def connect(self, websocket: WebSocket): await websocket.accept() self.active_connections.add(websocket) logger.info(f"WebSocket connected. Total: {len(self.active_connections)}") def disconnect(self, websocket: WebSocket): self.active_connections.discard(websocket) logger.info(f"WebSocket disconnected. Total: {len(self.active_connections)}") async def broadcast(self, message: dict): """ارسال پیام به همه کلاینت‌ها""" disconnected = set() for connection in self.active_connections: try: await connection.send_json(message) except Exception as e: logger.error(f"Error sending to client: {e}") disconnected.add(connection) # حذف اتصالات قطع شده for conn in disconnected: self.active_connections.discard(conn) manager = ConnectionManager() # Background task for broadcasting stats async def broadcast_stats(): """ارسال دوره‌ای آمار به کلاینت‌ها""" while True: try: if manager.active_connections: stats = get_stats_data() await manager.broadcast({ "type": "stats_update", "data": stats, "timestamp": datetime.now().isoformat() }) await asyncio.sleep(10) # هر 10 ثانیه except Exception as e: logger.error(f"Error in broadcast_stats: {e}") await asyncio.sleep(5) # Startup event @app.on_event("startup") async def startup_event(): """راه‌اندازی سرویس‌های پس‌زمینه""" logger.info("🚀 Starting Crypto Resources API...") logger.info(f"📦 Loaded {len([k for k,v in RESOURCES.items() if isinstance(v, list)])} categories") # شروع broadcast task asyncio.create_task(broadcast_stats()) logger.info("✅ Background tasks started") # Helper functions def get_stats_data(): """دریافت آمار کلی""" categories_count = {} total_resources = 0 for key, value in RESOURCES.items(): if isinstance(value, list): count = len(value) categories_count[key] = count total_resources += count return { "total_resources": total_resources, "total_categories": len(categories_count), "categories": categories_count } # HTML UI HTML_TEMPLATE = """ Crypto Resources API

🚀 Crypto Resources API

API جامع برای دسترسی به منابع داده کریپتوکارنسی

در حال اتصال...
مجموع منابع
0
دسته‌بندی‌ها
0
وضعیت سرور
⏳

📂 دسته‌بندی منابع

در حال بارگذاری...

📡 API Endpoints

GET /health - Health check
GET /api/resources/stats - آمار کلی منابع
GET /api/resources/list - لیست تمام منابع
GET /api/categories - لیست دسته‌بندی‌ها
GET /api/resources/category/{category} - منابع یک دسته خاص
WS /ws - WebSocket برای بروزرسانی لحظه‌ای

🔌 WebSocket Status: Disconnected

در انتظار اتصال...

💜 ساخته شده با عشق برای جامعه کریپتو

📚 مستندات کامل: /docs

""" # Routes @app.get("/", response_class=HTMLResponse) async def root(): """صفحه اصلی با UI""" return HTMLResponse(content=HTML_TEMPLATE) @app.get("/health") async def health(): """Health check""" return { "status": "healthy", "timestamp": datetime.now().isoformat(), "resources_loaded": len(RESOURCES) > 0, "total_categories": len([k for k, v in RESOURCES.items() if isinstance(v, list)]), "websocket_connections": len(manager.active_connections) } @app.get("/api/resources/stats") async def resources_stats(): """آمار منابع""" stats = get_stats_data() metadata = RESOURCES.get('metadata', {}) return { **stats, "metadata": metadata, "timestamp": datetime.now().isoformat() } @app.get("/api/resources/list") async def resources_list(): """لیست همه منابع""" all_resources = [] for category, resources in RESOURCES.items(): if isinstance(resources, list): for resource in resources: if isinstance(resource, dict): all_resources.append({ "category": category, "id": resource.get('id', 'unknown'), "name": resource.get('name', 'Unknown'), "base_url": resource.get('base_url', ''), "auth_type": resource.get('auth', {}).get('type', 'none') }) return { "total": len(all_resources), "resources": all_resources[:100], # اولین 100 مورد "note": f"Showing first 100 of {len(all_resources)} resources", "timestamp": datetime.now().isoformat() } @app.get("/api/resources/category/{category}") async def resources_by_category(category: str): """منابع یک دسته خاص""" if category not in RESOURCES: return JSONResponse( status_code=404, content={"error": f"Category '{category}' not found"} ) resources = RESOURCES.get(category, []) if not isinstance(resources, list): return JSONResponse( status_code=400, content={"error": f"Category '{category}' is not a resource list"} ) return { "category": category, "total": len(resources), "resources": resources, "timestamp": datetime.now().isoformat() } @app.get("/api/categories") async def list_categories(): """لیست دسته‌بندی‌ها""" categories = [] for key, value in RESOURCES.items(): if isinstance(value, list): categories.append({ "name": key, "count": len(value), "endpoint": f"/api/resources/category/{key}" }) return { "total": len(categories), "categories": categories, "timestamp": datetime.now().isoformat() } @app.websocket("/ws") async def websocket_endpoint(websocket: WebSocket): """WebSocket endpoint برای بروزرسانی لحظه‌ای""" await manager.connect(websocket) try: # ارسال آمار اولیه stats = get_stats_data() await websocket.send_json({ "type": "initial_stats", "data": stats, "timestamp": datetime.now().isoformat() }) # نگه داشتن اتصال while True: try: # دریافت پیام از کلاینت (اگر بفرستد) data = await websocket.receive_text() logger.info(f"Received from client: {data}") # پاسخ به کلاینت await websocket.send_json({ "type": "pong", "message": "Server is alive", "timestamp": datetime.now().isoformat() }) except Exception as e: logger.error(f"Error in websocket loop: {e}") break except WebSocketDisconnect: manager.disconnect(websocket) logger.info("Client disconnected normally") except Exception as e: logger.error(f"WebSocket error: {e}") manager.disconnect(websocket) # Run with uvicorn if __name__ == "__main__": import uvicorn print("=" * 80) print("🚀 راه‌اندازی Crypto Resources API Server") print("=" * 80) print(f"\nبارگذاری منابع...") print(f"✅ {len([k for k,v in RESOURCES.items() if isinstance(v, list)])} دسته بارگذاری شد") print(f"\n🌐 Server: http://0.0.0.0:7860") print(f"📚 Docs: http://0.0.0.0:7860/docs") print(f"🔌 WebSocket: ws://0.0.0.0:7860/ws") print(f"\nبرای توقف سرور: Ctrl+C") print("=" * 80 + "\n") uvicorn.run( app, host="0.0.0.0", port=7860, log_level="info", access_log=True )