句法分析：依存分析与成分分析

# 句法分析示例：同一个句子的两种视角
sentence = "猫 追 老鼠"

# 成分分析视角（短语结构）
#       S
#     /   \
#    NP    VP
#    |    /  \
#    猫   V    NP
#        |     |
#        追    老鼠
constituency_tree = {
    "type": "S",
    "children": [
        {"type": "NP", "word": "猫"},
        {"type": "VP", "children": [
            {"type": "V", "word": "追"},
            {"type": "NP", "word": "老鼠"}
        ]}
    ]
}

# 依存分析视角（依存关系）
#   追 <--root--
#  /  \
# 猫  老鼠
# ↑    ↑
# 主谓  动宾
dependency_tree = {
    "head": "追",
    "relations": [
        {"head": "追", "dep": "nsubj", "child": "猫"},
        {"head": "追", "dep": "dobj", "child": "老鼠"}
    ]
}

# 用 nltk 进行快速句法分析演示
import nltk
from nltk import CFG

# 定义一个简单的上下文无关文法
grammar = CFG.fromstring("""
  S -> NP VP
  NP -> Det N | Det N PP
  VP -> V NP | V NP PP
  PP -> P NP
  Det -> 'the' | 'a'
  N -> 'cat' | 'dog' | 'park' | 'ball'
  V -> 'chased' | 'saw'
  P -> 'in' | 'with'
""")

sentence = "the cat chased the dog".split()
print(f"句子: {' '.join(sentence)}")
print(f"词数: {len(sentence)}")

# 成分分析输出会显示所有可能的短语结构树
# 一个句子可能有多个合法的句法分析（歧义）

# CFG 的完整 Python 实现
from typing import List, Dict, Set, Tuple
from collections import defaultdict

class ContextFreeGrammar:
    """上下文无关文法的实现"""
    
    def __init__(self):
        self.productions: Dict[str, List[List[str]]] = defaultdict(list)
        self.start_symbol = "S"
    
    def add_rule(self, lhs: str, rhs: List[str]):
        """添加产生式规则：lhs → rhs"""
        self.productions[lhs].append(rhs)
    
    def get_nonterminals(self) -> Set[str]:
        """获取所有非终结符"""
        return set(self.productions.keys())
    
    def get_terminals(self) -> Set[str]:
        """获取所有终结符"""
        terminals = set()
        for rhs_list in self.productions.values():
            for rhs in rhs_list:
                for sym in rhs:
                    if sym not in self.get_nonterminals():
                        terminals.add(sym)
        return terminals
    
    def is_chomsky_normal(self) -> bool:
        """检查是否为乔姆斯基范式（CNF）"""
        nonterminals = self.get_nonterminals()
        for lhs, rhs_list in self.productions.items():
            for rhs in rhs_list:
                if len(rhs) == 1:
                    if rhs[0] not in nonterminals:
                        continue  # 允许终结符
                    return False
                elif len(rhs) == 2:
                    if not all(s in nonterminals for s in rhs):
                        return False
                else:
                    return False  # CNF 只允许 1 或 2 个符号
        return True

# 构建一个简单英语 CFG
grammar = ContextFreeGrammar()
grammar.add_rule("S", ["NP", "VP"])
grammar.add_rule("NP", ["Det", "N"])
grammar.add_rule("NP", ["Det", "N", "PP"])
grammar.add_rule("VP", ["V", "NP"])
grammar.add_rule("PP", ["P", "NP"])
print(f"CNF 检查: {grammar.is_chomsky_normal()}")

# 将 CFG 转换为乔姆斯基范式（CNF）
def to_cnf(grammar_rules: Dict[str, List[List[str]]]) -> Dict[str, List[List[str]]]:
    """将 CFG 转换为乔姆斯基范式
    CNF 要求每条规则要么是 A → BC（两个非终结符），要么是 A → a（一个终结符）
    """
    cnf = {}
    counter = [0]  # 用于生成新的非终结符名
    
    def new_symbol():
        counter[0] += 1
        return f"X{counter[0]}"
    
    # 第一步：处理长规则（>2个非终结符）
    for lhs, rhs_list in grammar_rules.items():
        cnf[lhs] = []
        for rhs in rhs_list:
            if len(rhs) > 2:
                # A → B C D E 变为：
                # A → B X1, X1 → C X2, X2 → D E
                current = rhs[0]
                for i in range(1, len(rhs) - 2):
                    new_sym = new_symbol()
                    cnf[lhs].append([current, new_sym])
                    lhs = new_sym
                    current = rhs[i]
                cnf[lhs].append([current, rhs[-2], rhs[-1]])
            else:
                cnf[lhs].append(rhs)
    
    return cnf

