标准编制依据库查询

README.md

@@ -10,13 +10,17 @@
     - uvicorn server.app:app --port=8001 --host=0.0.0.0
     - gunicorn -c gunicorn_config.py server.app:app       多进程启动
 
+    - python .\views\construction_review\app.py
+
+
+    
+
 
     pip install aioredis -i https://mirrors.aliyun.com/pypi/simple/
     pip install langgraph-checkpoint-postgres -i https://mirrors.aliyun.com/pypi/simple/
     pip install langchain-redis -i https://mirrors.aliyun.com/pypi/simple/
 
 
-
 ### 增加组件依赖
   pip install aiomysql -i https://mirrors.aliyun.com/pypi/simple/
 

config/config.ini

@@ -1,7 +1,7 @@
 
 
 [model]
-MODEL_TYPE=qwen_local_14b
+MODEL_TYPE=qwen
 
 
 

config/prompt/common_model_query.yaml

@@ -0,0 +1,10 @@
+
+# 任务提示词
+task_prompt: |
+  你是一个智能助手，根据提供的信息回答问题。
+
+
+
+# test
+template: |
+  ## 测试内容

config/prompt/intent_prompt.yaml

@@ -0,0 +1,22 @@
+
+# 系统提示词
+system_prompt: |
+  基于提供的样例，结合用户最近的对话历史上下文进行意图识别，精准匹配对应的业务场景指令。
+  必须优先参考最近的上下文语义及用户意图演变，若问题与样例中的任一业务场景相符，则返回对应指令；若无法匹配任何已定义场景，则返回 chat_box_generate。
+  严格遵守：仅输出指令字符串，不附加任何解释、说明或格式。
+  用户目前历史上下文信息：
+  {history}
+
+
+
+
+# 意图案例 准备few-shot样例;
+intent_examples: 
+  - inn: 你好;咨询.
+    out: chat_box_generate
+
+  - inn: 执行；操作；查询；处理；
+    out: common_agent
+
+
+           

config/prompt/system_prompt.yaml

@@ -0,0 +1,19 @@
+
+
+# 系统提示词
+system_prompt: |
+  分析专家于一身的AI助手，提供全方位的智能化指导。
+        你的建议要务实、经济、易操作，并能基于物联网数据提供精准预警和具体解决方案。
+            
+
+    
+# 用户上下文会话记录 摘要提示词
+summary_system_prompt: |
+  请总结以下对话内容，保留关键信息：
+  {history}
+
+
+
+# test
+template: |
+  ## 测试内容

config/sql/lq_db.sql

