db.py

import os
import json
import logging
import uuid
import bcrypt
import requests
import mysql.connector
from mysql.connector import pooling, errors
from datetime import datetime, timedelta
from urllib.parse import urlparse, urlsplit
import jwt
from dotenv import load_dotenv
import re
from html import unescape

load_dotenv()

DB_CONFIG = {
    "host": os.environ.get("DB_HOST"),
    "user": os.environ.get("DB_USER"),
    "password": os.environ.get("DB_PASS"),
    "database": os.environ.get("DB_NAME"),
    "pool_name": "opensum_pool",
    "pool_size": int(os.environ.get("DB_POOL_SIZE", 5)),
}

SESSION_DURATION_HOURS = int(os.environ.get("SESSION_DURATION_HOURS", 24))

logger = logging.getLogger("db")
logging.basicConfig(level=logging.INFO)

cnxpool = None

def get_connection():
    return cnxpool.get_connection()

def dict_from_row(cursor, row): # needs cur cause it contains sql informations like columns etc
    if row is None:
        return None
    return {col[0]: row[idx] for idx, col in enumerate(cursor.description)}

def rows_to_dicts(cursor, rows): # needs cur cause it contains sql informations like columns etc
    cols = [c[0] for c in cursor.description]
    return [dict(zip(cols, row)) for row in rows]

def init_db():
    global cnxpool

    # temp conn
    temp_conn = mysql.connector.connect(
        host=DB_CONFIG["host"],
        user=DB_CONFIG["user"],
        password=DB_CONFIG["password"]
    )
    temp_cursor = temp_conn.cursor()

    # create db
    try:
        temp_cursor.execute(f"CREATE DATABASE IF NOT EXISTS {DB_CONFIG['database']}")
        logger.info(f"'{DB_CONFIG['database']}' db exists")
    except errors.DatabaseError as e:
        logger.error(f"err creating db: {e}")
        temp_conn.close()
        return False

    # close temp conn
    temp_cursor.close()
    temp_conn.close()

    # create conn pool
    cnxpool = pooling.MySQLConnectionPool(
        pool_name=DB_CONFIG["pool_name"],
        pool_size=DB_CONFIG["pool_size"],
        host=DB_CONFIG["host"],
        user=DB_CONFIG["user"],
        password=DB_CONFIG["password"],
        database=DB_CONFIG["database"]
    )

    # create tables and admin
    conn = cnxpool.get_connection()
    cursor = conn.cursor()

    try:
        cursor.execute("""
            CREATE TABLE IF NOT EXISTS users (
                id VARCHAR(36) PRIMARY KEY,
                username VARCHAR(50) NOT NULL UNIQUE,
                password_hash VARCHAR(255) NOT NULL,
                email VARCHAR(100) NOT NULL UNIQUE,
                created_at DATETIME NOT NULL,
                user_status TINYINT DEFAULT 1
            )
        """)

        cursor.execute("""
            CREATE TABLE IF NOT EXISTS sessions (
                id VARCHAR(36) PRIMARY KEY,
                user_id VARCHAR(36) NOT NULL,
                token VARCHAR(255) NOT NULL,
                expires_at DATETIME NOT NULL,
                FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
            )
        """)

        cursor.execute("""
            CREATE TABLE IF NOT EXISTS summaries (
                id VARCHAR(36) PRIMARY KEY,
                title VARCHAR(255) NOT NULL,
                summary TEXT NOT NULL,
                user_id VARCHAR(36) NOT NULL,
                created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
                FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
            )
        """)

        # check if there's any users in the database
        cursor.execute("SELECT COUNT(*) FROM users")
        user_count = cursor.fetchone()[0]

        if user_count == 0:
            user_id = str(uuid.uuid4())
            created_at = datetime.utcnow()
            password_hash = bcrypt.hashpw(b"password", bcrypt.gensalt()).decode('utf-8')

            cursor.execute(
                """
                INSERT INTO users (id, username, password_hash, email, created_at, user_status)
                VALUES (%s, %s, %s, %s, %s, %s)
                """,
                (user_id, "admin", password_hash, "admin@buage.dev", created_at, 1)
            )
            conn.commit()
            logger.info("admin user created!")
        else:
            logger.info("users already exist, skipping admin creation")

    except errors.ProgrammingError as e:
        logger.error(f"an error occured while creating admin user or db: {e}")
        return False
    finally:
        cursor.close()
        conn.close()

    return True

