Spaces:
Sleeping
Sleeping
| 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) | |