database.py

# Commit test to ensure push
import sqlite3
import random
import string
import logging
import re
from typing import List, Dict, Any, Optional
from datetime import datetime

# Configure logging
logging.basicConfig(
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    level=logging.INFO,
    handlers=[
        logging.FileHandler("bot.log", encoding='utf-8'),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger(__name__)

class Database:
    def __init__(self, db_name: str = "cars.db"):
        try:
            self.conn = sqlite3.connect(db_name, check_same_thread=False)
            self.cursor = self.conn.cursor()
            self.create_tables()
            self.update_schema()
            logger.info(f"Successfully connected to database: {db_name}")
        except sqlite3.Error as e:
            logger.error(f"Failed to connect to database: {str(e)}")
            raise

    def create_tables(self):
        """Create database tables and indexes if they don't exist."""
        try:
            self.cursor.executescript('''
                -- Table for users
                CREATE TABLE IF NOT EXISTS users (
                    user_id INTEGER PRIMARY KEY,
                    username TEXT,
                    phone TEXT,
                    shares INTEGER DEFAULT 0,
                    daily_recommendations INTEGER DEFAULT 0,
                    last_recommendation_date TEXT
                );

                -- Table for ads
                CREATE TABLE IF NOT EXISTS ads (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    user_id INTEGER,
                    price TEXT,
                    car_name_ar TEXT,
                    car_name_en TEXT,
                    model TEXT,
                    engine_specs TEXT,
                    mileage TEXT,
                    origin TEXT,
                    plate TEXT,
                    accidents TEXT,
                    specs TEXT,
                    notes TEXT,
                    location TEXT,
                    phone TEXT,
                    main_photo TEXT,
                    photos TEXT,
                    status TEXT DEFAULT 'pending',
                    channel_message_id INTEGER,
                    created_at TEXT,
                    reactions TEXT DEFAULT '',
                    FOREIGN KEY(user_id) REFERENCES users(user_id)
                );

                -- Table for verification codes
                CREATE TABLE IF NOT EXISTS verification_codes (
                    user_id INTEGER,
                    phone TEXT,
                    code TEXT,
                    PRIMARY KEY(user_id, phone)
                );

                -- Table for reactions
                CREATE TABLE IF NOT EXISTS reactions (
                    ad_id INTEGER,
                    reaction_type TEXT,
                    count INTEGER,
                    last_updated TEXT,
                    PRIMARY KEY(ad_id, reaction_type),
                    FOREIGN KEY(ad_id) REFERENCES ads(id)
                );

                -- Table for stats
                CREATE TABLE IF NOT EXISTS stats (
                    id INTEGER PRIMARY KEY,
                    subscribers_count INTEGER DEFAULT 1300,
                    last_updated TEXT
                );

                -- Insert default stats
                INSERT OR IGNORE INTO stats (id, subscribers_count) VALUES (1, 1300);

                -- Indexes for faster queries
                CREATE INDEX IF NOT EXISTS idx_ads_user_id ON ads(user_id);
                CREATE INDEX IF NOT EXISTS idx_ads_status ON ads(status);
                CREATE INDEX IF NOT EXISTS idx_ads_car_name_ar ON ads(car_name_ar);
                CREATE INDEX IF NOT EXISTS idx_ads_location ON ads(location);
                CREATE INDEX IF NOT EXISTS idx_users_phone ON users(phone);
                -- Added index on created_at
                CREATE INDEX IF NOT EXISTS idx_ads_created_at ON ads(created_at);
            ''')
            self.conn.commit()
            logger.info("Database tables and indexes created successfully")
        except sqlite3.Error as e:
            logger.error(f"Error creating tables: {str(e)}")
            raise

    def update_schema(self):
        """Update database schema if needed (e.g., add new columns)."""
        try:
            # Check if 'username' column exists in users table
            self.cursor.execute("PRAGMA table_info(users)")
            columns = [info[1] for info in self.cursor.fetchall()]
            if 'username' not in columns:
                self.cursor.execute('ALTER TABLE users ADD COLUMN username TEXT')
                self.conn.commit()
                logger.info("Added 'username' column to users table")
        except sqlite3.Error as e:
            logger.error(f"Error updating schema: {str(e)}")
            raise

    def add_user(self, user_id: int, username: Optional[str] = None):
        """Add or update a user in the database."""
        try:
            self.cursor.execute('''
                INSERT OR IGNORE INTO users (user_id, username, shares, daily_recommendations)
                VALUES (?, ?, 0, 0)
            ''', (user_id, username))
            self.cursor.execute('''
                UPDATE stats SET subscribers_count = subscribers_count + 1, last_updated = ?
                WHERE id = 1
            ''', (datetime.now().strftime('%Y-%m-%d %H:%M:%S'),))
            self.conn.commit()
            logger.info(f"User {user_id} (username: {username}) added/updated")
        except sqlite3.Error as e:
            logger.error(f"Error adding user {user_id}: {str(e)}")
            raise

    def get_user_shares(self, user_id: int) -> int:
        """Get the number of shares for a user."""
        try:
            self.cursor.execute('SELECT shares FROM users WHERE user_id = ?', (user_id,))
            result = self.cursor.fetchone()
            return result[0] if result else 0
        except sqlite3.Error as e:
            logger.error(f"Error retrieving shares for user {user_id}: {str(e)}")
            return 0

    def increment_shares(self, user_id: int):
        """Increment the share count for a user."""
        try:
            self.cursor.execute('UPDATE users SET shares = shares + 1 WHERE user_id = ?', (user_id,))
            self.conn.commit()
            logger.info(f"Shares incremented for user {user_id}")
        except sqlite3.Error as e:
            logger.error(f"Error incrementing shares for user {user_id}: {str(e)}")

    def get_daily_recommendations(self, user_id: int) -> int:
        """Get the number of daily recommendations for a user."""
        try:
            self.cursor.execute('SELECT daily_recommendations, last_recommendation_date FROM users WHERE user_id = ?', (user_id,))
            result = self.cursor.fetchone()
            if result:
                count, date = result
                today = datetime.now().strftime('%Y-%m-%d')
                if date != today:
                    self.cursor.execute('UPDATE users SET daily_recommendations = 0, last_recommendation_date = ? WHERE user_id = ?', (today, user_id))
                    self.conn.commit()
                    logger.info(f"Reset daily recommendations for user {user_id} to 0")
                    return 0
                return count
            return 0
        except sqlite3.Error as e:
            logger.error(f"Error retrieving daily recommendations for user {user_id}: {str(e)}")
            return 0

    def increment_recommendations(self, user_id: int):
        """Increment the daily recommendation count for a user."""
        try:
            today = datetime.now().strftime('%Y-%m-%d')
            self.cursor.execute('''
                UPDATE users SET daily_recommendations = daily_recommendations + 1,
                last_recommendation_date = ? WHERE user_id = ?
            ''', (today, user_id))
            self.conn.commit()
            logger.info(f"Recommendations incremented for user {user_id}")
        except sqlite3.Error as e:
            logger.error(f"Error incrementing recommendations for user {user_id}: {str(e)}")

    def generate_verification_code(self, user_id: int, phone: str) -> str:
        """Generate a verification code for a user."""
        try:
            code = ''.join(random.choices(string.digits, k=6))
            self.cursor.execute('INSERT OR REPLACE INTO verification_codes (user_id, phone, code) VALUES (?, ?, ?)',
                               (user_id, phone, code))
            self.conn.commit()
            logger.info(f"Generated verification code for user {user_id} with phone {phone}")
            return code
        except sqlite3.Error as e:
            logger.error(f"Error generating verification code for user {user_id}: {str(e)}")
            raise

    def verify_phone_code(self, user_id: int, code: str) -> bool:
        """Verify a phone code for a user."""
        try:
            self.cursor.execute('SELECT code FROM verification_codes WHERE user_id = ?', (user_id,))
            result = self.cursor.fetchone()
            if result and result[0] == code:
                self.cursor.execute('DELETE FROM verification_codes WHERE user_id = ?', (user_id,))
                self.conn.commit()
                logger.info(f"Phone code verified for user {user_id}")
                return True
            logger.warning(f"Invalid phone code for user {user_id}")
            return False
        except sqlite3.Error as e:
            logger.error(f"Error verifying phone code for user {user_id}: {str(e)}")
            return False

    def create_car(self, model: str, price: float, status: str, description: Optional[str] = None) -> int:
        """Create a new car ad in the database."""
        try:
            created_at = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
            self.cursor.execute('''
                INSERT INTO ads (price, car_name_ar, car_name_en, model, status, notes, created_at)
                VALUES (?, ?, ?, ?, ?, ?, ?)
            ''', (
                str(price), model, model, model, status, description, created_at
            ))
            ad_id = self.cursor.lastrowid
            self.conn.commit()
            logger.info(f"Created car ad {ad_id} with model {model}")
            return ad_id
        except sqlite3.Error as e:
            logger.error(f"Error creating car ad: {str(e)}")
            raise

    def get_car_by_id(self, car_id: int) -> Optional[Dict[str, Any]]:
        """Retrieve a car ad by its ID."""
        try:
            self.cursor.execute('SELECT id, price, car_name_ar AS model, status, notes AS description, created_at FROM ads WHERE id = ?', (car_id,))
            car = self.cursor.fetchone()
            if car:
                columns = [desc[0] for desc in self.cursor.description]
                car_dict = dict(zip(columns, car))
                logger.info(f"Retrieved car {car_id}")
                return car_dict
            logger.warning(f"Car {car_id} not found")
            return None
        except sqlite3.Error as e:
            logger.error(f"Error retrieving car {car_id}: {str(e)}")
            raise

    def get_all_cars(self) -> List[Dict[str, Any]]:
        """Retrieve all approved car ads."""
        try:
            self.cursor.execute('SELECT id, price, car_name_ar AS model, status, notes AS description, created_at FROM ads WHERE status = "approved"')
            cars = self.cursor.fetchall()
            columns = [desc[0] for desc in self.cursor.description]
            cars_list = [dict(zip(columns, car)) for car in cars]
            logger.info(f"Retrieved {len(cars_list)} approved cars")
            return cars_list
        except sqlite3.Error as e:
            logger.error(f"Error retrieving all cars: {str(e)}")
            raise

    def save_ad(self, ad_data: Dict[str, Any], user_id: int) -> int:
        """Save a new ad in the database."""
        try:
            created_at = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
            self.cursor.execute('''
                INSERT INTO ads (user_id, price, car_name_ar, car_name_en, model, engine_specs, mileage, origin, plate,
                                accidents, specs, notes, location, phone, main_photo, photos, created_at)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            ''', (
                user_id, ad_data['price'], ad_data['car_name_ar'], ad_data['car_name_en'], ad_data['model'],
                ad_data['engine_specs'], ad_data['mileage'], ad_data['origin'], ad_data['plate'], ad_data['accidents'],
                ad_data['specs'], ad_data.get('notes'), ad_data['location'], ad_data['phone'], ad_data['main_photo'],
                ','.join(ad_data['photos']), created_at
            ))
            ad_id = self.cursor.lastrowid
            self.conn.commit()
            logger.info(f"Ad {ad_id} saved for user {user_id}")
            return ad_id
        except sqlite3.Error as e:
            logger.error(f"Error saving ad for user {user_id}: {str(e)}")
            raise

    def get_ad(self, ad_id: int) -> Optional[Dict[str, Any]]:
        """Retrieve an ad by its ID."""
        try:
            self.cursor.execute('SELECT * FROM ads WHERE id = ?', (ad_id,))
            ad = self.cursor.fetchone()
            if ad:
                columns = [desc[0] for desc in self.cursor.description]
                ad_dict = dict(zip(columns, ad))
                ad_dict['photos'] = ad_dict['photos'].split(',') if ad_dict['photos'] else []
                logger.info(f"Retrieved ad {ad_id}")
                return ad_dict
            logger.warning(f"Ad {ad_id} not found")
            return None
        except sqlite3.Error as e:
            logger.error(f"Error retrieving ad {ad_id}: {str(e)}")
            raise

    def update_ad_field(self, ad_id: int, field: str, value: Any):
        """Update a specific field for an ad."""
        try:
            self.cursor.execute(f'UPDATE ads SET {field} = ? WHERE id = ?', (value, ad_id))
            self.conn.commit()
            logger.info(f"Updated field {field} for ad {ad_id}")
        except sqlite3.Error as e:
            logger.error(f"Error updating field {field} for ad {ad_id}: {str(e)}")
            raise

    def update_ad_status(self, ad_id: int, status: str):
        """Update the status of an ad."""
        try:
            self.cursor.execute('UPDATE ads SET status = ? WHERE id = ?', (status, ad_id))
            self.conn.commit()
            logger.info(f"Updated status to {status} for ad {ad_id}")
        except sqlite3.Error as e:
            logger.error(f"Error updating status for ad {ad_id}: {str(e)}")
            raise

    def get_available_cars(self, user_id: Optional[int] = None) -> List[Dict[str, Any]]:
        """Retrieve approved ads, optionally filtered by user ID."""
        try:
            if user_id:
                self.cursor.execute('SELECT * FROM ads WHERE status = "approved" AND user_id = ?', (user_id,))
            else:
                self.cursor.execute('SELECT * FROM ads WHERE status = "approved"')
            ads = self.cursor.fetchall()
            columns = [desc[0] for desc in self.cursor.description]
            result = []
            for ad in ads:
                ad_dict = dict(zip(columns, ad))
                ad_dict['photos'] = ad_dict['photos'].split(',') if ad_dict['photos'] else []
                result.append(ad_dict)
            logger.info(f"Retrieved {len(result)} available cars")
            return result
        except sqlite3.Error as e:
            logger.error(f"Error retrieving available cars: {str(e)}")
            raise

    def save_channel_message(self, ad_id: int, message_id: int):
        """Save a channel message ID for an ad."""
        try:
            self.cursor.execute('UPDATE ads SET channel_message_id = ? WHERE id = ?', (message_id, ad_id))
            self.conn.commit()
            logger.info(f"Saved channel message ID {message_id} for ad {ad_id}")
        except sqlite3.Error as e:
            logger.error(f"Error saving channel message for ad {ad_id}: {str(e)}")
            raise

    def get_total_users(self) -> int:
        """Get the total number of users."""
        try:
            self.cursor.execute("SELECT subscribers_count FROM stats WHERE id = 1")
            result = self.cursor.fetchone()
            count = result[0] if result else 1300
            logger.info(f"Retrieved total users: {count}")
            return count
        except sqlite3.Error as e:
            logger.error(f"Error retrieving total users: {str(e)}")
            return 1300

    def search_ads(self, query: str) -> List[Dict[str, Any]]:
        """Search ads based on a query string."""
        try:
            query = query.lower().strip()
            search_terms = re.split(r'\s+', query)
            conditions = ["status = 'approved'"]
            params = []

            for term in search_terms:
                if term.isdigit() and len(term) == 4:  # Model year (e.g., 2020)
                    conditions.append("model LIKE ?")
                    params.append(f"%{term}%")
                elif re.match(r'(\d+\.?\d*)\s*(دولار|دينار|-)?(?:\s*(\d+\.?\d*))?', term):  # Price or range (e.g., 20 دولار, 20-30)
                    price_match = re.match(r'(\d+\.?\d*)\s*(دولار|دينار|-)?(?:\s*(\d+\.?\d*))?', term)
                    if price_match:
                        low_price = float(price_match.group(1))
                        currency = price_match.group(2)
                        high_price = float(price_match.group(3)) if price_match.group(3) else None
                        if currency == '-':
                            conditions.append("CAST(replace(replace(price, '$', ''), ' ألف دولار', '000') AS REAL) BETWEEN ? AND ?")
                            params.extend([low_price * 1000, high_price * 1000])
                        else:
                            conditions.append("LOWER(price) LIKE ?")
                            params.append(f'%{low_price}%{currency if currency else ""}%')
                else:  # Car name, location, or specs
                    conditions.append("(LOWER(car_name_ar) LIKE ? OR LOWER(car_name_en) LIKE ? OR LOWER(location) LIKE ? OR LOWER(specs) LIKE ?)")
                    params.extend([f"%{term}%", f"%{term}%", f"%{term}%", f"%{term}%"])

            where_clause = " AND ".join(conditions) if conditions else "1=1"
            self.cursor.execute(f'SELECT * FROM ads WHERE {where_clause}', params)
            ads = self.cursor.fetchall()
            columns = [desc[0] for desc in self.cursor.description]
            result = [dict(zip(columns, ad)) for ad in ads]
            for ad in result:
                ad['photos'] = ad['photos'].split(',') if ad['photos'] else []
            logger.info(f"Found {len(result)} ads for query: {query}")
            return result
        except sqlite3.Error as e:
            logger.error(f"Error searching ads for query {query}: {str(e)}")
            raise

    def add_reaction(self, ad_id: int, ad: Dict[str, Any]):
        """Add reactions to an ad."""
        try:
            from utils import generate_reactions
            reactions = generate_reactions(ad)
            self.cursor.execute('UPDATE ads SET reactions = ? WHERE id = ?', (reactions, ad_id))
            self.conn.commit()
            logger.info(f"Added reactions for ad {ad_id}")
        except sqlite3.Error as e:
            logger.error(f"Error adding reactions for ad {ad_id}: {str(e)}")
            raise
        except ImportError:
            logger.error(f"Failed to import generate_reactions from utils for ad {ad_id}")
            raise

    def extend_ad_duration(self, ad_id: int):
        """Extend the duration of an ad."""
        try:
            self.cursor.execute('UPDATE ads SET status = "approved" WHERE id = ?', (ad_id,))
            self.conn.commit()
            logger.info(f"Extended duration for ad {ad_id}")
        except sqlite3.Error as e:
            logger.error(f"Error extending duration for ad {ad_id}: {str(e)}")
            raise

    def close(self):
        """Close the database connection."""
        try:
            self.conn.close()
            logger.info("Database connection closed successfully")
        except sqlite3.Error as e:
            logger.error(f"Error closing database: {str(e)}")