How to Automate Sports Card Inventory Valuation Pipeline with Python

Managing a sports card collection of 100+ cards means hours of manual data entry, duplicate checking, and price lookups across eBay, PSA, Beckett, and other marketplaces. Collectors, dealers, and estate appraisers face the same repetitive workflow: scan or photograph cards, transcribe details into spreadsheets, search multiple pricing sources, record values, and repeat. A Python-based automation pipeline eliminates this busywork by extracting card data from PDFs or images, identifying duplicates, fetching pricing from six sources, and maintaining a master inventory with timestamped valuation reports—all without requiring manual CSV editing.

This sports cards tutorial walks through the design decisions, key components, and real code examples from a production system that automates sports card inventory valuation pipeline Python workflows. You’ll learn why certain architectural choices prevent data loss, how to fetch pricing without getting blocked, and how to integrate batch processing with orchestration platforms like n8n for hands-free portfolio management and insurance documentation.

Why Automate Sports Card Inventory Valuation Instead of Manual Tracking?

The manual workflow for sports card inventory management is deceptively time-intensive. A collector with 200 cards spends 2–3 minutes per card on data entry alone—that’s 6–10 hours of typing player names, years, sets, and card numbers into a spreadsheet. Add duplicate checking (searching the existing inventory for each new card) and pricing lookups (visiting eBay, PSA, Beckett, COMC, 130point, and PriceCharting for each card), and the total time balloons to 20–30 hours for a single batch.

Insurance documentation and estate appraisals require timestamped valuation reports that prove market values on specific dates. Manually updating 200 rows with six pricing columns every quarter means 1,200+ copy-paste operations—and a single typo can throw off portfolio totals by hundreds of dollars. Dealers preparing cards for sale need to identify duplicates to avoid listing the same card twice, but CSV find-and-replace searches break down when card descriptions vary slightly (“Topps Chrome” vs. “Topps Chrome Refractor”).

An automated valuation pipeline solves these problems by standardizing the workflow from scan to report. Drop a PDF or folder of card images into an incoming directory, and the system extracts data, generates unique identifiers, checks for duplicates, stages pricing lookups in batches (to respect rate limits), fetches valuations from all six sources, merges the data into a master inventory, and produces markdown reports—all without opening a spreadsheet. The result: 95% time savings and consistent, auditable pricing data for portfolio tracking and insurance claims.

System Architecture: PDF Extraction to Valuation Reports

The sports card inventory automation pipeline consists of thirteen Python scripts that form a linear workflow with conditional branching. The architecture separates extraction, duplicate detection, inventory management, batch staging, pricing fetches, aggregation, and reporting into discrete modules. This design allows each step to fail independently without corrupting downstream data—a critical requirement when dealing with rate-limited APIs and unreliable web scraping targets.

The workflow begins with extract_pdf_pages.py, which converts PDFs to per-page JPEGs using PyMuPDF (no external binaries required). The script creates session folders with front/back page pairing logic and writes sidecar JSON files to prevent reprocessing. Here’s the core PDF rendering logic:

import fitz  # PyMuPDF
import os

def extract_pdf_to_images(pdf_path, output_dir, target_dpi=300):
    doc = fitz.open(pdf_path)
    zoom = target_dpi / 72  # Calculate zoom factor from DPI
    matrix = fitz.Matrix(zoom, zoom)

    for page_num in range(len(doc)):
        page = doc[page_num]
        pix = page.get_pixmap(matrix=matrix)
        image_path = os.path.join(output_dir, f"page_{page_num + 1}.jpg")
        pix.save(image_path)

    doc.close()
    return len(doc)

This approach avoids the poppler-utils dependency (which requires system binaries like pdftocairo) and calculates zoom as target_dpi / 72 because PDF coordinate systems use 72 points per inch. The fitz.Matrix applies the scaling during rendering, producing high-quality JPEGs suitable for OCR or manual review.

