#!/usr/bin/env 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="%(asctime)s %(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]: 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"http://agir.april.org/projects/{self.name}/activity.atom?show_issues=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.exception("Exception raised to irc message parsing") target = None return " ".join(irc_msg.split()), actor, target except Exception: LOGGER.exception("Exception raised to irc message parsing") 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: str) -> None: self.send_raw("JOIN {} \n".format(project.channel)) # Joins channel self.projects.append(project) def get_project(self, name: str) -> str: 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() 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 for message in project.redmine(): self.send_privmsg(project.channel, message) # 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 !") # If the server pings us then we've got to respond! if irc_msg.find("PING :") != -1: self.pong(irc_msg) # 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() 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()