# 测试转换
rules = {"S": [["NP", "VP"]], "VP": [["V", "NP", "PP"]]}
cnf_rules = to_cnf(rules)
for lhs, rhs_list in cnf_rules.items():
    for rhs in rhs_list:
        print(f"{lhs} -> {' '.join(rhs)}")

# CKY 算法的完整实现
def cky_parse(words: list, grammar: dict, start: str = "S"):
    """CKY 解析算法
    
    Args:
        words: 输入句子（词列表）
        grammar: 文法规则 {A: [(B, C), ...]} 仅 CNF 形式
        start: 起始符号
    
    Returns:
        table: n×n 的解析表
        backpointers: 回溯指针（用于重建树）
    """
    n = len(words)
    # table[i][j] = 覆盖 words[i..j] 的所有非终结符
    table = [[set() for _ in range(n)] for _ in range(n)]
    # backpointers[i][j][A] = (B, C, k) 表示 A→BC 在 k 处分割
    backpointers = [[{} for _ in range(n)] for _ in range(n)]
    
    # 反转规则索引：(B, C) → [A]
    bin_rules = {}
    for A, rules in grammar.items():
        for B, C in rules:
            bin_rules.setdefault((B, C), []).append(A)
    
    # 步骤1: 填充对角线（单个词）
    for i, word in enumerate(words):
        # 假设 lexicon 将词映射到可能的词性
        lexicon = {"the": ["Det"], "cat": ["N"], "chased": ["V"], "dog": ["N"]}
        for tag in lexicon.get(word, []):
            table[i][i].add(tag)
    
    # 步骤2: 填充上层对角线
    for span in range(1, n):  # 跨度 1, 2, ..., n-1
        for i in range(n - span):
            j = i + span
            for k in range(i, j):
                for B in table[i][k]:
                    for C in table[k + 1][j]:
                        if (B, C) in bin_rules:
                            for A in bin_rules[(B, C)]:
                                table[i][j].add(A)
                                backpointers[i][j][A] = (B, C, k)
    
    return table, backpointers

# 测试
words = ["the", "cat", "chased", "the", "dog"]
grammar = {
    "S": [("NP", "VP")],
    "NP": [("Det", "N")],
    "VP": [("V", "NP")],
}
table, bp = cky_parse(words, grammar)
print(f"完整句子的可能分析: {table[0][4]}")

# 概率 CKY（PCFG 版本）- 选择最优分析树
import math

def probabilistic_cky(words, pcfg_rules, lexicon, start="S"):
    """概率 CKY 算法：为每个单元格保留最优概率"""
    n = len(words)
    # score[i][j][A] = 以 A 为根覆盖 words[i..j] 的最大概率
    score = [[{} for _ in range(n)] for _ in range(n)]
    bp = [[{} for _ in range(n)] for _ in range(n)]
    
    # 填充对角线：词 → 词性的概率
    for i, word in enumerate(words):
        for tag, prob in lexicon.get(word, {}).items():
            score[i][i][tag] = prob
    
    # 动态规划
    for span in range(1, n):
        for i in range(n - span):
            j = i + span
            for k in range(i, j):
                for (B, C), rules in pcfg_rules.items():
                    if B not in score[i][k] or C not in score[k+1][j]:
                        continue
                    for A, rule_prob in rules:
                        prob = rule_prob * score[i][k][B] * score[k+1][j][C]
                        if A not in score[i][j] or prob > score[i][j][A]:
                            score[i][j][A] = prob
                            bp[i][j][A] = (B, C, k)
    
    if start in score[0][n-1]:
        print(f"最优分析概率: {score[0][n-1][start]:.6f}")
    else:
        print("无法解析该句子")
    return score, bp

# PCFG 规则：(B, C) → [(A, 概率), ...]
pcfg = {
    ("Det", "N"): [("NP", 1.0)],
    ("V", "NP"): [("VP", 0.7), ("VP", 0.3)],
    ("NP", "VP"): [("S", 1.0)],
}
lex = {"the": {"Det": 1.0}, "cat": {"N": 1.0}, "chased": {"V": 1.0}, "dog": {"N": 1.0}}
probabilistic_cky(["the", "cat", "chased", "the", "dog"], pcfg, lex)

