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

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from datetime import datetime
from sqlalchemy.sql import func
from sqlalchemy.orm import Session
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

def proc():
    if settings.IS_PROD == False:
        return
    
    YzyApi.get_cache_access_token()

    
    
    lock_key = "yzy_job_proc"
    if redis_lock(lock_key):
        logger.info(datetime.now())

        detail_url = YzyApi.format_redirect_url("/leader/index")
        
        '''
        yzy_user_id = "etk4130970pjg84tgrhapo"
        description = "预案名称: 茂名市自然灾害救助应急预案\n响应级别: Ⅰ级响应\n消息内容: 单位您好！《茂名市自然灾害救助应急预案》现已全面启动，特此通知您单位迅速响应，全力做好预案工作要点：负责救灾工作宣传报道协调工作。"
        YzyApi.send_textcard_message(yzy_user_id, "预案响应消息", description, detail_url)
        yzy_user_id = "eb4kehgy6wj4qn0jhx1dk6"  # 暂时写死梦梅的账号
        description = "预案名称: 茂名市自然灾害救助应急预案\n响应级别: Ⅰ级响应\n消息内容: 单位您好！《茂名市自然灾害救助应急预案》现已全面启动，特此通知您单位迅速响应，全力做好预案工作要点：负责救灾工作宣传报道协调工作。"
        # ret = YzyApi.send_textcard_message(yzy_user_id, "预案响应消息", description, detail_url)
        # logger.info(ret)

        data = {
            "yzy_userid": yzy_user_id,
            "mobile": "13528373954",
            "content": description,
            "recorded_by": 1,
            "detail_url": detail_url,
            "foreign_key": "1",
            "from_scenario": "yjya",
            "title": "预案响应消息"
        }

        if settings.IS_STAGE:
            YzyApi.add_to_msg_queue(db, data)
        db.close()

        '''
        redis_unlock(lock_key)

def yzy_msg_queue_proc():
    print('yzy_msg_queue_proc ')
    if settings.IS_PROD == False:
        return
    
    with get_local_db() as db:
        print('yzy_msg_queue_proc starting...')
        lock_key = "yzy_msg_queue_job"
        if redis_lock(lock_key):
            logger.info(datetime.now())

            rows = db.query(YzyMsgQueue).filter(YzyMsgQueue.sent_status == 0).limit(20).all()
            for row in rows:
                try:
                    resp = YzyApi.send_textcard_message(row.yzy_userid, row.title, row.content, row.detail_url)
                    logger.info(resp)
                    row.sent_time = datetime.now()
                    if resp['errcode'] == 0:
                        row.sent_status = 1

                        if 'jobid' not in resp:
                            resp['jobid'] = ''
                        row.errmsg = resp['jobid']
                    else:
                        row.sent_status = 9
                        row.errmsg = resp['errmsg']
                    db.commit()
                except Exception as e:
                    traceback.print_exc()

            rows = db.query(YzyMsgQueueSk).filter(YzyMsgQueueSk.sent_status == 0).limit(30).all()
            for row in rows:
                try:
                    resp = YzyApi.send_textcard_message(row.yzy_userid, row.title, row.content, row.detail_url)
                    logger.info(resp)
                    row.sent_time = datetime.now()
                    if resp['errcode'] == 0:
                        row.sent_status = 1

                        if 'jobid' not in resp:
                            resp['jobid'] = ''
                        row.errmsg = resp['jobid']
                    else:
                        row.sent_status = 9
                        row.errmsg = resp['errmsg']
                    db.commit()
                except Exception as e:
                    traceback.print_exc()

            redis_unlock(lock_key)
        print('yzy_msg_queue_proc end.')

