python-fastapi-mm-zhcs-yj-api/routers/api/ThreeProofingResponsible/person.py

from fastapi import APIRouter, Request, Depends, HTTPException, Query
from sqlalchemy.exc import IntegrityError
from fastapi.responses import HTMLResponse, FileResponse
from fastapi.responses import JSONResponse
from database import get_db
from sqlalchemy import text, exists, and_, or_, not_
from sqlalchemy.orm import Session
from models import *
import json
import random
import os
from sqlalchemy import create_engine, select
from typing import Optional
from utils.StripTagsHTMLParser import *
from utils.three_proofing_responsible_util import *
from utils.ry_system_util import *
from common.db import db_event_management, db_user, db_area, db_emergency_plan
from common.security import valid_access_token
import traceback
from utils import *
from datetime import datetime, timedelta
import pandas as pd
from common.db import db_dept
from exceptions import AppException

router = APIRouter()



@router.post('/create')
async def create_contact(
        db: Session = Depends(get_db),
        body=Depends(remove_xss_json),
        user_id=Depends(valid_access_token)
):
    try:
        # 提取请求数据
        unit_name = body['unit_name']
        name = body['name']
        area_code = body['area_code']
        position = body['position']
        phone = body['phone']
        telephone= ''
        if 'telephone' in body:
            telephone=body['telephone']
        order_num = -1
        if 'order_num'in body:
            order_num=body['order_num']
        unit_id = db_dept.get_dept_id_by_name(db, unit_name)
        user_id_1 = db_user.get_user_id_by_phonenumber(db,phone)
        # 创建新的预案记录
        new_person = ThreeProofingResponsiblePerson(
            unit_id=unit_id,
            unit_name=unit_name,
            name=name,
            area_code=area_code,
            position=position,
            phone=phone,
            telephone=telephone,
            user_id = user_id_1,
            order_num=order_num,
            create_by=user_id
        )
        # 添加到数据库会话并提交
        type_list = body['type_list']
        if isinstance(type_list,list) and len(type_list)>0:

            db.add(new_person)
            db.commit()
        else:
            return JSONResponse(status_code=404,content={
                'code': 404,
                'msg': '责任类型不能为空'
            })
        try:
            for type_info in type_list:
                type_parent_id = type_info['type_parent_id']#get_type_parent_id_by_type_id(db,type_id)
                for type_id in type_info['children']:
                    new_person_type = ThreeProofingResponsiblePersonType(
                        type_parent_id = type_parent_id,
                        type_id = type_id,
                        person_id = new_person.id,
                        create_by=user_id
                    )
                    db.add(new_person_type)
                if type_parent_id in ('5','7','9'):
                    if 'children2' in type_info:
                        for other_type_id in type_info['children2']:
                            new_person_other_type = ThreeProofingResponsiblePersonOtherType(
                                type_parent_id=type_parent_id,
                                other_type_id=other_type_id,
                                person_id=new_person.id,
                                create_by=user_id
                            )
                            db.add(new_person_other_type)
                if type_parent_id in ('4','5','7','10','11'):
                    dept_name = None
                    if 'dept_name' in type_info:
                        dept_name = type_info['dept_name']
                    other_type_2_name = None
                    if 'other_type_2_name' in type_info:
                        other_type_2_name = type_info['other_type_2_name']
                    denger_point_name = None
                    if 'denger_point_name' in type_info:
                        denger_point_name = type_info['denger_point_name']
                    new_person_other_info = ThreeProofingResponsiblePersonOtherInfo(
                        type_parent_id = type_parent_id,
                        dept_name = dept_name,
                        other_type_2_name = other_type_2_name,
                        denger_point_name = denger_point_name,
                        person_id = new_person.id,
                        create_by=user_id
                    )
                    db.add(new_person_other_info)
        except:
            db.rollback()
            traceback.print_exc()
            new_person.del_flag='2'
        db.commit()
        # 返回创建成功的响应
        return {
            "code": 200,
            "msg": "创建成功",
            "data": None
        }
    except Exception as e:
        # 处理异常
        db.rollback()
        traceback.print_exc()
        raise HTTPException(status_code=500, detail=str(e))


