Python對城市距離自動化爬取【必學小型項目】

 

 

本地創建數據庫，將 excel 數據存儲到 city 表中，再取 | 湖北省 | 的所有地級市和縣、縣級市、區數據作為樣表數據記錄在樣表中。利用 python 的 xlrd 包，定義 process_data 包來存放操作 excel 數據，生成 sql 語句的類，定義 op_postgresql 包來存放數據庫的操作對象，定義各種方法

 PS::另外很多人在學習Python的過程中，往往因為沒有好的教程或者沒人指導從而導致自己容易放棄，為此我建了個Python全棧開發交流.裙 ：一久武其而而流一思（數字的諧音）轉換下可以找到了，里面有最新Python教程項目可拿,不懂的問題有老司機解決哦，一起相互監督共同進步

 

城市距離爬取 - 任務計划

本地創建數據庫，將 excel 數據存儲到 city 表中，再取 | 湖北省 | 的所有地級市和縣、縣級市、區數據作為樣表數據記錄在樣表中。

 

本地創建數據庫，將 excel 數據存儲到 city 表中，再取 | 湖北省 | 的所有地級市和縣、縣級市、區數據作為樣表數據記錄在樣表中。准備工作創建好 public/config.py 擴展包，到時候，利用 python 的 xlrd 包，定義 process_data 包來存放操作 excel 數據，生成 sql 語句的類，
定義 op_postgresql 包來存放數據庫的操作對象，定義各種方法
創建 crwler 包，來存放爬蟲的操作對象 -> 發現對方網站調用的地圖 api -> 更改為調用德地圖 api 的包 - 存放操作對象
創建 log 文件夾，存放數據庫操作的日志
創建 data 文件夾，存放初始 excel 數據

 

數據庫基本構造：

樣本數據表格式：
表名：sample_table

name	column	data type	length	分布	fk	必填域	備注
地域名	address	text				TRUE	地域名
地域類型	ad_type	integer				TRUE	0 - 為地級市；1 - 為縣、縣級市、區。
經緯度	coordinates	text				TRUE	地域名的經緯度
···

樣本 1-1 地點 route 表的格式

表名:sample_route

name	column	data type	length	分布	fk	必填域	備注
出發點	origin	text					出發點
目的點	destination	text					目的點
距離	distance	integer					距離
路線	route	text					路線
···

 

創建配置信息接口

方便存儲我們需要的特定變量和配置信息。

public/config.py

#!/usr/bin/python # -*- coding: utf-8 -*- #__author__: stray_camel import os,sys #當前package所在目錄的上級目錄 src_path = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))

 

創建讀取 excel 數據的接口

利用 python 的 xlrd 包，定義 process_data 包來存放操作 excel 數據，生成 sql 語句的類

參考 github 源碼 readme 文檔
並沒有發現在 PyPI 上有 document，所以只能去 github 上找源碼了，xlrd 處理 excel 基礎 guide

import xlrd book = xlrd.open_workbook("myfile.xls") print("The number of worksheets is {0}".format(book.nsheets)) print("Worksheet name(s): {0}".format(book.sheet_names())) sh = book.sheet_by_index(0) print("{0} {1} {2}".format(sh.name, sh.nrows, sh.ncols)) print("Cell D30 is {0}".format(sh.cell_value(rowx=29, colx=3))) for rx in range(sh.nrows): print(sh.row(rx))

創建 process_data/excel2sql.py 擴展包，方便后面 import
獲取 excel 的數據構造 sql 語句，創建 city 表（湖北省）樣表

process_data/excel2sql.py

#!/usr/bin/python # -*- coding: utf-8 -*- #__author__: stray_camel import xlrd,sys,os,logging from public import config class Excel2Sql(object): def __init__( self, url:"str類型的文件路徑", sheet:"excel中的表單名"): self.f_name = url # 將excel中特定表單名數據存儲起來 self.sh_data = xlrd.open_workbook(self.f_name).sheet_by_name(sheet) self.rows = self.sh_data.nrows self.cols = self.sh_data.ncols

當我們生成這個 Excel2Sql 對象的時候，我們希望按照類似

