Sternchen + AMI + Python

Ehrlich gesagt habe ich lange darüber nachgedacht, ob ich dieses Material veröffentlichen soll. Für diejenigen, die wissen, wie man mit AMI Asterisk arbeitet, gibt es hier nichts Interessantes. Für diejenigen, die gerade erst anfangen, etwas zu tun, ist es unwahrscheinlich, dass sie meinen Code verstehen (obwohl ich versucht habe, klar zu schreiben). Ich wang Kommentare wie: "Warum Habr für Ihre Notizen verwenden?". Andererseits kann das unter dem Schnitt gezeigte Skript der Ausgangspunkt einer Person sein. Das Skript führt nur alle Ereignisse vom AMI an die Konsole aus und kann sie filtern. Zum Beispiel zeige ich in der Konsole alle Aufrufe, die in einen der Kontexte "zadarma-in" oder "sibseti_in" fallen. Bei Interesse frage ich unter dem Schnitt:

In Echtzeit musste überprüft werden, von welcher Amtsleitung der Anruf kam, welche Tasten der Benutzer drückte, wer den Anruf entgegennahm usw. Lange wollte ich versuchen, mit AMI von Python zu arbeiten, hatte vorher ein wenig Erfahrung mit Bash und organisierte dann einen Rückruf.

Nachdem ich in verschiedenen vorgefertigten Bibliotheken gestöbert hatte, wurde mir schnell klar, dass keine von ihnen zu mir passt. Als Ergebnis wurde sein eigenes „Fahrrad“ in Form eines Skripts erfunden, das alle Informationen von AMI an json sendet. Standard-Python-Bibliotheken werden verwendet. Das Plus ist, dass es in dieser Form einfach ist, Ereignisse zu empfangen und zu analysieren, ohne die Verbindung zu einem bestimmten Anruf zu verlieren.

Das erste Skript druckt nur die Ereignisse aus, die in einen der Kontexte "zadarma-in" oder "sibseti_in" fallen.

Skript№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 = '' #   


Und das zweite Skript, das alle Ereignisse in die Konsole schreibt und beide Skripte betrachtet, macht deutlich, was geändert werden muss, um das gewünschte Ergebnis zu erzielen. Wenn es nicht ganz klar ist, müssen Sie json "string [mes]" in der Funktion "def telnet_for_string (string)" analysieren:

Skript Nummer 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/de415535/


All Articles