diff --git a/packaging/requirements-pack.txt b/packaging/requirements-pack.txt index 280b818..efc1763 100644 --- a/packaging/requirements-pack.txt +++ b/packaging/requirements-pack.txt @@ -1,9 +1,9 @@ pyinstaller==6.17.0 -numpy -pandas -matplotlib -scipy -openpyxl -requests -openai -pywebview +numpy>=1.24,<2.0 +pandas>=2.0,<3.0 +matplotlib>=3.7,<4.0 +scipy>=1.10,<2.0 +openpyxl>=3.1,<4.0 +requests>=2.28,<3.0 +openai>=1.0,<2.0 +pywebview>=4.0,<6.0 diff --git a/src/electrochem_v6/agent/agent_controller.py b/src/electrochem_v6/agent/agent_controller.py index 4e4198c..1cb321e 100644 --- a/src/electrochem_v6/agent/agent_controller.py +++ b/src/electrochem_v6/agent/agent_controller.py @@ -4,12 +4,15 @@ """ import json +import logging from typing import List, Dict, Any, Optional, Callable from electrochem_v6.llm.base_client import BaseLLMClient from .tool_executor import execute_tool from .tools import ALL_TOOLS from .prompts import SYSTEM_PROMPT +_logger = logging.getLogger(__name__) + def _debug_log(tag: str, payload): """统一的Debug输出，保证中文不会变成??""" @@ -20,7 +23,7 @@ def _debug_log(tag: str, payload): text = json.dumps(payload, ensure_ascii=False, default=str) except Exception: text = str(payload) - print(f"[DEBUG] {tag}: {text}") + _logger.debug("[%s]: %s", tag, text) class AgentController: diff --git a/src/electrochem_v6/agent/tool_executor.py b/src/electrochem_v6/agent/tool_executor.py index 880a08e..5034105 100644 --- a/src/electrochem_v6/agent/tool_executor.py +++ b/src/electrochem_v6/agent/tool_executor.py @@ -1,752 +1,129 @@ -"""Docstring""" -# Fixed problematic line -# Fixed problematic line -"""Docstring""" +"""Central tool dispatcher for the AI agent. + +All tool implementations live in dedicated modules (tools_data, tools_projects, +tools_analysis, tools_catalyst). This module wires them into a single +``execute_tool()`` entry-point. +""" import json -import os -import pandas as pd -from pathlib import Path -from typing import Dict, Any, List, Optional +import logging +from typing import Dict, Any + from .tools_analysis import tool_read_quality_report, tool_analyze_processing_results from .tools_catalyst import tool_get_catalyst_info +from .tools_data import ( + tool_analyze_data_characteristics, + tool_preview_data_file, + tool_scan_data_folder, +) +from .tools_projects import ( + tool_auto_process_with_smart_params, + tool_create_project, + tool_get_current_compare_selection, + tool_get_current_project_history, + tool_get_current_project_summary, + tool_get_processing_history, +) from electrochem_v6.llm.config import LLMConfig from electrochem_v6.llm.vision_client import VisionClient - -def execute_tool(tool_name: str, arguments: str | Dict[str, Any]) -> Dict: - """ - Execute tool function - - Args: - tool_name: Tool name - arguments: Arguments (JSON string or dict) - - Returns: - Result dict - """ - # 解析参数 - if isinstance(arguments, str): - try: - args = json.loads(arguments) - except json.JSONDecodeError: - return {"success": False, "error": "参数解析失败"} - else: - args = arguments - - # 路由到具体函? - tool_map = { - # 基础工具 - "query_lsv_summary": tool_query_lsv_summary, - "find_best_catalysts": tool_find_best_catalysts, - "compare_catalysts": tool_compare_catalysts, - "get_current_project_summary": tool_get_current_project_summary, - "get_current_project_history": tool_get_current_project_history, - "get_current_compare_selection": tool_get_current_compare_selection, - - # 增强工具(让AI能自主分析) - "scan_data_folder": tool_scan_data_folder, - "preview_data_file": tool_preview_data_file, - "analyze_data_characteristics": tool_analyze_data_characteristics, - "auto_process_with_smart_params": tool_auto_process_with_smart_params, - - # 管理工具 - "create_project": tool_create_project, - "get_processing_history": tool_get_processing_history, - - # 智能分析工具 - "read_quality_report": tool_read_quality_report, - "analyze_processing_results": tool_analyze_processing_results, - - # 催化剂中心工? - "get_catalyst_info": tool_get_catalyst_info, - - # 视觉诊断 - "analyze_waveform_image": tool_analyze_waveform_image, - } - - if tool_name in tool_map: - try: - return tool_map[tool_name](**args) - except Exception as e: - return {"success": False, "error": f"工具执行失败: {str(e)}"} - else: - return {"success": False, "error": f"未知工具: {tool_name}"} +_logger = logging.getLogger(__name__) -# ============== 基础工具实现 ============== +# ── LSV query tools (kept here – small, tightly coupled to history) ──────── def tool_query_lsv_summary(project_id: str = None, sort_by: str = "eta", top_n: int = None) -> Dict: - """Docstring""" + """查询LSV数据汇总。""" try: from electrochem_v6.store.legacy_runtime import get_history_manager_v6 hist_mgr = get_history_manager_v6() - - # ?关键修复:识?all"字符串并转换为None - if project_id and project_id.lower() in ['all', 'none', '']: + + if project_id and project_id.lower() in ("all", "none", ""): project_id = None - - # 查询数据 - if project_id: - summary = hist_mgr.get_lsv_summary(project_id=project_id) - else: - # 查询所有项? - summary = hist_mgr.get_lsv_summary() - - samples = summary.get('samples', []) - - # 调试信息 - print(f"[DEBUG] tool_query_lsv_summary: project_id={project_id}, 原始samples数量 = {len(samples)}") - - # 排序 + + summary = hist_mgr.get_lsv_summary(project_id=project_id) if project_id else hist_mgr.get_lsv_summary() + samples = summary.get("samples", []) + + _logger.debug("tool_query_lsv_summary: project_id=%s, 原始samples数量 = %d", project_id, len(samples)) + if sort_by == "tafel": - samples = sorted(samples, key=lambda x: x.get('tafel_slope', 999)) + samples = sorted(samples, key=lambda x: x.get("tafel_slope", 999)) else: - samples = sorted(samples, key=lambda x: x.get('overpotential_10', 999)) - - # 限制数量 + samples = sorted(samples, key=lambda x: x.get("overpotential_10", 999)) + if top_n: samples = samples[:top_n] - + return { "success": True, - "total_samples": len(summary.get('samples', [])), + "total_samples": len(summary.get("samples", [])), "returned_samples": len(samples), - "samples": samples + "samples": samples, } except Exception as e: return {"success": False, "error": str(e)} def tool_find_best_catalysts(project_id: str = None, count: int = 5) -> Dict: - """找出最优催化剂(修复:project_id可选)""" - # ?修复:project_id设为可选,None时查询所有项? + """找出最优催化剂。""" result = tool_query_lsv_summary(project_id=project_id, sort_by="eta", top_n=count) - - # 调试信息 - print(f"[DEBUG] tool_find_best_catalysts: result={result}") - - if result.get('success'): - samples = result.get('samples', []) - - # ?修复:检查samples而不是returned_samples + _logger.debug("tool_find_best_catalysts: result=%s", result) + + if result.get("success"): + samples = result.get("samples", []) if len(samples) == 0: return { "success": False, "message": "数据库中没有找到LSV记录。可能原因:1) 尚未处理任何LSV数据 2) 数据未正确保存到历史记录", - "suggestion": "Please check data processing tab for LSV data" + "suggestion": "Please check data processing tab for LSV data", } - return { "success": True, "count": len(samples), - "total_in_database": result.get('total_samples', len(samples)), - "best_catalysts": samples + "total_in_database": result.get("total_samples", len(samples)), + "best_catalysts": samples, } - else: - return result + return result -def tool_compare_catalysts(sample_names: List[str]) -> Dict: - """对比催化剂性能""" +def tool_compare_catalysts(sample_names: list[str]) -> Dict: + """对比催化剂性能。""" try: from electrochem_v6.store.legacy_runtime import get_history_manager_v6 hist_mgr = get_history_manager_v6() all_records = hist_mgr.get_all_records() - - # 查找指定样品 + comparison = [] for name in sample_names: for record in all_records: - if record.get('sample_name') == name and record.get('type') == 'LSV': - results = record.get('results', {}) - comparison.append({ - "sample_name": name, - "overpotential_10": results.get('overpotential_10'), - "tafel_slope": results.get('tafel_slope'), - "project": record.get('project_name', 'Unknown') - }) + if record.get("sample_name") == name and record.get("type") == "LSV": + results = record.get("results", {}) + comparison.append( + { + "sample_name": name, + "overpotential_10": results.get("overpotential_10"), + "tafel_slope": results.get("tafel_slope"), + "project": record.get("project_name", "Unknown"), + } + ) break - + if comparison: - # 找出最? - best_eta_sample = min(comparison, key=lambda x: x.get('overpotential_10', 999)) - + best_eta_sample = min(comparison, key=lambda x: x.get("overpotential_10", 999)) return { "success": True, "comparison": comparison, - "best_catalyst": best_eta_sample['sample_name'], - "best_overpotential": best_eta_sample['overpotential_10'] - } - else: - return {"success": False, "error": "未找到指定样品的数据"} - - except Exception as e: - return {"success": False, "error": str(e)} - - -def _resolve_v6_project(project_id: str = None, project_name: str = None) -> tuple[Optional[Dict[str, Any]], Optional[str]]: - from electrochem_v6.store.projects import list_projects - - projects = list_projects(status="active").get("projects") or [] - clean_id = str(project_id or "").strip() - clean_name = str(project_name or "").strip() - - if clean_id: - found = next((item for item in projects if str(item.get("id") or "").strip() == clean_id), None) - return found, None if found else f"未找到项目ID: {clean_id}" - - if clean_name: - found = next((item for item in projects if str(item.get("name") or "").strip() == clean_name), None) - if found: - return found, None - lowered = clean_name.casefold() - found = next((item for item in projects if str(item.get("name") or "").strip().casefold() == lowered), None) - return found, None if found else f"未找到项目名称: {clean_name}" - - return None, "请提供 project_id 或 project_name" - - -def _simplify_v6_history_record(record: Dict[str, Any]) -> Dict[str, Any]: - results = record.get("results") if isinstance(record.get("results"), dict) else {} - output_files = record.get("output_files") if isinstance(record.get("output_files"), list) else [] - return { - "timestamp": record.get("timestamp"), - "type": record.get("type"), - "sample_name": record.get("sample_name"), - "status": record.get("status"), - "file_name": record.get("file_name"), - "project_name": record.get("project_name"), - "summary_path": record.get("summary_path"), - "output_file_count": len([item for item in output_files if str(item or "").strip()]), - "results": { - "overpotential_10": results.get("overpotential_10"), - "potential_10": results.get("potential_10"), - "potential_at_10.0": results.get("potential_at_10.0"), - "tafel_slope": results.get("tafel_slope"), - }, - } - - -def _simplify_v6_compare_row(item: Dict[str, Any]) -> Dict[str, Any]: - return { - "sample_name": item.get("sample_name"), - "overpotential_10": item.get("overpotential_10"), - "potential_10": item.get("potential_10"), - "tafel_slope": item.get("tafel_slope"), - "record_count": item.get("record_count"), - "latest_time": item.get("latest_time"), - } - - -def tool_get_current_project_summary(project_id: str = None, project_name: str = None) -> Dict: - try: - from electrochem_v6.store.history import get_stats, list_history - from electrochem_v6.store.projects import get_lsv_summary - - project, err = _resolve_v6_project(project_id=project_id, project_name=project_name) - if not project: - return {"success": False, "error": err or "项目不存在"} - - pid = str(project.get("id") or "") - stats = (get_stats(project_id=pid, include_archived=False).get("data") or {}) - history = (list_history(project_id=pid, limit=5, include_archived=False).get("records") or []) - lsv = (get_lsv_summary(project_id=pid, page=1, page_size=5, sort_by="eta").get("lsv_summary") or {}) - return { - "success": True, - "project": { - "id": project.get("id"), - "name": project.get("name"), - "description": project.get("description"), - "updated_at": project.get("updated_at"), - }, - "stats": stats, - "recent_history": [_simplify_v6_history_record(item) for item in history[:5]], - "top_lsv_samples": [_simplify_v6_compare_row(item) for item in (lsv.get("samples") or [])[:5]], - } - except Exception as e: - return {"success": False, "error": str(e)} - - -def tool_get_current_project_history( - project_id: str = None, - project_name: str = None, - record_type: str = None, - limit: int = 10, -) -> Dict: - try: - from electrochem_v6.store.history import list_history - - project, err = _resolve_v6_project(project_id=project_id, project_name=project_name) - if not project: - return {"success": False, "error": err or "项目不存在"} - - pid = str(project.get("id") or "") - safe_limit = max(1, min(int(limit or 10), 50)) - records = list_history(project_id=pid, limit=max(safe_limit * 3, 20), include_archived=False).get("records") or [] - clean_type = str(record_type or "").strip().upper() - if clean_type: - records = [item for item in records if str(item.get("type") or "").upper() == clean_type] - records = records[:safe_limit] - return { - "success": True, - "project": {"id": project.get("id"), "name": project.get("name")}, - "record_type": clean_type or None, - "returned_count": len(records), - "records": [_simplify_v6_history_record(item) for item in records], - } - except Exception as e: - return {"success": False, "error": str(e)} - - -def tool_get_current_compare_selection( - project_id: str = None, - project_name: str = None, - sample_names: Optional[List[str]] = None, - limit: int = 5, -) -> Dict: - try: - from electrochem_v6.core import get_latest_project_lsv_compare_plot - from electrochem_v6.store.projects import get_lsv_summary - - project, err = _resolve_v6_project(project_id=project_id, project_name=project_name) - if not project: - return {"success": False, "error": err or "项目不存在"} - - pid = str(project.get("id") or "") - safe_limit = max(1, min(int(limit or 5), 20)) - summary = get_lsv_summary(project_id=pid, page=1, page_size=100, sort_by="eta").get("lsv_summary") or {} - samples = summary.get("samples") or [] - requested = [str(item).strip() for item in (sample_names or []) if str(item).strip()] - if requested: - requested_set = set(requested) - rows = [item for item in samples if str(item.get("sample_name") or "").strip() in requested_set] - selection_mode = "explicit_samples" - else: - rows = samples[:safe_limit] - selection_mode = "project_top_samples" - - latest_overlay = get_latest_project_lsv_compare_plot(project_id=pid, chart_type="overlay") - latest_overlay_plot = latest_overlay.get("plot") if latest_overlay.get("status") == "success" else None - latest_overlay_meta = None - if isinstance(latest_overlay_plot, dict): - latest_overlay_meta = { - "file_name": latest_overlay_plot.get("file_name"), - "plot_path": latest_overlay_plot.get("plot_path"), - "generated_at": latest_overlay_plot.get("generated_at"), - "selected_samples": latest_overlay_plot.get("selected_samples"), - } - - return { - "success": True, - "project": {"id": project.get("id"), "name": project.get("name")}, - "selection_mode": selection_mode, - "note": "当前 UI 选中的样品不会持久化到后端；未提供 sample_names 时返回项目内可对比的顶部样品。", - "compare_rows": [_simplify_v6_compare_row(item) for item in rows[:safe_limit]], - "latest_overlay_plot": latest_overlay_meta, - } - except Exception as e: - return {"success": False, "error": str(e)} - - -# ============== 增强工具实现(让AI??数据?============= - -def tool_scan_data_folder(folder_path: str) -> Dict: - """Docstring""" - try: - if not os.path.exists(folder_path): - return {"success": False, "error": f"文件夹不存在: {folder_path}"} - - files_info = [] - - for root, dirs, files in os.walk(folder_path): - for file in files: - if file.endswith(('.txt', '.csv')): - file_path = os.path.join(root, file) - - # 检测文件类? - file_upper = file.upper() - if "LSV" in file_upper or "TAFEL" in file_upper: - detected_type = "LSV" - elif "CV" in file_upper: - detected_type = "CV" - elif "EIS" in file_upper: - detected_type = "EIS" - elif "ECSA" in file_upper: - detected_type = "ECSA" - else: - detected_type = "Unknown" - - files_info.append({ - "file_name": file, - "file_path": file_path, - "detected_type": detected_type, - "folder": os.path.basename(root), - "size_kb": round(os.path.getsize(file_path) / 1024, 2) - }) - - # 统计 - by_type = {} - for f in files_info: - ftype = f['detected_type'] - by_type[ftype] = by_type.get(ftype, 0) + 1 - - return { - "success": True, - "folder_path": folder_path, - "total_files": len(files_info), - "files": files_info[:30], # 返回?0个避免太? - "statistics": { - "total": len(files_info), - "by_type": by_type, - "folders": len(set(f['folder'] for f in files_info)) + "best_catalyst": best_eta_sample["sample_name"], + "best_overpotential": best_eta_sample["overpotential_10"], } - } - except Exception as e: - return {"success": False, "error": str(e)} - - -def tool_preview_data_file(file_path: str, lines: int = 20) -> Dict: - """预览数据文件""" - try: - if not os.path.exists(file_path): - return {"success": False, "error": f"文件不存? {file_path}"} - - # 尝试多种编码 - encodings = ['utf-8', 'gbk', 'gb2312', 'latin-1'] - preview_lines = None - - for encoding in encodings: - try: - with open(file_path, 'r', encoding=encoding) as f: - preview_lines = [f.readline().strip() for _ in range(lines)] - break - except: - continue - - if preview_lines is None: - return {"success": False, "error": "无法读取文件(编码问题)"} - - # 分析内容 - content_str = '\n'.join(preview_lines) - has_potential = 'potential' in content_str.lower() - has_current = 'current' in content_str.lower() - has_freq = 'freq' in content_str.lower() - - # 推断类型 - if has_freq: - detected_type = "EIS" - elif has_potential and has_current: - detected_type = "LSV/CV" - else: - detected_type = "Unknown" - - return { - "success": True, - "file_path": file_path, - "preview_lines": preview_lines, - "detected_type": detected_type, - "has_header": has_potential or has_current, - "line_count_preview": len(preview_lines) - } - except Exception as e: - return {"success": False, "error": str(e)} - - -def tool_analyze_data_characteristics(file_path: str, data_type: str) -> Dict: - """分析数据特征(用于智能决定参数)""" - try: - # 自动检测数据起始行 - from electrochem_v6.core.processing_compat import auto_detect_data_start - - start_line = auto_detect_data_start(file_path) - - # ?修复:更健壮的数据读取,处理各种格式 - try: - # 尝试1:智能分隔符 - df = pd.read_csv(file_path, sep=r'\s+|,', skiprows=start_line-1, engine='python', nrows=1000, on_bad_lines='skip') - except: - try: - # 尝试2:仅空格 - df = pd.read_csv(file_path, delim_whitespace=True, skiprows=start_line-1, nrows=1000, on_bad_lines='skip') - except: - # 尝试3:仅逗号 - df = pd.read_csv(file_path, sep=',', skiprows=start_line-1, nrows=1000, on_bad_lines='skip') - - characteristics = { - "data_start_line": start_line, - "data_points": len(df) - } - - if data_type == "LSV": - # 查找电流? - current_col = next((col for col in df.columns if 'current' in col.lower()), None) - if current_col: - currents_mA = df[current_col].abs() * 1000 # 转为mA - - characteristics.update({ - "current_range_mA": { - "min": float(currents_mA.min()), - "max": float(currents_mA.max()) - }, - "suggested_tafel_range": "1-10" # 默认 - }) - - # 智能推荐Tafel范围 - max_current = currents_mA.max() - if max_current > 50: - characteristics["suggested_tafel_range"] = "5-50" - characteristics["reasoning"] = "电流较大,推荐使?-50 mA/cm²范围" - elif max_current < 5: - characteristics["suggested_tafel_range"] = "0.5-5" - characteristics["reasoning"] = "电流较小,推荐使?.5-5 mA/cm²范围" - else: - characteristics["reasoning"] = "电流范围正常,使用标?-10 mA/cm²范围" - - return { - "success": True, - "file_path": file_path, - "data_type": data_type, - "characteristics": characteristics - } - except Exception as e: - return {"success": False, "error": str(e)} - - -def tool_auto_process_with_smart_params( - folder_path: str, - data_type: str, - project_name: str = None, - potential_offset: float = None, - electrode_area: float = None, - target_current: str = None, - tafel_range: str = None, - extra_gui_params: Optional[Dict[str, Any]] = None, -) -> Dict: - """AI自主处理数据(核心功能)""" - try: - # Step 1: 扫描文件? - scan_result = tool_scan_data_folder(folder_path) - if not scan_result['success']: - return scan_result - - if scan_result['total_files'] == 0: - return {"success": False, "error": "文件夹中没有找到数据文件"} - - # Step 2: 使用用户指定或默认参? - print("Using default parameters for data processing") - - # ?支持用户自定义参? - if potential_offset is not None: - print("Processing...") - if electrode_area is not None: - print(f"📏 用户指定电极面积: {electrode_area} cm²") - if target_current is not None: - print(f"User-specified target current: {target_current} mA/cm^2") - if tafel_range is not None: - print("Processing...") - - # Step 3: 构建处理参数 - gui_vars = { - 'area': electrode_area if electrode_area is not None else 1.0, - 'potential_offset': potential_offset if potential_offset is not None else 0.0, # ?支持电位偏移 - 'auto_detect_start': True - } - - if data_type == "LSV": - final_target_current = target_current if target_current else '10,100' - enable_tafel = tafel_range is not None - final_tafel_range = tafel_range if tafel_range else '1-10' - gui_vars.update({ - 'lsv_enabled': True, - 'lsv_target_current': final_target_current, - 'tafel_enabled': enable_tafel, - 'tafel_range': final_tafel_range - }) - if not enable_tafel: - print("Processing LSV data only (Tafel analysis not enabled)") - elif data_type == "CV": - gui_vars.update({ - 'cv_enabled': True, - 'cv_match': 'prefix', - 'cv_prefix': 'CV', - 'cv_peaks_enabled': True, - 'cv_peaks_smooth': 5, - 'cv_peaks_min_height': 1.0, - 'cv_peaks_min_dist': 5, - 'cv_peaks_max': 3, - }) - print("Processing CV data with peak detection enabled") - elif data_type == "EIS": - gui_vars.update({ - 'eis_enabled': True, - 'eis_match': 'prefix', - 'eis_prefix': 'EIS', - 'plot_nyquist': True, - 'plot_bode': False, - 'eis_xlabel': "Z' (Ohm)", - 'eis_ylabel': "-Z'' (Ohm)", - }) - print("Processing EIS data (Nyquist plot enabled)") - elif data_type == "ECSA": - gui_vars.update({ - 'ecsa_enabled': True, - 'ecsa_match': 'prefix', - 'ecsa_prefix': 'ECSA', - 'ecsa_ev': 0.10, - 'ecsa_last_n': 1, - 'ecsa_avg_last_n': False, - 'ecsa_cs_value': 40.0, - 'ecsa_cs_unit': 'uF/cm^2', - 'ecsa_use_abs_delta': True, - }) - print("Processing ECSA dataset for double-layer analysis") - else: - return {"success": False, "error": f"Unsupported data_type: {data_type}"} - - if extra_gui_params: - gui_vars.update(extra_gui_params) - # Step 4: 创建或使用项? - if project_name: - from electrochem_v6.store.legacy_runtime import get_project_manager_v6 - proj_mgr = get_project_manager_v6() - - # 检查项目是否已存在 - existing = proj_mgr.get_all_projects() - project_id = None - for proj in existing: - if proj['name'] == project_name: - project_id = proj['id'] - break - - # 不存在则创建 - if not project_id: - project_id = proj_mgr.create_project( - name=project_name, - description=f"AI自动创建:{data_type}数据分析" - ) - - gui_vars['project_id'] = project_id - - # Step 5: 执行处理 - from electrochem_v6.core.processing_compat import run_pipeline - - result = run_pipeline(folder_path, gui_vars) - - # ?修复:正确统计处理文件数 - # 从scan_result的statistics中获取完整统计(不是files列表,那只有?0个) - messages = result.get('messages', []) - - # 使用完整的统计信? - by_type_stats = scan_result.get('statistics', {}).get('by_type', {}) - actual_processed = by_type_stats.get(data_type, 0) # ?从统计中获取真实数量 - - # Step 7: 汇总结果(包含质量分析? - quality_summary = result.get('quality_summary', {}) - vision_findings = [] - if isinstance(quality_summary, dict): - for report in (quality_summary.get('files', []) or []): - stats = (report or {}).get('stats') or {} - noise = stats.get('noise_analysis') or {} - vision = noise.get('vision_analysis') - if isinstance(vision, dict): - vision_findings.append({ - "file": report.get('filename') or report.get('file') or "unknown", - "success": bool(vision.get('success')), - "result": vision.get('result') or vision.get('error'), - "model": vision.get('model'), - "image_path": vision.get('image_path'), - }) - base_suggestion = "建议:处理完成后,可以问\"分析质量报告\"获取详细分析,或\"找出最优催化剂\"查看性能排名" - if vision_findings: - highlight = [] - for finding in vision_findings[:3]: - icon = "✅" if finding["success"] else "⚠️" - highlight.append(f"{icon} {finding['file']}: {finding.get('result','无视觉结论')}") - if len(vision_findings) > 3: - highlight.append(f"... 还有 {len(vision_findings) - 3} 个文件完成视觉诊断") - vision_block = "\n".join(highlight) - ai_suggestion = f"{base_suggestion}\n\n📷 视觉诊断:\n{vision_block}" - else: - ai_suggestion = base_suggestion - - return { - "success": True, - "message": "AI??????", - "summary": f"????{actual_processed}?{data_type}??", - "processing": { - "scanned_total": scan_result['total_files'], - "processed_count": actual_processed, - "data_type": data_type, - "output_files": messages # LSV_results.csv, quality_report.json? - }, - "parameters": { - "potential_offset": potential_offset if potential_offset else 0.0, - "electrode_area": electrode_area if electrode_area else 1.0, - "target_current": target_current if target_current else "10,100", - "tafel_enabled": gui_vars.get('tafel_enabled', False) - }, - "project_id": gui_vars.get('project_id'), - "ai_suggestion": ai_suggestion, - "vision_findings": vision_findings, - "quality_summary": quality_summary, - } - except Exception as e: - import traceback - return { - "success": False, - "error": str(e), - "traceback": traceback.format_exc() - } - - -def tool_create_project(name: str, description: str = "") -> Dict: - """创建项目""" - try: - from electrochem_v6.store.legacy_runtime import get_project_manager_v6 - - proj_mgr = get_project_manager_v6() - project_id = proj_mgr.create_project(name=name, description=description) - - return { - "success": True, - "project_id": project_id, - "message": f"项目'{name}'创建成功" - } + return {"success": False, "error": "未找到指定样品的数据"} except Exception as e: return {"success": False, "error": str(e)} -def tool_get_processing_history(project_id: str = None, record_type: str = None, limit: int = 20) -> Dict: - """获取处理历史""" - try: - from electrochem_v6.store.legacy_runtime import get_history_manager_v6 - - hist_mgr = get_history_manager_v6() - - if project_id: - records = hist_mgr.get_records_by_project(project_id) - else: - records = hist_mgr.get_all_records() - - if record_type: - records = [r for r in records if r.get('type', '').upper() == record_type.upper()] - - # 最新的在前 - records = sorted(records, key=lambda x: x.get('timestamp', ''), reverse=True) - - return { - "success": True, - "total_records": len(records), - "records": records[:limit] - } - except Exception as e: - return {"success": False, "error": str(e)} - - -# ============== 视觉分析工具 ============== +# ── Vision tool ──────────────────────────────────────────────────────────── _vision_client_cache: VisionClient | None = None @@ -782,17 +159,50 @@ def tool_analyze_waveform_image(image_path: str, context: str = "", max_tokens: return client.analyze_image(image_path=image_path, prompt=prompt, max_tokens=max_tokens) -# ============== 辅助函数 ============== +# ── Main dispatcher ──────────────────────────────────────────────────────── -def _detect_type_from_content(content: str) -> str: - """Function docstring""" - content_lower = content.lower() - if 'freq' in content_lower: - return "EIS" - elif 'potential' in content_lower and 'current' in content_lower: - return "LSV/CV" +def execute_tool(tool_name: str, arguments: str | Dict[str, Any]) -> Dict: + """Execute a named tool function with given arguments.""" + if isinstance(arguments, str): + try: + args = json.loads(arguments) + except json.JSONDecodeError: + return {"success": False, "error": "参数解析失败"} else: - return "Unknown" + args = arguments + + tool_map = { + # LSV query + "query_lsv_summary": tool_query_lsv_summary, + "find_best_catalysts": tool_find_best_catalysts, + "compare_catalysts": tool_compare_catalysts, + # Project / history + "get_current_project_summary": tool_get_current_project_summary, + "get_current_project_history": tool_get_current_project_history, + "get_current_compare_selection": tool_get_current_compare_selection, + "create_project": tool_create_project, + "get_processing_history": tool_get_processing_history, + # Data scanning + "scan_data_folder": tool_scan_data_folder, + "preview_data_file": tool_preview_data_file, + "analyze_data_characteristics": tool_analyze_data_characteristics, + "auto_process_with_smart_params": tool_auto_process_with_smart_params, + # Analysis + "read_quality_report": tool_read_quality_report, + "analyze_processing_results": tool_analyze_processing_results, + # Catalyst + "get_catalyst_info": tool_get_catalyst_info, + # Vision + "analyze_waveform_image": tool_analyze_waveform_image, + } + + if tool_name in tool_map: + try: + return tool_map[tool_name](**args) + except Exception as e: + _logger.warning("工具 %s 执行失败: %s", tool_name, e, exc_info=True) + return {"success": False, "error": f"工具执行失败: {str(e)}"} + return {"success": False, "error": f"未知工具: {tool_name}"} __all__ = ["execute_tool"] diff --git a/src/electrochem_v6/agent/tools_data.py b/src/electrochem_v6/agent/tools_data.py new file mode 100644 index 0000000..277ce17 --- /dev/null +++ b/src/electrochem_v6/agent/tools_data.py @@ -0,0 +1,182 @@ +"""Data scanning, preview and analysis tools for AI agent. + +These tools allow the AI to autonomously explore data folders, +preview file contents, and analyze data characteristics. +""" + +from __future__ import annotations + +import logging +import os +from typing import Any, Dict + +import pandas as pd + +_logger = logging.getLogger(__name__) + + +def tool_scan_data_folder(folder_path: str) -> Dict: + """扫描数据文件夹，返回文件列表和统计信息。""" + try: + resolved = os.path.realpath(folder_path) + if not os.path.isdir(resolved): + return {"success": False, "error": f"文件夹不存在: {folder_path}"} + folder_path = resolved + + files_info = [] + + for root, dirs, files in os.walk(folder_path): + for file in files: + if file.endswith((".txt", ".csv")): + file_path = os.path.join(root, file) + + file_upper = file.upper() + if "LSV" in file_upper or "TAFEL" in file_upper: + detected_type = "LSV" + elif "CV" in file_upper: + detected_type = "CV" + elif "EIS" in file_upper: + detected_type = "EIS" + elif "ECSA" in file_upper: + detected_type = "ECSA" + else: + detected_type = "Unknown" + + files_info.append( + { + "file_name": file, + "file_path": file_path, + "detected_type": detected_type, + "folder": os.path.basename(root), + "size_kb": round(os.path.getsize(file_path) / 1024, 2), + } + ) + + by_type: Dict[str, int] = {} + for f in files_info: + ftype = f["detected_type"] + by_type[ftype] = by_type.get(ftype, 0) + 1 + + MAX_RETURNED_FILES = 30 + + return { + "success": True, + "folder_path": folder_path, + "total_files": len(files_info), + "files": files_info[:MAX_RETURNED_FILES], + "statistics": { + "total": len(files_info), + "by_type": by_type, + "folders": len(set(f["folder"] for f in files_info)), + }, + } + except Exception as e: + return {"success": False, "error": str(e)} + + +def tool_preview_data_file(file_path: str, lines: int = 20) -> Dict: + """预览数据文件前 N 行。""" + try: + resolved = os.path.realpath(file_path) + ext = os.path.splitext(resolved)[1].lower() + if ext not in (".txt", ".csv", ".xlsx", ".xls", ".json"): + return {"success": False, "error": f"不支持的文件类型: {ext}"} + file_path = resolved + if not os.path.exists(file_path): + return {"success": False, "error": f"文件不存在: {file_path}"} + + encodings = ["utf-8", "gbk", "gb2312", "latin-1"] + preview_lines = None + + for encoding in encodings: + try: + with open(file_path, "r", encoding=encoding) as f: + preview_lines = [f.readline().strip() for _ in range(lines)] + break + except Exception: + continue + + if preview_lines is None: + return {"success": False, "error": "无法读取文件(编码问题)"} + + content_str = "\n".join(preview_lines) + has_potential = "potential" in content_str.lower() + has_current = "current" in content_str.lower() + has_freq = "freq" in content_str.lower() + + if has_freq: + detected_type = "EIS" + elif has_potential and has_current: + detected_type = "LSV/CV" + else: + detected_type = "Unknown" + + return { + "success": True, + "file_path": file_path, + "preview_lines": preview_lines, + "detected_type": detected_type, + "has_header": has_potential or has_current, + "line_count_preview": len(preview_lines), + } + except Exception as e: + return {"success": False, "error": str(e)} + + +def tool_analyze_data_characteristics(file_path: str, data_type: str) -> Dict: + """分析数据特征(用于智能决定参数)。""" + try: + from electrochem_v6.core.processing_compat import auto_detect_data_start + + start_line = auto_detect_data_start(file_path) + + try: + df = pd.read_csv( + file_path, sep=r"\s+|,", skiprows=start_line - 1, engine="python", nrows=1000, on_bad_lines="skip" + ) + except Exception: + try: + df = pd.read_csv( + file_path, delim_whitespace=True, skiprows=start_line - 1, nrows=1000, on_bad_lines="skip" + ) + except Exception: + df = pd.read_csv(file_path, sep=",", skiprows=start_line - 1, nrows=1000, on_bad_lines="skip") + + characteristics: Dict[str, Any] = { + "data_start_line": start_line, + "data_points": len(df), + } + + if data_type == "LSV": + current_col = next((col for col in df.columns if "current" in col.lower()), None) + if current_col: + currents_mA = df[current_col].abs() * 1000 + + characteristics.update( + { + "current_range_mA": { + "min": float(currents_mA.min()), + "max": float(currents_mA.max()), + }, + "suggested_tafel_range": "1-10", + } + ) + + max_current = currents_mA.max() + if max_current > 50: + characteristics["suggested_tafel_range"] = "5-50" + characteristics["reasoning"] = "电流较大,推荐使用5-50 mA/cm²范围" + elif max_current < 5: + characteristics["suggested_tafel_range"] = "0.5-5" + characteristics["reasoning"] = "电流较小,推荐使用0.5-5 mA/cm²范围" + else: + characteristics["reasoning"] = "电流范围正常,使用标准1-10 mA/cm²范围" + + return { + "success": True, + "file_path": file_path, + "data_type": data_type, + "characteristics": characteristics, + } + except Exception as e: + return {"success": False, "error": str(e)} diff --git a/src/electrochem_v6/agent/tools_projects.py b/src/electrochem_v6/agent/tools_projects.py new file mode 100644 index 0000000..758d538 --- /dev/null +++ b/src/electrochem_v6/agent/tools_projects.py @@ -0,0 +1,403 @@ +"""Project & history query tools, and AI auto-processing tool. + +These tools let the AI query project summaries, history records, +comparison selections, and run automated data processing. +""" + +from __future__ import annotations + +import logging +from typing import Any, Dict, List, Optional + +_logger = logging.getLogger(__name__) + + +# --------------------------------------------------------------------------- +# Internal helpers +# --------------------------------------------------------------------------- + +def _resolve_v6_project( + project_id: str = None, project_name: str = None +) -> tuple[Optional[Dict[str, Any]], Optional[str]]: + from electrochem_v6.store.projects import list_projects + + projects = list_projects(status="active").get("projects") or [] + clean_id = str(project_id or "").strip() + clean_name = str(project_name or "").strip() + + if clean_id: + found = next( + (item for item in projects if str(item.get("id") or "").strip() == clean_id), + None, + ) + return found, None if found else f"未找到项目ID: {clean_id}" + + if clean_name: + found = next( + (item for item in projects if str(item.get("name") or "").strip() == clean_name), + None, + ) + if found: + return found, None + lowered = clean_name.casefold() + found = next( + (item for item in projects if str(item.get("name") or "").strip().casefold() == lowered), + None, + ) + return found, None if found else f"未找到项目名称: {clean_name}" + + return None, "请提供 project_id 或 project_name" + + +def _simplify_v6_history_record(record: Dict[str, Any]) -> Dict[str, Any]: + results = record.get("results") if isinstance(record.get("results"), dict) else {} + output_files = record.get("output_files") if isinstance(record.get("output_files"), list) else [] + return { + "timestamp": record.get("timestamp"), + "type": record.get("type"), + "sample_name": record.get("sample_name"), + "status": record.get("status"), + "file_name": record.get("file_name"), + "project_name": record.get("project_name"), + "summary_path": record.get("summary_path"), + "output_file_count": len([item for item in output_files if str(item or "").strip()]), + "results": { + "overpotential_10": results.get("overpotential_10"), + "potential_10": results.get("potential_10"), + "potential_at_10.0": results.get("potential_at_10.0"), + "tafel_slope": results.get("tafel_slope"), + }, + } + + +def _simplify_v6_compare_row(item: Dict[str, Any]) -> Dict[str, Any]: + return { + "sample_name": item.get("sample_name"), + "overpotential_10": item.get("overpotential_10"), + "potential_10": item.get("potential_10"), + "tafel_slope": item.get("tafel_slope"), + "record_count": item.get("record_count"), + "latest_time": item.get("latest_time"), + } + + +# --------------------------------------------------------------------------- +# Tool implementations +# --------------------------------------------------------------------------- + + +def tool_get_current_project_summary( + project_id: str = None, project_name: str = None +) -> Dict: + try: + from electrochem_v6.store.history import get_stats, list_history + from electrochem_v6.store.projects import get_lsv_summary + + project, err = _resolve_v6_project(project_id=project_id, project_name=project_name) + if not project: + return {"success": False, "error": err or "项目不存在"} + + pid = str(project.get("id") or "") + stats = get_stats(project_id=pid, include_archived=False).get("data") or {} + history = list_history(project_id=pid, limit=5, include_archived=False).get("records") or [] + lsv = get_lsv_summary(project_id=pid, page=1, page_size=5, sort_by="eta").get("lsv_summary") or {} + return { + "success": True, + "project": { + "id": project.get("id"), + "name": project.get("name"), + "description": project.get("description"), + "updated_at": project.get("updated_at"), + }, + "stats": stats, + "recent_history": [_simplify_v6_history_record(item) for item in history[:5]], + "top_lsv_samples": [_simplify_v6_compare_row(item) for item in (lsv.get("samples") or [])[:5]], + } + except Exception as e: + return {"success": False, "error": str(e)} + + +def tool_get_current_project_history( + project_id: str = None, + project_name: str = None, + record_type: str = None, + limit: int = 10, +) -> Dict: + try: + from electrochem_v6.store.history import list_history + + project, err = _resolve_v6_project(project_id=project_id, project_name=project_name) + if not project: + return {"success": False, "error": err or "项目不存在"} + + pid = str(project.get("id") or "") + safe_limit = max(1, min(int(limit or 10), 50)) + records = ( + list_history(project_id=pid, limit=max(safe_limit * 3, 20), include_archived=False).get("records") or [] + ) + clean_type = str(record_type or "").strip().upper() + if clean_type: + records = [item for item in records if str(item.get("type") or "").upper() == clean_type] + records = records[:safe_limit] + return { + "success": True, + "project": {"id": project.get("id"), "name": project.get("name")}, + "record_type": clean_type or None, + "returned_count": len(records), + "records": [_simplify_v6_history_record(item) for item in records], + } + except Exception as e: + return {"success": False, "error": str(e)} + + +def tool_get_current_compare_selection( + project_id: str = None, + project_name: str = None, + sample_names: Optional[List[str]] = None, + limit: int = 5, +) -> Dict: + try: + from electrochem_v6.core import get_latest_project_lsv_compare_plot + from electrochem_v6.store.projects import get_lsv_summary + + project, err = _resolve_v6_project(project_id=project_id, project_name=project_name) + if not project: + return {"success": False, "error": err or "项目不存在"} + + pid = str(project.get("id") or "") + safe_limit = max(1, min(int(limit or 5), 20)) + summary = get_lsv_summary(project_id=pid, page=1, page_size=100, sort_by="eta").get("lsv_summary") or {} + samples = summary.get("samples") or [] + requested = [str(item).strip() for item in (sample_names or []) if str(item).strip()] + if requested: + requested_set = set(requested) + rows = [item for item in samples if str(item.get("sample_name") or "").strip() in requested_set] + selection_mode = "explicit_samples" + else: + rows = samples[:safe_limit] + selection_mode = "project_top_samples" + + latest_overlay = get_latest_project_lsv_compare_plot(project_id=pid, chart_type="overlay") + latest_overlay_plot = latest_overlay.get("plot") if latest_overlay.get("status") == "success" else None + latest_overlay_meta = None + if isinstance(latest_overlay_plot, dict): + latest_overlay_meta = { + "file_name": latest_overlay_plot.get("file_name"), + "plot_path": latest_overlay_plot.get("plot_path"), + "generated_at": latest_overlay_plot.get("generated_at"), + "selected_samples": latest_overlay_plot.get("selected_samples"), + } + + return { + "success": True, + "project": {"id": project.get("id"), "name": project.get("name")}, + "selection_mode": selection_mode, + "note": "当前 UI 选中的样品不会持久化到后端；未提供 sample_names 时返回项目内可对比的顶部样品。", + "compare_rows": [_simplify_v6_compare_row(item) for item in rows[:safe_limit]], + "latest_overlay_plot": latest_overlay_meta, + } + except Exception as e: + return {"success": False, "error": str(e)} + + +def tool_create_project(name: str, description: str = "") -> Dict: + """创建项目。""" + try: + from electrochem_v6.store.legacy_runtime import get_project_manager_v6 + + proj_mgr = get_project_manager_v6() + project_id = proj_mgr.create_project(name=name, description=description) + + return {"success": True, "project_id": project_id, "message": f"项目'{name}'创建成功"} + except Exception as e: + return {"success": False, "error": str(e)} + + +def tool_get_processing_history( + project_id: str = None, record_type: str = None, limit: int = 20 +) -> Dict: + """获取处理历史。""" + try: + from electrochem_v6.store.legacy_runtime import get_history_manager_v6 + + hist_mgr = get_history_manager_v6() + + if project_id: + records = hist_mgr.get_records_by_project(project_id) + else: + records = hist_mgr.get_all_records() + + if record_type: + records = [r for r in records if r.get("type", "").upper() == record_type.upper()] + + records = sorted(records, key=lambda x: x.get("timestamp", ""), reverse=True) + + return {"success": True, "total_records": len(records), "records": records[:limit]} + except Exception as e: + return {"success": False, "error": str(e)} + + +def tool_auto_process_with_smart_params( + folder_path: str, + data_type: str, + project_name: str = None, + potential_offset: float = None, + electrode_area: float = None, + target_current: str = None, + tafel_range: str = None, + extra_gui_params: Optional[Dict[str, Any]] = None, +) -> Dict: + """AI自主处理数据(核心功能)。""" + try: + from .tools_data import tool_scan_data_folder + + scan_result = tool_scan_data_folder(folder_path) + if not scan_result["success"]: + return scan_result + + if scan_result["total_files"] == 0: + return {"success": False, "error": "文件夹中没有找到数据文件"} + + gui_vars: Dict[str, Any] = { + "area": electrode_area if electrode_area is not None else 1.0, + "potential_offset": potential_offset if potential_offset is not None else 0.0, + "auto_detect_start": True, + } + + if data_type == "LSV": + final_target_current = target_current if target_current else "10,100" + enable_tafel = tafel_range is not None + final_tafel_range = tafel_range if tafel_range else "1-10" + gui_vars.update( + { + "lsv_enabled": True, + "lsv_target_current": final_target_current, + "tafel_enabled": enable_tafel, + "tafel_range": final_tafel_range, + } + ) + elif data_type == "CV": + gui_vars.update( + { + "cv_enabled": True, + "cv_match": "prefix", + "cv_prefix": "CV", + "cv_peaks_enabled": True, + "cv_peaks_smooth": 5, + "cv_peaks_min_height": 1.0, + "cv_peaks_min_dist": 5, + "cv_peaks_max": 3, + } + ) + elif data_type == "EIS": + gui_vars.update( + { + "eis_enabled": True, + "eis_match": "prefix", + "eis_prefix": "EIS", + "plot_nyquist": True, + "plot_bode": False, + "eis_xlabel": "Z' (Ohm)", + "eis_ylabel": "-Z'' (Ohm)", + } + ) + elif data_type == "ECSA": + gui_vars.update( + { + "ecsa_enabled": True, + "ecsa_match": "prefix", + "ecsa_prefix": "ECSA", + "ecsa_ev": 0.10, + "ecsa_last_n": 1, + "ecsa_avg_last_n": False, + "ecsa_cs_value": 40.0, + "ecsa_cs_unit": "uF/cm^2", + "ecsa_use_abs_delta": True, + } + ) + else: + return {"success": False, "error": f"Unsupported data_type: {data_type}"} + + if extra_gui_params: + gui_vars.update(extra_gui_params) + + if project_name: + from electrochem_v6.store.legacy_runtime import get_project_manager_v6 + + proj_mgr = get_project_manager_v6() + existing = proj_mgr.get_all_projects() + project_id = None + for proj in existing: + if proj["name"] == project_name: + project_id = proj["id"] + break + if not project_id: + project_id = proj_mgr.create_project( + name=project_name, description=f"AI自动创建:{data_type}数据分析" + ) + gui_vars["project_id"] = project_id + + from electrochem_v6.core.processing_compat import run_pipeline + + result = run_pipeline(folder_path, gui_vars) + + messages = result.get("messages", []) + by_type_stats = scan_result.get("statistics", {}).get("by_type", {}) + actual_processed = by_type_stats.get(data_type, 0) + + quality_summary = result.get("quality_summary", {}) + vision_findings = [] + if isinstance(quality_summary, dict): + for report in quality_summary.get("files", []) or []: + stats = (report or {}).get("stats") or {} + noise = stats.get("noise_analysis") or {} + vision = noise.get("vision_analysis") + if isinstance(vision, dict): + vision_findings.append( + { + "file": report.get("filename") or report.get("file") or "unknown", + "success": bool(vision.get("success")), + "result": vision.get("result") or vision.get("error"), + "model": vision.get("model"), + "image_path": vision.get("image_path"), + } + ) + + base_suggestion = "建议:处理完成后,可以问\"分析质量报告\"获取详细分析,或\"找出最优催化剂\"查看性能排名" + if vision_findings: + highlight = [] + for finding in vision_findings[:3]: + icon = "✅" if finding["success"] else "⚠️" + highlight.append(f"{icon} {finding['file']}: {finding.get('result', '无视觉结论')}") + if len(vision_findings) > 3: + highlight.append(f"... 还有 {len(vision_findings) - 3} 个文件完成视觉诊断") + vision_block = "\n".join(highlight) + ai_suggestion = f"{base_suggestion}\n\n📷 视觉诊断:\n{vision_block}" + else: + ai_suggestion = base_suggestion + + return { + "success": True, + "message": "AI自动处理完成", + "summary": f"已处理{actual_processed}个{data_type}文件", + "processing": { + "scanned_total": scan_result["total_files"], + "processed_count": actual_processed, + "data_type": data_type, + "output_files": messages, + }, + "parameters": { + "potential_offset": potential_offset if potential_offset else 0.0, + "electrode_area": electrode_area if electrode_area else 1.0, + "target_current": target_current if target_current else "10,100", + "tafel_enabled": gui_vars.get("tafel_enabled", False), + }, + "project_id": gui_vars.get("project_id"), + "ai_suggestion": ai_suggestion, + "vision_findings": vision_findings, + "quality_summary": quality_summary, + } + except Exception as e: + import traceback + + return {"success": False, "error": str(e), "traceback": traceback.format_exc()} diff --git a/src/electrochem_v6/core/path_security.py b/src/electrochem_v6/core/path_security.py new file mode 100644 index 0000000..e6bb14e --- /dev/null +++ b/src/electrochem_v6/core/path_security.py @@ -0,0 +1,87 @@ +"""Centralised path validation helpers to prevent directory traversal attacks.""" + +from __future__ import annotations + +import os +import re +from pathlib import Path +from typing import Optional, Sequence + + +# Default allowed extensions for data files +_DATA_FILE_EXTENSIONS = frozenset({".txt", ".csv", ".xlsx", ".xls", ".json", ".zip"}) + +# Default allowed image extensions +_IMAGE_FILE_EXTENSIONS = frozenset({".png", ".jpg", ".jpeg", ".gif", ".bmp", ".svg", ".webp"}) + + +def sanitize_filename(name: str) -> str: + """Convert arbitrary user-provided name to a filesystem-safe base name. + + Strips directory separators and traversal sequences, then restricts to + alphanumeric, underscore, hyphen, and dot characters. + + Returns ``"unknown"`` for empty inputs. + """ + if not name: + return "unknown" + # Take only the basename to strip any directory components + base = os.path.basename(name) + # Remove anything that isn't safe + safe = re.sub(r"[^A-Za-z0-9_.\-]+", "_", base) + # Prevent hidden files or bare dots + safe = safe.lstrip(".") + return safe or "unknown" + + +def validate_path_within( + user_path: str, + allowed_root: str | Path, + *, + must_exist: bool = True, + allowed_extensions: Optional[Sequence[str]] = None, +) -> Path: + """Resolve *user_path* and ensure it stays under *allowed_root*. + + Raises :class:`ValueError` when the path escapes the allowed root, has a + disallowed extension, or (when *must_exist* is True) does not exist. + """ + root = Path(allowed_root).resolve() + try: + resolved = Path(user_path).resolve() + except (OSError, ValueError) as exc: + raise ValueError(f"路径无法解析: {user_path}") from exc + + # Ensure the resolved path is under the allowed root + try: + resolved.relative_to(root) + except ValueError: + raise ValueError(f"路径不在允许范围内: {user_path}") + + if must_exist and not resolved.exists(): + raise ValueError(f"路径不存在: {user_path}") + + if allowed_extensions is not None: + ext = resolved.suffix.lower() + if ext not in allowed_extensions: + raise ValueError(f"不允许的文件类型: {ext}") + + return resolved + + +def is_safe_data_path(user_path: str, allowed_root: str | Path) -> bool: + """Return True when *user_path* is inside *allowed_root* and looks like a data file.""" + try: + validate_path_within(user_path, allowed_root, must_exist=True, allowed_extensions=_DATA_FILE_EXTENSIONS) + return True + except ValueError: + return False + + +def is_safe_image_path(image_path: str, allowed_root: str | Path) -> bool: + """Return True when *image_path* is inside *allowed_root* and is an image.""" + try: + validate_path_within(image_path, allowed_root, must_exist=True, allowed_extensions=_IMAGE_FILE_EXTENSIONS) + return True + except ValueError: + return False diff --git a/src/electrochem_v6/core/process_service.py b/src/electrochem_v6/core/process_service.py index e139023..759acaa 100644 --- a/src/electrochem_v6/core/process_service.py +++ b/src/electrochem_v6/core/process_service.py @@ -15,6 +15,7 @@ from electrochem_v6.config import APP_NAME, APP_VERSION, get_quality_report_file from electrochem_v6.core import processing_core_v6 as processing_core from electrochem_v6.core.processing_compat import run_pipeline +from electrochem_v6.core.utils import as_float as _as_float, as_int as _as_int, as_bool as _as_bool from electrochem_v6.store.legacy_runtime import get_history_manager_v6, get_project_manager_v6 from electrochem_v6.store.history import attach_run_outputs import matplotlib.pyplot as plt @@ -29,36 +30,6 @@ } -def _as_float(value: Any, default: float) -> float: - try: - return float(value) - except Exception: - return default - - -def _as_int(value: Any, default: int) -> int: - try: - return int(value) - except Exception: - return default - - -def _as_bool(value: Any, default: bool) -> bool: - if value is None: - return default - if isinstance(value, bool): - return value - if isinstance(value, (int, float)): - return value != 0 - if isinstance(value, str): - text = value.strip().lower() - if text in {"1", "true", "yes", "y", "on"}: - return True - if text in {"0", "false", "no", "n", "off", ""}: - return False - return default - - def _payload_get(payload: Dict[str, Any], key: str, default: Any = None) -> Any: params = payload.get("params") if isinstance(params, dict) and key in params: diff --git a/src/electrochem_v6/core/processing_core_v6.py b/src/electrochem_v6/core/processing_core_v6.py index 570dfc1..9bb695d 100644 --- a/src/electrochem_v6/core/processing_core_v6.py +++ b/src/electrochem_v6/core/processing_core_v6.py @@ -278,7 +278,12 @@ def _sanitize_filename(name: str) -> str: """Convert arbitrary filename to filesystem-friendly format.""" if not name: return "unknown" - return re.sub(r"[^A-Za-z0-9_.-]+", "_", name) + # Strip directory components to prevent path traversal + base = os.path.basename(name) + safe = re.sub(r"[^A-Za-z0-9_.\-]+", "_", base) + # Prevent hidden files + safe = safe.lstrip(".") + return safe or "unknown" def save_waveform_plot( diff --git a/src/electrochem_v6/core/processing_cv.py b/src/electrochem_v6/core/processing_cv.py index 7d21f42..f72b055 100644 --- a/src/electrochem_v6/core/processing_cv.py +++ b/src/electrochem_v6/core/processing_cv.py @@ -9,6 +9,7 @@ from . import processing_core_v6 as core from .processing_quality import DataQualityChecker +from .utils import read_file_with_fallback_encodings _resolve_plot_font = core._resolve_plot_font HISTORY_MANAGER_AVAILABLE = core.HISTORY_MANAGER_AVAILABLE @@ -24,15 +25,7 @@ def process_cv(subfolder, file, params, enable_quality_check=True): file_stem = os.path.splitext(os.path.basename(file))[0] encodings = ['utf-8', 'gbk', 'gb2312', 'ascii', 'latin-1', 'cp1252'] - lines = None - - for encoding in encodings: - try: - with open(filepath, 'r', encoding=encoding) as f: - lines = f.readlines()[int(params['start_line']) - 1:] - break - except UnicodeDecodeError: - continue + lines = read_file_with_fallback_encodings(filepath, start_line=int(params['start_line'])) if lines is None: print(f"无法读取CV文件 {filepath}，尝试了所有编码格式") @@ -45,7 +38,7 @@ def process_cv(subfolder, file, params, enable_quality_check=True): try: potential.append(float(parts[0])) current.append(float(parts[1]) * 1000) - except: + except (ValueError, TypeError): continue if not potential or not current: @@ -133,14 +126,16 @@ def process_cv(subfolder, file, params, enable_quality_check=True): xy=(xp, yp), xytext=(10,10), textcoords='offset points', bbox=dict(boxstyle='round,pad=0.2', fc='white', alpha=0.7)) except Exception: - pass - except Exception: - pass + pass # annotation cosmetics – non-critical + except Exception as exc: + log(f"CV峰值检测异常（继续处理）: {exc}") plt.tight_layout() # 避免覆盖：包含源文件名 - plt.savefig(os.path.join(subfolder, f"{subname}_{file_stem}_CV.png"), dpi=300, bbox_inches='tight') - plt.close() + try: + plt.savefig(os.path.join(subfolder, f"{subname}_{file_stem}_CV.png"), dpi=300, bbox_inches='tight') + finally: + plt.close() # ✅ 添加：保存CV历史记录 if HISTORY_MANAGER_AVAILABLE: @@ -175,7 +170,7 @@ def process_cv(subfolder, file, params, enable_quality_check=True): proj = proj_mgr.get_project(params['project_id']) if proj: record['project_name'] = proj['name'] - except: + except Exception: pass history_mgr.add_record(record) diff --git a/src/electrochem_v6/core/processing_ecsa.py b/src/electrochem_v6/core/processing_ecsa.py index 4dc76f2..4e178ab 100644 --- a/src/electrochem_v6/core/processing_ecsa.py +++ b/src/electrochem_v6/core/processing_ecsa.py @@ -253,7 +253,7 @@ def process_ecsa_for_subfolder(subfolder: str, files: list, params: dict, common proj = proj_mgr.get_project(params['project_id']) if proj: record['project_name'] = proj['name'] - except: + except Exception: pass history_mgr.add_record(record) diff --git a/src/electrochem_v6/core/processing_eis.py b/src/electrochem_v6/core/processing_eis.py index c55443b..70a5310 100644 --- a/src/electrochem_v6/core/processing_eis.py +++ b/src/electrochem_v6/core/processing_eis.py @@ -7,6 +7,7 @@ import matplotlib.pyplot as plt from . import processing_core_v6 as core +from .utils import read_file_with_fallback_encodings _resolve_plot_font = core._resolve_plot_font HISTORY_MANAGER_AVAILABLE = core.HISTORY_MANAGER_AVAILABLE @@ -22,15 +23,7 @@ def process_eis(subfolder, file, params): file_stem = os.path.splitext(os.path.basename(file))[0] encodings = ['utf-8', 'gbk', 'gb2312', 'ascii', 'latin-1', 'cp1252'] - lines = None - - for encoding in encodings: - try: - with open(filepath, 'r', encoding=encoding) as f: - lines = f.readlines()[int(params['start_line']) - 1:] - break - except UnicodeDecodeError: - continue + lines = read_file_with_fallback_encodings(filepath, start_line=int(params['start_line'])) if lines is None: print(f"无法读取EIS文件 {filepath}，尝试了所有编码格式") @@ -45,7 +38,7 @@ def process_eis(subfolder, file, params): freq.append(float(parts[0])) # 频率 Hz z_real.append(float(parts[1])) # Z' 实阻抗 z_imag.append(float(parts[2])) # Z'' 虚阻抗 - except: + except (ValueError, TypeError): continue if not z_real or not z_imag or not freq: @@ -76,8 +69,10 @@ def process_eis(subfolder, file, params): plt.grid(True, alpha=0.3) plt.axis('equal') # 等比例坐标轴，更好地显示圆弧 plt.tight_layout() - plt.savefig(os.path.join(subfolder, f"{subname}_{file_stem}_EIS_Nyquist.png"), dpi=300, bbox_inches='tight') - plt.close() + try: + plt.savefig(os.path.join(subfolder, f"{subname}_{file_stem}_EIS_Nyquist.png"), dpi=300, bbox_inches='tight') + finally: + plt.close() if plot_bode: # 绘制波特图（幅值图和相位图） @@ -115,8 +110,10 @@ def process_eis(subfolder, file, params): ax2.grid(True, alpha=0.3) plt.tight_layout() - plt.savefig(os.path.join(subfolder, f"{subname}_{file_stem}_EIS_Bode.png"), dpi=300, bbox_inches='tight') - plt.close() + try: + plt.savefig(os.path.join(subfolder, f"{subname}_{file_stem}_EIS_Bode.png"), dpi=300, bbox_inches='tight') + finally: + plt.close() # ✅ 添加：保存EIS历史记录 if HISTORY_MANAGER_AVAILABLE: @@ -159,7 +156,7 @@ def process_eis(subfolder, file, params): proj = proj_mgr.get_project(params['project_id']) if proj: record['project_name'] = proj['name'] - except: + except Exception: pass history_mgr.add_record(record) diff --git a/src/electrochem_v6/core/processing_lsv.py b/src/electrochem_v6/core/processing_lsv.py index c7ef1f1..3c6aa69 100644 --- a/src/electrochem_v6/core/processing_lsv.py +++ b/src/electrochem_v6/core/processing_lsv.py @@ -11,6 +11,35 @@ from . import processing_core_v6 as core from .processing_pipeline import _as_bool from .processing_quality import DataQualityChecker +from .utils import read_file_with_fallback_encodings + +# ── Module-level constants ───────────────────────────────────────────────── +# Outlier detection (MAD-based) +MAD_Z_SCORE_CONSTANT = 0.6745 +MAD_OUTLIER_THRESHOLD = 3.5 + +# Fitting defaults +MIN_FITTING_POINTS = 3 +TAFEL_RANGE_RATIO_MIN = 3.0 +MAX_EXTRAPOLATION_FACTOR = 2.0 +EXTRAPOLATION_LINE_POINTS = 50 + +# Current unit conversion +A_TO_MA = 1000.0 + +# Half-wave potential defaults +HALFWAVE_PERCENTILE = 95 +HALFWAVE_FRACTION = 0.5 + +# EIS high-frequency thresholds for Tafel-region selection +HF_THRESHOLD_UP = 0.7 # upper extrapolation direction +HF_THRESHOLD_DOWN = 0.3 # lower extrapolation direction + +# Relative threshold for imaginary impedance filtering (1% of range) +EIS_RELATIVE_IMAG_THRESHOLD = 0.01 + +# Software version stamp embedded in detail exports +SOFTWARE_VERSION = "2.3.2" get_logger = core.get_logger _resolve_plot_font = core._resolve_plot_font @@ -78,7 +107,7 @@ def parse_target_currents(target_current_str): except ValueError: continue return sorted(list(set(currents))) - except Exception: + except (ValueError, TypeError): return [] def _parse_tafel_range(raw_value): @@ -96,11 +125,11 @@ def _parse_tafel_range(raw_value): try: lo = float(lo_text) hi = float(hi_text) - except Exception: + except (ValueError, TypeError): return None return (lo, hi) -def _filter_outliers(real_vals, imag_vals, freq_vals=None, thresh=3.5): +def _filter_outliers(real_vals, imag_vals, freq_vals=None, thresh=MAD_OUTLIER_THRESHOLD): """Filter outliers in high-frequency region using MAD. Returns (real_vals, imag_vals, freq_vals) with outliers removed. @@ -117,7 +146,7 @@ def _mask(vals): mad = _np.nanmedian(_np.abs(v - med)) if mad <= 0: return _np.ones_like(v, dtype=bool) - z = 0.6745 * (v - med) / mad + z = MAD_Z_SCORE_CONSTANT * (v - med) / mad return _np.abs(z) <= thresh mask = _mask(real_vals) & _mask(imag_vals) @@ -142,7 +171,7 @@ def interpolate_multiple_potentials(potential, current, target_currents): return results def potential_at_current(potential_V, current_mAcm2, target_i=10.0, - min_pts=3, tafel_ratio=3.0, max_extrap_factor=2.0): + min_pts=MIN_FITTING_POINTS, tafel_ratio=TAFEL_RANGE_RATIO_MIN, max_extrap_factor=MAX_EXTRAPOLATION_FACTOR): """计算在电流密度 i=target_i (mA/cm²) 时的电位 E (V)。 逻辑： 1) 若曲线覆盖目标电流 -> 线性插值； @@ -182,12 +211,12 @@ def potential_at_current(potential_V, current_mAcm2, target_i=10.0, # 外推 going_up = target_i > I.max() if going_up: - thr = max(1.0, 0.7 * I.max()) + thr = max(1.0, HF_THRESHOLD_UP * I.max()) sel = np.where(I >= thr)[0] if sel.size < min_pts: sel = np.arange(max(0, len(I) - min_pts), len(I)) else: - thr = 0.3 * I.min() + thr = HF_THRESHOLD_DOWN * I.min() sel = np.where(I <= thr)[0] if sel.size < min_pts: sel = np.arange(0, min(min_pts, len(I))) @@ -206,7 +235,7 @@ def potential_at_current(potential_V, current_mAcm2, target_i=10.0, E10 = float(a + b * np.log10(max(target_i, 1e-12))) i0 = I_sel.max() if going_up else target_i i1 = target_i if going_up else I_sel.min() - i_ext = np.linspace(i0, i1, 50) + i_ext = np.linspace(i0, i1, EXTRAPOLATION_LINE_POINTS) E_ext = a + b * np.log10(np.clip(i_ext, 1e-12, None)) method = f"tafel (b={b:.3f} V/dec)" else: @@ -214,7 +243,7 @@ def potential_at_current(potential_V, current_mAcm2, target_i=10.0, E10 = float(m * target_i + c) i0 = I_sel.max() if going_up else target_i i1 = target_i if going_up else I_sel.min() - i_ext = np.linspace(i0, i1, 50) + i_ext = np.linspace(i0, i1, EXTRAPOLATION_LINE_POINTS) E_ext = m * i_ext + c method = "linear" @@ -240,24 +269,14 @@ def get_ir_from_eis(subfolder, eis_filename, start_line, method='auto', hf_point try: filepath = os.path.join(subfolder, eis_filename) if not os.path.exists(filepath): - print(f"EIS文件不存在: {filepath}") + log(f"EIS文件不存在: {filepath}") return None # 尝试多种编码读取 - encodings = ['utf-8', 'gbk', 'gb2312', 'ascii', 'latin-1', 'cp1252'] - lines = None - - for encoding in encodings: - try: - with open(filepath, 'r', encoding=encoding) as f: - lines = f.readlines()[int(start_line) - 1:] - print(f"成功读取 {encoding} 编码读取EIS文件: {eis_filename}") - break - except UnicodeDecodeError: - continue + lines = read_file_with_fallback_encodings(filepath, start_line=int(start_line)) if lines is None: - print(f"成功读取 {filepath}高频区选取 ????") + log(f"无法读取文件 {filepath}, 所有编码均失败") return None z_real, z_imag, frequencies = [], [], [] @@ -272,30 +291,30 @@ def get_ir_from_eis(subfolder, eis_filename, start_line, method='auto', hf_point z_real.append(real) z_imag.append(imag) except ValueError as e: - print(f"?{line_num + int(start_line)}???频率范围: {line.strip()}, 错误: {e}") + log(f"第{line_num + int(start_line)}行数据解析失败: {line.strip()}, 错误: {e}") continue if not z_real or not z_imag: - print(f"EIS文件 {eis_filename} 高频区选取 ????") + log(f"EIS文件 {eis_filename} 中未提取到有效数据") return None - print(f"成功读取 {len(z_real)} ?EIS???") - print(f"Z'范围: {min(z_real):.3f} ~ {max(z_real):.3f} Ω") - print(f"Z''??: {min(z_imag):.3f} ~ {max(z_imag):.3f} Ω") + log(f"成功读取 {len(z_real)} 个EIS数据点") + log(f"Z'范围: {min(z_real):.3f} ~ {max(z_real):.3f} Ω") + log(f"Z''范围: {min(z_imag):.3f} ~ {max(z_imag):.3f} Ω") method_key = (method or 'auto').strip().lower() supported_methods = {'auto', 'hf_intercept', 'hf_mean', 'linear_fit'} if method_key not in supported_methods: method_key = 'auto' - # 高频区选取 ?X???? + # 全局最小虚部作为初始IR值 min_imag_index = min(range(len(z_imag)), key=lambda i: abs(z_imag[i])) ir_value = z_real[min_imag_index] - print(f"???频率范围: Z'={ir_value:.3f}Ω, Z''={z_imag[min_imag_index]:.3f}?") + log(f"全局最小虚部: Z'={ir_value:.3f}Ω, Z''={z_imag[min_imag_index]:.3f}Ω") - # 尝试多种编码读取高频区选取 + # 高频区数据筛选与分析 if frequencies: - print(f"频率范围: {min(frequencies):.2f} ~ {max(frequencies):.2f} Hz") + log(f"频率范围: {min(frequencies):.2f} ~ {max(frequencies):.2f} Hz") try: hf_n = int(hf_points) if hf_points is not None else 0 except Exception: @@ -312,26 +331,29 @@ def get_ir_from_eis(subfolder, eis_filename, start_line, method='auto', hf_point high_freq_real, high_freq_imag, high_freq_freqs ) - print( - f"成功读取 {len(high_freq_real)} ?成功读取 {min(high_freq_freqs):.2f} ~ {max(high_freq_freqs):.2f} Hz") + if not high_freq_real or not high_freq_imag or not high_freq_freqs: + log("高频区数据经异常值过滤后为空, 使用全局初始IR值") + else: + log( + f"高频区筛选出 {len(high_freq_real)} 个数据点, " + f"频率范围: {min(high_freq_freqs):.2f} ~ {max(high_freq_freqs):.2f} Hz") - if high_freq_imag: min_imag_idx_in_high_freq = min(range(len(high_freq_imag)), key=lambda i: abs(high_freq_imag[i])) high_freq_ir_value = high_freq_real[min_imag_idx_in_high_freq] - print(f"?????频率范围: Z'={high_freq_ir_value:.3f}?") + log(f"高频区最小虚部: Z'={high_freq_ir_value:.3f}Ω") z_imag_range = max(abs(min(z_imag)), abs(max(z_imag))) - threshold = 0.01 * z_imag_range if z_imag_range > 0 else 0.01 + threshold = EIS_RELATIVE_IMAG_THRESHOLD * z_imag_range if z_imag_range > 0 else EIS_RELATIVE_IMAG_THRESHOLD if method_key == 'hf_intercept': if abs(high_freq_imag[min_imag_idx_in_high_freq]) < threshold: ir_value = high_freq_ir_value - print(f"??频率范围: Z'={ir_value:.3f}Ω") + log(f"高频截距法: Z'={ir_value:.3f}Ω") else: - print("尝试多种编码读取高频区选取 Ω") + log("高频截距法: 虚部偏大, 不满足截距条件") elif method_key == 'hf_mean': ir_value = sum(high_freq_real) / len(high_freq_real) - print(f"??频率范围: Z'???{ir_value:.3f}?") + log(f"高频均值法: Z'均值={ir_value:.3f}Ω") elif method_key == 'linear_fit': if len(high_freq_real) >= 2: import numpy as _np @@ -343,28 +365,28 @@ def get_ir_from_eis(subfolder, eis_filename, start_line, method='auto', hf_point x_min, x_max = min(high_freq_real), max(high_freq_real) if candidate < x_min or candidate > x_max: ir_value = sum(high_freq_real) / len(high_freq_real) - print(f"尝试多种编码读取?{ir_value:.3f}?") + log(f"线性拟合外推超出范围, 回退均值法: {ir_value:.3f}Ω") else: ir_value = candidate - print(f"????频率范围: Z'={ir_value:.3f}Ω") + log(f"线性拟合截距法: Z'={ir_value:.3f}Ω") else: ir_value = sum(high_freq_real) / len(high_freq_real) - print(f"尝试多种编码读取?{ir_value:.3f}?") + log(f"线性拟合斜率接近零, 回退均值法: {ir_value:.3f}Ω") else: - print("尝试多种编码读取?????") + log("线性拟合所需数据点不足") else: if abs(high_freq_imag[min_imag_idx_in_high_freq]) < threshold: ir_value = high_freq_ir_value - print(f"高频区选取 ??X高频区选取 Z'={ir_value:.3f}Ω") + log(f"自动法(截距): Z'={ir_value:.3f}Ω") else: ir_value = sum(high_freq_real) / len(high_freq_real) - print(f"???X高频区选取 ????Z'???{ir_value:.3f}?") + log(f"自动法(均值): Z'均值={ir_value:.3f}Ω") - print(f"最终确定IR值: {ir_value:.3f}?") + log(f"最终确定IR值: {ir_value:.3f}Ω") return ir_value except Exception as e: - print(f"??EIS文件不存在: {e}") + log(f"读取EIS文件失败: {e}") return None def process_lsv(subfolder, file, params, project_id=None, enable_quality_check=True): @@ -391,23 +413,11 @@ def process_lsv(subfolder, file, params, project_id=None, enable_quality_check=T lsv_quality_report = None # 初始化质量报告变量，确保始终存在 # 尝试多种编码格式读取文件 - encodings = ['utf-8', 'gbk', 'gb2312', 'ascii', 'latin-1', 'cp1252'] - lines = None - - for encoding in encodings: - try: - with open(filepath, 'r', encoding=encoding) as f: - lines = f.readlines()[int(params['start_line']) - 1:] - logger.debug(f"成功使用 {encoding} 编码读取文件") - break - except UnicodeDecodeError: - continue - except FileNotFoundError: - logger.error(f"文件未找到: {filepath}") - raise FileFormatError(f"LSV文件不存在: {file}") - except Exception as e: - logger.error(f"读取文件失败: {filepath} - {str(e)}") - raise FileFormatError(f"无法读取LSV文件: {file} - {str(e)}") + try: + lines = read_file_with_fallback_encodings(filepath, start_line=int(params['start_line'])) + except (FileNotFoundError, OSError) as e: + logger.error(f"文件读取失败: {filepath} - {e}") + raise FileFormatError(f"无法读取LSV文件: {file} - {e}") if lines is None: logger.error(f"所有编码格式均无法读取文件: {filepath}") @@ -422,7 +432,7 @@ def process_lsv(subfolder, file, params, project_id=None, enable_quality_check=T try: pot = float(parts[0]) + float(params['offset']) i_A = float(parts[1]) - cur_signed = i_A * 1000 / float(params['area']) + cur_signed = i_A * A_TO_MA / float(params['area']) cur_used = abs(cur_signed) if params.get('use_abs_current', True) else cur_signed potential.append(pot) current_signed.append(cur_signed) @@ -679,8 +689,8 @@ def process_lsv(subfolder, file, params, project_id=None, enable_quality_check=T 'r2': float(r2), 'range': (lo, hi) } - except Exception: - pass + except (ValueError, TypeError) as exc: + logger.debug("Tafel拟合跳过: %s", exc) plt.xlabel(params['xlabel']) plt.ylabel(params['ylabel']) title = params['title'].replace("{sample}", subname) @@ -694,8 +704,10 @@ def process_lsv(subfolder, file, params, project_id=None, enable_quality_check=T plt.legend() plt.tight_layout() # 避免同一文件夹内多文件相互覆盖：在文件名中加入源文件名 - plt.savefig(os.path.join(subfolder, f"{subname}_{file_stem}_LSV.png"), dpi=300, bbox_inches='tight') - plt.close() + try: + plt.savefig(os.path.join(subfolder, f"{subname}_{file_stem}_LSV.png"), dpi=300, bbox_inches='tight') + finally: + plt.close() # 如果进行了IR补偿，绘制补偿后的LSV曲线 if potential_compensated is not None: @@ -769,8 +781,8 @@ def process_lsv(subfolder, file, params, project_id=None, enable_quality_check=T 'range': (lo, hi), 'ir_compensation': ir_compensation } - except Exception: - pass + except (ValueError, TypeError) as exc: + logger.debug("IR补偿Tafel拟合跳过: %s", exc) plt.xlabel(params['xlabel']) plt.ylabel(params['ylabel']) title_compensated = params['title'].replace("{sample}", subname) + f" (IR: {ir_compensation:.2f}Ω)" @@ -783,8 +795,10 @@ def process_lsv(subfolder, file, params, project_id=None, enable_quality_check=T if params.get('mark_targets', True) and target_potentials_compensated: plt.legend() plt.tight_layout() - plt.savefig(os.path.join(subfolder, f"{subname}_{file_stem}_LSV_IR_compensated.png"), dpi=300, bbox_inches='tight') - plt.close() + try: + plt.savefig(os.path.join(subfolder, f"{subname}_{file_stem}_LSV_IR_compensated.png"), dpi=300, bbox_inches='tight') + finally: + plt.close() # 准备返回结果：包含所有目标电流密度的数据（不重复写入Rs） result_row = [subname, file_stem] @@ -908,7 +922,7 @@ def process_lsv(subfolder, file, params, project_id=None, enable_quality_check=T pd.DataFrame(tgt_records).to_excel(writer, sheet_name='targets', index=False) # info sheet info_rows = [ - {'Key':'SoftwareVersion','Value':'2.3.2'}, + {'Key':'SoftwareVersion','Value': SOFTWARE_VERSION}, {'Key':'GeneratedAt','Value': datetime.now().isoformat(timespec='seconds')}, {'Key':'Sample','Value': subname}, {'Key':'SourceFile','Value': file}, @@ -927,12 +941,12 @@ def process_lsv(subfolder, file, params, project_id=None, enable_quality_check=T # Tafel拟合图导出（如果启用） if params.get('export_tafel_plot', False): - print(f"开始导出Tafel图...") + log(f"开始导出Tafel图...") import numpy as _np # 用于log计算 # 优先导出IR补偿Tafel图（推荐使用） if tafel_fit_data_ir is not None: - print("找到IR补偿Tafel拟合数据，开始生成IR补偿Tafel图...") + log("找到IR补偿Tafel拟合数据，开始生成IR补偿Tafel图...") try: plt.figure(figsize=(8, 6)) plt.rcParams['font.sans-serif'] = [font_to_use] @@ -961,11 +975,11 @@ def process_lsv(subfolder, file, params, project_id=None, enable_quality_check=T tafel_ir_path = os.path.join(subfolder, f"{subname}_{file_stem}_Tafel_fit_IR.png") plt.savefig(tafel_ir_path, dpi=300, bbox_inches='tight') plt.close() - print(f"IR补偿Tafel拟合图已保存: {tafel_ir_path}") + log(f"IR补偿Tafel拟合图已保存: {tafel_ir_path}") except Exception as e: - print(f"导出IR补偿Tafel图失败: {e}") + log(f"导出IR补偿Tafel图失败: {e}") elif tafel_fit_data_original is not None: - print("未找到IR补偿数据，使用原始数据生成Tafel图（建议启用IR补偿获得更准确结果）...") + log("未找到IR补偿数据，使用原始数据生成Tafel图（建议启用IR补偿获得更准确结果）...") try: plt.figure(figsize=(8, 6)) plt.rcParams['font.sans-serif'] = [font_to_use] @@ -993,14 +1007,14 @@ def process_lsv(subfolder, file, params, project_id=None, enable_quality_check=T tafel_path = os.path.join(subfolder, f"{subname}_{file_stem}_Tafel_fit.png") plt.savefig(tafel_path, dpi=300, bbox_inches='tight') plt.close() - print(f"原始数据Tafel拟合图已保存: {tafel_path}") + log(f"原始数据Tafel拟合图已保存: {tafel_path}") except Exception as e: - print(f"导出原始Tafel图失败: {e}") + log(f"导出原始Tafel图失败: {e}") else: - print("警告：未找到可用的Tafel拟合数据！") - print("请确保：1) 勾选了'计算Tafel斜率' 2) 如需准确结果，建议同时启用IR补偿功能") + log("警告：未找到可用的Tafel拟合数据！") + log("请确保：1) 勾选了'计算Tafel斜率' 2) 如需准确结果，建议同时启用IR补偿功能") else: - print("未启用Tafel图导出功能") + log("未启用Tafel图导出功能") # 保存处理历史记录 if HISTORY_MANAGER_AVAILABLE: @@ -1181,7 +1195,7 @@ def process_lsv(subfolder, file, params, project_id=None, enable_quality_check=T pd.DataFrame(raw_records).to_excel(writer, sheet_name='raw', index=False) pd.DataFrame(tgt_records).to_excel(writer, sheet_name='targets', index=False) info_rows = [ - {'Key':'SoftwareVersion','Value':'2.3.2'}, + {'Key':'SoftwareVersion','Value': SOFTWARE_VERSION}, {'Key':'GeneratedAt','Value': datetime.now().isoformat(timespec='seconds')}, {'Key':'Sample','Value': subname}, {'Key':'SourceFile','Value': file}, diff --git a/src/electrochem_v6/core/processing_pipeline.py b/src/electrochem_v6/core/processing_pipeline.py index d7b437e..8e7505c 100644 --- a/src/electrochem_v6/core/processing_pipeline.py +++ b/src/electrochem_v6/core/processing_pipeline.py @@ -10,28 +10,13 @@ import numpy as np import pandas as pd +from electrochem_v6.core.utils import as_bool as _as_bool + def natural_sort_key(s): import re return [int(t) if t.isdigit() else t.lower() for t in re.split(r'(\d+)', str(s))] -def _as_bool(value, default=False): - if value is None: - return default - if isinstance(value, bool): - return value - if isinstance(value, (int, float)): - return value != 0 - if isinstance(value, str): - val = value.strip().lower() - if val in {'', 'none'}: - return default - if val in {'1', 'true', 'yes', 'y', 'on'}: - return True - if val in {'0', 'false', 'no', 'off', 'n'}: - return False - return bool(value) - def auto_detect_data_start(file_path: str, encodings: Optional[Sequence[str]] = None) -> int: """Auto-detect the first data line (1-based) in a text file.""" diff --git a/src/electrochem_v6/core/utils.py b/src/electrochem_v6/core/utils.py new file mode 100644 index 0000000..c3d75db --- /dev/null +++ b/src/electrochem_v6/core/utils.py @@ -0,0 +1,69 @@ +"""Shared type-coercion and I/O utilities used by multiple core modules.""" + +from __future__ import annotations + +import logging +from typing import Any, List, Optional, Sequence + +_logger = logging.getLogger(__name__) + +# Default fallback encoding list used across all processing modules. +DEFAULT_ENCODINGS: Sequence[str] = ("utf-8", "gbk", "gb2312", "ascii", "latin-1", "cp1252") + + +def read_file_with_fallback_encodings( + filepath: str, + *, + start_line: int = 1, + encodings: Sequence[str] | None = None, +) -> Optional[List[str]]: + """Read a text file trying multiple encodings in order. + + Returns the list of lines starting from *start_line* (1-based) or + ``None`` when all encodings fail. + """ + skip = max(0, int(start_line) - 1) + for enc in (encodings or DEFAULT_ENCODINGS): + try: + with open(filepath, "r", encoding=enc) as fh: + lines = fh.readlines()[skip:] + return lines + except UnicodeDecodeError: + continue + except (FileNotFoundError, PermissionError, OSError) as exc: + _logger.error("无法读取文件 %s: %s", filepath, exc) + raise + return None + + +def as_float(value: Any, default: float) -> float: + """Safely convert *value* to float, returning *default* on failure.""" + try: + return float(value) + except Exception: + return default + + +def as_int(value: Any, default: int) -> int: + """Safely convert *value* to int, returning *default* on failure.""" + try: + return int(value) + except Exception: + return default + + +def as_bool(value: Any, default: bool = False) -> bool: + """Coerce *value* to bool using common truthy/falsy strings.""" + if value is None: + return default + if isinstance(value, bool): + return value + if isinstance(value, (int, float)): + return value != 0 + if isinstance(value, str): + text = value.strip().lower() + if text in {"1", "true", "yes", "y", "on"}: + return True + if text in {"0", "false", "no", "n", "off", "", "none"}: + return False + return bool(value) diff --git a/src/electrochem_v6/llm/vision_client.py b/src/electrochem_v6/llm/vision_client.py index 229ca22..f26ed1b 100644 --- a/src/electrochem_v6/llm/vision_client.py +++ b/src/electrochem_v6/llm/vision_client.py @@ -46,8 +46,13 @@ def analyze_image(self, image_path: str, prompt: str, max_tokens: int = 800) -> return {"success": False, "error": str(exc), "type": exc.__class__.__name__} def _build_payload(self, image_path: str, prompt: str) -> Dict: - image_bytes = Path(image_path).read_bytes() - suffix = Path(image_path).suffix.lower().lstrip(".") or "png" + resolved = Path(image_path).resolve() + suffix = resolved.suffix.lower().lstrip(".") or "png" + if suffix not in ("png", "jpg", "jpeg", "gif", "bmp", "webp"): + raise ValueError(f"不支持的图像格式: {suffix}") + if not resolved.is_file(): + raise FileNotFoundError(f"图像文件不存在: {image_path}") + image_bytes = resolved.read_bytes() mime = f"image/{'jpeg' if suffix in ['jpg', 'jpeg'] else suffix}" data_url = f"data:{mime};base64,{base64.b64encode(image_bytes).decode('ascii')}" diff --git a/src/electrochem_v6/server/http_server.py b/src/electrochem_v6/server/http_server.py index 09e6f4e..70792b9 100644 --- a/src/electrochem_v6/server/http_server.py +++ b/src/electrochem_v6/server/http_server.py @@ -118,6 +118,7 @@ def _send_json(self, status_code: int, payload: Dict[str, Any]) -> None: self.send_response(response_status) self.send_header("Content-Type", "application/json; charset=utf-8") + self.send_header("X-Content-Type-Options", "nosniff") self.end_headers() self.wfile.write(response_body) log_event( @@ -149,6 +150,12 @@ def _send_static_file(self, file_path: Path) -> bool: content_type = "application/octet-stream" self.send_response(200) self.send_header("Content-Type", content_type) + self.send_header("X-Content-Type-Options", "nosniff") + if content_type == "text/html": + self.send_header( + "Content-Security-Policy", + "default-src 'self'; script-src 'self' 'unsafe-inline'; style-src 'self' 'unsafe-inline'; img-src 'self' data:;", + ) self.end_headers() self.wfile.write(resolved.read_bytes()) log_event( diff --git a/src/electrochem_v6/server/request_utils.py b/src/electrochem_v6/server/request_utils.py index a33cb65..5afc694 100644 --- a/src/electrochem_v6/server/request_utils.py +++ b/src/electrochem_v6/server/request_utils.py @@ -30,7 +30,7 @@ def read_json(handler: Any, max_json_body_bytes: int) -> Dict[str, Any]: try: payload = json.loads(body.decode("utf-8")) except json.JSONDecodeError as exc: - raise ValueError("??????? JSON") from exc + raise ValueError("请求体不是合法的 JSON") from exc if not isinstance(payload, dict): raise ValueError("JSON ??????????") return payload diff --git a/src/electrochem_v6/server/routes_get.py b/src/electrochem_v6/server/routes_get.py index 57d9af7..08fd36a 100644 --- a/src/electrochem_v6/server/routes_get.py +++ b/src/electrochem_v6/server/routes_get.py @@ -6,6 +6,15 @@ from typing import Any from urllib.parse import parse_qs, urlparse + +def _safe_int(raw: str, default: int, lo: int = 1, hi: int = 10000) -> int: + """Parse *raw* as int, clamping to [lo, hi]. Returns *default* on failure.""" + try: + return max(lo, min(int(raw), hi)) + except (ValueError, TypeError): + return default + + from electrochem_v6.core import ( build_project_lsv_compare_plot, export_project_report, @@ -59,9 +68,11 @@ def dispatch_get(handler: Any) -> bool: parts = path_parts(path) if len(parts) >= 5: project_id = parts[3] - page = int(query.get("page", ["1"])[0]) - page_size = int(query.get("page_size", ["15"])[0]) + page = _safe_int(query.get("page", ["1"])[0], 1) + page_size = _safe_int(query.get("page_size", ["15"])[0], 15, lo=1, hi=100) sort_by = query.get("sort", ["eta"])[0] + if sort_by not in ("eta", "tafel"): + sort_by = "eta" handler._send_json( 200, get_lsv_summary( @@ -98,6 +109,8 @@ def dispatch_get(handler: Any) -> bool: if text: selected_samples.append(text) chart_type = query.get("chart_type", ["overlay"])[0] + if chart_type not in ("overlay", "bar", "scatter", "radar"): + chart_type = "overlay" metric_key = query.get("metric", ["overpotential_10"])[0] target_current = query.get("target_current", ["10"])[0] payload = build_project_lsv_compare_plot( @@ -116,6 +129,8 @@ def dispatch_get(handler: Any) -> bool: if len(parts) >= 6: project_id = parts[3] chart_type = query.get("chart_type", ["overlay"])[0] + if chart_type not in ("overlay", "bar", "scatter", "radar"): + chart_type = "overlay" metric_key = query.get("metric", ["overpotential_10"])[0] target_current = query.get("target_current", ["10"])[0] payload = get_latest_project_lsv_compare_plot( @@ -151,7 +166,7 @@ def dispatch_get(handler: Any) -> bool: if path == "/api/v1/history": project_id = query.get("project", [None])[0] - limit = int(query.get("limit", ["100"])[0]) + limit = _safe_int(query.get("limit", ["100"])[0], 100, lo=1, hi=500) include_archived = (query.get("include_archived", ["0"])[0] or "").strip().lower() in {"1", "true", "yes"} handler._send_json(200, list_history(project_id=project_id, limit=limit, include_archived=include_archived)) return True @@ -177,8 +192,8 @@ def dispatch_get(handler: Any) -> bool: return True if path == "/api/v1/agent/conversations": - page = int(query.get("page", ["1"])[0]) - page_size = int(query.get("page_size", ["20"])[0]) + page = _safe_int(query.get("page", ["1"])[0], 1) + page_size = _safe_int(query.get("page_size", ["20"])[0], 20, lo=1, hi=100) filters = { "keyword": query.get("keyword", [""])[0], "project_name": query.get("project_name", [""])[0], diff --git a/src/electrochem_v6/server/routes_post.py b/src/electrochem_v6/server/routes_post.py index a576390..1a4c928 100644 --- a/src/electrochem_v6/server/routes_post.py +++ b/src/electrochem_v6/server/routes_post.py @@ -188,12 +188,17 @@ def dispatch_post(handler: Any, manager: Any) -> bool: ) handler._send_json(400, {"status": "error", "message": str(exc)}) return True - log_event(_LOGGER, "llm.config.update.request", {"path": path, "payload": payload}) + log_event(_LOGGER, "llm.config.update.request", { + "path": path, + "provider": payload.get("provider"), + "model": payload.get("model"), + # NOTE: payload 中可能包含 api_key，不记录完整 payload + }) result = update_provider(payload) log_event( _LOGGER, "llm.config.update.result", - {"path": path, "status": result.get("status"), "provider": result.get("provider"), "result": result}, + {"path": path, "status": result.get("status"), "provider": result.get("provider")}, level=logging.INFO if result.get("status") == "success" else logging.WARNING, ) handler._send_json(200 if result.get("status") == "success" else 400, result) @@ -292,6 +297,10 @@ def dispatch_post(handler: Any, manager: Any) -> bool: parts = path_parts(path) if len(parts) >= 6: template_name = unquote(parts[4]) + # Prevent path traversal in template name + if ".." in template_name or "/" in template_name or "\\" in template_name: + handler._send_json(400, {"status": "error", "message": "无效的模板名称"}) + return True result = delete_process_template(template_name) handler._send_json(200 if result.get("status") == "success" else 400, result) return True diff --git a/src/electrochem_v6/store/history.py b/src/electrochem_v6/store/history.py index c3c9a24..8f66ee7 100644 --- a/src/electrochem_v6/store/history.py +++ b/src/electrochem_v6/store/history.py @@ -3,10 +3,15 @@ from __future__ import annotations import json +import logging +import os +import shutil from typing import Any, Dict, Optional from .legacy_runtime import get_history_manager_v6 +_logger = logging.getLogger(__name__) + def _normalize_history_payload(payload: Any) -> Dict[str, Any]: if isinstance(payload, list): @@ -21,10 +26,17 @@ def _normalize_history_payload(payload: Any) -> Dict[str, Any]: def _write_history_payload(hist_mgr: Any, payload: Dict[str, Any]) -> None: safe_payload = hist_mgr._to_json_safe(payload) if hasattr(hist_mgr, "_to_json_safe") else payload + # Create backup before writing + history_file = str(hist_mgr.history_file) + if os.path.exists(history_file): + try: + shutil.copy2(history_file, history_file + ".bak") + except Exception: + _logger.warning("Failed to create history backup before write") if hasattr(hist_mgr, "_atomic_write_payload"): hist_mgr._atomic_write_payload(safe_payload) return - with open(hist_mgr.history_file, "w", encoding="utf-8") as f: + with open(history_file, "w", encoding="utf-8") as f: json.dump(safe_payload, f, ensure_ascii=False, indent=2) def _record_key(record: Dict[str, Any]) -> str: diff --git a/src/electrochem_v6/store/projects.py b/src/electrochem_v6/store/projects.py index f41f9b2..6d46607 100644 --- a/src/electrochem_v6/store/projects.py +++ b/src/electrochem_v6/store/projects.py @@ -3,7 +3,10 @@ from __future__ import annotations import json +import logging import os +import re +import shutil import tempfile import threading from datetime import datetime @@ -15,6 +18,27 @@ from .legacy_runtime import get_history_manager_v6, get_project_manager_v6 _PROJECTS_IO_LOCK = threading.RLock() +_logger = logging.getLogger(__name__) + +_MAX_PROJECT_NAME_LEN = 128 +_MAX_DESCRIPTION_LEN = 1024 +# Strip control characters (C0/C1) except common whitespace +_CONTROL_CHARS_RE = re.compile(r"[\x00-\x08\x0b\x0c\x0e-\x1f\x7f-\x9f]") + + +def _validate_project_name(name: str | None) -> tuple[str | None, str | None]: + """Return (clean_name, error_message). error_message is None on success.""" + clean = _CONTROL_CHARS_RE.sub("", str(name or "")).strip() + if not clean: + return None, "项目名称不能为空" + if len(clean) > _MAX_PROJECT_NAME_LEN: + return None, f"项目名称不能超过 {_MAX_PROJECT_NAME_LEN} 个字符" + return clean, None + + +def _sanitize_description(desc: str | None) -> str: + clean = _CONTROL_CHARS_RE.sub("", str(desc or "")).strip() + return clean[:_MAX_DESCRIPTION_LEN] def _safe_load_projects(projects_file: str) -> dict[str, Any]: @@ -23,13 +47,19 @@ def _safe_load_projects(projects_file: str) -> dict[str, Any]: parsed = json.load(f) if isinstance(parsed, dict): return parsed - except Exception: - pass + except Exception as exc: + _logger.warning("Failed to load projects file %s: %s", projects_file, exc) return {"version": "1.0", "projects": [], "default_project": None} def _atomic_write_json(path: str, data: dict[str, Any]) -> None: target = ensure_parent_dir(Path(path)) + # Create backup before writing + if target.exists(): + try: + shutil.copy2(str(target), str(target) + ".bak") + except Exception: + _logger.warning("Failed to create backup for %s", path) fd, tmp_path = tempfile.mkstemp(prefix=f"{target.stem}_", suffix=".tmp", dir=str(target.parent)) try: with os.fdopen(fd, "w", encoding="utf-8") as f: @@ -63,7 +93,8 @@ def _create_project_fallback( try: project_id = proj_mgr._generate_project_id() - except Exception: + except Exception as exc: + _logger.warning("Failed to generate project id via manager, using fallback: %s", exc) project_id = f"proj_fallback_{int(datetime.now().timestamp())}" now = datetime.now().strftime("%Y-%m-%d %H:%M:%S") if not color: @@ -89,7 +120,8 @@ def _create_project_fallback( try: _atomic_write_json(projects_file, data) return project_id - except Exception: + except Exception as exc: + _logger.error("Failed to write project file: %s", exc) return None @@ -109,8 +141,8 @@ def get_or_create_project_id_by_name( for proj in proj_mgr.get_all_projects("active"): if proj.get("name") == clean_name: return proj.get("id") - except Exception: - pass + except Exception as exc: + _logger.warning("Failed to lookup existing projects: %s", exc) return _create_project_fallback( proj_mgr, name=clean_name, @@ -130,8 +162,8 @@ def list_projects(status: str = "active") -> Dict[str, Any]: try: stats = proj_mgr.get_project_stats(item.get("id")) item["file_count"] = stats.get("total_files", item.get("file_count", 0)) - except Exception: - pass + except Exception as exc: + _logger.debug("Could not get stats for project %s: %s", item.get("id"), exc) safe_projects.append(item) return {"status": "success", "projects": safe_projects} @@ -142,12 +174,13 @@ def create_project( tags: Optional[List[str]] = None, color: Optional[str] = None, ) -> Dict[str, Any]: - clean_name = str(name or "").strip() - if not clean_name: - return {"status": "error", "message": "project name is required"} + clean_name, err = _validate_project_name(name) + if err: + return {"status": "error", "message": err} + clean_desc = _sanitize_description(description) project_id = get_or_create_project_id_by_name( clean_name, - description=str(description or "").strip(), + description=clean_desc, tags=tags or [], color=color, ) @@ -184,12 +217,12 @@ def update_project( return {"status": "error", "message": "missing project id"} payload: Dict[str, Any] = {} if name is not None: - clean_name = str(name).strip() - if not clean_name: - return {"status": "error", "message": "project name is required"} + clean_name, err = _validate_project_name(name) + if err: + return {"status": "error", "message": err} payload["name"] = clean_name if description is not None: - payload["description"] = str(description).strip() + payload["description"] = _sanitize_description(description) if tags is not None: payload["tags"] = [str(tag).strip() for tag in tags if str(tag).strip()] if color is not None: @@ -209,8 +242,8 @@ def update_project( stats = proj_mgr.get_project_stats(pid) project = dict(project) project["file_count"] = stats.get("total_files", project.get("file_count", 0)) - except Exception: - pass + except Exception as exc: + _logger.debug("Could not get stats for project %s: %s", pid, exc) return {"status": "success", "message": "project updated", "project_id": pid, "project": project} diff --git a/tests/test_v6_path_security.py b/tests/test_v6_path_security.py new file mode 100644 index 0000000..72daf03 --- /dev/null +++ b/tests/test_v6_path_security.py @@ -0,0 +1,120 @@ +"""Tests for electrochem_v6.core.path_security module.""" + +import os +import pytest +from pathlib import Path + +from electrochem_v6.core.path_security import ( + sanitize_filename, + validate_path_within, + is_safe_data_path, + is_safe_image_path, +) + + +# ── sanitize_filename ────────────────────────────────────────────────────── + +class TestSanitizeFilename: + def test_normal_name(self): + assert sanitize_filename("report.csv") == "report.csv" + + def test_strips_directory_components(self): + assert sanitize_filename("../../etc/passwd") == "passwd" + + def test_strips_windows_path(self): + assert sanitize_filename("C:\\Users\\hack\\evil.exe") == "evil.exe" + + def test_replaces_special_chars(self): + result = sanitize_filename("file name@#$.txt") + assert ".." not in result + assert "/" not in result + assert "\\" not in result + + def test_empty_returns_unknown(self): + assert sanitize_filename("") == "unknown" + + def test_none_returns_unknown(self): + assert sanitize_filename(None) == "unknown" # type: ignore[arg-type] + + def test_dots_only(self): + result = sanitize_filename("...") + assert result == "unknown" + + def test_leading_dot_stripped(self): + result = sanitize_filename(".hidden") + assert not result.startswith(".") + + def test_traversal_with_encoded_slashes(self): + result = sanitize_filename("..%2F..%2Fetc%2Fpasswd") + assert "/" not in result + assert "\\" not in result + + def test_unicode_name(self): + result = sanitize_filename("数据_报告.csv") + # Should not crash; result should be non-empty + assert result and result != "unknown" + + +# ── validate_path_within ─────────────────────────────────────────────────── + +class TestValidatePathWithin: + def test_valid_path(self, tmp_path): + f = tmp_path / "data.csv" + f.write_text("x") + result = validate_path_within(str(f), tmp_path) + assert result == f.resolve() + + def test_rejects_traversal(self, tmp_path): + evil = str(tmp_path / ".." / ".." / "etc" / "passwd") + with pytest.raises(ValueError, match="不在允许范围内"): + validate_path_within(evil, tmp_path) + + def test_rejects_nonexistent(self, tmp_path): + with pytest.raises(ValueError, match="不存在"): + validate_path_within(str(tmp_path / "nope.csv"), tmp_path) + + def test_allows_nonexistent_when_flag_off(self, tmp_path): + result = validate_path_within( + str(tmp_path / "future.csv"), tmp_path, must_exist=False + ) + assert result.name == "future.csv" + + def test_rejects_bad_extension(self, tmp_path): + f = tmp_path / "evil.py" + f.write_text("import os") + with pytest.raises(ValueError, match="不允许的文件类型"): + validate_path_within(str(f), tmp_path, allowed_extensions=[".csv", ".txt"]) + + def test_accepts_good_extension(self, tmp_path): + f = tmp_path / "data.csv" + f.write_text("a,b") + result = validate_path_within(str(f), tmp_path, allowed_extensions=[".csv"]) + assert result.suffix == ".csv" + + +# ── is_safe_data_path / is_safe_image_path ──────────────────────────────── + +class TestSafePathCheckers: + def test_safe_data_path(self, tmp_path): + f = tmp_path / "data.csv" + f.write_text("ok") + assert is_safe_data_path(str(f), tmp_path) is True + + def test_unsafe_data_path_traversal(self, tmp_path): + evil = str(tmp_path / ".." / ".." / "etc" / "passwd") + assert is_safe_data_path(evil, tmp_path) is False + + def test_unsafe_data_path_extension(self, tmp_path): + f = tmp_path / "script.py" + f.write_text("x") + assert is_safe_data_path(str(f), tmp_path) is False + + def test_safe_image_path(self, tmp_path): + f = tmp_path / "plot.png" + f.write_bytes(b"\x89PNG") + assert is_safe_image_path(str(f), tmp_path) is True + + def test_unsafe_image_extension(self, tmp_path): + f = tmp_path / "data.csv" + f.write_text("ok") + assert is_safe_image_path(str(f), tmp_path) is False diff --git a/tests/test_v6_projects_concurrency.py b/tests/test_v6_projects_concurrency.py index 9fd3c53..f9751a4 100644 --- a/tests/test_v6_projects_concurrency.py +++ b/tests/test_v6_projects_concurrency.py @@ -10,7 +10,8 @@ if str(SRC) not in sys.path: sys.path.insert(0, str(SRC)) -from electrochem_v6.store.projects import get_or_create_project_id_by_name # noqa: E402 +from electrochem_v6.store.projects import get_or_create_project_id_by_name, create_project # noqa: E402 +from electrochem_v6.store.projects import _validate_project_name, _sanitize_description # noqa: E402 def test_v6_get_or_create_project_concurrent_same_name(tmp_path): @@ -49,3 +50,55 @@ def test_v6_get_or_create_project_concurrent_same_name(tmp_path): else: os.environ["ELECTROCHEM_V6_CONVERSATION_FILE"] = old_conv + +# ── Project input validation tests ───────────────────────────────────────── + +class TestValidateProjectName: + def test_normal_name(self): + name, err = _validate_project_name("My Project") + assert name == "My Project" + assert err is None + + def test_empty_name(self): + name, err = _validate_project_name("") + assert name is None + assert err is not None + + def test_none_name(self): + name, err = _validate_project_name(None) + assert name is None + assert err is not None + + def test_whitespace_only(self): + name, err = _validate_project_name(" ") + assert name is None + assert err is not None + + def test_control_chars_stripped(self): + name, err = _validate_project_name("proj\x00ect\x07\x1fname") + assert name == "projectname" + assert err is None + + def test_too_long(self): + name, err = _validate_project_name("a" * 200) + assert name is None + assert "128" in err + + def test_unicode_ok(self): + name, err = _validate_project_name("电化学项目-1") + assert name == "电化学项目-1" + assert err is None + + +class TestSanitizeDescription: + def test_normal(self): + assert _sanitize_description("A description") == "A description" + + def test_control_chars(self): + assert _sanitize_description("foo\x00bar\x1f") == "foobar" + + def test_truncated(self): + assert len(_sanitize_description("x" * 2000)) == 1024 + + def test_none(self): + assert _sanitize_description(None) == "" diff --git a/tests/test_v6_utils.py b/tests/test_v6_utils.py new file mode 100644 index 0000000..6592bd6 --- /dev/null +++ b/tests/test_v6_utils.py @@ -0,0 +1,111 @@ +"""Tests for electrochem_v6.core.utils module.""" + +import os +import pytest + +from electrochem_v6.core.utils import ( + as_bool, + as_float, + as_int, + read_file_with_fallback_encodings, +) + + +# ── as_float ─────────────────────────────────────────────────────────────── + +class TestAsFloat: + def test_valid_string(self): + assert as_float("3.14", 0.0) == pytest.approx(3.14) + + def test_int_input(self): + assert as_float(42, 0.0) == 42.0 + + def test_invalid_returns_default(self): + assert as_float("abc", -1.0) == -1.0 + + def test_none_returns_default(self): + assert as_float(None, 5.5) == 5.5 + + +# ── as_int ───────────────────────────────────────────────────────────────── + +class TestAsInt: + def test_valid(self): + assert as_int("10", 0) == 10 + + def test_float_string_truncates(self): + assert as_int(3.9, 0) == 3 + + def test_garbage(self): + assert as_int("xyz", 99) == 99 + + +# ── as_bool ──────────────────────────────────────────────────────────────── + +class TestAsBool: + @pytest.mark.parametrize("val", ["1", "true", "True", "YES", "on", "y"]) + def test_truthy_strings(self, val): + assert as_bool(val) is True + + @pytest.mark.parametrize("val", ["0", "false", "False", "NO", "off", "n", "", "none"]) + def test_falsy_strings(self, val): + assert as_bool(val) is False + + def test_none_returns_default(self): + assert as_bool(None, True) is True + assert as_bool(None, False) is False + + def test_bool_passthrough(self): + assert as_bool(True) is True + assert as_bool(False) is False + + def test_numeric(self): + assert as_bool(1) is True + assert as_bool(0) is False + assert as_bool(0.0) is False + + +# ── read_file_with_fallback_encodings ────────────────────────────────────── + +class TestReadFileWithFallbackEncodings: + def test_reads_utf8(self, tmp_path): + f = tmp_path / "data.txt" + f.write_text("line1\nline2\nline3\n", encoding="utf-8") + lines = read_file_with_fallback_encodings(str(f)) + assert len(lines) == 3 + assert lines[0].strip() == "line1" + + def test_start_line_skips(self, tmp_path): + f = tmp_path / "data.txt" + f.write_text("header\nrow1\nrow2\n", encoding="utf-8") + lines = read_file_with_fallback_encodings(str(f), start_line=2) + assert len(lines) == 2 + assert lines[0].strip() == "row1" + + def test_gbk_fallback(self, tmp_path): + f = tmp_path / "data.txt" + # Write GBK-encoded Chinese text + f.write_bytes("你好世界\n第二行\n".encode("gbk")) + lines = read_file_with_fallback_encodings(str(f)) + assert lines is not None + assert len(lines) == 2 + + def test_nonexistent_file_raises(self, tmp_path): + with pytest.raises(FileNotFoundError): + read_file_with_fallback_encodings(str(tmp_path / "nope.txt")) + + def test_all_encodings_fail_returns_none(self, tmp_path): + # Binary file that can't be decoded + f = tmp_path / "binary.dat" + f.write_bytes(bytes(range(128, 256)) * 10) + # Use only a restrictive encoding list that will fail + result = read_file_with_fallback_encodings( + str(f), encodings=("ascii",) + ) + assert result is None + + def test_empty_file(self, tmp_path): + f = tmp_path / "empty.txt" + f.write_text("") + lines = read_file_with_fallback_encodings(str(f)) + assert lines == []