File size: 2,834 Bytes
d22bb02 | 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 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 | from abc import ABC, abstractmethod
from typing import List
import asyncio
from app.models.proxy import Proxy
from app.models.source import SourceConfig, SourceType
from app.grabber.patterns import ProxyPatterns
from app.grabber.parsers import VMessParser, VLESSParser, TrojanParser, SSParser
from app.utils.base64_decoder import SubscriptionDecoder
class BaseGrabber(ABC):
def __init__(
self, max_retries: int = 3, retry_delay: float = 1.0, timeout: int = 30
):
self.max_retries = max_retries
self.retry_delay = retry_delay
self.timeout = timeout
@abstractmethod
async def fetch_content(self, source: SourceConfig) -> str:
pass
async def extract_proxies(self, source: SourceConfig) -> List[Proxy]:
content = await self.fetch_content(source)
if source.selector:
proxies = await self._try_exact_selector(content, source)
if proxies:
return proxies
proxies = await self.parse_content(content, source.type)
return proxies
async def _try_exact_selector(
self, content: str, source: SourceConfig
) -> List[Proxy]:
return []
async def parse_content(self, content: str, source_type: SourceType) -> List[Proxy]:
proxies = []
if source_type == SourceType.SUBSCRIPTION_BASE64:
try:
content = SubscriptionDecoder.decode(content)
except ValueError:
pass
http_matches = ProxyPatterns.extract_http_proxies(content)
for ip, port in http_matches:
proxies.append(
Proxy(ip=ip, port=int(port), protocol="http", source=str(source_type))
)
vmess_urls = ProxyPatterns.extract_vmess_urls(content)
for url in vmess_urls:
try:
proxy = VMessParser.parse(url)
proxies.append(proxy)
except ValueError:
continue
vless_urls = ProxyPatterns.extract_vless_urls(content)
for url in vless_urls:
try:
proxy = VLESSParser.parse(url)
proxies.append(proxy)
except ValueError:
continue
trojan_urls = ProxyPatterns.extract_trojan_urls(content)
for url in trojan_urls:
try:
proxy = TrojanParser.parse(url)
proxies.append(proxy)
except ValueError:
continue
ss_urls = ProxyPatterns.extract_ss_urls(content)
for url in ss_urls:
try:
proxy = SSParser.parse(url)
proxies.append(proxy)
except ValueError:
continue
return proxies
|