concept_extractor.py

"""
ACN v4.0 - Enhanced Concept Extractor
增强概念提取器：
- 中文分词 + TF-IDF
- 官能团识别（完整扩展）
- 化学元素检测
- 水解/皂化/酯化/裂解反应引擎
- 概念风格映射（官能团效应）
"""

import re
import json
import logging
import random
import copy
from pathlib import Path
from typing import List, Dict, Set, Tuple, Optional
from collections import Counter, defaultdict
from dataclasses import dataclass, field

import jieba
import jieba.analyse
import numpy as np

logger = logging.getLogger(__name__)

CUSTOM_DICT_PATH = "custom_chem_dict.txt"


@dataclass
class Concept:
    """概念数据类"""
    term: str
    score: float
    frequency: int
    concept_type: str = "general"
    source_doc: str = ""
    element_mapping: str = ""       # 对应的化学元素映射
    functional_group: str = ""      # 对应的官能团映射
    risk_level: float = 0.0         # 风险/激进程度 (0-1)
    innovation_index: float = 0.0   # 创新指数 (0-1)

    def to_dict(self) -> Dict:
        return {
            "term": self.term,
            "score": self.score,
            "frequency": self.frequency,
            "type": self.concept_type,
            "source_doc": self.source_doc,
            "element_mapping": self.element_mapping,
            "functional_group": self.functional_group,
            "risk_level": self.risk_level,
            "innovation_index": self.innovation_index
        }


@dataclass
class ReactionResult:
    """反应结果"""
    success: bool
    message: str
    original_concepts: List[Concept] = field(default_factory=list)
    result_concepts: List[Concept] = field(default_factory=list)
    reaction_type: str = ""
    details: Dict = field(default_factory=dict)
    energy_change: float = 0.0      # 反应能量变化


class FunctionalGroupDetector:
    """官能团检测器（完整扩展版）"""
    FUNCTIONAL_GROUPS = {
        "alkane": {"patterns": [r"烷烃", r"甲烷", r"乙烷", r"丙烷", r"丁烷", r"alkane", r"CnH2n\+2"], "cn": "烷烃", "priority": 1, "risk": 0.1, "innovation": 0.1},
        "alkene": {"patterns": [r"烯烃", r"乙烯", r"丙烯", r"alkene", r"C=C"], "cn": "烯烃", "priority": 2, "risk": 0.3, "innovation": 0.3},
        "alkyne": {"patterns": [r"炔烃", r"乙炔", r"alkyne", r"C≡C"], "cn": "炔烃", "priority": 2, "risk": 0.4, "innovation": 0.4},
        "aromatic": {"patterns": [r"苯", r"芳香", r"芳烃", r"benzene", r"aromatic", r"苯环"], "cn": "芳香烃", "priority": 3, "risk": 0.2, "innovation": 0.5},
        "hydroxyl": {"patterns": [r"羟基", r"-OH", r"hydroxyl", r"醇"], "cn": "羟基", "priority": 5, "risk": 0.2, "innovation": 0.3},
        "carbonyl": {"patterns": [r"羰基", r"C=O", r"carbonyl"], "cn": "羰基", "priority": 4, "risk": 0.3, "innovation": 0.4},
        "aldehyde": {"patterns": [r"醛", r"-CHO", r"aldehyde"], "cn": "醛基", "priority": 5, "risk": 0.4, "innovation": 0.4},
        "ketone": {"patterns": [r"酮", r"ketone"], "cn": "酮基", "priority": 5, "risk": 0.3, "innovation": 0.3},
        "carboxyl": {"patterns": [r"羧基", r"-COOH", r"carboxyl", r"羧酸"], "cn": "羧基", "priority": 6, "risk": 0.3, "innovation": 0.5},
        "ester": {"patterns": [r"酯", r"-COO-", r"ester"], "cn": "酯基", "priority": 5, "risk": 0.2, "innovation": 0.3},
        "ether": {"patterns": [r"醚", r"-O-", r"ether"], "cn": "醚基", "priority": 4, "risk": 0.2, "innovation": 0.2},
        "amine": {"patterns": [r"胺", r"-NH2", r"amine", r"氨基"], "cn": "氨基", "priority": 5, "risk": 0.3, "innovation": 0.4},
        "amide": {"patterns": [r"酰胺", r"-CONH", r"amide"], "cn": "酰胺", "priority": 5, "risk": 0.3, "innovation": 0.4},
        "nitrile": {"patterns": [r"腈", r"-CN", r"nitrile", r"氰基"], "cn": "腈基", "priority": 5, "risk": 0.5, "innovation": 0.5},
        "nitro": {"patterns": [r"硝基", r"-NO2", r"nitro"], "cn": "硝基", "priority": 5, "risk": 0.8, "innovation": 0.6},
        "halide": {"patterns": [r"卤", r"氯", r"溴", r"碘", r"氟", r"halide"], "cn": "卤素", "priority": 4, "risk": 0.5, "innovation": 0.4},
        "sulfhydryl": {"patterns": [r"巯基", r"-SH", r"thiol", r"硫醇"], "cn": "巯基", "priority": 5, "risk": 0.4, "innovation": 0.4},
        "sulfonic": {"patterns": [r"磺酸", r"-SO3H", r"sulfonic"], "cn": "磺酸基", "priority": 5, "risk": 0.4, "innovation": 0.5},
        "phosphate": {"patterns": [r"磷酸", r"-PO4", r"phosphate"], "cn": "磷酸基", "priority": 5, "risk": 0.3, "innovation": 0.6},
    }

    def detect(self, text: str) -> List[Tuple[str, str, int, float, float]]:
        """返回 (name, cn_name, priority, risk, innovation)"""
        detected = []
        for name, info in self.FUNCTIONAL_GROUPS.items():
            for pattern in info["patterns"]:
                if re.search(pattern, text, re.IGNORECASE):
                    detected.append((name, info["cn"], info["priority"], info["risk"], info["innovation"]))
                    break
        detected = list(set(detected))
        detected.sort(key=lambda x: x[2], reverse=True)
        return detected


