Spaces:
Sleeping
Sleeping
| import copy | |
| from typing import Any, Dict, List, Optional | |
| class TicketingApp: | |
| def __init__(self, org_config: Dict[str, Any]): | |
| self.org = org_config | |
| self.tickets: Dict[str, Dict] = {} | |
| self.counter = 0 | |
| self._episode_created: List[str] = [] | |
| def snapshot(self) -> Dict[str, Any]: | |
| return copy.deepcopy({ | |
| "tickets": list(self.tickets.values()), | |
| "episode_created_ids": self._episode_created, | |
| }) | |
| def create_ticket(self, args: Dict[str, Any]) -> Dict[str, Any]: | |
| required = self.org.get("required_ticket_fields", ["summary"]) | |
| missing = [f for f in required if not args.get(f)] | |
| if missing: | |
| return {"ok": False, "error": f"Missing required fields: {missing}. Required: {required}"} | |
| self.counter += 1 | |
| ticket_id = f"T-{self.counter:04d}" | |
| ticket = { | |
| "id": ticket_id, | |
| "summary": args.get("summary", ""), | |
| "description": args.get("description", ""), | |
| "label": args.get("label"), | |
| "priority": args.get("priority"), | |
| "status": "open", | |
| "assigned_team": None, | |
| "linked_pr": args.get("linked_pr"), | |
| "comments": [], | |
| "created_this_episode": True, | |
| } | |
| valid_labels = list(self.org["label_taxonomy"].values()) | |
| if ticket["label"] and ticket["label"] not in valid_labels: | |
| return {"ok": False, "error": f"Invalid label '{ticket['label']}'. Valid labels: {valid_labels}"} | |
| valid_priorities = self.org["priority_levels"] | |
| if ticket["priority"] and ticket["priority"] not in valid_priorities: | |
| return {"ok": False, "error": f"Invalid priority '{ticket['priority']}'. Valid: {valid_priorities}"} | |
| self.tickets[ticket_id] = ticket | |
| self._episode_created.append(ticket_id) | |
| return {"ok": True, "data": copy.deepcopy(ticket), "side_effects": [f"ticket {ticket_id} created"]} | |
| def update_ticket(self, args: Dict[str, Any]) -> Dict[str, Any]: | |
| tid = args.get("ticket_id") | |
| if not tid or tid not in self.tickets: | |
| return {"ok": False, "error": f"Ticket '{tid}' not found"} | |
| t = self.tickets[tid] | |
| for k in ("summary", "description", "label", "priority", "linked_pr"): | |
| if k in args: | |
| t[k] = args[k] | |
| return {"ok": True, "data": copy.deepcopy(t), "side_effects": [f"ticket {tid} updated"]} | |
| def get_ticket(self, args: Dict[str, Any]) -> Dict[str, Any]: | |
| tid = args.get("ticket_id") | |
| t = self.tickets.get(tid) | |
| if not t: | |
| return {"ok": False, "error": f"Ticket '{tid}' not found"} | |
| return {"ok": True, "data": copy.deepcopy(t)} | |
| def list_tickets(self, args: Dict[str, Any]) -> Dict[str, Any]: | |
| tickets = list(self.tickets.values()) | |
| if args.get("status"): | |
| tickets = [t for t in tickets if t.get("status") == args["status"]] | |
| if args.get("label"): | |
| tickets = [t for t in tickets if t.get("label") == args["label"]] | |
| limit = min(args.get("limit", 20), 50) | |
| offset = args.get("offset", 0) | |
| page = tickets[offset: offset + limit] | |
| return {"ok": True, "data": page, "total": len(tickets)} | |
| def assign_ticket(self, args: Dict[str, Any]) -> Dict[str, Any]: | |
| tid = args.get("ticket_id") | |
| team = args.get("team") | |
| if not tid or tid not in self.tickets: | |
| return {"ok": False, "error": f"Ticket '{tid}' not found"} | |
| valid_teams = list(self.org["team_map"].values()) | |
| if team not in valid_teams: | |
| return {"ok": False, "error": f"Team '{team}' not found. Valid teams: {valid_teams}"} | |
| self.tickets[tid]["assigned_team"] = team | |
| return {"ok": True, "data": {"ticket_id": tid, "assigned_team": team}, "side_effects": [f"ticket {tid} assigned to {team}"]} | |
| def comment_ticket(self, args: Dict[str, Any]) -> Dict[str, Any]: | |
| tid = args.get("ticket_id") | |
| text = args.get("text", "") | |
| if not tid or tid not in self.tickets: | |
| return {"ok": False, "error": f"Ticket '{tid}' not found"} | |
| self.tickets[tid]["comments"].append({"text": text}) | |
| return {"ok": True, "data": {"comment": "added"}, "side_effects": [f"comment added to {tid}"]} | |
| def transition_ticket(self, args: Dict[str, Any]) -> Dict[str, Any]: | |
| tid = args.get("ticket_id") | |
| status = args.get("status", "closed") | |
| valid = ["open", "in_progress", "review", "closed", "wont_fix"] | |
| if status not in valid: | |
| return {"ok": False, "error": f"Invalid status. Valid: {valid}"} | |
| if not tid or tid not in self.tickets: | |
| return {"ok": False, "error": f"Ticket '{tid}' not found"} | |
| self.tickets[tid]["status"] = status | |
| return {"ok": True, "data": {"ticket_id": tid, "status": status}, "side_effects": [f"ticket {tid} transitioned to {status}"]} | |