excel_data = excel2sql.Excel2Sql("fiel_name","sheet_name")

的代碼形式來直接讀取 excel 文件並獲取某個表單的數據。所以在初始化對象的時候我們希望對其屬性進行賦值。

excel 表中，我們按照下面的形式進行存儲數據：

省 / 直轄市	地級市	縣、縣級市、區
北京市	北京市	東城區
...	...	...

 

 

 

之后我們希望通過調用這個類（接口）地時候能夠訪問其中一個函數，只獲取某個省 / 或者直轄市的所有數據，類似湖北省，我們指向獲取奇中 103 個縣、區。

在類 Excel2Sql 中定義方法：

def init_SampleViaProvince_name( self, Province_name:"省名" ) ->"insert的數據，列表形式[('地域名1','1','經緯度'),('地域名2','1','經緯度')]": geo_app = Geo_mapInterface(config.geo_key) all_data = [self.sh_data.row_values(i) for i in range(self.rows)] cities_data=[[["".join(i),1],["".join(i[1:len(i)]),1]][i[0]==i[1]] for i in all_data if i[0] == Province_name] for i in cities_data: i.append(geo_app.get_coordinatesViaaddress("".join(i[0]))) # cities_data=[[["".join(i),1,'test1'],["".join(i[1:len(i)]),1,'test2']][i[0]==i[1]] for i in all_data if i[0] == Province_name] return cities_data

之后我們可以測試類的構造是否正確，或進行調試：
在文件末端編寫：

if __name__ == "__main__": test = Excel2Sql(config.src_path+"\\data\\2019最新全國城市省市縣區行政級別對照表(194).xls","全國城市省市縣區域列表") print(test.init_SampleViaProvince_name("北京市"))

測試結果：

(env) PS F:\覽眾數據> & f:/覽眾數據/env/Scripts/python.exe f:/覽眾數據/城市距離爬取/process_data/excel2sql.py [['北京市東城區', 1, '116.416357,39.928353'], ['北京市西城區', 1, '116.365868,39.912289'], ['北京市崇文區', 1, '116.416357,39.928353'], ['北京市宣武區', 1, '116.365868,39.912289'], ['北京市朝陽區', 1, '116.601144,39.948574'], ['北京市豐台區', 1, '116.287149,39.858427'], ['北京市石景山區', 1, '116.222982,39.906611'], ['北京市海淀區', 1, '116.329519,39.972134'], ['北京市門頭溝區', 1, '116.102009,39.940646'], ['北京市房山區', 1, '116.143267,39.749144'], ['北京市通州區', 1, '116.656435,39.909946'], ['北京市順義區', 1, '116.654561,40.130347'], ['北京市昌 平區', 1, '116.231204,40.220660'], ['北京市大興區', 1, '116.341014,39.784747'], ['北京市平谷區', 1, '117.121383,40.140701'], ['北京市懷柔區', 1, '116.642349,40.315704'], ['北京市密雲縣', 1, '116.843177,40.376834'], ['北京 市延慶縣', 1, '115.974848,40.456951']]

 

創建 OP 數據庫 postgresql（其他數據庫也都一樣啦～）接口

定義 op_postgresql 包來存放數據庫的操作對象，定義各種方法

數據庫的 curd 真的是從大二寫到大四。
訪問 postgresql 數據庫一般用的包：psycopg2
訪問官網
在這個操作文檔網站中，使用的思路已經很清楚的寫出來了 http://initd.org/psycopg/docs/usage.html

 

 

希望大小少在網上走彎路（少看一些翻譯過來的文檔）。。。
http://initd.org/psycopg/

模式還是一樣，調用 postgresql 的驅動 / 接口，設置參數登陸，訪問數據庫。設置光標，注入 sql 數據，fetch 返回值。

• 這里需要注意的幾點是，默認防 xss 注入，寫代碼時一般設置參數訪問。
• 注意生成日志文件，打印日志

具體過程不贅述，直接上代碼

op_postgresql/opsql.py:

