Specification Extractor for Construction
Overview
Extract structured data from construction specification documents. Parse CSI MasterFormat sections, identify requirements, submittals, product standards, and compile actionable data for estimating and procurement.
Business Case
Automated spec extraction enables:
- Faster Estimating: Quickly identify scope and requirements
- Procurement Accuracy: Extract exact product specifications
- Submittal Tracking: Identify all required submittals
- Compliance Checking: Verify specs against standards
Technical Implementation
from dataclasses import dataclass, field
from typing import List, Dict, Any, Optional
import re
import pdfplumber
from pathlib import Path
@dataclass
class SpecSection:
number: str # e.g., "03 30 00"
title: str
part1_general: Dict[str, Any]
part2_products: Dict[str, Any]
part3_execution: Dict[str, Any]
raw_text: str
@dataclass
class ProductRequirement:
section: str
manufacturer: str
product_name: str
model: str
standards: List[str]
properties: Dict[str, str]
@dataclass
class SubmittalRequirement:
section: str
submittal_type: str # shop drawings, samples, product data, etc.
description: str
timing: str
copies: int
@dataclass
class SpecExtractionResult:
document_name: str
total_pages: int
sections: List[SpecSection]
products: List[ProductRequirement]
submittals: List[SubmittalRequirement]
standards_referenced: List[str]
class SpecificationExtractor:
"""Extract structured data from construction specifications."""
# CSI MasterFormat patterns
CSI_SECTION_PATTERN = r'^(\d{2}\s?\d{2}\s?\d{2})\s*[-–]\s*(.+?)$'
PART_PATTERN = r'^PART\s+(\d+)\s*[-–]\s*(.+?)$'
ARTICLE_PATTERN = r'^(\d+\.\d+)\s+([A-Z][A-Z\s]+)$'
# Submittal type keywords
SUBMITTAL_TYPES = {
'shop drawings': 'Shop Drawings',
'product data': 'Product Data',
'samples': 'Samples',
'certificates': 'Certificates',
'test reports': 'Test Reports',
'manufacturer instructions': 'Manufacturer Instructions',
'warranty': 'Warranty',
'maintenance data': 'Maintenance Data',
'mock-ups': 'Mock-ups',
}
# Common standard organizations
STANDARD_PATTERNS = [
r'ASTM\s+[A-Z]\d+',
r'ANSI\s+[A-Z]?\d+',
r'ACI\s+\d+',
r'AISC\s+\d+',
r'AWS\s+[A-Z]\d+',
r'ASCE\s+\d+',
r'UL\s+\d+',
r'FM\s+\d+',
r'NFPA\s+\d+',
r'IBC\s+\d+',
]
def __init__(self):
self.sections: Dict[str, SpecSection] = {}
def extract_from_pdf(self, pdf_path: str) -> SpecExtractionResult:
"""Extract specification data from PDF."""
path = Path(pdf_path)
all_text = ""
page_count = 0
with pdfplumber.open(pdf_path) as pdf:
page_count = len(pdf.pages)
for page in pdf.pages:
text = page.extract_text() or ""
all_text += text + "\n\n"
# Parse sections
sections = self._parse_sections(all_text)
# Extract products
products = self._extract_products(sections)
# Extract submittals
submittals = self._extract_submittals(sections)
# Extract standards
standards = self._extract_standards(all_text)
return SpecExtractionResult(
document_name=path.name,
total_pages=page_count,
sections=sections,
products=products,
submittals=submittals,
standards_referenced=standards
)
def _parse_sections(self, text: str) -> List[SpecSection]:
"""Parse CSI sections from specification text."""
sections = []
lines = text.split('\n')
current_section = None
current_part = None
current_content = []
for line in lines:
line = line.strip()
if not line:
continue
# Check for section header
section_match = re.match(self.CSI_SECTION_PATTERN, line, re.IGNORECASE)
if section_match:
# Save previous section
if current_section:
sections.append(self._finalize_section(current_section, current_content))
current_section = {
'number': section_match.group(1).replace(' ', ''),
'title': section_match.group(2).strip(),
'parts': {}
}
current_content = []
current_part = None
continue
# Check for part header
part_match = re.match(self.PART_PATTERN, line, re.IGNORECASE)
if part_match and current_section:
part_num = part_match.group(1)
part_name = part_match.group(2).strip()
current_part = f"part{part_num}"
current_section['parts'][current_part] = {
'name': part_name,
'content': []
}
continue
# Add content to current part
if current_section and current_part:
current_section['parts'][current_part]['content'].append(line)
elif current_section:
current_content.append(line)
# Save last section
if current_section:
sections.append(self._finalize_section(current_section, current_content))
return sections
def _finalize_section(self, section_data: Dict, general_content: List[str]) -> SpecSection:
"""Finalize a section with parsed parts."""
parts = section_data.get('parts', {})
part1 = self._parse_part_content(parts.get('part1', {}).get('content', []))
part2 = self._parse_part_content(parts.get('part2', {}).get('content', []))
part3 = self._parse_part_content(parts.get('part3', {}).get('content', []))
return SpecSection(
number=section_data['number'],
title=section_data['title'],
part1_general=part1,
part2_products=part2,
part3_execution=part3,
raw_text='\n'.join(general_content)
)
def _parse_part_content(self, content: List[str]) -> Dict[str, Any]:
"""Parse part content into structured data."""
result = {
'articles': {},
'items': []
}
current_article = None
for line in content:
# Check for article header
article_match = re.match(self.ARTICLE_PATTERN, line)
if article_match:
current_article = article_match.group(1)
result['articles'][current_article] = {
'title': article_match.group(2),
'items': []
}
continue
# Add to current article or general items
if current_article and current_article in result['articles']:
result['articles'][current_article]['items'].append(line)
else:
result['items'].append(line)
return result
def _extract_products(self, sections: List[SpecSection]) -> List[ProductRequirement]:
"""Extract product requirements from Part 2."""
products = []
for section in sections:
part2 = section.part2_products
for article_num, article in part2.get('articles', {}).items():
if 'MANUFACTURERS' in article['title'].upper():
for item in article['items']:
# Extract manufacturer names
if item.strip().startswith(('A.', 'B.', 'C.', '1.', '2.', '3.')):
mfr_name = re.sub(r'^[A-Z\d]+\.\s*', '', item).strip()
products.append(ProductRequirement(
s