Processing Excel Metadata at Scale: Tools and Techniques

import os

import hashlib

import sqlite3

from pathlib import Path

from datetime import datetime

def discover_xlsx_files(root: str, db_path: str) -> int:

    """Walk directory tree and register XLSX files in SQLite."""

    conn = sqlite3.connect(db_path)

    conn.execute("""

        CREATE TABLE IF NOT EXISTS files (

            path TEXT PRIMARY KEY,

            size INTEGER,

            mtime TEXT,

            discovered_at TEXT,

            processed INTEGER DEFAULT 0

        )

    """)

    count = 0

    for dirpath, _, filenames in os.walk(root):

        for fname in filenames:

            if not fname.lower().endswith(".xlsx"):

                continue

            full = os.path.join(dirpath, fname)

            stat = os.stat(full)

            conn.execute(

                "INSERT OR IGNORE INTO files "

                "(path, size, mtime, discovered_at) "

                "VALUES (?, ?, ?, ?)",

                (full, stat.st_size,

                 datetime.fromtimestamp(stat.st_mtime).isoformat(),

                 datetime.now().isoformat()),

            )

            count += 1

    conn.commit()

    conn.close()

    return count

import zipfile

import xml.etree.ElementTree as ET

from dataclasses import dataclass, field

from typing import Optional

NS = {

    "cp": "http://schemas.openxmlformats.org/package/2006/metadata/core-properties",

    "dc": "http://purl.org/dc/elements/1.1/",

    "dcterms": "http://purl.org/dc/terms/",

    "ep": "http://schemas.openxmlformats.org/officeDocument/2006/extended-properties",

}

@dataclass

class FileMetadata:

    path: str

    creator: str = ""

    last_modified_by: str = ""

    created: str = ""

    modified: str = ""

    revision: str = ""

    application: str = ""

    company: str = ""

    sheet_count: int = 0

    hidden_sheets: list[str] = field(default_factory=list)

    error: Optional[str] = None

def extract_metadata(filepath: str) -> FileMetadata:

    """Extract metadata from a single XLSX file."""

    result = FileMetadata(path=filepath)

    try:

        with zipfile.ZipFile(filepath, "r") as zf:

            # Core properties

            if "docProps/core.xml" in zf.namelist():

                tree = ET.fromstring(zf.read("docProps/core.xml"))

                result.creator = _text(tree, "dc:creator")

                result.last_modified_by = _text(tree, "cp:lastModifiedBy")

                result.created = _text(tree, "dcterms:created")

                result.modified = _text(tree, "dcterms:modified")

                result.revision = _text(tree, "cp:revision")

            # App properties

            if "docProps/app.xml" in zf.namelist():

                tree = ET.fromstring(zf.read("docProps/app.xml"))

                result.application = _text(tree, "ep:Application")

                result.company = _text(tree, "ep:Company")

            # Sheet visibility

            if "xl/workbook.xml" in zf.namelist():

                wb = ET.fromstring(zf.read("xl/workbook.xml"))

                ns = "http://schemas.openxmlformats.org/spreadsheetml/2006/main"

                sheets = wb.findall(f"{{{ns}}}sheets/{{{ns}}}sheet")

                result.sheet_count = len(sheets)

                result.hidden_sheets = [

                    s.get("name", "")

                    for s in sheets

                    if s.get("state") in ("hidden", "veryHidden")

                ]

    except zipfile.BadZipFile:

        result.error = "Not a valid ZIP/XLSX file"

    except Exception as e:

        result.error = str(e)

    return result

def _text(tree: ET.Element, tag: str) -> str:

    el = tree.find(tag, NS)

    return (el.text or "") if el is not None else ""

import sqlite3

from concurrent.futures import ProcessPoolExecutor, as_completed

from extractor import extract_metadata, FileMetadata

def process_all(db_path: str, workers: int = 8) -> dict:

    """Process all unprocessed files in the database."""

    conn = sqlite3.connect(db_path)

    rows = conn.execute(

        "SELECT path FROM files WHERE processed = 0"

    ).fetchall()

    paths = [r[0] for r in rows]

    conn.close()

    stats = {"total": len(paths), "success": 0, "errors": 0}

    results: list[FileMetadata] = []

    with ProcessPoolExecutor(max_workers=workers) as pool:

        futures = {

            pool.submit(extract_metadata, p): p

            for p in paths

        }

        for future in as_completed(futures):

            meta = future.result()

            results.append(meta)

            if meta.error:

                stats["errors"] += 1

            else:

                stats["success"] += 1

    # Store results back in database

    store_results(db_path, results)

    return stats

