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" # ⚡ THE GOLDILOCKS PARSER: Highly accurate, but 100% safe from false-positives 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" # 827KB Fallback Page Detection (Strict checking for the search row) 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 '')[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) # ⚡ THE GOLDILOCKS PARSER (Diagnostic Mirror) 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()