379 lines
13 KiB
Python
379 lines
13 KiB
Python
from pathlib import Path
|
||
import threading
|
||
import tkinter as tk
|
||
from tkinter import ttk, filedialog, scrolledtext
|
||
import queue
|
||
import requests
|
||
import time
|
||
from dataclasses import dataclass
|
||
from typing import Generator, Self
|
||
|
||
|
||
# Default values
|
||
DEFAULT_DISCORD_WEBHOOK_URL = "HOOOOOOOOOOOOOOOOK"
|
||
DEFAULT_LOG_FILE_PATH = Path(r"E:\RSI\StarCitizen\LIVE\Game.log")
|
||
DEBUG = False
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class Line:
|
||
line: str
|
||
is_realtime: bool
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class Kill:
|
||
killer: str
|
||
victim: str
|
||
zone: str
|
||
killed_by: str
|
||
damage_type: str
|
||
|
||
@staticmethod
|
||
def from_line(line: str) -> "Self":
|
||
# <2025-05-23T16:48:53.672Z> [Notice] <Actor Death> CActor::Kill: 'Neon0ne' [1895570109956] in zone 'ANVL_Ballista_3700492224326' killed by 'Aleksey31' [1978962253870] using 'MRCK_S07_ANVL_Ballista_Duel_3700492224372' [Class unknown] with damage type 'VehicleDestruction' from direction x: 0.000000, y: 0.000000, z: 0.000000 [Team_ActorTech][Actor]
|
||
splited = [i.strip("'") for i in line.split()]
|
||
return Kill(
|
||
victim=splited[5],
|
||
zone=splited[9],
|
||
killer=splited[12],
|
||
killed_by=splited[15],
|
||
damage_type=splited[21],
|
||
)
|
||
|
||
|
||
class App:
|
||
def __init__(
|
||
self,
|
||
log_file_path: Path,
|
||
discord_webhook_url: str,
|
||
do_backread: bool = False,
|
||
do_mock_discord: bool = False,
|
||
hide_npc_kills: bool = False,
|
||
kill_callback=None,
|
||
):
|
||
self.log_file_path = log_file_path
|
||
self.do_backread: bool = do_backread
|
||
self.do_mock_discord: bool = do_mock_discord
|
||
self.discord_webhook_url: str = discord_webhook_url
|
||
self.nickname: str | None = None
|
||
self.kill_callback = kill_callback
|
||
self.hide_npc_kills = hide_npc_kills
|
||
self.running = True
|
||
|
||
if not (discord_webhook_url.startswith("https://") or do_mock_discord):
|
||
msg = f"{discord_webhook_url} не вебхук, так то"
|
||
print(msg)
|
||
raise RuntimeError(msg)
|
||
|
||
def try_set_self_name(self, line: str) -> None:
|
||
# <2025-05-23T16:19:10.244Z> [Notice] <AccountLoginCharacterStatus_Character> Character: createdAt 1747220301029 - updatedAt 1747220301703 - geid 1978962253870 - accountId 2261831 - name Aleksey31 - state STATE_CURRENT [Team_GameServices][Login]
|
||
if self.nickname is not None:
|
||
print("Warning, AccountLoginCharacterStatus_Character duplicate", line)
|
||
return
|
||
|
||
try:
|
||
self.nickname = line.split()[17]
|
||
print(f"Extracted {self.nickname=}")
|
||
|
||
except IndexError:
|
||
print("Failed to extract nickname from line", line)
|
||
|
||
def process_lines(self) -> None:
|
||
for line in self.monitor_log_file():
|
||
if not self.running:
|
||
break
|
||
|
||
process_important_realtime_events = self.do_backread or line.is_realtime
|
||
|
||
if "<AccountLoginCharacterStatus_Character>" in line.line:
|
||
self.try_set_self_name(line.line)
|
||
continue
|
||
|
||
if (
|
||
"[Notice] <Actor Death> CActor::Kill:" in line.line
|
||
and process_important_realtime_events
|
||
):
|
||
kill = Kill.from_line(line.line)
|
||
# Skip processing if this is an NPC kill and hide_npc_kills is enabled
|
||
if self.hide_npc_kills and "_NPC_" in kill.victim:
|
||
continue
|
||
self.send_kill(kill)
|
||
|
||
def send_kill(self, kill: Kill) -> None:
|
||
if kill.victim == kill.killer:
|
||
msg = f"🤦 **{kill.victim}** РКН"
|
||
else:
|
||
msg = f"💀 **{kill.victim}** killed by 🔫 **{kill.killer}**"
|
||
|
||
self.send(msg)
|
||
|
||
# Call the callback if provided
|
||
if self.kill_callback:
|
||
self.kill_callback(kill, msg)
|
||
|
||
def send(self, message: str) -> None:
|
||
try:
|
||
if self.do_mock_discord:
|
||
print("Sending message to discord", message)
|
||
else:
|
||
requests.post(self.discord_webhook_url, json={"content": message})
|
||
|
||
except Exception as e:
|
||
print(f"Error sending to Discord: {e}")
|
||
|
||
def monitor_log_file(self) -> Generator[Line, None, None]:
|
||
is_realtime = False
|
||
try:
|
||
with self.log_file_path.open("r", encoding="utf-8") as file:
|
||
while self.running:
|
||
current_position = file.tell()
|
||
line = file.readline()
|
||
|
||
if not line:
|
||
is_realtime = True
|
||
time.sleep(5)
|
||
file.seek(current_position)
|
||
if DEBUG is True:
|
||
return
|
||
|
||
line = line.removesuffix("\n")
|
||
yield Line(line, is_realtime)
|
||
|
||
except FileNotFoundError:
|
||
print(f"Log file not found: {self.log_file_path}")
|
||
return
|
||
|
||
except Exception as e:
|
||
print(f"Error monitoring log file: {e}")
|
||
return
|
||
|
||
def stop(self):
|
||
self.running = False
|
||
|
||
|
||
class DeathLogGUI:
|
||
def __init__(self, root):
|
||
self.root = root
|
||
self.root.title("Star Citizen Death Log Monitor")
|
||
self.root.geometry("800x600")
|
||
self.root.minsize(600, 400)
|
||
|
||
self.log_path = tk.StringVar(value=str(DEFAULT_LOG_FILE_PATH))
|
||
self.webhook_url = tk.StringVar(value=DEFAULT_DISCORD_WEBHOOK_URL)
|
||
self.do_backread = tk.BooleanVar(value=False)
|
||
self.do_mock_discord = tk.BooleanVar(value=False)
|
||
self.hide_npc_kills = tk.BooleanVar(value=True)
|
||
|
||
self.app = None
|
||
self.monitor_thread = None
|
||
self.kill_queue = queue.Queue()
|
||
self.recent_kills: list[str] = []
|
||
self.max_kills_to_display = 100
|
||
|
||
self.create_widgets()
|
||
self.update_kill_display()
|
||
|
||
def create_widgets(self):
|
||
# Create main frame
|
||
main_frame = ttk.Frame(self.root, padding="10")
|
||
main_frame.pack(fill=tk.BOTH, expand=True)
|
||
|
||
# Configuration section
|
||
config_frame = ttk.LabelFrame(main_frame, text="Configuration", padding="10")
|
||
config_frame.pack(fill=tk.X, pady=5)
|
||
|
||
# Log file path
|
||
ttk.Label(config_frame, text="Log File Path:").grid(
|
||
row=0, column=0, sticky=tk.W, pady=2
|
||
)
|
||
ttk.Entry(config_frame, textvariable=self.log_path, width=50).grid(
|
||
row=0, column=1, sticky=tk.W + tk.E, pady=2, padx=5
|
||
)
|
||
ttk.Button(config_frame, text="Browse", command=self.browse_log_file).grid(
|
||
row=0, column=2, pady=2
|
||
)
|
||
|
||
# Discord webhook URL
|
||
ttk.Label(config_frame, text="Discord Webhook URL:").grid(
|
||
row=1, column=0, sticky=tk.W, pady=2
|
||
)
|
||
ttk.Entry(config_frame, textvariable=self.webhook_url, width=50).grid(
|
||
row=1, column=1, sticky=tk.W + tk.E, pady=2, padx=5
|
||
)
|
||
|
||
# Checkboxes
|
||
ttk.Checkbutton(
|
||
config_frame, text="Process existing log entries", variable=self.do_backread
|
||
).grid(row=2, column=0, columnspan=2, sticky=tk.W, pady=2)
|
||
ttk.Checkbutton(
|
||
config_frame,
|
||
text="Mock Discord (don't send actual messages)",
|
||
variable=self.do_mock_discord,
|
||
).grid(row=3, column=0, columnspan=2, sticky=tk.W, pady=2)
|
||
ttk.Checkbutton(
|
||
config_frame,
|
||
text="Hide NPC kills (_NPC_ in name)",
|
||
variable=self.hide_npc_kills,
|
||
).grid(row=4, column=0, columnspan=2, sticky=tk.W, pady=2)
|
||
|
||
# Control buttons
|
||
control_frame = ttk.Frame(main_frame)
|
||
control_frame.pack(fill=tk.X, pady=5)
|
||
|
||
self.start_button = ttk.Button(
|
||
control_frame, text="Start Monitoring", command=self.start_monitoring
|
||
)
|
||
self.start_button.pack(side=tk.LEFT, padx=5)
|
||
|
||
self.stop_button = ttk.Button(
|
||
control_frame,
|
||
text="Stop Monitoring",
|
||
command=self.stop_monitoring,
|
||
state=tk.DISABLED,
|
||
)
|
||
self.stop_button.pack(side=tk.LEFT, padx=5)
|
||
|
||
# Status indicator
|
||
self.status_var = tk.StringVar(value="Not monitoring")
|
||
status_frame = ttk.Frame(main_frame)
|
||
status_frame.pack(fill=tk.X, pady=5)
|
||
ttk.Label(status_frame, text="Status:").pack(side=tk.LEFT)
|
||
self.status_label = ttk.Label(
|
||
status_frame, textvariable=self.status_var, foreground="red"
|
||
)
|
||
self.status_label.pack(side=tk.LEFT, padx=5)
|
||
|
||
# Kill events display
|
||
kills_frame = ttk.LabelFrame(
|
||
main_frame, text="Recent Kill Events", padding="10"
|
||
)
|
||
kills_frame.pack(fill=tk.BOTH, expand=True, pady=5)
|
||
|
||
self.kill_display = scrolledtext.ScrolledText(
|
||
kills_frame, wrap=tk.WORD, height=10
|
||
)
|
||
self.kill_display.pack(fill=tk.BOTH, expand=True)
|
||
self.kill_display.config(state=tk.DISABLED)
|
||
|
||
def browse_log_file(self):
|
||
filename = filedialog.askopenfilename(
|
||
title="Select Star Citizen Log File",
|
||
filetypes=[("Log Files", "*.log"), ("All Files", "*.*")],
|
||
)
|
||
if filename:
|
||
self.log_path.set(filename)
|
||
|
||
def start_monitoring(self):
|
||
if self.monitor_thread and self.monitor_thread.is_alive():
|
||
return
|
||
|
||
try:
|
||
log_path = Path(self.log_path.get())
|
||
webhook_url = self.webhook_url.get()
|
||
|
||
self.app = App(
|
||
log_file_path=log_path,
|
||
discord_webhook_url=webhook_url,
|
||
do_backread=self.do_backread.get(),
|
||
do_mock_discord=self.do_mock_discord.get(),
|
||
hide_npc_kills=self.hide_npc_kills.get(),
|
||
kill_callback=self.on_kill,
|
||
)
|
||
|
||
self.monitor_thread = threading.Thread(
|
||
target=self.app.process_lines, daemon=True
|
||
)
|
||
self.monitor_thread.start()
|
||
|
||
self.status_var.set("Monitoring active")
|
||
self.status_label.config(foreground="green")
|
||
self.start_button.config(state=tk.DISABLED)
|
||
self.stop_button.config(state=tk.NORMAL)
|
||
|
||
except Exception as e:
|
||
self.status_var.set(f"Error: {str(e)}")
|
||
self.status_label.config(foreground="red")
|
||
|
||
def stop_monitoring(self):
|
||
if self.app:
|
||
self.app.stop()
|
||
|
||
if self.monitor_thread:
|
||
self.monitor_thread.join(timeout=1.0)
|
||
|
||
self.status_var.set("Monitoring stopped")
|
||
self.status_label.config(foreground="red")
|
||
self.start_button.config(state=tk.NORMAL)
|
||
self.stop_button.config(state=tk.DISABLED)
|
||
|
||
def on_kill(self, kill: Kill, message: str):
|
||
# No need to filter here as it's already filtered in the App class
|
||
self.kill_queue.put((kill, message))
|
||
|
||
def update_kill_display(self):
|
||
while not self.kill_queue.empty():
|
||
try:
|
||
kill, message = self.kill_queue.get_nowait()
|
||
timestamp = time.strftime("%Y-%m-%d %H:%M:%S")
|
||
# Simplified message with only nicknames and emojis
|
||
if kill.victim == kill.killer:
|
||
formatted_message = f"[{timestamp}] 🤦 {kill.victim} (self-kill)\n"
|
||
else:
|
||
formatted_message = (
|
||
f"[{timestamp}] 💀 {kill.victim} killed by 🔫 {kill.killer}\n"
|
||
)
|
||
|
||
self.recent_kills.append(formatted_message)
|
||
if len(self.recent_kills) > self.max_kills_to_display:
|
||
self.recent_kills.pop(0)
|
||
|
||
self.kill_display.config(state=tk.NORMAL)
|
||
self.kill_display.delete(1.0, tk.END)
|
||
for msg in self.recent_kills:
|
||
self.kill_display.insert(tk.END, msg)
|
||
self.kill_display.config(state=tk.DISABLED)
|
||
self.kill_display.see(tk.END)
|
||
except queue.Empty:
|
||
break
|
||
|
||
self.root.after(100, self.update_kill_display)
|
||
|
||
|
||
def main():
|
||
try:
|
||
# Check if we should run in GUI mode or CLI mode
|
||
import sys
|
||
|
||
if len(sys.argv) > 1 and sys.argv[1] == "--cli":
|
||
# CLI mode
|
||
if DEBUG is True:
|
||
app = App(
|
||
Path("./Game.log"),
|
||
DEFAULT_DISCORD_WEBHOOK_URL,
|
||
do_mock_discord=True,
|
||
do_backread=True,
|
||
hide_npc_kills=False,
|
||
)
|
||
|
||
else:
|
||
app = App(
|
||
DEFAULT_LOG_FILE_PATH,
|
||
DEFAULT_DISCORD_WEBHOOK_URL,
|
||
hide_npc_kills=False,
|
||
)
|
||
app.process_lines()
|
||
|
||
else:
|
||
# GUI mode
|
||
root = tk.Tk()
|
||
app = DeathLogGUI(root)
|
||
root.mainloop()
|
||
|
||
except KeyboardInterrupt:
|
||
print("\nMonitoring stopped")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|