python-fastapi-mm-zhcs-yj-api/routers/api/spatialAnalysis/point.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,literal
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 multiprocessing import Pool, cpu_count
import json
import time
import math


router = APIRouter()

@router.post("/get_info")
@router.get("/get_info")
async def get_infos(
        body = Depends(remove_xss_json),
        # zoom_level: float = Query(..., description="Zoom level for clustering"),
        # latitude_min: float = Query(..., description="Minimum latitude"),
        # latitude_max: float = Query(..., description="Maximum latitude"),
        # longitude_min: float = Query(..., description="Minimum longitude"),
        # longitude_max: float = Query(..., description="Maximum longitude"),
        # dict_value: str = Query(None),
        # option:str  = Query(None),
        db:  Session = Depends(get_db)
):
    try:
        # 根据缩放级别动态调整分组粒度
        zoom_level = float(body['zoom_level'])
        zoom_levels = {
            3: 10000,  # 全国范围
            4: 5000,
            5: 2500,
            6: 1250,
            7: 825,
            8: 412.5,
            9: 256.25,
            10: 178.125,
            11: 69.0625,
            12: 29.53125,
            13: 13.765625,
            14: 5.8828125,
            15: 2.44140625,
            16: 1.220703125,
            17: 0.6103515625,
            18: 0.30517578125
        }
        distance_threshold=zoom_levels[int(zoom_level-1)]
        # distance_threshold = 100000 / (2.2 ** zoom_level)  # 例如：每缩放一级，距离阈值减半
        dict_value= body['dict_value'].split(',')
        latitude_min = float(body['latitude_min'])
        latitude_max = float(body['latitude_max'])
        longitude_min = float(body['longitude_min'])
        longitude_max = float(body['longitude_max'])
        option = body['option'].split(',')
        print("1",time.time())
        videos = get_videos(db,dict_value,latitude_min,latitude_max,longitude_min,longitude_max)
        infos = get_points(db,option,latitude_min,latitude_max,longitude_min,longitude_max)
        # 动态分组逻辑
        groups = group_points(videos+infos, distance_threshold)

        print("4",time.time())

        return {"code": 200,
                "msg": "操作成功",
                "data": groups}
    except Exception as e:
        # 处理异常
        traceback.print_exc()
        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(e))
@router.post("/get_details")
@router.get("/get_details")
async def get_details(
        body = Depends(remove_xss_json),
    # center_latitude: float = Query(..., description="网格中心点的纬度"),
    # center_longitude: float = Query(..., description="网格中心点的经度"),
    # zoom_level: float = Query(..., description="缩放级别"),
    db: Session = Depends(get_db)
):
    try:
        # 计算网格大小
        zoom_level = float(body['zoom_level'])
        zoom_levels = {
            3: 10000,  # 全国范围
            4: 5000,
            5: 2500,
            6: 1250,
            7: 825,
            8: 412.5,
            9: 256.25,
            10: 178.125,
            11: 69.0625,
            12: 29.53125,
            13: 13.765625,
            14: 5.8828125,
            15: 2.44140625,
            16: 1.220703125,
            17: 0.6103515625,
            18: 0.30517578125
        }
        distance_threshold=zoom_levels[int(zoom_level-1)]
        # distance_threshold = 1000 / (1.5 ** zoom_level)  # 例如：每缩放一级，距离阈值减半
        grid_size = calculate_grid_size(distance_threshold)  # 地球半径为6371公里
        center_latitude = float(body['latitude'])
        center_longitude = float(body['longitude'])
        dict_value = body['dict_value'].split(',')
        option = body['option'].split(',')
        # 计算网格的经纬度范围
        latitude_min, latitude_max, longitude_min, longitude_max = get_grid_bounds_from_center(center_latitude, center_longitude, grid_size)
        videos = get_videos(db,dict_value,latitude_min,latitude_max,longitude_min,longitude_max)
        infos = get_points(db,option,latitude_min,latitude_max,longitude_min,longitude_max)
        return {"code": 200,
                "msg": "操作成功",
                "data": videos+infos }#{"videos":videos,"points":infos}}
    except Exception as e:
        # 处理异常
        traceback.print_exc()
        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(e))

def calculate_grid_size(distance_threshold):
    # 假设地球半径为6371公里，将距离阈值转换为经纬度的差值
    # 这里假设纬度变化对距离的影响较小，仅根据经度计算网格大小
    earth_radius = 6371  # 地球半径，单位为公里
    grid_size = distance_threshold / earth_radius
    return grid_size

def get_grid_key(latitude, longitude, grid_size):
    # 根据经纬度和网格大小计算网格键
    return (math.floor(latitude / grid_size), math.floor(longitude / grid_size))

def get_grid_bounds_from_center(center_latitude, center_longitude, grid_size):
    half_grid_size = grid_size / 2
    min_latitude = center_latitude - half_grid_size
    max_latitude = center_latitude + half_grid_size
    min_longitude = center_longitude - half_grid_size
    max_longitude = center_longitude + half_grid_size
    return min_latitude, max_latitude, min_longitude, max_longitude

