import os import sys import json import logging import shutil from pathlib import Path import faiss import numpy as np # 导入自定义模块 from graph_builder import build_rag_database_interactive from feature_retriever import FeatureRetriever from static_analyzer import run_static_analysis from issue_filter import run_issue_filtering # ====================== # 导入原有 config.py 配置 # ====================== from config import ( QWEN_API_KEY, QWEN_API_URL, PROJECT_ROOT, VECTOR_DB_PATH, METADATA_PATH, GRAPH_DATA_PATH, KNOWLEDGE_GRAPH_PATH, Default_PROJECT_ROOT, Default_VECTOR_DB_PATH, Default_METADATA_PATH, Default_GRAPH_DATA_PATH, Default_KNOWLEDGE_GRAPH_PATH, MAX_CODE_LENGTH, CPP_EXTENSIONS, IGNORE_DIRS, TOP_K, MAX_HOPS, MIN_SIMILARITY_THRESHOLD ) USER_PROJECT_ROOT = PROJECT_ROOT USER_VECTOR_DB_PATH = VECTOR_DB_PATH USER_METADATA_PATH = METADATA_PATH USER_GRAPH_DATA_PATH = GRAPH_DATA_PATH USER_KNOWLEDGE_GRAPH_PATH = KNOWLEDGE_GRAPH_PATH INTERNAL_DEFAULT_PROJECT_ROOT = Default_PROJECT_ROOT INTERNAL_DEFAULT_VECTOR_DB_PATH = Default_VECTOR_DB_PATH INTERNAL_DEFAULT_METADATA_PATH = Default_METADATA_PATH INTERNAL_DEFAULT_GRAPH_DATA_PATH = Default_GRAPH_DATA_PATH INTERNAL_DEFAULT_KNOWLEDGE_GRAPH_PATH = Default_KNOWLEDGE_GRAPH_PATH # 配置日志 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler("satellite_rag_system.log"), logging.StreamHandler() ] ) logger = logging.getLogger(__name__) # 保存用户运行时配置的文件名 RUNTIME_CONFIG_FILE = "user_runtime_config.json" def load_runtime_config(): """加载用户运行时保存的配置""" if os.path.exists(RUNTIME_CONFIG_FILE): try: with open(RUNTIME_CONFIG_FILE, 'r', encoding='utf-8') as f: return json.load(f) except Exception as e: logger.warning(f"加载运行时配置文件失败: {e}") return {} def save_runtime_config(config): """保存用户运行时配置""" try: with open(RUNTIME_CONFIG_FILE, 'w', encoding='utf-8') as f: json.dump(config, f, indent=2, ensure_ascii=False) logger.info(f"运行时配置已保存至: {RUNTIME_CONFIG_FILE}") return True except Exception as e: logger.error(f"保存运行时配置失败: {e}") return False def get_final_path(config_key, user_input, config_user_path, internal_default_path, runtime_config): """ 统一的路径解析函数 优先级: 1.用户本次输入 > 2.已保存的运行时配置 > 3.config.py中的主路径变量 > 4.config.py中的Default_*后备路径 """ # 优先级1: 用户本次运行的输入 if user_input and str(user_input).strip(): final_path_str = str(user_input).strip() # 立即更新运行时配置 runtime_config[config_key] = final_path_str final_path = Path(final_path_str) # 优先级2: 之前保存的运行时配置 elif config_key in runtime_config and runtime_config[config_key]: final_path_str = runtime_config[config_key] final_path = Path(final_path_str) # 优先级3: config.py中预设的主路径变量 (映射为USER_*) elif config_user_path and str(config_user_path).strip(): final_path = Path(config_user_path) # 优先级4: config.py中预设的Default_*后备路径 (映射为INTERNAL_DEFAULT_*) else: final_path = Path(internal_default_path) # 如果内部默认路径是相对路径，尝试基于项目根目录解析 if not final_path.is_absolute() and config_key != 'PROJECT_ROOT': project_root = get_final_path( 'PROJECT_ROOT', '', USER_PROJECT_ROOT, INTERNAL_DEFAULT_PROJECT_ROOT, runtime_config ) # 注意：避免创建不存在的路径，仅在 project_root 存在时拼接 if project_root.exists(): final_path = project_root / final_path else: # 如果 project_root 也不存在，回退到当前工作目录 final_path = Path.cwd() / final_path # 关键修复：确保不返回空路径 if not str(final_path).strip(): # 如果最终路径仍然是空的，使用当前目录的默认文件名 default_filenames = { 'VECTOR_DB_PATH': 'satellite_rag.faiss', 'METADATA_PATH': 'satellite_rag_metadata.json', 'GRAPH_DATA_PATH': 'code_knowledge_graph.json', 'PROJECT_ROOT': '.' } if config_key in default_filenames: final_path = Path(default_filenames[config_key]) else: final_path = Path('.') # 回退到当前目录 # 尝试解析为绝对路径，并规范化 try: return final_path.resolve() except: # 如果路径无效（如包含不存在的父目录），返回原始Path对象 return final_path def ensure_directory_exists(file_path): """确保文件所在的目录存在""" directory = os.path.dirname(file_path) if directory and not os.path.exists(directory): os.makedirs(directory, exist_ok=True) logger.info(f"创建目录: {directory}") return file_path def check_api_key(): """检查API Key配置""" print("=" * 60) print("检查API Key配置") print("=" * 60) if not QWEN_API_KEY or QWEN_API_KEY == "sk-your-api-key-here": print("⚠ 警告: API Key未配置或为默认值") print(f"当前API Key: {QWEN_API_KEY[:20]}..." if QWEN_API_KEY else "空") print(f"API URL: {QWEN_API_URL}") response = input("是否继续? (y/n): ").strip().lower() if response != 'y': print("请先配置 config.py 中的 QWEN_API_KEY 字段") sys.exit(1) else: print(f"✓ API Key已配置: {QWEN_API_KEY[:20]}...") print(f"✓ API URL: {QWEN_API_URL}") return True def check_knowledge_base_exists(runtime_config): """检查知识库是否已建设""" print("\n检查知识库文件...") # 使用统一的路径解析函数获取最终路径 vector_db_path = get_final_path( 'VECTOR_DB_PATH', '', USER_VECTOR_DB_PATH, INTERNAL_DEFAULT_VECTOR_DB_PATH, runtime_config ) metadata_path = get_final_path( 'METADATA_PATH', '', USER_METADATA_PATH, INTERNAL_DEFAULT_METADATA_PATH, runtime_config ) required_files = [vector_db_path, metadata_path] existing_files = [] missing_files = [] for file_path in required_files: if os.path.exists(file_path): existing_files.append(file_path) else: missing_files.append(file_path) if existing_files: print(f"✓ 找到 {len(existing_files)} 个知识库文件:") for file in existing_files: print(f" - {file}") if missing_files: print(f"⚠ 缺失 {len(missing_files)} 个知识库文件:") for file in missing_files: print(f" - {file}") return len(missing_files) == 0, existing_files, missing_files, vector_db_path, metadata_path def manual_configure_knowledge_base(runtime_config): """手动配置知识库文件地址""" print("\n" + "=" * 60) print("手动配置知识库文件") print("=" * 60) config_changes = {} # 获取当前默认值用于显示 current_vector_db = get_final_path( 'VECTOR_DB_PATH', '', USER_VECTOR_DB_PATH, INTERNAL_DEFAULT_VECTOR_DB_PATH, runtime_config ) current_metadata = get_final_path( 'METADATA_PATH', '', USER_METADATA_PATH, INTERNAL_DEFAULT_METADATA_PATH, runtime_config ) # 询问向量数据库路径 vector_input = input(f"请输入FAISS向量数据库文件路径 (当前: {current_vector_db}): ").strip() if vector_input: vector_path = Path(vector_input) config_changes['VECTOR_DB_PATH'] = str(vector_path) else: vector_path = current_vector_db # 询问元数据文件路径 metadata_input = input(f"请输入元数据文件路径 (当前: {current_metadata}): ").strip() if metadata_input: metadata_path = Path(metadata_input) config_changes['METADATA_PATH'] = str(metadata_path) else: metadata_path = current_metadata # 验证文件 print(f"\n验证配置的文件...") all_valid = True try: if os.path.exists(vector_path): index = faiss.read_index(str(vector_path)) dimension = index.d print(f"✓ FAISS索引: {vector_path} (维度: {dimension})") else: print(f"⚠ 文件不存在: {vector_path}") all_valid = False except Exception as e: print(f"⚠ 加载FAISS索引失败: {e}") all_valid = False try: if os.path.exists(metadata_path): with open(metadata_path, 'r', encoding='utf-8') as f: metadata = json.load(f) print(f"✓ 元数据文件: {metadata_path} (包含 {len(metadata)} 个函数)") else: print(f"⚠ 文件不存在: {metadata_path}") all_valid = False except Exception as e: print(f"⚠ 加载元数据失败: {e}") all_valid = False if config_changes and all_valid: # 更新运行时配置 runtime_config.update(config_changes) save_config = input("\n是否保存此配置供下次使用? (y/n): ").strip().lower() if save_config == 'y': if save_runtime_config(runtime_config): print(f"✓ 配置已保存到运行时文件: {RUNTIME_CONFIG_FILE}") return True elif not all_valid: print("\n配置验证失败，请检查文件路径。") return False else: print("\n使用现有配置，无更改。") return True def load_knowledge_base(runtime_config): """加载知识库文件""" print("\n" + "=" * 60) print("加载知识库") print("=" * 60) # 检查知识库 kb_exists, existing_files, missing_files, vector_db_path, metadata_path = check_knowledge_base_exists( runtime_config) if kb_exists: print("\n✓ 使用现有知识库配置") # 验证路径是否为空 if not str(vector_db_path).strip(): print("⚠ 错误: 向量数据库路径为空！") print("正在尝试使用默认值...") vector_db_path = Path("satellite_rag.faiss") if not str(metadata_path).strip(): print("⚠ 错误: 元数据文件路径为空！") print("正在尝试使用默认值...") metadata_path = Path("satellite_rag_metadata.json") return { 'vector_db_path': str(vector_db_path), 'metadata_path': str(metadata_path), 'graph_data_path': str(get_final_path( 'GRAPH_DATA_PATH', '', USER_GRAPH_DATA_PATH, INTERNAL_DEFAULT_GRAPH_DATA_PATH, runtime_config )) } else: print("\n知识库文件不完整，需要配置。") if manual_configure_knowledge_base(runtime_config): # 重新检查 kb_exists, existing_files, missing_files, vector_db_path, metadata_path = check_knowledge_base_exists( runtime_config) if kb_exists: return { 'vector_db_path': str(vector_db_path), 'metadata_path': str(metadata_path), 'graph_data_path': str(get_final_path( 'GRAPH_DATA_PATH', '', USER_GRAPH_DATA_PATH, INTERNAL_DEFAULT_GRAPH_DATA_PATH, runtime_config )) } return None def feature_retrieval_mode(runtime_config): """功能需求检索模式（主动引导路径输入版）""" print("\n" + "=" * 60) print("功能需求检索分析") print("=" * 60) # 第一步：主动引导用户配置或确认知识库路径 print("请配置知识库路径（直接按Enter键将使用括号内显示的默认路径）：") # 获取当前根据优先级计算出的默认路径（用于提示用户） default_vector_path = get_final_path( 'VECTOR_DB_PATH', '', USER_VECTOR_DB_PATH, INTERNAL_DEFAULT_VECTOR_DB_PATH, runtime_config ) default_metadata_path = get_final_path( 'METADATA_PATH', '', USER_METADATA_PATH, INTERNAL_DEFAULT_METADATA_PATH, runtime_config ) # 引导输入向量数据库路径 vector_prompt = f"请输入FAISS向量数据库文件路径 (默认: {default_vector_path}): " vector_input = input(vector_prompt).strip() # 引导输入元数据文件路径 metadata_prompt = f"请输入元数据文件路径 (默认: {default_metadata_path}): " metadata_input = input(metadata_prompt).strip() # 使用用户输入（或空输入）重新计算最终路径 # 注意：这里将用户输入传入 get_final_path，它会自动更新 runtime_config final_vector_path = get_final_path( 'VECTOR_DB_PATH', vector_input, USER_VECTOR_DB_PATH, INTERNAL_DEFAULT_VECTOR_DB_PATH, runtime_config ) final_metadata_path = get_final_path( 'METADATA_PATH', metadata_input, USER_METADATA_PATH, INTERNAL_DEFAULT_METADATA_PATH, runtime_config ) # 检查文件是否存在 print(f"\n检查知识库文件...") files_exist = True if not os.path.exists(final_vector_path): print(f"⚠ 向量数据库文件不存在: {final_vector_path}") files_exist = False else: print(f"✓ 向量数据库: {final_vector_path}") if not os.path.exists(final_metadata_path): print(f"⚠ 元数据文件不存在: {final_metadata_path}") files_exist = False else: print(f"✓ 元数据文件: {final_metadata_path}") if not files_exist: print("\n知识库文件缺失。") choice = input("请选择: 1. 手动指定其他路径 2. 返回主菜单\n选择 (1-2): ").strip() if choice == "1": # 递归调用自身，重新引导配置 return feature_retrieval_mode(runtime_config) else: print("功能检索取消，返回主菜单。") return # 构建传递给检索器的配置 config = { 'vector_db_path': str(final_vector_path), 'metadata_path': str(final_metadata_path), 'graph_data_path': str(get_final_path( 'GRAPH_DATA_PATH', '', USER_GRAPH_DATA_PATH, INTERNAL_DEFAULT_GRAPH_DATA_PATH, runtime_config )) } # 询问用户是否保存此次配置供下次使用 if vector_input or metadata_input: # 只有当用户本次有输入时才询问保存 save_choice = input(f"\n是否将此次的路径配置保存，供下次程序启动时默认使用? (y/n): ").strip().lower() if save_choice == 'y': if save_runtime_config(runtime_config): print(f"✓ 配置已保存至: {RUNTIME_CONFIG_FILE}") else: print("⚠ 配置保存失败。") # 第二步：加载知识库并进入检索循环 print("\n" + "=" * 60) print("加载知识库并初始化检索器...") print("=" * 60) try: retriever = FeatureRetriever(config) if not retriever.load_knowledge_base(): print("加载知识库失败！可能文件格式不正确或已损坏。") retry = input("是否重新配置路径? (y/n): ").strip().lower() if retry == 'y': return feature_retrieval_mode(runtime_config) # 重新开始 else: print("功能检索取消，返回主菜单。") return print(f"知识库加载成功: {len(retriever.metadatas)} 个函数") print("\n请输入自然语言描述功能需求进行分析") print("输入 'quit' 或 'exit' 返回主菜单") while True: query = input("\n功能需求: ").strip() if query.lower() in ["quit", "exit"]: break if not query: continue # 执行分析 result = retriever.analyze_with_multiple_constraints(query) # 显示结果 print(f"\n分析结果:") print(f"实现状态: {'✅ 已实现' if result['implemented'] else '❌ 未实现'}") print(f"综合评分: {result.get('total_score', 0):.3f}") print(f"判断理由: {result.get('reason', '无理由')}") if result.get("most_relevant_function"): rel_func = result["most_relevant_function"] print(f"最相关函数: {rel_func.get('name')}") except Exception as e: print(f"分析失败: {e}") logger.error(f"功能检索失败: {e}", exc_info=True) # 发生异常时，也提供重新配置的选项 retry = input("\n发生错误，是否重新配置知识库路径? (y/n): ").strip().lower() if retry == 'y': feature_retrieval_mode(runtime_config) def issue_filter_mode(runtime_config): """问题单过滤模式（修正知识库传递问题）""" print("\n" + "=" * 60) print("问题单过滤模式") print("=" * 60) # 第一步：首先引导用户配置知识库路径（先输入，不输入则默认） print("第一步：配置知识库路径") # 获取当前默认的知识库路径 default_vector_path = get_final_path( 'VECTOR_DB_PATH', '', USER_VECTOR_DB_PATH, INTERNAL_DEFAULT_VECTOR_DB_PATH, runtime_config ) default_metadata_path = get_final_path( 'METADATA_PATH', '', USER_METADATA_PATH, INTERNAL_DEFAULT_METADATA_PATH, runtime_config ) # 引导用户输入向量数据库路径 vector_prompt = f"请输入向量数据库文件路径 (FAISS索引文件)\n(直接按Enter使用当前默认值: {default_vector_path}): " vector_input = input(vector_prompt).strip() # 使用用户输入重新计算最终路径 final_vector_path = get_final_path( 'VECTOR_DB_PATH', vector_input, USER_VECTOR_DB_PATH, INTERNAL_DEFAULT_VECTOR_DB_PATH, runtime_config ) if not vector_input: print(f"使用默认向量数据库: {final_vector_path}") # 引导用户输入元数据文件路径 metadata_prompt = f"请输入元数据文件路径\n(直接按Enter使用当前默认值: {default_metadata_path}): " metadata_input = input(metadata_prompt).strip() # 使用用户输入重新计算最终路径 final_metadata_path = get_final_path( 'METADATA_PATH', metadata_input, USER_METADATA_PATH, INTERNAL_DEFAULT_METADATA_PATH, runtime_config ) if not metadata_input: print(f"使用默认元数据文件: {final_metadata_path}") # 检查知识库文件是否存在 print(f"\n检查知识库文件...") files_exist = True if not os.path.exists(final_vector_path): print(f"⚠ 向量数据库文件不存在: {final_vector_path}") files_exist = False else: print(f"✓ 向量数据库: {final_vector_path}") if not os.path.exists(final_metadata_path): print(f"⚠ 元数据文件不存在: {final_metadata_path}") files_exist = False else: print(f"✓ 元数据文件: {final_metadata_path}") if not files_exist: print("\n知识库文件缺失，无法进行问题单过滤。") print("请先使用主菜单选项1创建知识库，或确保文件路径正确。") return # 构建知识库配置字典 kb_config = { 'vector_db_path': str(final_vector_path), 'metadata_path': str(final_metadata_path), 'graph_data_path': str(get_final_path( 'GRAPH_DATA_PATH', '', USER_GRAPH_DATA_PATH, INTERNAL_DEFAULT_GRAPH_DATA_PATH, runtime_config )) } # 询问是否保存此次配置 if vector_input or metadata_input: save_choice = input(f"\n是否将此知识库配置保存供下次使用? (y/n): ").strip().lower() if save_choice == 'y': if save_runtime_config(runtime_config): print(f"✓ 配置已保存至运行时配置文件") print("\n✓ 知识库配置完成") print("\n第二步：配置问题单过滤参数") # 第二步：配置问题单过滤的其他参数 default_input_excel = "科代问题单样例.xlsx" # 首先检查默认文件是否存在 if not os.path.exists(default_input_excel): print(f"⚠ 注意: 默认输入文件 '{default_input_excel}' 不存在") input_prompt = f"请输入待过滤的问题单Excel文件路径\n(直接按Enter使用默认值: {default_input_excel}): " input_excel = input(input_prompt).strip() # 用户不输入时使用默认值 if not input_excel: input_excel = default_input_excel print(f"使用默认输入文件: {input_excel}") # 路径检查和修正逻辑... # [保持您原有的路径检查逻辑，但这里简化展示] if not os.path.exists(input_excel): print(f"文件不存在: {input_excel}") print("问题单过滤取消。") return # 输出文件路径 default_output_json = "filtered_defects.json" output_prompt = f"请输入输出JSON文件路径\n(直接按Enter使用默认值: {default_output_json}): " output_json = input(output_prompt).strip() if not output_json: output_json = default_output_json print(f"使用默认输出文件: {output_json}") # 项目根目录路径 default_project_root = get_final_path( 'PROJECT_ROOT', '', USER_PROJECT_ROOT, INTERNAL_DEFAULT_PROJECT_ROOT, runtime_config ) project_prompt = f"请输入问题单中代码文件所对应的项目根目录路径\n(直接按Enter使用当前默认值: {default_project_root}): " project_input = input(project_prompt).strip() final_project_root = get_final_path( 'PROJECT_ROOT', project_input, USER_PROJECT_ROOT, INTERNAL_DEFAULT_PROJECT_ROOT, runtime_config ) if not project_input: print(f"使用默认项目路径: {final_project_root}") if not os.path.isdir(final_project_root): print(f"项目路径无效: {final_project_root}") print("问题单过滤取消。") return # 保存项目路径配置（如果用户有输入） if project_input: runtime_config['PROJECT_ROOT'] = str(final_project_root) save_runtime_config(runtime_config) print("\n" + "=" * 60) print("开始问题单过滤分析...") print("=" * 60) print(f"输入文件: {input_excel}") print(f"输出文件: {output_json}") print(f"项目路径: {final_project_root}") print("=" * 60) try: # 关键：确保传递知识库路径参数 success = run_issue_filtering( input_xlsx_path=input_excel, output_json_path=output_json, project_path_override=str(final_project_root), # 明确传递知识库路径 kb_vector_path=str(final_vector_path), # 确保这个变量存在且正确 kb_metadata_path=str(final_metadata_path) # 确保这个变量存在且正确 ) if success: print(f"\n" + "=" * 60) print("✓ 问题单过滤完成！") print(f"结果已保存至: {output_json}") print("=" + "=" * 60) # 结果查看逻辑 while True: view_option = input( "\n请选择操作:\n1. 查看统计摘要\n2. 打开结果文件\n3. 返回主菜单\n选择 (1-3): ").strip() if view_option == "1": try: with open(output_json, 'r', encoding='utf-8') as f: data = json.load(f) if not data: print("结果文件为空。") continue true_positives = sum(1 for r in data if r["analysis_result"].get("is_real_defect") is True) false_positives = sum(1 for r in data if r["analysis_result"].get("is_real_defect") is False) unknown = len(data) - true_positives - false_positives high_urgency = sum(1 for r in data if r["analysis_result"].get("urgency_score", 0) >= 70) medium_urgency = sum(1 for r in data if 40 <= r["analysis_result"].get("urgency_score", 0) < 70) low_urgency = sum(1 for r in data if 0 < r["analysis_result"].get("urgency_score", 0) < 40) total_affected = sum(len(r["analysis_result"].get("affected_functions", [])) for r in data) defects_with_affected = sum(1 for r in data if r["analysis_result"].get("affected_functions")) print(f"\n统计摘要:") print(f" 总缺陷数: {len(data)}") print(f" 真实缺陷: {true_positives} 条 ({true_positives / len(data) * 100:.1f}%)") print(f" 误报缺陷: {false_positives} 条 ({false_positives / len(data) * 100:.1f}%)") print(f" 未知/错误: {unknown} 条 ({unknown / len(data) * 100:.1f}%)") print(f"\n紧急程度分布:") print(f" 高紧急(≥70): {high_urgency} 条") print(f" 中紧急(40-69): {medium_urgency} 条") print(f" 低紧急(1-39): {low_urgency} 条") print(f" 零分(非缺陷): {false_positives} 条") print(f"\n影响域分析:") print(f" {defects_with_affected} 个缺陷影响了 {total_affected} 个函数") sheet_counts = {} for item in data: sheet_name = item.get("sheet_name", "unknown") sheet_counts[sheet_name] = sheet_counts.get(sheet_name, 0) + 1 if sheet_counts: print(f"\n工作表分布:") for sheet, count in sheet_counts.items(): print(f" {sheet}: {count} 条") except Exception as e: print(f"读取结果文件失败: {e}") elif view_option == "2": try: import subprocess import platform with open(output_json, 'r', encoding='utf-8') as f: data = json.load(f) if not data: print("结果文件为空。") continue print(f"\n结果文件包含 {len(data)} 条记录。") view_method = input( "查看方式:\n1. 完整JSON文件\n2. 仅真实缺陷\n3. 高紧急缺陷(≥70分)\n选择 (1-3): ").strip() if view_method == "1": if platform.system() == "Windows": os.startfile(output_json) elif platform.system() == "Darwin": subprocess.run(["open", output_json]) else: subprocess.run(["xdg-open", output_json]) print(f"已使用默认程序打开: {output_json}") elif view_method in ["2", "3"]: if view_method == "2": filtered_data = [item for item in data if item["analysis_result"].get("is_real_defect") is True] print(f"\n真实缺陷 ({len(filtered_data)} 条):") else: filtered_data = [item for item in data if item["analysis_result"].get("urgency_score", 0) >= 70] print(f"\n高紧急缺陷 ({len(filtered_data)} 条):") for i, item in enumerate(filtered_data[:10], 1): print(f"\n{i}. 工作表: {item.get('sheet_name')}, 行: {item.get('row_index')}") print( f" 文件: {os.path.basename(item.get('file_path', ''))}:{item.get('line_number')}") print(f" 描述: {item.get('defect_description', '')[:80]}...") print(f" 真实缺陷: {item['analysis_result'].get('is_real_defect')}") print(f" 紧急分数: {item['analysis_result'].get('urgency_score', 0)}") if item['analysis_result'].get('affected_functions'): print( f" 影响函数: {', '.join(item['analysis_result'].get('affected_functions', []))}") if len(filtered_data) > 10: print(f"\n... 还有 {len(filtered_data) - 10} 条未显示") save_filtered = input(f"\n是否将筛选结果保存为新文件? (y/n): ").strip().lower() if save_filtered == 'y': filtered_file = output_json.replace('.json', '_filtered.json') with open(filtered_file, 'w', encoding='utf-8') as f: json.dump(filtered_data, f, indent=2, ensure_ascii=False) print(f"筛选结果已保存至: {filtered_file}") except Exception as e: print(f"打开或处理结果文件失败: {e}") elif view_option == "3": break else: print("无效选择，请重新输入") else: print("\n" + "=" * 60) print("✗ 问题单过滤过程失败。") print("=" + "=" * 60) except Exception as e: print(f"\n问题单过滤执行失败: {e}") import traceback traceback.print_exc() def static_analysis_mode(runtime_config): """静态分析模式（遵循"先输入，不输入则默认"原则）""" print("\n" + "=" * 60) print("静态分析模式") print("=" * 60) # 1. 首先检查并配置知识库（静态分析前必须指定知识库） print("第一步：配置知识库") # --- 修改开始：使用与项目路径相同的交互逻辑 --- # 使用get_final_path计算当前默认的知识库路径 default_vector_path = get_final_path( 'VECTOR_DB_PATH', '', USER_VECTOR_DB_PATH, INTERNAL_DEFAULT_VECTOR_DB_PATH, runtime_config ) default_metadata_path = get_final_path( 'METADATA_PATH', '', USER_METADATA_PATH, INTERNAL_DEFAULT_METADATA_PATH, runtime_config ) # 引导输入向量数据库路径 vector_prompt = f"请输入FAISS向量数据库文件路径\n(直接按Enter使用当前默认值: {default_vector_path}): " vector_input = input(vector_prompt).strip() # 引导输入元数据文件路径 metadata_prompt = f"请输入元数据文件路径\n(直接按Enter使用当前默认值: {default_metadata_path}): " metadata_input = input(metadata_prompt).strip() # 使用用户输入（或空输入）重新计算最终路径 final_vector_path = get_final_path( 'VECTOR_DB_PATH', vector_input, USER_VECTOR_DB_PATH, INTERNAL_DEFAULT_VECTOR_DB_PATH, runtime_config ) final_metadata_path = get_final_path( 'METADATA_PATH', metadata_input, USER_METADATA_PATH, INTERNAL_DEFAULT_METADATA_PATH, runtime_config ) # 检查文件是否存在 print(f"\n检查知识库文件...") files_exist = True if not os.path.exists(final_vector_path): print(f"⚠ 向量数据库文件不存在: {final_vector_path}") files_exist = False else: print(f"✓ 向量数据库: {final_vector_path}") if not os.path.exists(final_metadata_path): print(f"⚠ 元数据文件不存在: {final_metadata_path}") files_exist = False else: print(f"✓ 元数据文件: {final_metadata_path}") if not files_exist: print("\n知识库文件缺失。") choice = input("请选择: 1. 手动指定其他路径 2. 返回主菜单\n选择 (1-2): ").strip() if choice == "1": # 递归调用自身，重新引导配置 return static_analysis_mode(runtime_config) else: print("静态分析取消，返回主菜单。") return # 构建知识库配置字典 config = { 'vector_db_path': str(final_vector_path), 'metadata_path': str(final_metadata_path), 'graph_data_path': str(get_final_path( 'GRAPH_DATA_PATH', '', USER_GRAPH_DATA_PATH, INTERNAL_DEFAULT_GRAPH_DATA_PATH, runtime_config )) } # 询问是否保存此次配置（如果用户本次有输入） if vector_input or metadata_input: save_choice = input(f"\n是否将此次的知识库配置保存，供下次程序启动时默认使用? (y/n): ").strip().lower() if save_choice == 'y': if save_runtime_config(runtime_config): print(f"✓ 配置已保存至: {RUNTIME_CONFIG_FILE}") else: print("⚠ 配置保存失败。") # --- 修改结束 --- print("✓ 知识库配置成功") print("\n请配置静态分析参数：") # 2. 获取要分析的项目路径 - 先显示默认值，再提示输入 # 使用get_final_path计算当前默认项目路径 default_project_root = get_final_path( 'PROJECT_ROOT', '', USER_PROJECT_ROOT, INTERNAL_DEFAULT_PROJECT_ROOT, runtime_config ) project_prompt = f"请输入要分析的C/C++项目根目录路径\n(直接按Enter使用当前默认值: {default_project_root}): " project_input = input(project_prompt).strip() # ...（后续代码保持不变）... # 使用用户输入重新计算最终路径 project_path = get_final_path( 'PROJECT_ROOT', project_input, USER_PROJECT_ROOT, INTERNAL_DEFAULT_PROJECT_ROOT, runtime_config ) # 如果用户没有输入，使用默认值 if not project_input: print(f"使用默认项目路径: {project_path}") # 路径验证 if not project_path.exists(): print(f"⚠ 路径不存在: {project_path}") retry = input("是否重新输入? (y/n): ").strip().lower() if retry != 'y': print("静态分析取消。") return # 重新获取输入 while True: new_input = input("请输入正确的项目根目录路径: ").strip() if not new_input: print("路径不能为空") continue new_path = get_final_path( 'PROJECT_ROOT', new_input, USER_PROJECT_ROOT, INTERNAL_DEFAULT_PROJECT_ROOT, runtime_config ) if not new_path.exists(): print(f"路径不存在: {new_path}") continue if not new_path.is_dir(): print(f"路径不是目录: {new_path}") continue project_path = new_path project_input = new_input break if not project_path.is_dir(): print(f"⚠ 路径不是目录: {project_path}") retry = input("是否重新输入? (y/n): ").strip().lower() if retry != 'y': print("静态分析取消。") return # 重新获取输入 while True: new_input = input("请输入正确的项目根目录路径: ").strip() if not new_input: print("路径不能为空") continue new_path = get_final_path( 'PROJECT_ROOT', new_input, USER_PROJECT_ROOT, INTERNAL_DEFAULT_PROJECT_ROOT, runtime_config ) if not new_path.is_dir(): print(f"路径不是目录: {new_path}") continue project_path = new_path project_input = new_input break # 3. 获取规则文件路径 - 先显示默认值，再提示输入 default_rules = "审查规则.xlsx" # 首先检查默认文件是否存在 if not os.path.exists(default_rules): print(f"⚠ 注意: 默认规则文件 '{default_rules}' 不存在") rules_prompt = f"请输入审查规则Excel文件路径\n(直接按Enter使用默认值: {default_rules}): " rules_input = input(rules_prompt).strip() # 用户不输入时使用默认值 if not rules_input: rules_excel = default_rules print(f"使用默认规则文件: {rules_excel}") else: rules_excel = rules_input # 路径验证 rules_path = Path(rules_excel) if not rules_path.exists(): print(f"⚠ 规则文件不存在: {rules_path}") # 如果用户输入了但文件不存在，询问是否重新输入 if rules_excel != default_rules: # 用户输入了自定义路径 retry = input("文件不存在，是否重新输入路径? (y/n): ").strip().lower() if retry == 'y': # 重新获取输入 while True: new_input = input("请输入正确的规则文件路径: ").strip() if not new_input: print("路径不能为空") continue new_path = Path(new_input) if not new_path.exists(): print(f"文件不存在: {new_path}") continue rules_excel = new_input rules_path = new_path break else: print("静态分析取消。") return else: # 用户使用的是默认路径 print(f"默认规则文件 {default_rules} 不存在。") manual_input = input("是否手动输入文件路径? (y/n): ").strip().lower() if manual_input != 'y': print("静态分析取消。") return while True: new_input = input("请输入正确的规则文件路径: ").strip() if not new_input: print("路径不能为空") continue new_path = Path(new_input) if not new_path.exists(): print(f"文件不存在: {new_path}") continue rules_excel = new_input rules_path = new_path break # 4. 获取输出报告名称 - 先显示默认值，再提示输入 default_output = "audit_report" output_prompt = f"请输入输出报告文件的基础名称 (不包含扩展名)\n(直接按Enter使用默认值: {default_output}): " output_input = input(output_prompt).strip() # 用户不输入时使用默认值 if not output_input: output_base = default_output print(f"使用默认输出报告名称: {output_base}") else: output_base = output_input print("\n" + "=" * 60) print("开始静态分析...") print("=" * 60) print(f"项目路径: {project_path}") print(f"规则文件: {rules_path}") print(f"输出报告: {output_base}") print("=" * 60) try: success = run_static_analysis( project_path=str(project_path), rules_excel=str(rules_path), output_report=output_base ) if success: print("\n" + "=" * 60) print("✓ 静态分析执行成功！") print("=" + "=" * 60) # 保存更新后的运行时配置（如果用户有输入项目路径） if project_input: runtime_config['PROJECT_ROOT'] = str(project_path) save_runtime_config(runtime_config) print(f"✓ 项目路径已保存到运行时配置") # 提供查看结果的选项 result_files = [ f"{output_base}.html", f"{output_base}.pdf", f"{output_base}.json" ] existing_files = [] for file in result_files: if os.path.exists(file): existing_files.append(file) if existing_files: print(f"\n生成的报告文件:") for file in existing_files: print(f" - {file}") view_option = input(f"\n是否打开HTML报告文件? (y/n): ").strip().lower() if view_option == 'y' and os.path.exists(f"{output_base}.html"): try: import subprocess import platform html_file = f"{output_base}.html" if platform.system() == "Windows": os.startfile(html_file) elif platform.system() == "Darwin": subprocess.run(["open", html_file]) else: subprocess.run(["xdg-open", html_file]) print(f"已使用默认程序打开: {html_file}") except Exception as e: print(f"打开报告文件失败: {e}") else: print("\n" + "=" * 60) print("✗ 静态分析执行过程中遇到问题。") print("=" + "=" * 60) except Exception as e: print(f"\n静态分析执行失败: {e}") import traceback traceback.print_exc() def main_menu(): """主菜单界面""" print("\n" + "=" * 60) print("卫星星务软件代码RAG知识库系统") print("=" * 60) print("版本: 3.0") print("功能: 支持代码知识库构建、功能检索、问题单过滤、静态分析") print("=" * 60) # 加载运行时配置 runtime_config = load_runtime_config() if runtime_config: print(f"已加载运行时配置 ({len(runtime_config)} 项)") # 检查API Key if not check_api_key(): return while True: print("\n" + "=" * 60) print("主菜单") print("=" * 60) print("1. 创建项目知识库") print("2. 功能需求检索") print("3. 问题单过滤") print("4. 静态分析") print("5. 管理配置文件") print("6. 退出系统") print("=" * 60) choice = input("\n请选择功能 (1-6): ").strip() if choice == "1": print("\n" + "=" * 60) print("创建项目知识库") print("=" * 60) print("此功能将分析代码项目，构建知识图谱和向量数据库") confirm = input("\n确认开始知识库构建? (y/n): ").strip().lower() if confirm == 'y': try: result = build_rag_database_interactive() if result: print("\n知识库构建完成!") except Exception as e: print(f"知识库构建失败: {e}") logger.error(f"知识库构建失败: {e}", exc_info=True) elif choice == "2": print("\n进入功能需求检索模式...") kb_exists, _, _, _, _ = check_knowledge_base_exists(runtime_config) if not kb_exists: print("\n⚠ 尚未建设项目知识库!") response = input("是否先创建或配置知识库? (y/n): ").strip().lower() if response == 'y': if not manual_configure_knowledge_base(runtime_config): continue else: continue feature_retrieval_mode(runtime_config) elif choice == "3": print("\n进入问题单过滤模式...") kb_exists, _, _, _, _ = check_knowledge_base_exists(runtime_config) if not kb_exists: print("\n⚠ 尚未建设项目知识库!") response = input("是否配置已有知识库文件? (y/n): ").strip().lower() if response == 'y': if not manual_configure_knowledge_base(runtime_config): continue else: print("请先使用选项1创建知识库") continue issue_filter_mode(runtime_config) elif choice == "4": print("\n进入静态分析模式...") static_analysis_mode(runtime_config) elif choice == "5": print("\n" + "=" * 60) print("配置文件管理") print("=" * 60) print("1. 查看当前运行时配置") print("2. 清除运行时配置") print("3. 返回主菜单") config_choice = input("\n请选择操作 (1-3): ").strip() if config_choice == "1": print(f"\n当前运行时配置 ({RUNTIME_CONFIG_FILE}):") if runtime_config: for key, value in runtime_config.items(): print(f" {key}: {value}") else: print(" 无运行时配置") print(f"\n配置文件 (config.py) 中的主路径变量:") print(f" PROJECT_ROOT: {PROJECT_ROOT}") print(f" VECTOR_DB_PATH: {VECTOR_DB_PATH}") print(f" METADATA_PATH: {METADATA_PATH}") print(f" GRAPH_DATA_PATH: {GRAPH_DATA_PATH}") print(f" KNOWLEDGE_GRAPH_PATH: {KNOWLEDGE_GRAPH_PATH}") print(f"\n配置文件 (config.py) 中的默认后备路径 (Default_*):") print(f" Default_PROJECT_ROOT: {Default_PROJECT_ROOT}") print(f" Default_VECTOR_DB_PATH: {Default_VECTOR_DB_PATH}") print(f" Default_METADATA_PATH: {Default_METADATA_PATH}") input("\n按Enter键继续...") elif config_choice == "2": if os.path.exists(RUNTIME_CONFIG_FILE): confirm = input(f"确定要删除运行时配置文件 {RUNTIME_CONFIG_FILE} 吗? (y/n): ").strip().lower() if confirm == 'y': os.remove(RUNTIME_CONFIG_FILE) runtime_config.clear() print("运行时配置文件已删除。") else: print("运行时配置文件不存在。") elif config_choice == "3": continue else: print("无效选择") elif choice == "6": print("\n感谢使用，再见!") # 退出前保存配置 if runtime_config: save_runtime_config(runtime_config) break else: print("无效的选择，请重新输入") if __name__ == "__main__": try: main_menu() except KeyboardInterrupt: print("\n\n程序被用户中断。") except Exception as e: print(f"\n程序执行出错: {e}") import traceback traceback.print_exc() logger.error(f"主程序异常: {e}", exc_info=True)