gru/skrypty_narzedzia
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Update tvheadend_advenced.py

Browse Source
This commit is contained in:
gru
2025-06-18 23:09:50 +02:00
parent 2b32b51f70
commit 2b22c13af4
1 changed files with 32 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files

61
tvheadend_advenced.py
View File

@ -16,7 +16,6 @@ def get_input_status(host, port, user, password, input_uuid, timeout):
except Exception as e: except Exception as e:
print(f"CRITICAL - API returned invalid JSON: {e} | Response: {r.text}") print(f"CRITICAL - API returned invalid JSON: {e} | Response: {r.text}")
sys.exit(2) sys.exit(2)
# Obsługa odpowiedzi jako dict z kluczem 'entries'
if isinstance(data, dict) and "entries" in data: if isinstance(data, dict) and "entries" in data:
entries = data["entries"] entries = data["entries"]
elif isinstance(data, list): elif isinstance(data, list):
@ -33,26 +32,7 @@ def get_input_status(host, port, user, password, input_uuid, timeout):
print(f"CRITICAL - API error: {e}") print(f"CRITICAL - API error: {e}")
sys.exit(2) sys.exit(2)
def main(): def interpret_status(status, args):
parser = argparse.ArgumentParser(description="Nagios plugin for TVHeadend DVB-T advanced monitoring (Digest Auth)")
parser.add_argument("--host", required=True)
parser.add_argument("--port", type=int, default=9981)
parser.add_argument("--user", required=True)
parser.add_argument("--password", required=True)
parser.add_argument("--input_uuid", required=True, help="UUID tunera lub muxa (sprawdź przez API TVHeadend)")
parser.add_argument("--timeout", type=int, default=10)
parser.add_argument("--warning_ber", type=float, default=1e-5)
parser.add_argument("--critical_ber", type=float, default=1e-4)
parser.add_argument("--warning_unc", type=int, default=10)
parser.add_argument("--critical_unc", type=int, default=100)
args = parser.parse_args()
status = get_input_status(args.host, args.port, args.user, args.password, args.input_uuid, args.timeout)
if not status:
print("CRITICAL - Input (tuner/mux) not found or API error")
sys.exit(2)
# Pobieranie parametrów
signal = status.get("signal") signal = status.get("signal")
snr = status.get("snr") snr = status.get("snr")
ber = status.get("ber") ber = status.get("ber")
@ -62,7 +42,6 @@ def main():
unc = status.get("unc") unc = status.get("unc")
cc = status.get("cc") cc = status.get("cc")
# Ustalanie statusu
exit_code = 0 exit_code = 0
messages = [] messages = []
@ -84,8 +63,6 @@ def main():
messages.append(f"WARNING UNC={unc}") messages.append(f"WARNING UNC={unc}")
else: else:
messages.append(f"OK UNC={unc}") messages.append(f"OK UNC={unc}")
# Dodaj inne parametry do komunikatu
messages.append(f"Signal={signal}") messages.append(f"Signal={signal}")
messages.append(f"SNR={snr}") messages.append(f"SNR={snr}")
if mer is not None: if mer is not None:
@ -96,11 +73,37 @@ def main():
messages.append(f"EC_BLOCK={ec_block}") messages.append(f"EC_BLOCK={ec_block}")
if cc is not None: if cc is not None:
messages.append(f"CC={cc}") messages.append(f"CC={cc}")
return exit_code, messages
# Komunikat końcowy def main():
status_text = ["OK", "WARNING", "CRITICAL"][exit_code] parser = argparse.ArgumentParser(description="Nagios plugin for TVHeadend DVB-T advanced monitoring (Digest Auth, multi-card, optional second card)")
print(f"{status_text} - " + " | ".join(messages)) parser.add_argument("--host", required=True)
sys.exit(exit_code) parser.add_argument("--port", type=int, default=9981)
parser.add_argument("--user", required=True)
parser.add_argument("--password", required=True)
parser.add_argument("--input_uuid1", required=True, help="UUID tunera/muxa 1")
parser.add_argument("--input_uuid2", required=False, help="UUID tunera/muxa 2 (opcjonalnie)")
parser.add_argument("--timeout", type=int, default=10)
parser.add_argument("--warning_ber", type=float, default=1e-5)
parser.add_argument("--critical_ber", type=float, default=1e-4)
parser.add_argument("--warning_unc", type=int, default=10)
parser.add_argument("--critical_unc", type=int, default=100)
args = parser.parse_args()
status1 = get_input_status(args.host, args.port, args.user, args.password, args.input_uuid1, args.timeout)
code1, msg1 = interpret_status(status1, args)
if args.input_uuid2:
status2 = get_input_status(args.host, args.port, args.user, args.password, args.input_uuid2, args.timeout)
code2, msg2 = interpret_status(status2, args)
final_code = max(code1, code2)
status_text = ["OK", "WARNING", "CRITICAL"][final_code]
print(f"{status_text} - Karta1: {' | '.join(msg1)} || Karta2: {' | '.join(msg2)}")
sys.exit(final_code)
else:
status_text = ["OK", "WARNING", "CRITICAL"][code1]
print(f"{status_text} - Karta1: {' | '.join(msg1)}")
sys.exit(code1)
if __name__ == "__main__": if __name__ == "__main__":
main() main()