first commit

libs/libs_db.py

@@ -0,0 +1,86 @@
+# mysql数据库常用任务函数封装
+import pymysql, socket, time
+import config
+
+# 获取新的任务
+def get_task(task_type):
+    try:
+        with pymysql.connect(
+            host=config.mysql_local['host'],
+            port=config.mysql_local['port'],
+            user=config.mysql_local['user'],
+            password=config.mysql_local['password'],
+            db=config.mysql_local['db'],
+            charset=config.mysql_local['charset'],) as conn:
+            cursor = conn.cursor()
+
+            sql = f'select task_key from tasks where task_type = "{task_type}" and status = 0 order by id asc limit 1'
+            # print(f'sql: {sql}')
+            cursor.execute(sql)
+            data = cursor.fetchone()
+            if data:
+                return data[0]
+            else:
+                return ''
+    except Exception as e:
+        print(f"{time.strftime('%Y-%m-%d %H:%M:%S', time.localtime())} 执行get_task({task_type})异常: {str(e)}")
+        return ''
+
+# 新增新的任务
+def add_task(data):
+    try:
+        with pymysql.connect(
+            host=config.mysql_local['host'],
+            port=config.mysql_local['port'],
+            user=config.mysql_local['user'],
+            password=config.mysql_local['password'],
+            db=config.mysql_local['db'],
+            charset=config.mysql_local['charset'],) as conn:
+            cursor = conn.cursor()
+
+            sql = f'insert into tasks (task_type, task_key) values ("{data["task_type"]}", "{data["task_key"]}")'
+            # print(f'sql: {sql}')
+            cursor.execute(sql)
+            conn.commit()
+    except Exception as e:
+        print(f"{time.strftime('%Y-%m-%d %H:%M:%S', time.localtime())} 执行add_task({data})异常: {str(e)}")
+        
+# 开始任务
+def start_task(data):
+    try:
+        with pymysql.connect(
+            host=config.mysql_local['host'],
+            port=config.mysql_local['port'],
+            user=config.mysql_local['user'],
+            password=config.mysql_local['password'],
+            db=config.mysql_local['db'],
+            charset=config.mysql_local['charset'],) as conn:
+            cursor = conn.cursor()
+
+            hostname = socket.gethostname()
+            sql = f'update tasks set status = 1, hostname = "{hostname}", started_at = now(), updated_at = now() where task_type = "{data["task_type"]}" and task_key = "{data["task_key"]}"'
+            # print(f'sql: {sql}')
+            cursor.execute(sql)
+            conn.commit()
+    except Exception as e:
+        print(f"{time.strftime('%Y-%m-%d %H:%M:%S', time.localtime())} 执行start_task({data})异常: {str(e)}")
+
+# 完成任务
+def finish_task(data):
+    try:
+        with pymysql.connect(
+            host=config.mysql_local['host'],
+            port=config.mysql_local['port'],
+            user=config.mysql_local['user'],
+            password=config.mysql_local['password'],
+            db=config.mysql_local['db'],
+            charset=config.mysql_local['charset'],) as conn:
+            cursor = conn.cursor()
+
+            hostname = socket.gethostname()
+            sql = f'update tasks set status = 2, hostname = "{hostname}", finished_at = now(), updated_at = now() where task_type = "{data["task_type"]}" and task_key = "{data["task_key"]}"'
+            # print(f'sql: {sql}')
+            cursor.execute(sql)
+            conn.commit()
+    except Exception as e:
+        print(f"{time.strftime('%Y-%m-%d %H:%M:%S', time.localtime())} 执行finish_task({data})异常: {str(e)}")