| import json | |
| import os | |
| import logging | |
| from datetime import datetime | |
| logger = logging.getLogger("ONYX") | |
| KNOWLEDGE_FILE = "database/knowledge.json" | |
| class KnowledgeEngine: | |
| def __init__(self): | |
| if not os.path.exists("database"): | |
| os.makedirs("database") | |
| if not os.path.exists(KNOWLEDGE_FILE): | |
| with open(KNOWLEDGE_FILE, "w", encoding="utf-8") as f: | |
| json.dump([], f) | |
| # ========================= | |
| # LOAD KNOWLEDGE | |
| # ========================= | |
| def load(self): | |
| try: | |
| with open(KNOWLEDGE_FILE, "r", encoding="utf-8") as f: | |
| return json.load(f) | |
| except Exception as e: | |
| logger.error(f"Knowledge load error: {e}") | |
| return [] | |
| # ========================= | |
| # SAVE KNOWLEDGE | |
| # ========================= | |
| def save(self, data): | |
| try: | |
| with open(KNOWLEDGE_FILE, "w", encoding="utf-8") as f: | |
| json.dump(data, f, indent=2, ensure_ascii=False) | |
| except Exception as e: | |
| logger.error(f"Knowledge save error: {e}") | |
| # ========================= | |
| # ADD ENTRY | |
| # ========================= | |
| def add_entry(self, category, content): | |
| data = self.load() | |
| entry = { | |
| "category": category, | |
| "content": content, | |
| "timestamp": datetime.utcnow().isoformat() | |
| } | |
| data.append(entry) | |
| self.save(data) | |
| # ========================= | |
| # SEARCH | |
| # ========================= | |
| def search(self, keyword): | |
| data = self.load() | |
| results = [] | |
| for entry in data: | |
| if keyword.lower() in entry["content"].lower(): | |
| results.append(entry) | |
| return results |