@@ -0,0 +1,68 @@
+
+
+
+-- 1、编制依据基本信息表
+
+
+DROP TABLE IF EXISTS t_basis_of_preparation;
+CREATE TABLE IF NOT EXISTS t_basis_of_preparation (
+    id INT AUTO_INCREMENT PRIMARY KEY COMMENT '标准唯一标识符',
+    chinese_name VARCHAR(500) NOT NULL COMMENT '中文标准名称',
+    english_name VARCHAR(500) COMMENT '英文标准名称',
+    standard_no VARCHAR(100)  COMMENT '标准编号',
+    issuing_authority VARCHAR(200) COMMENT '发布机构',
+    release_date DATE COMMENT '发布日期',
+    implementation_date DATE COMMENT '实施日期',
+    drafting_unit VARCHAR(300) COMMENT '起草单位',
+    approving_department VARCHAR(200) COMMENT '批准部门',
+    document_type VARCHAR(10) COMMENT '标准类型: national-国家标准, industry-行业标准, local-地方标准, enterprise-企业标准',
+    professional_field VARCHAR(15) COMMENT '专业领域:Laws-法律,Technical-技术规范,Reference-参考规范,Internal-内部规范',
+    engineering_phase VARCHAR(100) COMMENT '工程阶段',
+    participating_units VARCHAR(800) COMMENT '参编单位',
+    reference_basis_list VARCHAR(1000) COMMENT '参考依据列表',
+    file_url VARCHAR(500) COMMENT '文件路径',
+		status VARCHAR(10) COMMENT '状态：current-现行，作废-void',
+    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP COMMENT '记录创建时间',
+    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '记录更新时间',
+    INDEX idx_standard_no (standard_no) COMMENT '标准编号索引',
+    INDEX idx_chinese_name (chinese_name(100)) COMMENT '中文名称索引',
+    INDEX idx_release_date (release_date) COMMENT '发布日期索引',
+    INDEX idx_document_type (document_type) COMMENT '标准类型索引',
+    INDEX idx_professional_field (professional_field) COMMENT '专业领域索引'
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='编制依据基本信息表';
+
+
+
+
+
+
+
+
+INSERT INTO t_basis_of_preparation (
+    chinese_name, english_name, standard_no, issuing_authority, 
+    release_date, implementation_date, drafting_unit, approving_department, 
+    document_type, professional_field, engineering_phase, participating_units, 
+    reference_basis_list, file_url, status
+) VALUES
+('中华人民共和国安全生产法', NULL, NULL, NULL, '2021-06-10', NULL, NULL, NULL, 'national', 'Laws', NULL, NULL, NULL, 'https://safety.jining.gov.cn/module/download/downfile.jsp?classid=0&showname=%E4%B8%AD%E5%8D%8E%E4%BA%BA%E6%B0%91%E5%85%B1%E5%92%8C%E5%9B%BD%E5%AE%89%E5%85%A8%E7%94%9F%E4%BA%A7%E6%B3%95%EF%BC%882021%E5%B9%B46%E6%9C%8810%E6%97%A5%E4%BF%AE%E8%AE%A2%E7%89%88%EF%BC%89.pdf&filename=3b0ee62a494049869e9361ec8ee4fb83.pdf', 'current'),
+('公路水运工程质量监督管理规定', NULL, NULL, '交通运输部', '2017-09-14', NULL, NULL, NULL, 'industry', 'Laws', NULL, NULL, NULL, 'https://xxgk.mot.gov.cn/2020/jigou/fgs/202006/t20200623_3307899.html', 'current'),
+('公路水运工程拟淘汰危及生产安全施工工艺、设备和材料目录', NULL, NULL, '交通运输部', NULL, NULL, NULL, NULL, 'industry', 'Technical', NULL, NULL, NULL, 'http://ztjfjt.jtgs.taizhou.gov.cn/cms_files/filemanager/1718223565/attach/20235/7485f997a006433f9d2530c46a4b9861.pdf', 'current'),
+('公路桥涵施工技术规范', NULL, 'JTG/T3650-2020', '交通运输部', NULL, NULL, NULL, NULL, 'industry', 'Technical', NULL, NULL, NULL, 'https://xxgk.mot.gov.cn/2020/jigou/glj/202006/P020200630665628060420.pdf', 'current'),
+('公路工程质量检验评定标准', NULL, 'JTGF80-1-2017', NULL, NULL, NULL, NULL, NULL, 'industry', 'Technical', NULL, NULL, NULL, 'https://jtst.mot.gov.cn/hb/search/stdHBDetailed?id=dd2ffc7d8c33835bad290e9d741f0634', 'current'),
+('公路工程施工安全技术规范', NULL, 'JTGF90-2015', NULL, NULL, NULL, NULL, NULL, 'industry', 'Technical', NULL, NULL, NULL, 'https://jtst.mot.gov.cn/hb/search/stdHBDetailed?id=4c4ab59797b5b4013c4089972fbb2290', 'current'),
+('混凝土结构工程施工质量验收规范', NULL, 'GB50204-2015', NULL, NULL, NULL, NULL, NULL, 'industry', 'Technical', NULL, NULL, NULL, 'http://www.cdapm.com.cn/upload/%E6%B7%B7%E5%87%9D%E5%9C%9F%E7%BB%93%E6%9E%84%E5%B7%A5%E7%A8%8B%E6%96%BD%E5%B7%A5%E8%B4%A8%E9%87%8F%E9%AA%8C%E6%94%B6%E8%A7%84%E8%8C%83GB%2050204-2015.pdf', 'current'),
+('施工现场临时用电安全技术规范', NULL, 'JGJ46-2016', NULL, NULL, NULL, NULL, NULL, 'industry', 'Technical', NULL, NULL, NULL, 'https://zjw.sh.gov.cn/cmsres/73/7320cf3c54aa4a34827bfecbe6ea293d/5a01c703dcca637c3b9247f4c001542f.pdf', 'current'),
+('建筑施工塔式起重机安装、使用、拆卸安全技术规范', NULL, 'JGJ196-2010', NULL, NULL, NULL, NULL, NULL, 'industry', 'Technical', NULL, NULL, NULL, 'https://zjw.sh.gov.cn/cmsres/99/99e29d723c8e49a488df5f787a529711/1314c992b03eb944fe2a020c26d457ae.pdf', 'current'),
+('建筑施工高空作业安全技术规范', NULL, 'JGJ80-2016', NULL, NULL, NULL, NULL, NULL, 'industry', 'Technical', NULL, NULL, NULL, 'https://zjw.sh.gov.cn/cmsres/dd/dd2874d657124e648b54c66a113fb0b1/2b641c95070e63127349d11cc3109bc6.pdf', 'current'),
+('混凝土结构设计规范2015 年版', NULL, 'GB50010-2010', NULL, NULL, NULL, NULL, NULL, 'industry', 'Technical', NULL, NULL, NULL, 'https://www.gbwindows.net/ow-content/uploads/download/gfbzdown/1.0.5%E6%9D%A1/%E5%85%B3%E8%81%94%E6%A0%87%E5%87%86/GB50010-2010(2015%E7%89%88)%20%20%E6%B7%B7%E5%87%9D%E5%9C%9F%E7%BB%93%E6%9E%84%E8%AE%BE%E8%AE%A1%E8%A7%84%E8%8C%83.pdf', 'current'),
+('混凝土结构工程施工质量验收规范', NULL, 'GB50204-2015', NULL, NULL, NULL, NULL, NULL, 'industry', 'Technical', NULL, NULL, NULL, 'http://www.cdapm.com.cn/upload/%E6%B7%B7%E5%87%9D%E5%9C%9F%E7%BB%93%E6%9E%84%E5%B7%A5%E7%A8%8B%E6%96%BD%E5%B7%A5%E8%B4%A8%E9%87%8F%E9%AA%8C%E6%94%B6%E8%A7%84%E8%8C%83GB%2050204-2015.pdf', 'current'),
+('建筑施工模板安全技术规程', NULL, 'JGJ162-2008', NULL, NULL, NULL, NULL, NULL, 'industry', 'Technical', NULL, NULL, NULL, 'http://www.cdapm.com.cn/upload/%E5%BB%BA%E7%AD%91%E6%96%BD%E5%B7%A5%E6%A8%A1%E6%9D%BF%E5%AE%89%E5%85%A8%E6%8A%80%E6%9C%AF%E8%A7%84%E8%8C%83JGJ162-2008.pdf', 'current'),
+('G4216 线屏山新市至金阳段高速公路 XJ4 标段两阶段施工设计图纸', NULL, NULL, NULL, NULL, NULL, NULL, NULL, 'enterprise', 'Reference', NULL, NULL, NULL, NULL, 'current'),
+('建设单位明确的工程施工工期、质量和环境保护要求以及关键工程控制要点', NULL, NULL, NULL, NULL, NULL, NULL, NULL, 'enterprise', 'Laws', NULL, NULL, NULL, NULL, 'current'),
+('本项目总体施工组织设计', NULL, NULL, NULL, NULL, NULL, NULL, NULL, 'enterprise', 'Reference', NULL, NULL, NULL, NULL, 'current'),
+('四川路桥集团《工程技术管理办法》及《工程质量管理办法》', NULL, NULL, NULL, NULL, NULL, NULL, NULL, 'enterprise', 'Internal', NULL, NULL, NULL, NULL, 'current'),
+('《起重机械安全规程》', NULL, 'B6067-2010', NULL, NULL, NULL, NULL, NULL, 'industry', 'Technical', NULL, NULL, NULL, 'https://openstd.samr.gov.cn/bzgk/gb/newGbInfo?hcno=9DED7058601D511BFD5EEE88677548D8', 'current'),
+('《架桥机通用技术条件》', NULL, 'GB/T26470-2011', NULL, NULL, NULL, NULL, NULL, 'industry', 'Technical', NULL, NULL, NULL, 'https://openstd.samr.gov.cn/bzgk/gb/newGbInfo?hcno=F8FC50E035D93142F37F28F0F5E8B678', 'current'),
+('《架桥机安全规程》', NULL, 'GB 26496-2011', NULL, NULL, NULL, NULL, NULL, 'industry', 'Technical', NULL, NULL, NULL, 'https://openstd.samr.gov.cn/bzgk/gb/newGbInfo?hcno=DF194527717A2C929434449D62FF8196', 'current'),
+('《公路水运工程安全生产监督管理办法》', NULL, '交通运输部令2017 年第25号', '交通运输部', NULL, NULL, NULL, NULL, 'industry', 'Laws', NULL, NULL, NULL, 'https://xxgk.mot.gov.cn/2020/gz/202112/t20211227_3633480.html', 'current'),
+('《危险性较大的分部分项工程安全管理规定》', NULL, '住建部令第37 号', '住房和城乡建设部', NULL, NULL, NULL, NULL, 'industry', 'Laws', NULL, NULL, NULL, 'https://www.gov.cn/gongbao/content/2018/content_5294422.htm', 'current');

config/sql/test.sql

@@ -2,7 +2,7 @@
 
 
  -- 测试信息表
-
+ DROP TABLE IF EXISTS test_tab;
 CREATE TABLE IF NOT EXISTS test_tab (
     id INT AUTO_INCREMENT PRIMARY KEY COMMENT '用户唯一标识符',
     name VARCHAR(100) NOT NULL COMMENT '用户姓名',

core/construction_review/component/reviewers/base_reviewer.py

@@ -9,7 +9,7 @@ import time
 from abc import ABC
 from typing import Dict, Any, Optional
 from dataclasses import dataclass
-from langfuse import obverse
+#from langfuse import obverse
 from foundation.agent.monitor.ai_trace_monitor import lf
 from foundation.agent.generate.model_generate import generate_model_client
 from core.construction_review.component.reviewers.utils.prompt_loader import prompt_loader
@@ -32,7 +32,7 @@ class BaseReviewer(ABC):
         self.model_client = generate_model_client
         self.prompt_loader = prompt_loader
     
-    @obverse
+    #@obverse
     async def review(self, name: str, trace_id: str, reviewer_type: str, prompt_name: str, review_content: str,review_references: str = None) -> ReviewResult:
         """
         执行审查

database/repositories/bus_data_query.py

@@ -0,0 +1,36 @@
+from typing import List, Tuple, Any, Optional, Dict
+from foundation.logger.loggering import server_logger
+from foundation.utils.common import handler_err
+from foundation.base.mysql.async_mysql_base_dao import AsyncBaseDAO
+
+
+class BasisOfPreparationDAO(AsyncBaseDAO):
+    """异步编制依据 对象"""
+    
+    
+    async def get_info_by_id(self, id: int) -> Optional[Dict]:
+        """根据ID获取编制依据"""
+        query = "SELECT * FROM t_basis_of_preparation WHERE id = %s"
+        return await self.fetch_one(query, (id,))
+    
+    async def get_list(self) -> List[Dict]:
+        """获取所有编制依据"""
+        query = "SELECT * FROM t_basis_of_preparation WHERE status = 'current' ORDER BY created_at DESC"
+        return await self.fetch_all(query)
+    
+
+    async def get_info_by_condition(self, conditions: Dict) -> List[Dict]:
+        """根据条件查询编制依据"""
+        if not conditions:
+            return await self.get_list()
+        
+        try:
+            where_clause = " AND ".join([f"{field} = %s" for field in conditions.keys()])
+            where_values = list(conditions.values())
+            
+            query = f"SELECT * FROM t_basis_of_preparation WHERE {where_clause} AND status = 'current' ORDER BY created_at DESC"
+            return await self.fetch_all(query, tuple(where_values))
+            
+        except Exception as err:
+            handler_err(logger=server_logger, err=err, err_name="条件查询失败")
+            raise

foundation/base/mysql/async_mysql_base_dao.py

@@ -216,161 +216,4 @@ class TestTabDAO(AsyncBaseDAO):
         return await self.fetch_all(query)
     
 
-    async def get_users_by_condition(self, conditions: Dict) -> List[Dict]:
-        """根据条件查询用户"""
-        if not conditions:
-            return await self.get_all_users()
-        
-        try:
-            where_clause = " AND ".join([f"{field} = %s" for field in conditions.keys()])
-            where_values = list(conditions.values())
-            
-            query = f"SELECT * FROM test_tab WHERE {where_clause} AND status = 'active' ORDER BY created_at DESC"
-            return await self.fetch_all(query, tuple(where_values))
-            
-        except Exception as err:
-            handler_err(logger=server_logger, err=err, err_name="条件查询用户失败")
-            raise
-    
-    # ========== 修改方法 ==========
-    
-    async def update_user(self, user_id: int, **updates) -> bool:
-        """
-        更新用户信息
-        
-        Args:
-            user_id: 用户ID
-            **updates: 要更新的字段，如 name='新名字', age=25, email='new@email.com'
-        
-        Returns:
-            bool: 更新是否成功
-        """
-        if not updates:
-            server_logger.warning("没有提供更新字段")
-            return False
-        
-        # 过滤允许更新的字段
-        allowed_fields = {'name', 'email', 'age', 'status'}
-        valid_updates = {k: v for k, v in updates.items() if k in allowed_fields}
-        
-        if not valid_updates:
-            server_logger.warning("没有有效的更新字段")
-            return False
-        
-        try:
-            return await self.update_by_id('test_tab', user_id, valid_updates)
-            
-        except Exception as err:
-            handler_err(logger=server_logger, err=err, err_name="更新用户失败")
-            raise
-    
-    async def update_user_by_email(self, email: str, **updates) -> bool:
-        """
-        根据邮箱更新用户信息
-        
-        Args:
-            email: 用户邮箱
-            **updates: 要更新的字段
-        
-        Returns:
-            bool: 更新是否成功
-        """
-        if not updates:
-            server_logger.warning("没有提供更新字段")
-            return False
-        
-        # 过滤允许更新的字段
-        allowed_fields = {'name', 'age', 'status'}
-        valid_updates = {k: v for k, v in updates.items() if k in allowed_fields}
-        
-        if not valid_updates:
-            server_logger.warning("没有有效的更新字段")
-            return False
-        
-        try:
-            return await self.update_record('test_tab', valid_updates, {'email': email})
-            
-        except Exception as err:
-            handler_err(logger=server_logger, err=err, err_name="根据邮箱更新用户失败")
-            raise
-    
-    async def update_user_status(self, user_id: int, status: str) -> bool:
-        """
-        更新用户状态
-        
-        Args:
-            user_id: 用户ID
-            status: 状态值 ('active' 或 'inactive')
-        
-        Returns:
-            bool: 更新是否成功
-        """
-        if status not in ('active', 'inactive'):
-            raise ValueError("状态值必须是 'active' 或 'inactive'")
-        
-        try:
-            return await self.update_user(user_id, status=status)
-            
-        except Exception as err:
-            handler_err(logger=server_logger, err=err, err_name="更新用户状态失败")
-            raise
-    
-    async def batch_update_users(self, updates_list: List[Dict]) -> bool:
-        """
-        批量更新用户信息
-        
-        Args:
-            updates_list: 更新数据列表，每个元素必须包含id字段
-        
-        Returns:
-            bool: 批量更新是否成功
-        """
-        try:
-            return await self.batch_update('test_tab', updates_list, 'id')
-            
-        except Exception as err:
-            handler_err(logger=server_logger, err=err, err_name="批量更新用户失败")
-            raise
-    
-    async def update_users_age_range(self, min_age: int, max_age: int, updates: Dict) -> bool:
-        """
-        更新年龄范围内的用户
-        
-        Args:
-            min_age: 最小年龄
-            max_age: 最大年龄
-            updates: 要更新的字段
-        
-        Returns:
-            bool: 更新是否成功
-        """
-        try:
-            where_sql = "age BETWEEN %s AND %s AND status = 'active'"
-            params = (min_age, max_age)
-            
-            return await self.update_with_condition('test_tab', updates, where_sql, params)
-            
-        except Exception as err:
-            handler_err(logger=server_logger, err=err, err_name="更新年龄范围用户失败")
-            raise
-    
-    async def increment_user_age(self, user_id: int, increment: int = 1) -> bool:
-        """
-        增加用户年龄
-        
-        Args:
-            user_id: 用户ID
-            increment: 增加的值，默认为1
-        
-        Returns:
-            bool: 更新是否成功
-        """
-        try:
-            sql = "UPDATE test_tab SET age = age + %s WHERE id = %s AND status = 'active'"
-            return await self.execute_query(sql, (increment, user_id))
-            
-        except Exception as err:
-            handler_err(logger=server_logger, err=err, err_name="增加用户年龄失败")
-            raise
-
 

foundation/utils/tool_utils.py

@@ -9,7 +9,7 @@ from foundation.logger.loggering import server_logger
 from foundation.utils.common import handler_err
 from foundation.base.config import config_handler
 import json
-from datetime import datetime
+from datetime import datetime, date
 # 获取当前文件的目录
 current_dir = os.path.dirname(__file__)
 # 构建到 .env 的相对路径
@@ -44,8 +44,11 @@ def get_system_prompt() -> str:
 
 
 
+
 class DateTimeEncoder(json.JSONEncoder):
     def default(self, obj):
         if isinstance(obj, datetime):
-            return obj.isoformat()  # 转换为 ISO 8601 格式字符串
+            return obj.strftime('%Y-%m-%d %H:%M:%S')
+        elif isinstance(obj, date):  # 添加对 date 类型的支持
+            return obj.strftime('%Y-%m-%d')
         return super().default(obj)

requirements.txt

@@ -182,4 +182,7 @@ langgraph-checkpoint-postgres==2.0.23
 langgraph-checkpoint-redis==0.0.8
 langchain-redis==0.2.3
 aiomysql==0.3.2
+celery=5.5.3
+pypdf==6.2.0
+grandalf==0.8
 

views/test_views.py

@@ -24,6 +24,7 @@ from views import test_router, get_operation_id
 from foundation.agent.workflow.test_workflow_graph import test_workflow_graph
 
 from foundation.base.mysql.async_mysql_base_dao import TestTabDAO
+from database.repositories.bus_data_query import BasisOfPreparationDAO
 from foundation.utils.tool_utils import DateTimeEncoder
 
 
@@ -511,3 +512,85 @@ async def test_mysql_add(
         handler_err(server_logger, trace_id=trace_id, err=err, err_name="/mysql/update")
         return JSONResponse(return_json(code=100500, msg=f"{err}", trace_id=trace_id))
     
+
+
+
+
+@test_router.post("/bop/get", response_model=TestForm)
+async def test_bop_get(
+        request: Request,
+        param: TestForm,
+        trace_id: str = Depends(get_operation_id)):
+    """
+    根据MySQL应用
+    """
+    try:
+        server_logger.info(trace_id=trace_id, msg=f"{param}")
+        # 验证参数
+
+        # 从字典中获取input
+        input_data = param.input
+        session_id = param.config.session_id
+        context = param.context
+        header_info = {
+        }
+        # 从app.state中获取数据库连接池
+        async_db_pool = request.app.state.async_db_pool
+        bop_dao = BasisOfPreparationDAO(async_db_pool)
+        test_id = input_data;
+        data = await bop_dao.get_info_by_id(id=test_id)
+        server_logger.info(trace_id=trace_id, msg=f"【result】: {data}", log_type="/bop/get")
+        json_str = json.dumps(data , cls=DateTimeEncoder, ensure_ascii=False, indent=2)
+        output = json_str
+        # 直接执行
+        server_logger.debug(trace_id=trace_id, msg=f"【result】: {output}", log_type="/bop/get")
+        # 返回字典格式的响应
+        return JSONResponse(
+            return_json(data={"output": output}, data_type="text", trace_id=trace_id))
+    except ValueError as err:
+        handler_err(server_logger, trace_id=trace_id, err=err, err_name="/bop/get")
+        return JSONResponse(return_json(code=100500, msg=f"{err}", trace_id=trace_id))
+
+    except Exception as err:
+        handler_err(server_logger, trace_id=trace_id, err=err, err_name="/bop/get")
+        return JSONResponse(return_json(code=100500, msg=f"{err}", trace_id=trace_id))
+    
+@test_router.post("/bop/list", response_model=TestForm)
+async def test_mysql_add(
+        request: Request,
+        param: TestForm,
+        trace_id: str = Depends(get_operation_id)):
+    """
+    根据MySQL应用
+    """
+    try:
+        server_logger.info(trace_id=trace_id, msg=f"{param}")
+        # 验证参数
+
+        # 从字典中获取input
+        input_data = param.input
+        session_id = param.config.session_id
+        context = param.context
+        header_info = {
+        }
+        # 从app.state中获取数据库连接池
+        async_db_pool = request.app.state.async_db_pool
+        from foundation.base.mysql.async_mysql_base_dao import TestTabDAO
+        bop_dao = BasisOfPreparationDAO(async_db_pool)
+        test_id = input_data;
+        data = await bop_dao.get_list()
+        server_logger.info(trace_id=trace_id, msg=f"【result】: {data}", log_type="/bop/list")
+        json_str = json.dumps(data , cls=DateTimeEncoder, ensure_ascii=False, indent=2)
+        output = json_str
+        # 直接执行
+        server_logger.debug(trace_id=trace_id, msg=f"【result】: {output}", log_type="/bop/list")
+        # 返回字典格式的响应
+        return JSONResponse(
+            return_json(data={"output": output}, data_type="text", trace_id=trace_id))
+    except ValueError as err:
+        handler_err(server_logger, trace_id=trace_id, err=err, err_name="/bop/list")
+        return JSONResponse(return_json(code=100500, msg=f"{err}", trace_id=trace_id))
+
+    except Exception as err:
+        handler_err(server_logger, trace_id=trace_id, err=err, err_name="/bop/list")
+        return JSONResponse(return_json(code=100500, msg=f"{err}", trace_id=trace_id))