jiachenlong/backend/app/routers/ocr.py

# ocr - OCR识别路由
# Version: 1.2.75
# 更新：

import os
# 更新：
import uuid
# 更新：
import base64
# 更新：
import httpx
# 更新：
from fastapi import APIRouter, Depends, HTTPException, UploadFile, File
# 更新：
from sqlalchemy.orm import Session
# 更新：
from app.core.database import get_db
# 更新：
from app.core.auth import get_current_user
# 更新：
from app.models.models import User
# 更新：

# 更新：
router = APIRouter(prefix="/api/ocr", tags=["OCR 识别"])
# 更新：

# 更新：
# 阿里云 DashScope API 配置
# 更新：
DASHSCOPE_API_KEY = os.getenv("DASHSCOPE_API_KEY", "sk-9389024a37da4f7bb455ac9a6b28776f")
# 更新：

# 更新：
# 阿里云 OSS 配置
# 更新：
OSS_CONFIG = {
# 更新：
    "access_key_id": os.getenv("OSS_ACCESS_KEY_ID", "LTAI5t6HUnpFBLEK9194kPVG"),
# 更新：
    "access_key_secret": os.getenv("OSS_ACCESS_KEY_SECRET", "LEr4Q8yRxb8D5b24cKfCwlt4MMoke1"),
# 更新：
    "bucket_name": "jiachenlong-oss",
# 更新：
    "endpoint": "oss-cn-hangzhou.aliyuncs.com",
# 更新：
    "public_url": "https://jiachenlong-oss.oss-cn-hangzhou.aliyuncs.com"
# 更新：
}
# 更新：

# 更新：
# 临时上传目录（用于OCR识别本地备选）
# 更新：
UPLOAD_DIR = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(__file__))), "uploads", "temp")
# 更新：
os.makedirs(UPLOAD_DIR, exist_ok=True)
# 更新：

# 更新：

# 更新：
def get_oss_path(file_type: str, user_id: str = None, collection_id: str = None, filename: str = None):
# 更新：
    """生成OSS路径 - 按年/月/日分类"""
# 更新：
    from datetime import datetime
# 更新：
    now = datetime.now()
# 更新：
    year = now.strftime("%Y")
# 更新：
    month = now.strftime("%m")
# 更新：
    day = now.strftime("%d")
# 更新：
    
# 更新：
    if file_type == "temp":
# 更新：
        # 临时文件: temp/{year}/{month}/{day}/{uuid}.{ext}
# 更新：
        import uuid
# 更新：
        unique_id = str(uuid.uuid4())
# 更新：
        ext = filename.split('.')[-1] if filename and '.' in filename else 'jpg'
# 更新：
        return f"temp/{year}/{month}/{day}/{unique_id}.{ext}", unique_id
# 更新：
    
# 更新：
    elif file_type == "collection":
# 更新：
        # 藏品文件: collections/{user_id}/{year}/{collection_id}/{filename}
# 更新：
        if not user_id or not collection_id:
# 更新：
            raise ValueError("user_id and collection_id required for collection")
# 更新：
        return f"collections/{user_id}/{year}/{collection_id}/{filename}"
# 更新：
    
# 更新：
    elif file_type == "avatar":
# 更新：
        # 头像: avatars/{user_id}/avatar.{ext}
# 更新：
        if not user_id:
# 更新：
            raise ValueError("user_id required for avatar")
# 更新：
        ext = filename.split('.')[-1] if filename and '.' in filename else 'jpg'
# 更新：
        return f"avatars/{user_id}/avatar.{ext}"
# 更新：
    
# 更新：
    return None
# 更新：

# 更新：

# 更新：
# 上传图片到OSS - 使用服务层（带压缩）
# 更新：
from app.services.oss import upload_to_oss as oss_upload
# 更新：

# 更新：
def upload_to_oss(file_data, oss_key):
# 更新：
    """上传文件到阿里云OSS（带自动压缩）"""
# 更新：
    return oss_upload(file_data, oss_key)
# 更新：

