ExcelデータをSQL Serverへインポートする方法：ウィザードとPythonで検証・重複防止

-- 検証用DBの所有者が別途準備する定義例。既存表へ上書きしない。
CREATE TABLE dbo.ExcelInventoryWizardStage (
    ItemCode varchar(20) COLLATE Latin1_General_100_BIN2 NOT NULL PRIMARY KEY,
    Quantity int NOT NULL CHECK (Quantity BETWEEN 0 AND 1000000),
    UnitPrice decimal(10,2) NOT NULL CHECK (UnitPrice >= 0),
    AsOfDate date NOT NULL
);

-- ウィザード用の専用表。元データの件数・数量合計と照合する。
SELECT COUNT_BIG(*) AS ImportedRows,
       SUM(CONVERT(bigint, Quantity)) AS TotalQuantity
FROM dbo.ExcelInventoryWizardStage;

ItemCode | Quantity | UnitPrice | AsOfDate
SKU001   | 12       | 19.90     | 2026-10-20
00002    | 0        | 0.00      | 2026-02-28

python -m pip install pandas openpyxl sqlalchemy pyodbc

-- Schema reference for a dedicated test database, provisioned separately by its owner.
-- Do not run against production or replace existing tables to make this example work.
CREATE TABLE dbo.ExcelImportBatch (
    BatchId char(64) NOT NULL,
    ImportedRowCount int NOT NULL,
    ImportedAt datetime2(0) NOT NULL
        CONSTRAINT DF_ExcelImportBatch_ImportedAt DEFAULT SYSUTCDATETIME(),
    CONSTRAINT PK_ExcelImportBatch PRIMARY KEY (BatchId),
    CONSTRAINT CK_ExcelImportBatch_ImportedRowCount CHECK (ImportedRowCount BETWEEN 1 AND 2000)
);

CREATE TABLE dbo.ExcelInventoryStage (
    BatchId char(64) NOT NULL,
    ItemCode varchar(20) COLLATE Latin1_General_100_BIN2 NOT NULL,
    Quantity int NOT NULL,
    UnitPrice decimal(10,2) NOT NULL,
    AsOfDate date NOT NULL,
    CONSTRAINT PK_ExcelInventoryStage PRIMARY KEY (BatchId, ItemCode),
    CONSTRAINT FK_ExcelInventoryStage_Batch FOREIGN KEY (BatchId)
        REFERENCES dbo.ExcelImportBatch (BatchId),
    CONSTRAINT CK_ExcelInventoryStage_Quantity CHECK (Quantity BETWEEN 0 AND 1000000),
    CONSTRAINT CK_ExcelInventoryStage_Price CHECK (UnitPrice >= 0)
);

"""Synthetic inventory example. Validation is local; --write opts into staging only."""
import argparse
from datetime import date
from decimal import Decimal
import hashlib
from io import BytesIO
import json
import os
from pathlib import Path
import re
import sys
from zipfile import ZipFile

import pandas as pd
from openpyxl import load_workbook

COLUMNS = ["ItemCode", "Quantity", "UnitPrice", "AsOfDate"]
MAX_ROWS = 2000
MAX_BYTES = 5 * 1024 * 1024
MAX_EXPANDED_BYTES = 50 * 1024 * 1024


class ImportRejected(Exception):
    """Messages contain fixed reason codes, never cell values or credentials."""


