Spaces:

chihsing
/

DrugOps

Sleeping

DrugOps / engines /molgate_module2_engineB.py

Upload 180 files

af894e0 verified 4 months ago

13.1 kB

"""
MolGate — Module 2: Ligand Enumeration (Engine B)
Stage 04: Protonation state (Dimorphite-DL @ pH 7.4)
Stage 05: Tautomer (RDKit canonical)
依賴：
  pip install dimorphite-dl rdkit
"""

import argparse
import json
import sys
from datetime import datetime
from pathlib import Path

_PKG_ROOT = Path(__file__).resolve().parent
_UI_ROOT = _PKG_ROOT.parent
if str(_PKG_ROOT) not in sys.path:
    sys.path.insert(0, str(_PKG_ROOT))
if str(_UI_ROOT) not in sys.path:
    sys.path.insert(0, str(_UI_ROOT))

from molgate_ui.console_utf8 import configure_console_utf8, should_prompt_interactive

configure_console_utf8()

from molgate_manifest import (
    append_registry,
    finalize_module2_engine_b_manifest,
    try_load_manifest_or_rebuild,
    validate_target_matches_manifest,
)

from rdkit import Chem
from rdkit.Chem.MolStandardize import rdMolStandardize

# ── 嘗試載入 Dimorphite-DL ───────────────────────────
try:
    from dimorphite_dl import protonate_smiles
    DIMORPHITE_OK = True
except ImportError:
    DIMORPHITE_OK = False
    print("[WARN] dimorphite-dl 未安裝，Stage 04 將降級為原始 SMILES")

# Stage 04/05 產物版本（與 Module 1 的 stage_01 schema 分開）
SCHEMA_VERSION = "0.2"
# Module 1 Stage 01 目前可能為 0.2、0.3…
MODULE1_STAGE01_SCHEMAS = frozenset({"0.2", "0.3"})

# ── Master Index risk tier 查詢 ───────────────────────
# 實際使用時從 JSON 或 DB 讀取，這裡示範內嵌
LIGAND_RISK = {
    "COX2": {
        "stage04_tier": "A",
        "stage04_rule": "neutral_only",
        "stage04_note": "羧酸藥，pH 7.4 主要去質子化，單一形態足夠",
        "stage05_tier": "A",
        "stage05_note": "無高風險互變核心",
    },
    "D2R": {
        "stage04_tier": "C",
        "stage04_rule": "manual_select",
        "stage04_note": "CNS 胺基藥，pKa 接近 7.4，需手選質子化態",
        "stage05_tier": "A",
        "stage05_note": "無高風險互變核心",
    },
}

# ── 工具函式 ──────────────────────────────────────────
def save_stage(session_dir: Path, stage: int, name: str, data: dict) -> dict:
    data["_stage"] = stage
    data["_name"] = name
    data["_timestamp"] = datetime.now().isoformat()
    data["_schema_version"] = SCHEMA_VERSION
    path = session_dir / f"stage_{stage:02d}_{name}.json"
    path.write_text(json.dumps(data, indent=2, ensure_ascii=False), encoding="utf-8")
    return data

def load_stage(session_dir: Path, stage: int, name: str) -> dict:
    """讀取本引擎產生的 Stage（04/05），schema 必須與 SCHEMA_VERSION 一致。"""
    path = session_dir / f"stage_{stage:02d}_{name}.json"
    if not path.exists():
        raise FileNotFoundError(f"找不到 Stage {stage} 暫存：{path}")
    data = json.loads(path.read_text(encoding="utf-8"))
    if data.get("_schema_version") != SCHEMA_VERSION:
        raise ValueError(f"Schema version mismatch at stage {stage}，請重跑此關卡")
    return data


