新的信息

libs/libs_db_task_rc.py

@@ -0,0 +1,30 @@
+# mysql数据库常用任务函数封装
+import pymysql, socket, time
+import config
+
+#公共连接库
+def pymysqlAlias():
+    return pymysql.connect(
+            host=config.mysql_task_rc['host'],
+            port=config.mysql_task_rc['port'],
+            user=config.mysql_task_rc['user'],
+            password=config.mysql_task_rc['password'],
+            db=config.mysql_task_rc['db'],
+            charset=config.mysql_task_rc['charset'],)
+
+# 获取新的任务
+def get_task_rc_count():
+    try:
+        with pymysqlAlias() as conn:
+            cursor = conn.cursor()
+            sql = f'select count(*) from task_distributed where status != 2 and created_at >= "2024-09-02 00:00:00"'
+            # 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_rc_count()异常: {str(e)}")
+        return ''