#!/usr/bin/python # -*- coding: utf-8 -*- #__author__: stray_camel ''' 定義對mysql數據庫基本操作的封裝 1.數據插入 2.表的清空 3.查詢表的所有數據 ''' import logging import psycopg2 from public import config class OperationDbInterface(object): #定義初始化連接數據庫 def __init__(self, host_db : '數據庫服務主機' = 'localhost', user_db: '數據庫用戶名' = 'postgres', passwd_db: '數據庫密碼' = '1026shenyang', name_db: '數據庫名稱' = 'linezone', port_db: '端口號，整型數字'=5432): try: self.conn=psycopg2.connect(database=name_db, user=user_db, password=passwd_db, host=host_db, port=port_db)#創建數據庫鏈接 except psycopg2.Error as e: print("創建數據庫連接失敗|postgresql Error %d: %s" % (e.args[0], e.args[1])) logging.basicConfig(stream=open(config.src_path + '/log/syserror.log', encoding="utf-8", mode="a"), level = logging.DEBUG,format='%(asctime)s %(filename)s[line:%(lineno)d] %(levelname)s %(message)s') logger = logging.getLogger(__name__) logger.exception(e) self.cur=self.conn.cursor() #定義在樣本表中插入數據操作 def insert_sample_data(self, condition : "insert語句" = "insert into sample_data(address,ad_type,coordinates) values (%s,%s,%s)", params : "insert數據，列表形式[('地域名1','1','經緯度'),('地域名2','1','經緯度')]" = [('地域名1','1','經緯度'),('地域名2','1','經緯度')] ) -> "字典形式的批量插入數據結果" : try: self.cur.executemany(condition,params) self.conn.commit() result={'code':'0000','message':'執行批量插入操作成功','data':len(params)} logging.basicConfig(stream=open(config.src_path + '/log/syserror.log', encoding="utf-8", mode="a"), level = logging.DEBUG, format='%(asctime)s %(filename)s[line:%(lineno)d] %(levelname)s %(message)s') logger = logging.getLogger(__name__) logger.info("在樣本表sample_data中插入數據{}條，操作：{}!".format(result['data'],result['message'])) except psycopg2.Error as e: self.conn.rollback() # 執行回滾操作 result={'code':'9999','message':'執行批量插入異常','data':[]} print ("數據庫錯誤|insert_data : %s" % (e.args[0])) logging.basicConfig(stream=open(config.src_path + '/log/syserror.log', encoding="utf-8", mode="a"), level = logging.DEBUG, format='%(asctime)s %(filename)s[line:%(lineno)d] %(levelname)s %(message)s') logger = logging.getLogger(__name__) logger.exception(e) return result 

