Create csv_util.py

csv_util.py

@@ -0,0 +1,66 @@
+# csv_util.py
+
+import os
+import pandas as pd
+import logging
+
+logging.basicConfig(
+    level=logging.INFO,
+    format='%(asctime)s [%(levelname)s] %(name)s: %(message)s',
+    datefmt='%Y-%m-%d %H:%M:%S'
+)
+logger = logging.getLogger(__name__)
+
+CSV_COLUMNS = ["DateTime", "Currency", "Impact", "Event", "Actual", "Forecast", "Previous", "Detail"]
+
+def ensure_csv_header(csv_file):
+    if not os.path.exists(csv_file) or os.path.getsize(csv_file) == 0:
+        pd.DataFrame(columns=CSV_COLUMNS).to_csv(csv_file, index=False)
+
+def read_existing_data(csv_file):
+    ensure_csv_header(csv_file)
+    try:
+        df = pd.read_csv(csv_file, dtype=str)
+        for col in CSV_COLUMNS:
+            if col not in df.columns:
+                df[col] = ""
+        return df[CSV_COLUMNS]
+    except Exception:
+        return pd.DataFrame(columns=CSV_COLUMNS)
+
+def write_data_to_csv(df: pd.DataFrame, csv_file: str):
+    df = df.sort_values(by="DateTime", ascending=True)
+    df.to_csv(csv_file, index=False)
+
+def merge_new_data(existing_df, new_df):
+    if existing_df.empty:
+        return new_df
+
+    def add_unique_key(df):
+        df = df.copy()
+        df['unique_key'] = (
+            df["DateTime"].astype(str).str.strip() + "_" +
+            df["Currency"].astype(str).str.strip() + "_" +
+            df["Event"].astype(str).str.strip()
+        )
+        return df
+
+    existing_df = add_unique_key(existing_df)
+    new_df = add_unique_key(new_df)
+    existing_df.set_index('unique_key', inplace=True)
+    new_df.set_index('unique_key', inplace=True)
+    new_rows_list = []
+    for key, new_row in new_df.iterrows():
+        if key in existing_df.index:
+            existing_detail = str(existing_df.at[key, "Detail"]).strip() if pd.notna(existing_df.at[key, "Detail"]) else ""
+            new_detail = str(new_row["Detail"]).strip() if pd.notna(new_row["Detail"]) else ""
+            if not existing_detail and new_detail:
+                existing_df.at[key, "Detail"] = new_detail
+        else:
+            new_rows_list.append(new_row)
+    if new_rows_list:
+        new_rows_df = pd.DataFrame(new_rows_list)
+        existing_df = pd.concat([existing_df, new_rows_df])
+    merged_df = existing_df.reset_index(drop=True)
+    merged_df = merged_df[CSV_COLUMNS]
+    return merged_df