Astérisque + AMI + Python

Honnêtement, je me suis longtemps demandé s'il fallait publier ce matériel. Pour ceux qui savent travailler avec AMI Asterisk, il n'y a rien d'intéressant ici. Pour ceux qui commencent tout juste à faire quelque chose, il est peu probable qu'ils comprennent mon code (même si j'ai essayé d'écrire clairement). J'ai wang des commentaires comme: "Pourquoi utiliser Habr pour vos notes?". D'un autre côté, le script affiché sous la coupe peut être le point de départ de quelqu'un. Le script ne fait rien sauf qu'il envoie tous les événements de l'AMI à la console et peut les filtrer. Par exemple, je montre dans la console tous les appels qui entrent dans l'un des contextes «zadarma-in» ou «sibseti_in». Si vous êtes intéressé, je demande sous la coupe:

Il y avait une tâche en temps réel pour regarder de quelle ligne téléphonique l'appel provenait, sur quels boutons de l'ivr l'utilisateur avait appuyé, qui avait répondu à l'appel, etc. Pendant longtemps, j'ai voulu essayer de travailler avec AMI de Python, avant cela j'avais une petite expérience de Bash puis d'organiser un rappel.

Après avoir fouillé dans diverses bibliothèques prêtes à l'emploi, je me suis vite rendu compte qu'aucune d'entre elles ne me convenait. En conséquence, son propre «vélo» a été inventé sous la forme d'un script qui envoie toutes les informations d'AMI à json. Des bibliothèques Python standard sont utilisées. Le plus est que sous cette forme, il est facile de recevoir et d'analyser tous les événements et de ne pas perdre le lien vers un appel spécifique.

Le premier script imprime uniquement les événements qui tombent dans l'un des contextes «zadarma-in» ou «sibseti_in».

Script№1
import telnetlib import json import re ## HOST = "192.168.10.10" PORT = "5038" user = "zabbix" password = "password" ## tn = telnetlib.Telnet(HOST,PORT) tn.write("Action: login".encode('ascii') + b"\n") username = "Username: " + user tn.write(username.encode('ascii') + b"\n") passWord = "Secret: " + password string_NOW = '' string_out = '' cd = 0 tn.write(passWord.encode('ascii') + b"\n\n") def telnet_for_string(string): global string_out string_out_def = '' for mes in string: try: if string[mes]['Context'] == 'zadarma-in' or string[mes]['Context'] == 'sibseti_in' or string[mes]['Context'] == 'IVR': Uniqueid = string[mes]['Uniqueid'] CallerIDNum = string[mes]['CallerIDNum'] Exten = string[mes]['Exten'] CallerIDName = string[mes]['CallerIDName'] try: Digit = string[mes]['Digit'] except KeyError: Digit = '' # if Exten == 's' or Exten == 'h': # Exten = '' # Context = string[mes]['Context'] string_out_def = json.dumps({'Uniqueid': Uniqueid, 'CallerIDNum':CallerIDNum, 'CallerIDName':CallerIDName, 'Exten':Exten, 'Digit':Digit}) # print(string_out_def) except UnboundLocalError: 1+1 except KeyError: 1+1 # print(string_out_def) if string_out_def: if string_out_def != string_out: print(string_out_def) string_out = string_out_def while True: string = '' event_string = '' elements_string = '' c = 0 read_some = tn.read_some() #    AMI string = read_some.decode('utf8', 'replace').replace('\r\n', '#') #        # # print(string) #       if not string.endswith('##'): string_NOW = string_NOW + string # print('1 --->',string_NOW) #   ,      #  ,          $, #       #, -      if string.endswith('##'): string_NOW = string_NOW + string string_NOW = string_NOW.replace('##', '$') #      $ string_NOW = string_NOW.replace('\n', '#') #    # string_NOW = string_NOW.replace('\r', '#') #    # string_NOW = string_NOW.replace('"', '') #   string_NOW = string_NOW.replace('\\', '') #    # print('string_NOW -->',string_NOW) # print() #      ..         events = re.findall(r'[AZ][\w]+:\s[^$]+', string_NOW) for event in events: c+=1 # print('event ---> ',event) event_elements = re.findall(r'[AZ][\w]+:\s[^#]+', event) #       for element in event_elements: element = '\"' + element.replace(': ', '\": "') + '\", '#    # print('element', element) elements_string = elements_string + element #   ,   # event_string = event_string + '\"' + elements_string.split(':')[1].split(',')[0].replace('"','') + '\": ' + '{' + elements_string + '}' # print(elements_string) # print(str(elements_string.split(':')[1].split(',')[0])) #      json: event_string = event_string + '\"' + str(c) + '\": ' + '{' + elements_string + '}' event_string = event_string.replace('}{', '},{') #     event_string = event_string.replace(', }', '}, ') # event_string = '{' + event_string + '}' event_string = event_string.replace('}, }', '}}') #     json,       json,      # json,         . try: parsed_string = json.loads(event_string) except json.decoder.JSONDecodeError: print('#############################################', '\n\n\n') print(event_string, '\n\n\n') print(string_NOW, '\n\n\n') print('#############################################', '\n\n\n') # print(event_string) # print(parsed_string['1']) #      "telnet_for_string",       . telnet_for_string(parsed_string) string_NOW = '' #   


Et le deuxième script, qui écrit tous les événements sur la console, en regardant les deux scripts, il devient clair ce qui doit être changé pour obtenir le résultat souhaité. Si ce n'est pas tout à fait clair, alors vous devez analyser json "string [mes]" dans la fonction "def telnet_for_string (string)":

Numéro de script 2
 import telnetlib import time import json import re ## HOST = "192.168.10.10" PORT = "5038" user = "zabbix" password = "password" ## tn = telnetlib.Telnet(HOST,PORT) tn.write("Action: login".encode('ascii') + b"\n") username = "Username: " + user tn.write(username.encode('ascii') + b"\n") passWord = "Secret: " + password string_NOW = '' string_out = '' tn.write(passWord.encode('ascii') + b"\n\n") def telnet_for_string(string): for mes in string: print(string[mes]) while True: # time.sleep(0.1) string = '' event_string = '' elements_string = '' c = 0 read_some = tn.read_some() string = read_some.decode('utf8', 'replace').replace('\r\n', '#') # print(string) if not string.endswith('##'): string_NOW = string_NOW + string if string.endswith('##'): string_NOW = string_NOW + string string_NOW = string_NOW.replace('##', '$') string_NOW = string_NOW.replace('\n', '#') string_NOW = string_NOW.replace('\r', '#') string_NOW = string_NOW.replace('"', '') string_NOW = string_NOW.replace('\\', '') events = re.findall(r'[AZ][\w]+:\s[^$]+', string_NOW) for event in events: c+=1 event_elements = re.findall(r'[AZ][\w]+:\s[^#]+', event) for element in event_elements: element = '\"' + element.replace(': ', '\": "') + '\", ' elements_string = elements_string + element event_string = event_string + '\"' + str(c) + '\": ' + '{' + elements_string + '}' event_string = event_string.replace('}{', '},{') event_string = event_string.replace(', }', '}, ') event_string = '{' + event_string + '}' event_string = event_string.replace('}, }', '}}') try: parsed_string = json.loads(event_string) except json.decoder.JSONDecodeError: print('#############################################', '\n\n\n') print(event_string, '\n\n\n') print(string_NOW, '\n\n\n') print('#############################################', '\n\n\n') telnet_for_string(parsed_string) string_NOW = '' 

Source: https://habr.com/ru/post/fr415535/


All Articles