雨情监测数据任务

jobs/__init__.py

@@ -7,11 +7,13 @@ from datetime import datetime, timedelta
 from utils import *
 from models import *
 from .yzy_job import proc as yzy_proc
+from .rainfall_conditions_job import proc as rainfall_proc
 
 def register_jobs(scheduler: BaseScheduler):
     # scheduler.add_job(yzy_proc, next_run_time=(datetime.now() + timedelta(seconds=3)))
     scheduler.add_job(yzy_proc, CronTrigger.from_crontab('0 */5 * * *'))
 
+    scheduler.add_job(rainfall_proc, CronTrigger.from_crontab('0 10 * * *'))
     # scheduler.add_job(wdyy_proc, next_run_time=(datetime.now() + timedelta(seconds=3)))
     # scheduler.add_job(wdyy_proc, CronTrigger.from_crontab('0 * * * *'))
 

jobs/rainfall_conditions_job.py

@@ -0,0 +1,105 @@
+#!/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 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=''):
+    # 获取当前时间
+    now = datetime.now()
+    # 计算 24 小时前的时间
+    twenty_four_hours_ago = now - timedelta(hours=24)
+    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:
+        db.query(GovdataRainDataInfo).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():
+    logger.info(datetime.now())
+    try:
+        put_data()
+    except:
+        pass

models/__init__.py

@@ -12,4 +12,5 @@ from .event_base import *
 from .taskRegistration_base import *
 from .video_base import *
 from .yzy_base import *
-from .pattern_base import *
+from .pattern_base import *
+from .rain_base import *

models/rain_base.py

@@ -0,0 +1,17 @@
+from sqlalchemy import Column, String, DateTime, Integer, BigInteger
+from sqlalchemy.ext.declarative import declarative_base
+from sqlalchemy.orm import sessionmaker
+from datetime import datetime
+
+Base = declarative_base()
+
+class GovdataRainDataInfo(Base):
+    __tablename__ = 'govdata_rain_data_info'
+
+    # 定义字段
+    code = Column(String(50), primary_key=True, nullable=False, comment='编码')
+    area_name = Column(String(50), nullable=True, comment='区域名称')
+    address = Column(String(255), nullable=True, comment='地址')
+    create_time = Column(DateTime, primary_key=True, nullable=False, comment='创建时间')
+    rainfall = Column(BigInteger, nullable=True, comment='降雨量')
+    update_time = Column(DateTime, nullable=True, comment='更新时间')