zuowei/erpnew
1
0
Fork 0
forked from erp-dev/erp
Code Pull Requests Activity

fix

Browse Source
This commit is contained in:
colaftc
2026-06-28 23:22:21 +08:00
parent 0e3c49ad73
commit 7c9b77afb0
6 changed files with 477 additions and 134 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
api_v1/admin.py
View File

@@ -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 = [

388
api_v1/management/commands/upload_mdy_plate_order_staging_images_to_qiniu.py
View File

@@ -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,61 +74,143 @@ 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)
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,
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
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,
)
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,
finally:
self._release_db_lock()
def _run_upload_loop(
self,
*,
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,
) -> 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',
]
)
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,
@@ -126,67 +221,154 @@ class Command(BaseCommand):
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 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
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
return stats
if limit is not None and processed_count >= limit:
self.stdout.write(self.style.SUCCESS(f'limited stop: {stats}'))
return
@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
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
@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)

53
api_v1/mdy_plate_order_staging_qiniu_upload.py
View File

@@ -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,

35
api_v1/migrations/0017_mdyplateorderstagingqiniuimageuploadstate.py Normal file
View File

@@ -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'],
},
),
]

22
api_v1/models.py
View File

@@ -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 同步失败记录。

48
docs/mdy_plate_order_staging_qiniu_image_upload_command_2026-06-28.md
View File

@@ -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