import cv2
import numpy as np
import pytesseract
from PIL import Image
from dataclasses import dataclass
from typing import Optional, List

@dataclass
class OCRResult:
    text: str
    confidence: float
    bbox: tuple
    level: int

class OCRPipeline:
    """生产级 OCR 流水线"""
    
    def __init__(self, lang='chi_sim+eng', psm=6, min_confidence=50):
        self.lang = lang
        self.psm = psm
        self.min_confidence = min_confidence
    
    def preprocess(self, image: np.ndarray) -> np.ndarray:
        """图像预处理"""
        # 灰度化
        if len(image.shape) == 3:
            gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)
        else:
            gray = image
        
        # 尺寸检查
        h, w = gray.shape
        if h < 50:
            scale = 100 / h
            gray = cv2.resize(gray, None, fx=scale, fy=scale, 
                             interpolation=cv2.INTER_CUBIC)
        
        # 去噪
        denoised = cv2.fastNlMeansDenoising(gray, None, 10, 7, 21)
        
        # 对比度增强
        clahe = cv2.createCLAHE(clipLimit=2.0, tileGridSize=(8, 8))
        enhanced = clahe.apply(denoised)
        
        # 二值化
        _, binary = cv2.threshold(enhanced, 0, 255, 
                                   cv2.THRESH_BINARY + cv2.THRESH_OTSU)
        
        return binary
    
    def ocr(self, image: np.ndarray) -> List[OCRResult]:
        """OCR 识别"""
        processed = self.preprocess(image)
        pil_img = Image.fromarray(processed)
        
        config = f'--psm {self.psm} --oem 1'
        data = pytesseract.image_to_data(
            pil_img, lang=self.lang, config=config,
            output_type=pytesseract.Output.DICT
        )
        
        results = []
        n = len(data['text'])
        for i in range(n):
            conf = int(data['conf'][i])
            text = data['text'][i].strip()
            
            if conf >= self.min_confidence and text:
                results.append(OCRResult(
                    text=text,
                    confidence=conf,
                    bbox=(data['left'][i], data['top'][i],
                          data['width'][i], data['height'][i]),
                    level=data['level'][i]
                ))
        
        return results
    
    def get_text(self, image: np.ndarray) -> str:
        """获取纯文本"""
        results = self.ocr(image)
        return ' '.join(r.text for r in results)
    
    def process_file(self, input_path: str, output_path: str):
        """处理单个文件"""
        image = cv2.imread(input_path)
        if image is None:
            raise ValueError(f"无法读取: {input_path}")
        
        text = self.get_text(image)
        
        with open(output_path, 'w', encoding='utf-8') as f:
            f.write(text)
        
        return text

# 使用
pipeline = OCRPipeline(lang='chi_sim+eng', min_confidence=60)
text = pipeline.process_file('input.png', 'output.txt')

import os
import json
from concurrent.futures import ProcessPoolExecutor, as_completed
from dataclasses import asdict
from datetime import datetime