# 对比同一句子的两种分析结果
sentence = "The big cat quickly chased the small dog"

# === 成分分析（短语结构树）===
#              S
#       _______|_______
#      NP              VP
#   ___|___        ____|_____
#  Det   AdjP     Adv       VP
#   |     |       |     ___|___
#  The  big     quickly V     NP
#               |      |   ___|___
#             chased Det  Adj    N
#                   |     |      |
#                  the  small   dog
constituency_repr = {
    "S": {
        "NP": {"Det": "The", "AdjP": "big", "N": "cat"},
        "VP": {
            "Adv": "quickly",
            "VP": {
                "V": "chased",
                "NP": {"Det": "the", "AdjP": "small", "N": "dog"}
            }
        }
    }
}

# === 依存分析（依存关系）===
#    chased (ROOT)
#    /  |   \
#  cat  |   dog
#  / \     / \
#The big quickly the small
dependency_repr = [
    ("chased", "ROOT", "ROOT"),
    ("cat", "nsubj", "chased"),
    ("dog", "dobj", "chased"),
    ("The", "det", "cat"),
    ("big", "amod", "cat"),
    ("quickly", "advmod", "chased"),
    ("the", "det", "dog"),
    ("small", "amod", "dog"),
]
print("=== 依存关系 ===")
for child, rel, head in dependency_repr:
    print(f"  {child} --[{rel}]--> {head}")

# 使用 spaCy 对比两种分析
import spacy

nlp = spacy.load("en_core_web_sm")
doc = nlp("The cat chased the dog in the park")

print("=== 依存分析 ===")
for token in doc:
    print(f"  {token.text:10s} | "
          f"{token.pos_:8s} | "
          f"head={token.head.text:10s} | "
          f"dep={token.dep_}")

print(" === 成分分析 ===")
# spaCy 的成分分析需要额外模型
# 这里展示如何提取名词短语作为成分的近似
for np in doc.noun_chunks:
    print(f"  NP: {np.text} (root={np.root.text})")

print(" === 依存树可视化 ===")
# 构建邻接表
adj = {}
for token in doc:
    if token.head != token:
        adj.setdefault(token.head.i, []).append(token.i)
for head_idx, children in adj.items():
    print(f"  {doc[head_idx].text} → {[doc[c].text for c in children]}")

# Arc-Standard 转移系统实现
from typing import List, Tuple

class ArcStandardParser:
    """Arc-Standard 转移依存分析器"""
    
    def __init__(self):
        self.stack = ["ROOT"]  # 栈，初始包含 ROOT
        self.buffer = []       # 缓冲区
        self.arcs = []         # 依存边集合
    
    def shift(self):
        """SHIFT: buffer[0] → stack"""
        if self.buffer:
            self.stack.append(self.buffer.pop(0))
            return True
        return False
    
    def left_arc(self, label: str = "dep") -> bool:
        """LEFT-ARC: stack[-2] → stack[-1]（stack[-1] 是头）"""
        if len(self.stack) >= 2 and self.stack[-2] != "ROOT":
            self.arcs.append((self.stack[-1], label, self.stack[-2]))
            self.stack.pop(-2)  # 弹出次顶元素
            return True
        return False
    
    def right_arc(self, label: str = "dep") -> bool:
        """RIGHT-ARC: stack[-1] → stack[-2]（stack[-1] 是头）"""
        if len(self.stack) >= 2:
            self.arcs.append((self.stack[-2], label, self.stack[-1]))
            self.stack.pop()  # 弹出栈顶
            return True
        return False
    
    def parse(self, words: List[str], oracle: List[str]):
        """使用 oracle 指导的解析过程
        
        Args:
            words: 输入词序列
            oracle: 转移动作序列，如 ["S", "S", "LA", "RA"]
        """
        self.buffer = list(words)
        self.stack = ["ROOT"]
        self.arcs = []
        
        for action in oracle:
            if action == "S":
                self.shift()
            elif action == "LA":
                self.left_arc()
            elif action == "RA":
                self.right_arc()
        
        return self.arcs

