228 lines
10 KiB
Python
228 lines
10 KiB
Python
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())
|