Révisions pour YAML 檔案讀寫與資料管理

timmy a révisé ce gist 10 months ago. Aller à la révision

Aucun changement

timmy a révisé ce gist 10 months ago. Aller à la révision

Aucun changement

timmy a révisé ce gist 10 months ago. Aller à la révision

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 a révisé ce gist 10 months ago. Aller à la révision

Aucun changement

timmy a révisé ce gist 10 months ago. Aller à la révision

1 file changed, 42 insertions

yaml_processor.py(fichier créé)

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

Plus récent Plus ancien