auth.py v1.0
[16:30:58] (%YnM-AI): 
YnM Media 
: @markus [■■■□□□□□□□] 30%
[16:31:00] (%YnM-AI): Kliens: YnM Web, Eszköz: Chrome, IP: 192.168.0.229, Dátum: 2024-12-27 16:30
[16:31:08] (%YnM-AI): YnM Media : @markus [■■■■■□□□□□] 50%
[16:31:10] (%YnM-AI): Kliens: YnM Web, Eszköz: Chrome, IP: 192.168.0.229, Dátum: 2024-12-27 16:31
[16:31:18] (%YnM-AI): YnM Media : @markus[■■■■■■■■■■] 100%nem sikerült belépni probléma megoldásáig. Belépési hibát nem fogok mutatni.
################################################
Jellyfin Webhook
{
"extras": {
"client::display": {
"contentType": "text/markdown"
}
},
"title": "🚫 Hitelesítési hiba 🚫",
"message": "\n\n**User:** {{Username}}\n\n**Client:** {{App}}\n\n**Device:** {{DeviceName}} \n\n**Client IP:** {{RemoteEndPoint}}",
"priority": {{Priority}}
}
################################################
auth.js
from datetime import datetime
from sopel import plugin
import requests
import re
import os
import html
import time
import json
from datetime import timedelta
# Gotify API beállításai
GOTIFY_URL = "http://192.168.0.150:81/message"
GOTIFY_CLIENT_TOKEN = "*******************************"
# App ID-k (csak az appid 2 üzeneteit olvassuk be)
APP_IDS = {2}
# A fájl, ahol a felhasználói adatokat tároljuk
USER_ERRORS_FILE = "/home/ai/.sopel/plugins/notify/user_errors.json"
def load_user_errors():
"""Betölti a felhasználói hibákat a JSON fájlból, ha nem található, új fájlt hoz létre."""
try:
with open(USER_ERRORS_FILE, 'r') as f:
return json.load(f)
except (FileNotFoundError, json.JSONDecodeError):
# Ha nem található a fájl, hozunk létre egy új, üres fájlt
save_user_errors({})
return {}
def save_user_errors(user_errors):
"""Elmenti a felhasználói hibákat a JSON fájlba."""
with open(USER_ERRORS_FILE, 'w') as f:
json.dump(user_errors, f, indent=4)
def load_last_message_id():
"""Betölti az utolsó üzenet ID-ját a fájlból."""
try:
with open('/home/ai/.sopel/plugins/notify/last_message_id.json', 'r') as f:
return json.load(f).get('last_message_id', None)
except (FileNotFoundError, json.JSONDecodeError):
return None
def save_last_message_id(last_message_id):
"""Elmenti az utolsó üzenet ID-ját a fájlba."""
with open('/home/ai/.sopel/plugins/notify/last_message_id.json', 'w') as f:
json.dump({'last_message_id': last_message_id}, f, indent=4)
def get_gotify_messages():
"""Lekéri a Gotify üzeneteket és szűri őket az appid alapján."""
response = requests.get(f"{GOTIFY_URL}?token={GOTIFY_CLIENT_TOKEN}")
if response.status_code == 200:
data = response.json()
# Szűrés App ID alapján
return [msg for msg in data.get('messages', []) if msg['appid'] in APP_IDS]
return []
def decode_html_entities(text):
"""Dekódolja a HTML entitásokat a szövegből."""
return html.unescape(text)
def is_user_error_recent(user, last_error_time):
"""Ellenőrzi, hogy a felhasználó hibaüzenete nem régebbi, mint 30 perc."""
now = datetime.now()
time_diff = now - datetime.strptime(last_error_time, "%Y-%m-%d %H:%M:%S")
return time_diff < timedelta(minutes=30)
def format_full_message(message, date_now, user_errors):
"""Formázza az üzenetet és kezeli a hibaüzeneteket."""
try:
cleaned_message = re.sub(r'\*\*(.*?)\*\*', r'\1', message['message']) # Markdown eltávolítása
cleaned_message = re.sub(r'[*_]{1,2}', '', cleaned_message) # További Markdown szimbólumok eltávolítása
# Kinyerjük az adatokat
user_text = re.search(r"User: (.+)", cleaned_message)
user_text = user_text.group(1).strip() if user_text else "N/A"
if user_text in user_errors and user_errors[user_text]["error_count"] >= 3:
return None # Ha a felhasználónak már 3 hibája van, nem küldünk semmit
client_text = re.search(r"Client: (.+)", cleaned_message)
client_text = client_text.group(1).strip() if client_text else "N/A"
device_text = re.search(r"Device: (.+)", cleaned_message)
device_text = device_text.group(1).strip() if device_text else "N/A"
client_ip_text = re.search(r"Client IP: (.+)", cleaned_message)
client_ip_text = client_ip_text.group(1).strip() if client_ip_text else "N/A"
# Csak akkor cseréljük le "YnM"-ra, ha a kliens neve tartalmazza a "Jellyfin"-t
if "Jellyfin" in client_text:
client_text = client_text.replace("Jellyfin", "YnM")
# HTML entitások dekódolása
user_text = decode_html_entities(user_text)
client_text = decode_html_entities(client_text)
device_text = decode_html_entities(device_text)
client_ip_text = decode_html_entities(client_ip_text)
# Hibaszámokhoz megfelelő üzenetek
error_count = user_errors.get(user_text, {}).get("error_count", 0) + 1
if error_count == 1:
message_1 = f"⚠️ YnM Media 🔐 ⚠️ 👤: @{user_text} [■■■□□□□□□□] 30%"
elif error_count == 2:
message_1 = f"⚠️ YnM Media 🔐 ⚠️ 👤: @{user_text} [■■■■■□□□□□] 50%"
elif error_count >= 3:
message_1 = f"⚠️ YnM Media 🔐 ⚠️ 👤: @{user_text}[■■■■■■■■■■] 100%"
message_3 = "Nem sikerült belépni, a probléma megoldásáig a belépés tiltásra került. Keresd meg @Markus-t!"
user_errors[user_text] = {"last_error_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "error_count": error_count}
save_user_errors(user_errors)
# Ha a hiba száma eléri a 3-at, akkor nem küldünk más üzenetet.
return message_1, "", message_3 # Üzenet 2 és 3 elhagyva
# Adatok frissítése
user_errors[user_text] = {"last_error_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "error_count": error_count}
save_user_errors(user_errors)
message_2 = (f"Kliens: {client_text}, Eszköz: {device_text}, IP: {client_ip_text}, Dátum: {date_now}")
return message_1, message_2, ""
except (KeyError, IndexError):
return f"Hibás üzenet formátum: {message['message']}", "", ""
last_message_id = load_last_message_id() or None # Ha nincs a fájlban, alapértelmezett None érték
@plugin.interval(10) # Minden 10 másodpercben fut
# Globális változó az utolsó üzenet ID tárolására
def gotify_reader(bot):
"""Lekéri a Gotify üzeneteket és elküldi őket IRC csatornára."""
global last_message_id
user_errors = load_user_errors() # Betöltjük a felhasználói hibákat
date_now = datetime.now().strftime('%Y-%m-%d %H:%M')
messages = get_gotify_messages()
if messages:
for message in messages:
# Csak új üzeneteket küldünk, amiket még nem láttunk
if last_message_id is None or message['id'] > last_message_id:
result = format_full_message(message, date_now, user_errors)
# Ha None-ot kapunk, akkor nem küldünk üzenetet
if result is None:
continue
message_1, message_2, message_3 = result
# Ellenőrizzük, hogy a felhasználó nem ért el 3 hibát
user_text = re.search(r"User: (.+)", message['message'])
if user_text:
user_text = user_text.group(1).strip()
if user_text in user_errors and user_errors[user_text]["error_count"] >= 3:
return None # Ha a felhasználónak már 3 hibája van, nem küldjük el az üzenetet
if message_1:
bot.say(message_1, '#Magyar') # Küldjük az első üzenetet
# 2 másodperces késleltetés a második üzenet előtt
time.sleep(2)
if message_2:
bot.say(message_2, '#Magyar') # Küldjük a második üzenetet
# Ha van harmadik üzenet, azt is küldjük
if message_3:
bot.say(message_3, '#Magyar')
last_message_id = message['id']
save_last_message_id(last_message_id)
def clear_user_login(user_text, user_errors):
"""Törli a felhasználó hibáit, csak az 'error_count' értékét állítja 0-ra."""
if user_text in user_errors:
# Csak az 'error_count' értékét állítjuk 0-ra
user_errors[user_text]["error_count"] = 0
save_user_errors(user_errors) # Elmentjük a módosított hibákat
return f"A {user_text} felhasználó hibaszámlálója [□□□□□□□□□□] 0%"
else:
return f"Nem található a {user_text} felhasználó hibái között."
@plugin.rule(r'\[Markus\]!clearlogin(?: (\w+))?')
@plugin.command('clearlogin')
def clear_login_command(bot, trigger):
"""Parancsot adhatunk meg a hibaszámláló törlésére."""
user_text = trigger.group(1) # Első csoport: felhasználó neve
if user_text:
user_text = user_text.strip() # Levágjuk a szóközöket
user_errors = load_user_errors() # Betöltjük a felhasználói hibákat
result = clear_user_login(user_text, user_errors)
bot.say(result, '#Magyar') # Küldjük a válasz üzenetet a csatornára
else:
bot.say("Hibás parancs. Használat: .clearlogin <felhasználó>", '#Magyar')