Hamza/maps-saas
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: sync scripts, scraper, and sql tools from server

Browse Source
This commit is contained in:
Hamza-Ayed
2026-08-16 13:19:10 +03:00
parent 32865e78df
commit 3136a2b7aa
14 changed files with 28844 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scraper/output.log
+8
View File
@@ -0,0 +1,8 @@
nohup: ignoring input
🏙️ [PROD] Starting full Damascus scan on Mac (Low-Resource)...
🚀 PASS: COMMERCIAL_PUBLIC (Low-Resource Mode)
📋 [DEBUG] Grid: 44x66 blocks. Starting...
🔎 [TRACE] Scanning Block 0,0 | 33.45000, 36.17000
🛡️ Bypassing Google Consent (ID: #L2AGLb)...
🛡️ Bypassing Google Consent (ID: #L2AGLb)...
scraper/scraper.py
+227
View File
@@ -0,0 +1,227 @@
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())
scraper/scraper_pro.py
+173
View File
@@ -0,0 +1,173 @@
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())