After extraction, check_duplicates.py compares extracted card_id values against the master inventory CSV. The duplicate checker builds an in-memory dictionary of card_id → count by reading the entire CSV once, then annotates each card_data.json in O(1) lookups—faster than repeated CSV scans for large inventories. write_inventory.py appends new cards to the master CSV and JSON, copying front/back images to a permanent output folder and maintaining a consistent 40-column schema with valuation placeholders.

Valuation begins when prepare_valuation_batch.py stages the next batch of unvalued cards (sorted oldest first by scan_date) into current_batch.json, respecting a configurable batch size to avoid rate limits. fetch_valuations.py queries six pricing sources with browser-like headers and shared requests.Session objects, writing raw JSON per source per card. aggregate_valuations.py parses these JSONs, updates the master inventory, and writes a deduplicated valuations.csv with valuation_date timestamps. Finally, generate_valuation_report.py produces dated markdown reports with summary tables and per-card breakdowns.

This modular pipeline integrates with n8n or similar orchestrators by printing structured status tokens (PAGES_EXTRACTED, BATCH_READY, etc.) to stdout. All scripts exit with code 0 even when no work is available, allowing orchestrators to parse stdout for conditional branching without workflow halts on expected empty states—a design decision that makes the system production-ready for unattended operation.

Preventing Duplicate Processing with Session Markers and card_id Generation

Duplicate detection operates at two levels: preventing reprocessing of input files and identifying duplicate cards in the inventory. Both rely on deterministic identifiers and sidecar metadata files that persist processing state across runs.

When extract_pdf_pages.py completes, it writes a {stem}_session.json file next to the source PDF in the incoming folder. The file contains extraction metadata (page count, timestamp, output paths) and serves as a permanent marker. On subsequent runs, scan_incoming_pdfs.py counts only files without a corresponding _session.json, allowing the system to safely retry failed extractions without re-scanning already-processed PDFs. This design supports incremental workflows: drop new PDFs into the incoming folder at any time, and only unprocessed files enter the pipeline.

Each card receives a unique card_id generated by combining player name, year, set, and card number into a deterministic string:

def generate_card_id(player, year, set_name, card_number):
    """Create deterministic card identifier from core attributes."""
    parts = [
        player.lower().replace(" ", "_"),
        str(year),
        set_name.lower().replace(" ", "_"),
        str(card_number).replace("#", "")
    ]
    return "_".join(parts)

# Example: generate_card_id("Mike Trout", 2011, "Topps Chrome", "175")
# Returns: "mike_trout_2011_topps_chrome_175"

This card_id enables duplicate detection, valuation deduplication, and stable image/markdown file paths without requiring a separate database. When check_duplicates.py loads the master inventory, it builds a dictionary mapping card_id to occurrence count:

import csv

def build_duplicate_map(inventory_csv_path):
    duplicate_map = {}

    with open(inventory_csv_path, 'r', encoding='utf-8') as f:
        reader = csv.DictReader(f)
        for row in reader:
            card_id = row.get('card_id', '').strip()
            if card_id:
                duplicate_map[card_id] = duplicate_map.get(card_id, 0) + 1

    return duplicate_map

For each newly extracted card, the script checks duplicate_map.get(card_id, 0) and writes is_duplicate: true and duplicate_count: N to the card’s card_data.json. This annotation survives inventory updates and enables filtering in reports: “Show only unique cards” or “List all duplicates with counts.” The design is intentionally read-heavy (build the full map once) rather than write-heavy (update a database on every insert), because inventory CSVs grow linearly while duplicate checks are batch operations performed once per extraction session.

Fetching Pricing Data from Six Sources Without Getting Blocked

Web scraping for pricing data faces two primary challenges: rate limiting (sites block repeated requests) and anti-bot measures (sites serve alternate responses to headless clients). The system addresses both through browser-like User-Agent headers, shared session warming, and isolated error handling that prevents one blocked source from stopping others.

