initial

.gitignore

@@ -0,0 +1,7 @@
+*/__pycache__
+__pycache__
+/test
+test*
+/dist
+/ocr_needs
+/ocr_workspace

client_test.py

@@ -0,0 +1,107 @@
+import requests
+import json
+
+class OCRClient:
+    def __init__(self, base_url):
+        self.base_url = base_url
+
+    def detect_barcode(self, image_path):
+        url = f"{self.base_url}/detect_barcode"
+        with open(image_path, 'rb') as f:
+            files = {'file': f}
+            response = requests.post(url, files=files)
+        return response.json()
+
+    def get_barcode(self, image_path=None, message=None):
+        url = f"{self.base_url}/get_barcode"
+        data = {}
+        files = {}
+        
+        if image_path:
+            with open(image_path, 'rb') as f:
+                files = {'file': f}
+        
+        if message:
+            data = {'message': json.dumps(message)}
+        
+        response = requests.post(url, data=data, files=files)
+        return response.json()
+
+    def search_info(self, search_params):
+        url = f"{self.base_url}/search"
+        response = requests.post(url, json=search_params)
+        return response.json()
+
+    def get_matio_id(self, image_path):
+        url = f"{self.base_url}/get_matio_id"
+        with open(image_path, 'rb') as f:
+            files = {'file': f}
+            response = requests.post(url, files=files)
+        return response.json()
+
+    def show_info(self, id):
+        url = f"{self.base_url}/show"
+        data = {'id': id}
+        response = requests.post(url, json=data)
+        return response.json()
+
+    def history_info(self, id):
+        url = f"{self.base_url}/history"
+        data = {'id': id}
+        response = requests.post(url, json=data)
+        return response.json()
+
+# 使用示例
+if __name__ == "__main__":
+    client = OCRClient("http://localhost:8000")  # 替换为实际的服务器地址和端口
+
+    # 检测条形码
+    barcode_result = client.detect_barcode("path/to/image.jpg")
+    print("条形码检测结果:", barcode_result)
+
+    # 获取条形码信息
+    barcode_info = client.get_barcode(image_path="path/to/image.jpg")
+    print("条形码信息:", barcode_info)
+
+    # 搜索信息
+    search_params = {
+        "barcode_type": "EAN13",
+        "matio_id": "123456",
+        "pageNum": 1,
+        "pageSize": 10
+    }
+    search_result = client.search_info(search_params)
+    print("搜索结果:", search_result)
+
+    # 获取matio_id
+    matio_id_result = client.get_matio_id("path/to/image.jpg")
+    print("Matio ID 结果:", matio_id_result)
+
+    # 显示信息
+    show_result = client.show_info("1")
+    print("显示信息:", show_result)
+
+    # 历史信息
+    history_result = client.history_info("1")
+    print("历史信息:", history_result)
+
+    # 使用图片路径调用get_barcode
+    barcode_info_with_image = client.get_barcode(image_path="path/to/image.jpg")
+    print("使用图片的条形码信息:", barcode_info_with_image)
+
+    # 使用message调用get_barcode
+    message = {
+        'barcode': '1C3JAA57000BL',
+        'image_origin_path': '/clt/ocr_workspace/tmp_images/origin_images_2024_09_29-16__43__55.png',
+        'upload_timestamp': '2024_09_29-16__43__55',
+        'barcode_type': '普通吊牌'
+    }
+    barcode_info_with_message = client.get_barcode(message=message)
+    print("使用message的条形码信息:", barcode_info_with_message)
+
+    # 同时使用图片和message调用get_barcode
+    barcode_info_combined = client.get_barcode(
+        image_path="path/to/image.jpg",
+        message=message
+    )
+    print("同时使用图片和message的条形码信息:", barcode_info_combined)

config.py

