start:utils:cockpit:p1:cpu:cockpit-itsm
CockpitITSM.py
Accès aux API REST de Cockpit ITSM et sauvegarde des informations dans la base de données.
- get_token_db - Récupère un tokent sauvegardé dans la base de données. Si le token n'est plus valide, en redemande un
- get_token(instance, clientId, clientSecret, username, password): - Essaie d'obtenir un tokent (test de connexion)
- update_organizations - Récupère les organisations qui sont actives et qui ont le module ticketing activé
- update_ticket_status - Récupère la liste des status de tickets
- update_ticket_priorities - Récupère la liste des priorités de tickets
- get_open_tickets - Transfert au sketch Arduino de la liste des tickets P1
- update_open_tickets - Récupère les tickets ouverts pour les organisations sélectionnées
- resolve_status_from_ticket - Récupère le status d'un ticket
- CockpitITSM.py
# CockpitITSM.py import requests import threading from datetime import datetime, timedelta, timezone from urllib.parse import quote from PersistentData import get_config, save_config from PersistentData import db # Access the db instance directly # Lock to prevent concurrent access between update and read _tickets_lock = threading.Lock() def get_token_db(): # Check if existing token in DB is still valid accessToken = get_config("accessToken") expireAtStr = get_config("expireAt") if accessToken and expireAtStr: try: expireAt = datetime.strptime(expireAtStr, "%Y-%m-%d %H:%M:%S") if datetime.now() < expireAt: return True except: pass # Token absent or expired: request a new one instance = get_config("instance") clientId = get_config("clientId") clientSecret = get_config("clientSecret") username = get_config("username") password = get_config("password") if not all([instance, clientId, clientSecret, username, password]): print("ERROR: Missing credentials in DB.") return False clean_instance = instance.replace("https://", "").replace("http://", "").strip("/") save_config("instance", clean_instance) url = f"https://{clean_instance}/oauth2/token" payload = { 'grant_type': 'password', 'client_id': clientId, 'client_secret': clientSecret, 'username': username, 'password': password, 'scope': 'public-api' } headers = {'Content-Type': 'application/x-www-form-urlencoded', 'Accept': 'application/json'} try: response = requests.post(url, data=payload, headers=headers, timeout=15) if response.status_code == 200: data = response.json() accessToken = data.get("access_token") expiresIn = data.get("expires_in") expireAt = datetime.now() + timedelta(seconds=int(expiresIn)) save_config("accessToken", accessToken) save_config("expireAt", expireAt.strftime("%Y-%m-%d %H:%M:%S")) return True return False except Exception as e: print(f"Connection Error: {str(e)}") return False def get_token(instance, clientId, clientSecret, username, password): """ Core logic to fetch a token from Cockpit ITSM API. Does NOT read from or write to DB (stateless). """ clean_instance = instance.replace("https://", "").replace("http://", "").strip("/") url = f"https://{clean_instance}/oauth2/token" payload = { 'grant_type': 'password', 'client_id': clientId, 'client_secret': clientSecret, 'username': username, 'password': password, 'scope': 'public-api' # Scope adjusted to your requirement } headers = { 'Content-Type': 'application/x-www-form-urlencoded', 'Accept': 'application/json' } try: response = requests.post(url, data=payload, headers=headers, timeout=15) if response.status_code == 200: return True, response.json() else: return False, f"Status {response.status_code}: {response.text}" except Exception as e: return False, str(e) def update_organizations(): if not get_token_db(): print("ERROR: Could not obtain a valid token.") return False instance = get_config("instance") accessToken = get_config("accessToken") if not accessToken: print("No token found. Please run get_token() first.") return False headers = { 'Authorization': f'Bearer {accessToken}', 'Accept': 'application/json' } try: # 1. Get ALL organizations: source of id, name and ticketingModuleEnabled list_url = f"https://{instance}/api/organization/list" res_all = requests.get(list_url, headers=headers, timeout=20) all_orgs = res_all.json() if res_all.status_code == 200 else [] # 2. Get ACTIVE organizations: source of enabled status (id only) module, obj_type = "CORE", "ORGANIZATION" core_url = f"https://{instance}/api/repository/objects?module={module}&types={obj_type}" res_active = requests.get(core_url, headers=headers, timeout=20) active_data = res_active.json() if res_active.status_code == 200 else [] # Extract the IDs of active organizations active_ids = {item['id'] for item in active_data if 'id' in item} # 3. Synchronize with Database for org in all_orgs: org_id = org['id'] org_data = { "id": org_id, "name": org['name'], "enabled": 1 if org_id in active_ids else 0, "ticketingModuleEnabled": 1 if org.get('ticketingModuleEnabled') else 0 } existing = db.read("organizations", condition=f"id = {org_id}") if existing: # Don't touch 'selected' to preserve user choice db.update("organizations", org_data, condition=f"id = {org_id}") else: # New organization: set 'selected' to 0 by default org_data["selected"] = 0 db.store("organizations", org_data) # 4. Disable organizations no longer present in all_orgs all_api_ids = [org['id'] for org in all_orgs] if all_api_ids: ids_str = ",".join(map(str, all_api_ids)) db.update("organizations", {"enabled": 0}, condition=f"id NOT IN ({ids_str})") return True except Exception as e: print(f"Error updating organizations: {str(e)}") return False def update_teams(): """ Fetches teams for each active organization via: GET {instance}/api/organization/by-name?{name} Stores them in the teams table preserving user selection. """ if not get_token_db(): print("ERROR: Could not obtain a valid token.") return False instance = get_config("instance") accessToken = get_config("accessToken") headers = {'Authorization': f'Bearer {accessToken}', 'Accept': 'application/json'} try: orgs = db.read("organizations", condition="selected = 1 AND enabled = 1 AND ticketingModuleEnabled = 1") or [] api_team_keys = set() for org in orgs: org_id = org["id"] org_name = org["name"] url = f"https://{instance}/api/organization/by-name?name={quote(org_name, safe='')}" response = requests.get(url, headers=headers, timeout=20) if response.status_code != 200: print(f"WARNING: Could not fetch teams for org {org_name}: HTTP {response.status_code}") continue data = response.json() teams = data.get("teams", []) for team in teams: team_id = team["id"] team_name = team["name"] api_team_keys.add(team_id) existing = db.read("teams", condition=f"id = {team_id}") if existing: db.update("teams", {"name": team_name, "org_id": org_id}, condition=f"id = {team_id}") else: db.store("teams", {"id": team_id, "org_id": org_id, "name": team_name, "selected": 0}) # Remove teams no longer in API all_db_teams = db.read("teams") or [] for t in all_db_teams: if t["id"] not in api_team_keys: db.execute_sql(f"DELETE FROM teams WHERE id = {t['id']}") print(f"DEBUG: Teams updated — {len(api_team_keys)} teams found.") return True except Exception as e: print(f"Error updating teams: {str(e)}") return False def update_ticket_status(): if not get_token_db(): print("ERROR: Could not obtain a valid token.") return False instance = get_config("instance") accessToken = get_config("accessToken") headers = {'Authorization': f'Bearer {accessToken}', 'Accept': 'application/json'} module, obj_type = "TICKETING", "TICKET_STATUS" # Update API Statuses url = f"https://{instance}/api/repository/objects?module={module}&types={obj_type}" try: response = requests.get(url, headers=headers, timeout=20) if response.status_code == 200: api_data = response.json() api_ids = [] for item in api_data: oid = item.get('id') itype = item.get('type') isub = item.get('subType') api_ids.append(oid) # Match against the composite primary key cond = f"id = {oid} AND type = '{itype}' AND subType = '{isub}'" existing = db.read("objects", condition=cond) data = { "id": oid, "module": item.get('module'), "type": itype, "subType": isub, "reference": item.get('reference'), "manual": 0 } if existing: db.update("objects", data, condition=cond) else: data["selected"] = 0 db.store("objects", data) # Cleanup: only remove API-sourced entries (manual = 0) not in latest API call. # Manually resolved entries (manual = 1) are preserved. if api_ids: ids_to_keep = ",".join(map(str, api_ids)) if ids_to_keep: db.execute_sql(f""" DELETE FROM objects WHERE type = '{obj_type}' AND manual = 0 AND id NOT IN ({ids_to_keep}) """) return True return False except Exception as e: print(f"Error updating ticket status: {str(e)}") return False def update_ticket_priorities(): if not get_token_db(): print("ERROR: Could not obtain a valid token.") return False instance = get_config("instance") accessToken = get_config("accessToken") headers = {'Authorization': f'Bearer {accessToken}', 'Accept': 'application/json'} # Static statuses that must always exist module, obj_type = "TICKETING", "TICKET_PRIORITY" # Update API Statuses url = f"https://{instance}/api/repository/objects?module={module}&types={obj_type}" try: response = requests.get(url, headers=headers, timeout=20) if response.status_code == 200: api_data = response.json() api_ids = [] for item in api_data: oid = item.get('id') itype = item.get('type') isub = item.get('subType') api_ids.append(oid) # Match against the composite primary key cond = f"id = {oid} AND type = '{itype}' AND subType = '{isub}'" existing = db.read("objects", condition=cond) data = { "id": oid, "module": item.get('module'), "type": itype, "subType": isub, "reference": item.get('reference') } if existing: db.update("objects", data, condition=cond) else: data["selected"] = 0 db.store("objects", data) # Cleanup: Delete IDs not present in the latest API call if api_ids: ids_str = ",".join(map(str, api_ids)) cleanup_sql = f""" DELETE FROM objects WHERE type = '{obj_type}' AND id NOT IN ({ids_str}) """ db.execute_sql(cleanup_sql) return True return False except Exception as e: print(f"Error updating ticket priorities: {str(e)}") return False def get_open_tickets(): """ Returns the list of tickets as a JSON string (no API call). Called by the Arduino Bridge. Format: [{"o":"org","t":"type","i":id,"n":"title","c":criticity}, ...] """ import json with _tickets_lock: rows = db.read("v_active_tickets") if not rows: return "[]" compact = [ { "o": r.get("organization_name", ""), "t": r.get("type", ""), "i": r.get("id", 0), "n": r.get("title", ""), "c": r.get("criticity", 0) } for r in rows ] return json.dumps(compact, ensure_ascii=False) def update_open_tickets(on_complete=None): """ Fetches open tickets for all selected organizations and stores them in the DB. Iterates pages until all results are retrieved, then truncates to totalNumberOfElements. Replaces all tickets in DB on each call (DELETE + INSERT). Called periodically (every 4 minutes) by the scheduler. on_complete: optional callback(count) called after a successful update. """ cycle_ts = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S") if not get_token_db(): print("ERROR: Could not obtain a valid token.") return False instance = get_config("instance") accessToken = get_config("accessToken") headers = { 'Authorization': f'Bearer {accessToken}', 'Accept': 'application/json', 'Content-Type': 'application/json' } url = f"https://{instance}/api/ticket/list/open" # Get selected organizations selected_orgs = db.read("organizations", condition="selected = 1 AND enabled = 1") if not selected_orgs: print("DEBUG: No selected organizations, skipping ticket fetch.") return True all_tickets = [] try: for org in selected_orgs: org_id = org["id"] org_name = org["name"] page = 0 while True: payload = { "organizationId": org_id, "page": page, "pageSize": 200 } response = requests.post(url, json=payload, headers=headers, timeout=30) if response.status_code != 200: print(f"WARNING: HTTP {response.status_code} fetching tickets for org {org_id} — skipping org") break data = response.json() total_pages = data.get("totalNumberOfPages", 1) total_elements = data.get("totalNumberOfElements", 0) page_tickets = data.get("content", []) if not page_tickets and page == 0: print(f"WARNING: Empty content for org {org_id} (totalElements={total_elements})") all_tickets.extend(page_tickets) if page == 0 and total_pages > 1: page += 1 continue elif page < total_pages - 1: page += 1 continue else: break # Replace all tickets in DB (lock to prevent reads during update) with _tickets_lock: db.execute_sql("DELETE FROM tickets") for t in all_tickets: org = t.get("organization") or {} status = t.get("status") or {} priority = t.get("priority") or {} assignedTeam = t.get("assignedTeam") or {} assignedUser = t.get("assignedUser") or {} ticket_data = { "id": t.get("id"), "type": t.get("type", ""), "title": t.get("title", ""), "organization_id": org.get("id", 0), "organization_name": org.get("name", ""), "status_id": status.get("id", 0), "status_reference": status.get("reference", ""), "priority_id": priority.get("id", 0), "priority_reference": priority.get("reference", ""), "assignedTeam_id": assignedTeam.get("id", 0), "assignedTeam_name": assignedTeam.get("name", ""), "assignedUser_id": assignedUser.get("id", 0), "assignedUser_name": assignedUser.get("name", ""), } db.store("tickets", ticket_data) count = len(all_tickets) if count == 0: print("WARNING: 0 tickets returned — skipping history storage (possible API issue)") else: print(f"DEBUG: {count} tickets fetched.") # Only persist statistics if we have tickets — never store 0 to avoid corrupting history if count > 0: from PersistentData import store_history from collections import defaultdict # Build per-(org_id, type) counts from all_tickets total_by_org_type = defaultdict(int) for t in all_tickets: org_id = t.get("organization", {}).get("id", 0) ttype = t.get("type", "UNKNOWN") total_by_org_type[(org_id, ttype)] += 1 # Build per-(org_id, type) counts from v_active_tickets (filtered) filtered_rows = db.read("v_active_tickets") filtered_by_org_type = defaultdict(int) for r in (filtered_rows or []): org_name = r.get("organization_name", "") ttype = r.get("type", "UNKNOWN") org_rows = db.read("organizations", condition=f"name = '{org_name}'") org_id = org_rows[0]["id"] if org_rows else 0 filtered_by_org_type[(org_id, ttype)] += 1 # Store TICKETS_TOTAL_{ORGID}_{TYPE} for (org_id, ttype), val in total_by_org_type.items(): store_history(f"TICKETS_TOTAL_{org_id}_{ttype}", val, cycle_ts) # Store TICKETS_FILTERED_{ORGID}_{TYPE} all_keys = set(total_by_org_type.keys()) | set(filtered_by_org_type.keys()) for (org_id, ttype) in all_keys: store_history(f"TICKETS_FILTERED_{org_id}_{ttype}", filtered_by_org_type.get((org_id, ttype), 0), cycle_ts) if on_complete: on_complete(count) return True except Exception as e: print(f"ERROR in get_open_tickets: {str(e)}") return False def resolve_status_from_ticket(ticket_id, sub_type): """ Fetches a single ticket by ID and returns its status id and reference. Used to resolve the real reference of a fixed status on a given instance. Returns: (True, {"status_id": int, "status_reference": str, "sub_type": str}) (False, error_message) """ if not get_token_db(): return False, "Could not obtain a valid token." instance = get_config("instance") accessToken = get_config("accessToken") headers = {'Authorization': f'Bearer {accessToken}', 'Accept': 'application/json'} url = f"https://{instance}/api/ticket/{ticket_id}" try: response = requests.get(url, headers=headers, timeout=15) if response.status_code == 200: data = response.json() status = data.get("status") or {} t_type = data.get("type", "") status_id = status.get("id") status_ref = status.get("reference", "") if not status_id: return False, f"Ticket #{ticket_id} has no status." if t_type != sub_type: return False, ( f"Ticket #{ticket_id} is of type '{t_type}', " f"expected '{sub_type}'." ) return True, { "status_id": status_id, "status_reference": status_ref, "sub_type": t_type } elif response.status_code == 404: return False, f"Ticket #{ticket_id} not found." else: return False, f"API error {response.status_code}." except Exception as e: return False, str(e)
start/utils/cockpit/p1/cpu/cockpit-itsm.txt · Last modified: by admin_wiki