From d9506337ff163e4321d6a70ce2b618e10204f233 Mon Sep 17 00:00:00 2001 From: fanilo Date: Sun, 28 Dec 2025 12:34:49 +0100 Subject: [PATCH] Mega enriched articles' data --- sage_connector.py | 2499 ++++++++++++++++++++++++++++++++++++--------- 1 file changed, 2014 insertions(+), 485 deletions(-) diff --git a/sage_connector.py b/sage_connector.py index a8435f9..d2a20d8 100644 --- a/sage_connector.py +++ b/sage_connector.py @@ -112,9 +112,6 @@ class SageConnector: try: with self._get_sql_connection() as conn: cursor = conn.cursor() - cursor.execute("SELECT COUNT(*) FROM F_COMPTET") - nb_tiers = cursor.fetchone()[0] - logger.info(f" Connexion SQL réussie: {nb_tiers} tiers détectés") except Exception as e: logger.warning(f"SQL non disponible: {e}") logger.warning(" Les lectures utiliseront COM (plus lent)") @@ -295,7 +292,6 @@ class SageConnector: "categorie_compta": row.N_CatCompta, } - # Récupérer les contacts (réutiliser la même méthode) fournisseur["contacts"] = self._get_contacts_client(row.CT_Num, conn) fournisseurs.append(fournisseur) @@ -461,7 +457,6 @@ class SageConnector: "categorie_compta": row.N_CatCompta, } - # Récupérer les contacts fournisseur["contacts"] = self._get_contacts_client(row.CT_Num, conn) logger.info(f"✅ SQL: Fournisseur {code_fournisseur} avec {len(fournisseur)} champs") @@ -955,27 +950,22 @@ class SageConnector: contacts = [] for row in rows: contact = { - # IDENTIFICATION "ct_num": self._safe_strip(row.CT_Num), "ct_no": row.CT_No, "n_contact": row.N_Contact, - # IDENTITÉ "civilite": self._safe_strip(row.CT_Civilite), "nom": self._safe_strip(row.CT_Nom), "prenom": self._safe_strip(row.CT_Prenom), "fonction": self._safe_strip(row.CT_Fonction), - # ORGANISATION "service_code": row.N_Service, - # COORDONNÉES "telephone": self._safe_strip(row.CT_Telephone), "portable": self._safe_strip(row.CT_TelPortable), "telecopie": self._safe_strip(row.CT_Telecopie), "email": self._safe_strip(row.CT_EMail), - # RÉSEAUX SOCIAUX "facebook": self._safe_strip(row.CT_Facebook), "linkedin": self._safe_strip(row.CT_LinkedIn), "skype": self._safe_strip(row.CT_Skype) @@ -1314,7 +1304,6 @@ class SageConnector: "categorie_compta": row.N_CatCompta, } - # Récupérer les contacts client["contacts"] = self._get_contacts_client(row.CT_Num, conn) logger.info(f"✅ SQL: Client {code_client} avec {len(client)} champs") @@ -1325,277 +1314,1653 @@ class SageConnector: return None - def lister_tous_articles(self, filtre="", avec_stock=True): + + def lister_tous_articles(self, filtre=""): try: with self._get_sql_connection() as conn: cursor = conn.cursor() - - query = """ - SELECT - AR_Ref, AR_Design, AR_PrixVen, AR_PrixAch, - AR_UniteVen, FA_CodeFamille, AR_Sommeil, - AR_CodeBarre, AR_Type - FROM F_ARTICLE - WHERE 1=1 - """ - + + logger.info("[SQL] Détection des colonnes de F_ARTICLE...") + cursor.execute("SELECT TOP 1 * FROM F_ARTICLE") + colonnes_disponibles = [column[0] for column in cursor.description] + logger.info(f"[SQL] Colonnes F_ARTICLE trouvées : {len(colonnes_disponibles)}") + + colonnes_config = { + "AR_Ref": "reference", + "AR_Design": "designation", + "AR_CodeBarre": "code_barre", + "AR_EdiCode": "edi_code", + "AR_Raccourci": "raccourci", + + "AR_PrixVen": "prix_vente", + "AR_PrixAch": "prix_achat", + "AR_Coef": "coef", + "AR_PUNet": "prix_net", + "AR_PrixAchNouv": "prix_achat_nouveau", + "AR_CoefNouv": "coef_nouveau", + "AR_PrixVenNouv": "prix_vente_nouveau", + "AR_DateApplication": "date_application_prix", + "AR_CoutStd": "cout_standard", + + "AR_UniteVen": "unite_vente", + "AR_UnitePoids": "unite_poids", + "AR_PoidsNet": "poids_net", + "AR_PoidsBrut": "poids_brut", + + "AR_Gamme1": "gamme_1", + "AR_Gamme2": "gamme_2", + + "FA_CodeFamille": "famille_code", + "AR_Type": "type_article", + "AR_Nature": "nature", + "AR_Garantie": "garantie", + "AR_CodeFiscal": "code_fiscal", + "AR_Pays": "pays", + + "CO_No": "fournisseur_principal", + "AR_Condition": "conditionnement", + "AR_NbColis": "nb_colis", + "AR_Prevision": "prevision", + + "AR_SuiviStock": "suivi_stock", + "AR_Nomencl": "nomenclature", + "AR_QteComp": "qte_composant", + "AR_QteOperatoire": "qte_operatoire", + + "AR_Sommeil": "sommeil", + "AR_Substitut": "article_substitut", + "AR_Escompte": "soumis_escompte", + "AR_Delai": "delai", + + "AR_Stat01": "stat_01", + "AR_Stat02": "stat_02", + "AR_Stat03": "stat_03", + "AR_Stat04": "stat_04", + "AR_Stat05": "stat_05", + "AR_HorsStat": "hors_statistique", + + "CL_No1": "categorie_1", + "CL_No2": "categorie_2", + "CL_No3": "categorie_3", + "CL_No4": "categorie_4", + + "AR_DateModif": "date_modification", + + "AR_VteDebit": "vente_debit", + "AR_NotImp": "non_imprimable", + "AR_Transfere": "transfere", + "AR_Publie": "publie", + "AR_Contremarque": "contremarque", + "AR_FactPoids": "fact_poids", + "AR_FactForfait": "fact_forfait", + "AR_SaisieVar": "saisie_variable", + "AR_Fictif": "fictif", + "AR_SousTraitance": "sous_traitance", + "AR_Criticite": "criticite", + + "RP_CodeDefaut": "reprise_code_defaut", + "AR_DelaiFabrication": "delai_fabrication", + "AR_DelaiPeremption": "delai_peremption", + "AR_DelaiSecurite": "delai_securite", + "AR_TypeLancement": "type_lancement", + "AR_Cycle": "cycle", + + "AR_Photo": "photo", + "AR_Langue1": "langue_1", + "AR_Langue2": "langue_2", + + "AR_Frais01FR_Denomination": "frais_01_denomination", + "AR_Frais02FR_Denomination": "frais_02_denomination", + "AR_Frais03FR_Denomination": "frais_03_denomination", + + "Marque commerciale": "marque_commerciale", + "Objectif / Qtés vendues": "objectif_qtes_vendues", + "Pourcentage teneur en or": "pourcentage_or", + "1ère commercialisation": "premiere_commercialisation", + "AR_InterdireCommande": "interdire_commande", + "AR_Exclure": "exclure", + } + + colonnes_a_lire = [ + col_sql for col_sql in colonnes_config.keys() + if col_sql in colonnes_disponibles + ] + + if not colonnes_a_lire: + logger.error("[SQL] Aucune colonne mappée trouvée !") + colonnes_a_lire = ["AR_Ref", "AR_Design", "AR_PrixVen"] + + logger.info(f"[SQL] Colonnes sélectionnées : {len(colonnes_a_lire)}") + + colonnes_sql = [] + for col in colonnes_a_lire: + if " " in col or "/" in col or "è" in col: + colonnes_sql.append(f"[{col}]") + else: + colonnes_sql.append(col) + + colonnes_str = ", ".join(colonnes_sql) + query = f"SELECT {colonnes_str} FROM F_ARTICLE WHERE 1=1" + params = [] - + if filtre: - query += " AND (AR_Ref LIKE ? OR AR_Design LIKE ?)" - params.extend([f"%{filtre}%", f"%{filtre}%"]) - + conditions = [] + if "AR_Ref" in colonnes_a_lire: + conditions.append("AR_Ref LIKE ?") + params.append(f"%{filtre}%") + if "AR_Design" in colonnes_a_lire: + conditions.append("AR_Design LIKE ?") + params.append(f"%{filtre}%") + if "AR_CodeBarre" in colonnes_a_lire: + conditions.append("AR_CodeBarre LIKE ?") + params.append(f"%{filtre}%") + + if conditions: + query += " AND (" + " OR ".join(conditions) + ")" + query += " ORDER BY AR_Ref" - + + logger.debug(f"[SQL] Requête : {query[:200]}...") cursor.execute(query, params) rows = cursor.fetchall() - + + logger.info(f"[SQL] {len(rows)} lignes récupérées") + articles = [] - + for row in rows: - article = { - "reference": self._safe_strip(row[0]), - "designation": self._safe_strip(row[1]), - "prix_vente": float(row[2]) if row[2] is not None else 0.0, - "prix_achat": float(row[3]) if row[3] is not None else 0.0, - "unite_vente": ( - str(row[4]).strip() if row[4] is not None else "" - ), - "famille_code": self._safe_strip(row[5]), - "est_actif": (row[6] == 0), - "code_ean": self._safe_strip(row[7]), - "type_article": row[8] if row[8] is not None else 0, - "stock_reel": 0.0, - "stock_mini": 0.0, - "stock_maxi": 0.0, - "stock_reserve": 0.0, - "stock_commande": 0.0, - "stock_disponible": 0.0, - } - article["code_barre"] = article["code_ean"] - articles.append(article) + row_data = {} + for idx, col_sql in enumerate(colonnes_a_lire): + valeur = row[idx] + if isinstance(valeur, str): + valeur = valeur.strip() + row_data[col_sql] = valeur + + if "Marque commerciale" in row_data: + logger.debug(f"[DEBUG] Marque commerciale trouvée: {row_data['Marque commerciale']}") + + article_data = self._mapper_article_depuis_row(row_data, colonnes_config) + articles.append(article_data) + + articles = self._enrichir_stocks_articles(articles, cursor) + articles = self._enrichir_familles_articles(articles, cursor) + articles = self._enrichir_fournisseurs_articles(articles, cursor) + articles = self._enrichir_tva_articles(articles, cursor) - if avec_stock and articles: - logger.info( - f" Enrichissement stock depuis F_ARTSTOCK pour {len(articles)} articles..." - ) - - try: - references = [ - a["reference"] for a in articles if a["reference"] - ] - - if not references: - return articles - - placeholders = ",".join(["?"] * len(references)) - stock_query = f""" - SELECT - AR_Ref, - SUM(ISNULL(AS_QteSto, 0)) as Stock_Total, - MIN(ISNULL(AS_QteMini, 0)) as Stock_Mini, - MAX(ISNULL(AS_QteMaxi, 0)) as Stock_Maxi, - SUM(ISNULL(AS_QteRes, 0)) as Stock_Reserve, - SUM(ISNULL(AS_QteCom, 0)) as Stock_Commande - FROM F_ARTSTOCK - WHERE AR_Ref IN ({placeholders}) - GROUP BY AR_Ref - """ - - cursor.execute(stock_query, references) - stock_rows = cursor.fetchall() - - stock_map = {} - for stock_row in stock_rows: - ref = self._safe_strip(stock_row[0]) - if ref: - stock_map[ref] = { - "stock_reel": ( - float(stock_row[1]) if stock_row[1] else 0.0 - ), - "stock_mini": ( - float(stock_row[2]) if stock_row[2] else 0.0 - ), - "stock_maxi": ( - float(stock_row[3]) if stock_row[3] else 0.0 - ), - "stock_reserve": ( - float(stock_row[4]) if stock_row[4] else 0.0 - ), - "stock_commande": ( - float(stock_row[5]) if stock_row[5] else 0.0 - ), - } - - logger.info( - f" Stocks chargés pour {len(stock_map)} articles depuis F_ARTSTOCK" - ) - - for article in articles: - if article["reference"] in stock_map: - stock_data = stock_map[article["reference"]] - article.update(stock_data) - article["stock_disponible"] = ( - article["stock_reel"] - article["stock_reserve"] - ) - else: - article["stock_reel"] = 0.0 - article["stock_mini"] = 0.0 - article["stock_maxi"] = 0.0 - article["stock_reserve"] = 0.0 - article["stock_commande"] = 0.0 - article["stock_disponible"] = 0.0 - - except Exception as e: - logger.error( - f" Erreur lecture F_ARTSTOCK: {e}", exc_info=True - ) - - logger.info( - f" SQL: {len(articles)} articles (stock={'oui' if avec_stock else 'non'})" - ) + articles = self._enrichir_stock_emplacements(articles, cursor) + articles = self._enrichir_gammes_articles(articles, cursor) + articles = self._enrichir_tarifs_clients(articles, cursor) + articles = self._enrichir_nomenclature(articles, cursor) + articles = self._enrichir_compta_articles(articles, cursor) + articles = self._enrichir_fournisseurs_multiples(articles, cursor) + articles = self._enrichir_depots_details(articles, cursor) + articles = self._enrichir_emplacements_details(articles, cursor) + articles = self._enrichir_gammes_enumeres(articles, cursor) + articles = self._enrichir_references_enumerees(articles, cursor) + articles = self._enrichir_medias_articles(articles, cursor) + articles = self._enrichir_prix_gammes(articles, cursor) + articles = self._enrichir_conditionnements(articles, cursor) + return articles - + except Exception as e: - logger.error(f" Erreur SQL articles: {e}", exc_info=True) + logger.error(f"✗ Erreur SQL articles: {e}", exc_info=True) raise RuntimeError(f"Erreur lecture articles: {str(e)}") - def lire_article(self, reference): + + def _enrichir_stock_emplacements(self, articles: List[Dict], cursor) -> List[Dict]: + """ + Enrichit avec le détail du stock par emplacement + Structure: articles[i]["emplacements"] = [{"depot": "01", "emplacement": "A1", "qte": 10}, ...] + """ try: - with self._get_sql_connection() as conn: - cursor = conn.cursor() - - cursor.execute( - """ - SELECT - AR_Ref, AR_Design, AR_PrixVen, AR_PrixAch, - AR_UniteVen, FA_CodeFamille, AR_Sommeil, - AR_CodeBarre, AR_Type - FROM F_ARTICLE - WHERE AR_Ref = ? - """, - (reference.upper(),), - ) - - row = cursor.fetchone() - - if not row: - return None - - article = { - "reference": self._safe_strip(row[0]), - "designation": self._safe_strip(row[1]), - "prix_vente": float(row[2]) if row[2] is not None else 0.0, - "prix_achat": float(row[3]) if row[3] is not None else 0.0, - "unite_vente": str(row[4]).strip() if row[4] is not None else "", - "famille_code": self._safe_strip(row[5]), - "est_actif": (row[6] == 0), - "code_ean": self._safe_strip(row[7]), - "code_barre": self._safe_strip(row[7]), - "type_article": row[8] if row[8] is not None else 0, - "type_article_libelle": { - 0: "Article", - 1: "Prestation", - 2: "Divers", - }.get(row[8] if row[8] is not None else 0, "Article"), - "description": "", - "designation_complementaire": "", - "poids": 0.0, - "volume": 0.0, - "tva_code": "", - "date_creation": "", - "date_modification": "", - "stock_reel": 0.0, - "stock_mini": 0.0, - "stock_maxi": 0.0, - "stock_reserve": 0.0, - "stock_commande": 0.0, - "stock_disponible": 0.0, - } - - article["tva_taux"] = 20.0 - - logger.info(f" Lecture stock depuis F_ARTSTOCK pour {reference}...") - - try: - cursor.execute( - """ - SELECT - SUM(ISNULL(AS_QteSto, 0)) as Stock_Total, - MIN(ISNULL(AS_QteMini, 0)) as Stock_Mini, - MAX(ISNULL(AS_QteMaxi, 0)) as Stock_Maxi, - SUM(ISNULL(AS_QteRes, 0)) as Stock_Reserve, - SUM(ISNULL(AS_QteCom, 0)) as Stock_Commande - FROM F_ARTSTOCK - WHERE AR_Ref = ? - GROUP BY AR_Ref - """, - (reference.upper(),), - ) - - stock_row = cursor.fetchone() - - if stock_row: - article["stock_reel"] = ( - float(stock_row[0]) if stock_row[0] else 0.0 - ) - article["stock_mini"] = ( - float(stock_row[1]) if stock_row[1] else 0.0 - ) - article["stock_maxi"] = ( - float(stock_row[2]) if stock_row[2] else 0.0 - ) - - stock_reserve_artstock = ( - float(stock_row[3]) if stock_row[3] else 0.0 - ) - stock_commande_artstock = ( - float(stock_row[4]) if stock_row[4] else 0.0 - ) - - if stock_reserve_artstock > 0: - article["stock_reserve"] = stock_reserve_artstock - if stock_commande_artstock > 0: - article["stock_commande"] = stock_commande_artstock - - article["stock_disponible"] = ( - article["stock_reel"] - article["stock_reserve"] - ) - - logger.info( - f" Stock trouvé dans F_ARTSTOCK: {article['stock_reel']} unités" - ) - else: - logger.info( - f"Aucun stock trouvé dans F_ARTSTOCK pour {reference}" - ) - - except Exception as e: - logger.error(f" Erreur lecture F_ARTSTOCK pour {reference}: {e}") - - if article["famille_code"]: - try: - cursor.execute( - "SELECT FA_Intitule FROM F_FAMILLE WHERE FA_CodeFamille = ?", - (article["famille_code"],), - ) - famille_row = cursor.fetchone() - if famille_row: - article["famille_libelle"] = self._safe_strip( - famille_row[0] - ) - else: - article["famille_libelle"] = "" - except: - article["famille_libelle"] = "" - else: - article["famille_libelle"] = "" - - return article - + logger.info(f" → Enrichissement stock emplacements...") + + references = [a["reference"] for a in articles if a["reference"]] + if not references: + return articles + + placeholders = ",".join(["?"] * len(references)) + query = f""" + SELECT + AR_Ref, + DE_No, + DP_No, + AE_QteSto, + AE_QtePrepa, + AE_QteAControler, + cbCreation, + cbModification + FROM F_ARTSTOCKEMPL + WHERE AR_Ref IN ({placeholders}) + ORDER BY AR_Ref, DE_No, DP_No + """ + + cursor.execute(query, references) + rows = cursor.fetchall() + + emplacements_map = {} + for row in rows: + ref = self._safe_strip(row[0]) + if not ref: + continue + + if ref not in emplacements_map: + emplacements_map[ref] = [] + + emplacements_map[ref].append({ + "depot": self._safe_strip(row[1]), + "emplacement": self._safe_strip(row[2]), + "qte_stockee": float(row[3]) if row[3] else 0.0, + "qte_preparee": float(row[4]) if row[4] else 0.0, + "qte_a_controler": float(row[5]) if row[5] else 0.0, + "date_creation": row[6], + "date_modification": row[7], + }) + + for article in articles: + article["emplacements"] = emplacements_map.get(article["reference"], []) + article["nb_emplacements"] = len(article["emplacements"]) + + logger.info(f" ✓ {len(emplacements_map)} articles avec emplacements") + return articles + except Exception as e: - logger.error(f" Erreur SQL article {reference}: {e}") + logger.error(f" ✗ Erreur stock emplacements: {e}") + for article in articles: + article["emplacements"] = [] + article["nb_emplacements"] = 0 + return articles + + def _enrichir_gammes_articles(self, articles: List[Dict], cursor) -> List[Dict]: + """ + Enrichit avec les gammes (taille, couleur, etc.) + Structure: articles[i]["gammes"] = [{"numero": 1, "enumere": "001", "type": 0}, ...] + """ + try: + logger.info(f" → Enrichissement gammes articles...") + + references = [a["reference"] for a in articles if a["reference"]] + if not references: + return articles + + placeholders = ",".join(["?"] * len(references)) + query = f""" + SELECT + AR_Ref, + AG_No, + EG_Enumere, + AG_Type, + cbCreation, + cbModification + FROM F_ARTGAMME + WHERE AR_Ref IN ({placeholders}) + ORDER BY AR_Ref, AG_No, EG_Enumere + """ + + cursor.execute(query, references) + rows = cursor.fetchall() + + gammes_map = {} + for row in rows: + ref = self._safe_strip(row[0]) + if not ref: + continue + + if ref not in gammes_map: + gammes_map[ref] = [] + + gammes_map[ref].append({ + "numero_gamme": int(row[1]) if row[1] else 0, + "enumere": self._safe_strip(row[2]), + "type_gamme": int(row[3]) if row[3] else 0, + "date_creation": row[4], + "date_modification": row[5], + }) + + for article in articles: + article["gammes"] = gammes_map.get(article["reference"], []) + article["nb_gammes"] = len(article["gammes"]) + + logger.info(f" ✓ {len(gammes_map)} articles avec gammes") + return articles + + except Exception as e: + logger.error(f" ✗ Erreur gammes: {e}") + for article in articles: + article["gammes"] = [] + article["nb_gammes"] = 0 + return articles + + def _enrichir_tarifs_clients(self, articles: List[Dict], cursor) -> List[Dict]: + """ + Enrichit avec les tarifs spécifiques par client/catégorie tarifaire + Structure: articles[i]["tarifs_clients"] = [{"client": "CLI001", "prix": 125.5}, ...] + """ + try: + logger.info(f" → Enrichissement tarifs clients...") + + references = [a["reference"] for a in articles if a["reference"]] + if not references: + return articles + + placeholders = ",".join(["?"] * len(references)) + query = f""" + SELECT + AR_Ref, + AC_Categorie, + CT_Num, + AC_PrixVen, + AC_Coef, + AC_PrixTTC, + AC_Arrondi, + AC_QteMont, + EG_Champ, + AC_PrixDev, + AC_Devise, + AC_Remise, + AC_Calcul, + AC_TypeRem, + AC_RefClient, + AC_CoefNouv, + AC_PrixVenNouv, + AC_PrixDevNouv, + AC_RemiseNouv, + AC_DateApplication, + cbCreation, + cbModification + FROM F_ARTCLIENT + WHERE AR_Ref IN ({placeholders}) + ORDER BY AR_Ref, AC_Categorie, CT_Num + """ + + cursor.execute(query, references) + rows = cursor.fetchall() + + tarifs_map = {} + for row in rows: + ref = self._safe_strip(row[0]) + if not ref: + continue + + if ref not in tarifs_map: + tarifs_map[ref] = [] + + tarifs_map[ref].append({ + "categorie": int(row[1]) if row[1] else 0, + "client_num": self._safe_strip(row[2]), + "prix_vente": float(row[3]) if row[3] else 0.0, + "coefficient": float(row[4]) if row[4] else 0.0, + "prix_ttc": float(row[5]) if row[5] else 0.0, + "arrondi": float(row[6]) if row[6] else 0.0, + "qte_montant": float(row[7]) if row[7] else 0.0, + "enumere_gamme": int(row[8]) if row[8] else 0, + "prix_devise": float(row[9]) if row[9] else 0.0, + "devise": int(row[10]) if row[10] else 0, + "remise": float(row[11]) if row[11] else 0.0, + "mode_calcul": int(row[12]) if row[12] else 0, + "type_remise": int(row[13]) if row[13] else 0, + "ref_client": self._safe_strip(row[14]), + "coef_nouveau": float(row[15]) if row[15] else 0.0, + "prix_vente_nouveau": float(row[16]) if row[16] else 0.0, + "prix_devise_nouveau": float(row[17]) if row[17] else 0.0, + "remise_nouvelle": float(row[18]) if row[18] else 0.0, + "date_application": row[19], + "date_creation": row[20], + "date_modification": row[21], + }) + + for article in articles: + article["tarifs_clients"] = tarifs_map.get(article["reference"], []) + article["nb_tarifs_clients"] = len(article["tarifs_clients"]) + + logger.info(f" ✓ {len(tarifs_map)} articles avec tarifs clients") + return articles + + except Exception as e: + logger.error(f" ✗ Erreur tarifs clients: {e}") + for article in articles: + article["tarifs_clients"] = [] + article["nb_tarifs_clients"] = 0 + return articles + + def _enrichir_nomenclature(self, articles: List[Dict], cursor) -> List[Dict]: + """ + Enrichit avec la nomenclature de production (composants, opérations) + Structure: articles[i]["composants"] = [{"operation": "OP10", "ressource": "RES01"}, ...] + """ + try: + logger.info(f" → Enrichissement nomenclature...") + + references = [a["reference"] for a in articles if a["reference"]] + if not references: + return articles + + placeholders = ",".join(["?"] * len(references)) + query = f""" + SELECT + AR_Ref, + AT_Operation, + RP_Code, + AT_Temps, + AT_Type, + AT_Description, + AT_Ordre, + AG_No1Comp, + AG_No2Comp, + AT_TypeRessource, + AT_Chevauche, + AT_Demarre, + AT_OperationChevauche, + AT_ValeurChevauche, + AT_TypeChevauche, + cbCreation, + cbModification + FROM F_ARTCOMPO + WHERE AR_Ref IN ({placeholders}) + ORDER BY AR_Ref, AT_Ordre, AT_Operation + """ + + cursor.execute(query, references) + rows = cursor.fetchall() + + composants_map = {} + for row in rows: + ref = self._safe_strip(row[0]) + if not ref: + continue + + if ref not in composants_map: + composants_map[ref] = [] + + composants_map[ref].append({ + "operation": self._safe_strip(row[1]), + "code_ressource": self._safe_strip(row[2]), + "temps": float(row[3]) if row[3] else 0.0, + "type": int(row[4]) if row[4] else 0, + "description": self._safe_strip(row[5]), + "ordre": int(row[6]) if row[6] else 0, + "gamme_1_comp": int(row[7]) if row[7] else 0, + "gamme_2_comp": int(row[8]) if row[8] else 0, + "type_ressource": int(row[9]) if row[9] else 0, + "chevauche": int(row[10]) if row[10] else 0, + "demarre": int(row[11]) if row[11] else 0, + "operation_chevauche": self._safe_strip(row[12]), + "valeur_chevauche": float(row[13]) if row[13] else 0.0, + "type_chevauche": int(row[14]) if row[14] else 0, + "date_creation": row[15], + "date_modification": row[16], + }) + + for article in articles: + article["composants"] = composants_map.get(article["reference"], []) + article["nb_composants"] = len(article["composants"]) + + logger.info(f" ✓ {len(composants_map)} articles avec nomenclature") + return articles + + except Exception as e: + logger.error(f" ✗ Erreur nomenclature: {e}") + for article in articles: + article["composants"] = [] + article["nb_composants"] = 0 + return articles + + def _enrichir_compta_articles(self, articles: List[Dict], cursor) -> List[Dict]: + """ + Enrichit avec les comptes comptables spécifiques par article + Structure: articles[i]["compta_vente/achat/stock"] = {...} + """ + try: + logger.info(f" → Enrichissement comptabilité articles...") + + references = [a["reference"] for a in articles if a["reference"]] + if not references: + return articles + + placeholders = ",".join(["?"] * len(references)) + query = f""" + SELECT + AR_Ref, + ACP_Type, + ACP_Champ, + ACP_ComptaCPT_CompteG, + ACP_ComptaCPT_CompteA, + ACP_ComptaCPT_Taxe1, + ACP_ComptaCPT_Taxe2, + ACP_ComptaCPT_Taxe3, + ACP_ComptaCPT_Date1, + ACP_ComptaCPT_Date2, + ACP_ComptaCPT_Date3, + ACP_ComptaCPT_TaxeAnc1, + ACP_ComptaCPT_TaxeAnc2, + ACP_ComptaCPT_TaxeAnc3, + ACP_TypeFacture, + cbCreation, + cbModification + FROM F_ARTCOMPTA + WHERE AR_Ref IN ({placeholders}) + ORDER BY AR_Ref, ACP_Type, ACP_Champ + """ + + cursor.execute(query, references) + rows = cursor.fetchall() + + compta_map = {} + for row in rows: + ref = self._safe_strip(row[0]) + if not ref: + continue + + if ref not in compta_map: + compta_map[ref] = {"vente": [], "achat": [], "stock": []} + + type_compta = int(row[1]) if row[1] else 0 + type_key = {0: "vente", 1: "achat", 2: "stock"}.get(type_compta, "autre") + + compta_entry = { + "champ": int(row[2]) if row[2] else 0, + "compte_general": self._safe_strip(row[3]), + "compte_auxiliaire": self._safe_strip(row[4]), + "taxe_1": self._safe_strip(row[5]), + "taxe_2": self._safe_strip(row[6]), + "taxe_3": self._safe_strip(row[7]), + "taxe_date_1": row[8], + "taxe_date_2": row[9], + "taxe_date_3": row[10], + "taxe_anc_1": self._safe_strip(row[11]), + "taxe_anc_2": self._safe_strip(row[12]), + "taxe_anc_3": self._safe_strip(row[13]), + "type_facture": int(row[14]) if row[14] else 0, + "date_creation": row[15], + "date_modification": row[16], + } + + if type_key in compta_map[ref]: + compta_map[ref][type_key].append(compta_entry) + + for article in articles: + compta = compta_map.get(article["reference"], {"vente": [], "achat": [], "stock": []}) + article["compta_vente"] = compta["vente"] + article["compta_achat"] = compta["achat"] + article["compta_stock"] = compta["stock"] + + logger.info(f" ✓ {len(compta_map)} articles avec compta spécifique") + return articles + + except Exception as e: + logger.error(f" ✗ Erreur comptabilité articles: {e}") + for article in articles: + article["compta_vente"] = [] + article["compta_achat"] = [] + article["compta_stock"] = [] + return articles + + def _enrichir_fournisseurs_multiples(self, articles: List[Dict], cursor) -> List[Dict]: + """ + Enrichit avec TOUS les fournisseurs (pas seulement le principal) + Structure: articles[i]["fournisseurs"] = [{"num": "F001", "ref": "REF123", "prix": 45.5}, ...] + """ + try: + logger.info(f" → Enrichissement fournisseurs multiples...") + + references = [a["reference"] for a in articles if a["reference"]] + if not references: + return articles + + placeholders = ",".join(["?"] * len(references)) + query = f""" + SELECT + AR_Ref, + CT_Num, + AF_RefFourniss, + AF_PrixAch, + AF_Unite, + AF_Conversion, + AF_DelaiAppro, + AF_Garantie, + AF_Colisage, + AF_QteMini, + AF_QteMont, + EG_Champ, + AF_Principal, + AF_PrixDev, + AF_Devise, + AF_Remise, + AF_ConvDiv, + AF_TypeRem, + AF_CodeBarre, + AF_PrixAchNouv, + AF_PrixDevNouv, + AF_RemiseNouv, + AF_DateApplication, + cbCreation, + cbModification + FROM F_ARTFOURNISS + WHERE AR_Ref IN ({placeholders}) + ORDER BY AR_Ref, AF_Principal DESC, CT_Num + """ + + cursor.execute(query, references) + rows = cursor.fetchall() + + fournisseurs_map = {} + for row in rows: + ref = self._safe_strip(row[0]) + if not ref: + continue + + if ref not in fournisseurs_map: + fournisseurs_map[ref] = [] + + fournisseurs_map[ref].append({ + "fournisseur_num": self._safe_strip(row[1]), + "ref_fournisseur": self._safe_strip(row[2]), + "prix_achat": float(row[3]) if row[3] else 0.0, + "unite": self._safe_strip(row[4]), + "conversion": float(row[5]) if row[5] else 0.0, + "delai_appro": int(row[6]) if row[6] else 0, + "garantie": int(row[7]) if row[7] else 0, + "colisage": int(row[8]) if row[8] else 0, + "qte_mini": float(row[9]) if row[9] else 0.0, + "qte_montant": float(row[10]) if row[10] else 0.0, + "enumere_gamme": int(row[11]) if row[11] else 0, + "est_principal": bool(row[12]), + "prix_devise": float(row[13]) if row[13] else 0.0, + "devise": int(row[14]) if row[14] else 0, + "remise": float(row[15]) if row[15] else 0.0, + "conversion_devise": float(row[16]) if row[16] else 0.0, + "type_remise": int(row[17]) if row[17] else 0, + "code_barre_fournisseur": self._safe_strip(row[18]), + "prix_achat_nouveau": float(row[19]) if row[19] else 0.0, + "prix_devise_nouveau": float(row[20]) if row[20] else 0.0, + "remise_nouvelle": float(row[21]) if row[21] else 0.0, + "date_application": row[22], + "date_creation": row[23], + "date_modification": row[24], + }) + + for article in articles: + article["fournisseurs"] = fournisseurs_map.get(article["reference"], []) + article["nb_fournisseurs"] = len(article["fournisseurs"]) + + logger.info(f" ✓ {len(fournisseurs_map)} articles avec fournisseurs multiples") + return articles + + except Exception as e: + logger.error(f" ✗ Erreur fournisseurs multiples: {e}") + for article in articles: + article["fournisseurs"] = [] + article["nb_fournisseurs"] = 0 + return articles + + def _enrichir_depots_details(self, articles: List[Dict], cursor) -> List[Dict]: + """ + Enrichit les stocks avec les informations détaillées des dépôts + Ajoute le nom du dépôt à chaque ligne de stock + """ + try: + logger.info(f" → Enrichissement détails dépôts...") + + query = """ + SELECT + DE_No, + DE_Intitule, + DE_Code, + DE_Adresse, + DE_Complement, + DE_CodePostal, + DE_Ville, + DE_Contact, + DE_Principal, + DE_CatCompta, + DE_Region, + DE_Pays, + DE_EMail, + DE_Telephone, + DE_Telecopie, + DP_NoDefaut, + DE_Exclure + FROM F_DEPOT + """ + + cursor.execute(query) + rows = cursor.fetchall() + + depots_map = {} + for row in rows: + de_no = self._safe_strip(row[0]) + if not de_no: + continue + + depots_map[de_no] = { + "depot_num": de_no, + "depot_nom": self._safe_strip(row[1]), + "depot_code": self._safe_strip(row[2]), + "depot_adresse": self._safe_strip(row[3]), + "depot_complement": self._safe_strip(row[4]), + "depot_code_postal": self._safe_strip(row[5]), + "depot_ville": self._safe_strip(row[6]), + "depot_contact": self._safe_strip(row[7]), + "depot_est_principal": bool(row[8]), + "depot_categorie_compta": int(row[9]) if row[9] else 0, + "depot_region": self._safe_strip(row[10]), + "depot_pays": self._safe_strip(row[11]), + "depot_email": self._safe_strip(row[12]), + "depot_telephone": self._safe_strip(row[13]), + "depot_fax": self._safe_strip(row[14]), + "depot_emplacement_defaut": self._safe_strip(row[15]), + "depot_exclu": bool(row[16]), + } + + logger.info(f" → {len(depots_map)} dépôts chargés") + + for article in articles: + for empl in article.get("emplacements", []): + depot_num = empl.get("depot") + if depot_num and depot_num in depots_map: + empl.update(depots_map[depot_num]) + + logger.info(f" ✓ Emplacements enrichis avec détails dépôts") + return articles + + except Exception as e: + logger.error(f" ✗ Erreur détails dépôts: {e}") + return articles + + def _enrichir_emplacements_details(self, articles: List[Dict], cursor) -> List[Dict]: + """ + Enrichit les emplacements avec leurs détails (zone, type, etc.) + """ + try: + logger.info(f" → Enrichissement détails emplacements...") + + query = """ + SELECT + DE_No, + DP_No, + DP_Code, + DP_Intitule, + DP_Zone, + DP_Type + FROM F_DEPOTEMPL + """ + + cursor.execute(query) + rows = cursor.fetchall() + + emplacements_map = {} + for row in rows: + de_no = self._safe_strip(row[0]) + dp_no = self._safe_strip(row[1]) + + if not de_no or not dp_no: + continue + + key = f"{de_no}_{dp_no}" + emplacements_map[key] = { + "emplacement_code": self._safe_strip(row[2]), + "emplacement_libelle": self._safe_strip(row[3]), + "emplacement_zone": self._safe_strip(row[4]), + "emplacement_type": int(row[5]) if row[5] else 0, + } + + logger.info(f" → {len(emplacements_map)} emplacements détaillés chargés") + + for article in articles: + for empl in article.get("emplacements", []): + depot = empl.get("depot") + emplacement = empl.get("emplacement") + if depot and emplacement: + key = f"{depot}_{emplacement}" + if key in emplacements_map: + empl.update(emplacements_map[key]) + + logger.info(f" ✓ Emplacements enrichis avec détails") + return articles + + except Exception as e: + logger.error(f" ✗ Erreur détails emplacements: {e}") + return articles + + def _enrichir_gammes_enumeres(self, articles: List[Dict], cursor) -> List[Dict]: + """ + Enrichit les gammes avec leurs libellés depuis F_ENUMGAMME et P_GAMME + """ + try: + logger.info(f" → Enrichissement énumérés gammes...") + + query_pgamme = "SELECT G_Intitule, G_Type FROM P_GAMME ORDER BY G_Type" + cursor.execute(query_pgamme) + pgamme_rows = cursor.fetchall() + + gammes_config = {} + for idx, row in enumerate(pgamme_rows): + gammes_config[idx + 1] = { + "nom": self._safe_strip(row[0]), + "type": int(row[1]) if row[1] else 0, + } + + logger.info(f" → Configuration gammes: {gammes_config}") + + query_enum = """ + SELECT + EG_Champ, + EG_Ligne, + EG_Enumere, + EG_BorneSup + FROM F_ENUMGAMME + ORDER BY EG_Champ, EG_Ligne + """ + + cursor.execute(query_enum) + enum_rows = cursor.fetchall() + + enumeres_map = {} + for row in enum_rows: + champ = int(row[0]) if row[0] else 0 + enumere = self._safe_strip(row[2]) + + if not enumere: + continue + + key = f"{champ}_{enumere}" + enumeres_map[key] = { + "ligne": int(row[1]) if row[1] else 0, + "enumere": enumere, + "borne_sup": float(row[3]) if row[3] else 0.0, + "gamme_nom": gammes_config.get(champ, {}).get("nom", f"Gamme {champ}"), + } + + logger.info(f" → {len(enumeres_map)} énumérés chargés") + + for article in articles: + for gamme in article.get("gammes", []): + num_gamme = gamme.get("numero_gamme") + enumere = gamme.get("enumere") + + if num_gamme and enumere: + key = f"{num_gamme}_{enumere}" + if key in enumeres_map: + gamme.update(enumeres_map[key]) + + logger.info(f" ✓ Gammes enrichies avec énumérés") + return articles + + except Exception as e: + logger.error(f" ✗ Erreur énumérés gammes: {e}") + return articles + + def _enrichir_references_enumerees(self, articles: List[Dict], cursor) -> List[Dict]: + """ + Enrichit avec les références énumérées (articles avec gammes) + Structure: articles[i]["refs_enumerees"] = [{"gamme1": 1, "gamme2": 3, "ref": "ART-R-B"}, ...] + """ + try: + logger.info(f" → Enrichissement références énumérées...") + + references = [a["reference"] for a in articles if a["reference"]] + if not references: + return articles + + placeholders = ",".join(["?"] * len(references)) + query = f""" + SELECT + AR_Ref, + AG_No1, + AG_No2, + AE_Ref, + AE_PrixAch, + AE_CodeBarre, + AE_PrixAchNouv, + AE_EdiCode, + AE_Sommeil, + cbCreation, + cbModification + FROM F_ARTENUMREF + WHERE AR_Ref IN ({placeholders}) + ORDER BY AR_Ref, AG_No1, AG_No2 + """ + + cursor.execute(query, references) + rows = cursor.fetchall() + + refs_enum_map = {} + for row in rows: + ref = self._safe_strip(row[0]) + if not ref: + continue + + if ref not in refs_enum_map: + refs_enum_map[ref] = [] + + refs_enum_map[ref].append({ + "gamme_1": int(row[1]) if row[1] else 0, + "gamme_2": int(row[2]) if row[2] else 0, + "reference_enumeree": self._safe_strip(row[3]), + "prix_achat": float(row[4]) if row[4] else 0.0, + "code_barre": self._safe_strip(row[5]), + "prix_achat_nouveau": float(row[6]) if row[6] else 0.0, + "edi_code": self._safe_strip(row[7]), + "en_sommeil": bool(row[8]), + "date_creation": row[9], + "date_modification": row[10], + }) + + for article in articles: + article["refs_enumerees"] = refs_enum_map.get(article["reference"], []) + article["nb_refs_enumerees"] = len(article["refs_enumerees"]) + + logger.info(f" ✓ {len(refs_enum_map)} articles avec références énumérées") + return articles + + except Exception as e: + logger.error(f" ✗ Erreur références énumérées: {e}") + for article in articles: + article["refs_enumerees"] = [] + article["nb_refs_enumerees"] = 0 + return articles + + def _enrichir_medias_articles(self, articles: List[Dict], cursor) -> List[Dict]: + """ + Enrichit avec les médias attachés (photos, documents, etc.) + Structure: articles[i]["medias"] = [{"fichier": "photo.jpg", "type": "image/jpeg"}, ...] + """ + try: + logger.info(f" → Enrichissement médias articles...") + + references = [a["reference"] for a in articles if a["reference"]] + if not references: + return articles + + placeholders = ",".join(["?"] * len(references)) + query = f""" + SELECT + AR_Ref, + ME_Commentaire, + ME_Fichier, + ME_TypeMIME, + ME_Origine, + ME_GedId, + cbCreation, + cbModification + FROM F_ARTICLEMEDIA + WHERE AR_Ref IN ({placeholders}) + ORDER BY AR_Ref, cbCreation + """ + + cursor.execute(query, references) + rows = cursor.fetchall() + + medias_map = {} + for row in rows: + ref = self._safe_strip(row[0]) + if not ref: + continue + + if ref not in medias_map: + medias_map[ref] = [] + + medias_map[ref].append({ + "commentaire": self._safe_strip(row[1]), + "fichier": self._safe_strip(row[2]), + "type_mime": self._safe_strip(row[3]), + "origine": int(row[4]) if row[4] else 0, + "ged_id": self._safe_strip(row[5]), + "date_creation": row[6], + "date_modification": row[7], + }) + + for article in articles: + article["medias"] = medias_map.get(article["reference"], []) + article["nb_medias"] = len(article["medias"]) + + logger.info(f" ✓ {len(medias_map)} articles avec médias") + return articles + + except Exception as e: + logger.error(f" ✗ Erreur médias: {e}") + for article in articles: + article["medias"] = [] + article["nb_medias"] = 0 + return articles + + def _enrichir_prix_gammes(self, articles: List[Dict], cursor) -> List[Dict]: + """ + Enrichit avec les prix spécifiques par combinaison de gammes + Structure: articles[i]["prix_gammes"] = [{"gamme1": 1, "gamme2": 3, "prix_net": 125.5}, ...] + """ + try: + logger.info(f" → Enrichissement prix par gammes...") + + references = [a["reference"] for a in articles if a["reference"]] + if not references: + return articles + + placeholders = ",".join(["?"] * len(references)) + query = f""" + SELECT + AR_Ref, + AG_No1, + AG_No2, + AR_PUNet, + AR_CoutStd, + cbCreation, + cbModification + FROM F_ARTPRIX + WHERE AR_Ref IN ({placeholders}) + ORDER BY AR_Ref, AG_No1, AG_No2 + """ + + cursor.execute(query, references) + rows = cursor.fetchall() + + prix_gammes_map = {} + for row in rows: + ref = self._safe_strip(row[0]) + if not ref: + continue + + if ref not in prix_gammes_map: + prix_gammes_map[ref] = [] + + prix_gammes_map[ref].append({ + "gamme_1": int(row[1]) if row[1] else 0, + "gamme_2": int(row[2]) if row[2] else 0, + "prix_net": float(row[3]) if row[3] else 0.0, + "cout_standard": float(row[4]) if row[4] else 0.0, + "date_creation": row[5], + "date_modification": row[6], + }) + + for article in articles: + article["prix_gammes"] = prix_gammes_map.get(article["reference"], []) + article["nb_prix_gammes"] = len(article["prix_gammes"]) + + logger.info(f" ✓ {len(prix_gammes_map)} articles avec prix par gammes") + return articles + + except Exception as e: + logger.error(f" ✗ Erreur prix gammes: {e}") + for article in articles: + article["prix_gammes"] = [] + article["nb_prix_gammes"] = 0 + return articles + + def _enrichir_conditionnements(self, articles: List[Dict], cursor) -> List[Dict]: + """ + Enrichit avec les conditionnements disponibles + """ + try: + logger.info(f" → Enrichissement conditionnements...") + + query = """ + SELECT + EC_Champ, + EC_Enumere, + EC_Quantite, + EC_EdiCode + FROM F_ENUMCOND + ORDER BY EC_Champ, EC_Enumere + """ + + cursor.execute(query) + rows = cursor.fetchall() + + cond_map = {} + for row in rows: + champ = int(row[0]) if row[0] else 0 + enumere = self._safe_strip(row[1]) + + if not enumere: + continue + + key = f"{champ}_{enumere}" + cond_map[key] = { + "champ": champ, + "enumere": enumere, + "quantite": float(row[2]) if row[2] else 0.0, + "edi_code": self._safe_strip(row[3]), + } + + logger.info(f" → {len(cond_map)} conditionnements chargés") + + for article in articles: + conditionnement = article.get("conditionnement") + if conditionnement: + for key, cond_data in cond_map.items(): + if cond_data["enumere"] == conditionnement: + article["conditionnement_qte"] = cond_data["quantite"] + article["conditionnement_edi"] = cond_data["edi_code"] + break + + logger.info(f" ✓ Conditionnements enrichis") + return articles + + except Exception as e: + logger.error(f" ✗ Erreur conditionnements: {e}") + return articles + + def _mapper_article_depuis_row(self, row_data: Dict, colonnes_config: Dict) -> Dict: + """ + Mappe une ligne SQL vers un dictionnaire article normalisé + + Args: + row_data: Dictionnaire avec noms de colonnes SQL comme clés + colonnes_config: Mapping SQL -> noms normalisés + + Returns: + Dictionnaire article avec noms normalisés + """ + article = {} + + def get_val(sql_col, default=None, convert_type=None): + val = row_data.get(sql_col, default) + if val is None: + return default + + if convert_type == float: + return float(val) if val not in (None, "") else (default or 0.0) + elif convert_type == int: + return int(val) if val not in (None, "") else (default or 0) + elif convert_type == bool: + return bool(val) if val not in (None, "") else (default or False) + elif convert_type == str: + return self._safe_strip(val) + + return val + + article["reference"] = get_val("AR_Ref", convert_type=str) + article["designation"] = get_val("AR_Design", convert_type=str) + article["code_ean"] = get_val("AR_CodeBarre", convert_type=str) + article["code_barre"] = get_val("AR_CodeBarre", convert_type=str) + article["edi_code"] = get_val("AR_EdiCode", convert_type=str) + article["raccourci"] = get_val("AR_Raccourci", convert_type=str) + + article["prix_vente"] = get_val("AR_PrixVen", 0.0, float) + article["prix_achat"] = get_val("AR_PrixAch", 0.0, float) + article["coef"] = get_val("AR_Coef", 0.0, float) + article["prix_net"] = get_val("AR_PUNet", 0.0, float) + article["prix_achat_nouveau"] = get_val("AR_PrixAchNouv", 0.0, float) + article["coef_nouveau"] = get_val("AR_CoefNouv", 0.0, float) + article["prix_vente_nouveau"] = get_val("AR_PrixVenNouv", 0.0, float) + + date_app = get_val("AR_DateApplication") + article["date_application_prix"] = str(date_app) if date_app else None + + article["cout_standard"] = get_val("AR_CoutStd", 0.0, float) + + article["unite_vente"] = get_val("AR_UniteVen", convert_type=str) + article["unite_poids"] = get_val("AR_UnitePoids", convert_type=str) + article["poids_net"] = get_val("AR_PoidsNet", 0.0, float) + article["poids_brut"] = get_val("AR_PoidsBrut", 0.0, float) + + article["gamme_1"] = get_val("AR_Gamme1", convert_type=str) + article["gamme_2"] = get_val("AR_Gamme2", convert_type=str) + + type_val = get_val("AR_Type", 0, int) + article["type_article"] = type_val + article["type_article_libelle"] = self._get_type_article_libelle(type_val) + article["famille_code"] = get_val("FA_CodeFamille", convert_type=str) + article["nature"] = get_val("AR_Nature", 0, int) + article["garantie"] = get_val("AR_Garantie", 0, int) + article["code_fiscal"] = get_val("AR_CodeFiscal", convert_type=str) + article["pays"] = get_val("AR_Pays", convert_type=str) + + article["fournisseur_principal"] = get_val("CO_No", 0, int) + article["conditionnement"] = get_val("AR_Condition", convert_type=str) + article["nb_colis"] = get_val("AR_NbColis", 0, int) + article["prevision"] = get_val("AR_Prevision", False, bool) + + article["suivi_stock"] = get_val("AR_SuiviStock", False, bool) + article["nomenclature"] = get_val("AR_Nomencl", False, bool) + article["qte_composant"] = get_val("AR_QteComp", 0.0, float) + article["qte_operatoire"] = get_val("AR_QteOperatoire", 0.0, float) + + sommeil = get_val("AR_Sommeil", 0, int) + article["est_actif"] = (sommeil == 0) + article["en_sommeil"] = (sommeil == 1) + article["article_substitut"] = get_val("AR_Substitut", convert_type=str) + article["soumis_escompte"] = get_val("AR_Escompte", False, bool) + article["delai"] = get_val("AR_Delai", 0, int) + + article["stat_01"] = get_val("AR_Stat01", convert_type=str) + article["stat_02"] = get_val("AR_Stat02", convert_type=str) + article["stat_03"] = get_val("AR_Stat03", convert_type=str) + article["stat_04"] = get_val("AR_Stat04", convert_type=str) + article["stat_05"] = get_val("AR_Stat05", convert_type=str) + article["hors_statistique"] = get_val("AR_HorsStat", False, bool) + + article["categorie_1"] = get_val("CL_No1", 0, int) + article["categorie_2"] = get_val("CL_No2", 0, int) + article["categorie_3"] = get_val("CL_No3", 0, int) + article["categorie_4"] = get_val("CL_No4", 0, int) + + date_modif = get_val("AR_DateModif") + article["date_modification"] = str(date_modif) if date_modif else None + + article["vente_debit"] = get_val("AR_VteDebit", False, bool) + article["non_imprimable"] = get_val("AR_NotImp", False, bool) + article["transfere"] = get_val("AR_Transfere", False, bool) + article["publie"] = get_val("AR_Publie", False, bool) + article["contremarque"] = get_val("AR_Contremarque", False, bool) + article["fact_poids"] = get_val("AR_FactPoids", False, bool) + article["fact_forfait"] = get_val("AR_FactForfait", False, bool) + article["saisie_variable"] = get_val("AR_SaisieVar", False, bool) + article["fictif"] = get_val("AR_Fictif", False, bool) + article["sous_traitance"] = get_val("AR_SousTraitance", False, bool) + article["criticite"] = get_val("AR_Criticite", 0, int) + + article["reprise_code_defaut"] = get_val("RP_CodeDefaut", convert_type=str) + article["delai_fabrication"] = get_val("AR_DelaiFabrication", 0, int) + article["delai_peremption"] = get_val("AR_DelaiPeremption", 0, int) + article["delai_securite"] = get_val("AR_DelaiSecurite", 0, int) + article["type_lancement"] = get_val("AR_TypeLancement", 0, int) + article["cycle"] = get_val("AR_Cycle", 1, int) + + article["photo"] = get_val("AR_Photo", convert_type=str) + article["langue_1"] = get_val("AR_Langue1", convert_type=str) + article["langue_2"] = get_val("AR_Langue2", convert_type=str) + + article["frais_01_denomination"] = get_val("AR_Frais01FR_Denomination", convert_type=str) + article["frais_02_denomination"] = get_val("AR_Frais02FR_Denomination", convert_type=str) + article["frais_03_denomination"] = get_val("AR_Frais03FR_Denomination", convert_type=str) + + article["marque_commerciale"] = get_val("Marque commerciale", convert_type=str) + + objectif_val = get_val("Objectif / Qtés vendues") + if objectif_val is not None: + article["objectif_qtes_vendues"] = str(float(objectif_val)) if objectif_val not in ("", 0, 0.0) else None + else: + article["objectif_qtes_vendues"] = None + + pourcentage_val = get_val("Pourcentage teneur en or") + if pourcentage_val is not None: + article["pourcentage_or"] = str(float(pourcentage_val)) if pourcentage_val not in ("", 0, 0.0) else None + else: + article["pourcentage_or"] = None + + date_com = get_val("1ère commercialisation") + article["premiere_commercialisation"] = str(date_com) if date_com else None + + article["interdire_commande"] = get_val("AR_InterdireCommande", False, bool) + article["exclure"] = get_val("AR_Exclure", False, bool) + + article["stock_reel"] = 0.0 + article["stock_mini"] = 0.0 + article["stock_maxi"] = 0.0 + article["stock_reserve"] = 0.0 + article["stock_commande"] = 0.0 + article["stock_disponible"] = 0.0 + + article["famille_libelle"] = None + article["famille_type"] = None + article["famille_unite_vente"] = None + article["famille_coef"] = None + article["famille_suivi_stock"] = None + article["famille_garantie"] = None + article["famille_unite_poids"] = None + article["famille_delai"] = None + article["famille_nb_colis"] = None + article["famille_code_fiscal"] = None + article["famille_escompte"] = None + article["famille_centrale"] = None + article["famille_nature"] = None + article["famille_hors_stat"] = None + article["famille_pays"] = None + + article["fournisseur_nom"] = None + article["tva_code"] = None + article["tva_taux"] = None + + return article + + def _enrichir_stocks_articles(self, articles: List[Dict], cursor) -> List[Dict]: + """Enrichit les articles avec les données de stock depuis F_ARTSTOCK""" + try: + logger.info(f" → Enrichissement stocks pour {len(articles)} articles...") + + references = [a["reference"] for a in articles if a["reference"]] + + if not references: + return articles + + placeholders = ",".join(["?"] * len(references)) + stock_query = f""" + SELECT + AR_Ref, + SUM(ISNULL(AS_QteSto, 0)) as Stock_Total, + MIN(ISNULL(AS_QteMini, 0)) as Stock_Mini, + MAX(ISNULL(AS_QteMaxi, 0)) as Stock_Maxi, + SUM(ISNULL(AS_QteRes, 0)) as Stock_Reserve, + SUM(ISNULL(AS_QteCom, 0)) as Stock_Commande + FROM F_ARTSTOCK + WHERE AR_Ref IN ({placeholders}) + GROUP BY AR_Ref + """ + + cursor.execute(stock_query, references) + stock_rows = cursor.fetchall() + + stock_map = {} + for stock_row in stock_rows: + ref = self._safe_strip(stock_row[0]) + if ref: + stock_map[ref] = { + "stock_reel": float(stock_row[1]) if stock_row[1] else 0.0, + "stock_mini": float(stock_row[2]) if stock_row[2] else 0.0, + "stock_maxi": float(stock_row[3]) if stock_row[3] else 0.0, + "stock_reserve": float(stock_row[4]) if stock_row[4] else 0.0, + "stock_commande": float(stock_row[5]) if stock_row[5] else 0.0, + } + + logger.info(f" → {len(stock_map)} articles avec stock trouvés dans F_ARTSTOCK") + + for article in articles: + if article["reference"] in stock_map: + stock_data = stock_map[article["reference"]] + article.update(stock_data) + article["stock_disponible"] = ( + article["stock_reel"] - article["stock_reserve"] + ) + + return articles + + except Exception as e: + logger.error(f" ✗ Erreur enrichissement stocks: {e}", exc_info=True) + return articles + + def _enrichir_fournisseurs_articles(self, articles: List[Dict], cursor) -> List[Dict]: + """Enrichit les articles avec le nom du fournisseur principal""" + try: + logger.info(f" → Enrichissement fournisseurs...") + + nums_fournisseurs = list(set([ + a["fournisseur_principal"] for a in articles + if a.get("fournisseur_principal") and a["fournisseur_principal"] > 0 + ])) + + if not nums_fournisseurs: + logger.warning(" ⚠ Aucun numéro de fournisseur trouvé dans les articles") + for article in articles: + article["fournisseur_nom"] = None + return articles + + logger.info(f" → {len(nums_fournisseurs)} fournisseurs uniques à chercher") + logger.info(f" → Exemples CO_No : {nums_fournisseurs[:5]}") + + placeholders = ",".join(["?"] * len(nums_fournisseurs)) + fournisseur_query = f""" + SELECT + CT_Num, + CT_Intitule, + CT_Type + FROM F_COMPTET + WHERE CT_Num IN ({placeholders}) + AND CT_Type = 1 + """ + + cursor.execute(fournisseur_query, nums_fournisseurs) + fournisseur_rows = cursor.fetchall() + + logger.info(f" → {len(fournisseur_rows)} fournisseurs trouvés dans F_COMPTET") + + if len(fournisseur_rows) == 0: + logger.warning(f" ⚠ Aucun fournisseur trouvé pour CT_Type=1 et CT_Num IN {nums_fournisseurs[:5]}") + cursor.execute(f"SELECT CT_Num, CT_Type FROM F_COMPTET WHERE CT_Num IN ({placeholders})", nums_fournisseurs) + tous_types = cursor.fetchall() + if tous_types: + logger.info(f" → Trouvé {len(tous_types)} comptes (tous types) : {[(r[0], r[1]) for r in tous_types[:5]]}") + + fournisseur_map = {} + for fourn_row in fournisseur_rows: + num = int(fourn_row[0]) # CT_Num + nom = self._safe_strip(fourn_row[1]) # CT_Intitule + type_ct = int(fourn_row[2]) # CT_Type + fournisseur_map[num] = nom + logger.debug(f" → Fournisseur mappé : {num} = {nom} (Type={type_ct})") + + nb_enrichis = 0 + for article in articles: + num_fourn = article.get("fournisseur_principal") + if num_fourn and num_fourn in fournisseur_map: + article["fournisseur_nom"] = fournisseur_map[num_fourn] + nb_enrichis += 1 + else: + article["fournisseur_nom"] = None + + logger.info(f" ✓ {nb_enrichis} articles enrichis avec nom fournisseur") + + return articles + + except Exception as e: + logger.error(f" ✗ Erreur enrichissement fournisseurs: {e}", exc_info=True) + for article in articles: + article["fournisseur_nom"] = None + return articles + + def _enrichir_familles_articles(self, articles: List[Dict], cursor) -> List[Dict]: + """Enrichit les articles avec les informations de famille depuis F_FAMILLE""" + try: + logger.info(f" → Enrichissement familles pour {len(articles)} articles...") + + codes_familles_bruts = [ + a.get("famille_code") for a in articles + if a.get("famille_code") not in (None, "", " ") + ] + + if codes_familles_bruts: + logger.info(f" → Exemples de codes familles : {codes_familles_bruts[:5]}") + + codes_familles = list(set([ + str(code).strip() for code in codes_familles_bruts if code + ])) + + if not codes_familles: + logger.warning(" ⚠ Aucun code famille trouvé dans les articles") + for article in articles: + self._init_champs_famille_vides(article) + return articles + + logger.info(f" → {len(codes_familles)} codes famille uniques") + + cursor.execute("SELECT TOP 1 * FROM F_FAMILLE") + colonnes_disponibles = [column[0] for column in cursor.description] + + colonnes_souhaitees = [ + "FA_CodeFamille", + "FA_Intitule", + "FA_Type", + "FA_UniteVen", + "FA_Coef", + "FA_SuiviStock", + "FA_Garantie", + "FA_UnitePoids", + "FA_Delai", + "FA_NbColis", + "FA_CodeFiscal", + "FA_Escompte", + "FA_Central", + "FA_Nature", + "FA_HorsStat", + "FA_Pays", + "FA_VteDebit", + "FA_NotImp", + "FA_Contremarque", + "FA_FactPoids", + "FA_FactForfait", + "FA_Publie", + "FA_RacineRef", + "FA_RacineCB", + "FA_Raccourci", + "FA_SousTraitance", + "FA_Fictif", + "FA_Criticite", + ] + + colonnes_a_lire = [col for col in colonnes_souhaitees if col in colonnes_disponibles] + + if "FA_CodeFamille" not in colonnes_a_lire or "FA_Intitule" not in colonnes_a_lire: + logger.error(" ✗ Colonnes essentielles manquantes !") + return articles + + logger.info(f" → Colonnes disponibles : {len(colonnes_a_lire)}") + + colonnes_str = ", ".join(colonnes_a_lire) + placeholders = ",".join(["?"] * len(codes_familles)) + + famille_query = f""" + SELECT {colonnes_str} + FROM F_FAMILLE + WHERE FA_CodeFamille IN ({placeholders}) + """ + + cursor.execute(famille_query, codes_familles) + famille_rows = cursor.fetchall() + + logger.info(f" → {len(famille_rows)} familles trouvées") + + famille_map = {} + for fam_row in famille_rows: + famille_data = {} + for idx, col in enumerate(colonnes_a_lire): + famille_data[col] = fam_row[idx] + + code = self._safe_strip(famille_data.get("FA_CodeFamille")) + if not code: + continue + + famille_map[code] = { + "famille_libelle": self._safe_strip(famille_data.get("FA_Intitule")), + "famille_type": int(famille_data.get("FA_Type", 0) or 0), + "famille_unite_vente": self._safe_strip(famille_data.get("FA_UniteVen")), + "famille_coef": float(famille_data.get("FA_Coef", 0) or 0), + "famille_suivi_stock": bool(famille_data.get("FA_SuiviStock", 0)), + "famille_garantie": int(famille_data.get("FA_Garantie", 0) or 0), + "famille_unite_poids": self._safe_strip(famille_data.get("FA_UnitePoids")), + "famille_delai": int(famille_data.get("FA_Delai", 0) or 0), + "famille_nb_colis": int(famille_data.get("FA_NbColis", 0) or 0), + "famille_code_fiscal": self._safe_strip(famille_data.get("FA_CodeFiscal")), + "famille_escompte": bool(famille_data.get("FA_Escompte", 0)), + "famille_centrale": bool(famille_data.get("FA_Central", 0)), + "famille_nature": int(famille_data.get("FA_Nature", 0) or 0), + "famille_hors_stat": bool(famille_data.get("FA_HorsStat", 0)), + "famille_pays": self._safe_strip(famille_data.get("FA_Pays")), + } + + logger.info(f" → {len(famille_map)} familles mappées") + + nb_enrichis = 0 + for article in articles: + code_fam = str(article.get("famille_code", "")).strip() + + if code_fam and code_fam in famille_map: + article.update(famille_map[code_fam]) + nb_enrichis += 1 + else: + self._init_champs_famille_vides(article) + + logger.info(f" ✓ {nb_enrichis} articles enrichis avec infos famille") + + return articles + + except Exception as e: + logger.error(f" Erreur enrichissement familles: {e}", exc_info=True) + for article in articles: + self._init_champs_famille_vides(article) + return articles + + def _init_champs_famille_vides(self, article: Dict): + """Initialise les champs famille à None/0""" + article["famille_libelle"] = None + article["famille_type"] = None + article["famille_unite_vente"] = None + article["famille_coef"] = None + article["famille_suivi_stock"] = None + article["famille_garantie"] = None + article["famille_unite_poids"] = None + article["famille_delai"] = None + article["famille_nb_colis"] = None + article["famille_code_fiscal"] = None + article["famille_escompte"] = None + article["famille_centrale"] = None + article["famille_nature"] = None + article["famille_hors_stat"] = None + article["famille_pays"] = None + + def _enrichir_tva_articles(self, articles: List[Dict], cursor) -> List[Dict]: + """Enrichit les articles avec le taux de TVA""" + try: + logger.info(f" → Enrichissement TVA...") + + codes_tva = list(set([ + a["code_fiscal"] for a in articles + if a.get("code_fiscal") + ])) + + if not codes_tva: + for article in articles: + article["tva_code"] = None + article["tva_taux"] = None + return articles + + placeholders = ",".join(["?"] * len(codes_tva)) + tva_query = f""" + SELECT + TA_Code, + TA_Taux + FROM F_TAXE + WHERE TA_Code IN ({placeholders}) + """ + + cursor.execute(tva_query, codes_tva) + tva_rows = cursor.fetchall() + + tva_map = {} + for tva_row in tva_rows: + code = self._safe_strip(tva_row[0]) + tva_map[code] = float(tva_row[1]) if tva_row[1] else 0.0 + + logger.info(f" → {len(tva_map)} codes TVA trouvés") + + for article in articles: + code_tva = article.get("code_fiscal") + if code_tva and code_tva in tva_map: + article["tva_code"] = code_tva + article["tva_taux"] = tva_map[code_tva] + else: + article["tva_code"] = code_tva + article["tva_taux"] = None + + return articles + + except Exception as e: + logger.error(f" ✗ Erreur enrichissement TVA: {e}", exc_info=True) + for article in articles: + article["tva_code"] = article.get("code_fiscal") + article["tva_taux"] = None + return articles + + def _get_type_article_libelle(self, type_val: int) -> str: + """Retourne le libellé du type d'article""" + types = { + 0: "Article", + 1: "Prestation", + 2: "Divers / Frais", + 3: "Nomenclature" + } + return types.get(type_val, f"Type {type_val}") + + def _safe_strip(self, value) -> Optional[str]: + """Nettoie une valeur string en toute sécurité""" + if value is None: return None + if isinstance(value, str): + stripped = value.strip() + return stripped if stripped else None + return str(value).strip() or None + def _convertir_type_pour_sql(self, type_doc: int) -> int: """COM → SQL : 0, 10, 20, 30... → 0, 1, 2, 3...""" @@ -10333,237 +11698,424 @@ class SageConnector: raise RuntimeError(f"Erreur technique Sage : {error_message}") - def lister_toutes_familles( + def lister_toutes_familles( self, filtre: str = "", inclure_totaux: bool = True ) -> List[Dict]: + """Liste toutes les familles avec leurs comptes comptables et fournisseur principal""" try: with self._get_sql_connection() as conn: cursor = conn.cursor() - logger.info("[SQL] Détection des colonnes de F_FAMILLE...") + logger.info("[SQL] Chargement des familles avec F_FAMCOMPTA et F_FAMFOURNISS...") - cursor.execute("SELECT TOP 1 * FROM F_FAMILLE") - colonnes_disponibles = [column[0] for column in cursor.description] - - logger.info(f"[SQL] Colonnes trouvées : {len(colonnes_disponibles)}") - - # Colonnes organisées par catégorie - colonnes_souhaitees = [ - # Identification - "FA_CodeFamille", - "FA_Intitule", - "FA_Type", - - # Vente et stock - "FA_UniteVen", - "FA_Coef", - "FA_SuiviStock", - "FA_Garantie", - "FA_UnitePoids", - "FA_Delai", - "FA_NbColis", - - # Comptabilité - "CG_NumAch", - "CG_NumVte", - "FA_CodeFiscal", - "FA_Escompte", - - # Organisation et classification - "FA_Central", - "FA_Nature", - "CL_No1", - "CL_No2", - "CL_No3", - "CL_No4", - - # Statistiques - "FA_Stat01", - "FA_Stat02", - "FA_Stat03", - "FA_Stat04", - "FA_Stat05", - "FA_HorsStat", - - # Paramètres commerciaux - "FA_Pays", - "FA_VteDebit", - "FA_NotImp", - "FA_Contremarque", - "FA_FactPoids", - "FA_FactForfait", - "FA_Publie", - - # Références et codes - "FA_RacineRef", - "FA_RacineCB", - "FA_Raccourci", - - # Gestion - "FA_SousTraitance", - "FA_Fictif", - "FA_Criticite" - ] - - colonnes_a_lire = [ - col for col in colonnes_souhaitees if col in colonnes_disponibles - ] - - if not colonnes_a_lire: - colonnes_a_lire = colonnes_disponibles - - logger.info(f"[SQL] Colonnes sélectionnées : {len(colonnes_a_lire)}") - - colonnes_str = ", ".join([f"f.{col}" for col in colonnes_a_lire]) - - # Requête avec LEFT JOIN pour compter les articles par famille - query = f""" - SELECT {colonnes_str}, - ISNULL(COUNT(a.AR_Ref), 0) as nb_articles + query = """ + SELECT + -- F_FAMILLE - Identification + f.FA_CodeFamille, + f.FA_Type, + f.FA_Intitule, + f.FA_UniteVen, + f.FA_Coef, + f.FA_SuiviStock, + f.FA_Garantie, + f.FA_Central, + + -- F_FAMILLE - Statistiques + f.FA_Stat01, + f.FA_Stat02, + f.FA_Stat03, + f.FA_Stat04, + f.FA_Stat05, + + -- F_FAMILLE - Fiscal et gestion + f.FA_CodeFiscal, + f.FA_Pays, + f.FA_UnitePoids, + f.FA_Escompte, + f.FA_Delai, + f.FA_HorsStat, + f.FA_VteDebit, + f.FA_NotImp, + + -- F_FAMILLE - Frais annexes (3 frais avec 3 remises chacun) + f.FA_Frais01FR_Denomination, + f.FA_Frais01FR_Rem01REM_Valeur, + f.FA_Frais01FR_Rem01REM_Type, + f.FA_Frais01FR_Rem02REM_Valeur, + f.FA_Frais01FR_Rem02REM_Type, + f.FA_Frais01FR_Rem03REM_Valeur, + f.FA_Frais01FR_Rem03REM_Type, + f.FA_Frais02FR_Denomination, + f.FA_Frais02FR_Rem01REM_Valeur, + f.FA_Frais02FR_Rem01REM_Type, + f.FA_Frais02FR_Rem02REM_Valeur, + f.FA_Frais02FR_Rem02REM_Type, + f.FA_Frais02FR_Rem03REM_Valeur, + f.FA_Frais02FR_Rem03REM_Type, + f.FA_Frais03FR_Denomination, + f.FA_Frais03FR_Rem01REM_Valeur, + f.FA_Frais03FR_Rem01REM_Type, + f.FA_Frais03FR_Rem02REM_Valeur, + f.FA_Frais03FR_Rem02REM_Type, + f.FA_Frais03FR_Rem03REM_Valeur, + f.FA_Frais03FR_Rem03REM_Type, + + -- F_FAMILLE - Options diverses + f.FA_Contremarque, + f.FA_FactPoids, + f.FA_FactForfait, + f.FA_Publie, + f.FA_RacineRef, + f.FA_RacineCB, + + -- F_FAMILLE - Catégories + f.CL_No1, + f.CL_No2, + f.CL_No3, + f.CL_No4, + + -- F_FAMILLE - Gestion avancée + f.FA_Nature, + f.FA_NbColis, + f.FA_SousTraitance, + f.FA_Fictif, + f.FA_Criticite, + + -- F_FAMILLE - Métadonnées système + f.cbMarq, + f.cbCreateur, + f.cbModification, + f.cbCreation, + f.cbCreationUser, + + -- F_FAMCOMPTA Vente (FCP_Type = 0) + vte.FCP_ComptaCPT_CompteG, + vte.FCP_ComptaCPT_CompteA, + vte.FCP_ComptaCPT_Taxe1, + vte.FCP_ComptaCPT_Taxe2, + vte.FCP_ComptaCPT_Taxe3, + vte.FCP_ComptaCPT_Date1, + vte.FCP_ComptaCPT_Date2, + vte.FCP_ComptaCPT_Date3, + vte.FCP_TypeFacture, + + -- F_FAMCOMPTA Achat (FCP_Type = 1) + ach.FCP_ComptaCPT_CompteG, + ach.FCP_ComptaCPT_CompteA, + ach.FCP_ComptaCPT_Taxe1, + ach.FCP_ComptaCPT_Taxe2, + ach.FCP_ComptaCPT_Taxe3, + ach.FCP_ComptaCPT_Date1, + ach.FCP_ComptaCPT_Date2, + ach.FCP_ComptaCPT_Date3, + ach.FCP_TypeFacture, + + -- F_FAMCOMPTA Stock (FCP_Type = 2) + sto.FCP_ComptaCPT_CompteG, + sto.FCP_ComptaCPT_CompteA, + + -- F_FAMFOURNISS (fournisseur principal FF_Principal=1) + ff.CT_Num, + ff.FF_Unite, + ff.FF_Conversion, + ff.FF_DelaiAppro, + ff.FF_Garantie, + ff.FF_Colisage, + ff.FF_QteMini, + ff.FF_QteMont, + ff.EG_Champ, + ff.FF_Devise, + ff.FF_Remise, + ff.FF_ConvDiv, + ff.FF_TypeRem, + + -- Nombre d'articles + ISNULL(COUNT(DISTINCT a.AR_Ref), 0) as nb_articles + FROM F_FAMILLE f + + -- Jointures comptables + LEFT JOIN F_FAMCOMPTA vte ON f.FA_CodeFamille = vte.FA_CodeFamille + AND vte.FCP_Type = 0 -- Vente + AND vte.FCP_Champ = 1 -- Compte principal + + LEFT JOIN F_FAMCOMPTA ach ON f.FA_CodeFamille = ach.FA_CodeFamille + AND ach.FCP_Type = 1 -- Achat + AND ach.FCP_Champ = 1 + + LEFT JOIN F_FAMCOMPTA sto ON f.FA_CodeFamille = sto.FA_CodeFamille + AND sto.FCP_Type = 2 -- Stock + AND sto.FCP_Champ = 1 + + -- Fournisseur principal + LEFT JOIN F_FAMFOURNISS ff ON f.FA_CodeFamille = ff.FA_CodeFamille + AND ff.FF_Principal = 1 + + -- Nombre d'articles LEFT JOIN F_ARTICLE a ON f.FA_CodeFamille = a.FA_CodeFamille + WHERE 1=1 """ params = [] - # Filtrage par FA_Type (si demandé) - if "FA_Type" in colonnes_disponibles and not inclure_totaux: - query += " AND f.FA_Type = 0" # Seulement Détail + if not inclure_totaux: + query += " AND f.FA_Type = 0" logger.info("[SQL] Filtre : FA_Type = 0 (Détail uniquement)") - else: - logger.info("[SQL] Filtre : TOUS les types (Détail + Total)") - # Filtrage par texte if filtre: - conditions_filtre = [] + query += """ + AND ( + f.FA_CodeFamille LIKE ? + OR f.FA_Intitule LIKE ? + ) + """ + params.extend([f"%{filtre}%", f"%{filtre}%"]) - if "FA_CodeFamille" in colonnes_a_lire: - conditions_filtre.append("f.FA_CodeFamille LIKE ?") - params.append(f"%{filtre}%") - - if "FA_Intitule" in colonnes_a_lire: - conditions_filtre.append("f.FA_Intitule LIKE ?") - params.append(f"%{filtre}%") - - if conditions_filtre: - query += " AND (" + " OR ".join(conditions_filtre) + ")" - - # GROUP BY pour le COUNT - query += f" GROUP BY {colonnes_str}" - - # ORDER BY - if "FA_Intitule" in colonnes_a_lire: - query += " ORDER BY f.FA_Intitule" - elif "FA_CodeFamille" in colonnes_a_lire: - query += " ORDER BY f.FA_CodeFamille" + query += """ + GROUP BY + f.FA_CodeFamille, f.FA_Type, f.FA_Intitule, f.FA_UniteVen, f.FA_Coef, + f.FA_SuiviStock, f.FA_Garantie, f.FA_Central, + f.FA_Stat01, f.FA_Stat02, f.FA_Stat03, f.FA_Stat04, f.FA_Stat05, + f.FA_CodeFiscal, f.FA_Pays, f.FA_UnitePoids, f.FA_Escompte, f.FA_Delai, + f.FA_HorsStat, f.FA_VteDebit, f.FA_NotImp, + f.FA_Frais01FR_Denomination, f.FA_Frais01FR_Rem01REM_Valeur, f.FA_Frais01FR_Rem01REM_Type, + f.FA_Frais01FR_Rem02REM_Valeur, f.FA_Frais01FR_Rem02REM_Type, + f.FA_Frais01FR_Rem03REM_Valeur, f.FA_Frais01FR_Rem03REM_Type, + f.FA_Frais02FR_Denomination, f.FA_Frais02FR_Rem01REM_Valeur, f.FA_Frais02FR_Rem01REM_Type, + f.FA_Frais02FR_Rem02REM_Valeur, f.FA_Frais02FR_Rem02REM_Type, + f.FA_Frais02FR_Rem03REM_Valeur, f.FA_Frais02FR_Rem03REM_Type, + f.FA_Frais03FR_Denomination, f.FA_Frais03FR_Rem01REM_Valeur, f.FA_Frais03FR_Rem01REM_Type, + f.FA_Frais03FR_Rem02REM_Valeur, f.FA_Frais03FR_Rem02REM_Type, + f.FA_Frais03FR_Rem03REM_Valeur, f.FA_Frais03FR_Rem03REM_Type, + f.FA_Contremarque, f.FA_FactPoids, f.FA_FactForfait, f.FA_Publie, + f.FA_RacineRef, f.FA_RacineCB, + f.CL_No1, f.CL_No2, f.CL_No3, f.CL_No4, + f.FA_Nature, f.FA_NbColis, f.FA_SousTraitance, f.FA_Fictif, f.FA_Criticite, + f.cbMarq, f.cbCreateur, f.cbModification, f.cbCreation, f.cbCreationUser, + vte.FCP_ComptaCPT_CompteG, vte.FCP_ComptaCPT_CompteA, + vte.FCP_ComptaCPT_Taxe1, vte.FCP_ComptaCPT_Taxe2, vte.FCP_ComptaCPT_Taxe3, + vte.FCP_ComptaCPT_Date1, vte.FCP_ComptaCPT_Date2, vte.FCP_ComptaCPT_Date3, + vte.FCP_TypeFacture, + ach.FCP_ComptaCPT_CompteG, ach.FCP_ComptaCPT_CompteA, + ach.FCP_ComptaCPT_Taxe1, ach.FCP_ComptaCPT_Taxe2, ach.FCP_ComptaCPT_Taxe3, + ach.FCP_ComptaCPT_Date1, ach.FCP_ComptaCPT_Date2, ach.FCP_ComptaCPT_Date3, + ach.FCP_TypeFacture, + sto.FCP_ComptaCPT_CompteG, sto.FCP_ComptaCPT_CompteA, + ff.CT_Num, ff.FF_Unite, ff.FF_Conversion, ff.FF_DelaiAppro, + ff.FF_Garantie, ff.FF_Colisage, ff.FF_QteMini, ff.FF_QteMont, + ff.EG_Champ, ff.FF_Devise, ff.FF_Remise, ff.FF_ConvDiv, ff.FF_TypeRem + ORDER BY f.FA_Intitule + """ cursor.execute(query, params) rows = cursor.fetchall() + def to_str(val): + """Convertit en string, gère None et int""" + if val is None: + return "" + return str(val).strip() if isinstance(val, str) else str(val) + + def to_float(val): + """Convertit en float, gère None""" + if val is None or val == "": + return 0.0 + try: + return float(val) + except (ValueError, TypeError): + return 0.0 + + def to_int(val): + """Convertit en int, gère None""" + if val is None or val == "": + return 0 + try: + return int(val) + except (ValueError, TypeError): + return 0 + + def to_bool(val): + """Convertit en bool""" + if val is None: + return False + if isinstance(val, bool): + return val + if isinstance(val, int): + return val != 0 + return bool(val) + familles = [] - + for row in rows: - famille = {} - - # Récupération des colonnes (sauf la dernière qui est nb_articles) - for idx, colonne in enumerate(colonnes_a_lire): - valeur = row[idx] - - if isinstance(valeur, str): - valeur = valeur.strip() - - famille[colonne] = valeur - - # Récupération du nb_articles (dernière colonne) - famille["nb_articles"] = row[-1] - - # Champs de base (compatibilité) - if "FA_CodeFamille" in famille: - famille["code"] = famille["FA_CodeFamille"] - - if "FA_Intitule" in famille: - famille["intitule"] = famille["FA_Intitule"] - - if "FA_Type" in famille: - type_val = famille["FA_Type"] - famille["type"] = type_val - famille["type_libelle"] = "Total" if type_val == 1 else "Détail" - famille["est_total"] = type_val == 1 - else: - famille["type"] = 0 - famille["type_libelle"] = "Détail" - famille["est_total"] = False - - # Vente et unités - famille["unite_vente"] = str(famille.get("FA_UniteVen", "")) - famille["unite_poids"] = str(famille.get("FA_UnitePoids", "")) - famille["coef"] = ( - float(famille.get("FA_Coef", 0.0)) - if famille.get("FA_Coef") is not None - else 0.0 - ) - - # Stock et logistique - famille["suivi_stock"] = bool(famille.get("FA_SuiviStock", 0)) - famille["garantie"] = int(famille.get("FA_Garantie", 0)) - famille["delai"] = int(famille.get("FA_Delai", 0)) - famille["nb_colis"] = int(famille.get("FA_NbColis", 0)) - - # Comptabilité - famille["compte_achat"] = famille.get("CG_NumAch", "") - famille["compte_vente"] = famille.get("CG_NumVte", "") - famille["code_fiscal"] = famille.get("FA_CodeFiscal", "") - famille["escompte"] = bool(famille.get("FA_Escompte", 0)) - - # Organisation - famille["est_centrale"] = bool(famille.get("FA_Central", 0)) - famille["nature"] = famille.get("FA_Nature", 0) - famille["pays"] = famille.get("FA_Pays", "") - - # Classifications - famille["categorie_1"] = famille.get("CL_No1", 0) - famille["categorie_2"] = famille.get("CL_No2", 0) - famille["categorie_3"] = famille.get("CL_No3", 0) - famille["categorie_4"] = famille.get("CL_No4", 0) - - # Statistiques - famille["stat_01"] = famille.get("FA_Stat01", "") - famille["stat_02"] = famille.get("FA_Stat02", "") - famille["stat_03"] = famille.get("FA_Stat03", "") - famille["stat_04"] = famille.get("FA_Stat04", "") - famille["stat_05"] = famille.get("FA_Stat05", "") - famille["hors_statistique"] = bool(famille.get("FA_HorsStat", 0)) - - # Paramètres commerciaux - famille["vente_debit"] = bool(famille.get("FA_VteDebit", 0)) - famille["non_imprimable"] = bool(famille.get("FA_NotImp", 0)) - famille["contremarque"] = bool(famille.get("FA_Contremarque", 0)) - famille["fact_poids"] = bool(famille.get("FA_FactPoids", 0)) - famille["fact_forfait"] = bool(famille.get("FA_FactForfait", 0)) - famille["publie"] = bool(famille.get("FA_Publie", 0)) - - # Références - famille["racine_reference"] = famille.get("FA_RacineRef", "") - famille["racine_code_barre"] = famille.get("FA_RacineCB", "") - famille["raccourci"] = famille.get("FA_Raccourci", "") - - # Gestion - famille["sous_traitance"] = bool(famille.get("FA_SousTraitance", 0)) - famille["fictif"] = bool(famille.get("FA_Fictif", 0)) - famille["criticite"] = int(famille.get("FA_Criticite", 0)) - + idx = 0 + + famille = { + "code": to_str(row[idx]), + "type": to_int(row[idx+1]), + "intitule": to_str(row[idx+2]), + "unite_vente": to_str(row[idx+3]), + "coef": to_float(row[idx+4]), + "suivi_stock": to_bool(row[idx+5]), + "garantie": to_int(row[idx+6]), + "est_centrale": to_bool(row[idx+7]), + } + idx += 8 + + famille.update({ + "stat_01": to_str(row[idx]), + "stat_02": to_str(row[idx+1]), + "stat_03": to_str(row[idx+2]), + "stat_04": to_str(row[idx+3]), + "stat_05": to_str(row[idx+4]), + }) + idx += 5 + + famille.update({ + "code_fiscal": to_str(row[idx]), + "pays": to_str(row[idx+1]), + "unite_poids": to_str(row[idx+2]), + "escompte": to_bool(row[idx+3]), + "delai": to_int(row[idx+4]), + "hors_statistique": to_bool(row[idx+5]), + "vente_debit": to_bool(row[idx+6]), + "non_imprimable": to_bool(row[idx+7]), + }) + idx += 8 + + famille.update({ + "frais_01_libelle": to_str(row[idx]), + "frais_01_remise_1_valeur": to_float(row[idx+1]), + "frais_01_remise_1_type": to_int(row[idx+2]), + "frais_01_remise_2_valeur": to_float(row[idx+3]), + "frais_01_remise_2_type": to_int(row[idx+4]), + "frais_01_remise_3_valeur": to_float(row[idx+5]), + "frais_01_remise_3_type": to_int(row[idx+6]), + "frais_02_libelle": to_str(row[idx+7]), + "frais_02_remise_1_valeur": to_float(row[idx+8]), + "frais_02_remise_1_type": to_int(row[idx+9]), + "frais_02_remise_2_valeur": to_float(row[idx+10]), + "frais_02_remise_2_type": to_int(row[idx+11]), + "frais_02_remise_3_valeur": to_float(row[idx+12]), + "frais_02_remise_3_type": to_int(row[idx+13]), + "frais_03_libelle": to_str(row[idx+14]), + "frais_03_remise_1_valeur": to_float(row[idx+15]), + "frais_03_remise_1_type": to_int(row[idx+16]), + "frais_03_remise_2_valeur": to_float(row[idx+17]), + "frais_03_remise_2_type": to_int(row[idx+18]), + "frais_03_remise_3_valeur": to_float(row[idx+19]), + "frais_03_remise_3_type": to_int(row[idx+20]), + }) + idx += 21 + + famille.update({ + "contremarque": to_bool(row[idx]), + "fact_poids": to_bool(row[idx+1]), + "fact_forfait": to_bool(row[idx+2]), + "publie": to_bool(row[idx+3]), + "racine_reference": to_str(row[idx+4]), + "racine_code_barre": to_str(row[idx+5]), + }) + idx += 6 + + famille.update({ + "categorie_1": to_int(row[idx]), + "categorie_2": to_int(row[idx+1]), + "categorie_3": to_int(row[idx+2]), + "categorie_4": to_int(row[idx+3]), + }) + idx += 4 + + famille.update({ + "nature": to_int(row[idx]), + "nb_colis": to_int(row[idx+1]), + "sous_traitance": to_bool(row[idx+2]), + "fictif": to_bool(row[idx+3]), + "criticite": to_int(row[idx+4]), + }) + idx += 5 + + famille.update({ + "cb_marq": to_int(row[idx]), + "cb_createur": to_str(row[idx+1]), + "cb_modification": row[idx+2], # datetime - garder tel quel + "cb_creation": row[idx+3], # datetime - garder tel quel + "cb_creation_user": to_str(row[idx+4]), + }) + idx += 5 + + famille.update({ + "compte_vente": to_str(row[idx]), + "compte_auxiliaire_vente": to_str(row[idx+1]), + "tva_vente_1": to_str(row[idx+2]), + "tva_vente_2": to_str(row[idx+3]), + "tva_vente_3": to_str(row[idx+4]), + "tva_vente_date_1": row[idx+5], # datetime + "tva_vente_date_2": row[idx+6], + "tva_vente_date_3": row[idx+7], + "type_facture_vente": to_int(row[idx+8]), + }) + idx += 9 + + famille.update({ + "compte_achat": to_str(row[idx]), + "compte_auxiliaire_achat": to_str(row[idx+1]), + "tva_achat_1": to_str(row[idx+2]), + "tva_achat_2": to_str(row[idx+3]), + "tva_achat_3": to_str(row[idx+4]), + "tva_achat_date_1": row[idx+5], + "tva_achat_date_2": row[idx+6], + "tva_achat_date_3": row[idx+7], + "type_facture_achat": to_int(row[idx+8]), + }) + idx += 9 + + famille.update({ + "compte_stock": to_str(row[idx]), + "compte_auxiliaire_stock": to_str(row[idx+1]), + }) + idx += 2 + + famille.update({ + "fournisseur_principal": to_str(row[idx]), + "fournisseur_unite": to_str(row[idx+1]), + "fournisseur_conversion": to_float(row[idx+2]), + "fournisseur_delai_appro": to_int(row[idx+3]), + "fournisseur_garantie": to_int(row[idx+4]), + "fournisseur_colisage": to_int(row[idx+5]), + "fournisseur_qte_mini": to_float(row[idx+6]), + "fournisseur_qte_mont": to_float(row[idx+7]), + "fournisseur_enumere_gamme": to_int(row[idx+8]), + "fournisseur_devise": to_int(row[idx+9]), + "fournisseur_remise": to_float(row[idx+10]), + "fournisseur_conv_div": to_float(row[idx+11]), + "fournisseur_type_remise": to_int(row[idx+12]), + }) + idx += 13 + + famille["nb_articles"] = to_int(row[idx]) + + famille["type_libelle"] = "Total" if famille["type"] == 1 else "Détail" + famille["est_total"] = famille["type"] == 1 + famille["est_detail"] = famille["type"] == 0 + + famille["FA_CodeFamille"] = famille["code"] + famille["FA_Intitule"] = famille["intitule"] + famille["FA_Type"] = famille["type"] + famille["CG_NumVte"] = famille["compte_vente"] + famille["CG_NumAch"] = famille["compte_achat"] + familles.append(famille) type_msg = "DÉTAIL uniquement" if not inclure_totaux else "TOUS types" - logger.info(f"SQL: {len(familles)} familles chargées ({type_msg})") + logger.info(f"✓ {len(familles)} familles chargées ({type_msg})") return familles except Exception as e: logger.error(f"Erreur SQL familles: {e}", exc_info=True) raise RuntimeError(f"Erreur lecture familles: {str(e)}") - + def lire_famille(self, code: str) -> Dict: """ @@ -10581,20 +12133,16 @@ class SageConnector: logger.info(f"[SQL] Lecture famille : {code}") - # Détection des colonnes disponibles cursor.execute("SELECT TOP 1 * FROM F_FAMILLE") colonnes_disponibles = [column[0] for column in cursor.description] logger.info(f"[SQL] Colonnes trouvées : {len(colonnes_disponibles)}") - # Colonnes organisées par catégorie (IDENTIQUE à lister_toutes_familles) colonnes_souhaitees = [ - # Identification "FA_CodeFamille", "FA_Intitule", "FA_Type", - # Vente et stock "FA_UniteVen", "FA_Coef", "FA_SuiviStock", @@ -10603,13 +12151,11 @@ class SageConnector: "FA_Delai", "FA_NbColis", - # Comptabilité "CG_NumAch", "CG_NumVte", "FA_CodeFiscal", "FA_Escompte", - # Organisation et classification "FA_Central", "FA_Nature", "CL_No1", @@ -10617,7 +12163,6 @@ class SageConnector: "CL_No3", "CL_No4", - # Statistiques "FA_Stat01", "FA_Stat02", "FA_Stat03", @@ -10625,7 +12170,6 @@ class SageConnector: "FA_Stat05", "FA_HorsStat", - # Paramètres commerciaux "FA_Pays", "FA_VteDebit", "FA_NotImp", @@ -10634,12 +12178,10 @@ class SageConnector: "FA_FactForfait", "FA_Publie", - # Références et codes "FA_RacineRef", "FA_RacineCB", "FA_Raccourci", - # Gestion "FA_SousTraitance", "FA_Fictif", "FA_Criticite" @@ -10656,7 +12198,6 @@ class SageConnector: colonnes_str = ", ".join([f"f.{col}" for col in colonnes_a_lire]) - # Requête avec LEFT JOIN pour compter les articles (IDENTIQUE à lister_toutes_familles) query = f""" SELECT {colonnes_str}, ISNULL(COUNT(a.AR_Ref), 0) as nb_articles @@ -10672,7 +12213,6 @@ class SageConnector: if not row: raise ValueError(f"Famille '{code}' introuvable dans Sage") - # Construction du dictionnaire (IDENTIQUE à lister_toutes_familles) famille = {} for idx, colonne in enumerate(colonnes_a_lire): @@ -10683,10 +12223,8 @@ class SageConnector: famille[colonne] = valeur - # Récupération du nb_articles (dernière colonne) famille["nb_articles"] = row[-1] - # Champs de base (compatibilité) if "FA_CodeFamille" in famille: famille["code"] = famille["FA_CodeFamille"] @@ -10703,7 +12241,6 @@ class SageConnector: famille["type_libelle"] = "Détail" famille["est_total"] = False - # Vente et unités famille["unite_vente"] = str(famille.get("FA_UniteVen", "")) famille["unite_poids"] = str(famille.get("FA_UnitePoids", "")) famille["coef"] = ( @@ -10712,30 +12249,25 @@ class SageConnector: else 0.0 ) - # Stock et logistique famille["suivi_stock"] = bool(famille.get("FA_SuiviStock", 0)) famille["garantie"] = int(famille.get("FA_Garantie", 0)) famille["delai"] = int(famille.get("FA_Delai", 0)) famille["nb_colis"] = int(famille.get("FA_NbColis", 0)) - # Comptabilité famille["compte_achat"] = famille.get("CG_NumAch", "") famille["compte_vente"] = famille.get("CG_NumVte", "") famille["code_fiscal"] = famille.get("FA_CodeFiscal", "") famille["escompte"] = bool(famille.get("FA_Escompte", 0)) - # Organisation famille["est_centrale"] = bool(famille.get("FA_Central", 0)) famille["nature"] = famille.get("FA_Nature", 0) famille["pays"] = famille.get("FA_Pays", "") - # Classifications famille["categorie_1"] = famille.get("CL_No1", 0) famille["categorie_2"] = famille.get("CL_No2", 0) famille["categorie_3"] = famille.get("CL_No3", 0) famille["categorie_4"] = famille.get("CL_No4", 0) - # Statistiques famille["stat_01"] = famille.get("FA_Stat01", "") famille["stat_02"] = famille.get("FA_Stat02", "") famille["stat_03"] = famille.get("FA_Stat03", "") @@ -10743,7 +12275,6 @@ class SageConnector: famille["stat_05"] = famille.get("FA_Stat05", "") famille["hors_statistique"] = bool(famille.get("FA_HorsStat", 0)) - # Paramètres commerciaux famille["vente_debit"] = bool(famille.get("FA_VteDebit", 0)) famille["non_imprimable"] = bool(famille.get("FA_NotImp", 0)) famille["contremarque"] = bool(famille.get("FA_Contremarque", 0)) @@ -10751,12 +12282,10 @@ class SageConnector: famille["fact_forfait"] = bool(famille.get("FA_FactForfait", 0)) famille["publie"] = bool(famille.get("FA_Publie", 0)) - # Références famille["racine_reference"] = famille.get("FA_RacineRef", "") famille["racine_code_barre"] = famille.get("FA_RacineCB", "") famille["raccourci"] = famille.get("FA_Raccourci", "") - # Gestion famille["sous_traitance"] = bool(famille.get("FA_SousTraitance", 0)) famille["fictif"] = bool(famille.get("FA_Fictif", 0)) famille["criticite"] = int(famille.get("FA_Criticite", 0))