python-fastapi-mm-zhcs-yj-api/jobs/duty_job.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from datetime import datetime, date
from sqlalchemy.orm import Session
from sqlalchemy import text, exists, and_, or_, not_
from utils import *
from utils.redis_util import *
from models import *
from exceptions import *
from database import get_local_db
from extensions import logger
from common import YzyApi
from config import settings
import traceback
from pprint import pprint
from common.db import db_user, db_yzy, db_msg_center
from config import settings

# 值班任务
# 每天00:00将当天值班人员推送一次
def proc():
    lock_key = "duty_job_proc"
    if redis_lock(lock_key):
        logger.info(datetime.now())

        with get_local_db() as db:

            __duty_old_proc(db)
            
            __duty_proc(db)

    
        redis_unlock(lock_key)


def format_time(time_diff) -> str:
    hours, minutes, seconds = str(time_diff).split(':')
    return str(hours).zfill(2) + ":" + str(minutes).zfill(2)

# 删除今天之前的值班提醒
def __duty_old_proc(db: Session):
    theDay = date.today()
    db.query(MsgCenter).filter(and_(MsgCenter.msg_type == '值班消息', MsgCenter.recv_time < theDay, MsgCenter.del_flag == '0', MsgCenter.recv_status == 0)).update({"del_flag": 1, "update_time": datetime.now()})
    db.commit()

# 新建今天的值班提醒
def __duty_proc(db: Session):
    theDay = date.today()
    rows = db.query(DutyShift).filter(and_(DutyShift.shift_date == theDay, DutyShift.del_flag == '0')).all()
    
    for row in rows:
        start_time = format_time(row.start_time)
        end_time = format_time(row.end_time)
        
        to_users = [row.leader_id, row.primary_staff_id, row.secondary_staff_id, row.standby_staff_id]
        for to_user_id in to_users:
            if(is_not_sent(db, to_user_id, theDay)):
                send_yzy_msg(db, to_user_id, start_time, end_time, str(row.shift_id))

def is_not_sent(db: Session, to_user_id: str, theDay: DateTime):
    user_info = db_user.get_user_info(db, to_user_id)
    yzy_account = user_info.yzy_account
    
    where = and_(YzyMsgQueue.mobile == yzy_account, 
                 YzyMsgQueue.from_scenario == 'duty_shift',
                 YzyMsgQueue.create_time >= theDay,
                 YzyMsgQueue.create_time < theDay + timedelta(days=1),
                )
    row = db.query(YzyMsgQueue).filter(where).first()
    return row is None

# 发送粤政易消息
def send_yzy_msg(db: Session, to_user_id: int, start_time: str, end_time: str, foreign_key: str) -> None:
    user_info = db_user.get_user_info(db, to_user_id)
    yzy_account = user_info.yzy_account

    yzy_userid = db_yzy.get_userid_by_account(db, yzy_account)
    detail_url = "{}{}".format(settings.YZY_WEB_ROOT, "/yjxp/#/worker/duty")

    description = f"您今天的值班时间为{start_time}-{end_time}，请准时到岗。"
    logger.info("{}, {}", yzy_account, description)
    data = {
        "yzy_userid": yzy_userid,
        "mobile": yzy_account,
        "content": description,
        "recorded_by": 0,
        "detail_url": detail_url,
        "foreign_key": foreign_key,
        "from_scenario": "duty_shift",
        "title": "值班消息"
    }
    YzyApi.add_to_msg_queue(db, data)
    db_msg_center.add_message(db, "值班消息", to_user_id, "值班提醒", description, foreign_key, 'duty_shift')