1
0
forked from erp-dev/erp

feat: haobuye

This commit is contained in:
2026-03-20 20:00:12 +08:00
parent 84dbb2ee10
commit 96cc67706a
10 changed files with 1162 additions and 0 deletions

View File

@@ -828,6 +828,31 @@ class CustomerAPITestCase(TestCase):
self.assertEqual(len(results), 1) self.assertEqual(len(results), 1)
self.assertEqual(results[0]['id'], customer_a.id) self.assertEqual(results[0]['id'], customer_a.id)
def test_filter_customers_by_visible_employee_name_excludes_unbound_customers(self):
"""测试 visible_employee_name 过滤不会返回没有绑定员工的客户"""
visible_employee = Employee.objects.create(
merchant=self.merchant,
name='李四业务',
)
unbound_customer = Customer.objects.create(
merchant=self.merchant,
name='未绑定客户',
created_by=self.employee,
)
bound_customer = Customer.objects.create(
merchant=self.merchant,
name='已绑定客户',
created_by=self.employee,
)
bound_customer.visible_employees.add(visible_employee)
response = self.client.get('/api/backend/customers/?visible_employee_name=李四')
self.assertEqual(response.status_code, status.HTTP_200_OK)
results = response.data['results'] if isinstance(response.data, dict) else response.data
returned_ids = {item['id'] for item in results}
self.assertIn(bound_customer.id, returned_ids)
self.assertNotIn(unbound_customer.id, returned_ids)
def test_filter_customers_by_visible_employee_name_returns_distinct_results(self): def test_filter_customers_by_visible_employee_name_returns_distinct_results(self):
"""测试 visible_employee_name 过滤不会因多名匹配员工导致客户重复""" """测试 visible_employee_name 过滤不会因多名匹配员工导致客户重复"""
customer = Customer.objects.create( customer = Customer.objects.create(

View File

@@ -0,0 +1,14 @@
from django.core.management.base import BaseCommand
from api_v1.tasks import sync_external_printing_records
class Command(BaseCommand):
help = '从外部 records 接口同步印染订单/任务'
def add_arguments(self, parser):
parser.add_argument('--limit', type=int, default=100)
def handle(self, *args, **options):
payload = sync_external_printing_records.run(limit=options['limit'])
self.stdout.write(self.style.SUCCESS(str(payload)))

View File

@@ -0,0 +1,44 @@
# Generated by Django 5.2.8 on 2026-03-20 00:00
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('api_v1', '0012_datasync_merchant'),
]
operations = [
migrations.CreateModel(
name='PrintingExternalSyncFailure',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('created_at', models.DateTimeField(auto_now_add=True, db_index=True, verbose_name='创建时间')),
('updated_at', models.DateTimeField(auto_now=True, verbose_name='更新时间')),
('run_date', models.DateField(db_index=True, verbose_name='任务日期')),
('external_record_id', models.PositiveBigIntegerField(db_index=True, verbose_name='外部记录ID')),
('external_order_id', models.CharField(blank=True, max_length=100, verbose_name='外部订单编号')),
('product_name', models.CharField(blank=True, max_length=200, verbose_name='外部产品名')),
('error', models.TextField(blank=True, verbose_name='错误摘要')),
('raw', models.JSONField(blank=True, default=dict, verbose_name='原始数据')),
('attempts', models.PositiveIntegerField(default=0, verbose_name='尝试次数')),
('last_attempt_at', models.DateTimeField(blank=True, null=True, verbose_name='最后尝试时间')),
],
options={
'verbose_name': '外部印染同步失败记录',
'verbose_name_plural': '外部印染同步失败记录',
'db_table': 'api_printing_external_sync_failure',
'ordering': ['-created_at'],
},
),
migrations.AddConstraint(
model_name='printingexternalsyncfailure',
constraint=models.UniqueConstraint(fields=('run_date', 'external_record_id'), name='uniq_printing_ext_sync_fail_run_date_record'),
),
migrations.AlterField(
model_name='datasync',
name='table_name',
field=models.CharField(choices=[('product', '产品'), ('customer', '客户'), ('plate_order', '开版(明道云)'), ('printing_external_record', '印染订单(外部 records'), ('mdy_plate_order_staging_tiia_upload', '开版暂存-腾讯图库上传')], max_length=50, verbose_name='同步目标'),
),
]

View File

@@ -93,6 +93,7 @@ class DataSync(ModelBase):
PRODUCT = 'product', '产品' PRODUCT = 'product', '产品'
CUSTOMER = 'customer', '客户' CUSTOMER = 'customer', '客户'
PLATE_ORDER = 'plate_order', '开版(明道云)' PLATE_ORDER = 'plate_order', '开版(明道云)'
PRINTING_EXTERNAL_RECORD = 'printing_external_record', '印染订单(外部 records'
MDY_PLATE_ORDER_STAGING_TIIA_UPLOAD = 'mdy_plate_order_staging_tiia_upload', '开版暂存-腾讯图库上传' MDY_PLATE_ORDER_STAGING_TIIA_UPLOAD = 'mdy_plate_order_staging_tiia_upload', '开版暂存-腾讯图库上传'
table_name = models.CharField(max_length=50, choices=TableName.choices, verbose_name='同步目标') table_name = models.CharField(max_length=50, choices=TableName.choices, verbose_name='同步目标')
@@ -221,6 +222,38 @@ class MDYPlateOrderStagingTiiaUploadFailure(ModelBase):
return f'{self.run_date} {self.mdy_rowid}' return f'{self.run_date} {self.mdy_rowid}'
class PrintingExternalSyncFailure(ModelBase):
"""
外部印染 records 同步失败记录。
以外部 record.ID 为追踪主键,允许同一天重复失败时累加 attempts。
"""
run_date = models.DateField(db_index=True, verbose_name='任务日期')
external_record_id = models.PositiveBigIntegerField(db_index=True, verbose_name='外部记录ID')
external_order_id = models.CharField(max_length=100, blank=True, verbose_name='外部订单编号')
product_name = models.CharField(max_length=200, blank=True, verbose_name='外部产品名')
error = models.TextField(blank=True, verbose_name='错误摘要')
raw = models.JSONField(default=dict, blank=True, verbose_name='原始数据')
attempts = models.PositiveIntegerField(default=0, verbose_name='尝试次数')
last_attempt_at = models.DateTimeField(null=True, blank=True, verbose_name='最后尝试时间')
class Meta:
db_table = 'api_printing_external_sync_failure'
verbose_name = '外部印染同步失败记录'
verbose_name_plural = '外部印染同步失败记录'
ordering = ['-created_at']
constraints = [
models.UniqueConstraint(
fields=['run_date', 'external_record_id'],
name='uniq_printing_ext_sync_fail_run_date_record',
)
]
def __str__(self):
return f'{self.external_record_id} @ {self.run_date}'
class ApiAuditLog(models.Model): class ApiAuditLog(models.Model):
"""API审计日志 - 记录创建操作的历史现场 """API审计日志 - 记录创建操作的历史现场

