add feature: react to launchpad bugs

.forgejo/workflows/maubot.yaml

@@ -15,15 +15,6 @@ jobs:
 
     steps:
       - uses: actions/checkout@v4
-      - run: cp ubottu/config.yaml.deploy ubottu/config.yaml
-      - run: rm -f ubottu/config.yaml.deploy
-      - run: rm -f ubottu/config.yaml.default
-      - run: sed -i "s/%%UNSHARED_SECRET%%/${UNSHARED_SECRET}/g" ubottu/config.yaml
-      - run: sed -i "s/%%HOMESERVER_URL%%/${HOMESERVER_URL}/g" ubottu/config.yaml
-      - run: sed -i "s/%%HOMESERVER_SECRET%%/${HOMESERVER_SECRET}/g" ubottu/config.yaml
-      - run: sed -i "s/%%HOMESERVER_DOMAIN%%/${HOMESERVER_DOMAIN}/g" ubottu/config.yaml
-      - run: sed -i "s/%%ADMIN_PW%%/${ADMIN_PW}/g" ubottu/config.yaml
-      - run: sed -i "s/%%PUBLIC_URL%%/${PUBLIC_URL}/g" ubottu/config.yaml
       - run: pip install maubot
       - run: mbc build # Build the project
       - run: mkdir -p output # Ensure output directory exists, `-p` to prevent error if already exists

USAGE.md

