# By: Md. Fahim Bin Amin (modified for Object-API endpoints) # This file contains constants and functions related to SolarWinds Service Desk integration. from dbqueries.integrations import db_solarwinds_servicedesk from utils import errors, logging, url_paths, var_names import json import requests # --------------------------------------------------------- # URL Templates (domain comes from vendor_endpoint_name) # using object-API style endpoints (no .json suffix) # --------------------------------------------------------- url_users = "{0}/users.json" url_user_by_id = "{0}/users/{1}.json" url_accounts = "{0}/accounts.json" url_companies = "{0}/companies.json" url_sites = "{0}/sites.json" url_categories = "{0}/categories.json" url_priorities = "{0}/priorities.json" url_states = "{0}/states.json" url_incidents = "{0}/incidents.json" url_incident_by_id = "{0}/incidents/{1}.json" url_incident_comments = "{0}/incidents/{1}/comments.json" # --------------------------------------------------------- # Valid Fallbacks (Since API often returns 404 for these) # --------------------------------------------------------- FALLBACK_PRIORITIES = [ {var_names.item_id: "Critical", var_names.name: "Critical"}, {var_names.item_id: "High", var_names.name: "High"}, {var_names.item_id: "Medium", var_names.name: "Medium"}, {var_names.item_id: "Low", var_names.name: "Low"} ] FALLBACK_STATES = [ {var_names.item_id: "New", var_names.name: "New"}, {var_names.item_id: "Assigned", var_names.name: "Assigned"}, {var_names.item_id: "Awaiting Input", var_names.name: "Awaiting Input"}, {var_names.item_id: "Resolved", var_names.name: "Resolved"}, {var_names.item_id: "Closed", var_names.name: "Closed"} ] # --------------------------------------------------------- # Helpers # --------------------------------------------------------- def get_auth_headers(api_key): return { "X-Samanage-Authorization": f"Bearer {api_key}", "Accept": "application/vnd.samanage.v2.1+json", "Content-Type": "application/json" } def extract_id_name(item): try: if not isinstance(item, dict): return None, None if "id" in item and "name" in item: return item["id"], item["name"] # sometimes nested under sub-object for v in item.values(): if isinstance(v, dict) and "id" in v and "name" in v: return v["id"], v["name"] return None, None except Exception: logging.exception("SolarWinds: extract_id_name failed") return None, None def flatten_category(node, out_list): if not isinstance(node, dict): return if "category" in node and isinstance(node["category"], dict): node = node["category"] cid, cname = extract_id_name(node) if cid is not None: out_list.append({var_names.item_id: cid, var_names.name: cname}) for child_key in ("children", "subcategories"): children = node.get(child_key) if isinstance(children, list): for c in children: flatten_category(c, out_list) # --------------------------------------------------------- # Admin Validation # using GET /users (or could use GET /users/{id}) # --------------------------------------------------------- def get_admin_uid(api_key, domain): base = domain.rstrip("/") url = url_users.format(base) try: resp = requests.get(url, headers=get_auth_headers(api_key)) except Exception: logging.exception("SolarWinds: /users request failed") return None if resp.status_code != 200: return None try: data = resp.json() except Exception: logging.exception("SolarWinds: /users invalid JSON") return None # Expecting list of users → return first user's id as admin if isinstance(data, list) and len(data) > 0 and "id" in data[0]: return str(data[0]["id"]) return None # --------------------------------------------------------- # Dropdown Fetching Functions # --------------------------------------------------------- def get_accounts(domain, api_key): base = domain.rstrip("/") headers = get_auth_headers(api_key) candidates = [url_accounts, url_companies, url_sites] for template in candidates: url = template.format(base) try: resp = requests.get(url, headers=headers) except Exception: logging.exception("SolarWinds: accounts request failed") continue if resp.status_code != 200: continue try: data = resp.json() except Exception: logging.exception("SolarWinds: accounts invalid JSON") return None out = [] if isinstance(data, list): for x in data: iid, nm = extract_id_name(x) if iid is not None: out.append({var_names.item_id: iid, var_names.name: nm}) elif isinstance(data, dict): for v in data.values(): if isinstance(v, list): for x in v: iid, nm = extract_id_name(x) if iid is not None: out.append({var_names.item_id: iid, var_names.name: nm}) return out return None def get_categories(domain, api_key): base = domain.rstrip("/") headers = get_auth_headers(api_key) url = url_categories.format(base) try: resp = requests.get(url, headers=headers) except Exception: logging.exception("SolarWinds: /categories request failed") return None if resp.status_code != 200: return None try: data = resp.json() except Exception: logging.exception("SolarWinds: /categories invalid JSON") return None out = [] if isinstance(data, list): for node in data: flatten_category(node, out) elif isinstance(data, dict): # maybe wrapped differently for v in data.values(): if isinstance(v, list): for node in v: flatten_category(node, out) break else: flatten_category(data, out) return out def get_priorities(domain, api_key): base = domain.rstrip("/") headers = get_auth_headers(api_key) url = url_priorities.format(base) try: resp = requests.get(url, headers=headers) except Exception: logging.exception("SolarWinds: /priorities request failed") return FALLBACK_PRIORITIES if resp.status_code != 200: return FALLBACK_PRIORITIES try: data = resp.json() except Exception: logging.exception("SolarWinds: /priorities invalid JSON") return FALLBACK_PRIORITIES out = [] if isinstance(data, list): for itm in data: iid, nm = extract_id_name(itm) if iid is not None: out.append({var_names.item_id: iid, var_names.name: nm}) elif isinstance(data, dict): for v in data.values(): if isinstance(v, list): for itm in v: iid, nm = extract_id_name(itm) if iid is not None: out.append({var_names.item_id: iid, var_names.name: nm}) if not out: return FALLBACK_PRIORITIES return out def get_statuses(domain, api_key): base = domain.rstrip("/") headers = get_auth_headers(api_key) url = url_states.format(base) try: resp = requests.get(url, headers=headers) except Exception: logging.exception("SolarWinds: /states request failed") return FALLBACK_STATES if resp.status_code != 200: return FALLBACK_STATES try: data = resp.json() except Exception: logging.exception("SolarWinds: /states invalid JSON") return FALLBACK_STATES out = [] if isinstance(data, list): for itm in data: iid, nm = extract_id_name(itm) if iid is not None: out.append({var_names.item_id: iid, var_names.name: nm}) elif isinstance(data, dict): for v in data.values(): if isinstance(v, list): for itm in v: iid, nm = extract_id_name(itm) if iid is not None: out.append({var_names.item_id: iid, var_names.name: nm}) if not out: return FALLBACK_STATES return out # --------------------------------------------------------- # Create Ticket # using POST /incidents # --------------------------------------------------------- def create_solarwinds_servicedesk_ticket( conn, timestamp, org_id, integ_key, integ_info, org_instance_id, task_title, text_msg, # basic fields extra_fields=None # new param: dict of optional SWSD fields ): """ extra_fields: optional dict. Example keys: site_id, department_id, priority, state_id, category (dict or id), subcategory (dict or id), due_at (string), custom_fields_values, etc. """ sw = db_solarwinds_servicedesk.get_solarwinds_servicedesk_account_details( conn, timestamp, org_id, integration_key=integ_key ) if not sw: raise LookupError(errors.err_integration_not_found) api_key = sw[var_names.password] domain = sw[var_names.vendor_endpoint_name] requester_id = sw.get(var_names.user_object_id) base = domain.rstrip("/") url = url_incidents.format(base) headers = get_auth_headers(api_key) body = { "incident": { "name": task_title, "description": text_msg or "" }} # Attach requester if requester_id: body["incident"]["requester"] = { "id": requester_id, "email": integ_info.get("email") # or stored admin email } # Merge optional fields if provided if extra_fields and isinstance(extra_fields, dict): for k, v in extra_fields.items(): # Only include valid, non-null values if v is not None: body["incident"][k] = v try: resp = requests.post(url, headers=headers, json=body) exe_status = resp.status_code try: exe_output = resp.json() except Exception: exe_output = {} except Exception as e: logging.exception("SolarWinds: create ticket failed") return None, 500, {"error": str(e)} ticket_id = None if exe_status in (200, 201): # New API: response usually returns incident object directly if isinstance(exe_output, dict) and "id" in exe_output: ticket_id = str(exe_output["id"]) else: inc = exe_output.get("incident") if isinstance(inc, dict) and "id" in inc: ticket_id = str(inc["id"]) return ticket_id, exe_status, exe_output # --------------------------------------------------------- # Update Ticket # using PUT /incidents/{id} + POST /incidents/{id}/comments # --------------------------------------------------------- def update_solarwinds_servicedesk_ticket( conn, timestamp, org_id, integ_key, integ_info, ticket_id, updates=None, note=None, voice_url=None ): sw = db_solarwinds_servicedesk.get_solarwinds_servicedesk_account_details( conn, timestamp, org_id, integration_key=integ_key ) if not sw: raise LookupError(errors.err_integration_not_found) api_key = sw[var_names.password] domain = sw[var_names.vendor_endpoint_name] base = domain.rstrip("/") url_get = url_incident_by_id.format(base, ticket_id) url_update = url_incident_by_id.format(base, ticket_id) url_comments = url_incident_comments.format(base, ticket_id) headers = get_auth_headers(api_key) # Fetch existing ticket try: resp = requests.get(url_get, headers=headers) if resp.status_code != 200: return ticket_id, resp.status_code, {} incident = resp.json() except Exception: return ticket_id, 500, {"error": "fetch_failed"} # ---------------------------------- # Fetch tenant-safe states # ---------------------------------- tenant_states = get_statuses(domain, api_key) state_name_to_id = { s[var_names.name]: s[var_names.item_id] for s in tenant_states if var_names.name in s and var_names.item_id in s } # ---------------------------------- # Build safe update payload # ---------------------------------- update_payload = {} if updates and isinstance(updates, dict): for k, v in updates.items(): if v is None: continue # SAFE STATE MAPPING if k == "state_id": if isinstance(v, int): logging.error(f"Numeric state_id rejected: {v}") continue real_state_id = state_name_to_id.get(v) if not real_state_id: logging.error(f"Invalid state name received: {v}") continue update_payload["state_id"] = real_state_id else: update_payload[k] = v # Voice update if voice_url: curr_desc = incident.get("description", "") or "" update_payload["description"] = ( curr_desc + f'
' ) # ---------------------------------- # SEND UPDATE # ---------------------------------- if update_payload: try: r = requests.put( url_update, headers=headers, json={"incident": update_payload} ) if r.status_code not in (200, 201): logging.error(f"SolarWinds: update failed status {r.status_code}") except Exception: logging.exception("SolarWinds: exception updating ticket") # ---------------------------------- # SAFE COMMENT ADD (NO user_id) # ---------------------------------- if note: comment_body = { "comment": { "body": note, "is_private": True } } try: r = requests.post(url_comments, headers=headers, json=comment_body) if r.status_code not in (200, 201): logging.error(f"SolarWinds: add note failed status {r.status_code}") except Exception: logging.exception("SolarWinds: exception adding note") return ticket_id, 200, {"updated_fields": update_payload}