250422-1代码。

routers/api/ThreeProofingResponsible/person.py

@@ -1,7 +1,7 @@
 from fastapi import APIRouter, Request, Depends, HTTPException, Query, BackgroundTasks
 from sqlalchemy.exc import IntegrityError
 from fastapi.responses import HTMLResponse, FileResponse
-from fastapi.responses import JSONResponse
+from fastapi.responses import JSONResponse,StreamingResponse
 from database import get_db
 from sqlalchemy import text, exists, and_, or_, not_
 from sqlalchemy.orm import Session
@@ -372,6 +372,147 @@ async def get_emergency_contact_list(
         raise HTTPException(status_code=500, detail=str(e))
 
 
+@router.get('/export')
+async def get_emergency_contact_list(
+        request: Request,
+        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),
+        auth_user: AuthUser = Depends(find_auth_user),
+        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}%'))
+
+        def get_area_chli(area_list: list, parent_id: int):
+
+            areas = parent_id_get_area_info(db, parent_id)
+            if areas:
+                for area in areas:
+                    area_list.append(area.id)
+                    get_area_chli(area_list, area.id)
+            return area_list
+
+        # if area_code:
+        #     query = query.filter(ThreeProofingResponsiblePerson.area_code.in_(get_area_chli([area_code],area_code)))
+        # '''440900000000'''
+        if area_code:
+            area_info = id_get_area_info(db, area_code)
+            if area_info:
+                area_code = area_info.area_code
+                area_code = area_code.replace('0000000000', '').replace('00000000', '').replace('000000', '').replace(
+                    '000', '')
+                query = query.filter(ThreeProofingResponsiblePerson.area_code2.like(f'%{area_code}%'))
+        # 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.all()
+
+        for info in contact_infos:
+            if three_proofing_responsible_person_data.sign_valid_row(info) == False:
+                raise HmacException(500, "三防责任人管理人员信息表验证异常，已被非法篡改")
+
+        # 将查询结果转换为列表形式的字典
+        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, [])
+            if info.order_num == -1:
+                order_num = ''
+            else:
+                order_num = str(info.order_num)
+            contact_infos_list.append({
+                "id": info.id,
+                "单位ID": info.unit_id,
+                "单位名称": info.unit_name,
+                "姓名": info.name,
+                "区划列表": area_list,
+                "区划编码": info.area_code,
+                "区划": area_info.area_name,
+                "职务": info.position,
+                "手机号码": mpfun.dec_data(info.phone),
+                "办公电话": mpfun.dec_data(info.telephone),
+                "排序顺序": order_num,
+                "创建时间": info.create_time.strftime('%Y-%m-%d %H:%M:%S'),
+                "责任类型": type_parent_list
+            })
+        # 返回结果+
+        import pandas as pd
+        from io import BytesIO
+        # 将查询结果转换为 DataFrame
+        df = pd.DataFrame(contact_infos_list)
+
+        # 将 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(db, auth_user, "三防责任人管理", f"三防责任人清单导出数据成功", request.client.host)
+
+        # 返回文件流
+        return StreamingResponse(output, headers=headers)
+        #
+        # return {
+        #     "code": 200,
+        #     "msg": "成功",
+        #     "data": contact_infos_list,
+        #     "total": total_items
+        # }
+
+    except HmacException as e:
+        return JSONResponse(status_code=e.code, content={
+            "code": e.code,
+            "msg": e.msg
+        })
+        # return {
+        #     "code": e.code,
+        #     "msg": e.msg
+        # }
+
+    except Exception as e:
+        # 处理异常
+        traceback.print_exc()
+        return JSONResponse(status_code=500, content={"code": 500, "msg": f"Internal server error: {str(e)}"})
+
+
 @router.get('/info/{id}')
 async def get_emergency_contact_id_info(
         id: str,