Create admin_patterns.py

admin_patterns.py

@@ -0,0 +1,191 @@
+# admin_patterns.py
+import os
+import json
+import re
+import shutil
+import pandas as pd
+from datetime import datetime
+
+BASE_DIR = os.path.dirname(os.path.abspath(__file__))
+PATTERN_FILE = os.path.join(BASE_DIR, "patterns.json")
+pattern_config = None
+DEFAULT_PATTERN_CONFIG = {"global": [], "states": {}}
+
+
+def _make_backup_of_patterns():
+    """
+    If PATTERN_FILE exists, copy it to patterns_backup_{timestamp}.json
+    Returns backup path or None on failure/if not exists.
+    """
+    try:
+        if os.path.exists(PATTERN_FILE):
+            ts = datetime.utcnow().strftime("%Y%m%dT%H%M%SZ")
+            backup_name = f"patterns_backup_{ts}.json"
+            backup_path = os.path.join(BASE_DIR, backup_name)
+            shutil.copyfile(PATTERN_FILE, backup_path)
+            return backup_path
+    except Exception as e:
+        print("ERROR creating patterns.json backup:", e)
+    return None
+
+
+def load_pattern_config(debug: bool = False):
+    """
+    Load patterns.json. If missing, create an empty skeleton.
+    """
+    global pattern_config
+    if os.path.exists(PATTERN_FILE):
+        try:
+            with open(PATTERN_FILE, "r", encoding="utf-8") as f:
+                cfg = json.load(f)
+            pattern_config = cfg
+            if debug:
+                print("DEBUG: Successfully loaded patterns.json")
+                print("DEBUG: global count:", len(cfg.get("global", [])))
+                print("DEBUG: state keys:", list(cfg.get("states", {}).keys()))
+            return cfg
+        except Exception as e:
+            print("ERROR: Failed to parse patterns.json:", e)
+            pattern_config = {"global": [], "states": {}}
+            return pattern_config
+    else:
+        # create empty file
+        try:
+            with open(PATTERN_FILE, "w", encoding="utf-8") as f:
+                json.dump(DEFAULT_PATTERN_CONFIG, f, indent=2, ensure_ascii=False)
+            print("DEBUG: Created empty patterns.json at", PATTERN_FILE)
+        except Exception as e:
+            print("ERROR: Could not create patterns.json:", e)
+        pattern_config = DEFAULT_PATTERN_CONFIG.copy()
+        return pattern_config
+
+
+def save_pattern_config(cfg: dict):
+    """
+    Save the given config to PATTERN_FILE with a timestamped backup of the previous file.
+    """
+    global pattern_config
+    # create backup first
+    backup_path = _make_backup_of_patterns()
+    if backup_path:
+        print(f"INFO: patterns.json backed up to {backup_path}")
+    # write new file
+    with open(PATTERN_FILE, "w", encoding="utf-8") as f:
+        json.dump(cfg, f, indent=2, ensure_ascii=False)
+    pattern_config = cfg.copy()
+    return True
+
+
+def build_patterns_from_config(cfg: dict, state_key: str | None):
+    global_list = [(p["pattern"], p["replacement"]) for p in cfg.get("global", [])]
+    state_list = []
+    if state_key:
+        state_key_up = state_key.upper().strip()
+        state_patterns = cfg.get("states", {}).get(state_key_up, [])
+        state_list = [(p["pattern"], p["replacement"]) for p in state_patterns]
+    return global_list, state_list
+
+
+def normalize_with_patterns_dynamic(s: str, state_key: str | None):
+    global pattern_config
+    if not isinstance(s, str):
+        return ""
+    s = s.upper()
+    if pattern_config is None:
+        load_pattern_config(debug=False)
+    cfg = pattern_config or DEFAULT_PATTERN_CONFIG
+    global_patterns, state_patterns = build_patterns_from_config(cfg, state_key)
+    for pat, repl in global_patterns:
+        try:
+            s = re.sub(pat, repl, s)
+        except re.error:
+            continue
+    for pat, repl in state_patterns:
+        try:
+            s = re.sub(pat, repl, s)
+        except re.error:
+            continue
+    s = re.sub(r"[^A-Z0-9]+", " ", s)
+    s = re.sub(r"\s+", " ", s).strip()
+    return s
+
+
+# Admin helpers for the UI
+def load_global_patterns_for_editor():
+    cfg = load_pattern_config(debug=False)
+    return pd.DataFrame(cfg.get("global", []))
+
+
+def load_state_patterns_for_editor(selected_state: str | None, new_state_name: str | None):
+    cfg = load_pattern_config(debug=False)
+    key = None
+    if new_state_name and new_state_name.strip():
+        key = new_state_name.strip().upper()
+    elif selected_state:
+        key = selected_state.strip().upper()
+    if not key:
+        return pd.DataFrame(columns=["pattern", "replacement"])
+    state_patterns = cfg.get("states", {}).get(key, [])
+    return pd.DataFrame(state_patterns)
+
+
+def save_global_patterns_from_editor(df: pd.DataFrame, admin_password: str, expected_password: str):
+    if expected_password is None:
+        return "❌ ADMIN password not configured in environment."
+    if admin_password != expected_password:
+        return "❌ Invalid admin password. Global patterns NOT saved."
+    cfg = load_pattern_config(debug=False)
+    cfg["global"] = df.fillna("").to_dict(orient="records")
+    save_pattern_config(cfg)
+    gcount = len(cfg.get("global", []))
+    skeys = sorted(list(cfg.get("states", {}).keys()))
+    return f"✅ Global patterns saved — global={gcount}, state_keys={skeys}"
+
+
+def save_state_patterns_from_editor(selected_state: str | None, new_state_name: str | None, df: pd.DataFrame, admin_password: str, expected_password: str):
+    if expected_password is None:
+        return "❌ ADMIN password not configured in environment."
+    if admin_password != expected_password:
+        return "❌ Invalid admin password. State patterns NOT saved."
+    key = None
+    if new_state_name and new_state_name.strip():
+        key = new_state_name.strip().upper()
+    elif selected_state:
+        key = selected_state.strip().upper()
+    if not key:
+        return "⚠ Please select a state or type a new state key."
+    cfg = load_pattern_config(debug=False)
+    cfg.setdefault("states", {})[key] = df.fillna("").to_dict(orient="records")
+    save_pattern_config(cfg)
+    gcount = len(cfg.get("global", []))
+    skeys = sorted(list(cfg.get("states", {}).keys()))
+    return f"✅ Patterns for {key} saved — global={gcount}, state_keys={skeys}"
+
+
+def refresh_pattern_config():
+    cfg = load_pattern_config(debug=True)
+    gcount = len(cfg.get("global", []))
+    skeys = sorted(list(cfg.get("states", {}).keys()))
+    return f"Refreshed patterns.json — global={gcount}, state_keys={skeys}"
+
+
+def show_patterns_file_info():
+    info_lines = []
+    info_lines.append(f"PATTERN_FILE: {PATTERN_FILE}")
+    info_lines.append(f"Exists: {os.path.exists(PATTERN_FILE)}")
+    if os.path.exists(PATTERN_FILE):
+        try:
+            size = os.path.getsize(PATTERN_FILE)
+            info_lines.append(f"Size (bytes): {size}")
+            with open(PATTERN_FILE, "r", encoding="utf-8") as f:
+                txt = f.read(1000)
+            info_lines.append("Preview (first 1000 chars):")
+            info_lines.append("```json\n" + txt + ("\n... (truncated)" if len(txt) >= 1000 else "") + "\n```")
+        except Exception as e:
+            info_lines.append("ERROR reading file: " + str(e))
+    try:
+        listing = os.listdir(BASE_DIR)
+        info_lines.append("Files in BASE_DIR: " + ", ".join(listing))
+    except Exception as e:
+        info_lines.append("ERROR listing BASE_DIR: " + str(e))
+    return "\n\n".join(info_lines)