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

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from datetime import datetime, timedelta
from sqlalchemy.orm import Session
from common.BigDataCenterAPI import *
from utils import *
from utils.redis_util import *
from models import *
from exceptions import *
from database import get_db_local
from extensions import logger
import pymysql
from config import settings



def get_stcd_data(stcd='',hours=24):
    # 获取当前时间
    now = datetime.now()
    # 计算 24 小时前的时间
    twenty_four_hours_ago = now - timedelta(hours=hours)
    formatted_now = now.strftime('%Y-%m-%d %H')
    formatted_twenty_four_hours_ago = twenty_four_hours_ago.strftime('%Y-%m-%d %H')
    url = 'https://19.15.75.180:8581/GatewayMsg/http/api/proxy/invoke'
    service_code = 'P0138'
    passid = 'C90-44004428'
    passtoken = 'f539f616b2834160bc47fda5f298512c'
    signTime = str(GetTime() // 1000)
    nonce = GetNonce(5)
    sign = GetSign(signTime, nonce, passtoken)
    headers = {
        'Content-Type': 'application/json',
        'x-tif-signature': sign,
        'x-tif-timestamp': signTime,
        'x-tif-nonce': nonce,
        'x-tif-paasid': passid,
        'x-tif-serviceId': service_code
    }
    data = {
        "system_id": "C90-44004428",
        "vender_id": "xxx",
        "department_id": "xxx",
        "query_timestamp": str(GetTime()),
        "UUID": GetNonce(4),
        "query": {
            "stcd":stcd,
            "startTime":f"{formatted_twenty_four_hours_ago}:00:00",
            "endTime":f"{formatted_now}:00:00"
        },
        "audit_info": {
            "operator_id": "xxxx",
            "operator_name": "xxx",
            "query_object_id": "xxxx",
            "query_object_id_type": "01",
            "item_id": "xxxx",
            "item_code": "xxx",
            "item_sequence": "xxx",
            "terminal_info": "xxxx",
            "query_timestamp": "xxxx"
        }

    }
    response = requests.post(url=url, headers=headers, json=data, verify=False)

    if response.status_code == 200:
        try:
            return response.json()['data']['data']
        except:
            return []


def put_data(db=get_db_local()):
    query = db.query(GovdataRainDataInfo)
    rainfull = get_stcd_data()
    if len(rainfull)>0:
        now = datetime.now()
        twenty_four_hours_ago = now - timedelta(hours=25)
        db.query(GovdataRainDataInfo).filter(GovdataRainDataInfo.create_time <twenty_four_hours_ago ).delete()
        db.commit()
        for info in rainfull:
            code = info['F3070220000034_000018001']
            create_time = info['F3070220000034_000018004']
            raininfo = query.filter(GovdataRainDataInfo.code == code).filter(GovdataRainDataInfo.create_time == create_time).first()
            if raininfo:
                raininfo.area_name = info['F3070220000034_000018002']
                raininfo.address = info['F3070220000034_000018003']
                raininfo.rainfall = info['F3070220000034_000018005']
                raininfo.update_time = info['F3070220000034_000018006']
            else:
                raindata=GovdataRainDataInfo(
                    code = info['F3070220000034_000018001'],
                    area_name = info['F3070220000034_000018002'],
                    address = info['F3070220000034_000018003'],
                    create_time = info['F3070220000034_000018004'],
                    rainfall = info['F3070220000034_000018005'],
                    update_time = info['F3070220000034_000018006']
                )
                db.add(
                    raindata
                )
            db.commit()

def proc():
    lock_key = "rainfall_conditions_job"
    if redis_lock(lock_key):
        logger.info(datetime.now())
        try:
            put_data()
        except:
            pass
        finally:
            redis_unlock(lock_key)