Local SE Agent Masterclass

from pathlib import Path
from collections import defaultdict
from typing import Dict, Set, List, Any, Optional
import tree_sitter_java as tsjava
from tree_sitter import Language, Parser, Node

class CallHierarchyGraph:
    """
    เอนจินวิเคราะห์ลำดับชั้นการเรียกฟังก์ชันข้ามไฟล์ (Call Hierarchy Directed Graph)
    จัดเก็บความสัมพันธ์สองทิศทาง (Bidirectional Graph):
    - outgoing_calls (Callees): caller -> {callees}
    - incoming_calls (Callers): callee -> {callers}
    """
    def __init__(self):
        self.language = Language(tsjava.language())
        self.parser = Parser(self.language)
        # กราฟขาออก: ใครเรียกใครบ้าง
        self.outgoing_calls: Dict[str, Set[str]] = defaultdict(set)
        # กราฟขาเข้า: ใครถูกใครเรียกบ้าง (สำคัญมากสำหรับหา Blast Radius!)
        self.incoming_calls: Dict[str, Set[str]] = defaultdict(set)

    def add_call_edge(self, caller_symbol: str, callee_symbol: str) -> None:
        """บันทึกเส้นความสัมพันธ์แบบมีทิศทางระหว่าง Caller และ Callee"""
        self.outgoing_calls[caller_symbol].add(callee_symbol)
        self.incoming_calls[callee_symbol].add(caller_symbol)

    def parse_and_index_file(self, filepath: Path) -> int:
        """
        ใช้ Tree-sitter แกะหาโหนด method_declaration และ method_invocation
        เพื่อผูกความสัมพันธ์เข้าสู่กราฟ
        :return: จำนวน Call edges ที่สกัดได้จากไฟล์นี้
        """
        source_code = filepath.read_text(encoding="utf-8", errors="replace")
        source_bytes = bytes(source_code, "utf8")
        tree = self.parser.parse(source_bytes)
        root_node = tree.root_node

        def get_text(node: Optional[Node]) -> str:
            if not node:
                return ""
            return source_bytes[node.start_byte:node.end_byte].decode("utf8")

        class_name = filepath.stem # เช่น "OrderController"
        edges_found = 0

        # ท่องต้นไม้ค้นหาเมธอด และคำสั่งเรียกเมธอดภายใน
        def visit_node(node: Node, current_method: Optional[str] = None):
            nonlocal edges_found
            method_context = current_method

            # เมื่อพบการประกาศเมธอด
            if node.type == "method_declaration":
                name_node = node.child_by_field_name("name")
                if name_node:
                    m_name = get_text(name_node)
                    method_context = f"{class_name}.{m_name}"

            # เมื่อพบการเรียกใช้งานเมธอด (Method Invocation)
            elif node.type == "method_invocation" and method_context:
                name_node = node.child_by_field_name("name")
                if name_node:
                    callee_name = get_text(name_node)
                    # บันทึก edge: caller -> callee
                    self.add_call_edge(method_context, callee_name)
                    edges_found += 1

            for child in node.children:
                visit_node(child, method_context)

        visit_node(root_node)
        return edges_found

    def index_directory(self, root_dir: Path) -> int:
        """สแกนอ่านไฟล์ Java ทั้งหมดในโปรเจกต์และสร้างกราฟรวม"""
        total_edges = 0
        for java_file in root_dir.rglob("*.java"):
            total_edges += self.parse_and_index_file(java_file)
        return total_edges

    def get_blast_radius(self, target_method_name: str) -> Dict[str, Any]:
        """
        [Impact Analysis Hero]:
        คำนวณรัศมีผลกระทบทันทีเมื่อเมธอดเป้าหมายเกิดการแก้ไข
        :param target_method_name: ชื่อเมธอด เช่น "processPayment"
        :return: รายชื่อ Callers ที่จะพัง และ Callees ที่พึ่งพา
        """
        # ค้นหาว่ามีใครเรียก target_method_name บ้าง
        callers = sorted(list(self.incoming_calls.get(target_method_name, set())))
        
        # ค้นหา dependencies ขาออก
        callees = []
        for caller, targets in self.outgoing_calls.items():
            if caller.endswith(f".{target_method_name}"):
                callees.extend(list(targets))
        callees = sorted(list(set(callees)))

        return {
            "target": target_method_name,
            "impact_severity": "HIGH" if len(callers) >= 2 else "MODERATE" if len(callers) == 1 else "LOW",
            "blast_radius_count": len(callers),
            "affected_callers": callers,      # ใครจะพังบ้างถ้าเราแก้ตรงนี้!
            "dependencies_callees": callees  # ตัวนี้พึ่งพาใครบ้าง (ต้อง Mock ในเทส)
        }

"""
Intelligence Package: รวบรวมเอนจิน AST Parsing, Code Skeletonization และ Symbol Analysis
[อัปเดต Day 19]: เพิ่ม CallHierarchyGraph
"""
from intelligence.ast_parser import JavaASTParser
from intelligence.call_graph import CallHierarchyGraph

__all__ = ["JavaASTParser", "CallHierarchyGraph"]

import time
import shutil
from pathlib import Path
from intelligence.call_graph import CallHierarchyGraph

