285 lines
15 KiB
Python
285 lines
15 KiB
Python
"""现货采样确认的导入模板、校验与原子写入。"""
|
||
import csv
|
||
import io
|
||
import zipfile
|
||
from datetime import date, datetime
|
||
from decimal import Decimal, InvalidOperation
|
||
from urllib.parse import urlsplit
|
||
|
||
from openpyxl import Workbook, load_workbook
|
||
from openpyxl.styles import Font, PatternFill
|
||
from services.spot_confirmation_edit import validate_confirmation_values
|
||
|
||
|
||
# (模板表头, 数据字段, 所属表, 字段长度上限)
|
||
FIELDS = [
|
||
('申请编号', 'application_id', None, None),
|
||
('日期', 'application_date', 'application', None),
|
||
('品牌', 'brand', 'application', 200),
|
||
('产品名称', 'product_name', 'application', 300),
|
||
('图片', 'image_path', 'application', 500),
|
||
('条形码', 'barcode', 'application', 100),
|
||
('采样数量', 'sample_quantity', 'application', None),
|
||
('产品链接', 'product_link', 'application', 1000),
|
||
('部门', 'department', 'application', 100),
|
||
('组别', 'group_name', 'application', 100),
|
||
('申请人', 'applicant_name', 'application', 200),
|
||
('厂家名称', 'manufacturer', 'application', 300),
|
||
('收件信息', 'recipient_info', 'application', 65535),
|
||
('样品状态', 'sample_status', 'confirmation', 50),
|
||
('产品单价', 'product_unit_price', 'confirmation', 1000),
|
||
('OA审批单号', 'oa_approval_no', 'confirmation', 100),
|
||
('寄样快递单号', 'express_tracking_no', 'confirmation', 100),
|
||
('到样日期', 'arrival_date', 'confirmation', None),
|
||
('备注', 'remark', 'confirmation', 65535),
|
||
('领样部门', 'receiving_department', 'confirmation', 100),
|
||
('领样人', 'recipient', 'confirmation', 100),
|
||
('样品确认', 'sample_confirmed', 'confirmation', 50),
|
||
('样品确认原因', 'sample_confirmation_reason', 'confirmation', 65535),
|
||
('产品编码', 'product_code', 'confirmation', 100),
|
||
('货品名称（最终名称）', 'final_product_name', 'confirmation', 300),
|
||
('平台', 'platform', 'confirmation', 200),
|
||
]
|
||
HEADER_MAP = {label: key for label, key, _, _ in FIELDS}
|
||
HEADER_MAP.update({'寄样数量': 'sample_quantity', '寄养数量': 'sample_quantity',
|
||
'厂家': 'manufacturer', '寄养单号': 'express_tracking_no',
|
||
'货品名称': 'final_product_name',
|
||
'样品备注': 'legacy_sample_remark', '样品位置': 'legacy_sample_location'})
|
||
MAX_ROWS = 5000
|
||
MAX_FILE_BYTES = 10 * 1024 * 1024
|
||
|
||
|
||
class ImportValidationError(ValueError):
|
||
def __init__(self, errors):
|
||
self.errors = errors
|
||
super().__init__('；'.join(errors[:20]))
|
||
|
||
|
||
def confirmation_template():
|
||
wb = Workbook()
|
||
sheet = wb.active
|
||
sheet.title = '采样确认'
|
||
sheet.append([f[0] for f in FIELDS])
|
||
sheet.freeze_panes = 'A2'
|
||
for col, (_, key, _, _) in enumerate(FIELDS, 1):
|
||
cell = sheet.cell(1, col)
|
||
cell.font = Font(bold=True, color='FFFFFF')
|
||
cell.fill = PatternFill('solid', fgColor='1677FF')
|
||
sheet.column_dimensions[cell.column_letter].width = 22
|
||
for row in range(2, 102):
|
||
sheet.cell(row, col).number_format = (
|
||
'0' if key in ('application_id', 'sample_quantity') else '@')
|
||
instructions = wb.create_sheet('填写说明')
|
||
for text in [
|
||
'请在“采样确认”工作表第一行表头下填写数据；空白行不导入。',
|
||
'申请编号留空：新增已通过的采样申请及关联确认记录，直接进入采样确认列表。',
|
||
'申请编号填写已有已通过申请的编号：更新该记录；空白字段保留原值，不覆盖为空。',
|
||
'申请编号可在现货采样确认列表的第一列查询；同一文件不能重复填写同一编号。',
|
||
'新增记录必须填写产品名称；日期留空为今天，采样数量留空为0。',
|
||
'申请人作为历史姓名文本保存，不要求对应现有账号。',
|
||
'日期及到样日期格式：YYYY-MM-DD，例如2026-10-14；数量须为不小于0的整数。',
|
||
'条形码、产品编码、快递单号按文本填写，避免前导零丢失；日期也可使用Excel日期单元格。',
|
||
'产品单价按文本保存，可填写价格说明，最多1000个字符；领样部门、领样人按模板填写导入。',
|
||
'样品确认仅支持做/不做；确认原因支持价格高/质感差/资质不全/同质化严重/其他：具体原因。',
|
||
'其他原因必须填写内容，且不能包含价格高、质感差、资质不全、同质化严重。',
|
||
'图片填写已上传的/static/uploads/spot_application/路径或HTTP(S)图片地址，不读取内嵌图片。',
|
||
'产品链接只填写HTTP(S)地址；表格中请使用实际值，不使用公式。',
|
||
'支持.xlsx和.csv，单个文件最多10MB、5000条数据；任何行校验失败时整份文件不写入。',
|
||
'不填写申请编号的记录每次导入均为新增；重复导入前请填写对应编号。',
|
||
]:
|
||
instructions.append([text])
|
||
instructions.column_dimensions['A'].width = 120
|
||
buffer = io.BytesIO()
|
||
wb.save(buffer)
|
||
buffer.seek(0)
|
||
return buffer
|
||
|
||
|
||
def _text(value):
|
||
return str(value).strip() if value is not None else ''
|
||
|
||
|
||
def _integer(value, label, minimum=0):
|
||
try:
|
||
number = Decimal(_text(value))
|
||
if not number.is_finite() or number != number.to_integral_value():
|
||
raise ValueError
|
||
if not minimum <= number <= 2147483647:
|
||
raise ValueError
|
||
return int(number)
|
||
except (InvalidOperation, ValueError):
|
||
raise ValueError(f'{label}须为{minimum}至2147483647之间的整数')
|
||
|
||
|
||
def _date(value, label):
|
||
if isinstance(value, datetime):
|
||
return value.date()
|
||
if isinstance(value, date):
|
||
return value
|
||
for fmt in ('%Y-%m-%d', '%Y/%m/%d', '%Y%m%d'):
|
||
try:
|
||
return datetime.strptime(_text(value), fmt).date()
|
||
except ValueError:
|
||
pass
|
||
raise ValueError(f'{label}格式错误，请填写YYYY-MM-DD')
|
||
|
||
|
||
def parse_confirmation_import(raw, filename):
|
||
if len(raw) > MAX_FILE_BYTES:
|
||
raise ImportValidationError(['文件不能超过10MB'])
|
||
if filename.lower().endswith('.xlsx'):
|
||
try:
|
||
with zipfile.ZipFile(io.BytesIO(raw)) as archive:
|
||
if sum(entry.file_size for entry in archive.infolist()) > 50 * 1024 * 1024:
|
||
raise ImportValidationError(['Excel解压后不能超过50MB'])
|
||
wb = load_workbook(io.BytesIO(raw), read_only=True, data_only=False)
|
||
try:
|
||
ws = wb['采样确认'] if '采样确认' in wb.sheetnames else wb.worksheets[0]
|
||
if (ws.max_column or 0) > 100:
|
||
raise ImportValidationError(['表格列数过多，请使用导入模板'])
|
||
rows = []
|
||
for row_number, row in enumerate(ws.iter_rows(), 1):
|
||
if row_number > 100000:
|
||
raise ImportValidationError(['表格空白行过多，请删除多余空行后重试'])
|
||
if any(cell.data_type == 'f' for cell in row):
|
||
raise ImportValidationError([f'第{row_number}行含公式，请转换为实际值'])
|
||
values = [cell.value for cell in row]
|
||
if row_number == 1 or any(_text(v) for v in values):
|
||
if len(rows) > MAX_ROWS:
|
||
raise ImportValidationError(['单次最多导入5000条数据'])
|
||
rows.append((row_number, values))
|
||
finally:
|
||
wb.close()
|
||
except ImportValidationError:
|
||
raise
|
||
except Exception:
|
||
raise ImportValidationError(['无法读取Excel，请上传有效的.xlsx文件'])
|
||
elif filename.lower().endswith('.csv'):
|
||
for encoding in ('utf-8-sig', 'gb18030'):
|
||
try:
|
||
text = raw.decode(encoding)
|
||
break
|
||
except UnicodeDecodeError:
|
||
continue
|
||
else:
|
||
raise ImportValidationError(['CSV编码无法识别，请使用UTF-8或GBK'])
|
||
try:
|
||
rows = []
|
||
for row_number, row in enumerate(csv.reader(io.StringIO(text)), 1):
|
||
if row_number == 1 or any(_text(v) for v in row):
|
||
if len(rows) > MAX_ROWS:
|
||
raise ImportValidationError(['单次最多导入5000条数据'])
|
||
rows.append((row_number, row))
|
||
except csv.Error:
|
||
raise ImportValidationError(['CSV格式错误，请使用导入模板'])
|
||
else:
|
||
raise ImportValidationError(['仅支持.xlsx或.csv文件'])
|
||
if not rows:
|
||
raise ImportValidationError(['文件为空'])
|
||
headers = [_text(v).lstrip('\ufeff') for v in rows[0][1]]
|
||
while headers and not headers[-1]:
|
||
headers.pop()
|
||
if not headers or any(h not in HEADER_MAP for h in headers):
|
||
raise ImportValidationError(['表头无法识别，请使用导入模板；未知表头：' +
|
||
'、'.join(h or '空白表头' for h in headers if h not in HEADER_MAP)])
|
||
keys = [HEADER_MAP[h] for h in headers]
|
||
if len(set(keys)) != len(keys):
|
||
raise ImportValidationError(['表头重复，请勿同时填写同一字段的新旧名称'])
|
||
if 'product_name' not in keys and 'application_id' not in keys:
|
||
raise ImportValidationError(['表头至少需要包含产品名称或申请编号'])
|
||
parsed, errors, seen_ids = [], [], set()
|
||
specs = {key: (label, limit) for label, key, _, limit in FIELDS}
|
||
specs.update(legacy_sample_remark=('样品备注', 65535), legacy_sample_location=('样品位置', 200))
|
||
for row_number, row in rows[1:]:
|
||
if not any(_text(v) for v in row):
|
||
continue
|
||
try:
|
||
if any(_text(v) for v in row[len(keys):]):
|
||
raise ValueError('数据列数多于表头列数')
|
||
values = {}
|
||
for key, value in zip(keys, row):
|
||
if not _text(value):
|
||
continue
|
||
label, limit = specs[key]
|
||
if key in ('application_id', 'sample_quantity'):
|
||
values[key] = _integer(value, label, 1 if key == 'application_id' else 0)
|
||
elif key in ('application_date', 'arrival_date'):
|
||
values[key] = _date(value, label)
|
||
else:
|
||
values[key] = _text(value)
|
||
if limit and (len(values[key].encode('utf-8')) if limit == 65535 else len(values[key])) > limit:
|
||
raise ValueError(f'{label}内容过长')
|
||
if key in ('image_path', 'product_link'):
|
||
local_image = key == 'image_path' and values[key].startswith('/static/uploads/spot_application/')
|
||
url = urlsplit(values[key])
|
||
if not local_image and not (url.scheme in ('http', 'https') and url.netloc):
|
||
raise ValueError(f'{label}请填写有效HTTP(S)地址' + ('或已上传图片路径' if key == 'image_path' else ''))
|
||
values.update(validate_confirmation_values(values))
|
||
application_id = values.get('application_id')
|
||
if not application_id and not values.get('product_name'):
|
||
raise ValueError('新增记录必须填写产品名称')
|
||
if application_id:
|
||
if application_id in seen_ids:
|
||
raise ValueError('申请编号在本文件中重复')
|
||
seen_ids.add(application_id)
|
||
parsed.append((row_number, values))
|
||
except ValueError as exc:
|
||
errors.append(f'第{row_number}行：{exc}')
|
||
if errors:
|
||
raise ImportValidationError(errors)
|
||
if not parsed:
|
||
raise ImportValidationError(['没有可导入的数据'])
|
||
return parsed
|
||
|
||
|
||
def import_confirmations(rows, actor_user_id, conn):
|
||
"""同一事务写入申请及确认表，校验失败或数据库异常时全部回滚。"""
|
||
added = updated = 0
|
||
try:
|
||
with conn.cursor() as cur:
|
||
cur.execute('SELECT id,username,real_name FROM dashboard_users')
|
||
prepared, errors = [], []
|
||
for row_number, values in rows:
|
||
values = dict(values)
|
||
application_id = values.get('application_id')
|
||
if application_id:
|
||
cur.execute('SELECT id,status FROM spot_applications WHERE id=%s FOR UPDATE', (application_id,))
|
||
existing = cur.fetchone()
|
||
if not existing or existing['status'] != '通过':
|
||
errors.append(f'第{row_number}行：申请编号不存在或未审核通过')
|
||
prepared.append(values)
|
||
if errors:
|
||
raise ImportValidationError(errors)
|
||
for values in prepared:
|
||
application_id = values.pop('application_id', None)
|
||
app_values = {key: values[key] for _, key, table, _ in FIELDS
|
||
if table == 'application' and key in values}
|
||
confirmation = {key: values[key] for _, key, table, _ in FIELDS
|
||
if table == 'confirmation' and key in values}
|
||
if application_id:
|
||
if app_values:
|
||
assignments = ','.join(f'`{key}`=%s' for key in app_values)
|
||
cur.execute(f'UPDATE spot_applications SET {assignments} WHERE id=%s',
|
||
tuple(app_values.values()) + (application_id,))
|
||
updated += 1
|
||
else:
|
||
app_values.setdefault('applicant_user_id', actor_user_id)
|
||
app_values.setdefault('application_date', date.today())
|
||
app_values.setdefault('sample_quantity', 0)
|
||
app_values['status'] = '通过'
|
||
columns = ','.join(f'`{key}`' for key in app_values)
|
||
placeholders = ','.join(['%s'] * len(app_values))
|
||
cur.execute(f'INSERT INTO spot_applications ({columns}) VALUES ({placeholders})', tuple(app_values.values()))
|
||
application_id = cur.lastrowid
|
||
added += 1
|
||
cur.execute('INSERT IGNORE INTO spot_confirmations(application_id) VALUES (%s)', (application_id,))
|
||
if confirmation:
|
||
assignments = ','.join(f'`{key}`=%s' for key in confirmation)
|
||
cur.execute(f'UPDATE spot_confirmations SET {assignments} WHERE application_id=%s',
|
||
tuple(confirmation.values()) + (application_id,))
|
||
conn.commit()
|
||
except Exception:
|
||
conn.rollback()
|
||
raise
|
||
return {'added': added, 'updated': updated, 'total': added + updated}
|