mirror of
https://github.com/norohind/EDMC-FCTracker.git
synced 2025-04-19 22:17:35 +03:00
427 lines
17 KiB
Python
427 lines
17 KiB
Python
# -*- coding: utf-8 -*-
|
||
import os
|
||
import requests
|
||
import threading
|
||
import json
|
||
import logging
|
||
import sys
|
||
|
||
from config import appname
|
||
import tkinter as tk
|
||
import myNotebook as nb
|
||
from config import config
|
||
from typing import Optional
|
||
|
||
# GPLv3, wrote by a31 aka norohind aka CMDR Aleksey31
|
||
# contact: a31#6403 (discord), a31@demb.design (email)
|
||
|
||
|
||
VERSION = "0.0.6-post"
|
||
force_beta = False # override Plugin_settings.SEND_IN_BETA
|
||
|
||
|
||
plugin_name = os.path.basename(os.path.dirname(__file__))
|
||
logger = logging.getLogger(f'{appname}.{plugin_name}')
|
||
|
||
this = sys.modules[__name__] # For holding module globals, thanks to edsm.py
|
||
|
||
|
||
class Messages:
|
||
"""Class that contains all using messages text"""
|
||
|
||
# take color in HEX and turn it into decimal
|
||
COLOR_JUMP_REQUEST = "1127128"
|
||
COLOR_JUMP_CANCELLED = "14177041"
|
||
COLOR_JUMP = "130816"
|
||
COLOR_PERMISSION_CHANGE = "5261068"
|
||
COLOR_CHANGE_NAME = "9355388"
|
||
|
||
TITLE_JUMP_REQUEST = "Запланирован прыжок"
|
||
TITLE_JUMP = "Прыжок совершён"
|
||
TITLE_JUMP_CANCELLED = "Прыжок отменён"
|
||
TITLE_CHANGE_DOCKING_PERMISSION = "Изменение разрешения на стыковку"
|
||
TITLE_IN_BETA = "Бета версия игры"
|
||
TITLE_CHANGE_NAME = "Изменение имени носителя"
|
||
|
||
TEXT_JUMP_REQUEST_BODY_DIFF = " к телу {body}"
|
||
TEXT_JUMP_REQUEST = "Запланирован прыжок носителя {name} в систему {system}"
|
||
TEXT_JUMP = "Носитель {name} совершил прыжок в систему {system}"
|
||
TEXT_JUMP_CANCELLED = "Прыжок носителя {name} отменён"
|
||
TEXT_PERMISSION_CHANGE = """Носитель {name} сменил разрешение на стыковку с {old_permission} на {new_permission}
|
||
Стыковка для преступников была {old_doc_for_crime}, сейчас {new_doc_for_crime}"""
|
||
TEXT_IN_BETA = "Внимание, данное сообщение относится только к бета версии игры!"
|
||
TEXT_CHANGE_NAME = "Имя носителя изменилось с {old_name} на {new_name}"
|
||
|
||
DOC_PERMISSION_ALL = "Для всех"
|
||
DOC_PERMISSION_NONE = "Никто"
|
||
DOC_PERMISSION_FRIENDS = "Только друзья"
|
||
DOC_PERMISSION_SQUADRON = "Только члены эскадрильи"
|
||
DOC_PERMISSION_FRIENDS_SQUADRON = "Только друзья и члены эскадрильи"
|
||
DOC_PERMISSION_ALLOW_NOTORIUS = "Разрешена"
|
||
DOC_PERMISSION_DISALLOW_NOTORIUS = "Запрещена"
|
||
|
||
|
||
class Carrier:
|
||
def __init__(self):
|
||
self.name = None
|
||
self.callsign = None
|
||
self.location = None
|
||
self.cID = None # CarrierID (aka MarketID)
|
||
self.docking_permission = None
|
||
self.allow_notorius = None
|
||
self.owner = None
|
||
|
||
|
||
class Embed:
|
||
"""Building completed and ready for send embed message for discord. Requires json lib"""
|
||
|
||
def __init__(self, **kwargs): # color, title, description, username
|
||
self.items = list()
|
||
|
||
if kwargs.get('username') is not None: # we can pass only username
|
||
self.username = kwargs.get('username')
|
||
kwargs.pop('username')
|
||
else:
|
||
self.username = None
|
||
|
||
if len(kwargs) == 0: # we can create new object without creating an item, just do not pass anything (
|
||
# exception is 'username') to constructor
|
||
return
|
||
|
||
self.add_item(**kwargs)
|
||
|
||
def add_item(self, **kwargs):
|
||
color = kwargs.get('color')
|
||
title = kwargs.get('title')
|
||
description = kwargs.get('description')
|
||
self.items.append(dict(title=title, color=color, description=description))
|
||
return len(self.items) - 1 # index of added item
|
||
|
||
def get_message(self):
|
||
"""Get full and ready for sending message"""
|
||
|
||
if self.username is not None:
|
||
return json.dumps(dict(username=self.username, embeds=self.items))
|
||
else:
|
||
return json.dumps(dict(embeds=self.items))
|
||
|
||
def __str__(self):
|
||
return str(self.get_message())
|
||
|
||
def update_item(self, item, key, new_value):
|
||
"""Replace value under 'key' in 'item'"""
|
||
self.items[item][key] = new_value
|
||
|
||
def concatenate_item(self, item, key, new_value):
|
||
"""Add to existing value new part"""
|
||
self.items[item][key] = self.items[item][key] + new_value
|
||
|
||
def set_username(self, username):
|
||
"""Will override current webhook username, for reset call this func with None"""
|
||
self.username = username
|
||
|
||
|
||
class Messages_sender(threading.Thread):
|
||
"""Sending embed message to discord "asynchronously" """
|
||
|
||
def __init__(self, embed, urls):
|
||
threading.Thread.__init__(self)
|
||
self.message = embed
|
||
self.urls = urls
|
||
self.start()
|
||
|
||
def run(self):
|
||
|
||
if isinstance(self.urls, list):
|
||
|
||
for url in self.urls:
|
||
|
||
if isinstance(url, str):
|
||
|
||
if url.startswith('https://'):
|
||
self.send(url)
|
||
else:
|
||
logger.debug(f'Skipping {url}')
|
||
else:
|
||
logger.warning(f'Unknown url type {type(url)}, {url}')
|
||
else:
|
||
logger.warning(f'Unknown urls type {type(self.urls)}, {self.urls}')
|
||
|
||
def send(self, single_url):
|
||
headers = {'Content-Type': 'application/json'}
|
||
try:
|
||
r = requests.post(single_url, data=self.message.encode('utf-8'), headers=headers)
|
||
if r.status_code != 204:
|
||
logger.warning(f"Status code: {r.status_code}!")
|
||
logger.warning(r.text)
|
||
|
||
except Exception as e:
|
||
logger.warning(f'An exception occurred when sending message to {single_url}; {e}')
|
||
|
||
|
||
def docking_permission2text(permission):
|
||
"""Convert one of all/none/friends/squadron/squadronfriends to user friendly message"""
|
||
options = {
|
||
"all": Messages.DOC_PERMISSION_ALL,
|
||
"none": Messages.DOC_PERMISSION_NONE,
|
||
"friends": Messages.DOC_PERMISSION_FRIENDS,
|
||
"squadron": Messages.DOC_PERMISSION_SQUADRON,
|
||
"squadronfriends": Messages.DOC_PERMISSION_FRIENDS_SQUADRON
|
||
}
|
||
|
||
return options.get(permission) # if key isn't valid then return None
|
||
|
||
|
||
def docking_permission4notorius2text(notorius):
|
||
"""As docking_permission2text() but for notorius (crime persons)"""
|
||
# in python True = 1, False = 0. So in our case False = docking disallow, True = docking allow
|
||
return (Messages.DOC_PERMISSION_DISALLOW_NOTORIUS, Messages.DOC_PERMISSION_ALLOW_NOTORIUS)[notorius]
|
||
|
||
|
||
def plugin_start3(plugin_dir):
|
||
logger.info(f"Plugin version: {VERSION}, enabled status: {config.get_bool('FCT_ENABLE_PLUGIN', default=True)}")
|
||
return 'FC tracker'
|
||
|
||
|
||
def journal_entry(cmdr, is_beta, system, station, entry, state):
|
||
if not config.get_bool('FCT_ENABLE_PLUGIN', default=True):
|
||
return
|
||
|
||
if is_beta and not config.get_bool('FCT_SEND_IN_BETA', default=False):
|
||
return
|
||
|
||
if state["Role"] is not None: # we don't work while you are multicrew passenger
|
||
logger.debug(f"Returning because multicrew, role: {state['Role']}")
|
||
return
|
||
|
||
event = entry["event"]
|
||
|
||
if event == "CarrierStats" and carrier.name is None:
|
||
carrier.name = entry["Name"]
|
||
carrier.cID = entry["CarrierID"]
|
||
carrier.docking_permission = entry["DockingAccess"]
|
||
carrier.allow_notorius = entry["AllowNotorious"]
|
||
carrier.callsign = entry["Callsign"]
|
||
return
|
||
|
||
if event in [
|
||
"CarrierJumpRequest",
|
||
"CarrierJumpCancelled",
|
||
"CarrierJump",
|
||
"CarrierDockingPermission",
|
||
"CarrierNameChange"
|
||
]:
|
||
|
||
message = Embed()
|
||
|
||
if event == "CarrierJumpRequest" and config.get_bool('FCT_SEND_JUMP_REQUESTS', default=True):
|
||
destination_system = entry["SystemName"]
|
||
message.add_item(color=Messages.COLOR_JUMP_REQUEST, title=Messages.TITLE_JUMP_REQUEST)
|
||
|
||
try:
|
||
destination_body = entry["Body"]
|
||
message.update_item(
|
||
item=0,
|
||
key="description",
|
||
new_value=Messages.TEXT_JUMP_REQUEST.format(
|
||
name=carrier.name,
|
||
system=destination_system) + Messages.TEXT_JUMP_REQUEST_BODY_DIFF.format(body=destination_body))
|
||
|
||
except KeyError:
|
||
message.update_item(
|
||
item=0,
|
||
key="description",
|
||
new_value=Messages.TEXT_JUMP_REQUEST.format(
|
||
name=carrier.name,
|
||
system=destination_system))
|
||
|
||
if event == "CarrierJumpCancelled" and config.get_bool('FCT_SEND_JUMP_CANCELING', default=True):
|
||
message.add_item(
|
||
color=Messages.COLOR_JUMP_CANCELLED,
|
||
title=Messages.TITLE_JUMP_CANCELLED,
|
||
description=Messages.TEXT_JUMP_CANCELLED.format(name=carrier.name))
|
||
|
||
if event == "CarrierJump" and config.get_bool('FCT_SEND_JUMPS', default=True):
|
||
# jump on not your own carrier case
|
||
if carrier.callsign != station:
|
||
# for case when you have your own carrier but now jumping on someone else's one
|
||
if config.get_bool('FCT_SEND_JUMPS_NOT_MY_OWN_CARRIER', default=False):
|
||
remember_carrier_name = carrier.name
|
||
carrier.name = station
|
||
else:
|
||
return
|
||
|
||
destination_system = entry["StarSystem"]
|
||
|
||
message.add_item(color=Messages.COLOR_JUMP, title=Messages.TITLE_JUMP)
|
||
|
||
try:
|
||
destination_body = entry["Body"]
|
||
message.update_item(
|
||
item=0,
|
||
key="description",
|
||
new_value=Messages.TEXT_JUMP.format(
|
||
system=destination_system,
|
||
name=carrier.name) + Messages.TEXT_JUMP_REQUEST_BODY_DIFF.format(body=destination_body))
|
||
|
||
except KeyError:
|
||
|
||
message.update_item(
|
||
item=0,
|
||
key="description",
|
||
new_value=Messages.TEXT_JUMP.format(
|
||
system=destination_system,
|
||
name=carrier.name))
|
||
|
||
if config.get_bool('FCT_SEND_JUMPS_NOT_MY_OWN_CARRIER', default=False) and carrier.callsign != station:
|
||
carrier.name = remember_carrier_name
|
||
|
||
if event == "CarrierDockingPermission" and \
|
||
config.get_bool('FCT_SEND_CHANGES_DOCKING_PERMISSIONS', default=True):
|
||
new_permission = entry["DockingAccess"]
|
||
new_doc_for_crime = entry["AllowNotorious"]
|
||
|
||
message.add_item(
|
||
title=Messages.TITLE_CHANGE_DOCKING_PERMISSION,
|
||
color=Messages.COLOR_PERMISSION_CHANGE,
|
||
description=Messages.TEXT_PERMISSION_CHANGE.format(
|
||
name=carrier.name,
|
||
old_permission=docking_permission2text(carrier.docking_permission),
|
||
new_permission=docking_permission2text(new_permission),
|
||
old_doc_for_crime=docking_permission4notorius2text(carrier.allow_notorius),
|
||
new_doc_for_crime=docking_permission4notorius2text(new_doc_for_crime)))
|
||
|
||
carrier.docking_permission = new_permission
|
||
carrier.allow_notorius = new_doc_for_crime
|
||
|
||
if event == "CarrierNameChange" and config.get_bool('FCT_SEND_CHANGES_NAME', default=True):
|
||
new_name = entry["Name"]
|
||
message.add_item(
|
||
title=Messages.TITLE_CHANGE_NAME,
|
||
description=Messages.TEXT_CHANGE_NAME.format(
|
||
old_name=carrier.name,
|
||
new_name=new_name),
|
||
color=Messages.COLOR_CHANGE_NAME)
|
||
carrier.name = new_name
|
||
|
||
if is_beta or force_beta: # TODO: just override is_beta
|
||
message.add_item(title=Messages.TITLE_IN_BETA, description=Messages.TEXT_IN_BETA)
|
||
|
||
# one Messages_sender instance per message
|
||
Messages_sender(message.get_message(), config.get_list('FCT_DISCORD_WEBHOOK_URLS', default=None))
|
||
|
||
|
||
def plugin_prefs(parent: nb.Notebook, cmdr: str, is_beta: bool) -> Optional[tk.Frame]:
|
||
row = 0
|
||
|
||
webhooks_urls = config.get_list('FCT_DISCORD_WEBHOOK_URLS', default=[None for i in range(0, 5)])
|
||
enable_plugin = tk.IntVar(value=config.get_bool('FCT_ENABLE_PLUGIN', default=True))
|
||
# this.webhook_url = tk.StringVar(value=config.get_str('FCT_DISCORD_WEBHOOK_URL', default=None))
|
||
send_jumps = tk.IntVar(value=config.get_bool('FCT_SEND_JUMPS', default=True))
|
||
send_in_beta = tk.IntVar(value=config.get_bool('FCT_SEND_IN_BETA', default=False))
|
||
send_jump_requests = tk.IntVar(value=config.get_bool('FCT_SEND_JUMP_REQUESTS', default=True))
|
||
send_jump_canceling = tk.IntVar(value=config.get_bool('FCT_SEND_JUMP_CANCELING', default=True))
|
||
send_changes_docking_permissions = tk.IntVar(value=config.get_bool('FCT_SEND_CHANGES_DOCKING_PERMISSIONS',
|
||
default=True))
|
||
send_changes_name = tk.IntVar(value=config.get_bool('FCT_SEND_CHANGES_NAME', default=True))
|
||
send_jumps_not_my_own_carrier = tk.IntVar(value=config.get_bool('FCT_SEND_JUMPS_NOT_MY_OWN_CARRIER', default=False))
|
||
|
||
frame = nb.Frame(parent)
|
||
|
||
nb.Checkbutton(
|
||
frame,
|
||
text="Enable plugin",
|
||
variable=enable_plugin,
|
||
command=lambda: config.set('FCT_ENABLE_PLUGIN', enable_plugin.get())).grid(
|
||
row=row, padx=10, pady=(10, 0), sticky=tk.W)
|
||
row += 1
|
||
|
||
nb.Label(
|
||
frame,
|
||
text="Enter your webhooks urls here, you can enter up to 5 hooks:").grid(
|
||
row=row, padx=10, pady=(10, 0), sticky=tk.W)
|
||
row += 1
|
||
|
||
this.webhooks_urls = [tk.StringVar(value=one_url) for one_url in webhooks_urls]
|
||
for i in range(0, 5):
|
||
|
||
nb.Entry(
|
||
frame,
|
||
textvariable=this.webhooks_urls[i],
|
||
width=115).grid(
|
||
row=row, padx=10, pady=(0, 5))
|
||
nb.Label(
|
||
frame,
|
||
text=f'#{i + 1}').grid(
|
||
row=row, column=1, sticky=tk.E
|
||
)
|
||
row += 1
|
||
|
||
nb.Checkbutton(
|
||
frame,
|
||
text='Send jumps',
|
||
variable=send_jumps,
|
||
command=lambda: config.set('FCT_SEND_JUMPS', send_jumps.get())).grid(
|
||
row=row, padx=10, pady=(5, 0), sticky=tk.W)
|
||
row += 1
|
||
|
||
nb.Checkbutton(
|
||
frame,
|
||
text='Send in beta',
|
||
variable=send_in_beta,
|
||
command=lambda: config.set('FCT_SEND_IN_BETA', send_in_beta.get())).grid(
|
||
row=row, padx=10, pady=(5, 0), sticky=tk.W)
|
||
|
||
row += 1
|
||
nb.Checkbutton(
|
||
frame,
|
||
text='Send jump requests',
|
||
variable=send_jump_requests,
|
||
command=lambda: config.set('FCT_SEND_JUMP_REQUESTS', send_jump_requests.get())).grid(
|
||
row=row, padx=10, pady=(5, 0), sticky=tk.W)
|
||
|
||
row += 1
|
||
nb.Checkbutton(
|
||
frame,
|
||
text='Send jump canceling',
|
||
variable=send_jump_canceling,
|
||
command=lambda: config.set('FCT_SEND_JUMP_CANCELING', send_jump_canceling.get())).grid(
|
||
row=row, padx=10, pady=(5, 0), sticky=tk.W)
|
||
|
||
row += 1
|
||
nb.Checkbutton(
|
||
frame,
|
||
text='Send changes docking permissions',
|
||
variable=send_changes_docking_permissions,
|
||
command=lambda: config.set('FCT_SEND_CHANGES_DOCKING_PERMISSIONS',
|
||
send_changes_docking_permissions.get())).grid(
|
||
row=row, padx=10, pady=(5, 0), sticky=tk.W)
|
||
|
||
row += 1
|
||
nb.Checkbutton(
|
||
frame,
|
||
text='Send changes name',
|
||
variable=send_changes_name,
|
||
command=lambda: config.set('FCT_SEND_CHANGES_NAME', send_changes_name.get())).grid(
|
||
row=row, padx=10, pady=(5, 0), sticky=tk.W)
|
||
|
||
row += 1
|
||
nb.Checkbutton(
|
||
frame,
|
||
text='Send jumps not my own carrier',
|
||
variable=send_jumps_not_my_own_carrier,
|
||
command=lambda: config.set('FCT_SEND_JUMPS_NOT_MY_OWN_CARRIER', send_jumps_not_my_own_carrier.get())).grid(
|
||
row=row, padx=10, pady=(5, 0), sticky=tk.W)
|
||
|
||
row += 1
|
||
return frame
|
||
|
||
|
||
def prefs_changed(cmdr: str, is_beta: bool) -> None:
|
||
config.set('FCT_DISCORD_WEBHOOK_URLS', [webhook_url.get() for webhook_url in this.webhooks_urls])
|
||
del this.webhooks_urls
|
||
config.save()
|
||
|
||
|
||
carrier = Carrier()
|