# 示例："cat chased dog" 的 Arc-Standard 解析
parser = ArcStandardParser()
# Oracle: S(shift cat), S(shift chased), LA(cat←chased), S(shift dog), RA(chased→dog)
arcs = parser.parse(["cat", "chased", "dog"], ["S", "S", "LA", "S", "RA"])
for head, label, child in arcs:
    print(f"  {child} --[{label}]--> {head}")

# Arc-Eager 转移系统（更激进）
class ArcEagerParser:
    """Arc-Eager 转移依存分析器
    
    四个动作：
    - LEFT-ARC: 建立依存并弹出栈顶
    - RIGHT-ARC: 建立依存但保留栈顶
    - REDUCE: 弹出已处理完的栈顶
    - SHIFT: 压入新词
    """
    
    def __init__(self):
        self.stack = ["ROOT"]
        self.buffer = []
        self.arcs = []
    
    def left_arc(self, label: str = "dep") -> bool:
        """stack[-2] 作为 stack[-1] 的孩子"""
        if len(self.stack) >= 2 and self.stack[-2] != "ROOT":
            self.arcs.append((self.stack[-1], label, self.stack[-2]))
            self.stack.pop(-2)
            return True
        return False
    
    def right_arc(self, label: str = "dep") -> bool:
        """stack[-1] 作为 buffer[0] 的头（buffer[0] 不弹出）"""
        if len(self.stack) >= 2 and self.buffer:
            self.arcs.append((self.stack[-1], label, self.buffer[0]))
            return True
        return False
    
    def reduce(self) -> bool:
        """弹出已有头的栈顶词"""
        if len(self.stack) >= 2:
            has_head = any(child == self.stack[-1] for _, _, child in self.arcs)
            if has_head:
                self.stack.pop()
                return True
        return False
    
    def shift(self):
        """buffer[0] → stack"""
        if self.buffer:
            self.stack.append(self.buffer.pop(0))
            return True
        return False

# 对比 Arc-Standard 和 Arc-Eager 的差异
print("Arc-Standard 特点：")
print("  - 动作数少（3个），状态空间小")
print("  - RIGHT-ARC 后立即弹出，效率高")
print("  - 某些结构需要特殊处理")
print("")
print("Arc-Eager 特点：")
print("  - 动作数多（4个），更灵活")
print("  - RIGHT-ARC 保留栈顶，支持后续 LEFT-ARC")
print("  - 更适合神经网络学习（动作选择更局部）")

# Chu-Liu/Edmonds MST 算法简化实现
def mst_parse(scores, n):
    """Chu-Liu/Edmonds 最大生成树算法
    
    Args:
        scores: n×n 矩阵，scores[i][j] = 词 i 作为词 j 的头的分数
        n: 词数（含 ROOT，索引 0 是 ROOT）
    
    Returns:
        heads: 每个词的头节点索引
    """
    # 简化版本：贪心选择每个节点的最高分入边
    # 注意：完整实现需要处理环的收缩
    heads = [-1] * n
    in_weights = [-float('inf')] * n
    
    for j in range(1, n):  # 每个非根节点
        for i in range(n):
            if i != j and scores[i][j] > in_weights[j]:
                in_weights[j] = scores[i][j]
                heads[j] = i
    
    # 检查是否有环（简化版，实际需要环收缩）
    visited = set()
    has_cycle = False
    for node in range(1, n):
        path = []
        curr = node
        while curr != -1 and curr not in visited:
            path.append(curr)
            visited.add(curr)
            curr = heads[curr]
        if curr in path:
            has_cycle = True
            break
    
    total_score = sum(in_weights[1:])
    return heads, total_score, has_cycle

# 示例打分矩阵
scores = [
    [0.0, 0.1, 0.8, 0.3],  # ROOT 的头分数
    [0.9, 0.0, 0.2, 0.1],  # "cat" 的头分数
    [0.1, 0.7, 0.0, 0.6],  # "chased" 的头分数
    [0.2, 0.1, 0.8, 0.0],  # "dog" 的头分数
]
words = ["ROOT", "cat", "chased", "dog"]
heads, score, cycle = mst_parse(scores, 4)
print(f"最优生成树分数: {score}")
for i, h in enumerate(heads):
    if i > 0:
        print(f"  {words[i]} --head--> {words[h]}")

