151 lines
4.1 KiB
Python
151 lines
4.1 KiB
Python
"""二维码相关接口 - API层"""
|
||
from fastapi import APIRouter, File, UploadFile
|
||
|
||
from app.schemas import ResultDomain
|
||
from app.services.workcase.qrcode import QrCodeService
|
||
|
||
router = APIRouter()
|
||
|
||
# 初始化服务
|
||
qrcode_service = QrCodeService()
|
||
|
||
|
||
@router.post(
|
||
"/generate",
|
||
response_model=ResultDomain[dict],
|
||
summary="生成二维码",
|
||
description="根据内容生成二维码"
|
||
)
|
||
async def generate_qrcode(
|
||
content: str,
|
||
size: int = 300,
|
||
error_correction: str = "H"
|
||
) -> ResultDomain[dict]:
|
||
"""
|
||
生成二维码
|
||
|
||
- **content**: 二维码内容
|
||
- **size**: 图片大小(像素,100-2000)
|
||
- **error_correction**: 纠错级别
|
||
- L: 7% 容错
|
||
- M: 15% 容错
|
||
- Q: 25% 容错
|
||
- H: 30% 容错 (推荐)
|
||
"""
|
||
result = await qrcode_service.generate_qrcode(
|
||
content=content,
|
||
size=size,
|
||
error_correction=error_correction
|
||
)
|
||
|
||
if result["success"]:
|
||
return ResultDomain.success(message="生成成功", data=result)
|
||
else:
|
||
return ResultDomain.fail(message=result.get("error", "生成失败"))
|
||
|
||
|
||
@router.post(
|
||
"/parse",
|
||
response_model=ResultDomain[dict],
|
||
summary="解析二维码",
|
||
description="解析二维码图片内容(支持URL、base64)"
|
||
)
|
||
async def parse_qrcode(
|
||
image_source: str,
|
||
strategy: str = "auto"
|
||
) -> ResultDomain[dict]:
|
||
"""
|
||
解析二维码
|
||
|
||
- **image_source**: 图片来源
|
||
- URL: http://... 或 https://...
|
||
- base64: data:image/...;base64,...
|
||
- 本地路径: /path/to/image.png
|
||
- **strategy**: 预处理策略
|
||
- basic: 基础模式,仅尝试原图和灰度图
|
||
- auto: 自动模式,尝试多种预处理方法 (推荐)
|
||
- enhanced: 增强模式,使用更多预处理技术
|
||
- all: 全部模式,尝试所有可能的预处理方法(包括多尺度)
|
||
"""
|
||
result = await qrcode_service.parse_qrcode(
|
||
image_source=image_source,
|
||
strategy=strategy
|
||
)
|
||
|
||
if result["success"]:
|
||
return ResultDomain.success(message="解析成功", data=result)
|
||
else:
|
||
return ResultDomain.fail(
|
||
message=result.get("error", "解析失败"),
|
||
data={"total_attempts": result.get("total_attempts", 0)}
|
||
)
|
||
|
||
|
||
@router.post(
|
||
"/parse-file",
|
||
response_model=ResultDomain[dict],
|
||
summary="解析二维码文件",
|
||
description="通过文件上传解析二维码"
|
||
)
|
||
async def parse_qrcode_file(
|
||
file: UploadFile = File(...),
|
||
strategy: str = "auto"
|
||
) -> ResultDomain[dict]:
|
||
"""
|
||
解析二维码文件上传
|
||
|
||
- **file**: 二维码图片文件(支持 png/jpg/jpeg/bmp 等格式)
|
||
- **strategy**: 预处理策略 (basic/auto/enhanced/all)
|
||
"""
|
||
# 读取文件内容
|
||
content = await file.read()
|
||
|
||
# 提取文件类型
|
||
if file.content_type:
|
||
file_type = file.content_type.split("/")[-1]
|
||
else:
|
||
# 从文件名提取扩展名
|
||
file_type = file.filename.split(".")[-1] if file.filename else "png"
|
||
|
||
# 调用服务
|
||
result = await qrcode_service.parse_qrcode_from_file(
|
||
file_content=content,
|
||
file_type=file_type,
|
||
strategy=strategy
|
||
)
|
||
|
||
if result["success"]:
|
||
return ResultDomain.success(message="解析成功", data=result)
|
||
else:
|
||
return ResultDomain.fail(
|
||
message=result.get("error", "解析失败"),
|
||
data={"total_attempts": result.get("total_attempts", 0)}
|
||
)
|
||
|
||
|
||
@router.post(
|
||
"/validate",
|
||
response_model=ResultDomain[dict],
|
||
summary="验证二维码内容",
|
||
description="验证内容是否适合生成二维码"
|
||
)
|
||
async def validate_qrcode_content(
|
||
content: str,
|
||
max_length: int = 2953
|
||
) -> ResultDomain[dict]:
|
||
"""
|
||
验证二维码内容
|
||
|
||
- **content**: 要验证的内容
|
||
- **max_length**: 最大长度(字节)
|
||
"""
|
||
result = qrcode_service.validate_qrcode_content(content, max_length)
|
||
|
||
if result["valid"]:
|
||
return ResultDomain.success(
|
||
message="内容有效",
|
||
data={"length": result["length"]}
|
||
)
|
||
else:
|
||
return ResultDomain.fail(message=result.get("error", "内容无效"))
|