You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
93 lines
4.1 KiB
93 lines
4.1 KiB
# 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 priority desc, 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() |
|
|
|
#判断是否是昆山教学的,是的话优先级设置为默认 |
|
if data["psid"] == '85': |
|
sql = f'insert into tasks (task_type, task_key,studio_id) values ("{data["task_type"]}", "{data["task_key"]}","{data["psid"]}")' |
|
elif data["psid"] == '1': |
|
#实验室的订单走分布式处理 |
|
sql = f'insert into task_distributed (task_type, task_key, priority,created_at,studio_id) values ("{data["task_type"]}", "{data["task_key"]}", 1,"{time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())}","{data["psid"]}")' |
|
else: |
|
sql = f'insert into tasks (task_type, task_key, priority,studio_id) values ("{data["task_type"]}", "{data["task_key"]}", 1,"{data["psid"]}")' |
|
# 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)}") |