fetch_valuations.py creates a requests.Session for each pricing source and preloads homepages to establish cookies before searching:

import requests
from bs4 import BeautifulSoup
import time

def fetch_pricecharting_value(player, year, set_name, card_number):
    """Scrape PriceCharting with warmed-up session."""
    session = requests.Session()

    # Lean headers: full headers trigger Brotli responses without price tables
    session.headers.update({
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
    })

    # Warm up session by loading homepage first
    session.get('https://www.pricecharting.com/')
    time.sleep(1)

    # Search with full query
    query = f"{player} {year} {set_name} {card_number}"
    search_url = f"https://www.pricecharting.com/search-products?q={query}"
    response = session.get(search_url)

    if response.status_code != 200:
        return {"error": "blocked", "status_code": response.status_code}

    soup = BeautifulSoup(response.text, 'lxml')
    price_elem = soup.select_one('.price')

    if not price_elem:
        # Retry without set_name as fallback
        query = f"{player} {year} {card_number}"
        search_url = f"https://www.pricecharting.com/search-products?q={query}"
        response = session.get(search_url)
        soup = BeautifulSoup(response.text, 'lxml')
        price_elem = soup.select_one('.price')

    if price_elem:
        return {"price": price_elem.text.strip(), "source": "pricecharting"}

    return {"error": "no_results"}

The comment “full headers trigger Brotli responses” refers to a real production issue: PriceCharting returns compressed responses with missing HTML tables when the request includes Accept-Encoding: gzip, deflate, br. Using a lean User-Agent without explicit encoding headers forces the site to send uncompressed HTML that BeautifulSoup can parse reliably.

The 130point scraper similarly warms up the session:

def fetch_130point_value(player, year, set_name):
    session = requests.Session()
    session.headers.update({
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
        'Referer': 'https://www.130point.com/'
    })

    # Load homepage to collect cookies and avoid CAPTCHA
    session.get('https://www.130point.com/')

    search_url = 'https://www.130point.com/sales/'
    params = {'q': f"{player} {year} {set_name}"}
    response = session.get(search_url, params=params)

    # Parse and extract pricing logic...

This design mimics real user behavior: load the homepage, wait briefly, then search. The session preserves cookies across requests, convincing the server that the client is a repeat visitor rather than a bot.

Each pricing source is wrapped in its own try/except block inside fetch_valuations.py:

import json

def fetch_all_sources(card_data, output_dir):
    results = {}

    # eBay
    try:
        results['ebay'] = fetch_ebay_value(card_data)
    except Exception as e:
        results['ebay'] = {"error": str(e)}

    # PSA
    try:
        results['psa'] = fetch_psa_value(card_data)
    except Exception as e:
        results['psa'] = {"error": str(e)}

    # COMC
    try:
        results['comc'] = fetch_comc_value(card_data)
    except Exception as e:
        results['comc'] = {"error": str(e)}

    # Continue for other sources...

    # Write raw JSON per source
    for source, data in results.items():
        output_path = os.path.join(output_dir, f"{source}.json")
        with open(output_path, 'w') as f:
            json.dump(data, f, indent=2)

    return results

This error isolation ensures that if COMC returns a 403 or PriceCharting times out, the other five sources still complete. aggregate_valuations.py merges whatever data was successfully fetched, writing — (em-dash) for missing sources in the final inventory CSV.

The eBay implementation supports both API and scraping modes:

import os
from bs4 import BeautifulSoup

def fetch_ebay_value(card_data):
    ebay_token = os.getenv('EBAY_OAUTH_TOKEN')

    if ebay_token:
        # Use Marketplace Insights API (search completed sales)
        return fetch_ebay_api(card_data, ebay_token)
    else:
        # Scrape public sold listings
        return fetch_ebay_scrape(card_data)

