jiachenlong/backend/app/routers/ocr.py

# OCR 识别路由 - 专业人民币生肖纪念钞鉴定
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")

# 上传目录配置
UPLOAD_DIR = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(__file__))), "uploads", "temp")
os.makedirs(UPLOAD_DIR, exist_ok=True)

# 专业提示词
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 图片识别 - 识别后自动保存图片到临时目录"""
    try:
        # 读取图片数据
        image_data = await image.read()
        image_base64 = base64.b64encode(image_data).decode('utf-8')
        
        # 生成临时文件名（识别后保存到服务器）
        temp_id = str(uuid.uuid4())
        ext = image.filename.split('.')[-1] if '.' in image.filename else 'jpg'
        temp_filename = f"{temp_id}.{ext}"
        temp_path = os.path.join(UPLOAD_DIR, temp_filename)
        
        # 保存图片到临时目录
        with open(temp_path, 'wb') as f:
            f.write(image_data)
        
        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 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)
            
            # 返回识别结果和临时图片路径
            return {
                "success": True,
                "text": text_content,
                "fields": fields,
                "temp_image": {
                    "id": temp_id,
                    "filename": temp_filename,
                    "path": f"uploads/temp/{temp_filename}",
                    "original_name": image.filename
                }
            }
            
    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)
):
    """将临时图片移动到藏品目录"""
    from app.models.models import Collection, CollectionImage
    import shutil
    
    # 验证藏品是否存在
    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="藏品不存在")
    
    # 临时文件路径
    temp_filename = f"{temp_id}.jpg"
    temp_path = os.path.join(UPLOAD_DIR, temp_filename)
    
    # 检查临时文件是否存在
    if not os.path.exists(temp_path):
        # 尝试其他扩展名
        for ext in ['jpeg', 'png', 'gif']:
            temp_filename = f"{temp_id}.{ext}"
            temp_path = os.path.join(UPLOAD_DIR, temp_filename)
            if os.path.exists(temp_path):
                break
    
    if 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)
    
    # 生成新文件名
    code = collection.f01_02_code or "0000"
    prefix = collection.f02_10_prefix_serial or ""
    username = current_user.f01_01_name
    new_filename = f"{username}-{code}-{prefix}.jpg"
    new_path = os.path.join(collection_dir, new_filename)
    
    # 如果文件已存在，添加随机后缀
    if os.path.exists(new_path):
        name, ext = os.path.splitext(new_filename)
        new_filename = f"{name}-{str(uuid.uuid4())[:8]}{ext}"
        new_path = os.path.join(collection_dir, new_filename)
    
    # 移动文件
    shutil.move(temp_path, new_path)
    
    # 创建图片记录
    image_record = CollectionImage(
        id=str(uuid.uuid4()),
        collection_id=collection.f99_90_id,
        filename=new_filename,
        original_name=temp_filename,
        path=f"uploads/collections/{new_filename}"
    )
    db.add(image_record)
    db.commit()
    
    return {
        "success": True,
        "image": {
            "id": image_record.id,
            "filename": image_record.filename,
            "path": image_record.path
        }
    }