class BatchOCRProcessor:
    """批量 OCR 处理器"""
    
    def __init__(self, pipeline: OCRPipeline, workers=4):
        self.pipeline = pipeline
        self.workers = workers
    
    def _process_single(self, args):
        """处理单个文件"""
        input_path, output_dir = args
        filename = os.path.basename(input_path)
        
        try:
            image = cv2.imread(input_path)
            results = self.pipeline.ocr(image)
            
            # 保存文本
            text = ' '.join(r.text for r in results)
            output_path = os.path.join(output_dir, filename.rsplit('.', 1)[0] + '.txt')
            with open(output_path, 'w', encoding='utf-8') as f:
                f.write(text)
            
            # 保存详细结果
            detail_path = os.path.join(output_dir, filename.rsplit('.', 1)[0] + '.json')
            with open(detail_path, 'w', encoding='utf-8') as f:
                json.dump([asdict(r) for r in results], f, ensure_ascii=False, indent=2)
            
            return {
                'file': filename,
                'status': 'success',
                'chars': len(text),
                'avg_confidence': sum(r.confidence for r in results) / len(results) if results else 0
            }
        
        except Exception as e:
            return {
                'file': filename,
                'status': 'error',
                'error': str(e)
            }
    
    def process_batch(self, input_dir: str, output_dir: str):
        """批量处理"""
        os.makedirs(output_dir, exist_ok=True)
        
        # 收集文件
        files = []
        for f in os.listdir(input_dir):
            if f.lower().endswith(('.png', '.jpg', '.jpeg', '.tif', '.tiff', '.bmp')):
                files.append((os.path.join(input_dir, f), output_dir))
        
        print(f"待处理: {len(files)} 个文件")
        
        results = []
        with ProcessPoolExecutor(max_workers=self.workers) as executor:
            futures = {executor.submit(self._process_single, f): f[0] for f in files}
            
            for future in as_completed(futures):
                result = future.result()
                results.append(result)
                
                status = "✓" if result['status'] == 'success' else "✗"
                print(f"{status} {result['file']}")
        
        # 保存报告
        report = {
            'timestamp': datetime.now().isoformat(),
            'total': len(results),
            'success': sum(1 for r in results if r['status'] == 'success'),
            'failed': sum(1 for r in results if r['status'] == 'error'),
            'results': results
        }
        
        report_path = os.path.join(output_dir, 'report.json')
        with open(report_path, 'w', encoding='utf-8') as f:
            json.dump(report, f, ensure_ascii=False, indent=2)
        
        return report

# 使用
pipeline = OCRPipeline(lang='chi_sim+eng')
processor = BatchOCRProcessor(pipeline, workers=4)
report = processor.process_batch('./input', './output')

import time
from functools import wraps

