346 lines
12 KiB
Python
346 lines
12 KiB
Python
# 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")
|
||
|
||
# 阿里云 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 upload_to_oss(file_data, filename, bucket_name=None):
|
||
"""上传文件到阿里云OSS"""
|
||
import oss2
|
||
bucket_name = bucket_name or OSS_CONFIG["bucket_name"]
|
||
|
||
auth = oss2.Auth(OSS_CONFIG["access_key_id"], OSS_CONFIG["access_key_secret"])
|
||
bucket = oss2.Bucket(auth, OSS_CONFIG["endpoint"], bucket_name)
|
||
|
||
# 上传文件
|
||
result = bucket.put_object(filename, file_data)
|
||
|
||
if result.status == 200:
|
||
return f"{OSS_CONFIG['public_url']}/{filename}"
|
||
else:
|
||
raise Exception(f"OSS上传失败: {result.status}")
|
||
|
||
# 专业提示词
|
||
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')
|
||
|
||
# 生成临时文件名（识别后保存到OSS）
|
||
temp_id = str(uuid.uuid4())
|
||
ext = image.filename.split('.')[-1] if '.' in image.filename else 'jpg'
|
||
oss_key = f"temp/{temp_id}.{ext}"
|
||
|
||
# 上传到OSS
|
||
try:
|
||
image_url = upload_to_oss(image_data, oss_key)
|
||
except Exception as oss_err:
|
||
# OSS失败时保存到本地作为备选
|
||
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)
|
||
image_url = f"/uploads/temp/{temp_filename}"
|
||
|
||
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": 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)
|
||
):
|
||
"""将临时图片移动到正式藏品目录"""
|
||
from app.models.models import Collection, CollectionImage
|
||
|
||
# 验证藏品是否存在
|
||
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="藏品不存在")
|
||
|
||
# 生成正式文件名
|
||
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 = f"collections/{final_filename}"
|
||
|
||
# 尝试从OSS获取临时图片 - 尝试多种扩展名
|
||
temp_extensions = ['jpg', 'jpeg', 'png', 'gif']
|
||
temp_content = None
|
||
found_ext = None
|
||
|
||
for ext in temp_extensions:
|
||
try:
|
||
temp_oss_key = f"temp/{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_ext = ext
|
||
break
|
||
except:
|
||
continue
|
||
|
||
if temp_content:
|
||
# 上传到正式目录
|
||
bucket.put_object(oss_key, temp_content)
|
||
|
||
# 删除临时图片
|
||
try:
|
||
bucket.delete_object(f"temp/{temp_id}.{found_ext}")
|
||
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
|
||
}
|
||
}
|