timmy / YAML 檔案讀寫與資料管理

import os import yaml import pandas as pd import logging from typing import Any, Dict, List # 設定 logging 基本參數 logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # 如果需要檔案鎖定，可以使用 filelock 模組（需先安裝：pip install filelock） try: from filelock import FileLock FILELOCK_AVAILABLE = True except ImportError: FILELOCK_AVAILABLE = False logger.warning("filelock 模組未安裝，未啟用檔案鎖定機制。") class YAMLProcessor: def __init__(self, yaml_path: str = "local_data.yaml") -> None: """ 處理 YAML 讀寫的類別。 Args: yaml_path (str): YAML 檔案路徑。 """ self.yaml_path = yaml_path # 檢查目錄是否存在，不存在則建立 directory = os.path.dirname(os.path.abspath(self.yaml_path)) if directory and not os.path.exists(directory): try: os.makedirs(directory, exist_ok=True) logger.info("建立目錄：%s", directory) except Exception as e: logger.error("建立目錄失敗：%s", e) # 如果檔案不存在，就建立一個空的 if not os.path.isfile(self.yaml_path): try: with open(self.yaml_path, "w", encoding="utf-8") as f: yaml.safe_dump([], f, allow_unicode=True) logger.info("建立 YAML 檔案：%s", self.yaml_path) except Exception as e: logger.error("建立 YAML 檔案失敗：%s", e) def load_local_data(self) -> pd.DataFrame: """ 從 YAML 中讀取資料並轉成 DataFrame。 Returns: pd.DataFrame: YAML 檔案內容。 """ data: List[Dict[str, Any]] = [] try: # 如果有啟用檔案鎖定，則使用 filelock 保護檔案讀取 if FILELOCK_AVAILABLE: lock_path = self.yaml_path + ".lock" with FileLock(lock_path): with open(self.yaml_path, "r", encoding="utf-8") as f: data = yaml.safe_load(f) else: with open(self.yaml_path, "r", encoding="utf-8") as f: data = yaml.safe_load(f) except Exception as e: logger.error("讀取 YAML 時發生錯誤：%s", e) data = [] if data is None: data = [] return pd.DataFrame(data) def save_local_data(self, df: pd.DataFrame) -> None: """ 將 DataFrame 內容回存到 YAML。 Args: df (pd.DataFrame): 要存回 YAML 的 DataFrame。 """ data_list = df.to_dict(orient="records") try: # 如果有啟用檔案鎖定，則使用 filelock 保護檔案寫入 if FILELOCK_AVAILABLE: lock_path = self.yaml_path + ".lock" with FileLock(lock_path): with open(self.yaml_path, "w", encoding="utf-8") as f: yaml.safe_dump(data_list, f, allow_unicode=True) else: with open(self.yaml_path, "w", encoding="utf-8") as f: yaml.safe_dump(data_list, f, allow_unicode=True) logger.info("成功儲存 YAML 檔案：%s", self.yaml_path) except Exception as e: logger.error("儲存 YAML 時發生錯誤：%s", e) def main(): # 指定 YAML 檔案路徑（可依需求調整） yaml_file = "local_data.yaml" processor = YAMLProcessor(yaml_file) # 載入 YAML 檔案內容為 DataFrame df = processor.load_local_data() print("目前 YAML 檔案資料：") print(df) # 定義一筆新的資料記錄 new_record = {"name": "Alice", "age": 30, "city": "Taipei"} # 如果 DataFrame 為空，建立新的 DataFrame；否則新增一筆資料 if df.empty: df = pd.DataFrame([new_record]) else: df.loc[len(df)] = new_record # 將更新後的 DataFrame 儲存回 YAML 檔案 processor.save_local_data(df) print("更新後的資料已儲存回 YAML 檔案。") if __name__ == "__main__": main()