# Eisner 算法：投射性依存分析的动态规划
def eisner(scores, n):
    """Eisner 算法寻找最优投射依存树
    
    核心思想：用两种类型的 span 来递归分解
    - complete[i][j]: words[i..j] 形成完整子树，i 或 j 是根
    - incomplete[i][j]: words[i..j] 形成不完整子树
    
    状态: dp[k][i][j]
    k=0: incomplete, j 是根 (i ≤ j)
    k=1: incomplete, i 是根 (i ≤ j)
    k=2: complete, i 是根 (i ≤ j)
    k=3: complete, j 是根 (i ≤ j)
    """
    # 初始化
    INF = float('-inf')
    dp = [[[[INF] * 4 for _ in range(n)] for _ in range(n)] for _ in range(n)]
    
    # 基线情况
    for i in range(n):
        for k in range(4):
            dp[0][i][i][k] = 0.0
    
    # 动态规划
    for span in range(1, n):
        for i in range(n - span):
            j = i + span
            for r in range(i, j):
                # incomplete: 分解为两个 complete
                dp[0][i][j][0] = max(dp[0][i][j][0],
                    dp[2][i][r][2] + dp[3][r+1][j][3])
                dp[1][i][j][1] = max(dp[1][i][j][1],
                    dp[2][i][r][2] + dp[3][r+1][j][3])
            
            for r in range(i, j + 1):
                # complete: 添加边 i→j 或 j→i
                dp[2][i][j][2] = max(dp[2][i][j][2],
                    dp[1][i][j][1] + scores[i][j])
                dp[3][i][j][3] = max(dp[3][i][j][3],
                    dp[0][i][j][0] + scores[j][i])
    
    return dp[2][0][n-1][2]

# 使用 BiLSTM 编码生成打分矩阵
import numpy as np
np.random.seed(42)
n_words = 5  # ROOT + 4 words
arc_scores = np.random.randn(n_words, n_words)
result = eisner(arc_scores, n_words)
print(f"Eisner 最优分数: {result:.4f}")

# spaCy 依存分析实战
import spacy
from spacy import displacy

# 加载英文模型
nlp = spacy.load("en_core_web_trf")  # Transformer 版本

# 分析句子
doc = nlp("The curious cat cautiously chased the tiny mouse")

# 获取依存关系
print(f"{'Token':<12} {'POS':<8} {'Head':<12} {'Dep Relation'}")
print("-" * 50)
for token in doc:
    print(f"{token.text:<12} {token.pos_:<8} {token.head.text:<12} {token.dep_}")

# 提取主语-动词-宾语三元组
def extract_svo(doc):
    """从依存树中提取 SVO 三元组"""
    svos = []
    for token in doc:
        if token.pos_ == "VERB":
            subject = None
            obj = None
            for child in token.children:
                if child.dep_ in ("nsubj", "nsubjpass"):
                    subject = child.text
                elif child.dep_ in ("dobj", "obj", "pobj"):
                    obj = child.text
            if subject and obj:
                svos.append((subject, token.text, obj))
    return svos

svos = extract_svo(doc)
print(f" SVO 三元组: {svos}")
# 输出: [('cat', 'chased', 'mouse')]

# Stanza 依存分析实战
import stanza

# 下载并加载英文模型
stanza.download("en")
nlp = stanza.Pipeline("en", processors="tokenize,pos,lemma,depparse")

# 分析句子
doc = nlp("The curious cat cautiously chased the tiny mouse")

# 遍历依存关系
print("=== Stanza 依存分析 ===")
for sentence in doc.sentences:
    print(f"句子: {' '.join(w.text for w in sentence.words)}")
    print(f"{'ID':<4} {'Form':<12} {'UPOS':<6} {'Head':<6} {'DepRel'}")
    print("-" * 45)
    for word in sentence.words:
        head_text = sentence.words[word.head - 1].text if word.head > 0 else "ROOT"
        print(f"{word.id:<4} {word.text:<12} {word.upos:<6} {head_text:<6} {word.deprel}")

# 成分分析（Stanza 也支持）
nlp_const = stanza.Pipeline("en", processors="tokenize,pos,constituency")
doc_const = nlp_const("The cat chased the dog")
for sent in doc_const.sentences:
    print(f" 成分树: {sent.constituency}")

# 批量处理
sentences = [
    "The cat chased the dog.",
    "The dog ran away quickly.",
    "Both animals lived happily.",
]
docs = nlp(" ".join(sentences))
for i, sent in enumerate(docs.sentences):
    print(f" 句子 {i+1}: {[w.text for w in sent.words]}")