423 lines
17 KiB
Python
423 lines
17 KiB
Python
import sqlite3
|
|
import mysql.connector
|
|
import json
|
|
from mysql.connector import Error as MySQLError
|
|
from botdiscord.config import get_db_config, get_db_type
|
|
|
|
_connection = None
|
|
|
|
def get_connection():
|
|
global _connection
|
|
db_type = get_db_type()
|
|
|
|
if db_type == "mysql":
|
|
if _connection is None or not _connection.is_connected():
|
|
db_config = get_db_config()
|
|
try:
|
|
_connection = mysql.connector.connect(
|
|
host=db_config.get("host", "localhost"),
|
|
port=db_config.get("port", 3306),
|
|
user=db_config.get("user", "root"),
|
|
password=db_config.get("password", ""),
|
|
database=db_config.get("name", "mi_red")
|
|
)
|
|
except MySQLError as e:
|
|
print(f"Error connecting to MySQL: {e}")
|
|
raise
|
|
return _connection
|
|
else:
|
|
import os
|
|
from botdiscord.config import get_db_path
|
|
db_path = get_db_path()
|
|
db_dir = os.path.dirname(db_path)
|
|
if db_dir and not os.path.exists(db_dir):
|
|
os.makedirs(db_dir, exist_ok=True)
|
|
return sqlite3.connect(db_path)
|
|
|
|
def close_connection():
|
|
global _connection
|
|
if _connection and _connection.is_connected():
|
|
_connection.close()
|
|
_connection = None
|
|
|
|
def _execute_query(query, params=None, fetch=False):
|
|
db_type = get_db_type()
|
|
|
|
if db_type == "mysql":
|
|
conn = get_connection()
|
|
cursor = conn.cursor()
|
|
try:
|
|
cursor.execute(query, params or ())
|
|
if fetch:
|
|
result = cursor.fetchall()
|
|
else:
|
|
conn.commit()
|
|
result = cursor.lastrowid
|
|
return result
|
|
finally:
|
|
cursor.close()
|
|
else:
|
|
conn = get_connection()
|
|
cursor = conn.cursor()
|
|
try:
|
|
cursor.execute(query, params or ())
|
|
if fetch:
|
|
result = cursor.fetchall()
|
|
else:
|
|
conn.commit()
|
|
result = cursor.lastrowid
|
|
return result
|
|
finally:
|
|
cursor.close()
|
|
|
|
def init_db():
|
|
db_type = get_db_type()
|
|
|
|
if db_type == "mysql":
|
|
conn = get_connection()
|
|
cursor = conn.cursor()
|
|
cursor.execute('''CREATE TABLE IF NOT EXISTS guild_languages
|
|
(guild_id BIGINT NOT NULL, lang_code VARCHAR(10) NOT NULL,
|
|
PRIMARY KEY (guild_id, lang_code)) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4''')
|
|
cursor.execute('''CREATE TABLE IF NOT EXISTS bot_config
|
|
(`key` VARCHAR(255) NOT NULL, value TEXT,
|
|
PRIMARY KEY (`key`)) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4''')
|
|
cursor.execute('''CREATE TABLE IF NOT EXISTS available_languages
|
|
(code VARCHAR(10) NOT NULL, name VARCHAR(100) NOT NULL, flag VARCHAR(20),
|
|
PRIMARY KEY (code)) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4''')
|
|
cursor.execute('''CREATE TABLE IF NOT EXISTS bot_languages
|
|
(bot_type VARCHAR(50) NOT NULL, lang_code VARCHAR(10) NOT NULL,
|
|
PRIMARY KEY (bot_type, lang_code)) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4''')
|
|
|
|
# Nuevas tablas para caché y persistencia
|
|
cursor.execute('''CREATE TABLE IF NOT EXISTS messages
|
|
(message_id BIGINT PRIMARY KEY,
|
|
guild_id BIGINT,
|
|
author_id BIGINT,
|
|
content TEXT NOT NULL,
|
|
mentions_map JSON,
|
|
bot_type ENUM('discord', 'telegram') NOT NULL,
|
|
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4''')
|
|
|
|
cursor.execute('''CREATE TABLE IF NOT EXISTS translations
|
|
(id INT AUTO_INCREMENT PRIMARY KEY,
|
|
message_id BIGINT NOT NULL,
|
|
target_lang VARCHAR(10) NOT NULL,
|
|
translated_text TEXT NOT NULL,
|
|
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
|
FOREIGN KEY (message_id) REFERENCES messages(message_id) ON DELETE CASCADE,
|
|
UNIQUE KEY idx_msg_lang (message_id, target_lang)) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4''')
|
|
|
|
# Tabla para traducciones de la interfaz web
|
|
cursor.execute('''CREATE TABLE IF NOT EXISTS ui_translations
|
|
(id INT AUTO_INCREMENT PRIMARY KEY,
|
|
original_text TEXT NOT NULL,
|
|
target_lang VARCHAR(10) NOT NULL,
|
|
translated_text TEXT NOT NULL,
|
|
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
|
UNIQUE KEY idx_ui_lang (original_text(255), target_lang)) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4''')
|
|
|
|
conn.commit()
|
|
cursor.close()
|
|
else:
|
|
conn = get_connection()
|
|
c = conn.cursor()
|
|
c.execute('''CREATE TABLE IF NOT EXISTS guild_languages
|
|
(guild_id INTEGER, lang_code TEXT, PRIMARY KEY (guild_id, lang_code))''')
|
|
c.execute('''CREATE TABLE IF NOT EXISTS bot_config
|
|
(key TEXT PRIMARY KEY, value TEXT)''')
|
|
c.execute('''CREATE TABLE IF NOT EXISTS available_languages
|
|
(code TEXT PRIMARY KEY, name TEXT, flag TEXT)''')
|
|
c.execute('''CREATE TABLE IF NOT EXISTS bot_languages
|
|
(bot_type TEXT, lang_code TEXT, PRIMARY KEY (bot_type, lang_code))''')
|
|
|
|
# SQLite equivalents
|
|
c.execute('''CREATE TABLE IF NOT EXISTS messages
|
|
(message_id INTEGER PRIMARY KEY,
|
|
guild_id INTEGER,
|
|
author_id INTEGER,
|
|
content TEXT NOT NULL,
|
|
mentions_map TEXT,
|
|
bot_type TEXT NOT NULL,
|
|
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP)''')
|
|
|
|
c.execute('''CREATE TABLE IF NOT EXISTS translations
|
|
(id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
message_id INTEGER NOT NULL,
|
|
target_lang TEXT NOT NULL,
|
|
translated_text TEXT NOT NULL,
|
|
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
|
UNIQUE(message_id, target_lang))''')
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
def set_available_languages(languages: list):
|
|
db_type = get_db_type()
|
|
|
|
if db_type == "mysql":
|
|
conn = get_connection()
|
|
cursor = conn.cursor()
|
|
cursor.execute("DELETE FROM available_languages")
|
|
for lang in languages:
|
|
cursor.execute("INSERT INTO available_languages (code, name, flag) VALUES (%s, %s, %s)",
|
|
(lang.get("code"), lang.get("name"), lang.get("flag", "")))
|
|
conn.commit()
|
|
cursor.close()
|
|
else:
|
|
conn = get_connection()
|
|
c = conn.cursor()
|
|
c.execute("DELETE FROM available_languages")
|
|
for lang in languages:
|
|
c.execute("INSERT OR REPLACE INTO available_languages (code, name, flag) VALUES (?, ?, ?)",
|
|
(lang.get("code"), lang.get("name"), lang.get("flag", "")))
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
def get_available_languages() -> list:
|
|
db_type = get_db_type()
|
|
|
|
if db_type == "mysql":
|
|
conn = get_connection()
|
|
cursor = conn.cursor()
|
|
cursor.execute("SELECT code, name, flag FROM available_languages ORDER BY name")
|
|
rows = cursor.fetchall()
|
|
cursor.close()
|
|
return [{"code": row[0], "name": row[1], "flag": row[2] or ""} for row in rows]
|
|
else:
|
|
conn = get_connection()
|
|
c = conn.cursor()
|
|
c.execute("SELECT code, name, flag FROM available_languages ORDER BY name")
|
|
langs = [{"code": row[0], "name": row[1], "flag": row[2] or ""} for row in c.fetchall()]
|
|
conn.close()
|
|
return langs
|
|
|
|
def set_bot_languages(bot_type: str, lang_codes: list):
|
|
db_type = get_db_type()
|
|
|
|
if db_type == "mysql":
|
|
conn = get_connection()
|
|
cursor = conn.cursor()
|
|
cursor.execute("DELETE FROM bot_languages WHERE bot_type = %s", (bot_type,))
|
|
for code in lang_codes:
|
|
cursor.execute("INSERT INTO bot_languages (bot_type, lang_code) VALUES (%s, %s)",
|
|
(bot_type, code))
|
|
conn.commit()
|
|
cursor.close()
|
|
else:
|
|
conn = get_connection()
|
|
c = conn.cursor()
|
|
c.execute("DELETE FROM bot_languages WHERE bot_type = ?", (bot_type,))
|
|
for code in lang_codes:
|
|
c.execute("INSERT INTO bot_languages (bot_type, lang_code) VALUES (?, ?)",
|
|
(bot_type, code))
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
def get_bot_languages(bot_type: str) -> list:
|
|
db_type = get_db_type()
|
|
|
|
if db_type == "mysql":
|
|
conn = get_connection()
|
|
cursor = conn.cursor()
|
|
cursor.execute("SELECT lang_code FROM bot_languages WHERE bot_type = %s", (bot_type,))
|
|
rows = cursor.fetchall()
|
|
cursor.close()
|
|
return [row[0] for row in rows]
|
|
else:
|
|
conn = get_connection()
|
|
c = conn.cursor()
|
|
c.execute("SELECT lang_code FROM bot_languages WHERE bot_type = ?", (bot_type,))
|
|
langs = [row[0] for row in c.fetchall()]
|
|
conn.close()
|
|
return langs
|
|
|
|
def get_active_languages(guild_id: int) -> list:
|
|
db_type = get_db_type()
|
|
|
|
if db_type == "mysql":
|
|
conn = get_connection()
|
|
cursor = conn.cursor()
|
|
cursor.execute("SELECT lang_code FROM guild_languages WHERE guild_id = %s", (guild_id,))
|
|
rows = cursor.fetchall()
|
|
cursor.close()
|
|
return [row[0] for row in rows]
|
|
else:
|
|
conn = get_connection()
|
|
c = conn.cursor()
|
|
c.execute("SELECT lang_code FROM guild_languages WHERE guild_id = ?", (guild_id,))
|
|
langs = [row[0] for row in c.fetchall()]
|
|
conn.close()
|
|
return langs
|
|
|
|
def set_active_languages(guild_id: int, lang_codes: list):
|
|
db_type = get_db_type()
|
|
|
|
if db_type == "mysql":
|
|
conn = get_connection()
|
|
cursor = conn.cursor()
|
|
cursor.execute("DELETE FROM guild_languages WHERE guild_id = %s", (guild_id,))
|
|
for code in lang_codes:
|
|
cursor.execute("INSERT INTO guild_languages (guild_id, lang_code) VALUES (%s, %s)",
|
|
(guild_id, code))
|
|
conn.commit()
|
|
cursor.close()
|
|
else:
|
|
conn = get_connection()
|
|
c = conn.cursor()
|
|
c.execute("DELETE FROM guild_languages WHERE guild_id = ?", (guild_id,))
|
|
for code in lang_codes:
|
|
c.execute("INSERT INTO guild_languages (guild_id, lang_code) VALUES (?, ?)", (guild_id, code))
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
def get_config_value(key: str) -> str:
|
|
db_type = get_db_type()
|
|
|
|
if db_type == "mysql":
|
|
conn = get_connection()
|
|
cursor = conn.cursor()
|
|
cursor.execute("SELECT value FROM bot_config WHERE `key` = %s", (key,))
|
|
row = cursor.fetchone()
|
|
cursor.close()
|
|
return row[0] if row else None
|
|
else:
|
|
conn = get_connection()
|
|
c = conn.cursor()
|
|
c.execute("SELECT value FROM bot_config WHERE key = ?", (key,))
|
|
row = c.fetchone()
|
|
conn.close()
|
|
return row[0] if row else None
|
|
|
|
def set_config_value(key: str, value: str):
|
|
db_type = get_db_type()
|
|
|
|
if db_type == "mysql":
|
|
conn = get_connection()
|
|
cursor = conn.cursor()
|
|
cursor.execute("INSERT INTO bot_config (`key`, value) VALUES (%s, %s) ON DUPLICATE KEY UPDATE value = %s",
|
|
(key, value, value))
|
|
conn.commit()
|
|
cursor.close()
|
|
else:
|
|
conn = get_connection()
|
|
c = conn.cursor()
|
|
c.execute("INSERT OR REPLACE INTO bot_config (key, value) VALUES (?, ?)", (key, value))
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
def save_message(message_id: int, guild_id: int, author_id: int, content: str, mentions_map: dict, bot_type: str):
|
|
db_type = get_db_type()
|
|
mentions_json = json.dumps(mentions_map)
|
|
|
|
if db_type == "mysql":
|
|
conn = get_connection()
|
|
cursor = conn.cursor()
|
|
query = """INSERT INTO messages (message_id, guild_id, author_id, content, mentions_map, bot_type)
|
|
VALUES (%s, %s, %s, %s, %s, %s)
|
|
ON DUPLICATE KEY UPDATE content = %s, mentions_map = %s"""
|
|
cursor.execute(query, (message_id, guild_id, author_id, content, mentions_json, bot_type, content, mentions_json))
|
|
conn.commit()
|
|
cursor.close()
|
|
else:
|
|
conn = get_connection()
|
|
c = conn.cursor()
|
|
c.execute("INSERT OR REPLACE INTO messages (message_id, guild_id, author_id, content, mentions_map, bot_type) VALUES (?, ?, ?, ?, ?, ?)",
|
|
(message_id, guild_id, author_id, content, mentions_json, bot_type))
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
def get_message(message_id: int):
|
|
db_type = get_db_type()
|
|
|
|
if db_type == "mysql":
|
|
conn = get_connection()
|
|
cursor = conn.cursor(dictionary=True)
|
|
cursor.execute("SELECT content, mentions_map, bot_type FROM messages WHERE message_id = %s", (message_id,))
|
|
row = cursor.fetchone()
|
|
cursor.close()
|
|
if row:
|
|
if row['mentions_map']:
|
|
row['mentions_map'] = json.loads(row['mentions_map'])
|
|
return row
|
|
else:
|
|
conn = get_connection()
|
|
c = conn.cursor()
|
|
c.execute("SELECT content, mentions_map, bot_type FROM messages WHERE message_id = ?", (message_id,))
|
|
row = c.fetchone()
|
|
conn.close()
|
|
if row:
|
|
return {
|
|
'content': row[0],
|
|
'mentions_map': json.loads(row[1]) if row[1] else {},
|
|
'bot_type': row[2]
|
|
}
|
|
return None
|
|
|
|
def save_translation(message_id: int, target_lang: str, translated_text: str):
|
|
db_type = get_db_type()
|
|
|
|
if db_type == "mysql":
|
|
conn = get_connection()
|
|
cursor = conn.cursor()
|
|
query = "INSERT INTO translations (message_id, target_lang, translated_text) VALUES (%s, %s, %s) ON DUPLICATE KEY UPDATE translated_text = %s"
|
|
cursor.execute(query, (message_id, target_lang, translated_text, translated_text))
|
|
conn.commit()
|
|
cursor.close()
|
|
else:
|
|
conn = get_connection()
|
|
c = conn.cursor()
|
|
c.execute("INSERT OR REPLACE INTO translations (message_id, target_lang, translated_text) VALUES (?, ?, ?)",
|
|
(message_id, target_lang, translated_text))
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
def get_cached_translation(message_id: int, target_lang: str) -> str:
|
|
db_type = get_db_type()
|
|
|
|
if db_type == "mysql":
|
|
conn = get_connection()
|
|
cursor = conn.cursor()
|
|
cursor.execute("SELECT translated_text FROM translations WHERE message_id = %s AND target_lang = %s", (message_id, target_lang))
|
|
row = cursor.fetchone()
|
|
cursor.close()
|
|
return row[0] if row else None
|
|
else:
|
|
conn = get_connection()
|
|
c = conn.cursor()
|
|
c.execute("SELECT translated_text FROM translations WHERE message_id = ? AND target_lang = ?", (message_id, target_lang))
|
|
row = c.fetchone()
|
|
conn.close()
|
|
return row[0] if row else None
|
|
|
|
def get_ui_translation(text: str, target_lang: str) -> str:
|
|
db_type = get_db_type()
|
|
if db_type == "mysql":
|
|
conn = get_connection()
|
|
cursor = conn.cursor()
|
|
cursor.execute("SELECT translated_text FROM ui_translations WHERE original_text = %s AND target_lang = %s", (text, target_lang))
|
|
row = cursor.fetchone()
|
|
cursor.close()
|
|
return row[0] if row else None
|
|
else:
|
|
conn = get_connection()
|
|
c = conn.cursor()
|
|
c.execute("SELECT translated_text FROM ui_translations WHERE original_text = ? AND target_lang = ?", (text, target_lang))
|
|
row = c.fetchone()
|
|
conn.close()
|
|
return row[0] if row else None
|
|
|
|
def save_ui_translation(text: str, target_lang: str, translated_text: str):
|
|
db_type = get_db_type()
|
|
if db_type == "mysql":
|
|
conn = get_connection()
|
|
cursor = conn.cursor()
|
|
cursor.execute("INSERT INTO ui_translations (original_text, target_lang, translated_text) VALUES (%s, %s, %s) ON DUPLICATE KEY UPDATE translated_text = %s", (text, target_lang, translated_text, translated_text))
|
|
conn.commit()
|
|
cursor.close()
|
|
else:
|
|
conn = get_connection()
|
|
c = conn.cursor()
|
|
c.execute("INSERT OR REPLACE INTO ui_translations (original_text, target_lang, translated_text) VALUES (?, ?, ?)", (text, target_lang, translated_text))
|
|
conn.commit()
|
|
conn.close()
|