View File

@@ -1,19 +1,27 @@
import asyncio import asyncio
import base64
import json import json
import logging import logging
import mimetypes
import os import os
import re
import shutil import shutil
import subprocess import subprocess
from collections import defaultdict
from datetime import datetime from datetime import datetime
from decimal import Decimal, InvalidOperation from decimal import Decimal, InvalidOperation
from pathlib import Path from pathlib import Path
import requests
from celery import shared_task from celery import shared_task
from django.conf import settings from django.conf import settings
from django.contrib.auth import get_user_model
from django.core.files.base import ContentFile
from django.utils import timezone from django.utils import timezone
from basic_info import models as basic_models from basic_info import models as basic_models
from api_v1 import models as api_models from api_v1 import models as api_models
from api_v1.views.printing.services import PrintingJobService, PrintingOrderService
from flower.utils import ( from flower.utils import (
fetch_products_from_mingdaoyun, fetch_products_from_mingdaoyun,
fetch_customers_from_mingdaoyun, fetch_customers_from_mingdaoyun,
@@ -23,9 +31,11 @@ from api_v1.mdy_plate_order_staging_tiia_upload import (
build_default_tiia_rate_limiter, build_default_tiia_rate_limiter,
upload_mdy_plate_order_staging_plate_images_to_tencent_tiia, upload_mdy_plate_order_staging_plate_images_to_tencent_tiia,
) )
from printing import models as printing_models
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
User = get_user_model()
def _ensure_backup_dir(output_dir: str | None) -> Path: def _ensure_backup_dir(output_dir: str | None) -> Path:
@@ -171,6 +181,417 @@ def _ensure_int(value):
return None return None
def _parse_external_datetime(value: str | None):
if not value:
return None
text = str(value).strip()
if not text:
return None
try:
dt = datetime.fromisoformat(text.replace('Z', '+00:00'))
except ValueError:
return None
if timezone.is_naive(dt):
dt = timezone.make_aware(dt, timezone.get_current_timezone())
return dt
def _extract_positive_int(value, *, field_name: str, allow_blank: bool = True):
if value is None or value == '':
return None if allow_blank else 0
try:
number = Decimal(str(value))
except (InvalidOperation, TypeError, ValueError) as exc:
raise RuntimeError(f'外部字段 {field_name} 无法解析为整数: {value!r}') from exc
if number != number.to_integral_value():
raise RuntimeError(f'外部字段 {field_name} 不是整数: {value!r}')
parsed = int(number)
if parsed < 0:
raise RuntimeError(f'外部字段 {field_name} 不能为负数: {value!r}')
return parsed
def _extract_pieces(value):
if value is None or value == '':
return None
match = re.search(r'(\d+)', str(value))
if not match:
return None
return int(match.group(1))
def _normalize_unit(value):
text = str(value or '').strip()
return text.lstrip('/').strip() or text
def _split_fabric_source_and_craft(value: str | None):
text = str(value or '').strip()
if not text:
return '', ''
parts = re.split(r'\s+', text)
fabric_source = parts[0] if parts else ''
craft = ' '.join(parts[1:]).strip() if len(parts) > 1 else ''
return fabric_source, craft
def _build_external_url(path: str) -> str:
base_url = str(getattr(settings, 'PRINTING_EXTERNAL_RECORDS_BASE_URL', '') or '').rstrip('/')
if not base_url:
raise RuntimeError('未配置 PRINTING_EXTERNAL_RECORDS_BASE_URL')
return f'{base_url}{path}'
def _get_external_headers() -> dict[str, str]:
secret = str(getattr(settings, 'PRINTING_EXTERNAL_RECORDS_AUTHORIZATION', '') or '').strip()
if not secret:
raise RuntimeError('未配置 PRINTING_EXTERNAL_RECORDS_AUTHORIZATION')
return {'Authorization': secret}
def _get_printing_sync_user():
user_id = int(getattr(settings, 'PRINTING_EXTERNAL_SYNC_USER_ID', 1) or 1)
user = User.objects.filter(id=user_id).select_related('employee__merchant').first()
if not user:
raise RuntimeError(f'未找到固定同步用户: {user_id}')
employee = getattr(user, 'employee', None)
merchant = getattr(employee, 'merchant', None)
if not merchant:
raise RuntimeError(f'固定同步用户 {user_id} 未绑定 employee.merchant无法确定同步商户')
return user
def _get_printing_sync_product_category(merchant):
category_id = getattr(settings, 'PRINTING_EXTERNAL_SYNC_PRODUCT_CATEGORY_ID', None)
qs = basic_models.ProductCategory.objects.filter(merchant=merchant).order_by('id')
if category_id:
qs = qs.filter(id=category_id)
category = qs.first()
if not category:
raise RuntimeError(
f'未找到用于外部印染同步的产品类别,请先创建类别或配置 '
f'PRINTING_EXTERNAL_SYNC_PRODUCT_CATEGORY_IDmerchant={merchant.id}'
)
return category
def _fetch_external_printing_records(*, limit: int = 100) -> dict:
url = _build_external_url('/api/v1/records')
response = requests.get(
url,
headers=_get_external_headers(),
params={'limit': max(1, int(limit)), 'update_cursor': 'false'},
timeout=30,
)
response.raise_for_status()
payload = response.json()
if not isinstance(payload, dict):
raise RuntimeError('外部 records 接口返回格式非法:非 JSON object')
if not isinstance(payload.get('records', []), list):
raise RuntimeError('外部 records 接口返回格式非法records 不是数组')
return payload
def _advance_external_printing_cursor(*, cursor_value: int) -> dict:
url = _build_external_url('/api/v1/cursor/set')
response = requests.post(
url,
headers=_get_external_headers(),
json={'value': int(cursor_value)},
timeout=30,
)
response.raise_for_status()
payload = response.json()
if not isinstance(payload, dict):
raise RuntimeError('外部 cursor/set 接口返回格式非法:非 JSON object')
return payload
def _group_external_records(records: list[dict]) -> dict[str, list[dict]]:
grouped: dict[str, list[dict]] = defaultdict(list)
for record in records:
external_order_id = str(record.get('BianHaoID') or '').strip()
if not external_order_id:
raise RuntimeError(f'外部 record 缺少 BianHaoID: {record!r}')
grouped[external_order_id].append(record)
return grouped
def _resolve_external_customer(*, merchant, record: dict):
customer_payload = record.get('customer') or {}
customer_name = str(customer_payload.get('KhName') or '').strip()
external_customer_id = str(record.get('KhID') or customer_payload.get('KhID') or '').strip()
if not customer_name:
raise RuntimeError(f'外部 record 缺少 customer.KhName: {record!r}')
customer = (
basic_models.Customer.objects.filter(merchant=merchant, name=customer_name)
.order_by('id')
.first()
)
if not customer:
raise RuntimeError(f'未找到客户: {customer_name}')
return customer, customer_name, external_customer_id
def _resolve_external_employee_user(*, merchant, external_employee_name: str, sync_user):
normalized = str(external_employee_name or '').strip()
if not normalized:
return sync_user, ''
employee = (
basic_models.Employee.objects.filter(merchant=merchant, name=normalized)
.select_related('sys_user')
.order_by('id')
.first()
)
if employee and employee.sys_user_id:
return employee.sys_user, ''
return sync_user, normalized
def _encode_external_image_name_b64(name: str) -> str:
raw = str(name or '').encode('utf-8')
return base64.urlsafe_b64encode(raw).decode('ascii').rstrip('=')
def _decode_external_image_data(data: str) -> bytes:
text = str(data or '').strip()
if not text:
raise RuntimeError('外部图片接口返回空 data')
padding = '=' * (-len(text) % 4)
try:
return base64.b64decode(text + padding)
except Exception as exc:
raise RuntimeError('外部图片接口返回的 data 不是合法 base64') from exc
def _fetch_external_product_image(external_product_name: str) -> dict:
url = _build_external_url('/api/v1/image')
response = requests.get(
url,
headers=_get_external_headers(),
params={'name_b64': _encode_external_image_name_b64(external_product_name)},
timeout=30,
)
if response.status_code == 404:
raise RuntimeError(f'外部图片不存在: {external_product_name}')
response.raise_for_status()
payload = response.json()
if not isinstance(payload, dict):
raise RuntimeError('外部图片接口返回格式非法:非 JSON object')
data = _decode_external_image_data(payload.get('data', ''))
content_type = str(payload.get('content_type') or '').strip() or 'application/octet-stream'
image_name = str(payload.get('name') or '').strip()
return {
'bytes': data,
'content_type': content_type,
'name': image_name or external_product_name,
}
def _build_product_image_filename(product_name: str, image_name: str, content_type: str) -> str:
suffix = Path(str(image_name or '').strip()).suffix
if not suffix:
suffix = mimetypes.guess_extension(content_type or '') or '.jpg'
safe_name = re.sub(r'[^0-9A-Za-z._-]+', '_', str(product_name or '').strip()) or 'external_product'
return f'{safe_name}{suffix}'
def _upload_product_image(product, image_payload: dict):
filename = _build_product_image_filename(
product_name=product.name,
image_name=image_payload.get('name', ''),
content_type=image_payload.get('content_type', ''),
)
content = ContentFile(image_payload['bytes'])
product.image.save(filename, content, save=False)
product.save(update_fields=['image', 'updated_at'])
def _record_printing_external_sync_failure(*, run_date, record: dict, error: str):
external_record_id = _extract_positive_int(record.get('ID'), field_name='ID', allow_blank=False)
obj, _created = api_models.PrintingExternalSyncFailure.objects.get_or_create(
run_date=run_date,
external_record_id=external_record_id,
defaults={
'external_order_id': str(record.get('BianHaoID') or '').strip(),
'product_name': str(record.get('YanSe') or '').strip(),
'error': error or '',
'raw': record,
'attempts': 0,
'last_attempt_at': timezone.now(),
},
)
obj.external_order_id = str(record.get('BianHaoID') or '').strip()
obj.product_name = str(record.get('YanSe') or '').strip()
obj.error = error or ''
obj.raw = record
obj.attempts = int(obj.attempts or 0) + 1
obj.last_attempt_at = timezone.now()
obj.save(
update_fields=[
'external_order_id',
'product_name',
'error',
'raw',
'attempts',
'last_attempt_at',
]
)
def _ensure_external_product(*, merchant, external_product_name: str, category):
normalized = str(external_product_name or '').strip()
if not normalized:
raise RuntimeError('外部产品名为空,无法创建/绑定产品')
existing_product = (
basic_models.Product.objects.filter(merchant=merchant, name=normalized)
.order_by('id')
.first()
)
if existing_product:
return existing_product, False
image_payload = _fetch_external_product_image(normalized)
product = basic_models.Product.objects.create(
merchant=merchant,
category=category,
name=normalized,
)
try:
_upload_product_image(product, image_payload)
except Exception:
if product.image:
try:
product.image.delete(save=False)
except Exception:
logger.warning('回滚外部产品图片失败: product_id=%s', product.id, exc_info=True)
product.delete()
raise
return product, True
def _build_order_external_raw(group_records: list[dict]) -> dict:
first = group_records[0] if group_records else {}
return {
'group_size': len(group_records),
'record_ids': [record.get('ID') for record in group_records],
'first_record': first,
}
def _build_external_order_data(
*,
merchant,
external_order_id: str,
group_records: list[dict],
sync_user,
):
first = group_records[0]
customer, external_customer_name, external_customer_id = _resolve_external_customer(merchant=merchant, record=first)
created_by_user, external_employee_name = _resolve_external_employee_user(
merchant=merchant,
external_employee_name=str(first.get('CaoZY') or '').strip(),
sync_user=sync_user,
)
fabric_source, craft = _split_fabric_source_and_craft(first.get('SHDZ'))
return {
'merchant': merchant,
'customer': customer,
'fabric': '',
'width': str(first.get('SeHao') or '').strip(),
'fabric_source': fabric_source or None,
'craft': craft or None,
'rolling_warn': str(first.get('BeiZhu') or '').strip() or None,
'curve': str(first.get('MeoA') or '').strip() or None,
'outgoing_date': _parse_external_datetime(first.get('KdRiQi')),
'created_by': created_by_user,
'external_order_id': external_order_id,
'external_customer_id': external_customer_id or None,
'external_customer_name': external_customer_name or None,
'external_employee_name': external_employee_name or None,
'external_raw': _build_order_external_raw(group_records),
}
def _upsert_external_printing_order(*, merchant, external_order_id: str, group_records: list[dict], sync_user):
order_data = _build_external_order_data(
merchant=merchant,
external_order_id=external_order_id,
group_records=group_records,
sync_user=sync_user,
)
created_by_user = order_data['created_by']
existing_order = (
printing_models.PrintingOrder.objects.filter(
merchant=merchant,
external_order_id=external_order_id,
)
.order_by('id')
.first()
)
if existing_order:
success, message, order = PrintingOrderService.update_printing_order(
existing_order,
order_data,
created_by_user,
)
if not success:
raise RuntimeError(f'更新印染订单失败: {message}')
return order, False
order = PrintingOrderService.create_printing_order(order_data, created_by_user)
return order, True
def _upsert_external_printing_job(*, merchant, printing_order, record: dict, sync_user, category):
record_id = _extract_positive_int(record.get('ID'), field_name='ID', allow_blank=False)
created_by_user, _external_employee_name = _resolve_external_employee_user(
merchant=merchant,
external_employee_name=str(record.get('CaoZY') or '').strip(),
sync_user=sync_user,
)
product, _ = _ensure_external_product(
merchant=merchant,
external_product_name=str(record.get('YanSe') or '').strip(),
category=category,
)
job_data = {
'merchant': merchant,
'printing_order': printing_order,
'product': product,
'quantity': _extract_positive_int(record.get('ShuLiang'), field_name='ShuLiang', allow_blank=False),
'unit': _normalize_unit(record.get('JiJiaDW')),
'size': str(record.get('ShuLiangZ') or '').strip() or None,
'pieces': _extract_pieces(record.get('BeiZhuC')),
'description': None,
'original_id': record_id,
'created_by': created_by_user,
'external_product_name': None,
'external_raw': record,
}
existing_job = (
printing_models.PrintingJob.objects.filter(
printing_order=printing_order,
original_id=record_id,
)
.order_by('id')
.first()
)
if existing_job:
success, message, job = PrintingJobService.update_printing_job(
existing_job,
job_data,
created_by_user,
)
if not success:
raise RuntimeError(f'更新印染任务失败: {message}')
return job, False
job = PrintingJobService.create_printing_job(job_data, created_by_user)
return job, True
def _run_fetch(page: int, page_size: int): def _run_fetch(page: int, page_size: int):
return asyncio.run(fetch_products_from_mingdaoyun(page=page, page_size=page_size)) return asyncio.run(fetch_products_from_mingdaoyun(page=page, page_size=page_size))
@@ -716,6 +1137,165 @@ def sync_mdy_plate_orders_for_merchant(
return payload return payload
@shared_task(bind=True)
def sync_external_printing_records(self, limit: int = 100):
"""
从外部 records 接口增量同步印染订单/任务。
策略:
- 拉取时强制 update_cursor=false
- 按外部 record.ID 处理失败并记录
- 本批处理结束后无论是否有部分失败,都推进远端 cursor
"""
limit = max(1, int(limit or 100))
sync_user = _get_printing_sync_user()
merchant = sync_user.employee.merchant
category = _get_printing_sync_product_category(merchant)
payload = _fetch_external_printing_records(limit=limit)
records = payload.get('records', [])
last_record_id = payload.get('last_record_id')
run_date = timezone.localdate()
orders_created = 0
orders_updated = 0
jobs_created = 0
jobs_updated = 0
failed_record_ids: list[int] = []
failed_records = 0
if records:
grouped_records = _group_external_records(records)
for external_order_id, group_records in grouped_records.items():
successful_records: list[dict] = []
for record in group_records:
try:
_ensure_external_product(
merchant=merchant,
external_product_name=str(record.get('YanSe') or '').strip(),
category=category,
)
except Exception as exc:
error_message = str(exc)
logger.warning(
'外部印染记录产品处理失败: external_record_id=%s external_order_id=%s error=%s',
record.get('ID'),
external_order_id,
error_message,
)
_record_printing_external_sync_failure(
run_date=run_date,
record=record,
error=error_message,
)
failed_records += 1
failed_record_ids.append(
_extract_positive_int(record.get('ID'), field_name='ID', allow_blank=False)
)
continue
successful_records.append(record)
if not successful_records:
continue
try:
order, order_created = _upsert_external_printing_order(
merchant=merchant,
external_order_id=external_order_id,
group_records=successful_records,
sync_user=sync_user,
)
if order_created:
orders_created += 1
else:
orders_updated += 1
except Exception as exc:
error_message = str(exc)
logger.warning(
'外部印染记录订单处理失败: external_order_id=%s error=%s',
external_order_id,
error_message,
)
for record in successful_records:
_record_printing_external_sync_failure(
run_date=run_date,
record=record,
error=error_message,
)
failed_records += 1
failed_record_ids.append(
_extract_positive_int(record.get('ID'), field_name='ID', allow_blank=False)
)
continue
for record in successful_records:
try:
_job, job_created = _upsert_external_printing_job(
merchant=merchant,
printing_order=order,
record=record,
sync_user=sync_user,
category=category,
)
if job_created:
jobs_created += 1
else:
jobs_updated += 1
except Exception as exc:
error_message = str(exc)
logger.warning(
'外部印染记录任务处理失败: external_record_id=%s external_order_id=%s error=%s',
record.get('ID'),
external_order_id,
error_message,
)
_record_printing_external_sync_failure(
run_date=run_date,
record=record,
error=error_message,
)
failed_records += 1
failed_record_ids.append(
_extract_positive_int(record.get('ID'), field_name='ID', allow_blank=False)
)
cursor_payload = None
if last_record_id is not None:
cursor_payload = _advance_external_printing_cursor(cursor_value=int(last_record_id))
else:
grouped_records = {}
cursor_payload = None
api_models.DataSync.objects.create(
table_name=api_models.DataSync.TableName.PRINTING_EXTERNAL_RECORD,
merchant=merchant,
page_index=1,
page_size=limit,
synced_rows=len(records),
total_count=int(payload.get('count') or 0),
note=f'cursor_after={payload.get("cursor_after")} last_record_id={last_record_id}',
)
result = {
'task_id': self.request.id,
'limit': limit,
'mode': payload.get('mode'),
'count': len(records),
'orders_created': orders_created,
'orders_updated': orders_updated,
'jobs_created': jobs_created,
'jobs_updated': jobs_updated,
'failed_records': failed_records,
'failed_record_ids': failed_record_ids,
'last_record_id': last_record_id,
'cursor_before': payload.get('cursor_before'),
'cursor_after': payload.get('cursor_after'),
'cursor_updated': bool(cursor_payload),
'group_count': len(grouped_records),
}
logger.info('外部印染 records 同步完成: %s', result)
return result
def _record_mdy_plate_order_staging_tiia_failure( def _record_mdy_plate_order_staging_tiia_failure(
*, *,
run_date, run_date,

View File

@@ -0,0 +1,194 @@
from unittest.mock import patch
from django.conf import settings
from django.contrib.auth.models import User
from django.test import TestCase
from api_v1 import models as api_models
from api_v1.tasks import sync_external_printing_records
from basic_info import models as basic_models
from printing import models as printing_models
from stateflow.models import Process, State
class ExternalPrintingRecordsSyncTaskTest(TestCase):
def setUp(self):
self.merchant = basic_models.Merchant.objects.create(
name='测试印染厂',
type=basic_models.MerchantTypeEnum.FACTORY,
)
self.sync_user = User.objects.create_user(username='sync-user', password='testpass')
self.sync_employee = basic_models.Employee.objects.create(
merchant=self.merchant,
sys_user=self.sync_user,
name='同步用户',
)
self.operator_user = User.objects.create_user(username='operator-user', password='testpass')
self.operator_employee = basic_models.Employee.objects.create(
merchant=self.merchant,
sys_user=self.operator_user,
name='钰涵',
)
self.customer = basic_models.Customer.objects.create(
merchant=self.merchant,
name='鸿烨服饰',
)
self.category = basic_models.ProductCategory.objects.create(
merchant=self.merchant,
name='同步产品类',
product_prefix='TP',
)
self.existing_product = basic_models.Product.objects.create(
merchant=self.merchant,
category=self.category,
name='Tj1712#12号色-24码',
)
state = State.objects.create(name='待生产')
self.process = Process.objects.create(name='默认印染流程')
self.process.replace_nodes([state])
settings.PRINTING_DEFAULT_PROCESS_ID = self.process.id
settings.PRINTING_EXTERNAL_SYNC_USER_ID = self.sync_user.id
settings.PRINTING_EXTERNAL_SYNC_PRODUCT_CATEGORY_ID = self.category.id
def _build_payload(self, records):
return {
'mode': 'incremental',
'cursor_before': 10,
'cursor_after': 12,
'last_record_id': records[-1]['ID'],
'updated': False,
'count': len(records),
'records': records,
}
def _build_record(self, *, record_id: int, product_name: str):
return {
'BeiZhu': '手感一定要柔软',
'BeiZhuC': '10件',
'BianHaoID': 'KD20432358',
'BianHaoKD': '1.27',
'CaoZY': '钰涵',
'FidJ': r'\\fw\\2026年-LWQ15\\2026\\H鸿烨\\Tj1712#',
'ID': record_id,
'JiJiaDW': '/段',
'KdRiQi': '2026-01-26T20:00:52Z',
'KhID': 'KH00999',
'MeoA': 'LWQ15',
'RiQi': '2026-01-26T00:00:00Z',
'SHDZ': '客户布 烧花',
'SeHao': '1.51',
'ShuLiang': '10.00',
'ShuLiangZ': '2.68',
'YanSe': product_name,
'customer': {
'KhID': 'KH00999',
'KhName': '鸿烨服饰',
},
}
@patch('api_v1.tasks._advance_external_printing_cursor')
@patch('api_v1.tasks._fetch_external_product_image')
@patch('api_v1.tasks._fetch_external_printing_records')
def test_existing_product_skips_image_fetch(
self,
mock_fetch_records,
mock_fetch_image,
mock_advance_cursor,
):
mock_fetch_records.return_value = self._build_payload(
[self._build_record(record_id=1000000, product_name=self.existing_product.name)]
)
mock_advance_cursor.return_value = {'updated': True}
result = sync_external_printing_records.run(limit=100)
self.assertEqual(result['jobs_created'], 1)
self.assertEqual(result['failed_records'], 0)
self.assertEqual(mock_fetch_image.call_count, 0)
mock_advance_cursor.assert_called_once_with(cursor_value=1000000)
job = printing_models.PrintingJob.objects.get(original_id=1000000)
self.assertEqual(job.product, self.existing_product)
self.assertIsNone(job.external_product_name)
@patch('api_v1.tasks._advance_external_printing_cursor')
@patch('api_v1.tasks._upload_product_image')
@patch('api_v1.tasks._fetch_external_product_image')
@patch('api_v1.tasks._fetch_external_printing_records')
def test_missing_product_creates_product_and_job(
self,
mock_fetch_records,
mock_fetch_image,
mock_upload_product_image,
mock_advance_cursor,
):
new_product_name = 'Tj1712#12号色-25码'
mock_fetch_records.return_value = self._build_payload(
[self._build_record(record_id=1000001, product_name=new_product_name)]
)
mock_fetch_image.return_value = {
'bytes': b'fake-image',
'content_type': 'image/jpeg',
'name': 'abc#123.jpg',
}
mock_advance_cursor.return_value = {'updated': True}
def _fake_upload(product, image_payload):
product.image = 'product_images/generated.jpg'
product.save(update_fields=['image'])
mock_upload_product_image.side_effect = _fake_upload
result = sync_external_printing_records.run(limit=100)
self.assertEqual(result['jobs_created'], 1)
self.assertEqual(result['failed_records'], 0)
mock_fetch_image.assert_called_once_with(new_product_name)
product = basic_models.Product.objects.get(name=new_product_name, merchant=self.merchant)
self.assertIsNotNone(product)
self.assertEqual(product.category, self.category)
job = printing_models.PrintingJob.objects.get(original_id=1000001)
self.assertEqual(job.product, product)
@patch('api_v1.tasks._advance_external_printing_cursor')
@patch('api_v1.tasks._fetch_external_product_image')
@patch('api_v1.tasks._fetch_external_printing_records')
def test_partial_failure_records_external_record_id_and_advances_cursor(
self,
mock_fetch_records,
mock_fetch_image,
mock_advance_cursor,
):
success_record = self._build_record(record_id=1000000, product_name=self.existing_product.name)
failed_record = self._build_record(record_id=1000001, product_name='不存在图片的产品#1')
mock_fetch_records.return_value = self._build_payload([success_record, failed_record])
mock_advance_cursor.return_value = {'updated': True}
def _fake_fetch_image(product_name):
raise RuntimeError(f'外部图片不存在: {product_name}')
mock_fetch_image.side_effect = _fake_fetch_image
result = sync_external_printing_records.run(limit=100)
self.assertEqual(result['jobs_created'], 1)
self.assertEqual(result['failed_records'], 1)
self.assertEqual(result['failed_record_ids'], [1000001])
mock_advance_cursor.assert_called_once_with(cursor_value=1000001)
self.assertTrue(
printing_models.PrintingJob.objects.filter(original_id=1000000).exists()
)
self.assertFalse(
printing_models.PrintingJob.objects.filter(original_id=1000001).exists()
)
failure = api_models.PrintingExternalSyncFailure.objects.get(
external_record_id=1000001
)
self.assertEqual(failure.external_order_id, 'KD20432358')
self.assertIn('外部图片不存在', failure.error)

View File

@@ -0,0 +1,170 @@
# 外部 records 同步到 PrintingOrder / PrintingJob 设计
日期2026-03-20
## 目标
将外部服务 `GET http://43.139.183.222:18080/api/v1/records` 返回的增量 records
按每 5 分钟一次的频率同步到本系统:
- 一组相同 `BianHaoID` 的 records -> 1 条 `printing.PrintingOrder`
- 每条 record -> 1 条 `printing.PrintingJob`
本设计文档用于固化本次实现的字段映射和兜底规则,后续若业务修正,以此为追溯基线。
## 外部 records 样例判断
外部 `records` 不是“订单主表”,而是“印花明细表”。
判断依据:
- 同一个 `BianHaoID` 下会出现多条 record
- 每条 record 含数量、尺寸、单位、产品名等明细信息
- 因此单条 record 更接近本系统的 `PrintingJob`
## 同步主规则
1.`BianHaoID` 对本次拉到的 records 分组
2. 每个分组 upsert 一条 `PrintingOrder`
3. 分组内每条 record 按 `ID` upsert 一条 `PrintingJob`
4. 仅当本地全部写入成功后,才推进外部 cursor
## 字段映射
### 一、外部分组 -> PrintingOrder
| 外部字段 | 含义 | 本系统字段 | 规则 |
| --- | --- | --- | --- |
| `BianHaoID` | 订单编号 | `PrintingOrder.external_order_id` | 分组键;原样保存 |
| `customer.KhName` | 客户名 | `PrintingOrder.customer` | 按本商户 `Customer.name` 精确匹配,取第一条 |
| `KhID` | 外部客户 ID | `PrintingOrder.external_customer_id` | 原样保存 |
| `customer.KhName` | 外部客户名 | `PrintingOrder.external_customer_name` | 原样保存 |
| `CaoZY` | 外部操作员/业务员 | `PrintingOrder.created_by` / `PrintingOrder.external_employee_name` | 若能匹配到内部员工且员工绑定了系统用户,则 `created_by` 取该用户,`external_employee_name` 置空;否则 `created_by` 取固定同步用户,`external_employee_name` 保留原文 |
| `SHDZ` | 布料来源 + 工艺 | `PrintingOrder.fabric_source` / `PrintingOrder.craft` | 以空白字符 split第 1 段为 `fabric_source`,剩余重新拼接为 `craft` |
| `SeHao` | 幅宽 | `PrintingOrder.width` | 原样保存 |
| `MeoA` | 曲线 | `PrintingOrder.curve` | 原样保存 |
| `BeiZhu` | 滚筒注意事项 | `PrintingOrder.rolling_warn` | 原样保存 |
| `KdRiQi` | 下单日期 | `PrintingOrder.outgoing_date` | 按 ISO 时间解析并保存 |
| 分组首条原始数据 | 外部来源快照 | `PrintingOrder.external_raw` | 保存 order 级别的原始数据,便于追溯 |
补充说明:
- 外部没有可靠的 “面料” 字段,因此 `PrintingOrder.fabric` 本次同步固定写空字符串 `""`
- `BianHaoKD``RiQi``FidJ` 等当前未单独属性化的字段,保留在 `external_raw`
- `BianHaoKD` 当前样例值类似 `"1.27"`,不按日期强解析
### 二、单条 record -> PrintingJob
| 外部字段 | 含义 | 本系统字段 | 规则 |
| --- | --- | --- | --- |
| `ID` | 明细顺序 ID | `PrintingJob.original_id` | 幂等键;原样保存 |
| `YanSe` | 外部产品名 | `PrintingJob.product` / `PrintingJob.external_product_name` | 若能按本商户 `Product.name` 精确匹配,直接绑定该产品且不再拉图;若本地不存在,则先创建产品、拉图并上传到 storage成功后绑定该产品 |
| `ShuLiang` | 下单数量 | `PrintingJob.quantity` | 按整数解析;当前预期外部值为整数字符串或 `10.00` 这类可安全转整数的值 |
| `JiJiaDW` | 单位 | `PrintingJob.unit` | 去掉前导 `/` 后保存,例如 `/段` -> `段` |
| `ShuLiangZ` | 一段尺寸 | `PrintingJob.size` | 原样保存 |
| `BeiZhuC` | 件数 | `PrintingJob.pieces` | 提取其中的整数,例如 `10件` -> `10` |
| `YanSe` | 外部产品名 | `PrintingJob.description` | 本次不单独映射到 description原始值由 `product.name``external_product_name` / `external_raw` 表达 |
| 单条原始数据 | 外部来源快照 | `PrintingJob.external_raw` | 保存 job 级别完整原始 record |
## 员工绑定规则
1. 读取外部 `CaoZY`
2. 在同步用户所属商户下按 `Employee.name` 精确匹配
3. 若匹配到的 `Employee.sys_user` 存在:
- `PrintingOrder.created_by = employee.sys_user`
- `PrintingOrder.external_employee_name = ""`
4. 若未匹配到,或员工未绑定系统用户:
- `PrintingOrder.created_by = settings.PRINTING_EXTERNAL_SYNC_USER_ID`
- `PrintingOrder.external_employee_name = 外部原文`
## 客户绑定规则
1. 读取 `customer.KhName`
2. 在同步用户所属商户下按 `Customer.name` 精确匹配
3. 取第一条结果作为 `PrintingOrder.customer`
4. 同时保留:
- `external_customer_id`
- `external_customer_name`
保护规则:
- 若本地找不到客户,则本次同步失败,不推进外部 cursor
- 原因:`PrintingOrder.customer` 在业务上不可空,且当前未授权把客户兜底到固定客户
## 产品绑定与图片规则
1. 读取 `YanSe`
2. 在同步用户所属商户下按 `Product.name` 精确匹配
3. 匹配成功:
- `PrintingJob.product = 匹配到的产品`
- 不请求外部图片 API
- `PrintingJob.external_product_name = ""`
4. 匹配失败:
- 先创建本地 `Product`
- 再调用外部图片 API
- `GET /api/v1/image?name_b64=<base64url(YanSe)>`
- 将返回的 base64 图片内容写入 `Product.image`
- 由 Django storage 自动上传到七牛
- 成功后 `PrintingJob.product = 新建产品`
- `PrintingJob.external_product_name = ""`
5. 若产品创建或图片拉取/上传失败:
- 该条外部 `record.ID` 记为失败
- 该 record 不创建 `PrintingJob`
- 本批次其它成功 records 继续处理
- cursor 仍然推进
## 外部接口与游标策略
外部配置写入 `settings.py`,不经 `.env`
- `PRINTING_EXTERNAL_RECORDS_BASE_URL`
- `PRINTING_EXTERNAL_RECORDS_AUTHORIZATION`
- `PRINTING_EXTERNAL_SYNC_USER_ID`
- `PRINTING_EXTERNAL_SYNC_PRODUCT_CATEGORY_ID`
游标策略:
1. 拉取 records 时显式使用 `update_cursor=false`
2. 本地成功 records / 失败 records 都处理完成后,调用 `/api/v1/cursor/set` 推进到 `last_record_id`
3. 失败 records 需单独落库,便于后续按外部 `record.ID` 补偿
## 调度策略
Celery Beat 每 5 分钟触发一次。
单次请求固定:
- `limit=100`
- 不传范围参数
即依赖外部 cursor 做增量拉取。
## 本次新增字段计划
### PrintingOrder
- `external_order_id`
- `external_customer_id`
- `external_customer_name`
- `external_employee_name`
- `external_raw`
### PrintingJob
- `external_product_name`
- `external_raw`
### API v1 失败记录
- `PrintingExternalSyncFailure`
- `external_record_id`
- `external_order_id`
- `product_name`
- `error`
- `raw`
## 已接受的业务风险
1. `BianHaoID` 是否跨外部系统唯一,不由本系统负责纠偏
2. 客户按名称取第一条,存在误绑风险,但这是本期接受的业务策略
3. 产品若本地不存在,则依赖“创建产品 + 拉图上传”流程;该流程失败时只影响对应 `record.ID`

View File

@@ -476,6 +476,10 @@ STATEFLOW_CURRENT_STATE_MODE = 'NEXT'
# Printing module settings # Printing module settings
PRINTING_DEFAULT_PROCESS_ID = 1 # 默认印染流程ID PRINTING_DEFAULT_PROCESS_ID = 1 # 默认印染流程ID
PRINTING_EXTERNAL_RECORDS_BASE_URL = 'http://43.139.183.222:18080'
PRINTING_EXTERNAL_RECORDS_AUTHORIZATION = 'your-fixed-authorization-secret'
PRINTING_EXTERNAL_SYNC_USER_ID = 1
PRINTING_EXTERNAL_SYNC_PRODUCT_CATEGORY_ID = None
PLATE_ORDER_DEFAULT_PROCESS_ID = 2 # 默认开版流程ID PLATE_ORDER_DEFAULT_PROCESS_ID = 2 # 默认开版流程ID
# 销售品自动创建配置PrintingJob 流程完成时) # 销售品自动创建配置PrintingJob 流程完成时)
@@ -581,6 +585,13 @@ CELERY_BEAT_SCHEDULE = {
'max_records': MDY_SYNC_MAX_RECORDS, 'max_records': MDY_SYNC_MAX_RECORDS,
}, },
}, },
'printing_external_records_sync': {
'task': 'api_v1.tasks.sync_external_printing_records',
'schedule': crontab(minute='*/5'),
'kwargs': {
'limit': 100,
},
},
# 明道云开版:同步到“暂存表” # 明道云开版:同步到“暂存表”
# - 仅在 02:00-08:00 时间窗内持续运行(每 N 分钟触发一次) # - 仅在 02:00-08:00 时间窗内持续运行(每 N 分钟触发一次)
# - 通过 request_interval_seconds 控制单次任务对明道云 API 的请求节奏,避免超 QPS # - 通过 request_interval_seconds 控制单次任务对明道云 API 的请求节奏,避免超 QPS

View File

@@ -0,0 +1,48 @@
# Generated by Django 5.2.8 on 2026-03-20 00:00
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('printing', '0035_merge_20260309_2247'),
]
operations = [
migrations.AddField(
model_name='printingjob',
name='external_product_name',
field=models.CharField(blank=True, max_length=200, null=True, verbose_name='外部产品名'),
),
migrations.AddField(
model_name='printingjob',
name='external_raw',
field=models.JSONField(blank=True, default=dict, help_text='保存外部 record 的完整原始数据', verbose_name='外部原始数据'),
),
migrations.AddField(
model_name='printingorder',
name='external_customer_id',
field=models.CharField(blank=True, max_length=100, null=True, verbose_name='外部客户ID'),
),
migrations.AddField(
model_name='printingorder',
name='external_customer_name',
field=models.CharField(blank=True, max_length=100, null=True, verbose_name='外部客户名'),
),
migrations.AddField(
model_name='printingorder',
name='external_employee_name',
field=models.CharField(blank=True, max_length=100, null=True, verbose_name='外部员工名'),
),
migrations.AddField(
model_name='printingorder',
name='external_order_id',
field=models.CharField(blank=True, db_index=True, max_length=100, null=True, verbose_name='外部订单编号'),
),
migrations.AddField(
model_name='printingorder',
name='external_raw',
field=models.JSONField(blank=True, default=dict, help_text='保存外部 records 聚合后的 order 级原始数据快照', verbose_name='外部原始数据'),
),
]

