75 lines
2.5 KiB
Python
75 lines
2.5 KiB
Python
# scripts/memory_store.py
|
||
import json
|
||
from pathlib import Path
|
||
from config import MEMORY_DIR
|
||
from datetime import datetime, timezone
|
||
|
||
def load_logs(date_str=None):
|
||
if date_str is None:
|
||
date_str = datetime.now().strftime("%Y-%m-%d")
|
||
path = MEMORY_DIR / f"{date_str}.json"
|
||
if path.exists():
|
||
with open(path, "r") as f:
|
||
return json.load(f)
|
||
return []
|
||
|
||
def save_message(sender, message):
|
||
date_str = datetime.now().strftime("%Y-%m-%d")
|
||
path = MEMORY_DIR / f"{date_str}.json"
|
||
logs = load_logs(date_str)
|
||
now = datetime.now(timezone.utc).isoformat()
|
||
logs.append({"timestamp": now, "sender": sender, "message": message})
|
||
with open(path, "w") as f:
|
||
json.dump(logs, f, indent=2, ensure_ascii=False)
|
||
|
||
def search_memory(query: str):
|
||
from glob import glob
|
||
all_logs = []
|
||
for file_path in sorted(MEMORY_DIR.glob("*.json")):
|
||
with open(file_path, "r") as f:
|
||
logs = json.load(f)
|
||
matched = [entry for entry in logs if query in entry["message"]]
|
||
all_logs.extend(matched)
|
||
return all_logs[-5:] # 最新5件だけ返す
|
||
|
||
|
||
# scripts/memory_store.py
|
||
import json
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
from config import MEMORY_DIR
|
||
|
||
# ログを読み込む（指定日または当日）
|
||
def load_logs(date_str=None):
|
||
if date_str is None:
|
||
date_str = datetime.now().strftime("%Y-%m-%d")
|
||
path = MEMORY_DIR / f"{date_str}.json"
|
||
if path.exists():
|
||
with open(path, "r") as f:
|
||
return json.load(f)
|
||
return []
|
||
|
||
# メッセージを保存する
|
||
def save_message(sender, message):
|
||
date_str = datetime.now().strftime("%Y-%m-%d")
|
||
path = MEMORY_DIR / f"{date_str}.json"
|
||
logs = load_logs(date_str)
|
||
#now = datetime.utcnow().isoformat() + "Z"
|
||
now = datetime.now(timezone.utc).isoformat()
|
||
logs.append({"timestamp": now, "sender": sender, "message": message})
|
||
with open(path, "w") as f:
|
||
json.dump(logs, f, indent=2, ensure_ascii=False)
|
||
|
||
# キーワードで過去のメモリを検索する（最新5件を返す）
|
||
def search_memory(query: str):
|
||
all_logs = []
|
||
for file_path in sorted(MEMORY_DIR.glob("*.json")):
|
||
try:
|
||
with open(file_path, "r") as f:
|
||
logs = json.load(f)
|
||
matched = [entry for entry in logs if query.lower() in entry["message"].lower()]
|
||
all_logs.extend(matched)
|
||
except Exception as e:
|
||
print(f"⚠️ 読み込み失敗: {file_path} ({e})")
|
||
return all_logs[-5:] # 最新5件を返す
|