建筑工程规范提取器

概述

从建筑工程规范文档中提取结构化数据。解析 CSI MasterFormat 章节，识别技术要求、报审文件（submittals）、产品标准，并汇总可用于成本估算与采购的可执行数据。

商业价值

自动化规范提取可实现：

• 更快的成本估算：快速识别工程范围与技术要求
• 更精准的采购执行：精确提取产品规格参数
• 报审文件追踪：识别全部必需提交的报审项
• 合规性核查：将规范条款与相关标准进行比对验证

技术实现

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  # 例如："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 等
    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:
    """从建筑工程规范中提取结构化数据。"""

    # CSI MasterFormat 匹配模式
    CSI_SECTION_PATTERN = r'^(d{2}s?d{2}s?d{2})s*[-–]s*(.+?)$'
    PART_PATTERN = r'^PARTs+(d+)s*[-–]s*(.+?)$'
    ARTICLE_PATTERN = r'^(d+.d+)s+([A-Z][A-Zs]+)$'

    # 报审类型关键词映射
    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',
    }

    # 常见标准制定机构正则模式
    STANDARD_PATTERNS = [
        r'ASTMs+[A-Z]d+',
        r'ANSIs+[A-Z]?d+',
        r'ACIs+d+',
        r'AISCs+d+',
        r'AWSs+[A-Z]d+',
        r'ASCEs+d+',
        r'ULs+d+',
        r'FMs+d+',
        r'NFPAs+d+',
        r'IBCs+d+',
    ]

    def __init__(self):
        self.sections: Dict[str, SpecSection] = {}

    def extract_from_pdf(self, pdf_path: str) -> SpecExtractionResult:
        """从 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 + "nn"

        # 解析章节
        sections = self._parse_sections(all_text)

        # 提取产品需求
        products = self._extract_products(sections)

        # 提取报审需求
        submittals = self._extract_submittals(sections)

        # 提取引用标准
        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]:
        """从规范文本中解析 CSI 章节。"""
        sections = []
        lines = text.split('n')

        current_section = None
        current_part = None
        current_content = []

        for line in lines:
            line = line.strip()
            if not line:
                continue

            # 检查是否为章节标题
            section_match = re.match(self.CSI_SECTION_PATTERN, line, re.IGNORECASE)
            if section_match:
                # 保存上一章节
                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

            # 检查是否为 Part 标题
            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

            # 将内容追加至当前 Part
            if current_section and current_part:
                current_section['parts'][current_part]['content'].append(line)
            elif current_section:
                current_content.append(line)

        # 保存最后一个章节
        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:
        """使用已解析的 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]:
        """将 Part 内容解析为结构化数据。"""
        result = {
            'articles': {},
            'items': []
        }

        current_article = None

        for line in content:
            # 检查是否为 Article 标题
            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

            # 追加至当前 Article 或通用 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]:
        """从 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']:
                        # 提取制造商名称
                        if item.strip().startswith(('A.', 'B.', 'C.', '1.', '2.', '3.')):
                            mfr_name = re.sub(r'^[A-Zd]+.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']:
                        # 提取材料要求
                        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]:
        """从 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]:
        """从文本中提取被引用的标准编号。"""
        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:
        """根据提取结果生成报审日志。"""
        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:
        """根据提取结果生成产品清单。"""
        lines = ["# Product Schedule", ""]

        # 按章节分组
        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:
        """生成完整的规范提取报告。"""
        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("")

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

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

        # 报审摘要
        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("")

        # 产品/制造商摘要
        lines.append("## Products/Manufacturers")
        lines.append(f"Total: {len(result.products)}")

        return "n".join(lines)

快速开始

# 初始化提取器
extractor = SpecificationExtractor()

# 从 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")

# 生成报审日志
submittal_log = extractor.generate_submittal_log(result)
print(submittal_log)

# 生成产品清单
product_schedule = extractor.generate_product_schedule(result)
print(product_schedule)

# 生成完整报告
report = extractor.generate_report(result)
print(report)

依赖项

pip install pdfplumber