guanghulab/tools/notion-corpus-loader/src/hldp_parser.py

# HLDP解析器 · 识别光湖语言结构标记
# HLDP://tools/notion-corpus-loader/hldp-parser

import re
from dataclasses import dataclass, field
from typing import List, Optional

@dataclass
class HLDPBlock:
    """一个HLDP结构块"""
    path: str = ""
    block_type: str = ""
    content: str = ""
    cognitive_jump: Optional[str] = None
    causal_chain: Optional[str] = None
    persona: Optional[str] = None
    quality_tag: Optional[str] = None
    metadata: dict = field(default_factory=dict)

class HLDPParser:
    """解析HLDP格式的Markdown/HTML内容"""
    
    HLDP_PATH_PATTERN = re.compile(r'HLDP://([\w/-]+)')
    
    COGNITIVE_JUMP_PATTERNS = [
        re.compile(r'【认知[跃迁升]*[级点]*】'),
        re.compile(r'⚠️\s*([^·\n]{3,30})'),
        re.compile(r'核心[认知跃迁]*[：:]\s*(.+)'),
        re.compile(r'_why[：:]\s*(.+)'),
    ]
    
    CAUSAL_PATTERNS = [
        re.compile(r'(起点|推导链|终点)[：:]\s*(.+)'),
        re.compile(r'(.+?)\s*→\s*(.+?)\s*→\s*(.+)'),
    ]
    
    PERSONA_PATTERNS = [
        re.compile(r'主权者[：:]\s*冰朔.*TCS-0002'),
        re.compile(r'人格体[：:]\s*(铸渊|霜砚|晨星|舒舒|曜冥|知秋)'),
        re.compile(r'(冰朔|霜砚|铸渊|晨星|舒舒)\s*(口述|整理|创建|记录|签发)'),
    ]
    
    QUALITY_PATTERNS = [
        re.compile(r'CORPUS-ID[：:]\s*(\S+)'),
        re.compile(r'语料类型[：:]\s*(.+)'),
        re.compile(r'(极高|高|中|低)\s*(?:权重|质量)'),
    ]
    
    def parse_markdown(self, text: str) -> List[HLDPBlock]:
        """解析Markdown文本为HLDP结构块"""
        blocks = []
        lines = text.split('\n')
        i = 0
        while i < len(lines):
            line = lines[i].strip()
            if not line:
                i += 1
                continue
            block = HLDPBlock()
            heading_match = re.match(r'^(#{1,4})\s+(.+)', line)
            if heading_match:
                block.block_type = f'heading_{len(heading_match.group(1))}'
                block.content = heading_match.group(2)
                blocks.append(block)
                i += 1
                continue
            if line.startswith('```'):
                code_lines = []
                i += 1
                while i < len(lines) and not lines[i].strip().startswith('```'):
                    code_lines.append(lines[i])
                    i += 1
                block.block_type = 'code_block'
                block.content = '\n'.join(code_lines)
                blocks.append(block)
                i += 1
                continue
            if line.startswith('> '):
                block.block_type = 'quote'
                block.content = line[2:]
                blocks.append(block)
                i += 1
                continue
            for emoji in ['📢', '💡', '⚠️']:
                if line.startswith(emoji):
                    block.block_type = 'callout'
                    block.content = line
                    blocks.append(block)
                    i += 1
                    continue
            if '[子页面]' in line:
                block.block_type = 'child_page'
                block.content = line
                blocks.append(block)
                i += 1
                continue
            if re.match(r'^[-*]\s', line):
                block.block_type = 'list_item'
                block.content = line
                blocks.append(block)
                i += 1
                continue
            if line == '---':
                block.block_type = 'divider'
                blocks.append(block)
                i += 1
                continue
            block.block_type = 'paragraph'
            block.content = line
            blocks.append(block)
            i += 1
        for block in blocks:
            self._extract_hldp_annotations(block)
        return blocks
    
    def _extract_hldp_annotations(self, block: HLDPBlock):
        text = block.content
        path_match = self.HLDP_PATH_PATTERN.search(text)
        if path_match:
            block.path = path_match.group(0)
        for pattern in self.COGNITIVE_JUMP_PATTERNS:
            match = pattern.search(text)
            if match:
                block.cognitive_jump = match.group(1) if match.lastindex else match.group(0)
                break
        for pattern in self.CAUSAL_PATTERNS:
            match = pattern.search(text)
            if match:
                block.causal_chain = match.group(0)
                break
        for pattern in self.PERSONA_PATTERNS:
            match = pattern.search(text)
            if match:
                block.persona = match.group(0)
                break
        for pattern in self.QUALITY_PATTERNS:
            match = pattern.search(text)
            if match:
                block.quality_tag = match.group(0)
                break
    
    def parse_html(self, html: str) -> List[HLDPBlock]:
        text = html
        text = re.sub(r'<[^>]+>', '', text)
        text = re.sub(r'&nbsp;', ' ', text)
        text = re.sub(r'&lt;', '<', text)
        text = re.sub(r'&gt;', '>', text)
        text = re.sub(r'&amp;', '&', text)
        return self.parse_markdown(text)
    
    def build_cognitive_chain(self, blocks: List[HLDPBlock]) -> List[dict]:
        jumps = []
        for block in blocks:
            if block.cognitive_jump:
                jumps.append({'jump_point': block.cognitive_jump, 'context': block.content[:200], 'hldp_path': block.path, 'persona': block.persona})
        return jumps
    
    def build_causal_chains(self, blocks: List[HLDPBlock]) -> List[str]:
        chains = []
        for block in blocks:
            if block.causal_chain:
                chains.append(block.causal_chain)
        return chains