Änderungen von YAML 檔案讀寫與資料管理

timmy hat die Gist bearbeitet 10 months ago. Zu Änderung gehen

Keine Änderungen

timmy hat die Gist bearbeitet 10 months ago. Zu Änderung gehen

Keine Änderungen

timmy hat die Gist bearbeitet 10 months ago. Zu Änderung gehen

1 file changed, 94 insertions, 16 deletions

yaml_processor.py

			@@ -1,42 +1,120 @@
1		-	# yaml_processor.py
2		-
	1	+	import os
3	2		import yaml
4	3		import pandas as pd
5		-	import os
	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 模組未安裝，未啟用檔案鎖定機制。")
6	18
7	19		class YAMLProcessor:
8		-	def __init__(self, yaml_path="local_data.yaml"):
	20	+	def __init__(self, yaml_path: str = "local_data.yaml") -> None:
9	21		"""
10	22		處理 YAML 讀寫的類別。
	23	+
11	24		Args:
12	25		yaml_path (str): YAML 檔案路徑。
13	26		"""
14	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	+
15	38		# 如果檔案不存在，就建立一個空的
16	39		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)
	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)
19	46
20		-	def load_local_data(self):
	47	+	def load_local_data(self) -> pd.DataFrame:
21	48		"""
22	49		從 YAML 中讀取資料並轉成 DataFrame。
	50	+
23	51		Returns:
24		-	pd.DataFrame: local_data.yaml 內容。
	52	+	pd.DataFrame: YAML 檔案內容。
25	53		"""
26		-	with open(self.yaml_path, "r", encoding="utf-8") as f:
27		-	data = yaml.safe_load(f)
	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	+
28	69		if data is None:
29	70		data = []
30		-	df = pd.DataFrame(data)
31		-	return df
32	71
33		-	def save_local_data(self, df):
	72	+	return pd.DataFrame(data)
	73	+
	74	+	def save_local_data(self, df: pd.DataFrame) -> None:
34	75		"""
35	76		將 DataFrame 內容回存到 YAML。
	77	+
36	78		Args:
37	79		df (pd.DataFrame): 要存回 YAML 的 DataFrame。
38	80		"""
39		-	# 將 df 轉成 list[dict] 方便寫回 YAML
40	81		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)
	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()

timmy hat die Gist bearbeitet 10 months ago. Zu Änderung gehen

Keine Änderungen

timmy hat die Gist bearbeitet 10 months ago. Zu Änderung gehen

1 file changed, 42 insertions

yaml_processor.py(Datei erstellt)

		@@ -0,0 +1,42 @@
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)

Neuer Älter