def setup_mock_project(base_dir: Path):
    """สร้างซอร์สโค้ด Java ข้าม 3 เลเยอร์ (Controller, Job, Service) จำลองสถานการณ์จริง"""
    base_dir.mkdir(parents=True, exist_ok=True)

    # 1. Controller Layer: สั่งชำระเงินเมื่อลูกค้ากดซื้อของ
    (base_dir / "OrderController.java").write_text("""
    package com.enterprise.controller;
    import com.enterprise.service.PaymentService;

    public class OrderController {
        private final PaymentService paymentService;

        public OrderController(PaymentService service) {
            this.paymentService = service;
        }

        public void checkoutOrder(Long orderId, Double totalAmount) {
            System.out.println("Processing order checkout...");
            // เรียกใช้ processPayment ข้ามไฟล์!
            paymentService.processPayment(orderId, totalAmount);
        }
    }
    """, encoding="utf-8")

    # 2. Batch Job Layer: สั่งตัดเงินอัตโนมัติรอบบิลรายเดือน
    (base_dir / "SubscriptionJob.java").write_text("""
    package com.enterprise.job;
    import com.enterprise.service.PaymentService;

    public class SubscriptionJob {
        private final PaymentService paymentService;

        public SubscriptionJob(PaymentService service) {
            this.paymentService = service;
        }

        public void executeMonthlyBilling(Long subId, Double fee) {
            // เรียกใช้ processPayment ข้ามไฟล์เช่นกัน!
            paymentService.processPayment(subId, fee);
        }
    }
    """, encoding="utf-8")

    # 3. Target Service Layer: คลาสเป้าหมายที่กำลังจะถูกแก้ไข
    (base_dir / "PaymentService.java").write_text("""
    package com.enterprise.service;

    public class PaymentService {
        public boolean processPayment(Long id, Double amount) {
            // เมธอดนี้พึ่งพา (Callees) สองคำสั่งด้านล่าง
            chargeGateway(amount);
            recordAuditLog(id);
            return true;
        }

        private void chargeGateway(Double amt) {
            // Gateway logic...
        }

        private void recordAuditLog(Long id) {
            // Audit DB logic...
        }
    }
    """, encoding="utf-8")

def main():
    print("=" * 70)
    print("🚀 เริ่มทดสอบ Day 19: Cross-File Architecture Analysis & Call Hierarchy")
    print("=" * 70)

    mock_dir = Path("./temp_call_hierarchy_test").resolve()
    setup_mock_project(mock_dir)
    print(f"📁 1. สร้างโปรเจกต์จำลอง 3 ไฟล์สำเร็จที่: {mock_dir.name}")

    try:
        # 2. เริ่มต้นระบบ CallHierarchyGraph และสแกนทั้งโฟลเดอร์
        call_graph = CallHierarchyGraph()
        start_time = time.perf_counter()
        total_edges = call_graph.index_directory(mock_dir)
        elapsed_ms = (time.perf_counter() - start_time) * 1000

        print(f"⚡ 2. สแกนและสร้างกราฟความสัมพันธ์สำเร็จในเวลา: {elapsed_ms:.2f} ms!")
        print(f"📊 ตรวจพบ Call Edges ทั้งหมดในระบบ: {total_edges} เส้นเชื่อม")

        # 3. ทดสอบคำนวณ Blast Radius เมื่อต้องการแก้ไขเมธอด 'processPayment'
        target_method = "processPayment"
        print(f"\n🎯 3. คำนวณ Blast Radius (รัศมีผลกระทบ) เมื่อจะแก้ไขเมธอด '{target_method}':")
        print("-" * 50)
        impact = call_graph.get_blast_radius(target_method)

        print(f"💥 ระดับความเสี่ยง (Severity): {impact['impact_severity']}")
        print(f"📈 จำนวนผู้ได้รับผลกระทบ: {impact['blast_radius_count']} จุด")
        print("🚨 รายชื่อ Callers ที่จะพังหากแก้ไข Signature นี้:")
        for caller in impact["affected_callers"]:
            print(f"   👉 {caller}")

        print("\n📦 รายชื่อ Dependencies ขาออก (Callees) ที่ต้องเขียน Mock:")
        for callee in impact["dependencies_callees"]:
            print(f"   🔗 {callee}()")
        print("-" * 50)

        # 4. Assertions ยืนยันความแม่นยำ 100%
        assert "OrderController.checkoutOrder" in impact["affected_callers"], "❌ ตรวจไม่พบ Caller จาก OrderController!"
        assert "SubscriptionJob.executeMonthlyBilling" in impact["affected_callers"], "❌ ตรวจไม่พบ Caller จาก SubscriptionJob!"
        assert "chargeGateway" in impact["dependencies_callees"], "❌ ตรวจไม่พบ Callee chargeGateway!"

        print("\n" + "=" * 70)
        print("🏆 ผ่านการทดสอบ Cross-File Call Hierarchy Graph สมบูรณ์แบบ 100%!")
        print("🚀 Agent ของคุณมองเห็นสถาปัตยกรรมทั้งระบบแล้ว พร้อมลุย Day 20: Hybrid Retrieval Engine!")
        print("=" * 70)

    finally:
        shutil.rmtree(mock_dir, ignore_errors=True)

if __name__ == "__main__":
    main()