Spaces:
Sleeping
Sleeping
File size: 4,973 Bytes
1f213fe | 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 102 103 104 105 106 107 108 | 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}"]}
|