def read_inventory(path):
    if path.suffix.lower() != ".xlsx":
        raise ImportRejected("E_FILE_TYPE")
    # Read once: validation and parsing use the same snapshot of the file.
    with path.open("rb") as source:
        payload = source.read(MAX_BYTES + 1)
    if not payload or len(payload) > MAX_BYTES:
        raise ImportRejected("E_FILE_SIZE")
    with ZipFile(BytesIO(payload)) as archive:
        entries = archive.infolist()
        if len(entries) > 1000 or sum(e.file_size for e in entries) > MAX_EXPANDED_BYTES:
            raise ImportRejected("E_ARCHIVE_SIZE")
        if any(e.flag_bits & 1 for e in entries):
            raise ImportRejected("E_ENCRYPTED_ARCHIVE")
    # This is a bounded, trusted sample-file reader, not a hostile-file sandbox.
    workbook = load_workbook(BytesIO(payload), data_only=False, keep_links=False)
    try:
        if workbook.sheetnames != ["Inventory"]:
            raise ImportRejected("E_SHEET")
        sheet = workbook["Inventory"]
        if sheet.sheet_state != "visible" or sheet.merged_cells.ranges:
            raise ImportRejected("E_LAYOUT")
        if any(d.hidden for d in sheet.row_dimensions.values()) or any(
            d.hidden for d in sheet.column_dimensions.values()
        ):
            raise ImportRejected("E_HIDDEN_DATA")
        if sheet.max_column != 4 or not 2 <= sheet.max_row <= MAX_ROWS + 1:
            raise ImportRejected("E_DIMENSIONS")
        if [cell.value for cell in sheet[1]] != COLUMNS:
            raise ImportRejected("E_HEADER")
        for row in sheet.iter_rows():
            if any(cell.value is None for cell in row):
                raise ImportRejected("E_EMPTY_CELL")
            if any(cell.data_type in {"f", "e"} for cell in row):
                raise ImportRejected("E_FORMULA_OR_EXCEL_ERROR")
    finally:
        workbook.close()
    frame = pd.read_excel(
        BytesIO(payload), sheet_name="Inventory", engine="openpyxl",
        dtype=object, keep_default_na=False, na_filter=False, nrows=MAX_ROWS + 1,
    )
    if list(frame.columns) != COLUMNS or not 1 <= len(frame) <= MAX_ROWS:
        raise ImportRejected("E_TABLE_SHAPE")
    records = []
    seen = set()
    for code, quantity, price, day in frame.itertuples(index=False, name=None):
        if not isinstance(code, str) or re.fullmatch(r"[A-Z0-9_-]{1,20}", code) is None:
            raise ImportRejected("E_ITEM_CODE")
        if code in seen:
            raise ImportRejected("E_DUPLICATE_ITEM_CODE")
        seen.add(code)
        if isinstance(quantity, bool) or not isinstance(quantity, (int, float)):
            raise ImportRejected("E_QUANTITY_TYPE")
        amount = Decimal(str(quantity))
        if not amount.is_finite() or amount != amount.to_integral_value() or not 0 <= amount <= 1000000:
            raise ImportRejected("E_QUANTITY_RANGE")
        if not isinstance(price, str) or re.fullmatch(r"(?:0|[1-9][0-9]{0,7})\.[0-9]{2}", price) is None:
            raise ImportRejected("E_UNIT_PRICE")
        if not isinstance(day, str) or re.fullmatch(r"[0-9]{4}-[0-9]{2}-[0-9]{2}", day) is None:
            raise ImportRejected("E_DATE_FORMAT")
        try:
            as_of_date = date.fromisoformat(day)
        except ValueError:
            raise ImportRejected("E_DATE_VALUE") from None
        records.append({"ItemCode": code, "Quantity": int(amount),
                        "UnitPrice": Decimal(price), "AsOfDate": as_of_date})
    return sorted(records, key=lambda item: item["ItemCode"])


def batch_id_for(records):
    # Canonical business values, not the XLSX ZIP bytes or its file name.
    canonical = [[r["ItemCode"], r["Quantity"], format(r["UnitPrice"], ".2f"),
                  r["AsOfDate"].isoformat()] for r in records]
    value = json.dumps(["inventory-v1", canonical], ensure_ascii=True, separators=(",", ":"))
    return hashlib.sha256(value.encode("utf-8")).hexdigest()


