Initial release

.gitignore

@@ -160,3 +160,9 @@ 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

LICENSE

@@ -1,6 +1,6 @@
 MIT License
 
-Copyright (c) 2024 buckbanzai
+Copyright (c) 2024 Liam Steckler
 
 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:
 

requirements.txt

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

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()