"""
儲存層 - JSON 檔案儲存實作
提供基於 JSON 檔案的資料持久化功能，包括：
- JSONStorage: 基礎 JSON 檔案儲存
- TransactionStorage: 交易記錄儲存
- AccountStorage: 帳戶資訊儲存
- CategoryStorage: 分類資訊儲存
- 原子寫入、備份復原、併發控制等特性
"""
import json
import os
import shutil
import threading
from pathlib import Path
from typing import Dict, Any, List, Optional, Union
from datetime import date, datetime
from decimal import Decimal
from .models import Transaction, Account, Category, create_default_categories
class StorageError(Exception):
"""儲存相關例外"""
pass
class JSONStorage:
"""基礎 JSON 儲存類"""
def __init__(self, file_path: Union[str, Path], default_data: Optional[Dict[str, Any]] = None):
self.file_path = Path(file_path)
self.default_data = default_data or {}
self._lock = threading.RLock() # 可重入鎖，防止併發問題
# 確保目錄存在
self.file_path.parent.mkdir(parents=True, exist_ok=True)
# 如果檔案不存在，建立預設資料
if not self.file_path.exists():
self.save_data(self.default_data)
def load_data(self) -> Dict[str, Any]:
"""載入資料，支援備份復原"""
with self._lock:
try:
return self._load_from_file(self.file_path)
except (json.JSONDecodeError, OSError) as e:
# 檔案損壞或不可讀，嘗試從備份復原
backup_path = self._get_backup_path()
if backup_path.exists():
try:
data = self._load_from_file(backup_path)
# 復原主檔案
self._write_file(self.file_path, data)
return data
except Exception:
pass
# 如果備份也失敗，回傳預設資料
return self.default_data.copy()
def save_data(self, data: Dict[str, Any]) -> None:
"""儲存資料，使用原子寫入"""
with self._lock:
try:
# 檢查檔案權限（如果檔案存在）
if self.file_path.exists() and not os.access(self.file_path, os.W_OK):
raise StorageError("權限錯誤: 檔案唯讀，無法寫入")
# 先建立備份（如果原檔案存在）
if self.file_path.exists():
self._create_backup()
# 原子寫入
self._atomic_write(self.file_path, data)
except Exception as e:
raise StorageError(f"儲存資料失敗: {e}")
def _load_from_file(self, file_path: Path) -> Dict[str, Any]:
"""從檔案載入 JSON 資料"""
with open(file_path, 'r', encoding='utf-8') as f:
return json.load(f)
def _write_file(self, file_path: Path, data: Dict[str, Any]) -> None:
"""寫入 JSON 資料到檔案"""
with open(file_path, 'w', encoding='utf-8') as f:
json.dump(data, f, indent=2, ensure_ascii=False)
def _atomic_write(self, file_path: Path, data: Dict[str, Any]) -> None:
"""原子寫入檔案"""
temp_path = file_path.with_suffix(file_path.suffix + '.tmp')
try:
# 寫入暫存檔案
self._write_file(temp_path, data)
# 原子重新命名
temp_path.replace(file_path)
except PermissionError as e:
# 清理暫存檔案
if temp_path.exists():
temp_path.unlink()
raise StorageError(f"權限錯誤: {e}")
except Exception as e:
# 清理暫存檔案
if temp_path.exists():
temp_path.unlink()
raise StorageError(f"原子寫入失敗: {e}")
def _create_backup(self) -> None:
"""建立備份檔案"""
if self.file_path.exists():
backup_path = self._get_backup_path()
shutil.copy2(self.file_path, backup_path)
def _get_backup_path(self) -> Path:
"""取得備份檔案路徑"""
# 正確的方法：使用 with_name 替換整個檔名
stem = self.file_path.stem # 檔名（不含副檔名）
suffix = self.file_path.suffix # 副檔名（如 .json）
backup_name = f"{stem}_backup{suffix}"
return self.file_path.with_name(backup_name)
class TransactionStorage:
"""交易記錄儲存"""
def __init__(self, file_path: Union[str, Path]):
self.storage = JSONStorage(file_path, {"transactions": []})
self._lock = threading.RLock() # 新增鎖用於併發控制
def load_transactions(self) -> List[Transaction]:
"""載入所有交易記錄"""
data = self.storage.load_data()
transactions = []
for trans_data in data.get("transactions", []):
try:
transactions.append(Transaction.from_dict(trans_data))
except Exception:
# 忽略損壞的記錄
continue
# 按時間戳倒序排序（最新的在前）
transactions.sort(key=lambda t: t.timestamp, reverse=True)
return transactions
def save_transactions(self, transactions: List[Transaction]) -> None:
"""儲存交易記錄清單"""
trans_data = [t.to_dict() for t in transactions]
self.storage.save_data({"transactions": trans_data})
def add_transaction(self, transaction: Transaction) -> None:
"""新增單筆交易（執行緒安全）"""
with self._lock:
transactions = self.load_transactions()
transactions.append(transaction)
self.save_transactions(transactions)
def get_transactions_by_category(self, category: str) -> List[Transaction]:
"""按分類篩選交易"""
transactions = self.load_transactions()
return [t for t in transactions if t.category == category]
def get_transactions_by_date_range(
self,
start_date: date,
end_date: date
) -> List[Transaction]:
"""按日期範圍篩選交易"""
transactions = self.load_transactions()
return [
t for t in transactions
if start_date <= t.date <= end_date
]
def get_transactions(
self,
limit: int = 20,
offset: int = 0,
category: Optional[str] = None,
start_date: Optional[date] = None,
end_date: Optional[date] = None
) -> List[Transaction]:
"""取得交易記錄（支援分頁與篩選）"""
transactions = self.load_transactions()
# 套用篩選條件
if category:
transactions = [t for t in transactions if t.category == category]
if start_date:
transactions = [t for t in transactions if t.date >= start_date]
if end_date:
transactions = [t for t in transactions if t.date <= end_date]
# 套用分頁
return transactions[offset:offset + limit]
class AccountStorage:
"""帳戶資訊儲存"""
def __init__(self, file_path: Union[str, Path]):
default_account = Account().to_dict()
self.storage = JSONStorage(file_path, default_account)
def load_account(self) -> Account:
"""載入帳戶資訊"""
data = self.storage.load_data()
return Account.from_dict(data)
def save_account(self, account: Account) -> None:
"""儲存帳戶資訊"""
self.storage.save_data(account.to_dict())
def update_balance(self, new_balance: Decimal, transaction_count: int) -> None:
"""更新帳戶餘額與交易數量"""
account = self.load_account()
account.balance = new_balance
account.total_transactions = transaction_count
account.last_updated = datetime.now()
self.save_account(account)
class CategoryStorage:
"""分類資訊儲存"""
def __init__(self, file_path: Union[str, Path]):
# 建立預設分類資料
default_categories = [cat.to_dict() for cat in create_default_categories()]
self.storage = JSONStorage(file_path, {"categories": default_categories})
def load_categories(self) -> List[Category]:
"""載入分類清單"""
data = self.storage.load_data()
categories = []
# 處理不同格式的資料：陣列格式或物件格式
if isinstance(data, list):
# 直接是分類陣列格式 (相容現有的 categories.json)
cat_list = data
else:
# 物件格式，從 categories 鍵取得
cat_list = data.get("categories", [])
for cat_data in cat_list:
try:
categories.append(Category.from_dict(cat_data))
except Exception:
# 忽略損壞的記錄
continue
# 如果沒有分類，回傳預設分類
if not categories:
categories = create_default_categories()
self.save_categories(categories)
return categories
def save_categories(self, categories: List[Category]) -> None:
"""儲存分類清單"""
cat_data = [cat.to_dict() for cat in categories]
self.storage.save_data({"categories": cat_data})
def is_valid_category(self, category_id: str) -> bool:
"""驗證分類是否有效"""
if not category_id or not category_id.strip():
return False
categories = self.load_categories()
valid_ids = {cat.id for cat in categories}
return category_id in valid_ids
def get_category(self, category_id: str) -> Optional[Category]:
"""根據 ID 取得分類"""
categories = self.load_categories()
for category in categories:
if category.id == category_id:
return category
return None
def add_category(self, category: Category) -> None:
"""新增新分類"""
categories = self.load_categories()
# 檢查 ID 是否已存在
existing_ids = {cat.id for cat in categories}
if category.id in existing_ids:
raise StorageError(f"分類 ID '{category.id}' 已存在")
categories.append(category)
self.save_categories(categories)
# 儲存管理器 - 統一管理所有儲存
class StorageManager:
"""儲存管理器，統一管理所有儲存元件"""
def __init__(self, data_dir: Union[str, Path] = "data"):
self.data_dir = Path(data_dir)
self.data_dir.mkdir(parents=True, exist_ok=True)
# 初始化各個儲存元件
self.transactions = TransactionStorage(self.data_dir / "transactions.json")
self.account = AccountStorage(self.data_dir / "account.json")
self.categories = CategoryStorage(self.data_dir / "categories.json")
def add_transaction(self, transaction: Transaction) -> None:
"""新增交易並更新帳戶"""
# 驗證分類
if not self.categories.is_valid_category(transaction.category):
raise StorageError(f"無效的分類: {transaction.category}")
# 新增交易
self.transactions.add_transaction(transaction)
# 更新帳戶
account = self.account.load_account()
account.add_transaction(transaction)
self.account.save_account(account)
def get_balance(self) -> Decimal:
"""取得目前餘額"""
account = self.account.load_account()
return account.balance
def get_transaction_count(self) -> int:
"""取得交易總數"""
account = self.account.load_account()
return account.total_transactions
def get_monthly_summary(self, year: int, month: int) -> Dict[str, Any]:
"""取得月度彙總"""
from calendar import monthrange
start_date = date(year, month, 1)
end_date = date(year, month, monthrange(year, month)[1])
transactions = self.transactions.get_transactions_by_date_range(start_date, end_date)
# 計算統計
total_income = sum(t.amount for t in transactions if t.amount > 0)
total_expense = sum(abs(t.amount) for t in transactions if t.amount < 0)
# 按分類統計
category_stats = {}
categories = self.categories.load_categories()
category_names = {cat.id: cat.name for cat in categories}
for transaction in transactions:
cat_name = category_names.get(transaction.category, transaction.category)
if cat_name not in category_stats:
category_stats[cat_name] = Decimal('0.00')
category_stats[cat_name] += abs(transaction.amount)
return {
"year": year,
"month": month,
"total_income": float(total_income),
"total_expense": float(total_expense),
"net_flow": float(total_income - total_expense),
"transaction_count": len(transactions),
"category_breakdown": {k: float(v) for k, v in category_stats.items()}
}
