54 lines
1.5 KiB
Python
54 lines
1.5 KiB
Python
import json
|
|
import uuid
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Dict, List, Optional
|
|
|
|
from .config import DATA_DIR, CONVERSATIONS_DIR
|
|
|
|
|
|
def _path(cid: str) -> Path:
|
|
return CONVERSATIONS_DIR / f"{cid}.json"
|
|
|
|
|
|
def make_conversation(title: str = "新对话") -> Dict:
|
|
cid = str(uuid.uuid4())
|
|
convo = {
|
|
"id": cid,
|
|
"title": (title or "新对话").strip()[:40],
|
|
"messages": [],
|
|
"updated_at": datetime.utcnow().isoformat(),
|
|
}
|
|
save(convo)
|
|
return convo
|
|
|
|
|
|
def save(convo: Dict) -> None:
|
|
_path(convo["id"]).write_text(json.dumps(convo, ensure_ascii=False), encoding="utf-8")
|
|
|
|
|
|
def load(cid: str) -> Optional[Dict]:
|
|
path = _path(cid)
|
|
legacy = DATA_DIR / f"{cid}.json"
|
|
if not path.exists() and legacy.exists():
|
|
path = legacy
|
|
if not path.exists():
|
|
return None
|
|
try:
|
|
return json.loads(path.read_text(encoding="utf-8"))
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def list_conversations() -> List[Dict]:
|
|
items = []
|
|
for root in (CONVERSATIONS_DIR, DATA_DIR):
|
|
for path in root.glob("*.json"):
|
|
try:
|
|
convo = json.loads(path.read_text(encoding="utf-8"))
|
|
if isinstance(convo, dict) and convo.get("id") and isinstance(convo.get("messages"), list):
|
|
items.append(convo)
|
|
except Exception:
|
|
continue
|
|
return sorted(items, key=lambda x: x.get("updated_at", ""), reverse=True)
|