@@ -0,0 +1,82 @@
+from ultralytics import YOLO
+from paddleocr import PaddleOCR
+import platform, torch, os, json
+import pydantic
+os.environ['KMP_DUPLICATE_LIB_OK'] = 'TRUE'
+barcode_type_list = ['RFID吊牌', '普通吊牌', '洗水唛', '饰品洗水唛', '饰品吊牌']
+if platform.system().lower() == "linux":
+    model_dir = "./ocr_needs/yolo-material/yolo_v8_model/"
+    ocr_det = "./ocr_needs/ocr_models/en_PP-OCRv3_det_infer"
+    ocr_rec = "./ocr_needs/ocr_models/en_PP-OCRv4_rec_infer"
+    ocr_det_ch = "./ocr_needs/ocr_models/ch_PP-OCRv4_det_infer"
+    ocr_rec_ch = "./ocr_needs/ocr_models/ch_PP-OCRv4_rec_infer"
+    ocr_cls = "./ocr_needs/ocr_models/ch_ppocr_mobile_v2.0_cls_infer"
+    ocr_images_dir = './ocr_workspace/'
+    ttf_path = './ocr_needs/MiSans-Heavy.ttf'
+header = '/data'
+address = 'ocr.gloria.com.cn'
+port = 8066
+file_url = f'http://{address}:{port}{header}'
+# 检查CUDA是否可用
+device = 'cuda' if torch.cuda.is_available() else 'cpu'
+print(f'device: {device} is used')
+
+dirs_to_check = [model_dir, ocr_images_dir]
+# ocr = PaddleOCR(use_angle_cls=True, lang='en', show_log=False, det_model_dir=ocr_det,
+#                rec_model_dir=ocr_rec, cls_model_dir=ocr_cls)
+ocr = PaddleOCR(use_angle_cls=True, lang='ch', show_log=False, det_model_dir=ocr_det_ch,
+               rec_model_dir=ocr_rec_ch, cls_model_dir=ocr_cls, use_gpu=True)
+print('ocr model is ready!')
+for directory in dirs_to_check:
+    if not os.path.exists(directory):
+        os.makedirs(directory)
+        print(f"Directory {directory} created.")
+    else:
+        print(f"Directory {directory} already exists.")
+
+# Define the model path
+model_path = os.path.join(model_dir, "obb-s-best-4.pt")
+model = YOLO(model_path).to(device)
+model.fuse()
+print('yolo model is ready !')
+
+
+sql_config = {
+    'user': 'root',
+    'password': 'Ywkj2020',
+    'host': '10.40.0.30',
+    'db': 'unique-code'
+}
+
+color_config = {
+    'user': 'IT',
+    'password': 'it@2022*',
+    'host': '10.40.0.81',
+    'database': 'goelia_erp',
+}
+
+class Search(pydantic.BaseModel):
+    matio_id: str = None
+    item_num : str = None
+    uploadStartTime : str = None
+    uploadEndTime : str = None
+    ocrStartTime : str = None
+    ocrEndTime : str = None
+    difference : str = None
+    pageSize : int = 20
+    pageNum : int = 1
+    barcode_type: str = None
+
+class Matio(pydantic.BaseModel):
+    barcode: str = ''
+    matio_id: str = ''
+    image_origin_path : str = 'None'
+    upload_time : str = 'None'
+    barcode_type : str = 'RFID吊牌'
+
+class ID(pydantic.BaseModel):
+    id: str = None
+
+class Item(pydantic.BaseModel):
+    name: str
+    description: str    

ocr_compare.py

