ute2e/tools/bridge-narration-quality.py

"""Check available clips against source audio and run an independent ASR screen.

ASR similarity is a triage aid, not a pronunciation or visual lip-sync verdict.
"""
from pathlib import Path
import difflib
import json
import re
import subprocess
import sys

import numpy as np
from scipy.signal import correlate, correlation_lags

ROOT=Path(__file__).resolve().parents[3]
sys.path.insert(0,str(ROOT/'ai_person/ai_person_api'))
from app.config import get_settings
from app.services.asr_service import AsrService

REPORT=ROOT/'unreal_tran/ute2e/reports/bridge-narration-junhao-20260921'
WEB=ROOT/'unreal_tran/unreal_tran_web/public'

def pcm(settings,path):
    return np.frombuffer(subprocess.run([settings.ffmpeg_binary,'-v','error','-i',str(path),
        '-vn','-ac','1','-ar','16000','-f','f32le','pipe:1'],capture_output=True,check=True).stdout,dtype=np.float32)

def normalize(text):
    numbers=['零','一','二','三','四','五','六','七','八','九','十','十一','十二','十三','十四','十五','十六']
    text=re.sub(r'\d+',lambda m:numbers[int(m[0])] if int(m[0])<len(numbers) else m[0],text)
    return re.sub(r'[^\u4e00-\u9fffA-Za-z0-9]','',text)

def main():
    settings=get_settings();asr=AsrService(settings)
    target=REPORT/'quality.json'
    results=json.loads(target.read_text('utf-8')) if target.exists() else []
    done={r['key']:r for r in results}
    progress=json.loads((REPORT/'generation.json').read_text('utf-8'))
    for clip in progress['clips']:
        if clip['key'] in done and done[clip['key']]['sha256']==clip['videoSha256'] and done[clip['key']].get('asrTailSeconds')==.8:continue
        audio=WEB/clip['audioUrl'].lstrip('/');video=WEB/clip['videoUrl'].lstrip('/')
        a,b=pcm(settings,audio),pcm(settings,video)
        lags=correlation_lags(len(b),len(a));corr=correlate(b,a,method='fft')
        mask=abs(lags)<=8000;lag=int(lags[mask][np.argmax(corr[mask])])
        score=float(np.max(corr[mask])/max(1e-9,np.linalg.norm(a)*np.linalg.norm(b)))
        recognizer=asr.load()
        stream=recognizer.create_stream()
        # Streaming Zipformer needs right context to emit the final syllables.
        stream.accept_waveform(16000,np.concatenate([b,np.zeros(12800,dtype=np.float32)]))
        stream.input_finished()
        while recognizer.is_ready(stream):recognizer.decode_stream(stream)
        transcript=recognizer.get_result(stream).strip()
        similarity=difflib.SequenceMatcher(None,normalize(clip['text']),normalize(transcript),autojunk=False).ratio()
        row={'key':clip['key'],'sha256':clip['videoSha256'],'text':clip['text'],'asrText':transcript,
            'asrSimilarity':round(similarity,4),'asrTailSeconds':.8,'audioLagMs':round(lag/16,2),'audioCorrelation':round(score,4),
            'sourceSeconds':len(a)/16000,'videoAudioSeconds':len(b)/16000,
            'technicalPass':abs(lag)<1600 and score>.9 and abs(len(a)-len(b))<16000*.7,
            'needsTranscriptReview':similarity<.85}
        results=[r for r in results if r['key']!=clip['key']]+[row]
        temp=target.with_suffix('.tmp');temp.write_text(json.dumps(results,ensure_ascii=False,indent=2),'utf-8');temp.replace(target)
        print(clip['key'],row['technicalPass'],'ASR',row['asrSimilarity'],'lag',row['audioLagMs'],flush=True)

if __name__=='__main__':main()