@@ -14,6 +14,15 @@ The available facts are available at [here](https://maubot.haxxors.com/factoids/
 **Example 2**: `!noble | Bob`
 **Response**: Bob: Ubuntu 24.04 (Noble Numbat) will be the 40th...
 
+## Launchpad Bugs
+
+The bot reacts to URLs from bugs.launchpad.net and to "bug [#]bugnumber" in messages.
+
+**Example 1**: got this issue right now https://bugs.launchpad.net/snapd/+bug/2052688 on my computer
+**Example 2**: i am affected by bug 2052688 on my computer
+**Example 3**: i am affected by bug #2052688 on my computer
+**Response**: Launchpad Bug #2052688 in snapd "run-snapd-ns-snapd\x2ddesktop [...] long time when system enter in shutdown" [Undecided, New]
+
 ## Time Commands
 
 ### `!time <city>`

ubottu/bot.py

@@ -28,34 +28,11 @@ class Ubottu(Plugin):
     return safe_string
 
   async def pre_start(self) -> None:
-    #if await self.get_ubottu_db('https://ubottu.com/ubuntu3.db'):
-    #  self.db = sqlite3.connect("/home/maubot/.ubottu/ubuntu3.db")
-    #else:
-    #  return False
     return True
 
   async def start(self) -> None:
     self.config.load_and_update()
     self.flood_protection = FloodProtection()
-    
-  async def get_ubottu_db(self, url):
-    """Load a file from a URL into an in-memory filesystem."""
-    # Create a filename if required
-    u = urlparse(url)
-    fn = "/home/maubot/.ubottu/" + os.path.basename(u.path)
-    #if os.path.isfile(fn):
-    #  return fn
-    requests.packages.urllib3.util.connection.HAS_IPV6 = False
-    with requests.get(url, stream=True) as r:
-        r.raise_for_status()  # Checks if the request was successful
-        # Open the local file in binary write mode
-        with open(fn, 'wb+') as f:
-            for chunk in r.iter_content(chunk_size=8192): 
-                # If you have a chunk of data, write it to the file
-                if chunk:
-                    f.write(chunk)
-        f.close()
-    return fn
 
   def check_access(self, sender, room_id):
       if sender in self.config["whitelist"] and room_id in self.config["rooms"]:
@@ -77,6 +54,14 @@ class Ubottu(Plugin):
         #print(data)
         await evt.reply(data['employees'][0]['email'])
 
+  async def lookup_launchpad_bug(self, bug_id):
+    url = 'http://127.0.0.1:8000/bugtracker/api/bugtracker/launchpad/' + str(bug_id) + '/'
+    resp = await self.http.get(url)
+    if resp.status == 200:
+      data = await resp.json()
+      return data
+    return False
+
   async def lookup_factoid(self, command_name, to_user, evt):
       api_url = 'http://127.0.0.1:8000/factoids/api/facts/'
       url = api_url + command_name + '/?format=json'
@@ -101,9 +86,27 @@ class Ubottu(Plugin):
             await evt.respond(value)
           return True
       return False
+  
+  @command.passive("bug #?(\d+)|https?:\/\/bugs\.launchpad\.net\/.+/(\d+)")
+  async def command_bug(self, evt: MessageEvent, match: Tuple[str]) -> None:
+    if match:
+        if match[1]:
+          bug_id = match[1]
+        if match[2]:
+          bug_id = match[2] 
+    if self.flood_protection.flood_check_bug(bug_id) and self.flood_protection.flood_check(evt.sender):
+        data = await self.lookup_launchpad_bug(bug_id)
+        if data:
+            if data['package'] != '':
+              package = ' in ' + '[' + data['package'] + '](' + data['target_link'] + ')'           
+            msg = f"Launchpad Bug [#{data['id']}]({data['link']}){package} \"{data['title']}\" [{data['importance']}, {data['status']}]"
+            await evt.respond(msg)
+            return True
+    return False
+
 
   @command.passive("^!(.+)$")
-  async def command(self, evt: MessageEvent, match: Tuple[str]) -> None:
+  async def command_e(self, evt: MessageEvent, match: Tuple[str]) -> None:
     # allow all rooms and users, only enable flood protection
     #if self.check_access(evt.sender, evt.room_id):
     if self.flood_protection.flood_check(evt.sender):
@@ -120,8 +123,7 @@ class Ubottu(Plugin):
       #block !tr factoid to allow translation
       if command_name == 'tr':
         return False
-
-
+        
       #reload stuff
       if command_name == 'reload' and self.check_access_sender(evt.sender):
         if self.pre_start():
@@ -142,6 +144,19 @@ class Ubottu(Plugin):
           if data:
             await evt.respond('The current time in ' + data['location'] + ' is ' + data['local_time'])
 
+      if command_name == 'bug':
+        if len(args) == 1:
+          package = ''
+          bug_id = int(args[0])
+          data = await self.lookup_launchpad_bug(bug_id)
+          if data:
+            if data['package'] != '':
+              package = ' in ' + '[' + data['package'] + '](' + data['target_link'] + ')'           
+            msg = f"Launchpad Bug [#{data['id']}]({data['link']}){package} \"{data['title']}\" [{data['importance']}, {data['status']}]"
+            await evt.respond(msg)
+            
+        return False
+
       #!package lookup command
       if command_name == 'package' or command_name == 'depends':
         apt = Apt()

ubottu/floodprotection.py

@@ -4,8 +4,11 @@ from time import time
 class FloodProtection:
     def __init__(self):
         self.user_commands = defaultdict(list)  # Stores timestamps of commands for each user
+        self.bug_ids = defaultdict(list)  # Stores timestamps of commands for each user
         self.max_commands = 3
         self.time_window = 60  # 60 seconds
+        self.time_window_bug = 180  # 3 minutes
+        self.max_commands_bugs = 1
 
     def flood_check(self, user_id):
         """Check if a user can send a command based on flood protection limits."""
@@ -22,4 +25,20 @@ class FloodProtection:
             return True  # Allow the command if under the limit
         
         # Otherwise, do not allow the command
+        return False
+    def flood_check_bug(self, bug_id):
+        """Check if a user can send a command based on flood protection limits."""
+        current_time = time()
+        if bug_id not in self.bug_ids:
+            self.bug_ids[bug_id] = [current_time]
+            return True 
+
+        # Remove bugs outside the time window
+        self.bug_ids[bug_id] = [timestamp for timestamp in self.bug_ids[bug_id] if current_time - timestamp < self.time_window_bug]
+
+        if len(self.bug_ids[bug_id]) < self.max_commands_bugs:
+            self.bug_ids[bug_id].append(current_time)
+            return True  # Allow the bug if under the limit
+        
+        # Otherwise, do not allow the bug
         return False