151 lines
4.7 KiB
Python
151 lines
4.7 KiB
Python
from pathlib import Path
|
|
|
|
import requests
|
|
import time
|
|
from dataclasses import dataclass
|
|
from typing import Generator, Self
|
|
|
|
|
|
DISCORD_WEBHOOK_URL = "HOOOOOOOOOOOOOOOOK"
|
|
LOG_FILE_PATH = Path(r"E:\RSI\StarCitizen\LIVE\Game.log")
|
|
INCLUDE_NPC = False
|
|
ONLY_PERSONAL_FEED = True
|
|
|
|
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,
|
|
):
|
|
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
|
|
|
|
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():
|
|
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)
|
|
|
|
if '_NPC_' in kill.killer + kill.victim and not INCLUDE_NPC:
|
|
continue
|
|
|
|
if self.nickname not in (kill.killer, kill.victim) and ONLY_PERSONAL_FEED is True:
|
|
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)
|
|
|
|
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
|
|
with self.log_file_path.open("r", encoding="utf-8") as file:
|
|
while True:
|
|
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)
|
|
|
|
|
|
def main():
|
|
try:
|
|
if DEBUG is True:
|
|
app = App(Path("./Game.log"), DISCORD_WEBHOOK_URL, do_mock_discord=True, do_backread=True)
|
|
|
|
else:
|
|
app = App(LOG_FILE_PATH, DISCORD_WEBHOOK_URL)
|
|
|
|
app.process_lines()
|
|
|
|
except KeyboardInterrupt:
|
|
print("\nMonitoring stopped")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|