import asyncio import os import requests import json import re import math import urllib.parse 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 Server Edition) DAMASCUS_BOUNDS = { "min_lat": 33.4500, "max_lat": 33.5800, "min_lng": 36.1700, "max_lng": 36.4000 } PASSES = { "COMMERCIAL_PUBLIC": { "categories": ["سوق", "محل", "مطعم", "كافيه", "صيدلية", "عيادة", "جامع", "وزارة", "مدرسة"], "lat_step": 0.0030, "lng_step": 0.0035, "zoom": 18, "scrolls": 3 }, "RESIDENTIAL": { "categories": ["منزل", "بيت", "بناء سكني", "دخلة", "حارة"], "lat_step": 0.0015, "lng_step": 0.0018, "zoom": 20, "scrolls": 4 } } 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+)") 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')"] for selector in selectors: if await page.locator(selector).count() > 0: await page.click(selector) await asyncio.sleep(1) return True return False except: return False async def extract_places(self, page, center_lat, center_lng, pass_name) -> list[dict]: await self.bypass_consent(page) places = [] try: elements = await page.query_selector_all('a[href*="/maps/place/"]') 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 = urllib.parse.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.5: continue if any(key in name.lower() for key in ["google", "بحث", "إبلاغ"]): 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 except: pass 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=15) if res.status_code in [200, 201]: print(f" ✅ [SYNC SUCCESS] {len(places)} points added.", flush=True) else: print(f" ❌ [SYNC ERROR] {res.status_code}", flush=True) except Exception as e: print(f" ⚠️ [SYNC EXCEPTION] {str(e)[:50]}", flush=True) async def run_pass(self, pass_name: str, test_mode=False): cfg = PASSES[pass_name] print(f"\n🚀 PASS: {pass_name} (Nuclear Mode Active)", flush=True) async with async_playwright() as p: user_data_dir = os.path.join(os.getcwd(), "browser_state_pro") context = await p.chromium.launch_persistent_context( user_data_dir, headless=True, viewport={"width": 1280, "height": 720}, user_agent="Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36" ) # حقن الكوكيز فوراً await context.add_cookies([ {"name": "CONSENT", "value": "YES+cb.20230531-17-p0.en+FX+908", "domain": ".google.com", "path": "/"}, {"name": "SOCS", "value": "CAISHAgBEhJnd3NfMjAyMzA4MzAtMF9SQzIaAnByIAE", "domain": ".google.com", "path": "/"} ]) page = await context.new_page() await page.set_extra_http_headers({"Accept-Language": "ar-SY,ar;q=0.9,en-US;q=0.8"}) rows = math.ceil((self.bounds['max_lat'] - self.bounds['min_lat']) / cfg['lat_step']) cols = math.ceil((self.bounds['max_lng'] - self.bounds['min_lng']) / cfg['lng_step']) for r in range(rows): for c in range(cols): lat = self.bounds['min_lat'] + (r * cfg['lat_step']) lng = self.bounds['min_lng'] + (c * cfg['lng_step']) try: async with asyncio.timeout(45): # مهلة 45 ثانية لكل كتلة print(f"🔎 [BLOCK {r},{c}] {lat:.4f}, {lng:.4f} | ", end="", flush=True) cat = cfg['categories'][0] # نكتفي بصنف واحد حالياً للسرعة الكلية url = f"https://www.google.com/maps/search/{cat}/@{lat},{lng},{cfg['zoom']}z?hl=ar" await page.goto(url, wait_until="domcontentloaded", timeout=30000) await self.bypass_consent(page) for _ in range(cfg['scrolls']): await page.mouse.wheel(0, 1500) await asyncio.sleep(0.3) found = await self.extract_places(page, lat, lng, pass_name) if found: self.send_batch(found) print(f"✅ Found {len(found)}", flush=True) else: print(f"⚪ Empty", flush=True) except Exception as e: print(f"⚠️ Skip: {str(e)[:20]}", flush=True) continue await context.close() if __name__ == "__main__": scanner = DamascusProScanner(DAMASCUS_BOUNDS) async def main(): print("🏙️ [PROD] Damascus Scanner PRO - Server Edition", flush=True) await scanner.run_pass("COMMERCIAL_PUBLIC") await scanner.run_pass("RESIDENTIAL") asyncio.run(main())