""" Database models for enhanced server monitoring system """ from sqlalchemy import Column, Integer, String, DateTime, Text, Boolean, ForeignKey, LargeBinary, Float, Table from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import relationship from datetime import datetime import json import hashlib from cryptography.fernet import Fernet import base64 import os Base = declarative_base() # Association table for many-to-many relationship between inventory groups and devices device_inventory_association = Table( 'device_inventory_groups', Base.metadata, Column('device_id', Integer, ForeignKey('devices.id'), primary_key=True), Column('group_id', Integer, ForeignKey('inventory_groups.id'), primary_key=True) ) # Export all models __all__ = [ 'Base', 'Device', 'MessageTemplate', 'LogEntry', 'FileUpload', 'AnsibleExecution', 'SystemStats', 'InventoryGroup', 'PlaybookExecution', 'PlaybookHostResult', 'ExecutionQueue', 'device_inventory_association', 'WMTGlobalConfig', 'WMTUpdateRequest', 'ExecutionFailureReport', ] class Device(Base): """Device information and metadata""" __tablename__ = 'devices' id = Column(Integer, primary_key=True) hostname = Column(String(255), nullable=False) device_ip = Column(String(45), nullable=False) # Support IPv6 nume_masa = Column(String(100), nullable=False) # Enhanced device metadata device_type = Column(String(50), default='unknown') os_version = Column(String(100)) last_seen = Column(DateTime, default=datetime.utcnow) status = Column(String(20), default='active') # active, inactive, maintenance location = Column(String(200)) description = Column(Text) # WMT (Workstation Management Terminal) integration fields mac_address = Column(String(17), unique=True, nullable=True, index=True) config_updated_at = Column(DateTime) info_reviewed_at = Column(DateTime, default=lambda: datetime(1970, 1, 1)) # Relationships logs = relationship("LogEntry", back_populates="device") files = relationship("FileUpload", back_populates="device") inventory_groups = relationship("InventoryGroup", secondary=device_inventory_association, back_populates="devices") update_requests = relationship('WMTUpdateRequest', back_populates='device', cascade='all, delete-orphan', order_by='WMTUpdateRequest.submitted_at.desc()') @property def device_name(self): """Alias for nume_masa – used by WMT module.""" return self.nume_masa def __repr__(self): return f"" class MessageTemplate(Base): """Message templates for compression and aliases""" __tablename__ = 'message_templates' id = Column(Integer, primary_key=True) template_hash = Column(String(64), unique=True, nullable=False) # SHA-256 hash template_text = Column(Text, nullable=False) category = Column(String(50), nullable=False) # error, info, warning, system alias = Column(String(20), unique=True) # Short alias like "SYS001" usage_count = Column(Integer, default=0) created_at = Column(DateTime, default=datetime.utcnow) # Relationships log_entries = relationship("LogEntry", back_populates="template") @staticmethod def create_hash(message): """Create hash for message template""" return hashlib.sha256(message.encode('utf-8')).hexdigest() def __repr__(self): return f"" class LogEntry(Base): """Compressed log entries with template references""" __tablename__ = 'log_entries' id = Column(Integer, primary_key=True) device_id = Column(Integer, ForeignKey('devices.id'), nullable=False) template_id = Column(Integer, ForeignKey('message_templates.id')) # Original fields timestamp = Column(DateTime, default=datetime.utcnow, nullable=False) severity = Column(String(20), default='info') # debug, info, warning, error, critical # Compressed message storage full_message = Column(Text) # Only for unique messages not in templates template_variables = Column(Text) # JSON for template variable substitution # Enhanced metadata source_file = Column(String(255)) # If log comes from file line_number = Column(Integer) process_id = Column(Integer) thread_id = Column(String(50)) # Relationships device = relationship("Device", back_populates="logs") template = relationship("MessageTemplate", back_populates="log_entries") @property def resolved_message(self): """Get the full resolved message""" if self.template: if self.template_variables: variables = json.loads(self.template_variables) return self.template.template_text.format(**variables) return self.template.template_text return self.full_message def __repr__(self): return f"" class FileUpload(Base): """File upload tracking and metadata""" __tablename__ = 'file_uploads' id = Column(Integer, primary_key=True) device_id = Column(Integer, ForeignKey('devices.id'), nullable=False) # File metadata filename = Column(String(255), nullable=False) original_filename = Column(String(255), nullable=False) file_path = Column(String(500), nullable=False) # Server storage path file_size = Column(Integer, nullable=False) file_hash = Column(String(64)) # SHA-256 for deduplication mime_type = Column(String(100)) # Upload metadata upload_date = Column(DateTime, default=datetime.utcnow) upload_ip = Column(String(45)) upload_user_agent = Column(String(500)) # Processing status processed = Column(Boolean, default=False) processing_status = Column(String(50), default='pending') # pending, processing, completed, error processing_error = Column(Text) # File content analysis (for logs/config files) is_log_file = Column(Boolean, default=False) log_entries_extracted = Column(Integer, default=0) # Relationships device = relationship("Device", back_populates="files") def __repr__(self): return f"" class AnsibleExecution(Base): """ DEPRECATED MODEL - DO NOT USE FOR NEW DEVELOPMENT This model is kept only for backward compatibility and data migration. All new automation functionality should use PlaybookExecution instead. This table will be removed in a future version after data migration. """ __tablename__ = 'ansible_executions' id = Column(Integer, primary_key=True) playbook_name = Column(String(200), nullable=False) target_devices = Column(Text) # JSON list of device IDs/IPs # Execution details command_line = Column(Text, nullable=False) execution_user = Column(String(100)) start_time = Column(DateTime, default=datetime.utcnow) end_time = Column(DateTime) status = Column(String(20), default='running') # running, completed, failed, cancelled exit_code = Column(Integer) # Output and logs stdout_log = Column(Text) stderr_log = Column(Text) ansible_log_file = Column(String(500)) # Results summary successful_hosts = Column(Integer, default=0) failed_hosts = Column(Integer, default=0) unreachable_hosts = Column(Integer, default=0) # Relationships - Note: This class is deprecated, use PlaybookExecution instead @classmethod def migrate_to_new_model(cls, session): """Migrate this execution to new PlaybookExecution model""" # This method is used by migration scripts pass def __repr__(self): return f"" class SystemStats(Base): """System statistics and metrics""" __tablename__ = 'system_stats' id = Column(Integer, primary_key=True) device_id = Column(Integer, ForeignKey('devices.id'), nullable=False) # System metrics timestamp = Column(DateTime, default=datetime.utcnow) cpu_usage = Column(Float) memory_usage = Column(Float) disk_usage = Column(Float) network_in = Column(Integer) network_out = Column(Integer) load_average = Column(Float) uptime = Column(Integer) # seconds # Process counts total_processes = Column(Integer) running_processes = Column(Integer) # Temperature (for Raspberry Pi) cpu_temperature = Column(Float) # Relationships device = relationship("Device") def __repr__(self): return f"" class InventoryGroup(Base): """Ansible inventory groups with encrypted SSH credentials""" __tablename__ = 'inventory_groups' id = Column(Integer, primary_key=True) name = Column(String(100), unique=True, nullable=False) description = Column(Text) # SSH Connection details ssh_user = Column(String(100), default='pi') ssh_port = Column(Integer, default=22) ssh_key_file = Column(String(500)) # Path to SSH key ssh_password_encrypted = Column(LargeBinary) # Encrypted password # Group settings ansible_vars = Column(Text) # JSON for group variables is_enabled = Column(Boolean, default=True) created_at = Column(DateTime, default=datetime.utcnow) updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow) # Relationships devices = relationship("Device", secondary=device_inventory_association, back_populates="inventory_groups") executions = relationship("PlaybookExecution", back_populates="inventory_group") def set_ssh_password(self, password: str): """Encrypt and store SSH password""" if password: # Generate or get encryption key key = self._get_encryption_key() f = Fernet(key) self.ssh_password_encrypted = f.encrypt(password.encode()) def get_ssh_password(self) -> str: """Decrypt and return SSH password""" if self.ssh_password_encrypted: key = self._get_encryption_key() f = Fernet(key) return f.decrypt(self.ssh_password_encrypted).decode() return None def _get_encryption_key(self) -> bytes: """Get or generate encryption key""" key_file = 'data/.ssh_encrypt_key' if os.path.exists(key_file): with open(key_file, 'rb') as f: return f.read() else: # Generate new key key = Fernet.generate_key() with open(key_file, 'wb') as f: f.write(key) return key def __repr__(self): return f"" class PlaybookExecution(Base): """ Enhanced playbook execution with queue management and comprehensive tracking. This is the primary model for all automation execution tracking. Replaces the deprecated AnsibleExecution model. """ __tablename__ = 'playbook_executions' id = Column(Integer, primary_key=True) execution_id = Column(String(36), unique=True, nullable=False) # UUID playbook_name = Column(String(200), nullable=False) playbook_description = Column(Text) # User-friendly description inventory_group_id = Column(Integer, ForeignKey('inventory_groups.id')) target_hosts = Column(Text) # JSON list of specific hosts # Execution details command_line = Column(Text) extra_vars = Column(Text) # JSON execution_user = Column(String(100)) execution_ip = Column(String(45)) # Enhanced timing queued_at = Column(DateTime, default=datetime.utcnow) started_at = Column(DateTime) completed_at = Column(DateTime) estimated_duration = Column(Integer) # Estimated seconds # Enhanced status tracking status = Column(String(20), default='queued') # queued, running, completed, failed, cancelled, timeout queue_position = Column(Integer, default=0) priority = Column(Integer, default=5) # 1-10, higher = more priority pid = Column(Integer) # Process ID when running exit_code = Column(Integer) retry_count = Column(Integer, default=0) max_retries = Column(Integer, default=0) # Enhanced output and logs stdout_log = Column(Text) stderr_log = Column(Text) ansible_log_file = Column(String(500)) summary_message = Column(Text) # User-friendly summary # Enhanced results summary total_hosts = Column(Integer, default=0) successful_hosts = Column(Integer, default=0) failed_hosts = Column(Integer, default=0) unreachable_hosts = Column(Integer, default=0) skipped_hosts = Column(Integer, default=0) changed_hosts = Column(Integer, default=0) # Hosts where changes were made # Relationships inventory_group = relationship("InventoryGroup", back_populates="executions") host_results = relationship("PlaybookHostResult", back_populates="execution", cascade="all, delete-orphan") # Properties for better UX @property def duration(self): """Calculate execution duration in seconds""" if self.started_at and self.completed_at: return (self.completed_at - self.started_at).total_seconds() return None @property def duration_formatted(self): """Human-readable duration""" duration = self.duration if duration is None: return "N/A" if duration < 60: return f"{int(duration)}s" elif duration < 3600: mins = int(duration // 60) secs = int(duration % 60) return f"{mins}m {secs}s" else: hours = int(duration // 3600) mins = int((duration % 3600) // 60) return f"{hours}h {mins}m" @property def success_rate(self): """Calculate success rate percentage""" if self.total_hosts > 0: return round((self.successful_hosts / self.total_hosts) * 100, 1) return 0 @property def status_display(self): """User-friendly status display""" status_map = { 'queued': '⏳ Queued', 'running': '🔄 Running', 'completed': '✅ Completed', 'failed': '❌ Failed', 'cancelled': '🚫 Cancelled', 'timeout': '⏰ Timeout' } return status_map.get(self.status, self.status) @property def is_running(self): """Check if execution is currently running""" return self.status in ['queued', 'running'] @property def is_finished(self): """Check if execution has finished""" return self.status in ['completed', 'failed', 'cancelled', 'timeout'] @property def can_retry(self): """Check if execution can be retried""" return (self.status in ['failed', 'timeout'] and self.retry_count < self.max_retries) def get_host_summary(self): """Get summary of host results""" return { 'total': self.total_hosts, 'successful': self.successful_hosts, 'failed': self.failed_hosts, 'unreachable': self.unreachable_hosts, 'skipped': self.skipped_hosts, 'changed': self.changed_hosts } def get_failed_hosts(self): """Get list of failed hosts for debugging""" return [result for result in self.host_results if result.status == 'failed'] def get_status_color(self): """Get CSS color class for status""" color_map = { 'queued': 'text-warning', 'running': 'text-info', 'completed': 'text-success', 'failed': 'text-danger', 'cancelled': 'text-secondary', 'timeout': 'text-warning' } return color_map.get(self.status, 'text-secondary') def update_summary(self): """Update summary message based on execution results""" if self.status == 'completed': if self.failed_hosts == 0: self.summary_message = f"✅ Successfully executed on all {self.successful_hosts} hosts" else: self.summary_message = f"⚠️ Completed with {self.failed_hosts} failures out of {self.total_hosts} hosts" elif self.status == 'failed': self.summary_message = f"❌ Execution failed: {self.failed_hosts}/{self.total_hosts} hosts failed" elif self.status == 'running': self.summary_message = f"🔄 Executing on {self.total_hosts} hosts..." elif self.status == 'queued': self.summary_message = f"⏳ Queued for execution on {self.total_hosts} hosts" def __repr__(self): return f"" class PlaybookHostResult(Base): """Individual host results for playbook executions""" __tablename__ = 'playbook_host_results' id = Column(Integer, primary_key=True) execution_id = Column(String(36), ForeignKey('playbook_executions.execution_id'), nullable=False) device_id = Column(Integer, ForeignKey('devices.id'), nullable=False) hostname = Column(String(255), nullable=False) # Result details status = Column(String(20), nullable=False) # ok, failed, unreachable, skipped changed = Column(Boolean, default=False) failed_tasks = Column(Integer, default=0) total_tasks = Column(Integer, default=0) # Timing start_time = Column(DateTime) end_time = Column(DateTime) # Output specific to this host host_output = Column(Text) error_message = Column(Text) # Task results summary task_results = Column(Text) # JSON with per-task results # Relationships execution = relationship("PlaybookExecution", back_populates="host_results") device = relationship("Device") @property def duration(self): """Calculate host execution duration""" if self.start_time and self.end_time: return (self.end_time - self.start_time).total_seconds() return None @property def success_rate(self): """Calculate task success rate for this host""" if self.total_tasks > 0: successful_tasks = self.total_tasks - self.failed_tasks return (successful_tasks / self.total_tasks) * 100 return 0 def __repr__(self): return f"" class ExecutionQueue(Base): """Queue management for background playbook executions""" __tablename__ = 'execution_queue' id = Column(Integer, primary_key=True) execution_id = Column(String(36), ForeignKey('playbook_executions.execution_id'), nullable=False) queue_position = Column(Integer, nullable=False, default=0) priority = Column(Integer, default=5) # 1-10, higher = more priority # Queue metadata queued_by = Column(String(100)) queued_at = Column(DateTime, default=datetime.utcnow) scheduled_for = Column(DateTime) # For scheduled executions # Dependencies depends_on = Column(String(36), ForeignKey('playbook_executions.execution_id')) # Wait for this execution # Status is_active = Column(Boolean, default=True) # Relationships execution = relationship("PlaybookExecution", foreign_keys=[execution_id]) dependency = relationship("PlaybookExecution", foreign_keys=[depends_on]) def __repr__(self): return f"" # --------------------------------------------------------------------------- # WMT (Workstation Management Terminal) configuration models # --------------------------------------------------------------------------- class WMTGlobalConfig(Base): """Global WMT application settings – one row shared by all devices.""" __tablename__ = 'wmt_global_config' id = Column(Integer, primary_key=True) # Chrome launch URLs chrome_url = Column(String(500), nullable=False, default='http://10.76.140.17/iweb_v2/index.php/traceability/production') chrome_local_url = Column(String(500)) # optional local / fallback URL chrome_insecure_origin = Column(String(200), default='http://10.76.140.17') # Card API card_api_base_url = Column(String(500), nullable=False, default='https://dataswsibiusb01.sibiusb.harting.intra/RO_Quality_PRD/api/record') # Server connectivity server_log_url = Column(String(500), default='http://rpi-ansible:80/logs') internet_check_host = Column(String(200), default='10.76.140.17') update_host = Column(String(200), default='rpi-ansible') update_user = Column(String(100), default='pi') # Metadata updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow) updated_by = Column(String(100), default='admin') notes = Column(Text) def to_dict(self): return { 'id': self.id, 'chrome_url': self.chrome_url, 'chrome_local_url': self.chrome_local_url, 'chrome_insecure_origin': self.chrome_insecure_origin, 'card_api_base_url': self.card_api_base_url, 'server_log_url': self.server_log_url, 'internet_check_host': self.internet_check_host, 'update_host': self.update_host, 'update_user': self.update_user, 'updated_at': self.updated_at.isoformat() if self.updated_at else None, 'updated_by': self.updated_by, 'notes': self.notes, } def __repr__(self): return f"" class WMTUpdateRequest(Base): """Device-initiated update request awaiting admin approval.""" __tablename__ = 'wmt_update_requests' id = Column(Integer, primary_key=True) # Foreign key to Device (nullable – device may not be registered yet) device_id = Column(Integer, ForeignKey('devices.id'), nullable=True) mac_address = Column(String(17), nullable=False, index=True) # always stored # Data proposed by the device proposed_device_name = Column(String(100)) proposed_hostname = Column(String(255)) proposed_device_ip = Column(String(45)) # Request metadata submitted_at = Column(DateTime, default=datetime.utcnow) client_config_mtime = Column(String(30)) # ISO timestamp from the client # Admin decision status = Column(String(20), default='pending') # pending | accepted | rejected admin_reviewed_at = Column(DateTime) admin_notes = Column(Text) # Relationship device = relationship('Device', back_populates='update_requests') def to_dict(self): return { 'id': self.id, 'device_id': self.device_id, 'mac_address': self.mac_address, 'proposed_device_name': self.proposed_device_name, 'proposed_hostname': self.proposed_hostname, 'proposed_device_ip': self.proposed_device_ip, 'submitted_at': self.submitted_at.isoformat() if self.submitted_at else None, 'client_config_mtime': self.client_config_mtime, 'status': self.status, 'admin_reviewed_at': self.admin_reviewed_at.isoformat() if self.admin_reviewed_at else None, 'admin_notes': self.admin_notes, } def __repr__(self): return f"" class ExecutionFailureReport(Base): """Saved report of failed/unreachable hosts from a playbook execution.""" __tablename__ = 'execution_failure_reports' id = Column(Integer, primary_key=True) execution_id = Column(String(36), nullable=False, index=True) playbook_name = Column(String(255), nullable=False) saved_at = Column(DateTime, default=datetime.utcnow, nullable=False) # Counts failed_count = Column(Integer, default=0) unreachable_count = Column(Integer, default=0) # JSON list of {hostname, status, reason} objects failed_hosts = Column(Text, nullable=False, default='[]') # Optional note added by user when saving note = Column(Text) @property def hosts_list(self): try: return json.loads(self.failed_hosts) except Exception: return [] def to_dict(self): return { 'id': self.id, 'execution_id': self.execution_id, 'playbook_name': self.playbook_name, 'saved_at': self.saved_at.isoformat() if self.saved_at else None, 'failed_count': self.failed_count, 'unreachable_count': self.unreachable_count, 'failed_hosts': self.hosts_list, 'note': self.note, } def __repr__(self): return (f"")