import datetime import os import sqlite3 import time from sqlalchemy import create_engine, Column, Integer, String, Boolean, DateTime, Text, Float, text from sqlalchemy.exc import OperationalError from sqlalchemy.orm import declarative_base, sessionmaker, scoped_session from config import Config engine = create_engine( Config.DATABASE_URL, connect_args={"check_same_thread": False, "timeout": 30} if "sqlite" in Config.DATABASE_URL else {}, pool_pre_ping=True ) db_session = scoped_session(sessionmaker(autocommit=False, autoflush=False, bind=engine)) Base = declarative_base() Base.query = db_session.query_property() def get_fresh_db(): """Create fresh SQLite DB session per operation to bypass thread-local transaction caching""" Session = sessionmaker(bind=engine) return Session() class Account(Base): __tablename__ = 'accounts' id = Column(Integer, primary_key=True) phone_number = Column(String(50), nullable=True) session_string = Column(Text, nullable=False, unique=True) name = Column(String(100), nullable=True) is_active = Column(Boolean, default=True) created_at = Column(DateTime, default=datetime.datetime.utcnow) def to_dict(self): return { "id": self.id, "phone_number": self.phone_number or "N/A", "name": self.name or "Telegram Account", "is_active": self.is_active, "created_at": self.created_at.strftime("%Y-%m-%d %H:%M:%S") if self.created_at else "" } class Setting(Base): __tablename__ = 'settings' id = Column(Integer, primary_key=True) key = Column(String(100), unique=True, nullable=False) value = Column(Text, nullable=True) @classmethod def get(cls, key, default=""): for attempt in range(5): session = sessionmaker(bind=engine)() try: row = session.query(cls).filter_by(key=key).first() return row.value if row else default except OperationalError: session.rollback() if attempt == 4: return default time.sleep(0.15 * (attempt + 1)) finally: session.close() @classmethod def set(cls, key, value): last_error = None for attempt in range(8): session = sessionmaker(bind=engine)() try: row = session.query(cls).filter_by(key=key).first() if not row: session.add(cls(key=key, value=str(value))) else: row.value = str(value) session.commit() return except OperationalError as exc: last_error = exc session.rollback() time.sleep(0.2 * (attempt + 1)) finally: session.close() raise last_error class LiveConfig(Base): __tablename__ = 'live_config' id = Column(Integer, primary_key=True) target_chat = Column(String(200), nullable=False, default="") # Username or Chat ID is_auto_join = Column(Boolean, default=False) is_muted = Column(Boolean, default=True) is_joined = Column(Boolean, default=False) last_status = Column(Text, default="Idle") # Unlimited text length for diagnostic reports updated_at = Column(DateTime, default=datetime.datetime.utcnow, onupdate=datetime.datetime.utcnow) def to_dict(self): return { "id": self.id, "target_chat": self.target_chat, "is_auto_join": self.is_auto_join, "is_muted": self.is_muted, "is_joined": self.is_joined, "last_status": self.last_status, "updated_at": self.updated_at.strftime("%Y-%m-%d %H:%M:%S") if self.updated_at else "" } class ReactionRule(Base): __tablename__ = 'reaction_rules' id = Column(Integer, primary_key=True) target_chat = Column(String(200), nullable=False) # @channel or chat_id emojis = Column(String(200), default="👍,❤️,🔥,😍") # Comma separated is_enabled = Column(Boolean, default=True) delay_seconds = Column(Float, default=2.0) created_at = Column(DateTime, default=datetime.datetime.utcnow) def to_dict(self): return { "id": self.id, "target_chat": self.target_chat, "emojis": self.emojis, "is_enabled": self.is_enabled, "delay_seconds": self.delay_seconds, "created_at": self.created_at.strftime("%Y-%m-%d %H:%M:%S") if self.created_at else "" } class SystemLog(Base): __tablename__ = 'system_logs' id = Column(Integer, primary_key=True) level = Column(String(20), default="INFO") message = Column(Text, nullable=False) timestamp = Column(DateTime, default=datetime.datetime.utcnow) def to_dict(self): return { "id": self.id, "level": self.level, "message": self.message, "timestamp": self.timestamp.strftime("%Y-%m-%d %H:%M:%S") if self.timestamp else "" } def log_system_event(message, level="INFO"): for attempt in range(5): session = sessionmaker(bind=engine)() try: session.add(SystemLog(level=level, message=message)) session.commit() return except OperationalError as exc: session.rollback() if attempt == 4: print(f"Error logging to DB: {exc}") return time.sleep(0.15 * (attempt + 1)) except Exception as exc: session.rollback() print(f"Error logging to DB: {exc}") return finally: session.close() def init_db(): Base.metadata.create_all(bind=engine) if "sqlite" in Config.DATABASE_URL: with engine.begin() as conn: conn.execute(text("PRAGMA journal_mode=WAL")) conn.execute(text("PRAGMA synchronous=NORMAL")) conn.execute(text("PRAGMA busy_timeout=30000")) # Automatic schema migration check for legacy system_logs table with engine.connect() as conn: try: conn.execute(text("SELECT timestamp FROM system_logs LIMIT 1")) except Exception: try: conn.execute(text("ALTER TABLE system_logs ADD COLUMN timestamp DATETIME")) conn.commit() except Exception: pass # Ensure initial default live config if not db_session.query(LiveConfig).first(): db_session.add(LiveConfig(target_chat="", is_auto_join=False, is_muted=True)) db_session.commit() # Keep recoverable snapshots outside release archives. A deployment must # never be able to destroy the only copy of Telegram string sessions. if "sqlite" in Config.DATABASE_URL: try: account_count = db_session.query(Account).count() if account_count: backup_dir = os.path.join(os.path.dirname(Config.DATABASE_URL.replace("sqlite:///", "")), "database_backups") os.makedirs(backup_dir, exist_ok=True) latest = os.path.join(backup_dir, "database-latest.db") source_path = Config.DATABASE_URL.replace("sqlite:///", "") with sqlite3.connect(source_path) as source, sqlite3.connect(latest) as destination: source.backup(destination) except Exception as exc: print(f"Database backup notice: {exc}")