基于腾讯云智能结构化 OCR 能力的最佳技术实践

pip install requests

import json
import base64
import hashlib
import hmacimport time
import requests

# 配置API密钥
SECRET_ID = 'YOUR_SECRET_ID'
SECRET_KEY = 'YOUR_SECRET_KEY'
ENDPOINT = 'ocr.tencentcloudapi.com'
REGION = 'ap-guangzhou'
ACTION = 'GeneralBasicOCR'
VERSION = '2022-03-23'

ef generate_signature(secret_key, params):    
      sorted_params = sorted(params.items())    
      canonical_query_string = '&'.join(['{}={}'.format(k, v) for k, v in sorted_params])    
      string_to_sign = 'GETocr.tencentcloudapi.com/?' + canonical_query_string    
      signature = hmac.new(secret_key.encode('utf-8'), string_to_sign.encode('utf-8'), hashlib.sha1).digest()    
      signature = base64.b64encode(signature).decode()    
      return signature
      
def call_ocr_api(image_path):    
      with open(image_path, 'rb') as f:        
          image_data = base64.b64encode(f.read()).decode('utf-8')        
      params = {        
          'Action': ACTION,        
          'Version': VERSION,        
          'Region': REGION,        
          'Timestamp': int(time.time()),        
          'Nonce': int(time.time()),        
          'SecretId': SECRET_ID,        
          'ImageBase64': image_data,        
          'LanguageType': 'CHN_ENG'    
      }        
      
      signature = generate_signature(SECRET_KEY, params)    
      params['Signature'] = signature        
      
      response = requests.get(f'https://{ENDPOINT}/', params=params)    
      return response.json()
      
  # 调用OCR API
  result = call_ocr_api('path_to_image.jpg')
  print(json.dumps(result, indent=2, ensure_ascii=False))

def parse_ocr_response(response):    
      if 'Response' in response and 'Error' not in response['Response']:        
          items = response['Response']['TextDetections']        
          for item in items:            
              print(f"Detected Text: {item['DetectedText']}")            
              print(f"Confidence: {item['Confidence']}")            
              print(f"Bounding Box: {item['Polygon']}")            
              print('-' * 30)    
          else:        
              print("Error:", response['Response']['Error']['Message'])
              
# 解析并打印识别结果
parse_ocr_response(result)

pip install tencentcloud-sdk-python

from tencentcloud.common import credential
from tencentcloud.ocr.v20181119 import ocr_client, models
import base64
import json

# 配置API密钥
cred = credential.Credential("YOUR_SECRET_ID", "YOUR_SECRET_KEY")
client = ocr_client.OcrClient(cred, "ap-guangzhou")

# 读取并编码图像with open("path_to_image.jpg", "rb") as image_file:    image_data = base64.b64encode(image_file.read()).decode('utf-8')# 创建请求对象req = models.GeneralBasicOCRRequest()params = {    "ImageBase64": image_data,    "LanguageType": "CHN_ENG"}req.from_json_string(json.dumps(params))# 调用OCR接口resp = client.GeneralBasicOCR(req)print(resp.to_json_string())

def parse_sdk_response(response):    
      for item in response.TextDetections:        
          print(f"Detected Text: {item.DetectedText}")        
          print(f"Confidence: {item.Confidence}")        
          print(f"Bounding Box: {item.Polygon}")        
          print('-' * 30)
# 解析并打印识别结果
parse_sdk_response(resp)

import cv2
import numpy as np

