From dc9eee72158d498538e94a366f309737e938851b Mon Sep 17 00:00:00 2001 From: Liam Steckler Date: Sat, 13 Jan 2024 14:24:00 -0800 Subject: [PATCH] Initial release --- scl.py | 191 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 191 insertions(+) create mode 100644 scl.py diff --git a/scl.py b/scl.py new file mode 100644 index 0000000..01a8804 --- /dev/null +++ b/scl.py @@ -0,0 +1,191 @@ +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()