Specification Extractor Agent Skill

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(
                                section=section.number,
                                manufacturer=mfr_name,
                                product_name='',
                                model='',
                                standards=[],
                                properties={}
                            ))

                elif 'MATERIALS' in article['title'].upper() or 'PRODUCTS' in article['title'].upper():
                    for item in article['items']:
                        # Extract material requirements
                        standards = self._extract_standards(item)
                        if standards:
                            products.append(ProductRequirement(
                                section=section.number,
                                manufacturer='',
                                product_name=item[:100],
                                model='',
                                standards=standards,
                                properties={}
                            ))

        return products

    def _extract_submittals(self, sections: List[SpecSection]) -> List[SubmittalRequirement]:
        """Extract submittal requirements from Part 1."""
        submittals = []

        for section in sections:
            part1 = section.part1_general

            for article_num, article in part1.get('articles', {}).items():
                if 'SUBMITTAL' in article['title'].upper():
                    for item in article['items']:
                        item_lower = item.lower()

                        for keyword, submittal_type in self.SUBMITTAL_TYPES.items():
                            if keyword in item_lower:
                                submittals.append(SubmittalRequirement(
                                    section=section.number,
                                    submittal_type=submittal_type,
                                    description=item.strip(),
                                    timing='Prior to fabrication',
                                    copies=3
                                ))
                                break

        return submittals

    def _extract_standards(self, text: str) -> List[str]:
        """Extract referenced standards from text."""
        standards = []

        for pattern in self.STANDARD_PATTERNS:
            matches = re.findall(pattern, text, re.IGNORECASE)
            standards.extend(matches)

        return list(set(standards))

    def generate_submittal_log(self, result: SpecExtractionResult) -> str:
        """Generate submittal log from extraction results."""
        lines = ["# Submittal Log", ""]
        lines.append(f"**Project Specs:** {result.document_name}")
        lines.append(f"**Total Submittals:** {len(result.submittals)}")
        lines.append("")

        lines.append("| # | Section | Type | Description | Status |")
        lines.append("|---|---------|------|-------------|--------|")

        for i, sub in enumerate(result.submittals, 1):
            desc = sub.description[:50] + "..." if len(sub.description) > 50 else sub.description
            lines.append(f"| {i} | {sub.section} | {sub.submittal_type} | {desc} | Pending |")

        return "\n".join(lines)

    def generate_product_schedule(self, result: SpecExtractionResult) -> str:
        """Generate product schedule from extraction results."""
        lines = ["# Product Schedule", ""]

        # Group by section
        by_section = {}
        for prod in result.products:
            if prod.section not in by_section:
                by_section[prod.section] = []
            by_section[prod.section].append(prod)

        for section, products in sorted(by_section.items()):
            lines.append(f"## Section {section}")
            lines.append("")

            for prod in products:
                if prod.manufacturer:
                    lines.append(f"- **Manufacturer:** {prod.manufacturer}")
                if prod.product_name:
                    lines.append(f"- **Product:** {prod.product_name}")
                if prod.standards:
                    lines.append(f"- **Standards:** {', '.join(prod.standards)}")
                lines.append("")

        return "\n".join(lines)

    def generate_report(self, result: SpecExtractionResult) -> str:
        """Generate comprehensive extraction report."""
        lines = ["# Specification Extraction Report", ""]
        lines.append(f"**Document:** {result.document_name}")
        lines.append(f"**Pages:** {result.total_pages}")
        lines.append(f"**Sections Found:** {len(result.sections)}")
        lines.append("")

        # Sections summary
        lines.append("## Sections Extracted")
        for section in result.sections:
            lines.append(f"- **{section.number}** - {section.title}")
        lines.append("")

        # Standards
        if result.standards_referenced:
            lines.append("## Standards Referenced")
            for std in sorted(set(result.standards_referenced)):
                lines.append(f"- {std}")
            lines.append("")

        # Submittals summary
        lines.append("## Submittals Required")
        lines.append(f"Total: {len(result.submittals)}")
        by_type = {}
        for sub in result.submittals:
            by_type[sub.submittal_type] = by_type.get(sub.submittal_type, 0) + 1
        for t, count in sorted(by_type.items()):
            lines.append(f"- {t}: {count}")
        lines.append("")

        # Products summary
        lines.append("## Products/Manufacturers")
        lines.append(f"Total: {len(result.products)}")

        return "\n".join(lines)

# Initialize extractor
extractor = SpecificationExtractor()

# Extract from PDF
result = extractor.extract_from_pdf("Project_Specifications.pdf")

print(f"Found {len(result.sections)} sections")
print(f"Found {len(result.submittals)} submittals")
print(f"Found {len(result.products)} product requirements")

# Generate submittal log
submittal_log = extractor.generate_submittal_log(result)
print(submittal_log)

# Generate product schedule
product_schedule = extractor.generate_product_schedule(result)
print(product_schedule)

# Full report
report = extractor.generate_report(result)
print(report)

pip install pdfplumber