@router.put('/update')
async def update_contact(
        db: Session = Depends(get_db),
        body=Depends(remove_xss_json),
        user_id=Depends(valid_access_token)
):
    try:
        # 提取请求数据
        id = body['id']
        person = get_person_info_by_id(db,id)

        if not person:
            return JSONResponse(status_code=404, content={
                'errcode': 404,
                'errmsg': '责任人不存在'
            })

        person.unit_name = body['unit_name']
        person.name = body['name']
        person.area_code = body['area_code']
        person.position = body['position']
        person.phone = body['phone']
        if 'telephone' in body:
            person.telephone=body['telephone']
        person.order_num = body['order_num']
        person.unit_id = db_dept.get_dept_id_by_name(db, body['unit_name'])
        type_list = body['type_list']
        old_person_type_list=get_person_type_by_person_id(db,person.id)
        for old_person_type in old_person_type_list:
            old_person_type.del_flag='2'
        old_person_other_type_list = get_person_other_info_by_person_id(db,person.id)
        for old_person_other_type in old_person_other_type_list:
            old_person_other_type.del_flag = '2'
        for type_info in type_list:
            type_parent_id = type_info['type_parent_id']  # get_type_parent_id_by_type_id(db,type_id)
            for type_id in type_info['children']:
                new_person_type = ThreeProofingResponsiblePersonType(
                    type_parent_id=type_parent_id,
                    type_id=type_id,
                    person_id=id,
                    create_by=user_id
                )
                db.add(new_person_type)
            if type_parent_id in ('5', '7', '9'):
                if 'children2' in type_info:
                    for other_type_id in type_info['children2']:
                        new_person_other_type = ThreeProofingResponsiblePersonOtherType(
                            type_parent_id=type_parent_id,
                            other_type_id=other_type_id,
                            person_id=person.id,
                            create_by=user_id
                        )
                        db.add(new_person_other_type)
            if type_parent_id in ('4', '5', '7', '10', '11'):
                dept_name = None
                if 'dept_name' in type_info:
                    dept_name = type_info['dept_name']
                other_type_2_name = None
                if 'other_type_2_name' in type_info:
                    other_type_2_name = type_info['other_type_2_name']
                denger_point_name = None
                if 'denger_point_name' in type_info:
                    denger_point_name = type_info['denger_point_name']
                new_person_other_info = ThreeProofingResponsiblePersonOtherInfo(
                    type_parent_id=type_parent_id,
                    dept_name=dept_name,
                    other_type_2_name=other_type_2_name,
                    denger_point_name=denger_point_name,
                    person_id=person.id,
                    create_by=user_id
                )
                db.add(new_person_other_info)
        # 更新到数据库会话并提交
        db.commit()

        # 返回创建成功的响应
        return {
            "code": 200,
            "msg": "更新成功",
            "data": None
        }
    except Exception as e:
        # 处理异常
        db.rollback()
        traceback.print_exc()
        raise HTTPException(status_code=500, detail=str(e))


@router.get('/list')
async def get_emergency_contact_list(
        type_parent_id: str = Query(None, description='单位名称'),
        area_code:str = Query(None, description='单位名称'),
        Name: str = Query(None, description='联系人'),
        page: int = Query(1, gt=0, description='页码'),
        pageSize: int = Query(10, gt=0, description='每页条目数量'),
        db: Session = Depends(get_db),
        user_id=Depends(valid_access_token)
):
    try:
        # 构建查询
        query = db.query(ThreeProofingResponsiblePerson)
        query = query.filter(ThreeProofingResponsiblePerson.del_flag == '0')
        # 应用查询条件
        if type_parent_id:
            person_list = get_person_list_by_type_parent_id(db,type_parent_id)
            query = query.filter(ThreeProofingResponsiblePerson.id.in_(person_list))
        if Name:
            query = query.filter(ThreeProofingResponsiblePerson.name.like(f'%{Name}%'))
        if area_code:
            query = query.filter(ThreeProofingResponsiblePerson.area_code==area_code)

        # 计算总条目数
        total_items = query.count()

        # 排序

        query = query.order_by(ThreeProofingResponsiblePerson.order_num.asc())
        query = query.order_by(ThreeProofingResponsiblePerson.create_time.desc())
        # 执行分页查询
        contact_infos = query.offset((page - 1) * pageSize).limit(pageSize).all()

        # 将查询结果转换为列表形式的字典
        contact_infos_list = []
        for info in contact_infos:
            type_parent_id_list = get_type_parent_id_by_person_id(db,info.id)
            type_parent_list = []
            type_parent_list2 = []
            for type_parent in type_parent_id_list:
                if type_parent not in type_parent_list2:
                    dict_data = get_dict_data_info(db,'three_proofing',type_parent)
                    type_parent_list2.append(type_parent)
                    type_parent_list.append({"type_parent_id":type_parent,"type_parent":dict_data.dict_label})
            area_info = id_get_area_info(db,info.area_code)
            user_info = user_id_get_user_info(db,info.create_by)
            area_list = db_area.id_get_area_parent_list(db,info.area_code,[])
            contact_infos_list.append({
                "id": info.id,
                "unit_id": info.unit_id,
                "unit_name": info.unit_name,
                "name": info.name,
                "area_list":area_list,
                "area_code": info.area_code,
                "area_name": area_info.area_name,
                "position": info.position,
                "phone": info.phone,
                "telephone":info.telephone,
                "order_num":info.order_num,
                "online_status":'0',
                "create_time": info.create_time.strftime('%Y-%m-%d %H:%M:%S'),
                "create_by":info.create_by,
                "create_user":user_info.nick_name,
                "type_parent_list":type_parent_list
            })
        # 返回结果+
        return {
            "code": 200,
            "msg": "成功",
            "data": contact_infos_list,
            "total": total_items
        }

    except Exception as e:
        # 处理异常
        traceback.print_exc()
        raise HTTPException(status_code=500, detail=str(e))