# 更新：
# 专业提示词
# 更新：
PROFESSIONAL_PROMPT = """你是一名专业的人民币生肖纪念钞鉴定专家。请严格按照以下步骤分析这张图片，并提取准确信息。
# 更新：

# 更新：
【识别流程】
# 更新：
1. 判断类型是否评级钞：首先确认是否为裸钞还是评级钞（有封装盒和标签）
# 更新：
2. 验证纪念钞特征：对照生肖纪念钞特征进行确认
# 更新：
3. 验证评级类型：有'标十'字眼的为标十，有'百连'字眼的为标百，其他为单张
# 更新：
4. 提取信息：仔细阅读标签上的所有文字内容
# 更新：

# 更新：
【版别格式要求】
# 更新：
只需要：年份 + 属相，例如：
# 更新：
- 2024 龙
# 更新：
- 2025 蛇
# 更新：
- 2026 马
# 更新：

# 更新：
【输出要求】
# 更新：
严格按照以下格式输出，每个字段必须填写具体值：
# 更新：
✅ 1 发行机构：中国人民银行
# 更新：
✅ 2 发行版别：2024 龙
# 更新：
✅ 3 面额：贰拾圆
# 更新：
✅ 4 是否评级：是/否
# 更新：
✅ 5 封装类型：裸钞/单张/标十/标百
# 更新：
✅ 6 冠字序号：J0xxxxxxxx
# 更新：
✅ 7 评级机构：ACG/PCGS/PMG
# 更新：
✅ 8 评级分数：67/68/69
# 更新：
✅ 9 是否三星：是/否
# 更新：
✅ 10 特殊标识：金山标/天马标/红绳版等
# 更新：
✅ 11 号码特征：金山号 2 张，天马号 3 张等
# 更新：

# 更新：
现在请仔细分析提供的图片，按上述格式输出结果。"""
# 更新：

# 更新：

# 更新：
@router.post("/recognize")
# 更新：
async def recognize_image(
# 更新：
    image: UploadFile = File(...),
# 更新：
    current_user: User = Depends(get_current_user),
# 更新：
    db: Session = Depends(get_db)
# 更新：
):
# 更新：
    """OCR 图片识别 - 识别后自动保存图片到OSS临时目录"""
# 更新：
    try:
# 更新：
        # 读取图片数据
# 更新：
        image_data = await image.read()
# 更新：
        image_base64 = base64.b64encode(image_data).decode('utf-8')
# 更新：
        
# 更新：
        # 生成OSS临时路径: temp/{year}/{month}/{day}/{uuid}.{ext}
# 更新：
        oss_key, temp_id = get_oss_path("temp", filename=image.filename)
# 更新：
        
# 更新：
        # 初始化temp_path为空
# 更新：
        temp_path = None
# 更新：
        
# 更新：
        # 上传到OSS
# 更新：
        try:
# 更新：
            image_url = upload_to_oss(image_data, oss_key)
# 更新：
        except Exception as oss_err:
# 更新：
            # OSS失败时保存到本地作为备选
# 更新：
            temp_path = os.path.join(UPLOAD_DIR, oss_key.split('/')[-1])
# 更新：
            os.makedirs(os.path.dirname(temp_path), exist_ok=True)
# 更新：
            with open(temp_path, 'wb') as f:
# 更新：
                f.write(image_data)
# 更新：
            image_url = f"/uploads/temp/{oss_key.split('/')[-1]}"
# 更新：
        
# 更新：
        headers = {
# 更新：
            "Authorization": f"Bearer {DASHSCOPE_API_KEY}",
# 更新：
            "Content-Type": "application/json"
# 更新：
        }
# 更新：
        
# 更新：
        # 阿里云 DashScope API 格式 (qwen-vl-plus 视觉模型)
# 更新：
        payload = {
# 更新：
            "model": "qwen-vl-plus",
# 更新：
            "input": {
# 更新：
                "messages": [{
# 更新：
                    "role": "user",
# 更新：
                    "content": [
# 更新：
                        {
# 更新：
                            "image": f"data:{image.content_type};base64,{image_base64}"
# 更新：
                        },
# 更新：
                        {
# 更新：
                            "text": PROFESSIONAL_PROMPT
# 更新：
                        }
# 更新：
                    ]
# 更新：
                }]
# 更新：
            },
# 更新：
            "parameters": {
# 更新：
                "max_tokens": 1000
# 更新：
            }
# 更新：
        }
