"""现货采样确认的导入模板、校验与原子写入。""" 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}