local_tools.py

import os
import pandas as pd
import json

# 全局上下文，用于存储当前用户名等信息，由调用方设置
CONTEXT = {
    "username": "default_user"
}

def get_user_workspace():
    """获取当前用户的专属工作目录"""
    base_dir = "user_workspaces"
    username = CONTEXT.get("username", "unknown_user")
    # 简单的清理，防止路径注入
    safe_username = "".join(c for c in username if c.isalnum() or c in ('-', '_'))
    workspace = os.path.join(base_dir, safe_username)
    
    if not os.path.exists(workspace):
        try:
            os.makedirs(workspace)
        except OSError as e:
            # 如果并发创建可能会报错，忽略
            pass
            
    return workspace

def get_safe_path(filepath: str) -> str:
    """
    将相对路径转换为用户工作区内的绝对路径，并进行安全检查。
    如果路径不安全（试图逃逸），抛出异常。
    """
    workspace = os.path.abspath(get_user_workspace())
    # 移除路径中的 .. 防止跳转，并确保是相对路径
    filepath = filepath.lstrip("/\\")
    
    # 构建绝对路径
    target_path = os.path.abspath(os.path.join(workspace, filepath))
    
    # 检查目标路径是否以工作区路径开头（防止 ../../../etc/passwd 攻击）
    if not target_path.startswith(workspace):
        raise ValueError(f"Access denied: Path '{filepath}' attempts to escape user workspace.")
        
    return target_path

# --- 本地工具函数 ---

def local_read_file(filepath: str) -> str:
    """读取本地文件内容 (限制在用户目录下)"""
    try:
        full_path = get_safe_path(filepath)
        
        if not os.path.exists(full_path):
            return f"Error: File '{filepath}' not found in workspace."
            
        # 根据扩展名判断读取方式
        _, ext = os.path.splitext(full_path)
        ext = ext.lower()
        
        if ext in ['.xlsx', '.xls']:
            # 读取 Excel 返回 JSON 字符串
            try:
                df = pd.read_excel(full_path)
                return df.to_json(orient="records", force_ascii=False)
            except Exception as e:
                return f"Error reading Excel file: {str(e)}"
        else:
            # 默认尝试文本读取
            with open(full_path, "r", encoding="utf-8") as f:
                return f.read()
                
    except ValueError as ve:
        return str(ve)
    except Exception as e:
        return f"Error reading file: {str(e)}"

def local_write_file(filepath: str, content: str) -> str:
    """在本地创建或覆盖写入文件 (限制在用户目录下)"""
    try:
        full_path = get_safe_path(filepath)
        
        # 确保父目录存在
        directory = os.path.dirname(full_path)
        if not os.path.exists(directory):
            os.makedirs(directory)

        # 根据扩展名判断写入方式
        _, ext = os.path.splitext(full_path)
        ext = ext.lower()

        if ext in ['.xlsx', '.xls']:
            # 尝试将 content 解析为 JSON 或 CSV 格式数据写入 Excel
            try:
                # 假设 content 是 JSON 字符串 (list of dicts)
                data = json.loads(content)
                df = pd.DataFrame(data)
                df.to_excel(full_path, index=False)
                return f"Success: Excel file '{filepath}' created with {len(df)} records."
            except json.JSONDecodeError:
                # 如果不是 JSON，尝试直接当文本处理（不太合理，但作为fallback）
                # 或者如果 content 是 CSV 格式
                try:
                    from io import StringIO
                    csv_io = StringIO(content)
                    df = pd.read_csv(csv_io)
                    df.to_excel(full_path, index=False)
                    return f"Success: Excel file '{filepath}' created from CSV data."
                except Exception as e:
                     return f"Error creating Excel: Content must be valid JSON list or CSV format. Details: {e}"
        else:
            # 默认文本写入
            with open(full_path, "w", encoding="utf-8") as f:
                f.write(content)
            return f"Success: File '{filepath}' written."
            
    except ValueError as ve:
        return str(ve)
    except Exception as e:
        return f"Error writing file: {str(e)}"

# --- 工具定义 (OpenAI Schema) ---
LOCAL_TOOLS_SCHEMA = [
    {
        "type": "function",
        "function": {
            "name": "read_file",
            "description": "Read content from a file in the user's workspace. Supports .txt, .md, .py, .json, .xlsx (returns JSON).",
            "parameters": {
                "type": "object",
                "properties": {
                    "filepath": {
                        "type": "string",
                        "description": "The relative path of the file to read (e.g., 'data.xlsx' or 'logs/error.txt')"
                    }
                },
                "required": ["filepath"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "write_file",
            "description": "Create or overwrite a file in the user's workspace. For .xlsx, content must be a JSON string (list of objects) or CSV format.",
            "parameters": {
                "type": "object",
                "properties": {
                    "filepath": {
                        "type": "string",
                        "description": "The relative path of the file to write (e.g., 'report.xlsx')"
                    },
                    "content": {
                        "type": "string",
                        "description": "The content to write. For Excel files, provide a JSON string representing a list of records."
                    }
                },
                "required": ["filepath", "content"]
            }
        }
    }
]