def yzy_unit_queue_proc():
    if settings.IS_PROD == False:
        return
    
    with get_local_db() as db:
        updatetime = db.query(func.max(YzyOrgUserEntity.updatetime)).scalar()
        if updatetime is None:
            updatetime = datetime(2000, 1, 1, 1, 1, 1)

        starttime = updatetime.strftime("%Y-%m-%d %H:%M:%S")
        endtime = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        
        # 这个看起来没用
        # YzyApi.getauthorizedusersbyupdatetime(starttime, endtime)
        
        result = YzyApi.getappusersbyupdatetime(starttime, endtime)
        
        '''
        result = {
            "success" : True,
            "errcode" : 0,
            "errmessage" : "请求成功",
            "data" : [ {
                "userid" : "imi85r9my4a9h1ls9c1fnz",
                "username" : "梁文龙",
                "gender" : 1,
                "mobile" : "49779889d18a2a9cadf132014e173116",
                "unitid" : "v5vd6t4zkbd9gz0ijlzj73",
                "position" : "主任",
                "priority" : "1",
                "order" : 1073741825,
                "status" : 0,
                "updateTime" : "2025-08-22 16:24:28"
            }, {
                "userid" : "28cm254t2kc5hjc70saq8x",
                "username" : "黄晓东",
                "gender" : 1,
                "mobile" : "97651f2cd1206c3ab205cd581606525f",
                "unitid" : "v5vd6t4zkbd9gz0ijlzj73",
                "position" : "一级科员",
                "priority" : "1",
                "order" : 1073741814,
                "status" : 0,
                "updateTime" : "2025-08-22 16:25:18"
            } ]
        }
        result = {
            "success" : True,
            "errcode" : 0,
            "errmessage" : "请求成功",
            "data" : [ {
                "userid" : "2ajk3d9w8qvncp8t5ekjqq",
                "username" : "邓思远",
                "gender" : 1,
                "mobile" : "a60d8514b5fa644e2e3d76d56902391c",
                "unitid" : "esesuh8rzc2jipjel27jcy",
                "position" : "专业技术十二级",
                "priority" : "1",
                "order" : 1073741813,
                "status" : 0,
                "updateTime" : "2025-08-13 09:18:11"
            } ]
        }
        '''
        if result['errcode'] == 0:
            data = result['data']

            for user_info in data:
                userid = user_info['userid']
                username = user_info['username']
                gender = user_info['gender']
                mobile = user_info['mobile']
                unitid = user_info['unitid']
                position = user_info['position']
                priority = user_info['priority']
                order = user_info['order']
                status = user_info['status']
                updateTime = user_info['updateTime']
                
                telephonenumber = ''    
                try:
                    # 敏感数据加密算法(DES 对称加密)
                    telephonenumber = YzyApi.desDecryptValue(settings.YZY_CORPSECRET, mobile)
                except:
                    traceback.print_exc()

                unitpath = None
                unit_row = db.query(YzyOrgUnitEntity).filter(YzyOrgUnitEntity.unitid == unitid).first()
                if unit_row is not None:
                    unitpath = unit_row.unitpath

                user_dict = {
                    "userid": userid,
                    "username": username,
                    "displayname": username,
                    "gender": gender,
                    "telephonenumber": telephonenumber,
                    "updatetime": updateTime,
                    "unitid": unitid,
                    "position": position,
                    "priority": priority,
                    "order_num": order,
                    "unitpath": unitpath
                }

                unit_row = db.query(YzyOrgUserEntity).filter(YzyOrgUserEntity.userid == userid).first()
                if unit_row is None:
                    new_unit = YzyOrgUserEntity(**user_dict)
                    new_unit.createtime = datetime.now()
                    db.add(new_unit)
                else:
                    db.query(YzyOrgUserEntity).filter(YzyOrgUserEntity.userid == userid).update(user_dict)
                db.commit()


        # result = YzyApi.getunitsbyuintidandupdatetime("v5vd6t4zkbd9gz0ijlzj73", starttime, endtime)
        
        updatetime = db.query(func.max(YzyOrgUnitEntity.updatetime)).scalar()
        if updatetime is None:
            updatetime = datetime(2000, 1, 1, 1, 1, 1)

        starttime = updatetime.strftime("%Y-%m-%d %H:%M:%S")
        endtime = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        
        result = YzyApi.getappunitsbyupdatetime(starttime, endtime)
        '''
        result = {
            "success" : True,
            "errcode" : 0,
            "errmessage" : "请求成功",
            "data" : [ {
                "unitid" : "0751a8jxfqajk90mt022n1",
                "unitname" : "茂名市政务服务和数据管理局",
                "parentunitid" : "0",
                "order" : 1073741788,
                "unittype" : 6,
                "isvirtual" : 0,
                "unitpath" : "茂名市政务服务和数据管理局",
                "unitfullpath" : "广东省/地市/茂名市/茂名市政府/茂名市政务服务和数据管理局",
                "unitidfullpath" : "449f3370-4cf6-11e8-85c3-45ad5e3a2bd7/073rxrx2fqajk90mt022la/073rxteufqajk90mqk6u8d/073ry80jfqajk90mspyvmi/0751a8jxfqajk90mt022n1",
                "status" : 0,
                "updateTime" : "2025-07-30 10:45:54"
            }, {
                "unitid" : "0755v3blfqajk90mr1g055",
                "unitname" : "数字广东网络建设有限公司",
                "parentunitid" : "0",
                "order" : 1073741823,
                "unittype" : 6,
                "isvirtual" : 0,
                "unitpath" : "数字广东网络建设有限公司",
                "unitfullpath" : "广东省/企业/其他企业/数字广东网络建设有限公司",
                "unitidfullpath" : "449f3370-4cf6-11e8-85c3-45ad5e3a2bd7/q0jp1qahizh3fbfhmt6xne/9zeh8xizv7242pof5ye5al/0755v3blfqajk90mr1g055",
                "status" : 1,
                "updateTime" : "2025-07-30 10:46:03"
            }, {
                "unitid" : "arhd5gsi3lvxn4zk3czm1m",
                "unitname" : "数字广东网络建设有限公司江门市分公司",
                "parentunitid" : "0",
                "order" : 1073741815,
                "unittype" : 6,
                "isvirtual" : 0,
                "unitpath" : "数字广东网络建设有限公司江门市分公司",
                "unitfullpath" : "广东省/地市/江门市/江门市其他单位/江门市企业单位/数字广东网络建设有限公司江门市分公司",
                "unitidfullpath" : "449f3370-4cf6-11e8-85c3-45ad5e3a2bd7/073rxrx2fqajk90mt022la/073rxt82fqajk90mrjbt9n/ypjhaxor3ts4a98wt65ltl/smflffowrhm03mvk5xr8sr/arhd5gsi3lvxn4zk3czm1m",
                "status" : 0,
                "updateTime" : "2025-07-30 10:52:27"
            }, {
                "unitid" : "yhyseu7xncbsrwtgg85x7z",
                "unitname" : "茂名市应急管理局",
                "parentunitid" : "0",
                "order" : 1073741797,
                "unittype" : 6,
                "isvirtual" : 0,
                "unitpath" : "茂名市应急管理局",
                "unitfullpath" : "广东省/地市/茂名市/茂名市政府/茂名市应急管理局",
                "unitidfullpath" : "449f3370-4cf6-11e8-85c3-45ad5e3a2bd7/073rxrx2fqajk90mt022la/073rxteufqajk90mqk6u8d/073ry80jfqajk90mspyvmi/yhyseu7xncbsrwtgg85x7z",
                "status" : 0,
                "updateTime" : "2025-08-13 09:18:11"
            } ]
        }
        '''

        if result['errcode'] == 0:
            data = result['data']
            for unit_info in data:
                unitid = unit_info['unitid']
                unitname = unit_info['unitname']
                unitpath = unit_info['unitpath']
                unitfullpath = unit_info['unitfullpath']
                unitidfullpath = unit_info['unitidfullpath']
                updateTime = unit_info['updateTime']
                order = unit_info['order']
                unittype = unit_info['unittype']
                status = unit_info['status']
                parentunitid = unit_info['parentunitid']

                unit_dict = {
                    "unitid": unitid,
                    "unitname": unitname,
                    "unitpath": unitpath,
                    "unitfullpath": unitfullpath,
                    "unitidfullpath": unitidfullpath,
                    "order": order,
                    "orgtype": unittype,
                    "updatetime": updateTime,
                    "priority": status,
                    "parentunitid": parentunitid,
                    "weworkpartyid": "ok"
                }
                unit_row = db.query(YzyOrgUnitEntity).filter(YzyOrgUnitEntity.unitid == unitid).first()
                if unit_row is None:
                    new_unit = YzyOrgUnitEntity(**unit_dict)
                    new_unit.createtime = datetime.now()
                    db.add(new_unit)
                else:
                    # 每次都是 0 
                    del unit_dict['parentunitid']
                    db.query(YzyOrgUnitEntity).filter(YzyOrgUnitEntity.unitid == unitid).update(unit_dict)
                db.commit()

                unit_path_list = unitfullpath.split("/")
                unitid_path_list = unitidfullpath.split("/")

                for i in range(len(unit_path_list)):
                    n = len(unit_path_list) - i - 1

                    unitname = unit_path_list[n]
                    unitid = unitid_path_list[n]

                    parentunitid = '0'
                    if n > 0:
                        parentunitid = unitid_path_list[n - 1]

                    unit_row = db.query(YzyOrgUnitEntity).filter(YzyOrgUnitEntity.unitid == unitid).first()
                    if unit_row is None:
                        new_unit = YzyOrgUnitEntity(unitid = unitid, unitname = unitname, unitpath = unitname, parentunitid = parentunitid, order = -1, weworkpartyid = '', updatetime=datetime.now())
                        new_unit.createtime = datetime.now()
                        db.add(new_unit)
                        db.commit()
                    else:
                        unit_row.parentunitid = parentunitid
                        db.commit()