python-fastapi-mm-zhcs-yj-api/utils/rainfall_util.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

from fastapi import APIRouter, Request, Depends, Query, HTTPException, status
from common.security import valid_access_token
from fastapi.responses import JSONResponse
from sqlalchemy.orm import Session
from sqlalchemy import and_, or_,text
from sqlalchemy.sql import func
from sqlalchemy.future import select
from common.auth_user import *
from pydantic import BaseModel
from database import get_db
from typing import List
from models import *
from utils import *
from utils.ry_system_util import *
from utils.video_util import *
from collections import defaultdict
import traceback
from concurrent.futures import ThreadPoolExecutor, as_completed
from jobs.rainfall_conditions_job import get_stcd_data
from multiprocessing import Pool, cpu_count
import json
import time
import math

def get_real_code(db,longitude,latitude):
    sql = text("""
            SELECT area_name, `code`, longitude, latitude, distance FROM (
                SELECT 
                    area_name, 
                    `code`, 
                    longitude, 
                    latitude,
                    ST_Distance_Sphere(
                        ST_GeomFromText(CONCAT('POINT(', longitude, ' ', latitude, ')')), 
                        ST_GeomFromText(:point)
                    ) AS distance
                FROM govdata_real_time_address 
                WHERE longitude IS NOT NULL AND latitude IS NOT NULL 
                ORDER BY distance ASC 
                LIMIT 1
            ) T
        """).bindparams(point=f"POINT({longitude} {latitude})")
    # 执行查询
    result = db.execute(sql).fetchone()

    # 处理结果
    if result:
        return dict(result)['code']
    else:
        return None

def get_rainfall(
    code: str,num:int,
    db: Session = Depends(get_db)
):
    value=0
    rainfulldata = get_stcd_data(code,num+1)
    for i in rainfulldata:
        value += i['F3070220000034_000018005']

    return value