python-fastapi-mm-zhcs-yj-api/routers/api/expert/__init__.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,StreamingResponse
from sqlalchemy.orm import Session
from sqlalchemy import and_, or_,text,literal
from common.db import db_czrz
from sqlalchemy.sql import func
from sqlalchemy.future import select
from common.auth_user import *
from pydantic import BaseModel
from database import get_db,get_db_share
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.put("/update/{id}")
async def update_info(
    id :str ,
    user_id=Depends(valid_access_token),
    body=Depends(remove_xss_json),
    db: Session = Depends(get_db_share)
):
    try:
        query = db.query(EmergencyExpertInfo)
        query = query.filter(EmergencyExpertInfo.id == id)
        update_info = query.first()
        if not update_info:
            return JSONResponse(status_code=404, content={'msg':'信息不存在','code':404})
        if 'name' in body:
            update_info.name = body['name']
        if 'county' in body:
            update_info.county = body['county']
        if 'expert_type' in body:
            update_info.expert_type = body['expert_type']
        if 'honorary_title' in body:
            update_info.honorary_title = body['honorary_title']
        if 'unit' in body:
            update_info.unit = body['unit']
        if 'position' in body:
            update_info.position = body['position']
        if 'professional_title' in body:
            update_info.professional_title = body['professional_title']
        if 'specialty' in body:
            update_info.specialty = body['specialty']
        if 'rescue_experience' in body:
            update_info.rescue_experience = body['rescue_experience']
        if 'birth_date' in body:
            update_info.birth_date = body['birth_date']
        if 'work_start_date' in body:
            update_info.work_start_date = body['work_start_date']
        if 'certificate_issue_date' in body:
            update_info.certificate_issue_date = body['certificate_issue_date']
        if 'professional_group' in body:
            update_info.professional_group = body['professional_group']
        if 'professional_field' in body:
            update_info.professional_field = body['professional_field']
        if 'work_phone' in body:
            update_info.work_phone = body['work_phone']
        if 'home_phone' in body:
            update_info.home_phone = body['home_phone']
        if 'mobile_phone' in body:
            update_info.mobile_phone = body['mobile_phone']
        if 'email' in body:
            update_info.email = body['email']
        if 'contact_address' in body:
            update_info.contact_address = body['contact_address']
        if 'longitude' in body:
            update_info.longitude = body['longitude']
        if 'latitude' in body:
            update_info.latitude = body['latitude']
        db.commit()
        return {"code": 200, "msg": "更新成功"}
    except Exception as e:
        traceback.print_exc()
        return JSONResponse(status_code=500, content={'msg': f"Internal server error: {str(e)}", 'code': 500})

@router.get("/info/{id}")
async def get_info(
    id: str,
    db: Session = Depends(get_db_share)
):
    try:
        query = db.query(EmergencyExpertInfo)
        query = query.filter(EmergencyExpertInfo.id == id)
        info = query.first()
        # pattern = db.query(TpPatternList).filter(TpPatternList.id == pattern_id).first()
        if not info:
            return JSONResponse(status_code=404, content={'msg':'信息不存在','code':404})

        return {"code": 200, "msg": "获取成功", "data": {'id':info.id,
                    'name':info.name,
                    'county':info.county,
                    'expert_type':info.expert_type,
                    'honorary_title':info.honorary_title,
                    'unit' :info.unit,
                    'position' :info.position,
                    'professional_title' :info.professional_title,
                    'specialty' :info.specialty,
                    'rescue_experience' :info.rescue_experience,
                    'birth_date' :info.birth_date,
                    'work_start_date' :info.work_start_date,
                    'certificate_issue_date' :info.certificate_issue_date,
                    'professional_group' :info.professional_group,
                    'professional_field' :info.professional_field,
                    'work_phone' :info.work_phone,
                    'home_phone' :info.home_phone,
                    'mobile_phone' :info.mobile_phone,
                    'email' :info.email,
                    'contact_address' :info.contact_address,
                    'longitude' :info.longitude,
                    'latitude' :info.latitude}}
    except Exception as e:
        traceback.print_exc()
        return JSONResponse(status_code=500, content={'msg': f"Internal server error: {str(e)}", 'code': 500})


