import os import pandas as pd from pydantic import BaseModel, Field from typing import Optional from datetime import datetime class Action: class Valves(BaseModel): pass class UserValves(BaseModel): show_status: bool = Field( default=True, description="Show status of the action." ) pass def __init__(self): self.valves = self.Valves() pass async def action( self, body: dict, __user__=None, __event_emitter__=None, __event_call__=None, ) -> Optional[dict]: print(f"action:{__name__}") user_valves = __user__.get("valves") if not user_valves: user_valves = self.UserValves() if __event_emitter__: last_assistant_message = body["messages"][-1] timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") content = last_assistant_message['content'] if user_valves.show_status: await __event_emitter__( { "type": "status", "data": {"description": "Saving to Excel", "done": False}, } ) try: directory = "/app/data" if not os.path.exists(directory): os.makedirs(directory) file_path = os.path.join(directory, "chat_history.xlsx") # 如果文件存在，则加载并追加内容；如果不存在，则创建新的文件 if os.path.exists(file_path): df = pd.read_excel(file_path) else: df = pd.DataFrame(columns=["时间戳", "对话内容"]) # 将新记录追加到DataFrame new_record = { "时间戳": timestamp, "对话内容": content } df = df.append(new_record, ignore_index=True) # 保存为Excel df.to_excel(file_path, index=False) print("Output saved to Excel file in the container, accessible on the host.") except Exception as e: print(f"Error saving output to Excel file: {str(e)}") if user_valves.show_status: await __event_emitter__( { "type": "status", "data": { "description": "Error Saving to Excel", "done": True, }, } ) if user_valves.show_status: await __event_emitter__( { "type": "status", "data": {"description": "Output Saved to Excel", "done": True}, } )