Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Sanitize filenames and handle member issues in attendance.py example script #155

Merged
merged 10 commits into from
Oct 20, 2024
58 changes: 34 additions & 24 deletions attendance.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@
import asyncio
import csv
import os
import re
from datetime import date

from config import password, username
Expand Down Expand Up @@ -31,16 +32,15 @@


async def main():
s = spond.Spond(username=username, password=password)
events = await s.get_events(min_start=args.f, max_start=args.t)
session = spond.Spond(username=username, password=password)
events = await session.get_events(min_start=args.f, max_start=args.t)

if not os.path.exists("./exports"):
os.makedirs("./exports")

for e in events:
filename = os.path.join(
"./exports", f"{e['startTimestamp']}-{e['heading']}.csv"
)
base_filename = _sanitise_filename(f"{e['startTimestamp']}-{e['heading']}")
filename = os.path.join("./exports", base_filename + ".csv")
with open(filename, "w", newline="") as csvfile:
spamwriter = csv.writer(
csvfile, delimiter=",", quotechar='"', quoting=csv.QUOTE_MINIMAL
Expand All @@ -50,81 +50,91 @@ async def main():
["Start", "End", "Description", "Name", "Answer", "Organizer"]
)
for o in e["owners"]:
person = await s.get_person(o["id"])
full_name = person["firstName"] + " " + person["lastName"]
name = await _derive_member_name(session, o["id"])
spamwriter.writerow(
[
e["startTimestamp"],
e["endTimestamp"],
e["heading"],
full_name,
name,
o["response"],
"X",
]
)
if args.a is True:
for r in e["responses"]["acceptedIds"]:
person = await s.get_person(r)
full_name = person["firstName"] + " " + person["lastName"]
name = await _derive_member_name(session, r)
spamwriter.writerow(
[
e["startTimestamp"],
e["endTimestamp"],
e["heading"],
full_name,
name,
"accepted",
]
)
for r in e["responses"]["declinedIds"]:
person = await s.get_person(r)
full_name = person["firstName"] + " " + person["lastName"]
name = await _derive_member_name(session, r)
spamwriter.writerow(
[
e["startTimestamp"],
e["endTimestamp"],
e["heading"],
full_name,
name,
"declined",
]
)
for r in e["responses"]["unansweredIds"]:
person = await s.get_person(r)
full_name = person["firstName"] + " " + person["lastName"]
name = await _derive_member_name(session, r)
spamwriter.writerow(
[
e["startTimestamp"],
e["endTimestamp"],
e["heading"],
full_name,
name,
"unanswered",
]
)
for r in e["responses"]["unconfirmedIds"]:
person = await s.get_person(r)
full_name = person["firstName"] + " " + person["lastName"]
name = await _derive_member_name(session, r)
spamwriter.writerow(
[
e["startTimestamp"],
e["endTimestamp"],
e["heading"],
full_name,
name,
"unconfirmed",
]
)
for r in e["responses"]["waitinglistIds"]:
person = await s.get_person(r)
full_name = person["firstName"] + " " + person["lastName"]
name = await _derive_member_name(session, r)
spamwriter.writerow(
[
e["startTimestamp"],
e["endTimestamp"],
e["heading"],
full_name,
name,
"waitinglist",
]
)

await s.clientsession.close()
await session.clientsession.close()


async def _derive_member_name(spond_session, member_id: str) -> str:
"""Return full name, or id if member can't be matched."""
try:
person = await spond_session.get_person(member_id)
except KeyError:
return member_id
return f"{person['firstName']} {person['lastName']}"


def _sanitise_filename(input_str: str) -> str:
"""Strip leading/trailing whitespace, spaces to underscores, remove chars not
alphanumeric/underscore/hyphen."""
output_str = str(input_str).strip().replace(" ", "_")
return re.sub(r"(?u)[^-\w.]", "", output_str)


loop = asyncio.new_event_loop()
Expand Down