def store_results(db_path: str, results: list) -> None:

    conn = sqlite3.connect(db_path)

    conn.execute("""

        CREATE TABLE IF NOT EXISTS metadata (

            path TEXT PRIMARY KEY,

            creator TEXT,

            last_modified_by TEXT,

            created TEXT,

            modified TEXT,

            revision TEXT,

            application TEXT,

            company TEXT,

            sheet_count INTEGER,

            hidden_sheets TEXT,

            error TEXT,

            extracted_at TEXT

        )

    """)

    for m in results:

        conn.execute(

            "INSERT OR REPLACE INTO metadata VALUES "

            "(?,?,?,?,?,?,?,?,?,?,?,?)",

            (m.path, m.creator, m.last_modified_by,

             m.created, m.modified, m.revision,

             m.application, m.company, m.sheet_count,

             ",".join(m.hidden_sheets),

             m.error, datetime.now().isoformat()),

        )

        # Mark file as processed in discovery table

        conn.execute(

            "UPDATE files SET processed = 1 WHERE path = ?",

            (m.path,),

        )

    conn.commit()

    conn.close()

import argparse

from discover import discover_xlsx_files

from parallel_processor import process_all

from reports import generate_report

def main():

    parser = argparse.ArgumentParser(

        description="Bulk Excel metadata processor"

    )

    sub = parser.add_subparsers(dest="command")

    # Discover command

    disc = sub.add_parser("discover",

        help="Scan directories for XLSX files")

    disc.add_argument("root",

        help="Root directory to scan")

    disc.add_argument("--db", default="metadata.db",

        help="SQLite database path")

    # Extract command

    ext = sub.add_parser("extract",

        help="Extract metadata from discovered files")

    ext.add_argument("--db", default="metadata.db")

    ext.add_argument("--workers", type=int, default=8)

    # Report command

    rep = sub.add_parser("report",

        help="Generate metadata analysis report")

    rep.add_argument("--db", default="metadata.db")

    rep.add_argument("--format",

        choices=["text", "csv", "json"],

        default="text")

    args = parser.parse_args()

    if args.command == "discover":

        count = discover_xlsx_files(args.root, args.db)

        print(f"Discovered {count} XLSX files")

    elif args.command == "extract":

        stats = process_all(args.db, args.workers)

        print(f"Processed: {stats['total']} files")

        print(f"Success: {stats['success']}")

        print(f"Errors: {stats['errors']}")

    elif args.command == "report":

        generate_report(args.db, args.format)

if __name__ == "__main__":

    main()

import sqlite3

import json

import csv

import sys

def generate_report(db_path: str, fmt: str) -> None:

    conn = sqlite3.connect(db_path)

    summary = {

        "total_files": _scalar(conn,

            "SELECT COUNT(*) FROM metadata"),

        "files_with_errors": _scalar(conn,

            "SELECT COUNT(*) FROM metadata "

            "WHERE error IS NOT NULL"),

        "files_with_author": _scalar(conn,

            "SELECT COUNT(*) FROM metadata "

            "WHERE creator != ''"),

        "files_with_hidden_sheets": _scalar(conn,

            "SELECT COUNT(*) FROM metadata "

            "WHERE hidden_sheets != ''"),

        "unique_authors": _scalar(conn,

            "SELECT COUNT(DISTINCT creator) "

            "FROM metadata WHERE creator != ''"),

        "unique_applications": _scalar(conn,

            "SELECT COUNT(DISTINCT application) "

            "FROM metadata WHERE application != ''"),

    }

    if fmt == "json":

        print(json.dumps(summary, indent=2))

    elif fmt == "csv":

        w = csv.writer(sys.stdout)

        w.writerow(summary.keys())

        w.writerow(summary.values())

    else:

        for k, v in summary.items():

            print(f"{k:<30} {v:>10}")

def _scalar(conn, query: str) -> int:

    return conn.execute(query).fetchone()[0]