forked from erp-dev/erp
135 lines
4.4 KiB
Python
135 lines
4.4 KiB
Python
import logging
|
||
from datetime import datetime, time, timedelta
|
||
|
||
from celery import shared_task
|
||
from django.conf import settings
|
||
from django.db import transaction
|
||
from django.utils import timezone
|
||
|
||
from printing.models import PlateOrder, PlateOrderTiiaUploadFailure
|
||
from printing.services import send_printing_job_production_completed_wecom
|
||
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
def _get_day_range(day) -> tuple[datetime, datetime]:
|
||
"""
|
||
Build [start, end) datetimes for a local date.
|
||
"""
|
||
tz = timezone.get_current_timezone()
|
||
start = datetime.combine(day, time.min)
|
||
end = start + timedelta(days=1)
|
||
if timezone.is_naive(start):
|
||
start = timezone.make_aware(start, tz)
|
||
if timezone.is_naive(end):
|
||
end = timezone.make_aware(end, tz)
|
||
return start, end
|
||
|
||
|
||
@transaction.atomic
|
||
def _record_failure(*, run_date, plate_order_id: int, error: str, details):
|
||
"""
|
||
Upsert failure record for (run_date, plate_order_id).
|
||
"""
|
||
now = timezone.now()
|
||
obj, created = PlateOrderTiiaUploadFailure.objects.select_for_update().get_or_create(
|
||
run_date=run_date,
|
||
plate_order_id=plate_order_id,
|
||
defaults={
|
||
'error': error or '',
|
||
'details': details or [],
|
||
'attempts': 1,
|
||
'last_attempt_at': now,
|
||
},
|
||
)
|
||
if not created:
|
||
obj.error = error or obj.error or ''
|
||
obj.details = details or obj.details or []
|
||
obj.attempts = (obj.attempts or 0) + 1
|
||
obj.last_attempt_at = now
|
||
obj.save(update_fields=['error', 'details', 'attempts', 'last_attempt_at', 'updated_at'])
|
||
return obj
|
||
|
||
|
||
@shared_task(bind=True)
|
||
def upload_yesterday_plate_order_images_to_tencent_tiia(self):
|
||
"""
|
||
每天凌晨 03:00 运行：
|
||
- 扫描“昨天创建”的 PlateOrder
|
||
- 将 plate_image(JSONField) 中的所有图片 URL 上传到腾讯云图库
|
||
- 遇错：记录失败的 plate_order_id（落库），继续处理下一个
|
||
|
||
注意：腾讯云调用限速 10 QPS（每秒 10 次）。
|
||
"""
|
||
from api_v1.utils.tencentcloud_tiia import (
|
||
SimpleRateLimiter,
|
||
upload_plate_order_images_to_tencent_tiia,
|
||
)
|
||
|
||
today = timezone.localdate()
|
||
run_date = today - timedelta(days=1)
|
||
start, end = _get_day_range(run_date)
|
||
|
||
qps = int(getattr(settings, 'TENCENTCLOUD_TIIA_QPS', 10) or 10)
|
||
limiter = SimpleRateLimiter(qps=float(qps))
|
||
|
||
qs = (
|
||
PlateOrder.objects.filter(created_at__gte=start, created_at__lt=end)
|
||
.only('id', 'plate_image')
|
||
.order_by('id')
|
||
)
|
||
|
||
total_orders = 0
|
||
failed_ids: list[int] = []
|
||
processed_images = 0
|
||
|
||
for po in qs.iterator(chunk_size=200):
|
||
total_orders += 1
|
||
try:
|
||
results = upload_plate_order_images_to_tencent_tiia(
|
||
plate_order_id=po.id,
|
||
rate_limiter=limiter,
|
||
)
|
||
except Exception as e:
|
||
failed_ids.append(po.id)
|
||
_record_failure(run_date=run_date, plate_order_id=po.id, error=str(e), details=[{'error': str(e)}])
|
||
logger.exception('[TIIA] PlateOrder %s 上传异常（任务继续）', po.id)
|
||
continue
|
||
|
||
processed_images += len(results)
|
||
bad = [r for r in results if not r.get('ok')]
|
||
if bad:
|
||
failed_ids.append(po.id)
|
||
summary = '; '.join([(b.get('error') or '') for b in bad][:5])
|
||
_record_failure(run_date=run_date, plate_order_id=po.id, error=summary, details=bad)
|
||
logger.warning('[TIIA] PlateOrder %s 部分图片上传失败: %s', po.id, summary)
|
||
|
||
payload = {
|
||
'task_id': getattr(getattr(self, 'request', None), 'id', None),
|
||
'run_date': str(run_date),
|
||
'created_range': {'start': start.isoformat(), 'end': end.isoformat()},
|
||
'total_orders': total_orders,
|
||
'processed_images': processed_images,
|
||
'failed_count': len(failed_ids),
|
||
'failed_ids': failed_ids,
|
||
}
|
||
logger.info('[TIIA] 昨日 PlateOrder 图片上传任务完成: %s', payload)
|
||
return payload
|
||
|
||
|
||
@shared_task(bind=True)
|
||
def notify_printing_job_production_completed_wecom(
|
||
self,
|
||
*,
|
||
printing_job_id: int,
|
||
triggered_by_id: int | None = None,
|
||
) -> dict:
|
||
payload = send_printing_job_production_completed_wecom(
|
||
printing_job_id=printing_job_id,
|
||
triggered_by_id=triggered_by_id,
|
||
)
|
||
payload["task_id"] = self.request.id
|
||
logger.info("[printing.tasks] 印染任务完成生产企业微信通知发送完成: %s", payload)
|
||
return payload
|