RSS consumer

consumerrss.py

@@ -0,0 +1,171 @@
+from maubot import Plugin, MessageEvent
+from maubot.handlers import command
+from maubot.handlers import web
+
+# Needed for configuration
+from typing import Type
+from mautrix.util.config import BaseProxyConfig, ConfigUpdateHelper
+from mautrix.util.async_db import UpgradeTable, Scheme, Connection
+from aiohttp.web import Request, Response, json_response
+
+import aiohttp
+import datetime
+
+import xml.etree.ElementTree as et
+
+upgrade_table = UpgradeTable()
+@upgrade_table.register(description="Initial revision")
+async def upgrade_v1(conn: Connection) -> None:
+	# Table contains:
+	# Unique ID (event ID) (Primary)
+	# Timestamp (TIMESTAMP type)
+	# Region (string)
+	# Message (string)
+	await conn.execute(
+		"""CREATE TABLE reports (
+			id TEXT PRIMARY KEY,
+			timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP NOT NULL,
+			region TEXT NOT NULL,
+			message TEXT
+		)"""
+	)
+
+# Ensures a running instance gets an updated config from the Maubot interface
+class Config(BaseProxyConfig):
+	def do_update(self, helper: ConfigUpdateHelper) -> None:
+		helper.copy("command_prefix")
+		helper.copy("allowed_regions")
+		helper.copy("serve_regions")
+		helper.copy("send_reaction")
+
+class ConsumerRSS(Plugin):
+	# Get configuration at startup
+	async def start(self) -> None:
+		self.config.load_and_update()
+	
+	# Get config
+	@classmethod
+	def get_config_class(cls) -> Type[BaseProxyConfig]:
+		return Config
+	
+	# Maubot system needs this to register the table
+	@classmethod
+	def get_db_upgrade_table(cls) -> UpgradeTable | None:
+		return upgrade_table
+	
+	# Get !command_name setting from config to register it
+	def get_command_name(self) -> str:
+		return self.config["command_prefix"]
+	
+	# Checks if a sender if allowed to send for a particular region
+	def validate_sender(self, region: str, sender: str):
+		# Mautrix isn't documented, like at all, so I'm just gonna catch the
+		# error because IDK how to see if a map is inside a map.
+		try: allowed_list = self.config["allowed_regions"][region]
+		except: return False
+		
+		if allowed_list is None: return True # Empty list
+		if sender in allowed_list: return True
+		
+		# Sender not allowed in region config
+		return False
+	
+	# Does the necessary config checks for the given event
+	# Returns a region name if sender is allowed for a region, or None
+	def validate_report(self, evt: MessageEvent, message: str):
+		# Split command (minus !command_name) into tokens
+		tokens = message.split()
+		region = tokens[0].lower()
+		
+		allowed_globally = self.validate_sender("__global__", evt.sender)
+		allowed_region = self.validate_sender(region, evt.sender)
+		if allowed_globally or allowed_region: return region
+		
+		return None
+	
+	# What gets called when !command_name message is sent
+	@command.new(name=get_command_name, help="Report Something")
+	@command.argument("message", pass_raw=True)
+	async def report(self, evt: MessageEvent, message: str) -> None:
+		try:
+			# Check if sender is allowed for a region
+			region = self.validate_report(evt, message)
+			if region is None: return
+			
+			# Feed information
+			split_message = message.split()
+			id = evt.event_id
+			timestamp = dt = datetime.datetime.fromtimestamp(evt.timestamp/1000)
+			self.log.debug(timestamp)
+			region = split_message[0]
+			message = ' '.join(split_message[1:])
+			
+			# Insert into database
+			self.log.debug(await self.database.execute("INSERT INTO reports (id, timestamp, region, message) VALUES ($1, $2, $3, $4)", id, timestamp, region, message))
+			
+			# Mark as inserted
+			await evt.react("👍")
+		except Exception as error:
+			self.log.fatal( f"An error occurred: {error}")
+	
+	@web.get("/feed.rss")
+	async def getfeed(self, req: Request) -> Response:
+		try:
+			# Get region from query
+			query = req.query
+			if not "region" in query:
+				return Response(text="Request must include ?region= query", status=405)
+			region = query["region"]
+			
+			# Check region against served regions
+			if not region in self.config["serve_regions"]:
+				return Response(text="Specified region not served on this consumer", status=404)
+			
+			if region == "__global__":
+				feed_items = await self.database.fetch("SELECT * FROM reports ORDER BY timestamp DESC LIMIT 50")
+			else:
+				feed_items = await self.database.fetch("SELECT * FROM reports where region=$1 ORDER BY timestamp DESC LIMIT 50", region)
+			self.log.debug(feed_items)
+			
+			# Build feed
+			feed = et.Element("rss")
+			feed.set("version", "2.0")
+			
+			# Channel metadata
+			channel = et.SubElement(feed, "channel")
+			et.SubElement(channel, "title").text = f"ICE reports {region}"
+			et.SubElement(channel, "description").text = f"ICE reports for {region}"
+			link = et.SubElement(channel, 'link')
+			link.text = self.config["feed_url"]
+			self_link = et.SubElement(channel, 'atom:link')
+			self_link.set('rel', 'self')
+			self_link.set('href', self.config["feed_url"])
+			self_link.set('type', 'application/rss+xml')
+
+			
+			# Add items
+			for item in feed_items:
+				# Data from feed
+				id = item["id"]
+				timestamp = item["timestamp"]
+				region = item["region"]
+				message = "[" + region + "] " + item["message"]
+				# Item
+				item_element = et.SubElement(channel, "item")
+				# GUID is different because the isPermaLink has to be false
+				guid = et.SubElement(item_element, 'guid')
+				guid.text = id
+				guid.set('isPermaLink', 'false')
+				# Other ite,s
+				et.SubElement(item_element, "title").text = message
+				et.SubElement(item_element, "description").text = message
+				et.SubElement(item_element, "pubDate").text = timestamp.strftime("%a, %d %b %Y %H:%M:%S +0000")
+				
+
+				
+			# Return feed
+			return aiohttp.web.Response(body=et.tostring(feed, encoding="unicode"), content_type='application/rss+xml')
+
+		except Exception as error:
+			self.log.fatal( f"An error occurred: {error}")
+			return Response(text="Internal server error", status=500)