如何在Python中从WebSocket解析JSON数据到变量

huangapple go评论100阅读模式
英文:

How to parse JSON data from websocket to variable in Python

问题

我使用AGI Asterisk进行语音识别。它通过WebSocket连接并接收JSON响应。
脚本收集单词并将它们粘合成短语。
到目前为止,一切都正常。

我需要进一步使用收集到的短语(发送到Telegram,然后通过API发送到帮助台系统),但如果我只使用text作为变量,它不起作用。
如何将text设置为变量?

#!/usr/bin/python3
from asterisk.agi import *
import os
from websocket import create_connection
import json
import traceback
import requests

AUDIO_FD = 3
CONTENT_TYPE = 'audio/l16; rate=8000; channels=1'
ACCEPT = 'audio/pcm'

def telegram_bot_sendtext(text):
    bot_token = '6069wxts_nWcA'
    bot_chatID = '-10300'
    bot_message = text
    send_text = 'https://api.telegram.org/bot' + bot_token + '/sendMessage?chat_id=' + bot_chatID + '&reply_to_message_id=2&parse_mode=Markdown&text=' + bot_message
    response = requests.get(send_text)
    return response.json()

def process_chunk(agi, ws, buf):
    agi.verbose("Processing chunk")
    ws.send_binary(buf)
    res = json.loads(ws.recv())
    agi.verbose("Result: " + str(res))
    if 'result' in res:
        text = " ".join([w['word'] for w in res['result']])
        os.system("espeak -w /tmp/response22.wav \"" + text.encode('utf-8') + "\"")
        os.system("sox /tmp/response22.wav -r 8000 /tmp/response.wav")
        agi.stream_file("/tmp/response")
        os.remove("/tmp/response.wav")

def startAGI():
    agi = AGI()
    agi.verbose("EAGI script started...")
    ani = agi.env['agi_callerid']
    did = agi.env['agi_extension']
    agi.verbose("Call answered from: %s to %s" % (ani, did))
    ws = create_connection("ws://localhost:2700")
    ws.send('{ "config" : { "sample_rate" : 8000 } }')
    agi.verbose("Connection created")
    try:
        while True:
            data = os.read(AUDIO_FD, 8000)
            if not data:
                break
            process_chunk(agi, ws, data)
    except Exception as err:
        agi.verbose(''.join(traceback.format_exception(type(err), err, err.__traceback__)).replace('\n', ' '))
    try:
        telegram_bot_sendtext(text)
    except Exception as exc:
        print("Post_Auth_Script_Telega : Error : ", str(exc))
    finally:
        ws.close()

startAGI()

JSON看起来像这样

Result: {
'result': [
{'conf': 1.0, 'end': 2.07, 'start': 1.71, 'word': 'One'},
{'conf': 1.0, 'end': 2.34, 'start': 2.07, 'word': 'Two'},
],
'text': 'One Two'}