繼續寫（代碼長了，怕顯示出錯）

    #定義在sample_route表中插入數據操作 def insert_sample_route(self, condition : "insert語句" , params : "insert語句的值" )->"字典形式的批量插入數據結果": try: self.cur.executemany(condition,params) self.conn.commit() result={'code':'0000','message':'執行批量插入操作成功','data':len(params)} logging.basicConfig(stream=open(config.src_path + '/log/syserror.log', encoding="utf-8", mode="a"), level = logging.DEBUG, format='%(asctime)s %(filename)s[line:%(lineno)d] %(levelname)s %(message)s') logger = logging.getLogger(__name__) logger.info("在樣本表sample_route中插入數據{}條，操作：{}!".format(result['data'],result['message'])) except psycopg2.Error as e: self.conn.rollback() # 執行回滾操作 result={'code':'9999','message':'執行批量插入異常','data':[]} print ("數據庫錯誤|insert_data : %s" % (e.args[0])) logging.basicConfig(stream=open(config.src_path + '/log/syserror.log', encoding="utf-8", mode="a"), level = logging.DEBUG, format='%(asctime)s %(filename)s[line:%(lineno)d] %(levelname)s %(message)s') logger = logging.getLogger(__name__) logger.exception(e) return result #定義對表的清空 def ini_table(self, tablename:"表名")->"清空表數據結果": try: rows_affect = self.cur.execute("select count(*) from {}".format(tablename)) test = self.cur.fetchone() # 獲取一條結果 self.cur.execute("truncate table {}".format(tablename)) self.conn.commit() result={'code':'0000','message':'執行清空表操作成功','data':test[0]} logging.basicConfig(stream=open(config.src_path + '/log/syserror.log', encoding="utf-8", mode="a"), level = logging.DEBUG, format='%(asctime)s %(filename)s[line:%(lineno)d] %(levelname)s %(message)s') logger = logging.getLogger(__name__) logger.info("清空{}表，操作數據{}條，操作：{}!".format(tablename,result['data'],result['message'])) except psycopg2.Error as e: self.conn.rollback() # 執行回滾操作 result={'code':'9999','message':'執行批量插入異常','data':[]} print ("數據庫錯誤|insert_data : %s" % (e.args[0])) logging.basicConfig(stream=open(config.src_path + '/log/syserror.log', encoding="utf-8", mode="a"), level = logging.DEBUG, format='%(asctime)s %(filename)s[line:%(lineno)d] %(levelname)s %(message)s') logger = logging.getLogger(__name__) logger.exception(e) return result #查詢表的所有數據 def select_all(self, tablename:"表名")->"返回list，存放查詢的結果": try: rows_affect = self.cur.execute("select * from {}".format(tablename)) test = self.cur.fetchall() # self.cur.execute("truncate table {}".format(tablename)) self.conn.commit() result={'code':'0000','message':'查詢表成功','data':test} logging.basicConfig(stream=open(config.src_path + '/log/syserror.log', encoding="utf-8", mode="a"), level = logging.DEBUG, format='%(asctime)s %(filename)s[line:%(lineno)d] %(levelname)s %(message)s') logger = logging.getLogger(__name__) logger.info("清空{}表，操作數據{}條，操作：{}!".format(tablename,result['data'],result['message'])) except psycopg2.Error as e: self.conn.rollback() # 執行回滾操作 result={'code':'9999','message':'執行批量插入異常','data':[]} print ("數據庫錯誤|insert_data : %s" % (e.args[0])) logging.basicConfig(stream=open(config.src_path + '/log/syserror.log', encoding="utf-8", mode="a"), level = logging.DEBUG, format='%(asctime)s %(filename)s[line:%(lineno)d] %(levelname)s %(message)s') logger = logging.getLogger(__name__) logger.exception(e) return result #數據庫關閉 def __del__(self): self.conn.close()

這里提出來想說一下的打印日志文件的操作，：
參考文件：
https://docs.python.org/zh-cn/3/library/lo...
https://docs.python.org/zh-cn/3/library/lo...
logging 作為 python 老牌庫，在 https://docs.python.org/zh-cn/3/library/in... 中一般都搜索的到，參數的說明不過多的贅述。
因為我的代碼都是用 utf-8 寫的所以在 basicConfig 配置時，加入了 utf-8 的信息。

result={'code':'9999','message':'執行批量插入異常','data':[]} print ("數據庫錯誤|insert_data : %s" % (e.args[0])) logging.basicConfig(stream=open(config.src_path + '/log/syserror.log', encoding="utf-8", mode="a"), level = logging.DEBUG, format='%(asctime)s %(filename)s[line:%(lineno)d] %(levelname)s %(message)s') logger = logging.getLogger(__name__) logger.exception(e)

 

測試爬取 https://licheng.supfree.net/ 網站

測試 https://licheng.supfree.net/ 網站是否可以傳參進行 post，獲取 request 后的兩地的地理距離

• 測試網站是否有反爬蟲機制，結果無。

通過測試 request，設置測試地點洪山區和江夏區，網站顯示距離為 16.5 公里
解析 html 發現
測試結果：網站的數據是通過 js 文件獲取傳參的。

