"""
回饋處理工具模組
Feedback processing utilities
"""
import os
import json
import base64
from typing import List, Dict, Any, Optional
from mcp.types import ImageContent, TextContent
from .resource_manager import create_temp_file
from ..debug import server_debug_log as debug_log
from .error_handler import ErrorHandler, ErrorType
def save_feedback_to_file(feedback_data: dict, file_path: Optional[str] = None) -> str:
"""
將回饋資料儲存到 JSON 文件
Args:
feedback_data: 回饋資料字典
file_path: 儲存路徑，若為 None 則自動產生臨時文件
Returns:
str: 儲存的文件路徑
"""
if file_path is None:
# 使用資源管理器創建臨時文件
file_path = create_temp_file(suffix=".json", prefix="feedback_")
# 確保目錄存在
directory = os.path.dirname(file_path)
if directory and not os.path.exists(directory):
os.makedirs(directory, exist_ok=True)
# 複製數據以避免修改原始數據
json_data = feedback_data.copy()
# 處理圖片數據：將 bytes 轉換為 base64 字符串以便 JSON 序列化
if "images" in json_data and isinstance(json_data["images"], list):
processed_images = []
for img in json_data["images"]:
if isinstance(img, dict) and "data" in img:
processed_img = img.copy()
# 如果 data 是 bytes，轉換為 base64 字符串
if isinstance(img["data"], bytes):
processed_img["data"] = base64.b64encode(img["data"]).decode(
"utf-8"
)
processed_img["data_type"] = "base64"
processed_images.append(processed_img)
else:
processed_images.append(img)
json_data["images"] = processed_images
# 儲存資料
with open(file_path, "w", encoding="utf-8") as f:
json.dump(json_data, f, ensure_ascii=False, indent=2)
debug_log(f"回饋資料已儲存至: {file_path}")
return file_path
def create_feedback_text(feedback_data: dict) -> str:
"""
建立格式化的回饋文字
Args:
feedback_data: 回饋資料字典
Returns:
str: 格式化後的回饋文字
"""
text_parts = []
# 基本回饋內容
if feedback_data.get("interactive_feedback"):
text_parts.append(f"=== 用戶回饋 ===\n{feedback_data['interactive_feedback']}")
# 命令執行日誌
if feedback_data.get("command_logs"):
text_parts.append(f"=== 命令執行日誌 ===\n{feedback_data['command_logs']}")
# 圖片附件概要
if feedback_data.get("images"):
images = feedback_data["images"]
text_parts.append(f"=== 圖片附件概要 ===\n用戶提供了 {len(images)} 張圖片：")
for i, img in enumerate(images, 1):
size = img.get("size", 0)
name = img.get("name", "unknown")
# 智能單位顯示
if size < 1024:
size_str = f"{size} B"
elif size < 1024 * 1024:
size_kb = size / 1024
size_str = f"{size_kb:.1f} KB"
else:
size_mb = size / (1024 * 1024)
size_str = f"{size_mb:.1f} MB"
img_info = f" {i}. {name} ({size_str})"
# 為提高兼容性，添加 base64 預覽信息
if img.get("data"):
try:
if isinstance(img["data"], bytes):
img_base64 = base64.b64encode(img["data"]).decode("utf-8")
elif isinstance(img["data"], str):
img_base64 = img["data"]
else:
img_base64 = None
if img_base64:
# 只顯示前50個字符的預覽
preview = (
img_base64[:50] + "..."
if len(img_base64) > 50
else img_base64
)
img_info += f"\n Base64 預覽: {preview}"
img_info += f"\n 完整 Base64 長度: {len(img_base64)} 字符"
# 如果客戶端不支援 MCP 圖片，可以提供完整 base64
debug_log(f"圖片 {i} Base64 已準備，長度: {len(img_base64)}")
# 檢查是否啟用 Base64 詳細模式（從 UI 設定中獲取）
include_full_base64 = feedback_data.get("settings", {}).get(
"enable_base64_detail", False
)
if include_full_base64:
# 根據檔案名推斷 MIME 類型
file_name = img.get("name", "image.png")
if file_name.lower().endswith((".jpg", ".jpeg")):
mime_type = "image/jpeg"
elif file_name.lower().endswith(".gif"):
mime_type = "image/gif"
elif file_name.lower().endswith(".webp"):
mime_type = "image/webp"
else:
mime_type = "image/png"
img_info += f"\n 完整 Base64: data:{mime_type};base64,{img_base64}"
except Exception as e:
debug_log(f"圖片 {i} Base64 處理失敗: {e}")
text_parts.append(img_info)
# 添加兼容性說明
text_parts.append(
"\n💡 注意：如果客戶端無法顯示圖片，圖片數據已包含在上述 Base64 信息中。"
)
return "\n\n".join(text_parts) if text_parts else "用戶未提供任何回饋內容。"
def process_images(images_data: List[Dict[str, Any]]) -> List[ImageContent]:
"""
處理圖片資料，轉換為 MCP 圖片內容對象
Args:
images_data: 圖片資料列表
Returns:
List[ImageContent]: MCP 圖片內容對象列表
"""
image_contents = []
for i, img in enumerate(images_data, 1):
try:
if not img.get("data"):
debug_log(f"圖片 {i} 沒有資料，跳過")
continue
# 檢查數據類型並相應處理
if isinstance(img["data"], bytes):
# 如果是原始 bytes 數據，轉換為 base64
image_base64 = base64.b64encode(img["data"]).decode("utf-8")
debug_log(
f"圖片 {i} 從 bytes 轉換為 base64，大小: {len(img['data'])} bytes"
)
elif isinstance(img["data"], str):
# 如果已經是 base64 字符串，直接使用
image_base64 = img["data"]
debug_log(f"圖片 {i} 已是 base64 字符串")
else:
debug_log(f"圖片 {i} 數據類型不支援: {type(img['data'])}")
continue
if len(image_base64) == 0:
debug_log(f"圖片 {i} 數據為空，跳過")
continue
# 根據文件名推斷 MIME 類型
file_name = img.get("name", "image.png")
if file_name.lower().endswith((".jpg", ".jpeg")):
mime_type = "image/jpeg"
elif file_name.lower().endswith(".gif"):
mime_type = "image/gif"
elif file_name.lower().endswith(".webp"):
mime_type = "image/webp"
elif file_name.lower().endswith(".bmp"):
mime_type = "image/bmp"
else:
mime_type = "image/png" # 默認使用 PNG
# 創建 ImageContent 對象
image_content = ImageContent(
type="image",
data=image_base64,
mimeType=mime_type
)
image_contents.append(image_content)
debug_log(f"圖片 {i} ({file_name}) 處理成功，MIME 類型: {mime_type}")
except Exception as e:
# 使用統一錯誤處理（不影響 JSON RPC）
error_id = ErrorHandler.log_error_with_context(
e,
context={"operation": "圖片處理", "image_index": i},
error_type=ErrorType.FILE_IO,
)
debug_log(f"圖片 {i} 處理失敗 [錯誤ID: {error_id}]: {e}")
debug_log(f"共處理 {len(image_contents)} 張圖片")
return image_contents