def make_engine():
    # Import only on the write path; validation does not connect to a database.
    from sqlalchemy import create_engine
    from sqlalchemy.engine import URL
    keys = ["SQLSERVER_HOST", "SQLSERVER_DATABASE", "SQLSERVER_USER", "SQLSERVER_PASSWORD"]
    if not all(os.environ.get(key) for key in keys):
        raise ImportRejected("E_CONNECTION_CONFIG")
    url = URL.create(
        "mssql+pyodbc", host=os.environ["SQLSERVER_HOST"], port=1433,
        database=os.environ["SQLSERVER_DATABASE"],
        username=os.environ["SQLSERVER_USER"], password=os.environ["SQLSERVER_PASSWORD"],
        query={"driver": "ODBC Driver 18 for SQL Server",
               "Encrypt": "yes", "TrustServerCertificate": "no"},
    )
    return create_engine(url, echo=False, hide_parameters=True,
                         connect_args={"timeout": 15})


def check_staged_rows(connection, batch_id, records):
    from sqlalchemy import text
    result = connection.execute(text("""
        SELECT ItemCode, Quantity, UnitPrice, AsOfDate
        FROM dbo.ExcelInventoryStage WHERE BatchId = :batch_id
    """), {"batch_id": batch_id})
    # Do not allow corrupted/incompatible pre-existing batches to use unbounded memory.
    observed = result.fetchmany(MAX_ROWS + 1)
    result.close()
    expected = [tuple(r[column] for column in COLUMNS) for r in records]
    if sorted(tuple(row) for row in observed) != sorted(expected):
        raise ImportRejected("E_STAGING_MISMATCH")


def write_staging(engine, records):
    from sqlalchemy import text
    batch_id = batch_id_for(records)
    outcome = "INSERTED"
    # No AUTOCOMMIT. Both tables must already exist with the documented schema.
    with engine.begin() as connection:
        connection.execute(text("SET XACT_ABORT ON"))
        previous = connection.execute(text("""
            SELECT ImportedRowCount FROM dbo.ExcelImportBatch WITH (UPDLOCK, HOLDLOCK)
            WHERE BatchId = :batch_id
        """), {"batch_id": batch_id}).scalar_one_or_none()
        if previous is not None:
            if previous != len(records):
                raise ImportRejected("E_BATCH_MISMATCH")
            check_staged_rows(connection, batch_id, records)
            outcome = "ALREADY_IMPORTED"
        else:
            connection.execute(text("""
                INSERT INTO dbo.ExcelImportBatch (BatchId, ImportedRowCount)
                VALUES (:batch_id, :row_count)
            """), {"batch_id": batch_id, "row_count": len(records)})
            insert = text("""
                INSERT INTO dbo.ExcelInventoryStage
                    (BatchId, ItemCode, Quantity, UnitPrice, AsOfDate)
                VALUES (:BatchId, :ItemCode, :Quantity, :UnitPrice, :AsOfDate)
            """)
            for offset in range(0, len(records), 200):
                connection.execute(insert, [dict(r, BatchId=batch_id)
                                            for r in records[offset:offset + 200]])
            check_staged_rows(connection, batch_id, records)
    # Only report success after the context manager has committed.
    return outcome


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("xlsx", type=Path)
    parser.add_argument("--write", action="store_true", help="Write only to the two staging tables")
    args = parser.parse_args()
    engine = None
    try:
        records = read_inventory(args.xlsx)
        if not args.write:
            print(f"VALIDATED rows={len(records)}; database not contacted")
            return 0
        engine = make_engine()
        outcome = write_staging(engine, records)
        print(f"{outcome} rows={len(records)}")
        return 0
    except ImportRejected as error:
        # Every ImportRejected in this script has a fixed, non-sensitive message.
        print(str(error), file=sys.stderr)
        return 1
    except Exception:
        # Do not log driver errors, input values, paths, connection URLs or secrets.
        print("E_OPERATION: inspect restricted diagnostics; do not assume commit failed", file=sys.stderr)
        return 1
    finally:
        if engine is not None:
            engine.dispose()


if __name__ == "__main__":
    sys.exit(main())

# まず合成データをローカル検証。DB接続なし。
python import_inventory.py synthetic_inventory.xlsx

# 検証用DB・権限・接続設定を確認してから行う明示的な書き込み。
python import_inventory.py synthetic_inventory.xlsx --write