| import asyncio |
| import aiohttp |
| import random |
| import time |
| import datetime |
| from utils.logger import log |
|
|
| class FragmentScraper: |
| def __init__(self): |
| self.session = None |
| self.cache = {} |
| self._rate_limited_until = 0.0 |
| |
| self.consecutive_errors = 0 |
| self.circuit_broken = False |
| self.circuit_opened_at = 0.0 |
| self.in_flight = {} |
| |
| self.latest_trace = { |
| "username": "None", "url": "None", "status": 0, |
| "reason": "None", "decision": False, "time": "Never" |
| } |
| |
| self._lock = None |
| self._cleanup_task = None |
| |
| self.user_agents = [ |
| "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/124.0.0.0 Safari/537.36", |
| "Mozilla/5.0 (Macintosh; Intel Mac OS X 14_4_1) AppleWebKit/605.1.15 Version/17.4.1 Safari/605.1.15", |
| "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 Chrome/123.0.0.0 Safari/537.36", |
| "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:125.0) Gecko/20100101 Firefox/125.0", |
| "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 Chrome/122.0.0.0 Safari/537.36" |
| ] |
|
|
| async def get_lock(self): |
| if self._lock is None: |
| self._lock = asyncio.Lock() |
| return self._lock |
|
|
| async def init_session(self): |
| if not self.session: |
| timeout = aiohttp.ClientTimeout(total=6, connect=3) |
| connector = aiohttp.TCPConnector(limit=50, ssl=False) |
| self.session = aiohttp.ClientSession(timeout=timeout, connector=connector) |
| self._cleanup_task = asyncio.create_task(self._cleanup_cache_loop()) |
| log("🌐 Fragment Scraper session initialized.") |
|
|
| async def close_session(self): |
| if self.session: |
| await self.session.close() |
| self.session = None |
| if self._cleanup_task: |
| self._cleanup_task.cancel() |
|
|
| async def _cleanup_cache_loop(self): |
| while True: |
| try: |
| await asyncio.sleep(120) |
| now = time.time() |
| lock = await self.get_lock() |
| async with lock: |
| expired = [k for k, (_, ts) in self.cache.items() if now - ts > 60] |
| for k in expired: del self.cache[k] |
| stuck = [k for k, expire_ts in self.in_flight.items() if now > expire_ts] |
| for k in stuck: del self.in_flight[k] |
| except asyncio.CancelledError: |
| break |
| except Exception: |
| pass |
|
|
| async def is_snipe_window(self, username: str) -> bool: |
| now = time.time() |
| |
| if self.circuit_broken: |
| if now - self.circuit_opened_at > 300: |
| self.circuit_broken = False |
| log("🔄 Half-Open Circuit: Testing Fragment marketplace connectivity...") |
| else: |
| return False |
| |
| if now < self._rate_limited_until: return False |
|
|
| lock = await self.get_lock() |
| |
| async with lock: |
| if username in self.cache and now - self.cache[username][1] < 60: |
| return self.cache[username][0] |
| if username in self.in_flight and now < self.in_flight[username]: |
| return False |
| self.in_flight[username] = now + 15 |
|
|
| try: |
| url = f"https://fragment.com/username/{username}" |
| headers = {"User-Agent": random.choice(self.user_agents)} |
| |
| async with self.session.get(url, headers=headers) as resp: |
| if resp.status == 429: |
| self._rate_limited_until = now + 15 + random.uniform(0, 3) |
| async with lock: |
| self.latest_trace = { |
| "username": username, "url": url, "status": 429, |
| "reason": "Cloudflare Rate Limit", "decision": False, |
| "time": datetime.datetime.utcnow().strftime("%H:%M:%S") |
| } |
| return False |
| |
| if resp.status == 404: |
| self.consecutive_errors = 0 |
| is_snipe = True |
| async with lock: |
| self.cache[username] = (is_snipe, time.time()) |
| self.latest_trace = { |
| "username": username, "url": url, "status": 404, |
| "reason": "404 Not Found (Dropped)", "decision": True, |
| "time": datetime.datetime.utcnow().strftime("%H:%M:%S") |
| } |
| return is_snipe |
| |
| if resp.status >= 500 or resp.status == 403: |
| self.consecutive_errors += 1 |
| if self.consecutive_errors >= 50 and not self.circuit_broken: |
| self.circuit_broken = True |
| self.circuit_opened_at = time.time() |
| log("🚨 CRITICAL: Scraper Circuit Broken! >= 50 consecutive errors.") |
| |
| async with lock: |
| self.latest_trace = { |
| "username": username, "url": url, "status": resp.status, |
| "reason": f"Server Error ({resp.status})", "decision": False, |
| "time": datetime.datetime.utcnow().strftime("%H:%M:%S") |
| } |
| return False |
|
|
| self.consecutive_errors = 0 |
| html = await resp.text() |
| is_snipe = False |
| reason = "No tags found" |
| |
| |
| if 'property="og:url" content="https://fragment.com/?query=' in html or f'/?query={username}' in str(resp.url): |
| is_snipe = True |
| reason = "State 4: Redirected to Search (Dropped)" |
| elif 'cf-browser-verification' in html or 'Just a moment...' in html: |
| is_snipe = False |
| reason = "State 6: Cloudflare Challenge Block" |
| elif 'tm-section-header-status tm-status-avail' in html: |
| is_snipe = False |
| reason = "State 1: Active Web3 Auction / Available" |
| elif 'tm-section-header-status tm-status-taken' in html or 'Make an offer for @' in html: |
| is_snipe = False |
| reason = "State 3: Explicitly Taken" |
| elif 'tm-section-header-status tm-status-unavail' in html: |
| if 'Sold for' in html or 'tm-username-usable' in html: |
| is_snipe = False |
| reason = "State 2: Sold on Web3" |
| else: |
| is_snipe = True |
| reason = "State 4: Explicit Unavailable Profile" |
| |
| elif 'tm-status-unavail' in html and (f'data-username="@{username}"' in html or f'value="{username}"' in html): |
| if 'Sold for' in html: |
| is_snipe = False |
| reason = "State 2: Sold on Web3 (Row)" |
| else: |
| is_snipe = True |
| reason = "State 4: Limbo/Dropped (Search Row Block)" |
| else: |
| is_snipe = False |
| reason = "Unknown HTML State" |
|
|
| async with lock: |
| self.cache[username] = (is_snipe, time.time()) |
| self.latest_trace = { |
| "username": username, "url": url, "status": resp.status, |
| "reason": reason, "decision": is_snipe, |
| "time": datetime.datetime.utcnow().strftime("%H:%M:%S") |
| } |
| return is_snipe |
|
|
| except Exception as e: |
| self.consecutive_errors += 1 |
| async with lock: |
| self.latest_trace = { |
| "username": username, "url": url, "status": 0, |
| "reason": "HTTP Exception", "decision": False, |
| "time": datetime.datetime.utcnow().strftime("%H:%M:%S") |
| } |
| return False |
| |
| finally: |
| async with lock: |
| self.in_flight.pop(username, None) |
|
|
| async def get_diagnostic(self, username: str) -> dict: |
| url = f"https://fragment.com/username/{username}" |
| headers = {"User-Agent": random.choice(self.user_agents)} |
| |
| result = { |
| "username": username, |
| "requested_url": url, |
| "final_url": "None", |
| "http_status": 0, |
| "html_size": 0, |
| "og_title": "None", |
| "is_snipe": False, |
| "reason": "Request Failed", |
| "markers": [] |
| } |
| |
| try: |
| async with self.session.get(url, headers=headers) as resp: |
| result["http_status"] = resp.status |
| html = await resp.text() |
| result["html_size"] = len(html) |
| result["final_url"] = str(resp.url) |
| |
| if '<meta property="og:title" content="' in html: |
| try: |
| title = html.split('<meta property="og:title" content="')[1].split('">')[0] |
| result["og_title"] = title |
| except: pass |
| |
| markers_to_check = ['tm-section-header-status', 'tm-status-unavail', 'tm-status-taken', 'tm-status-avail', 'Make an offer', 'Minimum bid', 'Highest bid', 'Sold for', 'Not for sale'] |
| for marker in markers_to_check: |
| if marker in html: result["markers"].append(marker) |
|
|
| |
| if 'property="og:url" content="https://fragment.com/?query=' in html or f'/?query={username}' in str(resp.url): |
| result["is_snipe"] = True |
| result["reason"] = "State 4: Redirected to Search (Dropped)" |
| elif 'cf-browser-verification' in html or 'Just a moment...' in html: |
| result["is_snipe"] = False |
| result["reason"] = "State 6: Cloudflare Challenge Block" |
| elif 'tm-section-header-status tm-status-avail' in html: |
| result["is_snipe"] = False |
| result["reason"] = "State 1: Active Web3 Auction / Available" |
| elif 'tm-section-header-status tm-status-taken' in html or 'Make an offer for @' in html: |
| result["is_snipe"] = False |
| result["reason"] = "State 3: Explicitly Taken" |
| elif 'tm-section-header-status tm-status-unavail' in html: |
| if 'Sold for' in html or 'tm-username-usable' in html: |
| result["is_snipe"] = False |
| result["reason"] = "State 2: Sold on Web3" |
| else: |
| result["is_snipe"] = True |
| result["reason"] = "State 4: Explicit Unavailable Profile" |
| elif 'tm-status-unavail' in html and (f'data-username="@{username}"' in html or f'value="{username}"' in html): |
| if 'Sold for' in html: |
| result["is_snipe"] = False |
| result["reason"] = "State 2: Sold on Web3 (Row)" |
| else: |
| result["is_snipe"] = True |
| result["reason"] = "State 4: Limbo/Dropped (Search Row Block)" |
| else: |
| result["is_snipe"] = False |
| result["reason"] = "Unknown HTML State" |
| |
| if resp.status == 404: |
| result["is_snipe"] = True |
| result["reason"] = "404 Not Found (Dropped)" |
| elif resp.status == 429: |
| result["is_snipe"] = False |
| result["reason"] = "Cloudflare Rate Limit (429)" |
|
|
| return result |
| except Exception as e: |
| result["reason"] = f"Exception: {str(e)[:40]}" |
| return result |
|
|
| scraper = FragmentScraper() |
