pingip.py

import pandas as pd
import os
import platform
import subprocess
import re
import time
from datetime import datetime
from concurrent.futures import ThreadPoolExecutor
from tabulate import tabulate

class CheckNetworkTool:
    def __init__(self, file_path="list_ip.xlsx"):
        self.file_path = file_path
        self.current_results =[]
        self.os_type = platform.system().lower()

    def check_file_exists(self):
        """Kiểm tra xem file Excel có tồn tại hay không."""
        if not os.path.exists(self.file_path):
            print(f"\n LỖI NGHIÊM TRỌNG: Không tìm thấy file '{self.file_path}'!")
            print(f" Vui lòng đảm bảo file '{self.file_path}' đang nằm CÙNG THƯ MỤC với ứng dụng này.")
            input("Nhấn Enter để quay lại menu...")
            return False
        return True

    def clean_ip(self, raw_ip):
        """Lọc địa chỉ IPv4 từ chuỗi rác bằng Regex."""
        ip_pattern = r'(\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})'
        match = re.search(ip_pattern, str(raw_ip))
        return match.group(1) if match else str(raw_ip).strip()

    def ping_worker(self, item):
        """Hàm thực thi ping cho một thiết bị đơn lẻ."""
        stt, name, raw_ip = item
        target_ip = self.clean_ip(raw_ip)
        
        param = "-n" if self.os_type == "windows" else "-c"
        timeout_param = "-w" if self.os_type == "windows" else "-W"
        
        try:
            # Tăng timeout lên 2000ms (2 giây) để tránh rớt mạng ảo
            command =["ping", param, "1", timeout_param, "2000", target_ip]
            result = subprocess.run(command, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
            status = " OK" if result.returncode == 0 else " FAILED"
        except:
            status = " ERROR"
        
        return {"STT": stt, "Tên": name, "IP Gốc": raw_ip, "IP Lọc": target_ip, "Trạng thái": status}

    def run_parallel_ping(self, sheet_name):
        """Đọc sheet và thực hiện ping song song."""
        # KÍCH HOẠT HÀM KIỂM TRA FILE Ở ĐÂY
        if not self.check_file_exists():
            return False

        try:
            df = pd.read_excel(self.file_path, sheet_name=sheet_name, header=None)
            tasks =[]
            for _, row in df.iterrows():
                if pd.isna(row.iloc[2]) or "IP" in str(row.iloc[2]).upper(): continue
                tasks.append((row.iloc[0], row.iloc[1], row.iloc[2]))

            print(f"\n Đang quét {len(tasks)} thiết bị trong nhóm {sheet_name}...")
            
            # Giới hạn tối đa 50 luồng để bảo vệ Router/Switch không bị quá tải
            max_threads = min(50, len(tasks))
            with ThreadPoolExecutor(max_workers=max_threads) as executor:
                self.current_results = list(executor.map(self.ping_worker, tasks))
            
            self.display_table(self.current_results)
            return True
        except ValueError:
            print(f"\n LỖI: Không tìm thấy Sheet có tên là '{sheet_name}' trong file Excel.")
            input("Nhấn Enter để tiếp tục...")
            return False
        except Exception as e:
            print(f"\n Lỗi khi đọc file hoặc xử lý dữ liệu: {e}")
            input("Nhấn Enter để tiếp tục...")
            return False

    def run_continuous_ping(self):
        """Chức năng Ping liên tục và thống kê số lần rớt gói."""
        # KÍCH HOẠT HÀM KIỂM TRA FILE Ở ĐÂY
        if not self.check_file_exists():
            return

        try:
            df = pd.read_excel(self.file_path, sheet_name="Ping_Continuous", header=None)
            targets =[]
            for _, row in df.iterrows():
                if pd.isna(row.iloc[2]) or "IP" in str(row.iloc[2]).upper(): continue
                targets.append({
                    "stt": row.iloc[0],
                    "name": row.iloc[1],
                    "ip": self.clean_ip(row.iloc[2]),
                    "raw_ip": row.iloc[2],
                    "drops": 0,
                    "total": 0
                })

            if not targets:
                print("\n Không có thiết bị nào hợp lệ để ping trong sheet Ping_Continuous!")
                input("Nhấn Enter để quay lại...")
                return

            # Nhập thời gian chạy
            while True:
                time_input = input(" Nhập tổng thời gian chạy liên tục (giây): ").strip()
                if time_input.isdigit() and int(time_input) > 0:
                    duration = int(time_input)
                    break
                print(" Vui lòng chỉ nhập số nguyên lớn hơn 0!")

            # Cho phép chọn giãn cách thời gian giữa các đợt quét
            while True:
                interval_input = input(" Nhập khoảng cách giữa các đợt quét (giây) [Gợi ý: 2-5s]: ").strip()
                if interval_input.isdigit() and int(interval_input) >= 0:
                    interval = int(interval_input)
                    break
                print(" Vui lòng chỉ nhập số nguyên >= 0!")

            print(f"\n Đang thực hiện ping liên tục trong {duration}s. Giãn cách: {interval}s...")
            start_time = time.time()
            param = "-n" if self.os_type == "windows" else "-c"

            def do_ping(target_dict):
                cmd =["ping", param, "1", "-w", "2000", target_dict['ip']]
                res = subprocess.run(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
                return target_dict['ip'], res.returncode == 0

            max_threads = min(50, len(targets))
            with ThreadPoolExecutor(max_workers=max_threads) as executor:
                while True:
                    current_elapsed = time.time() - start_time
                    if current_elapsed >= duration:
                        break
                    
                    futures =[executor.submit(do_ping, t) for t in targets]
                    
                    for future in futures:
                        ip, is_success = future.result()
                        for t in targets:
                            if t['ip'] == ip:
                                t['total'] += 1
                                if not is_success:
                                    t['drops'] += 1
                                break
                    
                    time.sleep(interval)

            # Tổng kết kết quả
            summary =[]
            for t in targets:
                drop_rate = (t['drops'] / t['total'] * 100) if t['total'] > 0 else 0
                summary.append({
                    "STT": t['stt'], "Tên": t['name'], "IP": t['ip'], 
                    "Tổng Ping": t['total'], "Số lần rớt": f"🔥 {t['drops']}", 
                    "Tỉ lệ rớt": f"{drop_rate:.1f}%"
                })
            
            self.current_results = summary
            self.display_table(summary)
            
        except ValueError:
            print(f"\n LỖI: Không tìm thấy Sheet có tên là 'Ping_Continuous' trong file Excel.")
            input("Nhấn Enter để tiếp tục...")
        except Exception as e:
            print(f"\n Lỗi Ping_Continuous: {e}")
            input("Nhấn Enter để tiếp tục...")

    def display_table(self, data):
        """Hiển thị dữ liệu dạng bảng đẹp mắt."""
        if not data: return
        print("\n" + tabulate(data, headers="keys", tablefmt="grid"))

    def export_excel(self, category):
        """Xuất kết quả ra file Excel."""
        if not self.current_results:
            print(" Không có dữ liệu để xuất!")
            return
        
        now = datetime.now()
        timestamp = now.strftime("%Y%m%d_%H%M%S")
        filename = f"Check_System_Leviva_{timestamp}.xlsx"
        
        try:
            df_export = pd.DataFrame(self.current_results)
            df_export.to_excel(filename, index=False)
            print(f"\n Đã xuất báo cáo: {filename}")
        except Exception as e:
            print(f"\n Lỗi xuất file (Hãy đảm bảo file không bị mở ở nơi khác): {e}")

    def sub_menu(self, category, is_continuous=False):
        """Menu phụ."""
        while True:
            print(f"\n--- TÙY CHỌN [{category}] ---")
            print("1. Check lại (Quét lại danh sách này)")
            print("2. In ra file Excel (Xuất báo cáo)")
            print("3. Thoát (Về menu chính)")
            
            choice = input(" Chọn (1-3): ").strip()
            if choice == '1':
                if is_continuous: self.run_continuous_ping()
                else: self.run_parallel_ping(category)
            elif choice == '2':
                self.export_excel(category)
            elif choice == '3':
                break
            else:
                print(" Lựa chọn không hợp lệ!")

    def main_menu(self):
        """Menu chính."""
        while True:
            os.system('cls' if self.os_type == 'windows' else 'clear')
            print("=============================================")
            print("   CHECK NETWORK AUTOMATION SYSTEM           ")
            print("=============================================")
            print("1. Check Camera  (Sheet: Camera)")
            print("2. Check Wifi    (Sheet: Wifi)")
            print("3. Check Switch  (Sheet: Switch)")
            print("4. Check Dropped Count (Ping_Continuous)")
            print("5. Thoát chương trình")
            print("=============================================")
            
            choice = input(" Lựa chọn của bạn: ").strip()
            
            if choice == '1':
                if self.run_parallel_ping("Camera"): self.sub_menu("Camera")
            elif choice == '2':
                if self.run_parallel_ping("Wifi"): self.sub_menu("Wifi")
            elif choice == '3':
                if self.run_parallel_ping("Switch"): self.sub_menu("Switch")
            elif choice == '4':
                self.run_continuous_ping()
                # Chỉ hiện sub-menu nếu việc chạy ping diễn ra thành công (file tồn tại)
                if self.current_results: 
                    self.sub_menu("Ping_Continuous", is_continuous=True)
            elif choice == '5':
                print(" Cảm ơn bạn đã sử dụng hệ thống Check Network. Tạm biệt!")
                break
            else:
                print(" Lựa chọn không đúng, nhấn Enter để thử lại...")
                input()

if __name__ == "__main__":
    tool = CheckNetworkTool()
    tool.main_menu()