108 lines
6.9 KiB
Python
108 lines
6.9 KiB
Python
#!/usr/bin/env python3
|
|
"""Explicit live fake-microphone qualification. Never physical microphone evidence.
|
|
|
|
Reads ignored protected deployment credentials; emits only safe qualification facts.
|
|
Requires local synthetic PCM fixture and deployed Hope. No deployment changes.
|
|
"""
|
|
import argparse
|
|
import json
|
|
import re
|
|
import time
|
|
from pathlib import Path
|
|
from playwright.sync_api import sync_playwright, expect
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument('--url', default='https://192.168.86.2:9178')
|
|
parser.add_argument('--credentials', default='.runtime/deployment.json')
|
|
parser.add_argument('--wav', default='.runtime/voice-qualification/microphone.wav')
|
|
parser.add_argument('--timeout', type=int, default=180)
|
|
args = parser.parse_args()
|
|
values = json.loads(Path(args.credentials).read_text())
|
|
fixture = Path(args.wav).resolve(strict=True)
|
|
receipt = {'mode': 'synthetic_browser_microphone_not_physical', 'connected': False,
|
|
'user_transcript': False, 'assistant_transcript': False,
|
|
'audio_track': False, 'audio_bytes': 0, 'errors': 0}
|
|
with sync_playwright() as playwright:
|
|
browser = playwright.chromium.launch(headless=True, args=[
|
|
'--use-fake-ui-for-media-stream', '--use-fake-device-for-media-stream',
|
|
'--use-file-for-fake-audio-capture=' + str(fixture),
|
|
'--autoplay-policy=no-user-gesture-required'])
|
|
context = browser.new_context(ignore_https_errors=True, permissions=['microphone'],
|
|
viewport={'width': 1440, 'height': 1000})
|
|
page = context.new_page()
|
|
page.add_init_script("window.__qualificationVoiceID=null;window.addEventListener('hope:voice-state',event=>{if(event.detail?.conversation_id)window.__qualificationVoiceID=event.detail.conversation_id;});")
|
|
page.on('pageerror', lambda error: receipt.update(errors=receipt['errors'] + 1))
|
|
# Read-only receiver stats: exposes no tokens, credentials or private payloads.
|
|
page.add_init_script('''window.__voiceStats={bytes:0,packets:0};
|
|
const original=RTCPeerConnection.prototype.getStats;
|
|
const connections=[]; const Native=window.RTCPeerConnection;
|
|
window.RTCPeerConnection=new Proxy(Native,{construct(target,args){
|
|
const connection=Reflect.construct(target,args); connections.push(connection); return connection;}});
|
|
setInterval(async()=>{for(const connection of connections){try{
|
|
const report=await original.call(connection);
|
|
report.forEach(item=>{if(item.type==='inbound-rtp'&&(item.kind==='audio'||item.mediaType==='audio')){
|
|
window.__voiceStats.bytes=Math.max(window.__voiceStats.bytes,item.bytesReceived||0);
|
|
window.__voiceStats.packets=Math.max(window.__voiceStats.packets,item.packetsReceived||0);}});
|
|
}catch{}}},500);''')
|
|
try:
|
|
page.goto(args.url.rstrip('/') + '/', wait_until='networkidle')
|
|
expect(page.locator('#loginSubmit')).to_be_enabled(timeout=30000)
|
|
page.locator('#loginUsername').fill('admin')
|
|
page.locator('#loginPassword').fill(values['admin_password'])
|
|
page.locator('#loginSubmit').click()
|
|
expect(page.locator('#primaryInput')).to_be_enabled(timeout=30000)
|
|
# Prefer restored prominent voice entry, tolerate current compatibility ID.
|
|
for selector in ['#talkToHope', '#talkVoice', '#openVoice']:
|
|
if page.locator(selector).count() and page.locator(selector).is_visible():
|
|
page.locator(selector).click()
|
|
break
|
|
expect(page.locator('#voiceModeBtn')).to_be_visible(timeout=10000)
|
|
page.locator('#voiceModeBtn').click()
|
|
expect(page.locator('#startBtn')).to_be_enabled(timeout=10000)
|
|
page.locator('#startBtn').click()
|
|
deadline = time.monotonic() + args.timeout
|
|
while time.monotonic() < deadline:
|
|
status = page.locator('#status').inner_text()
|
|
body = page.locator('#messages').inner_text() if page.locator('#messages').count() else ''
|
|
receipt['connected'] |= 'Microphone on' in status or 'Secure room' in status
|
|
# Synthetic question and expected answer facts, not transcript dumps.
|
|
user_text = ' '.join(page.locator('#messages .message-row.user').all_inner_texts()).lower()
|
|
receipt['user_transcript'] |= bool(re.search(r'(?:two|2)\s*(?:plus|\+)\s*(?:two|2)', user_text))
|
|
assistant_text = ' '.join(page.locator('#messages .message-row.agent').all_inner_texts()).lower()
|
|
receipt['assistant_transcript'] |= 'four' in assistant_text or '4' in assistant_text
|
|
receipt['audio_track'] |= page.locator('audio[id^="audio-"]').count() > 0
|
|
stats = page.evaluate('window.__voiceStats')
|
|
receipt['audio_bytes'] = stats['bytes']
|
|
if receipt['assistant_transcript'] and receipt['audio_bytes'] > 0:
|
|
break
|
|
page.wait_for_timeout(500)
|
|
receipt['audio_playing'] = page.evaluate('''Array.from(document.querySelectorAll('audio[id^="audio-"]')).some(el=>!el.paused&&el.srcObject&&el.srcObject.getAudioTracks().some(track=>track.readyState==='live'))''')
|
|
output = Path('.runtime/voice-qualification/live-receipt.json')
|
|
output.write_text(json.dumps(receipt, indent=2) + '\n')
|
|
print(json.dumps(receipt, sort_keys=True))
|
|
assert receipt['connected'], 'synthetic_voice_connection_failed'
|
|
assert receipt['user_transcript'], 'synthetic_voice_stt_not_observed'
|
|
assert receipt['assistant_transcript'], 'synthetic_voice_answer_not_observed'
|
|
assert receipt['audio_track'] and receipt['audio_bytes'] > 0, 'remote_tts_audio_not_observed'
|
|
finally:
|
|
if page.locator('#stopBtn').count() and page.locator('#stopBtn').is_visible() and page.locator('#stopBtn').is_enabled():
|
|
page.locator('#stopBtn').click()
|
|
# Delete only the ID captured for this newly created qualification
|
|
# voice conversation. Never enumerate or remove other account chats.
|
|
try:
|
|
cleanup = page.evaluate('''async()=>{let removed=false;const cid=window.__qualificationVoiceID;
|
|
if(cid){try{const result=await HopeAuth.request('/api/conversations/'+encodeURIComponent(cid),{method:'DELETE'});removed=result.ok===true;}catch{}}
|
|
try{await HopeAuth.request('/api/auth/logout',{method:'POST'});}catch{}
|
|
return {created_conversation_seen:!!cid,removed};}''')
|
|
print(json.dumps({'qualification_cleanup': cleanup}, sort_keys=True))
|
|
except Exception:
|
|
print('qualification_cleanup_unavailable')
|
|
context.close()
|
|
browser.close()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|