1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128
| # coding: utf-8
'''
Created on 23 mars 2018
@author:
'''
import csv
import sqlite3
import sys
import csv
import os
import encodings
from collections import namedtuple
from src.convertie import *
class record(namedtuple('record', 'ACT_CODE, ACT_LIBELLE')):
def __new__(cls, row):
if row[0] == 'NULL':
record.ACT_CODE = 'NULL'
if row[1] == 'NULL':
record.ACT_LIBELLE = 'NULL'
return super().__new__(cls, *row)
class Writer:
def __init__(self, path):
#def __init__(self, path='mem.db'):
cnx = self._cnx = sqlite3.connect(path)
commandesql = '''drop table if exists vsl_actes; '''
try:
cnx.execute(commandesql)
except sqlite3.Error as e:
print("Problème"+ " :" + e.args[0] + " sur commande /n"+ commandesql )
self._cnx.rollback()
else:
self._cnx.commit()
commandesql = """CREATE TABLE vsl_actes(
ACT_CODE TEXT(255),
ACT_LIBELLE TEXT(255),
CONSTRAINT actes_pk PRIMARY KEY(ACT_CODE));"""
try:
cnx.execute(commandesql)
except sqlite3.Error as e:
print("Problème"+ " :" + e.args[0] + " sur commande /n"+ commandesql )
self._cnx.rollback()
else:
self._cnx.commit()
commandesql = """CREATE INDEX vsl_actes_IDX ON vsl_actes(ACT_LIBELLE);"""
try:
cnx.execute(commandesql)
except sqlite3.Error as e:
print("Problème"+ " :" + e.args[0] + " sur commande /n"+ commandesql )
self._cnx.rollback()
else:
self._cnx.commit()
def writesmany(self, records):
rq = """INSERT OR IGNORE INTO vsl_actes
(ACT_CODE, ACT_LIBELLE)
VALUES(?, ?);"""
with self._cnx:
try:
self._cnx.executemany(rq, records)
except sqlite3.Error as e:
print("Problème"+ " :" + e.args[0] + " sur commande d'insertion /n" )
print("\t" +str(records))
self._cnx.rollback()
# dans le cas d'un problème d'injection multipls on repasse à la mono injection au cas par cas
# et on laisse les mécasnisme de sécurite de la base agir
self.writemono( records)
else:
self._cnx.commit()
def writemono(self, records):
rq = """INSERT OR IGNORE INTO vsl_actes
(ACT_CODE, ACT_LIBELLE)
VALUES(?, ?);"""
with self._cnx:
for enregistrement in records:
try:
self._cnx.execute(rq, enregistrement)
except sqlite3.Error as e:
print("Problème"+ " :" + e.args[0] + " sur commande d'insertion /n" )
print("\t" +str(enregistrement))
self._cnx.rollback()
else:
self._cnx.commit()
def dump(self):
with self._cnx:
for row in self._cnx.execute('select * from vsl_actes'):
print (row)
class vsl_acte(object):
def __init__(self, nomfichier , base):
self.nomfichier = nomfichier
self.base = base
print('Injection ' + self.nomfichier +' sur la base ' + self.base )
writer = Writer(self.base )
with open(self.nomfichier , encoding='utf-8') as file:
reader = csv.reader(file,'excel-fr')
reader.__next__() # bypass l'entete du fichier
packer = Packer(reader, record)
i = 1
for records in packer.get_records():
writer.writesmany(records)
#print(i, records)
i += 1
if __name__ == '__main__':
encodage ="utf-8"
BaseSQL='BaseVSL.db'
csv.register_dialect('excel-fr', ExcelFr())
vsl_acte('actes.csv', BaseSQL )
#writer.dump() |
Partager