2024 lines
90 KiB
Python
2024 lines
90 KiB
Python
# 修改后的主处理流程
|
||
import json
|
||
import os
|
||
import base64
|
||
import argparse
|
||
from collections import defaultdict, Counter
|
||
from datetime import timedelta
|
||
from tool.Ocr.tencentcloud_ocr import tencentcloud_ocr_api, tencent_ocr_video_all_frames, img2base64
|
||
from difflib import SequenceMatcher
|
||
import re
|
||
import random
|
||
import cv2
|
||
import jieba
|
||
from collections import deque
|
||
from concurrent.futures import ThreadPoolExecutor, as_completed
|
||
from config.config_utils import config
|
||
import numpy as np
|
||
from sklearn.cluster import DBSCAN
|
||
import math
|
||
|
||
|
||
def _jpeg_size(path):
|
||
"""只读 JPEG 文件头拿到 (宽,高), 不解码全图、不依赖 PIL。失败返回 None。"""
|
||
try:
|
||
with open(path, 'rb') as f:
|
||
if f.read(2) != b'\xff\xd8': # 不是 JPEG
|
||
return None
|
||
while True:
|
||
b = f.read(1)
|
||
while b and b != b'\xff': # 找到下一个标记起始 0xFF
|
||
b = f.read(1)
|
||
marker = f.read(1)
|
||
while marker == b'\xff': # 跳过填充 0xFF
|
||
marker = f.read(1)
|
||
if not marker:
|
||
return None
|
||
m = marker[0]
|
||
# SOF 标记(含尺寸): C0-CF, 排除 C4(DHT) C8(JPG) CC(DAC)
|
||
if 0xC0 <= m <= 0xCF and m not in (0xC4, 0xC8, 0xCC):
|
||
f.read(3) # 段长(2) + 精度(1)
|
||
h = int.from_bytes(f.read(2), 'big')
|
||
w = int.from_bytes(f.read(2), 'big')
|
||
return w, h
|
||
seg = int.from_bytes(f.read(2), 'big') # 其它段: 跳过
|
||
if seg < 2:
|
||
return None
|
||
f.seek(seg - 2, 1)
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
def _img2base64_capped(img_path, max_side):
|
||
"""给OCR的图片加【像素尺寸上限】: 最长边超过 max_side 就等比缩小再送。
|
||
防止区域检测过高导致的超大合成图(几千×上万像素)让单张OCR耗时>30s超时。
|
||
纯Python读图头判尺寸(无需PIL), 只有超限的才用 cv2 解码+缩放; 正常图原样送。"""
|
||
try:
|
||
if max_side and max_side > 0:
|
||
sz = _jpeg_size(img_path)
|
||
if sz and max(sz) > max_side:
|
||
im = cv2.imread(img_path)
|
||
if im is not None:
|
||
h, w = im.shape[:2]
|
||
sc = max_side / float(max(h, w))
|
||
im = cv2.resize(im, (max(1, int(w * sc)), max(1, int(h * sc))),
|
||
interpolation=cv2.INTER_AREA)
|
||
ok, buf = cv2.imencode('.jpg', im, [int(cv2.IMWRITE_JPEG_QUALITY), 90])
|
||
if ok:
|
||
return base64.b64encode(buf.tobytes()).decode()
|
||
except Exception:
|
||
pass
|
||
return img2base64(img_path=img_path)
|
||
|
||
|
||
def _safe_ocr_image(img_path):
|
||
"""对单张图片做 OCR，捕获所有异常，返回 (原始字符串结果, 错误信息或None)。
|
||
供并发 OCR 的 worker 调用——不直接写日志（worker运行在子线程，logger为线程本地，
|
||
日志会丢失），错误信息回传给主线程统一记录。任何一张图失败都不影响其他图。"""
|
||
empty = json.dumps({"TextDetections": [], "Language": "zh", "RequestId": "ocr_error"},
|
||
ensure_ascii=False)
|
||
try:
|
||
try:
|
||
_max_side = int(config.get_subtitle_config().get('ocr_max_image_side', 4000) or 0)
|
||
except (ValueError, TypeError, AttributeError):
|
||
_max_side = 4000
|
||
img = _img2base64_capped(img_path, _max_side)
|
||
result = tencentcloud_ocr_api(image_base64=img)
|
||
if not result:
|
||
return empty, f"OCR返回空结果: {img_path}"
|
||
return result, None
|
||
except Exception as e:
|
||
return empty, f"OCR调用异常 {img_path}: {e}"
|
||
|
||
|
||
import time
|
||
|
||
_EMPTY_OCR = json.dumps({"TextDetections": [], "Language": "zh", "RequestId": "ocr_error"},
|
||
ensure_ascii=False)
|
||
|
||
|
||
class OcrIncompleteError(Exception):
|
||
"""多轮重试后仍有图 OCR 失败。为"不丢帧": 不产出带洞的SRT, 让整段视频判失败去重跑。"""
|
||
pass
|
||
|
||
|
||
def _is_failed_ocr(raw):
|
||
"""判断某OCR结果是不是"失败"(全端点超时/异常)。
|
||
注意: TextDetections 为空但 RequestId 不是 ocr_error → 是合法的"图里确实没字", 不算失败, 不重试。"""
|
||
if not raw:
|
||
return True
|
||
try:
|
||
return json.loads(raw).get("RequestId") == "ocr_error"
|
||
except (json.JSONDecodeError, TypeError):
|
||
return True
|
||
|
||
|
||
def _ocr_pass(img_paths, workers):
|
||
"""对一组图并发 OCR 一遍, 返回 ({path: raw}, [错误信息])。单张失败返回空结构, 不抛异常。"""
|
||
results, errors = {}, []
|
||
with ThreadPoolExecutor(max_workers=workers) as executor:
|
||
future_to_path = {executor.submit(_safe_ocr_image, p): p for p in img_paths}
|
||
for future in as_completed(future_to_path):
|
||
path = future_to_path[future]
|
||
try:
|
||
ocr_raw, err = future.result()
|
||
results[path] = ocr_raw
|
||
if err:
|
||
errors.append(err)
|
||
except Exception as e: # _safe_ocr_image 已兜底, 这里双保险
|
||
errors.append(f"OCR任务异常 {path}: {e}")
|
||
results[path] = _EMPTY_OCR
|
||
return results, errors
|
||
|
||
|
||
def _batch_ocr_images(img_paths, logger=None):
|
||
"""并发对一组图片做 OCR，返回 {img_path: ocr_raw_string}。
|
||
【不丢帧】: 失败(全端点超时)的图【多轮重试】直到全部成功; 失败多是瞬时超时, 多轮基本都能救回。
|
||
若 ocr_no_drop=true(默认), 多轮后仍有失败 → 抛 OcrIncompleteError → 整段视频判失败去重跑,
|
||
绝不产出"带洞的SRT"标成功。"""
|
||
results = {}
|
||
if not img_paths:
|
||
return results
|
||
sub = config.get_subtitle_config()
|
||
try:
|
||
workers = int(sub.get('ocr_workers', 10))
|
||
except (ValueError, TypeError):
|
||
workers = 10
|
||
workers = max(1, min(workers, len(img_paths)))
|
||
try:
|
||
max_passes = max(1, int(sub.get('ocr_max_passes', 5) or 5))
|
||
except (ValueError, TypeError):
|
||
max_passes = 5
|
||
try:
|
||
sleep_s = max(0.0, float(sub.get('ocr_retry_sleep', 2) or 2))
|
||
except (ValueError, TypeError):
|
||
sleep_s = 2.0
|
||
no_drop = bool(sub.get('ocr_no_drop', True))
|
||
total = len(img_paths)
|
||
|
||
results, errors = _ocr_pass(img_paths, workers)
|
||
|
||
# —— 多轮重试: 只重试还失败的, 直到全部成功或用满轮数 ——
|
||
# 等待时间随轮数线性递增(sleep_s*轮数), 而不是固定间隔: 失败多半是端点拥堵/冷却,
|
||
# 固定短间隔等于一直硬撞同一批还没缓过来的端点; 递增给拥堵端点更多恢复时间。
|
||
pass_no = 1
|
||
while pass_no < max_passes:
|
||
failed = [p for p in img_paths if _is_failed_ocr(results.get(p))]
|
||
if not failed:
|
||
break
|
||
wait_s = sleep_s * pass_no
|
||
if logger:
|
||
logger.log(f"OCR第{pass_no}轮后仍{len(failed)}张失败, {wait_s}s后重试...",
|
||
level="warning", program="getsrt")
|
||
if wait_s:
|
||
time.sleep(wait_s)
|
||
retry_results, _ = _ocr_pass(failed, max(1, min(workers, len(failed))))
|
||
for p, raw in retry_results.items():
|
||
if not _is_failed_ocr(raw):
|
||
results[p] = raw
|
||
pass_no += 1
|
||
|
||
failed = [p for p in img_paths if _is_failed_ocr(results.get(p))]
|
||
if logger:
|
||
for msg in errors[:50]:
|
||
logger.log(msg, level="warning", program="getsrt")
|
||
logger.log(f"OCR完成: 共{total}张, {pass_no}轮后最终失败{len(failed)}张", program="getsrt")
|
||
|
||
# —— 不丢帧: 仍有失败就别产出带洞SRT, 抛异常让整段视频判失败去重跑 ——
|
||
if failed and no_drop:
|
||
raise OcrIncompleteError(
|
||
f"OCR多轮({max_passes})后仍有 {len(failed)}/{total} 张失败, 为不丢帧放弃本视频(待重跑)")
|
||
return results
|
||
def calculate_similarity(text1, text2):
|
||
"""计算两个文本的相似度"""
|
||
if not text1 or not text2:
|
||
return 0.0
|
||
return SequenceMatcher(None, text1, text2).ratio()
|
||
def process_composite_ocr_to_jsonl(metadata_path, composite_dir, output_jsonl_path,logger):
|
||
"""
|
||
处理拼接图片的OCR结果，生成与tencent_ocr_video_all_frames相同格式的jsonl文件
|
||
|
||
参数:
|
||
metadata_path: metadata.json文件路径
|
||
composite_dir: 拼接图片所在目录
|
||
output_jsonl_path: 输出jsonl文件路径
|
||
"""
|
||
# 读取metadata.json
|
||
with open(metadata_path, 'r', encoding='utf-8') as f:
|
||
metadata = json.load(f)
|
||
|
||
# 获取所有原始帧的路径（包括kept_first和kept_last）
|
||
all_frames = []
|
||
original_dir = os.path.dirname(metadata_path) # 假设原始帧与metadata在同一目录
|
||
|
||
# 添加kept_first帧
|
||
for frame_name in metadata['kept_first']:
|
||
all_frames.append({
|
||
'source_image': frame_name,
|
||
'is_blank': False,
|
||
'composite_source': None
|
||
})
|
||
|
||
# 添加composite中的帧
|
||
for composite in metadata['composites']:
|
||
composite_id = composite['composite_id']
|
||
composite_file = composite['file_name']
|
||
composite_path = os.path.join(composite_dir, composite_file)
|
||
|
||
for tile in composite['tiles']:
|
||
if tile['is_blank']:
|
||
continue
|
||
else:
|
||
all_frames.append({
|
||
'source_image': tile['source_image'],
|
||
'is_blank': False,
|
||
'composite_source': {
|
||
'composite_path': composite_path,
|
||
'x': tile['x'],
|
||
'y': tile['y'],
|
||
'width': tile['width'],
|
||
'height': tile['height']
|
||
}
|
||
})
|
||
|
||
# 添加kept_last帧
|
||
for frame_name in metadata['kept_last']:
|
||
all_frames.append({
|
||
'source_image': frame_name,
|
||
'is_blank': False,
|
||
'composite_source': None
|
||
})
|
||
|
||
# 对每个拼接图只进行一次OCR（并发）
|
||
composite_paths = [
|
||
os.path.join(composite_dir, c['file_name']) for c in metadata['composites']
|
||
]
|
||
logger.log(f"开始并发OCR {len(composite_paths)} 张合成图", program="getsrt")
|
||
logger.set_stat("composite_ocr_count", len(composite_paths))
|
||
composite_ocr_results = _batch_ocr_images(composite_paths, logger)
|
||
|
||
# 预先并发OCR所有需要单独处理的帧（kept_first / kept_last，去重且只取存在的文件）
|
||
single_frame_paths = []
|
||
seen_single = set()
|
||
for frame_info in all_frames:
|
||
if frame_info['is_blank'] or frame_info['composite_source']:
|
||
continue
|
||
fp = os.path.join(original_dir, frame_info['source_image'])
|
||
if fp in seen_single:
|
||
continue
|
||
seen_single.add(fp)
|
||
if os.path.exists(fp):
|
||
single_frame_paths.append(fp)
|
||
if single_frame_paths:
|
||
logger.log(f"开始并发OCR {len(single_frame_paths)} 张单独帧", program="getsrt")
|
||
logger.set_stat("single_frame_ocr_count", len(single_frame_paths))
|
||
logger.set_stat("total_ocr_count", len(composite_paths) + len(single_frame_paths))
|
||
single_ocr_results = _batch_ocr_images(single_frame_paths, logger)
|
||
blank = 0
|
||
# 生成jsonl文件
|
||
with open(output_jsonl_path, 'w', encoding='utf-8') as f:
|
||
for frame_info in all_frames:
|
||
frame_name = frame_info['source_image']
|
||
frame_path = os.path.join(original_dir, frame_name)
|
||
|
||
# 写入图片路径
|
||
f.write(f"pic_path: {frame_path}\n")
|
||
# 生成OCR结果
|
||
if frame_info['is_blank']:
|
||
if frame_info['composite_source']:
|
||
blank=blank+1
|
||
continue
|
||
elif frame_info['composite_source']:
|
||
# 从拼接图片的OCR结果中提取属于该区域的文字框
|
||
composite_info = frame_info['composite_source']
|
||
composite_ocr = composite_ocr_results[composite_info['composite_path']]
|
||
try:
|
||
composite_ocr = json.loads(composite_ocr)
|
||
except (json.JSONDecodeError, TypeError) as e:
|
||
logger.log(f"警告: 无法解析拼接图OCR结果: {e}",level="warning",program="getsrt")
|
||
composite_ocr = {"TextDetections": [], "Language": "zh", "RequestId": "parse_error"}
|
||
# 提取属于该区域的文字框
|
||
region_ocr = extract_ocr_for_region(
|
||
composite_ocr,
|
||
composite_info['x'],
|
||
composite_info['y'],
|
||
composite_info['width'],
|
||
composite_info['height']
|
||
)
|
||
ocr_result = region_ocr
|
||
else:
|
||
# 单独处理的帧（kept_first和kept_last），从预先并发OCR的结果中读取
|
||
if frame_path not in single_ocr_results:
|
||
logger.log(f"跳过已清理的帧: {frame_path}", level="warning", program="getsrt")
|
||
continue
|
||
ocr_result = single_ocr_results[frame_path]
|
||
try:
|
||
ocr_result = json.loads(ocr_result)
|
||
except (json.JSONDecodeError, TypeError) as e:
|
||
logger.log(f"警告: 无法解析单独帧OCR结果: {e}",level="warning",program="getsrt")
|
||
ocr_result = {"TextDetections": [], "Language": "zh", "RequestId": "parse_error"}
|
||
|
||
# 写入OCR结果
|
||
f.write(json.dumps(ocr_result, ensure_ascii=False) + "\n")
|
||
if blank>(len(all_frames)-18)/2:
|
||
return False
|
||
|
||
logger.log(f"已生成jsonl文件: {output_jsonl_path}",program="getsrt")
|
||
return True
|
||
|
||
def extract_ocr_for_region(composite_ocr, x, y, width, height):
|
||
"""
|
||
从拼接图的OCR结果中提取属于指定区域的文字框
|
||
|
||
参数:
|
||
composite_ocr: 拼接图的OCR结果
|
||
x, y: 区域左上角坐标
|
||
width, height: 区域宽高
|
||
|
||
返回:
|
||
该区域的OCR结果
|
||
"""
|
||
region_ocr = {
|
||
"TextDetections": [],
|
||
}
|
||
|
||
# 计算区域的右下角坐标
|
||
region_right = x + width
|
||
region_bottom = y + height
|
||
|
||
for detection in composite_ocr["TextDetections"]:
|
||
# 获取文字框的边界
|
||
if "Polygon" not in detection:
|
||
continue
|
||
|
||
polygon = detection["Polygon"]
|
||
if not polygon:
|
||
continue
|
||
|
||
# 计算文字框的中心点
|
||
xs = []
|
||
ys = []
|
||
for point in polygon:
|
||
if "X" in point:
|
||
xs.append(point["X"])
|
||
if "Y" in point:
|
||
ys.append(point["Y"])
|
||
|
||
if not xs or not ys:
|
||
continue
|
||
|
||
center_x = sum(xs) / len(xs)
|
||
center_y = sum(ys) / len(ys)
|
||
|
||
# 检查中心点是否在区域内
|
||
if (x <= center_x <= region_right and y <= center_y <= region_bottom):
|
||
# 调整坐标，使其相对于区域
|
||
adjusted_detection = detection.copy()
|
||
adjusted_polygon = []
|
||
for point in polygon:
|
||
new_point = {}
|
||
if "X" in point:
|
||
new_point["X"] = point["X"] - x
|
||
if "Y" in point:
|
||
new_point["Y"] = point["Y"] - y
|
||
adjusted_polygon.append(new_point)
|
||
|
||
adjusted_detection["Polygon"] = adjusted_polygon
|
||
region_ocr["TextDetections"].append(adjusted_detection)
|
||
|
||
return region_ocr
|
||
|
||
def analyze_subtitle_blocks(all_pic_ocr_pair, metadata_path,logger):
|
||
"""
|
||
分析所有图片，找到最可能是字幕的区块（10×100网格）
|
||
分别处理kept_first/kept_last和composite部分
|
||
"""
|
||
# 读取metadata.json获取帧的分类信息
|
||
with open(metadata_path, 'r', encoding='utf-8') as f:
|
||
metadata = json.load(f)
|
||
|
||
# 分离不同类型的帧
|
||
kept_first_names = set(metadata['kept_first'])
|
||
kept_last_names = set(metadata['kept_last'])
|
||
|
||
kept_first_ocr = []
|
||
composite_ocr = []
|
||
kept_last_ocr = []
|
||
for item in all_pic_ocr_pair:
|
||
pic_path, ocr_json = item
|
||
frame_name = os.path.basename(pic_path)
|
||
|
||
if frame_name in kept_first_names:
|
||
kept_first_ocr.append(item)
|
||
elif frame_name in kept_last_names:
|
||
kept_last_ocr.append(item)
|
||
else:
|
||
composite_ocr.append(item)
|
||
|
||
|
||
# 分别分析kept_first/composite/kept_last的最佳字幕位置
|
||
kept_first_grid = analyze_subtitle_blocks_for_group(kept_first_ocr, "kept_first",logger)
|
||
composite_grid = analyze_subtitle_blocks_for_group(composite_ocr, "composite",logger)
|
||
kept_last_grid = analyze_subtitle_blocks_for_group(kept_last_ocr, "kept_last",logger)
|
||
|
||
# 返回三个组的分析结果
|
||
return {
|
||
"kept_first": kept_first_grid,
|
||
"composite": composite_grid,
|
||
"kept_last": kept_last_grid
|
||
}
|
||
def calculate_grouped_similarity_simple(texts,grouped = True):
|
||
"""
|
||
简化的分组相似度计算方法
|
||
"""
|
||
if len(texts) < 2:
|
||
return 0
|
||
|
||
# 计算所有文本对之间的相似度
|
||
similarities = []
|
||
for i in range(len(texts)):
|
||
for j in range(i+1, len(texts)):
|
||
if texts[i]==None or texts[j]==None:
|
||
continue
|
||
similarity = SequenceMatcher(None, texts[i], texts[j]).ratio()
|
||
similarities.append(similarity)
|
||
|
||
# 如果有很多高相似度的对，即使有一些低相似度的对，也认为是高相似度
|
||
high_similarity_count = sum(1 for s in similarities if s > 0.7)
|
||
total_pairs = len(texts)*(len(texts)-1)/2
|
||
if not grouped:
|
||
if high_similarity_count / total_pairs > 0.7:
|
||
# 返回高相似度对的平均相似度
|
||
high_similarities = [s for s in similarities if s > 0.7]
|
||
return sum(high_similarities) / len(high_similarities)
|
||
else:
|
||
# 返回所有对的平均相似度
|
||
return sum(similarities) / total_pairs
|
||
# 如果超过一小部分的文本对有高相似度
|
||
if high_similarity_count / total_pairs > 0.2:
|
||
# 返回高相似度对的平均相似度
|
||
high_similarities = [s for s in similarities if s > 0.7]
|
||
return sum(high_similarities) / len(high_similarities)
|
||
else:
|
||
# 返回所有对的平均相似度
|
||
return sum(similarities) / total_pairs
|
||
|
||
def getaverageh(ocr_group):
|
||
h=0
|
||
tot=0
|
||
for item in ocr_group:
|
||
_, ocr_json = item
|
||
for text_detection in ocr_json['TextDetections']:
|
||
if text_detection['Confidence']<=60:
|
||
continue
|
||
if 'is_watermark' in text_detection:
|
||
if text_detection['is_watermark']:
|
||
continue
|
||
polygon = [[int(coord['X']), int(coord['Y'])] for coord in text_detection['Polygon']]
|
||
min_x = min(p[0] for p in polygon)
|
||
max_x = max(p[0] for p in polygon)
|
||
max_y = max(p[1] for p in polygon)
|
||
min_y = min(p[1] for p in polygon)
|
||
h =h+max_y-min_y
|
||
tot+=1
|
||
if tot==0:
|
||
return 0
|
||
if (tot>3*len(ocr_group)):
|
||
return 0
|
||
h = int(h/tot)
|
||
return h
|
||
def analyze_h(all_pic_ocr_pair, metadata_path):
|
||
"""
|
||
分析所有图片，找到最可能是字幕的区块（10×100网格）
|
||
分别处理kept_first/kept_last和composite部分
|
||
"""
|
||
# 读取metadata.json获取帧的分类信息
|
||
with open(metadata_path, 'r', encoding='utf-8') as f:
|
||
metadata = json.load(f)
|
||
|
||
# 分离不同类型的帧
|
||
kept_first_names = set(metadata['kept_first'])
|
||
kept_last_names = set(metadata['kept_last'])
|
||
|
||
kept_first_ocr = []
|
||
composite_ocr = []
|
||
kept_last_ocr = []
|
||
|
||
for item in all_pic_ocr_pair:
|
||
pic_path, ocr_json = item
|
||
frame_name = os.path.basename(pic_path)
|
||
|
||
if frame_name in kept_first_names:
|
||
kept_first_ocr.append(item)
|
||
elif frame_name in kept_last_names:
|
||
kept_last_ocr.append(item)
|
||
else:
|
||
composite_ocr.append(item)
|
||
# 分别分析kept_first/composite/kept_last的最佳字幕位置
|
||
kept_first_h = getaverageh(kept_first_ocr)
|
||
composite_h = getaverageh(composite_ocr)
|
||
kept_last_h = getaverageh(kept_last_ocr)
|
||
|
||
# 返回三个组的分析结果
|
||
return {
|
||
"kept_first": kept_first_h,
|
||
"composite": composite_h,
|
||
"kept_last": kept_last_h
|
||
}
|
||
def find_best_cluster_center(candidate_points):
|
||
"""
|
||
找到计数最高的网格点集群的中心
|
||
"""
|
||
# 使用DBSCAN聚类
|
||
points = np.array(candidate_points)
|
||
dbscan = DBSCAN(eps=1.5, min_samples=1) # 宽松参数确保所有点都能聚类
|
||
labels = dbscan.fit_predict(points)
|
||
|
||
# 找到最大的簇
|
||
clusters = {}
|
||
for i, label in enumerate(labels):
|
||
if label not in clusters:
|
||
clusters[label] = []
|
||
clusters[label].append(tuple(points[i]))
|
||
|
||
# 找出最大的簇
|
||
max_cluster = max(clusters.values(), key=len)
|
||
|
||
# 计算簇的中心（质心）
|
||
cluster_array = np.array(max_cluster)
|
||
center_x = np.mean(cluster_array[:, 0])
|
||
center_y = np.mean(cluster_array[:, 1])
|
||
|
||
return (round(center_x), round(center_y))
|
||
|
||
def analyze_subtitle_blocks_for_group(ocr_group, group_name,logger):
|
||
"""
|
||
分析特定组的图片，找到最可能是字幕的区块
|
||
改进版：位置检测使用所有帧，文字相似度检测使用抽样帧
|
||
"""
|
||
if not ocr_group:
|
||
print(f"{group_name}组没有OCR数据")
|
||
return None
|
||
|
||
# 网格划分参数
|
||
grid_rows = 50 # 纵向分30块
|
||
grid_cols = 200 # 横向分300块
|
||
global_left = float('inf')
|
||
global_right = 0
|
||
global_bottom = 0
|
||
global_top = float('inf')
|
||
if group_name=="composite":
|
||
grid_rows = 50
|
||
grid_cols = 100
|
||
# 计算全局边界（使用所有帧）
|
||
|
||
for item in ocr_group:
|
||
_, ocr_json = item
|
||
for text_detection in ocr_json['TextDetections']:
|
||
if text_detection['Confidence']<=60:
|
||
continue
|
||
polygon = [[int(coord['X']), int(coord['Y'])] for coord in text_detection['Polygon']]
|
||
min_x = min(p[0] for p in polygon)
|
||
max_x = max(p[0] for p in polygon)
|
||
max_y = max(p[1] for p in polygon)
|
||
min_y = min(p[1] for p in polygon)
|
||
if min_x < global_left:
|
||
global_left = min_x
|
||
if max_x > global_right:
|
||
global_right = max_x
|
||
if max_y > global_bottom:
|
||
global_bottom = max_y
|
||
if min_y < global_top:
|
||
global_top = min_y
|
||
# 设置固定边界
|
||
grid_left_bound = global_left
|
||
grid_right_bound = global_right
|
||
grid_bottom_bound = global_bottom
|
||
grid_top_bound = global_top
|
||
|
||
print(f"{group_name}组边界: bottom={global_bottom}, right={global_right}, left={global_left}, top={global_top}")
|
||
|
||
# 计算网格实际宽高
|
||
grid_width = (grid_right_bound - grid_left_bound) / grid_cols
|
||
grid_height = (grid_bottom_bound - grid_top_bound) / grid_rows
|
||
|
||
|
||
# 统计网格被包含次数（使用所有帧）
|
||
grid_counts = defaultdict(int)
|
||
left = 0
|
||
right = 0
|
||
middle = 0
|
||
for item in ocr_group:
|
||
_, ocr_json = item
|
||
for text_detection in ocr_json['TextDetections']:
|
||
polygon = [[int(coord['X']), int(coord['Y'])] for coord in text_detection['Polygon']]
|
||
min_x = min(p[0] for p in polygon)
|
||
max_x = max(p[0] for p in polygon)
|
||
min_y = min(p[1] for p in polygon)
|
||
max_y = max(p[1] for p in polygon)
|
||
if len(text_detection['DetectedText'])<5:
|
||
continue
|
||
if text_detection['Confidence']<=60:
|
||
continue
|
||
# 检查每个网格
|
||
for row in range(grid_rows):
|
||
for col in range(grid_cols):
|
||
cell_left = grid_left_bound + col * grid_width
|
||
cell_right = cell_left + grid_width
|
||
cell_top = grid_top_bound + row * grid_height
|
||
cell_bottom = cell_top + grid_height
|
||
|
||
# 检查网格中心是否在文字框内
|
||
center_x = (cell_left + cell_right) / 2
|
||
center_y = (cell_top + cell_bottom) / 2
|
||
|
||
if (min_x <= center_x <= max_x) and (min_y <= center_y <= max_y):
|
||
grid_counts[(row, col)] += 1
|
||
|
||
# 从所有帧中均匀抽样10帧用于文字相似度检测
|
||
sample_size = min(10, len(ocr_group))
|
||
if len(ocr_group) <= sample_size:
|
||
sample_frames = ocr_group
|
||
else:
|
||
# 计算抽样间隔
|
||
step = len(ocr_group) // sample_size
|
||
# 从不同位置开始抽样，确保覆盖整个视频
|
||
start_pos = min(10,len(ocr_group)-sample_size)+ random.randint(0, step - 1)
|
||
sample_frames = [ocr_group[i] for i in range(start_pos, len(ocr_group), step)][:sample_size]
|
||
|
||
print(f"从{group_name}组中抽样{sample_size}帧进行文字相似度分析")
|
||
|
||
# 统计抽样帧中每个网格的文字内容
|
||
grid_texts = defaultdict(list) # 存储每个网格的文字内容
|
||
|
||
for item in sample_frames:
|
||
_, ocr_json = item
|
||
for text_detection in ocr_json['TextDetections']:
|
||
polygon = [[int(coord['X']), int(coord['Y'])] for coord in text_detection['Polygon']]
|
||
min_x = min(p[0] for p in polygon)
|
||
max_x = max(p[0] for p in polygon)
|
||
min_y = min(p[1] for p in polygon)
|
||
max_y = max(p[1] for p in polygon)
|
||
|
||
# 检查每个网格
|
||
for row in range(grid_rows):
|
||
for col in range(grid_cols):
|
||
cell_left = grid_left_bound + col * grid_width
|
||
cell_right = cell_left + grid_width
|
||
cell_top = grid_top_bound + row * grid_height
|
||
cell_bottom = cell_top + grid_height
|
||
|
||
# 检查网格中心是否在文字框内
|
||
center_x = (cell_left + cell_right) / 2
|
||
center_y = (cell_top + cell_bottom) / 2
|
||
|
||
if (min_x <= center_x <= max_x) and (min_y <= center_y <= max_y):
|
||
if 'DetectedText' in text_detection and text_detection['Confidence']>=50:
|
||
grid_texts[(row, col)].append(text_detection['DetectedText'])
|
||
# 计算每个网格的文字相似度（仅对计数较高的网格）
|
||
adjusted_grid_counts = {}
|
||
watermark_cells=[]
|
||
for grid, texts in grid_texts.items():
|
||
if group_name=="composite":
|
||
similarity = calculate_grouped_similarity_simple(texts)
|
||
else:
|
||
similarity = calculate_grouped_similarity_simple(texts,grouped = False)
|
||
# 如果相似度超过阈值，可能是水印，降低权重
|
||
if similarity > 0.7: # 相似度阈值，可调整
|
||
adjusted_count = 1 # 相似度越高，权重越低
|
||
(row,col)=grid
|
||
cell_left = grid_left_bound + col * grid_width
|
||
cell_right = cell_left + grid_width
|
||
cell_top = grid_top_bound + row * grid_height
|
||
cell_bottom = cell_top + grid_height
|
||
watermark_cells.append({
|
||
'grid': (row, col),
|
||
'center_x': (cell_left + cell_right) / 2,
|
||
'center_y': (cell_top + cell_bottom) / 2
|
||
})
|
||
else:
|
||
adjusted_count = grid_counts[grid]
|
||
|
||
adjusted_grid_counts[grid] = adjusted_count
|
||
|
||
# 选择最佳网格
|
||
if not adjusted_grid_counts:
|
||
logger.log(f"{group_name}组未找到合适的字幕区域",level="warning",program="getsrt")
|
||
return None
|
||
|
||
max_count = max(adjusted_grid_counts.values())
|
||
print(f"最高调整计数: {max_count}")
|
||
|
||
# 获取最佳网格候选
|
||
best_grids = [grid for grid, count in adjusted_grid_counts.items() if count == max_count]
|
||
|
||
# 如果没有找到合适的网格，尝试使用原始计数
|
||
if group_name=="composite" and max_count < 2: # 阈值可调整
|
||
print("调整后计数太低，尝试使用原始计数")
|
||
max_count_original = max(grid_counts.values())
|
||
best_grids = [grid for grid, count in grid_counts.items() if count == max_count_original]
|
||
print(f"最高原始计数: {max_count_original}")
|
||
|
||
if not best_grids:
|
||
logger.log(f"{group_name}组未找到合适的字幕区域",level="warning",program="getsrt")
|
||
return None
|
||
|
||
if len(best_grids) > 1:
|
||
# 直接使用中心位置作为最佳网格
|
||
best_grid = find_best_cluster_center(best_grids)
|
||
logger.log(f"{group_name}组发现 {len(best_grids)} 个最佳网格，选择平均网格: {best_grid}",program="getsrt")
|
||
else:
|
||
best_grid = best_grids[0]
|
||
logger.log(f"{group_name}组最佳网格: {best_grid}",program="getsrt")
|
||
watermark_count=0
|
||
for frame_idx, item in enumerate(sample_frames):
|
||
pic_path, ocr_json = item
|
||
# 为每个text_detection添加水印标记
|
||
for text_detection in ocr_json['TextDetections']:
|
||
# 初始化水印标记
|
||
if 'is_watermark' not in text_detection:
|
||
text_detection['is_watermark'] = False
|
||
text_detection['watermark_grid'] = None
|
||
|
||
# 获取文本区域信息
|
||
if 'Polygon' not in text_detection:
|
||
continue
|
||
|
||
polygon = [[int(coord['X']), int(coord['Y'])] for coord in text_detection['Polygon']]
|
||
min_x = min(p[0] for p in polygon)
|
||
max_x = max(p[0] for p in polygon)
|
||
min_y = min(p[1] for p in polygon)
|
||
max_y = max(p[1] for p in polygon)
|
||
size = int((max_x-min_x)/grid_width) * int((max_y-min_y)/grid_height)
|
||
detected_text = text_detection.get('DetectedText', '')
|
||
confidence = text_detection.get('Confidence', 0)
|
||
|
||
if not detected_text or confidence < 50:
|
||
continue
|
||
cell_count = 0
|
||
# 检查文本是否在任何一个水印网格中
|
||
for cell in watermark_cells:
|
||
# 检查网格中心是否在文字框内
|
||
if (min_x <= cell['center_x'] <= max_x) and (min_y <= cell['center_y'] <= max_y):
|
||
# 直接修改text_detection标记为水印
|
||
cell_count +=1
|
||
if cell_count > size*0.5:
|
||
text_detection['is_watermark'] = True
|
||
text_detection['watermark_grid'] = cell['grid']
|
||
watermark_count += 1
|
||
|
||
print(f"已标记 {watermark_count} 个水印字幕")
|
||
return best_grid, grid_width, grid_height, grid_top_bound, grid_left_bound
|
||
|
||
def find_central_subtitles(ocr_json, grid_left, grid_right, grid_top, grid_bottom,frame_name):
|
||
"""
|
||
检测画面中央附近的字幕作为标题
|
||
适用于kept_first和kept_last组
|
||
"""
|
||
central_subtitles = []
|
||
nearest_subtitles = []
|
||
img = cv2.imread(frame_name)
|
||
height, width = img.shape[:2]
|
||
min_distance = float('inf')
|
||
for text_detection in ocr_json['TextDetections']:
|
||
if text_detection['Confidence'] <= 20:
|
||
continue
|
||
|
||
|
||
polygon = [[int(coord['X']), int(coord['Y'])] for coord in text_detection['Polygon']]
|
||
min_x = min(p[0] for p in polygon)
|
||
max_x = max(p[0] for p in polygon)
|
||
min_y = min(p[1] for p in polygon)
|
||
max_y = max(p[1] for p in polygon)
|
||
|
||
# 计算文字框中心坐标
|
||
box_center_x = (min_x + max_x) / 2
|
||
box_center_y = (min_y + max_y) / 2
|
||
|
||
# 检查是否在画面中央区域
|
||
# 中央区域定义为：水平方向在中间50%，垂直方向在中间40%-60%
|
||
screen_center_x = width/ 2
|
||
screen_center_y = height / 2
|
||
|
||
# 允许一定的偏移范围
|
||
horizontal_tolerance = width * 0.1 # 水平方向40%的容差
|
||
vertical_tolerance = height * 0.1 # 垂直方向30%的容差
|
||
distance = ((box_center_x - screen_center_x) ** 2 +
|
||
(box_center_y - screen_center_y) ** 2) ** 0.5
|
||
|
||
# 检查是否在中央区域附近
|
||
if (abs(box_center_x - screen_center_x) <= horizontal_tolerance and
|
||
abs(box_center_y - screen_center_y) <= vertical_tolerance):
|
||
if 'is_watermark' not in text_detection or not text_detection['is_watermark']:
|
||
central_subtitles.append({
|
||
'min_y': min_y,
|
||
'max_y': max_y,
|
||
'text': text_detection['DetectedText'],
|
||
'confidence': text_detection['Confidence']
|
||
})
|
||
subtitle_info = {
|
||
'min_y': min_y,
|
||
'max_y': max_y,
|
||
'text': text_detection['DetectedText'],
|
||
'confidence': text_detection['Confidence'],
|
||
'distance': distance
|
||
}
|
||
|
||
# 更新最近的字幕
|
||
if distance < min_distance:
|
||
min_distance = distance
|
||
nearest_subtitles = [subtitle_info] # 重置为新的最近字幕
|
||
elif distance == min_distance:
|
||
nearest_subtitles.append(subtitle_info) # 添加相同距离的字幕
|
||
|
||
# 优先返回中央区域内的字幕
|
||
if central_subtitles:
|
||
return central_subtitles
|
||
|
||
# 如果没有中央区域内的字幕，返回离中央最近的字幕
|
||
if nearest_subtitles:
|
||
print(f"未找到中央区域内的字幕，使用离中央最近的字幕（距离: {min_distance:.2f}）")
|
||
return nearest_subtitles
|
||
return central_subtitles
|
||
|
||
def evaluate_text_quality_rules(text):
|
||
"""
|
||
基于规则评估文本质量
|
||
返回一个质量评分（0-1）
|
||
"""
|
||
# 1. 计算中文字符比例
|
||
total_chars = len(text)
|
||
chinese_chars = len(re.findall(r'[\u4e00-\u9fff]', text))
|
||
chinese_ratio = chinese_chars / total_chars if total_chars > 0 else 0
|
||
|
||
# 2. 计算中英文混杂程度
|
||
# 统计连续中文字符和连续英文字符的数量
|
||
chinese_segments = len(re.findall(r'[\u4e00-\u9fff]+', text))
|
||
english_segments = len(re.findall(r'[a-zA-Z]+', text))
|
||
mixed_score = 1 - min(1, abs(chinese_segments - english_segments) / max(1, chinese_segments + english_segments))
|
||
|
||
# 3. 计算词语通顺度（使用jieba分词）
|
||
words = list(jieba.cut(text))
|
||
valid_words = [word for word in words if len(word) > 1] # 过滤掉单字
|
||
word_score = len(valid_words) / len(words) if words else 0
|
||
|
||
# 4. 计算特殊字符比例
|
||
special_chars = len(re.findall(r'[^a-zA-Z0-9\u4e00-\u9fff\s]', text))
|
||
special_ratio = 1 - min(1, special_chars / total_chars) if total_chars > 0 else 1
|
||
|
||
# 综合评分
|
||
quality_score = 0.2 * chinese_ratio + 0.7 * word_score + 0.1 * special_ratio
|
||
return quality_score
|
||
from pypinyin import pinyin, Style
|
||
import re
|
||
import zhconv # 新增导入
|
||
def text_to_pinyin(text):
|
||
"""
|
||
将文本转换为拼音（忽略标点符号和空格）
|
||
"""
|
||
# 去除标点符号和空格，只保留中文和英文
|
||
cleaned_text = re.sub(r'[^\w\u4e00-\u9fff]', '', text)
|
||
if not cleaned_text:
|
||
return ""
|
||
cleaned_text = zhconv.convert(cleaned_text, 'zh-hans')
|
||
# 中文转拼音
|
||
pinyin_list = []
|
||
for char in cleaned_text:
|
||
if '\u4e00' <= char <= '\u9fff': # 中文字符
|
||
char_pinyin = pinyin(char, style=Style.NORMAL)[0][0]
|
||
pinyin_list.append(char_pinyin)
|
||
else: # 英文字符或数字
|
||
pinyin_list.append(char.lower())
|
||
|
||
return "".join(pinyin_list)
|
||
|
||
def calculate_pinyin_similarity(text1, text2):
|
||
"""
|
||
计算两个文本的拼音相似度
|
||
"""
|
||
pinyin1 = text_to_pinyin(text1)
|
||
pinyin2 = text_to_pinyin(text2)
|
||
|
||
if not pinyin1 or not pinyin2:
|
||
return 0
|
||
if pinyin2 in pinyin1 and len(text2)>=3:
|
||
return 1.0
|
||
if pinyin1 in pinyin2 and len(text1)>=3:
|
||
return 0.6
|
||
# 使用编辑距离计算相似度
|
||
return SequenceMatcher(None, pinyin1, pinyin2).ratio()
|
||
def match_text(audio_text,detected_text):
|
||
if not detected_text:
|
||
return 0
|
||
similarity=0
|
||
audio_text = re.sub(r'[^\w\u4e00-\u9fff]', '', audio_text)
|
||
detected_text = re.sub(r'[^\w\u4e00-\u9fff]', '', detected_text)
|
||
if len(audio_text) == len(detected_text):
|
||
similarity = calculate_pinyin_similarity(audio_text,detected_text)
|
||
elif len(audio_text) >= len(detected_text):
|
||
for i in range(len(audio_text)-len(detected_text)):
|
||
similarity = max(similarity,calculate_pinyin_similarity(audio_text[i:i+len(detected_text)],detected_text))
|
||
else:
|
||
for i in range(len(detected_text)-len(audio_text)):
|
||
similarity = max(similarity,calculate_pinyin_similarity(detected_text[i:i+len(audio_text)],audio_text))
|
||
return similarity
|
||
def find_best_ocr_match(audio_text, ocr_detections,frame_name, min_confidence=75):
|
||
"""
|
||
在OCR结果中寻找与音频文字拼音最匹配的文本
|
||
"""
|
||
if not audio_text or not ocr_detections:
|
||
return None, 0
|
||
img = cv2.imread(frame_name)
|
||
height, width = img.shape[:2]
|
||
best_match = None
|
||
best_similarity = -1
|
||
second_match = None
|
||
second_similarity = -1
|
||
for detection in ocr_detections:
|
||
if detection.get('Confidence', 0) < min_confidence:
|
||
continue
|
||
polygon = [[int(coord['X']), int(coord['Y'])] for coord in detection['Polygon']]
|
||
min_x = min(p[0] for p in polygon)
|
||
max_x = max(p[0] for p in polygon)
|
||
min_y = min(p[1] for p in polygon)
|
||
max_y = max(p[1] for p in polygon)
|
||
box_center_y = (min_y + max_y) / 2
|
||
box_center_x = (min_x + max_x) / 2
|
||
detected_text = detection.get('DetectedText', '')
|
||
detected_text = re.sub(r'[^\w\u4e00-\u9fff]', '', detected_text)
|
||
if not detected_text :
|
||
continue
|
||
if len(detected_text) > len(audio_text)+3 and len(detected_text)<=5:
|
||
continue
|
||
# 计算拼音相似度
|
||
similarity = match_text(audio_text, detected_text)
|
||
if len(detected_text)>=10:
|
||
similarity = max(similarity,match_text(audio_text,detected_text[:len(detected_text)//2]))
|
||
if len(audio_text)>=10:
|
||
similarity = max(similarity,match_text(audio_text[:len(audio_text)//2],detected_text))
|
||
similarity = max(similarity,match_text(audio_text[len(audio_text)//2:],detected_text))
|
||
if len(audio_text)>=30:
|
||
similarity = max(similarity,calculate_pinyin_similarity(audio_text[3*len(audio_text)//4:],detected_text))
|
||
similarity = max(similarity,calculate_pinyin_similarity(audio_text[:len(audio_text)//4],detected_text))
|
||
if len(ocr_detections)==1 and similarity<0.8:
|
||
similarity+=0.2
|
||
if similarity<0.8 and (box_center_x>width*0.8 or box_center_x<width*0.2):
|
||
continue
|
||
elif box_center_y<height*0.1:
|
||
continue
|
||
if 'is_watermark' in detection:
|
||
if detection['is_watermark'] and similarity<0.85 and (box_center_x>width*0.9 or box_center_x<width*0.1):
|
||
continue
|
||
if detection['is_watermark'] and similarity<0.8 and box_center_y<height*0.3:
|
||
continue
|
||
if abs(best_similarity-similarity)<0.1 and len(best_match.get('DetectedText', '')) < len(detected_text) and similarity>0.85:
|
||
second_match = best_match
|
||
second_similarity = best_similarity
|
||
best_match = detection
|
||
elif similarity > best_similarity:
|
||
second_match = best_match
|
||
second_similarity = best_similarity
|
||
best_similarity = similarity
|
||
best_match = detection
|
||
else:
|
||
if similarity == best_similarity:
|
||
if len(best_match.get('DetectedText', '')) < len(detected_text):
|
||
second_match = best_match
|
||
second_similarity = best_similarity
|
||
best_match = detection
|
||
else:
|
||
second_similarity = similarity
|
||
second_match = detection
|
||
else:
|
||
if similarity == second_similarity:
|
||
join1 =f"{second_match['DetectedText']} {detection['DetectedText']}"
|
||
join2 =f"{detection['DetectedText']} {second_match['DetectedText']}"
|
||
similarity1 = match_text(audio_text,join1)
|
||
similarity2 = match_text(audio_text,join2)
|
||
if similarity1>similarity2:
|
||
second_match['DetectedTest'] = join1
|
||
else:
|
||
second_match['DetectedTest'] = join2
|
||
if similarity>second_similarity:
|
||
second_similarity = similarity
|
||
second_match = detection
|
||
if second_match:
|
||
#print(second_match)
|
||
polygon = [[int(coord['X']), int(coord['Y'])] for coord in best_match['Polygon']]
|
||
min_x = min(p[0] for p in polygon)
|
||
max_x = max(p[0] for p in polygon)
|
||
min_y = min(p[1] for p in polygon)
|
||
max_y = max(p[1] for p in polygon)
|
||
|
||
# 计算文字框中心Y坐标
|
||
box_center_y = (min_y + max_y) / 2
|
||
box_center_x = (min_x + max_x) / 2
|
||
polygon = [[int(coord['X']), int(coord['Y'])] for coord in second_match['Polygon']]
|
||
min_x2 = min(p[0] for p in polygon)
|
||
max_x2 = max(p[0] for p in polygon)
|
||
min_y2 = min(p[1] for p in polygon)
|
||
max_y2 = max(p[1] for p in polygon)
|
||
|
||
# 计算文字框中心Y坐标
|
||
box_center_y2 = (min_y2 + max_y2) / 2
|
||
box_center_x2 = (min_x2 + max_x2) / 2
|
||
if abs(box_center_y-box_center_y2)>2*(max_y-min_y) or min(min(abs(max_x2-min_x),abs(max_x-min_x2)),abs(box_center_x-box_center_x2))>4*(max_y-min_y):
|
||
return best_match, best_similarity
|
||
# 获取两个文本的Y坐标（使用Polygon的第一个点的Y值）
|
||
best_y = best_match['Polygon'][0]['Y']
|
||
second_y = second_match['Polygon'][0]['Y']
|
||
|
||
# 按Y坐标从小到大排序
|
||
if max_x<=min_x2:
|
||
combined_text = f"{best_match['DetectedText']} {second_match['DetectedText']}"
|
||
elif max_x2<=min_x:
|
||
combined_text = f"{second_match['DetectedText']} {best_match['DetectedText']}"
|
||
elif best_y < second_y :
|
||
combined_text = f"{best_match['DetectedText']} {second_match['DetectedText']}"
|
||
else:
|
||
combined_text = f"{second_match['DetectedText']} {best_match['DetectedText']}"
|
||
similarity = match_text(audio_text,combined_text)
|
||
#print(combined_text)
|
||
#print(similarity)
|
||
if similarity>=0.8 and len(combined_text)-len(audio_text)<5:
|
||
best_match['DetectedText'] = combined_text
|
||
if second_similarity >= 0.9 and similarity>=0.6:
|
||
best_match['DetectedText'] = combined_text
|
||
elif best_match is not None:
|
||
if 'is_watermark' in best_match:
|
||
if best_match['is_watermark'] and similarity<0.6:
|
||
return None, -1
|
||
return best_match, best_similarity
|
||
def calculate_english_ratio(text):
|
||
"""计算文本中英文字符的占比"""
|
||
if not text:
|
||
return 0
|
||
|
||
total_chars = len(text)
|
||
if total_chars == 0:
|
||
return 0
|
||
|
||
# 统计英文字符（字母、数字、英文标点）
|
||
english_chars = 0
|
||
for char in text:
|
||
# 判断是否为英文字符（包括数字和基本标点）
|
||
if ('a' <= char <= 'z') or ('A' <= char <= 'Z') or ('0' <= char <= '9'):
|
||
english_chars += 1
|
||
elif char in ' .,!?;:\'"-()[]{}':
|
||
english_chars += 1
|
||
|
||
return english_chars / total_chars
|
||
|
||
def _keep_main_cluster_per_line(boxes, grid_center_x, gap_char_mult=2.5, gap_min=80):
|
||
"""同一行内, 相邻框横向间隔过大(>gap_char_mult字宽 且 >gap_min像素)就断成多簇,
|
||
只保留【离字幕中心最近】的那一簇 → 把离得远的水印/角标从拼接里踢掉(不拼距离大的)。
|
||
多行字幕按行分别处理, 不影响正常换行。"""
|
||
if len(boxes) <= 1:
|
||
return boxes
|
||
avg_h = sum(b['max_y'] - b['min_y'] for b in boxes) / len(boxes)
|
||
lt = avg_h * 0.8
|
||
sb = sorted(boxes, key=lambda b: b['center_y'])
|
||
lines = [[sb[0]]]
|
||
for b in sb[1:]:
|
||
lc = sum(x['center_y'] for x in lines[-1]) / len(lines[-1])
|
||
if abs(b['center_y'] - lc) <= lt:
|
||
lines[-1].append(b)
|
||
else:
|
||
lines.append([b])
|
||
kept = []
|
||
for line in lines:
|
||
line.sort(key=lambda b: b['center_x'])
|
||
if len(line) == 1:
|
||
kept.append(line[0]); continue
|
||
ws = sorted((b['max_x'] - b['min_x']) / max(1, len(b['text'])) for b in line)
|
||
cw = ws[len(ws) // 2]
|
||
gap_thresh = max(cw * gap_char_mult, gap_min)
|
||
clusters = [[line[0]]]
|
||
for b in line[1:]:
|
||
if b['min_x'] - clusters[-1][-1]['max_x'] > gap_thresh:
|
||
clusters.append([b])
|
||
else:
|
||
clusters[-1].append(b)
|
||
if len(clusters) == 1:
|
||
kept.extend(line)
|
||
else:
|
||
def cc(c):
|
||
return (min(x['min_x'] for x in c) + max(x['max_x'] for x in c)) / 2
|
||
kept.extend(min(clusters, key=lambda c: abs(cc(c) - grid_center_x)))
|
||
return kept
|
||
|
||
|
||
def simple_smart_join(subtitle_boxes):
|
||
"""
|
||
智能排序：按Y分行(行间从上到下)，同行内按X从左到右。
|
||
分行判据用【垂直范围重叠率】而非"中心Y距离"：上下两行几乎不重叠→必分行→上行在前;
|
||
同一行的左右两块Y范围高度重叠→同行→按X排。
|
||
(原来用"中心Y距离<0.8行高",字高抖动时会把上下两行误判同行→按X排→双行顺序翻车。)
|
||
"""
|
||
if not subtitle_boxes:
|
||
return []
|
||
if len(subtitle_boxes) == 1:
|
||
return subtitle_boxes
|
||
|
||
# 按 center_y 排序后贪心分行：用【中心Y间隔】判同行(对OCR框高异常更稳健——
|
||
# 重叠率会被某个异常高的框骗到"同行"再按X排导致上下行翻转;中心Y间隔不受框高影响)。
|
||
# 中心Y间隔 < 0.5×较小框高 → 同行(左右两块,中心Y几乎相等);否则 → 上下两行。
|
||
sorted_boxes = sorted(subtitle_boxes, key=lambda b: b['center_y'])
|
||
lines = [[sorted_boxes[0]]]
|
||
for box in sorted_boxes[1:]:
|
||
prev = lines[-1][-1]
|
||
gap = abs(box['center_y'] - prev['center_y'])
|
||
min_h = min(box['max_y'] - box['min_y'], prev['max_y'] - prev['min_y'])
|
||
if min_h > 0 and gap < 0.5 * min_h:
|
||
lines[-1].append(box) # 中心Y很近 → 同一行
|
||
else:
|
||
lines.append([box]) # 中心Y拉开 → 新的一行(在下方)
|
||
|
||
# 行间已是上→下(按center_y排过)；同行内按 center_x 左→右
|
||
result = []
|
||
for line in lines:
|
||
line.sort(key=lambda b: b['center_x'])
|
||
result.extend(line)
|
||
return result
|
||
def extract_highest_subtitle(idx, all_pic_ocr_pair, best_grids, metadata_path,h,logger):
|
||
"""
|
||
提取纵向跨度最大的字幕框（基于Y坐标比较）
|
||
根据帧的类型使用不同的最佳网格参数
|
||
|
||
参数:
|
||
all_pic_ocr_pair: 图片OCR数据列表 [(pic_path1, ocr_json1), ...]
|
||
best_grids: 包含三个组的最佳网格参数的字典
|
||
metadata_path: metadata.json文件路径
|
||
|
||
返回:
|
||
纵向跨度最大的字幕文本（或None）
|
||
"""
|
||
# 读取metadata.json获取帧的分类信息
|
||
with open(metadata_path, 'r', encoding='utf-8') as f:
|
||
metadata = json.load(f)
|
||
kept_first_names = set(metadata['kept_first'])
|
||
kept_last_names = set(metadata['kept_last'])
|
||
|
||
pic_path, ocr_json = all_pic_ocr_pair[idx]
|
||
frame_name = os.path.basename(pic_path)
|
||
# 确定帧的类型并选择相应的最佳网格参数
|
||
if frame_name in kept_first_names:
|
||
grid_params = best_grids["kept_first"]
|
||
h = h["kept_first"]
|
||
group_name = "kept_first"
|
||
hh=0.8*h
|
||
elif frame_name in kept_last_names:
|
||
grid_params = best_grids["kept_last"]
|
||
h = h["kept_last"]
|
||
group_name = "kept_last"
|
||
hh=0.8*h
|
||
else:
|
||
grid_params = best_grids["composite"]
|
||
h = h["composite"]
|
||
group_name = "composite"
|
||
hh=1.5*h
|
||
|
||
if not grid_params:
|
||
return "",90, True,0
|
||
best_grid, grid_width, grid_height, grid_top, grid_left = grid_params
|
||
best_row, best_col = best_grid
|
||
|
||
# 计算网格边界
|
||
grid_left = grid_left + best_col * grid_width
|
||
grid_right = grid_left + grid_width
|
||
grid_top = grid_top + best_row * grid_height
|
||
grid_bottom = grid_top + grid_height
|
||
# composite字幕条只包含字幕文字，X方向使用全图宽度作为容差
|
||
img_half_width = 50 * grid_width
|
||
audio_text = metadata.get("audio_text", {})
|
||
|
||
# 如果是kept_first帧且有前三秒音频文字
|
||
if frame_name in kept_first_names and audio_text["first_3s"]!="!!" :
|
||
if not audio_text.get("first_3s"):
|
||
return "", 90, True,0
|
||
audio_str = audio_text["first_3s"]
|
||
logger.log(f"前三秒音频文字: {audio_str}",program="getsrt")
|
||
|
||
# 在OCR结果中寻找最佳匹配
|
||
best_detection, similarity = find_best_ocr_match(audio_str, ocr_json.get('TextDetections', []),pic_path)
|
||
|
||
if best_detection and similarity > 0.65: # 相似度阈值可调整
|
||
matched_text = best_detection['DetectedText']
|
||
confidence = best_detection['Confidence']
|
||
logger.log(f"找到拼音匹配的OCR文本: {matched_text} (相似度: {similarity:.2f})",program="getsrt")
|
||
return matched_text, similarity*50+confidence/2, True ,similarity
|
||
|
||
logger.log("未找到足够相似的OCR匹配",program="getsrt")
|
||
return "", 90, True, 0 # 音频文字置信度设为90
|
||
|
||
# 如果是kept_last帧且有后三秒音频文字
|
||
if frame_name in kept_last_names and audio_text["first_3s"]!="!!":
|
||
if not audio_text.get("last_3s"):
|
||
return "", 90, True, 0
|
||
audio_str = audio_text["last_3s"]
|
||
logger.log(f"后三秒音频文字: {audio_str}",program="getsrt")
|
||
|
||
# 在OCR结果中寻找最佳匹配
|
||
best_detection, similarity = find_best_ocr_match(audio_str, ocr_json.get('TextDetections', []),pic_path)
|
||
#如果音频文字短，说明几乎没说话，对相似度要求更高
|
||
if len(audio_str)<5:
|
||
if similarity<0.8:
|
||
return "", 90, True, 0
|
||
if best_detection and similarity > 0.65: # 相似度阈值可调整
|
||
matched_text = best_detection['DetectedText']
|
||
confidence = best_detection['Confidence']
|
||
logger.log(f"找到拼音匹配的OCR文本: {matched_text} (相似度: {similarity:.2f})",program="getsrt")
|
||
return matched_text, similarity*20+confidence*0.8, True,similarity
|
||
|
||
logger.log("未找到足够相似的OCR匹配",program="getsrt")
|
||
return "", 90, True,0
|
||
|
||
# 计算网格中心Y坐标
|
||
grid_center_y = (grid_top + grid_bottom) / 2
|
||
grid_center_x = (grid_left + grid_right) / 2
|
||
|
||
# 收集所有符合条件的字幕框
|
||
subtitle_boxes = []
|
||
if len(ocr_json['TextDetections'])==1:
|
||
text_detection = ocr_json['TextDetections'][0]
|
||
if text_detection['Confidence']<=60:
|
||
return "", 90, True,0
|
||
|
||
|
||
polygon = [[int(coord['X']), int(coord['Y'])] for coord in text_detection['Polygon']]
|
||
min_x = min(p[0] for p in polygon)
|
||
max_x = max(p[0] for p in polygon)
|
||
min_y = min(p[1] for p in polygon)
|
||
max_y = max(p[1] for p in polygon)
|
||
|
||
# 计算文字框中心Y坐标
|
||
box_center_y = (min_y + max_y) / 2
|
||
box_center_x = (min_x + max_x) / 2
|
||
if 'is_watermark' in text_detection:
|
||
if text_detection['is_watermark']:
|
||
if abs(box_center_y - grid_center_y) <= hh and not (max_x < grid_left or min_x > grid_right):
|
||
return text_detection['DetectedText'],text_detection['Confidence'],False,0
|
||
return "", 90, True,0
|
||
if abs(box_center_y - grid_center_y) <= 3*hh and abs(box_center_x - grid_center_x) <= 3*hh:
|
||
if text_detection['Confidence']<=80:
|
||
return text_detection['DetectedText'],text_detection['Confidence'],False,0
|
||
return text_detection['DetectedText'],text_detection['Confidence'],True,0
|
||
for text_detection in ocr_json['TextDetections']:
|
||
if text_detection['Confidence']<=60:
|
||
continue
|
||
if 'is_watermark' in text_detection:
|
||
if text_detection['is_watermark']:
|
||
continue
|
||
polygon = [[int(coord['X']), int(coord['Y'])] for coord in text_detection['Polygon']]
|
||
min_x = min(p[0] for p in polygon)
|
||
max_x = max(p[0] for p in polygon)
|
||
min_y = min(p[1] for p in polygon)
|
||
max_y = max(p[1] for p in polygon)
|
||
|
||
# 计算文字框中心Y坐标
|
||
box_center_y = (min_y + max_y) / 2
|
||
box_center_x = (min_x + max_x) / 2
|
||
if group_name in ["kept_first", "kept_last"]:
|
||
hh = 0.5*h
|
||
else:
|
||
hh = 1.5*h
|
||
hnow = max_y-min_y
|
||
# 检查是否在Y轴方向接近最佳网格
|
||
if abs(box_center_y - grid_center_y) <= hh:
|
||
# composite字幕条仅含字幕文字，使用全图宽度；其他组用2*hnow
|
||
x_tol = img_half_width if group_name == "composite" else 2 * hnow
|
||
if not (max_x < grid_left - x_tol or min_x > grid_right + x_tol):
|
||
subtitle_boxes.append({
|
||
'min_y': min_y,
|
||
'max_y': max_y,
|
||
'min_x': min_x,
|
||
'max_x': max_x,
|
||
'text': text_detection['DetectedText'],
|
||
'confidence': text_detection['Confidence'],
|
||
'center_x': box_center_x,
|
||
'center_y': box_center_y, # 添加中心Y坐标用于后续比较
|
||
'distance_to_center': abs(box_center_y - grid_center_y) # 添加距离中心的绝对值
|
||
})
|
||
else:
|
||
if abs(box_center_x - grid_center_x) <= hh and not (max_y <grid_top or min_y>grid_bottom):
|
||
subtitle_boxes.append({
|
||
'min_y': min_y,
|
||
'max_y': max_y,
|
||
'min_x': min_x,
|
||
'max_x': max_x,
|
||
'text': text_detection['DetectedText'],
|
||
'confidence': text_detection['Confidence'],
|
||
'center_x': box_center_x,
|
||
'center_y': box_center_y, # 添加中心Y坐标用于后续比较
|
||
'distance_to_center': abs(box_center_y - grid_center_y) # 添加距离中心的绝对值
|
||
})
|
||
|
||
# 如果没有找到符合条件的字幕框，返回None
|
||
if group_name in ["kept_first", "kept_last"] and not subtitle_boxes:
|
||
# 尝试检测画面中央附近的字幕作为标题
|
||
return "", 90, True,0
|
||
subtitle_boxes = find_central_subtitles(ocr_json, grid_left, grid_right, grid_top, grid_bottom,pic_path)
|
||
if (len(subtitle_boxes)==1):
|
||
center_y = (subtitle_boxes[0]['min_y']+subtitle_boxes[0]['max_y'])/2
|
||
diff_y = (-subtitle_boxes[0]['min_y']+subtitle_boxes[0]['max_y'])
|
||
length = (-subtitle_boxes[0]['min_x']+subtitle_boxes[0]['max_x'])/len(subtitle_boxes[0]['text'])
|
||
for text_detection in ocr_json['TextDetections']:
|
||
if text_detection['Confidence']<=20:
|
||
continue
|
||
if 'is_watermark' in text_detection:
|
||
if text_detection['is_watermark']:
|
||
continue
|
||
if text_detection['DetectedText']==subtitle_boxes[0]['text']:
|
||
continue
|
||
polygon = [[int(coord['X']), int(coord['Y'])] for coord in text_detection['Polygon']]
|
||
min_x = min(p[0] for p in polygon)
|
||
max_x = max(p[0] for p in polygon)
|
||
min_y = min(p[1] for p in polygon)
|
||
max_y = max(p[1] for p in polygon)
|
||
box_center_y = (min_y + max_y) / 2
|
||
box_center_x = (min_x + max_x) / 2
|
||
if abs(box_center_y - center_y)<=h/4 and abs(max_y-min_y-diff_y)<=h/2:
|
||
if abs(min_x-subtitle_boxes[0]['max_x'])<2*length or abs(subtitle_boxes[0]['min_x']-max_x)<=2*length:
|
||
subtitle_boxes.append({
|
||
'min_y': min_y,
|
||
'max_y': max_y,
|
||
'min_x': min_x,
|
||
'max_x': max_x,
|
||
'text': text_detection['DetectedText'],
|
||
'confidence':text_detection['Confidence'],
|
||
'center_x': box_center_x,
|
||
'center_y': box_center_y, # 添加中心Y坐标用于后续比较
|
||
'distance_to_center': abs(box_center_y - grid_center_y) # 添加距离中心的绝对值
|
||
})
|
||
|
||
filtered_boxes = []
|
||
if len(subtitle_boxes) > 1:
|
||
# 按距离网格中心的距离排序（从近到远）
|
||
# 过滤掉英文占比超过80%的box
|
||
filtered_boxes = []
|
||
for box in subtitle_boxes:
|
||
text = box['text']
|
||
english_ratio = calculate_english_ratio(text)
|
||
if (english_ratio <= 0.7 and box['confidence']>=75) : # 保留英文占比不超过80%的box
|
||
filtered_boxes.append(box)
|
||
subtitle_boxes = filtered_boxes
|
||
if len(subtitle_boxes) > 1:
|
||
# 先踢掉"离字幕中心太远"的框(水印/角标不与居中字幕拼接), 再排序拼接
|
||
if config.get_subtitle_config().get('split_far_boxes', True):
|
||
subtitle_boxes = _keep_main_cluster_per_line(subtitle_boxes, grid_center_x)
|
||
subtitle_boxes = simple_smart_join(subtitle_boxes)
|
||
best_subtitle = "".join(box['text'] for box in subtitle_boxes)
|
||
elif len(subtitle_boxes) == 1:
|
||
#print(subtitle_boxes)
|
||
best_subtitle = subtitle_boxes[0]['text']
|
||
if not subtitle_boxes:
|
||
return "", 90, True,0
|
||
|
||
confidence = sum(box['confidence'] for box in subtitle_boxes) / len(subtitle_boxes)
|
||
if best_subtitle:
|
||
# 如果有找到字幕，返回文本和置信度
|
||
if (len(subtitle_boxes)>1 and confidence<=90 ):
|
||
return best_subtitle, confidence,False,0
|
||
else:
|
||
return best_subtitle, confidence,True,0
|
||
else:
|
||
return "", 90, True,0
|
||
|
||
def _filter_watermark_by_position(all_pic_ocr_pair, logger, dom_frac=0.25, min_count=20,
|
||
max_distinct=40, bx=80, by=30):
|
||
"""坐标+单文本占比 删水印框(在逐帧提取字幕【之前】跑)。
|
||
把所有帧的OCR框按【位置分桶】,统计每个位置的文字:
|
||
某【单一文本】占该位置 ≥dom_frac 且出现≥min_count次 → 一个文本霸占此位置 = 水印/台标/固定标语 → 整框剔除。
|
||
真字幕在同一位置是【很多不同句子轮流出现】(单文本占比极低,实测~1%),不会被删。
|
||
注意: 不能用"多样性(不同/总数)"判, 因为字幕每句显示3-5帧, 多样性天然~0.3 会误删整条主字幕带。"""
|
||
def _norm(s):
|
||
return re.sub(r"[\s\W_]+", "", s)
|
||
|
||
def _key(det):
|
||
poly = det.get("Polygon") or []
|
||
if len(poly) < 4:
|
||
return None
|
||
cx = sum(p["X"] for p in poly) / len(poly)
|
||
cy = sum(p["Y"] for p in poly) / len(poly)
|
||
return (int(cx // bx), int(cy // by))
|
||
|
||
bucket = defaultdict(list)
|
||
for _pic, oj in all_pic_ocr_pair:
|
||
for det in oj.get("TextDetections", []):
|
||
k = _key(det)
|
||
t = _norm(det.get("DetectedText", ""))
|
||
if k and t:
|
||
bucket[k].append(t)
|
||
|
||
wm = set()
|
||
for k, texts in bucket.items():
|
||
if len(texts) < min_count:
|
||
continue
|
||
cnt = Counter(texts)
|
||
top = cnt.most_common(1)[0][1]
|
||
distinct = len(cnt)
|
||
# 水印: 单一文本占比高(>=dom_frac) 且 该位置不同文本少(<max_distinct);
|
||
# 字幕带即便有重复, "不同句子"也有几十上百种, distinct 大 → 不会被判水印。
|
||
if top / len(texts) >= dom_frac and distinct < max_distinct:
|
||
wm.add(k)
|
||
if not wm:
|
||
return 0
|
||
|
||
# 边缘列排除: 水印多固定在左/右下角, 字幕居中且与水印有间隔。若检测到的水印桶集中在某一边缘,
|
||
# 把【该边缘整列(到字幕区之前)】都排除 → 连被OCR打碎、漂移的水印碎片一起清掉(字幕居中不受影响)。
|
||
all_xb = [k[0] for k in bucket]
|
||
max_xb = max(all_xb) if all_xb else 0
|
||
edge_cols = set()
|
||
if max_xb > 0:
|
||
wm_xb = [k[0] for k in wm]
|
||
left_wm = [x for x in wm_xb if x <= max_xb * 0.35]
|
||
right_wm = [x for x in wm_xb if x >= max_xb * 0.65]
|
||
if len(left_wm) >= max(2, len(wm_xb) * 0.5): # 水印主要在左边
|
||
edge_cols = {x for x in range(0, max(left_wm) + 1)}
|
||
elif len(right_wm) >= max(2, len(wm_xb) * 0.5): # 水印主要在右边
|
||
edge_cols = {x for x in range(min(right_wm), max_xb + 1)}
|
||
|
||
removed = 0
|
||
for _pic, oj in all_pic_ocr_pair:
|
||
dets = oj.get("TextDetections", [])
|
||
keep = []
|
||
for det in dets:
|
||
k = _key(det)
|
||
if k is not None and (k in wm or k[0] in edge_cols):
|
||
removed += 1
|
||
else:
|
||
keep.append(det)
|
||
oj["TextDetections"] = keep
|
||
if logger:
|
||
logger.log(f"坐标水印过滤: {len(wm)}个固定位置判为水印(文字恒定), 删除{removed}个框", program="getsrt")
|
||
return removed
|
||
|
||
|
||
def _filter_static_text(results, logger, min_frac=0.25, min_prefix_len=6, min_count=10,
|
||
strip_prefix=False):
|
||
"""静态文字行过滤: 把"很多帧反复出现的同一文字"(水印/台标/固定标题)从逐帧文本里剔除。
|
||
真字幕每条只在少数连续帧出现,占总帧比例很低(实测<5%);水印/台标占比很高(实测30%+)。
|
||
- 全文静态(水印/台标): 某文本归一化后在 >=min_frac 的有字幕帧出现 → 整帧抹空。【默认开,已验证安全】
|
||
- 静态前缀(固定标题): 某长前缀在 >=min_frac 的帧出现 → 剥掉该前缀。【默认关,strip_prefix=True 才启用:
|
||
有误删风险——若真字幕碰巧共享开头(如"小明:…")会被误剥】
|
||
results: list of (frame_idx, text)。返回过滤后的同结构 list。"""
|
||
def _norm(s):
|
||
return re.sub(r"[\s\W_]+", "", s)
|
||
|
||
nonempty = [t.strip() for _, t in results if t and t.strip()]
|
||
n = len(nonempty)
|
||
if n < min_count:
|
||
return results
|
||
thr = max(min_count, int(min_frac * n))
|
||
|
||
static_full = {k for k, c in Counter(_norm(t) for t in nonempty).items() if k and c >= thr}
|
||
|
||
static_pref = []
|
||
if strip_prefix:
|
||
pref_cnt = Counter()
|
||
for t in nonempty:
|
||
for L in range(min_prefix_len, min(len(t), 40) + 1):
|
||
pref_cnt[t[:L]] += 1
|
||
for p in sorted((p for p, c in pref_cnt.items() if c >= thr), key=len, reverse=True):
|
||
if not any(q.startswith(p) for q in static_pref):
|
||
static_pref.append(p)
|
||
|
||
if not static_full and not static_pref:
|
||
return results
|
||
|
||
def _is_static_variant(key):
|
||
# 与水印互为子串(截断/拼接变体,如 "红警魔鬼蓝" / "…bilibili…")也算水印
|
||
if key in static_full:
|
||
return True
|
||
for s in static_full:
|
||
if len(key) >= 4 and (key in s or s in key):
|
||
return True
|
||
return False
|
||
|
||
removed = stripped = 0
|
||
out = []
|
||
for i, t in results:
|
||
if not t or not t.strip():
|
||
out.append((i, t)); continue
|
||
ts = t.strip()
|
||
if _is_static_variant(_norm(ts)):
|
||
out.append((i, "")); removed += 1; continue
|
||
for p in static_pref:
|
||
if ts == p:
|
||
ts = ""; removed += 1; break
|
||
if ts.startswith(p) and len(ts) > len(p):
|
||
ts = ts[len(p):].strip(); stripped += 1; break
|
||
out.append((i, ts))
|
||
|
||
if logger and (removed or stripped):
|
||
logger.log(f"静态文字过滤: 抹除{removed}帧(水印/台标), 剥固定前缀{stripped}帧; "
|
||
f"静态项={list(static_full)[:2]}{static_pref[:2]}", program="getsrt")
|
||
return out
|
||
|
||
|
||
def _collapse_repeat(s):
|
||
"""折叠一行内"紧挨着重复"的内容(OCR/拼接把同一片读了两遍):
|
||
整句重复: '追了的话就把那人头打了追了的话就把那人头打了' → '追了...打了'
|
||
前缀重复: 'Ok我就解释这一次Ok我就解释这一次再有...' → 'Ok我就解释这一次再有...'
|
||
通用做法: 反复去掉【相邻重复块】(≥5字, 从长到短优先), 直到稳定。"""
|
||
orig = (s or "").strip()
|
||
t = orig
|
||
if len(t) < 8:
|
||
return s
|
||
changed = True
|
||
while changed:
|
||
changed = False
|
||
n = len(t)
|
||
for L in range(n // 2, 4, -1): # 块长从长到短, 最短5字(避免误删短重复字)
|
||
hit = -1
|
||
for i in range(0, n - 2 * L + 1):
|
||
if t[i:i + L] == t[i + L:i + 2 * L]:
|
||
hit = i; break
|
||
if hit >= 0:
|
||
t = t[:hit + L] + t[hit + 2 * L:] # 删掉重复的那一份
|
||
changed = True
|
||
break
|
||
# 近似二重: 切两半, 一半是另一半的前缀(第二遍OCR少读几字)→ 取较完整那半
|
||
h = len(t) // 2
|
||
a, b = t[:h], t[h:]
|
||
if len(a) >= 5 and (a.startswith(b) or b.startswith(a)):
|
||
t = a if len(a) >= len(b) else b
|
||
return t if t != orig else s
|
||
|
||
|
||
def generate_srt_from_results(results, output_path, fps=30, interval_seconds=1):
|
||
"""
|
||
生成SRT字幕文件，合并相同的字幕文本
|
||
|
||
参数:
|
||
results: 字幕文本列表 [(frame_idx, text), ...]
|
||
output_path: 输出文件路径
|
||
fps: 视频帧率（用于精确时间计算，默认30帧/秒）
|
||
interval_seconds: 字幕时间间隔（秒），默认0.6秒
|
||
"""
|
||
if not results:
|
||
print("警告: 没有字幕数据可生成")
|
||
return
|
||
|
||
# 合并相同的字幕文本
|
||
merged_subtitles = []
|
||
current_text = None
|
||
start_time = 0
|
||
end_time = 0
|
||
|
||
for i, (frame_idx, text) in enumerate(results):
|
||
if (interval_seconds == 0.33):
|
||
cur_end = (i + 1) // 3 + (i + 1) % 3 * interval_seconds
|
||
else:
|
||
cur_end = (i + 1) * interval_seconds
|
||
|
||
# 完全相同 → 延长
|
||
if text == current_text:
|
||
end_time = cur_end
|
||
# 新文本以当前文本为前缀（渐进蹦字）→ 更新为更长的文本，延长时间
|
||
# 注意：要求 current_text 非空。空串是任何字符串的前缀，若允许会把
|
||
# 新字幕误判为"空白段的延续"，导致起始时间错误地继承前面空白的起点。
|
||
elif current_text and text and text.startswith(current_text):
|
||
current_text = text
|
||
end_time = cur_end
|
||
# 当前文本以新文本为前缀（字幕开始收缩/切换）→ 正常保存，开始新段
|
||
else:
|
||
if current_text is not None and i != 1 and end_time >= 1:
|
||
merged_subtitles.append({
|
||
'start': start_time,
|
||
'end': end_time,
|
||
'text': current_text
|
||
})
|
||
current_text = text
|
||
if (interval_seconds == 0.33):
|
||
start_time = i // 3 + (i % 3) * interval_seconds
|
||
else:
|
||
start_time = i * interval_seconds
|
||
end_time = cur_end
|
||
|
||
# 添加最后一个字幕段
|
||
if current_text is not None and end_time-start_time>0.4:
|
||
merged_subtitles.append({
|
||
'start': start_time,
|
||
'end': end_time,
|
||
'text': current_text
|
||
})
|
||
|
||
# 过滤空文本的字幕段
|
||
merged_subtitles = [sub for sub in merged_subtitles if sub['text'].strip()]
|
||
|
||
# 后处理：合并相邻高度相似的字幕（空帧打断 / OCR标点不一致）
|
||
def _norm_punct(t):
|
||
"""归一化标点（仅用于比对，不改输出）：
|
||
- ······ / ...... / … 统一为省略号占位符
|
||
- 横杠 - – — 与 '一'(yī) 统一：二者字形几乎相同，OCR 常互相误读"""
|
||
t = re.sub(r'[·•]{2,}', '…', t)
|
||
t = re.sub(r'\.{2,}', '…', t)
|
||
t = re.sub(r'[-–—一]', '—', t)
|
||
return t
|
||
|
||
def _is_subseq(short, long):
|
||
"""short 是否为 long 的子序列（每个字符按序出现在 long 中）。
|
||
用于判定 OCR 漏字的同一句字幕（如漏掉中间的 '……这样'）。"""
|
||
it = iter(long)
|
||
return all(ch in it for ch in short)
|
||
|
||
def _containment(short, long):
|
||
"""较短串被较长串覆盖的比例（基于公共匹配块总长 / 较短串长度）。
|
||
对截断 + 分散错字的同一句鲁棒（如 '同砚'→'同观'、'糗事'→'摸事' 等零散OCR错）。"""
|
||
if not short:
|
||
return 0.0
|
||
sm = SequenceMatcher(None, short, long)
|
||
matched = sum(blk.size for blk in sm.get_matching_blocks())
|
||
return matched / len(short)
|
||
|
||
def _strip_all_punct(t):
|
||
"""去掉所有标点/符号/空白，只保留文字，用于核心内容比对。"""
|
||
return re.sub(r'[\s·•.…，。！？、：；""‘’（）()\-–—~!?,.:; ]+', '', t)
|
||
|
||
def _common_prefix_len(x, y):
|
||
n = 0
|
||
for i in range(min(len(x), len(y))):
|
||
if x[i] == y[i]:
|
||
n = i + 1
|
||
else:
|
||
break
|
||
return n
|
||
|
||
def _similar(a, b):
|
||
"""两段文本高度相似：前缀关系 / 子序列关系 / 去标点后共享长前缀 / 相似度>=0.92
|
||
处理多种OCR差异：渐进蹦字、标点不一致(······vs......)、中间漏字、交叉缺失闪烁"""
|
||
if not a or not b:
|
||
return False
|
||
if b.startswith(a) or a.startswith(b):
|
||
return True
|
||
# 后缀关系: 一方=另一方【去掉开头几个字】(OCR漏读开头, 如 一刀两刀→刀两刀) → 同一句。
|
||
# 限制: 短串≥3字, 只差≤3个开头字, 避免误合不同短句。
|
||
_s, _l = (a, b) if len(a) <= len(b) else (b, a)
|
||
if len(_s) >= 3 and (len(_l) - len(_s)) <= 3 and _l.endswith(_s):
|
||
return True
|
||
# 标点归一化后再判断前缀关系（处理 ······ vs ...... 等OCR差异）
|
||
an, bn = _norm_punct(a), _norm_punct(b)
|
||
if bn.startswith(an) or an.startswith(bn):
|
||
return True
|
||
# 子序列关系：较短者是较长者的子序列 → OCR漏字的同一句（要求够长，避免短文本误判）
|
||
short, long = (an, bn) if len(an) <= len(bn) else (bn, an)
|
||
if len(short) >= 10 and _is_subseq(short, long):
|
||
return True
|
||
# 去掉所有标点后比对：处理"同一句的有损版、但损的部位不同"的OCR闪烁
|
||
ca, cb = _strip_all_punct(a), _strip_all_punct(b)
|
||
if ca and cb:
|
||
if ca == cb or ca.startswith(cb) or cb.startswith(ca):
|
||
return True
|
||
cs, cl = (ca, cb) if len(ca) <= len(cb) else (cb, ca)
|
||
# 较短者是较长者子序列
|
||
if len(cs) >= 10 and _is_subseq(cs, cl):
|
||
return True
|
||
# 共享足够长的公共前缀（开头稳定、中/尾各自丢失不同部分的闪烁特征）
|
||
# 绝对长度≥15字就是强信号（两条不同台词共享这么长的相同开头几乎不可能）；
|
||
# 较短时再要求占较短者较大比例，避免短文本误判。
|
||
cp = _common_prefix_len(ca, cb)
|
||
if cp >= 15 or (cp >= 12 and cp >= 0.6 * len(cs)):
|
||
return True
|
||
# 覆盖率：较短串的绝大部分字按块出现在较长串中 → 同一句的截断/分散错字变体
|
||
if len(cs) >= 12 and _containment(cs, cl) >= 0.9:
|
||
return True
|
||
# 截断开头：较短串与较长串的等长前缀高度相似（允许个别OCR错字，如 砚/观、钧/钩）
|
||
# → 短串是长串刚开始出现时的截断版。要求≥5字，保护"嗯""什么?"等短插话不被误合。
|
||
if 5 <= len(cs) < len(cl) and \
|
||
SequenceMatcher(None, cs, cl[:len(cs)]).ratio() >= 0.8:
|
||
return True
|
||
# 找公共前缀，去掉后各自剥离开头标点再判断前缀关系
|
||
prefix_len = _common_prefix_len(a, b)
|
||
if prefix_len >= 4:
|
||
# 剥离开头标点：含 ASCII句点. 中文标点 破折号 中点· 等
|
||
strip_re = r'^[-–—.…·•\s，。！？、：；]+'
|
||
a_rest = re.sub(strip_re, '', a[prefix_len:])
|
||
b_rest = re.sub(strip_re, '', b[prefix_len:])
|
||
if b_rest.startswith(a_rest) or a_rest.startswith(b_rest):
|
||
return True
|
||
if abs(len(an) - len(bn)) > 5:
|
||
return False
|
||
return SequenceMatcher(None, an, bn).ratio() >= 0.92
|
||
|
||
def _dedup_pass(subs):
|
||
out = []
|
||
for sub in subs:
|
||
if out and _similar(out[-1]['text'], sub['text']):
|
||
# 保留更长的文本，延伸结束时间
|
||
if len(sub['text']) >= len(out[-1]['text']):
|
||
out[-1]['text'] = sub['text']
|
||
out[-1]['end'] = sub['end']
|
||
else:
|
||
out.append(sub)
|
||
return out
|
||
|
||
# 反复去重直到稳定（定点迭代）。单遍不够：合并时"保留更长文本"会让 deduped[-1]
|
||
# 中途变长，某些段只有在它变长后才与之相似，但那时已被追加——下一遍才能合并。
|
||
prev_len = -1
|
||
while len(merged_subtitles) != prev_len:
|
||
prev_len = len(merged_subtitles)
|
||
merged_subtitles = _dedup_pass(merged_subtitles)
|
||
# 不再按"合并后字幕条数"卡最少数量——字幕少也放行, 只要真有内容(非空)就生成SRT。
|
||
if not merged_subtitles:
|
||
return False
|
||
# 生成SRT文件
|
||
with open(output_path, 'w', encoding='utf-8') as f:
|
||
for i, subtitle in enumerate(merged_subtitles):
|
||
# 格式化为SRT时间格式 (HH:MM:SS,mmm)
|
||
def format_time(seconds):
|
||
hours = int(seconds // 3600)
|
||
minutes = int((seconds % 3600) // 60)
|
||
seconds_remainder = seconds % 60
|
||
milliseconds = int((seconds_remainder - int(seconds_remainder)) * 1000)
|
||
return f"{hours:02d}:{minutes:02d}:{int(seconds_remainder):02d},{milliseconds:03d}"
|
||
|
||
srt_start = format_time(subtitle['start'])
|
||
srt_end = format_time(subtitle['end'])
|
||
|
||
# 写入SRT条目（折叠整句自我重复）
|
||
f.write(f"{i+1}\n")
|
||
f.write(f"{srt_start} --> {srt_end}\n")
|
||
f.write(f"{_collapse_repeat(subtitle['text'])}\n\n")
|
||
return True
|
||
|
||
|
||
import string
|
||
|
||
def check_subtitle_quality_enhanced(best_subtitle, strict_mode=True):
|
||
"""
|
||
增强版字幕质量检测
|
||
"""
|
||
if not best_subtitle or not best_subtitle.strip():
|
||
return True
|
||
|
||
subtitle = best_subtitle.strip()
|
||
total_chars = len(subtitle)
|
||
if total_chars == 0:
|
||
return True
|
||
|
||
digit_count = 0
|
||
symbol_count = 0
|
||
|
||
for char in subtitle:
|
||
if char.isdigit(): # 检测数字[2](@ref)
|
||
digit_count += 1
|
||
elif char in string.punctuation: # 检测标点符号
|
||
symbol_count += 1
|
||
|
||
total_special_chars = digit_count + symbol_count
|
||
ratio = total_special_chars / total_chars
|
||
english_ratio = calculate_english_ratio(subtitle)
|
||
# 严格模式：数字单独占比也不能过高
|
||
if strict_mode:
|
||
digit_ratio = digit_count / total_chars
|
||
if total_chars>=10 and digit_ratio<=0.9 and ratio<=0.9 and english_ratio<=0.6:
|
||
return True
|
||
# 数字占比超过30%或总特殊字符占比超过50%都视为不合格
|
||
flag = (digit_ratio <= 0.7) and (ratio <= 0.8)
|
||
else:
|
||
# 普通模式：只检查总特殊字符占比
|
||
flag = ratio <= 0.5
|
||
return flag
|
||
|
||
|
||
|
||
def _has_chinese(text):
|
||
"""文本是否含中文字符。"""
|
||
return any('一' <= ch <= '鿿' for ch in (text or ""))
|
||
|
||
|
||
def processsrt(root_path,jsonl_path, srt_path, interval_seconds,logger):
|
||
"""主处理流程"""
|
||
# 1. 加载数据
|
||
logger.start_program("getsrt")
|
||
_sub_cfg = config.get_subtitle_config()
|
||
require_chinese = _sub_cfg.get('require_chinese', True) # 只要中文字幕
|
||
redundant_ratio = float(_sub_cfg.get('redundant_merge_ratio', 0.85)) # 冗余阈值，越大越宽松
|
||
# metadata.json 和 composite 图片都在 composite 子目录中
|
||
composite_dir = os.path.join(root_path, "composite")
|
||
local_json_path = os.path.join(composite_dir, "metadata.json")
|
||
jsonl_path = os.path.join(composite_dir, jsonl_path)
|
||
srt_path = os.path.join(composite_dir, srt_path)
|
||
if not process_composite_ocr_to_jsonl(local_json_path, composite_dir, jsonl_path,logger):
|
||
logger.log(f"视频{root_path}将被过滤，原因:空白复合图超过一半",level="warning",program="getsrt")
|
||
with open(jsonl_path, "r", encoding='utf-8') as f:
|
||
contents = f.readlines()
|
||
return None,None,"空白复合图超过一半",contents,None,None,None
|
||
with open(jsonl_path, "r", encoding='utf-8') as f:
|
||
contents = f.readlines()
|
||
srt_path = os.path.abspath(srt_path) # 转换为绝对路径
|
||
all_pic_ocr_pair = []
|
||
idx = 0
|
||
while idx < len(contents):
|
||
try:
|
||
if idx % 2 == 0 and idx + 1 < len(contents):
|
||
pic_path = contents[idx].strip().split("pic_path: ")[-1]
|
||
ocr_json = json.loads(contents[idx+1])
|
||
all_pic_ocr_pair.append([pic_path, ocr_json])
|
||
idx += 2
|
||
else:
|
||
idx += 1
|
||
except (json.JSONDecodeError, KeyError, ValueError):
|
||
idx += 1
|
||
|
||
# 1.4 低置信占比过滤: 整段OCR里低置信检测占比过高 → 非字幕内容(幻灯片/代码/满屏文字), 整段丢弃。
|
||
# 口播字幕低置信率仅2-8%, 幻灯片/编程视频高达50%+ (OCR硬读点线目录/数字/代码出垃圾)。
|
||
# 这类视频还吃3-5倍OCR、拖垮吞吐, 直接判非字幕跳过。min_det 保证样本足够才判, 短视频不误杀。
|
||
if _sub_cfg.get('filter_lowconf', True):
|
||
_thr = int(_sub_cfg.get('lowconf_thr', 80))
|
||
_frac = float(_sub_cfg.get('lowconf_filter_frac', 0.35))
|
||
_min_det = int(_sub_cfg.get('lowconf_min_det', 200))
|
||
_confs = [d.get('Confidence', 0) for _p, _oj in all_pic_ocr_pair
|
||
for d in _oj.get('TextDetections', [])]
|
||
if len(_confs) >= _min_det:
|
||
_lowr = sum(1 for c in _confs if c < _thr) / len(_confs)
|
||
if _lowr > _frac:
|
||
logger.log(f"视频{root_path}将被过滤,原因:非字幕内容(低置信占比{_lowr*100:.0f}%>"
|
||
f"{_frac*100:.0f}%, 共{len(_confs)}检测)--多为幻灯片/代码/满屏文字",
|
||
level="warning", program="getsrt")
|
||
logger.end_program("getsrt")
|
||
return None, None, "非字幕内容(低置信过多)", contents, None, None, None
|
||
|
||
# 1.5 坐标水印过滤: 删掉"固定位置+文字恒定"的水印/台标框(在定位和提取之前)
|
||
if _sub_cfg.get('filter_watermark_pos', True):
|
||
_filter_watermark_by_position(
|
||
all_pic_ocr_pair, logger,
|
||
dom_frac=float(_sub_cfg.get('wm_dominant_frac', 0.5)),
|
||
min_count=int(_sub_cfg.get('wm_min_count', 20)))
|
||
|
||
# 2. 分析最佳字幕位置
|
||
best_grids = analyze_subtitle_blocks(all_pic_ocr_pair, local_json_path,logger)
|
||
if not best_grids["kept_first"] and not best_grids["composite"] and not best_grids["kept_last"]:
|
||
logger.log(f"视频{root_path}将被过滤，原因:未找到合适的字幕区域",level="warning",program="getsrt")
|
||
return None,None,"未找到合适的字幕区域",contents,best_grids,None,None
|
||
h = analyze_h(all_pic_ocr_pair, local_json_path)
|
||
if h["composite"]==0:
|
||
logger.log(f"视频{root_path}将被过滤，原因：冗余文字过多",level="warning",program="getsrt")
|
||
return None,None,"冗余文字过多",contents,best_grids,None,h["composite"]
|
||
# 3. 处理所有帧
|
||
results = []
|
||
recent_frames = []
|
||
tot_score = 0
|
||
tot_conf = 0
|
||
times = 0
|
||
merge_time = 0
|
||
last_text = ""
|
||
last_flag = False
|
||
blank=0
|
||
audio_frame = []
|
||
for idx in range(len(all_pic_ocr_pair)):
|
||
|
||
if os.environ.get("LOG_VERBOSE") == "1":
|
||
print(f"\r处理进度: {idx+1}/{len(all_pic_ocr_pair)}", end="")
|
||
subtitle_text, confidence, flag,similarity = extract_highest_subtitle(
|
||
idx,
|
||
all_pic_ocr_pair,
|
||
best_grids,
|
||
local_json_path,
|
||
h,
|
||
logger
|
||
)
|
||
if idx<9 or idx> len(all_pic_ocr_pair)-9:
|
||
if similarity!=0:
|
||
audio_frame.append(f"第{idx}帧找到拼音匹配的OCR文本: {subtitle_text} (相似度: {similarity:.2f})")
|
||
else:
|
||
audio_frame.append(f"第{idx}帧未找到拼音匹配的OCR文本")
|
||
if (not check_subtitle_quality_enhanced(subtitle_text)) and confidence<=90:
|
||
subtitle_text = ""
|
||
# 只要中文字幕：纯英文/数字的(无中文字符)丢弃
|
||
if require_chinese and subtitle_text and not _has_chinese(subtitle_text):
|
||
subtitle_text = ""
|
||
current_result = {
|
||
'frame_idx': idx,
|
||
'text': subtitle_text if subtitle_text else "",
|
||
'confidence': confidence if confidence else 75,
|
||
'is_merged': False
|
||
}
|
||
if not flag:
|
||
merge_time+=1
|
||
if (subtitle_text!=""):
|
||
times+=1
|
||
tot_score=tot_score+ evaluate_text_quality_rules(subtitle_text)
|
||
tot_conf=tot_conf + confidence
|
||
if (idx == 0):
|
||
if (subtitle_text==""):
|
||
current_result['confidence']=0
|
||
recent_frames.append(current_result)
|
||
results.append((current_result['frame_idx'],current_result['text']))
|
||
last_text = current_result['text']
|
||
last_confidence = current_result['confidence']
|
||
last_flag = flag
|
||
best_frame = current_result
|
||
continue
|
||
if len(recent_frames) > 0:
|
||
# 检查与最近帧的相似度
|
||
similarity = SequenceMatcher(None, subtitle_text, last_text).ratio()
|
||
if (similarity < 0.75 and last_text not in subtitle_text and subtitle_text not in last_text) or last_text=="" or (abs(len(last_text)-len(subtitle_text))>2 and confidence>=90 and last_confidence>=90 and last_flag and flag): # 相似度阈值，可调整
|
||
if last_confidence - confidence>15 and confidence<70:
|
||
current_result['text']=""
|
||
if (subtitle_text!=""):
|
||
last_text = subtitle_text
|
||
last_confidence = current_result['confidence']
|
||
last_flag = flag
|
||
blank=0
|
||
else:
|
||
current_result['confidence']=0
|
||
results.append((current_result['frame_idx'],current_result['text']))
|
||
recent_frames.append(current_result)
|
||
blank+=1
|
||
if blank >= 2:
|
||
recent_frames = []
|
||
recent_frames.append(current_result)
|
||
last_text = ""
|
||
last_confidence = 90
|
||
continue
|
||
if len(recent_frames) == 1:
|
||
if current_result['confidence']<75 and recent_frames[0]['confidence']<85:
|
||
idxx = recent_frames[0]['frame_idx']
|
||
results[idxx] = (idxx, "")
|
||
recent_frames = []
|
||
recent_frames.append(current_result)
|
||
if current_result['confidence']>=65:
|
||
results.append((current_result['frame_idx'],current_result['text']))
|
||
else:
|
||
results.append((current_result['frame_idx'],""))
|
||
else:
|
||
if subtitle_text in last_text and subtitle_text!="" and subtitle_text!=last_text:
|
||
current_result['text'] = last_text
|
||
if last_text in subtitle_text and last_text!="" and current_result['confidence']>=90:
|
||
current_result['is_merged']=True
|
||
if (subtitle_text!=""):
|
||
last_text = subtitle_text
|
||
last_confidence = current_result['confidence']
|
||
last_flag = flag
|
||
blank=0
|
||
else:
|
||
current_result['confidence']=0
|
||
results.append((current_result['frame_idx'],current_result['text']))
|
||
recent_frames.append(current_result)
|
||
blank+=1
|
||
if blank >= 2:
|
||
recent_frames = []
|
||
recent_frames.append(current_result)
|
||
last_text=""
|
||
last_confidence = 90
|
||
continue
|
||
recent_frames.append(current_result)
|
||
if current_result['is_merged']:
|
||
best_frame = current_result
|
||
best_frame['confidence'] = max(recent_frames, key=lambda x: x['confidence'])['confidence']
|
||
else:
|
||
best_frame = max(reversed(recent_frames), key=lambda x: x['confidence'])
|
||
if best_frame['confidence']<65:
|
||
best_frame['text']=""
|
||
for frame in recent_frames:
|
||
idxx = frame['frame_idx']
|
||
if idxx!=idx:
|
||
results[idxx] = (idxx, best_frame['text'])
|
||
last_text = best_frame['text']
|
||
last_confidence = best_frame['confidence']
|
||
current_result['text'] = best_frame['text']
|
||
results.append((current_result['frame_idx'],current_result['text']))
|
||
|
||
# 除零保护: 没有任何有效字幕帧(times==0)时直接判"字幕太少"(无字幕),
|
||
# 否则 tot_score/times 会抛 ZeroDivisionError, 被上层包成 RuntimeError 记为 error
|
||
# (线上这一类崩溃曾占全部 error 的 ~20%, 本质是无字幕, 应走 no_subtitle)。
|
||
if times == 0:
|
||
logger.log(f"\n视频{root_path}将被过滤,原因：字幕太少(无有效字幕帧)",
|
||
level="warning", program="getsrt")
|
||
logger.end_program("getsrt")
|
||
return None, None, "字幕太少", contents, best_grids, audio_frame, h["composite"]
|
||
score = tot_score/times
|
||
conf = tot_conf/times
|
||
logger.log(f"{times}个有效字幕综合得分：{score}",program="getsrt")
|
||
logger.log(f"{times}个有效字幕综合信任度：{conf/100}",program="getsrt")
|
||
if (merge_time > redundant_ratio*len(all_pic_ocr_pair)):
|
||
logger.log(f"\n视频{root_path}将被过滤，原因：冗余文字过多/水印占比过大",level="warning",program="getsrt")
|
||
logger.end_program("getsrt")
|
||
return None,None,"冗余文字太多",contents,best_grids,audio_frame,h["composite"]
|
||
# 字幕帧占比(times/总帧数)不再单独卡阈值——字幕数少也放行, 是否真的"没内容"交给
|
||
# 后面 generate_srt_from_results() 的实际产出结果判断(它生成不出SRT才归"字幕太少")。
|
||
if (conf<80):
|
||
logger.log(f"\n视频{root_path}将被过滤，原因：confidence太低",level="warning",program="getsrt")
|
||
logger.end_program("getsrt")
|
||
return None,conf,"可能是特殊字体，置信度太低",contents,best_grids,audio_frame,h["composite"]
|
||
# 3.5 静态文字行过滤(剔除水印/台标/固定标题；过滤后若没剩字幕→走"字幕太少"判无字幕)
|
||
if _sub_cfg.get('filter_static', True):
|
||
results = _filter_static_text(
|
||
results, logger,
|
||
min_frac=float(_sub_cfg.get('static_frac', 0.25)),
|
||
strip_prefix=_sub_cfg.get('strip_static_prefix', False))
|
||
|
||
# 4. 生成SRT文件
|
||
os.makedirs(os.path.dirname(srt_path), exist_ok=True)
|
||
if not generate_srt_from_results(results, srt_path,interval_seconds=interval_seconds):
|
||
logger.log(f"\n视频{root_path}将被过滤,原因：字幕太少",level="warning",program="getsrt")
|
||
logger.end_program("getsrt")
|
||
return None,None,"字幕太少",contents,best_grids,audio_frame,h["composite"]
|
||
#generate_json_from_results(results, srt_path,interval_seconds=interval_seconds)
|
||
logger.log(f"\n处理完成！字幕已保存到: {srt_path}",program="getsrt")
|
||
logger.end_program("getsrt")
|
||
return score,conf,srt_path,contents,best_grids,audio_frame,h["composite"]
|
||
if __name__ == "__main__":
|
||
processsrt("./split/6833783390639033604/composite","ocr_result.jsonl", "test.srt", 0.33) |