<details>
<summary>英文:</summary>
I am using agi asterisk for speech recognition. It connects via websocket and receives a response in json.
The script collects words and glues them into a phrase.
Everything is working at this stage.
I need to use the collected phrase further (send to telegram and then via api to the helpdesk system), but it doesn&#39;t work if i just use **text** as a variable.
How to make ***text*** a variable?
#!/usr/bin/python3
from asterisk.agi import *
import os
from websocket import create_connection
import json
import traceback
import requests
AUDIO_FD = 3
CONTENT_TYPE = &#39;audio/l16; rate=8000; channels=1&#39;
ACCEPT = &#39;audio/pcm&#39;
def telegram_bot_sendtext(text):
bot_token = &#39;6069wxts_nWcA&#39;
bot_chatID = &#39;-10300&#39;
bot_message = text
send_text = &#39;https://api.telegram.org/bot&#39; + bot_token + &#39;/sendMessage?chat_id=&#39; + bot_chatID + &#39;&amp;reply_to_message_id=2&amp;parse_mode=Markdown&amp;text=&#39; + bot_message
response = requests.get(send_text)
return response.json()
def process_chunk(agi, ws, buf):
agi.verbose(&quot;Processing chunk&quot;)
ws.send_binary(buf)
res = json.loads(ws.recv())
agi.verbose(&quot;Result: &quot; + str(res))
if &#39;result&#39; in res:
**text = &quot; &quot;.join([w[&#39;word&#39;] for w in res[&#39;result&#39;]])**
os.system(&quot;espeak -w /tmp/response22.wav \&quot;&quot; + text.encode(&#39;utf-8&#39;) + &quot;\&quot;&quot;)
os.system(&quot;sox /tmp/response22.wav -r 8000 /tmp/response.wav&quot;)
agi.stream_file(&quot;/tmp/response&quot;)
os.remove(&quot;/tmp/response.wav&quot;)
def startAGI():
agi = AGI()
agi.verbose(&quot;EAGI script started...&quot;)
ani = agi.env[&#39;agi_callerid&#39;]
did = agi.env[&#39;agi_extension&#39;]
agi.verbose(&quot;Call answered from: %s to %s&quot; % (ani, did))
ws = create_connection(&quot;ws://localhost:2700&quot;)
ws.send(&#39;{ &quot;config&quot; : { &quot;sample_rate&quot; : 8000 } }&#39;)
agi.verbose(&quot;Connection created&quot;)
try:
while True:
data = os.read(AUDIO_FD, 8000)
if not data:
break
process_chunk(agi, ws, data)
except Exception as err:
agi.verbose(&#39;&#39;.join(traceback.format_exception(type(err), err, err.__traceback__)).replace(&#39;\n&#39;, &#39; &#39;))
try:
telegram_bot_sendtext(text)
except Exception as exc:
print(&quot;Post_Auth_Script_Telega : Error : &quot;, str(exc))
finally:
ws.close()
startAGI()
json looks like
Result: {
&#39;result&#39;: [
{&#39;conf&#39;: 1.0, &#39;end&#39;: 2.07, &#39;start&#39;: 1.71, &#39;word&#39;: &#39;One&#39;}, 
{&#39;conf&#39;: 1.0, &#39;end&#39;: 2.34, &#39;start&#39;: 2.07, &#39;word&#39;: &#39;Two&#39;}, 
], 
&#39;text&#39;: &#39;One Two&#39;}
</details>
# 答案1
**得分**: 0
以下是代码的翻译部分:
```python
#!/usr/bin/python3
from asterisk.agi import *
import os
from websocket import create_connection
import json
import traceback
import requests
AUDIO_FD = 3
CONTENT_TYPE = 'audio/l16; rate=8000; channels=1'
ACCEPT = 'audio/pcm'
def telegram_bot_sendtext(text):
bot_token = '606922aNwxts_nWcA'
bot_chatID = '-100100'
bot_message = text 
send_text = 'https://api.telegram.org/bot' + bot_token + '/sendMessage?chat_id=' + bot_chatID + '&amp;reply_to_message_id=2&amp;parse_mode=Markdown&amp;text=' + bot_message
response = requests.get(send_text)
return response.json()
def process_chunk(agi, ws, buf):
agi.verbose("Processing chunk")
ws.send_binary(buf)
res = json.loads(ws.recv())
agi.verbose("Result: " + str(res))
if 'result' in res:
text = " ".join([w['word'] for w in res['result']])
try:
telegram_bot_sendtext(text)
except Exception as exc:
print("Error: ", str(exc))
os.system("espeak -w /tmp/response22.wav \"" + text.encode('utf-8') + "\"")
os.system("sox /tmp/response22.wav -r 8000 /tmp/response.wav")
agi.stream_file("/tmp/response")
os.remove("/tmp/response.wav")
def startAGI():
agi = AGI()
agi.verbose("EAGI script started...")
ani = agi.env['agi_callerid']
did = agi.env['agi_extension']
agi.verbose("Call answered from: %s to %s" % (ani, did))
ws = create_connection("ws://localhost:2700")
ws.send('{ "config" : { "sample_rate" : 8000 } }')
agi.verbose("Connection created")
try:
while True:
data = os.read(AUDIO_FD, 8000)
if not data:
break
process_chunk(agi, ws, data)
except Exception as err:
agi.verbose(' '.join(traceback.format_exception(type(err), err, err.__traceback__)).replace('\n', ' '))
#    try:
#        telegram_bot_sendtext(text)
#    except Exception as exc:
#        print("Error: ", str(exc))
finally:
ws.close()
startAGI()

希望这对你有所帮助。如果你需要进一步的信息或翻译,请告诉我。

英文:

Final working verison

#!/usr/bin/python3
from asterisk.agi import *
import os
from websocket import create_connection
import json
import traceback
import requests
AUDIO_FD = 3
CONTENT_TYPE = &#39;audio/l16; rate=8000; channels=1&#39;
ACCEPT = &#39;audio/pcm&#39;
def telegram_bot_sendtext(text):
bot_token = &#39;606922aNwxts_nWcA&#39;
bot_chatID = &#39;-100100&#39;
bot_message = text 
send_text = &#39;https://api.telegram.org/bot&#39; + bot_token + &#39;/sendMessage?chat_id=&#39; + bot_chatID + &#39;&amp;reply_to_message_id=2&amp;parse_mode=Markdown&amp;text=&#39; + bot_message
response = requests.get(send_text)
return response.json()
def process_chunk(agi, ws, buf):
agi.verbose(&quot;Processing chunk&quot;)
ws.send_binary(buf)
res = json.loads(ws.recv())
agi.verbose(&quot;Result: &quot; + str(res))
if &#39;result&#39; in res:
text = &quot; &quot;.join([w[&#39;word&#39;] for w in res[&#39;result&#39;]])
try:
telegram_bot_sendtext(text)
except Exception as exc:
print(&quot;rror : &quot;, str(exc))
os.system(&quot;espeak -w /tmp/response22.wav \&quot;&quot; + text.encode(&#39;utf-8&#39;) + &quot;\&quot;&quot;)
os.system(&quot;sox /tmp/response22.wav -r 8000 /tmp/response.wav&quot;)
agi.stream_file(&quot;/tmp/response&quot;)
os.remove(&quot;/tmp/response.wav&quot;)
def startAGI():
agi = AGI()
agi.verbose(&quot;EAGI script started...&quot;)
ani = agi.env[&#39;agi_callerid&#39;]
did = agi.env[&#39;agi_extension&#39;]
agi.verbose(&quot;Call answered from: %s to %s&quot; % (ani, did))
ws = create_connection(&quot;ws://localhost:2700&quot;)
ws.send(&#39;{ &quot;config&quot; : { &quot;sample_rate&quot; : 8000 } }&#39;)
agi.verbose(&quot;Connection created&quot;)
try:
while True:
data = os.read(AUDIO_FD, 8000)
if not data:
break
process_chunk(agi, ws, data)
except Exception as err:
agi.verbose(&#39;&#39;.join(traceback.format_exception(type(err), err, err.__traceback__)).replace(&#39;\n&#39;, &#39; &#39;))
#    try:
#        telegram_bot_sendtext(text)
#    except Exception as exc:
#        print(&quot;rror : &quot;, str(exc))
finally:
ws.close()
startAGI()

huangapple
  • 本文由 发表于 2023年7月27日 20:38:19
  • 转载请务必保留本文链接:https://go.coder-hub.com/76779811.html
匿名

发表评论

匿名网友

:?: :razz: :sad: :evil: :!: :smile: :oops: :grin: :eek: :shock: :???: :cool: :lol: :mad: :twisted: :roll: :wink: :idea: :arrow: :neutral: :cry: :mrgreen:

确定