turns-00030.parquet:38398
6a244a3c3834cd0df7f4fd86degenerate_repetitionAbsentFinal dense release
USER
import sys, os, time, subprocess, threading, psutil, socket, requests, platform, base64, json, ipaddress
import csv, io, winsound, tempfile, ffmpeg, pyexpat, shutil, ast, operator, re, msvcrt, atexit
from queue import Queue, Empty
from PyQt5.QtWidgets import (QApplication, QWidget, QVBoxLayout, QLineEdit, QLabel,
QTabWidget, QPushButton, QMessageBox, QTableWidget, QTableWidgetItem,
QFormLayout, QCheckBox, QTextEdit, QComboBox, QInputDialog, QHBoxLayout,
QAbstractItemView, QHeaderView, QDialog, QTableView, QStatusBar,
QDateEdit, QFileDialog, QMenu, QAction, QStyledItemDelegate, QStackedWidget,
QColorDialog, QSizePolicy)
from PyQt5.QtCore import QTimer, Qt, QByteArray, QThread, pyqtSignal, QObject, QDate, QTime, QRect, QLocale
from PyQt5 import QtGui
from PyQt5.QtGui import QIcon, QPixmap, QClipboard, QColor, QBrush, QFont, QIntValidator, QPainter
from io import BytesIO
from datetime import datetime, timedelta
from filelock import FileLock, Timeout
from playsound import playsound
from threading import Thread, Lock
import pygame, fnmatch
import urllib.parse
import urllib.request
from collections import deque
from concurrent.futures import ThreadPoolExecutor, as_completed
ALERT_TIME = 60 # seconds
ALERT_REPEAT = 300 # seconds
ALERT_TIMEOUT = 5 # seconds
OFFLINE_DURATION = 60 # seconds
MAX_WORKERS = 10 # Maximum number of threads
MAX_PINGS = 50 # Maximum number of concurrent pings
GROUP_SIZE = 30 # Make groups of hostname
LOG_NORMAL = 60 # Write host retrun online after 60 seconds
FONT_SIZE = 14
english_locale = QLocale(QLocale.English, QLocale.UnitedStates)
class UpdateLogWidget(QWidget):
def __init__(self, parent=None):
super().__init__(parent)
layout = QVBoxLayout(self)
# Create a QTextEdit for displaying the update log
self.text_edit = QTextEdit(self)
self.text_edit.setReadOnly(True) # Make it read-only
self.text_edit.setPlainText(
"""Update Log
v49.6
- fixed กำหนดให้เปิดโปรแกรมได้ 1 instance
v49.5
- เล่นเสียงเมื่อ host offine หรือ online จำกัดจำนวน 10 คิว
- บันทึก history_online by date และ backup temp file
- support thai font, export csv
- backup offline_state.json ทุกครั้งที่มีการเปิดโปรแกรม
- กำหนดให้เปิดโปรแกรมได้ 1 instance
v49.4
- สามารถ input ได้เช่น 60*2 เพื่อคำนวณเวลาได้ง่าย
v49.3
- highligh host ที่ offline time มากกว่าที่กำหนดและเลือกสี
v49
- การอ่าน return_online log คอมพิวเตอร์ที่ตั้งค่ารูปแบบวันที่ที่ต่างกัน
- save offline state ป้องกันการ damage จากการเขียนไฟล์หรือโปรแกรมขัดข้องด้วยการสร้าง temp"""
)
# Adjust fonts if necessary
font = QFont()
font.setPointSize(11)
self.text_edit.setFont(font)
layout.addWidget(self.text_edit)
self.setLayout(layout)
class ExpressionEvaluator(ast.NodeVisitor):
allowed_nodes = (ast.Expression, ast.BinOp, ast.UnaryOp,
ast.Add, ast.Sub, ast.Mult, ast.Div, ast.FloorDiv,
ast.Pow, ast.USub, ast.UAdd, ast.Constant,)
def visit(self, node):
if not isinstance(node, self.allowed_nodes):
raise ValueError("Unsupported expression")
return super().visit(node)
def evaluate(self, expression):
node = ast.parse(expression, mode='eval')
return self.visit(node.body)
def visit_BinOp(self, node):
left = self.visit(node.left)
right = self.visit(node.right)
operator = node.op
if isinstance(operator, ast.Add):
return left + right
elif isinstance(operator, ast.Sub):
return left - right
elif isinstance(operator, ast.Mult):
return left * right
elif isinstance(operator, ast.Div):
return left / right
elif isinstance(operator, ast.FloorDiv):
return left // right
elif isinstance(operator, ast.Pow):
return left ** right
else:
raise ValueError("Unsupported operator")
def visit_UnaryOp(self, node):
operand = self.visit(node.operand)
operator = node.op
if isinstance(operator, ast.UAdd):
return +operand
elif isinstance(operator, ast.USub):
return -operand
else:
raise ValueError("Unsupported unary operator")
def visit_Num(self, node):
return node.n
def visit_Constant(self, node): # สำหรับ Python 3.8 ขึ้นไป
if isinstance(node.value, (int, float)):
return node.value
else:
raise ValueError("Constants must be numbers")
def safe_eval(expr):
evaluator = ExpressionEvaluator()
try:
value = evaluator.evaluate(expr.strip())
return int(value)
except Exception as e:
raise ValueError(f"Invalid expression: {expr}")
class SettingsWidget(QWidget):
update_exception_hosts_signal = pyqtSignal(set)
SETTINGS_FILE='settings.json'
def __init__(self, queue, offline_dict, stop_event, semaphore, host_info_list, tab_widget, main_window):
super().__init__()
self.tab_widget = tab_widget
self.queue = queue
self.offline_dict = offline_dict
self.stop_event = stop_event
self.semaphore = semaphore
self.host_info_list = host_info_list
self.main_window = main_window
self.group_size = GROUP_SIZE
self.pop_up_alert_enabled = False
self.sound_alert_enabled = True
self.remote_alert = ''
self.exception_hosts = set()
self.sound_alert_patterns = set()
self.font_size = FONT_SIZE
self.red_alert_time = 15 # Default red alert time in minutes
self.red_alert_color = QColor() # Default red alert color
self.initUI()
self.update_exception_hosts_signal.connect(self.update_exception_hosts_slot)
self.load_settings()
def initUI(self):
layout = QFormLayout()
self.alert_time_input = QLineEdit(self)
self.alert_time_input.setText(str(ALERT_TIME))
layout.addRow("Alert Time (seconds):", self.alert_time_input)
self.alert_repeat_input = QLineEdit(self)
self.alert_repeat_input.setText(str(ALERT_REPEAT))
layout.addRow("Alert Repeat (seconds):", self.alert_repeat_input)
self.alert_timeout_input = QLineEdit(self)
self.alert_timeout_input.setText(str(ALERT_TIMEOUT))
layout.addRow("Alert Timeout (seconds):", self.alert_timeout_input)
self.offline_duration_input = QLineEdit(self)
self.offline_duration_input.setText(str(OFFLINE_DURATION))
layout.addRow("Offline Duration (seconds):", self.offline_duration_input)
self.max_workers_input = QLineEdit(self)
self.max_workers_input.setText(str(MAX_WORKERS))
layout.addRow("Max Workers:", self.max_workers_input)
self.max_pings_input = QLineEdit(self)
self.max_pings_input.setText(str(MAX_PINGS))
layout.addRow("Max Pings:", self.max_pings_input)
self.group_size_input = QLineEdit(self)
self.group_size_input.setText(str(GROUP_SIZE))
layout.addRow("Group Size:", self.group_size_input)
self.remote_alert_input = QLineEdit(self)
layout.addRow("Remote Alert:", self.remote_alert_input)
self.log_normal_input = QLineEdit(self)
self.log_normal_input.setText(str(LOG_NORMAL))
layout.addRow("Log normal if over seconds:", self.log_normal_input)
self.font_size_input = QLineEdit(self)
self.font_size_input.setText(str(self.font_size))
layout.addRow("Font Size:", self.font_size_input)
# Create a label for Red Alert Time
red_alert_time_label = QLabel("Red Alert Time (minutes):", self)
# Create a horizontal layout for Red Alert Time Input and Color Picker
red_alert_layout = QHBoxLayout()
# Add the Red Alert Time Input
self.red_alert_time_input = QLineEdit(self)
self.red_alert_time_input.setText(str(self.red_alert_time)) # default value
# Make the input expand to fill space
self.red_alert_time_input.setSizePolicy(QSizePolicy.Expanding, QSizePolicy.Fixed)
red_alert_layout.addWidget(self.red_alert_time_input)
# Add the Red Alert Color Picker Button
self.red_alert_color_button = QPushButton("", self)
self.red_alert_color_button.setFixedSize(25, 25) # Make it a square
# Set the size policy to fixed to prevent it from expanding
self.red_alert_color_button.setSizePolicy(QSizePolicy.Fixed, QSizePolicy.Fixed)
# Adjust the height to match the line edit
self.red_alert_color_button.setFixedHeight(self.red_alert_time_input.sizeHint().height())
self.red_alert_color_button.clicked.connect(self.choose_red_alert_color)
red_alert_layout.addWidget(self.red_alert_color_button)
# Ensure that the layout spacing is consistent
red_alert_layout.setContentsMargins(0, 0, 0, 0)
red_alert_layout.setSpacing(5)
# Create a QWidget to hold the horizontal layout
red_alert_widget = QWidget()
red_alert_widget.setLayout(red_alert_layout)
# Now, add the row to the form layout
layout.addRow(red_alert_time_label, red_alert_widget)
self.exception_hosts_text_edit = QTextEdit(self)
self.exception_hosts_text_edit.setPlaceholderText("Hostnames separated by commas")
self.exception_hosts_text_edit.textChanged.connect(self.update_exception_hosts)
layout.addRow("Exception Alert Hosts:", self.exception_hosts_text_edit)
self.sound_alert_hosts_text_edit = QTextEdit(self)
self.sound_alert_hosts_text_edit.setPlaceholderText("Host patterns (wildcards allowed) separated by commas")
self.sound_alert_hosts_text_edit.textChanged.connect(self.update_sound_alert_patterns)
layout.addRow("Sound Alert Hosts:", self.sound_alert_hosts_text_edit)
self.pop_up_alert_checkbox = QCheckBox("Pop-up Alert", self)
self.pop_up_alert_checkbox.setChecked(self.pop_up_alert_enabled)
self.sound_alert_checkbox = QCheckBox("Sound Alert", self)
self.sound_alert_checkbox.setChecked(self.sound_alert_enabled)
checkboxes_widget = QWidget(self)
checkboxes_layout = QHBoxLayout(checkboxes_widget)
checkboxes_layout.addWidget(self.pop_up_alert_checkbox)
checkboxes_layout.addWidget(self.sound_alert_checkbox)
checkboxes_widget.setLayout(checkboxes_layout)
layout.addRow("", checkboxes_widget)
self.apply_button = QPushButton("Apply", self)
self.apply_button.clicked.connect(self.apply_settings)
layout.addRow("", self.apply_button)
self.setLayout(layout)
def choose_red_alert_color(self):
color = QColorDialog.getColor(self.red_alert_color, self, "Select Red Alert Color")
if color.isValid():
self.red_alert_color = color
self.red_alert_color_button.setStyleSheet(f"background-color: {color.name()};")
else:
# User canceled the color selection; set to no color
self.red_alert_color = QColor()
self.red_alert_color_button.setStyleSheet("")
def apply_settings(self):
self.update_global_settings()
self.apply_font_size()
# Update the red_alert_time and red_alert_color in OfflineHostMonitor
for i in range(self.tab_widget.count()):
widget = self.tab_widget.widget(i)
if isinstance(widget, OfflineHostMonitor):
widget.set_red_alert_settings(self.red_alert_time, self.red_alert_color)
#self.start_monitoring()
#self.restart_monitoring()
self.show_settings_applied()
def update_exception_hosts_slot(self, exception_hosts):
self.exception_hosts = exception_hosts
self.exception_hosts_text_edit.setPlainText(', '.join(exception_hosts))
def update_sound_alert_patterns(self):
text = self.sound_alert_hosts_text_edit.toPlainText()
pattern_list = [pattern.strip() for pattern in text.split(',') if pattern.strip()]
self.sound_alert_patterns = set(pattern_list)
self.save_settings()
print("Updated sound alert patterns:", self.sound_alert_patterns)
def load_settings(self):
if os.path.exists(self.SETTINGS_FILE):
with open(self.SETTINGS_FILE, 'r') as f:
settings = json.load(f)
sound_alert_patterns = settings.get('sound_alert_patterns', '')
self.sound_alert_hosts_text_edit.setPlainText(sound_alert_patterns)
pattern_list = [pattern.strip() for pattern in sound_alert_patterns.split(',') if pattern.strip()]
self.sound_alert_patterns = set(pattern_list)
self.red_alert_time_input.setText(str(settings.get('red_alert_time', '15')))
color_name = settings.get('red_alert_color', '')
if color_name:
self.red_alert_color = QColor(color_name)
if self.red_alert_color.isValid():
self.red_alert_color_button.setStyleSheet(f"background-color: {self.red_alert_color.name()};")
else:
self.red_alert_color = QColor()
self.red_alert_color_button.setStyleSheet("")
else:
self.red_alert_color = QColor()
self.red_alert_color_button.setStyleSheet("")
else:
print("Settings file not found. Using default settings.")
def save_settings(self):
settings = {}
sound_alert_patterns = self.sound_alert_hosts_text_edit.toPlainText()
settings['sound_alert_patterns'] = sound_alert_patterns
settings['red_alert_time'] = self.red_alert_time_input.text()
settings['red_alert_color'] = self.red_alert_color.name() if self.red_alert_color.isValid() else ''
with open(self.SETTINGS_FILE, 'w') as f:
json.dump(settings, f, indent=4)
def closeEvent(self, event):
self.save_settings()
event.accept()
def update_global_settings(self):
global ALERT_TIME, ALERT_REPEAT, ALERT_TIMEOUT, OFFLINE_DURATION, MAX_WORKERS, MAX_PINGS, GROUP_SIZE, LOG_NORMAL
try:
ALERT_TIME = safe_eval(self.alert_time_input.text())
ALERT_REPEAT = safe_eval(self.alert_repeat_input.text())
ALERT_TIMEOUT = safe_eval(self.alert_timeout_input.text())
OFFLINE_DURATION = safe_eval(self.offline_duration_input.text())
MAX_WORKERS = safe_eval(self.max_workers_input.text())
MAX_PINGS = safe_eval(self.max_pings_input.text())
GROUP_SIZE = safe_eval(self.group_size_input.text())
LOG_NORMAL = safe_eval(self.log_normal_input.text())
self.font_size = safe_eval(self.font_size_input.text())
self.red_alert_time = safe_eval(self.red_alert_time_input.text())
self.pop_up_alert_enabled = self.pop_up_alert_checkbox.isChecked()
self.sound_alert_enabled = self.sound_alert_checkbox.isChecked()
self.exception_hosts = set(hostname.strip().upper() for hostname in self.exception_hosts_text_edit.toPlainText().split(',') if hostname.strip())
self.sound_alert_hosts = set(hostname.strip().upper() for hostname in self.sound_alert_hosts_text_edit.toPlainText().split(',') if hostname.strip())
self.remote_alert = self.remote_alert_input.text()
self.update_exception_hosts_signal.emit(self.exception_hosts)
except ValueError as e:
QMessageBox.warning(self, "Invalid Input", str(e))
def update_exception_hosts(self):
self.exception_hosts = set(hostname.strip().upper() for hostname in self.exception_hosts_text_edit.toPlainText().split(',') if hostname.strip())
def apply_font_size(self):
for i in range(self.tab_widget.count()):
widget = self.tab_widget.widget(i)
if isinstance(widget, OfflineHostMonitor):
widget.set_font_size(self.font_size)
#if isinstance(widget, ReturnOnlineWidget):
# widget.set_font_size(self.font_size)
monitoring_threads = {}
def start_monitoring(self):
for host_info in self.host_info_list:
hostname, _, _, _, _ = host_info
normalized_hostname = hostname.upper()
if normalized_hostname not in self.monitoring_threads or not self.monitoring_threads[normalized_hostname].is_alive():
thread = threading.Thread(
target=monitor_host,
args=(host_info, self.queue, self.offline_dict, self.stop_event, self.semaphore, self.tab_widget.widget(0)),
daemon=True
)
thread.start()
self.monitoring_threads[normalized_hostname] = thread
def restart_monitoring(self):
self.stop_event.set()
time.sleep(1)
self.stop_event.clear()
self.semaphore = threading.Semaphore(MAX_PINGS)
self.queue.queue.clear()
self.monitoring_threads.clear()
self.start_monitoring()
def check_alerts(self):
current_time = time.time()
for hostname, details in self.offline_dict.items():
duration_seconds = int(current_time - details['start_time'])
if duration_seconds > ALERT_TIME and current_time - details['last_alert_time'] > ALERT_REPEAT:
normalized_hostname = hostname.lower()
exception_hosts_lower = {h.lower() for h in self.exception_hosts}
if normalized_hostname not in exception_hosts_lower:
threading.Thread(target=self.trigger_alert, args=(hostname, details), daemon=True).start()
details['last_alert_time'] = current_time
def show_settings_applied(self):
self.tab_widget.setCurrentIndex(0)
class OfflineHostMonitor(QWidget):
update_inspect_status = pyqtSignal(str, str)
def __init__(self, queue, offline_dict, stop_event, semaphore, remote_alert=''):
super().__init__()
self.queue = queue
self.offline_dict = offline_dict
self.stop_event = stop_event
self.semaphore = semaphore
self.remote_alert = remote_alert
self.settings_widget = None
self.attempts_dict = {}
self.update_inspect_status.connect(self.on_update_inspect_status)
self.red_alert_time = 15 # default value in minutes
self.red_alert_color = QColor() # default alert color
self.initUI()
self.update_signal = pyqtSignal()
threading.Thread(target=update_offline_list, args=(self.queue, self.offline_dict, self.stop_event, self.update_signal, self.settings_widget, self), daemon=True).start()
self.timer = QTimer(self)
self.timer.timeout.connect(self.refresh_table)
self.timer.start(1000) # Refresh every second
self.inspect_filter_state = 0 # 0: All, 1: WAN down & Server up, 2: Server down & Server up
self.sound_manager = SoundManager()
def set_red_alert_settings(self, red_alert_time, red_alert_color):
self.red_alert_time = red_alert_time
self.red_alert_color = red_alert_color
def show_context_menu(self, position):
index = self.table_widget.horizontalHeader().logicalIndexAt(position)
if index == 6:
menu = QMenu(self)
filter_all_action = QAction('Show All', self)
filter_all_action.setCheckable(True)
filter_wan_down_server_up_action = QAction('WAN down', self)
filter_wan_down_server_up_action.setCheckable(True)
filter_server_down_server_up_action = QAction('Server down', self)
filter_server_down_server_up_action.setCheckable(True)
filter_all_action.triggered.connect(lambda: self.apply_filter(0))
filter_wan_down_server_up_action.triggered.connect(lambda: self.apply_filter(1))
filter_server_down_server_up_action.triggered.connect(lambda: self.apply_filter(2))
self.update_check_states(filter_all_action, filter_wan_down_server_up_action, filter_server_down_server_up_action)
menu.addAction(filter_all_action)
menu.addAction(filter_wan_down_server_up_action)
menu.addAction(filter_server_down_server_up_action)
menu.exec_(self.table_widget.horizontalHeader().viewport().mapToGlobal(position))
def update_check_states(self, filter_all_action, filter_wan_down_server_up_action, filter_server_down_server_up_action):
filter_all_action.setChecked(self.inspect_filter_state == 0)
filter_wan_down_server_up_action.setChecked(self.inspect_filter_state == 1)
filter_server_down_server_up_action.setChecked(self.inspect_filter_state == 2)
def apply_filter(self, state):
self.inspect_filter_state = state
self.refresh_table()
def closeEvent(self, event):
self.stop_all_threads()
self.sound_manager.stop()
event.accept()
def stop_all_threads(self):
if hasattr(self, 'stop_event'):
self.stop_event.set()
def set_font_size(self, font_size):
font = self.table_widget.font()
font.setPointSize(font_size)
self.table_widget.setFont(font)
# Apply to headers
header_font = self.table_widget.horizontalHeader().font()
header_font.setPointSize(font_size)
self.table_widget.horizontalHeader().setFont(header_font)
def adjust_column_widths(self):
table_width = self.table_widget.viewport().width()
num_columns = self.table_widget.columnCount()
total_column_width = sum(self.table_widget.columnWidth(i) for i in range(num_columns))
if total_column_width > table_width:
scale_factor = table_width / total_column_width
else:
scale_factor = 1.0
for i in range(num_columns):
current_width = self.table_widget.columnWidth(i)
new_width = int(current_width * scale_factor)
self.table_widget.setColumnWidth(i, new_width)
min_column_width = 50
for i in range(num_columns):
if self.table_widget.columnWidth(i) < min_column_width:
self.table_widget.setColumnWidth(i, min_column_width)
self.table_widget.resizeRowsToContents()
def on_update_inspect_status(self, hostname, inspect_status):
try:
row = self.table_widget.findItems(hostname, Qt.MatchExactly)[0].row()
# do something with the row
except IndexError:
pass #print(f"Hostname '{hostname}' not found in table widget")
def initUI(self):
layout = QVBoxLayout()
self.table_widget = QTableWidget(self)
self.table_widget.setColumnCount(8)
self.table_widget.setHorizontalHeaderLabels(["No", "Hostname", "Store Name", "Telephone", "Offline Time", "Duration Offline", "Inspect", "Comment"])
self.table_widget.verticalHeader().setVisible(False)
font = QFont()
font.setPointSize(FONT_SIZE)
self.table_widget.setFont(font)
header_font = QFont()
header_font.setPointSize(FONT_SIZE)
header = self.table_widget.horizontalHeader()
header.setFont(header_font)
# Set resize modes for columns
#header.setSectionResizeMode(QHeaderView.ResizeToContents)
header.setSectionResizeMode(7, QHeaderView.Stretch) # Stretch the 'Comment' column
self.table_widget.setEditTriggers(QAbstractItemView.NoEditTriggers)
self.table_widget.cellDoubleClicked.connect(self.on_cell_double_clicked)
header.setContextMenuPolicy(Qt.CustomContextMenu)
header.customContextMenuRequested.connect(self.show_context_menu)
layout.addWidget(self.table_widget)
self.setLayout(layout)
self.setWindowTitle('Server Monitor')
self.setGeometry(300, 300, 1100, 400)
def resizeEvent(self, event):
super().resizeEvent(event)
table_width = self.table_widget.width()
scrollbar_width = 20
col_percentages = [0.01, 0.09, 0.20, 0.10, 0.15, 0.12, 0.09, 0.21]
column_widths = [table_width * p for p in col_percentages]
total_column_width = sum(column_widths)
available_width = table_width - scrollbar_width
if total_column_width > available_width:
scale_factor = available_width / total_column_width
column_widths = [width * scale_factor for width in column_widths]
min_last_col_width = 5
column_widths[-1] = max(column_widths[-1], min_last_col_width)
for i, width in enumerate(column_widths):
self.table_widget.setColumnWidth(i, int(width))
def on_cell_double_clicked(self, row, column):
if column == 1: # Hostname column
hostname = self.table_widget.item(row, 1).text()
hostname_upper = hostname.upper()
if hostname_upper not in self.settings_widget.exception_hosts:
self.settings_widget.exception_hosts.add(hostname_upper)
self.settings_widget.exception_hosts_text_edit.setPlainText(
', '.join(sorted(self.settings_widget.exception_hosts))
)
self.show_toast("Exception")
else:
self.show_toast("In Except")
elif column == 7: # Comment column
hostname = self.table_widget.item(row, 1).text()
current_comment = self.table_widget.item(row, column).text()
new_comment, ok = QInputDialog.getText(self, f"{hostname}", "Comment:", text=current_comment)
if ok:
self.update_comment(hostname, new_comment)
elif column == 3: # Telephone column
hostname = self.table_widget.item(row, 1).text()
telephone = self.table_widget.item(row, column).text()
clipboard = QApplication.clipboard()
clipboard.setText(telephone)
self.show_toast("Copied")
def show_toast(self, message):
toast_label = QLabel(message, self)
toast_label.setStyleSheet("background-color: rgba(0, 0, 0, 0.8); color: white; padding: 10px; border-radius: 6px;")
toast_label.setAlignment(Qt.AlignCenter)
toast_label.setGeometry(0, 0, 100, 33) # Adjust size and position as needed
toast_label.move(self.width() // 2 - toast_label.width() // 2, self.height() - toast_label.height() - 50)
toast_label.show()
QTimer.singleShot(1000, toast_label.hide)
def set_row_color(self, row, color):
for column in range(self.table_widget.columnCount()):
item = self.table_widget.item(row, column)
if item:
item.setBackground(QBrush(color))
def set_settings_widget(self, settings_widget):
self.settings_widget = settings_widget
def refresh_table(self):
current_time = time.time()
self.table_widget.setRowCount(0)
filtered_offline = {
hostname: details for hostname, details in self.offline_dict.items()
if (current_time - details['start_time']) > OFFLINE_DURATION
}
if self.inspect_filter_state == 1:
filtered_offline = {hostname: details for hostname, details in filtered_offline.items() if details.get('inspect_status') in ['WAN down', 'Server up']}
elif self.inspect_filter_state == 2:
filtered_offline = {hostname: details for hostname, details in filtered_offline.items() if details.get('inspect_status') in ['Server down', 'Server up']}
sorted_offline = sorted(filtered_offline.items(), key=lambda item: current_time - item[1]['start_time'], reverse=True)
if self.settings_widget:
exception_hosts = self.settings_widget.exception_hosts
for row, (hostname, details) in enumerate(sorted_offline):
duration_seconds = int(current_time - details['start_time'])
duration_minutes = duration_seconds / 60
hours, remainder = divmod(duration_seconds, 3600)
minutes, seconds = divmod(remainder, 60)
duration = f"{hours:02}:{minutes:02}:{seconds:02}"
comment = details.get('comment', '')
inspect_status = details.get('inspect_status', 'Unknown')
if duration_seconds > ALERT_TIME and current_time - details['last_alert_time'] > ALERT_REPEAT:
if self.settings_widget:
normalized_hostname = hostname.upper()
exception_hosts_upper = {h.upper() for h in self.settings_widget.exception_hosts}
if normalized_hostname not in exception_hosts_upper:
if self.settings_widget.pop_up_alert_enabled:
threading.Thread(target=self.trigger_pop_up_alert, args=(hostname, details, duration), daemon=True).start()
if self.settings_widget.sound_alert_enabled and row < 10: # check if hostname is less than 10
threading.Thread(target=self.trigger_sound_alert, args=(hostname, details, duration), daemon=True).start()
details['last_alert_time'] = current_time
self.table_widget.insertRow(row)
item_no = QTableWidgetItem(str(row + 1))
item_hostname = QTableWidgetItem(hostname)
item_store_name = QTableWidgetItem(details['store_name'])
item_telephone = QTableWidgetItem(details['telephone'])
item_offline_time = QTableWidgetItem(details['offline_time'])
item_duration = QTableWidgetItem(duration)
item_inspect = QTableWidgetItem(inspect_status)
item_comment = QTableWidgetItem(comment)
item_no.setTextAlignment(Qt.AlignCenter)
item_hostname.setTextAlignment(Qt.AlignCenter)
item_offline_time.setTextAlignment(Qt.AlignCenter)
item_duration.setTextAlignment(Qt.AlignCenter)
item_inspect.setTextAlignment(Qt.AlignCenter)
self.table_widget.setItem(row, 0, item_no)
self.table_widget.setItem(row, 1, item_hostname)
self.table_widget.setItem(row, 2, item_store_name)
self.table_widget.setItem(row, 3, item_telephone)
self.table_widget.setItem(row, 4, item_offline_time)
self.table_widget.setItem(row, 5, item_duration)
self.table_widget.setItem(row, 6, item_inspect)
self.table_widget.setItem(row, 7, item_comment)
#if hostname.startswith(("PTH11","PTH15")) or hostname.startswith(("169","s")):
hostname_lower = hostname.lower()
if any(fnmatch.fnmatch(hostname_lower, pattern.lower()) for pattern in self.settings_widget.sound_alert_patterns):
self.set_row_color(row, QColor('lightyellow'))
if duration_minutes >= self.red_alert_time and self.red_alert_color.isValid():
self.set_row_color(row, self.red_alert_color)
if inspect_status == "Server up":
self.set_row_color(row, QColor('palegreen'))
# Update the comment
details['comment'] = self.table_widget.item(row, 7).text()
self.adjust_row_heights()
def adjust_row_heights(self):
num_rows = self.table_widget.rowCount()
self.table_widget.resizeRowsToContents()
for row in range(num_rows):
current_height = self.table_widget.rowHeight(row)
new_height = int(current_height * 1.15)
self.table_widget.setRowHeight(row, new_height)
def update_comment(self, hostname, text):
if hostname in self.offline_dict:
self.offline_dict[hostname]['comment'] = text
self.refresh_table()
def trigger_alert(self, hostname, details, duration):
if self.sound_alert_enabled:
self.trigger_sound_alert(hostname, details)
if self.pop_up_alert_enabled:
self.trigger_pop_up_alert(hostname, details, duration)
def trigger_pop_up_alert(self, hostname, details, duration):
comment = details.get('comment', '')
message = (f"Server: {hostname} Store Name: {details['store_name']}\n\n"
f"Offline Time: {duration}\n\n"
f"Comment: {comment}")
try:
subprocess.run(["msg", "*", f"/TIME:{ALERT_TIMEOUT}", message], check=True, creationflags=subprocess.CREATE_NO_WINDOW)
except subprocess.CalledProcessError as e:
print(f"Failed to send pop-up message: {e}")
except subprocess.TimeoutExpired:
print("Pop-up message command timed out.")
except Exception as e:
print(f"An unexpected error occurred while sending pop-up alert: {e}")
def trigger_sound_alert(self, hostname, details, inspect_status):
hostname_lower = hostname.lower()
if any(fnmatch.fnmatch(hostname_lower, pattern.lower()) for pattern in self.settings_widget.sound_alert_patterns):
inspect_status = details.get('inspect_status', 'link down')
store_name = details.get('store_name', 'Unknown Store')
if inspect_status == 'WAN down':
inspect_status = 'Link down'
valid_statuses = {'Link down', 'Server down'}
fallback_sound_files = {'Link down': 'linkdown.mp3','Server down': 'serverdown.mp3'}
message = f"Big C {store_name} {inspect_status}"
audio_filename = f'{message}.mp3'
audio_path = os.path.join('sounds', audio_filename)
if os.path.exists(audio_path):
print(f"Sound file found: {audio_path}")
self.sound_manager.play(audio_path)
else:
print(f"Sound file '{audio_filename}' does not exist. Attempting to download...")
downloaded_file = download_audio(message)
if downloaded_file and os.path.exists(downloaded_file):
print(f"Downloaded sound file: {downloaded_file}")
self.sound_manager.play(downloaded_file)
else:
fallback_file = fallback_sound_files.get(inspect_status, None)
if fallback_file and os.path.exists(fallback_file):
print(f"Failed to download. Playing fallback sound file: {fallback_file}")
self.sound_manager.play(fallback_file)
else:
print(f"No valid sound file available for '{inspect_status}'.")
print(f"Sound alert triggered for {hostname} {store_name} ({inspect_status})")
class SoundManager:
def __init__(self, max_sounds=10):
pygame.init()
self.max_sounds = max_sounds
self.sound_queue = deque() # Use deque to easily limit the queue size
self.playing_lock = threading.Lock()
self.program_start_time = time.time() # Track the program start time
self.initial_delay_seconds = 30 # 30-second delay for initial sound ignore
self.thread = threading.Thread(target=self._play_sounds, daemon=True)
self.thread.start()
def _play_sounds(self):
while True:
if not self.sound_queue:
time.sleep(0.1) # Sleep for a short while if no sounds are in the queue
continue
sound_file = self.sound_queue.popleft() # Pop from the left
if sound_file is None:
break
with self.playing_lock:
try:
pygame.mixer.music.load(sound_file)
pygame.mixer.music.play()
while pygame.mixer.music.get_busy():
pygame.time.Clock().tick(10)
except Exception as e:
print(f"Error playing sound {sound_file}: {e}")
def play(self, sound_file):
# Check if the current time is beyond the 30-second initial delay
if time.time() - self.program_start_time > self.initial_delay_seconds:
if len(self.sound_queue) < self.max_sounds:
self.sound_queue.append(sound_file)
def stop(self):
self.sound_queue = deque() # Clear the queue to stop any further sounds from playing
pygame.quit()
class ReturnOnlineWidget(QWidget):
dataUpdated = pyqtSignal() # Define a signal to notify when data is updated
def __init__(self, parent=None):
super().__init__(parent)
self.initUI()
def set_font_size(self, font_size):
font = self.table_widget.font()
font.setPointSize(font_size)
self.table_widget.setFont(font)
header_font = self.table_widget.horizontalHeader().font()
header_font.setPointSize(font_size)
self.table_widget.horizontalHeader().setFont(header_font)
def initUI(self):
layout = QVBoxLayout(self)
# Create filter controls
self.start_date_edit = QDateEdit(self)
self.start_date_edit.setDate(QDate.currentDate())
self.start_date_edit.setCalendarPopup(True)
self.start_date_edit.setDisplayFormat("yyyy-MM-dd")
self.start_date_edit.dateChanged.connect(self.load_data)
self.start_date_edit.setFixedWidth(120) # Set fixed width
self.start_date_edit.setLocale(english_locale)
self.end_date_edit = QDateEdit(self)
self.end_date_edit.setDate(QDate.currentDate())
self.end_date_edit.setCalendarPopup(True)
self.end_date_edit.setDisplayFormat("yyyy-MM-dd")
self.end_date_edit.dateChanged.connect(self.load_data)
self.end_date_edit.setFixedWidth(120) # Set fixed width
self.end_date_edit.setLocale(english_locale)
self.status_filter = QComboBox(self)
self.status_filter.addItems(['All', 'WAN down', 'Server down'])
self.status_filter.currentTextChanged.connect(self.load_data)
self.duration_edit = QLineEdit(self)
self.duration_edit.setPlaceholderText(">= Minutes")
self.duration_edit.textChanged.connect(self.load_data)
self.duration_edit.setValidator(QIntValidator(0, 999999, self))
self.hostname_edit = QLineEdit(self)
self.hostname_edit.setPlaceholderText("Hostname (All)")
self.hostname_edit.textChanged.connect(self.load_data)
self.export_button = QPushButton("Export to CSV", self)
self.export_button.clicked.connect(self.export_to_csv)
self.export_button.setFixedWidth(120) # Set fixed width
# Layout for filter controls
filter_layout = QHBoxLayout()
filter_layout.addWidget(QLabel("Start Date:", self))
filter_layout.addWidget(self.start_date_edit)
filter_layout.addWidget(QLabel("End Date:", self))
filter_layout.addWidget(self.end_date_edit)
filter_layout.addWidget(QLabel("Status:", self))
filter_layout.addWidget(self.status_filter)
filter_layout.addWidget(QLabel("Duration:", self))
filter_layout.addWidget(self.duration_edit)
filter_layout.addWidget(QLabel("Hostname:", self))
filter_layout.addWidget(self.hostname_edit)
filter_layout.addWidget(self.export_button)
layout.addLayout(filter_layout)
# Status bar
self.status_bar = QStatusBar(self)
layout.addWidget(self.status_bar)
# Create the table widget
self.table_widget = QTableWidget(self)
self.table_widget.setEditTriggers(QAbstractItemView.NoEditTriggers)
# Set up table columns and headers
self.table_widget.setColumnCount(7)
self.table_widget.setHorizontalHeaderLabels(['Hostname', 'Storename', 'Start Offline', 'End Offline', 'Duration', 'Status', 'Comment'])
# Set resize modes for the table columns
header = self.table_widget.horizontalHeader()
#header.setSectionResizeMode(QHeaderView.ResizeToContents)
header.setStretchLastSection(True) # Makes the 'Comment' column stretch
# Optionally, adjust resize modes for specific columns
# header.setSectionResizeMode(0, QHeaderView.ResizeToContents) # 'Hostname'
# header.setSectionResizeMode(1, QHeaderView.ResizeToContents) # 'Storename'
# header.setSectionResizeMode(2, QHeaderView.Stretch) # 'Start Offline'
# header.setSectionResizeMode(3, QHeaderView.Stretch) # 'End Offline'
# header.setSectionResizeMode(4, QHeaderView.ResizeToContents) # 'Duration'
# header.setSectionResizeMode(5, QHeaderView.ResizeToContents) # 'Status'
# header.setSectionResizeMode(6, QHeaderView.Stretch) # 'Comment'
# Adjust fonts
font = QFont()
font.setPointSize(12)
self.table_widget.setFont(font)
header.setFont(font)
# Add table to layout
layout.addWidget(self.table_widget)
self.setLayout(layout)
# Connect data updated signal
self.dataUpdated.connect(self.load_data)
# Initial data load
self.load_data()
def convert_to_seconds(self, duration):
try:
if duration.isdigit():
return int(duration) * 60 # Convert minutes to seconds
parts = duration.split(':')
if len(parts) == 3:
hours, minutes, seconds = map(int, parts)
return hours * 3600 + minutes * 60 + seconds
elif len(parts) == 2: # Handle mm:ss format if needed
minutes, seconds = map(int, parts)
return minutes * 60 + seconds
elif len(parts) == 1: # Handle ss format if needed
return int(parts[0])
except ValueError:
return None
def load_data(self):
folder_path = 'history_log'
lock_path = 'history_log.lock'
temp_file_path = 'history_log_temp.txt'
lock = FileLock(lock_path)
def is_older_than_180_days(date_str):
date_format = "%Y-%m-%d"
try:
date_obj = datetime.strptime(date_str, date_format)
return date_obj < datetime.now() - timedelta(days=180)
except ValueError:
return True
try:
with lock.acquire(timeout=5): # Wait up to 5 seconds for the lock
files_to_process = []
for filename in os.listdir(folder_path):
# Exclude history_all.txt from processing
if filename.endswith(".txt") and filename != 'history_all.txt' and is_older_than_180_days(filename[:10]):
files_to_process.append(os.path.join(folder_path, filename))
lines = []
for file_path in files_to_process:
with open(file_path, 'r', encoding='utf-8') as file:
lines.extend(file.readlines())
lines_before_cleanup = len(lines)
lines_after_cleanup = 0
removed_dates = set()
with open(temp_file_path, 'w', encoding='utf-8') as temp_file:
for line in lines:
parts = line.strip().split('|')
if len(parts) >= 11:
end_offline_date_str = parts[6] # parts[6] is online_date in 'YYYY-MM-DD' format
if is_older_than_180_days(end_offline_date_str):
removed_dates.add(end_offline_date_str)
else:
temp_file.write(line)
lines_after_cleanup += 1
del lines # Remove reference to lines
if os.path.exists(temp_file_path):
try:
os.replace(temp_file_path, os.path.join(folder_path, 'history_all.txt'))
except Exception as e:
print(f"Error during file replacement: {e}")
return
else:
print(f"Temporary file {temp_file_path} does not exist.")
return
# Retrieve user inputs for filtering
start_date = self.start_date_edit.date().toPyDate() # Returns a datetime.date object
end_date = self.end_date_edit.date().toPyDate() # Returns a datetime.date object
status_filter = self.status_filter.currentText()
hostname_filter = self.hostname_edit.text().strip().lower()
# Parse the duration filter
user_input_duration = self.duration_edit.text().strip()
if user_input_duration.isdigit():
duration_filter = self.convert_to_seconds(user_input_duration)
else:
duration_filter = None # If invalid, don't filter on duration
rows = []
with open(os.path.join(folder_path, 'history_all.txt'), 'r', encoding='utf-8') as file:
for line in file:
parts = line.strip().split('|')
if len(parts) >= 11:
# Parse end_offline_date_str into a datetime.date object
try:
end_offline_date_str = parts[6] # online_date
end_offline_date = datetime.strptime(end_offline_date_str, '%Y-%m-%d').date()
except ValueError:
continue # Skip this line if date can't be parsed
# Check if the end_offline_date is within the selected date range
if start_date <= end_offline_date <= end_date:
# Apply status filter
if status_filter == 'All' or parts[9] == status_filter:
# Apply hostname filter
hostname = parts[2]
if hostname_filter == "" or hostname_filter.lower() in hostname.lower():
# Convert duration from the file to seconds for comparison
duration_str = parts[8]
duration = self.convert_to_seconds(duration_str)
if duration_filter is None or (duration is not None and duration >= duration_filter):
start_offline = f"{parts[4]} {parts[5]}" # start_date and start_time_str
end_offline = f"{parts[6]} {parts[7]}" # online_date and online_time_str
rows.append([hostname, parts[3], start_offline, end_offline, duration_str, parts[9], parts[10]])
rows.reverse() # To show latest entries first
# Set up table widget
self.table_widget.setRowCount(len(rows))
self.table_widget.setColumnCount(7)
self.table_widget.setHorizontalHeaderLabels(['Hostname', 'Storename', 'Start Offline', 'End Offline', 'Duration', 'Status', 'Comment'])
for row_idx, (hostname, storename, start_offline, end_offline, duration, status, comment) in enumerate(rows):
self.table_widget.setItem(row_idx, 0, QTableWidgetItem(hostname))
self.table_widget.setItem(row_idx, 1, QTableWidgetItem(storename))
self.table_widget.setItem(row_idx, 2, QTableWidgetItem(start_offline))
self.table_widget.setItem(row_idx, 3, QTableWidgetItem(end_offline))
self.table_widget.setItem(row_idx, 4, QTableWidgetItem(duration))
self.table_widget.setItem(row_idx, 5, QTableWidgetItem(status))
self.table_widget.setItem(row_idx, 6, QTableWidgetItem(comment))
self.table_widget.verticalHeader().setVisible(False)
# Update status bar
total_count = len(rows)
wan_down_count = sum(1 for row in rows if row[5] == 'WAN down')
server_down_count = sum(1 for row in rows if row[5] == 'Server down')
self.status_bar.showMessage(f"Total: {total_count} | WAN down: {wan_down_count} | Server down: {server_down_count}")
except Timeout:
print("Failed to acquire file lock")
except Exception as e:
print(f"An error occurred: {e}")
def resizeEvent(self, event):
super().resizeEvent(event)
table_width = self.table_widget.width()
scrollbar_width = 20
col_percentages = [0.13, 0.31, 0.20, 0.20, 0.10, 0.13, 0.13]
column_widths = [table_width * p for p in col_percentages]
total_column_width = sum(column_widths)
if total_column_width > table_width:
scale_factor = (table_width - scrollbar_width) / total_column_width
column_widths = [width * scale_factor for width in column_widths]
min_last_col_width = 40
column_widths[-1] = max(column_widths[-1], min_last_col_width)
for i, width in enumerate(column_widths):
self.table_widget.setColumnWidth(i, int(width))
#self.adjust_row_heights()
def adjust_row_heights(self):
num_rows = self.table_widget.rowCount()
self.table_widget.resizeRowsToContents()
for row in range(num_rows):
current_height = self.table_widget.rowHeight(row)
new_height = int(current_height * 1.1)
self.table_widget.setRowHeight(row, new_height)
def export_to_csv(self):
start_date_str = self.start_date_edit.date().toString("yyyy-MM-dd")
end_date_str = self.end_date_edit.date().toString("yyyy-MM-dd")
default_filename = f"history_online_{start_date_str}-{end_date_str}.csv"
file_path, _ = QFileDialog.getSaveFileName(self, "Save CSV File", default_filename, "CSV Files (*.csv)")
if not file_path:
return
if not file_path.endswith(".csv"):
file_path += ".csv"
lock_path = 'history_online.lock'
lock = FileLock(lock_path)
try:
with lock.acquire(timeout=10): # Wait up to 10 seconds for the lock
with open(file_path, 'w', newline='', encoding='utf-8-sig') as file:
writer = csv.writer(file)
writer.writerow(['Hostname', 'Storename', 'Start Offline', 'End Offline', 'Duration', 'Status', 'Comment'])
for row_idx in range(self.table_widget.rowCount()):
row_data = [self.table_widget.item(row_idx, col_idx).text() for col_idx in range(self.table_widget.columnCount())]
writer.writerow(row_data)
print(f"Data exported to {file_path}")
except Timeout:
print("Failed to acquire file lock")
def read_host_list(file_path):
host_info = []
hostname_count = 0
try:
with open(file_path, 'r', encoding='utf-8-sig') as file:
for line in file:
line = line.strip()
if line:
normalized_line = line.upper()
if normalized_line.startswith("#"):
print(f"Skipping line: {line}")
continue
parts = line.split('|') # Change delimiter to |
if len(parts) > 5:
print(f"Warning: Line has too many parts, using only the first 5: {line}")
parts = parts[:5] # Truncate to the first 5 parts
parts = parts + [''] * (5 - len(parts))
hostname = parts[0].strip()
store_name = parts[1].strip() if len(parts) > 1 else ""
telephone = parts[2].strip() if len(parts) > 2 else ""
wan_a = parts[3].strip() if len(parts) > 3 else ""
wan_b = parts[4].strip() if len(parts) > 4 else ""
host_info.append((hostname, store_name, telephone, wan_a, wan_b))
hostname_count += 1
except Exception as e:
print(f"Error reading file '{file_path}': {e}")
group_size = max(1,(hostname_count + 19) // GROUP_SIZE)
grouped_host_info = [host_info[i:i + group_size] for i in range(0, hostname_count, group_size)]
# Reorder the groups to prevent consecutive repeats of the last hostname
host_info = []
for i in range(group_size):
for group in grouped_host_info:
if i < len(group):
host_info.append(group[i])
# Rotate the groups to prevent consecutive repeats
grouped_host_info = grouped_host_info[1:] + [grouped_host_info[0]]
#return reordered_host_info, hostname_count
return host_info, hostname_count
def count_hostnames(file_path):
try:
with open(file_path, 'r', encoding='utf-8') as file:
return sum(1 for line in file if line.strip())
except Exception as e:
print(f"Error reading file '{file_path}': {e}")
return 0
prev_ping_time = 0
def check_port(hostname, port, timeout):
try:
with socket.create_connection((hostname, port), timeout=timeout):
return port
except (socket.timeout, ConnectionRefusedError, OSError):
return None
def tcp_socket(hostname, wan_a=None, wan_b=None, timeout=2):
global prev_ping_time
n = QTime.currentTime().toString("hh:mm:ss")
inspect_status = ""
ports = [80, 8080, 22, 23, 3389]
if hostname.startswith("169.254.163.169"):
current_time = time.time()
time_diff = current_time - prev_ping_time if prev_ping_time != 0 else 0
print(f"{n} Ping Checked (Diff previous ping: {time_diff:.2f}s)")
prev_ping_time = current_time
# Check the first port (80) sequentially
if check_port(hostname, ports[0], timeout):
inspect_status = "Server up"
return True, inspect_status
# Check remaining ports in parallel
with ThreadPoolExecutor() as executor:
future_to_port = {executor.submit(check_port, hostname, port, timeout): port for port in ports[1:]}
for future in as_completed(future_to_port):
result = future.result()
if result is not None:
inspect_status = "Server up"
# Cancel any outstanding futures
for pending_future in future_to_port:
if pending_future != future:
pending_future.cancel()
return True, inspect_status
# Second Checking
if hostname.startswith(("PTH2", "PTH3")):
try:
ip = socket.gethostbyname(hostname)
wan_ip = '.'.join(ip.split('.')[:-1] + ['10'])
try:
with socket.create_connection((wan_ip, 23), timeout=timeout):
inspect_status = "Server down"
return False, inspect_status
except (socket.timeout, ConnectionRefusedError, OSError):
inspect_status = "WAN down"
return False, inspect_status
except (socket.gaierror, socket.timeout, ConnectionRefusedError, OSError):
inspect_status = "WAN down"
return False, inspect_status
elif hostname.startswith(("PTH11", "PTH15")) or hostname.startswith(("169", "s")):
for ip, port in [(wan_a, 23), (wan_b, 23)]:
if ip:
try:
with socket.create_connection((ip, port), timeout=timeout):
inspect_status = "Server down"
return False, inspect_status
except (socket.timeout, ConnectionRefusedError, OSError):
pass
inspect_status = "WAN down"
return False, inspect_status
else:
inspect_status = "Server down"
return False, inspect_status
def get_google_translate_tts_url(text, lang='th'):
"""Get the URL for Google Translate TTS."""
url = 'https://translate.google.com/translate_tts'
params = {
'ie': 'UTF-8',
'q': text,
'tl': lang,
'client': 'gtx'
}
return url, params
def download_audio(text, lang='th'):
try:
url, params = get_google_translate_tts_url(text, lang)
response = requests.get(url, params=params, headers={'User-Agent': 'Mozilla/5.0'})
response.raise_for_status()
sounds_folder = 'sounds'
if not os.path.exists(sounds_folder):
os.mkdir(sounds_folder)
audio_filename = f'{text}.mp3'
audio_path = os.path.join(sounds_folder, audio_filename)
with open(audio_path, 'wb') as audio_file:
audio_file.write(response.content)
print(f"Audio saved to {audio_path}")
return audio_path
except requests.RequestException as e:
print(f"Failed to download audio: {e}")
return None
offline_dict_lock = Lock()
def schedule_host_deletion(hostname, offline_dict, update_signal, settings_widget, lock, main_window):
def delete_host():
normalized_hostname = hostname.upper()
with lock:
if normalized_hostname in offline_dict:
store_name = offline_dict[normalized_hostname]['store_name']
log_online(
hostname,
store_name,
offline_dict,
normalized_hostname,
offline_dict[normalized_hostname]['inspect_status']
)
offline_duration = time.time() - offline_dict[normalized_hostname]['start_time']
if settings_widget.sound_alert_enabled and offline_duration > OFFLINE_DURATION and len(main_window.sound_manager.sound_queue) < main_window.sound_manager.max_sounds:
hostname_lower = hostname.lower()
if any(fnmatch.fnmatch(hostname_lower, pattern.lower()) for pattern in settings_widget.sound_alert_patterns):
if hasattr(main_window, 'sound_manager'):
message = f"Big C {store_name} return to Online"
audio_filename = f"{message}.mp3"
audio_path = os.path.join('sounds', audio_filename)
fallback_sound_file = "returnonline.mp3"
def play_sound():
if os.path.exists(audio_path):
main_window.sound_manager.play(audio_path)
else:
print(f"Sound file '{audio_filename}' does not exist. Attempting to download...")
downloaded_file = download_audio(message)
if downloaded_file and os.path.exists(downloaded_file):
main_window.sound_manager.play(downloaded_file)
else:
if os.path.exists(fallback_sound_file):
print(f"Failed to download. Playing fallback sound file: {fallback_sound_file}")
main_window.sound_manager.play(fallback_sound_file)
else:
print("Failed to download sound file and no fallback sound available.")
sound_thread = threading.Thread(target=play_sound)
sound_thread.start()
del offline_dict[normalized_hostname]
save_offline_state('offline_state.json', offline_dict)
if normalized_hostname in settings_widget.exception_hosts:
settings_widget.exception_hosts.remove(normalized_hostname)
settings_widget.update_exception_hosts_signal.emit(settings_widget.exception_hosts)
#else:
#print(f"Hostname {normalized_hostname} was not in offline_dict")
update_signal.emit()
timer = threading.Timer(5.0, delete_host)
timer.start()
def update_offline_list(queue, offline_dict, stop_event, update_signal, settings_widget, main_window):
if stop_event is None:
print("Stop event is None")
return
while not stop_event.is_set():
try:
hostname, store_name, telephone, is_online, inspect_status = queue.get(timeout=1)
normalized_hostname = hostname.upper()
if is_online:
if normalized_hostname in offline_dict:
# Schedule the deletion after 1 second without blocking the loop
schedule_host_deletion(hostname, offline_dict, update_signal, settings_widget, offline_dict_lock, main_window)
elif not is_online and normalized_hostname not in offline_dict:
with offline_dict_lock:
offline_dict[normalized_hostname] = {
'store_name': store_name,
'telephone': telephone,
'start_time': time.time(),
'offline_time': time.strftime('%Y-%m-%d %H:%M:%S', time.localtime()),
'last_alert_time': 0,
'inspect_status': inspect_status
}
update_signal.emit()
queue.task_done()
except Empty:
time.sleep(1)
except Exception as e:
print(f"Error in update_offline_list: {e}")
def monitor_host(host_info, queue, offline_dict, stop_event, semaphore, main_window):
global previous_status
previous_status = {}
hostname, store_name, telephone, wan_a, wan_b = host_info
normalized_hostname = hostname.upper()
while not stop_event.is_set():
with semaphore:
is_online, inspect_status = tcp_socket(hostname, wan_a, wan_b)
if main_window is not None and hasattr(main_window, 'update_inspect_status'):
main_window.update_inspect_status.emit(hostname, inspect_status)
#main_window.update_inspect_status.emit(hostname, inspect_status)
if not is_online:
if normalized_hostname not in offline_dict:
offline_dict[normalized_hostname] = {
'store_name': store_name,
'telephone': telephone,
'start_time': time.time(),
'offline_time': time.strftime('%Y-%m-%d %H:%M:%S', time.localtime()),
'last_alert_time': 0,
'inspect_status': inspect_status
}
else:
offline_dict[normalized_hostname]['inspect_status'] = inspect_status
previous_status[normalized_hostname] = inspect_status # store previous status
else:
if normalized_hostname in offline_dict:
offline_dict[normalized_hostname]['inspect_status'] = inspect_status
queue.put((hostname, store_name, telephone, is_online, inspect_status))
time.sleep(3)
file_lock = threading.Lock()
previous_status = {}
def log_online(hostname, store_name, offline_dict, normalized_hostname, inspect_status):
global previous_status
start_time = offline_dict[normalized_hostname]['start_time']
online_time = time.time()
offline_duration = online_time - start_time
duration_time = time.strftime('%H:%M:%S', time.gmtime(offline_duration))
start_date = time.strftime('%Y-%m-%d', time.localtime(start_time))
start_time_str = time.strftime('%H:%M:%S', time.localtime(start_time))
online_date = time.strftime('%Y-%m-%d', time.localtime(online_time))
online_time_str = time.strftime('%H:%M:%S', time.localtime(online_time))
previous_status_str = previous_status.get(normalized_hostname, "Server down") # Get previous status
comment = offline_dict[normalized_hostname].get('comment', '')
if offline_duration > LOG_NORMAL: # Only write to file if offline duration exceeds LOG_NORMAL
log_entry = (
f"{online_date}|{online_time_str}|{hostname}|{store_name}|"
f"{start_date}|{start_time_str}|{online_date}|{online_time_str}|"
f"{duration_time}|{previous_status_str}|{comment}\n"
)
def backup_history():
backup_dir = 'backup'
os.makedirs(backup_dir, exist_ok=True) # Create backup folder if it doesn't exist
timestamp = time.strftime('%Y%m%d_%H%M%S', time.localtime(online_time))
backup_filename = f"history_online_backup_{timestamp}.txt"
backup_path = os.path.join(backup_dir, backup_filename)
# Get a list of existing backup files
backup_files = [f for f in os.listdir(backup_dir) if f.startswith('history_online_backup_')]
# If there are more than 1 backup files, remove the oldest one
if len(backup_files) >= 1:
oldest_file = min(backup_files, key=lambda f: os.path.getctime(os.path.join(backup_dir, f)))
os.remove(os.path.join(backup_dir, oldest_file))
try:
if os.path.exists(os.path.join('history_log', f"history_online_{online_date}.txt")):
shutil.copy(os.path.join('history_log', f"history_online_{online_date}.txt"), backup_path)
#print(f"Backup successful: {backup_path}")
else:
print(f"No history_online_{online_date}.txt file available for backup.")
except Exception as e:
print(f"Error occurred while backing up the file: {e}")
# Call the backup function before writing new data
backup_history()
# Write new data to the file
try:
log_dir = 'history_log'
os.makedirs(log_dir, exist_ok=True) # Create log folder if it doesn't exist
with open(os.path.join(log_dir, f"history_online_{online_date}.txt"), 'a', encoding='utf-8') as file:
file.write(log_entry)
#print(f"Log entry written successfully to {log_dir}/history_online_{online_date}.txt")
except Exception as e:
print(f"Error occurred while writing the log entry: {e}")
save_lock = threading.Lock()
def save_offline_state(file_path, offline_dict):
backup_folder = os.path.join(os.path.dirname(file_path), 'backup')
if not os.path.exists(backup_folder):
os.makedirs(backup_folder)
backup_file_path = os.path.join(backup_folder, os.path.basename(file_path) + '.bak')
if os.path.exists(file_path):
shutil.copyfile(file_path, backup_file_path)
temp_file_path = file_path + '.tmp'
with save_lock:
try:
with open(temp_file_path, 'w', encoding='utf-8') as temp_file:
json.dump(offline_dict, temp_file, ensure_ascii=False, indent=4)
os.replace(temp_file_path, file_path)
#print(f"Offline state saved to '{file_path}'")
except Exception as e:
#print(f"Error saving offline state to '{file_path}': {e}")
if os.path.exists(temp_file_path):
os.remove(temp_file_path)
def load_offline_state(file_path):
current_time = time.time()
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
backup_filename = f"offline_state_as_open_{timestamp}.json"
# Ensure the history_log directory exists
history_log_dir = "history_log"
os.makedirs(history_log_dir, exist_ok=True)
if os.path.exists(file_path):
try:
# Copy the file to the history_log directory with the new name
backup_file_path = os.path.join(history_log_dir, backup_filename)
shutil.copy(file_path, backup_file_path)
print(f"Offline state backed up to '{backup_file_path}'")
with open(file_path, 'r', encoding='utf-8') as file:
offline_dict = json.load(file)
offline_dict = {hostname.upper(): details for hostname, details in offline_dict.items()}
for details in offline_dict.values():
details['last_alert_time'] = current_time
print(f"Offline state loaded from '{file_path}'")
return offline_dict
except Exception as e:
print(f"Error loading offline state from '{file_path}': {e}")
return {}
else:
print(f"File '{file_path}' does not exist.")
return {}
def setup_application():
return QApplication(sys.argv)
LOCK_FILE = "program.lock"
class LockManager:
def __init__(self, filename):
self.filename = filename
self.file = None
def acquire(self):
self.file = open(self.filename, 'w')
try:
msvcrt.locking(self.file.fileno(), msvcrt.LK_NBLCK, 1)
except (IOError, OSError):
self.file.close()
raise
def release(self):
try:
if self.file:
msvcrt.locking(self.file.fileno(), msvcrt.LK_UNLCK, 1)
self.file.close()
os.remove(self.filename)
except Exception as e:
print(f"Error releasing lock: {e}")
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
class HostnameFileEventHandler(FileSystemEventHandler):
def __init__(self, main_window, filename):
super().__init__()
self.main_window = main_window
self.filename = filename
def on_modified(self, event):
if event.src_path.endswith(self.filename):
self.reload_hosts()
def reload_hosts(self):
try:
host_info_list, hostname_count = read_host_list(self.filename)
print(f"Reloaded: Number of hostnames in file: {hostname_count}")
# Ensure that active threads are stopped and restarted with new hosts
if not host_info_list:
sys.exit()
# Clear existing host info and restart monitoring
self.main_window.stop_all_threads()
self.main_window.host_info_list = host_info_list
self.main_window.start_monitoring()
except Exception as e:
QMessageBox.warning(None, "Error", f"Failed to reload hostnames: {e}")
def monitor_hostname_file(main_window, filename='hostname.txt'):
event_handler = HostnameFileEventHandler(main_window, filename)
observer = Observer()
observer.schedule(event_handler, os.path.dirname(os.path.abspath(filename)), recursive=False)
observer.start()
return observer
def main():
# Initialize PyQt5 application
app = QApplication(sys.argv)
lock_manager = LockManager(LOCK_FILE)
try:
lock_manager.acquire()
except (IOError, OSError):
QMessageBox.warning(None, "", "The program is already running.")
if lock_manager.file:
lock_manager.file.close()
sys.exit()
atexit.register(lock_manager.release)
# Your main program logic here
print("Program is running...")
global MAX_PINGS, MAX_WORKERS
tab_widget = QTabWidget()
# Initialize SoundManager (assuming this is defined elsewhere in your code)
sound_manager = SoundManager()
# Main widget and layout setup
main_widget = QWidget()
main_layout = QVBoxLayout(main_widget)
# UI setup (e.g., labels for status information)
host_label = QLabel()
host_label.setAlignment(Qt.AlignRight | Qt.AlignVCenter)
host_label.setStyleSheet("font-size: 12pt; color: black; padding-right: 10px;")
clock_label = QLabel()
clock_label.setAlignment(Qt.AlignRight | Qt.AlignVCenter)
clock_label.setStyleSheet("font-size: 12pt;")
# Arrange layout
main_layout.addWidget(host_label)
main_layout.addWidget(tab_widget)
main_layout.addWidget(clock_label)
main_widget.setLayout(main_layout)
main_widget.setWindowTitle("Server Monitor - Alert")
main_widget.setGeometry(300, 300, 1100, 600)
# Function to update status labels
def update_labels():
current_time = QTime.currentTime().toString("hh:mm:ss")
current_date = QDate.currentDate().toString("dd MMM yyyy")
# Assuming host_info_dict is defined/available
total_hosts = len(host_info_dict)
offline_hosts = 0 # Implement logic to determine online/offline hosts
offline_hosts_wan = 0 # Implement logic to specifically track WAN status
offline_hosts_serv = 0 # Implement logic for Server status
# Example label updates
host_label.setText(f"Total: {total_hosts} Online: {total_hosts - offline_hosts} Offline: {offline_hosts} [ Server: {offline_hosts_serv} Wan: {offline_hosts_wan} ]")
clock_label.setText(f"{current_date} - {current_time} ")
# Timer for updating labels
timer = QTimer()
timer.timeout.connect(update_labels)
timer.start(1000) # Update every second
# Create main UI components
main_window = OfflineHostMonitor(None, None, None, None)
tab_widget.addTab(main_window, "Monitor")
return_online_widget = ReturnOnlineWidget()
tab_widget.addTab(return_online_widget, "Return Online")
main_window.sound_manager = sound_manager # Attach the sound manager to the main window
settings_widget = SettingsWidget(None, None, None, None, None, tab_widget, main_window)
tab_widget.addTab(settings_widget, "Settings")
update_log_widget = UpdateLogWidget()
tab_widget.addTab(update_log_widget, "Update Log")
main_window.settings_widget = settings_widget
observer = monitor_hostname_file(main_window)
# Main event loop
try:
main_window.show()
sys.exit(app.exec_())
finally:
observer.stop()
observer.join()
# Prepare application exit cleanup
def cleanup():
stop_event.set() # Signal all threads to stop
for thread in active_threads:
thread.join() # Wait for threads to finish
# Connect cleanup on application exit
app.aboutToQuit.connect(cleanup)
# File path setup and initial checks
file_path = 'hostname.txt'
offline_state_file = 'offline_state.json'
if not os.path.exists(file_path):
QMessageBox.critical(None, "File Not Found", "The file 'hostname.txt' was not found. Exiting the application.")
sys.exit()
host_info_list, hostname_count = read_host_list(file_path)
print(f"Number of hostnames in file: {hostname_count}")
if not host_info_list:
sys.exit()
# Load offline state
offline_dict = load_offline_state(offline_state_file)
host_info_dict = {hostname.upper(): (hostname, store_name, telephone, wan_a, wan_b) for hostname, store_name, telephone, wan_a, wan_b in host_info_list}
offline_dict = {hostname: details for hostname, details in offline_dict.items() if hostname in host_info_dict}
# Queue and threading setup
queue = Queue()
stop_event = threading.Event()
active_threads = []
cpu_count = psutil.cpu_count(logical=True)
memory = psutil.virtual_memory()
available_memory_gb = memory.available / (1024 ** 2)
max_pings_based_on_memory = int(available_memory_gb / 0.1)
MAX_PINGS = min(MAX_PINGS, max_pings_based_on_memory)
semaphore = threading.Semaphore(MAX_PINGS)
# Assign globals to widgets
main_window.queue = queue
main_window.offline_dict = offline_dict
main_window.stop_event = stop_event
main_window.semaphore = semaphore
settings_widget.queue = queue
settings_widget.offline_dict = offline_dict
settings_widget.stop_event = stop_event
settings_widget.semaphore = semaphore
settings_widget.host_info_list = host_info_list
return_online_signal = return_online_widget.dataUpdated
# Start monitoring threads for each host
for host_info in host_info_list:
thread = threading.Thread(
target=monitor_host,
args=(host_info, queue, offline_dict, stop_event, semaphore, main_window)
)
thread.start()
active_threads.append(thread)
# Start thread to handle offline list updates
update_thread = threading.Thread(
target=update_offline_list,
args=(queue, offline_dict, stop_event, return_online_signal, settings_widget, main_window),
daemon=True
)
update_thread.start()
active_threads.append(update_thread)
# Show main widget
main_widget.show()
# Ensure state is saved on app quit
app.aboutToQuit.connect(lambda: save_offline_state(offline_state_file, offline_dict))
sys.exit(app.exec_())
if __name__ == "__main__":
main()
i got error
Traceback (most recent call last):
File "C:\Users\Noname\phyper\phyper.v49.7.py", line 1534, in update_labels
total_hosts = len(host_info_dict)
^^^^^^^^^^^^^^
NameError: cannot access free variable 'host_info_dict' where it is not associated with a value in enclosing scopeASSISTANT
The error you're encountering occurs because the `host_info_dict` variable is being referenced in the `update_labels` function, but it hasn't been defined in the scope of that function yet. To resolve this issue, you need to ensure that `host_info_dict` is properly initialized and accessible in the `update_labels` function.
Here's how you can address this issue:
1. Ensure proper initialization of `host_info_dict`. This should be done before you use it in the `update_labels` function.
2. Pass `host_info_dict` to the `update_labels` function, or ensure that it is in a scope where it can be accessed by that function.
Below is your `main()` function with modifications to ensure `host_info_dict` is correctly defined and passed where needed:
```python
def main():
# Initialize PyQt5 application
app = QApplication(sys.argv)
lock_manager = LockManager(LOCK_FILE)
try:
lock_manager.acquire()
except (IOError, OSError):
QMessageBox.warning(None, "", "The program is already running.")
if lock_manager.file:
lock_manager.file.close()
sys.exit()
atexit.register(lock_manager.release)
# Your main program logic here
print("Program is running...")
global MAX_PINGS, MAX_WORKERS
tab_widget = QTabWidget()
# Initialize SoundManager (assuming this is defined elsewhere in your code)
sound_manager = SoundManager()
# Configure file paths and host related data
file_path = 'hostname.txt'
offline_state_file = 'offline_state.json'
if not os.path.exists(file_path):
QMessageBox.critical(None, "File Not Found", "The file 'hostname.txt' was not found. Exiting the application.")
sys.exit()
host_info_list, hostname_count = read_host_list(file_path)
print(f"Number of hostnames in file: {hostname_count}")
if not host_info_list:
sys.exit()
# Load offline state
offline_dict = load_offline_state(offline_state_file)
host_info_dict = {hostname.upper(): (hostname, store_name, telephone, wan_a, wan_b) for hostname, store_name, telephone, wan_a, wan_b in host_info_list}
offline_dict = {hostname: details for hostname, details in offline_dict.items() if hostname in host_info_dict}
# Main widget and layout setup
main_widget = QWidget()
main_layout = QVBoxLayout(main_widget)
# UI setup (e.g., labels for status information)
host_label = QLabel()
host_label.setAlignment(Qt.AlignRight | Qt.AlignVCenter)
host_label.setStyleSheet("font-size: 12pt; color: black; padding-right: 10px;")
clock_label = QLabel()
clock_label.setAlignment(Qt.AlignRight | Qt.AlignVCenter)
clock_label.setStyleSheet("font-size: 12pt;")
# Arrange layout
main_layout.addWidget(host_label)
main_layout.addWidget(tab_widget)
main_layout.addWidget(clock_label)
main_widget.setLayout(main_layout)
main_widget.setWindowTitle("Server Monitor - Alert")
main_widget.setGeometry(300, 300, 1100, 600)
# Function to update status labels
def update_labels():
current_time = QTime.currentTime().toString("hh:mm:ss")
current_date = QDate.currentDate().toString("dd MMM yyyy")
total_hosts = len(host_info_dict)
offline_hosts = 0 # Implement logic to determine online/offline hosts
offline_hosts_wan = 0 # Implement logic to specifically track WAN status
offline_hosts_serv = 0 # Implement logic for Server status
# Example label updates
host_label.setText(f"Total: {total_hosts} Online: {total_hosts - offline_hosts} Offline: {offline_hosts} [ Server: {offline_hosts_serv} Wan: {offline_hosts_wan} ]")
clock_label.setText(f"{current_date} - {current_time} ")
# Timer for updating labels
timer = QTimer()
timer.timeout.connect(update_labels)
timer.start(1000) # Update every second
# Create main UI components
main_window = OfflineHostMonitor(None, None, None, None)
tab_widget.addTab(main_window, "Monitor")
return_online_widget = ReturnOnlineWidget()
tab_widget.addTab(return_online_widget, "Return Online")
main_window.sound_manager = sound_manager # Attach the sound manager to the main window
settings_widget = SettingsWidget(None, None, None, None, None, tab_widget, main_window)
tab_widget.addTab(settings_widget, "Settings")
update_log_widget = UpdateLogWidget()
tab_widget.addTab(update_log_widget, "Update Log")
main_window.settings_widget = settings_widget
observer = monitor_hostname_file(main_window)
# Main event loop
try:
main_window.show()
sys.exit(app.exec_())
finally:
observer.stop()
observer.join()
# Prepare application exit cleanup
def cleanup():
stop_event.set() # Signal all threads to stop
for thread in active_threads:
thread.join() # Wait for threads to finish
# Connect cleanup on application exit
app.aboutToQuit.connect(cleanup)
# Queue and threading setup
queue = Queue()
stop_event = threading.Event()
active_threads = []
cpu_count = psutil.cpu_count(logical=True)
memory = psutil.virtual_memory()
available_memory_gb = memory.available / (1024 ** 2)
max_pings_based_on_memory = int(available_memory_gb / 0.1)
MAX_PINGS = min(MAX_PINGS, max_pings_based_on_memory)
semaphore = threading.Semaphore(MAX_PINGS)
# Assign globals to widgets
main_window.queue = queue
main_window.offline_dict = offline_dict
main_window.stop_event = stop_event
main_window.semaphore = semaphore
settings_widget.queue = queue
settings_widget.offline_dict = offline_dict
settings_widget.stop_event = stop_event
settings_widget.semaphore = semaphore
settings_widget.host_info_list = host_info_list
return_online_signal = return_online_widget.dataUpdated
# Start monitoring threads for each host
for host_info in host_info_list:
thread = threading.Thread(
target=monitor_host,
args=(host_info, queue, offline_dict, stop_event, semaphore, main_window)
)
thread.start()
active_threads.append(thread)
# Start thread to handle offline list updates
update_thread = threading.Thread(
target=update_offline_list,
args=(queue, offline_dict, stop_event, return_online_signal, settings_widget, main_window),
daemon=True
)
update_thread.start()
active_threads.append(update_thread)
# Show main widget
main_widget.show()
# Ensure state is saved on app quit
app.aboutToQuit.connect(lambda: save_offline_state(offline_state_file, offline_dict))
sys.exit(app.exec_())
if __name__ == "__main__":
main()
```
### Changes Made:
- Moved the `host_info_list`, `host_info_dict`, and `offline_dict` initialization before creating the UI, so they are accessible to the `update_labels()` function.
- Defined `host_info_dict` before using it in `update_labels`.
- Adjusted comments and ensured consistent handling of program logic.
Now `host_info_dict` is properly initialized and scoped so that it's available for the `update_labels()` function to use.