def get_user_by_email(email):
    conn = get_connection()

    try:
        cur = conn.cursor() # database connection, required to exec sql
        cur.execute("SELECT * FROM users WHERE email = %s", (email,)) # sql to exec
        row = cur.fetchone() # fetch only one result ( can use fetchall() too but useless here )

        if not row:
            return None

        user = dict_from_row(cur, row)

        user_obj = {
            "id": user.get("id"), # user id
            "username": user.get("username"),
            "email": user.get("email"),
            "password_hash": user.get("password_hash"),
            "createdAt": user.get("created_at"),
            "profile": json.loads(user.get("profile")) if user.get("profile") else {}, # user raw public profile data, will return something like {'bio': 'Test', 'id': '1' etc...}
            "publicKey": user.get("public_key"),
            "privateKey": user.get("private_key")
        }

        return user_obj
    finally:
        cur.close() # close the cursor, required to prevent memory leak or locks
        conn.close() # close connection (conn) in a clean way to prevent crashes

def authenticate(token):
    if not token:
        return None
    
    conn = get_connection()
    
    try:
        cur = conn.cursor(dictionary=True)
        
        cur.execute(
            'SELECT user_id FROM sessions WHERE token = %s AND expires_at > NOW()',
            (token,)
        )
        row = cur.fetchone()
        
        if not row:
            return None
        
        return get_user_by_id(row["user_id"])
    
    finally:
        cur.close()
        conn.close()

def get_user_by_session(token): # old version function
    return authenticate(token)

def login(email, password):
    user = get_user_by_email(email)
    
    if not user:
        return None
    
    password_hash = user.get("password_hash")
    
    if not password_hash or not bcrypt.checkpw(password.encode('utf-8'), password_hash.encode('utf-8')):
        return None
    
    session_token = os.urandom(32).hex()
    expires_at = datetime.utcnow() + timedelta(hours=int(SESSION_DURATION_HOURS))
    
    rid = str(uuid.uuid4())

    conn = get_connection()
    
    try:
        cur = conn.cursor()
        cur.execute(
            'INSERT INTO sessions (id, user_id, token, expires_at) VALUES (%s, %s, %s, %s)',
            (rid, user['id'], session_token, expires_at)
        )
        conn.commit()
    finally:
        cur.close()
        conn.close()
    
    return {"token": session_token, "expiresAt": expires_at}

def get_user_by_id(user_id):
    conn = get_connection()

    try:
        cur = conn.cursor()
        cur.execute('SELECT * FROM users WHERE id=%s', (user_id,))
        row = cur.fetchone()

        if not row:
            return None

        user = dict_from_row(cur, row)

        return {
            "id": user.get("id"),
            "username": user.get("username"),
            "email": user.get("email"),
            "password_hash": user.get("password_hash"),
            "createdAt": user.get("created_at"),
            "user_stats": user.get("user_status")
        }
    finally:
        cur.close()
        conn.close()

def update_username(user_id, new_username):
    conn = get_connection()

    try:
        cur = conn.cursor()
        cur.execute('UPDATE users SET username=%s WHERE id=%s', (new_username, user_id))
        row = cur.fetchone()
        conn.commit()

        if not row:
            return None

        user = dict_from_row(cur, row)

        return {
            "id": user.get("id"),
            "username": user.get("username"),
            "email": user.get("email"),
            "password_hash": user.get("password_hash"),
            "createdAt": user.get("created_at"),
            "user_stats": user.get("user_status")
        }
    finally:
        cur.close()
        conn.close()

def update_email(user_id, new_email):
    conn = get_connection()

    try:
        cur = conn.cursor()
        cur.execute('UPDATE users SET email=%s WHERE id=%s', (new_email, user_id))
        row = cur.fetchone()
        conn.commit()

        if not row:
            return None

        user = dict_from_row(cur, row)

        return {
            "id": user.get("id"),
            "username": user.get("username"),
            "email": user.get("email"),
            "password_hash": user.get("password_hash"),
            "createdAt": user.get("created_at"),
            "user_stats": user.get("user_status")
        }
    finally:
        cur.close()
        conn.close()

