学员资料批量导入总丢行：行级校验、错误定位与重试幂等的落地实践

CREATE TABLE import_batch (
  batch_id       BIGINT PRIMARY KEY AUTO_INCREMENT,
  file_name      VARCHAR(255) NOT NULL,
  total_rows     INT NOT NULL,
  status         VARCHAR(16) NOT NULL DEFAULT 'PROCESSING',
  created_at     DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
);

CREATE TABLE import_row (
  id            BIGINT PRIMARY KEY AUTO_INCREMENT,
  batch_id      BIGINT NOT NULL,
  row_no        INT NOT NULL,           -- Excel 里的行号
  raw_data      JSON NOT NULL,          -- 原始行数据
  status        VARCHAR(16) NOT NULL,   -- PENDING/VALID/INVALID/INSERTED/FAILED
  error_msg     VARCHAR(500),           -- 校验/写入失败原因
  updated_at    DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  KEY idx_batch_status (batch_id, status),
  UNIQUE KEY uk_batch_row (batch_id, row_no)
);

def validate_row(row, row_no, context):
    errors = []
    phone = str(row.get("phone", "")).strip()
    if not re.fullmatch(r"1[3-9]\d{9}", phone):
        errors.append(f"手机号格式不正确：{phone}")
    student_no = str(row.get("student_no", "")).strip()
    if context.student_no_exists(student_no):
        errors.append(f"学号已存在：{student_no}")
    class_id = row.get("class_id")
    if class_id and not context.class_exists(class_id):
        errors.append(f"班级不存在：{class_id}")
    if errors:
        return {"status": "INVALID", "errors": errors}
    return {"status": "VALID"}

def create_or_reuse_batch(file_name):
    row = db.fetchone("SELECT * FROM import_batch WHERE file_name=%s", (file_name,))
    if row and row["status"] in ("PROCESSING", "DONE"):
        return row["batch_id"]
    return db.insert("INSERT INTO import_batch(file_name) VALUES(%s)", (file_name,))

INSERT INTO student (student_no, name, phone, class_id, batch_id)
VALUES (%s, %s, %s, %s, %s)
ON DUPLICATE KEY UPDATE
  name = VALUES(name),
  phone = VALUES(phone),
  updated_at = NOW();

def insert_rows(batch_id, valid_rows, page_size=200):
    for offset in range(0, len(valid_rows), page_size):
        page = valid_rows[offset:offset + page_size]
        try:
            with db.transaction():
                for row in page:
                    db.execute(INSERT_STUDENT_SQL, row_to_params(row))
                db.execute(
                    "UPDATE import_row SET status='INSERTED' WHERE batch_id=%s AND row_no IN (%s)",
                    batch_id, [r["row_no"] for r in page],
                )
        except Exception as e:
            # 本页失败，回滚本页，把行状态置为 FAILED 并记录原因，不阻塞后续页
            db.execute(
                "UPDATE import_row SET status='FAILED', error_msg=%s WHERE batch_id=%s AND row_no IN (%s)",
                str(e)[:200], batch_id, [r["row_no"] for r in page],
            )