def fetch_ebay_scrape(card_data):
    query = f"{card_data['player']} {card_data['year']} {card_data['set_name']}"
    search_url = f"https://www.ebay.com/sch/i.html?_nkw={query}&LH_Sold=1"

    response = requests.get(search_url, headers={'User-Agent': '...'})
    soup = BeautifulSoup(response.text, 'lxml')

    prices = []
    for item in soup.select('.s-item__price'):
        price_text = item.text.strip().replace('$', '').replace(',', '')
        try:
            prices.append(float(price_text))
        except ValueError:
            continue

    if prices:
        prices.sort()
        return {
            "low": prices[0],
            "mid": prices[len(prices) // 2],
            "high": prices[-1],
            "count": len(prices)
        }

    return {"error": "no_results"}

Both modes return the same result schema (low/mid/high/count), allowing aggregate_valuations.py to process eBay data without checking which fetch method was used.

Atomic File Writes and Error Isolation for Production Reliability

Data loss during inventory updates is unacceptable—corruption of the master CSV means running the recovery script to rebuild from per-card JSON files. The system prevents this through atomic file writes using temporary files and os.replace().

When aggregate_valuations.py updates the master inventory with new pricing data, it writes to a .tmp file first, then atomically replaces the original:

import os
import csv

def update_inventory_atomic(inventory_csv_path, updated_rows):
    """Write updated inventory with atomic replacement."""
    tmp_path = inventory_csv_path + '.tmp'

    # Write all rows to temporary file
    with open(tmp_path, 'w', newline='', encoding='utf-8') as f:
        fieldnames = updated_rows[0].keys()
        writer = csv.DictWriter(f, fieldnames=fieldnames)
        writer.writeheader()
        writer.writerows(updated_rows)

    # Atomic replace: succeeds completely or not at all
    os.replace(tmp_path, inventory_csv_path)

os.replace() is atomic on POSIX and Windows systems (as of Python 3.3), meaning the operation either completes fully or fails without modifying the destination file. If the script crashes after writing the temporary file but before replacing, the original inventory remains intact and the .tmp file can be inspected or discarded. This design pattern is critical for unattended automation where human intervention may be delayed by hours.

The valuation reports follow the same pattern:

def write_valuation_report_atomic(report_path, report_content):
    tmp_path = report_path + '.tmp'

    with open(tmp_path, 'w', encoding='utf-8') as f:
        f.write(report_content)

    os.replace(tmp_path, report_path)

Markdown reports include dash-formatted empty values using a helper function:

def dash(value):
    """Return em-dash for empty/None values."""
    return '—' if value in (None, '', 'N/A') else str(value)

def dollar(value):
    """Format currency with dollar sign."""
    if value in (None, '', 'N/A'):
        return '—'
    return f"${value}"

# Usage in report generation
row = f"| {card_id} | {dollar(ebay_mid)} | {dollar(psa_value)} | {dash(beckett_value)} |"

This formatting produces clean markdown tables where missing data is clearly distinguished from zero values:

| Card ID              | eBay Mid | PSA Value | Beckett |
|----------------------|----------|-----------|---------|
| mike_trout_2011_175  | $42.50   | $65.00    | —       |
| bryce_harper_2012_1  | $38.00   | —         | $45.00  |

Error isolation extends to the batch processing logic. prepare_valuation_batch.py sorts unvalued cards by scan_date ascending (oldest first) to ensure newly scanned cards wait in queue while older inventory gets priced first:

def prepare_valuation_batch(inventory_csv_path, batch_size=50):
    """Stage next batch of unvalued cards, sorted oldest first."""
    unvalued_cards = []

    with open(inventory_csv_path, 'r', encoding='utf-8') as f:
        reader = csv.DictReader(f)
        for row in reader:
            if not row.get('valuation_status') or row['valuation_status'] == 'pending':
                unvalued_cards.append(row)

    # Sort by scan_date ascending (oldest first)
    unvalued_cards.sort(key=lambda x: x.get('scan_date', ''))

    batch = unvalued_cards[:batch_size]

    with open('current_batch.json', 'w') as f:
        json.dump(batch, f, indent=2)

    return len(batch)

This prevents stale cards from accumulating at the bottom of the inventory—a problem that occurs when newest cards are always prioritized, leaving old inventory unpriced indefinitely.

Staging Valuation Batches and Integrating with n8n Orchestrators

The system is designed for integration with orchestration platforms like n8n through structured stdout tokens and integer-based conditional branching. Every script exits with code 0 and prints status messages that n8n can parse without relying on exit codes (which would halt workflows on expected empty states).

For example, scan_valuation_queue.py counts cards with empty or “pending” valuation_status and prints the count:

import csv

def scan_valuation_queue(inventory_csv_path):
    """Count cards needing valuation."""
    count = 0

    with open(inventory_csv_path, 'r', encoding='utf-8') as f:
        reader = csv.DictReader(f)
        for row in reader:
            status = row.get('valuation_status', '').strip()
            if not status or status == 'pending':
                count += 1

    print(f"VALUATION_QUEUE_COUNT:{count}")
    return 0  # Always exit 0

An n8n workflow can execute this script, capture stdout, and use a Switch node to branch:

IF stdout contains "VALUATION_QUEUE_COUNT:0"
  → Skip valuation (no work)
ELSE IF stdout contains "VALUATION_QUEUE_COUNT:[1-9]"
  → Execute prepare_valuation_batch.py
  → Execute fetch_valuations.py
  → Execute aggregate_valuations.py

This design allows the workflow to check for work, conditionally execute batch processing, and loop back to the queue scanner—all without custom exit codes that might conflict with orchestrator expectations.

check_md_config.py follows the same pattern for conditional markdown generation:

import json

def check_md_config(config_path='sports_cards_config.json'):
    """Output MD_ENABLED or MD_DISABLED based on config flag."""
    with open(config_path, 'r') as f:
        config = json.load(f)

    if config.get('generate_card_markdown', False):
        print("MD_ENABLED")
    else:
        print("MD_DISABLED")

    return 0

The n8n workflow parses stdout and conditionally executes generate_card_md.py only when markdown generation is enabled—avoiding unnecessary file I/O for users who only need CSV inventory without per-card documentation.

Batch size configuration lives in sports_cards_config.json:

{
  "batch_size": 50,
  "generate_card_markdown": true,
  "enabled_sources": ["ebay", "psa", "beckett", "comc", "130point", "pricecharting"],
  "output_base_path": "output/sports-cards/",
  "reports_path": "reports/"
}

This JSON file acts as the single source of configuration truth, allowing users to adjust batch sizes (higher for fast connections, lower to avoid rate limits) and disable specific pricing sources without editing Python code. The enabled_sources list filters which pricing functions are called in fetch_valuations.py.

Getting Started: Deploy Your Own Valuation Pipeline

To deploy this sports card inventory automation pipeline, clone the repository and install dependencies:

pip install pymupdf Pillow requests beautifulsoup4 lxml

Create sports_cards_config.json with your preferred batch size and output paths. Optionally, add EBAY_OAUTH_TOKEN to a .env file to enable eBay Marketplace Insights API mode (requires developer account).

Drop PDFs or images into incoming/sports-cards/. The pipeline will extract card data, check for duplicates, append to output/sports-cards/master_inventory.csv, and stage the next valuation batch. Run fetch_valuations.py to query all six pricing sources, then aggregate_valuations.py to merge data and generate_valuation_report.py to produce markdown summaries.

For hands-free automation, integrate with n8n using Execute Command nodes that capture stdout and branch on VALUATION_QUEUE_COUNT or MD_ENABLED tokens. Schedule the workflow to run nightly or weekly, and the system will maintain up-to-date pricing data for portfolio tracking, insurance documentation, or sale preparation—with zero manual CSV editing required.