View File

@@ -363,6 +363,37 @@ class PrintingOrder(ModelBase):
related_name='printing_orders', related_name='printing_orders',
verbose_name='创建人', verbose_name='创建人',
) )
external_order_id = models.CharField(
max_length=100,
blank=True,
null=True,
db_index=True,
verbose_name='外部订单编号',
)
external_customer_id = models.CharField(
max_length=100,
blank=True,
null=True,
verbose_name='外部客户ID',
)
external_customer_name = models.CharField(
max_length=100,
blank=True,
null=True,
verbose_name='外部客户名',
)
external_employee_name = models.CharField(
max_length=100,
blank=True,
null=True,
verbose_name='外部员工名',
)
external_raw = models.JSONField(
default=dict,
blank=True,
verbose_name='外部原始数据',
help_text='保存外部 records 聚合后的 order 级原始数据快照',
)
def __str__(self): def __str__(self):
return self.human_id return self.human_id
@@ -462,6 +493,18 @@ class PrintingJob(ModelBase):
related_name='printing_jobs', related_name='printing_jobs',
verbose_name='创建人', verbose_name='创建人',
) )
external_product_name = models.CharField(
max_length=200,
blank=True,
null=True,
verbose_name='外部产品名',
)
external_raw = models.JSONField(
default=dict,
blank=True,
verbose_name='外部原始数据',
help_text='保存外部 record 的完整原始数据',
)
def __str__(self): def __str__(self):
return self.printing_order.human_id return self.printing_order.human_id