# ----------------------------------- # 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 sys 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 bcolors: HEADER = '\033[95m' OKBLUE = '\033[94m' OKCYAN = '\033[96m' OKGREEN = '\033[92m' WARNING = '\033[93m' FAIL = '\033[91m' ENDC = '\033[0m' BOLD = '\033[1m' UNDERLINE = '\033[4m' class Holvia(TS_OCR): sqlUser = 'toosmart' sqlHost = '' sqlPasswd = 'i5Hf7MSHKkPN1hy' sqlDb = '' idAbattoir = 24 linesPoids = [] linesLots = [] lineMerged = {} dateLivraison = "" typeLivraison = "" enAnomalie = 0 poidsTotal = 0.000 cleCoche = "COCHE[ ]?FILIE[ ]?RE[ ]?LBC[ ]?[Vv]" ocrDpi = 500 ocrFolder = '/home/serge/www/toosmart_digicoche/data/tmp/' outputFileName = '' motsASurligner = [] dateTuerie = "" poidsTotalCalcule = 0.000 poidsTotalDetecte = 0 qttCocheCalcule = 0 qttCocheDetecte = 0 nonIntegrable = False mergedImage = [] def __init__(self, pdf, outputFileName,ocrDirectory,dbName,sqlHost,debug): 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.debug = debug except: self.printDebug("Erreur de connexion à la base de données") def printDebug(self,pTxt): self._debugOCR = self._debugOCR + '
' + pTxt if self.debug: print(pTxt) # Analyse du BL de l'abattoir def read(self): try: # conversion en fichier jpeg try: imageAmelioreeSurlignee = [] # lecture pagesPdf = convert_from_path(self.pdf, self.ocrDpi) iPage = 0 ocrReturnArray = [] except Exception as err: self.printDebug(f"Unexpected {err=}, {type(err)=}") self.printDebug("ERROR") """ 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 + 'Page_' + str(iPage) + '.jpg', "JPEG") # --------------------------------- # Optimisation de l'image img = cv2.imread(self.ocrFolder + 'Page_' + str(iPage) + '.jpg') result = img.copy() # On supprime les lignes horizontales gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY) thresh = cv2.threshold(gray, 6, 255, cv2.THRESH_BINARY_INV + cv2.THRESH_OTSU)[1] horizontal_kernel = cv2.getStructuringElement(cv2.MORPH_RECT, (40, 1)) remove_horizontal = cv2.morphologyEx(thresh, cv2.MORPH_OPEN, horizontal_kernel, iterations=1) cnts = cv2.findContours(remove_horizontal, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE) cnts = cnts[0] if len(cnts) == 2 else cnts[1] for c in cnts: cv2.drawContours(result, [c], -1, (255, 255, 255), 1) # --------------------------------- # On lance la reconaissance OCR sur l'image df = pytesseract.image_to_string(np.array(result), lang='fra',config=' --oem 0 --psm 4') cv2.imwrite(self.ocrFolder + 'Page_' + str(iPage) + '_result.jpg', result) ocrReturnArray.append(df) # --------------------------------- imageAmelioreeSurlignee.append(Image.open(self.ocrFolder + 'Page_' + str(iPage) + '_result.jpg')) iPage = iPage + 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) self.printDebug(f"Unexpected {e=}, {type(e)=}") self.printDebug("ERROR") # Recherche du type de livraison m = re.search(self.cleCoche, ocrReturnArray[0]) if m is not None: self.typeLivraison = m.group() else: m = re.search(r"COCHE[ ]?FIL[ \/]?([IE])[BR]E[ ]?LBC[ ]?[Vv]?", ocrReturnArray[0]) if m is not None: self.typeLivraison = "COCHE FILIERE LBC VPF" else: # BL Non intégrable car la clef n'est pas correcte # On retourne dans le bash l'information puis on quitte le script # (NE PAS SUPPRIMER CETTE LIGNE !!!) print('{"blNonIntegrable":1}') self.nonIntegrable = True self.printDebug(ocrReturnArray[0]) return False # --------------------------------- # Recherche de la date de livraison # MB - 25/10/24 - Modification de la regex pour prendre en compte le cas 1 1/05/2024 en plus # ancienne regex [0-9]{8} (?:([0-9]) )?([0-9]{1,2})/([0-9]{2})/([0-9]{4}) # nouvelle regex Livré le (?:([0-9]) )?([0-9]{1,2})\/([0-9]{2})\/([0-9]{4}) m = re.findall(r"Livré le ([0-9]{1,2}\/[0-9]{2}\/[0-9]{4})", ocrReturnArray[0]) if len(m) > 0: self.dateLivraison = str(m[0]) # --------------------------------- self.printDebug(ocrReturnArray[0]) # Recherche du numéro du BL try: m = re.findall(r"Bon de livraison[\n]+[0-9]{3}[a-zA-Z ]+ ([0-9]{8})", ocrReturnArray[0],re.MULTILINE) 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.nonIntegrable = True return False # --------------------------------- j = 0 """ On parcours les pages du pdf """ for Sfile in ocrReturnArray: self.printDebug(Sfile) # Recherche du nombre de coches, du poids et du lot m = re.findall(r"([0-9]{1,3}[.,][0-9]{1,2}).*([0-9]{10})", Sfile) if len(m) > 0: self.linesPoids.extend(m) self.qttCocheCalcule = len(m) + self.qttCocheCalcule for poid in m: # on ajoute le poids détecté au poids total self.poidsTotalCalcule = round(self.poidsTotalCalcule + round(float(poid[0].replace(",", ".")), 3), 3) # --------------------------------- # Recherche des numéros de lots, numéro de tuerie et eleveur #m = re.findall(r" ([0-9]{5}) [0-9].* ([0-9]{1,2}/[0-9]{1,2}/[0-9]{4}) ([0-9A-Za-z]*)", Sfile,re.MULTILINE) #if len(m) > 0: # self.linesLots.extend(m) # --------------------------------- # Recherche du poids total et du nombre de pièces try: m = re.findall(r"TOTAL MARCHANDISES\D+(([0-9])+)\D+(([0-9.,])+)\D+KG", ocrReturnArray[0], re.MULTILINE) if len(m) > 0: for element in m: self.poidsTotalDetecte = element[2].replace(".", "") self.qttCocheDetecte = element[0] except Exception as err: self.printDebug(f"Unexpected1 {err=}, {type(err)=}") self.printDebug("ERROR1") # --------------------------------- # Gestion de l'image pour surlignage des zones try: tesseract_output = pytesseract.image_to_data(imageAmelioreeSurlignee[j],output_type=pytesseract.Output.DICT) self.surlignerZones(tesseract_output,j) except: self.printDebug("ERROR") j = j + 1 # --------------------------------- # Merge des fichiers PDF try: im = Image.open(self.ocrFolder + self.outputFileName + '-color-0.jpg').convert('RGB') im.save(self.ocrFolder + self.outputFileName + "-BL.pdf", save_all=True, append_images=self.mergedImage) #im.save(self.ocrFolder+"BL.pdf", save_all=True, append_images=self.mergedImage) 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) self.printDebug(f"Unexpected {e=}, {type(e)=}") self.printDebug("ERROR") # --------------------------------- 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) self.printDebug(f"Unexpected {e=}, {type(e)=}") self.printDebug("ERROR") """ Surligne les zones détectées par l'OCR dans le fichier """ def surlignerZones(self, tesseract_output, iPage): try: imageASurligner = cv2.imread(self.ocrFolder + 'Page_' + str(iPage) + '_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") width, height = img.size new_size = (width // 4, height // 4) resized_image = img.resize(new_size).convert('RGB') if iPage > 0: self.mergedImage.append(resized_image) resized_image.save(self.ocrFolder + self.outputFileName + '-color-'+str(iPage)+'.jpg') except Exception as err: print("erreur") print(f"UnexpectedE {err=}, {type(err)=}") return True def mergeData(self): try: if self.nonIntegrable: return False # Parcours des lignes de poids for linePoids in self.linesPoids: self.lineMerged[linePoids[1][-5:]] = {"poids": linePoids[0], "date": str(self.quantiemeVersDate(int(linePoids[1][2:5]))), "eleveur": "", "tuerie": linePoids[1][-5:], "lot": linePoids[1]} # Parcours des lots #for lineLots in self.linesLots: # try: # self.lineMerged[lineLots[0]]["date"] = lineLots[1] # self.lineMerged[lineLots[0]]["eleveur"] = lineLots[2] # except: # continue 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 créé l'entête du BL dans eteko # Date de création de la ligne currentDate = datetime.now().strftime('%Y%m%d%H%M') 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) " "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("'", "\\'") +"');" try: self.cursor.execute(sql) except Exception as err: print(f"UnexpectedA {err=}, {type(err)=}") print("ERROR A2") # 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) + "}") # ------------------------------------------ for key in self.lineMerged: # Initialisation des anomalies de la ligne à 0 self.enAnomalie = 0 try: # On stocke la date de livraison dans le format YYYYMMDD self.dateLivraisonLigne = self.lineMerged[key]["date"] except: self.dateLivraisonLigne = "" self.enAnomalie = 1 # Récupération du numéro de tuerie numeroTuerie = str(self.lineMerged[key]["lot"])[-4:] # On créé la ligne du BL dans eteko 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, attf101,attf104) VALUES (1, 11, NULL, 11, 1, 0, null, null, null, null, null, '" + str( blId) + "','" + str( self.lineMerged[key]["lot"]) + "', '" + str(numeroTuerie) + "', '" + str( self.lineMerged[key]["poids"]) + "', '" + str(self.dateLivraisonLigne) + "', '" + str( self.enAnomalie) + "', '" + self.dateLivraison + "');") self.cursor.execute(sql) # Fermeture de la connexion à MySQL self.conn.close() 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) self.printDebug(f"Unexpected {e=}, {type(e)=}") self.printDebug("ERROR") 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) self.printDebug(f"Unexpected {e=}, {type(e)=}") self.printDebug("ERROR") 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 err: print(f"UnexpectedA {err=}, {type(err)=}") print("ERROR A4 ")