zalo-icon
facebook-icon
phone-icon
techdata
Xử lý hàng trăm file Excel bằng Python | Automating Excel with Python
Hoàng Minh
September 20, 2026
Data Engineer Python TechData.AI

Mỗi tháng, bộ phận kế toán hoặc vận hành có thể nhận hàng trăm file Excel từ các chi nhánh. Mở từng file, kiểm tra sheet, sao chép dữ liệu và gộp vào một workbook tổng là công việc tốn thời gian, khó kiểm soát và rất dễ sai. Bài viết này xây dựng một ứng dụng Python có thể quét cả thư mục, đọc hàng loạt workbook, kiểm tra schema, gộp dữ liệu, tách file lỗi và tạo báo cáo đối soát.

Mục lục

  1. Bài toán thực tế
  2. Sau bài này bạn sẽ làm được gì?
  3. Kiến thức Python cần dùng
  4. Chuẩn bị môi trường
  5. Hiểu dữ liệu đầu vào
  6. Xây dựng ứng dụng từng bước
  7. Ghép thành chương trình hoàn chỉnh
  8. Kiểm tra kết quả
  9. Những lỗi thường gặp
  10. Nâng cấp thành ứng dụng thực tế
  11. Bài tập thực hành
  12. Project mở rộng
  13. Tổng kết
  14. Tài liệu tham khảo
Bài toán tổng hợp hàng trăm file Excel từ nhiều chi nhánh

1. Bài toán thực tế

Một công ty có 80 cửa hàng. Đầu tháng, mỗi cửa hàng gửi từ hai đến bốn file Excel gồm doanh số, tồn kho và danh sách đơn hàng hoàn trả. Tên file không hoàn toàn thống nhất, một số workbook đổi tên sheet, vài file thiếu cột và có file đang mở nên xuất hiện bản tạm. Nhân viên tổng hợp thường mất gần một ngày để kiểm tra và ghép dữ liệu. Nếu một file bị bỏ sót, báo cáo quản trị có thể thiếu cả doanh thu của một cửa hàng mà không tạo ra lỗi rõ ràng.

Ứng dụng trong bài không chỉ nối các file với nhau. Chương trình lập danh sách file đã nhận, đọc đúng sheet, kiểm tra cột bắt buộc, gắn tên file nguồn vào từng dòng, tách những file không đạt yêu cầu và tạo bảng audit. Nhờ vậy, người vận hành có thể trả lời file nào đã được xử lý, file nào thất bại, nguyên nhân là gì và số lượng bản ghi thay đổi ra sao.

2. Sau bài này bạn sẽ làm được gì?

Project nhận các workbook trong thư mục data/incoming, đọc sheet Sales, chuẩn hóa dữ liệu và tạo file output/monthly_sales_report.xlsx. Workbook kết quả gồm dữ liệu đã gộp, dữ liệu không hợp lệ, báo cáo theo cửa hàng và nhật ký xử lý file. Một file lỗi không làm toàn bộ chương trình mất kết quả; lỗi được ghi lại để người phụ trách xử lý riêng.

3. Kiến thức Python cần dùng

Project sử dụng pathlib để tìm file, Pandas để đọc và gộp DataFrame, openpyxl làm engine Excel và try except để cô lập lỗi theo từng workbook. Function được dùng để tách logic đọc file, kiểm tra schema, chuẩn hóa, tổng hợp và xuất báo cáo.

4. Chuẩn bị môi trường

python -m venv .venv
source .venv/bin/activate
python -m pip install pandas openpyxl
excel-batch-processing/
    data/
        incoming/
    output/
    main.py
    requirements.txt

5. Hiểu dữ liệu đầu vào

Mỗi sheet Sales cần có sáu cột: order_id, order_date, store_id, product_id, quantity và unit_price. Tổ hợp mã đơn hàng và mã sản phẩm được dùng làm khóa dòng. Số lượng phải là số nguyên dương, đơn giá không âm và ngày bán phải chuyển được sang datetime. Tên file nguồn được giữ lại như một cột lineage.

Quy trình gộp nhiều file Excel bằng Python

6. Xây dựng ứng dụng từng bước

Bước 1: Tìm file Excel hợp lệ

from pathlib import Path
import pandas as pd

INPUT_DIR = Path("data/incoming")
OUTPUT_FILE = Path(
    "output/monthly_sales_report.xlsx"
)

REQUIRED_COLUMNS = {
    "order_id",
    "order_date",
    "store_id",
    "product_id",
    "quantity",
    "unit_price",
}


def discover_files(
    input_dir: Path,
) -> list[Path]:
    return sorted(
        path
        for path in input_dir.glob("*.xlsx")
        if not path.name.startswith("~$")
    )

File bắt đầu bằng ~$ thường là file tạm do Excel tạo khi workbook đang mở. Loại file này ngay từ bước discovery giúp tránh lỗi quyền truy cập và dữ liệu không hoàn chỉnh.

Bước 2: Đọc và kiểm tra từng workbook

