buckbanzai/seattlecitylight-mastodon-bot
1
0
Fork 0
Code Issues 7 Pull requests Projects Releases Packages Wiki Activity

Compare commits

base: buckbanzai:3fd95e037ce03845e3d734fc4a86419deb1d17ac
Branches Tags
buckbanzai:main
buckbanzai:edit-posts
buckbanzai:quieter-restoration-time
buckbanzai:switch-map-generator
..
compare: buckbanzai:f4b00c1bcac14bebe87c913432b022fa76fdde87
Branches Tags
buckbanzai:edit-posts
buckbanzai:main
buckbanzai:quieter-restoration-time
buckbanzai:switch-map-generator

No commits in common. "3fd95e037ce03845e3d734fc4a86419deb1d17ac" and "f4b00c1bcac14bebe87c913432b022fa76fdde87" have entirely different histories.
3fd95e037c ... f4b00c1bca

4 changed files with 1 additions and 201 deletions
Show stats Expand all files Collapse all files

6
.gitignore vendored
View file

@ -160,9 +160,3 @@ cython_debug/
# option (not recommended) you can uncomment the following to ignore the entire idea folder.
#.idea/
.vscode/
# Exclude items that may be used during development
*.secret
*.db
.DS_Store

2
LICENSE
View file

@ -1,6 +1,6 @@
MIT License
Copyright (c) 2024 Liam Steckler
Copyright (c) 2024 buckbanzai
Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:

3
requirements.txt
View file

@ -1,3 +0,0 @@
Mastodon.py==1.8.1
requests==2.28.2
sqlalchemy==2.0.25

191
scl.py
View file

@ -1,191 +0,0 @@
from datetime import datetime
import requests
from bs4 import BeautifulSoup
from mastodon import Mastodon
import sqlite3
from typing import List
from typing import Optional
from sqlalchemy import ForeignKey, select
from sqlalchemy import String
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import Session
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import relationship
from sqlalchemy.exc import NoResultFound
from sqlalchemy import create_engine
post_datetime_format = "%b %e %l:%M %p"
scl_events_url = "https://utilisocial.io/datacapable/v2/p/scl/map/events"
scl_events_response = requests.get(scl_events_url)
try:
scl_events = scl_events_response.json()
except requests.JSONDecodeError:
print("JSON could not be loaded from SCL API")
raise
mastodon = Mastodon(access_token="scl_bot_mastodon.secret")
class Base(DeclarativeBase):
pass
class SclOutage(Base):
__tablename__ = "scl_outages"
scl_outage_id: Mapped[int] = mapped_column(primary_key=True, unique=True)
outage_user_id: Mapped[str] = mapped_column()
most_recent_post_id: Mapped[str] = mapped_column()
last_updated_time: Mapped[datetime] = mapped_column()
estimated_restoration_time: Mapped[datetime] = mapped_column()
cause: Mapped[str] = mapped_column()
outage_size: Mapped[str] = mapped_column()
status: Mapped[Optional[str]] = mapped_column()
no_longer_in_response_time: Mapped[
Optional[datetime]
] = (
mapped_column()
) # If the event is no longer being returned in the response, this will be set to the current time
def __repr__(self) -> str:
return f"User(scl_outage_id={self.scl_outage_id!r}, most_recent_post_id={self.most_recent_post_id!r}, last_updated_time={self.last_updated_time!r}, no_longer_in_response_time={self.no_longer_in_response_time!r})"
engine = create_engine("sqlite:///scl.db")
Base.metadata.create_all(engine)
with Session(engine) as session:
for event in scl_events:
print("Processing outage with Internal ID {}".format(event["id"]))
start_time = datetime.fromtimestamp(event["startTime"] / 1000)
last_updated_time = datetime.fromtimestamp(event["lastUpdatedTime"] / 1000)
estimated_restoration_time = datetime.fromtimestamp(event["etrTime"] / 1000)
lookup_statement = select(SclOutage).where(
SclOutage.scl_outage_id == event["id"]
)
lookup_result = session.scalars(lookup_statement)
if event["numPeople"] < 250:
outage_size = "Small"
elif event["numPeople"] < 1000:
outage_size = "Medium"
else:
outage_size = "Large"
if "status" in event:
status = event["status"]
else:
status = None
hashtag_string = "#SeattleCityLightOutage #SCLOutage #SCLOutage{}".format(
event["identifier"]
)
try:
existing_record = lookup_result.one()
updated_properties = []
updated_entries = []
if estimated_restoration_time != existing_record.estimated_restoration_time:
existing_record.estimated_restoration_time = estimated_restoration_time
updated_properties.append("estimated restoration")
updated_entries.append(
"Est. Restoration: {}".format(
estimated_restoration_time.strftime(post_datetime_format)
)
)
if event["cause"] != existing_record.cause:
existing_record.cause = event["cause"]
updated_properties.append("cause")
updated_entries.append("Cause: {}".format(event["cause"]))
if outage_size != existing_record.outage_size:
existing_record.outage_size = outage_size
updated_properties.append("outage size")
updated_entries.append("Outage Size: {}".format(outage_size))
if status != existing_record.status:
existing_record.status = status
updated_properties.append("status")
updated_entries.append("Status: {}".format(status))
if updated_properties:
updated_properties.sort()
updated_entries.sort()
if len(updated_properties) == 1:
updated_entries.insert(
0,
"The {} of this outage has been updated.\n".format(
updated_properties[0]
),
)
else:
# TODO: this currently just smashes all of the properties together with commas, it'd be nice to make it actually format it like a sentence
updated_entries.insert(
0,
"The {} of this outage have been updated.\n".format(
", ".join(updated_properties)
),
)
updated_entries.append("")
updated_entries.append(hashtag_string)
mastodon_post_result = mastodon.status_post(
status="\n".join(updated_entries),
in_reply_to_id=existing_record.most_recent_post_id,
visibility="public",
)
existing_record.most_recent_post_id = mastodon_post_result["id"]
session.commit()
except NoResultFound:
# TODO: Logic to post initial to Mastodon,
print("Existing record not found")
post_text = """Seattle City Light is reporting a {} outage in {}.
Start Date: {}
Est. Restoration: {}
Cause: {}
{}""".format(
outage_size.lower(),
event["city"],
start_time.strftime(post_datetime_format),
estimated_restoration_time.strftime(post_datetime_format),
event["cause"],
hashtag_string,
)
mastodon_post_result = mastodon.status_post(
status=post_text,
visibility="public",
)
new_outage_record = SclOutage(
scl_outage_id=event["id"],
outage_user_id=event["identifier"],
most_recent_post_id=mastodon_post_result["id"],
last_updated_time=last_updated_time,
estimated_restoration_time=estimated_restoration_time,
cause=event["cause"],
status=status,
outage_size=outage_size,
)
session.add(new_outage_record)
session.commit()
lookup_active_outages_statement = select(SclOutage).where(
SclOutage.no_longer_in_response_time == None
)
for active_outage in session.scalars(lookup_active_outages_statement):
if not any(event["id"] == active_outage.scl_outage_id for event in scl_events):
# Event ID no longer exists in response
mastodon_post_result = mastodon.status_post(
status="This outage is reported to be resolved.\n\n#SeattleCityLightOutage #SCLOutage #SCLOutage{}".format(
active_outage.outage_user_id
),
in_reply_to_id=active_outage.most_recent_post_id,
visibility="public",
)
active_outage.most_recent_post_id = mastodon_post_result["id"]
active_outage.no_longer_in_response_time = datetime.now()
session.commit()