From 7c9b77afb0c9d1c6ad46187a379f64ee99029597 Mon Sep 17 00:00:00 2001 From: colaftc Date: Sun, 28 Jun 2026 23:22:21 +0800 Subject: [PATCH] fix --- api_v1/admin.py | 33 ++ ...mdy_plate_order_staging_images_to_qiniu.py | 420 +++++++++++++----- .../mdy_plate_order_staging_qiniu_upload.py | 53 +++ ...yplateorderstagingqiniuimageuploadstate.py | 35 ++ api_v1/models.py | 22 + ...g_qiniu_image_upload_command_2026-06-28.md | 48 +- 6 files changed, 477 insertions(+), 134 deletions(-) create mode 100644 api_v1/migrations/0017_mdyplateorderstagingqiniuimageuploadstate.py diff --git a/api_v1/admin.py b/api_v1/admin.py index dc3faf6..dec7e1a 100644 --- a/api_v1/admin.py +++ b/api_v1/admin.py @@ -6,6 +6,7 @@ from api_v1.models import ( MDYPlateOrderStaging, MDYPlateOrderStagingTiiaUploadFailure, MDYPlateOrderStagingQiniuImageUploadAudit, + MDYPlateOrderStagingQiniuImageUploadState, PrintingExternalSyncFailure, ApiAuditLog, ) @@ -114,6 +115,38 @@ class MDYPlateOrderStagingQiniuImageUploadAuditAdmin(admin.ModelAdmin): return False +@admin.register(MDYPlateOrderStagingQiniuImageUploadState) +class MDYPlateOrderStagingQiniuImageUploadStateAdmin(admin.ModelAdmin): + list_display = [ + 'id', + 'name', + 'last_staging_id', + 'staging_finished', + 'scanned_staging_rows', + 'discovered_image_items', + 'created_audit_rows', + 'existing_audit_rows', + 'last_heartbeat_at', + 'updated_at', + ] + readonly_fields = [ + 'name', + 'last_staging_id', + 'staging_finished', + 'scanned_staging_rows', + 'discovered_image_items', + 'created_audit_rows', + 'existing_audit_rows', + 'last_heartbeat_at', + 'created_at', + 'updated_at', + ] + ordering = ['name'] + + 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 index f71db37..427f6e1 100644 --- 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 @@ -1,6 +1,9 @@ +import time from datetime import timedelta -from django.core.management.base import BaseCommand +from django.core.management.base import BaseCommand, CommandError +from django.db import connection +from django.db.models import Q from django.utils import timezone from api_v1.mdy_plate_order_staging_qiniu_upload import ( @@ -8,27 +11,37 @@ from api_v1.mdy_plate_order_staging_qiniu_upload import ( DEFAULT_DOWNLOAD_TIMEOUT, DEFAULT_MAX_IMAGE_SIZE, DEFAULT_SPOOL_MAX_SIZE, + DEFAULT_UPLOAD_STATE_NAME, build_default_qiniu_rate_limiter, - is_qiniu_url, + get_or_create_audit_for_work_item, + get_or_create_default_upload_state, iter_plate_image_work_items, - process_plate_image_work_item, - should_process_audit, + process_qiniu_image_upload_audit, ) from api_v1.models import ( MDYPlateOrderStaging, MDYPlateOrderStagingQiniuImageUploadAudit, + MDYPlateOrderStagingQiniuImageUploadState, ) class Command(BaseCommand): - help = '将明道云开版暂存 raw.开版图 图片流式上传到七牛，并记录逐图审计' + help = '将明道云开版暂存 raw.开版图 图片分批上传到七牛，并记录逐图审计' + lock_key = 'api_v1.upload_mdy_plate_order_staging_images_to_qiniu' 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('--dry-run', action='store_true', help='只输出配置和当前游标，不扫描、不写库、不下载、不上传') + parser.add_argument('--scan-batch-size', type=int, default=20, help='每轮最多扫描多少条 staging') + parser.add_argument('--upload-batch-size', type=int, default=5, help='每轮最多上传多少条审计记录') + parser.add_argument( + '--max-upload-attempts', + type=int, + default=None, + help='最多尝试处理多少条审计上传记录；测试/灰度用，不传则持续运行到结束', + ) parser.add_argument('--qps', type=float, default=None, help='下载/上传限速；默认 settings 或 2 qps；<=0 表示不限速') + parser.add_argument('--loop-sleep-seconds', type=float, default=0.2, help='每轮结束后的短暂休眠，降低 CPU 压力') + parser.add_argument('--reset-cursor', action='store_true', help='重置 staging 扫描游标，从头重新发现审计记录') parser.add_argument('--retry-failed', action='store_true', help='重试 status=failed 的审计记录') parser.add_argument( '--retry-stale-processing', @@ -61,132 +74,301 @@ class Command(BaseCommand): help='流式下载的单块大小', ) parser.add_argument('--timeout', type=int, default=DEFAULT_DOWNLOAD_TIMEOUT, help='单次下载超时时间（秒）') - parser.add_argument('--progress-every', type=int, default=100, help='每处理多少张候选图片输出一次进度') + parser.add_argument('--progress-every-loops', type=int, default=1, help='每多少轮输出一次心跳，默认每轮输出') def handle(self, *args, **options): dry_run = options['dry_run'] - limit = options['limit'] - start_id = options['start_id'] + scan_batch_size = options['scan_batch_size'] + upload_batch_size = options['upload_batch_size'] + max_upload_attempts = options['max_upload_attempts'] + loop_sleep_seconds = options['loop_sleep_seconds'] retry_failed = options['retry_failed'] retry_stale_processing = options['retry_stale_processing'] - stale_after_minutes = options['stale_after_minutes'] + stale_before = timezone.now() - timedelta(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']) + if scan_batch_size <= 0: + raise CommandError('--scan-batch-size 必须大于 0') + if upload_batch_size <= 0: + raise CommandError('--upload-batch-size 必须大于 0') + if max_upload_attempts is not None and max_upload_attempts <= 0: + raise CommandError('--max-upload-attempts 必须大于 0') - qs = MDYPlateOrderStaging.objects.only('id', 'mdy_rowid', 'raw').order_by('id') - if start_id is not None: - qs = qs.filter(id__gte=start_id) + if dry_run: + state = MDYPlateOrderStagingQiniuImageUploadState.objects.filter( + name=DEFAULT_UPLOAD_STATE_NAME, + ).first() + config = { + 'dry_run': True, + 'scan_batch_size': scan_batch_size, + 'upload_batch_size': upload_batch_size, + 'max_upload_attempts': max_upload_attempts, + 'qps': options['qps'], + 'loop_sleep_seconds': loop_sleep_seconds, + 'retry_failed': retry_failed, + 'retry_stale_processing': retry_stale_processing, + 'max_attempts': max_attempts, + 'state_exists': state is not None, + 'state_last_staging_id': state.last_staging_id if state else None, + 'state_staging_finished': state.staging_finished if state else None, + } + self.stdout.write(f'start: {config}') + self.stdout.write(self.style.SUCCESS('dry-run stop: no database writes, no upload')) + return - stats = { - 'staging_rows': 0, - 'image_items': 0, - 'planned': 0, - 'uploaded': 0, - 'reused': 0, - 'failed': 0, - 'skipped': 0, - 'already_qiniu': 0, - } - last_progress_count = 0 + if not self._try_acquire_db_lock(): + raise CommandError('已有 upload_mdy_plate_order_staging_images_to_qiniu 命令正在运行') + try: + self._run_upload_loop( + options=options, + scan_batch_size=scan_batch_size, + upload_batch_size=upload_batch_size, + max_upload_attempts=max_upload_attempts, + loop_sleep_seconds=loop_sleep_seconds, + retry_failed=retry_failed, + retry_stale_processing=retry_stale_processing, + stale_before=stale_before, + max_attempts=max_attempts, + ) + finally: + self._release_db_lock() - 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( + def _run_upload_loop( self, - work_item, *, + options, + scan_batch_size: int, + upload_batch_size: int, + max_upload_attempts: int | None, + loop_sleep_seconds: float, 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, + ) -> None: + state = get_or_create_default_upload_state() + if options['reset_cursor']: + state.last_staging_id = 0 + state.staging_finished = False + state.scanned_staging_rows = 0 + state.discovered_image_items = 0 + state.created_audit_rows = 0 + state.existing_audit_rows = 0 + state.last_heartbeat_at = timezone.now() + state.save( + update_fields=[ + 'last_staging_id', + 'staging_finished', + 'scanned_staging_rows', + 'discovered_image_items', + 'created_audit_rows', + 'existing_audit_rows', + 'last_heartbeat_at', + 'updated_at', + ] ) - .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, + config = { + 'dry_run': False, + 'scan_batch_size': scan_batch_size, + 'upload_batch_size': upload_batch_size, + 'max_upload_attempts': max_upload_attempts, + 'qps': options['qps'], + 'loop_sleep_seconds': loop_sleep_seconds, + 'retry_failed': retry_failed, + 'retry_stale_processing': retry_stale_processing, + 'max_attempts': max_attempts, + 'state_last_staging_id': state.last_staging_id, + 'state_staging_finished': state.staging_finished, + } + self.stdout.write(f'start: {config}') + + rate_limiter = build_default_qiniu_rate_limiter(options['qps']) + totals = { + 'loops': 0, + 'scanned_staging_rows': 0, + 'discovered_image_items': 0, + 'created_audit_rows': 0, + 'existing_audit_rows': 0, + 'uploaded': 0, + 'reused': 0, + 'failed': 0, + 'skipped': 0, + } + + while True: + totals['loops'] += 1 + effective_upload_batch_size = upload_batch_size + if max_upload_attempts is not None: + remaining_attempts = max_upload_attempts - self._upload_attempt_count(totals) + if remaining_attempts <= 0: + self.stdout.write(self.style.SUCCESS(f'max-upload-attempts stop: {totals}')) + return + effective_upload_batch_size = min(upload_batch_size, remaining_attempts) + + discover_stats = self._discover_next_staging_batch( + scan_batch_size=scan_batch_size, + ) + upload_stats = self._upload_next_audit_batch( + upload_batch_size=effective_upload_batch_size, + 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'], + ) + self._add_stats(totals, discover_stats) + self._add_stats(totals, upload_stats) + + state = get_or_create_default_upload_state() + heartbeat = { + 'loop': totals['loops'], + 'state_last_staging_id': state.last_staging_id, + 'state_staging_finished': state.staging_finished, + 'discover': discover_stats, + 'upload': upload_stats, + 'totals': totals, + } + state.last_heartbeat_at = timezone.now() + state.save(update_fields=['last_heartbeat_at', 'updated_at']) + + if options['progress_every_loops'] > 0 and totals['loops'] % options['progress_every_loops'] == 0: + self.stdout.write(f'progress: {heartbeat}') + + if state.staging_finished and upload_stats['selected_audit_rows'] == 0: + self.stdout.write(self.style.SUCCESS(f'finished: {heartbeat}')) + return + + if max_upload_attempts is not None and self._upload_attempt_count(totals) >= max_upload_attempts: + self.stdout.write(self.style.SUCCESS(f'max-upload-attempts stop: {heartbeat}')) + return + + if loop_sleep_seconds > 0: + time.sleep(loop_sleep_seconds) + + def _try_acquire_db_lock(self) -> bool: + if connection.vendor != 'postgresql': + return True + with connection.cursor() as cursor: + cursor.execute('select pg_try_advisory_lock(hashtext(%s))', [self.lock_key]) + return bool(cursor.fetchone()[0]) + + def _release_db_lock(self) -> None: + if connection.vendor != 'postgresql': + return + with connection.cursor() as cursor: + cursor.execute('select pg_advisory_unlock(hashtext(%s))', [self.lock_key]) + + def _discover_next_staging_batch(self, *, scan_batch_size: int) -> dict[str, int]: + state = get_or_create_default_upload_state() + stats = { + 'scanned_staging_rows': 0, + 'discovered_image_items': 0, + 'created_audit_rows': 0, + 'existing_audit_rows': 0, + } + if state.staging_finished: + return stats + + rows = list( + MDYPlateOrderStaging.objects.filter(id__gt=state.last_staging_id) + .only('id', 'mdy_rowid', 'raw') + .order_by('id')[:scan_batch_size] ) - if can_process: - return 'planned', reason - return 'skipped', reason + if not rows: + state.staging_finished = True + state.last_heartbeat_at = timezone.now() + state.save(update_fields=['staging_finished', 'last_heartbeat_at', 'updated_at']) + return stats + + for staging in rows: + stats['scanned_staging_rows'] += 1 + for work_item in iter_plate_image_work_items(staging): + stats['discovered_image_items'] += 1 + _audit, created = get_or_create_audit_for_work_item(work_item) + if created: + stats['created_audit_rows'] += 1 + else: + stats['existing_audit_rows'] += 1 + + state.last_staging_id = rows[-1].id + state.scanned_staging_rows += stats['scanned_staging_rows'] + state.discovered_image_items += stats['discovered_image_items'] + state.created_audit_rows += stats['created_audit_rows'] + state.existing_audit_rows += stats['existing_audit_rows'] + state.last_heartbeat_at = timezone.now() + state.save( + update_fields=[ + 'last_staging_id', + 'scanned_staging_rows', + 'discovered_image_items', + 'created_audit_rows', + 'existing_audit_rows', + 'last_heartbeat_at', + 'updated_at', + ] + ) + return stats + + def _upload_next_audit_batch(self, **kwargs) -> dict[str, int]: + upload_batch_size = kwargs.pop('upload_batch_size') + retry_failed = kwargs['retry_failed'] + retry_stale_processing = kwargs['retry_stale_processing'] + stale_before = kwargs['stale_before'] + + statuses = MDYPlateOrderStagingQiniuImageUploadAudit.Status + status_filter = Q(status=statuses.PENDING) + if retry_failed: + status_filter |= Q(status=statuses.FAILED) + if retry_stale_processing: + status_filter |= Q(status=statuses.PROCESSING, last_attempt_at__lte=stale_before) + + audits = list( + MDYPlateOrderStagingQiniuImageUploadAudit.objects.filter(status_filter) + .select_related('staging') + .order_by('id')[:upload_batch_size] + ) + stats = { + 'selected_audit_rows': len(audits), + 'uploaded': 0, + 'reused': 0, + 'failed': 0, + 'skipped': 0, + } + + process_kwargs = dict(kwargs) + process_kwargs.pop('retry_failed', None) + process_kwargs.pop('retry_stale_processing', None) + process_kwargs.pop('stale_before', None) + for audit in audits: + result = process_qiniu_image_upload_audit( + audit, + retry_failed=retry_failed, + retry_stale_processing=retry_stale_processing, + stale_before=stale_before, + **process_kwargs, + ) + if result.action == 'uploaded': + stats['uploaded'] += 1 + elif result.action == 'reused': + stats['reused'] += 1 + elif result.action == 'failed': + stats['failed'] += 1 + else: + stats['skipped'] += 1 + + return stats + + @staticmethod + def _add_stats(target: dict[str, int], source: dict[str, int]) -> None: + for key, value in source.items(): + if key in target: + target[key] += value + + @staticmethod + def _upload_attempt_count(stats: dict[str, int]) -> int: + return stats.get('uploaded', 0) + stats.get('reused', 0) + stats.get('failed', 0) + stats.get('skipped', 0) diff --git a/api_v1/mdy_plate_order_staging_qiniu_upload.py b/api_v1/mdy_plate_order_staging_qiniu_upload.py index d5f59af..c8f10c3 100644 --- a/api_v1/mdy_plate_order_staging_qiniu_upload.py +++ b/api_v1/mdy_plate_order_staging_qiniu_upload.py @@ -21,6 +21,7 @@ from api_v1.mdy_plate_order_staging_tiia_upload import ( from api_v1.models import ( MDYPlateOrderStaging, MDYPlateOrderStagingQiniuImageUploadAudit, + MDYPlateOrderStagingQiniuImageUploadState, ) from api_v1.utils.tencentcloud_tiia import SimpleRateLimiter @@ -31,6 +32,7 @@ DEFAULT_DOWNLOAD_CHUNK_SIZE = 1024 * 1024 DEFAULT_SPOOL_MAX_SIZE = 8 * 1024 * 1024 DEFAULT_MAX_IMAGE_SIZE = 100 * 1024 * 1024 DEFAULT_DOWNLOAD_TIMEOUT = 30 +DEFAULT_UPLOAD_STATE_NAME = 'raw_plate_images_qiniu_upload' @dataclass(frozen=True) @@ -109,6 +111,24 @@ def get_or_create_audit_for_work_item(work_item: PlateImageWorkItem): ) +def get_or_create_default_upload_state(): + return MDYPlateOrderStagingQiniuImageUploadState.objects.get_or_create( + name=DEFAULT_UPLOAD_STATE_NAME, + )[0] + + +def build_work_item_from_audit(audit: MDYPlateOrderStagingQiniuImageUploadAudit) -> PlateImageWorkItem: + return PlateImageWorkItem( + staging=audit.staging, + attachment_index=audit.attachment_index, + source_field=audit.source_field, + original_url=audit.original_url, + original_host=audit.original_host, + raw_attachment=audit.raw_attachment if isinstance(audit.raw_attachment, dict) else {}, + source=audit.source, + ) + + def should_process_audit( audit: MDYPlateOrderStagingQiniuImageUploadAudit, *, @@ -151,6 +171,39 @@ def process_plate_image_work_item( timeout: int = DEFAULT_DOWNLOAD_TIMEOUT, ) -> ProcessResult: audit, _created = get_or_create_audit_for_work_item(work_item) + return process_qiniu_image_upload_audit( + audit, + 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, + upload_func=upload_func, + max_image_size=max_image_size, + spool_max_size=spool_max_size, + chunk_size=chunk_size, + timeout=timeout, + ) + + +def process_qiniu_image_upload_audit( + audit: MDYPlateOrderStagingQiniuImageUploadAudit, + *, + work_item: PlateImageWorkItem | None = None, + 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: + if work_item is None: + work_item = build_work_item_from_audit(audit) can_process, reason = should_process_audit( audit, retry_failed=retry_failed, diff --git a/api_v1/migrations/0017_mdyplateorderstagingqiniuimageuploadstate.py b/api_v1/migrations/0017_mdyplateorderstagingqiniuimageuploadstate.py new file mode 100644 index 0000000..4160d53 --- /dev/null +++ b/api_v1/migrations/0017_mdyplateorderstagingqiniuimageuploadstate.py @@ -0,0 +1,35 @@ +# Generated by Codex on 2026-06-28 + +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('api_v1', '0016_mdyplateorderstagingqiniuimageuploadaudit'), + ] + + operations = [ + migrations.CreateModel( + name='MDYPlateOrderStagingQiniuImageUploadState', + 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='更新时间')), + ('name', models.CharField(max_length=80, unique=True, verbose_name='任务名称')), + ('last_staging_id', models.PositiveBigIntegerField(default=0, verbose_name='已扫描到的暂存表ID')), + ('staging_finished', models.BooleanField(default=False, verbose_name='暂存表是否扫描完成')), + ('scanned_staging_rows', models.PositiveBigIntegerField(default=0, verbose_name='已扫描暂存行数')), + ('discovered_image_items', models.PositiveBigIntegerField(default=0, verbose_name='发现图片数')), + ('created_audit_rows', models.PositiveBigIntegerField(default=0, verbose_name='创建审计数')), + ('existing_audit_rows', models.PositiveBigIntegerField(default=0, verbose_name='已存在审计数')), + ('last_heartbeat_at', models.DateTimeField(blank=True, null=True, verbose_name='最后心跳时间')), + ], + options={ + 'verbose_name': '开版暂存开版图七牛上传状态', + 'verbose_name_plural': '开版暂存开版图七牛上传状态', + 'db_table': 'api_mdy_plate_order_staging_qiniu_image_upload_state', + 'ordering': ['name'], + }, + ), + ] diff --git a/api_v1/models.py b/api_v1/models.py index d8f5062..01e51a4 100644 --- a/api_v1/models.py +++ b/api_v1/models.py @@ -291,6 +291,28 @@ class MDYPlateOrderStagingQiniuImageUploadAudit(ModelBase): return f'{self.mdy_rowid}#{self.attachment_index} {self.status}' +class MDYPlateOrderStagingQiniuImageUploadState(ModelBase): + """明道云开版暂存开版图上传任务的持久扫描游标。""" + + name = models.CharField(max_length=80, unique=True, verbose_name='任务名称') + last_staging_id = models.PositiveBigIntegerField(default=0, verbose_name='已扫描到的暂存表ID') + staging_finished = models.BooleanField(default=False, verbose_name='暂存表是否扫描完成') + scanned_staging_rows = models.PositiveBigIntegerField(default=0, verbose_name='已扫描暂存行数') + discovered_image_items = models.PositiveBigIntegerField(default=0, verbose_name='发现图片数') + created_audit_rows = models.PositiveBigIntegerField(default=0, verbose_name='创建审计数') + existing_audit_rows = models.PositiveBigIntegerField(default=0, verbose_name='已存在审计数') + last_heartbeat_at = models.DateTimeField(null=True, blank=True, verbose_name='最后心跳时间') + + class Meta: + db_table = 'api_mdy_plate_order_staging_qiniu_image_upload_state' + verbose_name = '开版暂存开版图七牛上传状态' + verbose_name_plural = '开版暂存开版图七牛上传状态' + ordering = ['name'] + + def __str__(self): + return f'{self.name}: staging_id={self.last_staging_id}' + + class PrintingExternalSyncFailure(ModelBase): """ 外部印染 records 同步失败记录。 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 index 370c817..70b499c 100644 --- 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 @@ -4,10 +4,11 @@ ## 目标 -将 `api_v1.MDYPlateOrderStaging.raw` 中明道云“开版图”字段的原始图片 URL 上传到七牛云，并把上传结果记录到独立审计表： +将 `api_v1.MDYPlateOrderStaging.raw` 中明道云“开版图”字段的原始图片 URL 上传到七牛云，并把上传结果记录到独立审计表和任务状态表： - 暂存表：`api_mdy_plate_order_staging` - 审计表：`api_mdy_plate_order_staging_qiniu_image_upload_audit` +- 状态表：`api_mdy_plate_order_staging_qiniu_image_upload_state` - 处理字段：`raw["62d52f4b8d2972284492dd27"]` - URL 优先级：`original_file_full_path`、`original_file_path`、`DownloadUrl`、`download_url`、`downloadUrl` - 目标域名：`image.yuwen.cloud` @@ -26,27 +27,32 @@ python manage.py upload_mdy_plate_order_staging_images_to_qiniu python manage.py upload_mdy_plate_order_staging_images_to_qiniu --dry-run ``` -`--dry-run` 不写数据库、不下载、不上传，只统计会处理和会跳过的图片。 +`--dry-run` 不写数据库、不扫描 staging、不下载、不上传，只输出当前 cursor 和运行配置。 ## 推荐发布后执行方式 -先小批量执行： - -```bash -python manage.py upload_mdy_plate_order_staging_images_to_qiniu --limit 100 --qps 1 -``` - -确认审计表中的 `success/failed` 分布和七牛 URL 正常后，再放大运行： +默认命令会持续运行到 staging cursor 扫描完成，并且审计表中没有可处理记录为止： ```bash python manage.py upload_mdy_plate_order_staging_images_to_qiniu --qps 2 ``` +命令在 PostgreSQL 上会使用 advisory lock，同一数据库同一时间只允许一个实例运行该命令。 + +如果需要更保守： + +```bash +python manage.py upload_mdy_plate_order_staging_images_to_qiniu --qps 1 --scan-batch-size 10 --upload-batch-size 2 --loop-sleep-seconds 0.5 +``` + ## 重要参数 -- `--limit N`：最多处理 N 张候选图片，用于灰度运行。 -- `--start-id ID`：只扫描 `MDYPlateOrderStaging.id >= ID` 的记录。 +- `--scan-batch-size N`：每轮最多扫描 N 条 staging，默认 `20`。 +- `--upload-batch-size N`：每轮最多上传 N 条审计记录，默认 `5`。 +- `--max-upload-attempts N`：最多尝试处理 N 条审计上传记录，达到后退出；用于本地测试/灰度，不传则持续运行到结束。 +- `--loop-sleep-seconds N`：每轮结束后的短暂休眠，默认 `0.2` 秒。 - `--qps N`：限制下载/上传速率；默认读取 `MDY_PLATE_ORDER_STAGING_QINIU_UPLOAD_QPS`，没有配置时为 `2`。传 `0` 表示不限速。 +- `--reset-cursor`：重置 staging 扫描 cursor，从头重新发现审计记录；已有审计记录不会重复创建。 - `--retry-failed`：重试已经明确失败的审计记录。 - `--retry-stale-processing`：重试长时间停在 `processing` 的审计记录。 - `--stale-after-minutes N`：配合 `--retry-stale-processing`，默认 `120` 分钟。 @@ -54,12 +60,18 @@ python manage.py upload_mdy_plate_order_staging_images_to_qiniu --qps 2 - `--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` 秒。 +- `--progress-every-loops N`：每多少轮输出一次心跳，默认每轮输出。 ## 中断与恢复 -审计表就是任务游标： +状态表保存 staging 扫描游标： + +- `last_staging_id`：已经扫描到的 staging id。 +- `staging_finished`：staging 是否已经扫描完成。 +- `last_heartbeat_at`：命令最近一次心跳。 + +审计表保存逐图上传状态： - `success`：已成功，有 `qiniu_key` 和 `qiniu_url`，后续默认跳过。 - `failed`：明确失败，默认跳过；需要传 `--retry-failed` 才会重试。 @@ -80,7 +92,13 @@ python manage.py upload_mdy_plate_order_staging_images_to_qiniu --retry-stale-pr ## 资源占用策略 -命令按 staging 行使用 `iterator()` 分批读取，不会一次性加载全表。 +命令不会对 staging 开长时间全表 iterator。它使用持久 cursor 做 keyset 分批： + +1. 每轮只查询 `id > last_staging_id` 的少量 staging，默认 20 条。 +2. 从这批 staging 中发现图片并创建 `pending` 审计。 +3. 更新状态表 `last_staging_id`。 +4. 每轮只消费少量审计记录，默认 5 条。 +5. 输出心跳，再短暂 sleep。 每张图片独立处理： @@ -92,7 +110,7 @@ python manage.py upload_mdy_plate_order_staging_images_to_qiniu --retry-stale-pr 6. 上传到七牛后关闭响应和临时文件。 7. 单张图成功或失败都立即落库，不开启长事务。 -因此命令的内存占用主要受 `--spool-max-size-mb`、`--download-chunk-size-mb` 和单张图片大小影响，不随总记录量线性增长。 +因此命令的内存占用主要受 `--scan-batch-size`、`--upload-batch-size`、`--spool-max-size-mb`、`--download-chunk-size-mb` 和单张图片大小影响，不随总记录量线性增长。 ## 数据核查 SQL