From 5f182db1481384faff0dbfde56603e5c548d71da Mon Sep 17 00:00:00 2001 From: colaftc Date: Sun, 28 Jun 2026 22:19:02 +0800 Subject: [PATCH] feat: upload mdy stagging to qiniu beta --- api_v1/admin.py | 46 +++ ...mdy_plate_order_staging_images_to_qiniu.py | 192 ++++++++++ .../mdy_plate_order_staging_qiniu_upload.py | 341 ++++++++++++++++++ ...yplateorderstagingqiniuimageuploadaudit.py | 96 +++++ api_v1/models.py | 69 ++++ ...st_mdy_plate_order_staging_qiniu_upload.py | 133 +++++++ ...g_qiniu_image_upload_command_2026-06-28.md | 127 +++++++ 7 files changed, 1004 insertions(+) create mode 100644 api_v1/management/commands/upload_mdy_plate_order_staging_images_to_qiniu.py create mode 100644 api_v1/mdy_plate_order_staging_qiniu_upload.py create mode 100644 api_v1/migrations/0016_mdyplateorderstagingqiniuimageuploadaudit.py create mode 100644 api_v1/test_mdy_plate_order_staging_qiniu_upload.py create mode 100644 docs/mdy_plate_order_staging_qiniu_image_upload_command_2026-06-28.md diff --git a/api_v1/admin.py b/api_v1/admin.py index 6d35e57..dc3faf6 100644 --- a/api_v1/admin.py +++ b/api_v1/admin.py @@ -5,6 +5,7 @@ from api_v1.models import ( DataSync, MDYPlateOrderStaging, MDYPlateOrderStagingTiiaUploadFailure, + MDYPlateOrderStagingQiniuImageUploadAudit, PrintingExternalSyncFailure, ApiAuditLog, ) @@ -68,6 +69,51 @@ class MDYPlateOrderStagingTiiaUploadFailureAdmin(admin.ModelAdmin): ordering = ['-created_at'] +@admin.register(MDYPlateOrderStagingQiniuImageUploadAudit) +class MDYPlateOrderStagingQiniuImageUploadAuditAdmin(admin.ModelAdmin): + list_display = [ + 'id', + 'mdy_rowid', + 'staging_id', + 'attachment_index', + 'original_host', + 'status', + 'attempts', + 'uploaded_at', + 'last_attempt_at', + 'created_at', + ] + list_filter = ['status', 'original_host', 'created_at', 'uploaded_at'] + search_fields = ['mdy_rowid', 'original_url', 'qiniu_url', 'error'] + readonly_fields = [ + 'staging', + 'mdy_rowid', + 'source', + 'attachment_index', + 'source_field', + 'original_url', + 'original_host', + 'qiniu_key', + 'qiniu_url', + 'status', + 'attempts', + 'error', + 'file_size', + 'content_type', + 'raw_attachment', + 'started_at', + 'uploaded_at', + 'last_attempt_at', + 'created_at', + 'updated_at', + ] + date_hierarchy = 'created_at' + ordering = ['-created_at'] + + def has_add_permission(self, request): + return False + + @admin.register(PrintingExternalSyncFailure) class PrintingExternalSyncFailureAdmin(admin.ModelAdmin): list_display = [ diff --git a/api_v1/management/commands/upload_mdy_plate_order_staging_images_to_qiniu.py b/api_v1/management/commands/upload_mdy_plate_order_staging_images_to_qiniu.py new file mode 100644 index 0000000..f71db37 --- /dev/null +++ b/api_v1/management/commands/upload_mdy_plate_order_staging_images_to_qiniu.py @@ -0,0 +1,192 @@ +from datetime import timedelta + +from django.core.management.base import BaseCommand +from django.utils import timezone + +from api_v1.mdy_plate_order_staging_qiniu_upload import ( + DEFAULT_DOWNLOAD_CHUNK_SIZE, + DEFAULT_DOWNLOAD_TIMEOUT, + DEFAULT_MAX_IMAGE_SIZE, + DEFAULT_SPOOL_MAX_SIZE, + build_default_qiniu_rate_limiter, + is_qiniu_url, + iter_plate_image_work_items, + process_plate_image_work_item, + should_process_audit, +) +from api_v1.models import ( + MDYPlateOrderStaging, + MDYPlateOrderStagingQiniuImageUploadAudit, +) + + +class Command(BaseCommand): + help = '将明道云开版暂存 raw.开版图 图片流式上传到七牛，并记录逐图审计' + + def add_arguments(self, parser): + parser.add_argument('--dry-run', action='store_true', help='只扫描和汇总，不写数据库、不下载、不上传') + parser.add_argument('--limit', type=int, default=None, help='最多处理多少张候选图片；不传则处理全部') + parser.add_argument('--start-id', type=int, default=None, help='仅扫描 id >= start-id 的暂存记录') + parser.add_argument('--db-chunk-size', type=int, default=500, help='Django iterator 每批读取的 staging 行数') + parser.add_argument('--qps', type=float, default=None, help='下载/上传限速；默认 settings 或 2 qps；<=0 表示不限速') + parser.add_argument('--retry-failed', action='store_true', help='重试 status=failed 的审计记录') + parser.add_argument( + '--retry-stale-processing', + action='store_true', + help='重试长时间停留在 processing 的审计记录', + ) + parser.add_argument( + '--stale-after-minutes', + type=int, + default=120, + help='processing 超过多少分钟视为可重试的陈旧记录', + ) + parser.add_argument('--max-attempts', type=int, default=3, help='单张图最多尝试次数；不限制请传 0') + parser.add_argument( + '--max-image-size-mb', + type=int, + default=DEFAULT_MAX_IMAGE_SIZE // 1024 // 1024, + help='单张图片最大下载体积，超过则失败落审计', + ) + parser.add_argument( + '--spool-max-size-mb', + type=int, + default=DEFAULT_SPOOL_MAX_SIZE // 1024 // 1024, + help='单张图内存缓冲上限，超过后临时落盘', + ) + parser.add_argument( + '--download-chunk-size-mb', + type=int, + default=DEFAULT_DOWNLOAD_CHUNK_SIZE // 1024 // 1024, + help='流式下载的单块大小', + ) + parser.add_argument('--timeout', type=int, default=DEFAULT_DOWNLOAD_TIMEOUT, help='单次下载超时时间（秒）') + parser.add_argument('--progress-every', type=int, default=100, help='每处理多少张候选图片输出一次进度') + + def handle(self, *args, **options): + dry_run = options['dry_run'] + limit = options['limit'] + start_id = options['start_id'] + retry_failed = options['retry_failed'] + retry_stale_processing = options['retry_stale_processing'] + stale_after_minutes = options['stale_after_minutes'] + max_attempts = options['max_attempts'] or None + db_chunk_size = options['db_chunk_size'] + progress_every = options['progress_every'] + + stale_before = timezone.now() - timedelta(minutes=stale_after_minutes) + rate_limiter = None if dry_run else build_default_qiniu_rate_limiter(options['qps']) + + qs = MDYPlateOrderStaging.objects.only('id', 'mdy_rowid', 'raw').order_by('id') + if start_id is not None: + qs = qs.filter(id__gte=start_id) + + stats = { + 'staging_rows': 0, + 'image_items': 0, + 'planned': 0, + 'uploaded': 0, + 'reused': 0, + 'failed': 0, + 'skipped': 0, + 'already_qiniu': 0, + } + last_progress_count = 0 + + for staging in qs.iterator(chunk_size=db_chunk_size): + stats['staging_rows'] += 1 + for work_item in iter_plate_image_work_items(staging): + stats['image_items'] += 1 + if dry_run: + action, reason = self._classify_dry_run( + work_item, + retry_failed=retry_failed, + retry_stale_processing=retry_stale_processing, + stale_before=stale_before, + max_attempts=max_attempts, + ) + if action == 'planned': + stats['planned'] += 1 + elif reason == 'original_url_already_qiniu': + stats['already_qiniu'] += 1 + stats['skipped'] += 1 + else: + stats['skipped'] += 1 + processed_count = stats['planned'] + else: + result = process_plate_image_work_item( + work_item, + retry_failed=retry_failed, + retry_stale_processing=retry_stale_processing, + stale_before=stale_before, + max_attempts=max_attempts, + rate_limiter=rate_limiter, + max_image_size=options['max_image_size_mb'] * 1024 * 1024, + spool_max_size=options['spool_max_size_mb'] * 1024 * 1024, + chunk_size=options['download_chunk_size_mb'] * 1024 * 1024, + timeout=options['timeout'], + ) + if result.action == 'uploaded': + stats['uploaded'] += 1 + elif result.action == 'reused': + stats['reused'] += 1 + elif result.action == 'failed': + stats['failed'] += 1 + elif result.reason == 'original_url_already_qiniu': + stats['already_qiniu'] += 1 + stats['skipped'] += 1 + else: + stats['skipped'] += 1 + processed_count = stats['uploaded'] + stats['reused'] + stats['failed'] + + if ( + progress_every > 0 + and processed_count > 0 + and processed_count % progress_every == 0 + and processed_count != last_progress_count + ): + self.stdout.write(f'progress: {stats}') + last_progress_count = processed_count + + if limit is not None and processed_count >= limit: + self.stdout.write(self.style.SUCCESS(f'limited stop: {stats}')) + return + + self.stdout.write(self.style.SUCCESS(str(stats))) + + def _classify_dry_run( + self, + work_item, + *, + retry_failed: bool, + retry_stale_processing: bool, + stale_before, + max_attempts: int | None, + ) -> tuple[str, str]: + if is_qiniu_url(work_item.original_url): + return 'skipped', 'original_url_already_qiniu' + + audit = ( + MDYPlateOrderStagingQiniuImageUploadAudit.objects.filter( + staging=work_item.staging, + source=work_item.source, + attachment_index=work_item.attachment_index, + source_field=work_item.source_field, + original_url=work_item.original_url, + ) + .only('id', 'status', 'attempts', 'last_attempt_at') + .first() + ) + if audit is None: + return 'planned', 'new_audit' + + can_process, reason = should_process_audit( + audit, + retry_failed=retry_failed, + retry_stale_processing=retry_stale_processing, + stale_before=stale_before, + max_attempts=max_attempts, + ) + if can_process: + return 'planned', reason + return 'skipped', reason diff --git a/api_v1/mdy_plate_order_staging_qiniu_upload.py b/api_v1/mdy_plate_order_staging_qiniu_upload.py new file mode 100644 index 0000000..d5f59af --- /dev/null +++ b/api_v1/mdy_plate_order_staging_qiniu_upload.py @@ -0,0 +1,341 @@ +from __future__ import annotations + +import hashlib +import mimetypes +import os +import tempfile +from dataclasses import dataclass +from typing import Any, Callable +from urllib.parse import urlparse +from urllib.request import Request, urlopen + +from django.conf import settings +from django.core.files import File +from django.core.files.storage import default_storage +from django.utils import timezone + +from api_v1.mdy_plate_order_staging_tiia_upload import ( + _extract_plate_image_attachment_items, + _pick_attachment_url, +) +from api_v1.models import ( + MDYPlateOrderStaging, + MDYPlateOrderStagingQiniuImageUploadAudit, +) +from api_v1.utils.tencentcloud_tiia import SimpleRateLimiter + + +RAW_PLATE_IMAGES_SOURCE = MDYPlateOrderStagingQiniuImageUploadAudit.Source.RAW_PLATE_IMAGES +DEFAULT_QINIU_PREFIX = 'mdy_plate_order_staging/plate_images' +DEFAULT_DOWNLOAD_CHUNK_SIZE = 1024 * 1024 +DEFAULT_SPOOL_MAX_SIZE = 8 * 1024 * 1024 +DEFAULT_MAX_IMAGE_SIZE = 100 * 1024 * 1024 +DEFAULT_DOWNLOAD_TIMEOUT = 30 + + +@dataclass(frozen=True) +class PlateImageWorkItem: + staging: MDYPlateOrderStaging + attachment_index: int + source_field: str + original_url: str + original_host: str + raw_attachment: dict[str, Any] + source: str = RAW_PLATE_IMAGES_SOURCE + + +@dataclass(frozen=True) +class UploadedImage: + qiniu_key: str + qiniu_url: str + file_size: int + content_type: str + + +@dataclass(frozen=True) +class ProcessResult: + action: str + audit: MDYPlateOrderStagingQiniuImageUploadAudit | None = None + reason: str = '' + + +UploadFunc = Callable[..., UploadedImage] + + +def iter_plate_image_work_items(staging: MDYPlateOrderStaging): + for idx, item in enumerate(_extract_plate_image_attachment_items(staging)): + picked = _pick_attachment_url(item) + if not picked.url: + continue + parsed = urlparse(picked.url) + yield PlateImageWorkItem( + staging=staging, + attachment_index=idx, + source_field=picked.source_field or '', + original_url=picked.url, + original_host=parsed.hostname or '', + raw_attachment=dict(item), + ) + + +def is_qiniu_url(url: str) -> bool: + configured_domain = getattr(settings, 'QINIU_BUCKET_DOMAIN', '') or '' + if not configured_domain: + return False + return (urlparse(url).hostname or '').lower() == configured_domain.lower() + + +def build_qiniu_key(work_item: PlateImageWorkItem, *, content_type: str = '', prefix: str = DEFAULT_QINIU_PREFIX) -> str: + digest = hashlib.sha256(work_item.original_url.encode('utf-8')).hexdigest() + ext = _pick_extension(work_item.original_url, content_type) + return ( + f'{prefix}/staging-{work_item.staging.id}/' + f'attachment-{work_item.attachment_index}-{digest[:32]}{ext}' + ) + + +def get_or_create_audit_for_work_item(work_item: PlateImageWorkItem): + return MDYPlateOrderStagingQiniuImageUploadAudit.objects.get_or_create( + staging=work_item.staging, + source=work_item.source, + attachment_index=work_item.attachment_index, + source_field=work_item.source_field, + original_url=work_item.original_url, + defaults={ + 'mdy_rowid': work_item.staging.mdy_rowid, + 'original_host': work_item.original_host, + 'raw_attachment': work_item.raw_attachment, + }, + ) + + +def should_process_audit( + audit: MDYPlateOrderStagingQiniuImageUploadAudit, + *, + retry_failed: bool = False, + retry_stale_processing: bool = False, + stale_before=None, + max_attempts: int | None = None, +) -> tuple[bool, str]: + status = audit.status + statuses = MDYPlateOrderStagingQiniuImageUploadAudit.Status + + if status == statuses.SUCCESS: + return False, 'already_success' + if status == statuses.SKIPPED: + return False, 'already_skipped' + if max_attempts is not None and audit.attempts >= max_attempts: + return False, 'max_attempts_reached' + if status == statuses.FAILED and not retry_failed: + return False, 'failed_needs_retry_failed' + if status == statuses.PROCESSING: + if not retry_stale_processing: + return False, 'processing' + if stale_before is not None and audit.last_attempt_at and audit.last_attempt_at > stale_before: + return False, 'processing_not_stale' + return True, '' + + +def process_plate_image_work_item( + work_item: PlateImageWorkItem, + *, + retry_failed: bool = False, + retry_stale_processing: bool = False, + stale_before=None, + max_attempts: int | None = None, + rate_limiter: SimpleRateLimiter | None = None, + upload_func: UploadFunc | None = None, + max_image_size: int = DEFAULT_MAX_IMAGE_SIZE, + spool_max_size: int = DEFAULT_SPOOL_MAX_SIZE, + chunk_size: int = DEFAULT_DOWNLOAD_CHUNK_SIZE, + timeout: int = DEFAULT_DOWNLOAD_TIMEOUT, +) -> ProcessResult: + audit, _created = get_or_create_audit_for_work_item(work_item) + can_process, reason = should_process_audit( + audit, + retry_failed=retry_failed, + retry_stale_processing=retry_stale_processing, + stale_before=stale_before, + max_attempts=max_attempts, + ) + if not can_process: + return ProcessResult(action='skipped', audit=audit, reason=reason) + + if is_qiniu_url(work_item.original_url): + _mark_audit_skipped(audit, reason='original_url_already_qiniu') + return ProcessResult(action='skipped', audit=audit, reason='original_url_already_qiniu') + + _mark_audit_processing(audit) + + try: + reused = _find_reusable_success_audit(audit) + if reused is not None: + _mark_audit_success( + audit, + qiniu_key=reused.qiniu_key, + qiniu_url=reused.qiniu_url, + file_size=reused.file_size, + content_type=reused.content_type, + ) + return ProcessResult(action='reused', audit=audit) + + if rate_limiter is not None: + rate_limiter.wait() + uploader = upload_func or upload_original_url_to_qiniu + uploaded = uploader( + work_item, + max_image_size=max_image_size, + spool_max_size=spool_max_size, + chunk_size=chunk_size, + timeout=timeout, + ) + _mark_audit_success( + audit, + qiniu_key=uploaded.qiniu_key, + qiniu_url=uploaded.qiniu_url, + file_size=uploaded.file_size, + content_type=uploaded.content_type, + ) + return ProcessResult(action='uploaded', audit=audit) + except Exception as exc: + _mark_audit_failed(audit, str(exc)) + return ProcessResult(action='failed', audit=audit, reason=str(exc)) + + +def upload_original_url_to_qiniu( + work_item: PlateImageWorkItem, + *, + max_image_size: int = DEFAULT_MAX_IMAGE_SIZE, + spool_max_size: int = DEFAULT_SPOOL_MAX_SIZE, + chunk_size: int = DEFAULT_DOWNLOAD_CHUNK_SIZE, + timeout: int = DEFAULT_DOWNLOAD_TIMEOUT, +) -> UploadedImage: + request = Request( + work_item.original_url, + headers={'User-Agent': 'flower-mdy-plate-image-qiniu-migration/1.0'}, + ) + with urlopen(request, timeout=timeout) as response: + status = getattr(response, 'status', 200) + if status >= 400: + raise RuntimeError(f'download failed with status {status}') + + content_type = response.headers.get('Content-Type', '').split(';', 1)[0].strip() + content_length = _parse_content_length(response.headers.get('Content-Length')) + if content_length is not None and content_length > max_image_size: + raise RuntimeError(f'image too large: {content_length} > {max_image_size}') + + qiniu_key = build_qiniu_key(work_item, content_type=content_type) + with tempfile.SpooledTemporaryFile(max_size=spool_max_size, mode='w+b') as tmp: + total = 0 + while True: + chunk = response.read(chunk_size) + if not chunk: + break + total += len(chunk) + if total > max_image_size: + raise RuntimeError(f'image too large: {total} > {max_image_size}') + tmp.write(chunk) + + tmp.seek(0) + saved_key = default_storage.save(qiniu_key, File(tmp, name=os.path.basename(qiniu_key))) + return UploadedImage( + qiniu_key=saved_key, + qiniu_url=default_storage.url(saved_key), + file_size=total, + content_type=content_type, + ) + + +def build_default_qiniu_rate_limiter(qps: float | None = None) -> SimpleRateLimiter | None: + effective_qps = qps + if effective_qps is None: + effective_qps = float(getattr(settings, 'MDY_PLATE_ORDER_STAGING_QINIU_UPLOAD_QPS', 2) or 2) + if effective_qps <= 0: + return None + return SimpleRateLimiter(qps=effective_qps) + + +def _find_reusable_success_audit( + audit: MDYPlateOrderStagingQiniuImageUploadAudit, +) -> MDYPlateOrderStagingQiniuImageUploadAudit | None: + statuses = MDYPlateOrderStagingQiniuImageUploadAudit.Status + return ( + MDYPlateOrderStagingQiniuImageUploadAudit.objects.filter( + original_url=audit.original_url, + status=statuses.SUCCESS, + ) + .exclude(pk=audit.pk) + .exclude(qiniu_url='') + .order_by('id') + .first() + ) + + +def _mark_audit_processing(audit: MDYPlateOrderStagingQiniuImageUploadAudit) -> None: + now = timezone.now() + audit.status = MDYPlateOrderStagingQiniuImageUploadAudit.Status.PROCESSING + audit.attempts += 1 + audit.error = '' + audit.started_at = now + audit.last_attempt_at = now + audit.save(update_fields=['status', 'attempts', 'error', 'started_at', 'last_attempt_at', 'updated_at']) + + +def _mark_audit_success( + audit: MDYPlateOrderStagingQiniuImageUploadAudit, + *, + qiniu_key: str, + qiniu_url: str, + file_size: int | None, + content_type: str, +) -> None: + now = timezone.now() + audit.status = MDYPlateOrderStagingQiniuImageUploadAudit.Status.SUCCESS + audit.qiniu_key = qiniu_key + audit.qiniu_url = qiniu_url + audit.file_size = file_size + audit.content_type = content_type or '' + audit.error = '' + audit.uploaded_at = now + audit.save( + update_fields=[ + 'status', + 'qiniu_key', + 'qiniu_url', + 'file_size', + 'content_type', + 'error', + 'uploaded_at', + 'updated_at', + ] + ) + + +def _mark_audit_failed(audit: MDYPlateOrderStagingQiniuImageUploadAudit, error: str) -> None: + audit.status = MDYPlateOrderStagingQiniuImageUploadAudit.Status.FAILED + audit.error = error[:4000] + audit.save(update_fields=['status', 'error', 'updated_at']) + + +def _mark_audit_skipped(audit: MDYPlateOrderStagingQiniuImageUploadAudit, *, reason: str) -> None: + audit.status = MDYPlateOrderStagingQiniuImageUploadAudit.Status.SKIPPED + audit.error = reason + audit.save(update_fields=['status', 'error', 'updated_at']) + + +def _parse_content_length(value: str | None) -> int | None: + if not value: + return None + try: + return int(value) + except ValueError: + return None + + +def _pick_extension(url: str, content_type: str = '') -> str: + path = urlparse(url).path + ext = os.path.splitext(path)[1].lower() + if ext and len(ext) <= 10: + return ext + guessed = mimetypes.guess_extension(content_type) if content_type else None + return guessed or '' diff --git a/api_v1/migrations/0016_mdyplateorderstagingqiniuimageuploadaudit.py b/api_v1/migrations/0016_mdyplateorderstagingqiniuimageuploadaudit.py new file mode 100644 index 0000000..7eb2858 --- /dev/null +++ b/api_v1/migrations/0016_mdyplateorderstagingqiniuimageuploadaudit.py @@ -0,0 +1,96 @@ +# Generated by Codex on 2026-06-28 + +import django.db.models.deletion +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('api_v1', '0015_alter_apiauditlog_method_and_more'), + ] + + operations = [ + migrations.CreateModel( + name='MDYPlateOrderStagingQiniuImageUploadAudit', + fields=[ + ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')), + ('created_at', models.DateTimeField(auto_now_add=True, verbose_name='创建时间')), + ('updated_at', models.DateTimeField(auto_now=True, verbose_name='更新时间')), + ('mdy_rowid', models.CharField(db_index=True, max_length=64, verbose_name='明道云 RowID')), + ( + 'source', + models.CharField( + choices=[('raw_plate_images', 'raw.开版图')], + default='raw_plate_images', + max_length=50, + verbose_name='图片来源', + ), + ), + ('attachment_index', models.PositiveIntegerField(verbose_name='附件序号')), + ('source_field', models.CharField(blank=True, max_length=64, verbose_name='原始 URL 字段')), + ('original_url', models.URLField(max_length=2048, verbose_name='原始图片 URL')), + ('original_host', models.CharField(blank=True, db_index=True, max_length=255, verbose_name='原始图片域名')), + ('qiniu_key', models.CharField(blank=True, max_length=500, verbose_name='七牛对象 Key')), + ('qiniu_url', models.URLField(blank=True, max_length=2048, verbose_name='七牛图片 URL')), + ( + 'status', + models.CharField( + choices=[ + ('pending', '待处理'), + ('processing', '处理中'), + ('success', '成功'), + ('failed', '失败'), + ('skipped', '跳过'), + ], + db_index=True, + default='pending', + max_length=20, + verbose_name='上传状态', + ), + ), + ('attempts', models.PositiveIntegerField(default=0, verbose_name='尝试次数')), + ('error', models.TextField(blank=True, verbose_name='错误信息')), + ('file_size', models.PositiveBigIntegerField(blank=True, null=True, verbose_name='文件大小')), + ('content_type', models.CharField(blank=True, max_length=100, verbose_name='Content-Type')), + ('raw_attachment', models.JSONField(blank=True, default=dict, verbose_name='原始附件快照')), + ('started_at', models.DateTimeField(blank=True, null=True, verbose_name='开始处理时间')), + ('uploaded_at', models.DateTimeField(blank=True, null=True, verbose_name='上传成功时间')), + ('last_attempt_at', models.DateTimeField(blank=True, null=True, verbose_name='最后尝试时间')), + ( + 'staging', + models.ForeignKey( + on_delete=django.db.models.deletion.CASCADE, + related_name='qiniu_image_upload_audits', + to='api_v1.mdyplateorderstaging', + verbose_name='开版暂存记录', + ), + ), + ], + options={ + 'verbose_name': '开版暂存开版图七牛上传审计', + 'verbose_name_plural': '开版暂存开版图七牛上传审计', + 'db_table': 'api_mdy_plate_order_staging_qiniu_image_upload_audit', + 'ordering': ['-created_at'], + }, + ), + migrations.AddConstraint( + model_name='mdyplateorderstagingqiniuimageuploadaudit', + constraint=models.UniqueConstraint( + fields=('staging', 'source', 'attachment_index', 'source_field', 'original_url'), + name='uniq_mdy_qiniu_staging_src_idx_url', + ), + ), + migrations.AddIndex( + model_name='mdyplateorderstagingqiniuimageuploadaudit', + index=models.Index(fields=['status', 'id'], name='mdy_qiniu_status_id_idx'), + ), + migrations.AddIndex( + model_name='mdyplateorderstagingqiniuimageuploadaudit', + index=models.Index(fields=['original_url'], name='mdy_qiniu_orig_url_idx'), + ), + migrations.AddIndex( + model_name='mdyplateorderstagingqiniuimageuploadaudit', + index=models.Index(fields=['staging', 'source'], name='mdy_qiniu_staging_src_idx'), + ), + ] diff --git a/api_v1/models.py b/api_v1/models.py index 4fd187a..d8f5062 100644 --- a/api_v1/models.py +++ b/api_v1/models.py @@ -222,6 +222,75 @@ class MDYPlateOrderStagingTiiaUploadFailure(ModelBase): return f'{self.run_date} {self.mdy_rowid}' +class MDYPlateOrderStagingQiniuImageUploadAudit(ModelBase): + """明道云开版暂存"开版图"迁移到七牛的逐图审计记录。""" + + class Status(models.TextChoices): + PENDING = 'pending', '待处理' + PROCESSING = 'processing', '处理中' + SUCCESS = 'success', '成功' + FAILED = 'failed', '失败' + SKIPPED = 'skipped', '跳过' + + class Source(models.TextChoices): + RAW_PLATE_IMAGES = 'raw_plate_images', 'raw.开版图' + + staging = models.ForeignKey( + MDYPlateOrderStaging, + on_delete=models.CASCADE, + related_name='qiniu_image_upload_audits', + verbose_name='开版暂存记录', + ) + mdy_rowid = models.CharField(max_length=64, db_index=True, verbose_name='明道云 RowID') + source = models.CharField( + max_length=50, + choices=Source.choices, + default=Source.RAW_PLATE_IMAGES, + verbose_name='图片来源', + ) + attachment_index = models.PositiveIntegerField(verbose_name='附件序号') + source_field = models.CharField(max_length=64, blank=True, verbose_name='原始 URL 字段') + original_url = models.URLField(max_length=2048, verbose_name='原始图片 URL') + original_host = models.CharField(max_length=255, blank=True, db_index=True, verbose_name='原始图片域名') + qiniu_key = models.CharField(max_length=500, blank=True, verbose_name='七牛对象 Key') + qiniu_url = models.URLField(max_length=2048, blank=True, verbose_name='七牛图片 URL') + status = models.CharField( + max_length=20, + choices=Status.choices, + default=Status.PENDING, + db_index=True, + verbose_name='上传状态', + ) + attempts = models.PositiveIntegerField(default=0, verbose_name='尝试次数') + error = models.TextField(blank=True, verbose_name='错误信息') + file_size = models.PositiveBigIntegerField(null=True, blank=True, verbose_name='文件大小') + content_type = models.CharField(max_length=100, blank=True, verbose_name='Content-Type') + raw_attachment = models.JSONField(default=dict, blank=True, verbose_name='原始附件快照') + started_at = models.DateTimeField(null=True, blank=True, verbose_name='开始处理时间') + uploaded_at = models.DateTimeField(null=True, blank=True, verbose_name='上传成功时间') + last_attempt_at = models.DateTimeField(null=True, blank=True, verbose_name='最后尝试时间') + + class Meta: + db_table = 'api_mdy_plate_order_staging_qiniu_image_upload_audit' + verbose_name = '开版暂存开版图七牛上传审计' + verbose_name_plural = '开版暂存开版图七牛上传审计' + ordering = ['-created_at'] + constraints = [ + models.UniqueConstraint( + fields=['staging', 'source', 'attachment_index', 'source_field', 'original_url'], + name='uniq_mdy_qiniu_staging_src_idx_url', + ) + ] + indexes = [ + models.Index(fields=['status', 'id'], name='mdy_qiniu_status_id_idx'), + models.Index(fields=['original_url'], name='mdy_qiniu_orig_url_idx'), + models.Index(fields=['staging', 'source'], name='mdy_qiniu_staging_src_idx'), + ] + + def __str__(self): + return f'{self.mdy_rowid}#{self.attachment_index} {self.status}' + + class PrintingExternalSyncFailure(ModelBase): """ 外部印染 records 同步失败记录。 diff --git a/api_v1/test_mdy_plate_order_staging_qiniu_upload.py b/api_v1/test_mdy_plate_order_staging_qiniu_upload.py new file mode 100644 index 0000000..4328f60 --- /dev/null +++ b/api_v1/test_mdy_plate_order_staging_qiniu_upload.py @@ -0,0 +1,133 @@ +from django.test import SimpleTestCase, TestCase + +from api_v1.mdy_plate_order_staging_qiniu_upload import ( + PlateImageWorkItem, + UploadedImage, + build_qiniu_key, + iter_plate_image_work_items, + process_plate_image_work_item, +) +from api_v1.models import ( + MDYPlateOrderStaging, + MDYPlateOrderStagingQiniuImageUploadAudit, +) + + +PLATE_IMAGES_CONTROL_ID = '62d52f4b8d2972284492dd27' + + +class MDYPlateOrderStagingQiniuUploadWorkItemTest(SimpleTestCase): + def test_iter_plate_image_work_items_prefers_original_full_path(self): + staging = MDYPlateOrderStaging( + id=123, + mdy_rowid='row-1', + raw={ + PLATE_IMAGES_CONTROL_ID: [ + { + 'original_file_full_path': 'https://p1.mingdaoyun.cn/file-a.jpg', + 'DownloadUrl': 'https://p1.mingdaoyun.cn/download-a.jpg', + } + ] + }, + ) + + items = list(iter_plate_image_work_items(staging)) + + self.assertEqual(len(items), 1) + self.assertEqual(items[0].attachment_index, 0) + self.assertEqual(items[0].source_field, 'original_file_full_path') + self.assertEqual(items[0].original_url, 'https://p1.mingdaoyun.cn/file-a.jpg') + self.assertEqual(items[0].original_host, 'p1.mingdaoyun.cn') + + def test_build_qiniu_key_is_deterministic_and_keeps_extension(self): + work_item = PlateImageWorkItem( + staging=MDYPlateOrderStaging(id=456, mdy_rowid='row-2'), + attachment_index=3, + source_field='original_file_full_path', + original_url='https://p1.mingdaoyun.cn/path/image.png', + original_host='p1.mingdaoyun.cn', + raw_attachment={}, + ) + + key1 = build_qiniu_key(work_item) + key2 = build_qiniu_key(work_item) + + self.assertEqual(key1, key2) + self.assertTrue(key1.startswith('mdy_plate_order_staging/plate_images/staging-456/attachment-3-')) + self.assertTrue(key1.endswith('.png')) + + +class MDYPlateOrderStagingQiniuUploadProcessTest(TestCase): + def test_process_uploads_and_marks_success(self): + staging = MDYPlateOrderStaging.objects.create( + mdy_rowid='row-success', + raw={PLATE_IMAGES_CONTROL_ID: [{'original_file_full_path': 'https://p1.mingdaoyun.cn/a.jpg'}]}, + ) + work_item = next(iter_plate_image_work_items(staging)) + + result = process_plate_image_work_item(work_item, upload_func=self._successful_upload) + + self.assertEqual(result.action, 'uploaded') + audit = MDYPlateOrderStagingQiniuImageUploadAudit.objects.get() + self.assertEqual(audit.status, MDYPlateOrderStagingQiniuImageUploadAudit.Status.SUCCESS) + self.assertEqual(audit.attempts, 1) + self.assertEqual(audit.qiniu_url, 'https://image.yuwen.cloud/a.jpg') + self.assertEqual(audit.file_size, 12) + + def test_failed_audit_requires_retry_failed_flag(self): + staging = MDYPlateOrderStaging.objects.create( + mdy_rowid='row-failed', + raw={PLATE_IMAGES_CONTROL_ID: [{'original_file_full_path': 'https://p1.mingdaoyun.cn/b.jpg'}]}, + ) + work_item = next(iter_plate_image_work_items(staging)) + + first = process_plate_image_work_item(work_item, upload_func=self._failing_upload) + second = process_plate_image_work_item(work_item, upload_func=self._successful_upload) + third = process_plate_image_work_item(work_item, retry_failed=True, upload_func=self._successful_upload) + + audit = MDYPlateOrderStagingQiniuImageUploadAudit.objects.get() + self.assertEqual(first.action, 'failed') + self.assertEqual(second.action, 'skipped') + self.assertEqual(second.reason, 'failed_needs_retry_failed') + self.assertEqual(third.action, 'uploaded') + self.assertEqual(audit.status, MDYPlateOrderStagingQiniuImageUploadAudit.Status.SUCCESS) + self.assertEqual(audit.attempts, 2) + + def test_reuses_successful_upload_for_same_original_url(self): + url = 'https://p1.mingdaoyun.cn/shared.jpg' + first_staging = MDYPlateOrderStaging.objects.create( + mdy_rowid='row-first', + raw={PLATE_IMAGES_CONTROL_ID: [{'original_file_full_path': url}]}, + ) + second_staging = MDYPlateOrderStaging.objects.create( + mdy_rowid='row-second', + raw={PLATE_IMAGES_CONTROL_ID: [{'original_file_full_path': url}]}, + ) + first_item = next(iter_plate_image_work_items(first_staging)) + second_item = next(iter_plate_image_work_items(second_staging)) + + process_plate_image_work_item(first_item, upload_func=self._successful_upload) + result = process_plate_image_work_item(second_item, upload_func=self._unexpected_upload) + + self.assertEqual(result.action, 'reused') + audits = MDYPlateOrderStagingQiniuImageUploadAudit.objects.order_by('id') + self.assertEqual(audits.count(), 2) + self.assertEqual(audits[1].qiniu_url, audits[0].qiniu_url) + self.assertEqual(audits[1].qiniu_key, audits[0].qiniu_key) + + @staticmethod + def _successful_upload(work_item, **kwargs): + return UploadedImage( + qiniu_key='mdy_plate_order_staging/plate_images/a.jpg', + qiniu_url='https://image.yuwen.cloud/a.jpg', + file_size=12, + content_type='image/jpeg', + ) + + @staticmethod + def _failing_upload(work_item, **kwargs): + raise RuntimeError('download failed') + + @staticmethod + def _unexpected_upload(work_item, **kwargs): + raise AssertionError('upload should have been reused') diff --git a/docs/mdy_plate_order_staging_qiniu_image_upload_command_2026-06-28.md b/docs/mdy_plate_order_staging_qiniu_image_upload_command_2026-06-28.md new file mode 100644 index 0000000..370c817 --- /dev/null +++ b/docs/mdy_plate_order_staging_qiniu_image_upload_command_2026-06-28.md @@ -0,0 +1,127 @@ +# 明道云开版暂存开版图上传七牛命令 + +日期：2026-06-28 + +## 目标 + +将 `api_v1.MDYPlateOrderStaging.raw` 中明道云“开版图”字段的原始图片 URL 上传到七牛云，并把上传结果记录到独立审计表： + +- 暂存表：`api_mdy_plate_order_staging` +- 审计表：`api_mdy_plate_order_staging_qiniu_image_upload_audit` +- 处理字段：`raw["62d52f4b8d2972284492dd27"]` +- URL 优先级：`original_file_full_path`、`original_file_path`、`DownloadUrl`、`download_url`、`downloadUrl` +- 目标域名：`image.yuwen.cloud` + +命令不会回填或修改 `MDYPlateOrderStaging.raw`。 + +## 命令 + +```bash +python manage.py upload_mdy_plate_order_staging_images_to_qiniu +``` + +建议先在生产/测试副本环境做只读扫描： + +```bash +python manage.py upload_mdy_plate_order_staging_images_to_qiniu --dry-run +``` + +`--dry-run` 不写数据库、不下载、不上传，只统计会处理和会跳过的图片。 + +## 推荐发布后执行方式 + +先小批量执行： + +```bash +python manage.py upload_mdy_plate_order_staging_images_to_qiniu --limit 100 --qps 1 +``` + +确认审计表中的 `success/failed` 分布和七牛 URL 正常后，再放大运行： + +```bash +python manage.py upload_mdy_plate_order_staging_images_to_qiniu --qps 2 +``` + +## 重要参数 + +- `--limit N`：最多处理 N 张候选图片，用于灰度运行。 +- `--start-id ID`：只扫描 `MDYPlateOrderStaging.id >= ID` 的记录。 +- `--qps N`：限制下载/上传速率；默认读取 `MDY_PLATE_ORDER_STAGING_QINIU_UPLOAD_QPS`，没有配置时为 `2`。传 `0` 表示不限速。 +- `--retry-failed`：重试已经明确失败的审计记录。 +- `--retry-stale-processing`：重试长时间停在 `processing` 的审计记录。 +- `--stale-after-minutes N`：配合 `--retry-stale-processing`，默认 `120` 分钟。 +- `--max-attempts N`：单张图最多尝试次数，默认 `3`；传 `0` 表示不限制。 +- `--max-image-size-mb N`：单张图最大下载大小，默认 `100MB`。 +- `--spool-max-size-mb N`：单张图内存缓冲上限，默认 `8MB`；超过后自动使用临时文件。 +- `--download-chunk-size-mb N`：流式下载块大小，默认 `1MB`。 +- `--db-chunk-size N`：Django `iterator()` 每批读取 staging 行数，默认 `500`。 +- `--timeout N`：单次下载超时时间，默认 `30` 秒。 + +## 中断与恢复 + +审计表就是任务游标： + +- `success`：已成功，有 `qiniu_key` 和 `qiniu_url`，后续默认跳过。 +- `failed`：明确失败，默认跳过；需要传 `--retry-failed` 才会重试。 +- `processing`：命令中断可能留下的处理中状态，默认跳过；需要传 `--retry-stale-processing` 才会按陈旧时间重试。 +- `skipped`：例如原始 URL 已经是七牛域名，默认跳过。 + +重试所有明确失败： + +```bash +python manage.py upload_mdy_plate_order_staging_images_to_qiniu --retry-failed --qps 1 +``` + +重试中断留下的陈旧 `processing`： + +```bash +python manage.py upload_mdy_plate_order_staging_images_to_qiniu --retry-stale-processing --stale-after-minutes 120 --qps 1 +``` + +## 资源占用策略 + +命令按 staging 行使用 `iterator()` 分批读取，不会一次性加载全表。 + +每张图片独立处理： + +1. 创建或读取一条图片审计记录。 +2. 将状态置为 `processing` 并递增 `attempts`。 +3. 先查找同 `original_url` 的成功审计；如存在，直接复用已有七牛 URL，不重复下载上传。 +4. 如无可复用记录，则流式下载原图。 +5. 使用 `SpooledTemporaryFile`：小文件留在内存，超过 `--spool-max-size-mb` 自动落临时文件。 +6. 上传到七牛后关闭响应和临时文件。 +7. 单张图成功或失败都立即落库，不开启长事务。 + +因此命令的内存占用主要受 `--spool-max-size-mb`、`--download-chunk-size-mb` 和单张图片大小影响，不随总记录量线性增长。 + +## 数据核查 SQL + +查看状态分布： + +```sql +select status, count(*) +from api_mdy_plate_order_staging_qiniu_image_upload_audit +group by status +order by status; +``` + +查看最近失败： + +```sql +select id, staging_id, mdy_rowid, attachment_index, attempts, original_url, error, updated_at +from api_mdy_plate_order_staging_qiniu_image_upload_audit +where status = 'failed' +order by updated_at desc +limit 50; +``` + +查看成功但 URL 非目标域名的异常记录： + +```sql +select id, staging_id, mdy_rowid, qiniu_url +from api_mdy_plate_order_staging_qiniu_image_upload_audit +where status = 'success' + and qiniu_url not like 'http://image.yuwen.cloud/%' + and qiniu_url not like 'https://image.yuwen.cloud/%' +limit 50; +```