389379d8bb
- 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
315 lines
10 KiB
Python
315 lines
10 KiB
Python
"""
|
|
Sedo Official API Client
|
|
|
|
This service provides access to Sedo's official API for:
|
|
- Domain search and auctions
|
|
- Marketplace listings
|
|
- Domain pricing
|
|
|
|
API Documentation: https://api.sedo.com/apidocs/v1/
|
|
Type: XML-RPC based API
|
|
|
|
SECURITY:
|
|
- Credentials are loaded from environment variables
|
|
- NEVER hardcode credentials in this file
|
|
|
|
WHERE TO FIND YOUR CREDENTIALS:
|
|
1. Login to https://sedo.com
|
|
2. Go to "Mein Sedo" / "My Sedo"
|
|
3. Navigate to "API-Zugang" / "API Access"
|
|
4. You'll find:
|
|
- Partner ID (your user ID)
|
|
- SignKey (signature key for authentication)
|
|
|
|
Usage:
|
|
from app.services.sedo_api import sedo_client
|
|
|
|
# Search domains for sale
|
|
listings = await sedo_client.search_domains(keyword="tech")
|
|
"""
|
|
import logging
|
|
import hashlib
|
|
import time
|
|
from datetime import datetime, timedelta
|
|
from typing import Optional, List, Dict, Any
|
|
import httpx
|
|
from xml.etree import ElementTree
|
|
|
|
from app.config import get_settings
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class SedoAPIClient:
|
|
"""
|
|
Official Sedo API Client.
|
|
|
|
Sedo uses an XML-RPC style API with signature-based authentication.
|
|
Each request must include:
|
|
- partnerid: Your partner ID
|
|
- signkey: Your signature key (or hashed signature)
|
|
"""
|
|
|
|
def __init__(self):
|
|
self.settings = get_settings()
|
|
self.base_url = self.settings.sedo_api_base or "https://api.sedo.com/api/v1/"
|
|
self.partner_id = self.settings.sedo_partner_id
|
|
self.sign_key = self.settings.sedo_sign_key
|
|
|
|
# HTTP client
|
|
self._client: Optional[httpx.AsyncClient] = None
|
|
|
|
@property
|
|
def is_configured(self) -> bool:
|
|
"""Check if API credentials are configured."""
|
|
return bool(self.partner_id and self.sign_key)
|
|
|
|
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/x-www-form-urlencoded",
|
|
"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
|
|
|
|
def _generate_signature(self, params: Dict[str, Any]) -> str:
|
|
"""
|
|
Generate request signature for Sedo API.
|
|
|
|
The signature is typically: MD5(signkey + sorted_params)
|
|
Check Sedo docs for exact implementation.
|
|
"""
|
|
# Simple implementation - may need adjustment based on actual Sedo requirements
|
|
sorted_params = "&".join(f"{k}={v}" for k, v in sorted(params.items()))
|
|
signature_base = f"{self.sign_key}{sorted_params}"
|
|
return hashlib.md5(signature_base.encode()).hexdigest()
|
|
|
|
async def _request(
|
|
self,
|
|
endpoint: str,
|
|
params: Optional[Dict] = None
|
|
) -> Dict[str, Any]:
|
|
"""Make an authenticated API request."""
|
|
if not self.is_configured:
|
|
raise ValueError("Sedo API credentials not configured")
|
|
|
|
client = await self._get_client()
|
|
|
|
# Base params for all requests
|
|
request_params = {
|
|
"partnerid": self.partner_id,
|
|
"signkey": self.sign_key,
|
|
**(params or {})
|
|
}
|
|
|
|
url = f"{self.base_url.rstrip('/')}/{endpoint.lstrip('/')}"
|
|
|
|
try:
|
|
response = await client.get(url, params=request_params)
|
|
response.raise_for_status()
|
|
|
|
# Sedo API can return XML or JSON depending on endpoint
|
|
content_type = response.headers.get("content-type", "")
|
|
|
|
if "xml" in content_type:
|
|
return self._parse_xml_response(response.text)
|
|
elif "json" in content_type:
|
|
return response.json()
|
|
else:
|
|
# Try JSON first, fallback to XML
|
|
try:
|
|
return response.json()
|
|
except:
|
|
return self._parse_xml_response(response.text)
|
|
|
|
except httpx.HTTPError as e:
|
|
logger.error(f"Sedo API request failed: {e}")
|
|
raise
|
|
|
|
def _parse_xml_response(self, xml_text: str) -> Dict[str, Any]:
|
|
"""Parse XML response from Sedo API."""
|
|
try:
|
|
root = ElementTree.fromstring(xml_text)
|
|
return self._xml_to_dict(root)
|
|
except Exception as e:
|
|
logger.warning(f"Failed to parse XML: {e}")
|
|
return {"raw": xml_text}
|
|
|
|
def _xml_to_dict(self, element) -> Dict[str, Any]:
|
|
"""Convert XML element to dictionary."""
|
|
result = {}
|
|
for child in element:
|
|
if len(child) > 0:
|
|
result[child.tag] = self._xml_to_dict(child)
|
|
else:
|
|
result[child.tag] = child.text
|
|
return result
|
|
|
|
# =========================================================================
|
|
# DOMAIN SEARCH ENDPOINTS
|
|
# =========================================================================
|
|
|
|
async def search_domains(
|
|
self,
|
|
keyword: Optional[str] = None,
|
|
tld: Optional[str] = None,
|
|
min_price: Optional[float] = None,
|
|
max_price: Optional[float] = None,
|
|
page: int = 1,
|
|
page_size: int = 100,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Search for domains listed on Sedo marketplace.
|
|
|
|
Returns domains for sale (not auctions).
|
|
"""
|
|
params = {
|
|
"output_method": "json", # Request JSON response
|
|
}
|
|
|
|
if keyword:
|
|
params["keyword"] = 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 page:
|
|
params["page"] = page
|
|
if page_size:
|
|
params["pagesize"] = min(page_size, 100)
|
|
|
|
return await self._request("DomainSearch", params)
|
|
|
|
async def search_auctions(
|
|
self,
|
|
keyword: Optional[str] = None,
|
|
tld: Optional[str] = None,
|
|
ending_within_hours: Optional[int] = None,
|
|
page: int = 1,
|
|
page_size: int = 100,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Search for active domain auctions on Sedo.
|
|
"""
|
|
params = {
|
|
"output_method": "json",
|
|
"auction": "true", # Only auctions
|
|
}
|
|
|
|
if keyword:
|
|
params["keyword"] = keyword
|
|
if tld:
|
|
params["tld"] = tld.lstrip(".")
|
|
if page:
|
|
params["page"] = page
|
|
if page_size:
|
|
params["pagesize"] = min(page_size, 100)
|
|
|
|
return await self._request("DomainSearch", params)
|
|
|
|
async def get_domain_details(self, domain: str) -> Dict[str, Any]:
|
|
"""Get detailed information about a specific domain."""
|
|
params = {
|
|
"domain": domain,
|
|
"output_method": "json",
|
|
}
|
|
return await self._request("DomainDetails", params)
|
|
|
|
async def get_ending_soon_auctions(
|
|
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
|
|
)
|
|
|
|
# =========================================================================
|
|
# 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,
|
|
"hint": "Find your credentials at: Sedo.com → Mein Sedo → API-Zugang"
|
|
}
|
|
|
|
try:
|
|
# Try a simple search to test connection
|
|
result = await self.search_domains(keyword="test", page_size=1)
|
|
return {
|
|
"success": True,
|
|
"configured": True,
|
|
"partner_id": self.partner_id,
|
|
"authenticated_at": datetime.utcnow().isoformat()
|
|
}
|
|
except Exception as e:
|
|
return {
|
|
"success": False,
|
|
"error": str(e),
|
|
"configured": True
|
|
}
|
|
|
|
def transform_to_pounce_format(self, sedo_listing: Dict) -> Dict[str, Any]:
|
|
"""
|
|
Transform Sedo listing to Pounce internal format.
|
|
|
|
Maps Sedo fields to our DomainAuction model.
|
|
"""
|
|
domain = sedo_listing.get("domain") or sedo_listing.get("domainname", "")
|
|
tld = domain.rsplit(".", 1)[1] if "." in domain else ""
|
|
|
|
# Parse end time if auction
|
|
end_time_str = sedo_listing.get("auctionend") or sedo_listing.get("enddate")
|
|
if end_time_str:
|
|
try:
|
|
end_time = datetime.fromisoformat(end_time_str.replace("Z", "+00:00"))
|
|
except:
|
|
end_time = datetime.utcnow() + timedelta(days=7)
|
|
else:
|
|
end_time = datetime.utcnow() + timedelta(days=7)
|
|
|
|
# Price handling
|
|
price = sedo_listing.get("price") or sedo_listing.get("currentbid") or 0
|
|
if isinstance(price, str):
|
|
price = float(price.replace(",", "").replace("$", "").replace("€", ""))
|
|
|
|
return {
|
|
"domain": domain,
|
|
"tld": tld,
|
|
"platform": "Sedo",
|
|
"current_bid": price,
|
|
"buy_now_price": sedo_listing.get("buynow") or sedo_listing.get("bin"),
|
|
"currency": sedo_listing.get("currency", "EUR"),
|
|
"num_bids": sedo_listing.get("numbids") or sedo_listing.get("bidcount", 0),
|
|
"end_time": end_time,
|
|
"auction_url": f"https://sedo.com/search/details/?domain={domain}",
|
|
"age_years": None,
|
|
"reserve_met": sedo_listing.get("reservemet"),
|
|
"traffic": sedo_listing.get("traffic"),
|
|
"is_auction": sedo_listing.get("isaution") == "1" or sedo_listing.get("auction") == True,
|
|
}
|
|
|
|
|
|
# Singleton instance
|
|
sedo_client = SedoAPIClient()
|
|
|