def load_stage01_from_module1(session_dir: Path) -> dict:
    """
    讀取 molgate_module1 產生的 stage_01_smiles.json。
    不與 Engine B 的 SCHEMA_VERSION 綁定（Module 1 獨立演進 0.2、0.3…）。
    """
    path = session_dir / "stage_01_smiles.json"
    if not path.exists():
        raise FileNotFoundError(f"找不到 Module 1 輸出：{path}")
    data = json.loads(path.read_text(encoding="utf-8"))
    ver = data.get("_schema_version")
    if ver is not None and ver not in MODULE1_STAGE01_SCHEMAS:
        print(
            f"[WARN] Stage 01 schema 為 {ver!r}，未在已知清單 {sorted(MODULE1_STAGE01_SCHEMAS)}，仍嘗試繼續"
        )
    if "canonical_smiles" not in data:
        raise ValueError("stage_01_smiles.json 缺少 canonical_smiles")
    return data

def prompt_select(
    candidates: list[str],
    label: str,
    *,
    non_interactive: bool = False,
) -> str:
    """人工選擇介面；subprocess / --non-interactive 時採用候選 [0]。"""
    if not candidates:
        raise ValueError(f"無候選可選：{label}")
    if not should_prompt_interactive(non_interactive=non_interactive):
        print(f"  [AUTO] 非互動：{label} → 候選 [0]")
        return candidates[0]
    print(f"\n  ── 需要人工選擇：{label} ──")
    for i, c in enumerate(candidates):
        print(f"  [{i}] {c}")
    while True:
        try:
            idx = int(input("  請輸入編號選擇："))
            if 0 <= idx < len(candidates):
                return candidates[idx]
        except (ValueError, KeyboardInterrupt):
            pass
        print("  請輸入有效編號")

# ── Stage 04：質子化狀態 ──────────────────────────────
def stage04_protonation(
    smiles: str,
    target: str,
    session_dir: Path,
    *,
    non_interactive: bool = False,
) -> dict:
    print("\n[Stage 04] 質子化狀態（pH 7.4）...")

    risk = LIGAND_RISK.get(target.upper(), {})
    tier = risk.get("stage04_tier", "B")
    rule = risk.get("stage04_rule", "dimorphite")
    note = risk.get("stage04_note", "")

    print(f"  → Tier: {tier} | Rule: {rule}")
    if note:
        print(f"  → Note: {note}")

    # Dimorphite-DL 產生候選
    candidates = []

    if DIMORPHITE_OK:
        try:
            from dimorphite_dl import protonate_smiles
            results = protonate_smiles(smiles, ph_min=7.35, ph_max=7.45)
            candidates = list(results) if results else [smiles]
        except Exception as e:
            print(f"  ⚠ Dimorphite-DL 執行錯誤：{e}，降級為原始 SMILES")
            candidates = [smiles]
    else:
        # 降級：直接用輸入 SMILES
        candidates = [smiles]

    print(f"  ✓ 候選數：{len(candidates)}")
    for c in candidates:
        print(f"    {c}")

    # 選擇邏輯
    if tier == "C" and should_prompt_interactive(non_interactive=non_interactive):
        selected = prompt_select(candidates, "質子化形態", non_interactive=non_interactive)
        status = "MANUAL"
    elif tier == "C":
        selected = candidates[0]
        status = "AUTO"
        print("  [AUTO] tier C 但非互動模式：採用候選 [0]")
    elif len(candidates) > 2 and should_prompt_interactive(non_interactive=non_interactive):
        print(f"  ⚠ 候選數 > 2，需手動選擇")
        selected = prompt_select(candidates, "質子化形態", non_interactive=non_interactive)
        status = "MANUAL"
        tier = "C_auto"
    elif len(candidates) > 2:
        selected = candidates[0]
        status = "AUTO"
        tier = "C_auto"
        print("  [AUTO] 候選數 > 2，非互動：採用候選 [0]")
    else:
        # 自動取第一個（主要形態）
        selected = candidates[0]
        status = "AUTO" if tier == "A" else "CONFIRM"

    print(f"  ✓ 選定：{selected}")
    print(f"  → Status: {status}")

    result = {
        "status": status,
        "tier": tier,
        "input_smiles": smiles,
        "candidates": candidates,
        "selected_smiles": selected,
        "ph": 7.4,
        "note": note,
    }
    return save_stage(session_dir, 4, "protonation", result)

