99 lines
3.3 KiB
Python
99 lines
3.3 KiB
Python
"""Prompt management — view, edit, reload system prompts."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from fastapi import APIRouter, HTTPException
|
|
from pydantic import BaseModel
|
|
|
|
from Project.Sanad.config import SCRIPTS_DIR
|
|
from Project.Sanad.core.config_loader import section as _cfg_section
|
|
from Project.Sanad.dashboard.routes._safe_io import (
|
|
atomic_write_text, MAX_UPLOAD_BYTES,
|
|
)
|
|
|
|
router = APIRouter()
|
|
|
|
# Filenames — SINGLE SOURCE in core.script_files
|
|
_SCRIPTS = _cfg_section("core", "script_files")
|
|
SCRIPT_PROMPT_PATH = SCRIPTS_DIR / _SCRIPTS.get("persona", "sanad_script.txt")
|
|
RULE_PROMPT_PATH = SCRIPTS_DIR / _SCRIPTS.get("rules", "sanad_rule.txt")
|
|
MAX_PROMPT_BYTES = MAX_UPLOAD_BYTES
|
|
|
|
# Default system prompt — SINGLE SOURCE in core.gemini_defaults
|
|
DEFAULT_SYSTEM_PROMPT = _cfg_section("core", "gemini_defaults").get(
|
|
"default_system_prompt",
|
|
"You are Sanad (Bousandah), a wise and friendly Emirati assistant. "
|
|
"Speak strictly in the UAE dialect (Khaleeji). "
|
|
"Be helpful, concise, and use local greetings like 'Marhaba' and 'Ya Khoy'."
|
|
)
|
|
|
|
|
|
def _load_system_prompt() -> str:
|
|
try:
|
|
content = SCRIPT_PROMPT_PATH.read_text(encoding="utf-8-sig").strip()
|
|
if content:
|
|
return content
|
|
except FileNotFoundError:
|
|
pass
|
|
return DEFAULT_SYSTEM_PROMPT
|
|
|
|
|
|
def _load_rule_prompts() -> dict[str, str]:
|
|
result = {"system_prompt": "", "replay_prompt": ""}
|
|
try:
|
|
content = RULE_PROMPT_PATH.read_text(encoding="utf-8-sig").strip()
|
|
sections: dict[str, list[str]] = {}
|
|
current = None
|
|
for line in content.splitlines():
|
|
stripped = line.strip()
|
|
if stripped.startswith("[") and stripped.endswith("]"):
|
|
current = stripped[1:-1].strip()
|
|
sections[current] = []
|
|
elif current is not None:
|
|
sections[current].append(line.rstrip())
|
|
result["system_prompt"] = "\n".join(sections.get("SYSTEM_PROMPT", [])).strip()
|
|
result["replay_prompt"] = "\n".join(sections.get("REPLAY_SYSTEM_PROMPT", [])).strip()
|
|
except FileNotFoundError:
|
|
pass
|
|
if not result["system_prompt"]:
|
|
result["system_prompt"] = _load_system_prompt()
|
|
return result
|
|
|
|
|
|
@router.get("/")
|
|
async def get_prompt():
|
|
return {
|
|
"script_path": str(SCRIPT_PROMPT_PATH),
|
|
"rule_path": str(RULE_PROMPT_PATH),
|
|
"system_prompt": _load_system_prompt(),
|
|
"rules": _load_rule_prompts(),
|
|
}
|
|
|
|
|
|
class PromptUpdate(BaseModel):
|
|
content: str
|
|
|
|
|
|
@router.post("/update")
|
|
async def update_prompt(payload: PromptUpdate):
|
|
if len(payload.content.encode("utf-8")) > MAX_PROMPT_BYTES:
|
|
raise HTTPException(413, f"Prompt too large (max {MAX_PROMPT_BYTES} bytes).")
|
|
try:
|
|
SCRIPTS_DIR.mkdir(parents=True, exist_ok=True)
|
|
atomic_write_text(SCRIPT_PROMPT_PATH, payload.content.rstrip() + "\n")
|
|
except OSError as exc:
|
|
raise HTTPException(500, f"Could not write prompt: {exc}")
|
|
return {"ok": True, "path": str(SCRIPT_PROMPT_PATH), "length": len(payload.content)}
|
|
|
|
|
|
@router.post("/reload")
|
|
async def reload_prompts():
|
|
rules = _load_rule_prompts()
|
|
return {
|
|
"ok": True,
|
|
"system_prompt": rules["system_prompt"],
|
|
"replay_prompt": rules["replay_prompt"],
|
|
"script_path": str(SCRIPT_PROMPT_PATH),
|
|
"rule_path": str(RULE_PROMPT_PATH),
|
|
}
|