30 lines
1.0 KiB
Python
30 lines
1.0 KiB
Python
# 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 '' |