""" Database Manager for B2B Sales AI Agent Handles database initialization, migrations, and session management """ from sqlalchemy import create_engine, event from sqlalchemy.orm import sessionmaker, scoped_session from sqlalchemy.pool import StaticPool import os import logging from pathlib import Path from contextlib import contextmanager logger = logging.getLogger(__name__) class DatabaseManager: """ Manages SQLite database connections and sessions """ def __init__(self, db_path: str = None): """ Initialize database manager Args: db_path: Path to SQLite database file """ if db_path is None: # Default to data/cx_agent.db # For HuggingFace Spaces, try /data first (persistent), fallback to /tmp default_path = os.getenv('DATABASE_PATH', './data/cx_agent.db') # Check if we're on HuggingFace Spaces if os.path.exists('/data'): # HF Spaces with persistent storage default_path = '/data/cx_agent.db' elif os.path.exists('/tmp'): # Fallback to tmp if data dir not available default_path = '/tmp/cx_agent.db' db_path = default_path self.db_path = db_path self.engine = None self.Session = None def initialize(self): """Initialize database connection and create tables""" try: print(f"📂 Initializing database at: {self.db_path}") logger.info(f"Initializing database at: {self.db_path}") # Ensure data directory exists db_dir = Path(self.db_path).parent db_dir.mkdir(parents=True, exist_ok=True) print(f"📁 Database directory: {db_dir}") logger.info(f"Database directory created/verified: {db_dir}") # Create engine self.engine = create_engine( f'sqlite:///{self.db_path}', connect_args={'check_same_thread': False}, poolclass=StaticPool, echo=False # Set to True for SQL debugging ) # Enable foreign keys for SQLite @event.listens_for(self.engine, "connect") def set_sqlite_pragma(dbapi_conn, connection_record): cursor = dbapi_conn.cursor() cursor.execute("PRAGMA foreign_keys=ON") cursor.close() # Create session factory # expire_on_commit=False keeps objects accessible after commit session_factory = sessionmaker(bind=self.engine, expire_on_commit=False) self.Session = scoped_session(session_factory) # Import models and create tables try: from models.database import Base as EnterpriseBase EnterpriseBase.metadata.create_all(self.engine) print("✅ Enterprise tables created") logger.info("Enterprise tables created") except ImportError as e: print(f"⚠️ Could not import enterprise models: {e}") logger.warning(f"Could not import enterprise models: {e}") logger.info(f"Database initialized at {self.db_path}") # Initialize with default data self._initialize_default_data() return True except Exception as e: logger.error(f"Failed to initialize database: {str(e)}") raise def _initialize_default_data(self): """Insert default data for new databases""" try: from models.database import Setting, Sequence, SequenceEmail, Template session = self.Session() # Check if already initialized existing_settings = session.query(Setting).first() if existing_settings: session.close() return # Default settings default_settings = [ Setting(key='company_name', value='Your Company', description='Company name for email footers'), Setting(key='company_address', value='123 Main St, City, State 12345', description='Physical address for CAN-SPAM compliance'), Setting(key='sender_name', value='Sales Team', description='Default sender name'), Setting(key='sender_email', value='hello@example.com', description='Default sender email'), Setting(key='daily_email_limit', value='1000', description='Max emails per day'), Setting(key='enable_tracking', value='1', description='Enable email tracking'), ] session.add_all(default_settings) # Default sequence template: Cold Outreach (3-touch) cold_outreach = Sequence( name='Cold Outreach - 3 Touch', description='Standard 3-email cold outreach sequence', category='outbound', is_template=True ) session.add(cold_outreach) session.flush() sequence_emails = [ SequenceEmail( sequence_id=cold_outreach.id, step_number=1, wait_days=0, subject='Quick question about {{company_name}}', body='''Hi {{first_name}}, I noticed {{company_name}} is in the {{industry}} space with {{company_size}} employees. Companies like yours often face challenges with {{pain_points}}. We've helped similar companies reduce support costs by 35% and improve customer satisfaction significantly. Would you be open to a brief 15-minute call to explore if we might be able to help? Best regards, {{sender_name}}''' ), SequenceEmail( sequence_id=cold_outreach.id, step_number=2, wait_days=3, subject='Re: Quick question about {{company_name}}', body='''Hi {{first_name}}, I wanted to follow up on my previous email. I understand you're busy, so I'll keep this brief. We recently helped a company similar to {{company_name}} achieve: • 40% reduction in support ticket volume • 25% improvement in customer satisfaction scores • 30% faster response times I'd love to share how we did it. Are you available for a quick call this week? Best, {{sender_name}}''' ), SequenceEmail( sequence_id=cold_outreach.id, step_number=3, wait_days=7, subject='Last attempt - {{company_name}}', body='''Hi {{first_name}}, This is my last attempt to reach you. I completely understand if now isn't the right time. If you're interested in learning how we can help {{company_name}} improve customer experience, I'm happy to send over some quick resources. Otherwise, I'll assume this isn't a priority right now and won't bother you again. Thanks for your time, {{sender_name}} P.S. If you'd prefer to be removed from my list, just reply "Not interested" and I'll make sure you don't hear from me again.''' ), ] session.add_all(sequence_emails) # Default email templates templates = [ Template( name='Meeting Request', category='meeting_request', subject='Meeting invitation - {{company_name}}', body='''Hi {{first_name}}, Thank you for your interest! I'd love to schedule a call to discuss how we can help {{company_name}}. Here are a few time slots that work for me: • {{time_slot_1}} • {{time_slot_2}} • {{time_slot_3}} Let me know which works best for you, or feel free to suggest another time. Looking forward to speaking with you! Best, {{sender_name}}''', variables='["first_name", "company_name", "time_slot_1", "time_slot_2", "time_slot_3", "sender_name"]' ), Template( name='Follow-up After Meeting', category='follow_up', subject='Great speaking with you, {{first_name}}', body='''Hi {{first_name}}, Thanks for taking the time to speak with me today about {{company_name}}'s customer experience goals. As discussed, here are the next steps: • {{next_step_1}} • {{next_step_2}} I'll follow up on {{follow_up_date}} as we agreed. Please don't hesitate to reach out if you have any questions in the meantime. Best regards, {{sender_name}}''', variables='["first_name", "company_name", "next_step_1", "next_step_2", "follow_up_date", "sender_name"]' ), ] session.add_all(templates) session.commit() session.close() logger.info("Default data initialized successfully") except Exception as e: logger.error(f"Failed to initialize default data: {str(e)}") if session: session.rollback() session.close() @contextmanager def get_session(self): """ Context manager for database sessions Usage: with db_manager.get_session() as session: session.query(Contact).all() """ session = self.Session() try: yield session session.commit() except Exception: session.rollback() raise finally: session.close() def close(self): """Close database connection""" if self.Session: self.Session.remove() if self.engine: self.engine.dispose() logger.info("Database connection closed") # Global database manager instance _db_manager = None def get_db_manager() -> DatabaseManager: """Get or create global database manager instance""" global _db_manager if _db_manager is None: _db_manager = DatabaseManager() _db_manager.initialize() return _db_manager def init_database(db_path: str = None): """Initialize database with custom path""" global _db_manager _db_manager = DatabaseManager(db_path) _db_manager.initialize() return _db_manager