""" 样本中心服务层 从 sample_view.py 提取的SQL查询逻辑 """ import logging import uuid from datetime import datetime from typing import Optional, List, Dict, Any, Tuple from app.base.async_mysql_connection import get_db_connection from app.base.minio_connection import get_minio_manager from app.core.config import config_handler from app.services.milvus_service import milvus_service from app.services.task_service import task_service logger = logging.getLogger(__name__) # 表名映射 TABLE_MAP = { 'standard': 't_samp_standard_base_info', 'construction_plan': 't_samp_construction_plan_base_info', 'regulation': 't_samp_office_regulations' } # 文档类型映射 (前端显示 -> 数据库存储) DOCUMENT_TYPE_MAP = { "国家标准": "GB", "行业标准": "HY", "部门规章": "BM", "地方标准": "DB", "企业标准": "QY", "管理制度": "GL", "技术规范": "GF", "团体标准": "TT", "国际标准": "GJ", "国家法律": "FL", "地方法规": "LR", "其他": "QT" } # 文档类型反向映射 (数据库存储 -> 前端显示) # 包含旧版本简写以保证存量数据兼容性 DOCUMENT_TYPE_REVERSE_MAP = {v: k for k, v in DOCUMENT_TYPE_MAP.items()} DOCUMENT_TYPE_REVERSE_MAP.update({ "HB": "行业标准", "QB": "企业标准", "GLZD": "管理制度", "JSGF": "技术规范", "TB": "团体标准", "IB": "国际标准", "OTHER": "其他" }) # 专业领域映射 (前端显示 -> 数据库存储) PROFESSIONAL_FIELD_MAP = { "法律法规": "FL", "通用标准": "TY", "勘察钻探": "KC", "地基基础": "DJ", "路基路面": "LJ", "桥梁工程": "QL", "隧道工程": "SD", "交通工程": "JT", "建筑工程": "JZ", "市政工程": "SZ", "机电安装": "JD", "路桥工程": "LB", "装饰装修": "ZS", "港口航道": "GK", "铁路工程": "TL", "房建工程": "FJ", "水利电力": "SL", "信息化": "XX", "试验检测": "SY", "安全环保": "AQ", "其他": "QT" } # 专业领域反向映射 (数据库存储 -> 前端显示) PROFESSIONAL_FIELD_REVERSE_MAP = {v: k for k, v in PROFESSIONAL_FIELD_MAP.items()} # 方案类别映射 (前端显示 -> 数据库存储) PLAN_CATEGORY_MAP = { "超危大方案": "CH", "超危大方案较大Ⅱ级": "CH2", "超危大方案特大Ⅳ级": "CH4", "超危大方案一般Ⅰ级": "CH1", "超危大方案重大Ⅲ级": "CH3", "危大方案": "WD", "一般方案": "YB" } # 方案类别反向映射 (数据库存储 -> 前端显示) PLAN_CATEGORY_REVERSE_MAP = {v: k for k, v in PLAN_CATEGORY_MAP.items()} # 一级分类映射 (前端显示 -> 数据库存储) FIRST_LEVEL_MAP = { "施工方案": "SC" } # 一级分类反向映射 (数据库存储 -> 前端显示) FIRST_LEVEL_REVERSE_MAP = {v: k for k, v in FIRST_LEVEL_MAP.items()} # 二级分类映射 (前端显示 -> 数据库存储) SECOND_LEVEL_MAP = { "临建工程": "LZ", "路基工程": "LJ", "桥梁工程": "QL", "隧道工程": "SD", "其他": "QT" } # 二级分类反向映射 (数据库存储 -> 前端显示) SECOND_LEVEL_REVERSE_MAP = {v: k for k, v in SECOND_LEVEL_MAP.items()} # 三级分类映射 (前端显示 -> 数据库存储) THIRD_LEVEL_MAP = { "TBM施工": "TM", "拌和站安、拆施工": "BH", "不良地质隧道施工": "BL", "常规桥梁": "CG", "挡土墙工程类": "DT", "辅助坑道施工": "FB", "复杂洞口工程施工": "FD", "钢筋加工场安、拆": "GG", "钢栈桥施工": "GZ", "拱桥": "GH", "涵洞工程类": "HD", "滑坡体处理类": "HP", "路堤": "LT", "路堑": "LQ", "其他": "QT", "深基坑": "JK", "隧道总体施工": "ZT", "特殊结构隧道": "TS", "斜拉桥": "XL", "悬索桥": "XS" } # 三级分类反向映射 (数据库存储 -> 前端显示) THIRD_LEVEL_REVERSE_MAP = {v: k for k, v in THIRD_LEVEL_MAP.items()} # 四级分类映射 (前端显示 -> 数据库存储) FOURTH_LEVEL_MAP = { "挡土墙": "DT", "顶管": "DG", "断层破碎带及软弱围岩": "DL", "钢筋砼箱涵": "GX", "高填路堤": "GT", "抗滑桩": "KH", "软岩大变形隧道": "RY", "上部结构": "SB", "深基坑开挖与支护": "JK", "深挖路堑": "LC", "隧道TBM": "TM", "隧道进洞": "JD", "隧道竖井": "SJ", "隧道斜井": "XJ", "特种设备": "TZ", "瓦斯隧道": "WS", "下部结构": "XB", "小净距隧道": "NJ", "岩爆隧道": "YB", "岩溶隧道": "YR", "涌水突泥隧道": "YN", "桩基础": "ZJ", "其他": "QT" } # 四级分类反向映射 (数据库存储 -> 前端显示) FOURTH_LEVEL_REVERSE_MAP = {v: k for k, v in FOURTH_LEVEL_MAP.items()} def get_table_name(source_type: str) -> Optional[str]: """根据source_type获取表名""" return TABLE_MAP.get(source_type) class SampleService: """样本中心服务类 - 使用 SQL 查询方式""" _initialized = False def __init__(self): """初始化服务""" # 使用统一的 MinIO 管理器 self.minio_manager = get_minio_manager() # 使用全局 Milvus 服务 self.milvus_service = milvus_service # 确保集合已创建 (仅在首次初始化时执行) if not SampleService._initialized: try: self.milvus_service.ensure_collections() SampleService._initialized = True except Exception as e: logger.error(f"初始化 Milvus 集合失败: {e}") def _format_document_row(self, item: Dict[str, Any]) -> Dict[str, Any]: """格式化文档行数据，处理URL、日期映射和文件名显示""" if not item: return item # 1. 处理 URL 转换 for key in ['file_url', 'md_url', 'json_url']: if item.get(key): item[key] = self.minio_manager.get_full_url(item[key]) # 2. 映射字段以适配前端通用显示 source_type = item.get('source_type') # 处理文档类型显示 (简写 -> 中文) doc_type_code = item.get('document_type') if doc_type_code in DOCUMENT_TYPE_REVERSE_MAP: item['document_type'] = DOCUMENT_TYPE_REVERSE_MAP[doc_type_code] # 处理专业领域显示 (简写 -> 中文) prof_field_code = item.get('professional_field') if prof_field_code in PROFESSIONAL_FIELD_REVERSE_MAP: item['professional_field'] = PROFESSIONAL_FIELD_REVERSE_MAP[prof_field_code] # 处理一级分类显示 (简写 -> 中文) level_1_code = item.get('level_1_classification') if level_1_code in FIRST_LEVEL_REVERSE_MAP: item['level_1_classification'] = FIRST_LEVEL_REVERSE_MAP[level_1_code] # 处理二级分类显示 (简写 -> 中文) level_2_code = item.get('level_2_classification') if level_2_code in SECOND_LEVEL_REVERSE_MAP: item['level_2_classification'] = SECOND_LEVEL_REVERSE_MAP[level_2_code] # 处理三级分类显示 (简写 -> 中文) level_3_code = item.get('level_3_classification') if level_3_code in THIRD_LEVEL_REVERSE_MAP: item['level_3_classification'] = THIRD_LEVEL_REVERSE_MAP[level_3_code] # 处理四级分类显示 (简写 -> 中文) level_4_code = item.get('level_4_classification') if level_4_code in FOURTH_LEVEL_REVERSE_MAP: item['level_4_classification'] = FOURTH_LEVEL_REVERSE_MAP[level_4_code] # 处理方案类别显示 (简写 -> 中文) plan_category_code = item.get('plan_category') if plan_category_code in PLAN_CATEGORY_REVERSE_MAP: item['plan_category'] = PLAN_CATEGORY_REVERSE_MAP[plan_category_code] if source_type == 'standard': # 标准规范特有映射 if 'standard_number' in item: item['standard_no'] = item.get('standard_number') elif source_type == 'construction_plan': # 施工方案特有映射 item['issuing_authority'] = item.get('compiling_unit') item['release_date'] = item.get('compiling_date') elif source_type == 'regulation': # 办公制度特有映射 item['issuing_authority'] = item.get('issuing_department') item['release_date'] = item.get('publish_date') # 3. 格式化时间 date_keys = [ 'created_time', 'updated_time', 'release_date', 'publish_date', 'compiling_date', 'implementation_date', 'effective_start_date', 'effective_end_date' ] for key in date_keys: val = item.get(key) if val and hasattr(val, 'isoformat'): item[key] = val.isoformat() elif val is not None: item[key] = str(val) # 4. 增加格式化文件名供前端显示 if item.get('conversion_status') == 2: title = item.get('title', 'document') item['md_display_name'] = f"{title}.md" item['json_display_name'] = f"{title}.json" return item async def get_upload_url(self, filename: str, content_type: str, prefix: str = None) -> Tuple[bool, str, Dict[str, Any]]: """获取 MinIO 预签名上传 URL""" try: data = self.minio_manager.get_upload_url(filename, content_type, prefix=prefix) return True, "成功获取上传链接", data except Exception as e: logger.exception("生成上传链接失败") return False, f"生成上传链接失败: {str(e)}", {} # ==================== 文档管理 ==================== async def batch_enter_knowledge_base(self, doc_ids: List[str], username: str, kb_method: str = "general", chunk_size: int = 500, separator: str = "。") -> Tuple[int, str]: """批量将文档入库到知识库 Args: doc_ids: 文档ID列表 username: 操作人 kb_method: 切分方法 chunk_size: 切分长度 separator: 切分符号 """ conn = get_db_connection() if not conn: return 0, "数据库连接失败，请检查数据库服务状态" cursor = conn.cursor() success_count = 0 already_entered_count = 0 failed_count = 0 error_details = [] try: # 1. 获取所有选中选中的文档详情 placeholders = ','.join(['%s']*len(doc_ids)) fetch_sql = f""" SELECT id, title, source_type, md_url, conversion_status, whether_to_enter, created_time, kb_id, file_url FROM t_samp_document_main WHERE id IN ({placeholders}) """ cursor.execute(fetch_sql, tuple(doc_ids)) selected_docs = cursor.fetchall() if not selected_docs: return 0, "选中的文档在数据库中不存在" # 2. 逐份处理 for doc in selected_docs: doc_id = doc['id'] title = doc.get('title', '未命名文档') status = doc.get('conversion_status') whether_to_enter = doc.get('whether_to_enter', 0) md_url = doc.get('md_url') source_type = doc.get('source_type') # A. 检查是否已入库 if whether_to_enter == 1: logger.info(f"文档 {title}({doc_id}) 已入库，跳过二次入库") already_entered_count += 1 error_details.append(f"· {title}: 已在知识库中，无需重复入库") continue # B. 检查转换状态 if status != 2: reason = "尚未转换成功" if status == 0 else "正在转换中" if status == 1 else "转换失败" logger.warning(f"文档 {title}({doc_id}) 状态为 {status}，入库失败: {reason}") failed_count += 1 error_details.append(f"· {title}: {reason}") continue if not md_url: logger.warning(f"文档 {title}({doc_id}) 缺少 md_url，入库失败") failed_count += 1 error_details.append(f"· {title}: 转换结果地址丢失") continue # C. 确定入库策略 (从数据库读取已绑定的知识库) current_kb_id = doc.get('kb_id') current_kb_method = kb_method # 直接使用前端传来的切分方式 if not current_kb_id: logger.warning(f"文档 {title}({doc_id}) 未指定知识库，跳过入库") failed_count += 1 error_details.append(f"· {title}: 未指定目标知识库") continue if not current_kb_method: logger.warning(f"文档 {title}({doc_id}) 未指定切分方式，跳过入库") failed_count += 1 error_details.append(f"· {title}: 未指定切分策略") continue # 获取知识库信息 (collection_name_parent, collection_name_children) kb_sql = "SELECT collection_name_parent, collection_name_children FROM t_samp_knowledge_base WHERE id = %s AND is_deleted = 0" cursor.execute(kb_sql, (current_kb_id,)) kb_res = cursor.fetchone() if not kb_res: logger.warning(f"找不到指定的知识库: id={current_kb_id}") failed_count += 1 error_details.append(f"· {title}: 指定的知识库不存在或已被删除") continue collection_name_parent = kb_res['collection_name_parent'] collection_name_children = kb_res['collection_name_children'] # D. 从 MinIO 获取 Markdown 内容 try: md_content = self.minio_manager.get_object_content(md_url) if not md_content: raise ValueError(f"无法从 MinIO 读取内容 (URL: {md_url})") except Exception as minio_err: logger.error(f"读取文档 {title} 内容失败: {minio_err}") failed_count += 1 error_details.append(f"· {title}: 读取云端文件失败") continue # E. 调用 MilvusService 进行切分和入库 try: # 获取业务元数据 business_metadata = {} if source_type in ['standard', 'basis']: std_sql = """ SELECT id as basic_info_id, chinese_name, standard_number, issuing_authority, document_type, professional_field, validity FROM t_samp_standard_base_info WHERE id = %s """ cursor.execute(std_sql, (doc_id,)) business_metadata = cursor.fetchone() or {} elif source_type == 'construction_plan': plan_sql = """ SELECT id as basic_info_id, plan_name, project_name, project_section, compiling_unit, compiling_date, plan_summary, plan_category, level_1_classification, level_2_classification, level_3_classification, level_4_classification FROM t_samp_construction_plan_base_info WHERE id = %s """ cursor.execute(plan_sql, (doc_id,)) business_metadata = cursor.fetchone() or {} elif source_type == 'regulation': reg_sql = """ SELECT id as basic_info_id, file_name, issuing_department as issuing_authority, document_type, publish_date as release_date, note FROM t_samp_office_regulations WHERE id = %s """ cursor.execute(reg_sql, (doc_id,)) business_metadata = cursor.fetchone() or {} # 强制转换 document_type 为简写 (处理存量数据或异常情况) if 'document_type' in business_metadata and business_metadata['document_type']: doc_type = business_metadata['document_type'] business_metadata['document_type'] = DOCUMENT_TYPE_MAP.get(doc_type, doc_type) # 强制转换 professional_field 为简写 if 'professional_field' in business_metadata and business_metadata['professional_field']: prof_field = business_metadata['professional_field'] business_metadata['professional_field'] = PROFESSIONAL_FIELD_MAP.get(prof_field, prof_field) # 强制转换 plan_category 为简写 if 'plan_category' in business_metadata and business_metadata['plan_category']: plan_cat = business_metadata['plan_category'] business_metadata['plan_category'] = PLAN_CATEGORY_MAP.get(plan_cat, plan_cat) # 强制转换 level_1_classification 为简写 if 'level_1_classification' in business_metadata and business_metadata['level_1_classification']: l1_class = business_metadata['level_1_classification'] business_metadata['level_1_classification'] = FIRST_LEVEL_MAP.get(l1_class, l1_class) # 强制转换 level_2_classification 为简写 if 'level_2_classification' in business_metadata and business_metadata['level_2_classification']: l2_class = business_metadata['level_2_classification'] business_metadata['level_2_classification'] = SECOND_LEVEL_MAP.get(l2_class, l2_class) # 强制转换 level_3_classification 为简写 if 'level_3_classification' in business_metadata and business_metadata['level_3_classification']: l3_class = business_metadata['level_3_classification'] business_metadata['level_3_classification'] = THIRD_LEVEL_MAP.get(l3_class, l3_class) # 强制转换 level_4_classification 为简写 if 'level_4_classification' in business_metadata and business_metadata['level_4_classification']: l4_class = business_metadata['level_4_classification'] business_metadata['level_4_classification'] = FOURTH_LEVEL_MAP.get(l4_class, l4_class) # 统一使用主表的 file_url，确保数据来源一致，入库时使用完整 URL raw_file_url = doc.get('file_url', '') business_metadata['file_url'] = self.minio_manager.get_full_url(raw_file_url) if raw_file_url else '' # 准备元数据 current_date = int(datetime.now().strftime('%Y%m%d')) doc_info = { "doc_id": doc_id, "file_name": title, "doc_version": int(doc['created_time'].strftime('%Y%m%d')) if doc.get('created_time') else current_date, "tags": "", "user_id": username, # 传递操作人作为 created_by "kb_id": current_kb_id, "kb_method": current_kb_method, "collection_name_parent": collection_name_parent, "collection_name_children": collection_name_children, "chunk_size": chunk_size, "separator": separator, "source_type": source_type, "business_metadata": business_metadata # 注入业务元数据 } await self.milvus_service.insert_knowledge(md_content, doc_info) # F. 添加到任务管理中心 (类型为 data) try: await task_service.add_task(doc_id, 'data') except Exception as task_err: logger.error(f"添加文档 {title} 到任务中心失败: {task_err}") # G. 更新数据库状态 update_sql = "UPDATE t_samp_document_main SET whether_to_enter = 1, kb_id = %s, kb_method = %s, updated_by = %s, updated_time = NOW() WHERE id = %s" cursor.execute(update_sql, (current_kb_id, current_kb_method, username, doc_id)) success_count += 1 except Exception as milvus_err: logger.exception(f"文档 {title} 写入向量库失败") failed_count += 1 error_details.append(f"· {title}: 写入向量库失败 ({str(milvus_err)})") continue conn.commit() # 构造详细的消息 if success_count == len(doc_ids) and failed_count == 0 and already_entered_count == 0: msg = f"✅ 入库成功！共处理 {success_count} 份文档。" else: msg = f"📊 入库处理完成：\n· 成功：{success_count} 份\n" if already_entered_count > 0: msg += f"· 跳过：{already_entered_count} 份 (已入库)\n" if failed_count > 0: msg += f"· 失败：{failed_count} 份\n" if error_details: detailed_msg = msg + "\n⚠️ 详情说明：\n" + "\n".join(error_details) return success_count, detailed_msg return success_count, msg except Exception as e: logger.exception(f"文档批量入库异常: {e}") conn.rollback() return 0, f"操作异常: {str(e)}" finally: cursor.close() conn.close() async def batch_add_to_task(self, doc_ids: List[str], username: str, project_name: str, task_tags: Optional[List[str]] = None) -> Tuple[bool, str]: """批量将文档加入标注任务中心 (单表化) Args: doc_ids: 文档ID列表 username: 操作人 project_name: 项目名称 (作为 project_id) task_tags: 标注任务标签列表 (例如 ["标签1", "标签2"]) """ conn = get_db_connection() if not conn: return False, "数据库连接失败" cursor = conn.cursor() try: if not doc_ids: return False, "未指定要加入任务的文档 ID" # 0. 每次点击都生成一个新的项目 UUID (不再检查重名复用) import uuid project_id = str(uuid.uuid4()) # 处理标签：转换为 JSON 字符串存储 import json tag_str = json.dumps(task_tags, ensure_ascii=False) if task_tags else None # 1. 过滤掉未入库的文档 placeholders = ', '.join(['%s'] * len(doc_ids)) check_entered_sql = f"SELECT id FROM t_samp_document_main WHERE id IN ({placeholders}) AND whether_to_enter = 1" cursor.execute(check_entered_sql, doc_ids) entered_ids = [row['id'] for row in cursor.fetchall()] unentered_count = len(doc_ids) - len(entered_ids) if not entered_ids: return False, "所选文档均未入库，无法加入标注任务中心" # 2. 更新 whether_to_task 状态 ids_to_add = entered_ids add_placeholders = ', '.join(['%s'] * len(ids_to_add)) sql = f"UPDATE t_samp_document_main SET whether_to_task = 1, updated_by = %s, updated_time = NOW() WHERE id IN ({add_placeholders})" cursor.execute(sql, (username, *ids_to_add)) # 3. 写入任务管理表 (单表逻辑) for doc_id in ids_to_add: try: # 获取业务元数据 metadata_dict = {} try: # 定义需要过滤掉的非业务/内部状态字段 EXCLUDE_FIELDS = { 'id', 'created_time', 'updated_time', 'created_by', 'updated_by', 'conversion_status', 'whether_to_enter', 'whether_to_task', 'kb_method', 'whether_to_delete' } # 查询主表和子表信息 cursor.execute("SELECT * FROM t_samp_document_main WHERE id = %s", (doc_id,)) doc_main = cursor.fetchone() if doc_main: # 基础元数据 (仅保留标题和来源类型等核心信息) for k, v in doc_main.items(): if v is not None and v != '' and k not in EXCLUDE_FIELDS: metadata_dict[k] = v # 子表元数据 source_type = doc_main.get('source_type') table_name = TABLE_MAP.get(source_type) if table_name: cursor.execute(f"SELECT * FROM {table_name} WHERE id = %s", (doc_id,)) sub_data = cursor.fetchone() if sub_data: for k, v in sub_data.items(): if v is not None and v != '' and k not in EXCLUDE_FIELDS: metadata_dict[k] = v # 递归格式化时间 metadata_dict = task_service._serialize_datetime(metadata_dict) except Exception as meta_err: logger.warning(f"获取文档 {doc_id} 元数据失败: {meta_err}") await task_service.add_task( business_id=doc_id, task_type='data', project_id=project_id, project_name=project_name, tag=tag_str, metadata=json.dumps(metadata_dict, ensure_ascii=False) if metadata_dict else None ) except Exception as e: logger.exception(f"添加文档 {doc_id} 到任务中心失败: {e}") conn.commit() # 4. 自动推送至外部标注平台 push_success, push_msg = await task_service.send_to_external_platform(project_id) msg = f"成功将 {len(ids_to_add)} 份文档加入项目: {project_name}" if push_success: msg += f" (已推送: {push_msg})" else: msg += f" (推送失败: {push_msg})" if unentered_count > 0: msg += f"，{unentered_count} 份文档因未入库被跳过" return True, msg except Exception as e: logger.exception("批量加入任务失败") if conn: conn.rollback() return False, f"操作失败: {str(e)}" finally: if cursor: cursor.close() if conn: conn.close() async def clear_knowledge_base_data(self, doc_ids: List[str], username: str) -> Tuple[int, str]: """批量清空文档在知识库（Milvus）中的数据片段""" conn = get_db_connection() if not conn: return 0, "数据库连接失败" cursor = conn.cursor() success_count = 0 error_details = [] try: # 1. 获取文档的知识库信息 placeholders = ', '.join(['%s'] * len(doc_ids)) sql = f""" SELECT id, title, kb_id, whether_to_enter FROM t_samp_document_main WHERE id IN ({placeholders}) """ cursor.execute(sql, doc_ids) docs = cursor.fetchall() for doc in docs: doc_id = doc['id'] title = doc.get('title', '未命名') kb_id = doc.get('kb_id') # 只有入库了的才需要清空 if doc.get('whether_to_enter') != 1: continue if not kb_id: error_details.append(f"· {title}: 未关联知识库，无法清空") continue # 获取集合名称 cursor.execute("SELECT collection_name_parent, collection_name_children FROM t_samp_knowledge_base WHERE id = %s", (kb_id,)) kb_info = cursor.fetchone() if not kb_info: error_details.append(f"· {title}: 找不到关联的知识库配置") continue # 2. 从 Milvus 删除 try: collections = [kb_info.get('collection_name_parent'), kb_info.get('collection_name_children')] for coll_name in collections: if coll_name and self.milvus_service.client.has_collection(coll_name): # 使用 document_id 进行过滤删除 self.milvus_service.client.delete( collection_name=coll_name, filter=f'document_id == "{doc_id}"' ) # 3. 更新数据库状态 update_sql = "UPDATE t_samp_document_main SET whether_to_enter = 0, updated_by = %s, updated_time = NOW() WHERE id = %s" cursor.execute(update_sql, (username, doc_id)) success_count += 1 except Exception as milvus_err: logger.error(f"清空文档 {title} Milvus 数据失败: {milvus_err}") error_details.append(f"· {title}: 清空向量库数据失败") conn.commit() msg = f"成功清空 {success_count} 份文档的知识库片段" if error_details: msg += "\n部分操作受限：\n" + "\n".join(error_details) return success_count, msg except Exception as e: logger.exception("批量清空知识库数据失败") conn.rollback() return 0, f"操作失败: {str(e)}" finally: cursor.close() conn.close() async def batch_delete_documents(self, doc_ids: List[str]) -> Tuple[int, str]: """批量删除文档 (仅限未入库或已清空的文档)""" conn = get_db_connection() if not conn: return 0, "数据库连接失败" cursor = conn.cursor() try: if not doc_ids: return 0, "未指定要删除的文档 ID" # 1. 检查文档状态：已入库的文档不允许直接删除 placeholders = ', '.join(['%s'] * len(doc_ids)) check_sql = f"SELECT id, title FROM t_samp_document_main WHERE id IN ({placeholders}) AND whether_to_enter = 1" cursor.execute(check_sql, doc_ids) entered_docs = cursor.fetchall() if entered_docs: titles = [d['title'] for d in entered_docs] return 0, f"删除失败：以下文档已入库，请先执行‘清空数据’操作后再删除：\n{', '.join(titles[:5])}{'...' if len(titles)>5 else ''}" # 2. 执行物理删除 # 尝试同步删除子表中的数据 try: for sub_table in TABLE_MAP.values(): sub_sql = f"DELETE FROM {sub_table} WHERE id IN ({placeholders})" try: cursor.execute(sub_sql, doc_ids) except Exception as sub_e: logger.error(f"删除子表 {sub_table} 数据失败: {sub_e}") except Exception as sync_e: logger.error(f"同步删除子表数据失败: {sync_e}") # 删除主表 t_samp_document_main 中的数据 sql_main = f"DELETE FROM t_samp_document_main WHERE id IN ({placeholders})" cursor.execute(sql_main, doc_ids) affected_rows = cursor.rowcount # 同步删除任务管理中心的数据 try: from app.services.task_service import task_service for doc_id in doc_ids: # 注意：task_service 的删除方法可能叫 delete_task_by_business_id 或类似，这里假设存在 # 如果没有，我们需要查出主键 ID 再删 cursor.execute("DELETE FROM t_task_management WHERE business_id = %s", (doc_id,)) except Exception as task_err: logger.error(f"同步删除任务中心数据失败: {task_err}") conn.commit() return affected_rows, f"成功删除 {affected_rows} 条文档数据" except Exception as e: logger.exception("批量删除失败") conn.rollback() return 0, f"批量删除失败: {str(e)}" finally: cursor.close() conn.close() async def get_tag_tree(self) -> List[Dict[str, Any]]: """获取标签层级树 (从 t_samp_tag_category 查询)""" conn = get_db_connection() if not conn: return [] cursor = conn.cursor() try: sql = """ SELECT id, parent_id, name, level, type FROM t_samp_tag_category WHERE is_deleted = 0 AND status = 1 ORDER BY level ASC, sort_no ASC """ cursor.execute(sql) tags = cursor.fetchall() # 构建树形结构 tag_dict = {tag['id']: {**tag, 'children': []} for tag in tags} tree = [] for tag_id, tag_item in tag_dict.items(): parent_id = tag_item['parent_id'] if parent_id == 0: tree.append(tag_item) elif parent_id in tag_dict: tag_dict[parent_id]['children'].append(tag_item) return tree except Exception as e: logger.error(f"获取标签树失败: {e}") return [] finally: cursor.close() conn.close() async def get_document_list( self, whether_to_enter: Optional[int] = None, conversion_status: Optional[int] = None, keyword: Optional[str] = None, table_type: Optional[str] = None, plan_category: Optional[str] = None, level_2_classification: Optional[str] = None, level_3_classification: Optional[str] = None, level_4_classification: Optional[str] = None, page: int = 1, size: int = 50 ) -> Tuple[List[Dict[str, Any]], int, int, int]: """获取文档列表 (支持关联查询子表)""" conn = get_db_connection() if not conn: return [], 0, 0, 0 cursor = conn.cursor() try: where_clauses = [] params = [] # 基础查询 if table_type: where_clauses.append("m.source_type = %s") params.append(table_type) if table_type in TABLE_MAP: # 如果指定了类型且在 TABLE_MAP 中，使用 LEFT JOIN 关联查询，以便搜索子表字段 sub_table = TABLE_MAP[table_type] from_sql = f""" t_samp_document_main m LEFT JOIN {sub_table} s ON m.id = s.id LEFT JOIN t_sys_user u1 ON m.created_by = u1.id LEFT JOIN t_sys_user u2 ON m.updated_by = u2.id LEFT JOIN t_samp_knowledge_base kb ON m.kb_id = kb.id """ fields_sql = "m.*, s.*, u1.username as creator_name, u2.username as updater_name, kb.name as kb_name, m.id as id" # 施工方案特有的过滤字段 if table_type == 'construction_plan': if plan_category: where_clauses.append("s.plan_category = %s") params.append(plan_category) if level_2_classification: where_clauses.append("s.level_2_classification = %s") params.append(SECOND_LEVEL_MAP.get(level_2_classification, level_2_classification)) if level_3_classification: where_clauses.append("s.level_3_classification = %s") params.append(THIRD_LEVEL_MAP.get(level_3_classification, level_3_classification)) if level_4_classification: where_clauses.append("s.level_4_classification = %s") params.append(FOURTH_LEVEL_MAP.get(level_4_classification, level_4_classification)) else: from_sql = "t_samp_document_main m LEFT JOIN t_sys_user u1 ON m.created_by = u1.id LEFT JOIN t_sys_user u2 ON m.updated_by = u2.id LEFT JOIN t_samp_knowledge_base kb ON m.kb_id = kb.id" fields_sql = "m.*, u1.username as creator_name, u2.username as updater_name, kb.name as kb_name" else: from_sql = "t_samp_document_main m LEFT JOIN t_sys_user u1 ON m.created_by = u1.id LEFT JOIN t_sys_user u2 ON m.updated_by = u2.id LEFT JOIN t_samp_knowledge_base kb ON m.kb_id = kb.id" fields_sql = "m.*, u1.username as creator_name, u2.username as updater_name, kb.name as kb_name" order_sql = "m.created_time DESC" title_field = "m.title" # 分离 whether_to_enter 与 conversion_status 的过滤逻辑 if whether_to_enter is not None: where_clauses.append("m.whether_to_enter = %s") params.append(whether_to_enter) if conversion_status is not None: where_clauses.append("m.conversion_status = %s") params.append(conversion_status) if keyword: where_clauses.append(f"{title_field} LIKE %s") params.append(f"%{keyword}%") where_sql = " WHERE " + " AND ".join(where_clauses) if where_clauses else "" offset = (page - 1) * size sql = f"SELECT {fields_sql} FROM {from_sql} {where_sql} ORDER BY {order_sql} LIMIT %s OFFSET %s" params.extend([size, offset]) cursor.execute(sql, tuple(params)) items = [self._format_document_row(row) for row in cursor.fetchall()] # 总数 count_sql = f"SELECT COUNT(*) as count FROM {from_sql} {where_sql}" cursor.execute(count_sql, tuple(params[:-2])) res = cursor.fetchone() total = res['count'] if res else 0 # 统计数据 cursor.execute("SELECT COUNT(*) as count FROM t_samp_document_main") res = cursor.fetchone() all_total = res['count'] if res else 0 cursor.execute("SELECT COUNT(*) as count FROM t_samp_document_main WHERE whether_to_enter = 1") res = cursor.fetchone() total_entered = res['count'] if res else 0 return items, total, all_total, total_entered except Exception as e: logger.exception("获取文档列表失败") return [], 0, 0, 0 finally: cursor.close() conn.close() async def get_document_detail(self, doc_id: str) -> Optional[Dict[str, Any]]: """获取文档详情 (关联查询子表)""" conn = get_db_connection() if not conn: return None cursor = conn.cursor() try: # 1. 查询主表 (关联用户表获取姓名) cursor.execute(""" SELECT m.*, u1.username as creator_name, u2.username as updater_name FROM t_samp_document_main m LEFT JOIN t_sys_user u1 ON m.created_by = u1.id LEFT JOIN t_sys_user u2 ON m.updated_by = u2.id WHERE m.id = %s """, (doc_id,)) doc = cursor.fetchone() if not doc: return None # 2. 根据 source_type 查询对应的子表信息 source_type = doc.get('source_type') table_name = TABLE_MAP.get(source_type) if table_name: # 关联子表的所有字段 sub_sql = f"SELECT * FROM {table_name} WHERE id = %s" cursor.execute(sub_sql, (doc_id,)) sub_data = cursor.fetchone() if sub_data: # 将子表字段合并到 doc 中，方便前端使用 # 注意：如果字段名冲突，子表字段会覆盖主表字段（除了 id） sub_data.pop('id', None) doc.update(sub_data) return self._format_document_row(doc) except Exception as e: logger.exception("获取文档详情失败") return None finally: cursor.close() conn.close() def _to_int(self, value: Any) -> Optional[int]: """安全转换为整数""" if value is None or value == '' or str(value).lower() == 'null': return None try: return int(value) except (ValueError, TypeError): return None def _to_date(self, value: Any) -> Optional[str]: """安全处理日期字符串""" if value is None or value == '' or str(value).lower() == 'null': return None # 如果已经是 datetime.date 或 datetime.datetime 对象 if hasattr(value, 'strftime'): return value.strftime('%Y-%m-%d') return str(value) async def add_document(self, doc_data: Dict[str, Any], user_id: str) -> Tuple[bool, str, Optional[str]]: """添加新文档（先主表后子表，解耦触发器）""" conn = get_db_connection() if not conn: return False, "数据库连接失败", None cursor = conn.cursor() try: doc_id = str(uuid.uuid4()) table_type = doc_data.get('table_type', 'standard') table_name = TABLE_MAP.get(table_type) # 安全转换字段 release_date = self._to_date(doc_data.get('release_date')) # 处理 URL 存储（转为相对路径） file_url = self.minio_manager.get_relative_path(doc_data.get('file_url')) # 处理文档类型 (中文 -> 简写) doc_type_cn = doc_data.get('document_type') doc_type_code = DOCUMENT_TYPE_MAP.get(doc_type_cn, doc_type_cn) # 找不到则保持原样 # 处理专业领域 (中文 -> 简写) prof_field_cn = doc_data.get('professional_field') prof_field_code = PROFESSIONAL_FIELD_MAP.get(prof_field_cn, prof_field_cn) # 处理方案类别 (中文 -> 简写) plan_category_cn = doc_data.get('plan_category') plan_category_code = PLAN_CATEGORY_MAP.get(plan_category_cn, plan_category_cn) # 处理一级分类 (中文 -> 简写) level_1_cn = doc_data.get('level_1_classification') level_1_code = FIRST_LEVEL_MAP.get(level_1_cn, level_1_cn) # 处理二级分类 (中文 -> 简写) level_2_cn = doc_data.get('level_2_classification') level_2_code = SECOND_LEVEL_MAP.get(level_2_cn, level_2_cn) # 处理三级分类 (中文 -> 简写) level_3_cn = doc_data.get('level_3_classification') level_3_code = THIRD_LEVEL_MAP.get(level_3_cn, level_3_cn) # 处理四级分类 (中文 -> 简写) level_4_cn = doc_data.get('level_4_classification') level_4_code = FOURTH_LEVEL_MAP.get(level_4_cn, level_4_cn) # 处理四级分类 (中文 -> 简写) level_4_cn = doc_data.get('level_4_classification') level_4_code = FOURTH_LEVEL_MAP.get(level_4_cn, level_4_cn) # 处理四级分类 (中文 -> 简写) level_4_cn = doc_data.get('level_4_classification') level_4_code = FOURTH_LEVEL_MAP.get(level_4_cn, level_4_cn) # 1. 插入主表 (作为资产中心) cursor.execute( """ INSERT INTO t_samp_document_main ( id, title, source_type, file_url, file_extension, created_by, updated_by, created_time, updated_time, conversion_status, whether_to_task, kb_id ) VALUES (%s, %s, %s, %s, %s, %s, %s, NOW(), NOW(), 0, 0, %s) """, ( doc_id, doc_data.get('title'), table_type, file_url, doc_data.get('file_extension'), user_id, user_id, doc_data.get('kb_id') ) ) # 2. 插入子表 (仅存储业务字段) if table_type == 'standard': cursor.execute( f""" INSERT INTO {table_name} ( id, chinese_name, english_name, standard_number, issuing_authority, release_date, implementation_date, drafting_unit, approving_department, participating_units, document_type, professional_field, engineering_phase, validity, reference_basis, source_url, note, created_by, updated_by, created_time, updated_time ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, NOW(), NOW()) """, ( doc_id, doc_data.get('title'), doc_data.get('english_name'), doc_data.get('standard_no'), doc_data.get('issuing_authority'), release_date, self._to_date(doc_data.get('implementation_date')), doc_data.get('drafting_unit'), doc_data.get('approving_department'), doc_data.get('participating_units'), doc_type_code, prof_field_code, doc_data.get('engineering_phase'), doc_data.get('validity', '现行'), doc_data.get('reference_basis'), doc_data.get('source_url'), doc_data.get('note'), user_id, user_id ) ) elif table_type == 'construction_plan': cursor.execute( f""" INSERT INTO {table_name} ( id, plan_name, project_name, project_section, compiling_unit, compiling_date, plan_summary, compilation_basis, plan_category, level_1_classification, level_2_classification, level_3_classification, level_4_classification, note, created_by, updated_by, created_time, updated_time ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, NOW(), NOW()) """, ( doc_id, doc_data.get('title'), doc_data.get('project_name'), doc_data.get('project_section'), doc_data.get('issuing_authority'), release_date, doc_data.get('plan_summary'), doc_data.get('compilation_basis'), plan_category_code, level_1_code or 'SC', level_2_code, level_3_code, level_4_code, doc_data.get('note'), user_id, user_id ) ) elif table_type == 'regulation': cursor.execute( f""" INSERT INTO {table_name} ( id, file_name, issuing_department, document_type, publish_date, effective_start_date, effective_end_date, note, created_by, updated_by, created_time, updated_time ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, NOW(), NOW()) """, ( doc_id, doc_data.get('title'), doc_data.get('issuing_authority'), doc_type_code, release_date, self._to_date(doc_data.get('effective_start_date')), self._to_date(doc_data.get('effective_end_date')), doc_data.get('note'), user_id, user_id ) ) # 3. 添加到任务管理中心 (类型为 data) try: await task_service.add_task(doc_id, 'data') except Exception as task_err: logger.error(f"添加文档 {doc_data.get('title')} 到任务中心失败: {task_err}") conn.commit() return True, "文档添加成功", doc_id except Exception as e: logger.exception("添加文档失败") conn.rollback() return False, str(e), None finally: cursor.close() conn.close() async def edit_document(self, doc_data: Dict[str, Any], updater_id: str) -> Tuple[bool, str]: """编辑文档（同步主表和子表，解耦触发器）""" conn = get_db_connection() if not conn: return False, "数据库连接失败" cursor = conn.cursor() try: doc_id = doc_data.get('id') table_type = doc_data.get('table_type', 'standard') table_name = TABLE_MAP.get(table_type) # 安全转换字段 release_date = self._to_date(doc_data.get('release_date')) # 处理 URL 存储（转为相对路径） file_url = self.minio_manager.get_relative_path(doc_data.get('file_url')) # 处理文档类型 (中文 -> 简写) doc_type_cn = doc_data.get('document_type') doc_type_code = DOCUMENT_TYPE_MAP.get(doc_type_cn, doc_type_cn) # 找不到则保持原样 # 处理专业领域 (中文 -> 简写) prof_field_cn = doc_data.get('professional_field') prof_field_code = PROFESSIONAL_FIELD_MAP.get(prof_field_cn, prof_field_cn) # 处理方案类别 (中文 -> 简写) plan_category_cn = doc_data.get('plan_category') plan_category_code = PLAN_CATEGORY_MAP.get(plan_category_cn, plan_category_cn) # 处理一级分类 (中文 -> 简写) level_1_cn = doc_data.get('level_1_classification') level_1_code = FIRST_LEVEL_MAP.get(level_1_cn, level_1_cn) # 处理二级分类 (中文 -> 简写) level_2_cn = doc_data.get('level_2_classification') level_2_code = SECOND_LEVEL_MAP.get(level_2_cn, level_2_cn) # 处理三级分类 (中文 -> 简写) level_3_cn = doc_data.get('level_3_classification') level_3_code = THIRD_LEVEL_MAP.get(level_3_cn, level_3_cn) # 处理四级分类 (中文 -> 简写) level_4_cn = doc_data.get('level_4_classification') level_4_code = FOURTH_LEVEL_MAP.get(level_4_cn, level_4_cn) # 1. 更新主表 cursor.execute( """ UPDATE t_samp_document_main SET title = %s, file_url = %s, file_extension = %s, updated_by = %s, updated_time = NOW(), kb_id = %s WHERE id = %s """, ( doc_data.get('title'), file_url, doc_data.get('file_extension'), updater_id, doc_data.get('kb_id'), doc_id ) ) # 2. 更新子表 if table_type == 'standard': cursor.execute( f""" UPDATE {table_name} SET chinese_name = %s, english_name = %s, standard_number = %s, issuing_authority = %s, release_date = %s, implementation_date = %s, drafting_unit = %s, approving_department = %s, participating_units = %s, document_type = %s, professional_field = %s, engineering_phase = %s, validity = %s, reference_basis = %s, source_url = %s, note = %s, updated_by = %s, updated_time = NOW() WHERE id = %s """, ( doc_data.get('title'), doc_data.get('english_name'), doc_data.get('standard_no'), doc_data.get('issuing_authority'), release_date, self._to_date(doc_data.get('implementation_date')), doc_data.get('drafting_unit'), doc_data.get('approving_department'), doc_data.get('participating_units'), doc_type_code, prof_field_code, doc_data.get('engineering_phase'), doc_data.get('validity'), doc_data.get('reference_basis'), doc_data.get('source_url'), doc_data.get('note'), updater_id, doc_id ) ) elif table_type == 'construction_plan': cursor.execute( f""" UPDATE {table_name} SET plan_name = %s, project_name = %s, project_section = %s, compiling_unit = %s, compiling_date = %s, plan_summary = %s, compilation_basis = %s, plan_category = %s, level_1_classification = %s, level_2_classification = %s, level_3_classification = %s, level_4_classification = %s, note = %s, updated_by = %s, updated_time = NOW() WHERE id = %s """, ( doc_data.get('title'), doc_data.get('project_name'), doc_data.get('project_section'), doc_data.get('issuing_authority'), release_date, doc_data.get('plan_summary'), doc_data.get('compilation_basis'), plan_category_code, level_1_code, level_2_code, level_3_code, level_4_code, doc_data.get('note'), updater_id, doc_id ) ) elif table_type == 'regulation': cursor.execute( f""" UPDATE {table_name} SET file_name = %s, issuing_department = %s, document_type = %s, publish_date = %s, effective_start_date = %s, effective_end_date = %s, note = %s, updated_by = %s, updated_time = NOW() WHERE id = %s """, ( doc_data.get('title'), doc_data.get('issuing_authority'), doc_type_code, release_date, self._to_date(doc_data.get('effective_start_date')), self._to_date(doc_data.get('effective_end_date')), doc_data.get('note'), updater_id, doc_id ) ) conn.commit() return True, "文档更新成功" except Exception as e: logger.exception("编辑文档失败") conn.rollback() return False, str(e) finally: cursor.close() conn.close() async def enter_document(self, doc_id: str, username: str) -> Tuple[bool, str]: """文档入库（单个）""" affected_rows, message = await self.batch_enter_knowledge_base([doc_id], username) return affected_rows > 0, message async def get_basic_info_list( self, type: str, page: int, size: int, keyword: Optional[str] = None, **filters ) -> Tuple[List[Dict[str, Any]], int]: """获取基本信息列表（关联主表获取文件和转换状态）""" conn = get_db_connection() if not conn: return [], 0 cursor = conn.cursor() try: # 根据类型选择表名和字段映射 if type == 'standard': table_name = "t_samp_standard_base_info" # 关联主表字段：file_url, conversion_status, md_url, json_url fields = """ s.id, s.chinese_name as title, s.standard_number as standard_no, s.issuing_authority, s.release_date, s.document_type, s.professional_field, s.validity, s.note, s.participating_units, s.reference_basis, s.created_by, u1.username as creator_name, s.created_time, s.updated_by, u2.username as updater_name, s.updated_time, m.file_url, m.conversion_status, m.md_url, m.json_url, m.kb_id, m.whether_to_enter """ field_map = { 'title': 's.chinese_name', 'standard_no': 's.standard_number', 'issuing_authority': 's.issuing_authority', 'release_date': 's.release_date', 'document_type': 's.document_type', 'professional_field': 's.professional_field', 'validity': 's.validity' } elif type == 'construction_plan': table_name = "t_samp_construction_plan_base_info" fields = """ s.id, s.plan_name as title, NULL as standard_no, s.project_name, s.project_section, s.compiling_unit as issuing_authority, s.compiling_date as release_date, NULL as document_type, NULL as professional_field, NULL as validity, s.plan_summary, s.compilation_basis, s.plan_category, s.level_1_classification, s.level_2_classification, s.level_3_classification, s.level_4_classification, s.note, s.created_by, u1.username as creator_name, s.created_time, s.updated_by, u2.username as updater_name, s.updated_time, m.file_url, m.conversion_status, m.md_url, m.json_url, m.kb_id, m.whether_to_enter """ field_map = { 'title': 's.plan_name', 'issuing_authority': 's.compiling_unit', 'release_date': 's.compiling_date', 'plan_category': 's.plan_category', 'level_1_classification': 's.level_1_classification', 'level_2_classification': 's.level_2_classification', 'level_3_classification': 's.level_3_classification', 'level_4_classification': 's.level_4_classification' } elif type == 'regulation': table_name = "t_samp_office_regulations" fields = """ s.id, s.file_name as title, NULL as standard_no, s.issuing_department as issuing_authority, s.publish_date as release_date, s.document_type, NULL as professional_field, NULL as validity, s.note, s.created_by, u1.username as creator_name, s.created_time, s.updated_by, u2.username as updater_name, s.updated_time, m.file_url, m.conversion_status, m.md_url, m.json_url, m.kb_id, m.whether_to_enter """ field_map = { 'title': 's.file_name', 'issuing_authority': 's.issuing_department', 'release_date': 's.publish_date', 'document_type': 's.document_type' } else: return [], 0 where_clauses = [] params = [] # 统一关键字搜索 if keyword: if type == 'standard': where_clauses.append("(s.chinese_name LIKE %s OR s.standard_number LIKE %s)") params.extend([f"%{keyword}%", f"%{keyword}%"]) elif type == 'construction_plan': where_clauses.append("s.plan_name LIKE %s") params.append(f"%{keyword}%") field_map['issuing_authority'] = 's.compiling_unit' field_map['release_date'] = 's.compiling_date' elif type == 'regulation': where_clauses.append("s.file_name LIKE %s") params.append(f"%{keyword}%") field_map['issuing_authority'] = 's.issuing_department' field_map['release_date'] = 's.publish_date' # 精细化检索 for filter_key, filter_value in filters.items(): if not filter_value: continue # 特殊处理日期范围 date_field = field_map.get('release_date', 's.release_date') if filter_key == 'release_date_start': where_clauses.append(f"{date_field} >= %s") params.append(filter_value) continue if filter_key == 'release_date_end': where_clauses.append(f"{date_field} <= %s") params.append(filter_value) continue db_field = field_map.get(filter_key) if db_field: # 处理文档类型查询 (前端传中文 -> 数据库查简写) if filter_key == 'document_type': filter_value = DOCUMENT_TYPE_MAP.get(filter_value, filter_value) # 处理专业领域查询 (前端传中文 -> 数据库查简写) if filter_key == 'professional_field': filter_value = PROFESSIONAL_FIELD_MAP.get(filter_value, filter_value) # 处理方案类别查询 (前端传中文 -> 数据库查简写) if filter_key == 'plan_category': filter_value = PLAN_CATEGORY_MAP.get(filter_value, filter_value) # 处理一级分类查询 (前端传中文 -> 数据库查简写) if filter_key == 'level_1_classification': filter_value = FIRST_LEVEL_MAP.get(filter_value, filter_value) # 处理二级分类查询 (前端传中文 -> 数据库查简写) if filter_key == 'level_2_classification': filter_value = SECOND_LEVEL_MAP.get(filter_value, filter_value) # 处理三级分类查询 (前端传中文 -> 数据库查简写) if filter_key == 'level_3_classification': filter_value = THIRD_LEVEL_MAP.get(filter_value, filter_value) # 处理四级分类查询 (前端传中文 -> 数据库查简写) if filter_key == 'level_4_classification': filter_value = FOURTH_LEVEL_MAP.get(filter_value, filter_value) # 如果是 title, standard_no, issuing_authority，支持模糊查询 if filter_key in ['title', 'standard_no', 'issuing_authority']: where_clauses.append(f"{db_field} LIKE %s") params.append(f"%{filter_value}%") else: where_clauses.append(f"{db_field} = %s") params.append(filter_value) where_sql = " WHERE " + " AND ".join(where_clauses) if where_clauses else "" offset = (page - 1) * size # 使用 LEFT JOIN 关联主表和用户表获取姓名 sql = f""" SELECT {fields}, kb.name as kb_name FROM {table_name} s LEFT JOIN t_samp_document_main m ON s.id = m.id LEFT JOIN t_sys_user u1 ON s.created_by = u1.id LEFT JOIN t_sys_user u2 ON s.updated_by = u2.id LEFT JOIN t_samp_knowledge_base kb ON m.kb_id = kb.id {where_sql} ORDER BY s.created_time DESC LIMIT %s OFFSET %s """ params = params + [size, offset] cursor.execute(sql, tuple(params)) items = cursor.fetchall() # 处理 URL 转换 for item in items: # 处理文档类型显示 (简写 -> 中文) doc_type_code = item.get('document_type') if doc_type_code in DOCUMENT_TYPE_REVERSE_MAP: item['document_type'] = DOCUMENT_TYPE_REVERSE_MAP[doc_type_code] # 处理专业领域显示 (简写 -> 中文) prof_field_code = item.get('professional_field') if prof_field_code in PROFESSIONAL_FIELD_REVERSE_MAP: item['professional_field'] = PROFESSIONAL_FIELD_REVERSE_MAP[prof_field_code] # 处理方案类别显示 (简写 -> 中文) plan_category_code = item.get('plan_category') if plan_category_code in PLAN_CATEGORY_REVERSE_MAP: item['plan_category'] = PLAN_CATEGORY_REVERSE_MAP[plan_category_code] # 处理一级分类显示 (简写 -> 中文) level_1_code = item.get('level_1_classification') if level_1_code in FIRST_LEVEL_REVERSE_MAP: item['level_1_classification'] = FIRST_LEVEL_REVERSE_MAP[level_1_code] # 处理二级分类显示 (简写 -> 中文) level_2_code = item.get('level_2_classification') if level_2_code in SECOND_LEVEL_REVERSE_MAP: item['level_2_classification'] = SECOND_LEVEL_REVERSE_MAP[level_2_code] # 处理三级分类显示 (简写 -> 中文) level_3_code = item.get('level_3_classification') if level_3_code in THIRD_LEVEL_REVERSE_MAP: item['level_3_classification'] = THIRD_LEVEL_REVERSE_MAP[level_3_code] # 处理四级分类显示 (简写 -> 中文) level_4_code = item.get('level_4_classification') if level_4_code in FOURTH_LEVEL_REVERSE_MAP: item['level_4_classification'] = FOURTH_LEVEL_REVERSE_MAP[level_4_code] for key in ['file_url', 'md_url', 'json_url']: if item.get(key): item[key] = self.minio_manager.get_full_url(item[key]) # 总数 count_sql = f"SELECT COUNT(*) as count FROM {table_name} s {where_sql}" cursor.execute(count_sql, tuple(params[:-2])) res = cursor.fetchone() total = res['count'] if res else 0 return items, total except Exception as e: logger.exception(f"获取 {type} 列表失败") return [], 0 finally: cursor.close() conn.close() # ==================== 文档转换 ==================== async def get_document_source_type(self, doc_id: str) -> Optional[str]: """获取文档的source_type""" conn = get_db_connection() if not conn: return None cursor = conn.cursor() try: cursor.execute("SELECT source_type FROM t_samp_document_main WHERE id = %s", (doc_id,)) res = cursor.fetchone() return res['source_type'] if res else None except Exception as e: logger.exception(f"获取文档source_type失败: {e}") return None finally: cursor.close() conn.close() async def get_document_title(self, doc_id: str) -> str: """获取文档标题""" conn = get_db_connection() if not conn: return "文档" cursor = conn.cursor() try: cursor.execute("SELECT title FROM t_samp_document_main WHERE id = %s", (doc_id,)) res = cursor.fetchone() return res['title'] if res else "文档" except Exception as e: logger.exception(f"获取文档标题失败: {e}") return "文档" finally: cursor.close() conn.close() async def update_conversion_status(self, doc_id: str, status: int, md_url: Optional[str] = None, json_url: Optional[str] = None, error_message: Optional[str] = None) -> bool: """更新文档转换状态 Args: doc_id: 文档ID status: 转换状态 (0=未转换, 1=转换中, 2=已完成, 3=失败) md_url: Markdown文件URL json_url: JSON文件URL error_message: 错误信息 """ conn = get_db_connection() if not conn: return False cursor = conn.cursor() try: update_clauses = ["conversion_status = %s"] params = [status] if error_message: update_clauses.append("conversion_error = %s") params.append(error_message) if md_url: update_clauses.append("md_url = %s") params.append(md_url) if json_url: update_clauses.append("json_url = %s") params.append(json_url) sql = f"UPDATE t_samp_document_main SET {', '.join(update_clauses)}, updated_time = NOW() WHERE id = %s" params.append(doc_id) cursor.execute(sql, tuple(params)) conn.commit() return True except Exception as e: logger.exception(f"更新转换进度失败: {e}") conn.rollback() return False finally: cursor.close() conn.close() # ==================== 基础信息管理 ==================== async def add_basic_info(self, type: str, data: Dict[str, Any], user_id: str) -> Tuple[bool, str, Optional[str]]: """新增基本信息""" logger.info(f"Adding basic info for type {type}: {data}") conn = get_db_connection() if not conn: return False, "数据库连接失败", None cursor = conn.cursor() try: table_name = TABLE_MAP.get(type) if not table_name: return False, "无效的类型", None doc_id = str(uuid.uuid4()) file_url = data.get('file_url') file_extension = file_url.split('.')[-1] if file_url and '.' in file_url else None # 处理文档类型 (中文 -> 简写) doc_type_cn = data.get('document_type') doc_type_code = DOCUMENT_TYPE_MAP.get(doc_type_cn, doc_type_cn) # 处理专业领域 (中文 -> 简写) prof_field_cn = data.get('professional_field') prof_field_code = PROFESSIONAL_FIELD_MAP.get(prof_field_cn, prof_field_cn) # 处理方案类别 (中文 -> 简写) plan_category_cn = data.get('plan_category') plan_category_code = PLAN_CATEGORY_MAP.get(plan_category_cn, plan_category_cn) # 处理一级分类 (中文 -> 简写) level_1_cn = data.get('level_1_classification') level_1_code = FIRST_LEVEL_MAP.get(level_1_cn, level_1_cn) # 处理二级分类 (中文 -> 简写) level_2_cn = data.get('level_2_classification') level_2_code = SECOND_LEVEL_MAP.get(level_2_cn, level_2_cn) # 处理三级分类 (中文 -> 简写) level_3_cn = data.get('level_3_classification') level_3_code = THIRD_LEVEL_MAP.get(level_3_cn, level_3_cn) # 处理四级分类 (中文 -> 简写) level_4_cn = data.get('level_4_classification') level_4_code = FOURTH_LEVEL_MAP.get(level_4_cn, level_4_cn) # 处理四级分类 (中文 -> 简写) level_4_cn = data.get('level_4_classification') level_4_code = FOURTH_LEVEL_MAP.get(level_4_cn, level_4_cn) # 1. 插入主表 (解耦触发器，手动同步) cursor.execute( """ INSERT INTO t_samp_document_main ( id, title, source_type, file_url, file_extension, created_by, updated_by, created_time, updated_time, conversion_status, whether_to_task, kb_id ) VALUES (%s, %s, %s, %s, %s, %s, %s, NOW(), NOW(), 0, 0, %s) """, ( doc_id, data.get('title'), type, file_url, file_extension, user_id, user_id, data.get('kb_id') ) ) # 2. 插入子表 (移除 file_url，因为它现在只存储 in 主表中) if type == 'standard': sql = f""" INSERT INTO {table_name} ( id, chinese_name, english_name, standard_number, issuing_authority, release_date, implementation_date, drafting_unit, approving_department, participating_units, document_type, professional_field, engineering_phase, validity, reference_basis, source_url, note, created_by, updated_by, created_time, updated_time ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, NOW(), NOW()) """ params = ( doc_id, data.get('title'), data.get('english_name'), data.get('standard_no'), data.get('issuing_authority'), self._to_date(data.get('release_date')), self._to_date(data.get('implementation_date')), data.get('drafting_unit'), data.get('approving_department'), data.get('participating_units'), doc_type_code, prof_field_code, data.get('engineering_phase'), data.get('validity', '现行'), data.get('reference_basis'), data.get('source_url'), data.get('note'), user_id, user_id ) elif type == 'construction_plan': sql = f""" INSERT INTO {table_name} ( id, plan_name, project_name, project_section, compiling_unit, compiling_date, plan_summary, compilation_basis, plan_category, level_1_classification, level_2_classification, level_3_classification, level_4_classification, note, created_by, updated_by, created_time, updated_time ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, NOW(), NOW()) """ params = ( doc_id, data.get('title'), data.get('project_name'), data.get('project_section'), data.get('issuing_authority'), self._to_date(data.get('release_date')), data.get('plan_summary'), data.get('compilation_basis'), plan_category_code, level_1_code or 'SC', level_2_code, level_3_code, level_4_code, data.get('note'), user_id, user_id ) elif type == 'regulation': sql = f""" INSERT INTO {table_name} ( id, file_name, issuing_department, document_type, publish_date, effective_start_date, effective_end_date, note, created_by, updated_by, created_time, updated_time ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, NOW(), NOW()) """ params = ( doc_id, data.get('title'), data.get('issuing_authority'), doc_type_code, self._to_date(data.get('release_date')), self._to_date(data.get('effective_start_date')), self._to_date(data.get('effective_end_date')), data.get('note'), user_id, user_id ) else: return False, "不支持的类型", None cursor.execute(sql, params) # 3. 添加到任务管理中心 (类型为 data) try: # 尝试调用异步方法，如果报错则记录日志 import asyncio try: # 检查是否有正在运行的事件循环 loop = asyncio.get_event_loop() if loop.is_running(): # 在运行的循环中创建任务 loop.create_task(task_service.add_task(doc_id, 'data')) else: # 否则使用 run 运行（不推荐在 web 环境下这样做，但这里是兜底） loop.run_until_complete(task_service.add_task(doc_id, 'data')) except RuntimeError: # 没有事件循环时 asyncio.run(task_service.add_task(doc_id, 'data')) except Exception as task_err: logger.error(f"添加基本信息 {data.get('title')} 到任务中心失败: {task_err}") conn.commit() return True, "新增成功", doc_id except Exception as e: logger.exception("新增基本信息失败") conn.rollback() return False, str(e), None finally: cursor.close() conn.close() async def edit_basic_info(self, type: str, doc_id: str, data: Dict[str, Any], updater_id: str) -> Tuple[bool, str]: """编辑基本信息""" logger.info(f"Editing basic info for type {type}, id {doc_id}: {data}") conn = get_db_connection() if not conn: return False, "数据库连接失败" cursor = conn.cursor() try: table_name = TABLE_MAP.get(type) if not table_name: return False, "无效的类型" # 处理 URL 存储（转为相对路径） file_url = self.minio_manager.get_relative_path(data.get('file_url')) file_extension = file_url.split('.')[-1] if file_url and '.' in file_url else None # 处理文档类型 (中文 -> 简写) doc_type_cn = data.get('document_type') doc_type_code = DOCUMENT_TYPE_MAP.get(doc_type_cn, doc_type_cn) # 处理专业领域 (中文 -> 简写) prof_field_cn = data.get('professional_field') prof_field_code = PROFESSIONAL_FIELD_MAP.get(prof_field_cn, prof_field_cn) # 处理方案类别 (中文 -> 简写) plan_category_cn = data.get('plan_category') plan_category_code = PLAN_CATEGORY_MAP.get(plan_category_cn, plan_category_cn) # 处理一级分类 (中文 -> 简写) level_1_cn = data.get('level_1_classification') level_1_code = FIRST_LEVEL_MAP.get(level_1_cn, level_1_cn) # 处理二级分类 (中文 -> 简写) level_2_cn = data.get('level_2_classification') level_2_code = SECOND_LEVEL_MAP.get(level_2_cn, level_2_cn) # 处理三级分类 (中文 -> 简写) level_3_cn = data.get('level_3_classification') level_3_code = THIRD_LEVEL_MAP.get(level_3_cn, level_3_cn) # 处理四级分类 (中文 -> 简写) level_4_cn = data.get('level_4_classification') level_4_code = FOURTH_LEVEL_MAP.get(level_4_cn, level_4_cn) # 1. 更新主表 (解耦触发器) cursor.execute( """ UPDATE t_samp_document_main SET title = %s, file_url = %s, file_extension = %s, updated_by = %s, updated_time = NOW(), kb_id = %s WHERE id = %s """, (data.get('title'), file_url, file_extension, updater_id, data.get('kb_id'), doc_id) ) # 2. 更新子表 (移除 file_url) if type == 'standard': sql = f""" UPDATE {table_name} SET chinese_name = %s, standard_number = %s, issuing_authority = %s, release_date = %s, document_type = %s, professional_field = %s, validity = %s, english_name = %s, implementation_date = %s, drafting_unit = %s, approving_department = %s, engineering_phase = %s, participating_units = %s, reference_basis = %s, source_url = %s, note = %s, updated_by = %s, updated_time = NOW() WHERE id = %s """ params = ( data.get('title'), data.get('standard_no'), data.get('issuing_authority'), self._to_date(data.get('release_date')), doc_type_code, prof_field_code, data.get('validity'), data.get('english_name'), self._to_date(data.get('implementation_date')), data.get('drafting_unit'), data.get('approving_department'), data.get('engineering_phase'), data.get('participating_units'), data.get('reference_basis'), data.get('source_url'), data.get('note'), updater_id, doc_id ) elif type == 'construction_plan': sql = f""" UPDATE {table_name} SET plan_name = %s, project_name = %s, project_section = %s, compiling_unit = %s, compiling_date = %s, plan_summary = %s, compilation_basis = %s, plan_category = %s, level_1_classification = %s, level_2_classification = %s, level_3_classification = %s, level_4_classification = %s, note = %s, updated_by = %s, updated_time = NOW() WHERE id = %s """ params = ( data.get('title'), data.get('project_name'), data.get('project_section'), data.get('issuing_authority'), self._to_date(data.get('release_date')), data.get('plan_summary'), data.get('compilation_basis'), plan_category_code, level_1_code, level_2_code, level_3_code, level_4_code, data.get('note'), updater_id, doc_id ) elif type == 'regulation': sql = f""" UPDATE {table_name} SET file_name = %s, issuing_department = %s, document_type = %s, publish_date = %s, effective_start_date = %s, effective_end_date = %s, note = %s, updated_by = %s, updated_time = NOW() WHERE id = %s """ params = ( data.get('title'), data.get('issuing_authority'), doc_type_code, self._to_date(data.get('release_date')), self._to_date(data.get('effective_start_date')), self._to_date(data.get('effective_end_date')), data.get('note'), updater_id, doc_id ) else: return False, "不支持的类型" cursor.execute(sql, params) conn.commit() return True, "编辑成功" except Exception as e: logger.exception("编辑基本信息失败") conn.rollback() return False, str(e) finally: cursor.close() conn.close() async def delete_basic_info(self, type: str, doc_id: str) -> Tuple[bool, str]: """删除基本信息""" if not doc_id: return False, "缺少 ID 参数" logger.info(f"Deleting basic info: type={type}, id={doc_id}") conn = get_db_connection() if not conn: return False, "数据库连接失败" cursor = conn.cursor() try: table_name = TABLE_MAP.get(type) if not table_name: return False, "无效的类型" # 1. 显式删除子表记录 (防止 CASCADE 未生效) try: cursor.execute(f"DELETE FROM {table_name} WHERE id = %s", (doc_id,)) logger.info(f"Deleted from sub-table {table_name}, affected: {cursor.rowcount}") except Exception as sub_e: logger.warning(f"删除子表 {table_name} 记录失败 (可能不存在): {sub_e}") # 2. 同步删除任务管理中心的数据 (优先删除关联数据) try: # 使用当前事务删除任务记录（如果 task_service 支持的话，目前它自建连接） # 这里我们直接在当前 cursor 中也执行一次，确保事务一致性 cursor.execute("DELETE FROM t_task_management WHERE business_id = %s", (doc_id,)) logger.info(f"Deleted from t_task_management, affected: {cursor.rowcount}") except Exception as task_e: logger.warning(f"在主事务中删除任务记录失败: {task_e}") # 3. 删除主表记录 cursor.execute("DELETE FROM t_samp_document_main WHERE id = %s", (doc_id,)) affected_main = cursor.rowcount logger.info(f"Deleted from t_samp_document_main, affected: {affected_main}") if affected_main == 0: logger.warning(f"未找到主表记录: {doc_id}") # 即使主表没找到，我们也 commit 之前的操作并返回成功（幂等性） conn.commit() # 4. 再次确保任务中心数据已删除 (调用原有服务) try: await task_service.delete_task(doc_id) except Exception as task_err: logger.error(f"调用 task_service 删除任务失败: {task_err}") return True, "删除成功" except Exception as e: logger.exception(f"删除基本信息异常 (ID: {doc_id})") conn.rollback() return False, f"删除失败: {str(e)}" finally: cursor.close() conn.close()