# ── Stage 05：互變異構體 ──────────────────────────────
def stage05_tautomer(
    smiles: str,
    target: str,
    session_dir: Path,
    *,
    non_interactive: bool = False,
) -> dict:
    print("\n[Stage 05] 互變異構體（canonical tautomer）...")

    risk = LIGAND_RISK.get(target.upper(), {})
    tier = risk.get("stage05_tier", "A")
    note = risk.get("stage05_note", "")

    print(f"  → Tier: {tier}")
    if note:
        print(f"  → Note: {note}")

    mol = Chem.MolFromSmiles(smiles)
    if mol is None:
        result = {"status": "FAIL", "reason": "無法解析 Stage 04 輸出的 SMILES"}
        save_stage(session_dir, 5, "tautomer", result)
        raise ValueError(f"Stage 05 FAIL: {result['reason']}")

    # RDKit canonical tautomer
    enumerator = rdMolStandardize.TautomerEnumerator()
    canon = enumerator.Canonicalize(mol)
    canon_smiles = Chem.MolToSmiles(canon)

    # 列出所有候選
    tauts = enumerator.Enumerate(mol)
    all_tauts = [Chem.MolToSmiles(t) for t in tauts]

    print(f"  ✓ Canonical tautomer：{canon_smiles}")
    print(f"  ✓ 所有候選數：{len(all_tauts)}")

    # tier C 才停下來手選
    if tier == "C" and should_prompt_interactive(non_interactive=non_interactive):
        print(f"  ⚠ 高風險互變標記，需手動確認")
        for t in all_tauts:
            print(f"    {t}")
        selected = prompt_select(all_tauts, "互變異構體", non_interactive=non_interactive)
        status = "MANUAL"
    elif tier == "C":
        selected = canon_smiles
        status = "AUTO"
        print("  [AUTO] tier C 互變異構，非互動：採用 canonical")
    else:
        selected = canon_smiles
        status = "AUTO" if tier == "A" else "CONFIRM"

    print(f"  ✓ 選定：{selected}")
    print(f"  → Status: {status}")

    result = {
        "status": status,
        "tier": tier,
        "input_smiles": smiles,
        "canonical_tautomer": canon_smiles,
        "all_tautomers": all_tauts,
        "selected_smiles": selected,
        "tautomer_count": len(all_tauts),
        "note": note,
    }
    return save_stage(session_dir, 5, "tautomer", result)

# ── 主流程 ────────────────────────────────────────────
def run_engine_b(session_dir: Path, target: str, *, non_interactive: bool = False):
    print("=" * 55)
    print("  MolGate — Module 2 Engine B")
    print("  Stage 04: Protonation | Stage 05: Tautomer")
    print("=" * 55)

    try:
        try_load_manifest_or_rebuild(session_dir)
        validate_target_matches_manifest(session_dir, target)
    except ValueError as e:
        print(f"[ERROR] {e}")
        sys.exit(1)
    except FileNotFoundError as e:
        print(f"[ERROR] {e}")
        sys.exit(1)

    # 讀取 Stage 01 輸出（Module 1，schema 與本引擎分離）
    try:
        s1 = load_stage01_from_module1(session_dir)
    except FileNotFoundError:
        print("[ERROR] 找不到 Module 1 輸出，請先跑 molgate_module1.py")
        sys.exit(1)

    smiles = s1["canonical_smiles"]
    print(f"\n  輸入 SMILES（來自 Stage 01）：{smiles}")
    print(f"  Target：{target}")

    if s1.get("recommend_resolve_stereo_before_module2"):
        print(
            "\n  ⚠ WARN：Stage 01 標記「建議先處理立體再進 Module 2」"
            f"（未定義手性 {s1.get('n_undefined_chiral_centers', '?')} | "
            f"未定義 E/Z {s1.get('n_undefined_ez_bonds', '?')}）"
        )
        print("     仍可繼續 Engine B；結果之解讀風險自負。\n")

    try:
        r4 = stage04_protonation(
            smiles, target, session_dir, non_interactive=non_interactive
        )
        r5 = stage05_tautomer(
            r4["selected_smiles"], target, session_dir, non_interactive=non_interactive
        )
    except ValueError as e:
        print(f"\n[ERROR] Pipeline stopped: {e}")
        sys.exit(1)

    finalize_module2_engine_b_manifest(session_dir, r4, r5)
    append_registry(
        session_dir.parent,
        {
            "event": "module2_engine_b_complete",
            "session_id": session_dir.name,
            "session_dir": str(session_dir.resolve()),
            "target": target,
            "final_smiles": r5["selected_smiles"],
            "timestamp": datetime.now().isoformat(),
        },
    )

    print("\n" + "=" * 55)
    print("  Engine B 完成 [OK]")
    print(f"  最終 SMILES → Stage 09 使用：{r5['selected_smiles']}")
    print(f"  Session log：{session_dir}")
    print(f"  Manifest：{session_dir / 'molgate_run_manifest.json'}")
    print("=" * 55)

    return {
        "stage04": r4,
        "stage05": r5,
        "final_smiles": r5["selected_smiles"],
    }

