Initial commit
This commit is contained in:
247
main.py
Normal file
247
main.py
Normal file
@@ -0,0 +1,247 @@
|
||||
# Copyright 2026 Philipp Matthias Schäfer <philipp.matthias.schaefer@posteo.de>
|
||||
#
|
||||
# This file is part of the Spielerplus Chat Watcher application.
|
||||
#
|
||||
# The Spielerplus Chat Watcher application is free software: you can redistribute
|
||||
# it and/or modify it under the terms of the GNU Affero General Public License
|
||||
# as published by the Free Software Foundation, either version 3 of the
|
||||
# License, or (at your option) any later version.
|
||||
#
|
||||
# The Spielerplus Chat Watcher application is distributed in the hope that it
|
||||
# will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Affero
|
||||
# General Public License for more details.
|
||||
#
|
||||
# You should have received a copy of the GNU Affero General Public License
|
||||
# along with the Clean CommonMark library. If not, see
|
||||
# <https://www.gnu.org/licenses/>.
|
||||
"""Spielerplus Chat Watcher.
|
||||
|
||||
Regularly checks for new chat messages in Spielerplus (https://www.spielerplus.de).
|
||||
If a new message is encountered, a notifying email is sent to a configured address.
|
||||
"""
|
||||
|
||||
import logging
|
||||
from datetime import timedelta
|
||||
from email.message import EmailMessage
|
||||
from hashlib import sha256
|
||||
from smtplib import SMTP
|
||||
from time import sleep
|
||||
from types import TracebackType
|
||||
|
||||
from pydantic import EmailStr
|
||||
from pydantic_settings import BaseSettings, SettingsConfigDict
|
||||
from selenium.common import NoSuchElementException, WebDriverException
|
||||
from selenium.webdriver import Firefox
|
||||
from selenium.webdriver.common.by import By
|
||||
from selenium.webdriver.firefox.options import Options
|
||||
from selenium.webdriver.support import expected_conditions
|
||||
from selenium.webdriver.support.ui import WebDriverWait
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
# Interval between two consecutive checks for new chat messages.
|
||||
POLL_INTERVAL = timedelta(minutes=15)
|
||||
|
||||
|
||||
class Configuration(BaseSettings):
|
||||
"""Pydantic Settings based configuration.
|
||||
|
||||
Note that the variables in the .env file that is read to fill an object
|
||||
of this class need to have capitalized names with a prefix as defined
|
||||
below in the model configuration.
|
||||
|
||||
Attributes
|
||||
----------
|
||||
username : str
|
||||
Username of the account for which chat messages should be checked.
|
||||
password : str
|
||||
Password of the account for which chat messages should be checked.
|
||||
form_address : EmailStr
|
||||
Email address used in the From-field of notification emails.
|
||||
to_address : EmailStr
|
||||
Email address to which notification emails are sent.
|
||||
"""
|
||||
|
||||
username: str
|
||||
password: str
|
||||
|
||||
from_address: EmailStr
|
||||
to_address: EmailStr
|
||||
|
||||
model_config = SettingsConfigDict(env_prefix="spcw_", env_file=".env")
|
||||
|
||||
|
||||
def send_email(configuration: Configuration, subject: str) -> None:
|
||||
"""Send an empty email with the given subject to the configured email address.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
configuration : Configuration
|
||||
The configuration object which holds the email address for the From- and
|
||||
To-fields of the to be sent email.
|
||||
subject : str
|
||||
The subject used in the to be sent email.
|
||||
"""
|
||||
message = EmailMessage()
|
||||
|
||||
message["From"] = configuration.from_address
|
||||
message["To"] = configuration.to_address
|
||||
message["Subject"] = subject
|
||||
|
||||
smtp = SMTP("localhost")
|
||||
smtp.send_message(message)
|
||||
logger.debug(
|
||||
'Sent email from "%s" to "%s" with subject "%s"',
|
||||
message["From"],
|
||||
message["To"],
|
||||
message["Subject"],
|
||||
)
|
||||
|
||||
|
||||
class FirefoxDriver:
|
||||
"""Context manager wrapper for Selenium’s Firefox webdriver."""
|
||||
|
||||
def __enter__(self) -> Firefox:
|
||||
"""Create a headless Firefox webdriver object and return it.
|
||||
|
||||
Returns
|
||||
-------
|
||||
Firefox
|
||||
The created webdriver object.
|
||||
"""
|
||||
options = Options()
|
||||
options.add_argument("--headless")
|
||||
self.driver = Firefox(options=options)
|
||||
logger.debug("Created Firefox webdriver")
|
||||
return self.driver
|
||||
|
||||
def __exit__(
|
||||
self,
|
||||
_: type[BaseException] | None,
|
||||
__: BaseException | None,
|
||||
___: TracebackType | None,
|
||||
) -> bool | None:
|
||||
"""Close the previously created webdriver object.
|
||||
|
||||
Returns
|
||||
-------
|
||||
None
|
||||
Nothing.
|
||||
"""
|
||||
self.driver.quit()
|
||||
logger.debug("Closed Firefox webdriver")
|
||||
return None
|
||||
|
||||
|
||||
def get_last_chat_message(configuration: Configuration) -> str | None:
|
||||
"""Use Selenium to retrieve the text of the last posted chat message.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
configuration : Configuration
|
||||
Configuration object that provides the login credentials for
|
||||
Spielerplus.
|
||||
|
||||
Returns
|
||||
-------
|
||||
str | None
|
||||
Either the text of the last chat message, which may be empty, if it was
|
||||
deleted, or None, which indicates that no messages where found.
|
||||
"""
|
||||
with FirefoxDriver() as driver:
|
||||
driver.get("https://www.spielerplus.de/en/site/login")
|
||||
logger.debug("Opened login page")
|
||||
|
||||
current_url = driver.current_url
|
||||
|
||||
try:
|
||||
driver.find_element(By.CSS_SELECTOR, ".cmpclose a").click()
|
||||
logger.debug("Closed cookie banner")
|
||||
except NoSuchElementException:
|
||||
logger.debug("No cookie banner present")
|
||||
except WebDriverException as e:
|
||||
logger.warning(
|
||||
"Caught unexpected exception while trying to close cookie banner: %s", e
|
||||
)
|
||||
|
||||
# login
|
||||
driver.find_element(By.ID, "loginform-email").send_keys(configuration.username)
|
||||
driver.find_element(By.ID, "loginform-password").send_keys(
|
||||
configuration.password
|
||||
)
|
||||
driver.find_element(By.CSS_SELECTOR, "#login-form button").click()
|
||||
|
||||
logger.debug("Submitted login form and wait for login")
|
||||
|
||||
WebDriverWait(driver, 15).until(expected_conditions.url_changes(current_url))
|
||||
|
||||
driver.get("https://www.spielerplus.de/en/chat")
|
||||
|
||||
logger.debug("Navigated to chat")
|
||||
|
||||
messages = driver.find_elements(By.CSS_SELECTOR, ".chat-message-item")
|
||||
if not messages:
|
||||
logger.warning("Found no messages on chat page")
|
||||
return None
|
||||
|
||||
last_message = messages[-1]
|
||||
paragraphs = last_message.find_elements(By.CSS_SELECTOR, ".chat-bubble p")
|
||||
if not paragraphs:
|
||||
paragraphs = []
|
||||
text = "\n".join([p.text for p in paragraphs])
|
||||
|
||||
logger.debug("Scraped last message: %s", text)
|
||||
|
||||
return text
|
||||
|
||||
|
||||
def hash_string(string: str) -> str:
|
||||
"""Calculate and return hash of string.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
string : str
|
||||
The string that should be hashed.
|
||||
|
||||
Returns
|
||||
-------
|
||||
str
|
||||
Hex encoded binary hash of string.
|
||||
"""
|
||||
m = sha256()
|
||||
m.update(string.encode())
|
||||
return m.hexdigest()
|
||||
|
||||
|
||||
def main() -> None:
|
||||
"""Entry point of script."""
|
||||
logging.basicConfig(level=logging.WARNING)
|
||||
|
||||
# The values are read from the .env file, which mypy does not know.
|
||||
configuration = Configuration() # type: ignore[call-arg]
|
||||
|
||||
logger.debug("Loaded configuration.")
|
||||
|
||||
# When the script is started, we will miss all messages until a non-empty
|
||||
# one is posted. An empty message is, as far as we know, a deleted message.
|
||||
last_hash = hash_string("")
|
||||
while True:
|
||||
last_message = get_last_chat_message(configuration)
|
||||
|
||||
if not last_message:
|
||||
send_email(configuration, "Could not fetch last message")
|
||||
continue
|
||||
|
||||
new_hash = hash_string(last_message)
|
||||
|
||||
if last_hash != new_hash:
|
||||
send_email(configuration, "New message(s) in Spielerplus")
|
||||
last_hash = new_hash
|
||||
|
||||
sleep(POLL_INTERVAL.seconds)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user