257 lines
8.0 KiB
Python
257 lines
8.0 KiB
Python
import subprocess
|
|
from dotenv import load_dotenv
|
|
import os
|
|
from dom5game import Dom5game
|
|
import re
|
|
|
|
load_dotenv()
|
|
SERVER_PATH = os.getenv("SERVERPATH")
|
|
|
|
|
|
def create_server(name: str, port: int, channel: int, command: str):
|
|
if is_port_in_use(port):
|
|
return "ERROR_PORT_IN_USE"
|
|
os.mkdir("games/" + name)
|
|
game = Dom5game(name, channel, [], 0, {})
|
|
game.to_json()
|
|
try:
|
|
p = subprocess.Popen(
|
|
command,
|
|
stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.STDOUT,
|
|
)
|
|
except Exception as e:
|
|
return "EXCEPTION OCCURED: " + str(e)
|
|
return "SUCCESS"
|
|
# p.wait()
|
|
|
|
|
|
def get_nations():
|
|
# TODO mod support
|
|
nation_string = subprocess.check_output([SERVER_PATH, "--listnations"]).decode()
|
|
nations = {}
|
|
current_era = None
|
|
for line in nation_string.splitlines():
|
|
line = line.strip()
|
|
|
|
era_match = re.match(r"-+ Era (\d+) -+", line)
|
|
if era_match:
|
|
current_era = f"{era_match.group(1)}"
|
|
nations[current_era] = {}
|
|
continue
|
|
|
|
entry_match = re.match(r"(\d+)\s+(.*?),\s+(.*)", line)
|
|
if entry_match and current_era:
|
|
nation_id = int(entry_match.group(1))
|
|
name = entry_match.group(2), entry_match.group(3)
|
|
|
|
nations[current_era][nation_id] = {"name": name}
|
|
return nations
|
|
|
|
|
|
# This is terrible.
|
|
def server_command_builder(
|
|
name: str,
|
|
port: int,
|
|
era: int, # 1 EA, 2 MA, 3 LA
|
|
ai_slots: list[
|
|
tuple[int, int]
|
|
] = [], # list of nations that will be ai tuple format: NationID, AI Level (1-6), 0 being closed
|
|
client_start: bool = True,
|
|
teams: list[tuple[int, int, int]] = [],
|
|
clustered_start: bool = False,
|
|
team_game: bool = False,
|
|
mapfile: str = "",
|
|
random_map: int = 0,
|
|
research_difficulty: int = 2,
|
|
random_start_research: bool = True,
|
|
hof_size: int = 10,
|
|
global_slots: int = 5,
|
|
inde_strength: int = 5,
|
|
magic_sites: int = 40,
|
|
event_rarity: int = 1,
|
|
richness: int = 100,
|
|
resources: int = 100,
|
|
recruitment: int = 100,
|
|
supplies: int = 100,
|
|
masterpass: str = "",
|
|
start_prov: int = 1,
|
|
renaming: bool = True,
|
|
score_graphs: bool = False,
|
|
no_nation_info: bool = False,
|
|
no_cheat_det: bool = False,
|
|
no_artifact_rest: bool = False,
|
|
story_events: int = 1,
|
|
new_ai_lvl: int = 2,
|
|
no_new_ai: bool = False,
|
|
conq_all: bool = False,
|
|
thrones: tuple[int, int, int] = (1, 1, 1),
|
|
required_apoints: int = 0,
|
|
cataclysm: int = 0,
|
|
vwrap: bool = True,
|
|
hwrap: bool = True,
|
|
):
|
|
available_nations = get_nations()
|
|
|
|
# TODO change < ranges to =<
|
|
if (not ((random_map == 0) ^ (mapfile == ""))) or random_map not in {0, 10, 15, 20}:
|
|
return "ERROR_MAP"
|
|
|
|
if not (0 < era < 4):
|
|
return "ERROR_ERA"
|
|
|
|
ai_indexes = [t[1] for t in ai_slots]
|
|
if not (len(ai_indexes) == len(set(ai_indexes))):
|
|
return "ERROR_AIS"
|
|
|
|
# if bool(set(closed_slots) & set(ai_indexes)):
|
|
# return "ERROR_CLOSED_AIS"
|
|
|
|
if required_apoints == 0:
|
|
required_apoints = thrones[0] + (2 * thrones[1]) + (3 * thrones[2]) - 1
|
|
|
|
if not (0 <= research_difficulty < 5):
|
|
return "ERROR_RESEARCH"
|
|
if not (4 < hof_size < 16):
|
|
return "ERROR_HOF"
|
|
if not (2 < global_slots < 10):
|
|
return "ERROR_GLOBALS"
|
|
if not (0 <= inde_strength <= 9):
|
|
return "ERROR_INDEP"
|
|
if not (0 <= magic_sites <= 75):
|
|
return "ERROR_MAGICSITES"
|
|
if not (1 <= event_rarity <= 2):
|
|
return "ERROR_EVENTS"
|
|
if not (50 <= richness <= 300):
|
|
return "ERROR_RICHNESS"
|
|
if not (50 <= recruitment <= 300):
|
|
return "ERROR_RECRUITMENT"
|
|
if not (50 <= resources <= 300):
|
|
return "ERROR_RESOURCES"
|
|
if not (50 <= supplies <= 300):
|
|
return "ERROR_SUPPLIES"
|
|
if not (1 <= start_prov <= 9):
|
|
return "ERROR_STARTPROV"
|
|
if not (0 <= story_events <= 2):
|
|
return "ERROR_STORYEVENTS"
|
|
if not (1 <= new_ai_lvl <= 6):
|
|
return "ERROR_NEWAIS"
|
|
command = [
|
|
SERVER_PATH,
|
|
" -TS",
|
|
name,
|
|
" --port ",
|
|
str(port),
|
|
" --statuspage ",
|
|
os.getcwd() + "/games/" + name + "/turnstats.html",
|
|
" --era ",
|
|
str(era),
|
|
]
|
|
if team_game:
|
|
for team in teams:
|
|
# TODO: Can team have more than one disciple and pretender?
|
|
if team[0] not in available_nations[str(era)]:
|
|
return "ERROR_INVALID_NATION"
|
|
|
|
command.append(
|
|
" --team " + str(team[0]) + " " + str(team[1]) + " " + str(team[2])
|
|
)
|
|
if clustered_start:
|
|
command.append(" --clustered")
|
|
|
|
if not client_start:
|
|
command.append(" --noclientstart")
|
|
if cataclysm > 0:
|
|
command.append(" --cataclysm " + str(cataclysm))
|
|
if conq_all:
|
|
command.append(" --conqall")
|
|
if not research_difficulty == 2:
|
|
command.append(" --research " + str(research_difficulty))
|
|
if not random_start_research:
|
|
command.append(" --norandres")
|
|
if not hof_size == 10:
|
|
command.append(" --hof_size " + str(hof_size))
|
|
if not global_slots == 5:
|
|
command.append(" --globals " + str(global_slots))
|
|
if not inde_strength == 5:
|
|
command.append(" --indepstr " + str(inde_strength))
|
|
if not magic_sites == 40:
|
|
command.append(" --magisites " + str(magic_sites))
|
|
command.append(" --eventrarity " + str(event_rarity))
|
|
if not richness == 100:
|
|
command.append(" --richness " + str(richness))
|
|
if not resources == 100:
|
|
command.append(" --resources" + str(resources))
|
|
if not recruitment == 100:
|
|
command.append(" --recruitment" + str(recruitment))
|
|
if not supplies == 100:
|
|
command.append(" --supplies" + str(supplies))
|
|
if not masterpass == "":
|
|
command.append(" --masterpass " + masterpass)
|
|
# i am going insane
|
|
if not start_prov == 1:
|
|
command.append(" --startprov " + str(start_prov))
|
|
if renaming:
|
|
command.append(" --renaming")
|
|
if score_graphs:
|
|
command.append(" --scoregraphs")
|
|
if no_nation_info:
|
|
command.append(" --nonationinfo")
|
|
if no_cheat_det:
|
|
command.append(" --nocheatdet")
|
|
if no_artifact_rest:
|
|
command.append(" --noartrest")
|
|
if story_events == 1:
|
|
command.append(" --storyevents")
|
|
elif story_events == 2:
|
|
command.append(" --allstoryevents")
|
|
command.append(" --newailvl " + str(new_ai_lvl))
|
|
if no_new_ai:
|
|
command.append(" --nonewai")
|
|
if conq_all:
|
|
command.append(" --conqall")
|
|
command.append(
|
|
" --thrones " + str(thrones[0]) + " " + str(thrones[1]) + " " + str(thrones[2])
|
|
)
|
|
command.append(" --requiredap " + str(required_apoints))
|
|
if not cataclysm == 0:
|
|
command.append(" --cataclysm " + str(cataclysm))
|
|
if vwrap:
|
|
command.append(" --vwrap")
|
|
if not hwrap:
|
|
command.append(" --nohwrap")
|
|
if ai_slots:
|
|
added_ais = []
|
|
for ai in ai_slots:
|
|
if ai[0] not in available_nations[str(era)]:
|
|
return "ERROR_INVALID_NATION"
|
|
if ai[0] in added_ais:
|
|
return "ERROR_DUPLICATE_AI"
|
|
match ai[1]:
|
|
case 0:
|
|
command.append(" --closed " + str(ai[0]))
|
|
case 1:
|
|
command.append(" --easyai " + str(ai[0]))
|
|
case 2:
|
|
command.append(" --normai " + str(ai[0]))
|
|
case 3:
|
|
command.append(" --diffai " + str(ai[0]))
|
|
case 4:
|
|
command.append(" --mightyai " + str(ai[0]))
|
|
case 5:
|
|
command.append(" --masterai " + str(ai[0]))
|
|
case 6:
|
|
command.append(" --impai " + str(ai[0]))
|
|
case _:
|
|
return "ERROR_INVALID_AI"
|
|
added_ais.append(ai[0])
|
|
|
|
return "".join(command)
|
|
|
|
|
|
def is_port_in_use(port: int) -> bool:
|
|
import socket
|
|
|
|
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
|
|
return s.connect_ex(("localhost", port)) == 0
|