import bcrypt

def update_password(user_id, new_password):
    if not new_password:
        return None
    
    conn = get_connection()
    
    try:
        cur = conn.cursor()
        
        password_hash = bcrypt.hashpw(
            new_password.encode('utf-8'),
            bcrypt.gensalt()
        ).decode('utf-8')
        
        cur.execute(
            'UPDATE users SET password_hash=%s WHERE id=%s',
            (password_hash, user_id)
        )
        conn.commit()
        
        cur.execute('SELECT * FROM users WHERE id=%s', (user_id,))
        row = cur.fetchone()
        
        if not row:
            return None
        
        user = dict_from_row(cur, row)
        
        return {
            "id": user.get("id"),
            "username": user.get("username"),
            "email": user.get("email"),
            "password_hash": user.get("password_hash"),
            "createdAt": user.get("created_at"),
            "user_stats": user.get("user_stats")
        }
    except Exception as e:
        conn.rollback()
        raise
    finally:
        cur.close()
        conn.close()

def verify_password(user_id, pswrd):
    if not pswrd:
        return None
    
    conn = get_connection()
    
    try:
        cur = conn.cursor(dictionary=True)
        cur.execute(
            'SELECT password_hash FROM users WHERE id = %s',
            (user_id,)
        )
        row = cur.fetchone()
        
        if not row:
            return None
        
        password_hash = row["password_hash"]
        if not bcrypt.checkpw(pswrd.encode('utf-8'), password_hash.encode('utf-8')):
            return None
        
        return get_user_by_id(user_id)
    
    finally:
        cur.close()
        conn.close()

# summaries

def saveSummary(title, summary, userId):
    summary_id = str(uuid.uuid4())
    created_at = datetime.utcnow()

    conn = get_connection()

    try:
        cur = conn.cursor()
        cur.execute(
            """
            INSERT INTO summaries (id, title, summary, user_id, created_at)
            VALUES (%s, %s, %s, %s, %s)
            """,
            (summary_id, title, summary, userId, created_at)
        )
        conn.commit()
        logger.info(f"saved : {title}")
        return {"id": summary_id, "title": title, "createdAt": created_at}
    except errors.DatabaseError as e:
        logger.error(f"error saving summary: {e}")
        return None
    finally:
        cur.close()
        conn.close()

def getSummary(summary_id):
    conn = get_connection()

    try:
        cur = conn.cursor()
        cur.execute(
            "SELECT * FROM summaries WHERE id = %s",
            (summary_id,)
        )
        row = cur.fetchone()

        if not row:
            return None

        summary = dict_from_row(cur, row)
        return {
            "id": summary.get("id"),
            "title": summary.get("title"),
            "summary": summary.get("summary"),
            "userId": summary.get("user_id"),
            "createdAt": summary.get("created_at")
        }
    except errors.DatabaseError as e:
        logger.error(f"Erreur lors de la récupération du résumé: {e}")
        return None
    finally:
        cur.close()
        conn.close()

def getSummariesByUser(user_id):
    conn = get_connection()

    try:
        cur = conn.cursor()
        cur.execute(
            "SELECT * FROM summaries WHERE user_id = %s ORDER BY created_at DESC",
            (user_id,)
        )
        rows = cur.fetchall()

        if not rows:
            return []

        summaries = rows_to_dicts(cur, rows)
        return [{
            "id": s.get("id"),
            "title": s.get("title"),
            "summary": s.get("summary"),
            "userId": s.get("user_id"),
            "createdAt": s.get("created_at")
        } for s in summaries]
    except errors.DatabaseError as e:
        logger.error(f"Erreur lors de la récupération des résumés: {e}")
        return None
    finally:
        cur.close()
        conn.close()

def deleteSummary(summary_id):
    conn = get_connection()

    try:
        cur = conn.cursor()
        cur.execute(
            "DELETE FROM summaries WHERE id = %s",
            (summary_id,)
        )
        
        if cur.rowcount > 0:
            conn.commit() 
            return True
        else:
            return False 

    except Exception as e:
        logger.error(f"error deleting summary: {e}")
        conn.rollback()
        return None
    finally:
        cur.close()
        conn.close()


if not init_db():
    logger.error("error")
    raise RuntimeError("couldnt init the db")