mirror of
https://github.com/Noettore/lagomareGateKeeperBot.git
synced 2025-10-15 03:26:40 +02:00
Che schifo
This commit is contained in:
@@ -1,11 +1,11 @@
|
||||
from .credential import Credential
|
||||
from .gates import Gates
|
||||
from .gate import Gate
|
||||
from .status import Status
|
||||
from .users import Users, Role
|
||||
|
||||
__all__ = [
|
||||
"Credential",
|
||||
"Gates",
|
||||
"Gate",
|
||||
"Status",
|
||||
"Users",
|
||||
"Role"
|
||||
|
18
models/gate.py
Normal file
18
models/gate.py
Normal file
@@ -0,0 +1,18 @@
|
||||
import json
|
||||
from services import AVConnectAPI
|
||||
from .status import Status
|
||||
from .credential import Credential
|
||||
|
||||
class Gate:
|
||||
def __init__(self, id: str, name: str, status: Status = Status.ENABLED):
|
||||
self.id = id
|
||||
self.name = name
|
||||
self.status = status if isinstance(status, Status) else Status(status)
|
||||
|
||||
def to_dict(self):
|
||||
return {"id": self.id, "name": self.name, "status": self.status.value}
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: dict):
|
||||
return cls(data["id"], data["name"], Status(data.get("status", Status.ENABLED)))
|
||||
|
@@ -1,48 +0,0 @@
|
||||
import json
|
||||
from services import AVConnectAPI
|
||||
from .status import Status
|
||||
from .credential import Credential
|
||||
|
||||
class Gate:
|
||||
def __init__(self, id: str, name: str, status: Status = Status.ENABLED):
|
||||
self.id = id
|
||||
self.name = name
|
||||
self.status = status if isinstance(status, Status) else Status(status)
|
||||
|
||||
def to_dict(self):
|
||||
return {"id": self.id, "name": self.name, "status": self.status.value}
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: dict):
|
||||
return cls(data["id"], data["name"], Status(data.get("status", Status.ENABLED)))
|
||||
|
||||
class Gates:
|
||||
def __init__(self, json_path: str = "./data/gates.json"):
|
||||
self._json_path: str = json_path
|
||||
self._gates: dict[str, Gate] = self._load_gates()
|
||||
|
||||
def _load_gates(self) -> dict[str, Gate]:
|
||||
try:
|
||||
with open(self._json_path, "r") as file:
|
||||
gates_data = json.load(file)
|
||||
return {gate: Gate.from_dict(data) for gate, data in gates_data.items()}
|
||||
except Exception:
|
||||
return {}
|
||||
|
||||
def get_name(self, gate: str) -> str | None:
|
||||
return self._gates[gate].name if gate in self._gates else None
|
||||
|
||||
def get_all_gates_id(self) -> list[str]:
|
||||
return [gate for gate in self._gates.keys() if self._gates[gate].status == Status.ENABLED]
|
||||
|
||||
def open_gate(self, gate: str, credentials: Credential) -> bool:
|
||||
if gate not in self._gates:
|
||||
return False
|
||||
if self._gates[gate].status == Status.DISABLED:
|
||||
return False
|
||||
try:
|
||||
api = AVConnectAPI(credentials)
|
||||
return api.exec_gate_macro(self._gates[gate].id)
|
||||
except Exception as e:
|
||||
print(f"Failed to open gate {gate}: {e}")
|
||||
return False
|
21
repository/gates_repository.py
Normal file
21
repository/gates_repository.py
Normal file
@@ -0,0 +1,21 @@
|
||||
import json
|
||||
from models import Status, Gate
|
||||
|
||||
class GatesRepository:
|
||||
def __init__(self, json_path: str = "./data/gates.json"):
|
||||
self._json_path: str = json_path
|
||||
self._gates: dict[str, Gate] = self._load_gates()
|
||||
|
||||
def _load_gates(self) -> dict[str, Gate]:
|
||||
try:
|
||||
with open(self._json_path, "r") as file:
|
||||
gates_data = json.load(file)
|
||||
return {gate: Gate.from_dict(data) for gate, data in gates_data.items()}
|
||||
except Exception:
|
||||
return {}
|
||||
|
||||
def get_by_key(self, key: str) -> Gate | None:
|
||||
return self._gates[key] if key in self._gates else None
|
||||
|
||||
def get_all_gates_id(self) -> list[str]:
|
||||
return [gate for gate in self._gates.keys() if self._gates[gate].status == Status.ENABLED]
|
14
services/gates.py
Normal file
14
services/gates.py
Normal file
@@ -0,0 +1,14 @@
|
||||
from models import Credential, Status
|
||||
|
||||
class GatesService:
|
||||
def open_gate(self, gate: str, credentials: Credential) -> bool:
|
||||
if gate not in self._gates:
|
||||
return False
|
||||
if self._gates[gate].status == Status.DISABLED:
|
||||
return False
|
||||
try:
|
||||
api = AVConnectAPI(credentials)
|
||||
return api.exec_gate_macro(self._gates[gate].id)
|
||||
except Exception as e:
|
||||
print(f"Failed to open gate {gate}: {e}")
|
||||
return False
|
Reference in New Issue
Block a user