# -*- coding: utf-8 -*- import psycopg2 as psypg import time as time import sys as sys import psycopg2.extras as psyex import datetime as dtime import json import glob as glob import pandas as pd import os as os import argparse as agp def connectdb(): print(os.environ) connDb=psypg.connect(dbname=os.environ['PG_DATABASE'], user=os.environ['PG_USERNAME'],password=os.environ['PG_PASSWORD'],host=os.environ['PG_HOST'],port=os.environ['PG_PORT']) return connDb def disconnectdb(connexion): connexion.close() def createTempTable(curInsert,nameTable): dropTempTable(curInsert,nameTable) if nameTable=="temp_data_pcv": sql="CREATE TABLE "+nameTable+ \ """(identifiant_cvi CHARACTER VARYING(15), commune_id_cvi CHARACTER VARYING(20), section_cvi CHARACTER VARYING(10), prefix_cvi INTEGER, numero_cvi INTEGER, dsf_cvi INTEGER, owner_name CHARACTER VARYING(256), lieudit_id CHARACTER VARYING(10), contenance_cvi NUMERIC, commune_id_cvi_insee CHARACTER VARYING(20),identifiant_cadastre CHARACTER VARYING(14), commune_id_cadastre CHARACTER VARYING(10), prefix_cadastre CHARACTER VARYING(10), section_cadastre CHARACTER VARYING(10), numero_cadastre INTEGER, contenance_cadastre NUMERIC, date_created TIMESTAMP, date_updated TIMESTAMP, geom_4326 CHARACTER VARYING, commune_id INTEGER);""" elif nameTable=="temp_data_spcv": sql="CREATE table "+nameTable+ \ """ (identifiant_pcv_cvi CHARACTER VARYING(15), num_spcv INTEGER, exploit_id INTEGER, cepage CHARACTER VARYING(8), portegreffe CHARACTER VARYING(8), bailleur_nature_id INTEGER, owner_farmer_bailleur_id INTEGER, plant_year INTEGER, superficie NUMERIC, ecart_rang NUMERIC, ecart_pied NUMERIC,id INTEGER, code CHARACTER VARYING(25), EVV CHARACTER VARYING(25), name CHARACTER VARYING(255), email CHARACTER VARYING(255), parcel_id INTEGER, farmer_id INTEGER);""" elif nameTable=="temp_data_exploit": sql="CREATE UNLOGGED TABLE "+nameTable+ \ """ (id INTEGER, code CHARACTER VARYING(25), EVV CHARACTER VARYING(25), name CHARACTER VARYING(255), email CHARACTER VARYING(255), civilite CHARACTER VARYING(10), address_name CHARACTER VARYING(255), address_name2 CHARACTER VARYING(255), address CHARACTER VARYING(255), address_suppl CHARACTER VARYING(255), code_insee INTEGER, city CHARACTER VARYING(255));""" curInsert.execute(sql) def dropTempTable(curInsert,nameTable): sql="DROP TABLE IF EXISTS "+nameTable+";" curInsert.execute(sql) def insertExploit(dataExploit,curInsert): time_start=time.time() createTempTable(curInsert,"temp_data_exploit") sql = """INSERT INTO temp_data_exploit VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s);""" psyex.execute_batch(curInsert,sql,dataExploit.values) print("process of insert into temp temp_data_exploit in " + str(time.time()-time_start) +"s") #TODO update information time_start=time.time() sql = """INSERT INTO users (id,username,num_evv,firstname,email,settings,role,address,address_suppl,code_insee,city,address_name, address_name2, civilite) SELECT nextval('users_id_seq'), LPAD(code,5,'0'), EVV, name,email,'[]','ROLE_USER',address,address_suppl,code_insee,city,address_name, address_name2, civilite from temp_data_exploit ON CONFLICT (username) DO UPDATE SET firstname = excluded.firstname, address = excluded.address , address_suppl = excluded.address_suppl, code_insee=excluded.code_insee, city=excluded.city, address_name=excluded.address_name, address_name2=excluded.address_name2, civilite=excluded.civilite;""" curInsert.execute(sql) dropTempTable(curInsert,"temp_data_exploit") print("process of exploit in " + str(time.time()-time_start) +"s") def insertParcel(dataPCV,curInsert): time_start=time.time() createTempTable(curInsert,"temp_data_pcv") # UPDATE Active to false for existing parcel sql = "UPDATE parcel SET active = false"; curInsert.execute(sql) sql = """INSERT INTO temp_data_pcv (identifiant_cvi , commune_id_cvi , section_cvi , prefix_cvi , numero_cvi , dsf_cvi , owner_name , lieudit_id , contenance_cvi , commune_id_cvi_insee ,identifiant_cadastre , commune_id_cadastre , prefix_cadastre , section_cadastre , numero_cadastre , contenance_cadastre , date_created , date_updated , geom_4326) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s);""" psyex.execute_batch(curInsert,sql,dataPCV.values) print("process of insert into temp temp_data_pcv in " + str(time.time()-time_start) +"s") #update commune sql = """UPDATE temp_data_pcv SET commune_id=comm.id FROM commune comm WHERE temp_data_pcv.commune_id_cvi_insee::INTEGER=comm.insee::INTEGER ;;""" curInsert.execute(sql) sql = """UPDATE temp_data_pcv SET commune_id=comm.id FROM commune comm WHERE commune_id IS NULL and temp_data_pcv.commune_id_cadastre::NUMERIC::INTEGER=comm.insee::INTEGER ;""" curInsert.execute(sql) time_start=time.time() sql = """INSERT INTO parcel (id,commune_id,identifiant_cvi, section_cvi , prefix_cvi , numero_cvi , dsf_cvi , owner_name , contenance_cvi , identifiant_cadastre , prefix_cadastre , section_cadastre , numero_cadastre , contenance_cadastre, date_created , date_updated , geom_4326, active, date_insertion ) SELECT nextval('parcel_id_seq'),commune_id, identifiant_cvi, section_cvi , prefix_cvi , numero_cvi , dsf_cvi , owner_name , contenance_cvi , identifiant_cadastre , prefix_cadastre , section_cadastre , numero_cadastre , contenance_cadastre, date_created , date_updated , ST_GeomFromText(geom_4326), TRUE, CURRENT_TIMESTAMP from temp_data_pcv ;""" curInsert.execute(sql) dropTempTable(curInsert,"temp_data_pcv") print("process of parcels in " + str(time.time()-time_start) +"s") def insertSubParcel(dataSPCV,curInsert): time_start=time.time() createTempTable(curInsert,"temp_data_spcv") sql = """INSERT INTO temp_data_spcv VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s);""" psyex.execute_batch(curInsert,sql,dataSPCV.values) print("process of insert into temp temp_data_spcv in " + str(time.time()-time_start) +"s") #update parcel sql = """UPDATE temp_data_spcv SET parcel_id=p.id FROM parcel p WHERE temp_data_spcv.identifiant_pcv_cvi=p.identifiant_cvi AND p.dataset_id IS NULL ;;""" curInsert.execute(sql) #update farmer sql = """UPDATE temp_data_spcv SET farmer_id=u.id FROM users u WHERE temp_data_spcv.code IS NOT NULL and LPAD(temp_data_spcv.code::numeric::integer::varchar,5,'0') =u.username ;;""" curInsert.execute(sql) time_start=time.time() sql = """INSERT INTO sous_parcel (id,farmer_id, parcel_id , num_spcv , plant_year , cepage , portegreffe , superficie , ecart_pied , ecart_rang ) SELECT nextval('sous_parcel_id_seq'), farmer_id , parcel_id , num_spcv , plant_year , cepage , portegreffe , superficie , ecart_pied , ecart_rang from temp_data_spcv ;""" curInsert.execute(sql) dropTempTable(curInsert,'temp_data_spcv') print("process of subparcels in " + str(time.time()-time_start) +"s") def checkCsvFiles(listCsvFiles): if len(listCsvFiles)>3: print('too many csv files to process') sys.exit() listTest=[] for row in listCsvFiles: listTest.append(row.split('/')[-1].split('_')[1]) print(listTest) if 'PCV' not in listTest or 'EXPL' not in listTest or 'SPCV' not in listTest: return False else: return True if __name__ == '__main__': parser=agp.ArgumentParser('sentinel process') parser.add_argument("-d", "--dir", help="directory with files to import",default='', type=str) args = parser.parse_args() print(args) if args.dir == '': print('you must define directory that contains csv files') sys.exit() listCsvFiles = glob.glob(args.dir + '/*.csv') if not checkCsvFiles(listCsvFiles): print('you must have three files PCV, SPCV, EXPLOIT') else: print('three files will be processed : ', listCsvFiles) for row in listCsvFiles: print(row) if 'EXPL' in row: dfExploit = pd.read_csv(row, delimiter=';') dfExploit = dfExploit.fillna(psypg.extensions.AsIs('NULL')) if 'SPCV' in row: dfSPCV = pd.read_csv(row, delimiter=';', dtype={'cepage': int, 'portegreffe': int}) dfSPCV.drop(columns=['MMME', 'nomofficiel', 'nomofficiel2', 'adresse', 'adresse2', 'CP', 'ville'], inplace=True) dfSPCV = dfSPCV.fillna(psypg.extensions.AsIs('NULL')) if '_PCV' in row: parse_dates = ['date_created', 'date_updated'] dfPCV = pd.read_csv(row, delimiter=';', dtype={'commune_id_cvi_insee': str, 'commune_id_cvi': str,'commune_id_cadastre': str}) dfPCV = pd.read_csv(row, delimiter=';') dfPCV = dfPCV.astype('object') dfPCV[dfPCV.isnull()] = None connDb = connectdb() print("Database connexion ok") curInsert = connDb.cursor() insertExploit(dfExploit, curInsert) insertParcel(dfPCV, curInsert) insertSubParcel(dfSPCV, curInsert) connDb.commit() curInsert.close() disconnectdb(connDb)