43 lines
3.1 KiB
Python
43 lines
3.1 KiB
Python
"""Synthetic-only HTTPS acceptance; Python 3.6+, never disables TLS checks."""
|
|
import hashlib,json,pathlib,re,ssl,sys,time,urllib.error,urllib.request,uuid
|
|
ENV=pathlib.Path(sys.argv[1])
|
|
WAV=pathlib.Path(sys.argv[2])
|
|
MODE=sys.argv[3] if len(sys.argv)>3 else 'all'
|
|
CANARIES=json.loads(pathlib.Path(sys.argv[4]).read_text(encoding='utf-8')) if len(sys.argv)>4 else ['紫色海豚','金色河流','银色树叶','红色石榴']
|
|
env=dict(line.split('=',1) for line in ENV.read_text(encoding='utf-8').splitlines() if '=' in line)
|
|
base=env['DIFY_ASR_BASE_URL'];key=env['DIFY_ASR_APP_KEY']
|
|
ctx=ssl.create_default_context()
|
|
def request(path,data,content_type,auth=True):
|
|
h={'Content-Type':content_type}
|
|
if auth:h['Authorization']='Bearer '+key
|
|
req=urllib.request.Request(base+path,data=data,headers=h,method='POST')
|
|
start=time.monotonic()
|
|
try:
|
|
with urllib.request.urlopen(req,context=ctx,timeout=200) as response:
|
|
code=response.status;body=response.read().decode()
|
|
except urllib.error.HTTPError as error:
|
|
code=error.code;body=error.read().decode()
|
|
return {'status':code,'elapsed_seconds':round(time.monotonic()-start,3),'body':json.loads(body)}
|
|
def audio():
|
|
boundary='synthetic'+uuid.uuid4().hex
|
|
body=('--'+boundary+'\r\nContent-Disposition: form-data; name="user"\r\n\r\nsynthetic-asr-acceptance-20261009\r\n--'+boundary+'\r\nContent-Disposition: form-data; name="file"; filename="synthetic.wav"\r\nContent-Type: audio/wav\r\n\r\n').encode()+WAV.read_bytes()+('\r\n--'+boundary+'--\r\n').encode()
|
|
return request('/audio-to-text',body,'multipart/form-data; boundary='+boundary)
|
|
r={'synthetic':True,'tls_verify':True,'check_hostname':ctx.check_hostname,'ssl':ssl.OPENSSL_VERSION,'mode':MODE,'app_id':env['DIFY_ASR_APP_ID'],'base_url':base,'binding_revision':env['DIFY_BINDING_REVISION']}
|
|
if MODE in ('all','asr'):
|
|
r['unauthorized']=request('/audio-to-text',b'','application/octet-stream',False)
|
|
r['asr']=audio()
|
|
text=r['asr']['body'].get('text','')
|
|
r['audio_only_canaries']={x:x in text for x in CANARIES}
|
|
r['asr_pass']=r['unauthorized']['status']==401 and r['asr']['status']==200 and all(r['audio_only_canaries'].values())
|
|
if MODE in ('all','chat'):
|
|
query='这是独立的合成 JSON 验收，不包含患者数据。请只输出 JSON 对象：{"canary":"BLUE_KITE_20261009","sum":42,"preview_only":true}。不要添加其他字段、解释或 Markdown。'
|
|
payload={'inputs':{},'query':query,'response_mode':'blocking','user':'synthetic-chat-acceptance-20261009','auto_generate_name':False}
|
|
r['chat']=request('/chat-messages',json.dumps(payload,ensure_ascii=False).encode(),'application/json')
|
|
answer=r['chat']['body'].get('answer','')
|
|
try:parsed=json.loads(answer)
|
|
except ValueError:parsed=None
|
|
r['chat_pass']=r['chat']['status']==200 and parsed=={'canary':'BLUE_KITE_20261009','sum':42,'preview_only':True} and bool(r['chat']['body'].get('message_id')) and '<think>' not in answer
|
|
r['passed']=all(r.get(k,True) for k in ['asr_pass','chat_pass'])
|
|
print(json.dumps(r,ensure_ascii=True,indent=2))
|
|
sys.exit(0 if r['passed'] else 1)
|