@router.get("/list")
async def get_list_info(
        keyword : str = Query(None),
        county : str = Query(None),
        professional_group : str = Query(None),
        page: int = Query(1, gt=0, description='页码'),
        pageSize: int = Query(5, gt=0, description='每页条目数量'),
        db: Session = Depends(get_db_share)
):
    try:
        query = db.query(EmergencyExpertInfo)
        if keyword:
            query = query.filter(or_(EmergencyExpertInfo.name.like(f'%{keyword}%'),
                                 EmergencyExpertInfo.unit.like(f'%{keyword}%'),
                                 EmergencyExpertInfo.professional_title.like(f'%{keyword}%'),
                                 EmergencyExpertInfo.professional_group.like(f'%{keyword}%'),
                                 EmergencyExpertInfo.professional_field.like(f'%{keyword}%')))
        print(query)
        if county:
            query = query.filter(EmergencyExpertInfo.county==county)
        if professional_group:
            query = query.filter(EmergencyExpertInfo.professional_group==professional_group)
        total_items = query.count()
        query = query.offset((page - 1) * pageSize).limit(pageSize)
        data = query.all()

        return {"code": 200, "msg": "获取成功",
                "data": [{'id':info.id,
                    'name':info.name,
                    'county':info.county,
                    'expert_type':info.expert_type,
                    'honorary_title':info.honorary_title,
                    'unit' :info.unit,
                    'position' :info.position,
                    'professional_title' :info.professional_title,
                    'specialty' :info.specialty,
                    'rescue_experience' :info.rescue_experience,
                    'birth_date' :info.birth_date,
                    'work_start_date' :info.work_start_date,
                    'certificate_issue_date' :info.certificate_issue_date,
                    'professional_group' :info.professional_group,
                    'professional_field' :info.professional_field,
                    'work_phone' :info.work_phone,
                    'home_phone' :info.home_phone,
                    'mobile_phone' :info.mobile_phone,
                    'email' :info.email,
                    'contact_address' :info.contact_address,
                    'longitude' :info.longitude,
                    'latitude' :info.latitude} for info in data],
                "total": total_items,
                "page": page,
                "pageSize": pageSize,
                "totalPages": (total_items + pageSize - 1) // pageSize}
    except Exception as e:
        traceback.print_exc()
        return JSONResponse(status_code=500, content={'msg': f"Internal server error: {str(e)}", 'code': 500})

@router.get("/export")
async def get_list_info(
        request: Request,
        keyword: str = Query(None),
        auth_user: AuthUser = Depends(find_auth_user),
        db: Session = Depends(get_db_share),
        mmdb:Session = Depends(get_db)
):
    try:
        query = db.query(EmergencyExpertInfo)
        if keyword:
            query = query.filter(or_(EmergencyExpertInfo.name.like(f'%{keyword}%'),
                                 EmergencyExpertInfo.unit.like(f'%{keyword}%'),
                                 EmergencyExpertInfo.professional_title.like(f'%{keyword}%'),
                                 EmergencyExpertInfo.professional_group.like(f'%{keyword}%'),
                                 EmergencyExpertInfo.professional_field.like(f'%{keyword}%')))
        data = query.all()
        data = [{'序号':info.id,
                    '姓名':info.name,
                    '所属区县':info.county,
                    '专家类型':info.expert_type,
                    '荣誉称号':info.honorary_title,
                    '单位' :info.unit,
                    '职位' :info.position,
                    '职称' :info.professional_title,
                    '擅长事故类型' :info.specialty,
                    '救援经历' :info.rescue_experience,
                    '出生日期' :info.birth_date,
                    '工作时间' :info.work_start_date,
                    '发证日期' :info.certificate_issue_date,
                    '专业分组' :info.professional_group,
                    '专业领域' :info.professional_field,
                    '工作电话' :info.work_phone,
                    '住宅电话' :info.home_phone,
                    '移动电话' :info.mobile_phone,
                    '电子邮箱' :info.email,
                    '联系地址' :info.contact_address,
                    '经度' :info.longitude,
                    '纬度' :info.latitude} for info in data]
        # 返回结果
        import pandas as pd
        from io import BytesIO
        # 将查询结果转换为 DataFrame
        df = pd.DataFrame(data)

        # 将 DataFrame 导出为 Excel 文件
        output = BytesIO()
        with pd.ExcelWriter(output, engine='openpyxl') as writer:
            df.to_excel(writer, index=False)

        # 设置响应头
        output.seek(0)
        from urllib.parse import quote
        encoded_filename = f'应急专家管理导出{datetime.now().strftime("%Y%m%d%H%mi%s")}.xlsx'
        encoded_filename = quote(encoded_filename, encoding='utf-8')
        headers = {
            'Content-Disposition': f'attachment; filename*=UTF-8\'\'{encoded_filename}',
            'Content-Type': 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet'
        }

        db_czrz.log(mmdb, auth_user, "应急专家管理", f"应急专家管理导出数据成功", request.client.host)

        # 返回文件流
        return StreamingResponse(output, headers=headers)
    except Exception as e:
        traceback.print_exc()
        return JSONResponse(status_code=500, content={'msg': f"Internal server error: {str(e)}", 'code': 500})