def read_workbook(
    file_path: Path,
) -> pd.DataFrame:
    frame = pd.read_excel(
        file_path,
        sheet_name="Sales",
        dtype={
            "order_id": "string",
            "store_id": "string",
            "product_id": "string",
        },
    )

    frame.columns = [
        str(column)
        .strip()
        .lower()
        .replace(" ", "_")
        for column in frame.columns
    ]

    missing = sorted(
        REQUIRED_COLUMNS
        - set(frame.columns)
    )

    if missing:
        raise ValueError(
            f"Thiếu cột: {missing}"
        )

    frame["source_file"] = file_path.name
    return frame

Mỗi workbook được kiểm tra độc lập. Nếu sheet không tồn tại hoặc thiếu cột, hàm phát sinh lỗi có nội dung cụ thể. Cột source_file bảo đảm một dòng dữ liệu sau khi gộp vẫn truy ngược được về file ban đầu.

Bước 3: Đọc hàng loạt nhưng không che giấu lỗi

def extract_batch(
    files: list[Path],
) -> tuple[pd.DataFrame, pd.DataFrame]:
    frames = []
    audit_rows = []

    for file_path in files:
        try:
            frame = read_workbook(file_path)
            frames.append(frame)
            audit_rows.append({
                "file_name": file_path.name,
                "status": "success",
                "records": len(frame),
                "error": "",
            })
        except Exception as exc:
            audit_rows.append({
                "file_name": file_path.name,
                "status": "failed",
                "records": 0,
                "error": str(exc),
            })

    if not frames:
        raise RuntimeError(
            "Không có workbook nào đọc thành công"
        )

    combined = pd.concat(
        frames,
        ignore_index=True,
    )

    audit = pd.DataFrame(audit_rows)
    return combined, audit

try except được đặt bên trong vòng lặp để một workbook lỗi không làm mất toàn bộ lô. Tuy nhiên, chương trình không bỏ qua lỗi âm thầm. Mọi lỗi đều xuất hiện trong audit và pipeline dừng nếu không có file nào đọc thành công.

Bước 4: Chuẩn hóa và kiểm tra dữ liệu

def transform(
    frame: pd.DataFrame,
) -> pd.DataFrame:
    cleaned = frame.copy()

    for column in [
        "order_id",
        "store_id",
        "product_id",
    ]:
        cleaned[column] = (
            cleaned[column]
            .astype("string")
            .str.strip()
            .replace("", pd.NA)
        )

    cleaned["order_date"] = pd.to_datetime(
        cleaned["order_date"],
        errors="coerce",
        dayfirst=True,
    )

    cleaned["quantity"] = pd.to_numeric(
        cleaned["quantity"],
        errors="coerce",
    )

    cleaned["unit_price"] = pd.to_numeric(
        cleaned["unit_price"],
        errors="coerce",
    )

    cleaned["revenue"] = (
        cleaned["quantity"]
        * cleaned["unit_price"]
    )

    return cleaned
def validate(
    frame: pd.DataFrame,
) -> tuple[pd.DataFrame, pd.DataFrame]:
    checked = frame.copy()

    invalid = (
        checked["order_id"].isna()
        | checked["store_id"].isna()
        | checked["product_id"].isna()
        | checked["order_date"].isna()
        | checked["quantity"].isna()
        | (checked["quantity"] <= 0)
        | checked["unit_price"].isna()
        | (checked["unit_price"] < 0)
    )

    return (
        checked.loc[~invalid].copy(),
        checked.loc[invalid].copy(),
    )

Bước 5: Tổng hợp và xuất workbook

def build_store_report(
    valid_rows: pd.DataFrame,
) -> pd.DataFrame:
    return (
        valid_rows
        .groupby(
            "store_id",
            as_index=False,
        )
        .agg(
            orders=("order_id", "nunique"),
            quantity=("quantity", "sum"),
            revenue=("revenue", "sum"),
        )
        .sort_values(
            "revenue",
            ascending=False,
        )
    )

7. Ghép thành chương trình hoàn chỉnh

def export_report(
    valid_rows: pd.DataFrame,
    invalid_rows: pd.DataFrame,
    store_report: pd.DataFrame,
    audit: pd.DataFrame,
) -> None:
    OUTPUT_FILE.parent.mkdir(
        parents=True,
        exist_ok=True,
    )

    with pd.ExcelWriter(
        OUTPUT_FILE,
        engine="openpyxl",
    ) as writer:
        valid_rows.to_excel(
            writer,
            sheet_name="clean_sales",
            index=False,
        )
        invalid_rows.to_excel(
            writer,
            sheet_name="invalid_rows",
            index=False,
        )
        store_report.to_excel(
            writer,
            sheet_name="store_report",
            index=False,
        )
        audit.to_excel(
            writer,
            sheet_name="file_audit",
            index=False,
        )


