yaml_processor.py
· 1.3 KiB · Python
原始文件
# yaml_processor.py
import yaml
import pandas as pd
import os
class YAMLProcessor:
def __init__(self, yaml_path="local_data.yaml"):
"""
處理 YAML 讀寫的類別。
Args:
yaml_path (str): YAML 檔案路徑。
"""
self.yaml_path = yaml_path
# 如果檔案不存在，就建立一個空的
if not os.path.isfile(self.yaml_path):
with open(self.yaml_path, "w", encoding="utf-8") as f:
yaml.safe_dump([], f, allow_unicode=True)
def load_local_data(self):
"""
從 YAML 中讀取資料並轉成 DataFrame。
Returns:
pd.DataFrame: local_data.yaml 內容。
"""
with open(self.yaml_path, "r", encoding="utf-8") as f:
data = yaml.safe_load(f)
if data is None:
data = []
df = pd.DataFrame(data)
return df
def save_local_data(self, df):
"""
將 DataFrame 內容回存到 YAML。
Args:
df (pd.DataFrame): 要存回 YAML 的 DataFrame。
"""
# 將 df 轉成 list[dict] 方便寫回 YAML
data_list = df.to_dict(orient="records")
with open(self.yaml_path, "w", encoding="utf-8") as f:
yaml.safe_dump(data_list, f, allow_unicode=True)
| 1 | # yaml_processor.py |
| 2 | |
| 3 | import yaml |
| 4 | import pandas as pd |
| 5 | import os |
| 6 | |
| 7 | class YAMLProcessor: |
| 8 | def __init__(self, yaml_path="local_data.yaml"): |
| 9 | """ |
| 10 | 處理 YAML 讀寫的類別。 |
| 11 | Args: |
| 12 | yaml_path (str): YAML 檔案路徑。 |
| 13 | """ |
| 14 | self.yaml_path = yaml_path |
| 15 | # 如果檔案不存在，就建立一個空的 |
| 16 | if not os.path.isfile(self.yaml_path): |
| 17 | with open(self.yaml_path, "w", encoding="utf-8") as f: |
| 18 | yaml.safe_dump([], f, allow_unicode=True) |
| 19 | |
| 20 | def load_local_data(self): |
| 21 | """ |
| 22 | 從 YAML 中讀取資料並轉成 DataFrame。 |
| 23 | Returns: |
| 24 | pd.DataFrame: local_data.yaml 內容。 |
| 25 | """ |
| 26 | with open(self.yaml_path, "r", encoding="utf-8") as f: |
| 27 | data = yaml.safe_load(f) |
| 28 | if data is None: |
| 29 | data = [] |
| 30 | df = pd.DataFrame(data) |
| 31 | return df |
| 32 | |
| 33 | def save_local_data(self, df): |
| 34 | """ |
| 35 | 將 DataFrame 內容回存到 YAML。 |
| 36 | Args: |
| 37 | df (pd.DataFrame): 要存回 YAML 的 DataFrame。 |
| 38 | """ |
| 39 | # 將 df 轉成 list[dict] 方便寫回 YAML |
| 40 | data_list = df.to_dict(orient="records") |
| 41 | with open(self.yaml_path, "w", encoding="utf-8") as f: |
| 42 | yaml.safe_dump(data_list, f, allow_unicode=True) |
| 43 |