@router.get('/info/{id}')
async def get_emergency_contact_id_info(
        id: str,
        db: Session = Depends(get_db),
        user_id=Depends(valid_access_token)
):
    try:

        contact = get_person_info_by_id(db,id)
        if not contact:
            return JSONResponse(status_code=404, content={
                'errcode': 404,
                'errmsg': '联系人不存在'
            })
        # 将查询结果转换为列表形式的字典
        area_info = id_get_area_info(db,contact.area_code)
        user_info = user_id_get_user_info(db,contact.create_by)

        area_list = db_area.id_get_area_parent_list(db, contact.area_code, [])
        contact_result = {
            "id": contact.id,
            "unit_id": contact.unit_id,
            "unit_name": contact.unit_name,
            "name": contact.name,
            "area_list":area_list,
            "area_code":contact.area_code,
            "area_name": area_info.area_name,
            "position": contact.position,
            "phone": contact.phone,
            "telephone":contact.telephone,
            "order_num":contact.order_num,
            "online_status":'0',
            "create_time": contact.create_time.strftime('%Y-%m-%d %H:%M:%S'),
            "create_by":contact.create_by,
            "create_user":user_info.nick_name,
            "type_list":[]
        }
        type_parent_id_list = []
        type_list = get_person_type_by_person_id(db,contact.id)
        for type_info in type_list:
            if type_info.type_parent_id not in type_parent_id_list:
                type_parent_id_list.append(type_info.type_parent_id)

        for type_parent_id in type_parent_id_list:
            dict_data = get_dict_data_info(db, 'three_proofing', type_parent_id)
            type_data = {"type_parent_id": type_parent_id, "type_parent": dict_data.dict_label,"children":[]}
            for type_info in get_person_type_by_person_id_and_type_parent_id(db,contact.id,type_parent_id):
                type_data_info = get_type_info_by_id(db,type_info.type_id)
                type_data['children'].append({"id": type_info.type_id, "label": type_data_info.type_name})
            other_info = get_person_other_info_by_person_id_and_type_parent_id(db, contact.id, type_parent_id)
            if other_info:
                type_data['dept_name'] = other_info.dept_name
                type_data['denger_point_name'] = other_info.denger_point_name
                type_data['other_type_2_name'] = other_info.other_type_2_name
            other_type_list = get_person_other_type_by_person_id_and_type_parent_id(db, contact.id, type_parent_id)
            if other_type_list:
                type_data['children2'] = []
            for other_type in other_type_list:
                other_type_info = get_other_type_info_by_id(db, other_type.other_type_id)
                label= ''
                if other_type_info:
                    label = other_type_info.type_name
                type_data['children2'].append({"id": other_type.other_type_id, "label": label})

            contact_result['type_list'].append(type_data)
        # 返回结果
        return {
            "code": 200,
            "msg": "成功",
            "data": contact_result
        }
    except Exception as e:
        # 处理异常
        traceback.print_exc()
        raise HTTPException(status_code=500, detail=str(e))




@router.delete('/delete')
async def delete_emergency_plans(
        ids: list,
        db: Session = Depends(get_db),
        body=Depends(remove_xss_json),
        user_id=Depends(valid_access_token)
):
    try:
        # 提取请求数据
        query = db.query(ThreeProofingResponsiblePerson)
        query = query.filter(ThreeProofingResponsiblePerson.del_flag != '2')
        query = query.filter(ThreeProofingResponsiblePerson.id.in_(ids))
        contacts = query.all()
        if not contacts:
            return JSONResponse(status_code=404, content={
                'errcode': 404,
                'errmsg': '联系人不存在'
            })
        for contact in contacts:
            contact.del_flag = '2'
            contact.create_by = user_id
        # 更新到数据库会话并提交
        db.commit()

        # 返回创建成功的响应
        return {
            "code": 200,
            "msg": "删除成功",
            "data": None
        }
    except Exception as e:
        # 处理异常
        traceback.print_exc()
        raise HTTPException(status_code=500, detail=str(e))