def main() -> None:
    files = discover_files(INPUT_DIR)

    if not files:
        raise FileNotFoundError(
            "Không tìm thấy file Excel đầu vào"
        )

    raw_data, audit = extract_batch(files)
    cleaned = transform(raw_data)

    cleaned = cleaned.drop_duplicates(
        subset=["order_id", "product_id"],
        keep="last",
    )

    valid_rows, invalid_rows = validate(
        cleaned
    )

    store_report = build_store_report(
        valid_rows
    )

    export_report(
        valid_rows,
        invalid_rows,
        store_report,
        audit,
    )

    print(f"Files discovered: {len(files)}")
    print(f"Input records: {len(raw_data)}")
    print(f"Valid records: {len(valid_rows)}")
    print(f"Invalid records: {len(invalid_rows)}")


if __name__ == "__main__":
    main()
Kiểm tra cấu trúc và chất lượng file Excel trước khi gộp

8. Kiểm tra kết quả

Files discovered: 214
Files successful: 209
Files failed: 5
Input records: 842350
Duplicate removed: 1260
Valid records: 839814
Invalid records: 1276

Trước tiên, kiểm tra sheet file_audit để bảo đảm số file nhận được khớp với danh sách cửa hàng. Sau đó đối soát số bản ghi đầu vào với duplicate, dữ liệu hợp lệ và dữ liệu lỗi. Tổng doanh thu trong store_report phải bằng tổng cột doanh thu của sheet clean_sales.

9. Những lỗi thường gặp

Sheet không tồn tại. Tên sheet có thể có khoảng trắng hoặc thay đổi cách viết. Không nên tự động lấy sheet đầu tiên nếu chưa xác nhận quy ước nghiệp vụ.

PermissionError. File có thể đang mở hoặc output đang được người dùng sử dụng. Hãy đóng workbook hoặc xuất kết quả với tên có thời gian chạy.

Định dạng cũ XLS. OpenPyXL xử lý XLSX, không xử lý định dạng XLS cũ. Doanh nghiệp nên chuyển đổi định dạng hoặc cài engine phù hợp sau khi đánh giá rủi ro phụ thuộc.

Workbook quá lớn. Excel không phải định dạng tối ưu cho dữ liệu hàng triệu dòng. Khi dữ liệu tăng, nên chuyển vùng xử lý trung gian sang CSV, Parquet hoặc database.

Tự động hóa báo cáo Excel định kỳ bằng Python

10. Nâng cấp thành ứng dụng thực tế

Production cần có thư mục incoming, processed và quarantine. File chỉ được chuyển sang processed khi toàn bộ bước kiểm tra hoàn tất. File lỗi được chuyển vào quarantine cùng thông báo dễ hiểu. Checksum giúp phát hiện file đã xử lý trước đó ngay cả khi tên file thay đổi.

Pipeline cần logging, run ID, thời gian xử lý, schema version và ngưỡng chất lượng. Khi tỷ lệ file lỗi vượt ngưỡng, hệ thống nên chặn báo cáo hoặc gửi cảnh báo. Dữ liệu sạch có thể được nạp vào Data Warehouse để Power BI đọc từ nguồn quản trị thay vì phụ thuộc vào workbook tổng.

11. Bài tập thực hành

  1. Bổ sung khả năng đọc nhiều sheet có cùng schema trong một workbook.
  2. Tạo mapping để chấp nhận một số biến thể tên cột nhưng vẫn ghi lại schema gốc.
  3. Thêm checksum, thư mục quarantine và cơ chế chạy lại an toàn.
Nâng cấp project Python xử lý Excel thành công cụ dùng thật

12. Project mở rộng

Xây hệ thống Consolidation Service cho dữ liệu từ 100 chi nhánh. Ứng dụng cần có giao diện tải file, lịch sử xử lý, audit theo chi nhánh, cảnh báo schema drift, lưu dữ liệu vào PostgreSQL và dashboard theo dõi file đến đúng hạn. Project portfolio cần kèm unit test, dữ liệu mẫu và tài liệu business rule.

13. Tổng kết

Bạn vừa xây một ứng dụng Python xử lý hàng loạt workbook thay cho thao tác mở và sao chép thủ công. Sản phẩm không chỉ gộp dữ liệu mà còn kiểm tra schema, cô lập file lỗi, giữ lineage, xác thực bản ghi và tạo audit phục vụ đối soát.

14. Tài liệu tham khảo

  1. Pandas documentation, pandas.read_excel: https://pandas.pydata.org/docs/reference/api/pandas.read_excel.html
  2. Pandas documentation, pandas.ExcelFile: https://pandas.pydata.org/docs/reference/api/pandas.ExcelFile.html
  3. Pandas documentation, pandas.concat: https://pandas.pydata.org/docs/reference/api/pandas.concat.html
  4. Pandas documentation, ExcelWriter: https://pandas.pydata.org/docs/reference/api/pandas.ExcelWriter.html
  5. OpenPyXL documentation: https://openpyxl.readthedocs.io/en/stable/
  6. Python documentation, pathlib: https://docs.python.org/3/library/pathlib.html
🔖Tags: Excel Automation python
Share on Facebook
Share on Twitter
Scroll to Top