e9f06d1cbf
- DropCatch API Client mit OAuth2 Authentifizierung - Sedo API Client (bereit für Credentials) - Tier 1 APIs → Tier 2 Scraping Fallback-Logik - Admin Endpoints: /test-apis, /trigger-scrape, /scrape-status - MARKET_CONCEPT.md komplett überarbeitet: - Realistische Bestandsaufnahme - 3-Säulen-Konzept (Auktionen, Pounce Direct, Drops) - API-Realität dokumentiert (DropCatch = nur eigene Aktivitäten) - Roadmap und nächste Schritte
335 lines
11 KiB
Python
335 lines
11 KiB
Python
"""
|
|
DropCatch Official API Client
|
|
|
|
This service provides access to DropCatch's official API for:
|
|
- Searching domain auctions
|
|
- Getting auction details
|
|
- Backorder management
|
|
|
|
API Documentation: https://www.dropcatch.com/hiw/dropcatch-api
|
|
Interactive Docs: https://api.dropcatch.com/swagger
|
|
|
|
SECURITY:
|
|
- Credentials are loaded from environment variables
|
|
- NEVER hardcode credentials in this file
|
|
|
|
Usage:
|
|
from app.services.dropcatch_api import dropcatch_client
|
|
|
|
# Get active auctions
|
|
auctions = await dropcatch_client.search_auctions(keyword="tech")
|
|
"""
|
|
import logging
|
|
from datetime import datetime, timedelta
|
|
from typing import Optional, List, Dict, Any
|
|
import httpx
|
|
from functools import lru_cache
|
|
|
|
from app.config import get_settings
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class DropCatchAPIClient:
|
|
"""
|
|
Official DropCatch API Client.
|
|
|
|
This uses the V2 API endpoints (V1 is deprecated).
|
|
Authentication is via OAuth2 client credentials.
|
|
"""
|
|
|
|
def __init__(self):
|
|
self.settings = get_settings()
|
|
self.base_url = self.settings.dropcatch_api_base or "https://api.dropcatch.com"
|
|
self.client_id = self.settings.dropcatch_client_id
|
|
self.client_secret = self.settings.dropcatch_client_secret
|
|
|
|
# Token cache
|
|
self._access_token: Optional[str] = None
|
|
self._token_expires_at: Optional[datetime] = None
|
|
|
|
# HTTP client
|
|
self._client: Optional[httpx.AsyncClient] = None
|
|
|
|
@property
|
|
def is_configured(self) -> bool:
|
|
"""Check if API credentials are configured."""
|
|
return bool(self.client_id and self.client_secret)
|
|
|
|
async def _get_client(self) -> httpx.AsyncClient:
|
|
"""Get or create HTTP client."""
|
|
if self._client is None or self._client.is_closed:
|
|
self._client = httpx.AsyncClient(
|
|
timeout=30.0,
|
|
headers={
|
|
"Content-Type": "application/json",
|
|
"User-Agent": "Pounce/1.0 (Domain Intelligence Platform)"
|
|
}
|
|
)
|
|
return self._client
|
|
|
|
async def close(self):
|
|
"""Close the HTTP client."""
|
|
if self._client and not self._client.is_closed:
|
|
await self._client.aclose()
|
|
self._client = None
|
|
|
|
async def _authenticate(self) -> str:
|
|
"""
|
|
Authenticate with DropCatch API and get access token.
|
|
|
|
POST https://api.dropcatch.com/authorize
|
|
Body: { "clientId": "...", "clientSecret": "..." }
|
|
|
|
Returns: Access token string
|
|
"""
|
|
if not self.is_configured:
|
|
raise ValueError("DropCatch API credentials not configured")
|
|
|
|
# Check if we have a valid cached token
|
|
if self._access_token and self._token_expires_at:
|
|
if datetime.utcnow() < self._token_expires_at - timedelta(minutes=5):
|
|
return self._access_token
|
|
|
|
client = await self._get_client()
|
|
|
|
try:
|
|
response = await client.post(
|
|
f"{self.base_url}/authorize",
|
|
json={
|
|
"clientId": self.client_id,
|
|
"clientSecret": self.client_secret
|
|
}
|
|
)
|
|
|
|
if response.status_code != 200:
|
|
logger.error(f"DropCatch auth failed: {response.status_code} - {response.text}")
|
|
raise Exception(f"Authentication failed: {response.status_code}")
|
|
|
|
data = response.json()
|
|
|
|
# Extract token - the response format may vary
|
|
# Common formats: { "token": "...", "expiresIn": 3600 }
|
|
# or: { "accessToken": "...", "expiresIn": 3600 }
|
|
self._access_token = data.get("token") or data.get("accessToken") or data.get("access_token")
|
|
|
|
# Calculate expiry (default 1 hour if not specified)
|
|
expires_in = data.get("expiresIn") or data.get("expires_in") or 3600
|
|
self._token_expires_at = datetime.utcnow() + timedelta(seconds=expires_in)
|
|
|
|
logger.info("DropCatch API: Successfully authenticated")
|
|
return self._access_token
|
|
|
|
except httpx.HTTPError as e:
|
|
logger.error(f"DropCatch auth HTTP error: {e}")
|
|
raise
|
|
|
|
async def _request(
|
|
self,
|
|
method: str,
|
|
endpoint: str,
|
|
params: Optional[Dict] = None,
|
|
json_data: Optional[Dict] = None
|
|
) -> Dict[str, Any]:
|
|
"""Make an authenticated API request."""
|
|
token = await self._authenticate()
|
|
client = await self._get_client()
|
|
|
|
headers = {
|
|
"Authorization": f"Bearer {token}"
|
|
}
|
|
|
|
url = f"{self.base_url}{endpoint}"
|
|
|
|
try:
|
|
response = await client.request(
|
|
method=method,
|
|
url=url,
|
|
params=params,
|
|
json=json_data,
|
|
headers=headers
|
|
)
|
|
|
|
if response.status_code == 401:
|
|
# Token expired, re-authenticate
|
|
self._access_token = None
|
|
token = await self._authenticate()
|
|
headers["Authorization"] = f"Bearer {token}"
|
|
response = await client.request(
|
|
method=method,
|
|
url=url,
|
|
params=params,
|
|
json=json_data,
|
|
headers=headers
|
|
)
|
|
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
except httpx.HTTPError as e:
|
|
logger.error(f"DropCatch API request failed: {e}")
|
|
raise
|
|
|
|
# =========================================================================
|
|
# AUCTION ENDPOINTS (V2)
|
|
# =========================================================================
|
|
|
|
async def search_auctions(
|
|
self,
|
|
keyword: Optional[str] = None,
|
|
tld: Optional[str] = None,
|
|
min_price: Optional[float] = None,
|
|
max_price: Optional[float] = None,
|
|
ending_within_hours: Optional[int] = None,
|
|
page_size: int = 100,
|
|
page_token: Optional[str] = None,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Search for domain auctions.
|
|
|
|
Endpoint: GET /v2/auctions (or similar - check interactive docs)
|
|
|
|
Returns:
|
|
{
|
|
"auctions": [...],
|
|
"cursor": {
|
|
"next": "...",
|
|
"previous": "..."
|
|
}
|
|
}
|
|
"""
|
|
params = {
|
|
"pageSize": page_size,
|
|
}
|
|
|
|
if keyword:
|
|
params["searchTerm"] = keyword
|
|
if tld:
|
|
params["tld"] = tld.lstrip(".")
|
|
if min_price is not None:
|
|
params["minPrice"] = min_price
|
|
if max_price is not None:
|
|
params["maxPrice"] = max_price
|
|
if ending_within_hours:
|
|
params["endingWithinHours"] = ending_within_hours
|
|
if page_token:
|
|
params["pageToken"] = page_token
|
|
|
|
return await self._request("GET", "/v2/auctions", params=params)
|
|
|
|
async def get_auction(self, auction_id: int) -> Dict[str, Any]:
|
|
"""Get details for a specific auction."""
|
|
return await self._request("GET", f"/v2/auctions/{auction_id}")
|
|
|
|
async def get_ending_soon(
|
|
self,
|
|
hours: int = 24,
|
|
page_size: int = 50
|
|
) -> Dict[str, Any]:
|
|
"""Get auctions ending soon."""
|
|
return await self.search_auctions(
|
|
ending_within_hours=hours,
|
|
page_size=page_size
|
|
)
|
|
|
|
async def get_hot_auctions(self, page_size: int = 50) -> Dict[str, Any]:
|
|
"""
|
|
Get hot/popular auctions (high bid activity).
|
|
Note: The actual endpoint may vary - check interactive docs.
|
|
"""
|
|
# This might be a different endpoint or sort parameter
|
|
params = {
|
|
"pageSize": page_size,
|
|
"sortBy": "bidCount", # or "popularity" - check docs
|
|
"sortOrder": "desc"
|
|
}
|
|
return await self._request("GET", "/v2/auctions", params=params)
|
|
|
|
# =========================================================================
|
|
# BACKORDER ENDPOINTS (V2)
|
|
# =========================================================================
|
|
|
|
async def search_backorders(
|
|
self,
|
|
keyword: Optional[str] = None,
|
|
page_size: int = 100,
|
|
page_token: Optional[str] = None,
|
|
) -> Dict[str, Any]:
|
|
"""Search for available backorders (domains dropping soon)."""
|
|
params = {"pageSize": page_size}
|
|
|
|
if keyword:
|
|
params["searchTerm"] = keyword
|
|
if page_token:
|
|
params["pageToken"] = page_token
|
|
|
|
return await self._request("GET", "/v2/backorders", params=params)
|
|
|
|
# =========================================================================
|
|
# UTILITY METHODS
|
|
# =========================================================================
|
|
|
|
async def test_connection(self) -> Dict[str, Any]:
|
|
"""Test the API connection and credentials."""
|
|
if not self.is_configured:
|
|
return {
|
|
"success": False,
|
|
"error": "API credentials not configured",
|
|
"configured": False
|
|
}
|
|
|
|
try:
|
|
await self._authenticate()
|
|
return {
|
|
"success": True,
|
|
"configured": True,
|
|
"client_id": self.client_id.split(":")[0] if ":" in self.client_id else self.client_id,
|
|
"authenticated_at": datetime.utcnow().isoformat()
|
|
}
|
|
except Exception as e:
|
|
return {
|
|
"success": False,
|
|
"error": str(e),
|
|
"configured": True
|
|
}
|
|
|
|
def transform_to_pounce_format(self, dc_auction: Dict) -> Dict[str, Any]:
|
|
"""
|
|
Transform DropCatch auction to Pounce internal format.
|
|
|
|
Maps DropCatch fields to our DomainAuction model.
|
|
"""
|
|
domain = dc_auction.get("domainName") or dc_auction.get("domain", "")
|
|
tld = domain.rsplit(".", 1)[1] if "." in domain else ""
|
|
|
|
# Parse end time (format may vary)
|
|
end_time_str = dc_auction.get("auctionEndTime") or dc_auction.get("endTime")
|
|
if end_time_str:
|
|
try:
|
|
end_time = datetime.fromisoformat(end_time_str.replace("Z", "+00:00"))
|
|
except:
|
|
end_time = datetime.utcnow() + timedelta(days=1)
|
|
else:
|
|
end_time = datetime.utcnow() + timedelta(days=1)
|
|
|
|
return {
|
|
"domain": domain,
|
|
"tld": tld,
|
|
"platform": "DropCatch",
|
|
"current_bid": dc_auction.get("currentBid") or dc_auction.get("price", 0),
|
|
"currency": "USD",
|
|
"num_bids": dc_auction.get("bidCount") or dc_auction.get("numberOfBids", 0),
|
|
"end_time": end_time,
|
|
"auction_url": f"https://www.dropcatch.com/domain/{domain}",
|
|
"age_years": dc_auction.get("yearsOld") or dc_auction.get("age"),
|
|
"buy_now_price": dc_auction.get("buyNowPrice"),
|
|
"reserve_met": dc_auction.get("reserveMet"),
|
|
"traffic": dc_auction.get("traffic"),
|
|
"external_id": str(dc_auction.get("auctionId") or dc_auction.get("id", "")),
|
|
}
|
|
|
|
|
|
# Singleton instance
|
|
dropcatch_client = DropCatchAPIClient()
|
|
|