forked from erp-dev/erp
217 lines
7.4 KiB
Python
217 lines
7.4 KiB
Python
"""
|
||
Tencent Cloud TIIA helpers.
|
||
|
||
We use Tencent Cloud "Image and Video AI" (TIIA) CreateImage API to upload an image
|
||
referenced by URL into a gallery group.
|
||
|
||
This module only provides a single upload function (not a scheduled task).
|
||
|
||
SDK reference: https://github.com/TencentCloud/tencentcloud-sdk-python
|
||
"""
|
||
|
||
import json
|
||
import uuid
|
||
from urllib.parse import urlparse
|
||
|
||
from django.conf import settings
|
||
|
||
from api_v1.utils.media import build_public_media_url
|
||
|
||
def _truncate(s: str, max_len: int) -> str:
|
||
s = s or ''
|
||
if len(s) <= max_len:
|
||
return s
|
||
return s[:max_len]
|
||
|
||
|
||
def upload_image_url_to_tencent_tiia(*, image_url: str, entity_id: str) -> dict:
|
||
"""
|
||
Upload an image by URL to Tencent Cloud TIIA gallery using CreateImage.
|
||
|
||
Only `image_url` and `entity_id` are required; other fields are derived from Django settings.
|
||
|
||
Required settings:
|
||
- TENCENTCLOUD_SECRET_ID
|
||
- TENCENTCLOUD_SECRET_KEY
|
||
- TENCENTCLOUD_TIIA_GROUP_ID
|
||
|
||
Optional settings:
|
||
- TENCENTCLOUD_TOKEN (STS token, if applicable)
|
||
- TENCENTCLOUD_TIIA_REGION (default: ap-guangzhou)
|
||
- TENCENTCLOUD_TIIA_ENDPOINT (default: tiia.tencentcloudapi.com)
|
||
- TENCENTCLOUD_TIIA_PIC_NAME_PREFIX (default: plate_order)
|
||
|
||
Returns:
|
||
Parsed JSON response dict from TencentCloud SDK.
|
||
"""
|
||
image_url = (image_url or '').strip()
|
||
if not image_url:
|
||
raise ValueError('image_url 不能为空')
|
||
|
||
entity_id = (entity_id or '').strip()
|
||
if not entity_id:
|
||
raise ValueError('entity_id 不能为空')
|
||
|
||
parsed = urlparse(image_url)
|
||
if not (parsed.scheme and parsed.netloc):
|
||
raise ValueError('image_url 必须为绝对 URL（包含 scheme 与 host）')
|
||
|
||
group_id = (getattr(settings, 'TENCENTCLOUD_TIIA_GROUP_ID', '') or '').strip()
|
||
if not group_id:
|
||
raise ValueError('缺少 settings: TENCENTCLOUD_TIIA_GROUP_ID')
|
||
|
||
secret_id = (getattr(settings, 'TENCENTCLOUD_SECRET_ID', '') or '').strip()
|
||
secret_key = (getattr(settings, 'TENCENTCLOUD_SECRET_KEY', '') or '').strip()
|
||
token = (getattr(settings, 'TENCENTCLOUD_TOKEN', '') or '').strip() or None
|
||
if not secret_id or not secret_key:
|
||
raise ValueError('缺少 settings: TENCENTCLOUD_SECRET_ID / TENCENTCLOUD_SECRET_KEY')
|
||
|
||
region = (getattr(settings, 'TENCENTCLOUD_TIIA_REGION', None) or 'ap-guangzhou').strip()
|
||
endpoint = (getattr(settings, 'TENCENTCLOUD_TIIA_ENDPOINT', None) or 'tiia.tencentcloudapi.com').strip()
|
||
|
||
pic_prefix = (getattr(settings, 'TENCENTCLOUD_TIIA_PIC_NAME_PREFIX', None) or 'plate_order').strip() or 'plate_order'
|
||
|
||
# CreateImage requires PicName; we auto-generate it.
|
||
uid = uuid.uuid4().hex
|
||
# NOTE: 云端对字段长度通常有限制，这里做一个保守截断，避免超长导致 400。
|
||
entity_id = _truncate(entity_id, 128)
|
||
basename = (parsed.path.rsplit('/', 1)[-1] or '').strip()
|
||
pic_name = f'{pic_prefix}_{basename}' if basename else f'{pic_prefix}_{uid}'
|
||
pic_name = _truncate(pic_name, 128)
|
||
|
||
try:
|
||
from tencentcloud.common import credential
|
||
from tencentcloud.common.profile.client_profile import ClientProfile
|
||
from tencentcloud.common.profile.http_profile import HttpProfile
|
||
from tencentcloud.tiia.v20190529 import tiia_client, models
|
||
except Exception as e: # pragma: no cover
|
||
raise RuntimeError(
|
||
"TencentCloud SDK 未安装或不可用，请先安装依赖：tencentcloud-sdk-python"
|
||
) from e
|
||
|
||
cred = credential.Credential(secret_id, secret_key, token)
|
||
http_profile = HttpProfile(endpoint=endpoint, reqTimeout=30)
|
||
client_profile = ClientProfile(httpProfile=http_profile)
|
||
client = tiia_client.TiiaClient(cred, region, client_profile)
|
||
|
||
req = models.CreateImageRequest()
|
||
req.GroupId = str(group_id)
|
||
req.EntityId = str(entity_id)
|
||
req.PicName = str(pic_name)
|
||
req.ImageUrl = str(image_url)
|
||
|
||
resp = client.CreateImage(req)
|
||
return json.loads(resp.to_json_string())
|
||
|
||
def _extract_urls_from_plate_image(plate_image) -> list[str]:
|
||
"""
|
||
从 PlateOrder.plate_image(JSON) 中提取所有可能的图片 URL/Path。
|
||
|
||
plate_image 在本项目中通常为 list[dict]，dict 里常见字段：
|
||
- url: 绝对 URL 或相对路径
|
||
- path: 相对路径（uploads/...）
|
||
也兼容：
|
||
- imageUrl/ImageURL: 外部系统字段名
|
||
- 直接 string（url/path）
|
||
"""
|
||
if not plate_image:
|
||
return []
|
||
|
||
if isinstance(plate_image, dict):
|
||
items = [plate_image]
|
||
elif isinstance(plate_image, list):
|
||
items = plate_image
|
||
elif isinstance(plate_image, str):
|
||
items = [plate_image]
|
||
else:
|
||
return []
|
||
|
||
urls: list[str] = []
|
||
for it in items:
|
||
raw = None
|
||
if isinstance(it, str):
|
||
raw = it
|
||
elif isinstance(it, dict):
|
||
raw = (
|
||
it.get('url')
|
||
or it.get('imageUrl')
|
||
or it.get('ImageUrl')
|
||
or it.get('imageURL')
|
||
or it.get('ImageURL')
|
||
or it.get('path')
|
||
)
|
||
if not raw:
|
||
continue
|
||
normalized = build_public_media_url(str(raw), request=None)
|
||
if normalized:
|
||
urls.append(normalized)
|
||
|
||
# 去重（保序）
|
||
seen = set()
|
||
deduped: list[str] = []
|
||
for u in urls:
|
||
if u in seen:
|
||
continue
|
||
seen.add(u)
|
||
deduped.append(u)
|
||
return deduped
|
||
|
||
|
||
def upload_plate_order_images_to_tencent_tiia(*, plate_order_id: int) -> list[dict]:
|
||
"""
|
||
包装函数：给定 plate_order_id，读取 PlateOrder.plate_image(JSONField) 中所有图片并上传到腾讯云图库。
|
||
|
||
- entity_id 使用 plate_order_id（字符串化）
|
||
- 逐张上传；单张失败不会中断整体，会返回 errors 供人工排查
|
||
|
||
Returns:
|
||
list[dict] 每张图片一条结果：
|
||
- image_url: str
|
||
- ok: bool
|
||
- response: dict | None
|
||
- error: str | None
|
||
"""
|
||
if plate_order_id is None:
|
||
raise ValueError('plate_order_id 不能为空')
|
||
try:
|
||
plate_order_id_int = int(plate_order_id)
|
||
except (TypeError, ValueError):
|
||
raise ValueError('plate_order_id 必须为整数')
|
||
|
||
from printing.models import PlateOrder
|
||
|
||
try:
|
||
po = PlateOrder.objects.get(id=plate_order_id_int)
|
||
except PlateOrder.DoesNotExist:
|
||
raise ValueError(f'PlateOrder {plate_order_id_int} 不存在')
|
||
|
||
urls = _extract_urls_from_plate_image(getattr(po, 'plate_image', None))
|
||
if not urls:
|
||
return []
|
||
|
||
entity_id = str(plate_order_id_int)
|
||
results: list[dict] = []
|
||
for url in urls:
|
||
url_str = (url or '').strip()
|
||
if not url_str:
|
||
continue
|
||
parsed = urlparse(url_str)
|
||
if not (parsed.scheme and parsed.netloc):
|
||
results.append({
|
||
'image_url': url_str,
|
||
'ok': False,
|
||
'response': None,
|
||
'error': '图片 URL 不是绝对地址（请确保存储了绝对 url，或正确配置 QINIU_BUCKET_DOMAIN）',
|
||
})
|
||
continue
|
||
try:
|
||
resp = upload_image_url_to_tencent_tiia(image_url=url_str, entity_id=entity_id)
|
||
results.append({'image_url': url_str, 'ok': True, 'response': resp, 'error': None})
|
||
except Exception as e:
|
||
results.append({'image_url': url_str, 'ok': False, 'response': None, 'error': str(e)})
|
||
return results
|
||
|
||
|
||
__all__ = ['upload_image_url_to_tencent_tiia', 'upload_plate_order_images_to_tencent_tiia']
|
||
|