Spaces:
Sleeping
Sleeping
File size: 2,056 Bytes
68025ee | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 | import logging
import os
import requests
logger = logging.getLogger(__name__)
CRYPTOPANIC_API_KEY = os.getenv("CRYPTOPANIC_API_KEY", "")
CRYPTOCOMPARE_API_KEY = os.getenv("CRYPTOCOMPARE_API_KEY", "")
def fetch_news(asset: str, date: str = None, limit: int = 10) -> list:
"""Fetch crypto news. Returns list of {title, url, published_at}."""
currency = asset.split("/")[0] # BTC, ETH
# Try CryptoPanic first
if CRYPTOPANIC_API_KEY:
try:
return _fetch_cryptopanic(currency, limit)
except Exception as e:
logger.warning(f"CryptoPanic failed: {e}")
# Try CryptoCompare
try:
return _fetch_cryptocompare(currency, limit)
except Exception as e:
logger.warning(f"CryptoCompare failed: {e}")
return []
def _fetch_cryptopanic(currency: str, limit: int) -> list:
url = "https://cryptopanic.com/api/v1/posts/"
params = {
"auth_token": CRYPTOPANIC_API_KEY,
"currencies": currency,
"kind": "news",
"limit": limit,
}
resp = requests.get(url, params=params, timeout=15)
resp.raise_for_status()
data = resp.json()
results = []
for item in data.get("results", [])[:limit]:
results.append({
"title": item.get("title", ""),
"url": item.get("url", ""),
"published_at": item.get("published_at", ""),
})
return results
def _fetch_cryptocompare(currency: str, limit: int) -> list:
url = "https://min-api.cryptocompare.com/data/v2/news/"
params = {"categories": currency, "lang": "EN", "limit": limit}
if CRYPTOCOMPARE_API_KEY:
params["api_key"] = CRYPTOCOMPARE_API_KEY
resp = requests.get(url, params=params, timeout=15)
resp.raise_for_status()
data = resp.json()
results = []
for item in data.get("Data", [])[:limit]:
results.append({
"title": item.get("title", ""),
"url": item.get("url", ""),
"published_at": str(item.get("published_on", "")),
})
return results
|