# ----------------------------------- # Imports des librairies # ----------------------------------- try: from PIL import Image from uuid import uuid4 import json except ImportError: import Image import pytesseract from pytesseract import Output import cv2 import re from pdf2image import convert_from_path from TSLibOCR import * import time import os import numpy as np from TS_OCR import * import MySQLdb from datetime import datetime from collections import Counter debug = False """ Retourne le temps passé depuis le lancement du programme """ def getTime(): return "" class Loudeac(TS_OCR): sqlUser = 'toosmart' sqlHost = '' sqlPasswd = 'i5Hf7MSHKkPN1hy' sqlDb = '' ocrDpi = 350 ocrFolder = '/home/serge/www/toosmart_digicoche/data/tmp/' outputFileName = '' linesPoids = [] linesLots = [] lineMerged = {} motsASurligner = [] dateLivraison = "" dateTuerie = "" dateAbattageMultiple = False cleCoche = "COCHE L[R]?[B]?C VPF" typeLivraison = "" enAnomalie = 0 poidsTotalCalcule = 0.000 poidsTotalDetecte = 0 qttCocheCalcule = 0 idAbattoir = 22 qttCocheDetecte = 0 nonIntegrable = False def __init__(self, pdf, outputFileName,ocrDirectory,dbName,sqlHost): try: self.sqlDb = dbName self.sqlHost = sqlHost """ Connexion à la base de données MySQL de Digicoche (Eteko)""" self.conn = MySQLdb.connect(host=self.sqlHost, user=self.sqlUser, passwd=self.sqlPasswd, db=self.sqlDb) self.cursor = self.conn.cursor() self.pdf = pdf self.outputFileName = outputFileName self.ocrFolder = ocrDirectory self.printDebug("Connexion à la base de données OK") self.printDebug("outputFileName = "+str(outputFileName)) except: self.printDebug("Erreur de connexion à la base de données") def printDebug(self,pTxt): self._debugOCR = self._debugOCR + '
' + pTxt if debug: print(pTxt) # Analyse du BL de l'abattoir Loudeac def read(self): # lecture pagesPdf = convert_from_path(self.pdf, self.ocrDpi) iPage = 1 ocrReturnArray = [] """ On parcours les pages du pdf """ for page in pagesPdf: try: # conversion de l'image en fichier jpeg image_name = "Page_" + str(iPage) + ".jpg" page.save(self.ocrFolder + image_name, "JPEG") # --------------------------------- # Optimisation de l'image img = cv2.imread(self.ocrFolder + image_name) imageAmelioree = img.copy() imageAmelioree = self.increase_brightness(imageAmelioree,30) # Augmente la luminositée de l'image afin d'eleminer les surimpressions du BL Loudeac cv2.imwrite(self.ocrFolder + 'Page_' + str(iPage) + '_result.jpg', imageAmelioree) imageAmelioreeSurlignee = Image.open(self.ocrFolder + 'Page_' + str(iPage) + '_result.jpg') # --------------------------------- # On lance la reconaissance OCR sur l'image self.printDebug("%s - Lecture OCR" % getTime()) df = pytesseract.image_to_string(np.array(imageAmelioree), lang='fra', config=' ') ocrReturnArray.append(df) self.printDebug("%s - Lecture OCR Terminée" % getTime()) # --------------------------------- iPage = iPage + 1 except Exception as err: self.printDebug(f"Unexpected {err=}, {type(err)=}") self.printDebug("ERROR") # Recherche de la date de livraison m = re.findall(r"[0-9]{5} ([0-9]{1,2}/[0-9]{1,2}/[0-9]{2})", ocrReturnArray[0]) if len(m) > 0: self.dateLivraison = str(m[0]) self.motsASurligner.append(m[0]) self.printDebug("Date de livraison : " + self.dateLivraison) # --------------------------------- # Recherche de la date de la tuerie m = re.findall(r"DATE ABATTAG\D+([0-9]{1,2}\/[0-9]{1,2}\/[0-9]{2})", ocrReturnArray[0]) if len(m) > 0: self.dateTuerie = str(m[0]) self.motsASurligner.append(m[0]) self.printDebug("Date de la tuerie : " + self.dateTuerie) # --------------------------------- # On vérifie si il y a plus d'une date de tuerie m = re.findall(r"DATE ABATTAG\D+([0-9]{1,2}\/[0-9]{1,2}\/[0-9]{2})[,. ]*([0-9]{1,2}\/[0-9]{1,2}\/[0-9]{2})", ocrReturnArray[0]) if len(m) > 0: # Il existe plus d'une date de tuerie sur le BL self.dateAbattageMultiple = True # --------------------------------- # Recherche du poids total try: m = re.findall(r"TOTAL[\W]+GENERAL\D+([\d]+)\D+([\d\,\. ]+)", ocrReturnArray[0]) if len(m) > 0: for element in m: self.poidsTotalDetecte = element[1].replace('.', '') self.qttCocheDetecte = element[0] except Exception as err: self.printDebug(f"Unexpected1 {err=}, {type(err)=}") self.printDebug("ERROR1") # --------------------------------- # Recherche du type de livraison m = re.search(self.cleCoche, ocrReturnArray[0]) if m is not None: self.typeLivraison = m.group() self.motsASurligner.append("COCHE LBC VPF") self.printDebug("Type de livraison : " + self.typeLivraison) else: m = re.search(r"POCHP LDC VPF", ocrReturnArray[0]) if m is not None: self.typeLivraison = m.group() self.motsASurligner.append("POCHP LDC VPF") self.printDebug("Type de livraison : " + self.typeLivraison) else: print('{"blNonIntegrable":1}') self.nonIntegrable = True return False # --------------------------------- # Recherche du numéro du BL self.printDebug('1 : ' + ocrReturnArray[0]) try: m = re.findall(r"BON LIVRAISON[ ]?N°[\D]+([0-9]{5})", ocrReturnArray[0],re.IGNORECASE) if len(m) > 0: for element in m: self._numBl = element except Exception as err: self._numBl = "" self.printDebug(f"Unexpected1 {err=}, {type(err)=}") self.printDebug("ERROR1") if self._numBl != "": if self.blExist(): # Le BL Existe déjà en base de donneés print('{"blEnDoublon":1}') self.printDebug(self._numBl) self.nonIntegrable = True return False # --------------------------------- # On parcours les pages du BL pour rechercher les données for Sfile in ocrReturnArray: Sfile = re.sub(r', ([0-9])', r',\1',str(ocrReturnArray[0])) # Recherche du nombre de coches, du poids et du lot if self.typeLivraison != "": # On vérifie si le type de livraison du BL est bien autorisé m = re.findall(r"([0-9]{4,5})[ -.]([0-9.,]{5})", Sfile) if len(m) > 0: self.qttCocheCalcule = len(m) for poid in m: # On ajoute le N° de tuerie et le poids au texte à surligner self.motsASurligner.append(poid[0]) # N° de tuerie self.motsASurligner.append(poid[1]) # Poids self.printDebug("Nombre de coches detectées : " + str(len(m))) self.printDebug("Détail des poids : " + str(m)) # On sauvegarde les lignes détectées self.linesPoids.extend(m) # Gestion de l'image pour surlignage des zones tesseract_output = pytesseract.image_to_data(np.array(imageAmelioreeSurlignee),output_type=pytesseract.Output.DICT) try: self.surlignerZones(tesseract_output) except: print(f"Unexpected2 {err=}, {type(err)=}") print("ERROR2") # --------------------------------- def mergeData(self): if self.nonIntegrable: return False for linePoids in self.linesPoids: self.lineMerged[linePoids[1]] = {"poids" : linePoids[1], "date" : "","lot":linePoids[0]} try: # On stocke la date de livraison dans le format YYYYMMDD self.dateLivraison = datetime.strptime(self.dateLivraison, "%d/%m/%y").strftime("%Y%m%d") except: self.dateLivraison = "" try: # On stocke la date de la tuerie dans le format YYYYMMDD self.dateTuerie = datetime.strptime(self.dateTuerie, "%d/%m/%y").strftime("%Y%m%d") except: self.dateTuerie = "" self.qttCocheCalcule = len(self.lineMerged) try: for key in self.lineMerged: # on ajoute le poids détecté au poids total self.poidsTotalCalcule = round(self.poidsTotalCalcule + round(float(self.lineMerged[key]["poids"].replace(",", ".")), 3), 3) except Exception as err: print(f"UnexpectedA {err=}, {type(err)=}") print("ERROR A1") # On créé l'entête du BL dans eteko # Date de création de la ligne currentDate = datetime.now().strftime('%Y%m%d%H%M') if self.enAnomalie: statutBL = 6 # En anomalie else: statutBL = 2 # Non controlé sql = "INSERT INTO tier_att(id_base, id_type, id_tier,attdatcre, id_work, actif, newobj, att1, att2, att3,att6,att7,att8,att9,attf100,att17,att90,att15) VALUES (1, 10, NULL,'"+str(currentDate)+"', 16, 1, 0, "+str(self.idAbattoir)+", '" + self.typeLivraison + "', '" + str( self.poidsTotalCalcule) + "','"+str(self.poidsTotalDetecte)+"','"+str(self.poidsTotalCalcule)+"','"+str(self.qttCocheDetecte)+"','"+str(self.qttCocheCalcule)+"', '" + self.dateLivraison + "','"+ self._numBl+"','"+ self._debugOCR.replace("'", "\\'") +"',"+str(statutBL)+");" try: self.cursor.execute(sql) except Exception as err: print(f"UnexpectedA {err=}, {type(err)=}") print("ERROR A2") print(sql) # On récupère l'id du BL créé dans Eteko blId = self.cursor.lastrowid # On retourne dans le bash l'id du BL créé # (NE PAS SUPPRIMER CETTE LIGNE !!!) print('{"blId":' + str(blId) + "}") # ------------------------------------------ # Contrôle des longueurs des numéros de tuerie # On vérifie la longueur, on en déduit une majorité, # puis on met en anomalie celle qui ne sont pas de la même longueur que la majorité tuerieLongueur = {} for key in self.lineMerged: try: tuerieLongueur[len(self.lineMerged[key]["lot"])] = tuerieLongueur[len(self.lineMerged[key]["lot"])] + 1 except: tuerieLongueur[len(self.lineMerged[key]["lot"])] = 1 longueurNumTueriePlusUtilise = max(tuerieLongueur, key = lambda k: tuerieLongueur[k]) #------------------------------------------- # On parcours les lignes du BL pour les ajouter à la base de données for key in self.lineMerged: numeroTuerie = str(self.lineMerged[key]["lot"]) anomalieLigne = "" if(len(numeroTuerie) != longueurNumTueriePlusUtilise): anomalieLigne = "anomalie" sql = ( "INSERT INTO prod_att(id_base, id_type, id_prod, id_work, actif, newobj, fid_prod, type, name, revision, `desc`, att1, att2, att3, att4, attf100, att8,attf104) VALUES (1, 11, NULL, 11, 1, 0, null, null, null, null, null, '" + str( blId) + "','', '" + str(numeroTuerie) + "', '" + str( self.lineMerged[key]["poids"]) + "', '" + str(self.dateTuerie) + "','" + str( anomalieLigne) + "','" + self.dateLivraison + "');") try: self.cursor.execute(sql) except Exception as err: print(f"UnexpectedB {err=}, {type(err)=}") print("ERRORB") self.conn.close() """ Surligne les zones détectées par l'OCR dans le fichier """ def surlignerZones(self,tesseract_output): try: imageASurligner = cv2.imread(self.ocrFolder + 'Page_1_result.jpg') blockASurligner = [] except Exception as err: print(f"UnexpectedC {err=}, {type(err)=}") print("ERRORC") for i, level_idx in enumerate(tesseract_output['level']): if level_idx == 5: #if tesseract_output['text'][i] in self.motsASurligner: bbox = { 'x': 1 * tesseract_output['left'][i] , 'y': 1 * tesseract_output['top'][i] , 'width': 1 * tesseract_output['width'][i] , 'height': 1 * tesseract_output['height'][i], 'rotation': 0, 'confidence':tesseract_output['conf'][i], 'text':tesseract_output['text'][i] } if int(tesseract_output['conf'][i]) >= 85: color = (42, 219, 151) elif int(tesseract_output['conf'][i]) >= 50: color = (0, 165, 255) else: color = (0, 0, 255) try: overlay = imageASurligner.copy() cv2.rectangle(overlay, (bbox['x'], bbox['y']), (bbox['x'] + bbox['width'], bbox['y'] + bbox['height']), color, cv2.FILLED) alpha = 0.5 imageASurligner = cv2.addWeighted(overlay, alpha, imageASurligner, 1 - alpha, 0) blockASurligner.append(bbox) except Exception as err: print(f"UnexpectedD {err=}, {type(err)=}") print("ERRORD") # On sauvegarde le fichier try: imageASurligner = cv2.cvtColor(imageASurligner, cv2.COLOR_BGR2RGB) img = Image.fromarray(imageASurligner, "RGB") img.save(self.ocrFolder+self.outputFileName+"-BL.pdf") except Exception as err: print("erreur") print(f"UnexpectedE {err=}, {type(err)=}") return True def increase_brightness(self,img, value=30): hsv = cv2.cvtColor(img, cv2.COLOR_BGR2HSV) h, s, v = cv2.split(hsv) lim = 255 - value v[v > lim] = 255 v[v <= lim] += value final_hsv = cv2.merge((h, s, v)) img = cv2.cvtColor(final_hsv, cv2.COLOR_HSV2BGR) img = cv2.addWeighted(img, 1.2, np.zeros(img.shape, img.dtype), 0, 5) return img def quantiemeVersDate(self,n): try: a = datetime.now().year """Donne la date d=[j,m,a] qui est le nième jour de l'année a""" if ((a%4==0 and a%100!=0) or a%400==0): # bissextile? jm = (0,31,60,91,121,152,182,213,244,274,305,335,366) else: jm = (0,31,59,90,120,151,181,212,243,273,304,334,365) for m in range(1,13): if jm[m]>=n: #return str(n)+str(jm[m-1])+ str(m)+ str(a) return str(a) + str("{:02d}".format(m)) + str("{:02d}".format(n-jm[m-1])) except Exception as e: exception_type, exception_object, exception_traceback = sys.exc_info() filename = exception_traceback.tb_frame.f_code.co_filename line_number = exception_traceback.tb_lineno print("Exception type: ", exception_type) print("File name: ", filename) print("Line number: ", line_number) print("n: ", n) self.printDebug(f"Unexpected {e=}, {type(e)=}") self.printDebug("ERROR")