PaddleOCR的API开放了！撸⼀套视频字幕提取⼯具，体验完美！

def ocr_image(file_path: str, frame: int) -> str:
    try:
        with open(file_path, "rb") as f:
            file_data = base64.b64encode(f.read()).decode("ascii")
        headers = {"Authorization": f"token {TOKEN}", "Content-Type": "application/json"}
        payload = {
            "file": file_data,
            "fileType": 1,
            "useDocOrientationClassify": False,
            "useDocUnwarping": False,
            "useTextlineOrientation": False,
        }
        resp = requests.post(API_URL, json=payload, headers=headers, timeout=20)
        if resp.status_code != 200:
            return ""
        result = resp.json().get("result", {})
        if not result.get("ocrResults"):
            return ""
        texts, height = [], result.get("dataInfo", {}).get("height", 720)
        boxes = []
        y_threshold = height * 6 / 7
        if not os.path.exists("output"):
            os.makedirs("output")
        for res in result["ocrResults"]:
            pruned = res.get("prunedResult", {})
            ocr_res_image_url = res.get("ocrImage", "")
            if ocr_res_image_url:
                response = requests.get(ocr_res_image_url)
                if response.status_code == 200:
                    with open(f"output/frame_ocr_{frame}.jpg", "wb") as f:
                        f.write(response.content)
                    with open(f"output/frame_ocr_{frame}.json", "w") as f:
                        json.dump(pruned, f)
                else:
                    print("图片下载失败:", response.status_code)
            else:
                print("没有ocrImage字段")
            for txt, box in zip(pruned.get("rec_texts", []), pruned.get("rec_boxes", [])):
                if box[1] >= y_threshold or box[3] >= y_threshold:
                    texts.append(txt)
                    boxes.append(box)
        src_img = cv2.imread(file_path)
        for text, box in zip(texts, boxes):
            color = get_random_color()
            draw_transparent_rectangle(src_img, (box[0], box[1]), (box[2], box[3]), color, alpha=0.5)
            src_img = draw_text_with_pil(src_img, text, (box[0], box[1]-20),color=(0,0,255))

        result_text = " ".join(texts).strip()

        cv2.imwrite(f"output/frame_result_{frame}.jpg", src_img, [int(cv2.IMWRITE_JPEG_QUALITY), 100])
        return result_text
    except Exception as e:
        print(f"An error occurred: {e}")
        return ""
        
 def extract_subtitles(video_path: str, interval: float, progress_cb=None, ui_queue=None, cancel_event=None):
    cap = cv2.VideoCapture(video_path)
    fps = cap.get(cv2.CAP_PROP_FPS) or 25
    total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
    frame_gap = int(fps * interval)
    temp_dir = tempfile.mkdtemp()
    tasks, frame_num = [], 0
    while True:
        if cancel_event and cancel_event.is_set():
            break
        ret, frame = cap.read()
        if not ret:
            break
        if frame_num % frame_gap == 0:
            img_path = os.path.join(temp_dir, f"f{frame_num}.jpg")
            cv2.imwrite(img_path, frame)
            tasks.append((frame_num, img_path))
        frame_num += 1
    cap.release()
    if cancel_event and cancel_event.is_set():
        return []
    sorted_map = SortedDict()
    lock = threading.Lock()

    def _ocr(task):
        fnum, fpath = task
        time.sleep(0.1)
        txt = ocr_image(fpath, fnum)
        with lock:
            if not txt:
                return
            start = fnum / fps
            end = (fnum + frame_gap) / fps
            sorted_map[fnum] = (start, end, txt)
            merge_neighbor(sorted_map, fnum, fps)
            if progress_cb:
                progress_cb(len(sorted_map) / len(tasks) if tasks else 0)
            if ui_queue is not None:
                ui_queue.put(list(sorted_map.values()))

    with ThreadPoolExecutor(max_workers=1) as exe:
        futures = [exe.submit(_ocr, t) for t in tasks]
        for f in as_completed(futures):
            if cancel_event and cancel_event.is_set():
                for fut in futures:
                    fut.cancel()
                break
            try:
                f.result()
            except Exception:
                pass
    # 清理
    for f in os.listdir(temp_dir):
        try:
            os.remove(os.path.join(temp_dir, f))
        except Exception:
            pass
    try:
        os.rmdir(temp_dir)
    except Exception:
        pass
    return [{"start": v[0], "end": v[1], "text": v[2]} for v in sorted_map.values()]