def preprocess_image(image_path):    
      # 读取图像    
      image = cv2.imread(image_path)        
      
      # 灰度化    
      gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)        
      
      # 二值化    
      _, binary = cv2.threshold(gray, 150, 255, cv2.THRESH_BINARY_INV)        
      
      # 去噪    
      kernel = np.ones((1,1), np.uint8)    
      denoised = cv2.morphologyEx(binary, cv2.MORPH_OPEN, kernel)        
      
      # 旋转校正    
      coords = np.column_stack(np.where(denoised > 0))    
      angle = cv2.minAreaRect(coords)[-1]    
      if angle < -45:        
          angle = -(90 + angle)    
      else:        
          angle = -angle    
      (h, w) = denoised.shape    
      center = (w // 2, h // 2)    
      M = cv2.getRotationMatrix2D(center, angle, 1.0)    
      rotated = cv2.warpAffine(denoised, M, (w, h), flags=cv2.INTER_CUBIC, borderMode=cv2.BORDER_REPLICATE)        
      return rotated
      
# 示例
preprocessed_image = preprocess_image('path_to_image.jpg')
cv2.imwrite('preprocessed_image.jpg', preprocessed_image)

def extract_fields(ocr_result):    
      fields = {        
          'Name': '',        
          'ID Number': '',        
          'Income': '',        
          'Loan Amount': '',        
          'Date': ''    
      }        
      
      for item in ocr_result['Response']['TextDetections']:        text = item['DetectedText']        if '姓名' in text:            fields['Name'] = text.split('姓名')[-1].strip()        elif '身份证号' in text:            fields['ID Number'] = text.split('身份证号')[-1].strip()        elif '收入' in text:            fields['Income'] = text.split('收入')[-1].strip()        elif '贷款金额' in text:            fields['Loan Amount'] = text.split('贷款金额')[-1].strip()        elif '日期' in text:            fields['Date'] = text.split('日期')[-1].strip()        return fields# 示例fields = extract_fields(result)print(fields)

def safe_call_ocr_api(image_path):    
      try:        
          response = call_ocr_api(image_path)        
          if 'Response' in response and 'Error' not in response['Response']:            
              return response        
          else:           
              error_message = response['Response']['Error']['Message']            
              print(f"OCR API Error: {error_message}")            
              return None    
              
       except Exception as e:        
           print(f"Exception occurred: {e}")        
           return None
           
# 示例
response = safe_call_ocr_api('path_to_image.jpg')
if response:    
      fields = extract_fields(response)    
      print(fields)
else:    
      print("OCR processing failed.")

import threading

def process_batch(image_paths):    
      results = []    
      threads = []        
      
      def worker(image_path):        
          response = safe_call_ocr_api(image_path)        
          if response:            
              fields = extract_fields(response)            
              results.append(fields)        
      for path in image_paths:        
          t = threading.Thread(target=worker, args=(path,))        
          threads.append(t)        
          t.start()        
      for t in threads:        
          t.join()        
      return results
      
# 示例
image_paths = ['image1.jpg', 'image2.jpg', 'image3.jpg']
batch_results = process_batch(image_paths)print(batch_results)

import logging

# 配置日志
logging.basicConfig(filename='ocr_processing.log', level=logging.INFO,                    
                    format='%(asctime)s %(levelname)s:%(message)s')
def safe_call_ocr_api_with_logging(image_path):    
      try:        
          response = call_ocr_api(image_path)        
          if 'Response' in response and 'Error' not in response['Response']:            
              logging.info(f"OCR Success: {image_path}")            
              return response        
          else:            
              error_message = response['Response']['Error']['Message']            
              logging.error(f"OCR API Error for {image_path}: {error_message}")           
              return None    
              
      except Exception as e:        
          logging.error(f"Exception for {image_path}: {e}")        
          return 
          None
          
# 示例
response = safe_call_ocr_api_with_logging('path_to_image.jpg')
if response:    
      fields = extract_fields(response)    
      print(fields)
else:    
      print("OCR processing failed.")

import json
import base64
import hashlib
import hmac
import time
import requests
import cv2import numpy as np
import threading
import logging
from tencentcloud.common import credential
from tencentcloud.ocr.v20181119 import ocr_client, models

# 配置日志
logging.basicConfig(filename='ocr_processing.log', level=logging.INFO,                    
                    format='%(asctime)s %(levelname)s:%(message)s')
                    
# 配置API密钥
SECRET_ID = 'YOUR_SECRET_ID'
SECRET_KEY ='YOUR_SECRET_KEY'
ENDPOINT = 'ocr.tencentcloudapi.com'
REGION = 'ap-guangzhou'
ACTION = 'GeneralBasicOCR'
VERSION = '2022-03-23'

def preprocess_image(image_path):    
    # 读取图像    
    image = cv2.imread(image_path) 
           
    # 灰度化    
    gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)        
    
    # 二值化    
    _, binary = cv2.threshold(gray, 150, 255, cv2.THRESH_BINARY_INV)        
    
    # 去噪    
    kernel = np.ones((1,1), np.uint8)    
    denoised = cv2.morphologyEx(binary, cv2.MORPH_OPEN, kernel)        
    
    # 旋转校正    
    coords = np.column_stack(np.where(denoised > 0))    
    angle = cv2.minAreaRect(coords)[-1]    
    if angle < -45:        
        angle = -(90 + angle)    else:        
        angle = -angle    
    (h, w) = denoised.shape    
    center = (w // 2, h // 2)    
    M = cv2.getRotationMatrix2D(center, angle, 1.0)    
    rotated = cv2.warpAffine(denoised, M, (w, h), flags=cv2.INTER_CUBIC, borderMode=cv2.BORDER_REPLICATE)        
    # 保存预处理后的图像    
    preprocessed_path = 'preprocessed_' + image_path    
    cv2.imwrite(preprocessed_path, rotated)    
    return preprocessed_path
      
def generate_signature(secret_key, params):    
    sorted_params = sorted(params.items())    
    canonical_query_string = '&'.join(['{}={}'.format(k, v) for k, v in sorted_params])    
    string_to_sign = 'GETocr.tencentcloudapi.com/?' + canonical_query_string    
    signature = hmac.new(secret_key.encode('utf-8'), string_to_sign.encode('utf-8'), hashlib.sha1).digest()    
    signature = base64.b64encode(signature).decode()    
    return signature
      
def call_ocr_api(image_path):    
    preprocessed_image = preprocess_image(image_path)    
    with open(preprocessed_image, 'rb') as f:        
        image_data = base64.b64encode(f.read()).decode('utf-8')        
      
    params = {        
        'Action': ACTION,        
        'Version': VERSION,        
        'Region': REGION,        
        'Timestamp': int(time.time()),        
        'Nonce': int(time.time()),        
        'SecretId': SECRET_ID,        
        'ImageBase64': image_data,        
        'LanguageType': 'CHN_ENG'    
    }        
    
    signature = generate_signature(SECRET_KEY, params)    
    params['Signature'] = signature        
    
    try:        
        response = requests.get(f'https://{ENDPOINT}/', params=params, timeout=10)        
        response.raise_for_status()        
        return response.json()    
    except requests.exceptions.RequestException as e:        
        logging.error(f"HTTP Request failed: {e}")        
        return None

def extract_fields(ocr_result):    
    fields = {        
        'Name': '',        
        'ID Number': '',        
        'Income': '',        
        'Loan Amount': '',        
        'Date': ''    
    }        
    
    for item in ocr_result.get('Response', {}).get('TextDetections', []):        
        text = item.get('DetectedText', '')        
        if '姓名' in text:            
            fields['Name'] = text.split('姓名')[-1].strip()        
        elif '身份证号' in text:            
            fields['ID Number'] = text.split('身份证号')[-1].strip()        
        elif '收入' in text:            
            fields['Income'] = text.split('收入')[-1].strip()        
        elif '贷款金额' in text:            
            fields['Loan Amount'] = text.split('贷款金额')[-1].strip()        
        elif '日期' in text:            
            fields['Date'] = text.split('日期')[-1].strip()        
    return fields
    
def safe_call_ocr_api_with_logging(image_path):    
    response = call_ocr_api(image_path)   
    if response and 'Response' in response and 'Error' not in response['Response']:        
        logging.info(f"OCR Success: {image_path}")        
        fields = extract_fields(response)        
        return fields    
    else:        
        error_message = response.get('Response', {}).get('Error', {}).get('Message', 'Unknown Error')
        logging.error(f"OCR API Error for {image_path}: {error_message}")        
        return None
        
def process_image(image_path):    
    fields = safe_call_ocr_api_with_logging(image_path)    
    if fields:        
        print(f"Processed {image_path}: {fields}")    
    else:        
        print(f"Failed to process {image_path}")
        
def main():    
      image_paths = ['loan_application1.jpg', 'loan_application2.jpg', 'loan_application3.jpg']    
      threads = []        
      
      for path in image_paths:        
          t = threading.Thread(target=process_image, args=(path,))        
          threads.append(t)        
          t.start()        
          
      for t in threads:        
      t.join()
      
if __name__ == "__main__":    
      main()