import sqlite3 import pandas as pd import os import re from typing import List, Tuple, Dict, Any import PyPDF2 import glob from datetime import datetime import warnings warnings.filterwarnings('ignore') class UniversalProductionDB: def __init__(self, db_name: str = 'universal_production_company.db'): self.db_path = os.path.join(os.getcwd(), db_name) self.conn = None self.project_folder = 'material_db_project' self.data_sources = { 'materials': 'Materials_import.xlsx', 'material_suppliers': 'Material_suppliers_import.xlsx', 'material_types': 'Material_type_import.xlsx', 'product_types': 'Product_type_import.xlsx', 'suppliers': 'Suppliers_import.xlsx' } def connect(self): """Подключение к базе данных""" try: self.conn = sqlite3.connect(self.db_path) print(f"✅ Успешное подключение к базе данных: {self.db_path}") return True except sqlite3.Error as e: print(f"❌ Ошибка подключения к базе данных: {e}") return False def get_file_path(self, filename: str) -> str: """Получение полного пути к файлу с учетом папки проекта""" project_path = os.path.join(os.getcwd(), self.project_folder) if os.path.exists(project_path) and os.path.isdir(project_path): file_path = os.path.join(project_path, filename) if os.path.exists(file_path): return file_path return filename def analyze_excel_structure(self, filename: str) -> Dict[str, Any]: """Анализ структуры Excel файла""" try: file_path = self.get_file_path(filename) if not os.path.exists(file_path): return {'error': f'Файл {filename} не найден'} print(f"📊 Анализирую структуру файла: {filename}") excel_file = pd.ExcelFile(file_path) sheets_info = {} for sheet_name in excel_file.sheet_names: df = pd.read_excel(file_path, sheet_name=sheet_name) sheet_info = { 'columns': df.columns.tolist(), 'data_types': df.dtypes.astype(str).to_dict(), 'sample_data': df.head(2).to_dict('records'), 'row_count': len(df), 'null_counts': df.isnull().sum().to_dict() } sheets_info[sheet_name] = sheet_info return { 'file_path': file_path, 'sheets': sheets_info, 'sheet_names': excel_file.sheet_names } except Exception as e: return {'error': f'Ошибка анализа файла: {e}'} def detect_column_mapping(self, columns: List[str]) -> Dict[str, str]: """Автоматическое определение соответствия колонок""" column_mapping = {} russian_mappings = { 'материал': ['наименование материала', 'материал', 'название материала'], 'тип': ['тип материала', 'тип'], 'поставщик': ['поставщик', 'наименование поставщика'], 'цена': ['цена', 'цена единицы', 'стоимость'], 'количество': ['количество', 'количество на складе'], 'рейтинг': ['рейтинг', 'rating'], 'инн': ['инн', 'inn'], 'дата': ['дата', 'date'], 'артикул': ['артикул', 'article'], 'продукция': ['тип продукции', 'продукция'] } for col in columns: col_lower = col.lower() for eng_key, ru_values in russian_mappings.items(): if any(ru_val in col_lower for ru_val in ru_values): column_mapping[eng_key] = col break if 'name' in col_lower and 'material' in col_lower: column_mapping['material_name'] = col elif 'type' in col_lower and 'material' in col_lower: column_mapping['material_type'] = col elif 'supplier' in col_lower: column_mapping['supplier_name'] = col elif 'price' in col_lower or 'cost' in col_lower: column_mapping['price'] = col elif 'quantity' in col_lower or 'count' in col_lower: column_mapping['quantity'] = col return column_mapping def create_complete_database_structure(self): """Создание полной структуры базы данных производственной компании""" try: cursor = self.conn.cursor() # 1. Таблица партнеров cursor.execute(''' CREATE TABLE IF NOT EXISTS partners ( id INTEGER PRIMARY KEY AUTOINCREMENT, partner_type TEXT NOT NULL, company_name TEXT NOT NULL UNIQUE, legal_address TEXT NOT NULL, inn TEXT NOT NULL UNIQUE, director_name TEXT NOT NULL, phone TEXT, email TEXT, logo_path TEXT, rating INTEGER DEFAULT 5, sales_places TEXT, total_sales_volume DECIMAL(15,2) DEFAULT 0, discount_percent DECIMAL(5,2) DEFAULT 0, created_date TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_date TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ''') # 2. Таблица истории рейтинга партнеров cursor.execute(''' CREATE TABLE IF NOT EXISTS partner_rating_history ( id INTEGER PRIMARY KEY AUTOINCREMENT, partner_id INTEGER NOT NULL, old_rating INTEGER, new_rating INTEGER, change_reason TEXT, changed_by TEXT, change_date TIMESTAMP DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (partner_id) REFERENCES partners (id) ) ''') # 3. Таблица менеджеров cursor.execute(''' CREATE TABLE IF NOT EXISTS managers ( id INTEGER PRIMARY KEY AUTOINCREMENT, full_name TEXT NOT NULL, birth_date DATE, passport_data TEXT, bank_details TEXT, family_status TEXT, health_status TEXT, hire_date DATE DEFAULT CURRENT_DATE, is_active BOOLEAN DEFAULT TRUE ) ''') # 4. Таблица заявок cursor.execute(''' CREATE TABLE IF NOT EXISTS orders ( id INTEGER PRIMARY KEY AUTOINCREMENT, partner_id INTEGER NOT NULL, manager_id INTEGER NOT NULL, order_status TEXT DEFAULT 'created', total_amount DECIMAL(15,2), prepayment_amount DECIMAL(15,2) DEFAULT 0, prepayment_received BOOLEAN DEFAULT FALSE, full_payment_received BOOLEAN DEFAULT FALSE, order_date TIMESTAMP DEFAULT CURRENT_TIMESTAMP, prepayment_due_date DATE, production_start_date DATE, expected_completion_date DATE, actual_completion_date DATE, delivery_method TEXT, cancellation_reason TEXT, FOREIGN KEY (partner_id) REFERENCES partners (id), FOREIGN KEY (manager_id) REFERENCES managers (id) ) ''') # 5. Таблица продукции в заявках cursor.execute(''' CREATE TABLE IF NOT EXISTS order_products ( id INTEGER PRIMARY KEY AUTOINCREMENT, order_id INTEGER NOT NULL, product_article TEXT NOT NULL, product_name TEXT NOT NULL, quantity INTEGER NOT NULL, unit_price DECIMAL(10,2) NOT NULL, total_price DECIMAL(10,2) NOT NULL, production_date DATE, is_ready BOOLEAN DEFAULT FALSE, FOREIGN KEY (order_id) REFERENCES orders (id) ) ''') # 6. Таблица сотрудников cursor.execute(''' CREATE TABLE IF NOT EXISTS employees ( id INTEGER PRIMARY KEY AUTOINCREMENT, full_name TEXT NOT NULL, birth_date DATE, passport_data TEXT, bank_details TEXT, family_status TEXT, health_status TEXT, position TEXT NOT NULL, hire_date DATE DEFAULT CURRENT_DATE, workshop_number INTEGER, is_active BOOLEAN DEFAULT TRUE ) ''') # 7. Таблица оборудования и допусков cursor.execute(''' CREATE TABLE IF NOT EXISTS equipment_access ( id INTEGER PRIMARY KEY AUTOINCREMENT, employee_id INTEGER NOT NULL, equipment_type TEXT NOT NULL, equipment_name TEXT NOT NULL, access_granted_date DATE, access_expiry_date DATE, certification_number TEXT, FOREIGN KEY (employee_id) REFERENCES employees (id) ) ''') # 8. Таблица контроля доступа cursor.execute(''' CREATE TABLE IF NOT EXISTS access_logs ( id INTEGER PRIMARY KEY AUTOINCREMENT, employee_id INTEGER NOT NULL, door_id TEXT, access_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP, access_type TEXT, success BOOLEAN DEFAULT TRUE, FOREIGN KEY (employee_id) REFERENCES employees (id) ) ''') # 9. Универсальная таблица материалов cursor.execute(''' CREATE TABLE IF NOT EXISTS materials ( id INTEGER PRIMARY KEY AUTOINCREMENT, material_name TEXT NOT NULL, material_type TEXT, unit_price DECIMAL(15,4), current_quantity INTEGER, min_quantity INTEGER, package_quantity INTEGER, unit_of_measure TEXT, description TEXT, image_path TEXT, loss_percent DECIMAL(8,4) DEFAULT 0, created_date TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_date TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ''') # 10. Таблица истории изменений материалов cursor.execute(''' CREATE TABLE IF NOT EXISTS material_inventory_history ( id INTEGER PRIMARY KEY AUTOINCREMENT, material_name TEXT NOT NULL, old_quantity INTEGER, new_quantity INTEGER, change_reason TEXT, document_number TEXT, change_date TIMESTAMP DEFAULT CURRENT_TIMESTAMP, changed_by TEXT ) ''') # 11. Универсальная таблица поставщиков cursor.execute(''' CREATE TABLE IF NOT EXISTS suppliers ( id INTEGER PRIMARY KEY AUTOINCREMENT, supplier_name TEXT NOT NULL UNIQUE, supplier_type TEXT, inn TEXT, contact_person TEXT, phone TEXT, email TEXT, address TEXT, quality_rating INTEGER DEFAULT 5, start_date DATE, is_active BOOLEAN DEFAULT TRUE, created_date TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ''') # 12. Универсальная таблица связей материалов и поставщиков cursor.execute(''' CREATE TABLE IF NOT EXISTS material_suppliers ( id INTEGER PRIMARY KEY AUTOINCREMENT, material_name TEXT NOT NULL, supplier_name TEXT NOT NULL, is_primary BOOLEAN DEFAULT FALSE, delivery_time_days INTEGER, last_supply_date DATE, created_date TIMESTAMP DEFAULT CURRENT_TIMESTAMP, UNIQUE(material_name, supplier_name) ) ''') # 13. Таблица истории поставок cursor.execute(''' CREATE TABLE IF NOT EXISTS supply_history ( id INTEGER PRIMARY KEY AUTOINCREMENT, supplier_name TEXT NOT NULL, material_name TEXT NOT NULL, supply_date DATE, quantity INTEGER, unit_price DECIMAL(10,2), total_amount DECIMAL(10,2), delivery_note_number TEXT, quality_score INTEGER, received_by TEXT ) ''') # 14. Таблица продукции cursor.execute(''' CREATE TABLE IF NOT EXISTS products ( id INTEGER PRIMARY KEY AUTOINCREMENT, article TEXT NOT NULL UNIQUE, product_type TEXT NOT NULL, product_name TEXT NOT NULL, description TEXT, image_path TEXT, min_partner_price DECIMAL(10,2), package_length DECIMAL(8,2), package_width DECIMAL(8,2), package_height DECIMAL(8,2), net_weight DECIMAL(8,2), gross_weight DECIMAL(8,2), certificate_path TEXT, standard_number TEXT, production_time_hours INTEGER, cost_price DECIMAL(10,2), workshop_number INTEGER, required_workers INTEGER, is_active BOOLEAN DEFAULT TRUE, created_date TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ''') # 15. Таблица истории цен продукции cursor.execute(''' CREATE TABLE IF NOT EXISTS product_price_history ( id INTEGER PRIMARY KEY AUTOINCREMENT, product_article TEXT NOT NULL, old_price DECIMAL(10,2), new_price DECIMAL(10,2), change_date TIMESTAMP DEFAULT CURRENT_TIMESTAMP, changed_by TEXT, change_reason TEXT ) ''') # 16. Таблица материалов для производства продукции cursor.execute(''' CREATE TABLE IF NOT EXISTS product_materials ( id INTEGER PRIMARY KEY AUTOINCREMENT, product_article TEXT NOT NULL, material_name TEXT NOT NULL, material_quantity DECIMAL(10,3) NOT NULL, waste_percent DECIMAL(5,2) DEFAULT 0 ) ''') # 17. Таблица складских операций cursor.execute(''' CREATE TABLE IF NOT EXISTS warehouse_operations ( id INTEGER PRIMARY KEY AUTOINCREMENT, operation_type TEXT NOT NULL, material_name TEXT, product_article TEXT, quantity INTEGER, document_number TEXT, operation_date TIMESTAMP DEFAULT CURRENT_TIMESTAMP, performed_by TEXT, reference_id INTEGER, notes TEXT ) ''') # 18. Таблица типов материалов cursor.execute(''' CREATE TABLE IF NOT EXISTS material_types ( id INTEGER PRIMARY KEY AUTOINCREMENT, type_name TEXT NOT NULL UNIQUE, loss_percent DECIMAL(8,4) DEFAULT 0, description TEXT, created_date TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ''') # 19. Таблица типов продукции cursor.execute(''' CREATE TABLE IF NOT EXISTS product_types ( id INTEGER PRIMARY KEY AUTOINCREMENT, type_name TEXT NOT NULL UNIQUE, coefficient DECIMAL(10,4) DEFAULT 1.0, description TEXT, created_date TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ''') self.conn.commit() print("✅ Полная универсальная структура базы данных создана!") return True except sqlite3.Error as e: print(f"❌ Ошибка создания структуры базы данных: {e}") return False def import_materials_data(self): """Умный импорт данных о материалах""" try: file_info = self.analyze_excel_structure(self.data_sources['materials']) if 'error' in file_info: print(f"❌ {file_info['error']}") return False sheet_name = file_info['sheet_names'][0] sheet_data = file_info['sheets'][sheet_name] columns = sheet_data['columns'] print(f"📋 Колонки в файле материалов: {columns}") mapping = self.detect_column_mapping(columns) print(f"🔍 Определенные колонки: {mapping}") df = pd.read_excel(file_info['file_path'], sheet_name=sheet_name) # Автоматическое переименование колонок column_renames = {} if 'material_name' in mapping: column_renames[mapping['material_name']] = 'material_name' if 'material_type' in mapping: column_renames[mapping['material_type']] = 'material_type' if 'price' in mapping: column_renames[mapping['price']] = 'unit_price' if 'quantity' in mapping: # Ищем разные типы количеств for col in columns: col_lower = col.lower() if 'количество' in col_lower and 'упаковк' not in col_lower: column_renames[col] = 'current_quantity' elif 'минимальн' in col_lower: column_renames[col] = 'min_quantity' elif 'упаковк' in col_lower: column_renames[col] = 'package_quantity' if column_renames: df = df.rename(columns=column_renames) cursor = self.conn.cursor() imported_count = 0 for _, row in df.iterrows(): try: material_name = row.get('material_name', '') if material_name: cursor.execute(''' INSERT OR REPLACE INTO materials (material_name, material_type, unit_price, current_quantity, min_quantity, package_quantity, unit_of_measure, description) VALUES (?, ?, ?, ?, ?, ?, ?, ?) ''', ( material_name, row.get('material_type', ''), self.safe_convert_float(row.get('unit_price', 0)), self.safe_convert_int(row.get('current_quantity', 0)), self.safe_convert_int(row.get('min_quantity', 0)), self.safe_convert_int(row.get('package_quantity', 0)), row.get('unit_of_measure', ''), row.get('description', '') )) imported_count += 1 except Exception as e: print(f"⚠️ Ошибка импорта материала {row.get('material_name', '')}: {e}") continue self.conn.commit() print(f"✅ Импортировано материалов: {imported_count}") return True except Exception as e: print(f"❌ Ошибка импорта материалов: {e}") return False def import_suppliers_data(self): """Умный импорт данных о поставщиках""" try: file_info = self.analyze_excel_structure(self.data_sources['suppliers']) if 'error' in file_info: print(f"❌ {file_info['error']}") return False sheet_name = file_info['sheet_names'][0] df = pd.read_excel(file_info['file_path'], sheet_name=sheet_name) cursor = self.conn.cursor() imported_count = 0 for _, row in df.iterrows(): try: # Автоматически находим нужные колонки supplier_name = None supplier_type = None inn = None rating = 5 start_date = None for col in df.columns: col_lower = col.lower() val = row[col] if 'поставщик' in col_lower and 'наименован' in col_lower: supplier_name = val elif 'тип' in col_lower and 'поставщик' in col_lower: supplier_type = val elif 'инн' in col_lower: inn = str(val) elif 'рейтинг' in col_lower: rating = self.safe_convert_int(val) elif 'дата' in col_lower: start_date = self.safe_convert_date(val) if supplier_name: cursor.execute(''' INSERT OR REPLACE INTO suppliers (supplier_name, supplier_type, inn, quality_rating, start_date) VALUES (?, ?, ?, ?, ?) ''', (supplier_name, supplier_type, inn, rating, start_date)) imported_count += 1 except Exception as e: print(f"⚠️ Ошибка импорта поставщика: {e}") continue self.conn.commit() print(f"✅ Импортировано поставщиков: {imported_count}") return True except Exception as e: print(f"❌ Ошибка импорта поставщиков: {e}") return False def import_material_suppliers_data(self): """Умный импорт связей материалов и поставщиков""" try: file_info = self.analyze_excel_structure(self.data_sources['material_suppliers']) if 'error' in file_info: print(f"❌ {file_info['error']}") return False sheet_name = file_info['sheet_names'][0] df = pd.read_excel(file_info['file_path'], sheet_name=sheet_name) cursor = self.conn.cursor() imported_count = 0 for _, row in df.iterrows(): try: # Автоматически находим колонки material_name = None supplier_name = None for col in df.columns: col_lower = col.lower() val = row[col] if 'материал' in col_lower: material_name = val elif 'поставщик' in col_lower: supplier_name = val if material_name and supplier_name: cursor.execute(''' INSERT OR IGNORE INTO material_suppliers (material_name, supplier_name) VALUES (?, ?) ''', (material_name, supplier_name)) if cursor.rowcount > 0: imported_count += 1 except Exception as e: print(f"⚠️ Ошибка импорта связи: {e}") continue self.conn.commit() print(f"✅ Импортировано связей: {imported_count}") return True except Exception as e: print(f"❌ Ошибка импорта связей: {e}") return False def import_material_types_data(self): """Импорт типов материалов""" try: file_info = self.analyze_excel_structure(self.data_sources['material_types']) if 'error' in file_info: print(f"❌ {file_info['error']}") return False sheet_name = file_info['sheet_names'][0] df = pd.read_excel(file_info['file_path'], sheet_name=sheet_name) cursor = self.conn.cursor() imported_count = 0 for _, row in df.iterrows(): try: type_name = None loss_percent = 0 for col in df.columns: col_lower = col.lower() if 'тип' in col_lower: type_name = row[col] elif 'процент' in col_lower or 'потер' in col_lower: loss_percent = self.safe_convert_float(row[col]) if type_name: cursor.execute(''' INSERT OR REPLACE INTO material_types (type_name, loss_percent) VALUES (?, ?) ''', (type_name, loss_percent)) imported_count += 1 except Exception as e: print(f"⚠️ Ошибка импорта типа материала: {e}") continue self.conn.commit() print(f"✅ Импортировано типов материалов: {imported_count}") return True except Exception as e: print(f"❌ Ошибка импорта типов материалов: {e}") return False def import_product_types_data(self): """Импорт типов продукции""" try: file_info = self.analyze_excel_structure(self.data_sources['product_types']) if 'error' in file_info: print(f"❌ {file_info['error']}") return False sheet_name = file_info['sheet_names'][0] df = pd.read_excel(file_info['file_path'], sheet_name=sheet_name) cursor = self.conn.cursor() imported_count = 0 for _, row in df.iterrows(): try: type_name = None coefficient = 1.0 for col in df.columns: col_lower = col.lower() if 'тип' in col_lower and 'продукц' in col_lower: type_name = row[col] elif 'коэффициент' in col_lower: coefficient = self.safe_convert_float(row[col]) if type_name: cursor.execute(''' INSERT OR REPLACE INTO product_types (type_name, coefficient) VALUES (?, ?) ''', (type_name, coefficient)) imported_count += 1 except Exception as e: print(f"⚠️ Ошибка импорта типа продукции: {e}") continue self.conn.commit() print(f"✅ Импортировано типов продукции: {imported_count}") return True except Exception as e: print(f"❌ Ошибка импорта типов продукции: {e}") return False def import_all_excel_data(self): """Импорт всех данных из Excel файлов""" print("🚀 Начинаю умный импорт всех данных...") imports = [ ('Материалы', self.import_materials_data), ('Поставщики', self.import_suppliers_data), ('Типы материалов', self.import_material_types_data), ('Типы продукции', self.import_product_types_data), ('Связи материалы-поставщики', self.import_material_suppliers_data) ] success_count = 0 for name, import_func in imports: print(f"\n📥 Импорт {name}...") if import_func(): success_count += 1 print(f"\n🎉 Импорт завершен! Успешно: {success_count}/{len(imports)}") return success_count == len(imports) def safe_convert_float(self, value) -> float: """Безопасное преобразование в float""" try: if pd.isna(value) or value == '': return 0.0 return float(value) except (ValueError, TypeError): return 0.0 def safe_convert_int(self, value) -> int: """Безопасное преобразование в int""" try: if pd.isna(value) or value == '': return 0 return int(float(value)) except (ValueError, TypeError): return 0 def safe_convert_date(self, value): """Безопасное преобразование в дату""" try: if pd.isna(value) or value == '': return None if isinstance(value, str): return datetime.strptime(value, '%Y-%m-%d %H:%M:%S').date() return value.date() if hasattr(value, 'date') else value except (ValueError, TypeError): return None def generate_sample_data(self): """Генерация демонстрационных данных для тестирования""" try: cursor = self.conn.cursor() # Добавляем партнеров partners_data = [ ( 'Розничная сеть', 'СтройДом', 'Москва, ул. Строителей, 1', '0987654321', 'Смирнов А.А.', '+79991111111', 'stroy@dom.ru', 5, 'Магазины, интернет', 1500000, 10), ('Оптовик', 'КерамикТрейд', 'СПб, ул. Керамиков, 5', '0987654322', 'Кузнецов В.В.', '+79992222222', 'keram@trade.ru', 4, 'Оптовые продажи', 800000, 7), ] cursor.executemany(''' INSERT OR IGNORE INTO partners (partner_type, company_name, legal_address, inn, director_name, phone, email, rating, sales_places, total_sales_volume, discount_percent) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ''', partners_data) # Добавляем менеджеров managers_data = [ ('Иванов Иван Иванович', '1985-05-15', '4510 123456', '40702810000000001234', 'Женат', 'Здоров', '2020-01-15'), ('Петрова Мария Сергеевна', '1990-08-20', '4510 654321', '40702810000000005678', 'Не замужем', 'Здоров', '2021-03-10'), ] cursor.executemany(''' INSERT OR IGNORE INTO managers (full_name, birth_date, passport_data, bank_details, family_status, health_status, hire_date) VALUES (?, ?, ?, ?, ?, ?, ?) ''', managers_data) # Добавляем продукцию products_data = [ ( 'PL-001', 'Плитка', 'Плитка керамическая 30x30', 'Плитка для пола', '', 450.00, 30, 30, 5, 2.5, 3.0, '', 'ГОСТ 123', 24, 280.00, 1, 3), ('PL-002', 'Плитка', 'Плитка керамическая 20x20', 'Плитка для ванной', '', 380.00, 20, 20, 4, 1.8, 2.2, '', 'ГОСТ 124', 20, 220.00, 1, 2), ] cursor.executemany(''' INSERT OR IGNORE INTO products (article, product_type, product_name, description, image_path, min_partner_price, package_length, package_width, package_height, net_weight, gross_weight, certificate_path, standard_number, production_time_hours, cost_price, workshop_number, required_workers) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ''', products_data) self.conn.commit() print("✅ Демонстрационные данные успешно добавлены!") return True except sqlite3.Error as e: print(f"❌ Ошибка добавления демонстрационных данных: {e}") return False def analyze_database_schema(self): """Анализ текущей схемы базы данных""" try: cursor = self.conn.cursor() cursor.execute("SELECT name FROM sqlite_master WHERE type='table'") tables = cursor.fetchall() print("\n📊 АНАЛИЗ СХЕМЫ БАЗЫ ДАННЫХ:") print("=" * 50) for table in tables: table_name = table[0] print(f"\n📋 Таблица: {table_name}") print("-" * 30) cursor.execute(f"PRAGMA table_info({table_name})") columns = cursor.fetchall() for column in columns: col_name, col_type, not_null, default_val, pk = column[1], column[2], column[3], column[4], column[ 5] print(f" {col_name:25} {col_type:15} {'NOT NULL' if not_null else 'NULL':8} {'PK' if pk else ''}") return True except sqlite3.Error as e: print(f"❌ Ошибка анализа схемы: {e}") return False def generate_er_diagram(self): """Генерация текстовой ER-диаграммы""" print("\n🔗 ER-ДИАГРАММА БАЗЫ ДАННЫХ:") print("=" * 60) er_diagram = """ ╔════════════════════════════════════════════════════════════════╗ ║ УНИВЕРСАЛЬНАЯ ПРОИЗВОДСТВЕННАЯ СИСТЕМА ║ ╚════════════════════════════════════════════════════════════════╝ 📊 ОСНОВНЫЕ СУЩНОСТИ: 1. 🏢 ПАРТНЕРЫ (partners) - клиенты компании 2. 📋 ЗАЯВКИ (orders) - заказы от партнеров 3. 👨‍💼 СОТРУДНИКИ (employees, managers) - персонал 4. 🏭 ПРОИЗВОДСТВО (products, product_materials) - продукция 5. 📦 СКЛАД (materials, warehouse_operations) - материалы 6. 🏪 ПОСТАВЩИКИ (suppliers, material_suppliers) - поставщики 🔄 АВТОМАТИЧЕСКИЙ ИМПОРТ ИЗ EXCEL: • Материалы и их характеристики • Поставщики и их рейтинги • Связи материалов и поставщиков • Типы материалов и продукции • Адаптация к изменениям в файлах 🎯 ВОЗМОЖНОСТИ СИСТЕМЫ: • Автоанализ структуры Excel файлов • Умное определение колонок • Поддержка изменений в данных • Полная производственная логика • Анализ PDF заданий """ print(er_diagram) return True def get_database_stats(self): """Получение статистики базы данных""" try: cursor = self.conn.cursor() stats = {} tables = ['materials', 'suppliers', 'material_suppliers', 'material_types', 'product_types', 'partners', 'managers', 'products', 'employees'] for table in tables: cursor.execute(f'SELECT COUNT(*) FROM {table}') count = cursor.fetchone()[0] stats[table] = count return stats except sqlite3.Error as e: print(f"❌ Ошибка получения статистики: {e}") return {} def search_materials(self, keyword: str): """Поиск материалов по ключевому слову""" try: cursor = self.conn.cursor() cursor.execute(''' SELECT m.material_name, m.material_type, m.unit_price, m.current_quantity, m.unit_of_measure, GROUP_CONCAT(DISTINCT ms.supplier_name) as suppliers FROM materials m LEFT JOIN material_suppliers ms ON m.material_name = ms.material_name WHERE m.material_name LIKE ? OR m.material_type LIKE ? OR ms.supplier_name LIKE ? GROUP BY m.material_name ''', (f'%{keyword}%', f'%{keyword}%', f'%{keyword}%')) results = cursor.fetchall() return results except sqlite3.Error as e: print(f"❌ Ошибка поиска: {e}") return [] def close(self): """Закрытие соединения с базой данных""" if self.conn: self.conn.close() print("🔌 Соединение с базой данных закрыто") class PDFDomainAnalyzer: def __init__(self, db): self.db = db self.entities = {} self.relationships = [] def extract_text_from_pdf(self, pdf_path: str) -> str: """Извлечение текста из PDF файла""" try: with open(pdf_path, 'rb') as file: pdf_reader = PyPDF2.PdfReader(file) text = "" for page in pdf_reader.pages: text += page.extract_text() + "\n" return text except Exception as e: print(f"❌ Ошибка чтения PDF файла: {e}") return "" def analyze_domain_description(self, text: str): """Анализ описания предметной области""" print("🧠 АНАЛИЗИРУЮ ОПИСАНИЕ ПРЕДМЕТНОЙ ОБЛАСТИ...") print("=" * 60) entities_sections = re.split(r'\d+\.\s+', text) for section in entities_sections[1:]: self._analyze_entity_section(section) self._print_analysis_results() def _analyze_entity_section(self, section: str): """Анализ секции с описанием сущности""" lines = section.split('\n') if not lines: return entity_name = lines[0].strip().split(':')[0] attributes = [] print(f"🔍 Анализ сущности: {entity_name}") for line in lines[1:]: line = line.strip() if line and not line.startswith('–') and not line.startswith('====='): attributes.extend(self._extract_attributes(line)) self.entities[entity_name] = attributes def _extract_attributes(self, text: str) -> List[str]: """Извлечение атрибутов из текста""" attributes = [] patterns = [ r'([А-Я][а-я]+\s+[а-я]+\s*:)', r'(ИНН|ФИО|адрес|телефон|email|рейтинг|скидк|стоимость|количество)', r'(дата\s+[а-я]+|время\s+[а-я]+)', r'(артикул|описание|изображение|сертификат|стандарт)', r'(вес|размер|длина|ширина|высота)', r'(история\s+[а-я]+|изменен[ия]+\s+[а-я]+)' ] for pattern in patterns: matches = re.findall(pattern, text, re.IGNORECASE) attributes.extend(matches) return list(set(attributes)) def _print_analysis_results(self): """Вывод результатов анализа""" print("\n📋 РЕЗУЛЬТАТЫ АНАЛИЗА ПРЕДМЕТНОЙ ОБЛАСТИ:") print("=" * 60) for entity, attributes in self.entities.items(): print(f"\n🏷️ {entity.upper()}:") if attributes: for attr in attributes[:10]: print(f" 📌 {attr}") else: print(" 📌 Атрибуты не найдены в тексте") def generate_database_recommendations(self): """Генерация рекомендаций по базе данных на основе анализа""" print("\n💡 РЕКОМЕНДАЦИИ ПО РАЗРАБОТКЕ БАЗЫ ДАННЫХ:") print("=" * 60) recommendations = [] if 'Партнеры' in self.entities: recommendations.append("✅ Создана таблица partners с рейтингом и системой скидок") recommendations.append("✅ Добавлена история изменений рейтинга partner_rating_history") if any('менеджер' in entity.lower() for entity in self.entities): recommendations.append("✅ Создана таблица managers для работы с партнерами") if 'Заявка' in self.entities: recommendations.append("✅ Реализована сложная логика заявок с статусами и оплатами") recommendations.append("✅ Добавлены таблицы orders и order_products") if 'Сотрудники' in self.entities: recommendations.append("✅ Создана таблица employees с полными данными сотрудников") recommendations.append("✅ Реализована система допусков equipment_access") if 'Материалы' in self.entities: recommendations.append("✅ Создана универсальная таблица materials с автозаполнением") recommendations.append("✅ Реализована история изменений material_inventory_history") if 'Поставщики' in self.entities: recommendations.append("✅ Создана универсальная таблица suppliers с автозаполнением") recommendations.append("✅ Добавлена история поставок supply_history") if 'Продукция' in self.entities: recommendations.append("✅ Создана таблица products с полными характеристиками") recommendations.append("✅ Реализована система рецептур product_materials") for rec in recommendations: print(f" {rec}") print(f"\n🎯 Всего реализовано сущностей: {len(self.entities)}") print("🚀 База данных готова к использованию!") def main(): print("=== 🏭 УНИВЕРСАЛЬНАЯ ПРОИЗВОДСТВЕННАЯ СИСТЕМА ===") print("📊 Автоимпорт Excel + Анализ PDF + Полная БД") db = UniversalProductionDB() if not db.connect(): return analyzer = PDFDomainAnalyzer(db) try: while True: print("\n" + "=" * 60) print("Введите команду:") print(" 'создать' - создать полную структуру БД") print(" 'импорт' - умный импорт всех Excel данных") print(" 'примеры' - добавить демо-данные") print(" 'статистика' - статистика БД") print(" 'поиск <ключ>' - поиск материалов") print(" 'схема' - схема БД") print(" 'er' - ER-диаграмма") print(" 'анализ pdf' - анализ PDF с описанием") print(" 'файл <имя.pdf>' - анализ конкретного PDF") print(" 'анализ excel <файл>' - анализ структуры Excel") print(" 'выход' - завершить работу") command = input("\n➡️ Ваша команда: ").strip() if command.lower() == 'выход': break elif command.lower() == 'создать': db.create_complete_database_structure() elif command.lower() == 'импорт': db.import_all_excel_data() elif command.lower() == 'примеры': db.generate_sample_data() elif command.lower() == 'статистика': stats = db.get_database_stats() print("\n📊 СТАТИСТИКА БАЗЫ ДАННЫХ:") for table, count in stats.items(): print(f" {table}: {count} записей") elif command.startswith('поиск '): keyword = command[6:].strip() if keyword: results = db.search_materials(keyword) print(f"\n🔍 Результаты поиска по '{keyword}':") for material, m_type, price, quantity, measure, suppliers in results: print(f" 🏷️ {material} ({m_type})") print(f" 💰 {price} ₽, 📦 {quantity} {measure}") print(f" 🏢 Поставщики: {suppliers}") elif command.lower() == 'схема': db.analyze_database_schema() elif command.lower() == 'er': db.generate_er_diagram() elif command.lower() == 'анализ pdf': pdf_text = analyzer.extract_text_from_pdf('1.pdf') if pdf_text: analyzer.analyze_domain_description(pdf_text) analyzer.generate_database_recommendations() elif command.startswith('файл '): filename = command[5:].strip() if filename.endswith('.pdf'): if os.path.exists(filename): pdf_text = analyzer.extract_text_from_pdf(filename) if pdf_text: analyzer.analyze_domain_description(pdf_text) analyzer.generate_database_recommendations() else: print(f"❌ Файл {filename} не найден!") elif command.startswith('анализ excel '): filename = command[13:].strip() analysis = db.analyze_excel_structure(filename) if 'error' in analysis: print(f"❌ {analysis['error']}") else: print(f"\n📋 Анализ файла {filename}:") for sheet_name, sheet_info in analysis['sheets'].items(): print(f" 📄 Лист: {sheet_name}") print(f" Колонки: {sheet_info['columns']}") print(f" Записей: {sheet_info['row_count']}") else: print("❌ Неизвестная команда") finally: db.close() print("\n🎉 Работа системы завершена!") if __name__ == "__main__": main()