class ChemicalElementDetector:
    """化学元素检测器（扩展版）"""
    ELEMENTS = {
        'H': ('氢', 0.1, 0.1), 'He': ('氦', 0.0, 0.0), 'Li': ('锂', 0.3, 0.5),
        'Be': ('铍', 0.2, 0.3), 'B': ('硼', 0.3, 0.4), 'C': ('碳', 0.2, 0.5),
        'N': ('氮', 0.3, 0.4), 'O': ('氧', 0.2, 0.3), 'F': ('氟', 0.6, 0.5),
        'Ne': ('氖', 0.0, 0.0), 'Na': ('钠', 0.3, 0.3), 'Mg': ('镁', 0.2, 0.3),
        'Al': ('铝', 0.2, 0.3), 'Si': ('硅', 0.2, 0.5), 'P': ('磷', 0.4, 0.5),
        'S': ('硫', 0.4, 0.4), 'Cl': ('氯', 0.5, 0.4), 'Ar': ('氩', 0.0, 0.0),
        'K': ('钾', 0.3, 0.3), 'Ca': ('钙', 0.2, 0.3), 'Fe': ('铁', 0.3, 0.4),
        'Cu': ('铜', 0.3, 0.4), 'Zn': ('锌', 0.3, 0.4), 'Br': ('溴', 0.5, 0.4),
        'I': ('碘', 0.4, 0.4), 'Au': ('金', 0.2, 0.6), 'Ag': ('银', 0.2, 0.5),
        'Pt': ('铂', 0.2, 0.7),
    }

    def detect(self, text: str) -> List[Tuple[str, str, float, float]]:
        """返回 (symbol, cn_name, risk, innovation)"""
        found = []
        for symbol, (name, risk, innov) in self.ELEMENTS.items():
            if symbol in text or name in text:
                found.append((symbol, name, risk, innov))
        return list(set(found))


