from collections import Counter, defaultdict from django.core.management.base import BaseCommand from django.db import connection from app_files.models import models_sales_documents, models_sales_document_lines LINE_FIELD_PREFIXES = [ 'sales_documents_index_line', 'sales_documents_quantity', 'sales_documents_description', 'sales_documents_description_free', 'sales_documents_price', 'sales_documents_discount', 'sales_documents_calculated_price', 'sales_documents_discounted_price', 'sales_documents_vat', ] def as_str(value): if value is None: return '' return str(value).strip() def safe_int(value, default=None): try: if value is None or value == '': return default return int(value) except (TypeError, ValueError): return default class Command(BaseCommand): help = "Diagnostic non destructif de cohérence des lignes de documents avant front dynamique." def add_arguments(self, parser): parser.add_argument('--show-details', action='store_true', help='Affiche les détails des écarts par document.') parser.add_argument('--doc', type=str, default='', help='Limiter le diagnostic à un id_sales_documents précis.') def handle(self, *args, **options): show_details = options['show_details'] doc_filter = options['doc'].strip() self.stdout.write('Diagnostic lignes documents : cohérence legacy 25 lignes / table dynamique') self.stdout.write('------------------------------------------------------------------------') legacy_columns = self._get_table_columns('table_sales_documents') line_columns = self._get_table_columns('table_sales_document_lines') self.stdout.write('Structure détectée') self.stdout.write('------------------') self.stdout.write(f"table_sales_documents présente : {'oui' if legacy_columns else 'non'}") self.stdout.write(f"table_sales_document_lines présente : {'oui' if line_columns else 'non'}") legacy_line_columns = [c for c in legacy_columns if self._is_legacy_line_column(c)] self.stdout.write(f"Colonnes legacy lignes document détectées : {len(legacy_line_columns)}") self.stdout.write('') docs_qs = models_sales_documents.objects.all().order_by('id_sales_documents') if doc_filter: docs_qs = docs_qs.filter(id_sales_documents=doc_filter) total_docs = docs_qs.count() total_lines = models_sales_document_lines.objects.count() if not doc_filter else models_sales_document_lines.objects.filter(sales_document_id=doc_filter).count() warnings = [] errors = [] info = [] stats = { 'docs_without_lines': 0, 'docs_with_not_25_lines': 0, 'docs_with_more_than_25_lines': 0, 'docs_with_position_gaps': 0, 'docs_with_position_duplicates': 0, 'docs_with_line_number_duplicates': 0, 'docs_with_legacy_new_mismatch': 0, 'legacy_non_empty_lines': 0, 'new_non_empty_lines': 0, } for doc in docs_qs: doc_id = str(doc.id_sales_documents) lines = list( models_sales_document_lines.objects .filter(sales_document_id=doc.id_sales_documents) .order_by('sales_documents_line_position', 'sales_documents_line_number', 'id_sales_document_line') ) if not lines: stats['docs_without_lines'] += 1 warnings.append(f"document sans ligne dynamique | doc={doc_id}") continue if len(lines) != 25: stats['docs_with_not_25_lines'] += 1 info.append(f"document avec {len(lines)} lignes dynamiques | doc={doc_id}") if len(lines) > 25: stats['docs_with_more_than_25_lines'] += 1 info.append(f"document déjà au-delà de 25 lignes | doc={doc_id} | lignes={len(lines)}") positions = [safe_int(line.sales_documents_line_position) for line in lines] line_numbers = [safe_int(line.sales_documents_line_number) for line in lines] position_counter = Counter(positions) duplicate_positions = [p for p, count in position_counter.items() if p is not None and count > 1] if duplicate_positions: stats['docs_with_position_duplicates'] += 1 errors.append(f"positions dupliquées | doc={doc_id} | positions={duplicate_positions}") line_number_counter = Counter(line_numbers) duplicate_line_numbers = [n for n, count in line_number_counter.items() if n is not None and count > 1] if duplicate_line_numbers: stats['docs_with_line_number_duplicates'] += 1 errors.append(f"numéros de ligne dupliqués | doc={doc_id} | lignes={duplicate_line_numbers}") expected_positions = list(range(1, len(lines) + 1)) sorted_positions = sorted([p for p in positions if p is not None]) if sorted_positions != expected_positions: stats['docs_with_position_gaps'] += 1 warnings.append(f"positions non séquentielles | doc={doc_id} | attendu=1..{len(lines)} | trouvé={sorted_positions[:40]}") for line in lines: if self._line_has_content(line): stats['new_non_empty_lines'] += 1 legacy_non_empty_indexes = [] for i in range(1, 26): legacy_values = self._legacy_values_for_line(doc, i) if self._legacy_values_have_content(legacy_values): legacy_non_empty_indexes.append(i) stats['legacy_non_empty_lines'] += 1 mismatches = self._compare_first_25_legacy_and_new(doc, lines) if mismatches: stats['docs_with_legacy_new_mismatch'] += 1 warnings.append(f"écart legacy/nouvelle table sur premières 25 lignes | doc={doc_id} | écarts={len(mismatches)}") if show_details: for mismatch in mismatches[:30]: warnings.append(f" - doc={doc_id} | {mismatch}") if show_details: info.append( f"doc={doc_id} | lignes_dynamiques={len(lines)} | legacy_non_vides={legacy_non_empty_indexes}" ) self.stdout.write('Données détectées') self.stdout.write('----------------') self.stdout.write(f"Documents analysés : {total_docs}") self.stdout.write(f"Lignes dynamiques analysées : {total_lines}") self.stdout.write(f"Documents sans ligne dynamique : {stats['docs_without_lines']}") self.stdout.write(f"Documents avec nombre de lignes différent de 25 : {stats['docs_with_not_25_lines']}") self.stdout.write(f"Documents avec plus de 25 lignes : {stats['docs_with_more_than_25_lines']}") self.stdout.write(f"Documents avec positions dupliquées : {stats['docs_with_position_duplicates']}") self.stdout.write(f"Documents avec numéros de ligne dupliqués : {stats['docs_with_line_number_duplicates']}") self.stdout.write(f"Documents avec positions non séquentielles : {stats['docs_with_position_gaps']}") self.stdout.write(f"Documents avec écarts legacy/nouvelle table : {stats['docs_with_legacy_new_mismatch']}") self.stdout.write(f"Lignes legacy non vides détectées : {stats['legacy_non_empty_lines']}") self.stdout.write(f"Lignes dynamiques non vides détectées : {stats['new_non_empty_lines']}") self.stdout.write('') if errors: self.stdout.write('ERREURS BLOQUANTES') self.stdout.write('------------------') for item in errors[:120]: self.stdout.write(f"- {item}") if len(errors) > 120: self.stdout.write(f"... {len(errors) - 120} erreur(s) masquée(s)") self.stdout.write('') if warnings: self.stdout.write('AVERTISSEMENTS') self.stdout.write('--------------') for item in warnings[:160]: self.stdout.write(f"- {item}") if len(warnings) > 160: self.stdout.write(f"... {len(warnings) - 160} avertissement(s) masqué(s)") self.stdout.write('') if info and show_details: self.stdout.write('DÉTAILS') self.stdout.write('-------') for item in info[:200]: self.stdout.write(f"- {item}") if len(info) > 200: self.stdout.write(f"... {len(info) - 200} détail(s) masqué(s)") self.stdout.write('') self.stdout.write('Lecture du résultat') self.stdout.write('-------------------') if errors: self.stdout.write('Résultat : corrections nécessaires avant ajout/suppression/réordonnancement dynamique.') elif warnings: self.stdout.write('Résultat : pas de blocage critique, mais des points doivent être compris avant bascule front dynamique.') else: self.stdout.write('Résultat : cohérence lignes document OK pour préparer le front dynamique.') def _get_table_columns(self, table_name): with connection.cursor() as cursor: try: return [col.name for col in connection.introspection.get_table_description(cursor, table_name)] except Exception: return [] def _is_legacy_line_column(self, column_name): return any(column_name.startswith(prefix + '_') for prefix in LINE_FIELD_PREFIXES) def _legacy_values_for_line(self, doc, i): return { 'position': as_str(getattr(doc, f'sales_documents_index_line_{i}', '')), 'quantity': as_str(getattr(doc, f'sales_documents_quantity_{i}', '')), 'product_id': as_str(getattr(doc, f'sales_documents_description_{i}', '')), 'description_free': as_str(getattr(doc, f'sales_documents_description_free_{i}', '')), 'price': as_str(getattr(doc, f'sales_documents_price_{i}', '')), 'discount': as_str(getattr(doc, f'sales_documents_discount_{i}', '')), 'calculated_price': as_str(getattr(doc, f'sales_documents_calculated_price_{i}', '')), 'discounted_price': as_str(getattr(doc, f'sales_documents_discounted_price_{i}', '')), 'vat': as_str(getattr(doc, f'sales_documents_vat_{i}', '')), } def _line_values(self, line): return { 'position': as_str(line.sales_documents_line_position), 'quantity': as_str(line.sales_documents_quantity), 'product_id': as_str(line.product_id or ''), 'description_free': as_str(line.sales_documents_description_free), 'price': as_str(line.sales_documents_price), 'discount': as_str(line.sales_documents_discount), 'calculated_price': as_str(line.sales_documents_calculated_price), 'discounted_price': as_str(line.sales_documents_discounted_price), 'vat': as_str(line.sales_documents_vat), } def _legacy_values_have_content(self, values): ignored_empty = {'position': '', 'vat': '', 'product_id': '', 'quantity': '', 'price': '', 'discount': '', 'calculated_price': '', 'discounted_price': '', 'description_free': ''} for key, value in values.items(): if key == 'position': continue if key == 'vat' and value in ('', '0'): continue if key == 'product_id' and value in ('', '0'): continue if value not in ('', '0'): return True return False def _line_has_content(self, line): values = self._line_values(line) return self._legacy_values_have_content(values) def _compare_first_25_legacy_and_new(self, doc, lines): by_line_number = {safe_int(line.sales_documents_line_number): line for line in lines} mismatches = [] for i in range(1, 26): legacy_values = self._legacy_values_for_line(doc, i) line = by_line_number.get(i) if line is None: if self._legacy_values_have_content(legacy_values): mismatches.append(f"ligne {i} présente côté legacy mais absente côté nouvelle table") continue new_values = self._line_values(line) fields_to_compare = [ 'position', 'quantity', 'product_id', 'description_free', 'price', 'discount', 'calculated_price', 'discounted_price', 'vat' ] for field in fields_to_compare: legacy_value = legacy_values[field] new_value = new_values[field] if field == 'vat': legacy_value = str(safe_int(legacy_value, 0)) new_value = str(safe_int(new_value, 0)) if field == 'product_id': legacy_value = '' if legacy_value == '0' else legacy_value new_value = '' if new_value == '0' else new_value if legacy_value != new_value: # Ignore fully empty legacy line compared with empty new default line. if not self._legacy_values_have_content(legacy_values) and not self._line_has_content(line): continue mismatches.append(f"ligne {i} champ {field} | legacy='{legacy_value}' | nouvelle_table='{new_value}'") return mismatches