Build a Supplier Intelligence System
Build a modular ETL pipeline that consolidates supplier data into a PostgreSQL matrix.
Introduction
30 Second Summary
Large organizations track thousands of suppliers across spreadsheets, Word documents, and presentations. Finding which supplier offers the best price, fastest delivery, or greenest footprint means manually digging through files that were never designed to talk to each other.
In this project, you will build a supplier intelligence system that ingests Excel, Word, and PowerPoint files, extracts supplier data, resolves duplicates, and generates a consolidated evaluation matrix across 35+ categories.
What You'll Build
A modular ETL pipeline backed by PostgreSQL that processes heterogeneous documents into a standardized supplier database and exportable evaluation matrix.
By the end of this project, you'll have:
- A PostgreSQL database with a hybrid relational/JSONB schema supporting 35+ dynamic supplier categories.
- An abstract parser framework that extracts structured data from Excel, Word, and PowerPoint files.
- A Pydantic-powered validation and normalization pipeline with idempotent database storage and full audit trails.
- A supplier matrix generator that exports consolidated data to CSV and Excel formats.
- Secret Mission: A CLI interface with progress reporting, dry-run mode, and pipeline statistics.
Are there any prerequisites?
This project assumes strong Python experience including type hints, object-oriented programming, and package structure. No prior database or ETL experience is required.
Set Up Your Environment
Before you can build the supplier intelligence pipeline, you need a working database, a clean Python environment, and a code editor.
In this step, you will install PostgreSQL 17 as the database backend, set up a Python virtual environment, and install Cursor as your code editor.
In this step, get ready to:
- Install PostgreSQL 17 on Windows and verify it is running.
- Create a Python virtual environment and verify PostgreSQL connectivity from Python.
- Install Cursor editor and open it.
Install PostgreSQL 17 on Windows
PostgreSQL is an open-source relational database that will store all the supplier data your pipeline extracts. You need to install it locally so your Python code can connect to it.
- Open your browser and navigate to the PostgreSQL Windows download page:
- https://www.postgresql.org/download/windows/
- Click Download the installer certified by EDB.
- On the EDB downloads page, select PostgreSQL version 17 for Windows x86-64.
- Run the downloaded installer file.
- Accept the default installation directory.
- When prompted to select components, keep all defaults selected (PostgreSQL Server, pgAdmin 4, Stack Builder, Command Line Tools).
- When prompted for a password, enter a password for the postgres superuser. Remember this password. For this project, using postgres as the password keeps things simple.
- Keep the default port 5432.
- Keep the default locale and click through the remaining screens to finish installation.
- When the installer completes, uncheck Launch Stack Builder and click Finish.
What did the installer just do?
The EDB installer set up PostgreSQL 17 as a Windows service. This means the database server starts automatically whenever your computer boots.
It also added pgAdmin 4 (a graphical database manager) and command-line tools like psql for interacting with the database from your terminal.
Now verify that PostgreSQL is running. The installer should have added psql to your system PATH.
- Open a new terminal window and check the PostgreSQL version by running:
psql --version✔️ I see a version number starting with 17
PostgreSQL is installed and the command-line tools are on your PATH. You are ready to move on.
ⓧ I see a version lower than 17
You have an older version of PostgreSQL installed. You need to uninstall the old version first, then re-run the PostgreSQL 17 installer.
- Press the Windows key and type Apps & features.
- Open Apps & features from the results.
- Find the old PostgreSQL entry, select it, and click Uninstall.
- After uninstalling, re-run the PostgreSQL 17 installer you downloaded earlier.
- Close and reopen your terminal, then run psql --version again.
ⓧ Command not recognized
The PostgreSQL bin directory is not in your system PATH. You need to add it manually.
- Press the Windows key and type Environment Variables.
- Click Edit the system environment variables.
- Click Environment Variables.
- Under System variables, select Path and click Edit.
- Click New and add C:\Program Files\PostgreSQL\17\bin.
- Click OK on all dialogs to save.
- Close and reopen your terminal, then run psql --version again.
Create a Python virtual environment and verify connectivity
A virtual environment isolates your project's packages from your system Python. This prevents version conflicts between projects.
- Move to your Desktop by running this command:
cd ~/Desktop- Create your project folder and move into it by running these commands:
mkdir supplier-intel
cd supplier-intel- Create the virtual environment by running:
python -m venv .venv- Activate the virtual environment by running:
.venv\Scripts\activateYou should see (.venv) at the start of your terminal prompt. This confirms the virtual environment is active.
Not seeing (.venv) in your prompt?
If you get an error about running scripts being disabled, you need to adjust your execution policy. Run this command first, then try activating again:
Set-ExecutionPolicy -Scope CurrentUser -ExecutionPolicy RemoteSignedHelp me fix the virtual environment activation error
Now verify that Python can talk to PostgreSQL. You need to install the psycopg2-binary driver and test the connection.
- Install the PostgreSQL driver by running:
pip install psycopg2-binary- Create the database your project will use by running:
psql -U postgres -c "CREATE DATABASE supplier_intel;"What does this command do?
This runs psql (the PostgreSQL command-line client) as the postgres superuser and executes a SQL command to create a new database called supplier_intel.
When prompted for a password, enter the password you set during PostgreSQL installation.
- Test the Python connection by running:
python -c "import psycopg2; conn = psycopg2.connect(dbname='supplier_intel', user='postgres', password='postgres', host='localhost', port='5432'); print('Connected successfully!'); conn.close()"You should see Connected successfully! printed in your terminal.
Seeing a connection refused error?
Make sure the PostgreSQL service is running. Press the Windows key, type Services, and open the Services app. Look for a service named postgresql-x64-17 and confirm its status is Running. If it is stopped, right-click it and select Start.
If your password is different from postgres, replace the password value in the connection command with the one you chose during installation.
Help me fix the PostgreSQL connection error
Install Cursor editor
Cursor is a code editor built on VS Code with AI features baked in. You will use it to write and navigate the project code.
- Open your browser and go to the Cursor download page:
- https://cursor.com/download
- Click the Windows download button.
- Run the downloaded installer and follow the prompts.
- When the installer finishes, press the Windows key, type Cursor, and press Enter to open it.
- Sign up for a free Cursor account when prompted, or sign in if you already have one.
Do I need to pay for Cursor?
No. Cursor has a free tier that works for everything in this project. You do not need a paid plan.
- In Cursor, click File in the top menu bar.
- Click Open Folder.
- Navigate to your Desktop and select the supplier-intel folder, then click Select Folder.
You should see the supplier-intel folder open in Cursor's Explorer sidebar on the left.
Cursor installer not working?
Make sure you are running Windows 10 or later. The Cursor installer does not support older versions of Windows.
If you see a security warning, click More info and then click Run anyway to proceed.
Help me fix the Cursor installation issue
Your environment is ready. Next up, you will create the project structure and install all the Python dependencies.
Create Project Structure and Install Dependencies
Your environment is ready, but right now it is just an empty virtual environment with PostgreSQL running in the background. To build a production-grade pipeline, you need a modular project structure where each concern (parsing, validation, normalization, storage) lives in its own module.
In this step, you will scaffold the directory layout, pin all dependencies, and create the configuration module that every other module will import. This foundation ensures that future steps slot cleanly into place without refactoring.
In this step, get ready to:
- Create the modular directory structure with proper Python packaging.
- Install all project dependencies into your virtual environment.
- Create the configuration module with database URL and structured logging.
Create the modular project directory structure
A well-organized Python project uses packages (directories with __init__.py files) to group related modules together. Your pipeline will live inside a supplier_intel package, and raw document files will go into a data directory.
- In your terminal (with the virtual environment still activated), create the project directories by running these commands:
mkdir supplier_intel
mkdir data- Confirm the directories exist by running:
lsYou should see supplier_intel and data listed alongside your virtual environment folder.
- In Cursor, create a new file inside the supplier_intel folder called __init__.py and paste the following code:
"""Supplier Intelligence System - Phase 1: Data Extraction and Matrix Generation."""
__version__ = "1.0.0"What does __init__.py do?
This file marks the supplier_intel directory as a Python package. Without it, Python cannot import modules from that folder. The __version__ variable gives your system a version identifier for tracking releases.
- Save supplier_intel/__init__.py.
Install all dependencies into the virtual environment
Rather than installing packages one at a time, a requirements.txt file pins every dependency to an exact version. This guarantees that anyone running your project gets identical behaviour.
- In Cursor, create a new file in the project root called requirements.txt and paste the following:
SQLAlchemy==2.0.50
alembic==1.18.4
pydantic==2.13.4
openpyxl==3.1.5
python-docx==1.2.0
python-pptx==1.0.2
psycopg2-binary==2.9.10
python-dotenv==1.1.0What are these packages?
- SQLAlchemy is the Python ORM for defining database models and querying PostgreSQL.
- Alembic handles database schema migrations so you can evolve your tables safely.
- Pydantic validates and structures data at every stage of the pipeline.
- openpyxl, python-docx, and python-pptx parse Excel, Word, and PowerPoint files respectively.
- psycopg2-binary is the PostgreSQL database driver that SQLAlchemy uses under the hood.
- python-dotenv loads environment variables from a .env file so secrets stay out of your code.
- Save requirements.txt.
- Install all the dependencies by running this command in your terminal:
pip install -r requirements.txtYou should see each package download and install with a final line saying Successfully installed followed by the package list.
Installation failing?
- If you see a permissions error, confirm your virtual environment is activated. You should see (venv) or similar at the start of your terminal prompt.
- If psycopg2-binary fails to install, ensure PostgreSQL 17 is still running. The driver needs to detect the PostgreSQL installation during build.
- help me debug a pip install failure for my supplier intelligence project
Create the configuration module
Every module in the pipeline needs to know where the database lives and how to log its activity. A central configuration module solves this by loading settings from environment variables with sensible defaults.
- In Cursor, create a new file in the project root called .env and paste the following:
DATABASE_URL=postgresql+psycopg2://postgres:postgres@localhost:5432/supplier_intel
LOG_LEVEL=INFOWhy use a .env file?
The .env file keeps sensitive values like database credentials separate from your code. The python-dotenv library reads this file at startup and injects the values into os.getenv() calls automatically.
If you set a different password for your postgres user during installation, update the password in the DATABASE_URL value now.
- Save .env.
- Now create a new file at supplier_intel/config.py and paste the module-level configuration:
"""Configuration module for the supplier intelligence system."""
import logging
import os
from pathlib import Path
from dotenv import load_dotenv
load_dotenv()
BASE_DIR = Path(__file__).resolve().parent.parent
DATA_DIR = BASE_DIR / "data"
LOG_DIR = BASE_DIR / "logs"
DATABASE_URL = os.getenv(
"DATABASE_URL",
"postgresql+psycopg2://postgres:postgres@localhost:5432/supplier_intel",
)
LOG_LEVEL = os.getenv("LOG_LEVEL", "INFO")What does this code do?
- load_dotenv() reads your .env file and makes its values available via os.getenv().
- BASE_DIR resolves to the project root (one level above the supplier_intel package).
- DATABASE_URL pulls the connection string from the environment, falling back to a local default if not set.
- LOG_LEVEL controls how verbose your pipeline logs are (INFO by default).
- Below the LOG_LEVEL line, add the logging setup function:
def setup_logging() -> logging.Logger:
"""Configure structured logging for the pipeline."""
LOG_DIR.mkdir(exist_ok=True)
logger = logging.getLogger("supplier_intel")
logger.setLevel(getattr(logging, LOG_LEVEL))
formatter = logging.Formatter(
"%(asctime)s | %(levelname)-8s | %(name)s | %(message)s",
datefmt="%Y-%m-%d %H:%M:%S",
)
file_handler = logging.FileHandler(LOG_DIR / "pipeline.log")
file_handler.setFormatter(formatter)
logger.addHandler(file_handler)
console_handler = logging.StreamHandler()
console_handler.setFormatter(formatter)
logger.addHandler(console_handler)
return loggerWhat does setup_logging do?
- LOG_DIR.mkdir(exist_ok=True) creates the logs/ directory on first run so log files have somewhere to live.
- The formatter produces timestamped, structured log lines (e.g. 2026-06-01 14:30:00 | INFO | supplier_intel | message).
- Two handlers write logs to both a file (pipeline.log) and the terminal simultaneously so you can watch progress live.
- Save supplier_intel/config.py.
- Verify the configuration loads correctly by running this in your terminal:
python -c "from supplier_intel.config import DATABASE_URL; print(DATABASE_URL)"You should see postgresql+psycopg2://postgres:postgres@localhost:5432/supplier_intel printed to the terminal.
Seeing a ModuleNotFoundError?
- Make sure you are running the command from the project root directory (the folder containing supplier_intel/).
- Check that supplier_intel/__init__.py exists. Without it, Python does not recognize the folder as an importable package.
- Confirm your virtual environment is activated by checking for (venv) in your terminal prompt.
- help me fix a ModuleNotFoundError when importing supplier_intel.config
✔️ Awesome, I've got everything!
Great. Double check you have saved both supplier_intel/__init__.py and supplier_intel/config.py.
ⓧ I'd like to double check the full code
Here is the complete supplier_intel/config.py for comparison:
"""Configuration module for the supplier intelligence system."""
import logging
import os
from pathlib import Path
from dotenv import load_dotenv
load_dotenv()
BASE_DIR = Path(__file__).resolve().parent.parent
DATA_DIR = BASE_DIR / "data"
LOG_DIR = BASE_DIR / "logs"
DATABASE_URL = os.getenv(
"DATABASE_URL",
"postgresql+psycopg2://postgres:postgres@localhost:5432/supplier_intel",
)
LOG_LEVEL = os.getenv("LOG_LEVEL", "INFO")
def setup_logging() -> logging.Logger:
"""Configure structured logging for the pipeline."""
LOG_DIR.mkdir(exist_ok=True)
logger = logging.getLogger("supplier_intel")
logger.setLevel(getattr(logging, LOG_LEVEL))
formatter = logging.Formatter(
"%(asctime)s | %(levelname)-8s | %(name)s | %(message)s",
datefmt="%Y-%m-%d %H:%M:%S",
)
file_handler = logging.FileHandler(LOG_DIR / "pipeline.log")
file_handler.setFormatter(formatter)
logger.addHandler(file_handler)
console_handler = logging.StreamHandler()
console_handler.setFormatter(formatter)
logger.addHandler(console_handler)
return loggerAnd here is the complete .env:
DATABASE_URL=postgresql+psycopg2://postgres:postgres@localhost:5432/supplier_intel
LOG_LEVEL=INFOAnd the complete requirements.txt:
SQLAlchemy==2.0.50
alembic==1.18.4
pydantic==2.13.4
openpyxl==3.1.5
python-docx==1.2.0
python-pptx==1.0.2
psycopg2-binary==2.9.10
python-dotenv==1.1.0Your project skeleton is in place with all dependencies installed and configuration loading from environment variables. Next up, you will design the database schema using SQLAlchemy's hybrid relational/JSONB pattern to handle 35+ dynamic supplier categories without rigid table definitions.
Design the Database Schema
Your pipeline needs a permanent home for supplier data. Raw files contain scattered information across dozens of categories, and every re-run of the pipeline should update existing records without creating duplicates.
In this step, you will design a PostgreSQL schema that combines relational columns for core fields with a flexible JSONB column for dynamic attributes. You will then use Alembic to version-control your schema and apply it to the database.
In this step, get ready to:
- Define SQLAlchemy 2.0 models with a hybrid relational/JSONB pattern.
- Initialize Alembic and generate the initial migration.
- Run the migration to create all tables in PostgreSQL.
Define the SQLAlchemy models
The database needs three tables. A suppliers table stores one row per supplier with core relational columns plus a JSONB attributes column for the 35+ dynamic categories. A source_documents table tracks every file the pipeline has processed. An extraction_records table links each extracted data point back to its source for full audit traceability.
Why hybrid relational plus JSONB?
Storing all 35+ supplier categories as separate columns would force a schema migration every time a new category appears. A pure JSONB approach loses the ability to enforce constraints on core fields like name and country.
The hybrid pattern gives you the best of both. Relational columns handle frequently-queried fields with proper types and indexes. The JSONB column absorbs any new attribute category without a schema change.
- Create a new file called models.py inside your supplier_intel folder.
- Add the imports and base class by pasting this code:
"""SQLAlchemy models for the supplier intelligence database."""
from datetime import datetime, timezone
from typing import Any
from sqlalchemy import (
Boolean,
DateTime,
Float,
Index,
Integer,
String,
Text,
UniqueConstraint,
func,
)
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
class Base(DeclarativeBase):
"""Base class for all database models."""
passWhat does this code do?
- The imports pull in SQLAlchemy 2.0's type-annotated mapping system (Mapped, mapped_column) and PostgreSQL-specific types (JSONB).
- Base is the declarative base that all model classes inherit from. Alembic reads its metadata attribute to detect which tables to create.
- Below the Base class, add the Supplier model:
class Supplier(Base):
"""Master supplier record with core relational fields and flexible JSONB attributes."""
__tablename__ = "suppliers"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
name: Mapped[str] = mapped_column(String(500), nullable=False)
normalized_name: Mapped[str] = mapped_column(String(500), nullable=False, unique=True)
country: Mapped[str | None] = mapped_column(String(100))
city: Mapped[str | None] = mapped_column(String(200))
primary_category: Mapped[str | None] = mapped_column(String(200))
status: Mapped[str] = mapped_column(String(50), default="active")
# Flexible attributes stored as JSONB for dynamic categories
attributes: Mapped[dict[str, Any]] = mapped_column(JSONB, default=dict)
# Metadata
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), server_default=func.now()
)
updated_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), server_default=func.now(), onupdate=func.now()
)
confidence_score: Mapped[float] = mapped_column(Float, default=0.0)
requires_review: Mapped[bool] = mapped_column(Boolean, default=False)
# Relationships
extraction_records: Mapped[list["ExtractionRecord"]] = relationship(
back_populates="supplier", cascade="all, delete-orphan"
)
__table_args__ = (
Index("ix_suppliers_attributes_gin", "attributes", postgresql_using="gin"),
Index("ix_suppliers_normalized_name", "normalized_name"),
Index("ix_suppliers_country", "country"),
)What does this model do?
- Core relational columns (name, normalized_name, country) give you fast indexed queries on the fields you filter most.
- The attributes JSONB column stores any supplier category (price, certifications, sustainability scores) as key-value pairs without schema changes.
- The GIN index on attributes enables efficient containment queries (@>) and key-existence checks (?) across the JSONB data.
- normalized_name has a unique constraint so the pipeline can upsert suppliers by their deduplicated name.
- Below the Supplier class, add the SourceDocument model:
class SourceDocument(Base):
"""Registry of all source documents processed by the system."""
__tablename__ = "source_documents"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
file_path: Mapped[str] = mapped_column(String(1000), nullable=False, unique=True)
file_name: Mapped[str] = mapped_column(String(500), nullable=False)
file_type: Mapped[str] = mapped_column(String(10), nullable=False)
file_hash: Mapped[str] = mapped_column(String(64), nullable=False)
file_size_bytes: Mapped[int] = mapped_column(Integer)
# Processing metadata
processed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
processing_status: Mapped[str] = mapped_column(String(50), default="pending")
error_message: Mapped[str | None] = mapped_column(Text)
records_extracted: Mapped[int] = mapped_column(Integer, default=0)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), server_default=func.now()
)
# Relationships
extraction_records: Mapped[list["ExtractionRecord"]] = relationship(
back_populates="source_document", cascade="all, delete-orphan"
)What does this model do?
- file_hash stores a SHA-256 hash of each file so the pipeline can detect unchanged documents and skip reprocessing.
- processing_status tracks whether a document is pending, completed, or failed.
- The unique constraint on file_path prevents the same document from being registered twice.
- Below the SourceDocument class, add the ExtractionRecord model:
class ExtractionRecord(Base):
"""Individual data extraction linking a supplier attribute to its source."""
__tablename__ = "extraction_records"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
supplier_id: Mapped[int] = mapped_column(Integer, nullable=False)
source_document_id: Mapped[int] = mapped_column(Integer, nullable=False)
# What was extracted
attribute_name: Mapped[str] = mapped_column(String(200), nullable=False)
raw_value: Mapped[str | None] = mapped_column(Text)
normalized_value: Mapped[str | None] = mapped_column(Text)
value_type: Mapped[str] = mapped_column(String(50), default="string")
# Extraction quality
confidence: Mapped[float] = mapped_column(Float, default=1.0)
extraction_method: Mapped[str] = mapped_column(String(100))
source_location: Mapped[str | None] = mapped_column(String(500))
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), server_default=func.now()
)
# Relationships
supplier: Mapped["Supplier"] = relationship(back_populates="extraction_records")
source_document: Mapped["SourceDocument"] = relationship(
back_populates="extraction_records"
)
__table_args__ = (
UniqueConstraint(
"supplier_id",
"source_document_id",
"attribute_name",
name="uq_extraction_supplier_doc_attr",
),
Index("ix_extraction_supplier_id", "supplier_id"),
Index("ix_extraction_source_document_id", "source_document_id"),
)What does this model do?
- Each row records one attribute extracted from one document for one supplier. This gives you a full audit trail from any data point back to its source.
- The UniqueConstraint on (supplier_id, source_document_id, attribute_name) prevents duplicate extraction records when the pipeline is re-run.
- confidence scores how reliable the extraction was. Lower scores can trigger manual review.
- Save supplier_intel/models.py.
✔️ Awesome, I've got everything!
Great. Double check you have saved the file before moving on.
ⓧ I'd like to double check the full code
"""SQLAlchemy models for the supplier intelligence database."""
from datetime import datetime, timezone
from typing import Any
from sqlalchemy import (
Boolean,
DateTime,
Float,
Index,
Integer,
String,
Text,
UniqueConstraint,
func,
)
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
class Base(DeclarativeBase):
"""Base class for all database models."""
pass
class Supplier(Base):
"""Master supplier record with core relational fields and flexible JSONB attributes."""
__tablename__ = "suppliers"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
name: Mapped[str] = mapped_column(String(500), nullable=False)
normalized_name: Mapped[str] = mapped_column(String(500), nullable=False, unique=True)
country: Mapped[str | None] = mapped_column(String(100))
city: Mapped[str | None] = mapped_column(String(200))
primary_category: Mapped[str | None] = mapped_column(String(200))
status: Mapped[str] = mapped_column(String(50), default="active")
# Flexible attributes stored as JSONB for dynamic categories
attributes: Mapped[dict[str, Any]] = mapped_column(JSONB, default=dict)
# Metadata
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), server_default=func.now()
)
updated_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), server_default=func.now(), onupdate=func.now()
)
confidence_score: Mapped[float] = mapped_column(Float, default=0.0)
requires_review: Mapped[bool] = mapped_column(Boolean, default=False)
# Relationships
extraction_records: Mapped[list["ExtractionRecord"]] = relationship(
back_populates="supplier", cascade="all, delete-orphan"
)
__table_args__ = (
Index("ix_suppliers_attributes_gin", "attributes", postgresql_using="gin"),
Index("ix_suppliers_normalized_name", "normalized_name"),
Index("ix_suppliers_country", "country"),
)
class SourceDocument(Base):
"""Registry of all source documents processed by the system."""
__tablename__ = "source_documents"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
file_path: Mapped[str] = mapped_column(String(1000), nullable=False, unique=True)
file_name: Mapped[str] = mapped_column(String(500), nullable=False)
file_type: Mapped[str] = mapped_column(String(10), nullable=False)
file_hash: Mapped[str] = mapped_column(String(64), nullable=False)
file_size_bytes: Mapped[int] = mapped_column(Integer)
# Processing metadata
processed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
processing_status: Mapped[str] = mapped_column(String(50), default="pending")
error_message: Mapped[str | None] = mapped_column(Text)
records_extracted: Mapped[int] = mapped_column(Integer, default=0)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), server_default=func.now()
)
# Relationships
extraction_records: Mapped[list["ExtractionRecord"]] = relationship(
back_populates="source_document", cascade="all, delete-orphan"
)
class ExtractionRecord(Base):
"""Individual data extraction linking a supplier attribute to its source."""
__tablename__ = "extraction_records"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
supplier_id: Mapped[int] = mapped_column(Integer, nullable=False)
source_document_id: Mapped[int] = mapped_column(Integer, nullable=False)
# What was extracted
attribute_name: Mapped[str] = mapped_column(String(200), nullable=False)
raw_value: Mapped[str | None] = mapped_column(Text)
normalized_value: Mapped[str | None] = mapped_column(Text)
value_type: Mapped[str] = mapped_column(String(50), default="string")
# Extraction quality
confidence: Mapped[float] = mapped_column(Float, default=1.0)
extraction_method: Mapped[str] = mapped_column(String(100))
source_location: Mapped[str | None] = mapped_column(String(500))
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), server_default=func.now()
)
# Relationships
supplier: Mapped["Supplier"] = relationship(back_populates="extraction_records")
source_document: Mapped["SourceDocument"] = relationship(
back_populates="extraction_records"
)
__table_args__ = (
UniqueConstraint(
"supplier_id",
"source_document_id",
"attribute_name",
name="uq_extraction_supplier_doc_attr",
),
Index("ix_extraction_supplier_id", "supplier_id"),
Index("ix_extraction_source_document_id", "source_document_id"),
)Initialize Alembic and generate the initial migration
Alembic is a database migration tool that version-controls your schema. It compares your SQLAlchemy models against the current database state and generates migration scripts to bring the database up to date.
- First, create the supplier_intel database in PostgreSQL by running this command:
createdb -U postgres supplier_intelWhat about the password prompt?
PostgreSQL will prompt you for a password. Enter the password you set for the postgres user during PostgreSQL installation. The characters will not appear as you type.
- Initialize the Alembic migration environment by running:
alembic init alembicWhat does this command do?
This creates an alembic.ini config file in your project root and an alembic/ directory containing env.py, a versions/ folder for migration scripts, and a template file.
- In alembic.ini, find this line:
sqlalchemy.url = driver://user:pass@localhost/dbname- Replace it with your actual database URL so it looks like this:
sqlalchemy.url = postgresql+psycopg2://postgres:postgres@localhost:5432/supplier_intel- Save alembic.ini.
- Now open alembic/env.py and replace its entire contents with this code:
"""Alembic environment configuration."""
from logging.config import fileConfig
from alembic import context
from sqlalchemy import engine_from_config, pool
from supplier_intel.models import Base
config = context.config
if config.config_file_name is not None:
fileConfig(config.config_file_name)
target_metadata = Base.metadata
def run_migrations_offline() -> None:
"""Run migrations in 'offline' mode."""
url = config.get_main_option("sqlalchemy.url")
context.configure(url=url, target_metadata=target_metadata, literal_binds=True)
with context.begin_transaction():
context.run_migrations()
def run_migrations_online() -> None:
"""Run migrations in 'online' mode."""
connectable = engine_from_config(
config.get_section(config.config_ini_section, {}),
prefix="sqlalchemy.",
poolclass=pool.NullPool,
)
with connectable.connect() as connection:
context.configure(connection=connection, target_metadata=target_metadata)
with context.begin_transaction():
context.run_migrations()
if context.is_offline_mode():
run_migrations_offline()
else:
run_migrations_online()What does this file do?
- The key line is from supplier_intel.models import Base. This tells Alembic about your models so it can compare them against the database.
- target_metadata = Base.metadata provides Alembic with the full definition of every table, column, and index you declared in models.py.
- The online/offline split lets Alembic either connect directly or generate SQL scripts for review.
- Save alembic/env.py.
- Generate the initial migration by running:
alembic revision --autogenerate -m "initial schema"- You should see output indicating a new migration file was created in alembic/versions/.
Seeing a connection error?
- Make sure the PostgreSQL service is running. You can check with pg_isready in your terminal.
- Verify the sqlalchemy.url in alembic.ini matches the password you set during PostgreSQL installation.
- Confirm you created the supplier_intel database in the previous step. Run createdb -U postgres supplier_intel again if needed.
Why is my Alembic migration failing to connect?
Run the migration to create the tables
The migration file contains the SQL instructions to create your tables. Now you will apply it to the live database.
- Apply the migration by running:
alembic upgrade head- You should see output showing that the migration was applied successfully.
- Verify the tables exist by running:
psql -U postgres -d supplier_intel -c "\dt"You should see a table listing that includes suppliers, source_documents, extraction_records, and alembic_version (Alembic's internal tracking table).
Tables not showing up?
- If alembic upgrade head reported success but you see no tables, make sure you are connecting to the correct database (supplier_intel, not postgres).
- If you see a "no such table" error in the migration, check that alembic/env.py imports Base from supplier_intel.models correctly.
Why are my tables missing after running alembic upgrade head?
Your database schema is live and ready to receive supplier data. Next up, you will build the document parsers that feed data into these tables.
Build the Document Parsers
Your database is ready to store supplier data, but you have no way to get data out of the source files yet. Supplier information lives inside Excel spreadsheets, Word documents, and PowerPoint presentations.
In this step, you will build a parser framework that reads each document format and outputs a consistent data structure. The framework uses an abstract base class so that adding new file formats later requires zero changes to existing code.
In this step, get ready to:
- Build the abstract base parser class that defines the interface all parsers must implement.
- Implement the Excel parser using openpyxl to extract tabular supplier data.
- Implement the Word and PowerPoint parsers to extract text-based supplier information.
Build the abstract base parser class
Every parser in the system must produce the same output shape regardless of the file format it reads. An abstract base class enforces this contract at the code level.
- Create the parsers package directory inside supplier_intel by running this command:
mkdir supplier_intel\parsers- Create a new file called base.py inside supplier_intel/parsers/ and add the following code:
"""Abstract base parser defining the interface for all document parsers."""
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from pathlib import Path
@dataclass
class ParsedContent:
file_path: str
file_type: str
tables: list[list[dict[str, str]]] = field(default_factory=list)
text_blocks: list[str] = field(default_factory=list)
metadata: dict[str, str] = field(default_factory=dict)What does this code do?
ParsedContent is a dataclass that serves as the universal output format for every parser. It holds three types of extracted content:
- tables stores rows of key-value pairs extracted from tabular data (spreadsheet rows, document tables).
- text_blocks stores paragraphs and free-text content from documents.
- metadata stores extra information about the parse (sheet names, row counts, slide numbers).
- Now add the BaseParser abstract class below the ParsedContent dataclass in the same file:
class BaseParser(ABC):
@property
@abstractmethod
def supported_extensions(self) -> list[str]: ...
@abstractmethod
def parse(self, file_path: Path) -> ParsedContent: ...
def can_handle(self, file_path: Path) -> bool:
return file_path.suffix.lower() in self.supported_extensionsWhat does this code do?
BaseParser defines the contract every parser must follow. Any subclass must implement supported_extensions (which file types it handles) and parse (how it reads a file).
The can_handle method is a concrete helper that checks if a file's extension matches what the parser supports. This is inherited by all subclasses for free.
- Save base.py.
- Verify the module loads by running this command:
python -c "from supplier_intel.parsers.base import BaseParser, ParsedContent; print('Base parser loaded')"You should see Base parser loaded printed in the terminal.
Seeing a ModuleNotFoundError?
- Make sure you created the parsers folder inside supplier_intel/, not in the project root.
- Check that your virtual environment is activated. You should see (venv) at the start of your terminal prompt.
Help me debug my base parser import error
✔️ Awesome, I've got everything!
Great. Double check you have saved base.py and move on to the next parser.
ⓧ I'd like to double check the full code
"""Abstract base parser defining the interface for all document parsers."""
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from pathlib import Path
@dataclass
class ParsedContent:
file_path: str
file_type: str
tables: list[list[dict[str, str]]] = field(default_factory=list)
text_blocks: list[str] = field(default_factory=list)
metadata: dict[str, str] = field(default_factory=dict)
class BaseParser(ABC):
@property
@abstractmethod
def supported_extensions(self) -> list[str]: ...
@abstractmethod
def parse(self, file_path: Path) -> ParsedContent: ...
def can_handle(self, file_path: Path) -> bool:
return file_path.suffix.lower() in self.supported_extensionsImplement the Excel parser
Excel files are the most common source of structured supplier data. Each sheet in a workbook can contain a table of suppliers with different attributes in each column.
- Create a new file called excel_parser.py inside supplier_intel/parsers/ and add the following code:
"""Excel file parser using openpyxl."""
import logging
from pathlib import Path
from openpyxl import load_workbook
from supplier_intel.parsers.base import BaseParser, ParsedContent
logger = logging.getLogger("supplier_intel.parsers.excel")
class ExcelParser(BaseParser):
@property
def supported_extensions(self) -> list[str]:
return [".xlsx", ".xls"]
def parse(self, file_path: Path) -> ParsedContent:
logger.info(f"Parsing Excel file: {file_path.name}")
content = ParsedContent(file_path=str(file_path), file_type="excel")
workbook = load_workbook(str(file_path), read_only=True, data_only=True)
for sheet_name in workbook.sheetnames:
sheet = workbook[sheet_name]
rows = list(sheet.iter_rows(values_only=True))
if len(rows) < 2:
continue
# First row is headers
headers = [str(cell).strip() if cell else f"col_{i}" for i, cell in enumerate(rows[0])]
table_data: list[dict[str, str]] = []
for row in rows[1:]:
if all(cell is None for cell in row):
continue
row_dict = {}
for i, cell in enumerate(row):
if i < len(headers):
row_dict[headers[i]] = str(cell).strip() if cell is not None else ""
table_data.append(row_dict)
if table_data:
content.tables.append(table_data)
content.metadata[f"sheet_{sheet_name}_rows"] = str(len(table_data))
workbook.close()
return contentWhat does this code do?
The ExcelParser opens a workbook in read-only mode and iterates through every sheet. For each sheet with at least 2 rows, it treats the first row as column headers and converts each subsequent row into a dictionary.
The data_only=True flag tells openpyxl to read calculated formula results instead of the formula strings themselves. Sheets with fewer than 2 rows (header only or empty) are skipped.
- Save excel_parser.py.
- Verify the parser loads by running:
python -c "from supplier_intel.parsers.excel_parser import ExcelParser; print(ExcelParser().supported_extensions)"You should see ['.xlsx', '.xls'] printed in the terminal.
Seeing an ImportError for openpyxl?
- Make sure your virtual environment is activated and you installed all dependencies from requirements.txt in the previous step.
- Check that the import path at the top of excel_parser.py says from supplier_intel.parsers.base import BaseParser, ParsedContent exactly.
Help me fix my Excel parser import
✔️ Awesome, I've got everything!
Great. Save the file and move on to the Word and PowerPoint parsers.
ⓧ I'd like to double check the full code
"""Excel file parser using openpyxl."""
import logging
from pathlib import Path
from openpyxl import load_workbook
from supplier_intel.parsers.base import BaseParser, ParsedContent
logger = logging.getLogger("supplier_intel.parsers.excel")
class ExcelParser(BaseParser):
@property
def supported_extensions(self) -> list[str]:
return [".xlsx", ".xls"]
def parse(self, file_path: Path) -> ParsedContent:
logger.info(f"Parsing Excel file: {file_path.name}")
content = ParsedContent(file_path=str(file_path), file_type="excel")
workbook = load_workbook(str(file_path), read_only=True, data_only=True)
for sheet_name in workbook.sheetnames:
sheet = workbook[sheet_name]
rows = list(sheet.iter_rows(values_only=True))
if len(rows) < 2:
continue
headers = [str(cell).strip() if cell else f"col_{i}" for i, cell in enumerate(rows[0])]
table_data: list[dict[str, str]] = []
for row in rows[1:]:
if all(cell is None for cell in row):
continue
row_dict = {}
for i, cell in enumerate(row):
if i < len(headers):
row_dict[headers[i]] = str(cell).strip() if cell is not None else ""
table_data.append(row_dict)
if table_data:
content.tables.append(table_data)
content.metadata[f"sheet_{sheet_name}_rows"] = str(len(table_data))
workbook.close()
return contentImplement the Word and PowerPoint parsers
Word and PowerPoint files contain a mix of paragraphs, bullet lists, and embedded tables. Both parsers extract tables into the same format as the Excel parser, plus capture loose text blocks for downstream extraction.
- Create a new file called word_parser.py inside supplier_intel/parsers/ and add the following code:
"""Word document parser using python-docx."""
import logging
from pathlib import Path
from docx import Document
from supplier_intel.parsers.base import BaseParser, ParsedContent
logger = logging.getLogger("supplier_intel.parsers.word")
class WordParser(BaseParser):
@property
def supported_extensions(self) -> list[str]:
return [".docx"]
def parse(self, file_path: Path) -> ParsedContent:
logger.info(f"Parsing Word file: {file_path.name}")
content = ParsedContent(file_path=str(file_path), file_type="word")
doc = Document(str(file_path))
# Extract paragraphs
for paragraph in doc.paragraphs:
text = paragraph.text.strip()
if text:
content.text_blocks.append(text)
# Extract tables
for table in doc.tables:
if not table.rows:
continue
headers = [cell.text.strip() for cell in table.rows[0].cells]
table_data: list[dict[str, str]] = []
for row in table.rows[1:]:
row_dict = {}
for i, cell in enumerate(row.cells):
if i < len(headers):
row_dict[headers[i]] = cell.text.strip()
table_data.append(row_dict)
if table_data:
content.tables.append(table_data)
return contentWhat does this code do?
The WordParser handles two extraction paths. First, it iterates through all paragraphs and collects non-empty text into text_blocks.
Second, it finds embedded tables and converts them into the same header-to-row dictionary format the ExcelParser uses. This means downstream code never has to know which file type the data came from.
- Save word_parser.py.
- Now create pptx_parser.py inside supplier_intel/parsers/ and add the following code:
"""PowerPoint parser using python-pptx."""
import logging
from pathlib import Path
from pptx import Presentation
from pptx.table import Table
from supplier_intel.parsers.base import BaseParser, ParsedContent
logger = logging.getLogger("supplier_intel.parsers.pptx")
class PowerPointParser(BaseParser):
@property
def supported_extensions(self) -> list[str]:
return [".pptx"]
def parse(self, file_path: Path) -> ParsedContent:
logger.info(f"Parsing PowerPoint file: {file_path.name}")
content = ParsedContent(file_path=str(file_path), file_type="pptx")
prs = Presentation(str(file_path))
for slide_num, slide in enumerate(prs.slides, 1):
for shape in slide.shapes:
# Extract text from text frames
if shape.has_text_frame:
for paragraph in shape.text_frame.paragraphs:
text = paragraph.text.strip()
if text:
content.text_blocks.append(text)
# Extract tables
if shape.has_table:
table: Table = shape.table
if len(table.rows) < 2:
continue
headers = [table.cell(0, col).text.strip() for col in range(len(table.columns))]
table_data: list[dict[str, str]] = []
for row_idx in range(1, len(table.rows)):
row_dict = {}
for col_idx in range(len(table.columns)):
if col_idx < len(headers):
row_dict[headers[col_idx]] = table.cell(row_idx, col_idx).text.strip()
table_data.append(row_dict)
if table_data:
content.tables.append(table_data)
return contentWhat does this code do?
The PowerPointParser iterates through every slide and every shape on each slide. Shapes can be text boxes, images, or tables.
For text frames, it pulls each paragraph's text. For table shapes, it uses cell-level access (table.cell(row, col)) to extract rows into dictionaries. This handles presentation decks that embed supplier comparison tables in slides.
- Save pptx_parser.py.
- Finally, create the package initializer __init__.py inside supplier_intel/parsers/ and add the following code:
"""Document parsers package."""
from supplier_intel.parsers.base import BaseParser, ParsedContent
from supplier_intel.parsers.excel_parser import ExcelParser
from supplier_intel.parsers.pptx_parser import PowerPointParser
from supplier_intel.parsers.word_parser import WordParser
__all__ = [
"BaseParser",
"ParsedContent",
"ExcelParser",
"WordParser",
"PowerPointParser",
]Why create an __init__.py?
This file makes supplier_intel/parsers/ a proper Python package and re-exports all parser classes from one location. Other modules can now import everything from supplier_intel.parsers directly instead of reaching into individual files.
- Save __init__.py.
- Verify all parsers load correctly by running:
python -c "from supplier_intel.parsers import ExcelParser, WordParser, PowerPointParser; print('All parsers loaded')"You should see All parsers loaded printed in the terminal.
Seeing an import error for docx or pptx?
- The package name on PyPI is python-docx but you import it as from docx import Document. Make sure you installed python-docx, not docx.
- Similarly, python-pptx is installed via pip but imported as from pptx import Presentation.
- If you see errors about lxml, both libraries depend on it. Run pip install lxml to install it manually.
Help me fix my parser imports
✔️ Awesome, I've got everything!
Great. Double check you have saved all files and move on to the next step.
ⓧ I'd like to double check the full code
"""Word document parser using python-docx."""
import logging
from pathlib import Path
from docx import Document
from supplier_intel.parsers.base import BaseParser, ParsedContent
logger = logging.getLogger("supplier_intel.parsers.word")
class WordParser(BaseParser):
@property
def supported_extensions(self) -> list[str]:
return [".docx"]
def parse(self, file_path: Path) -> ParsedContent:
logger.info(f"Parsing Word file: {file_path.name}")
content = ParsedContent(file_path=str(file_path), file_type="word")
doc = Document(str(file_path))
for paragraph in doc.paragraphs:
text = paragraph.text.strip()
if text:
content.text_blocks.append(text)
for table in doc.tables:
if not table.rows:
continue
headers = [cell.text.strip() for cell in table.rows[0].cells]
table_data: list[dict[str, str]] = []
for row in table.rows[1:]:
row_dict = {}
for i, cell in enumerate(row.cells):
if i < len(headers):
row_dict[headers[i]] = cell.text.strip()
table_data.append(row_dict)
if table_data:
content.tables.append(table_data)
return content"""PowerPoint parser using python-pptx."""
import logging
from pathlib import Path
from pptx import Presentation
from pptx.table import Table
from supplier_intel.parsers.base import BaseParser, ParsedContent
logger = logging.getLogger("supplier_intel.parsers.pptx")
class PowerPointParser(BaseParser):
@property
def supported_extensions(self) -> list[str]:
return [".pptx"]
def parse(self, file_path: Path) -> ParsedContent:
logger.info(f"Parsing PowerPoint file: {file_path.name}")
content = ParsedContent(file_path=str(file_path), file_type="pptx")
prs = Presentation(str(file_path))
for slide_num, slide in enumerate(prs.slides, 1):
for shape in slide.shapes:
if shape.has_text_frame:
for paragraph in shape.text_frame.paragraphs:
text = paragraph.text.strip()
if text:
content.text_blocks.append(text)
if shape.has_table:
table: Table = shape.table
if len(table.rows) < 2:
continue
headers = [table.cell(0, col).text.strip() for col in range(len(table.columns))]
table_data: list[dict[str, str]] = []
for row_idx in range(1, len(table.rows)):
row_dict = {}
for col_idx in range(len(table.columns)):
if col_idx < len(headers):
row_dict[headers[col_idx]] = table.cell(row_idx, col_idx).text.strip()
table_data.append(row_dict)
if table_data:
content.tables.append(table_data)
return content"""Document parsers package."""
from supplier_intel.parsers.base import BaseParser, ParsedContent
from supplier_intel.parsers.excel_parser import ExcelParser
from supplier_intel.parsers.pptx_parser import PowerPointParser
from supplier_intel.parsers.word_parser import WordParser
__all__ = [
"BaseParser",
"ParsedContent",
"ExcelParser",
"WordParser",
"PowerPointParser",
]Why this approach instead of a single universal parser?
A unified parser library like MarkItDown converts everything to Markdown, which loses structural information (table cell boundaries, column headers). For a data extraction system that needs precise tabular data, direct parsing with format-specific libraries preserves structure better.
The abstract base class pattern means you can add a PDF parser or CSV parser in the future by creating one new file. No existing code needs to change.
Your parser framework is complete. Next up, you will build the extraction engine that converts the raw parsed content into validated supplier records using Pydantic.
Extract and Validate Supplier Data
Your parsers produce structured ParsedContent objects containing tables and text blocks. Raw parsed data is messy though. Column headers vary between documents, supplier names contain inconsistent suffixes, and numeric values mix units.
This step builds the transformation layer that sits between parsing and storage. You will define strict validation schemas with Pydantic, create an extraction engine that maps raw columns to standardized attributes, and build normalizers that clean company names, countries, and numeric values into consistent forms.
In this step, get ready to:
- Build Pydantic validation models for raw supplier data.
- Create an extraction engine that maps parsed table columns to standardized attributes.
- Build a normalization engine for company names, countries, and numeric values.
Define the validation schemas
Every supplier record extracted from a document must pass through a validation gate before it enters the database. Pydantic models enforce type constraints, required fields, and custom cleaning logic at the boundary between raw data and your system.
- Create a new file called validation.py inside the supplier_intel folder by running:
cd ~/Desktop/supplier-intel
New-Item -ItemType File -Path supplier_intel/validation.py- Open supplier_intel/validation.py in Cursor and paste the following code:
"""Pydantic models for data validation and extraction schemas."""
from pydantic import BaseModel, Field, field_validator
class RawSupplierData(BaseModel):
"""Raw supplier data as extracted from a document before normalization."""
name: str = Field(..., min_length=1)
country: str | None = Field(None)
city: str | None = Field(None)
category: str | None = Field(None)
# Dynamic attributes extracted from the document
attributes: dict[str, str | float | int | None] = Field(default_factory=dict)
# Extraction metadata
source_file: str = Field(...)
source_location: str = Field("")
extraction_method: str = Field("table")
@field_validator("name")
@classmethod
def clean_name(cls, v: str) -> str:
"""Remove excessive whitespace from supplier names."""
return " ".join(v.split())
class ValidationResult(BaseModel):
"""Result of validating a batch of extracted records."""
valid_records: list[RawSupplierData] = Field(default_factory=list)
invalid_records: list[dict] = Field(default_factory=list)
warnings: list[str] = Field(default_factory=list)
total_processed: int = 0
confidence_scores: dict[str, float] = Field(default_factory=dict)What does this code do?
RawSupplierData defines the shape of a single supplier record before normalization. The name field is required (min_length=1), while country, city, and category are optional. The attributes dict holds any dynamic data that does not fit the core columns.
The @field_validator on name collapses runs of whitespace so that "Acme Corp" becomes "Acme Corp" at the moment of validation. ValidationResult collects both valid and invalid records so the pipeline can report on extraction quality.
- Save supplier_intel/validation.py.
- Confirm the file is importable by running this in your terminal:
python -c "from supplier_intel.validation import RawSupplierData, ValidationResult; print('Validation models loaded')"You should see Validation models loaded printed to the terminal.
Seeing a ModuleNotFoundError?
- Make sure your virtual environment is activated. The prompt should show (.venv) at the start of the line.
- Confirm you are in the supplier-intel project folder (the one containing the supplier_intel package folder).
- Check that pydantic is installed by running pip list | findstr pydantic.
Help me debug a ModuleNotFoundError when importing my validation models
Create the extraction engine
The extraction engine bridges the gap between raw parsed tables and validated Pydantic models. It needs to identify which column holds supplier names, map remaining columns to standardized attribute names, and construct validated records for each row.
- Create supplier_intel/extraction.py and paste the column-mapping configuration and helper functions:
"""Extraction engine that converts parsed content into validated supplier records."""
import logging
from typing import Any
from supplier_intel.parsers.base import ParsedContent
from supplier_intel.validation import RawSupplierData, ValidationResult
logger = logging.getLogger("supplier_intel.extraction")
# Common header patterns that indicate supplier-relevant columns
SUPPLIER_NAME_PATTERNS = [
"supplier", "vendor", "company", "manufacturer", "name", "firm",
]
ATTRIBUTE_MAPPING: dict[str, list[str]] = {
"price": ["price", "cost", "unit_price", "rate"],
"delivery_time": ["delivery", "lead_time", "delivery_time", "lead time"],
"co2_emissions": ["co2", "carbon", "emissions", "co2_emissions"],
"production_capacity": ["capacity", "production_capacity", "volume"],
"certifications": ["certification", "certifications", "cert", "iso"],
"materials": ["material", "materials", "raw_material"],
"country": ["country", "location", "region", "origin"],
"city": ["city", "town", "site"],
"category": ["category", "product", "product_category", "type"],
"quality_metrics": ["quality", "defect_rate", "quality_score"],
"sustainability": ["sustainability", "esg", "green"],
"margins": ["margin", "margins", "profit_margin"],
"risk_indicators": ["risk", "risk_score", "risk_level"],
}
def _find_name_column(headers: list[str]) -> str | None:
"""Identify which column contains supplier names."""
headers_lower = [h.lower().replace(" ", "_") for h in headers]
for pattern in SUPPLIER_NAME_PATTERNS:
for header in headers_lower:
if pattern in header:
return headers[headers_lower.index(header)]
return None
def _map_column_to_attribute(column_name: str) -> str | None:
"""Map a column header to a standardized attribute name."""
col_lower = column_name.lower().replace(" ", "_")
for attr_name, patterns in ATTRIBUTE_MAPPING.items():
for pattern in patterns:
if pattern in col_lower:
return attr_name
return col_lower # Use cleaned column name as-is if no mapping foundWhat does this code do?
SUPPLIER_NAME_PATTERNS lists keywords that typically appear in a column header when it contains the supplier's name. The engine tries each pattern against the table's headers to find the name column.
ATTRIBUTE_MAPPING maps standardized internal attribute names (like delivery_time) to the many variations you will encounter across different spreadsheets ("delivery", "lead_time", "lead time"). This lets the system handle inconsistent headers without manual mapping per file.
- Save extraction.py, then add the main extraction functions below the helper functions you just pasted:
def extract_from_table(
table: list[dict[str, str]], source_file: str
) -> list[RawSupplierData]:
"""Extract supplier records from a parsed table."""
if not table:
return []
headers = list(table[0].keys())
name_column = _find_name_column(headers)
if not name_column:
logger.debug(f"No supplier name column found in table from {source_file}")
return []
records: list[RawSupplierData] = []
for row_idx, row in enumerate(table):
supplier_name = row.get(name_column, "").strip()
if not supplier_name:
continue
attributes: dict[str, Any] = {}
country = None
city = None
category = None
for col_name, value in row.items():
if col_name == name_column or not value:
continue
attr_name = _map_column_to_attribute(col_name)
if attr_name == "country":
country = value
elif attr_name == "city":
city = value
elif attr_name == "category":
category = value
else:
attributes[attr_name] = value
try:
record = RawSupplierData(
name=supplier_name,
country=country,
city=city,
category=category,
attributes=attributes,
source_file=source_file,
source_location=f"row_{row_idx + 1}",
extraction_method="table",
)
records.append(record)
except Exception as e:
logger.warning(f"Validation failed for row {row_idx}: {e}")
return records
def extract_from_content(content: ParsedContent) -> ValidationResult:
"""Extract all supplier records from parsed document content."""
result = ValidationResult()
for table in content.tables:
records = extract_from_table(table, content.file_path)
result.valid_records.extend(records)
result.total_processed = len(result.valid_records) + len(result.invalid_records)
logger.info(
f"Extracted {len(result.valid_records)} valid records from {content.file_path}"
)
return resultWhat does this code do?
extract_from_table iterates through each row of a parsed table. For every row that has a supplier name, it maps remaining columns to standardized attribute names and constructs a RawSupplierData object. If Pydantic validation rejects a row, the error is logged and that row is skipped.
extract_from_content is the public entry point. It loops through all tables inside a ParsedContent object and aggregates results into a ValidationResult with counts of valid and invalid records.
- Save supplier_intel/extraction.py.
- Verify the extraction module loads by running:
python -c "from supplier_intel.extraction import extract_from_content; print('Extraction engine loaded')"You should see Extraction engine loaded with no errors.
Seeing an ImportError about validation?
- The extraction module imports from supplier_intel.validation. Make sure you saved validation.py in the previous substep.
- Check for typos in the import line. It should read from supplier_intel.validation import RawSupplierData, ValidationResult.
Help me fix an ImportError in my extraction.py file
Build the normalization engine
Extracted data from different documents uses inconsistent naming. "Siemens AG", "siemens", and "SIEMENS Corporation" all refer to the same supplier. Countries appear as "US", "USA", or "United States". The normalization engine standardizes these variations so the pipeline can deduplicate records accurately.
- Create supplier_intel/normalization.py and paste the reference data constants:
"""Normalization engine for standardizing supplier data."""
import logging
import re
import unicodedata
from supplier_intel.validation import RawSupplierData
logger = logging.getLogger("supplier_intel.normalization")
# Common company suffixes to normalize
COMPANY_SUFFIXES = [
r"\bltd\.?\b", r"\blimited\b", r"\binc\.?\b", r"\bincorporated\b",
r"\bcorp\.?\b", r"\bcorporation\b", r"\bgmbh\b", r"\bag\b",
r"\bs\.?a\.?\b", r"\bb\.?v\.?\b", r"\bn\.?v\.?\b", r"\bllc\b",
r"\bco\.?\b", r"\bcompany\b", r"\b& co\.?\b", r"\bplc\b",
]
# Country name normalization map
COUNTRY_ALIASES: dict[str, str] = {
"us": "United States",
"usa": "United States",
"united states of america": "United States",
"uk": "United Kingdom",
"gb": "United Kingdom",
"great britain": "United Kingdom",
"de": "Germany",
"deutschland": "Germany",
"cn": "China",
"prc": "China",
"jp": "Japan",
"fr": "France",
"nl": "Netherlands",
"holland": "Netherlands",
"kr": "South Korea",
"south korea": "South Korea",
"republic of korea": "South Korea",
}
# Unit normalization patterns
UNIT_CONVERSIONS: dict[str, dict[str, float]] = {
"weight": {"kg": 1.0, "g": 0.001, "lb": 0.453592, "lbs": 0.453592, "t": 1000.0},
"currency": {"usd": 1.0, "$": 1.0, "eur": 1.08, "gbp": 1.27},
"time_days": {"days": 1.0, "weeks": 7.0, "months": 30.0, "d": 1.0, "w": 7.0},
}What are these constants for?
COMPANY_SUFFIXES contains regex patterns for legal entity suffixes (Ltd, Inc, GmbH, etc.) that get stripped during name normalization. This ensures "Bosch GmbH" and "Bosch" resolve to the same normalized key.
COUNTRY_ALIASES maps abbreviations and alternate names to a single canonical form. UNIT_CONVERSIONS provides conversion factors so values like "14 days" and "2 weeks" can be compared on the same scale.
- Below the constants in the same file, add the normalization functions:
def normalize_company_name(name: str) -> str:
"""Normalize a company name for deduplication matching."""
# Lowercase and strip
normalized = name.lower().strip()
# Remove accents/diacritics
normalized = unicodedata.normalize("NFKD", normalized)
normalized = normalized.encode("ascii", "ignore").decode("ascii")
# Remove common suffixes
for suffix_pattern in COMPANY_SUFFIXES:
normalized = re.sub(suffix_pattern, "", normalized, flags=re.IGNORECASE)
# Remove punctuation and extra whitespace
normalized = re.sub(r"[^\w\s]", "", normalized)
normalized = " ".join(normalized.split())
return normalized
def normalize_country(country: str | None) -> str | None:
"""Normalize country name to a standard form."""
if not country:
return None
country_lower = country.lower().strip()
return COUNTRY_ALIASES.get(country_lower, country.strip().title())
def normalize_numeric_value(value: str, unit_category: str | None = None) -> str:
"""Extract and normalize numeric values with units."""
# Try to extract a number
match = re.search(r"([\d,]+\.?\d*)", value.replace(",", ""))
if not match:
return value
number = float(match.group(1).replace(",", ""))
if unit_category and unit_category in UNIT_CONVERSIONS:
value_lower = value.lower()
conversions = UNIT_CONVERSIONS[unit_category]
for unit, factor in conversions.items():
if unit in value_lower:
number = number * factor
break
return str(round(number, 4))How does name normalization work?
normalize_company_name applies four transformations in sequence: lowercasing, stripping Unicode diacritics (so "Müller" becomes "muller"), removing legal suffixes via regex, and collapsing punctuation. The result is a stable key that matches across name variations.
normalize_numeric_value extracts the numeric portion from strings like "14 days" or "$25.50", then applies a unit conversion factor if the unit_category is specified. This converts all time values to days and all currencies to a base unit.
- Still in normalization.py, add the record-level normalization and deduplication functions at the bottom:
def normalize_record(record: RawSupplierData) -> RawSupplierData:
"""Apply all normalization rules to a supplier record."""
record.country = normalize_country(record.country)
# Normalize numeric attributes where applicable
numeric_attrs = ["price", "delivery_time", "co2_emissions", "production_capacity", "margins"]
for attr_name in numeric_attrs:
if attr_name in record.attributes and isinstance(record.attributes[attr_name], str):
unit_cat = None
if "time" in attr_name:
unit_cat = "time_days"
elif "price" in attr_name or "margin" in attr_name:
unit_cat = "currency"
record.attributes[attr_name] = normalize_numeric_value(
record.attributes[attr_name], unit_cat
)
return record
def find_duplicates(
records: list[RawSupplierData],
) -> dict[str, list[RawSupplierData]]:
"""Group records by normalized company name to identify duplicates."""
groups: dict[str, list[RawSupplierData]] = {}
for record in records:
key = normalize_company_name(record.name)
if key not in groups:
groups[key] = []
groups[key].append(record)
return groupsWhat does this code do?
normalize_record is the single entry point for cleaning one supplier record. It normalizes the country field and applies numeric normalization to attributes that represent quantities (prices, delivery times, emissions).
find_duplicates groups records by their normalized company name. If "Bosch GmbH" and "Robert Bosch" both normalize to "bosch", they end up in the same group. The pipeline uses these groups to merge duplicates during storage.
- Save supplier_intel/normalization.py.
- Test the normalization logic by running:
python -c "from supplier_intel.normalization import normalize_company_name; print(normalize_company_name('Siemens AG'))"You should see siemens printed. The suffix "AG" was stripped and the name was lowercased.
Seeing something other than 'siemens'?
- Make sure the COMPANY_SUFFIXES list includes the r"\bag\b" pattern. Without it, "AG" will not be removed.
- Check that re and unicodedata are imported at the top of the file. Both are standard library modules.
Help me debug my normalize_company_name function output
✔️ Awesome, I've got everything!
Great. Double check you have saved all three files: validation.py, extraction.py, and normalization.py.
ⓧ I'd like to double check the full code
Here is the complete code for each file created in this step.
"""Pydantic models for data validation and extraction schemas."""
from pydantic import BaseModel, Field, field_validator
class RawSupplierData(BaseModel):
"""Raw supplier data as extracted from a document before normalization."""
name: str = Field(..., min_length=1)
country: str | None = Field(None)
city: str | None = Field(None)
category: str | None = Field(None)
# Dynamic attributes extracted from the document
attributes: dict[str, str | float | int | None] = Field(default_factory=dict)
# Extraction metadata
source_file: str = Field(...)
source_location: str = Field("")
extraction_method: str = Field("table")
@field_validator("name")
@classmethod
def clean_name(cls, v: str) -> str:
"""Remove excessive whitespace from supplier names."""
return " ".join(v.split())
class ValidationResult(BaseModel):
"""Result of validating a batch of extracted records."""
valid_records: list[RawSupplierData] = Field(default_factory=list)
invalid_records: list[dict] = Field(default_factory=list)
warnings: list[str] = Field(default_factory=list)
total_processed: int = 0
confidence_scores: dict[str, float] = Field(default_factory=dict)"""Extraction engine that converts parsed content into validated supplier records."""
import logging
from typing import Any
from supplier_intel.parsers.base import ParsedContent
from supplier_intel.validation import RawSupplierData, ValidationResult
logger = logging.getLogger("supplier_intel.extraction")
# Common header patterns that indicate supplier-relevant columns
SUPPLIER_NAME_PATTERNS = [
"supplier", "vendor", "company", "manufacturer", "name", "firm",
]
ATTRIBUTE_MAPPING: dict[str, list[str]] = {
"price": ["price", "cost", "unit_price", "rate"],
"delivery_time": ["delivery", "lead_time", "delivery_time", "lead time"],
"co2_emissions": ["co2", "carbon", "emissions", "co2_emissions"],
"production_capacity": ["capacity", "production_capacity", "volume"],
"certifications": ["certification", "certifications", "cert", "iso"],
"materials": ["material", "materials", "raw_material"],
"country": ["country", "location", "region", "origin"],
"city": ["city", "town", "site"],
"category": ["category", "product", "product_category", "type"],
"quality_metrics": ["quality", "defect_rate", "quality_score"],
"sustainability": ["sustainability", "esg", "green"],
"margins": ["margin", "margins", "profit_margin"],
"risk_indicators": ["risk", "risk_score", "risk_level"],
}
def _find_name_column(headers: list[str]) -> str | None:
"""Identify which column contains supplier names."""
headers_lower = [h.lower().replace(" ", "_") for h in headers]
for pattern in SUPPLIER_NAME_PATTERNS:
for header in headers_lower:
if pattern in header:
return headers[headers_lower.index(header)]
return None
def _map_column_to_attribute(column_name: str) -> str | None:
"""Map a column header to a standardized attribute name."""
col_lower = column_name.lower().replace(" ", "_")
for attr_name, patterns in ATTRIBUTE_MAPPING.items():
for pattern in patterns:
if pattern in col_lower:
return attr_name
return col_lower # Use cleaned column name as-is if no mapping found
def extract_from_table(
table: list[dict[str, str]], source_file: str
) -> list[RawSupplierData]:
"""Extract supplier records from a parsed table."""
if not table:
return []
headers = list(table[0].keys())
name_column = _find_name_column(headers)
if not name_column:
logger.debug(f"No supplier name column found in table from {source_file}")
return []
records: list[RawSupplierData] = []
for row_idx, row in enumerate(table):
supplier_name = row.get(name_column, "").strip()
if not supplier_name:
continue
attributes: dict[str, Any] = {}
country = None
city = None
category = None
for col_name, value in row.items():
if col_name == name_column or not value:
continue
attr_name = _map_column_to_attribute(col_name)
if attr_name == "country":
country = value
elif attr_name == "city":
city = value
elif attr_name == "category":
category = value
else:
attributes[attr_name] = value
try:
record = RawSupplierData(
name=supplier_name,
country=country,
city=city,
category=category,
attributes=attributes,
source_file=source_file,
source_location=f"row_{row_idx + 1}",
extraction_method="table",
)
records.append(record)
except Exception as e:
logger.warning(f"Validation failed for row {row_idx}: {e}")
return records
def extract_from_content(content: ParsedContent) -> ValidationResult:
"""Extract all supplier records from parsed document content."""
result = ValidationResult()
for table in content.tables:
records = extract_from_table(table, content.file_path)
result.valid_records.extend(records)
result.total_processed = len(result.valid_records) + len(result.invalid_records)
logger.info(
f"Extracted {len(result.valid_records)} valid records from {content.file_path}"
)
return result"""Normalization engine for standardizing supplier data."""
import logging
import re
import unicodedata
from supplier_intel.validation import RawSupplierData
logger = logging.getLogger("supplier_intel.normalization")
# Common company suffixes to normalize
COMPANY_SUFFIXES = [
r"\bltd\.?\b", r"\blimited\b", r"\binc\.?\b", r"\bincorporated\b",
r"\bcorp\.?\b", r"\bcorporation\b", r"\bgmbh\b", r"\bag\b",
r"\bs\.?a\.?\b", r"\bb\.?v\.?\b", r"\bn\.?v\.?\b", r"\bllc\b",
r"\bco\.?\b", r"\bcompany\b", r"\b& co\.?\b", r"\bplc\b",
]
# Country name normalization map
COUNTRY_ALIASES: dict[str, str] = {
"us": "United States",
"usa": "United States",
"united states of america": "United States",
"uk": "United Kingdom",
"gb": "United Kingdom",
"great britain": "United Kingdom",
"de": "Germany",
"deutschland": "Germany",
"cn": "China",
"prc": "China",
"jp": "Japan",
"fr": "France",
"nl": "Netherlands",
"holland": "Netherlands",
"kr": "South Korea",
"south korea": "South Korea",
"republic of korea": "South Korea",
}
# Unit normalization patterns
UNIT_CONVERSIONS: dict[str, dict[str, float]] = {
"weight": {"kg": 1.0, "g": 0.001, "lb": 0.453592, "lbs": 0.453592, "t": 1000.0},
"currency": {"usd": 1.0, "$": 1.0, "eur": 1.08, "gbp": 1.27},
"time_days": {"days": 1.0, "weeks": 7.0, "months": 30.0, "d": 1.0, "w": 7.0},
}
def normalize_company_name(name: str) -> str:
"""Normalize a company name for deduplication matching."""
# Lowercase and strip
normalized = name.lower().strip()
# Remove accents/diacritics
normalized = unicodedata.normalize("NFKD", normalized)
normalized = normalized.encode("ascii", "ignore").decode("ascii")
# Remove common suffixes
for suffix_pattern in COMPANY_SUFFIXES:
normalized = re.sub(suffix_pattern, "", normalized, flags=re.IGNORECASE)
# Remove punctuation and extra whitespace
normalized = re.sub(r"[^\w\s]", "", normalized)
normalized = " ".join(normalized.split())
return normalized
def normalize_country(country: str | None) -> str | None:
"""Normalize country name to a standard form."""
if not country:
return None
country_lower = country.lower().strip()
return COUNTRY_ALIASES.get(country_lower, country.strip().title())
def normalize_numeric_value(value: str, unit_category: str | None = None) -> str:
"""Extract and normalize numeric values with units."""
# Try to extract a number
match = re.search(r"([\d,]+\.?\d*)", value.replace(",", ""))
if not match:
return value
number = float(match.group(1).replace(",", ""))
if unit_category and unit_category in UNIT_CONVERSIONS:
value_lower = value.lower()
conversions = UNIT_CONVERSIONS[unit_category]
for unit, factor in conversions.items():
if unit in value_lower:
number = number * factor
break
return str(round(number, 4))
def normalize_record(record: RawSupplierData) -> RawSupplierData:
"""Apply all normalization rules to a supplier record."""
record.country = normalize_country(record.country)
# Normalize numeric attributes where applicable
numeric_attrs = ["price", "delivery_time", "co2_emissions", "production_capacity", "margins"]
for attr_name in numeric_attrs:
if attr_name in record.attributes and isinstance(record.attributes[attr_name], str):
unit_cat = None
if "time" in attr_name:
unit_cat = "time_days"
elif "price" in attr_name or "margin" in attr_name:
unit_cat = "currency"
record.attributes[attr_name] = normalize_numeric_value(
record.attributes[attr_name], unit_cat
)
return record
def find_duplicates(
records: list[RawSupplierData],
) -> dict[str, list[RawSupplierData]]:
"""Group records by normalized company name to identify duplicates."""
groups: dict[str, list[RawSupplierData]] = {}
for record in records:
key = normalize_company_name(record.name)
if key not in groups:
groups[key] = []
groups[key].append(record)
return groupsYour extraction and normalization layers are ready. The pipeline can now take raw parsed content, validate it into structured records, and standardize names, countries, and numeric values. Next up, you will wire these modules into a pipeline orchestrator that stores everything in PostgreSQL with full audit trails.
Build the Pipeline Orchestrator
You now have all the individual processing modules ready. Parsers extract raw content from documents. The extraction engine converts that content into validated records. The normalization layer standardizes values and detects duplicates.
The missing piece is the orchestrator that chains these modules into a single, reliable pipeline. In this step, you will build a Pipeline class that coordinates the entire flow from file ingestion through to database storage.
In this step, get ready to:
- Build the pipeline runner that chains all processing modules together.
- Implement idempotent storage using PostgreSQL upserts to prevent duplicates on re-runs.
- Add structured logging with traceability from extracted data back to source documents.
Create the Pipeline class foundation
The pipeline orchestrator needs to know how to connect to the database, select the right parser for each file type, and detect whether a file has changed since its last processing run.
- Create a new file called pipeline.py inside your supplier_intel folder.
- Add the imports, logger, and the Pipeline class with its initialization and helper methods by pasting this code:
"""Pipeline orchestrator that chains all processing modules together."""
import hashlib
import logging
from datetime import datetime, timezone
from pathlib import Path
from sqlalchemy import create_engine, text
from sqlalchemy.dialects.postgresql import insert
from sqlalchemy.orm import Session, sessionmaker
from supplier_intel.config import DATABASE_URL
from supplier_intel.extraction import extract_from_content
from supplier_intel.models import Base, ExtractionRecord, SourceDocument, Supplier
from supplier_intel.normalization import find_duplicates, normalize_company_name, normalize_record
from supplier_intel.parsers import BaseParser, ExcelParser, PowerPointParser, WordParser
from supplier_intel.validation import RawSupplierData
logger = logging.getLogger("supplier_intel.pipeline")
class Pipeline:
def __init__(self, database_url: str | None = None):
self.engine = create_engine(database_url or DATABASE_URL, echo=False)
self.SessionLocal = sessionmaker(bind=self.engine)
self.parsers: list[BaseParser] = [ExcelParser(), WordParser(), PowerPointParser()]
def _get_parser(self, file_path: Path) -> BaseParser | None:
"""Find the appropriate parser for a file."""
for parser in self.parsers:
if parser.can_handle(file_path):
return parser
return None
def _compute_file_hash(self, file_path: Path) -> str:
"""Compute SHA-256 hash of a file for change detection."""
sha256 = hashlib.sha256()
with open(file_path, "rb") as f:
for chunk in iter(lambda: f.read(8192), b""):
sha256.update(chunk)
return sha256.hexdigest()What does this code do?
The Pipeline class is the central coordinator. It creates a database engine from the connection URL, sets up a session factory, and instantiates all three parsers (Excel, Word, PowerPoint).
The _get_parser helper iterates through parsers and returns the one that can handle a given file extension. The _compute_file_hash method generates a SHA-256 hash of a file's contents. This lets the pipeline detect whether a file has changed since its last run.
- Save pipeline.py.
You won't see a visible change yet. The next chunk adds the document registration logic that uses these helpers.
Add document registration with change detection
Before the pipeline processes a file, it needs to check whether that file is new or has changed since the last run. This prevents wasted work on re-runs and forms the first layer of idempotency.
- In pipeline.py, add the _register_document method inside the Pipeline class, below _compute_file_hash:
def _register_document(self, session: Session, file_path: Path) -> SourceDocument | None:
"""Register or update a source document. Returns None if unchanged."""
file_hash = self._compute_file_hash(file_path)
existing = session.query(SourceDocument).filter_by(file_path=str(file_path)).first()
if existing and existing.file_hash == file_hash:
return None
if existing:
existing.file_hash = file_hash
existing.processing_status = "pending"
existing.processed_at = None
return existing
doc = SourceDocument(
file_path=str(file_path), file_name=file_path.name,
file_type=file_path.suffix.lower().lstrip("."),
file_hash=file_hash, file_size_bytes=file_path.stat().st_size,
)
session.add(doc)
session.flush()
return docWhat does this code do?
This method implements three-way logic for document registration:
- If the file exists in the database with the same hash, it returns None (skip processing, nothing changed).
- If the file exists but its hash differs, it resets the document status so it gets reprocessed.
- If the file is brand new, it creates a SourceDocument record and flushes it to get an ID for linking extraction records later.
- Save pipeline.py.
You won't see a change yet. The next chunk adds the upsert methods that write supplier data to the database.
Implement idempotent supplier upserts
The pipeline must be safely re-runnable. If you process the same document twice, the database should end up in the same state as if you processed it once. PostgreSQL's INSERT ON CONFLICT pattern (also called an upsert) makes this possible.
- Add the _upsert_supplier method below _register_document:
def _upsert_supplier(self, session: Session, record: RawSupplierData) -> Supplier:
"""Insert or update a supplier using PostgreSQL upsert."""
normalized_name = normalize_company_name(record.name)
stmt = insert(Supplier).values(
name=record.name, normalized_name=normalized_name,
country=record.country, city=record.city,
primary_category=record.category, attributes=record.attributes, confidence_score=1.0,
)
# On conflict, merge attributes (existing + new)
stmt = stmt.on_conflict_do_update(
index_elements=["normalized_name"],
set_={
"country": stmt.excluded.country, "city": stmt.excluded.city,
"primary_category": stmt.excluded.primary_category,
"attributes": text("suppliers.attributes || EXCLUDED.attributes"),
"updated_at": datetime.now(timezone.utc),
},
)
session.execute(stmt)
session.flush()
return session.query(Supplier).filter_by(normalized_name=normalized_name).first()What does this code do?
This is the core idempotency pattern. The insert(...).on_conflict_do_update() call tells PostgreSQL: try to insert a new supplier, but if one already exists with the same normalized_name, update it instead.
The suppliers.attributes || EXCLUDED.attributes expression merges the existing JSONB attributes with the new ones. Existing keys get overwritten with newer data, while keys unique to either side are preserved.
Why upserts instead of check-then-insert?
A naive approach would query first, then insert or update. That creates a race condition in concurrent systems and requires two round trips to the database. The upsert is atomic. PostgreSQL handles the conflict detection and resolution in a single statement.
- Save pipeline.py.
You won't see a visible change yet. The next chunk applies the same upsert pattern to extraction records.
- Add the _store_extraction_record method below _upsert_supplier:
def _store_extraction_record(self, session: Session, supplier: Supplier, doc: SourceDocument, record: RawSupplierData) -> None:
"""Store extraction audit trail."""
for attr_name, attr_value in record.attributes.items():
stmt = insert(ExtractionRecord).values(
supplier_id=supplier.id, source_document_id=doc.id,
attribute_name=attr_name, raw_value=str(attr_value) if attr_value else None,
normalized_value=str(attr_value) if attr_value else None,
extraction_method=record.extraction_method, source_location=record.source_location, confidence=1.0,
)
stmt = stmt.on_conflict_do_update(
constraint="uq_extraction_supplier_doc_attr",
set_={"raw_value": stmt.excluded.raw_value, "normalized_value": stmt.excluded.normalized_value},
)
session.execute(stmt)What does this code do?
This method creates one extraction record per attribute, linking the supplier to the source document. Each record stores both the raw and normalized values, plus metadata about how and where the data was extracted.
The upsert uses the uq_extraction_supplier_doc_attr unique constraint you defined in Step 3. If the same attribute from the same document for the same supplier already exists, it updates the values instead of creating a duplicate.
- Save pipeline.py.
You won't see a change yet. The next chunk ties everything together with the public processing methods.
Add the orchestration methods with structured logging
The final piece is the public API: a method that processes a single file through all pipeline stages, and a batch method that processes an entire directory. These methods provide traceability by updating document status and tracking statistics.
- Add the process_file method below _store_extraction_record:
def process_file(self, file_path: Path) -> int:
"""Process a single file through the full pipeline. Returns records extracted."""
parser = self._get_parser(file_path)
if not parser:
return 0
session = self.SessionLocal()
try:
# Register document (skips if unchanged)
doc = self._register_document(session, file_path)
if doc is None:
session.commit()
return 0
# Parse the document
content = parser.parse(file_path)
# Extract and validate
result = extract_from_content(content)
# Normalize all records
normalized_records = [normalize_record(r) for r in result.valid_records]
# Store each record with upsert
records_stored = 0
for record in normalized_records:
supplier = self._upsert_supplier(session, record)
self._store_extraction_record(session, supplier, doc, record)
records_stored += 1
# Update document processing status
doc.processing_status = "completed"
doc.processed_at = datetime.now(timezone.utc)
doc.records_extracted = records_stored
session.commit()
return records_stored
except Exception as e:
session.rollback()
raise
finally:
session.close()What does this code do?
This is the full pipeline flow in one method: ingest, parse, extract, validate, normalize, store. Each step feeds into the next.
- The method opens a database session, registers the document, and skips early if nothing changed.
- It calls the parser, extraction engine, and normalizer in sequence.
- It commits all changes atomically. If anything fails, the entire transaction is rolled back.
- The document's processing_status and records_extracted fields provide traceability for auditing.
- Save pipeline.py.
You won't see a change yet. The next chunk adds batch directory processing.
- Add the process_directory method below process_file:
def process_directory(self, directory: Path) -> dict[str, int]:
"""Process all supported files in a directory."""
stats = {"processed": 0, "skipped": 0, "failed": 0, "total_records": 0}
supported_extensions = set()
for parser in self.parsers:
supported_extensions.update(parser.supported_extensions)
files = [f for f in directory.rglob("*") if f.suffix.lower() in supported_extensions and f.is_file()]
for file_path in files:
try:
records = self.process_file(file_path)
if records > 0:
stats["processed"] += 1
stats["total_records"] += records
else:
stats["skipped"] += 1
except Exception:
stats["failed"] += 1
return statsWhat does this code do?
This method recursively scans a directory for supported file types using rglob("*"). It processes each file and tracks statistics: how many files were processed, skipped (unchanged), or failed.
Failed files don't halt the batch. The pipeline continues processing remaining files and reports failures in the stats dictionary at the end.
- Save pipeline.py.
- Verify the file exists by running:
python -c "from supplier_intel.pipeline import Pipeline; print('Pipeline module loaded successfully')"You should see Pipeline module loaded successfully printed to the terminal.
Seeing an ImportError?
- Make sure your virtual environment is activated. The prompt should show (venv) or similar.
- Check that all prior modules exist: supplier_intel/models.py, supplier_intel/extraction.py, supplier_intel/normalization.py, and the parsers package.
- Confirm pipeline.py is inside the supplier_intel folder (not the project root).
Why is my Pipeline import failing?
✔️ Awesome, I've got everything!
Great. Double check you have saved pipeline.py with all six methods inside the Pipeline class.
ⓧ I'd like to double check the full code
"""Pipeline orchestrator that chains all processing modules together."""
import hashlib
import logging
from datetime import datetime, timezone
from pathlib import Path
from sqlalchemy import create_engine, text
from sqlalchemy.dialects.postgresql import insert
from sqlalchemy.orm import Session, sessionmaker
from supplier_intel.config import DATABASE_URL
from supplier_intel.extraction import extract_from_content
from supplier_intel.models import Base, ExtractionRecord, SourceDocument, Supplier
from supplier_intel.normalization import find_duplicates, normalize_company_name, normalize_record
from supplier_intel.parsers import BaseParser, ExcelParser, PowerPointParser, WordParser
from supplier_intel.validation import RawSupplierData
logger = logging.getLogger("supplier_intel.pipeline")
class Pipeline:
def __init__(self, database_url: str | None = None):
self.engine = create_engine(database_url or DATABASE_URL, echo=False)
self.SessionLocal = sessionmaker(bind=self.engine)
self.parsers: list[BaseParser] = [ExcelParser(), WordParser(), PowerPointParser()]
def _get_parser(self, file_path: Path) -> BaseParser | None:
for parser in self.parsers:
if parser.can_handle(file_path):
return parser
return None
def _compute_file_hash(self, file_path: Path) -> str:
sha256 = hashlib.sha256()
with open(file_path, "rb") as f:
for chunk in iter(lambda: f.read(8192), b""):
sha256.update(chunk)
return sha256.hexdigest()
def _register_document(self, session: Session, file_path: Path) -> SourceDocument | None:
file_hash = self._compute_file_hash(file_path)
existing = session.query(SourceDocument).filter_by(file_path=str(file_path)).first()
if existing and existing.file_hash == file_hash:
return None
if existing:
existing.file_hash = file_hash
existing.processing_status = "pending"
existing.processed_at = None
return existing
doc = SourceDocument(
file_path=str(file_path), file_name=file_path.name,
file_type=file_path.suffix.lower().lstrip("."),
file_hash=file_hash, file_size_bytes=file_path.stat().st_size,
)
session.add(doc)
session.flush()
return doc
def _upsert_supplier(self, session: Session, record: RawSupplierData) -> Supplier:
normalized_name = normalize_company_name(record.name)
stmt = insert(Supplier).values(
name=record.name, normalized_name=normalized_name,
country=record.country, city=record.city,
primary_category=record.category, attributes=record.attributes, confidence_score=1.0,
)
stmt = stmt.on_conflict_do_update(
index_elements=["normalized_name"],
set_={
"country": stmt.excluded.country, "city": stmt.excluded.city,
"primary_category": stmt.excluded.primary_category,
"attributes": text("suppliers.attributes || EXCLUDED.attributes"),
"updated_at": datetime.now(timezone.utc),
},
)
session.execute(stmt)
session.flush()
return session.query(Supplier).filter_by(normalized_name=normalized_name).first()
def _store_extraction_record(self, session: Session, supplier: Supplier, doc: SourceDocument, record: RawSupplierData) -> None:
for attr_name, attr_value in record.attributes.items():
stmt = insert(ExtractionRecord).values(
supplier_id=supplier.id, source_document_id=doc.id,
attribute_name=attr_name, raw_value=str(attr_value) if attr_value else None,
normalized_value=str(attr_value) if attr_value else None,
extraction_method=record.extraction_method, source_location=record.source_location, confidence=1.0,
)
stmt = stmt.on_conflict_do_update(
constraint="uq_extraction_supplier_doc_attr",
set_={"raw_value": stmt.excluded.raw_value, "normalized_value": stmt.excluded.normalized_value},
)
session.execute(stmt)
def process_file(self, file_path: Path) -> int:
parser = self._get_parser(file_path)
if not parser:
return 0
session = self.SessionLocal()
try:
doc = self._register_document(session, file_path)
if doc is None:
session.commit()
return 0
content = parser.parse(file_path)
result = extract_from_content(content)
normalized_records = [normalize_record(r) for r in result.valid_records]
records_stored = 0
for record in normalized_records:
supplier = self._upsert_supplier(session, record)
self._store_extraction_record(session, supplier, doc, record)
records_stored += 1
doc.processing_status = "completed"
doc.processed_at = datetime.now(timezone.utc)
doc.records_extracted = records_stored
session.commit()
return records_stored
except Exception as e:
session.rollback()
raise
finally:
session.close()
def process_directory(self, directory: Path) -> dict[str, int]:
stats = {"processed": 0, "skipped": 0, "failed": 0, "total_records": 0}
supported_extensions = set()
for parser in self.parsers:
supported_extensions.update(parser.supported_extensions)
files = [f for f in directory.rglob("*") if f.suffix.lower() in supported_extensions and f.is_file()]
for file_path in files:
try:
records = self.process_file(file_path)
if records > 0:
stats["processed"] += 1
stats["total_records"] += records
else:
stats["skipped"] += 1
except Exception:
stats["failed"] += 1
return statsYour pipeline orchestrator is complete. It chains every module you have built so far into a single, idempotent flow. Next up, you will build the matrix generator that queries all this stored data and exports the final supplier matrix.
Generate the Supplier Matrix
Your pipeline can now ingest documents, parse them, extract supplier data, normalize it, and store everything in PostgreSQL with full audit trails. The final piece is the deliverable itself.
In this step, you will build the matrix generator that queries all stored supplier data and assembles it into a consolidated matrix across 35+ evaluation categories. You will also create the command-line entry point that ties the entire system together.
In this step, get ready to:
- Build the matrix query engine that assembles supplier attributes across 35+ categories.
- Implement export to both CSV and Excel formats.
- Create the entry point script and summary report with coverage statistics.
Build the matrix query engine
The matrix generator queries every supplier in the database and flattens their JSONB attributes into a single row per supplier. Each row spans 35+ evaluation categories, giving procurement teams a complete view of every supplier in one spreadsheet.
- Create a new file called matrix.py inside your supplier_intel folder.
- Add the imports, category list, and the MatrixGenerator class with its generate_matrix method by pasting the following code:
"""Supplier matrix generation module."""
import csv
import logging
from pathlib import Path
from openpyxl import Workbook
from sqlalchemy import create_engine, func
from sqlalchemy.orm import sessionmaker
from supplier_intel.config import DATABASE_URL
from supplier_intel.models import ExtractionRecord, Supplier
logger = logging.getLogger("supplier_intel.matrix")
# The 35+ evaluation categories for the supplier matrix
MATRIX_CATEGORIES = [
"price", "co2_emissions", "delivery_time", "production_capacity",
"margins", "production_possibilities", "materials", "certifications",
"sustainability", "geographic_location", "lead_times", "product_categories",
"quality_metrics", "risk_indicators", "minimum_order_quantity",
"payment_terms", "warranty", "return_policy", "communication_rating",
"financial_stability", "years_in_business", "employee_count",
"revenue", "insurance_coverage", "compliance_status",
"technology_capability", "innovation_score", "flexibility",
"after_sales_support", "packaging", "logistics_capability",
"inventory_management", "scalability", "ethical_practices",
"diversity_inclusion",
]
class MatrixGenerator:
"""Generates the consolidated supplier matrix from the database."""
def __init__(self, database_url: str | None = None):
self.engine = create_engine(database_url or DATABASE_URL)
self.SessionLocal = sessionmaker(bind=self.engine)
def generate_matrix(self) -> list[dict[str, str]]:
"""Query all suppliers and build the matrix data structure."""
session = self.SessionLocal()
try:
suppliers = session.query(Supplier).order_by(Supplier.name).all()
matrix_rows: list[dict[str, str]] = []
for supplier in suppliers:
# Core relational fields form the first columns
row: dict[str, str] = {
"supplier_name": supplier.name,
"normalized_name": supplier.normalized_name,
"country": supplier.country or "",
"city": supplier.city or "",
"primary_category": supplier.primary_category or "",
"status": supplier.status,
"confidence_score": str(supplier.confidence_score),
"requires_review": str(supplier.requires_review),
}
# Fill in all matrix categories from JSONB attributes
for category in MATRIX_CATEGORIES:
value = supplier.attributes.get(category, "")
row[category] = str(value) if value else ""
matrix_rows.append(row)
return matrix_rows
finally:
session.close()What does this code do?
MATRIX_CATEGORIES defines the 35 evaluation dimensions that procurement teams care about. Each supplier row will have a column for every one of these categories.
The generate_matrix method queries all suppliers ordered alphabetically, then flattens each supplier's relational fields (name, country, city) and JSONB attributes column into a single flat dictionary. This is what makes the hybrid schema pay off. The relational columns give you fast filtering, while the JSONB column holds every dynamic category without requiring schema changes.
- Save matrix.py.
Seeing an import error?
Make sure you are importing ExtractionRecord and Supplier from supplier_intel.models. If Python cannot find the module, confirm your virtual environment is activated and that models.py exists in the supplier_intel folder.
Help me debug an import error in matrix.py
Implement CSV and Excel export
The matrix data structure is useless without a way to deliver it. Procurement teams need the data in spreadsheet formats they already use daily.
- In matrix.py, add the export_csv and export_excel methods inside the MatrixGenerator class, below the generate_matrix method:
def export_csv(self, output_path: Path) -> None:
"""Export the supplier matrix to a CSV file."""
matrix = self.generate_matrix()
if not matrix:
return
fieldnames = list(matrix[0].keys())
with open(output_path, "w", newline="", encoding="utf-8") as f:
writer = csv.DictWriter(f, fieldnames=fieldnames)
writer.writeheader()
writer.writerows(matrix)
def export_excel(self, output_path: Path) -> None:
"""Export the supplier matrix to an Excel file."""
matrix = self.generate_matrix()
if not matrix:
return
wb = Workbook()
ws = wb.active
ws.title = "Supplier Matrix"
# Write headers as the first row
headers = list(matrix[0].keys())
for col, header in enumerate(headers, 1):
ws.cell(row=1, column=col, value=header)
# Write each supplier as a data row
for row_idx, row_data in enumerate(matrix, 2):
for col_idx, header in enumerate(headers, 1):
ws.cell(row=row_idx, column=col_idx, value=row_data.get(header, ""))
wb.save(str(output_path))What do these methods do?
Both methods call generate_matrix() to get the full supplier data, then serialize it into the target format.
- export_csv uses Python's built-in csv.DictWriter to produce a standards-compliant CSV with proper encoding.
- export_excel uses openpyxl to write a proper .xlsx file with the sheet named "Supplier Matrix".
- Save matrix.py.
IndentationError when adding these methods?
Both export_csv and export_excel must be indented at the same level as generate_matrix. They are methods of the MatrixGenerator class, so they need exactly 4 spaces of indentation for the def keyword.
Help me fix an indentation error in my MatrixGenerator class
Add coverage statistics and create the entry point
A raw matrix is helpful, but knowing which categories have gaps is critical for procurement decisions. The coverage report shows exactly which suppliers have which attributes populated.
- In matrix.py, add the get_coverage_stats method inside the MatrixGenerator class, between generate_matrix and export_csv:
def get_coverage_stats(self) -> dict[str, dict[str, int]]:
"""Calculate coverage statistics for each matrix category."""
session = self.SessionLocal()
try:
total_suppliers = session.query(func.count(Supplier.id)).scalar() or 0
stats: dict[str, dict[str, int]] = {}
for category in MATRIX_CATEGORIES:
# Count suppliers that have this attribute populated
count = (
session.query(func.count(Supplier.id))
.filter(
Supplier.attributes[category].astext != "",
Supplier.attributes.has_key(category),
)
.scalar()
or 0
)
stats[category] = {
"populated": count,
"total": total_suppliers,
"coverage_pct": round(count / total_suppliers * 100, 1) if total_suppliers > 0 else 0.0,
}
return stats
finally:
session.close()What does get_coverage_stats do?
This method iterates over every category in MATRIX_CATEGORIES and queries PostgreSQL to count how many suppliers have that attribute populated. It leverages SQLAlchemy's JSONB operators (.astext and .has_key()) to check the JSONB column directly in SQL.
The result tells you things like "32 out of 500 suppliers have delivery_time populated (6.4% coverage)." This is how you identify data gaps that need more document ingestion.
- Save matrix.py.
Now you need a command-line entry point that ties the pipeline and matrix generator together into one runnable script.
- Create a new file called run_pipeline.py in your project root (the same level as requirements.txt).
- Paste the following code into run_pipeline.py:
"""Entry point script for running the supplier intelligence pipeline."""
import sys
from pathlib import Path
from supplier_intel.config import setup_logging
from supplier_intel.matrix import MatrixGenerator
from supplier_intel.pipeline import Pipeline
def main():
"""Run the supplier intelligence pipeline."""
logger = setup_logging()
if len(sys.argv) < 2:
print("Usage: python run_pipeline.py <input_directory> [output_file]")
sys.exit(1)
input_dir = Path(sys.argv[1])
output_file = Path(sys.argv[2]) if len(sys.argv) > 2 else Path("supplier_matrix.xlsx")
if not input_dir.exists():
print(f"Error: Directory not found: {input_dir}")
sys.exit(1)
# Run the pipeline
pipeline = Pipeline()
stats = pipeline.process_directory(input_dir)
print(f"\n{'='*50}")
print("PIPELINE RESULTS")
print(f"{'='*50}")
print(f" Files processed: {stats['processed']}")
print(f" Files skipped: {stats['skipped']}")
print(f" Files failed: {stats['failed']}")
print(f" Total records: {stats['total_records']}")
# Generate matrix
generator = MatrixGenerator()
if output_file.suffix == ".csv":
generator.export_csv(output_file)
else:
generator.export_excel(output_file)
# Print coverage stats
coverage = generator.get_coverage_stats()
print(f"\n{'='*50}")
print("COVERAGE STATISTICS")
print(f"{'='*50}")
for category, data in sorted(coverage.items()):
bar = "#" * int(data["coverage_pct"] / 5) if data["coverage_pct"] > 0 else ""
print(f" {category:<30} {data['populated']:>4}/{data['total']:<4} ({data['coverage_pct']:>5.1f}%) {bar}")
print(f"\nMatrix exported to: {output_file}")
if __name__ == "__main__":
main()What does run_pipeline.py do?
- It accepts a directory path as a command-line argument and an optional output file name (defaults to supplier_matrix.xlsx).
- It runs the full pipeline on every supported file in that directory.
- After processing, it generates the supplier matrix and prints a coverage report showing which categories have data and which have gaps.
- Save run_pipeline.py.
- Verify the script loads without errors by running:
python run_pipeline.pyYou should see the usage message: Usage: python run_pipeline.py <input_directory> [output_file]. This confirms all imports resolve correctly and the system is wired together.
Seeing a ModuleNotFoundError?
Confirm your virtual environment is activated. If the error mentions supplier_intel, make sure you are running the command from your project root directory (the folder that contains the supplier_intel folder and run_pipeline.py).
If the error mentions openpyxl or sqlalchemy, re-run pip install -r requirements.txt to make sure all dependencies are installed.
Help me fix a ModuleNotFoundError when running run_pipeline.py
To test the full pipeline end-to-end, create a sample data directory and run the pipeline against it:
- Create a data folder in your project root by running:
mkdir data- Run the pipeline against the empty data directory to confirm it completes without errors:
python run_pipeline.py dataYou should see the pipeline results showing 0 files processed and a coverage statistics section. The system is working end-to-end. When you drop real Excel, Word, or PowerPoint files with supplier data into the data folder and re-run, the pipeline will extract, normalize, and store supplier records, then produce a full matrix.
✔️ Awesome, I've got everything!
Great. Make sure both matrix.py and run_pipeline.py are saved.
ⓧ I'd like to double check the full code
Here is the complete supplier_intel/matrix.py:
"""Supplier matrix generation module."""
import csv
import logging
from pathlib import Path
from openpyxl import Workbook
from sqlalchemy import create_engine, func
from sqlalchemy.orm import sessionmaker
from supplier_intel.config import DATABASE_URL
from supplier_intel.models import ExtractionRecord, Supplier
logger = logging.getLogger("supplier_intel.matrix")
MATRIX_CATEGORIES = [
"price", "co2_emissions", "delivery_time", "production_capacity",
"margins", "production_possibilities", "materials", "certifications",
"sustainability", "geographic_location", "lead_times", "product_categories",
"quality_metrics", "risk_indicators", "minimum_order_quantity",
"payment_terms", "warranty", "return_policy", "communication_rating",
"financial_stability", "years_in_business", "employee_count",
"revenue", "insurance_coverage", "compliance_status",
"technology_capability", "innovation_score", "flexibility",
"after_sales_support", "packaging", "logistics_capability",
"inventory_management", "scalability", "ethical_practices",
"diversity_inclusion",
]
class MatrixGenerator:
def __init__(self, database_url: str | None = None):
self.engine = create_engine(database_url or DATABASE_URL)
self.SessionLocal = sessionmaker(bind=self.engine)
def generate_matrix(self) -> list[dict[str, str]]:
session = self.SessionLocal()
try:
suppliers = session.query(Supplier).order_by(Supplier.name).all()
matrix_rows: list[dict[str, str]] = []
for supplier in suppliers:
row: dict[str, str] = {
"supplier_name": supplier.name,
"normalized_name": supplier.normalized_name,
"country": supplier.country or "",
"city": supplier.city or "",
"primary_category": supplier.primary_category or "",
"status": supplier.status,
"confidence_score": str(supplier.confidence_score),
"requires_review": str(supplier.requires_review),
}
for category in MATRIX_CATEGORIES:
value = supplier.attributes.get(category, "")
row[category] = str(value) if value else ""
matrix_rows.append(row)
return matrix_rows
finally:
session.close()
def get_coverage_stats(self) -> dict[str, dict[str, int]]:
session = self.SessionLocal()
try:
total_suppliers = session.query(func.count(Supplier.id)).scalar() or 0
stats: dict[str, dict[str, int]] = {}
for category in MATRIX_CATEGORIES:
count = (
session.query(func.count(Supplier.id))
.filter(
Supplier.attributes[category].astext != "",
Supplier.attributes.has_key(category),
)
.scalar() or 0
)
stats[category] = {
"populated": count, "total": total_suppliers,
"coverage_pct": round(count / total_suppliers * 100, 1) if total_suppliers > 0 else 0.0,
}
return stats
finally:
session.close()
def export_csv(self, output_path: Path) -> None:
matrix = self.generate_matrix()
if not matrix:
return
fieldnames = list(matrix[0].keys())
with open(output_path, "w", newline="", encoding="utf-8") as f:
writer = csv.DictWriter(f, fieldnames=fieldnames)
writer.writeheader()
writer.writerows(matrix)
def export_excel(self, output_path: Path) -> None:
matrix = self.generate_matrix()
if not matrix:
return
wb = Workbook()
ws = wb.active
ws.title = "Supplier Matrix"
headers = list(matrix[0].keys())
for col, header in enumerate(headers, 1):
ws.cell(row=1, column=col, value=header)
for row_idx, row_data in enumerate(matrix, 2):
for col_idx, header in enumerate(headers, 1):
ws.cell(row=row_idx, column=col_idx, value=row_data.get(header, ""))
wb.save(str(output_path))And here is the complete run_pipeline.py:
"""Entry point script for running the supplier intelligence pipeline."""
import sys
from pathlib import Path
from supplier_intel.config import setup_logging
from supplier_intel.matrix import MatrixGenerator
from supplier_intel.pipeline import Pipeline
def main():
logger = setup_logging()
if len(sys.argv) < 2:
print("Usage: python run_pipeline.py <input_directory> [output_file]")
sys.exit(1)
input_dir = Path(sys.argv[1])
output_file = Path(sys.argv[2]) if len(sys.argv) > 2 else Path("supplier_matrix.xlsx")
if not input_dir.exists():
print(f"Error: Directory not found: {input_dir}")
sys.exit(1)
pipeline = Pipeline()
stats = pipeline.process_directory(input_dir)
print(f"\n{'='*50}\nPIPELINE RESULTS\n{'='*50}")
print(f" Files processed: {stats['processed']}")
print(f" Files skipped: {stats['skipped']}")
print(f" Files failed: {stats['failed']}")
print(f" Total records: {stats['total_records']}")
generator = MatrixGenerator()
if output_file.suffix == ".csv":
generator.export_csv(output_file)
else:
generator.export_excel(output_file)
coverage = generator.get_coverage_stats()
print(f"\n{'='*50}\nCOVERAGE STATISTICS\n{'='*50}")
for category, data in sorted(coverage.items()):
bar = "#" * int(data["coverage_pct"] / 5) if data["coverage_pct"] > 0 else ""
print(f" {category:<30} {data['populated']:>4}/{data['total']:<4} ({data['coverage_pct']:>5.1f}%) {bar}")
print(f"\nMatrix exported to: {output_file}")
if __name__ == "__main__":
main()Your supplier intelligence system is now fully operational. You have a complete pipeline that ingests documents, extracts supplier data, normalizes it, stores it with full traceability, and produces the final deliverable. Next up, the Secret Mission challenges you to wrap this in a professional CLI interface.
Secret mission
Add a CLI Interface with Click
Your pipeline works, but running it requires editing scripts or remembering long command-line arguments. In this secret mission, you will build a proper CLI using Click and Rich that gives you a polished developer experience with progress bars, colored output, dry-run mode, and pipeline statistics at your fingertips.
Clean Up Your Resources
Clean Up Your Resources
This project runs entirely locally with a free PostgreSQL database. There are no ongoing cloud costs. Decide whether to keep your resources running, pause them to come back later, or delete them entirely.
Resources you used:
- PostgreSQL supplier_intel database (with tables: suppliers, source_documents, extraction_records).
- PostgreSQL 17 service running on your machine.
- Project folder on your Desktop containing all Python source files, virtual environment, and generated output files.
Keep everything running
No action needed. Choose this if you plan to continue developing the pipeline, adding new parsers, or extending the system for Phase 2.
- The PostgreSQL database persists on your machine and costs nothing.
- The PostgreSQL service uses minimal memory when idle.
- Your project files remain on your Desktop ready for future work.
Pause - I'll come back to this later
Stop the PostgreSQL service to free up system memory. Your database and project files stay intact for when you return.
- Open a terminal as Administrator (right-click your terminal app and select Run as administrator).
- Stop the PostgreSQL service by running:
net stop postgresql-x64-17- When you want to resume, open an elevated terminal and run:
net start postgresql-x64-17What does stopping the service do?
Stopping the PostgreSQL service shuts down the database server process. Your data remains on disk. Starting the service again brings everything back exactly as you left it.
Delete - I don't want to use this again
Remove all project resources and the PostgreSQL database. Follow these steps to clean up completely.
First, drop the database. Open your terminal and run:
psql -U postgres -c "DROP DATABASE supplier_intel;"- Enter the password you set for the postgres user during installation when prompted.
Next, delete your project folder. Navigate to your Desktop and run:
cd ~/Desktop
rmdir /s /q supplier-intelWhat if your folder has a different name?
Replace supplier-intel with whatever you named your project folder. You can check the name by running ls or dir on your Desktop first.
Optionally, uninstall PostgreSQL entirely:
- Press the Windows key to open your search bar.
- Type Add or remove programs and press Enter.
- Find PostgreSQL 17 in the list.
- Click Uninstall and follow the prompts to remove it.
Data directory cleanup
The PostgreSQL uninstaller may leave behind the data directory. After uninstalling, check C:\Program Files\PostgreSQL\17\data and delete it manually if it still exists.
Nice Work!
Nice Work!
You just built a production-grade supplier intelligence system from scratch. This is the kind of architecture that powers enterprise procurement decisions at companies processing thousands of supplier documents.
Here's what you learned:
- Designed a modular ETL pipeline with 7 distinct layers (ingestion, parsing, extraction, validation, normalization, storage, and matrix generation) following clear separation of concerns.
- Built a hybrid PostgreSQL schema combining relational columns with JSONB attributes to flexibly handle 35+ dynamic supplier categories without requiring schema changes.
- Implemented idempotent upserts with full audit trails, ensuring the pipeline can be safely re-run without duplicates and every data point traces back to its source document.
- Secret Mission: Added a CLI interface with Click and Rich providing progress reporting, dry-run mode, and pipeline statistics with colored terminal output.
Ready to quiz yourself?