lookingWeb/dataService/MySqlService.py

import pymysql
from loguru import logger

class DatabaseConfig:
    HOST = "192.168.123.199"
    PORT = 3306
    USER = "root"
    PASSWORD = "GUOxy5157"
    DB = "LookingWeb"


class MySQLDatabase:
    def __init__(self, host, prot, user, password, db):
        self.port = prot
        self.host = host
        self.user = user
        self.password = password
        self.db = db
        self.connection = None
        self.connect()

    def __enter__(self):
        self.connect()
        return self
    def __exit__(self, exc_type, exc_val, exc_tb):
        if self.connection and self.connection.open:
            self.connection.close()

    @classmethod
    def create_connection(cls):
        return cls(DatabaseConfig.HOST, DatabaseConfig.PORT, DatabaseConfig.USER, DatabaseConfig.PASSWORD,
                   DatabaseConfig.DB)


    def connect(self):
        if not self.connection or not self.connection.open:
            try:
                self.connection = pymysql.connect(
                    host=self.host, port=self.port, user=self.user,
                    password=self.password, db=self.db, charset='utf8mb4',
                    cursorclass=pymysql.cursors.DictCursor)
                logger.info("数据库连接成功：{}@{}", self.user, self.host)
            except pymysql.MySQLError as e:
                logger.error("数据库连接失败：{}", str(e))

    def close(self):
        if self.connection and self.connection.open:
            self.connection.close()
        self.connection = None
    def query(self, sql, params=None):
        self.connect()
        if not self.connection:
            logger.error("查询失败：{}，无数据库连接", sql)
            return None
        with self.connection.cursor() as cursor:
            try:
                cursor.execute(sql, params or ())
                result = cursor.fetchall()
                logger.debug("查询成功：{}", sql)
                return result
            except pymysql.MySQLError as e:
                logger.error("查询失败：{}，错误：{}", sql, str(e))
                return None

    def execute(self, sql, params=None):
        self.connect()
        if not self.connection:
            logger.error("执行失败：{}，无数据库连接", sql)
            return
        with self.connection.cursor() as cursor:
            cursor.execute(sql, params or ())
            self.connection.commit()


def is_connected():
    logger.debug("正在检查数据库连接...")
    with MySQLDatabase.create_connection() as mysql:
        try:
            mysql.query("SELECT * FROM seeds_info where id = 1")
            logger.info("数据库连接检查通过")
            return True
        except pymysql.MySQLError as e:
            logger.error("数据库连接检查失败：{}", str(e))
            return False


def seed_is_exist(seed):
    with MySQLDatabase.create_connection() as mysql:
        try:
            sql = "SELECT * FROM seeds_info where seed_url = %s"
            result = mysql.query(sql, (seed.seedUrl,))
            if not result:
                logger.info("种子不存在: {}", seed.seedUrl)
                return False
            else:
                logger.info("种子已存在: {}", seed.seedUrl)
                return True
        except pymysql.MySQLError as e:
            logger.error("检查种子存在失败: {}，错误: {}", seed.seedUrl, str(e))
            return False

def save_seed(seed_info):
    with MySQLDatabase.create_connection() as connection:
        query = """
        INSERT INTO seeds_info (name, seed_url, have_download, create_time, update_time, url)
        VALUES (%s, %s, %s, %s, %s, %s)
        """
        values = (seed_info.name, seed_info.seedUrl, seed_info.haveDownload, seed_info.createTime, seed_info.updateTime,
                  seed_info.url)
        try:
            connection.execute(query, values)
            logger.info("种子信息保存成功: {}", seed_info.seedUrl)
            return True
        except pymysql.MySQLError as e:
            logger.error("保存种子信息失败: {}，错误: {}", seed_info.seedUrl, str(e))
            return False

def delete_seed(seed_info):
    query = "DELETE FROM seeds_info WHERE seed_url = %s"
    value = (seed_info.seedUrl,)
    with MySQLDatabase.create_connection() as connection:
        try:
            connection.execute(query, value)
            logger.info("种子删除成功: {}", seed_info.seedUrl)
            return True
        except pymysql.MySQLError as e:
            logger.error("删除种子失败: {}，错误: {}", seed_info.seedUrl, str(e))
            return False