import os import re from collections import 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, models_sales_document_deposits LINE_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_', ] DEPOSIT_LEGACY_COLUMNS = [ 'sales_documents_deposit_id', 'sales_documents_deposit_number', 'sales_documents_deposit_date', 'sales_documents_deposit_amount', 'sales_documents_deposit_method', ] SCAN_EXTENSIONS = {'.py', '.js', '.html'} IGNORED_DIR_NAMES = { '__pycache__', '.git', '.idea', '__MACOSX', 'node_modules', 'venv', '.venv', } class Command(BaseCommand): help = "Audit Palier D : lignes documents dynamiques + acomptes" def add_arguments(self, parser): parser.add_argument( '--show-lines', action='store_true', help='Affiche toutes les références trouvées au lieu du résumé par fichier.', ) parser.add_argument( '--max-lines', type=int, default=25, help='Nombre maximum de lignes détaillées affichées par catégorie avec --show-lines.', ) def handle(self, *args, **options): root_dir = os.getcwd() show_lines = options['show_lines'] max_lines = options['max_lines'] self.stdout.write('Audit Palier D : lignes documents + acomptes') self.stdout.write('------------------------------------------------') self.stdout.write(f'Dossier analysé : {root_dir}') self.stdout.write('') table_columns = self._get_table_columns() self._print_db_schema_summary(table_columns) self._print_data_summary() self._print_code_summary(root_dir, show_lines=show_lines, max_lines=max_lines) self._print_recommended_plan() def _get_table_columns(self): result = {} with connection.cursor() as cursor: table_names = connection.introspection.table_names(cursor) for table_name in [ 'table_sales_documents', 'table_sales_document_lines', 'table_sales_document_deposits', ]: if table_name not in table_names: result[table_name] = None continue description = connection.introspection.get_table_description(cursor, table_name) result[table_name] = [column.name for column in description] return result def _print_db_schema_summary(self, table_columns): self.stdout.write('1) Structure MySQL détectée') self.stdout.write('---------------------------') sales_doc_columns = table_columns.get('table_sales_documents') or [] line_columns = [] for column in sales_doc_columns: if any(column.startswith(prefix) for prefix in LINE_PREFIXES): line_columns.append(column) deposit_columns = [column for column in DEPOSIT_LEGACY_COLUMNS if column in sales_doc_columns] self.stdout.write(f"table_sales_documents : {'présente' if table_columns.get('table_sales_documents') is not None else 'ABSENTE'}") self.stdout.write(f"table_sales_document_lines : {'présente' if table_columns.get('table_sales_document_lines') is not None else 'ABSENTE'}") self.stdout.write(f"table_sales_document_deposits : {'présente' if table_columns.get('table_sales_document_deposits') is not None else 'ABSENTE'}") self.stdout.write('') self.stdout.write(f'Colonnes legacy lignes document dans table_sales_documents : {len(line_columns)}') if line_columns: self.stdout.write(' Exemples : ' + ', '.join(line_columns[:12]) + (' ...' if len(line_columns) > 12 else '')) self.stdout.write(f'Colonnes legacy acomptes dans table_sales_documents : {len(deposit_columns)}') if deposit_columns: self.stdout.write(' ' + ', '.join(deposit_columns)) self.stdout.write('') if table_columns.get('table_sales_document_lines'): self.stdout.write('Colonnes nouvelle table lignes :') self.stdout.write(' ' + ', '.join(table_columns['table_sales_document_lines'])) if table_columns.get('table_sales_document_deposits'): self.stdout.write('Colonnes nouvelle table acomptes :') self.stdout.write(' ' + ', '.join(table_columns['table_sales_document_deposits'])) self.stdout.write('') def _print_data_summary(self): self.stdout.write('2) Données détectées') self.stdout.write('--------------------') documents_count = models_sales_documents.objects.count() lines_count = models_sales_document_lines.objects.count() deposits_count = models_sales_document_deposits.objects.count() self.stdout.write(f'Documents : {documents_count}') self.stdout.write(f'Lignes dans table_sales_document_lines : {lines_count}') self.stdout.write(f'Acomptes dans table_sales_document_deposits : {deposits_count}') if documents_count: line_counts_by_doc = defaultdict(int) for sales_document_id in models_sales_document_lines.objects.values_list('sales_document_id', flat=True): line_counts_by_doc[str(sales_document_id)] += 1 docs_without_lines = [] docs_with_not_25 = [] docs_with_more_than_25 = [] min_lines = None max_lines = 0 for doc_id in models_sales_documents.objects.values_list('id_sales_documents', flat=True): count = line_counts_by_doc.get(str(doc_id), 0) min_lines = count if min_lines is None else min(min_lines, count) max_lines = max(max_lines, count) if count == 0: docs_without_lines.append(doc_id) if count != 25: docs_with_not_25.append((doc_id, count)) if count > 25: docs_with_more_than_25.append((doc_id, count)) self.stdout.write(f'Lignes par document : min={min_lines}, max={max_lines}') self.stdout.write(f'Documents sans ligne nouvelle table : {len(docs_without_lines)}') self.stdout.write(f'Documents avec un nombre de lignes différent de 25 : {len(docs_with_not_25)}') self.stdout.write(f'Documents déjà avec plus de 25 lignes : {len(docs_with_more_than_25)}') if docs_without_lines[:5]: self.stdout.write(' Exemples sans ligne : ' + ', '.join(str(x) for x in docs_without_lines[:5])) if docs_with_not_25[:5]: self.stdout.write(' Exemples != 25 : ' + ', '.join(f'{doc}:{count}' for doc, count in docs_with_not_25[:5])) if docs_with_more_than_25[:5]: self.stdout.write(' Exemples > 25 : ' + ', '.join(f'{doc}:{count}' for doc, count in docs_with_more_than_25[:5])) self.stdout.write('') def _print_code_summary(self, root_dir, show_lines=False, max_lines=25): self.stdout.write('3) Références code/front') self.stdout.write('-----------------------') references = self._scan_code(root_dir) category_counts = defaultdict(int) file_counts = defaultdict(lambda: defaultdict(int)) for ref in references: category_counts[ref['category']] += 1 file_counts[ref['path']][ref['category']] += 1 for category in sorted(category_counts): self.stdout.write(f'- {category}: {category_counts[category]}') self.stdout.write('') self.stdout.write('Résumé par fichier important') for rel_path in sorted(file_counts): if not self._is_interesting_file(rel_path): continue parts = [f'{category}={count}' for category, count in sorted(file_counts[rel_path].items())] self.stdout.write(f'- {rel_path}: ' + ', '.join(parts)) if show_lines: self.stdout.write('') self.stdout.write('Détail des références') by_category = defaultdict(list) for ref in references: by_category[ref['category']].append(ref) for category in sorted(by_category): self.stdout.write('') self.stdout.write(category) self.stdout.write('-' * len(category)) for ref in by_category[category][:max_lines]: self.stdout.write(f"{ref['path']}:L{ref['line_no']} | {ref['token']}") self.stdout.write(f" {ref['line']}") remaining = len(by_category[category]) - max_lines if remaining > 0: self.stdout.write(f' ... {remaining} référence(s) masquée(s)') self.stdout.write('') def _scan_code(self, root_dir): refs = [] token_regexes = [] # Ligne fixe 1..25 : on cherche les familles avec un suffixe numérique. for prefix in LINE_PREFIXES: token_regexes.append(('line_legacy_contract', prefix, re.compile(r'\b' + re.escape(prefix) + r'\d+\b'))) # Acomptes legacy dans table_sales_documents ou nouvelle table deposit. for column in DEPOSIT_LEGACY_COLUMNS: token_regexes.append(('deposit_reference', column, re.compile(r'\b' + re.escape(column) + r'\b'))) # Nouvelle table / modèle lignes et acomptes. new_table_tokens = [ 'models_sales_document_lines', 'table_sales_document_lines', 'sales_document_lines', 'models_sales_document_deposits', 'table_sales_document_deposits', 'sales_document_deposits', ] for token in new_table_tokens: token_regexes.append(('new_table_reference', token, re.compile(r'\b' + re.escape(token) + r'\b'))) for dirpath, dirnames, filenames in os.walk(root_dir): dirnames[:] = [d for d in dirnames if d not in IGNORED_DIR_NAMES] rel_dir = os.path.relpath(dirpath, root_dir) if rel_dir.startswith('__MACOSX'): continue for filename in filenames: ext = os.path.splitext(filename)[1] if ext not in SCAN_EXTENSIONS: continue path = os.path.join(dirpath, filename) rel_path = os.path.relpath(path, root_dir) if rel_path.startswith('__MACOSX'): continue try: with open(path, 'r', encoding='utf-8') as f: lines = f.readlines() except UnicodeDecodeError: continue for index, raw_line in enumerate(lines, start=1): line = raw_line.strip() if not line: continue for base_category, token, regex in token_regexes: if not regex.search(line): continue refs.append({ 'path': rel_path, 'line_no': index, 'line': line[:240], 'token': token, 'category': self._classify_reference(rel_path, line, base_category, token), }) return refs def _classify_reference(self, rel_path, line, base_category, token): if rel_path.startswith('app_files/migrations/'): return 'migration_historique' if base_category == 'new_table_reference': return 'nouvelle_table_ou_modele' if base_category == 'deposit_reference': if 'models_sales_document_deposits' in line or 'table_sales_document_deposits' in line: return 'acompte_nouvelle_table' if rel_path == 'app_files/models.py' and 'class models_sales_document_deposits' in line: return 'acompte_nouvelle_table' if rel_path == 'app_files/models.py' and token in line and 'class models_sales_documents' not in line: return 'acompte_modele_ou_legacy' if rel_path == 'app_files/views.py': if 'models_sales_document_deposits' in line: return 'acompte_nouvelle_table' if 'doc_instance.' in line or "getattr(doc_instance" in line or "update_fields" in line: return 'acompte_miroir_legacy_views' return 'acompte_vue_active' if rel_path.endswith('.html'): return 'acompte_template_front' if rel_path.endswith('.js'): return 'acompte_js_front' return 'acompte_reference_autre' # Legacy contrat lignes 1..25 if rel_path == 'app_files/models.py': return 'ligne_legacy_modele_sales_documents' if rel_path == 'app_files/forms.py': return 'ligne_legacy_formulaire' if rel_path == 'app_files/views.py': if any(x in line for x in ['post_data.get', 'getattr(doc_instance', 'setattr(doc_instance', 'line_payload', 'sales_documents_line']): return 'ligne_pont_views_legacy_vers_table' return 'ligne_vue_reference' if rel_path.endswith('sales_documents_products.html'): return 'ligne_template_contrat_front_25' if rel_path.endswith('sales_documents_edit.js'): return 'ligne_js_contrat_front_25' if rel_path.endswith('.html'): return 'ligne_template_autre' if rel_path.endswith('.js'): return 'ligne_js_autre' return 'ligne_reference_autre' def _is_interesting_file(self, rel_path): interesting = [ 'app_files/models.py', 'app_files/forms.py', 'app_files/views.py', 'templates/sales_documents_products.html', 'js/sales_documents_edit.js', 'templates/update_record.html', 'app_files/management/commands/check_new_tables_integrity.py', ] return rel_path in interesting def _print_recommended_plan(self): self.stdout.write('4) Lecture du résultat / plan conseillé') self.stdout.write('--------------------------------------') self.stdout.write('Objectif final : ne plus dépendre des champs sales_documents_*_1..25 côté front ni côté BDD.') self.stdout.write('') self.stdout.write('Plan recommandé :') self.stdout.write('D2 - Ajouter une API JSON dédiée aux lignes document depuis table_sales_document_lines.') self.stdout.write('D3 - Rendre les lignes dans le front depuis JSON, sans boucle HTML 1..25 fixe.') self.stdout.write('D4 - Ajouter les actions + ligne / supprimer ligne / réordonner dynamiquement.') self.stdout.write('D5 - Sauvegarder les lignes dynamiques vers table_sales_document_lines, sans limite 25.') self.stdout.write('D6 - Quand le front dynamique est validé, supprimer les colonnes legacy sales_documents_*_1..25 de table_sales_documents.') self.stdout.write('D7 - Nettoyer forms.py, sales_documents_products.html, sales_documents_edit.js et views.py.') self.stdout.write('') self.stdout.write('Note : les acomptes sont déjà en table dédiée. Ils doivent être vérifiés séparément pour distinguer les vrais champs legacy') self.stdout.write('dans table_sales_documents des champs normaux de table_sales_document_deposits.')