Update obsidian-kb skill and add find-skills

- obsidian-kb: Add OCR processors (easyocr, tesseract)
- obsidian-kb: Fix subtitle detection with test files
- obsidian-kb: Add cookie verification script
- Add find-skills skill for discovering capabilities
- Add video processing scripts
- Update clawhub lock file
This commit is contained in:
OpenClaw Agent
2026-02-09 11:53:42 +08:00
parent d66eade77e
commit e0a2b323d9
33 changed files with 3555 additions and 36 deletions
@@ -9,11 +9,14 @@ B站视频内容提取器
import asyncio
import json
import logging
import re
from datetime import datetime
from typing import Dict, Any, Optional
from urllib.parse import urlparse
logger = logging.getLogger(__name__)
from playwright.async_api import async_playwright
from bs4 import BeautifulSoup
@@ -0,0 +1,355 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
OCR 文字识别处理器(EasyOCR 版本)
使用 EasyOCR 实现视频帧文字识别功能,作为 PaddleOCR 的替代方案
主要优势:不依赖 PaddlePaddle,避免 oneDNN 兼容性问题
"""
import re
import numpy as np
from typing import List, Dict, Any, Tuple
import logging
logger = logging.getLogger(__name__)
class EasyOCRProcessor:
"""
EasyOCR 文字识别处理器
使用 EasyOCR 从视频帧中识别文字,支持批量识别和结果过滤
相比 PaddleOCR 的优势:
- 不依赖 PaddlePaddle,避免 oneDNN 兼容性问题
- 安装更简单,依赖更少
- 支持多种语言(70+ 种)
"""
# 相似度阈值:用于判断两个文本是否相似(基于编辑距离)
SIMILARITY_THRESHOLD = 0.85
# 置信度阈值
CONFIDENCE_THRESHOLD = 0.7
# 最小字幕长度(字符数)
MIN_SUBTITLE_LENGTH = 2
def __init__(self, lang: str = 'ch_sim', use_gpu: bool = False, detail: int = 1):
"""
初始化 OCR 处理器
Args:
lang: 语言支持,'ch_sim' 为简体中文,'ch_tra' 为繁体中文
use_gpu: 是否使用 GPU 加速
detail: 识别结果详细程度(0=只返回文本,1=返回文本+坐标+置信度)
"""
self.lang = lang
self.use_gpu = use_gpu
self.detail = detail
# 初始化 EasyOCR 实例
self._ocr = None
self._initialize_ocr()
def _initialize_ocr(self):
"""初始化 EasyOCR 实例"""
try:
from easyocr import Reader
self._ocr = Reader(
[self.lang],
gpu=self.use_gpu,
verbose=False # 不显示详细日志
)
logger.info("EasyOCR 初始化成功")
except Exception as e:
logger.error(f"EasyOCR 初始化失败: {e}")
raise RuntimeError(f"无法初始化 EasyOCR: {e}")
def _initialize_ocr(self):
"""初始化 EasyOCR 实例"""
try:
from easyocr import Reader
self._ocr = Reader(
[self.lang],
gpu=self.use_gpu,
verbose=False # 不显示详细日志
)
logger.info("EasyOCR 初始化成功")
except Exception as e:
logger.error(f"EasyOCR 初始化失败: {e}")
raise RuntimeError(f"无法初始化 EasyOCR: {e}")
def recognize_frames(self, frames: List[Tuple[np.ndarray, float]]) -> List[Dict[str, Any]]:
"""
批量识别帧中的文字
Args:
frames: 帧列表,每个元素为 (图像数组, 时间戳) 元组
- 图像数组:numpy.ndarray,形状为 (H, W, 3) 的 RGB 图像
- 时间戳:float,单位为秒
Returns:
List[Dict[str, Any]]: 识别结果列表,每个结果包含:
- time: float, 时间戳(秒)
- text: str, 识别的文本
- confidence: float, 置信度 (0-1)
"""
if not frames:
logger.warning("输入帧列表为空")
return []
results = []
for image, timestamp in frames:
try:
# 调用 EasyOCR 进行识别
# EasyOCR 返回格式: [[[x1, y1], [x2, y2], [x3, y3], [x4, y4]], (text, confidence, ...)]
# 根据 detail 参数,返回格式不同:
# detail=0: 返回 [text1, text2, ...]
# detail=1: 返回 [[bbox, (text, confidence)], ...]
ocr_result = self._ocr.readtext(image, detail=self.detail)
if ocr_result:
if self.detail == 0:
# 简单模式:只返回文本列表
for i, text in enumerate(ocr_result):
if text:
results.append({
'time': timestamp,
'text': str(text),
'confidence': 1.0 # 简单模式下没有置信度
})
else:
# 详细模式:返回边界框+文本+置信度
for item in ocr_result:
if len(item) >= 2:
bbox = item[0]
text_info = item[1]
if len(text_info) >= 2:
text, confidence = text_info[0], text_info[1]
# 跳过空文本
if not text:
continue
# 处理置信度:可能是数字或特殊字符
try:
if isinstance(confidence, str):
# 如果是 '?' 或其他特殊字符,使用默认置信度
conf_float = 0.5
else:
conf_float = float(confidence)
except (ValueError, TypeError):
conf_float = 0.5
results.append({
'time': timestamp,
'text': text,
'confidence': conf_float
})
except Exception as e:
logger.warning(f"帧识别失败 (时间戳 {timestamp}s): {e}")
continue
logger.info(f"原始识别结果数量: {len(results)}")
# 过滤和清洗结果
filtered_results = self._filter_and_clean(results)
logger.info(f"过滤后结果数量: {len(filtered_results)}")
return filtered_results
def _filter_and_clean(self, results: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""
过滤和清洗识别结果
Args:
results: 原始识别结果列表
Returns:
List[Dict[str, Any]]: 过滤后的结果列表
"""
if not results:
return []
# 步骤 1: 按置信度过滤
filtered = self._filter_by_confidence(results)
logger.debug(f"置信度过滤后: {len(filtered)}")
# 步骤 2: 过滤噪声(纯符号、单字符等)
filtered = self._filter_noise(filtered)
logger.debug(f"噪声过滤后: {len(filtered)}")
# 步骤 3: 去除重复或相似的文本
filtered = self._remove_duplicates(filtered)
logger.debug(f"去重后: {len(filtered)}")
# 步骤 4: 按时间戳排序
filtered.sort(key=lambda x: x['time'])
return filtered
def _filter_by_confidence(self, results: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""
按置信度过滤结果
Args:
results: 识别结果列表
Returns:
List[Dict[str, Any]]: 高置信度结果
"""
return [r for r in results if r['confidence'] >= self.CONFIDENCE_THRESHOLD]
def _filter_noise(self, results: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""
过滤噪声结果
过滤条件:
- 文本长度小于最小长度
- 纯符号或空白字符
Args:
results: 识别结果列表
Returns:
List[Dict[str, Any]]: 过滤后的结果
"""
filtered = []
for result in results:
text = result['text'].strip()
# 过滤过短的文本
if len(text) < self.MIN_SUBTITLE_LENGTH:
continue
# 过滤纯符号或空白字符
if self._is_noise(text):
continue
filtered.append(result)
return filtered
def _is_noise(self, text: str) -> bool:
"""
判断文本是否为噪声
Args:
text: 要判断的文本
Returns:
bool: 如果是噪声返回 True
"""
# 使用正则判断是否只包含符号和空白
try:
# 移除标点符号后检查是否为空
# 先替换常见的中文标点
cleaned = re.sub(r'[^\w\u4e00-\u9fff]', '', text)
return len(cleaned) == 0
except:
# 如果正则出错,保守处理,不算噪声
return False
def _remove_duplicates(self, results: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""
去除重复或相似的文本
基于编辑距离(Levenshtein distance)判断文本相似度
Args:
results: 识别结果列表(假设已按时间戳排序)
Returns:
List[Dict[str, Any]]: 去重后的结果
"""
if not results:
return []
deduped = [results[0]]
for current in results[1:]:
last = deduped[-1]
# 计算相似度
similarity = self._calculate_similarity(current['text'], last['text'])
# 如果相似度低于阈值,则保留
if similarity < self.SIMILARITY_THRESHOLD:
deduped.append(current)
else:
# 相似文本:保留置信度更高的那个
if current['confidence'] > last['confidence']:
deduped[-1] = current
return deduped
def _calculate_similarity(self, text1: str, text2: str) -> float:
"""
计算两个文本的相似度(基于归一化编辑距离)
Args:
text1: 文本1
text2: 文本2
Returns:
float: 相似度 (0-1),1 表示完全相同
"""
if text1 == text2:
return 1.0
len1, len2 = len(text1), len(text2)
if len1 == 0 or len2 == 0:
return 0.0
# 动态规划计算编辑距离
dp = np.zeros((len1 + 1, len2 + 1))
for i in range(len1 + 1):
dp[i, 0] = i
for j in range(len2 + 1):
dp[0, j] = j
for i in range(1, len1 + 1):
for j in range(1, len2 + 1):
if text1[i-1] == text2[j-1]:
dp[i, j] = dp[i-1, j-1]
else:
dp[i, j] = min(
dp[i-1, j] + 1, # 删除
dp[i, j-1] + 1, # 插入
dp[i-1, j-1] + 1 # 替换
)
edit_distance = dp[len1, len2]
max_len = max(len1, len2)
# 归一化相似度:1 - (编辑距离 / 最大长度)
similarity = 1.0 - (edit_distance / max_len)
return similarity
def get_ocr_info(self) -> Dict[str, Any]:
"""
获取 OCR 处理器信息
Returns:
Dict[str, Any]: OCR 配置信息
"""
return {
'engine': 'easyocr',
'lang': self.lang,
'use_gpu': self.use_gpu,
'detail': self.detail,
'confidence_threshold': self.CONFIDENCE_THRESHOLD,
'similarity_threshold': self.SIMILARITY_THRESHOLD,
'min_subtitle_length': self.MIN_SUBTITLE_LENGTH
}
@@ -0,0 +1,309 @@
"""
OCR 文字识别处理器
使用 PaddleOCR 实现视频帧文字识别功能,主要用于提取硬字幕
"""
import os
import re
import numpy as np
from typing import List, Dict, Any, Tuple
from paddleocr import PaddleOCR
import logging
logger = logging.getLogger(__name__)
class OCRProcessor:
"""
OCR 文字识别处理器
使用 PaddleOCR 从视频帧中识别文字,支持批量识别和结果过滤
"""
# 相似度阈值:用于判断两个文本是否相似(基于编辑距离)
SIMILARITY_THRESHOLD = 0.85
# 置信度阈值
CONFIDENCE_THRESHOLD = 0.7
# 最小字幕长度(字符数)
MIN_SUBTITLE_LENGTH = 2
# 注意:Python re 模块不支持 \p{P} 等属性类,在 _is_noise 方法中使用自定义逻辑
def __init__(self, use_textline_orientation: bool = True, lang: str = 'ch',
use_gpu: bool = False):
"""
初始化 OCR 处理器
Args:
use_textline_orientation: 是否使用文本行方向分类器(PaddleOCR 3.x 参数名)
lang: 语言支持,'ch' 为中英文混合
use_gpu: 是否使用 GPU 加速(通过环境变量设置,如需启用请先设置 CUDA_VISIBLE_DEVICES)
"""
self.use_textline_orientation = use_textline_orientation
self.lang = lang
self.use_gpu = use_gpu
# 初始化 PaddleOCR 实例
self._ocr = None
self._initialize_ocr()
def _initialize_ocr(self):
"""初始化 PaddleOCR 实例"""
try:
# 强制禁用 oneDNN 和 MKL
os.environ['FLAGS_use_mkldnn'] = '0'
os.environ['MKL_SERVICE_FORCE_INTEL'] = '1'
os.environ['OMP_NUM_THREADS'] = '1'
os.environ['PADDLE_ON_INFER'] = '1'
# PaddleOCR 3.x 版本参数
self._ocr = PaddleOCR(
use_textline_orientation=self.use_textline_orientation,
lang=self.lang
)
logger.info("PaddleOCR 初始化成功(oneDNN 已禁用)")
except Exception as e:
logger.error(f"PaddleOCR 初始化失败: {e}")
raise RuntimeError(f"无法初始化 PaddleOCR: {e}")
def recognize_frames(self, frames: List[Tuple[np.ndarray, float]]) -> List[Dict[str, Any]]:
"""
批量识别帧中的文字
Args:
frames: 帧列表,每个元素为 (图像数组, 时间戳) 元组
- 图像数组:numpy.ndarray,形状为 (H, W, 3) 的 RGB 图像
- 时间戳:float,单位为秒
Returns:
List[Dict[str, Any]]: 识别结果列表,每个结果包含:
- time: float, 时间戳(秒)
- text: str, 识别的文本
- confidence: float, 置信度 (0-1)
"""
if not frames:
logger.warning("输入帧列表为空")
return []
results = []
for image, timestamp in frames:
try:
# 调用 PaddleOCR 进行识别
# 注意:PaddleOCR 3.x 版本的 ocr() 方法参数已简化
ocr_result = self._ocr.ocr(image)
# PaddleOCR 返回格式: [[[box], (text, confidence)], ...]
if ocr_result and ocr_result[0]:
for line in ocr_result[0]:
box, (text, confidence) = line
if text and confidence:
results.append({
'time': timestamp,
'text': text,
'confidence': float(confidence)
})
except Exception as e:
logger.warning(f"帧识别失败 (时间戳 {timestamp}s): {e}")
continue
logger.info(f"原始识别结果数量: {len(results)}")
# 过滤和清洗结果
filtered_results = self._filter_and_clean(results)
logger.info(f"过滤后结果数量: {len(filtered_results)}")
return filtered_results
def _filter_and_clean(self, results: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""
过滤和清洗识别结果
Args:
results: 原始识别结果列表
Returns:
List[Dict[str, Any]]: 过滤后的结果列表
"""
if not results:
return []
# 步骤 1: 按置信度过滤
filtered = self._filter_by_confidence(results)
logger.debug(f"置信度过滤后: {len(filtered)}")
# 步骤 2: 过滤噪声(纯符号、单字符等)
filtered = self._filter_noise(filtered)
logger.debug(f"噪声过滤后: {len(filtered)}")
# 步骤 3: 去除重复或相似的文本
filtered = self._remove_duplicates(filtered)
logger.debug(f"去重后: {len(filtered)}")
# 步骤 4: 按时间戳排序
filtered.sort(key=lambda x: x['time'])
return filtered
def _filter_by_confidence(self, results: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""
按置信度过滤结果
Args:
results: 识别结果列表
Returns:
List[Dict[str, Any]]: 高置信度结果
"""
return [r for r in results if r['confidence'] >= self.CONFIDENCE_THRESHOLD]
def _filter_noise(self, results: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""
过滤噪声结果
过滤条件:
- 文本长度小于最小长度
- 纯符号或空白字符
Args:
results: 识别结果列表
Returns:
List[Dict[str, Any]]: 过滤后的结果
"""
filtered = []
for result in results:
text = result['text'].strip()
# 过滤过短的文本
if len(text) < self.MIN_SUBTITLE_LENGTH:
continue
# 过滤纯符号或空白字符
if self._is_noise(text):
continue
filtered.append(result)
return filtered
def _is_noise(self, text: str) -> bool:
"""
判断文本是否为噪声
Args:
text: 要判断的文本
Returns:
bool: 如果是噪声返回 True
"""
# 使用正则判断是否只包含符号和空白
# 注意:这里用简单的正则,更复杂的判断可以后续优化
try:
# 移除标点符号后检查是否为空
# 先替换常见的中文标点
cleaned = re.sub(r'[^\w\u4e00-\u9fff]', '', text)
return len(cleaned) == 0
except:
# 如果正则出错,保守处理,不算噪声
return False
def _remove_duplicates(self, results: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""
去除重复或相似的文本
基于编辑距离(Levenshtein distance)判断文本相似度
Args:
results: 识别结果列表(假设已按时间戳排序)
Returns:
List[Dict[str, Any]]: 去重后的结果
"""
if not results:
return []
deduped = [results[0]]
for current in results[1:]:
last = deduped[-1]
# 计算相似度
similarity = self._calculate_similarity(current['text'], last['text'])
# 如果相似度低于阈值,则保留
if similarity < self.SIMILARITY_THRESHOLD:
deduped.append(current)
else:
# 相似文本:保留置信度更高的那个
if current['confidence'] > last['confidence']:
deduped[-1] = current
return deduped
def _calculate_similarity(self, text1: str, text2: str) -> float:
"""
计算两个文本的相似度(基于归一化编辑距离)
Args:
text1: 文本1
text2: 文本2
Returns:
float: 相似度 (0-1),1 表示完全相同
"""
if text1 == text2:
return 1.0
len1, len2 = len(text1), len(text2)
if len1 == 0 or len2 == 0:
return 0.0
# 动态规划计算编辑距离
dp = np.zeros((len1 + 1, len2 + 1))
for i in range(len1 + 1):
dp[i, 0] = i
for j in range(len2 + 1):
dp[0, j] = j
for i in range(1, len1 + 1):
for j in range(1, len2 + 1):
if text1[i-1] == text2[j-1]:
dp[i, j] = dp[i-1, j-1]
else:
dp[i, j] = min(
dp[i-1, j] + 1, # 删除
dp[i, j-1] + 1, # 插入
dp[i-1, j-1] + 1 # 替换
)
edit_distance = dp[len1, len2]
max_len = max(len1, len2)
# 归一化相似度:1 - (编辑距离 / 最大长度)
similarity = 1.0 - (edit_distance / max_len)
return similarity
def get_ocr_info(self) -> Dict[str, Any]:
"""
获取 OCR 处理器信息
Returns:
Dict[str, Any]: OCR 配置信息
"""
return {
'use_textline_orientation': self.use_textline_orientation,
'lang': self.lang,
'use_gpu': self.use_gpu,
'confidence_threshold': self.CONFIDENCE_THRESHOLD,
'similarity_threshold': self.SIMILARITY_THRESHOLD,
'min_subtitle_length': self.MIN_SUBTITLE_LENGTH
}
@@ -35,6 +35,10 @@ class DrissionBrowser:
from DrissionPage import ChromiumOptions, ChromiumPage
options = ChromiumOptions()
# 设置浏览器路径(服务器环境需要)
options.set_browser_path('/usr/bin/chromium-browser')
if self.headless:
options.headless()
options.set_user_agent('Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36')
@@ -116,10 +120,17 @@ class ZhihuDrissionExtractor:
self.browser = DrissionBrowser(headless=self.headless, timeout=self.timeout)
self.markdown_converter = MarkdownConverter()
self.cookie_file = os.path.join(
os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))),
'data', 'cookies', 'zhihu.json'
)
# 从配置中读取 Cookie 文件路径,支持环境变量展开
# 直接读取 cookie_file(因为 self.config 已经是 zhihu 配置)
cookie_file = self.config.get('cookie_file')
if cookie_file:
self.cookie_file = os.path.expanduser(cookie_file)
else:
# 默认路径
self.cookie_file = os.path.join(
os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))),
'data', 'cookies', 'zhihu.json'
)
def identify_url_type(self, url: str) -> Optional[str]:
"""
@@ -180,9 +191,11 @@ class ZhihuDrissionExtractor:
cookies = []
# 1. 尝试从配置加载(优先级最高)
if self.config.get('zhihu.cookie_str'):
# 支持扁平和嵌套配置结构
cookie_str = self.config.get('cookie_str')
if cookie_str:
try:
cookies = json.loads(self.config.get('zhihu.cookie_str'))
cookies = json.loads(cookie_str)
print(f"从配置加载 Cookie,共 {len(cookies)} 个")
except json.JSONDecodeError as e:
print(f"配置中的 Cookie JSON 格式错误: {e}")
@@ -197,7 +210,16 @@ class ZhihuDrissionExtractor:
# 3. 尝试从文件加载(优先级最低)
else:
cookie_file = self.config.get('zhihu.cookie_file', self.cookie_file)
# 直接从配置中读取 cookie_file(因为 self.config 已经是 zhihu 配置)
cookie_file = self.config.get('cookie_file')
if not cookie_file:
# 如果没有配置,使用默认路径
cookie_file = self.cookie_file
else:
# 展开环境变量(如 ~)
cookie_file = os.path.expanduser(cookie_file)
if cookie_file and Path(cookie_file).exists():
try:
with open(cookie_file, 'r', encoding='utf-8') as f:
@@ -206,7 +228,8 @@ class ZhihuDrissionExtractor:
except (json.JSONDecodeError, FileNotFoundError, IOError) as e:
print(f"从文件加载 Cookie 失败: {e}")
else:
print("未找到 Cookie 配置,将使用未登录状态访问知乎")
print(f"未找到 Cookie 配置,将使用未登录状态访问知乎")
print(f" 期望路径: {cookie_file}")
return cookies