def retry(max_retries=3, delay=1):
    """重试装饰器"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if attempt == max_retries - 1:
                        raise
                    time.sleep(delay * (attempt + 1))
            return None
        return wrapper
    return decorator

@retry(max_retries=3)
def safe_ocr(image_path, lang='chi_sim+eng'):
    """带重试的 OCR"""
    img = Image.open(image_path)
    return pytesseract.image_to_string(img, lang=lang)

def ocr_with_fallback(image_path, lang='chi_sim+eng'):
    """带降级的 OCR"""
    img = Image.open(image_path)
    
    # 策略 1: 完整配置
    try:
        config = '--psm 6 --oem 1 -c tessedit_min_confidence=60'
        text = pytesseract.image_to_string(img, lang=lang, config=config)
        if len(text.strip()) > 10:
            return text, 'full'
    except:
        pass
    
    # 策略 2: 简化配置
    try:
        config = '--psm 3'
        text = pytesseract.image_to_string(img, lang=lang, config=config)
        if len(text.strip()) > 5:
            return text, 'simple'
    except:
        pass
    
    # 策略 3: 最基本配置
    try:
        text = pytesseract.image_to_string(img, lang='eng')
        return text, 'basic'
    except:
        return '', 'failed'

def validate_ocr_result(text, expected_type='general'):
    """验证 OCR 结果质量"""
    issues = []
    
    if not text or not text.strip():
        issues.append("空结果")
        return False, issues
    
    # 检查乱码（连续特殊字符）
    import re
    if re.search(r'[^\w\s\u4e00-\u9fff]{5,}', text):
        issues.append("可能包含乱码")
    
    # 检查重复（同一字符重复多次）
    if re.search(r'(.)\1{10,}', text):
        issues.append("存在重复字符")
    
    # 根据类型检查
    if expected_type == 'number':
        if not re.search(r'\d', text):
            issues.append("未识别到数字")
    
    elif expected_type == 'chinese':
        if not re.search(r'[\u4e00-\u9fff]', text):
            issues.append("未识别到中文")
    
    is_valid = len(issues) == 0
    return is_valid, issues

def analyze_confidence(results: List[OCRResult]):
    """分析置信度分布"""
    if not results:
        return {}
    
    confs = [r.confidence for r in results]
    
    stats = {
        'count': len(confs),
        'mean': sum(confs) / len(confs),
        'min': min(confs),
        'max': max(confs),
        'low_confidence': sum(1 for c in confs if c < 60),
        'medium_confidence': sum(1 for c in confs if 60 <= c < 80),
        'high_confidence': sum(1 for c in confs if c >= 80),
    }
    
    return stats

import time
import statistics

def benchmark_ocr(image_path, iterations=10, lang='chi_sim+eng'):
    """OCR 性能基准测试"""
    img = Image.open(image_path)
    
    times = []
    for _ in range(iterations):
        start = time.time()
        pytesseract.image_to_string(img, lang=lang)
        times.append(time.time() - start)
    
    stats = {
        'iterations': iterations,
        'mean': statistics.mean(times),
        'median': statistics.median(times),
        'stdev': statistics.stdev(times) if len(times) > 1 else 0,
        'min': min(times),
        'max': max(times),
    }
    
    print(f"平均: {stats['mean']:.3f}s, 中位数: {stats['median']:.3f}s")
    return stats

def recommend_ocr_solution(requirements):
    """OCR 方案推荐"""
    scores = {
        'Tesseract': 0,
        'PaddleOCR': 0,
        'EasyOCR': 0,
        'Cloud API': 0,
    }
    
    # 语言因素
    if requirements.get('primary_lang') == 'chinese':
        scores['PaddleOCR'] += 3
        scores['Tesseract'] += 1
    elif requirements.get('primary_lang') == 'english':
        scores['Tesseract'] += 3
        scores['PaddleOCR'] += 2
    
    # 精度要求
    if requirements.get('accuracy') == 'high':
        scores['Cloud API'] += 3
        scores['PaddleOCR'] += 2
    elif requirements.get('accuracy') == 'medium':
        scores['Tesseract'] += 2
        scores['PaddleOCR'] += 2
    
    # 成本考虑
    if requirements.get('budget') == 'free':
        scores['Cloud API'] -= 5
        scores['Tesseract'] += 2
        scores['PaddleOCR'] += 2
    
    # 部署复杂度
    if requirements.get('deployment') == 'simple':
        scores['Tesseract'] += 2
        scores['EasyOCR'] += 2
    
    # 推荐
    best = max(scores.items(), key=lambda x: x[1])
    return best[0], scores

# 使用
req = {
    'primary_lang': 'chinese',
    'accuracy': 'high',
    'budget': 'free',
    'deployment': 'moderate'
}
recommendation, scores = recommend_ocr_solution(req)
print(f"推荐方案: {recommendation}")
print(f"评分: {scores}")

# 1. 查看 Tesseract 版本和语言
tesseract --version
tesseract --list-langs

# 2. 输出调试图像
tesseract image.png output --psm 6 -c tessedit_write_images=true

# 3. 尝试不同 PSM
for psm in 3 4 6 11; do
    echo "=== PSM $psm ==="
    tesseract image.png stdout --psm $psm -l chi_sim+eng | head -5
done

# 4. 查看详细输出
tesseract image.png stdout -l chi_sim+eng --psm 6 hocr | grep 'ocrx_word'

class OCREvaluator:
    """OCR 评估体系"""
    
    def __init__(self, gt_dir):
        self.gt_dir = gt_dir
    
    def evaluate(self, ocr_results):
        """评估 OCR 结果"""
        metrics = {
            'char_accuracy': self.char_accuracy(ocr_results),
            'word_accuracy': self.word_accuracy(ocr_results),
            'confidence_stats': self.confidence_stats(ocr_results),
        }
        return metrics
    
    def char_accuracy(self, results):
        """字符准确率"""
        total = 0
        correct = 0
        for result in results:
            gt = result['ground_truth']
            pred = result['prediction']
            for g, p in zip(gt, pred):
                total += 1
                if g == p:
                    correct += 1
        return correct / total if total > 0 else 0