#!/usr/bin/python3 import socket import sys import re import select import os.path import configparser import logging from datetime import datetime, timedelta, timezone from typing import Generator, Tuple, List import feedparser LOGGER = logging.getLogger(__name__) logging.basicConfig(level=logging.INFO, format="%(levelname)s %(threadName)s %(name)s %(message)s") # Help configuration. HELP_LIST = ["help", "faq", "aide"] HELLO_LIST = ["hello", "yo", "bonjour", "salut"] # Load configuration. CONFIG_FILENAME = "/etc/redminebot/redminebot.conf" if len(sys.argv) == 2: CONFIG_FILENAME = sys.argv[1] if os.path.isfile(CONFIG_FILENAME): LOGGER.info('Using configuration file: %s', CONFIG_FILENAME) CONFIG = configparser.ConfigParser() CONFIG.read(CONFIG_FILENAME) DEFAULT_SERVER = CONFIG.get("IRCSection", "irc.server") DEFAULT_NICKNAME = CONFIG.get("IRCSection", "irc.nickname") REGISTERED = CONFIG.get("IRCSection", "irc.registered") PASSWORD = CONFIG.get("IRCSection", "irc.password") PROJECT_ID_1 = CONFIG.get("IRCSection", "irc.projects.1.id") PROJECT_CHANNEL_1 = CONFIG.get("IRCSection", "irc.projects.1.channel") PROJECT_ID_2 = CONFIG.get("IRCSection", "irc.projects.2.id") PROJECT_CHANNEL_2 = CONFIG.get("IRCSection", "irc.projects.2.channel") PROJECT_ID_3 = CONFIG.get("IRCSection", "irc.projects.3.id") PROJECT_CHANNEL_3 = CONFIG.get("IRCSection", "irc.projects.3.channel") else: LOGGER.error("Missing configuration file.") sys.exit() class Project: def __init__(self, project: str, channel: str): self.name = project self.channel = channel self.redmine_next = datetime.now(tz=timezone.utc) self.redmine_latest = datetime.now(tz=timezone.utc) def generate_messages(self) -> Generator[str, None, None]: print("Generating messages for project", self.name) now = datetime.now(tz=timezone.utc) if now < self.redmine_next: return latest_new = self.redmine_latest self.redmine_next = self.redmine_next + timedelta(seconds=10) redmine = feedparser.parse(f"https://agir.april.org/projects/{self.name}/activity.atom?show_issues=1&show_changesets=1") for i in reversed(range(len(redmine.entries))): redmine_entry = redmine.entries[i] entry_updated = parse_redmine_datetime(redmine_entry.updated) if entry_updated > self.redmine_latest: yield f"Redmine: ({redmine_entry.link}): {redmine_entry.title}" # find the new latest time if entry_updated > latest_new: latest_new = entry_updated self.redmine_latest = latest_new def generate_command_regex(options: List[str]) -> str: pattern = "(" for option in options: pattern += option pattern += '|' pattern = pattern[:-1] pattern += ")" return pattern def parse_irc_messages(irc_msg: str) -> Tuple[str, str, str]: try: actor = irc_msg.split(":")[1].split("!")[0] try: target = irc_msg.split(":")[1].split(" ")[2] except Exception: LOGGER.warning("Failed to parse target for actor='{}' in the following incoming irc message: {}".format(actor, irc_msg)) target = None return " ".join(irc_msg.split()), actor, target except Exception: LOGGER.exception("Failed to parse the following incoming irc message: {}".format(irc_msg)) return None, None, None def parse_redmine_datetime(redmine_date: str) -> datetime: date_with_iso_tz = redmine_date[:-1] + '+00:00' return datetime.fromisoformat(date_with_iso_tz) class Bot: def __init__(self, server: str, bot_nick: str): self.bot_nick = bot_nick self.hello_regex = re.compile(generate_command_regex(HELLO_LIST), re.I) self.help_regex = re.compile(generate_command_regex(HELP_LIST), re.I) self.server = server self.projects = [] self.irc_socket = None def send_raw(self, msg: str) -> None: self.irc_socket.send(msg.encode("utf-8", "ignore")) def send_privmsg(self, channel: str, message: str) -> None: self.send_raw(f"PRIVMSG {channel} :{message}\n") def connect(self) -> None: self.irc_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.irc_socket.connect((self.server, 6667)) self.send_raw(f"USER {self.bot_nick} {self.bot_nick} {self.bot_nick} :Robot Agir April" f".\n") # bot authentication self.send_raw(f"NICK {self.bot_nick}\n") # Assign the nick to the bot. if not PASSWORD.strip() and REGISTERED is True: self.send_privmsg("NickServ", f"IDENTIFY {self.bot_nick} {PASSWORD}") def add_project(self, project: Project) -> None: self.send_raw("JOIN {} \n".format(project.channel)) # Joins channel self.projects.append(project) def get_project(self, name: str) -> Project: for project in self.projects: if name[0] != '#' and project.name == name: return project if name[0] == '#' and project.channel == name: return project return None # Main loop def loop_forever(self) -> None: last_read = datetime.utcnow() while True: # Loop forever ready_to_read, _fd2, _fd3 = select.select([self.irc_socket], [], [], 1) if ready_to_read: last_read = datetime.utcnow() irc_msg = self.msg_handler() # If the server pings us then we've got to respond! if irc_msg.find("PING :") != -1: self.pong(irc_msg) LOGGER.info("Responding to ping message: {}".format(irc_msg)) continue irc_msg, actor, channel = parse_irc_messages(irc_msg) if irc_msg is not None: self.message_response(irc_msg, actor, channel) if datetime.utcnow() - last_read > timedelta(minutes=10): raise Exception('timeout: nothing to read on socket since 10 minutes') # Responds to server Pings. def pong(self, irc_msg: str) -> None: response = f"PONG :{irc_msg.split('PING :')[1]}\n" self.send_raw(response) self._check_global_updates() def _check_global_updates(self) -> None: LOGGER.info("Checking all projects updates...") for project in self.projects: for message in project.generate_messages(): self.send_privmsg(project.channel, message) def _check_project_updates(self, channel_name: str) -> None: LOGGER.info("Checking project of channel %s updates...", channel_name) project = self.get_project(channel_name) if not project: LOGGER.error("Cannot find project for channel %s", channel_name) return print(project) try: for message in project.generate_messages(): self.send_privmsg(project.channel, message) except Exception as e: LOGGER.error("Error while checking project %s updates: %s", project.name, e) # Parses messages and responds to them appropriately. def message_response(self, irc_msg: str, actor: str, channel: str) -> None: # If someone talks to (or refers to) the bot. if (':!' in irc_msg or self.bot_nick.lower() in irc_msg.lower()) and \ actor != self.bot_nick and \ "PRIVMSG".lower() in irc_msg.lower(): if self.hello_regex.search(irc_msg): self.bot_hello(channel, 'Yo!') if self.help_regex.search(irc_msg): self.bot_help(channel) if 'refresh' in irc_msg.lower(): self.send_privmsg(channel, "Raffraîchissement en cours") self._check_project_updates(channel) self.send_privmsg(channel, "Fait !") # Responds to a user that inputs "Hello Mybot". def bot_hello(self, channel: str, greeting: str) -> None: self.send_privmsg(channel, greeting) # Explains what the bot is when queried. def bot_help(self, channel: str) -> None: self.send_privmsg(channel, "Bonjour, je suis un bot qui reconnaît les options !help, !refresh et !bonjour") project = self.get_project(channel) if not project: LOGGER.warning("Project of channel %s not found", channel) return self.send_privmsg(channel, f"Périodiquement, je vais afficher les actualités " f"de http://agir.april.org/projects/{project.name}.") # Reads the messages from the server and adds them to the Queue and prints # them to the console. This function will be run in a thread, see below. def msg_handler(self) -> str: # pragma: no cover (this excludes this function from testing) new_msg = self.irc_socket.recv(2048) # receive data from the server if not new_msg: LOGGER.error("Empty recv. It seems I’ve lost my mind. I stop to be reborn.") sys.exit(7) else: new_msg = new_msg.decode("utf-8").strip('\n\r') # removing any unnecessary linebreaks if new_msg != '' and new_msg.find("PING :") == -1: LOGGER.debug(new_msg) return new_msg def main() -> None: LOGGER.info("redmine bot starting…") redmine_bot = Bot(DEFAULT_SERVER, DEFAULT_NICKNAME) redmine_bot.connect() LOGGER.info("redmine bot connected") if PROJECT_ID_1 and PROJECT_CHANNEL_1: redmine_bot.add_project(Project(PROJECT_ID_1, PROJECT_CHANNEL_1)) if PROJECT_ID_2 and PROJECT_CHANNEL_2: redmine_bot.add_project(Project(PROJECT_ID_2, PROJECT_CHANNEL_2)) if PROJECT_ID_3 and PROJECT_CHANNEL_3: redmine_bot.add_project(Project(PROJECT_ID_3, PROJECT_CHANNEL_3)) redmine_bot.loop_forever() if __name__ == "__main__": main()