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()
|