import asyncio import os import requests import json import re import math import time from urllib.parse import unquote from playwright.async_api import async_playwright from dotenv import load_dotenv # Environment Setup load_dotenv() API_URL = os.getenv("API_URL", "http://localhost:3200/api") API_KEY = os.getenv("API_KEY", "intaleq_secret_2026") # PRODUCTION SCAN CALIBRATION (Damascus Pro - High Performance Mac Edition) DAMASCUS_BOUNDS = { "min_lat": 33.4500, "max_lat": 33.5800, "min_lng": 36.1700, "max_lng": 36.4000 } PASSES = { "RESIDENTIAL": { "categories": ["منزل", "بيت", "بناء سكني", "دخلة", "حارة"], "lat_step": 0.0015, "lng_step": 0.0018, "zoom": 20, "scrolls": 4 }, "COMMERCIAL_PUBLIC": { "categories": ["سوق", "محل", "مطعم", "كافيه", "صيدلية", "عيادة", "جامع", "وزارة", "مدرسة"], "lat_step": 0.0030, "lng_step": 0.0035, "zoom": 18, "scrolls": 3 } } class DamascusProScanner: def __init__(self, bounds: dict): self.bounds = bounds self.global_seen = set() self.coord_pattern = re.compile(r"!3d(-?\d+\.\d+)!4d(-?\d+\.\d+)|/@(-?\d+\.\d+),(-?\d+\.\d+)") async def route_interceptor(self, route): """السماح بكافة الموارد على الـ Mac لضمان دقة النتائج""" return await route.continue_() def calculate_distance(self, lat1, lon1, lat2, lon2): R = 6371 d_lat = math.radians(lat2 - lat1) d_lon = math.radians(lon2 - lon1) a = math.sin(d_lat / 2)**2 + math.cos(math.radians(lat1)) * math.cos(math.radians(lat2)) * math.sin(d_lon / 2)**2 return R * (2 * math.asin(math.sqrt(a))) async def bypass_consent(self, page): """تخطي صفحة موافقة جوجل بمختلف أشكالها (للعمل بسلاسة في السيرفر)""" try: selectors = [ "#L2AGLb", "button[aria-label*='Accept']", "button:has-text('أوافق')", "button:has-text('I agree')", "form[action*='consent'] button" ] for selector in selectors: if await page.locator(selector).count() > 0: print(f" 🛡️ Bypassing Google Consent (Selector: {selector})...", flush=True) await page.click(selector) await page.wait_for_timeout(2000) return True return False except: return False async def wait_for_results(self, page): """الانتظار الذكي لنتائج البحث أو صفحة الخريطة""" try: await page.wait_for_selector('a[href*="/maps/place/"]', timeout=15000) return True except: return False async def extract_places(self, page, center_lat, center_lng, pass_name) -> list[dict]: await self.bypass_consent(page) # التمرير لأسفل لتحميل المزيد من النتائج for _ in range(PASSES[pass_name].get('scrolls', 3)): await page.mouse.wheel(0, 4000) await page.wait_for_timeout(1000) # الانتظار حتى تظهر النتائج فعلياً await self.wait_for_results(page) places = [] elements = await page.query_selector_all('a[href*="/maps/place/"]') if len(elements) > 0: print(f" ✨ Found {len(elements)} potential results on page", flush=True) for el in elements: try: href = await el.get_attribute('href') if not href: continue name_match = re.search(r"/place/([^/]+)", href) if not name_match: continue name = unquote(name_match.group(1)).replace('+', ' ').strip() lat, lng = None, None matches = self.coord_pattern.findall(href) for m in matches: valid = [val for val in m if val] if len(valid) == 2: lat, lng = float(valid[0]), float(valid[1]) break if name and lat and lng: dist = self.calculate_distance(center_lat, center_lng, lat, lng) if dist > 3.0: continue # زيادة المدى قليلاً FORBIDDEN = ["google", "تكبير", "تصغير", "بحث", "مساحة", "إبلاغ", "عن"] if any(key in name.lower() for key in FORBIDDEN): continue dedup_key = f"{name.lower()}|{round(lat, 5)}|{round(lng, 5)}" if dedup_key not in self.global_seen: self.global_seen.add(dedup_key) places.append({ "name": name, "name_ar": name, "latitude": lat, "longitude": lng, "category": pass_name.title(), "city": "Damascus", "source": "scraper_admin" }) except: continue return places def send_batch(self, places: list[dict]): if not places: return try: res = requests.post( f"{API_URL}/geocoding/upsert-batch", headers={"X-API-Key": API_KEY}, json={"places": places}, timeout=20 ) if res.status_code in [200, 201]: print(f" ✅ [SYNC SUCCESS] {len(places)} points added. Response: {res.text[:50]}", flush=True) else: print(f" ❌ [SYNC ERROR] Status: {res.status_code} | Body: {res.text}", flush=True) except Exception as e: print(f" ⚠️ [SYNC EXCEPTION] {str(e)}", flush=True) async def run_pass(self, pass_name: str, test_mode=False): cfg = PASSES[pass_name] print(f"\n🚀 PASS: {pass_name} (Low-Resource Mode)", flush=True) async with async_playwright() as p: user_agent = "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36" # استخدام مجلد دائم لحفظ الـ Cookies والحالة لضمان تجاوز الحجب user_data_dir = os.path.join(os.getcwd(), "browser_state") context = await p.chromium.launch_persistent_context( user_data_dir, headless=not test_mode, viewport={"width": 1024, "height": 768}, locale="ar-SA", user_agent=user_agent ) # حدود البحث (تصغير المنطقة جداً في وضع الاختبار) bounds = { "min_lat": 33.5100, "max_lat": 33.5110, "min_lng": 36.2900, "max_lng": 36.2915 } if test_mode else self.bounds rows = math.ceil((bounds['max_lat'] - bounds['min_lat']) / cfg['lat_step']) cols = math.ceil((bounds['max_lng'] - bounds['min_lng']) / cfg['lng_step']) print(f"📋 [DEBUG] Grid: {rows}x{cols} blocks. Starting...", flush=True) for r in range(rows): for c in range(cols): lat = bounds['min_lat'] + (r * cfg['lat_step']) lng = bounds['min_lng'] + (c * cfg['lng_step']) print(f"🔎 [TRACE] Scanning Block {r},{c} | {lat:.5f}, {lng:.5f}", flush=True) page = await context.new_page() await page.route("**/*", self.route_interceptor) for cat in cfg['categories']: url = f"https://www.google.com/maps/search/{cat}/@{lat},{lng},{cfg['zoom']}z?hl=ar" try: await page.goto(url, wait_until="domcontentloaded", timeout=40000) # Trace & Handle Consent via CSS ID (More reliable) title = await page.title() if "قبل المتابعة" in title or "Before you continue" in title or "Consent" in title: print(f" 🛡️ Bypassing Google Consent (ID: #L2AGLb)...", flush=True) try: await page.click("button#L2AGLb") await page.wait_for_load_state("networkidle", timeout=10000) except: pass # الانتظار حتى تظهر أول نتيجة بحث أو انتهاء الوقت try: await page.wait_for_selector('a[href*="/maps/place/"]', timeout=8000) except: pass for _ in range(cfg['scrolls']): await page.mouse.wheel(0, 1000) await asyncio.sleep(0.5 if test_mode else 0.4) found = await self.extract_places(page, lat, lng, pass_name) if found: self.send_batch(found) print(f" ✅ [{pass_name}] Block {r},{c} | Found {len(found)} results", flush=True) except Exception as e: print(f" ⚠️ [ERROR] {str(e)[:40]}", flush=True) await page.close() await asyncio.sleep(1.0 if test_mode else 0.6) await browser.close() if __name__ == "__main__": scanner = DamascusProScanner(DAMASCUS_BOUNDS) async def main(): print("🏙️ [PROD] Starting full Damascus scan on Mac (Low-Resource)...", flush=True) # تشغيل المسح التجاري أولاً await scanner.run_pass("COMMERCIAL_PUBLIC", test_mode=False) # ثم المسح السكني الكثيف await scanner.run_pass("RESIDENTIAL", test_mode=False) asyncio.run(main())