Create utils.py

utils.py

@@ -0,0 +1,83 @@
+import os
+import json
+import re
+import logging
+
+from memory_logic import add_rule_entry, add_memory_entry
+
+logger = logging.getLogger(__name__)
+
+def format_insights_for_prompt(retrieved_insights_list: list[str]) -> tuple[str, list[dict]]:
+    if not retrieved_insights_list: return "No specific guiding principles or learned insights retrieved.", []
+    parsed = []
+    for text in retrieved_insights_list:
+        match = re.match(r"\[(CORE_RULE|RESPONSE_PRINCIPLE|BEHAVIORAL_ADJUSTMENT|GENERAL_LEARNING)\|([\d\.]+?)\](.*)", text.strip(), re.DOTALL | re.IGNORECASE)
+        if match: parsed.append({"type": match.group(1).upper().replace(" ", "_"), "score": match.group(2), "text": match.group(3).strip(), "original": text.strip()})
+        else: parsed.append({"type": "GENERAL_LEARNING", "score": "0.5", "text": text.strip(), "original": text.strip()})
+    parsed.sort(key=lambda x: float(x["score"]) if x["score"].replace('.', '', 1).isdigit() else -1.0, reverse=True)
+    grouped = {"CORE_RULE": [], "RESPONSE_PRINCIPLE": [], "BEHAVIORAL_ADJUSTMENT": [], "GENERAL_LEARNING": []}
+    for p_item in parsed: grouped.get(p_item["type"], grouped["GENERAL_LEARNING"]).append(f"- (Score: {p_item['score']}) {p_item['text']}")
+    sections = [f"{k.replace('_', ' ').title()}:\n" + "\n".join(v) for k, v in grouped.items() if v]
+    return "\n\n".join(sections) if sections else "No guiding principles retrieved.", parsed
+
+def load_rules_from_file(filepath: str | None, progress_callback=None):
+    if not filepath or not os.path.exists(filepath): return 0, 0, 0
+    added, skipped, errors = 0, 0, 0
+    with open(filepath, 'r', encoding='utf-8') as f: content = f.read()
+    if not content.strip(): return 0, 0, 0
+    
+    potential_rules = []
+    if filepath.lower().endswith(".txt"):
+        potential_rules = content.split("\n\n---\n\n")
+        if len(potential_rules) == 1 and "\n" in content: potential_rules = content.splitlines()
+    elif filepath.lower().endswith(".jsonl"):
+        for line in content.splitlines():
+            if line.strip():
+                try: potential_rules.append(json.loads(line))
+                except json.JSONDecodeError: errors += 1
+    
+    valid_rules = [r.strip() for r in potential_rules if isinstance(r, str) and r.strip()]
+    total = len(valid_rules)
+    if not total: return 0, 0, errors
+
+    for idx, rule_text in enumerate(valid_rules):
+        success, status_msg = add_rule_entry(rule_text)
+        if success: added += 1
+        elif status_msg == "duplicate": skipped += 1
+        else: errors += 1
+        if progress_callback: progress_callback((idx + 1) / total, f"Processed {idx+1}/{total} rules...")
+    
+    logger.info(f"Loaded rules from {filepath}: Added {added}, Skipped {skipped}, Errors {errors}.")
+    return added, skipped, errors
+
+def load_memories_from_file(filepath: str | None, progress_callback=None):
+    if not filepath or not os.path.exists(filepath): return 0, 0, 0
+    added, format_err, save_err = 0, 0, 0
+    with open(filepath, 'r', encoding='utf-8') as f: content = f.read()
+    if not content.strip(): return 0, 0, 0
+    
+    mem_objects = []
+    if filepath.lower().endswith(".json"):
+        try:
+            data = json.loads(content)
+            mem_objects = data if isinstance(data, list) else [data]
+        except json.JSONDecodeError: format_err = 1
+    elif filepath.lower().endswith(".jsonl"):
+        for line in content.splitlines():
+            if line.strip():
+                try: mem_objects.append(json.loads(line))
+                except json.JSONDecodeError: format_err += 1
+
+    total = len(mem_objects)
+    if not total: return 0, format_err, 0
+
+    for idx, mem in enumerate(mem_objects):
+        if isinstance(mem, dict) and all(k in mem for k in ["user_input", "bot_response", "metrics"]):
+            success, _ = add_memory_entry(mem["user_input"], mem["metrics"], mem["bot_response"])
+            if success: added += 1
+            else: save_err += 1
+        else: format_err += 1
+        if progress_callback: progress_callback((idx + 1) / total, f"Processed {idx+1}/{total} memories...")
+        
+    logger.info(f"Loaded memories from {filepath}: Added {added}, Format Errors {format_err}, Save Errors {save_err}.")
+    return added, format_err, save_err