# 更新：
        
# 更新：
        async with httpx.AsyncClient(timeout=60.0) as client:
# 更新：
            response = await client.post(
# 更新：
                "https://dashscope.aliyuncs.com/api/v1/services/aigc/multimodal-generation/generation",
# 更新：
                json=payload,
# 更新：
                headers=headers
# 更新：
            )
# 更新：
            
# 更新：
            if response.status_code != 200:
# 更新：
                # 识别失败，删除临时文件
# 更新：
                if temp_path and os.path.exists(temp_path):
# 更新：
                    os.remove(temp_path)
# 更新：
                raise HTTPException(status_code=500, detail=f"OCR API 调用失败：{response.text[:200]}")
# 更新：
            
# 更新：
            ocr_result = response.json()
# 更新：
            text_content = ""
# 更新：
            # 新版API返回格式
# 更新：
            if "output" in ocr_result and "choices" in ocr_result["output"]:
# 更新：
                choices = ocr_result["output"]["choices"]
# 更新：
                if choices and len(choices) > 0:
# 更新：
                    content = choices[0].get("message", {}).get("content", [])
# 更新：
                    if content and len(content) > 0:
# 更新：
                        text_content = content[0].get("text", "")
# 更新：
            
# 更新：
            fields = extract_fields(text_content)
# 更新：
            
# 更新：
            # 更新用户AI识别次数
# 更新：
            current_user.f99_95_ai_count = (current_user.f99_95_ai_count or 0) + 1
# 更新：
            db.commit()
# 更新：
            
# 更新：
            # 返回识别结果和临时图片路径
# 更新：
            return {
# 更新：
                "success": True,
# 更新：
                "text": text_content,
# 更新：
                "fields": fields,
# 更新：
                "aiCount": current_user.f99_95_ai_count,
# 更新：
                "temp_image": {
# 更新：
                    "id": temp_id,
# 更新：
                    "filename": oss_key.split('/')[-1],
# 更新：
                    "path": image_url,
# 更新：
                    "original_name": image.filename,
# 更新：
                    "is_oss": image_url.startswith("https://")
# 更新：
                }
# 更新：
            }
# 更新：
            
# 更新：
    except Exception as e:
# 更新：
        import traceback
# 更新：
        error_detail = f"识别失败：{str(e)}\n{traceback.format_exc()}"
# 更新：
        raise HTTPException(status_code=500, detail=error_detail)
# 更新：

# 更新：

# 更新：
def extract_fields(text: str) -> dict:
# 更新：
    """从 OCR 文本中提取字段 - 直接返回 AI 识别结果"""
# 更新：
    import re
# 更新：
    fields = {}
# 更新：
    
# 更新：
    # 解析结构化输出
# 更新：
    patterns = {
# 更新：
        'issuer': r'✅.*?1.*?发行机构.*?[:：]\s*(.+?)(?:\n|$)',
# 更新：
        'version': r'✅.*?2.*?发行版别.*?[:：]\s*(.+?)(?:\n|$)',
# 更新：
        'denomination': r'✅.*?3.*?面额.*?[:：]\s*(.+?)(?:\n|$)',
# 更新：
        'is_graded_text': r'✅.*?4.*?是否评级.*?[:：]\s*(.+?)(?:\n|$)',
# 更新：
        'packaging': r'✅.*?5.*?封装类型.*?[:：]\s*(.+?)(?:\n|$)',
# 更新：
        'prefix_serial': r'✅.*?6.*?冠字序号.*?[:：]\s*(.+?)(?:\n|$)',
# 更新：
        'grading_company': r'✅.*?7.*?评级机构.*?[:：]\s*(.+?)(?:\n|$)',
# 更新：
        'grading_score': r'✅.*?8.*?评级分数.*?[:：]\s*(.+?)(?:\n|$)',
# 更新：
        'three_star_text': r'✅.*?9.*?是否三星.*?[:：]\s*(.+?)(?:\n|$)',
# 更新：
        'special_mark': r'✅.*?10.*?特殊标识.*?[:：]\s*(.+?)(?:\n|$)',
# 更新：
        'serial_feature': r'✅.*?11.*?号码特征.*?[:：]\s*(.+?)(?:\n|$)'
# 更新：
    }
