从视频文字检索到原帧引用

前置是媒体文件处理以及文档摄取的真实OCR。本节把这些机制接起来：真正编码视频、解码两帧、OCR裁切图、建立倒排索引、过滤后检索，最后把引用映射回原帧坐标和媒体时间。文本查询命中视觉证据，不靠手写“识别结果”冒充OCR。

检索文本与媒体证据是两层对象

OCR文字便于检索，但同一句话可以出现在不同租户、不同文件版本或不同时间帧。来源身份至少包含原文件版本、帧定位、解析器版本与区域；文字内容相同不能合并这些身份。返回给模型的证据应带source_id，模型选择ID后由应用从本轮授权快照取回坐标，不能任它捏造bbox或时间。

这里选择文字→OCR区域的检索支路，适用于订单屏幕、课件、字幕牌。没有文字的画面需要视觉描述或图文联合embedding：图像编码器与文本编码器要来自同一对齐模型，归一化后在同一向量空间比较；OCR文本向量与图像向量若来自不同模型，不能直接点积。要融合独立检索器，先各自排序再融合名次，保留来源；算法与真实向量模型实验见检索与重排。本节FTS命中不能证明“红色汽车”这类视觉语义可检索。

首次运行：真视频、真OCR、真索引

沿第一章环境增加rapidocr-onnxruntime==1.4.4。完整基线为Python3.12、Pillow11.3.0、av16.0.1、numpy2.2.6和RapidOCR1.4.4；OCR使用包内ONNX模型，无云端调用。该旧包的返回值为result, elapsed，不要直接套用新版rapidocr包的输出对象。

bash

python -m pip install 'Pillow==11.3.0' 'av==16.0.1' 'numpy==2.2.6' 'rapidocr-onnxruntime==1.4.4'

保存为media_retrieval.py，在专用实验目录执行python media_retrieval.py。将生成verified-box.png供人工核对；临时媒体和索引在结束后清理。首次ONNX加载较慢。OCR在独立子进程运行，避免部分macOS轮子同时加载PyAV与OpenCV各自FFmpeg库导致符号冲突；子进程只接收本例生成的图片路径，不执行模型返回的代码。

python

from pathlib import Path
from fractions import Fraction
import hashlib,json,re,sqlite3,tempfile,subprocess,sys
import av
import numpy as np
from PIL import Image,ImageDraw,ImageFont

# 视频真实编码，OCR读取解码后的裁切图；没有把图中文字直接放进索引。
def build(root):
    video=root/'orders.mkv'
    with av.open(str(video),'w') as container:
        stream=container.add_stream('mpeg4',rate=1)
        stream.width=640;stream.height=180;stream.pix_fmt='yuv420p'
        for i,text in enumerate(['ORDER 42 READY','ORDER 42 HOLD']):
            image=Image.new('RGB',(640,180),'white')
            ImageDraw.Draw(image).text((80,50),text,font=ImageFont.load_default(size=42),fill='black')
            frame=av.VideoFrame.from_ndarray(np.asarray(image),format='rgb24')
            frame.pts=i;frame.time_base=Fraction(1,1)
            for packet in stream.encode(frame):container.mux(packet)
        for packet in stream.encode():container.mux(packet)
    version=hashlib.sha256(video.read_bytes()).hexdigest()
    rows=[];crop_box=(40,20,600,160);scale=2
    with av.open(str(video)) as container:
        frames=list(container.decode(video=0))
    for i,frame in enumerate(frames):
        if frame.pts is None:raise ValueError('missing media time')
        t=float(frame.pts*frame.time_base)
        original=frame.to_image().convert('RGB')
        region=original.crop(crop_box).resize((1120,280))
        crop_path=root/f'crop-{i}.png';region.save(crop_path)
        # 独立OCR进程避免PyAV与OpenCV的FFmpeg库在同进程碰撞。
        child="from rapidocr_onnxruntime import RapidOCR; import json,sys; r,_=RapidOCR(intra_op_num_threads=1,inter_op_num_threads=1)(sys.argv[1]); print(json.dumps(r))"
        run=subprocess.run([sys.executable,'-c',child,str(crop_path)],capture_output=True,text=True,timeout=60,check=True)
        result=json.loads(run.stdout)
        if not result:raise ValueError('OCR empty')
        for j,(polygon,text,confidence) in enumerate(result):
            if confidence<.8:continue
            xs=[p[0] for p in polygon];ys=[p[1] for p in polygon]
            bbox=[crop_box[0]+min(xs)/scale,crop_box[1]+min(ys)/scale,
                  crop_box[0]+max(xs)/scale,crop_box[1]+max(ys)/scale]
            assert 0<=bbox[0]<bbox[2]<=640 and 0<=bbox[1]<bbox[3]<=180
            # PTS标注的是这一帧，不推断物体在整秒内一直存在。
            rows.append(dict(id=f'{version}:rapidocr1.4.4:crop40-20-600-160:s2:frame{i}:ocr{j}',version=version,tenant='A',
                text=text,bbox=bbox,time=t,frame=i,deleted=False,source=str(video),
                confidence=float(confidence),ocr='rapidocr-onnxruntime@1.4.4',
                transform={'crop':crop_box,'scale':scale,'original_size':[640,180]}))
    return rows,version

def normalize(text):return re.sub('[^a-z0-9]','',text.lower())
def index(rows):
    db=sqlite3.connect(':memory:')
    # FTS5真实倒排检索；不是图像语义向量，这条分支检索OCR可见文字。
    db.execute('CREATE VIRTUAL TABLE search USING fts5(id UNINDEXED,terms)')
    db.execute('CREATE TABLE metadata(id TEXT PRIMARY KEY, tenant TEXT, version TEXT, deleted INT, body TEXT)')
    for row in rows:
        db.execute('INSERT INTO search VALUES(?,?)',(row['id'],normalize(row['text'])))
        db.execute('INSERT INTO metadata VALUES(?,?,?,?,?)',
            (row['id'],row['tenant'],row['version'],row['deleted'],json.dumps(row)))
    db.commit();return db

