技能备份 - 2026-04-15 (40个技能)
This commit is contained in:
root
@@ -0,0 +1 @@
|
||||
# Engine module
|
||||
@@ -0,0 +1 @@
|
||||
# MacVision OCR module
|
||||
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
@@ -0,0 +1,153 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
MacVisionOCR - macOS Vision Text Recognition via Swift script
|
||||
|
||||
Uses Apple's Vision Framework via Swift script (confirmed working).
|
||||
|
||||
Requirements:
|
||||
- macOS 10.15+ (Catalina)
|
||||
- Swift (built-in on macOS)
|
||||
"""
|
||||
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from typing import Dict, List
|
||||
|
||||
# Path to Swift script
|
||||
SWIFT_SCRIPT_PATH = Path(__file__).parent / "macvision_swift.swift"
|
||||
|
||||
|
||||
class MacVisionOCR:
|
||||
"""macOS Vision OCR wrapper using Swift script"""
|
||||
|
||||
def __init__(self, verbose: bool = False):
|
||||
"""Initialize MacVisionOCR processor."""
|
||||
self.verbose = verbose
|
||||
|
||||
if sys.platform != 'darwin':
|
||||
raise RuntimeError("MacVisionOCR only works on macOS")
|
||||
|
||||
if self.verbose:
|
||||
print("[INFO] MacVisionOCR initialized (via Swift)")
|
||||
|
||||
def _estimate_confidence(self, text: str) -> float:
|
||||
"""Estimate OCR confidence from text quality."""
|
||||
if not text:
|
||||
return 0.0
|
||||
|
||||
printable = sum(1 for c in text if c.isprintable() or c in '\n\r\t')
|
||||
length_factor = min(1.0, len(text) / 100)
|
||||
quality = printable / max(len(text), 1)
|
||||
|
||||
confidence = 0.85 + (0.15 * quality * length_factor)
|
||||
return round(min(confidence, 1.0), 4)
|
||||
|
||||
def extract(self, image_path: str) -> Dict:
|
||||
"""Extract text from image using Mac Vision OCR via Swift."""
|
||||
from time import time
|
||||
|
||||
if sys.platform != 'darwin':
|
||||
return {
|
||||
'text': '',
|
||||
'confidence': 0.0,
|
||||
'error': 'MacVisionOCR only works on macOS',
|
||||
'processing_time_ms': 0
|
||||
}
|
||||
|
||||
start_time = time()
|
||||
|
||||
try:
|
||||
# Run Swift script with image path as argument
|
||||
result = subprocess.run(
|
||||
['swift', str(SWIFT_SCRIPT_PATH), image_path],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=60
|
||||
)
|
||||
|
||||
if result.returncode != 0:
|
||||
return {
|
||||
'text': '',
|
||||
'confidence': 0.0,
|
||||
'error': result.stderr.strip(),
|
||||
'processing_time_ms': (time() - start_time) * 1000
|
||||
}
|
||||
|
||||
# Parse Chinese output
|
||||
lines = result.stdout.strip().split('\n')
|
||||
text_lines = []
|
||||
all_confidences = []
|
||||
|
||||
current_text = None
|
||||
for line in lines:
|
||||
if line.startswith('文本：'):
|
||||
current_text = line[3:].strip()
|
||||
elif line.startswith('置信度：') and current_text:
|
||||
try:
|
||||
conf_str = line[4:].strip()
|
||||
# Swift returns confidence as percentage (0-100)
|
||||
# Output format: "置信度：50.00%" -> need to divide by 100
|
||||
numeric_part = conf_str.rstrip('%').strip()
|
||||
confidence = float(numeric_part) / 100.0
|
||||
text_lines.append(current_text)
|
||||
all_confidences.append(confidence)
|
||||
current_text = None
|
||||
except (ValueError, IndexError):
|
||||
pass
|
||||
|
||||
full_text = '\n'.join(text_lines)
|
||||
processing_time = (time() - start_time) * 1000
|
||||
avg_confidence = sum(all_confidences) / len(all_confidences) if all_confidences else 0.0
|
||||
|
||||
return {
|
||||
'text': full_text.strip(),
|
||||
'confidence': round(avg_confidence, 4),
|
||||
'line_count': len(text_lines),
|
||||
'processing_time_ms': round(processing_time, 2)
|
||||
}
|
||||
|
||||
except subprocess.TimeoutExpired:
|
||||
return {
|
||||
'text': '',
|
||||
'confidence': 0.0,
|
||||
'error': 'MacVisionOCR timeout',
|
||||
'processing_time_ms': 60000
|
||||
}
|
||||
except Exception as e:
|
||||
processing_time = (time() - start_time) * 1000
|
||||
return {
|
||||
'text': '',
|
||||
'confidence': 0.0,
|
||||
'error': f"MacVision error: {str(e)}",
|
||||
'processing_time_ms': round(processing_time, 2)
|
||||
}
|
||||
|
||||
def batch_extract(self, image_paths: List[str]) -> List[Dict]:
|
||||
"""Process multiple images"""
|
||||
return [self.extract(path) for path in image_paths]
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
import argparse
|
||||
|
||||
if sys.platform != 'darwin':
|
||||
print("[ERROR] MacVisionOCR only works on macOS")
|
||||
sys.exit(1)
|
||||
|
||||
if not SWIFT_SCRIPT_PATH.exists():
|
||||
print(f"[ERROR] Swift script not found: {SWIFT_SCRIPT_PATH}")
|
||||
sys.exit(1)
|
||||
|
||||
parser = argparse.ArgumentParser(description='macOS Vision OCR via Swift')
|
||||
parser.add_argument('image', help='Image file to process')
|
||||
parser.add_argument('--verbose', '-v', action='store_true')
|
||||
args = parser.parse_args()
|
||||
|
||||
processor = MacVisionOCR(verbose=args.verbose)
|
||||
result = processor.extract(args.image)
|
||||
|
||||
print(f"\nText:\n{result['text']}")
|
||||
print(f"\nConfidence: {result['confidence']:.4f}")
|
||||
print(f"Lines: {result.get('line_count', 0)}")
|
||||
print(f"Time: {result.get('processing_time_ms', 0):.2f}ms")
|
||||
@@ -0,0 +1,70 @@
|
||||
import Foundation
|
||||
import Vision
|
||||
import AppKit
|
||||
|
||||
// Swift 5.9+ 可以用 CommandLine.arguments，兼容旧版本用下面的方式
|
||||
let args = ProcessInfo.processInfo.arguments
|
||||
|
||||
guard args.count >= 2 else {
|
||||
print("Usage: swift macvision_swift.swift <image_path>")
|
||||
exit(1)
|
||||
}
|
||||
|
||||
let imagePath = args[1]
|
||||
|
||||
// 1. 加载图片
|
||||
guard let image = NSImage(contentsOfFile: imagePath) else {
|
||||
print("❌ 无法加载图片: \(imagePath)")
|
||||
exit(1)
|
||||
}
|
||||
|
||||
// 2. 转换为 CGImage（Vision 需要）
|
||||
guard let tiffData = image.tiffRepresentation,
|
||||
let cgImage = NSBitmapImageRep(data: tiffData)?.cgImage else {
|
||||
print("❌ 无法转换图片")
|
||||
exit(1)
|
||||
}
|
||||
|
||||
// 3. 设置信号量等待异步完成
|
||||
let semaphore = DispatchSemaphore(value: 0)
|
||||
var results: [(text: String, confidence: Float)] = []
|
||||
|
||||
// 4. 创建 OCR 请求
|
||||
let request = VNRecognizeTextRequest { request, error in
|
||||
defer { semaphore.signal() } // 关键：完成后释放信号量
|
||||
|
||||
if let error = error {
|
||||
print("❌ 识别错误：\(error)")
|
||||
return
|
||||
}
|
||||
|
||||
guard let observations = request.results as? [VNRecognizedTextObservation] else {
|
||||
print("❌ 无结果")
|
||||
return
|
||||
}
|
||||
|
||||
for observation in observations {
|
||||
guard let candidate = observation.topCandidates(1).first else { continue }
|
||||
results.append((candidate.string, candidate.confidence))
|
||||
}
|
||||
}
|
||||
|
||||
// 5. 配置识别参数
|
||||
request.recognitionLanguages = ["zh-Hans", "zh-Hant", "en-US"] // 支持中文
|
||||
request.usesLanguageCorrection = true // 启用语言校正
|
||||
|
||||
// 6. 执行识别
|
||||
let handler = VNImageRequestHandler(cgImage: cgImage, options: [:])
|
||||
|
||||
do {
|
||||
try handler.perform([request])
|
||||
_ = semaphore.wait(timeout: .now() + 30) // 等待最多 30 秒
|
||||
|
||||
// 输出结果
|
||||
for (text, conf) in results {
|
||||
print("文本：\(text)")
|
||||
print("置信度：\(String(format: "%.2f", conf * 100))%")
|
||||
}
|
||||
} catch {
|
||||
print("❌ 执行错误：\(error)")
|
||||
}
|
||||
@@ -0,0 +1,220 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
PaddleOCR wrapper - High accuracy Chinese OCR engine
|
||||
|
||||
Compatible with PaddleOCR 3.4.0+ API (predict() method returns list)
|
||||
Based on Emma's testing: result[0][0][1][0] for text, result[0][0][1][1] for confidence
|
||||
"""
|
||||
|
||||
import time
|
||||
from pathlib import Path
|
||||
from typing import Dict, List, Optional
|
||||
|
||||
|
||||
class PaddleOCR:
|
||||
"""PaddleOCR processor for high-accuracy text extraction"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
verbose: bool = False,
|
||||
lang: str = 'ch',
|
||||
use_angle_cls: bool = True
|
||||
):
|
||||
"""Initialize PaddleOCR - lang and use_angle_cls parameters"""
|
||||
self.verbose = verbose
|
||||
self.lang = lang
|
||||
self.use_angle_cls = use_angle_cls
|
||||
self._init_ocr()
|
||||
|
||||
def _init_ocr(self) -> None:
|
||||
"""Initialize PaddleOCR model"""
|
||||
from paddleocr import PaddleOCR
|
||||
|
||||
self.ocr = PaddleOCR(lang=self.lang, use_angle_cls=self.use_angle_cls)
|
||||
|
||||
def extract(self, image_path: str) -> Dict:
|
||||
"""
|
||||
Extract text from image using PaddleOCR.
|
||||
|
||||
Args:
|
||||
image_path: Path to input image
|
||||
|
||||
Returns:
|
||||
Dict with text, confidence, results, processing time
|
||||
"""
|
||||
if self.verbose:
|
||||
print(f"[PaddleOCR] Processing: {image_path}")
|
||||
|
||||
start_time = time.time()
|
||||
|
||||
try:
|
||||
# Run OCR - using predict() for PaddleOCR 3.4.0+
|
||||
result = self.ocr.predict(image_path)
|
||||
|
||||
processing_time = time.time() - start_time
|
||||
|
||||
# Parse results - legacy format: [[box], [text, confidence]]
|
||||
parsed = self._parse_results(result)
|
||||
parsed['processing_time_ms'] = round(processing_time * 1000, 2)
|
||||
|
||||
return parsed
|
||||
|
||||
except Exception as e:
|
||||
processing_time = time.time() - start_time
|
||||
return {
|
||||
'text': '',
|
||||
'confidence': 0.0,
|
||||
'error': str(e),
|
||||
'results': [],
|
||||
'processing_time_ms': round(processing_time * 1000, 2)
|
||||
}
|
||||
|
||||
def _parse_results(self, results) -> Dict:
|
||||
"""
|
||||
Parse PaddleOCR 3.4.0+ output format.
|
||||
|
||||
PaddleOCR 3.4.0+ predict() returns dict with:
|
||||
- rec_texts: list of detected text strings
|
||||
- rec_scores: list of confidence scores
|
||||
- dt_polys: list of text polygon coordinates
|
||||
|
||||
Returns:
|
||||
Dict with text, confidence, results, line_count
|
||||
"""
|
||||
if not results or not results[0]:
|
||||
return {
|
||||
'text': '',
|
||||
'confidence': 0.0,
|
||||
'results': [],
|
||||
'error': 'No text detected'
|
||||
}
|
||||
|
||||
# New format: result[0] is a dict with 'rec_texts' and 'rec_scores'
|
||||
result_dict = results[0]
|
||||
|
||||
if isinstance(result_dict, dict):
|
||||
texts = result_dict.get('rec_texts', [])
|
||||
scores = result_dict.get('rec_scores', [])
|
||||
|
||||
if not texts:
|
||||
return {
|
||||
'text': '',
|
||||
'confidence': 0.0,
|
||||
'results': [],
|
||||
'error': 'No text detected'
|
||||
}
|
||||
|
||||
# Combine all text (PaddleOCR already splits by lines)
|
||||
full_text = '\n'.join(texts)
|
||||
|
||||
# Calculate average confidence
|
||||
avg_confidence = sum(scores) / len(scores) if scores else 0.0
|
||||
|
||||
# Build detailed results
|
||||
detailed_results = []
|
||||
for i, (text, score) in enumerate(zip(texts, scores)):
|
||||
# Get bbox if available (dt_polys is list of arrays)
|
||||
dt_polys = result_dict.get('dt_polys', [])
|
||||
bbox = None
|
||||
if i < len(dt_polys):
|
||||
try:
|
||||
poly = dt_polys[i]
|
||||
if hasattr(poly, 'tolist'):
|
||||
bbox = poly.tolist()
|
||||
else:
|
||||
bbox = poly
|
||||
except:
|
||||
bbox = None
|
||||
|
||||
detailed_results.append({
|
||||
'text': text,
|
||||
'confidence': round(score, 4),
|
||||
'bbox': bbox
|
||||
})
|
||||
|
||||
return {
|
||||
'text': full_text.strip(),
|
||||
'confidence': round(avg_confidence, 4),
|
||||
'results': detailed_results,
|
||||
'line_count': len(texts)
|
||||
}
|
||||
else:
|
||||
# Fallback: handle legacy format if still used
|
||||
return self._parse_results_legacy(results)
|
||||
|
||||
def _parse_results_legacy(self, results) -> Dict:
|
||||
"""
|
||||
Legacy parser for backward compatibility.
|
||||
Old format: [[bbox], [text, confidence]]
|
||||
"""
|
||||
if not results or not results[0]:
|
||||
return {
|
||||
'text': '',
|
||||
'confidence': 0.0,
|
||||
'results': [],
|
||||
'error': 'No text detected'
|
||||
}
|
||||
|
||||
text_lines = []
|
||||
all_confidences = []
|
||||
|
||||
for line in results:
|
||||
if len(line) < 2:
|
||||
continue
|
||||
|
||||
box = line[0]
|
||||
|
||||
if isinstance(line[1], (list, tuple)) and len(line[1]) >= 2:
|
||||
text, confidence = line[1][0], line[1][1]
|
||||
else:
|
||||
text, confidence = str(line[1]), 0.0
|
||||
|
||||
text_lines.append(text)
|
||||
all_confidences.append(confidence)
|
||||
|
||||
full_text = '\n'.join(text_lines)
|
||||
avg_confidence = sum(all_confidences) / len(all_confidences) if all_confidences else 0.0
|
||||
|
||||
detailed_results = []
|
||||
for line in results:
|
||||
if len(line) >= 2:
|
||||
box = line[0]
|
||||
if isinstance(line[1], (list, tuple)) and len(line[1]) >= 2:
|
||||
text, confidence = line[1][0], line[1][1]
|
||||
else:
|
||||
text, confidence = str(line[1]), 0.0
|
||||
|
||||
detailed_results.append({
|
||||
'text': text,
|
||||
'confidence': confidence,
|
||||
'bbox': box
|
||||
})
|
||||
|
||||
return {
|
||||
'text': full_text.strip(),
|
||||
'confidence': round(avg_confidence, 4),
|
||||
'results': detailed_results,
|
||||
'line_count': len(detailed_results)
|
||||
}
|
||||
|
||||
def batch_extract(self, image_paths: List[str]) -> List[Dict]:
|
||||
"""Process multiple images"""
|
||||
return [self.extract(path) for path in image_paths]
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
import argparse
|
||||
|
||||
parser = argparse.ArgumentParser(description='PaddleOCR wrapper')
|
||||
parser.add_argument('image', help='Image file to process')
|
||||
parser.add_argument('--lang', default='ch', help='Language code (ch, en, etc.)')
|
||||
parser.add_argument('--verbose', '-v', action='store_true')
|
||||
args = parser.parse_args()
|
||||
|
||||
processor = PaddleOCR(verbose=args.verbose, lang=args.lang)
|
||||
result = processor.extract(args.image)
|
||||
|
||||
print(f"\nText:\n{result['text']}")
|
||||
print(f"\nConfidence: {result['confidence']:.4f}")
|
||||
print(f"Lines detected: {result.get('line_count', 0)}")
|
||||
print(f"Time: {result.get('processing_time_ms', 0):.2f}ms")
|
||||
@@ -0,0 +1,183 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Engine selector - Intelligent OCR engine selection logic
|
||||
|
||||
Rules:
|
||||
1. Image contains Chinese characters → PaddleOCR (better Chinese support)
|
||||
2. Image is simple text, English only → Tesseract (faster, lighter)
|
||||
3. User explicitly requests engine → Use requested engine
|
||||
4. Auto mode, mixed/unknown → PaddleOCR (higher accuracy)
|
||||
|
||||
Multi-engine parallel mode:
|
||||
- Tesseract + PaddleOCR + MacVision (macOS only)
|
||||
- Agent择优 based on confidence, language support, and speed
|
||||
"""
|
||||
|
||||
import re
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from typing import Literal, List, Dict
|
||||
|
||||
|
||||
def detect_chinese(content: str) -> bool:
|
||||
"""Check if content contains Chinese characters"""
|
||||
# Chinese range: \u4e00-\u9fff
|
||||
return bool(re.search(r'[\u4e00-\u9fff]', content))
|
||||
|
||||
|
||||
def analyze_image_complexity(image_path: str) -> str:
|
||||
"""
|
||||
Analyze image complexity to determine optimal engine.
|
||||
|
||||
Returns:
|
||||
'simple' or 'complex'
|
||||
"""
|
||||
image_path_lower = Path(image_path).name.lower()
|
||||
|
||||
complex_patterns = [
|
||||
'menu', 'invoice', 'contract', 'certificate',
|
||||
'exam', 'form', 'table', 'receipt'
|
||||
]
|
||||
|
||||
if any(p in image_path_lower for p in complex_patterns):
|
||||
return 'complex'
|
||||
|
||||
return 'simple'
|
||||
|
||||
|
||||
def get_available_engines(image_path: str) -> List[str]:
|
||||
"""
|
||||
Get list of available engines for the current platform.
|
||||
|
||||
Args:
|
||||
image_path: Path to image (for platform-specific hints)
|
||||
|
||||
Returns:
|
||||
List of engine names: ['tesseract', 'paddle', 'macvision']
|
||||
"""
|
||||
engines = ['tesseract', 'paddle'] # Always available
|
||||
|
||||
# Add MacVision on macOS
|
||||
if sys.platform == 'darwin':
|
||||
engines.append('macvision')
|
||||
|
||||
return engines
|
||||
|
||||
|
||||
def select_engine(
|
||||
image_path: str,
|
||||
requested_engine: Literal['auto', 'tesseract', 'paddle', 'macvision'] = 'auto'
|
||||
) -> List[str]:
|
||||
"""
|
||||
Select engines for OCR processing.
|
||||
|
||||
In multi-engine mode, returns list of engines to run in parallel.
|
||||
|
||||
Args:
|
||||
image_path: Path to image being processed
|
||||
requested_engine: User request or 'auto'
|
||||
|
||||
Returns:
|
||||
List of engine names to use
|
||||
"""
|
||||
available = get_available_engines(image_path)
|
||||
|
||||
# Rule 1: User explicitly requested single engine
|
||||
if requested_engine != 'auto':
|
||||
if requested_engine in available:
|
||||
return [requested_engine]
|
||||
else:
|
||||
return available # Fallback to all available
|
||||
|
||||
# Rule 2: Check image path for hints
|
||||
path_lower = Path(image_path).name.lower()
|
||||
|
||||
# Simple screenshots → Tesseract (fastest)
|
||||
simple_indicators = ['screenshot', 'snap', 'capture', 'screen']
|
||||
if any(ind in path_lower for ind in simple_indicators):
|
||||
return ['tesseract']
|
||||
|
||||
# Complex documents → All engines (max accuracy)
|
||||
complex_indicators = ['menu', 'invoice', 'certificate', 'contract', 'receipt']
|
||||
if any(ind in path_lower for ind in complex_indicators):
|
||||
return available # Run all available engines
|
||||
|
||||
# Default: Run all available engines for best results
|
||||
return available
|
||||
|
||||
|
||||
def select_best_result(
|
||||
results: List[Dict],
|
||||
preferred_engine: str = 'paddle'
|
||||
) -> Dict:
|
||||
"""
|
||||
Select the best OCR result from multiple engines.
|
||||
|
||||
Args:
|
||||
results: List of OCR results from different engines
|
||||
preferred_engine: Preferred engine for tie-breaking
|
||||
|
||||
Returns:
|
||||
Dict with selected result and metadata
|
||||
"""
|
||||
if not results:
|
||||
return {
|
||||
'text': '',
|
||||
'confidence': 0.0,
|
||||
'error': 'No results',
|
||||
'selected_engine': None
|
||||
}
|
||||
|
||||
# Filter valid results (with text)
|
||||
valid_results = [r for r in results if r.get('text', '').strip()]
|
||||
|
||||
if not valid_results:
|
||||
return {
|
||||
'text': '',
|
||||
'confidence': 0.0,
|
||||
'error': 'All engines failed',
|
||||
'selected_engine': None
|
||||
}
|
||||
|
||||
# Calculate weighted score
|
||||
for r in valid_results:
|
||||
engine = r.get('engine', 'unknown')
|
||||
|
||||
# Engine quality weights
|
||||
quality_weights = {
|
||||
'paddle': 1.0,
|
||||
'macvision': 0.95,
|
||||
'tesseract': 0.9
|
||||
}
|
||||
|
||||
# Language support weights (optional, can be extended)
|
||||
language_weights = {
|
||||
'paddle': 1.0, # Best for Chinese
|
||||
'macvision': 0.85, # Good for English, fair for Chinese
|
||||
'tesseract': 0.8 # Good for English, fair for Chinese
|
||||
}
|
||||
|
||||
# Combined score
|
||||
base_confidence = r.get('confidence', 0)
|
||||
quality = quality_weights.get(engine, 0.8)
|
||||
language = language_weights.get(engine, 0.8)
|
||||
|
||||
r['_score'] = base_confidence * quality * language
|
||||
r['_quality_weight'] = quality
|
||||
r['_language_weight'] = language
|
||||
|
||||
# Sort by score
|
||||
sorted_results = sorted(valid_results, key=lambda x: x.get('_score', 0), reverse=True)
|
||||
|
||||
# Select best
|
||||
best = sorted_results[0]
|
||||
|
||||
return {
|
||||
'text': best.get('text', ''),
|
||||
'confidence': best.get('confidence', 0),
|
||||
'engine': best.get('engine', 'unknown'),
|
||||
'selected_engine': best.get('engine', 'unknown'),
|
||||
'processing_time_ms': sum(r.get('processing_time_ms', 0) for r in valid_results),
|
||||
'score': best.get('_score', 0),
|
||||
'other_results': sorted_results[1:] # Include runner-ups for verification
|
||||
}
|
||||
@@ -0,0 +1,269 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Tesseract OCR wrapper with optimized configuration for mixed Chinese/English content
|
||||
"""
|
||||
|
||||
import logging
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
from pathlib import Path
|
||||
from typing import Dict, List, Optional
|
||||
|
||||
try:
|
||||
import cv2
|
||||
import numpy as np
|
||||
from PIL import Image
|
||||
except ImportError:
|
||||
print("[ERROR] Install dependencies: pip install opencv-python numpy pillow")
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
logging.basicConfig(level=logging.INFO)
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class TesseractOCR:
|
||||
"""Tesseract OCR processor with preprocessing pipeline"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
verbose: bool = False,
|
||||
lang: str = 'chi_sim+eng',
|
||||
psm: int = 6,
|
||||
oem: int = 3
|
||||
):
|
||||
"""
|
||||
Initialize Tesseract processor.
|
||||
|
||||
Args:
|
||||
verbose: Enable detailed logging
|
||||
lang: Tesseract language code (e.g., 'eng', 'chi_sim', 'chi_sim+eng')
|
||||
psm: Page segmentation mode (default: 6, uniform block)
|
||||
oem: OCR engine mode (default: 3, LSTM only)
|
||||
"""
|
||||
self.verbose = verbose
|
||||
self.lang = lang
|
||||
self.psm = psm
|
||||
self.oem = oem
|
||||
|
||||
# Check tesseract availability
|
||||
self._check_tesseract()
|
||||
|
||||
def _check_tesseract(self) -> bool:
|
||||
"""Check if tesseract is installed"""
|
||||
try:
|
||||
result = subprocess.run(
|
||||
['tesseract', '--version'],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=5
|
||||
)
|
||||
if self.verbose:
|
||||
logger.info(f"Tesseract version: {result.stdout.split()[2]}")
|
||||
return True
|
||||
except (subprocess.TimeoutExpired, FileNotFoundError, IndexError):
|
||||
logger.error("Tesseract not found. Install with:")
|
||||
logger.error(" macOS: brew install tesseract")
|
||||
logger.error(" Ubuntu: sudo apt install tesseract-ocr")
|
||||
logger.error(" Windows: Download from https://github.com/UB-Mannheim/tesseract/wiki")
|
||||
return False
|
||||
|
||||
def _preprocess_image(self, image_path: str) -> str:
|
||||
"""
|
||||
Apply preprocessing pipeline.
|
||||
|
||||
Returns:
|
||||
Path to processed image
|
||||
"""
|
||||
if self.verbose:
|
||||
logger.info("Preprocessing image...")
|
||||
|
||||
start_time = time.time()
|
||||
|
||||
# Read image
|
||||
img = cv2.imread(image_path)
|
||||
if img is None:
|
||||
raise ValueError(f"Could not load image: {image_path}")
|
||||
|
||||
# Convert to grayscale
|
||||
gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)
|
||||
|
||||
# Apply bilateral filter (noise reduction + edge preservation)
|
||||
bilateral = cv2.bilateralFilter(gray, 9, 75, 75)
|
||||
|
||||
# Adaptive thresholding for low contrast images
|
||||
min_val, max_val, _, _ = cv2.minMaxLoc(bilateral)
|
||||
contrast = max_val - min_val
|
||||
|
||||
if contrast < 100: # Low contrast
|
||||
processed = cv2.adaptiveThreshold(
|
||||
bilateral, 255, cv2.ADAPTIVE_THRESH_GAUSSIAN_C,
|
||||
cv2.THRESH_BINARY, 11, 2
|
||||
)
|
||||
else:
|
||||
processed = bilateral
|
||||
|
||||
# Save processed image (temp file)
|
||||
input_path = Path(image_path)
|
||||
processed_path = input_path.parent / f"{input_path.stem}_tess_processed.png"
|
||||
cv2.imwrite(str(processed_path), processed)
|
||||
|
||||
if self.verbose:
|
||||
logger.info(f"Preprocessing completed in {time.time() - start_time:.2f}s")
|
||||
|
||||
return str(processed_path)
|
||||
|
||||
def _run_tesseract(
|
||||
self,
|
||||
image_path: str,
|
||||
lang: Optional[str] = None,
|
||||
psm: Optional[int] = None,
|
||||
oem: Optional[int] = None
|
||||
) -> Dict:
|
||||
"""
|
||||
Run Tesseract OCR.
|
||||
|
||||
Returns:
|
||||
Dict with text, confidence, timing
|
||||
"""
|
||||
if lang is None:
|
||||
lang = self.lang
|
||||
if psm is None:
|
||||
psm = self.psm
|
||||
if oem is None:
|
||||
oem = self.oem
|
||||
|
||||
# Run tesseract
|
||||
cmd = [
|
||||
'tesseract', image_path, 'stdout',
|
||||
'-l', lang,
|
||||
'--psm', str(psm),
|
||||
'--oem', str(oem),
|
||||
'_stdout'
|
||||
]
|
||||
|
||||
if self.verbose:
|
||||
logger.info(f"Running: {' '.join(cmd)}")
|
||||
|
||||
start_time = time.time()
|
||||
|
||||
try:
|
||||
result = subprocess.run(
|
||||
cmd,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=60
|
||||
)
|
||||
|
||||
processing_time = time.time() - start_time
|
||||
|
||||
if result.returncode != 0:
|
||||
logger.error(f"Tesseract error: {result.stderr}")
|
||||
return {
|
||||
'text': '',
|
||||
'confidence': 0.0,
|
||||
'error': result.stderr,
|
||||
'processing_time_ms': processing_time * 1000
|
||||
}
|
||||
|
||||
# Extract confidence if available
|
||||
text = result.stdout.strip()
|
||||
confidence = self._estimate_confidence(text)
|
||||
|
||||
return {
|
||||
'text': text,
|
||||
'confidence': confidence,
|
||||
'processing_time_ms': processing_time * 1000
|
||||
}
|
||||
|
||||
except subprocess.TimeoutExpired:
|
||||
return {
|
||||
'text': '',
|
||||
'confidence': 0.0,
|
||||
'error': 'Tesseract timeout',
|
||||
'processing_time_ms': 60000
|
||||
}
|
||||
|
||||
def _estimate_confidence(self, text: str) -> float:
|
||||
"""
|
||||
Estimate OCR confidence from output quality.
|
||||
|
||||
Simple heuristic: longer text with fewer garbage characters = higher confidence
|
||||
"""
|
||||
if not text:
|
||||
return 0.0
|
||||
|
||||
# Count printable characters
|
||||
printable = sum(1 for c in text if c.isprintable() or c in '\n\r\t')
|
||||
|
||||
# Length factor (more text = more reliable)
|
||||
length_factor = min(1.0, len(text) / 100)
|
||||
|
||||
# Quality factor
|
||||
quality = printable / max(len(text), 1)
|
||||
|
||||
# Combined score
|
||||
confidence = (0.6 * quality) + (0.4 * length_factor)
|
||||
|
||||
return round(confidence, 2)
|
||||
|
||||
def extract(self, image_path: str) -> Dict:
|
||||
"""
|
||||
Extract text from image using Tesseract.
|
||||
|
||||
Args:
|
||||
image_path: Path to input image
|
||||
|
||||
Returns:
|
||||
Dict with text, confidence, processing time
|
||||
"""
|
||||
if self.verbose:
|
||||
logger.info(f"Processing: {image_path}")
|
||||
|
||||
start_time = time.time()
|
||||
|
||||
# Preprocess
|
||||
processed_path = self._preprocess_image(image_path)
|
||||
|
||||
try:
|
||||
# Run OCR
|
||||
result = self._run_tesseract(
|
||||
processed_path,
|
||||
lang=self.lang,
|
||||
psm=self.psm,
|
||||
oem=self.oem
|
||||
)
|
||||
|
||||
# Clean up temp file
|
||||
Path(processed_path).unlink(missing_ok=True)
|
||||
|
||||
result['processing_time_ms'] = round(time.time() - start_time, 2)
|
||||
|
||||
return result
|
||||
|
||||
except Exception as e:
|
||||
Path(processed_path).unlink(missing_ok=True)
|
||||
return {
|
||||
'text': '',
|
||||
'confidence': 0.0,
|
||||
'error': str(e),
|
||||
'processing_time_ms': (time.time() - start_time) * 1000
|
||||
}
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
import argparse
|
||||
|
||||
parser = argparse.ArgumentParser(description='Tesseract OCR wrapper')
|
||||
parser.add_argument('image', help='Image file to process')
|
||||
parser.add_argument('--lang', default='chi_sim+eng', help='Language code')
|
||||
parser.add_argument('--verbose', '-v', action='store_true')
|
||||
args = parser.parse_args()
|
||||
|
||||
processor = TesseractOCR(verbose=args.verbose, lang=args.lang)
|
||||
result = processor.extract(args.image)
|
||||
|
||||
print(f"\nText:\n{result['text']}")
|
||||
print(f"\nConfidence: {result['confidence']:.2%}")
|
||||
print(f"Time: {result.get('processing_time_ms', 0):.2f}ms")
|
||||
Reference in New Issue
Block a user