277 lines
9.5 KiB
Python
277 lines
9.5 KiB
Python
import dataclasses
|
|
import logging
|
|
from typing import TypedDict
|
|
|
|
from django_q.tasks import async_task
|
|
|
|
from doorcontrol.hid.Credential import Credential
|
|
from doorcontrol.hid.DoorController import ROOT, E
|
|
from doorcontrol.models import AttributeScheduleRule, Door, FlagScheduleRule
|
|
from membershipworks.models import Member
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class CardholderAttribs(TypedDict):
|
|
forename: str
|
|
middleName: str
|
|
surname: str
|
|
email: str
|
|
phone: str
|
|
custom1: str
|
|
custom2: str
|
|
|
|
|
|
@dataclasses.dataclass
|
|
class DoorMember:
|
|
door: Door
|
|
attribs: CardholderAttribs
|
|
credentials: set[Credential]
|
|
schedules: set[str]
|
|
cardholderID: str | None = None
|
|
|
|
@classmethod
|
|
def from_membershipworks_member(cls, member: Member, door: Door):
|
|
if member.access_card_facility_code and member.access_card_number:
|
|
credentials = {
|
|
Credential(
|
|
code=(
|
|
member.access_card_facility_code,
|
|
member.access_card_number,
|
|
)
|
|
)
|
|
}
|
|
else:
|
|
credentials = set()
|
|
|
|
reasons_and_schedules: dict[str, str] = {}
|
|
if (
|
|
member.is_active
|
|
or member.flags.filter(name="Misc. Access", type="folder").exists()
|
|
) and getattr(member, door.access_field):
|
|
# TODO: could this be annotated?
|
|
reasons_and_schedules |= FlagScheduleRule.objects.filter(
|
|
doors=door, flag__members=member
|
|
).values_list("flag__name", "schedule__name")
|
|
|
|
# TODO: this seems like it could be cleaner
|
|
reasons_and_schedules |= {
|
|
attribute_rule.access_field: attribute_rule.schedule.name
|
|
for attribute_rule in AttributeScheduleRule.objects.filter(doors=door)
|
|
if getattr(member, attribute_rule.access_field)
|
|
}
|
|
|
|
reasons = sorted(reasons_and_schedules.keys())
|
|
|
|
return cls(
|
|
door=door,
|
|
attribs={
|
|
"forename": member.first_name,
|
|
"middleName": "",
|
|
"surname": member.last_name,
|
|
"email": member.email,
|
|
"phone": member.phone,
|
|
"custom1": "|".join(reasons).replace("&", "and"),
|
|
"custom2": member.uid,
|
|
},
|
|
credentials=credentials,
|
|
schedules=set(reasons_and_schedules.values()),
|
|
)
|
|
|
|
@classmethod
|
|
def from_cardholder(cls, data, door: Door):
|
|
return cls(
|
|
door=door,
|
|
attribs={
|
|
"forename": data.get("forename", ""),
|
|
"middleName": data.attrib.get("middleName", ""),
|
|
"surname": data.get("surname", ""),
|
|
"email": data.attrib.get("email", ""),
|
|
"phone": data.attrib.get("phone", ""),
|
|
"custom1": data.attrib.get("custom1", ""),
|
|
"custom2": data.attrib.get("custom2", ""),
|
|
},
|
|
cardholderID=data.attrib["cardholderID"],
|
|
credentials={
|
|
Credential(hex=(c.attrib["rawCardNumber"]))
|
|
for c in data.findall("{*}Credential")
|
|
},
|
|
schedules={r.attrib["scheduleName"] for r in data.findall("{*}Role")},
|
|
)
|
|
|
|
@property
|
|
def membershipworks_id(self):
|
|
return self.attribs["custom2"]
|
|
|
|
@property
|
|
def full_name(self):
|
|
return f"{self.attribs['forename']} {self.attribs['surname']}"
|
|
|
|
def update_attribs(self):
|
|
self.door.controller.doXMLRequest(
|
|
ROOT(
|
|
E.Cardholders(
|
|
{"action": "UD", "cardholderID": self.cardholderID},
|
|
E.CardHolder(self.attribs),
|
|
)
|
|
)
|
|
)
|
|
|
|
def update_credentials(
|
|
self,
|
|
existing_door_credentials: set[Credential],
|
|
all_members: list["DoorMember"],
|
|
old_credentials: set[Credential] = set(),
|
|
):
|
|
# cardholderID should be set on a member before this is called
|
|
assert self.cardholderID is not None
|
|
|
|
other_assigned_cards = {
|
|
card for m in all_members if m != self for card in m.credentials
|
|
}
|
|
|
|
# cards removed, and won't be reassigned to someone else
|
|
for card in (old_credentials - self.credentials) - other_assigned_cards:
|
|
self.door.controller.update_credential(card.hex, "")
|
|
|
|
if self.credentials - old_credentials: # cards added
|
|
for card in (
|
|
self.credentials & other_assigned_cards
|
|
): # new card exists in another member
|
|
logger.info(
|
|
[
|
|
m
|
|
for m in all_members
|
|
for card in m.credentials
|
|
if card in self.credentials
|
|
]
|
|
)
|
|
raise Exception(f"Duplicate Card in input data! {card}")
|
|
|
|
# card existed in door.controller, and needs to be reassigned
|
|
for card in self.credentials & existing_door_credentials:
|
|
self.door.controller.update_credential(card.hex, self.cardholderID)
|
|
|
|
# cards that never existed, and need to be created
|
|
if self.credentials - existing_door_credentials:
|
|
xml_credentials = [
|
|
E.Credential(
|
|
{
|
|
"formatName": str(credential.code[0]),
|
|
"cardNumber": str(credential.code[1]),
|
|
"formatID": self.door.card_formats[str(credential.code[0])],
|
|
"isCard": "true",
|
|
"cardholderID": self.cardholderID,
|
|
}
|
|
)
|
|
for credential in self.credentials - existing_door_credentials
|
|
]
|
|
|
|
self.door.controller.doXMLRequest(
|
|
ROOT(E.Credentials({"action": "AD"}, *xml_credentials))
|
|
)
|
|
|
|
def update_schedules(self):
|
|
roles = [
|
|
E.Role(
|
|
{
|
|
"roleID": self.cardholderID,
|
|
"scheduleID": self.door.schedules_map[schedule],
|
|
"resourceID": "0",
|
|
}
|
|
)
|
|
for schedule in self.schedules
|
|
]
|
|
|
|
self.door.controller.doXMLRequest(
|
|
ROOT(
|
|
E.RoleSet(
|
|
{"action": "UD", "roleSetID": self.cardholderID}, E.Roles(*roles)
|
|
)
|
|
)
|
|
)
|
|
|
|
|
|
def update_door(door: Door, dry_run: bool = False):
|
|
members = [
|
|
DoorMember.from_membershipworks_member(membershipworks_member, door)
|
|
for membershipworks_member in (Member.objects.with_is_active()).all()
|
|
]
|
|
|
|
cardholders = {
|
|
member.membershipworks_id: member
|
|
for member in [
|
|
DoorMember.from_cardholder(ch, door)
|
|
for ch in door.controller.get_cardholders()
|
|
]
|
|
}
|
|
|
|
existing_door_credentials = {
|
|
Credential(hex=c.attrib["rawCardNumber"])
|
|
for c in door.controller.get_credentials()
|
|
}
|
|
|
|
# TODO: can I combine requests?
|
|
for member in members:
|
|
# cardholder did not exist, so add them
|
|
if member.membershipworks_id not in cardholders:
|
|
logger.info(f"Adding Member {member.full_name}: {member}")
|
|
if not dry_run:
|
|
resp = door.controller.doXMLRequest(
|
|
ROOT(E.Cardholders({"action": "AD"}, E.Cardholder(member.attribs)))
|
|
)
|
|
member.cardholderID = resp.find("{*}Cardholders/{*}Cardholder").attrib[
|
|
"cardholderID"
|
|
]
|
|
|
|
member.update_attribs()
|
|
member.update_credentials(existing_door_credentials, members)
|
|
member.update_schedules()
|
|
|
|
# cardholder exists, compare contents
|
|
else:
|
|
existing_cardholder = cardholders.pop(member.membershipworks_id)
|
|
member.cardholderID = existing_cardholder.cardholderID
|
|
|
|
if member.attribs != existing_cardholder.attribs:
|
|
changes = {
|
|
k: f'"{existing_cardholder.attribs[k]}" -> "{v}"'
|
|
for k, v in member.attribs.items()
|
|
if existing_cardholder.attribs[k] != v
|
|
}
|
|
logger.info(f"Updating profile for {member.full_name}: {changes}")
|
|
if not dry_run:
|
|
member.update_attribs()
|
|
|
|
if member.credentials != existing_cardholder.credentials:
|
|
logger.info(
|
|
f"Updating card for {member.full_name}:"
|
|
f" {existing_cardholder.credentials} -> {member.credentials}"
|
|
)
|
|
if not dry_run:
|
|
member.update_credentials(
|
|
existing_door_credentials,
|
|
members,
|
|
old_credentials=existing_cardholder.credentials,
|
|
)
|
|
|
|
if member.schedules != existing_cardholder.schedules:
|
|
logger.info(
|
|
f"Updating schedule for {member.full_name}:"
|
|
f" {existing_cardholder.schedules} -> {member.schedules}"
|
|
)
|
|
if not dry_run:
|
|
member.update_schedules()
|
|
|
|
# TODO: delete cardholders that are no longer members?
|
|
|
|
|
|
def q_update_all_doors():
|
|
for door in Door.objects.all():
|
|
async_task(
|
|
update_door,
|
|
door,
|
|
cluster="internal",
|
|
group=f"Update HID Door Controller - {door.name}",
|
|
)
|