@@ -0,0 +1,311 @@
+import os, io, time, json, csv, ast, traceback, cv2, asyncio, datetime
+from typing import Optional
+from concurrent.futures import ThreadPoolExecutor
+
+import numpy as np
+import pandas as pd
+from PIL import Image
+from fastapi import FastAPI, File, UploadFile, Form
+from fastapi.responses import JSONResponse
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.staticfiles import StaticFiles
+
+from config import (
+    ocr_images_dir, model, header, port, file_url, 
+    Search, ID, Matio, Item
+)
+from utils import get_time, detection, sql_product, image_handle, Compare
+
+detect_instance = detection()
+ocr = detect_instance.ocr
+
+app = FastAPI()
+executor = ThreadPoolExecutor(max_workers=15)
+camera_connections = {}
+app.mount(header, StaticFiles(directory=ocr_images_dir), name="static")
+app.add_middleware(
+   CORSMiddleware,
+   allow_origins=["*"],
+   allow_credentials=True,
+   allow_methods=["*"],
+   allow_headers=["*"],
+)
+@app.post('/detect_barcode')
+async def detect_barcode(file: UploadFile = File(...)):
+    try:
+        # 确保目录存在
+        tmp_dir = os.path.join(ocr_images_dir, 'tmp_images')
+        os.makedirs(tmp_dir, exist_ok=True)
+        
+        # 读取并处理上传的图像
+        contents = await file.read()
+        upload_timestamp = datetime.datetime.now().strftime("%Y_%m_%d-%H__%M__%S")
+        image = Image.open(io.BytesIO(contents))
+        image = image_handle.correct_image_orientation(image)
+        image = cv2.cvtColor(np.array(image.convert("RGB")), cv2.COLOR_RGB2BGR)
+        
+        # 保存原始图像
+        image_origin_path = os.path.join(tmp_dir, f"origin_images_{upload_timestamp}.png")
+        cv2.imwrite(image_origin_path, image)
+        time1 = time.time()
+        # 预测并处理结果
+        results = model.predict(image, conf=0.3)
+        print(f"predict time is:{time.time() - time1}")
+        
+        for result in results:
+            for obb in result.obb:
+                if int(obb.cls.item()) == 15 and float(obb.conf.item()) > 0:
+                    points = obb.xyxyxyxy[0].cpu().numpy()
+                    cropped_image = image_handle.crop_image_second(image.copy(), points)
+                    barcode_image_path = os.path.join(tmp_dir, f"barcode_images_{upload_timestamp}.png")
+                    cv2.imwrite(barcode_image_path, cropped_image)
+                    barcode = await asyncio.get_event_loop().run_in_executor(
+                        executor, detect_instance.detect_barcode_ocr, barcode_image_path)
+                    if barcode:
+                        return {'barcode': barcode, 'image_origin_path': image_origin_path, 'upload_timestamp': upload_timestamp}
+        
+        return {'barcode': None, 'image_origin_path': image_origin_path, 'upload_timestamp': upload_timestamp}
+    except Exception as e:
+        print(e)
+        return None, None, None
+
+@app.post('/get_barcode')
+async def get_barcode(message: Optional[str] = Form(None), file: Optional[UploadFile] = File(None)):
+    try:
+        result_dir = os.path.join(ocr_images_dir, 'results')
+        os.makedirs(result_dir, exist_ok=True)
+
+        if message:
+            item_dict = json.loads(message)
+            message_new = Matio(**item_dict)
+            barcode, image_origin_path, upload_timestamp, barcode_type, matio_id = message_new.barcode, message_new.image_origin_path, message_new.upload_time, message_new.barcode_type, message_new.matio_id
+        elif file:
+            barcode_data = await detect_barcode(file)
+            barcode, image_origin_path, upload_timestamp = barcode_data.get('barcode'), barcode_data.get('image_origin_path'), barcode_data.get('upload_timestamp')
+            barcode_type, matio_id = Matio().barcode_type, Matio().matio_id
+        
+        if not barcode:
+            return JSONResponse(content={"code": 0, "decs": "未识别到barcode，请重新上传"}, status_code=500)
+        
+        image = cv2.imread(image_origin_path)
+        image_width, image_height = image.shape[1], image.shape[0]
+        
+        detection_timestamp = datetime.datetime.now().strftime("%Y_%m_%d-%H__%M__%S")
+        data_result, matio_id, color_id, _ = sql_product.sql_information(barcode=barcode, barcode_type=barcode_type, matio_id=matio_id)
+        
+        results_dir = os.path.join(result_dir, matio_id)
+        os.makedirs(results_dir, exist_ok=True)
+        
+        result_image_path = os.path.join(results_dir, f"ocr_result_images_{detection_timestamp}.png")
+        regular_image_path = os.path.join(results_dir, f"regular_images_{upload_timestamp}.png")
+        cv2.imwrite(regular_image_path, image)
+        
+        ocr_result = []
+        result = ocr.ocr(regular_image_path, cls=True)
+        
+        for line in result[0]:
+            bbox = [[int(x) for x in point] for point in line[0]]
+            text = line[1][0]
+            image = image_handle.draw_box_and_text(image, bbox, text)
+            ocr_result.append([text])
+        
+        resize = cv2.resize(image, (image_width, image_height), interpolation=cv2.INTER_AREA)
+        cv2.imwrite(result_image_path, resize)
+        
+        ocr_image_url = file_url + result_image_path.replace(ocr_images_dir, '/')
+        regular_image_url = file_url + regular_image_path.replace(ocr_images_dir, '/')
+        
+        data_set, log = Compare.compare(ocr_result=ocr_result, dataset=data_result)
+
+        color_set = {k: data_set.get(k) for k in ['color_name', 'color_id', '产品名称', 'language']}
+        color_set['颜色'] = color_set.pop('color_name') if 'color_name' in color_set else ''
+        color_set['色号'] = color_set.pop('color_id') if 'color_id' in color_set else ''
+        color_set['语言'] = color_set.pop('language') if 'language' in color_set else ''
+
+        name_set = {k: data_set.get(k) for k in ['号型', 'size_id', '产品名称', 'language']}
+        if 'size_id' in name_set:
+            name_set.pop('size_id')
+        name_set['语言'] = name_set.pop('language') if 'language' in name_set else ''
+        for k in ['color_name', 'color_id', '号型', 'size_id', '产品名称', 'language']:
+            data_set.pop(k, None)
+        
+        size_compare_flag, size_compare_logs = sql_product.size_information(matio_id, color_id)
+        color_compare_flag, color_compare_logs = sql_product.color_information(matio_id)
+        
+        csv_file_path = os.path.join(ocr_images_dir, 'history.csv')
+        dicts = {
+            "id": str(len(pd.read_csv(csv_file_path)) if os.path.exists(csv_file_path) else '0'),
+            "matio_id": matio_id,
+            "item_num": matio_id.split('-')[0],
+            "difference": '1' if log else '0',
+            "upload_time": upload_timestamp.replace('__', ':').replace('_', '/').replace('-', ' '),
+            "ocr_time": detection_timestamp.replace('__', ':').replace('_', '/').replace('-', ' '),
+            "logs": log,
+            "regular_image": regular_image_url,
+            "ocr_image": ocr_image_url,
+            "size_compare_flag": size_compare_flag,
+            "size_compare_logs": size_compare_logs if size_compare_logs else "'None'",
+            "color_compare_flag": color_compare_flag,
+            "color_compare_logs": color_compare_logs if color_compare_logs else "'None'",
+            "barcode_type": barcode_type,
+            "data_set": data_set,
+            "color_set": color_set,
+            "name_set": name_set
+        }
+        
+        with open(csv_file_path, 'a', newline='', encoding='utf-8') as csv_file:
+            writer = csv.DictWriter(csv_file, fieldnames=dicts.keys())
+            if csv_file.tell() == 0:
+                writer.writeheader()
+            writer.writerow(dicts)
+        
+        return JSONResponse(content={"code": 1, "decs": "识别成功"}, status_code=200)
+    
+    except Exception as e:
+        traceback.print_exc()
+        return JSONResponse(content={"code": 0, "decs": "失败，出现错误，请重试"}, status_code=500)
+
+@app.post("/test/")
+async def test(file: Optional[UploadFile] = File(None),
+    item: Optional[str] = Form(None)):
+    if item:
+        item_dict = json.loads(item)
+        item_model = Item(**item_dict)
+        return {"item": item_model.name + item_model.description, "filename": file}
+    else:
+        return {'status': 'succeed', "filename":file.filename}
+
+@app.post('/search')
+async def search_info(message: Search):
+    try:
+        csv_path = os.path.join(ocr_images_dir, 'history.csv')
+        if not os.path.exists(csv_path):
+            return {"code": "1", "decs": None, "data": None}
+        
+        df = pd.read_csv(csv_path)
+        df_length = len(df)
+        
+        # 应用过滤条件
+        filters = { 
+            'barcode_type': message.barcode_type,
+            'matio_id': message.matio_id,
+            'item_num': message.item_num,
+            'difference': int(message.difference) if message.difference else None
+        }
+        for col, condition in filters.items():
+            
+            if condition or condition==0:
+                df = df[df[col] == condition]
+        # 时间过滤
+        for time_col, start_time, end_time in [
+            ('upload_time', message.uploadStartTime, message.uploadEndTime),
+            ('ocr_time', message.ocrStartTime, message.ocrEndTime)
+        ]:
+            if start_time and end_time:
+                df[time_col] = pd.to_datetime(df[time_col])
+                start = datetime.datetime(**get_time(start_time))
+                end = datetime.datetime(**get_time(end_time))
+                df = df[(df[time_col] >= start) & (df[time_col] <= end)]
+                df[time_col] = df[time_col].dt.strftime("%Y/%m/%d %H:%M:%S")
+        
+        # 排序和分页
+        df = df.sort_values(by="upload_time", ascending=False)
+        start = (message.pageNum - 1) * message.pageSize
+        end = start + message.pageSize
+        page_data = df.iloc[start:end].to_dict(orient="records")
+        
+        # 处理数据
+        for record in page_data:
+            for key, value in record.items():
+                try:
+                    record[key] = ast.literal_eval(value)
+                except (SyntaxError, ValueError):
+                    pass
+        
+        data = {
+            "records": page_data,
+            "total": str(df_length),
+            "size": str(message.pageSize),
+            "current": str(message.pageNum),
+            "orders": [],
+            "optimizeCountSql": True,
+            "searchCount": True,
+            "countId": '',
+            "maxLimit": '',
+            "pages": str(df_length // message.pageSize + 1)
+        }
+        
+        return {"code": "1", "decs": None, "data": data}
+    
+    except Exception as e:
+        traceback.print_exc()
+        return JSONResponse(content={"code": "0", "decs": str(e)}, status_code=500)
+
+@app.post('/get_matio_id')
+async def get_matio_id(file: UploadFile = File(...)):
+    try:
+        if not file:
+            return {"code": "1", "matio_list": []}
+        
+        barcode_data = await detect_barcode(file)
+        barcode = barcode_data.get('barcode')
+        prefix_code_list = sql_product.sql_matio_id(prefix_code=barcode)
+        
+        return {
+            "code": "1", 
+            "matio_list": prefix_code_list, 
+            'image_origin_path': barcode_data.get('image_origin_path'),
+            'upload_timestamp': barcode_data.get('upload_timestamp'),
+            'barcode': barcode
+        }
+            
+    except Exception as e:
+        return JSONResponse(content={"code": "0", "decs": str(e)}, status_code=500)
+@app.post('/show')
+async def show_info(message: ID):
+    try:
+        csv_path = os.path.join(ocr_images_dir, 'history.csv')
+        df = pd.read_csv(csv_path)
+        data = df[df['id'] == int(message.id)].iloc[0].to_dict()
+        
+        for key, value in data.items():
+            try:
+                data[key] = ast.literal_eval(value)
+            except (ValueError, SyntaxError):
+                if key == 'logs':
+                    data[key] = json.loads(value.replace("'", '"'))
+        
+        return {"code": "1", "decs": None, "data": {"records": data}}
+    except Exception as e:
+        return JSONResponse(content={"code": "0", "decs": str(e)}, status_code=500)
+
+@app.post('/history')
+async def history_info(message: ID):
+    try:
+        csv_path = os.path.join(ocr_images_dir, 'history.csv')
+        df = pd.read_csv(csv_path)
+        
+        # 获取指定 id 对应的 matio_id
+        matio_id = df.loc[df['id'] == int(message.id), 'matio_id'].iloc[0]
+        
+        # 筛选并排序数据
+        data = df[df['matio_id'] == matio_id].sort_values(by="upload_time", ascending=False)
+        
+        # 格式化时间并转换为字典
+        data['upload_time'] = pd.to_datetime(data["upload_time"]).dt.strftime("%Y/%m/%d %H:%M:%S")
+        records = data.to_dict(orient='records')
+        
+        # 处理 logs 字段
+        for record in records:
+            record['logs'] = json.loads(record['logs'].replace("'", '"'))
+        
+        return {"code": "1", "decs": None, "data": {"records": records}}
+    
+    except Exception as e:
+        traceback.print_exc()
+        return JSONResponse(content={"code": "0", "decs": str(e)}, status_code=500)
+
+
+if __name__ == "__main__":
+    import uvicorn
+    uvicorn.run(app, host="0.0.0.0", port=port)

requirements.txt

@@ -0,0 +1,5 @@
+fastapi
+paddleocr
+ultralytics
+pyzbar
+paddlepaddle

sql_query.py

@@ -0,0 +1,22 @@
+QUERY_BASE = """SELECT goods_no 货号, clothes_type 产品名称, CONCAT(color_id, color_name) AS 色号, style_batch 批次, hao_type 号型,
+    exec_standard 执行标准, safe_standard 安全类别, element 纤维成分, maintaindes 保养说明, Maintaindes1 保养说明1, Maintaindes2 
+    保养说明2, Maintaindes3 保养说明3, Maintaindes4 保养说明4, price 零售价, size 尺码, desc11, desc9 产地, desc4, desc5, size_id, color_name, color_id, language,
+    warm 温馨提示, warm1 温馨提示1, warm2 温馨提示2, warm3 温馨提示3, warm4 温馨提示4, warm5 温馨提示5
+    FROM water_mark_info WHERE matio_id = '{matio_id}' AND COLOR_ID = '{color_id}' AND SIZE = '{size}' AND language = 'CN'"""
+
+QUERY_MATIO_ID = """SELECT BAR_CODE, matio_id, COLOR_ID, SIZE FROM tag_code WHERE BAR_CODE LIKE '{barcode}%'"""
+query_condition_rfid = "AND print_type LIKE '%主打%'"
+query_condition_normal = "AND print_type NOT LIKE '%主打%'"
+
+color_sql = """SELECT language, color_id, color_name, clothes_type FROM water_mark_info WHERE matio_id = '{matio_id}' AND language = 'CN'"""
+color_code_sql = """SELECT code, name FROM color WHERE code = '{code}'"""
+
+size_first_sql = """SELECT SIZE, hao_type, size_id, clothes_type, language FROM water_mark_info WHERE matio_id = '{matio_id}' AND COLOR_ID = '{color_id}' AND language = 'CN'"""
+size_check_sql = """SELECT d.dict_label, s.size_code, s.sizes 
+                 FROM base_size s 
+                 LEFT JOIN sys_dict_data d ON d.dict_value = s.size_name
+                 WHERE d.dict_type = 'base_size_group' 
+                 AND dict_label IN (SELECT group_size FROM water_mark_info WHERE group_size IS NOT NULL AND goods_no = '{item_id}' ORDER BY pkid ASC) 
+                 AND s.language = 'CN'"""
+
+matio_id_sql = """SELECT distinct matio_id FROM tag_code WHERE prefix_code = '{prefix_code}' order by matio_id"""

utils.py

@@ -0,0 +1,422 @@
+import re
+import cv2
+from PIL import Image, ImageDraw, ImageFont
+import numpy as np
+import pymysql
+import pandas as pd
+import pymssql
+import mysql.connector
+import datetime
+from config import ocr, ttf_path, sql_config, color_config
+import time, os
+from PIL import ExifTags
+from sql_query import *
+
+class detection:
+    def __init__(self) -> None:
+        self.ocr = ocr
+    def rotate_image(self, image, angle):
+        # 获取原图尺寸
+        (h, w) = image.shape[:2]
+        
+        # 计算旋转后的图像需要的尺寸
+        diagonal = int(np.sqrt(h**2 + w**2))
+        
+        # 创建一个更大的正方形画布
+        square = np.zeros((diagonal, diagonal, 3), dtype=np.uint8)
+        
+        # 计算原图需要平移的距离
+        offset_x = (diagonal - w) // 2
+        offset_y = (diagonal - h) // 2
+        
+        # 将原图放在新画布中心
+        square[offset_y:offset_y+h, offset_x:offset_x+w] = image
+        
+        # 旋转图像
+        center = (diagonal // 2, diagonal // 2)
+        M = cv2.getRotationMatrix2D(center, angle, 1.0)
+        rotated = cv2.warpAffine(square, M, (diagonal, diagonal))
+        
+        return rotated
+    def rotate_image_second(self, image, angle):
+        (h, w) = image.shape[:2]
+        center = (w / 2, h / 2)
+        M = cv2.getRotationMatrix2D(center, angle, 1.0)
+        rotated = cv2.warpAffine(image, M, (w, h))
+        return rotated
+
+    # 检查是否符合条码的基本格式
+    def is_valid_barcode(self, s: str):
+        # 检查字符串长度是否大于15
+        # if len(s) <= 15 or len(s) > 22:
+        #     return False
+        # 检查字符串开头是否为"1" 或 "I"（有时候1会误识别为I，在这个场景下通常其实为1）
+        # if not s.startswith('1'):
+        if not (s.startswith('1') or s.startswith("I")):
+            return False
+        # 检查字符串中是否包含指定的尺码之一
+        size_codes = {'XXS', 'XS', 'S', 'M', 'L', 'XL', 'XXL', 'F'}
+        if any(code in s for code in size_codes):
+            return True
+        return False
+
+
+    def check_and_return_string(self, data):
+        try:
+            # 将OCR结果转换为字符串
+            data_str = str(data)
+            # 使用正则表达式查找以"1"开头并被引号括起来的字符串
+            # matches = re.findall(r"'(1.*?)'", data_str)
+            matches = re.findall(r"'([1I].*?)'", data_str)
+            for match in matches:
+                if self.is_valid_barcode(match):
+                    # 如果条形码以"I"开头，将其转换为以"1"开头
+                    if match.startswith("I"):
+                        match = "1" + match[1:]
+                    return True, match
+            return False, None
+        except Exception as e:
+            print(e)
+            return False, None
+
+
+    def detect_barcode_ocr(self, img_path):
+        time1 = time.time()
+        image = cv2.imread(img_path)
+        # 定义旋转角度
+        angles = [0, 180, 90, 270, 45, 135, 225, 315, 5, 10, 20, 30, 355, 350, 340, 330, 60, 75, 105, 120, 150, 165, 195, 210, 240, 255, 285]
+        # 遍历角度进行 OCR 识别
+        for angle in angles:
+            rotated_image = self.rotate_image(image, angle)
+            result = self.ocr.ocr(rotated_image, cls=True)
+            has_barcode, barcode = self.check_and_return_string(result)
+            if has_barcode:
+                time2 = time.time()
+                return barcode
+        else:
+            time2 = time.time()
+            return None
+
+class image_handle:
+    
+    @staticmethod
+    def order_points(points):
+        # 初始化坐标点
+        rect = np.zeros((4, 2), dtype="float32")
+        
+        # 顶点和
+        s = points.sum(axis=1)
+        rect[0] = points[np.argmin(s)]
+        rect[2] = points[np.argmax(s)]
+        
+        # 顶点差
+        diff = np.diff(points, axis=1)
+        rect[1] = points[np.argmin(diff)]
+        rect[3] = points[np.argmax(diff)]
+        
+        return rect
+    
+    @staticmethod
+    def draw_box_and_text(image, bbox, text):
+        # 使用 CV2 绘制矩形框
+        cv2.rectangle(image, tuple(bbox[0]), tuple(bbox[2]), (0, 255, 0), 2)
+        
+        # 转换图像从 BGR 到 RGB
+        rgb_image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)
+        pil_image = Image.fromarray(rgb_image)
+        
+        # 创建 ImageDraw 对象
+        draw = ImageDraw.Draw(pil_image)
+        
+        # 设置字体
+        font = ImageFont.truetype(ttf_path, 10, encoding="utf-8")
+        
+        # 计算文本位置（在框的上方）
+        text_position = (bbox[0][0], bbox[0][1] - 15)
+        
+        # 绘制文本
+        draw.text(text_position, text, (255, 0, 0), font=font)
+        
+        # 将图像转回 BGR 颜色空间
+        result_image = cv2.cvtColor(np.array(pil_image), cv2.COLOR_RGB2BGR)
+        
+        return result_image
+
+    @staticmethod
+    def crop_image_second(image, points):
+        time1 = time.time()
+        # 获取坐标点顺序
+        rect = image_handle.order_points(points)
+        (tl, tr, br, bl) = rect
+
+        # 计算新图像的宽度
+        widthA = np.sqrt(((br[0] - bl[0]) ** 2) + ((br[1] - bl[1]) ** 2))
+        widthB = np.sqrt(((tr[0] - tl[0]) ** 2) + ((tr[1] - tl[1]) ** 2))
+        maxWidth = max(int(widthA), int(widthB))
+
+        # 计算新图像的高度
+        heightA = np.sqrt(((tr[0] - br[0]) ** 2) + ((tr[1] - br[1]) ** 2))
+        heightB = np.sqrt(((tl[0] - bl[0]) ** 2) + ((tl[1] - bl[1]) ** 2))
+        maxHeight = max(int(heightA), int(heightB))
+
+        # 构建目标点
+        dst = np.array([
+            [0, 0],
+            [maxWidth - 1, 0],
+            [maxWidth - 1, maxHeight - 1],
+            [0, maxHeight - 1]], dtype="float32")
+
+        # 计算透视变换矩阵
+        M = cv2.getPerspectiveTransform(rect, dst)
+        
+        # 执行透视变换
+        warped = cv2.warpPerspective(image, M, (maxWidth, maxHeight))
+        time2 = time.time()
+        # print(f"crop_second time: {time2 - time1} s!!")
+        return warped
+    
+    @staticmethod
+    def clean_old_images(directory, lifetime):
+        current_time = time.time()
+        for filename in os.listdir(directory):
+            file_path = os.path.join(directory, filename)
+            if os.path.isfile(file_path):
+                file_creation_time = os.path.getctime(file_path)
+                if current_time - file_creation_time > lifetime:
+                    os.remove(file_path)
+                    # print(f"Deleted old image: {file_path}")
+    @staticmethod
+    def correct_image_orientation(image):
+        try:
+            for orientation in ExifTags.TAGS.keys():
+                if ExifTags.TAGS[orientation] == 'Orientation':
+                    break
+
+            exif = image._getexif()
+            if exif is not None:
+                orientation = exif.get(orientation, 1)
+                if orientation == 3:
+                    image = image.rotate(180, expand=True)
+                elif orientation == 6:
+                    image = image.rotate(270, expand=True)
+                elif orientation == 8:
+                    image = image.rotate(90, expand=True)
+        except (AttributeError, KeyError, IndexError):
+            # cases: image don't have getexif
+            pass
+        return image
+
+
+class Compare:
+
+    @staticmethod
+    def remove_whitespace(s):
+        # 使用正则表达式匹配所有空白字符并替换为空字符串
+        return re.sub(r'\s+', '', s)
+    
+    @staticmethod
+    def replace_bracket(s):
+        if s:
+            # 定义一个正则表达式模式，匹配所有类型的括号
+            pattern = r'[\(\)\[\]\{\}\（\）\【\】\{\}〈〉《》]'
+            # 使用空字符串替换所有匹配到的括号
+            result = re.sub(pattern, '', s)
+            return result
+        else:
+            return s
+    @staticmethod
+    def convert_new_dic(dataset):
+        new_dic = {}
+        special_keys = {
+            '保养说明': ('K\d{3}', ''),
+            '温馨提示': ('H\d{3}', '')
+        }
+
+        for key, value in dataset.items():
+            if value:
+                value = Compare.remove_whitespace(value)
+                for special_key, (pattern, initial) in special_keys.items():
+                    if key.startswith(special_key):
+                        value = re.sub(pattern, '', value)
+                        new_dic[special_key] = new_dic.get(special_key, initial) + value
+                        break
+                else:
+                    new_dic[key] = value
+            elif key.startswith('温馨提示') and '温馨提示' not in new_dic:
+                new_dic['温馨提示'] = ''
+
+        return new_dic
+
+    @staticmethod
+    def en_to_zh_punctuation(s):
+        # 英文标点到中文标点的映射
+        en_to_zh_map = {
+            ',': '，', ';': '；', '：': ':', '~': '～', '（': '(', '）': ')'
+        }
+        # 替换所有出现的英文标点为对应的中文标点
+        for en, zh in en_to_zh_map.items():
+            s = s.replace(en, zh)
+        return s
+
+    @staticmethod
+    def compare(ocr_result, dataset):
+        extra = set(['XXS', 'XS', 'S', 'M', 'L', 'XL', 'XXL'])
+        desc = set(['Z', 'G'])
+        keyword = set(dataset.keys())
+        dataset = Compare.convert_new_dic(dataset)
+        dic = {}
+        i = 0
+
+        while i < len(ocr_result):
+            text = ocr_result[i][0]
+            
+            if '零售价' in text:
+                dic['零售价'] = next((l[0].strip('￥') for l in ocr_result if l[0].startswith('￥')), '')
+            elif '：' in text:
+                key, value = text.split('：', 1)
+                if key in keyword:
+                    if value:
+                        dic[key] = Compare.en_to_zh_punctuation(value)
+                    else:
+                        j = i + 1
+                        while j < len(ocr_result) and ocr_result[j][0].split('：')[0] not in keyword | extra | desc:
+                            value += ocr_result[j][0]
+                            j += 1
+                        dic[key] = Compare.en_to_zh_punctuation(re.sub('·', '', value))
+                        i = j - 1
+            elif text in extra:
+                dic['尺码'] = text
+            elif text[0] in ['(', '（'] and text[1].isalpha() and '合格证' not in text:
+                dic['desc11' if 'desc11' not in dic and u'\u4e00' <= text[1] <= u'\u9fa5' else 'desc5'] = text
+            elif text in desc:
+                dic['desc4'] = text
+            
+            i += 1
+        if dic.get('desc5','') == dataset.get('desc11', ''):
+            dic['desc5'] = dataset.get('desc11', '')
+        if dic.get('desc11','') == dataset.get('desc5', ''):
+            dic['desc11'] = dataset.get('desc5', '')
+
+        dataset['价钱下的产品名'] = dataset.pop('desc5') if 'desc5' in dataset else ''
+        dataset['备注二'] = dataset.pop('desc4') if 'desc4' in dataset else ''
+        dataset['齐码'] = dataset.pop('desc11') if 'desc11' in dataset else ''
+
+        dic['价钱下的产品名'] = dic.pop('desc5') if 'desc5' in dic else ''
+        dic['备注二'] = dic.pop('desc4') if 'desc4' in dic else ''
+        dic['齐码'] = dic.pop('desc11') if 'desc11' in dic else ''
+
+        log = [
+            {'name': key, 'value': [dic[key], dataset.get(key, '')]}
+            for key in dic
+            if Compare.replace_bracket(dic[key]) != Compare.replace_bracket(dataset.get(key, ''))
+        ]
+
+        return dataset, log
+
+class sql_product:
+    @staticmethod
+    def size_information(matio_id, color_id):
+        item_id = matio_id.split('-')[0]
+        
+        with pymysql.connect(**sql_config) as conn:
+            with conn.cursor(pymysql.cursors.DictCursor) as cur:
+                cur.execute(size_first_sql.format(matio_id=matio_id, color_id=color_id))
+                production = [
+                    {**({d['SIZE']: d['hao_type'].split('(' if '(' in d['hao_type'] else '（')[1].strip(')）').strip()}),
+                    **{k: d[k] for k in ['size_id', 'clothes_type', 'language']}}
+                    for d in cur.fetchall()
+                ]
+                cur.execute(size_check_sql.format(item_id=item_id))
+                base_size = {d["size_code"]: d['sizes'] for d in cur.fetchall()}
+        logs = [d for d in production if any(d[k] != base_size.get(k) for k in d if k not in ['size_id', 'clothes_type', 'language'])]
+        return '0' if logs else '1', logs or "'None'"
+
+    @staticmethod
+    def color_information(matio_id):
+        with pymysql.connect(**sql_config) as mysql_conn, pymssql.connect(**color_config) as sql_server_conn:
+            with mysql_conn.cursor(pymysql.cursors.DictCursor) as mysql_cur, sql_server_conn.cursor(as_dict=True) as sql_server_cur:
+                mysql_cur.execute(color_sql.format(matio_id=matio_id))
+                water_mark_result = mysql_cur.fetchall()
+                
+                if not water_mark_result:
+                    return '1', "'None'"
+
+                sql_server_cur.execute(color_code_sql.format(code=water_mark_result[0]['color_id']))
+                base_result = sql_server_cur.fetchone()
+
+        logs = [
+            {k: d[k] for k in ('color_name', 'language', 'color_id', 'clothes_type')}
+            for d in water_mark_result
+            if d['color_id'] != base_result['code'] or d['color_name'] != base_result['name']
+        ]
+
+        return '0' if logs else '1', logs or "'None'"
+
+    @staticmethod
+    def sql_information(barcode, barcode_type='RFID吊牌', matio_id=''):
+
+        conn = mysql.connector.connect(**sql_config)
+
+        if barcode_type == 'RFID吊牌':
+            df_1 = pd.read_sql_query(QUERY_MATIO_ID.format(barcode=barcode), conn)
+            if df_1.empty:
+                conn.close()
+                return {}, matio_id, '', datetime.datetime.now().strftime("%Y/%m/%d %H:%M:%S")
+            matio_id, color_id, size = df_1.iloc[0][['matio_id', 'COLOR_ID', 'SIZE']]
+            query_condition = query_condition_rfid
+        elif barcode_type == '普通吊牌':
+            color_id, size = barcode[9:12], barcode[12:]
+            query_condition = query_condition_normal
+        print(barcode)
+        print(matio_id)
+        print(color_id)
+        print(size)
+        conn.close()
+        query_second = QUERY_BASE.format(matio_id=matio_id, color_id=color_id, size=size) + " " + query_condition
+        print(query_second)
+        with pymysql.connect(**sql_config) as connn:
+            curr = connn.cursor()
+            curr.execute(query_second)
+            columns = [desc[0] for desc in curr.description]
+            res = curr.fetchall()
+
+        sql_time = datetime.datetime.now().strftime("%Y/%m/%d %H:%M:%S")
+
+        return (dict(zip(columns, res[0])), matio_id, color_id, sql_time) if res else ({}, matio_id, color_id, sql_time)
+    
+    @staticmethod
+    def sql_matio_id(prefix_code):
+        with pymysql.connect(**sql_config) as conn:
+            with conn.cursor() as cur:
+                cur.execute(matio_id_sql.format(prefix_code=prefix_code))
+                return [row[0] for row in cur.fetchall()]
+
+
+
+def get_time(strs):
+    dicts = {}
+    dicts['year'] = int(strs.split(' ')[0].split('-')[0])
+    dicts['month'] = int(strs.split(' ')[0].split('-')[1])
+    dicts['day'] = int(strs.split(' ')[0].split('-')[2])
+    dicts['hour'] = int(strs.split(' ')[1].split(':')[0])
+    dicts['minute'] = int(strs.split(' ')[1].split(':')[1])
+    dicts['second'] = int(strs.split(' ')[1].split(':')[2])
+    return dicts
+
+
+
+
+
+
+if __name__ == '__main__':
+    data_result, matio_id, color_id, _ = sql_product.sql_information(barcode='1CCCAB05005WS', barcode_type='普通吊牌', matio_id='1CCCAB050-01')
+    print(data_result)
+    # size_compare_flag, size_compare_logs = size_information(matio_id, color_id)
+
+
+    # result = sql_matio_id('1C3JAA57000BL')
+    # print(result)
+    # from config import Matio
+    # x = Matio()
+    # print(x)
+    pass