# ── CLI ──────────────────────────────────────────────
if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="MolGate Module 2 Engine B")
    parser.add_argument("--session-dir", required=True, help="Module 1 的 session 目錄路徑")
    parser.add_argument("--target", required=True, help="Target 名稱（如 COX2）")
    parser.add_argument(
        "--non-interactive",
        action="store_true",
        help="不呼叫 input()（Web/API subprocess 必用）",
    )
    args = parser.parse_args()

    session_dir = Path(args.session_dir)
    if not session_dir.exists():
        print(f"[ERROR] Session 目錄不存在：{session_dir}")
        sys.exit(1)

    run_engine_b(session_dir, args.target, non_interactive=args.non_interactive)

	"""
	MolGate — Module 2: Ligand Enumeration (Engine B)
	Stage 04: Protonation state (Dimorphite-DL @ pH 7.4)
	Stage 05: Tautomer (RDKit canonical)

	依賴：
	pip install dimorphite-dl rdkit
	"""

	import argparse
	import json
	import sys
	from datetime import datetime
	from pathlib import Path

	_PKG_ROOT = Path(__file__).resolve().parent
	_UI_ROOT = _PKG_ROOT.parent
	if str(_PKG_ROOT) not in sys.path:
	sys.path.insert(0, str(_PKG_ROOT))
	if str(_UI_ROOT) not in sys.path:
	sys.path.insert(0, str(_UI_ROOT))

	from molgate_ui.console_utf8 import configure_console_utf8, should_prompt_interactive

	configure_console_utf8()

	from molgate_manifest import (
	append_registry,
	finalize_module2_engine_b_manifest,
	try_load_manifest_or_rebuild,
	validate_target_matches_manifest,
	)

	from rdkit import Chem
	from rdkit.Chem.MolStandardize import rdMolStandardize

	# ── 嘗試載入 Dimorphite-DL ───────────────────────────
	try:
	from dimorphite_dl import protonate_smiles
	DIMORPHITE_OK = True
	except ImportError:
	DIMORPHITE_OK = False
	print("[WARN] dimorphite-dl 未安裝，Stage 04 將降級為原始 SMILES")

	# Stage 04/05 產物版本（與 Module 1 的 stage_01 schema 分開）
	SCHEMA_VERSION = "0.2"
	# Module 1 Stage 01 目前可能為 0.2、0.3…
	MODULE1_STAGE01_SCHEMAS = frozenset({"0.2", "0.3"})

	# ── Master Index risk tier 查詢 ───────────────────────
	# 實際使用時從 JSON 或 DB 讀取，這裡示範內嵌
	LIGAND_RISK = {
	"COX2": {
	"stage04_tier": "A",
	"stage04_rule": "neutral_only",
	"stage04_note": "羧酸藥，pH 7.4 主要去質子化，單一形態足夠",
	"stage05_tier": "A",
	"stage05_note": "無高風險互變核心",
	},
	"D2R": {
	"stage04_tier": "C",
	"stage04_rule": "manual_select",
	"stage04_note": "CNS 胺基藥，pKa 接近 7.4，需手選質子化態",
	"stage05_tier": "A",
	"stage05_note": "無高風險互變核心",
	},
	}

	# ── 工具函式 ──────────────────────────────────────────
	def save_stage(session_dir: Path, stage: int, name: str, data: dict) -> dict:
	data["_stage"] = stage
	data["_name"] = name
	data["_timestamp"] = datetime.now().isoformat()
	data["_schema_version"] = SCHEMA_VERSION
	path = session_dir / f"stage_{stage:02d}_{name}.json"
	path.write_text(json.dumps(data, indent=2, ensure_ascii=False), encoding="utf-8")
	return data

	def load_stage(session_dir: Path, stage: int, name: str) -> dict:
	"""讀取本引擎產生的 Stage（04/05），schema 必須與 SCHEMA_VERSION 一致。"""
	path = session_dir / f"stage_{stage:02d}_{name}.json"
	if not path.exists():
	raise FileNotFoundError(f"找不到 Stage {stage} 暫存：{path}")
	data = json.loads(path.read_text(encoding="utf-8"))
	if data.get("_schema_version") != SCHEMA_VERSION:
	raise ValueError(f"Schema version mismatch at stage {stage}，請重跑此關卡")
	return data


	def load_stage01_from_module1(session_dir: Path) -> dict:
	"""
	讀取 molgate_module1 產生的 stage_01_smiles.json。
	不與 Engine B 的 SCHEMA_VERSION 綁定（Module 1 獨立演進 0.2、0.3…）。
	"""
	path = session_dir / "stage_01_smiles.json"
	if not path.exists():
	raise FileNotFoundError(f"找不到 Module 1 輸出：{path}")
	data = json.loads(path.read_text(encoding="utf-8"))
	ver = data.get("_schema_version")
	if ver is not None and ver not in MODULE1_STAGE01_SCHEMAS:
	print(
	f"[WARN] Stage 01 schema 為 {ver!r}，未在已知清單 {sorted(MODULE1_STAGE01_SCHEMAS)}，仍嘗試繼續"
	)
	if "canonical_smiles" not in data:
	raise ValueError("stage_01_smiles.json 缺少 canonical_smiles")
	return data

	def prompt_select(
	candidates: list[str],
	label: str,
	*,
	non_interactive: bool = False,
	) -> str:
	"""人工選擇介面；subprocess / --non-interactive 時採用候選 [0]。"""
	if not candidates:
	raise ValueError(f"無候選可選：{label}")
	if not should_prompt_interactive(non_interactive=non_interactive):
	print(f" [AUTO] 非互動：{label} → 候選 [0]")
	return candidates[0]
	print(f"\n ── 需要人工選擇：{label} ──")
	for i, c in enumerate(candidates):
	print(f" [{i}] {c}")
	while True:
	try:
	idx = int(input(" 請輸入編號選擇："))
	if 0 <= idx < len(candidates):
	return candidates[idx]
	except (ValueError, KeyboardInterrupt):
	pass
	print(" 請輸入有效編號")

	# ── Stage 04：質子化狀態 ──────────────────────────────
	def stage04_protonation(
	smiles: str,
	target: str,
	session_dir: Path,
	*,
	non_interactive: bool = False,
	) -> dict:
	print("\n[Stage 04] 質子化狀態（pH 7.4）...")

	risk = LIGAND_RISK.get(target.upper(), {})
	tier = risk.get("stage04_tier", "B")
	rule = risk.get("stage04_rule", "dimorphite")
	note = risk.get("stage04_note", "")

	print(f" → Tier: {tier} \| Rule: {rule}")
	if note:
	print(f" → Note: {note}")

	# Dimorphite-DL 產生候選
	candidates = []

	if DIMORPHITE_OK:
	try:
	from dimorphite_dl import protonate_smiles
	results = protonate_smiles(smiles, ph_min=7.35, ph_max=7.45)
	candidates = list(results) if results else [smiles]
	except Exception as e:
	print(f" ⚠ Dimorphite-DL 執行錯誤：{e}，降級為原始 SMILES")
	candidates = [smiles]
	else:
	# 降級：直接用輸入 SMILES
	candidates = [smiles]

	print(f" ✓ 候選數：{len(candidates)}")
	for c in candidates:
	print(f" {c}")

	# 選擇邏輯
	if tier == "C" and should_prompt_interactive(non_interactive=non_interactive):
	selected = prompt_select(candidates, "質子化形態", non_interactive=non_interactive)
	status = "MANUAL"
	elif tier == "C":
	selected = candidates[0]
	status = "AUTO"
	print(" [AUTO] tier C 但非互動模式：採用候選 [0]")
	elif len(candidates) > 2 and should_prompt_interactive(non_interactive=non_interactive):
	print(f" ⚠ 候選數 > 2，需手動選擇")
	selected = prompt_select(candidates, "質子化形態", non_interactive=non_interactive)
	status = "MANUAL"
	tier = "C_auto"
	elif len(candidates) > 2:
	selected = candidates[0]
	status = "AUTO"
	tier = "C_auto"
	print(" [AUTO] 候選數 > 2，非互動：採用候選 [0]")
	else:
	# 自動取第一個（主要形態）
	selected = candidates[0]
	status = "AUTO" if tier == "A" else "CONFIRM"

	print(f" ✓ 選定：{selected}")
	print(f" → Status: {status}")

	result = {
	"status": status,
	"tier": tier,
	"input_smiles": smiles,
	"candidates": candidates,
	"selected_smiles": selected,
	"ph": 7.4,
	"note": note,
	}
	return save_stage(session_dir, 4, "protonation", result)

	# ── Stage 05：互變異構體 ──────────────────────────────
	def stage05_tautomer(
	smiles: str,
	target: str,
	session_dir: Path,
	*,
	non_interactive: bool = False,
	) -> dict:
	print("\n[Stage 05] 互變異構體（canonical tautomer）...")

	risk = LIGAND_RISK.get(target.upper(), {})
	tier = risk.get("stage05_tier", "A")
	note = risk.get("stage05_note", "")

	print(f" → Tier: {tier}")
	if note:
	print(f" → Note: {note}")

	mol = Chem.MolFromSmiles(smiles)
	if mol is None:
	result = {"status": "FAIL", "reason": "無法解析 Stage 04 輸出的 SMILES"}
	save_stage(session_dir, 5, "tautomer", result)
	raise ValueError(f"Stage 05 FAIL: {result['reason']}")

	# RDKit canonical tautomer
	enumerator = rdMolStandardize.TautomerEnumerator()
	canon = enumerator.Canonicalize(mol)
	canon_smiles = Chem.MolToSmiles(canon)

	# 列出所有候選
	tauts = enumerator.Enumerate(mol)
	all_tauts = [Chem.MolToSmiles(t) for t in tauts]

	print(f" ✓ Canonical tautomer：{canon_smiles}")
	print(f" ✓ 所有候選數：{len(all_tauts)}")

	# tier C 才停下來手選
	if tier == "C" and should_prompt_interactive(non_interactive=non_interactive):
	print(f" ⚠ 高風險互變標記，需手動確認")
	for t in all_tauts:
	print(f" {t}")
	selected = prompt_select(all_tauts, "互變異構體", non_interactive=non_interactive)
	status = "MANUAL"
	elif tier == "C":
	selected = canon_smiles
	status = "AUTO"
	print(" [AUTO] tier C 互變異構，非互動：採用 canonical")
	else:
	selected = canon_smiles
	status = "AUTO" if tier == "A" else "CONFIRM"

	print(f" ✓ 選定：{selected}")
	print(f" → Status: {status}")

	result = {
	"status": status,
	"tier": tier,
	"input_smiles": smiles,
	"canonical_tautomer": canon_smiles,
	"all_tautomers": all_tauts,
	"selected_smiles": selected,
	"tautomer_count": len(all_tauts),
	"note": note,
	}
	return save_stage(session_dir, 5, "tautomer", result)

	# ── 主流程 ────────────────────────────────────────────
	def run_engine_b(session_dir: Path, target: str, *, non_interactive: bool = False):
	print("=" * 55)
	print(" MolGate — Module 2 Engine B")
	print(" Stage 04: Protonation \| Stage 05: Tautomer")
	print("=" * 55)

	try:
	try_load_manifest_or_rebuild(session_dir)
	validate_target_matches_manifest(session_dir, target)
	except ValueError as e:
	print(f"[ERROR] {e}")
	sys.exit(1)
	except FileNotFoundError as e:
	print(f"[ERROR] {e}")
	sys.exit(1)

	# 讀取 Stage 01 輸出（Module 1，schema 與本引擎分離）
	try:
	s1 = load_stage01_from_module1(session_dir)
	except FileNotFoundError:
	print("[ERROR] 找不到 Module 1 輸出，請先跑 molgate_module1.py")
	sys.exit(1)

	smiles = s1["canonical_smiles"]
	print(f"\n 輸入 SMILES（來自 Stage 01）：{smiles}")
	print(f" Target：{target}")

	if s1.get("recommend_resolve_stereo_before_module2"):
	print(
	"\n ⚠ WARN：Stage 01 標記「建議先處理立體再進 Module 2」"
	f"（未定義手性 {s1.get('n_undefined_chiral_centers', '?')} \| "
	f"未定義 E/Z {s1.get('n_undefined_ez_bonds', '?')}）"
	)
	print(" 仍可繼續 Engine B；結果之解讀風險自負。\n")

	try:
	r4 = stage04_protonation(
	smiles, target, session_dir, non_interactive=non_interactive
	)
	r5 = stage05_tautomer(
	r4["selected_smiles"], target, session_dir, non_interactive=non_interactive
	)
	except ValueError as e:
	print(f"\n[ERROR] Pipeline stopped: {e}")
	sys.exit(1)

	finalize_module2_engine_b_manifest(session_dir, r4, r5)
	append_registry(
	session_dir.parent,
	{
	"event": "module2_engine_b_complete",
	"session_id": session_dir.name,
	"session_dir": str(session_dir.resolve()),
	"target": target,
	"final_smiles": r5["selected_smiles"],
	"timestamp": datetime.now().isoformat(),
	},
	)

	print("\n" + "=" * 55)
	print(" Engine B 完成 [OK]")
	print(f" 最終 SMILES → Stage 09 使用：{r5['selected_smiles']}")
	print(f" Session log：{session_dir}")
	print(f" Manifest：{session_dir / 'molgate_run_manifest.json'}")
	print("=" * 55)

	return {
	"stage04": r4,
	"stage05": r5,
	"final_smiles": r5["selected_smiles"],
	}

	# ── CLI ──────────────────────────────────────────────
	if __name__ == "__main__":
	parser = argparse.ArgumentParser(description="MolGate Module 2 Engine B")
	parser.add_argument("--session-dir", required=True, help="Module 1 的 session 目錄路徑")
	parser.add_argument("--target", required=True, help="Target 名稱（如 COX2）")
	parser.add_argument(
	"--non-interactive",
	action="store_true",
	help="不呼叫 input()（Web/API subprocess 必用）",
	)
	args = parser.parse_args()

	session_dir = Path(args.session_dir)
	if not session_dir.exists():
	print(f"[ERROR] Session 目錄不存在：{session_dir}")
	sys.exit(1)

	run_engine_b(session_dir, args.target, non_interactive=args.non_interactive)