var map = new BMap.Map("container"); map.centerAndZoom(new BMap.Point(116.404, 39.915), 14); var oGl = document.getElementById("div_gongli"); var ofname = document.getElementById("tbxArea"); var otname = document.getElementById("tbxAreaTo"); if (ofname.value != "" && otname.value != "") { var output = "全程："; var searchComplete = function(results) { if (transit.getStatus() != BMAP_STATUS_SUCCESS) { return; } var plan = results.getPlan(0); output += plan.getDistance(true); //獲取距離 output += "/"; output += plan.getDuration(true); //獲取時間 } var transit = new BMap.DrivingRoute(map, { renderOptions: { map: map, panel: "results", autoViewport: true }, onSearchComplete: searchComplete, onPolylinesSet: function() { oGl.innerText = output; } }); transit.search(ofname.value, otname.value); } ... 

我們查看網站加載的 js 文件，發現獲取 Bmap 這個對象原來是來自於

https://api.map.baidu.com/?qt=nav&c=131&sn=2%24%24%24%24%24%24%E6%B4%AA%E5%B1%B1%E5%8C%BA%24%240%24%24%24%24&en=2%24%24%24%24%24%24%E6%B1%9F%E5%A4%8F%E5%8C%BA%24%240%24%24%24%24&sy=0&ie=utf-8&oue=1&fromproduct=jsapi&res=api&callback=BMap._rd._cbk35162&ak=zS6eHWhoEwXMUrQKkaaTlvY65XsVykFf

很明顯，這個網站也是調用的百度的 api。
我們查看 js 文件傳遞的部分參數：

content: {dis: 16538,…} dis: 16538 kps: [{a: 7, dr: "", dw: 0, ett: 17, ic: "", iw: 0, pt: ".=zl83LBgOCJVA;", rt: 1, tt: 1},…] rss: [{d: 0, g: "", n: "", rr: 0, t: 0, tr: 0},…] taxi: {detail: [{desc: "白天(05:00-23:00)", kmPrice: "2.3", startPrice: "14.0", totalPrice: "47"},…],…} time: 1516 toll: 0 ...

核實 content 里的 dis 和 time 是否就是網站顯示的距離和時間
當我們更換測試地點后，顯示的距離和 https://api.map.baidu.com 中 content 的內容一樣
time：1516%60=25.26666666666667‬，和顯示的 26 分鍾也是核對的。

測試結果：網站沒有反爬蟲機制，但是調用的是百度地圖 pai 獲取數。

• 網站儲存地址的數據是按照編碼來的，對應的下級城市為小數
  比如熱門城市：

  hot_city: ["北京市|131", "上海市|289", "廣州市|257", "深圳市|340", "成都市|75", "天津市|332", "南京市|315", "杭州市|179", "武漢市|218",…] 0: "北京市|131" 1: "上海市|289" 2: "廣州市|257" 3: "深圳市|340" 4: "成都市|75" 5: "天津市|332" 6: "南京市|315" 7: "杭州市|179" 8: "武漢市|218" 9: "重慶市|132"

  當測試區級地點：（洪山區、江夏區）

  map.centerAndZoom(new BMap.Point(116.404, 39.915), 14);

• 如果不行能否調用高德地圖 api？

 

創建接口 - 調用高德地圖 api

在高德的管理平台注冊個人開發：https://lbs.amap.com/dev/key/app

 

 

申請個人的 key。每日調用量有上線，所以只能一點點的做。
我們將申請到的 key 寫入配置信息文件中：
public/config.py

#!/usr/bin/python # -*- coding: utf-8 -*- #__author__: stray_camel import os,sys #當前package所在目錄的上級目錄 src_path = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) geo_key = '3e2235273ddtestdef4' #key我已經打馬賽克了，請自己去申請

完成功能：
通過地域名查詢經緯度；
對出發 / 目的地點 - 路程 - 路線，數據進行查詢，並插入到數據庫中，現已實現。但對於數據量較多的情況，數據庫的操作較慢。

首先前往高德地圖注冊個人用戶，獲取一個 key，之后我們可以通過構造 url，通過 request 來獲取數據。

通過 address 獲取經緯度：

def get_coordinatesViaaddress(self, address:"地點名" ) -> "返回str類型的經緯度": url='https://restapi.amap.com/v3/geocode/geo?address='+address+'&output=json&key='+self.key #將一些符號進行URL編碼 newUrl = parse.quote(url, safe="/:=&?#+!$,;'@()* ") coor = json.loads(urllib.request.urlopen(newUrl).read())['geocodes'][0]['location'] logging.basicConfig(stream=open(config.src_path + '/log/syserror.log', encoding="utf-8", mode="a"), level = logging.DEBUG, format='%(asctime)s %(filename)s[line:%(lineno)d] %(levelname)s %(message)s') logger = logging.getLogger(__name__) logger.info("查詢{}的經緯度：{}！".format(address,coor)) # print() return coor

通過城市 list 獲取兩點之間距離和出行方式：

def get_disViaCoordinates(self, addList:"一個列表存放地址數據" ) -> "{'origin':[],'destination':[],'distance':[],'route':[]}": dict_route = {'origin':[],'destination':[],'distance':[],'route':[]} for m in range(len(addList)): for n in range(m,len(addList)): if m!=n: print('get_tetst',m,n) #從addList中得到地址的名稱，經緯度 origin = addList[m][2] destination = addList[n][2] url2='https://restapi.amap.com/v3/direction/driving?origin='+origin+'&destination='+destination+'&extensions=all&output=json&key=3e2235273dd2c0ca2421071fbb96def4' #編碼 newUrl2 = parse.quote(url2, safe="/:=&?#+!$,;'@()*[]") #發送請求 response2 = urllib.request.urlopen(newUrl2) #接收數據 data2 = response2.read() #解析json文件 jsonData2 = json.loads(data2) #輸出該json中所有road的值 # print(jsonData2) road=jsonpath.jsonpath(jsonData2,'$..road') #從json文件中提取距離 distance = jsonData2['route']['paths'][0]['distance'] #字典dict_route中追加數據 dict_route.setdefault("origin",[]).append(addList[m][0]) dict_route.setdefault("destination",[]).append(addList[n][0]) dict_route.setdefault("distance",[]).append(distance) dict_route.setdefault("route",[]).append(road) return dict_route

數據庫樣品：
sample_table

數據庫的內容我就用 json 表示了哈：

[ { "address": "湖北省武漢市江岸區", "ad_type": 1, "coordinates": "114.278760,30.592688" }, { "address": "湖北省武漢市江漢區", "ad_type": 1, "coordinates": "114.270871,30.601430" }, { "address": "湖北省武漢市喬口區", "ad_type": 1, "coordinates": "114.214920,30.582202" }, ...共103條地點數據

sample_route，以 sample_table 前三個數據為例做出查詢，和返回。

[ { "origin": "湖北省武漢市江岸區", "destination": "湖北省武漢市江漢區", "route": "['台北一路', '新華路']", "distance": "1520" }, { "origin": "湖北省武漢市江岸區", "destination": "湖北省武漢市喬口區", "route": "['台北一路', '台北路', '解放大道', '解放大道', '解放大道', '解放大道', '解放大道', '解放大道', '解放大道', '解放大道', '二環線輔路', '沿河大道']", "distance": "9197" }, { "origin": "湖北省武漢市江漢區", "destination": "湖北省武漢市喬口區", "route": "['新華路', '建設大道', '建設大道', '建設大道', '建設大道', '沿河大道']", "distance": "7428" } ]

---

BUG:
問題：在 process_data/excel2sql.py，調用格比 public/config.py 接口
問題：當我們訪問隔壁文件夾的接口時，如果發現調用不了，可以在當前文件的頭部加入：

import sys,os absPath = os.path.abspath(__file__) #返回代碼段所在的位置，肯定是在某個.py文件中 temPath = os.path.dirname(absPath) #往上返回一級目錄，得到文件所在的路徑 temPath = os.path.dirname(temPath) #在往上返回一級，得到文件夾所在的路徑 sys.path.append(temPath)

將當前文件夾所在的路徑加入到 python 系統路徑中

以上就是本次分享；另外很多人在學習Python的過程中，往往因為沒有好的教程或者沒人指導從而導致自己容易放棄，為此我建了個Python全棧開發交流.裙 ：一久武其而而流一思（數字的諧音）轉換下可以找到了，里面有最新Python教程項目可拿,不懂的問題有老司機解決哦，一起相互監督共同進步

本文的文字及圖片來源於網絡加上自己的想法,僅供學習、交流使用,不具有任何商業用途,版權歸原作者所有,如有問題請及時聯系我們以作處理。

城市距離爬取 - 任務計划