# 更新：
    
# 更新：
    for field, pattern in patterns.items():
# 更新：
        match = re.search(pattern, text, re.IGNORECASE | re.DOTALL)
# 更新：
        if match:
# 更新：
            value = match.group(1).strip()
# 更新：
            # 保留所有值，包括"无"和"未识别"，让前端处理
# 更新：
            fields[field] = value
# 更新：
    
# 更新：
    # 处理是否评级
# 更新：
    if 'is_graded_text' in fields:
# 更新：
        fields['is_graded'] = '是' in fields.pop('is_graded_text')
# 更新：
    
# 更新：
    # 处理是否三星
# 更新：
    if 'three_star_text' in fields:
# 更新：
        fields['three_star'] = '是' in fields.pop('three_star_text')
# 更新：
    
# 更新：
    # 简化版别字段（2024 龙年贺岁纪念钞（标十） → 2024 龙）
# 更新：
    if 'version' in fields:
# 更新：
        version = fields['version']
# 更新：
        # 提取年份和生肖
# 更新：
        year_match = re.search(r'(20\d{2})', version)
# 更新：
        animal = ''
# 更新：
        if '龙' in version:
# 更新：
            animal = '龙'
# 更新：
        elif '蛇' in version:
# 更新：
            animal = '蛇'
# 更新：
        elif '马' in version:
# 更新：
            animal = '马'
# 更新：
        elif '羊' in version:
# 更新：
            animal = '羊'
# 更新：
        elif '猴' in version:
# 更新：
            animal = '猴'
# 更新：
        elif '鸡' in version:
# 更新：
            animal = '鸡'
# 更新：
        elif '狗' in version:
# 更新：
            animal = '狗'
# 更新：
        elif '猪' in version:
# 更新：
            animal = '猪'
# 更新：
        elif '鼠' in version:
# 更新：
            animal = '鼠'
# 更新：
        elif '牛' in version:
# 更新：
            animal = '牛'
# 更新：
        elif '虎' in version:
# 更新：
            animal = '虎'
# 更新：
        elif '兔' in version:
# 更新：
            animal = '兔'
# 更新：
        
# 更新：
        if year_match and animal:
# 更新：
            fields['version'] = f"{year_match.group(1)}{animal}"
# 更新：
    
# 更新：
    return fields
# 更新：

# 更新：

# 更新：
@router.post("/claim-temp-image")
# 更新：
async def claim_temp_image(
# 更新：
    temp_id: str,
# 更新：
    collection_id: str,
# 更新：
    current_user: User = Depends(get_current_user),
# 更新：
    db: Session = Depends(get_db)
# 更新：
):
# 更新：
    """将临时图片移动到正式藏品目录 - 按用户/年/藏品ID分类"""
# 更新：
    from app.models.models import Collection, CollectionImage
# 更新：
    from datetime import datetime
# 更新：
    
# 更新：
    # 验证藏品是否存在
# 更新：
    collection = db.query(Collection).filter(
# 更新：
        Collection.f99_90_id == collection_id,
# 更新：
        Collection.f99_91_user_id == current_user.f99_90_id
# 更新：
    ).first()
# 更新：
    
# 更新：
    if not collection:
# 更新：
        raise HTTPException(status_code=404, detail="藏品不存在")
# 更新：
    
# 更新：
    # 生成正式文件名和OSS路径: collections/{user_id}/{year}/{collection_id}/{filename}
# 更新：
    code = collection.f01_02_code or "0000"
# 更新：
    prefix = collection.f02_10_prefix_serial or ""
