327 lines
9.4 KiB
Python
327 lines
9.4 KiB
Python
import os
|
||
import requests
|
||
from pymongo import MongoClient
|
||
from bson.objectid import ObjectId
|
||
import datetime
|
||
|
||
_var = "/opt/Inventarsystem/"
|
||
_cmd = "manage-tenant.sh"
|
||
|
||
MONGO_URI = os.environ.get("MONGO_URI", "mongodb://localhost:27017")
|
||
MONGO_DB_NAME = os.environ.get("MONGO_DB_NAME", "Invario_Website")
|
||
|
||
|
||
def _get_users_collection():
|
||
client = MongoClient(MONGO_URI, serverSelectionTimeoutMS=1500)
|
||
db = client[MONGO_DB_NAME]
|
||
return client, db["packages"]
|
||
|
||
def clear_special(var_:str) -> str:
|
||
"""
|
||
Clears the variable of any special carakters
|
||
|
||
Input:
|
||
- var -> String
|
||
|
||
Output:
|
||
- str cleared of the speacial carakters
|
||
"""
|
||
# Refactor the name varible to match the subdomain requirements
|
||
try:
|
||
var_ = var_.lower()
|
||
special_caracters1 = '^°!"§$%&/()=?\ß{[]}´`*~#,;.:<>|@€µ'
|
||
special_caracters2 = 'äüö'
|
||
for char in special_caracters1:
|
||
var_ = var_.replace(char, "")
|
||
for i in special_caracters2:
|
||
match i:
|
||
case "ü":
|
||
var_ = var_.replace(i, "ue")
|
||
case "ä":
|
||
var_ = var_.replace(i, "ae")
|
||
case "ö":
|
||
var_ = var_.replace(i, "oe")
|
||
except:
|
||
return False
|
||
|
||
def execute_script(wd_: str, file_: str, com_: str, com2_: str="None", com3_: str="None", com4_: str="None") -> str:
|
||
"""
|
||
executes a script with the option of to extra inputs
|
||
|
||
Input:
|
||
- wd_ = working directory of the Inventorysystem -> String
|
||
- file_ = working file that youre targeting -> String
|
||
- com_ = first option -> String
|
||
- com2_ = second option (Optional if needet)-> String
|
||
- com3_ = third option (Optional if needet)-> String
|
||
- com4_ = fourth option (Optional if needet)-> String
|
||
Output:
|
||
- ether False if failed -> bool
|
||
- or result.stdout output of the executed process -> str
|
||
"""
|
||
import subprocess
|
||
update_path = os.path.join(wd_, file_)
|
||
if not update_path:
|
||
return False
|
||
if com2_ != "None":
|
||
cmd = f'bash "{update_path}" {com_} {com2_}'
|
||
elif com3_ != "None":
|
||
cmd = f'bash "{update_path}" {com_} {com2_} {com3_}'
|
||
elif com4_ != "None":
|
||
cmd = f'bash "{update_path}" {com_} {com2_} {com3_} {com4_}'
|
||
else:
|
||
cmd = f'bash "{update_path}" {com_}'
|
||
try:
|
||
result = subprocess.run(
|
||
["bash", "-lc", cmd],
|
||
capture_output=True,
|
||
text=True,
|
||
cwd=wd_,
|
||
)
|
||
except Exception as e:
|
||
print(f"KRITISCHER FEHLER bei execute_script im Ordner {wd_}: {e}") # Jetzt wird der Fehler ins Log geschrieben
|
||
return False
|
||
return result.stdout
|
||
|
||
class versions:
|
||
"""
|
||
This will give access to anything like:
|
||
- Version of the Inventorysystem
|
||
- Version of the Server
|
||
|
||
modules:
|
||
- inventorysystem()
|
||
- server()
|
||
"""
|
||
def inventorysystem(name: str) -> str:
|
||
"""
|
||
Version of the Inventorysystem
|
||
|
||
Output:
|
||
- version -> String
|
||
"""
|
||
return Instance.edit(name, "inventarsystem")
|
||
|
||
def bibliothek(name: str) -> str:
|
||
"""
|
||
Version of the Inventorysystem
|
||
|
||
Output:
|
||
- version -> String
|
||
"""
|
||
return Instance.edit(name, "buecherei")
|
||
|
||
def terminplanen(name: str) -> str:
|
||
"""
|
||
Version of the Inventorysystem
|
||
|
||
Output:
|
||
- version -> String
|
||
"""
|
||
return Instance.edit(name, "terminverwaltung")
|
||
|
||
def emailversand(name: str) -> str:
|
||
"""
|
||
Version of the Inventorysystem
|
||
|
||
Output:
|
||
- version -> String
|
||
"""
|
||
return Instance.edit(name, "emailversand")
|
||
|
||
def starter(name: str) -> str:
|
||
"""
|
||
Version of the Inventorysystem
|
||
|
||
Output:
|
||
- version -> String
|
||
"""
|
||
return Instance.edit(name, "starter")
|
||
|
||
def advanced(name: str) -> str:
|
||
"""
|
||
Version of the Inventorysystem
|
||
|
||
Output:
|
||
- version -> String
|
||
"""
|
||
return Instance.edit(name, "advanced")
|
||
|
||
def testversion(name: str) -> str:
|
||
"""
|
||
Version of the Inventorysystem
|
||
|
||
Output:
|
||
- version -> String
|
||
"""
|
||
client = None
|
||
try:
|
||
client, packages = _get_users_collection()
|
||
packages.insert_one(
|
||
{
|
||
'client_name': name,
|
||
'inventarsystem': False,
|
||
'buecherei': False,
|
||
'terminverwaltung': False,
|
||
'emailversand': False,
|
||
'starter': False,
|
||
'advanced': False,
|
||
'registered_on': datetime.datetime.utcnow()
|
||
}
|
||
)
|
||
return True
|
||
finally:
|
||
if client:
|
||
client.close()
|
||
return Instance.edit(name, "starter")
|
||
|
||
def _get_auth(name: str) -> int:
|
||
"""
|
||
Check if the if it is older than 5 days Test Version
|
||
|
||
Output:
|
||
- days since registration -> int
|
||
"""
|
||
client = None
|
||
try:
|
||
client, packages = _get_users_collection()
|
||
package = packages.find_one({'client_name': name}) or {}
|
||
for key in package:
|
||
if key == "registered_on":
|
||
if package[key]:
|
||
instance_age = (datetime.datetime.utcnow() - package[key]).days
|
||
else:
|
||
instance_age = None
|
||
finally:
|
||
if client:
|
||
client.close()
|
||
return int(instance_age) if int(instance_age) is not None and instance_age <= 5 else False
|
||
|
||
|
||
class Instance:
|
||
@staticmethod
|
||
def list() -> list:
|
||
"""
|
||
Lists all existing tenants along with their mapped ports.
|
||
Returns: list of dicts -> [{'name': 'school1', 'port': 10002}, ...]
|
||
"""
|
||
result = execute_script(_var, _cmd, "list")
|
||
if not isinstance(result, str):
|
||
return []
|
||
|
||
tenants = []
|
||
pattern = re.compile(r"^-\s+([^\s]+)\s+\(port\s+(\d+)\)")
|
||
for line in result.splitlines():
|
||
match = pattern.match(line.strip())
|
||
if match:
|
||
tenants.append({
|
||
"name": match.group(1),
|
||
"port": int(match.group(2))
|
||
})
|
||
return tenants
|
||
|
||
@classmethod
|
||
def get_next_available_port(cls, start_port=10002) -> int:
|
||
tenants = cls.list()
|
||
if not tenants:
|
||
return start_port
|
||
used_ports = [t["port"] for t in tenants]
|
||
return max(used_ports) + 1
|
||
|
||
@classmethod
|
||
def new(cls, name: str, port: int = None, password: str = "admin123") -> bool:
|
||
safe_name = clear_special(name)
|
||
if port is None:
|
||
port = cls.get_next_available_port()
|
||
|
||
return execute_script(_var, _cmd, "add", safe_name, str(port), password)
|
||
|
||
@staticmethod
|
||
def edit(name: str, module_preset: str) -> bool:
|
||
safe_name = clear_special(name)
|
||
presets = {
|
||
"inventarsystem": "library=off inventory=on student_cards=off terminplan=off",
|
||
"buecherei": "library=on inventory=off student_cards=off terminplan=off",
|
||
"terminverwaltung": "library=off inventory=off student_cards=off terminplan=on",
|
||
"emailversand": "library=off inventory=on student_cards=off terminplan=off",
|
||
"starter": "library=on inventory=on student_cards=on terminplan=off",
|
||
"advanced": "library=on inventory=on student_cards=on terminplan=on",
|
||
}
|
||
|
||
config = presets.get(module_preset)
|
||
if not config:
|
||
return False
|
||
|
||
return execute_script(_var, _cmd, "module", safe_name, config)
|
||
|
||
@staticmethod
|
||
def remove(name: str) -> bool:
|
||
safe_name = clear_special(name)
|
||
if not execute_script(_var, _cmd, "remove", safe_name):
|
||
return False
|
||
|
||
try:
|
||
client, packages = _get_users_collection()
|
||
packages.delete_one({"client_name": safe_name})
|
||
except Exception:
|
||
pass
|
||
finally:
|
||
if 'client' in locals() and client:
|
||
client.close()
|
||
return True
|
||
|
||
@staticmethod
|
||
def status(name: str) -> bool:
|
||
safe_name = clear_special(name)
|
||
try:
|
||
res = requests.get(f"https://{safe_name}.invario-software.de/health", timeout=5)
|
||
return res.status_code == 200
|
||
except requests.RequestException:
|
||
return False
|
||
|
||
@staticmethod
|
||
def restart(name: str) -> bool:
|
||
return execute_script(_var, _cmd, "restart-tenant", clear_special(name))
|
||
|
||
class ussage:
|
||
"""
|
||
This will give informations about anything like:
|
||
- RAM Ussage of the server
|
||
- CPU Ussage of the server
|
||
- Strorage that is in use
|
||
|
||
modules:
|
||
- ram()
|
||
- cpu()
|
||
- storage()
|
||
"""
|
||
def ram() -> int:
|
||
"""
|
||
RAM ussage of the complete system
|
||
|
||
Output:
|
||
- ram ussage -> interger in GB
|
||
"""
|
||
#print("RAM usage (%):", ram.percent)
|
||
import psutil
|
||
ram = psutil.virtual_memory()
|
||
return int(round(ram.used / 1e9, 2))
|
||
|
||
def cpu() -> int:
|
||
"""
|
||
System cpu ussage
|
||
|
||
Output:
|
||
- cpu ussage -> integer in Percent
|
||
"""
|
||
import psutil
|
||
return int(psutil.cpu_percent(interval=1))
|
||
|
||
def storage() -> int:
|
||
"""
|
||
System storage ussage
|
||
|
||
Output:
|
||
- storager ussage -> integer in Percent
|
||
"""
|
||
pass |