250424-1代码。

routers/api/Knowledge/__init__.py

@@ -2,15 +2,19 @@ from fastapi import APIRouter, Request, Depends, HTTPException, Query,Response
 from sqlalchemy.exc import IntegrityError
 
 from common.security import valid_access_token
-from fastapi.responses import JSONResponse
+from fastapi.responses import JSONResponse,StreamingResponse
+from common.db import db_czrz
+from common.auth_user import *
 from database import get_db
 from sqlalchemy.orm import Session
 from models import *
+from utils.ry_system_util import *
 import json
 import random
 from sqlalchemy import create_engine, select, or_
 from typing import Optional
 import uuid
+import traceback
 router = APIRouter()
 
 
@@ -170,11 +174,117 @@ async def select_knowledge(
     return result
 
 
+@router.get('/export')
+async def select_knowledge(
+        request: Request,
+        db: Session = Depends(get_db),
+        sortBy: str = Query(..., description="排序字段"),
+        sortOrder: str = Query(..., description="排序顺序"),
+        pageNum: int = Query(1, gt=0, description="页码"),
+        pageSize: int = Query(10, gt=0, le=100, description="每页大小"),
+        eventType: str = Query(None, description="事件类型"),
+        publishDateRange: str = Query(None, description="发布日期范围"),
+        auth_user: AuthUser = Depends(find_auth_user),
+        query: str = Query(None, description="查询关键字", user_id=Depends(valid_access_token))
+):
+    try:
+        data_query = db.query(KnowledgeBase)
+        data_query = data_query.filter(KnowledgeBase.del_flag != '2')
+
+        if eventType:
+            data_query = data_query.filter(KnowledgeBase.eventType == eventType)
+
+        if publishDateRange:
+            start_date, end_date = publishDateRange.split('-')
+            data_query = data_query.filter(KnowledgeBase.publishDate.between(start_date, end_date))
+
+        if query:
+            search_fields = [getattr(KnowledgeBase, field) for field in ('reportName', 'publishingUnit', 'reportId') if
+                             hasattr(KnowledgeBase, field)]
+            search_conditions = [field.like(f'%{query}%') for field in search_fields]
+            data_query = data_query.filter(or_(*search_conditions))
+
+        if hasattr(KnowledgeBase, sortBy):
+            sort_attr = getattr(KnowledgeBase, sortBy)
+            data_query = data_query.order_by(sort_attr.asc() if sortOrder == 'asc' else sort_attr.desc())
+        total_count = data_query.count()
+        offset = (pageNum - 1) * pageSize
+        # data_query = data_query.offset(offset).limit(pageSize)
+
+        fields = ['reportId', 'reportName', 'eventType', 'publishDate', 'publishingUnit', 'summary', 'subject',
+                  'notificationType', 'base_code']
+        entities = [getattr(KnowledgeBase, field) for field in fields if hasattr(KnowledgeBase, field)]
+        data = data_query.with_entities(*entities).all()
+        # mm_event_type_list = dict_type_get_dict_data_info(db,'mm_event_type')
+        mm_event_type = {}
+        for i in dict_type_get_dict_data_info(db,'mm_event_type'):
+            mm_event_type[i.dict_value]=i.dict_label
+        result_items = []
+        for item in data:
+            item_dict = {field: getattr(item, field) for field in fields}
+            # print(item_dict)
+            # base_code = item_dict['base_code']
+            # kf_entries = db.query(KnowledgeFile).filter(KnowledgeFile.knowledge_base_code == base_code).filter(
+            #     KnowledgeFile.is_deleted != '2').all()
+            # item_dict['files'] = [
+            #     {
+            #         "fileIdentifier": file.file_identifier,
+            #         "fileName": file.file_name,
+            #         "url": file.storage_file_name,
+            #         "status": "success"
+            #     }
+            #     for file in kf_entries
+            # ]
+            item_dict = {"报告编号":item_dict["reportId"],
+                         "报告名称":item_dict["reportName"],
+                         "主题词":item_dict["subject"],
+                         "事件类型":mm_event_type[item_dict["eventType"]],
+                         "摘要":item_dict["summary"],
+                         "来源单位":item_dict["publishingUnit"],
+                         "发布日期":item_dict["publishDate"],
+                         "知识类型":item_dict["notificationType"]}
+            result_items.append(item_dict)
+        import pandas as pd
+        from io import BytesIO
+        # 将查询结果转换为 DataFrame
+        df = pd.DataFrame(result_items)
+
+        # 将 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)
+        # result = {
+        #     "code": 200,
+        #     'msg': '查询成功',
+        #     'pages': (total_count + pageSize - 1) // pageSize,
+        #     'total': total_count,
+        #     "currentPage": pageNum,
+        #     "pageSize": pageSize,
+        #     'data': result_items
+        # }
+        #
+        # return result
 
-
-
-
+    except Exception as e:
+        traceback.print_exc()
+        # 处理异常
+        db.rollback()
+        raise HTTPException(status_code=500, detail=str(e))
 
 @router.get('/detail')
 async def get_knowledge_detail(db: Session = Depends(get_db), reportID: Optional[str] = Query(None, description="报告ID"), user_id=Depends(valid_access_token)):