# 更新：
    username = current_user.f01_01_name
# 更新：
    import time
# 更新：
    final_filename = f"{username}-{code}-{prefix}-{int(time.time())}.jpg"
# 更新：
    oss_key = get_oss_path("collection", user_id=current_user.f99_90_id, collection_id=collection_id, filename=final_filename)
# 更新：
    
# 更新：
    # 尝试从OSS获取临时图片 - 尝试多种扩展名和日期路径
# 更新：
    temp_extensions = ['jpg', 'jpeg', 'png', 'gif', 'JPG', 'JPEG', 'PNG', 'GIF']
# 更新：
    temp_content = None
# 更新：
    found_key = None
# 更新：
    
# 更新：
    # 尝试最近7天的路径
# 更新：
    from datetime import timedelta
# 更新：
    for i in range(7):
# 更新：
        date = datetime.now() - timedelta(days=i)
# 更新：
        year = date.strftime("%Y")
# 更新：
        month = date.strftime("%m")
# 更新：
        day = date.strftime("%d")
# 更新：
        
# 更新：
        for ext in temp_extensions:
# 更新：
            try:
# 更新：
                temp_oss_key = f"temp/{year}/{month}/{day}/{temp_id}.{ext}"
# 更新：
                import oss2
# 更新：
                auth = oss2.Auth(OSS_CONFIG["access_key_id"], OSS_CONFIG["access_key_secret"])
# 更新：
                bucket = oss2.Bucket(auth, OSS_CONFIG["endpoint"], OSS_CONFIG["bucket_name"])
# 更新：
                temp_content = bucket.get_object(temp_oss_key).read()
# 更新：
                found_key = temp_oss_key
# 更新：
                break
# 更新：
            except:
# 更新：
                continue
# 更新：
        if temp_content:
# 更新：
            break
# 更新：
    
# 更新：
    if temp_content:
# 更新：
        # 上传到正式目录
# 更新：
        bucket.put_object(oss_key, temp_content)
# 更新：
        
# 更新：
        # 删除临时图片
# 更新：
        try:
# 更新：
            bucket.delete_object(found_key)
# 更新：
        except:
# 更新：
            pass
# 更新：
        
# 更新：
        # OSS URL
# 更新：
        image_path = f"{OSS_CONFIG['public_url']}/{oss_key}"
# 更新：
        
# 更新：
    else:
# 更新：
        # OSS失败，使用本地文件
# 更新：
        temp_path = None
# 更新：
        for ext in temp_extensions:
# 更新：
            temp_path = os.path.join(UPLOAD_DIR, f"{temp_id}.{ext}")
# 更新：
            if os.path.exists(temp_path):
# 更新：
                break
# 更新：
        
# 更新：
        if not temp_path or not os.path.exists(temp_path):
# 更新：
            raise HTTPException(status_code=404, detail="临时图片不存在或已过期")
# 更新：
        
# 更新：
        # 保存到本地
# 更新：
        collection_dir = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(__file__))), "uploads", "collections")
# 更新：
        os.makedirs(collection_dir, exist_ok=True)
# 更新：
        
# 更新：
        new_path = os.path.join(collection_dir, final_filename)
# 更新：
        import shutil
# 更新：
        shutil.move(temp_path, new_path)
# 更新：
        image_path = f"uploads/collections/{final_filename}"
# 更新：
    
# 更新：
    # 创建图片记录
# 更新：
    image_record = CollectionImage(
# 更新：
        id=str(uuid.uuid4()),
# 更新：
        collection_id=collection.f99_90_id,
# 更新：
        filename=final_filename,
# 更新：
        original_name=temp_id,
# 更新：
        path=image_path
# 更新：
    )
# 更新：
    db.add(image_record)
# 更新：
    db.commit()
# 更新：
    
# 更新：
    return {
# 更新：
        "success": True,
# 更新：
        "image": {
# 更新：
            "id": image_record.id,
# 更新：
            "filename": image_record.filename,
# 更新：
            "path": image_record.path
# 更新：
        }
# 更新：
    }
# 更新：