Spaces:
Sleeping
Sleeping
File size: 3,162 Bytes
06671aa | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 | from __future__ import annotations
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
from fastapi import HTTPException
from models import PlaybookDetail, PlaybookSummary
ROOT_DIR = Path(__file__).resolve().parent
PLAYBOOK_DIR = ROOT_DIR / "config" / "playbooks"
from prompts.chat_role_briefs import CHAT_ROLE_BRIEFS, chat_role_brief as _chat_role_brief
PLAYBOOKS: dict[str, dict[str, Any]] = {
"buyer": {
"name": "Zamil Buyer-Side Review",
"description": "Buyer/customer-side review using config/playbooks/buyer.yaml.",
"filename": "buyer.yaml",
"is_editable": False,
"chat_brief": CHAT_ROLE_BRIEFS["buyer"],
},
"seller": {
"name": "Zamil Seller-Side Review",
"description": "Seller/supplier-side review using config/playbooks/seller.yaml.",
"filename": "seller.yaml",
"is_editable": False,
"chat_brief": CHAT_ROLE_BRIEFS["seller"],
},
}
def _path_for(playbook_id: str) -> Path:
spec = PLAYBOOKS.get(playbook_id)
if not spec:
raise HTTPException(status_code=404, detail=f"Unknown playbook: {playbook_id}")
return PLAYBOOK_DIR / spec["filename"]
def _updated_at(path: Path) -> str | None:
if not path.exists():
return None
return datetime.fromtimestamp(path.stat().st_mtime, timezone.utc).isoformat()
def _is_editable(playbook_id: str) -> bool:
return bool(PLAYBOOKS.get(playbook_id, {}).get("is_editable"))
def chat_role_brief(playbook_id: str) -> str:
return _chat_role_brief(playbook_id)
def list_playbooks() -> list[PlaybookSummary]:
items: list[PlaybookSummary] = []
for playbook_id, spec in PLAYBOOKS.items():
path = _path_for(playbook_id)
items.append(
PlaybookSummary(
id=playbook_id,
name=spec["name"],
description=spec["description"],
source_path=str(path),
updated_at=_updated_at(path),
is_editable=_is_editable(playbook_id),
)
)
return items
def get_playbook(playbook_id: str) -> PlaybookDetail:
spec = PLAYBOOKS.get(playbook_id)
path = _path_for(playbook_id)
if not path.exists():
raise HTTPException(status_code=500, detail=f"Playbook file missing: {path}")
return PlaybookDetail(
id=playbook_id,
name=spec["name"],
description=spec["description"],
source_path=str(path),
updated_at=_updated_at(path),
is_editable=_is_editable(playbook_id),
yaml_text=path.read_text(encoding="utf-8"),
)
def update_playbook(playbook_id: str, yaml_text: str) -> PlaybookDetail:
if not _is_editable(playbook_id):
raise HTTPException(
status_code=403,
detail="This base playbook is locked. Create a new playbook instead of editing the original.",
)
path = _path_for(playbook_id)
if not yaml_text.strip():
raise HTTPException(status_code=400, detail="Playbook YAML cannot be empty.")
path.write_text(yaml_text, encoding="utf-8")
return get_playbook(playbook_id)
|