def retrieve(db,query,tenant,version,k=3):
    term=normalize(query)
    if not term:return []
    # 身份来自服务端；先过滤，再LIMIT，不能先取混租TopK再掩盖。
    sql='''SELECT m.body FROM search JOIN metadata m ON m.id=search.id
           WHERE search MATCH ? AND m.tenant=? AND m.version=? AND m.deleted=0
           ORDER BY bm25(search),m.id LIMIT ?'''
    return [json.loads(r[0]) for r in db.execute(sql,('"'+term+'"',tenant,version,k))]

def citation(db,source_id,evidence,tenant,version):
    matches=[r for r in evidence if r['id']==source_id and r['tenant']==tenant
             and r['version']==version and not r['deleted']]
    if len(matches)!=1:raise ValueError('citation outside authorized snapshot')
    row=matches[0]
    live=db.execute('SELECT tenant,version,deleted FROM metadata WHERE id=?',(source_id,)).fetchone()
    if live!=(tenant,version,0):raise ValueError('authorization or version revoked')
    if hashlib.sha256(Path(row['source']).read_bytes()).hexdigest()!=version:
        raise ValueError('source version changed; retrieve again')
    return {key:row[key] for key in ['id','text','source','version','bbox','time','frame']}

def main():
    with tempfile.TemporaryDirectory() as directory:
        rows,version=build(Path(directory))
        # 同文字别租户、已删除/旧版本，证明不能用文本作为来源身份。
        base=rows[0]
        rows += [dict(base,id='other-tenant',tenant='B'),dict(base,id='old-version',version='old'),
                 dict(base,id='deleted-copy',deleted=True)]
        db=index(rows)
        hits=retrieve(db,'ORDER 42 READY','A',version)
        assert hits and len(hits)==1,(hits,rows)
        evidence=citation(db,hits[0]['id'],hits,'A',version)
        assert evidence['time']==0 and evidence['frame']==0
        assert evidence['bbox'][0]>=40 and evidence['bbox'][1]>=20
        assert retrieve(db,'UNKNOWN','A',version)==[]
        for wrong in ['other-tenant','old-version','deleted-copy','invented']:
            try:citation(db,wrong,hits,'A',version)
            except ValueError:pass
            else:raise AssertionError('invalid citation')
        # 把裁切缩放结果画回解码原图供人工核验，不覆盖原媒体。
        with av.open(evidence['source']) as container:
            image=next(container.decode(video=0)).to_image()
        ImageDraw.Draw(image).rectangle(evidence['bbox'],outline='red',width=3)
        image.save('verified-box.png')
        print(json.dumps(evidence,ensure_ascii=False))
        db.close()
    print('real video → OCR → FTS → authorized source, time and original coordinates')
if __name__=='__main__':main()

正常结果包含OCR实际识别文字ORDER42READY、第0帧、时间0秒，以及映射回640×180原帧的bbox；不同平台识别框可有小差异，不断言每个像素完全相同。OCR输入是裁切框(40,20,600,160)放大2倍后的图，所以原坐标为crop_origin + recognized_coordinate / 2。直接把识别坐标画在原图上会出现平移和缩放错误。

第1帧的另一条状态不能因相同订单号而被合并。当前索引只检索规范化后的整串英文数字，故意用完整标牌查询固定机制；这不是通用分词方案。自然语言、中文或错别字查询需要分词/字符n-gram/向量召回和单独的召回评估。SQLite构建了真实FTS倒排索引，权限、版本、删除状态在SQL的LIMIT前过滤，未授权行不会占用已授权结果的TopK名额。

引用分两次核验：先确认ID属于本轮证据，再检查当前元数据未撤权/删掉、文件内容哈希仍匹配。这里读取本地文件即校验一次；生产对象存储应使用不可变版本与授权下载地址，避免可变路径检查后又被替换。bbox和PTS来自可信摄取记录，模型只选ID；它仍可能选错依据，所以来源有效不等于语义支持。看图检查标牌，才能确认文字与位置一致。

接进生成与跨模态上下文

传入生成器时，每个证据项序列化id,text,time,bbox,version，需要视觉判断时同时传相应原帧或裁切图。若只传OCR文字，就只能回答文字支持的问题，不能断言颜色、人物动作或整段视频状态。响应结构用answer和sources:[id]，沿结构化输出校验结构，再逐个调用本页citation；没有命中就返回依据不足，不生成猜测ID。

音频片段同理：保存原WAV版本、采样率、区间起止样本，重采样后用媒体时间映射回原样本，不能拿到达时刻当证据时间。转录final只提供片段文字，不自动提供词级时间戳。图片中的注入文字也只是待核对证据，不是应用指令；引用、原图访问和删除传播共享租户边界。

练习

改查ORDER 42 HOLD，应定位第1帧、1秒；运行真实OCR，不手工填入识别结果。
在取回证据后修改原文件字节，旧引用应拒绝；在数据库打删除墓碑后，即使保留旧快照也不能再次引用。
解释为什么OCR框的位置正确仍不能证明“订单现在已准备好”，以及为什么只改查询成“准备好了没”可能检索为空。

本章答案

从视频文字检索到原帧引用 ​

检索文本与媒体证据是两层对象 ​

首次运行：真视频、真OCR、真索引 ​

接进生成与跨模态上下文 ​

练习 ​

从视频文字检索到原帧引用

检索文本与媒体证据是两层对象

首次运行：真视频、真OCR、真索引

接进生成与跨模态上下文

练习