除此之外, 这篇博客: [python] 旅游中的 处理内容 部分也许能够解决你的问题, 你可以仔细阅读以下内容或者直接跳转源博客中阅读:
- 利用高德地图API进行旅游景点的坐标获取
- 利用高德地图驾车API对爬虫获取的路线进行路线设计
- 储存到mongodb
#! /usr/bin/env python
# -*- coding: utf-8 -*-
# __file__: lv_data
import requests
import json
import geojson
from geomet import wkt
import pymongo
import geojsonio
GD_API_KEY = '93a19a8f482c4d60d98c180425e4967f'
def do_gd_location(address, city):
"""
高德坐标数据获取
:param address:
:param city:
:return:
"""
url = "https://restapi.amap.com/v3/geocode/geo?key={}&address={}&city={}".format(GD_API_KEY, address.split("(")[0],
city)
jd = eval(requests.get(url=url).text)
if jd.get("status") =="1":
# 默认获取第一个
if len(jd.get('geocodes')) > 0:
location = jd.get('geocodes')[0].get('location')
return location.split(",")
else:
return None
def do_gonlve(gonlve, city):
"""
景点处理
:param gonlve:
:param city:
:return:
"""
result = []
for item in gonlve:
location = do_gd_location(item, city)
if location == None:
location = ["0", "0"]
point_path = "POINT(" + " ".join(location) + ")"
data = {
"addr": item,
"location": location,
"wkt": point_path,
}
result.append(data)
return result
pass
def gd_driver(start_point, end_point, waypoints, city):
"""
高德驾车路线导航
:param start_point:起点
:param end_point:终点
:param waypoints:途径点
:param city:城市
:return:
"""
url = "https://restapi.amap.com/v3/direction/driving"
origin = do_gd_location(start_point, city)
destination = do_gd_location(end_point, city)
ac = None
if origin == None or destination == None:
return None
else:
waypoints_data = []
for i in waypoints:
location = do_gd_location(i, city)
if location:
waypoints_data.append(",".join(location))
if len(waypoints_data) > 1:
ac = ";".join(waypoints_data)
elif len(waypoints_data) == 1:
ac = waypoints_data[0]
else:
return None
params = {'key': GD_API_KEY,
'origin': ",".join(origin),
'destination': ",".join(destination),
'waypoints': ac,
}
req = requests.get(url=url, params=params)
dic = eval(req.text)
if dic.get("status") == "1":
steps = dic.get("route").get('paths')[0].get('steps')
try:
if steps:
point_list = []
for i in steps:
polyline = i.get('polyline')
point_list.append(polyline)
a = ';'.join(point_list).replace(",", ' ').replace(";", ',')
return 'LINESTRING(' + a + ')'
except Exception as e:
print(e)
def rote_link_location(rote_link, city):
"""
路线规划
:param rote_link:
:param city:
:return:
"""
ls = [x.split("(")[0] for x in rote_link]
path = gd_driver(
start_point=ls[0],
end_point=ls[(len(ls) - 1)],
waypoints=ls[1:-1],
city=city
)
return {
"path": ls,
"wkt": path,
}
def do_rote_link(rote_link_list, city):
"""
路线处理
:param rote_link_list:
:param city:
:return:
"""
res = []
for items in rote_link_list:
rote_name = items.get('rote_name')
rote_url = items.get('rote_url')
di = {
"rote_name": rote_name,
"rote_url": rote_url,
"rote_link": list()
}
for item in items.get('rote_link'):
rote_link = item.get("rote_link")
rs = rote_link_location(rote_link, city)
di.get("rote_link").append(rs)
res.append(di)
return res
pass
def city_gonglve_data(load_dict):
"""
主要运行函数
:return:
"""
one_data = {
"city_id": None,
"city_name": None,
"url": None,
"route": None,
"gonglve": None,
"gonlve": None,
"rotes": None
}
city_name = load_dict.get('city_name')
# gonlve 坐标数据
gonlve = load_dict.get('ly').get('gonlve')
gonlve_data = do_gonlve(gonlve, city=city_name)
# 路线数据
rote_link_list = load_dict.get('ly').get('rote')
rote_link_list_data = do_rote_link(rote_link_list, city=city_name)
one_data['city_id'] = load_dict.get('city_id')
one_data['city_name'] = load_dict.get('city_name')
one_data['url'] = load_dict.get('url')
one_data['route'] = load_dict.get('route')
one_data['gonglve'] = load_dict.get('gonglve')
one_data['gonlve'] = gonlve_data
one_data['rotes'] = rote_link_list_data
# with open('shanghai-shuju-gs.json', 'w') as result_file:
# json.dump(one_data, result_file, indent=2, ensure_ascii=False)
# save_mg_gdshuju(one_data)
print("数据生成完成")
return one_data
pass
def get_mg_client():
client = pymongo.MongoClient(host='localhost', port=27017)
db = client.lvyou
return db
def save_mg_gdshuju(l):
"""
保存坐标信息
:param l:
:return:
"""
db = get_mg_client()
collection = db.gdshj
rs = collection.find(l)
if list(rs) == []:
collection.insert_one(l)
def save_mg_gdllx(l):
"""
保存高德路线
:param l:
:return:
"""
db = get_mg_client()
collection = db.gdlx
rs = collection.find(l)
if list(rs) == []:
collection.insert_one(l)
def gd_geojson_mg(load_dict):
"""
高德geojson生产
:return:
"""
gonlve = load_dict.get('gonlve')
rotes = load_dict.get('rotes')
feature_list = []
for i in gonlve:
geom = eval(json.dumps(wkt.loads(i.get("wkt"))))
atr = {
"addr": i.get('addr'),
"marker-color": "#e0b91d",
"marker-size": "medium",
"marker-symbol": "circle",
}
feature_list.append(
dict(type="Feature", geometry=geom, properties=atr)
)
for i in rotes:
item = i.get('rote_link')
rote_name = i.get("rote_name")
for it in item:
if it.get("wkt"):
geom = eval(json.dumps(wkt.loads(it.get("wkt"))))
atr = {
"rote_name": rote_name + str(list(item).index(it) + 1),
"rote_path": ",".join(it.get("path")),
}
feature_list.append(
dict(type="Feature", geometry=geom, properties=atr)
)
data = {"type": "FeatureCollection", "features": feature_list}
return data
def run_mg_ins_gd():
db = get_mg_client()
# 路由路线
con1 = db.lsa
# 查询所有旅游路线 ly !=null
r1 = con1.find({'ly': {"$ne": None}})
for r in r1:
dt1 = city_gonglve_data(r)
dt2 = gd_geojson_mg(dt1)
save_mg_gdshuju(dt1)
save_mg_gdllx(dt2)
if __name__ == '__main__':
# city_gonglve_data()
# main()
run_mg_ins_gd()
pass
