Create generate_ticket.py

generate_ticket.py

@@ -0,0 +1,112 @@
+
+from dotenv import load_dotenv
+import requests
+import time
+import os
+from typing import Optional, Dict, Any
+
+load_dotenv()
+
+# --- ServiceNow OAuth / instance config ---
+INSTANCE_URL = os.getenv("SERVICENOW_INSTANCE_URL")            # e.g., https://your-instance.service-now.com
+CLIENT_ID = os.getenv("SERVICENOW_CLIENT_ID")
+CLIENT_SECRET = os.getenv("SERVICENOW_CLIENT_SECRET")
+
+# Optional but recommended: provide defaults for create payload (helps BR/ACL allow quick resolution)
+DEFAULT_ASSIGNMENT_GROUP_SYSID = os.getenv("SERVICENOW_ASSIGNMENT_GROUP_SYSID")  # e.g., Service Desk group sys_id
+DEFAULT_CATEGORY = os.getenv("SERVICENOW_DEFAULT_CATEGORY", "inquiry")
+DEFAULT_SUBCATEGORY = os.getenv("SERVICENOW_DEFAULT_SUBCATEGORY", "general")
+
+# OAuth token storage (simple in-memory cache)
+_token_info: Dict[str, Any] = {"access_token": None, "expires_at": 0}
+
+# Read SSL verify flag from env (default True)
+VERIFY_SSL = os.getenv("SERVICENOW_SSL_VERIFY", "true").lower() in ("1", "true", "yes")
+
+
+def get_oauth_token() -> str:
+    """
+    Fetch a new OAuth token using client credentials flow.
+    """
+    if not INSTANCE_URL or not CLIENT_ID or not CLIENT_SECRET:
+        raise RuntimeError(
+            "ServiceNow OAuth env vars are missing (SERVICENOW_INSTANCE_URL / CLIENT_ID / CLIENT_SECRET)."
+        )
+    url = f"{INSTANCE_URL}/oauth_token.do"
+    data = {
+        "grant_type": "client_credentials",
+        "client_id": CLIENT_ID,
+        "client_secret": CLIENT_SECRET,
+        "scope": "global",
+    }
+    # NOTE: verify=False is OK for local dev; set verify=True in production
+    resp = requests.post(url, data=data, verify=VERIFY_SSL, timeout=25)  # <-- use VERIFY_SSL
+    if resp.status_code == 200:
+        tok = resp.json()
+        _token_info["access_token"] = tok.get("access_token")
+        _token_info["expires_at"] = time.time() + tok.get("expires_in", 0)
+        return _token_info["access_token"]
+    raise Exception(f"Failed to get token: {resp.status_code} - {resp.text}")
+
+
+def get_valid_token() -> str:
+    """
+    Return a valid token, refreshing if expired.
+    """
+    if _token_info["access_token"] and time.time() < _token_info["expires_at"]:
+        return _token_info["access_token"]
+    return get_oauth_token()
+
+
+def create_incident(short_description: str, description: str) -> Dict[str, Any]:
+    """
+    Create an incident in ServiceNow using OAuth token.
+
+    Returns:
+      - {"number": "<INC...>", "sys_id": "<sys_id>"} on success
+      - {"error": "<message>", "status_code": <int>} on failure
+
+    NOTE:
+      - Includes optional defaults (category/subcategory/assignment_group) to satisfy common BR/ACL requirements
+        so your subsequent PATCH to Resolved succeeds.
+    """
+    token = get_valid_token()
+    url = f"{INSTANCE_URL}/api/now/table/incident"
+
+    headers = {
+        "Authorization": f"Bearer {token}",
+        "Content-Type": "application/json",
+        "Accept": "application/json",
+    }
+
+    payload: Dict[str, Any] = {
+        "short_description": short_description,
+        "description": description,
+        "urgency": "3",   # Medium
+        "priority": "3",
+        # Optional defaults—adjust for your instance if needed
+        "category": DEFAULT_CATEGORY,
+        "subcategory": DEFAULT_SUBCATEGORY,
+        "caller_id": os.getenv("SERVICENOW_CALLER_SYSID"),
+    }
+
+    # Only include assignment_group if the env var is set
+    if DEFAULT_ASSIGNMENT_GROUP_SYSID:
+        payload["assignment_group"] = DEFAULT_ASSIGNMENT_GROUP_SYSID
+
+    # NOTE: verify=False is OK for local dev; set verify=True in production
+    resp = requests.post(url, headers=headers, json=payload, verify=VERIFY_SSL, timeout=30)  # <-- use VERIFY_SSL
+
+    if resp.status_code in (200, 201):
+        try:
+            result = resp.json().get("result", {})
+        except Exception:
+            # Some instances respond with bare object
+            result = resp.json()
+        number = result.get("number")
+        sys_id = result.get("sys_id")
+        if number and sys_id:
+            return {"number": number, "sys_id": sys_id}
+        # If SN responded but fields missing, surface an error
+        return {"error": f"Missing Incident fields in response: {resp.text}", "status_code": resp.status_code}
+        # Failure: return structured error for caller (main.py /incident & /chat will handle this gracefully)