原文地址：http://www.kingpika.top:5599/post/79
一苏研、準(zhǔn)備好數(shù)據(jù)并組織成以下格式：
點(diǎn)線面格式均可
二、代碼（需要安裝相應(yīng)的庫）
#coding:utf-8
import geopandas as gpd
import pandas as pd
from sqlalchemy import create_engine
from geoalchemy2 import Geometry,WKTElement
import numpy as np
import os
import re
import json
from osgeo import ogr
#數(shù)據(jù)寫入函數(shù)：
def write_gis(path,engine):
geoType=getGeoTypeFromDir(path)
if(geoType):
map_data = gpd.GeoDataFrame.from_file(path)
map_data['geometry'] = map_data['geometry'].apply(lambda x: WKTElement(x.wkt,3857))
# map_data.drop(['center','parent'], axis = 1, inplace=True)
map_data.to_sql(
name? = re.split('\\.',path)[0],
con?? = engine,
if_exists= 'replace',
dtype = {'geometry':Geometry(geometry_type =geoType,srid = 3857)}
)
return None
#創(chuàng)建批量任務(wù)
def to_do(file_path,username,password,dbname):
os.chdir(file_path)
link = "postgresql://{0}:{1}@localhost:3351/{2}".format(username,password,dbname)
print(file_path)
engine = create_engine(link,encoding = 'utf-8')
file_list = os.listdir(file_path)
print(file_list)
map(lambda x: write_gis(x,engine),file_list)
return None
#從文件夾中讀取shp腮郊，獲得其類型
def getGeoTypeFromDir(dirPath):![]()
"""從文件夾中讀取shp摹蘑，獲得其類型"""
# os.path.abspath(path)
if os.path.isdir(dirPath):
file_list = os.listdir(dirPath)
for file in file_list:
ext=os.path.splitext(file)[1]
if(ext=='.shp'):
driver = ogr.GetDriverByName('ESRI Shapefile')
file=os.path.abspath(file)#返回文件的絕對路徑，這里返回的是錯(cuò)的轧飞，缺少了上級文件夾
names=os.path.split(file)
file=os.path.join(names[0],dirPath,names[1])
dataSource = driver.Open(file,0)
layer = dataSource.GetLayer(0)
feat = layer.GetFeature(0)
geom = feat.GetGeometryRef()
geoCode=geom.GetGeometryType()
return deGeoTypeCode(geoCode)
return None
#解譯ogr的geometry code
def deGeoTypeCode(code):
"""解譯ogr的geometry code"""
if code==1:
return 'POINT'
elif code==2:
return 'LINESTRING'
elif code==3:
return 'POLYGON'
return None
#執(zhí)行任務(wù)計(jì)劃
if __name__ == '__main__':
file_path = '/Users/jinming/Desktop/multipleImport/3857shp'
username = 'postgres'
password = 'pwd'
dbname = 'gistest'
to_do(file_path,username,password,dbname)
print('DONE')
