"""
MCP 資源實作 - 記帳資料資源
實作所有記帳相關的 MCP Resources：
- transactions://all - 所有交易記錄
- categories://list - 分類清單
- summary://current - 目前彙總資訊
"""
from typing import Dict, Any, List, Optional
import json
from datetime import datetime
from .storage import StorageManager
class AccountingResources:
"""記帳資源類 - 實作所有 MCP Resources"""
def __init__(self, storage_manager: Optional[StorageManager] = None):
self.storage = storage_manager or StorageManager()
async def get_resource(self, uri: str) -> Dict[str, Any]:
"""
根據 URI 取得資源內容
Args:
uri: 資源 URI
Returns:
Dict containing resource content and metadata
"""
try:
if uri == "transactions://all":
return await self._get_all_transactions()
elif uri == "categories://list":
return await self._get_categories_list()
elif uri == "summary://current":
return await self._get_current_summary()
else:
return {
"success": False,
"error": f"未知的資源 URI: {uri}",
"error_type": "invalid_uri"
}
except Exception as e:
return {
"success": False,
"error": f"取得資源失敗: {str(e)}",
"error_type": "resource_error"
}
async def _get_all_transactions(self) -> Dict[str, Any]:
"""取得所有交易記錄資源"""
transactions = self.storage.transactions.load_transactions()
# 轉換為資源格式
transactions_data = []
for t in transactions:
transactions_data.append({
"id": t.id,
"amount": float(t.amount),
"category": t.category,
"description": t.description,
"date": t.date.isoformat(),
"timestamp": t.timestamp.isoformat(),
"type": t.transaction_type.value
})
# 計算統計資訊
total_income = sum(t["amount"] for t in transactions_data if t["amount"] > 0)
total_expense = sum(abs(t["amount"]) for t in transactions_data if t["amount"] < 0)
return {
"success": True,
"content": json.dumps({
"transactions": transactions_data,
"metadata": {
"total_count": len(transactions_data),
"total_income": total_income,
"total_expense": total_expense,
"date_range": {
"earliest": min(t["date"] for t in transactions_data) if transactions_data else None,
"latest": max(t["date"] for t in transactions_data) if transactions_data else None
},
"generated_at": datetime.now().isoformat()
}
}, indent=2, ensure_ascii=False),
"mimeType": "application/json"
}
async def _get_categories_list(self) -> Dict[str, Any]:
"""取得分類清單資源"""
categories = self.storage.categories.load_categories()
# 轉換為資源格式
categories_data = []
for cat in categories:
categories_data.append({
"id": cat.id,
"name": cat.name,
"description": cat.description,
"type": cat.category_type.value,
"color": cat.color,
"icon": cat.icon
})
# 按類型分組
expense_categories = [cat for cat in categories_data if cat["type"] == "expense"]
income_categories = [cat for cat in categories_data if cat["type"] == "income"]
return {
"success": True,
"content": json.dumps({
"categories": {
"all": categories_data,
"by_type": {
"expense": expense_categories,
"income": income_categories
}
},
"metadata": {
"total_count": len(categories_data),
"expense_count": len(expense_categories),
"income_count": len(income_categories),
"generated_at": datetime.now().isoformat()
}
}, indent=2, ensure_ascii=False),
"mimeType": "application/json"
}
async def _get_current_summary(self) -> Dict[str, Any]:
"""取得目前彙總資訊資源"""
# 取得帳戶資訊
account = self.storage.account.load_account()
# 取得本月彙總
now = datetime.now()
monthly_summary = self.storage.get_monthly_summary(now.year, now.month)
# 取得最近交易
recent_transactions = self.storage.transactions.get_transactions(limit=5)
recent_data = []
for t in recent_transactions:
recent_data.append({
"id": t.id,
"amount": float(t.amount),
"category": t.category,
"description": t.description,
"date": t.date.isoformat(),
"type": t.transaction_type.value
})
# 取得分類資訊用於統計
categories = self.storage.categories.load_categories()
category_names = {cat.id: cat.name for cat in categories}
return {
"success": True,
"content": json.dumps({
"account_summary": {
"balance": float(account.balance),
"total_transactions": account.total_transactions,
"last_updated": account.last_updated.isoformat() if account.last_updated else None,
"created_at": account.created_at.isoformat() if account.created_at else None
},
"monthly_summary": {
"year": now.year,
"month": now.month,
"month_name": now.strftime("%B"),
"total_income": monthly_summary["total_income"],
"total_expense": monthly_summary["total_expense"],
"net_flow": monthly_summary["net_flow"],
"transaction_count": monthly_summary["transaction_count"],
"category_breakdown": monthly_summary["category_breakdown"]
},
"recent_transactions": recent_data,
"quick_stats": {
"avg_transaction": float(account.balance / account.total_transactions) if account.total_transactions > 0 else 0,
"transactions_this_month": monthly_summary["transaction_count"],
"days_since_last_transaction": (datetime.now().date() - recent_transactions[0].date).days if recent_transactions else None
},
"metadata": {
"generated_at": datetime.now().isoformat(),
"currency": "TWD",
"timezone": "Asia/Shanghai"
}
}, indent=2, ensure_ascii=False),
"mimeType": "application/json"
}
def get_resource_definitions() -> List[Dict[str, Any]]:
"""回傳所有資源的 MCP 定義"""
return [
{
"uri": "transactions://all",
"name": "所有交易記錄",
"description": "包含所有交易記錄的完整資料集，包括統計中繼資料",
"mimeType": "application/json"
},
{
"uri": "categories://list",
"name": "分類清單",
"description": "所有可用的收入與支出分類定義，包括圖示與顏色資訊",
"mimeType": "application/json"
},
{
"uri": "summary://current",
"name": "目前彙總",
"description": "帳戶餘額、本月統計、最近交易等彙總資訊",
"mimeType": "application/json"
}
]
