268 lines
10 KiB
Python
268 lines
10 KiB
Python
|
"""
|
|||
|
|
高德地图高精度坐标获取工具类
|
|||
|
|
"""
|
|||
|
|
import json
|
|||
|
|
import os
|
|||
|
|
import random
|
|||
|
|
import re
|
|||
|
|
import time
|
|||
|
|
from selenium import webdriver
|
|||
|
|
from selenium.webdriver.chrome.options import Options
|
|||
|
|
from selenium.webdriver.chrome.service import Service
|
|||
|
|
from selenium.webdriver.common.by import By
|
|||
|
|
from selenium.webdriver.common.keys import Keys
|
|||
|
|
from selenium.webdriver.support import expected_conditions as EC
|
|||
|
|
from selenium.webdriver.support.ui import WebDriverWait
|
|||
|
|
|
|||
|
|
class PaChongGaoDeKit:
|
|||
|
|
def __init__(self, headless=True, cookie_file='amap_cookies.json'):
|
|||
|
|
self.headless = headless
|
|||
|
|
self.cookie_file = cookie_file
|
|||
|
|
self.driver = None
|
|||
|
|
|
|||
|
|
# Ensure cookie file path is absolute if relative
|
|||
|
|
if not os.path.isabs(self.cookie_file):
|
|||
|
|
# Assuming relative to this file or current working directory
|
|||
|
|
# For now, let's assume it's relative to the project root or current execution path
|
|||
|
|
# Better to use absolute path if possible.
|
|||
|
|
# If the file is in the same directory as this script:
|
|||
|
|
current_dir = os.path.dirname(os.path.abspath(__file__))
|
|||
|
|
potential_path = os.path.join(current_dir, self.cookie_file)
|
|||
|
|
if os.path.exists(potential_path):
|
|||
|
|
self.cookie_file = potential_path
|
|||
|
|
else:
|
|||
|
|
# Fallback to checking root/Config or similar if needed, or just leave as is
|
|||
|
|
pass
|
|||
|
|
|
|||
|
|
def _init_driver(self):
|
|||
|
|
if self.driver:
|
|||
|
|
return
|
|||
|
|
|
|||
|
|
print(f"正在启动 Chrome 浏览器 (无头模式={self.headless})...")
|
|||
|
|
chrome_options = Options()
|
|||
|
|
if self.headless:
|
|||
|
|
chrome_options.add_argument('--headless')
|
|||
|
|
|
|||
|
|
# 解决驱动与浏览器不匹配的常见配置
|
|||
|
|
chrome_options.add_argument('--no-sandbox')
|
|||
|
|
chrome_options.add_argument('--disable-dev-shm-usage')
|
|||
|
|
chrome_options.add_argument('--window-size=1920,1080')
|
|||
|
|
chrome_options.add_argument('--lang=zh-CN')
|
|||
|
|
chrome_options.add_argument("--disable-blink-features=AutomationControlled")
|
|||
|
|
chrome_options.add_experimental_option('excludeSwitches', ['enable-automation'])
|
|||
|
|
chrome_options.add_experimental_option('useAutomationExtension', False)
|
|||
|
|
|
|||
|
|
service = Service()
|
|||
|
|
try:
|
|||
|
|
from webdriver_manager.chrome import ChromeDriverManager
|
|||
|
|
from webdriver_manager.core.os_manager import ChromeType
|
|||
|
|
# 使用 ChromeDriverManager 自动下载匹配版本的 ChromeDriver
|
|||
|
|
driver_path = ChromeDriverManager().install()
|
|||
|
|
service = Service(driver_path)
|
|||
|
|
print(f"已成功加载 ChromeDriver: {driver_path}")
|
|||
|
|
except ImportError:
|
|||
|
|
print("提示：未安装 webdriver_manager 库，建议安装以获得更好的驱动管理体验：pip install webdriver-manager")
|
|||
|
|
except Exception as e:
|
|||
|
|
print(f"提示：自动管理驱动失败，尝试使用系统默认驱动。错误原因: {e}")
|
|||
|
|
|
|||
|
|
try:
|
|||
|
|
self.driver = webdriver.Chrome(service=service, options=chrome_options)
|
|||
|
|
|
|||
|
|
if not self.headless:
|
|||
|
|
self.driver.maximize_window()
|
|||
|
|
|
|||
|
|
self.driver.execute_cdp_cmd("Page.addScriptToEvaluateOnNewDocument", {
|
|||
|
|
"source": "Object.defineProperty(navigator, 'webdriver', {get: () => undefined})"
|
|||
|
|
})
|
|||
|
|
except Exception as e:
|
|||
|
|
print(f"致命错误：无法启动 Chrome 浏览器。请确保已安装 Chrome 且版本与驱动匹配。详情: {e}")
|
|||
|
|
raise e
|
|||
|
|
|
|||
|
|
def _load_cookies(self):
|
|||
|
|
if not self.driver:
|
|||
|
|
return False
|
|||
|
|
|
|||
|
|
if not os.path.exists(self.cookie_file):
|
|||
|
|
print(f"Cookie file not found: {self.cookie_file}")
|
|||
|
|
return False
|
|||
|
|
|
|||
|
|
try:
|
|||
|
|
# Visit domain first
|
|||
|
|
self.driver.get('https://www.amap.com')
|
|||
|
|
time.sleep(1)
|
|||
|
|
|
|||
|
|
with open(self.cookie_file, 'r', encoding='utf-8') as f:
|
|||
|
|
cookies = json.load(f)
|
|||
|
|
|
|||
|
|
for cookie in cookies:
|
|||
|
|
if 'expiry' in cookie:
|
|||
|
|
cookie['expiry'] = int(cookie['expiry'])
|
|||
|
|
if 'sameSite' in cookie and cookie['sameSite'] not in ['Strict', 'Lax', 'None']:
|
|||
|
|
del cookie['sameSite']
|
|||
|
|
try:
|
|||
|
|
self.driver.add_cookie(cookie)
|
|||
|
|
except:
|
|||
|
|
pass
|
|||
|
|
return True
|
|||
|
|
except Exception as e:
|
|||
|
|
print(f"Failed to load cookies: {e}")
|
|||
|
|
return False
|
|||
|
|
|
|||
|
|
def ensure_logged_in(self):
|
|||
|
|
self._init_driver()
|
|||
|
|
if not self._load_cookies():
|
|||
|
|
print("Warning: Failed to load cookies, coordinate query might fail.")
|
|||
|
|
|
|||
|
|
# Navigate to picker tool
|
|||
|
|
self.driver.get('https://lbs.amap.com/tools/picker')
|
|||
|
|
time.sleep(2)
|
|||
|
|
|
|||
|
|
def parse_coordinates_info(self, text):
|
|||
|
|
if not text:
|
|||
|
|
return None
|
|||
|
|
try:
|
|||
|
|
clean_text = re.sub(r'[\u3000\t\n\r ]+', ',', text)
|
|||
|
|
clean_text = re.sub(r'[,，]+', ',', clean_text).strip(',')
|
|||
|
|
pattern = r'([7-9]?[0-9]\.[0-9]{1,}|1[0-3][0-9]\.[0-9]{1,})[,，]\s*([1-8]?[0-9]\.[0-9]{1,}|90\.[0-9]{1,})'
|
|||
|
|
matches = re.finditer(pattern, clean_text)
|
|||
|
|
|
|||
|
|
valid_coordinates = []
|
|||
|
|
for match in matches:
|
|||
|
|
lng = match.group(1).strip()
|
|||
|
|
lat = match.group(2).strip()
|
|||
|
|
try:
|
|||
|
|
lng_float = float(lng)
|
|||
|
|
lat_float = float(lat)
|
|||
|
|
if 70 <= lng_float <= 140 and 0 <= lat_float <= 60:
|
|||
|
|
valid_coordinates.append((lng_float, lat_float))
|
|||
|
|
except ValueError:
|
|||
|
|
continue
|
|||
|
|
|
|||
|
|
if valid_coordinates:
|
|||
|
|
# Return the first valid one
|
|||
|
|
return valid_coordinates[0]
|
|||
|
|
return None
|
|||
|
|
except Exception:
|
|||
|
|
return None
|
|||
|
|
|
|||
|
|
def get_coordinate(self, address: str):
|
|||
|
|
"""
|
|||
|
|
Query coordinate for a single address.
|
|||
|
|
Returns: (lng, lat) tuple or (None, None)
|
|||
|
|
"""
|
|||
|
|
if not address:
|
|||
|
|
return None, None
|
|||
|
|
|
|||
|
|
if not self.driver:
|
|||
|
|
self.ensure_logged_in()
|
|||
|
|
|
|||
|
|
try:
|
|||
|
|
# Search input
|
|||
|
|
search_input = WebDriverWait(self.driver, 10).until(
|
|||
|
|
EC.presence_of_element_located((By.ID, 'txtSearch'))
|
|||
|
|
)
|
|||
|
|
search_input.clear()
|
|||
|
|
search_input.send_keys(address)
|
|||
|
|
|
|||
|
|
# Click search
|
|||
|
|
try:
|
|||
|
|
search_button = WebDriverWait(self.driver, 5).until(
|
|||
|
|
EC.presence_of_element_located((By.CLASS_NAME, 'btn-search'))
|
|||
|
|
)
|
|||
|
|
search_button.click()
|
|||
|
|
except:
|
|||
|
|
search_input.send_keys(Keys.ENTER)
|
|||
|
|
|
|||
|
|
time.sleep(2) # Wait for result
|
|||
|
|
|
|||
|
|
# Get coordinate
|
|||
|
|
coordinate_text = None
|
|||
|
|
try:
|
|||
|
|
coord_elem = WebDriverWait(self.driver, 10).until(
|
|||
|
|
EC.presence_of_element_located((By.ID, 'txtCoordinate'))
|
|||
|
|
)
|
|||
|
|
val = coord_elem.get_attribute('value')
|
|||
|
|
if val and val.strip():
|
|||
|
|
coordinate_text = val.strip()
|
|||
|
|
else:
|
|||
|
|
coordinate_text = coord_elem.text.strip()
|
|||
|
|
except:
|
|||
|
|
pass
|
|||
|
|
|
|||
|
|
if coordinate_text:
|
|||
|
|
result = self.parse_coordinates_info(coordinate_text)
|
|||
|
|
if result:
|
|||
|
|
return result
|
|||
|
|
|
|||
|
|
return None, None
|
|||
|
|
|
|||
|
|
except Exception as e:
|
|||
|
|
print(f"Error querying address '{address}': {e}")
|
|||
|
|
# Try to recover session if needed, or just return None
|
|||
|
|
return None, None
|
|||
|
|
|
|||
|
|
def close(self):
|
|||
|
|
if self.driver:
|
|||
|
|
try:
|
|||
|
|
self.driver.quit()
|
|||
|
|
except:
|
|||
|
|
pass
|
|||
|
|
self.driver = None
|
|||
|
|
|
|||
|
|
def __enter__(self):
|
|||
|
|
return self
|
|||
|
|
|
|||
|
|
def __exit__(self, exc_type, exc_val, exc_tb):
|
|||
|
|
self.close()
|
|||
|
|
|
|||
|
|
def login_and_save_cookies(self):
|
|||
|
|
"""
|
|||
|
|
打开浏览器进行手动登录，并将 Cookie 保存到文件。
|
|||
|
|
该方法将阻塞，直到用户确认登录完成。
|
|||
|
|
"""
|
|||
|
|
# 强制切换为非无头模式以便手动登录
|
|||
|
|
if self.headless:
|
|||
|
|
print("切换至可视化模式进行登录...")
|
|||
|
|
self.close()
|
|||
|
|
self.headless = False
|
|||
|
|
|
|||
|
|
self._init_driver()
|
|||
|
|
|
|||
|
|
print("正在跳转至高德地图登录页面...")
|
|||
|
|
# 跳转至通常需要登录的页面
|
|||
|
|
self.driver.get('https://lbs.amap.com/tools/picker')
|
|||
|
|
|
|||
|
|
print("\n" + "="*60)
|
|||
|
|
print("操作指引：")
|
|||
|
|
print("1. 浏览器窗口已打开，请在窗口中手动完成登录。")
|
|||
|
|
print("2. 您可以选择扫码登录或账号密码登录。")
|
|||
|
|
print("3. 登录成功并跳转到控制台页面后，请回到此窗口。")
|
|||
|
|
print("4. 在下方按 [回车键(Enter)]，程序将自动抓取并保存 Cookie。")
|
|||
|
|
print("="*60 + "\n")
|
|||
|
|
|
|||
|
|
input("等待登录完成，确认后请按 [Enter] 继续...")
|
|||
|
|
|
|||
|
|
# 保存 Cookie
|
|||
|
|
try:
|
|||
|
|
cookies = self.driver.get_cookies()
|
|||
|
|
with open(self.cookie_file, 'w', encoding='utf-8') as f:
|
|||
|
|
json.dump(cookies, f, indent=4)
|
|||
|
|
print(f"成功：Cookie 已保存至: {self.cookie_file}")
|
|||
|
|
return True
|
|||
|
|
except Exception as e:
|
|||
|
|
print(f"错误：保存 Cookie 失败: {e}")
|
|||
|
|
return False
|
|||
|
|
|
|||
|
|
if __name__ == "__main__":
|
|||
|
|
# Test code
|
|||
|
|
# Ensure you have 'amap_cookies.json' in the same directory or provide full path
|
|||
|
|
# If no cookies, it might fail to load the picker page correctly if login is required.
|
|||
|
|
|
|||
|
|
kit = PaChongGaoDeKit(headless=False)
|
|||
|
|
try:
|
|||
|
|
addr = "北京市天安门"
|
|||
|
|
lng, lat = kit.get_coordinate(addr)
|
|||
|
|
print(f"{addr}: {lng}, {lat}")
|
|||
|
|
finally:
|
|||
|
|
kit.close()
|