maps-saas/scraper/scraper_pro.py

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())