Initial commit: LookingWeb project

dataService/MySqlService.py

@@ -0,0 +1,122 @@
+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):
+        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()
+
+    def query(self, sql, params=None):
+        self.connect()
+        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()
+        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