Spaces:
Paused
Paused
File size: 1,605 Bytes
29348f5 81ebf36 29348f5 8584387 29348f5 8584387 29348f5 8584387 29348f5 8584387 29348f5 | 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 | from abc import ABC, abstractmethod
from typing import Dict, Tuple, Optional
import httpx
from mediaflow_proxy.configs import settings
class BaseExtractor(ABC):
"""Base class for all URL extractors."""
def __init__(self, proxy_enabled: bool, request_headers: dict):
self.proxy_url = settings.proxy_url if proxy_enabled else None
self.base_headers = {
"user-agent": settings.user_agent,
"accept-language": "en-US,en;q=0.5",
}
self.base_headers.update(request_headers)
async def _make_request(
self, url: str, headers: Optional[Dict] = None, follow_redirects: bool = True, **kwargs
) -> httpx.Response:
"""Make HTTP request with error handling."""
try:
async with httpx.AsyncClient(proxy=self.proxy_url) as client:
request_headers = self.base_headers
request_headers.update(headers or {})
response = await client.get(
url,
headers=request_headers,
follow_redirects=follow_redirects,
timeout=30,
**kwargs,
)
response.raise_for_status()
return response
except httpx.HTTPError as e:
raise ValueError(f"HTTP request failed: {str(e)}")
except Exception as e:
raise ValueError(f"Request failed: {str(e)}")
@abstractmethod
async def extract(self, url: str) -> Tuple[str, Dict[str, str]]:
"""Extract final URL and required headers."""
pass
|