YAML 檔案讀寫與資料管理

Änderung f49d0965260947f543d647866caed7367cdf8ccb

yaml_processor.py · 4.1 KiB · Python Originalformat

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()

1	import os
2	import yaml
3	import pandas as pd
4	import logging
5	from typing import Any, Dict, List
6
7	# 設定 logging 基本參數
8	logging.basicConfig(level=logging.INFO)
9	logger = logging.getLogger(__name__)
10
11	# 如果需要檔案鎖定，可以使用 filelock 模組（需先安裝：pip install filelock）
12	try:
13	from filelock import FileLock
14	FILELOCK_AVAILABLE = True
15	except ImportError:
16	FILELOCK_AVAILABLE = False
17	logger.warning("filelock 模組未安裝，未啟用檔案鎖定機制。")
18
19	class YAMLProcessor:
20	def __init__(self, yaml_path: str = "local_data.yaml") -> None:
21	"""
22	處理 YAML 讀寫的類別。
23
24	Args:
25	yaml_path (str): YAML 檔案路徑。
26	"""
27	self.yaml_path = yaml_path
28
29	# 檢查目錄是否存在，不存在則建立
30	directory = os.path.dirname(os.path.abspath(self.yaml_path))
31	if directory and not os.path.exists(directory):
32	try:
33	os.makedirs(directory, exist_ok=True)
34	logger.info("建立目錄：%s", directory)
35	except Exception as e:
36	logger.error("建立目錄失敗：%s", e)
37
38	# 如果檔案不存在，就建立一個空的
39	if not os.path.isfile(self.yaml_path):
40	try:
41	with open(self.yaml_path, "w", encoding="utf-8") as f:
42	yaml.safe_dump([], f, allow_unicode=True)
43	logger.info("建立 YAML 檔案：%s", self.yaml_path)
44	except Exception as e:
45	logger.error("建立 YAML 檔案失敗：%s", e)
46
47	def load_local_data(self) -> pd.DataFrame:
48	"""
49	從 YAML 中讀取資料並轉成 DataFrame。
50
51	Returns:
52	pd.DataFrame: YAML 檔案內容。
53	"""
54	data: List[Dict[str, Any]] = []
55	try:
56	# 如果有啟用檔案鎖定，則使用 filelock 保護檔案讀取
57	if FILELOCK_AVAILABLE:
58	lock_path = self.yaml_path + ".lock"
59	with FileLock(lock_path):
60	with open(self.yaml_path, "r", encoding="utf-8") as f:
61	data = yaml.safe_load(f)
62	else:
63	with open(self.yaml_path, "r", encoding="utf-8") as f:
64	data = yaml.safe_load(f)
65	except Exception as e:
66	logger.error("讀取 YAML 時發生錯誤：%s", e)
67	data = []
68
69	if data is None:
70	data = []
71
72	return pd.DataFrame(data)
73
74	def save_local_data(self, df: pd.DataFrame) -> None:
75	"""
76	將 DataFrame 內容回存到 YAML。
77
78	Args:
79	df (pd.DataFrame): 要存回 YAML 的 DataFrame。
80	"""
81	data_list = df.to_dict(orient="records")
82	try:
83	# 如果有啟用檔案鎖定，則使用 filelock 保護檔案寫入
84	if FILELOCK_AVAILABLE:
85	lock_path = self.yaml_path + ".lock"
86	with FileLock(lock_path):
87	with open(self.yaml_path, "w", encoding="utf-8") as f:
88	yaml.safe_dump(data_list, f, allow_unicode=True)
89	else:
90	with open(self.yaml_path, "w", encoding="utf-8") as f:
91	yaml.safe_dump(data_list, f, allow_unicode=True)
92	logger.info("成功儲存 YAML 檔案：%s", self.yaml_path)
93	except Exception as e:
94	logger.error("儲存 YAML 時發生錯誤：%s", e)
95
96	def main():
97	# 指定 YAML 檔案路徑（可依需求調整）
98	yaml_file = "local_data.yaml"
99	processor = YAMLProcessor(yaml_file)
100
101	# 載入 YAML 檔案內容為 DataFrame
102	df = processor.load_local_data()
103	print("目前 YAML 檔案資料：")
104	print(df)
105
106	# 定義一筆新的資料記錄
107	new_record = {"name": "Alice", "age": 30, "city": "Taipei"}
108
109	# 如果 DataFrame 為空，建立新的 DataFrame；否則新增一筆資料
110	if df.empty:
111	df = pd.DataFrame([new_record])
112	else:
113	df.loc[len(df)] = new_record
114
115	# 將更新後的 DataFrame 儲存回 YAML 檔案
116	processor.save_local_data(df)
117	print("更新後的資料已儲存回 YAML 檔案。")
118
119	if __name__ == "__main__":
120	main()
121