def calculate_distance(point1, point2):
    # 使用 Haversine 公式计算两点之间的距离
    from math import radians, sin, cos, sqrt, atan2
    R = 6371  # 地球半径（公里）
    lat1, lon1 = radians(point1.latitude), radians(point1.longitude)
    lat2, lon2 = radians(point2.latitude), radians(point2.longitude)
    dlat = lat2 - lat1
    dlon = lon2 - lon1
    a = sin(dlat / 2) ** 2 + cos(lat1) * cos(lat2) * sin(dlon / 2) ** 2
    c = 2 * atan2(sqrt(a), sqrt(1 - a))
    return R * c

def group_points(points, distance_threshold):
    grid_size = calculate_grid_size(distance_threshold)
    grid = defaultdict(lambda:{"count":0}) #,"list":[]
    groups = []
    tmp = defaultdict(list)
    for point in points:
        grid_key = get_grid_key(float(point.latitude), float(point.longitude), grid_size)
        lovalue = str(point.latitude)+str(point.longitude)
        if lovalue not in tmp['%s-%s'%grid_key]:
            tmp['%s-%s'%grid_key].append(lovalue)
        grid['%s-%s'%grid_key]['count']+=1
        if grid['%s-%s'%grid_key]['count']>1 and len(tmp['%s-%s'%grid_key])<3:
            grid['%s-%s'%grid_key]['dataType'] = ''
            grid['%s-%s'%grid_key]['id'] = ""
            if len(tmp['%s-%s'%grid_key])<2:
                grid['%s-%s'%grid_key]['name'] = '多数据点位'
                grid['%s-%s' % grid_key]['type'] ='1'
            else:
                grid['%s-%s' % grid_key]['name'] = '聚合点位'
                grid['%s-%s' % grid_key]['type'] = '3'
            # grid['%s-%s'%grid_key]['latitude'] = float(point.latitude) #(grid_key[0] + 0.5) * grid_size
            # grid['%s-%s'%grid_key]['longitude'] = float(point.longitude) #(grid_key[1] + 0.5) * grid_size
        elif grid['%s-%s'%grid_key]['count']==1:
            if point.dataType=='video':
                grid['%s-%s' % grid_key]['id'] = point.gbIndexCode
            else:
                grid['%s-%s' % grid_key]['id'] = point.id
            grid['%s-%s'%grid_key]['dataType'] = point.dataType
            grid['%s-%s'%grid_key]['name'] = point.name
            grid['%s-%s'%grid_key]['type'] ='2'
            grid['%s-%s'%grid_key]['latitude'] = float(point.latitude)
            grid['%s-%s'%grid_key]['longitude'] = float(point.longitude)

    groups = list(grid.values())

    return groups

def get_videos(db:Session,dict_value,latitude_min,latitude_max,longitude_min,longitude_max):
    que = True
    if len(dict_value)>0:
        videolist = []
        for value in dict_value:
            tag_info = get_dict_data_info(db, 'video_type', value)
            if tag_info:
                if tag_info.dict_label == '全量视频':
                    break
                else:
                    videolist += [i.video_code for i in tag_get_video_tag_list(db, value)]
        else:
            que = TPVideoInfo.gbIndexCode.in_(videolist)
    # 查询分组
    query = (
        select(
            TPVideoInfo.gbIndexCode,
            TPVideoInfo.latitude,
            TPVideoInfo.longitude,
            TPVideoInfo.name,
            TPVideoInfo.status,
            literal('video').label("dataType")
        )
            .select_from(TPVideoInfo).where(
            and_(
                TPVideoInfo.latitude >= latitude_min,
                TPVideoInfo.latitude <= latitude_max,
                TPVideoInfo.longitude >= longitude_min,
                TPVideoInfo.longitude <= longitude_max,
                TPVideoInfo.longitude > 0,
                TPVideoInfo.latitude > 0, que
            )
        )
            .order_by(TPVideoInfo.status.asc())
    )

    result = db.execute(query)
    videos = result.fetchall()
    return videos


def get_points(db:Session,option,latitude_min,latitude_max,longitude_min,longitude_max):
    # 使用参数化查询避免 SQL 注入
    if isinstance(option, list):
        option = tuple(option)
    query = text("""
        SELECT 
            A.`name`,A.`id`,A.dataType,A.longitude,A.latitude,A.infoType
        FROM (
            SELECT 
                *,
                ROW_NUMBER() OVER (PARTITION BY longitude, latitude, `name` 
                                   ORDER BY longitude, latitude, `name`) AS rn
            FROM 
                `point_data` 
            WHERE 
                longitude > 0 
                AND latitude BETWEEN :latitude_min AND :latitude_max 
                AND longitude BETWEEN :longitude_min AND :longitude_max 
                AND dataType IN :option
        ) AS A 
        WHERE rn = 1
    """)

    # 执行查询并传递参数
    result = db.execute(query, {
        'latitude_min': latitude_min,
        'latitude_max': latitude_max,
        'longitude_min': longitude_min,
        'longitude_max': longitude_max,
        'option': option
    })
    infos = result.fetchall()
    return infos