@router.delete('/delete/{id}')
async def delete_emergency_plans(
        id: int,
        db: Session = Depends(get_db),
        body=Depends(remove_xss_json),
        user_id=Depends(valid_access_token)
):
    try:
        contact = get_person_info_by_id(db, id)
        if not contact:
            return JSONResponse(status_code=404, content={
                'errcode': 404,
                'errmsg': '联系人不存在'
            })
        contact.del_flag = '2'
        contact.create_by = user_id
        # 更新到数据库会话并提交
        db.commit()

        # 返回创建成功的响应
        return {
            "code": 200,
            "msg": "删除成功",
            "data": None
        }
    except Exception as e:
        # 处理异常
        traceback.print_exc()
        raise HTTPException(status_code=500, detail=str(e))

#
# @router.post('/createImport')
# async def create_contact(
#         db: Session = Depends(get_db),
#         body=Depends(remove_xss_json),
#         user_id=Depends(valid_access_token)
# ):
#     try:
#         # 提取请求数据
#         filename = body['filename']
#         if len(filename) == 0:
#             raise Exception()
#
#         file_name = filename[0]['url']
#         file_path = f'/data/upload/mergefile/uploads/{file_name}'
#
#         # 检查文件是否存在
#         if not os.path.isfile(file_path):
#             return JSONResponse(status_code=404, content={
#                 'errcode': 404,
#                 'errmsg': f'{file_name}不存在'
#             })
#
#         # 定义预期的列名和对应的最大长度
#         expected_columns = {
#             '单位名称': 255,
#             '联系人': 255,
#             '职务': 255,
#             '粤政易手机号码': 20
#         }
#         try:
#             df = pd.read_excel(file_path, engine='openpyxl', header=0)
#         except Exception as e:
#             raise AppException(500, "请按模板上传！")
#
#         # 读取Excel文件
#         # try:
#         #     df = pd.read_excel(file_path, engine='openpyxl', header=0)
#         # except Exception as e:
#         #     return f"读取Excel文件时出错: {e}"
#
#         # 获取前两行的数据
#         first_two_rows = df
#         # print(first_two_rows)
#         # 检查列名是否匹配
#         actual_columns = first_two_rows.columns.tolist()
#         # print('actual_columns', actual_columns)
#         missing_columns = [col for col in expected_columns.keys() if col not in actual_columns]
#
#         if len(missing_columns) > 0:
#             raise AppException(500, "请按模板上传")
#
#         head1 = df.head(1)
#         head1data = head1.to_dict(orient='records')[0]
#         print(head1data)
#         if head1data['单位名称'] == '填写单位名称' and head1data['联系人'] == '填写单位联系人' and head1data['职务'] == '填写联系人职务' and \
#                 head1data['粤政易手机号码'] == '填写联系人的粤政易注册手机号码':
#             df = df.drop(index=0)
#         else:
#             raise AppException(500, "请按模板上传。")
#
#         # 检查前两行的字段长度
#         for index, row in first_two_rows.iterrows():
#             for col, max_length in expected_columns.items():
#                 # if pd.isna(row[col]):
#                 #     return f"第{index + 1}行的'{col}'字段不能为空。"
#                 if pd.notna(row[col]) and len(str(row[col])) > max_length:
#                     raise AppException(500, f"第{index + 1}行的'{col}'字段长度超过{max_length}字符。")
#
#         data = df.to_dict(orient='records')
#         # print(data)
#         infos = []
#         for info in data:
#             if pd.isna(info['单位名称']) and pd.isna(info['联系人']) and pd.isna(info['粤政易手机号码']):
#                 continue
#             if pd.isna(info['单位名称']) or pd.isna(info['联系人']) or pd.isna(info['粤政易手机号码']):
#                 return "单位名称、联系人、粤政易手机号码为必填"
#             if pd.isna(info['职务']):
#                 info['职务'] = None
#             infos.append(info)
#
#         # 创建新的预案记录
#         for contact in infos:
#             unit_id = db_dept.get_dept_id_by_name(db, contact['单位名称'])
#             if unit_id == '':
#                 raise AppException(500, "单位名称不正确")
#
#             # 删除之前同一个部门的人员
#             db.query(EmergencyContactInfo).filter(
#                 and_(EmergencyContactInfo.del_flag == "0", EmergencyContactInfo.unit_id == unit_id, )).update(
#                 {"del_flag": "2"})
#
#             new_contact = EmergencyContactInfo(
#                 unit_id=unit_id,
#                 unit_name=contact['单位名称'],
#                 contact_name=contact['联系人'],
#                 position=contact['职务'],
#                 yue_gov_ease_phone=contact['粤政易手机号码'],
#                 create_by=user_id
#             )
#
#             # 添加到数据库会话
#             db.add(new_contact)
#         # 提交
#         db.commit()
#
#         # 返回创建成功的响应
#         return {
#             "code": 200,
#             "msg": "创建成功",
#             "data": None
#         }
#
#     except AppException as e:
#         return {
#             "code": 500,
#             "msg": e.msg
#         }
#
#     except Exception as e:
#         traceback.print_exc()
#         # 处理异常
#         raise HTTPException(status_code=500, detail=str(e))