class ReactionEngine:
    """化学反应引擎 v4（水解/皂化/酯化/裂解/共轭加权）"""

    def __init__(self):
        self.history = []

    def hydrolysis(self, concepts: List[Concept], intensity: float = 0.5) -> ReactionResult:
        """水解反应：松散化概念"""
        original = copy.deepcopy(concepts)
        result_concepts = []
        split_count = 0
        energy_released = 0

        for c in concepts:
            new_score = c.score * (1 - intensity * 0.6)
            new_freq = max(1, int(c.frequency * (1 - intensity * 0.4)))
            energy_released += (c.score - new_score)

            if len(c.term) > 4 and random.random() < intensity * 0.3:
                parts = list(jieba.cut(c.term))
                if len(parts) > 1:
                    for part in parts:
                        if len(part) >= 2:
                            result_concepts.append(Concept(
                                term=part, score=new_score / len(parts),
                                frequency=new_freq, concept_type=c.concept_type,
                                source_doc=c.source_doc
                            ))
                    split_count += 1
                    continue

            c.score = new_score
            c.frequency = new_freq
            result_concepts.append(c)

        result_concepts = self._deduplicate(result_concepts)
        details = {"split": split_count, "intensity": intensity, "energy_released": round(energy_released, 2)}
        self.history.append({"type": "hydrolysis", "details": details})

        return ReactionResult(
            success=True,
            message=f"✅ 水解反应完成（强度 {intensity:.0%}，释放能量 {energy_released:.1f}）",
            original_concepts=original, result_concepts=result_concepts,
            reaction_type="hydrolysis", details=details, energy_change=-energy_released
        )

    def saponification(self, concepts: List[Concept], intensity: float = 0.5) -> ReactionResult:
        """皂化反应：紧密结合概念"""
        original = copy.deepcopy(concepts)
        term_counter = Counter(c.term for c in concepts)
        merged = 0
        energy_absorbed = 0
        result_concepts = []
        seen = set()

        for c in concepts:
            if c.term in seen:
                continue
            new_score = c.score * (1 + intensity * 0.8)
            energy_absorbed += (new_score - c.score)

            if intensity > 0.6:
                for other in concepts:
                    if other.term != c.term and (c.term in other.term or other.term in c.term):
                        new_score += other.score * intensity * 0.3
                        seen.add(other.term)
                        merged += 1

            result_concepts.append(Concept(
                term=c.term, score=min(new_score, 100.0),
                frequency=c.frequency, concept_type=c.concept_type,
                source_doc=c.source_doc
            ))
            seen.add(c.term)

        result_concepts = sorted(result_concepts, key=lambda x: x.score, reverse=True)
        details = {"merged": merged, "intensity": intensity, "energy_absorbed": round(energy_absorbed, 2)}
        self.history.append({"type": "saponification", "details": details})

        return ReactionResult(
            success=True,
            message=f"✅ 皂化反应完成（强度 {intensity:.0%}，吸收能量 {energy_absorbed:.1f}）",
            original_concepts=original, result_concepts=result_concepts,
            reaction_type="saponification", details=details, energy_change=energy_absorbed
        )

    def cracking(self, concepts: List[Concept], intensity: float = 0.5) -> ReactionResult:
        """裂解反应：研究方向分裂"""
        original = copy.deepcopy(concepts)
        result_concepts = []
        cracked = 0

        for c in concepts:
            if c.score > 50 and random.random() < intensity * 0.5:
                parts = list(jieba.cut(c.term))
                if len(parts) >= 2:
                    for i, part in enumerate(parts):
                        if len(part) >= 2:
                            result_concepts.append(Concept(
                                term=part,
                                score=c.score * (0.4 + random.random() * 0.3),
                                frequency=max(1, c.frequency // 2),
                                concept_type=c.concept_type,
                                source_doc=c.source_doc,
                                risk_level=min(1.0, c.risk_level + 0.2)
                            ))
                    cracked += 1
                    continue
            result_concepts.append(c)

        result_concepts = self._deduplicate(result_concepts)
        details = {"cracked": cracked, "intensity": intensity}
        self.history.append({"type": "cracking", "details": details})

        return ReactionResult(
            success=True,
            message=f"✅ 裂解反应完成（裂解 {cracked} 个概念）",
            original_concepts=original, result_concepts=result_concepts,
            reaction_type="cracking", details=details
        )

    def conjugation_weighting(self, concepts: List[Concept],
                               boost_factor: float = 1.5) -> ReactionResult:
        """共轭加权：多概念叠加产生的非线性增强效应"""
        original = copy.deepcopy(concepts)
        term_freq = Counter(c.term for c in concepts)
        enhanced = 0

        for c in concepts:
            if term_freq[c.term] > 1:
                c.score *= boost_factor
                c.innovation_index = min(1.0, c.innovation_index + 0.1)
                enhanced += 1

        concepts = sorted(concepts, key=lambda x: x.score, reverse=True)
        details = {"enhanced": enhanced, "boost_factor": boost_factor}
        self.history.append({"type": "conjugation", "details": details})

        return ReactionResult(
            success=True,
            message=f"✅ 共轭加权完成（增强 {enhanced} 个概念）",
            original_concepts=original, result_concepts=concepts,
            reaction_type="conjugation", details=details
        )

    def _deduplicate(self, concepts: List[Concept]) -> List[Concept]:
        seen = {}
        for c in concepts:
            key = c.term.lower()
            if key not in seen or c.score > seen[key].score:
                seen[key] = c
        return sorted(seen.values(), key=lambda x: x.score, reverse=True)


class EnhancedConceptExtractor:
    """增强概念提取器 v4"""

    def __init__(self):
        self.functional_detector = FunctionalGroupDetector()
        self.element_detector = ChemicalElementDetector()
        self.reaction_engine = ReactionEngine()
        self.stopwords = set(list("的了和是在与或等为于以有无不也都而及被将从到对把让给"))

        if Path(CUSTOM_DICT_PATH).exists():
            jieba.load_userdict(CUSTOM_DICT_PATH)
            logger.info(f"已加载自定义词典: {CUSTOM_DICT_PATH}")

    def extract_concepts(self, text: str, top_n: int = 100, min_length: int = 2,
                         include_functional_groups: bool = True,
                         source_doc: str = "") -> List[Concept]:
        if not text.strip():
            return []

        concepts = []
        concepts.extend(self._extract_chinese_concepts(text, top_n, min_length, source_doc))

        if include_functional_groups:
            fg_detected = self.functional_detector.detect(text)
            for name, fg_name, priority, risk, innov in fg_detected[:20]:
                freq = text.count(fg_name)
                if freq > 0:
                    concepts.append(Concept(
                        term=fg_name, score=80.0 + freq * 5, frequency=freq,
                        concept_type="functional_group", source_doc=source_doc,
                        functional_group=name, risk_level=risk, innovation_index=innov
                    ))

        elements = self.element_detector.detect(text)
        for symbol, cn_name, risk, innov in elements:
            freq = text.count(cn_name)
            if freq > 0:
                concepts.append(Concept(
                    term=cn_name, score=70.0 + freq * 3, frequency=freq,
                    concept_type="element", source_doc=source_doc,
                    element_mapping=symbol, risk_level=risk, innovation_index=innov
                ))

        concepts = self._deduplicate(concepts, top_n)
        logger.info(f"提取概念完成: {len(concepts)} 个")
        return concepts

    def _extract_chinese_concepts(self, text: str, top_n: int, min_length: int,
                                   source_doc: str = "") -> List[Concept]:
        concepts = []
        try:
            keywords = jieba.analyse.extract_tags(
                text, topK=top_n, withWeight=True,
                allowPOS=('n', 'nz', 'v', 'vn', 'an', 'eng', 'ns', 'nt')
            )
            for word, weight in keywords:
                if len(word) < min_length or word in self.stopwords or word.isdigit():
                    continue
                freq = text.count(word)
                concepts.append(Concept(
                    term=word, score=float(weight * 100), frequency=freq,
                    concept_type="general", source_doc=source_doc
                ))
        except Exception as e:
            logger.warning(f"中文概念提取失败: {e}")
        return concepts

    def _deduplicate(self, concepts: List[Concept], top_n: int) -> List[Concept]:
        seen = {}
        for c in concepts:
            key = c.term.lower()
            if key not in seen or c.score > seen[key].score:
                seen[key] = c
        return sorted(seen.values(), key=lambda x: x.score, reverse=True)[:top_n]

    def hydrolysis(self, concepts, intensity=0.5):
        return self.reaction_engine.hydrolysis(concepts, intensity)

    def saponification(self, concepts, intensity=0.5):
        return self.reaction_engine.saponification(concepts, intensity)

    def cracking(self, concepts, intensity=0.5):
        return self.reaction_engine.cracking(concepts, intensity)

    def conjugation_weighting(self, concepts, boost=1.5):
        return self.reaction_engine.conjugation_weighting(concepts, boost)

    def get_reaction_history(self):
        return self.reaction_engine.history


def create_custom_dict(output_path: str = CUSTOM_DICT_PATH):
    """创建化学领域自定义词典"""
    chem_terms = [
        "苯环", "羟基", "羧基", "酯基", "醚基", "醛基", "酮基", "氨基", "酰胺",
        "硝基", "氰基", "巯基", "磺酸基", "烷烃", "烯烃", "炔烃", "芳香烃",
        "氧化物", "氢氧化物", "酸根", "盐酸", "硫酸", "硝酸", "磷酸",
        "取代反应", "加成反应", "消除反应", "氧化反应", "还原反应",
        "水解反应", "皂化反应", "酯化反应", "缩合反应", "聚合反应",
        "裂解反应", "共轭", "手性", "对映异构", "键能", "电负性",
        "势能面", "过渡态", "活化能", "零点能", "涌现", "相变",
        "电子云", "概率密度", "波函数", "量子跃迁", "光谱",
    ]
    with open(output_path, 'w', encoding='utf-8') as f:
        for term in chem_terms:
            f.write(f"{term} 10 n\n")
    logger.info(f"自定义词典已创建: {output_path}")
    return output_path