Add showdiff

.drone.yml

@@ -0,0 +1,29 @@
+---
+kind: pipeline
+type: exec
+name: build
+
+platform:
+  os: linux
+  arch: arm64
+
+steps:
+- name: build
+  commands:
+    - docker build -t gcrkrause/mastodon-blocklist-deploy .
+- name: push
+  environment:
+    USERNAME:
+      from_secret: docker-hub-user
+    PASSWORD:
+      from_secret: docker-hub-pw
+  commands:
+    - docker login -u $USERNAME -p $PASSWORD
+    - docker push gcrkrause/mastodon-blocklist-deploy
+    - docker image prune -a -f
+  when:
+    event:
+      exclude:
+      - pull_request
+    branch: 
+    - main

README.md

@@ -9,11 +9,11 @@ the mastodon server, this is supposed to be automated with Drone CI.
 
 In order to compare the list entries, we can read the whole blocklist
 using [the get endpoint](https://docs.joinmastodon.org/methods/admin/domain_blocks/#get). At the same time we read the
-whole file in the repository, make a comparision
+whole file in the repository, make a comparison
 and [remove](https://docs.joinmastodon.org/methods/admin/domain_blocks/#delete) unblocked domains from the blocklist
 and [add](https://docs.joinmastodon.org/methods/admin/domain_blocks/#create) newly added.
 
-Since we have several attributes for a domain blog, a simple `.txt` file might not be sufficient. We probably want to
+Since we have several attributes for a domain block, a simple `.txt` file might not be sufficient. We probably want to
 set the severity, reject_media, reject_reports and comments. This means we need a human-readable, easily python-readable
 and structured file format. Since Python 3.11 got native support for [toml](https://toml.io/) and it
 supports [Array of Tables](https://toml.io/en/v1.0.0#array-of-tables), I'd prefer to use this.
@@ -25,13 +25,15 @@ supports [Array of Tables](https://toml.io/en/v1.0.0#array-of-tables), I'd prefe
 ##
 
 ```
-$ mastodon_blocklist_deploy -h
-usage: mastodon_blocklist_deploy [-h] [-s SERVER] [-t TOKEN] [-i INPUT_FILE] [-r REMOTE_BLOCKLIST] [-o OUTPUT] [-v] [-n] {diff,deploy,export}
+usage: mastodon_blocklist_deploy [-h] [-s SERVER] [-t TOKEN] [-i INPUT_FILE] [-r REMOTE_BLOCKLIST] [-o OUTPUT] [-v] [-n]
+                                 [--format FORMAT] [--private]
+                                 {diff,deploy,export}
 
 Deploy blocklist updates to a mastodon server
 
 positional arguments:
-  {diff,deploy,export}  Either use 'diff' to check the difference between local blockĺist and the blocklist on the server, 'deploy' to apply the current local blocklist or 'export' to export the remote blocklist into a local file.
+  {diff,deploy,export}  Either use 'diff' to check the difference between local blockĺist and the blocklist on the server, 'deploy'
+                        to apply the current local blocklist or 'export' to export the remote blocklist into a local file.
 
 options:
   -h, --help            show this help message and exit
@@ -47,6 +49,8 @@ options:
                         Filename where to export the blocklist
   -v, --verbose
   -n, --no-delete       Do not delete existing blocks
+  --format FORMAT       Export format: toml|markdown|csv|json
+  --private             When the flag is set, private comment will also be exported.
 ```
 
 ## Obtain a server token
@@ -87,4 +91,4 @@ mastodon_blocklist_deploy diff -s yourserver -t yourtoken -i blocklist.toml
 
 ```
 mastodon_blocklist_deploy apply -s yourserver -t yourtoken -i blocklist.toml
-```
+```

mastodon_blocklist_deploy/cli.py

@@ -7,6 +7,8 @@ import os
 import toml
 
 from mastodon_blocklist_deploy.models import Instance
+from mastodon_blocklist_deploy.helpers import blocklist_to_markdown, blocklist_to_toml, blocklist_to_csv, \
+    blocklist_to_json
 
 
 def load_blocklist_file(filename: str) -> [Instance]:
@@ -38,16 +40,64 @@ def load_blocklist_from_instance(server: str, token: str) -> [Instance]:
     else:
         raise ConnectionError(f"Could not connect to the server ({response.status_code}: {response.reason})")
 
+
 def remove_key_from_dict(dict, key):
     del dict[key]
     return dict
 
+
+def exporter(blocklist, output=None, format: str = "toml", private: bool = False):
+    if format == "toml":
+        exported_text = blocklist_to_toml(blocklist, private)
+    if format == "csv":
+        exported_text = blocklist_to_csv(blocklist, private)
+    if format == "markdown":
+        exported_text = blocklist_to_markdown(blocklist, private)
+    if format == "json":
+        exported_text = blocklist_to_json(blocklist, private)
+
+    # Output the text
+    if output is not None:
+        with open(output, "w") as f:
+            f.write(exported_text)
+    else:
+        print(exported_text)
+
+
+def merge(input_file, merge_target, format: str = "toml", private: bool = False, overwrite=False):
+    """Shows a table in the CLI comparing the local and remote blocklist"""
+    from rich.table import Table
+    from rich.console import Console
+    input_blocklist = load_blocklist_file(input_file)
+    merge_target_blocklist = load_blocklist_file(merge_target)
+    for input_instance in input_blocklist:
+        # If the block is already there with the same parameters we do nothing
+        if input_instance in merge_target_blocklist:
+            continue
+        # Check if there is a domain in the merge target where the input domain is similar
+        try:
+            merge_target_instance = [merge_target_instance for merge_target_instance in merge_target_blocklist if input_instance.domain == merge_target_instance.domain][0]
+            if not overwrite:
+                key_input = ""
+                while key_input not in ("i", "O"):
+                    print(f"Different settings for {input_instance.domain} detected.")
+                    Instance.show_diff(input_instance, merge_target_instance)
+                    key_input = input("Keep input (i) or original (o) [i/O]")
+            elif key_input == "i":
+                merge_target_blocklist.append(merge_target_instance)
+            else:
+                merge_target_blocklist.append(input_instance)
+        except KeyError:
+            pass
+
+
 def cli():
     parser = argparse.ArgumentParser(description='Deploy blocklist updates to a mastodon server')
-    parser.add_argument('action', choices=['diff', 'deploy', 'export'],
+    parser.add_argument('action', choices=['diff', 'deploy', 'export', 'merge'],
                         help="Either use 'diff' to check the difference between local blockĺist and the blocklist on "
                              "the server, 'deploy' to apply the current local blocklist or 'export' to export the remote "
-                             "blocklist into a local file.")
+                             "blocklist into a local file. merge can be used to merge a blocklist (given by -i) into "
+                             "another (-o)")
     parser.add_argument('-s', '--server', help="The address of the server where you want to deploy (e.g. "
                                                "mastodon.social)")
     parser.add_argument('-t', '--token', help="Authorization token")
@@ -56,6 +106,9 @@ def cli():
     parser.add_argument('-o', '--output', help="Filename where to export the blocklist")
     parser.add_argument('-v', '--verbose', action='store_true')
     parser.add_argument('-n', '--no-delete', action='store_true', help="Do not delete existing blocks")
+    parser.add_argument('--format', default="toml", type=str, help="Export format: toml|markdown|csv|json")
+    parser.add_argument('--private', action='store_true', help="When the flag is set, private comment will also be "
+                                                               "exported.")
     args = parser.parse_args()
     if args.verbose:
         logging.basicConfig(level=logging.DEBUG)
@@ -67,17 +120,17 @@ def cli():
     else:
         token = os.getenv('MBD_TOKEN')
 
-
-
-    """if there is a remote blocklist provided load this instead of fetching it from a server (for debugging reasons)"""
-    if args.remote_blocklist:
-        with open(args.remote_blocklist) as f:
-            remote_blocklist = blocklist_json_to_instances(json.load(f))
-    else:
-        remote_blocklist = load_blocklist_from_instance(server=args.server, token=token)
+    """Get a remote blocklist only when necessary"""
+    if args.action in ["diff", "deploy"]:
+        """if there is a remote blocklist provided load this instead of fetching it from a server (for debugging reasons)"""
+        if args.remote_blocklist:
+            with open(args.remote_blocklist) as f:
+                remote_blocklist = blocklist_json_to_instances(json.load(f))
+        else:
+            remote_blocklist = load_blocklist_from_instance(server=args.server, token=token)
 
     """Load local blocklist only when needed"""
-    if args.action in ["diff", "deploy"]:
+    if args.action in ["diff", "deploy", "merge"]:
         if args.input_file:
             blocklist_filename = args.input_file
         else:
@@ -94,11 +147,9 @@ def cli():
         diffs = Instance.list_diffs(local_blocklist, remote_blocklist)
         Instance.apply_blocks_from_diff(diffs, args.server, token, args.no_delete)
     elif args.action == "export":
-        if not args.output:
-            print(toml.dumps({"instances": [remove_key_from_dict(b.__dict__, 'id') for b in remote_blocklist]}))
-        else:
-            with open(args.output, "w") as f:
-                toml.dump({"instances": [remove_key_from_dict(b.__dict__, 'id') for b in remote_blocklist]}, f)
+        exporter(remote_blocklist, args.output, args.format, args.private)
+    elif args.action == "merge":
+        merge(args.input_file, args.output, args.format, args.private)
 
 
 if __name__ == "__main__":

mastodon_blocklist_deploy/helpers.py

@@ -0,0 +1,36 @@
+from mastodon_blocklist_deploy.models import Instance
+import toml
+import io
+import csv
+import json
+
+def blocklist_to_markdown(blocklist: [Instance], private: bool = False):
+    if private:
+        markdown_string = "| Instance | Status | Reason | Private Comment |\n | --- | --- | --- |\n"
+    else:
+        markdown_string = "| Instance | Status | Reason |\n | --- | --- | --- |\n"
+    for instance in blocklist:
+
+        if private:
+            markdown_string += f"| {instance.domain} | {instance.severity} | {instance.public_comment} | {instance.private_comment} |\n"
+        else:
+            markdown_string += f"| {instance.domain} | {instance.severity} | {instance.public_comment} |\n"
+
+    return markdown_string
+
+def blocklist_to_toml(blocklist: [Instance], private: bool = False):
+    toml_string = toml.dumps({"instances": [b.as_dict(private) for b in blocklist]})
+    return toml_string
+
+def blocklist_to_csv(blocklist: [Instance], private: bool = False):
+    csv_string = io.StringIO()
+    blocklist_as_dict = [b.as_dict(private) for b in blocklist]
+    keys = blocklist_as_dict[0].keys()
+    w = csv.DictWriter(csv_string, keys)
+    w.writeheader()
+    w.writerows(blocklist_as_dict)
+    return csv_string.getvalue()
+
+def blocklist_to_json(blocklist: [Instance], private: bool = False):
+    json_string = json.dumps([b.as_dict(private) for b in blocklist])
+    return json_string

mastodon_blocklist_deploy/models.py

@@ -11,12 +11,7 @@ class Instance:
         self.reject_reports = False
         self.id = None
 
-        """Remote blocks and local blocks are parsed differently"""
-        try:
-            instance_dict["id"]
-            self.parse_remote_block(instance_dict)
-        except KeyError:
-            self.parse_local_block(instance_dict)
+        self.parse_block(instance_dict)
 
     def __str__(self):
         return f"{self.domain}: {self.severity}"
@@ -27,37 +22,29 @@ class Instance:
     def status_str(self):
         return f"{self.severity}\nReject reports: {self.reject_reports}\nReject media: {self.reject_media}\nObfuscate: {self.obfuscate}"
 
-    def parse_remote_block(self, instance_dict):
-        self.domain = instance_dict["domain"]
-        self.id = instance_dict["id"]
-        self.severity = instance_dict["severity"]
-        self.public_comment = instance_dict["public_comment"]
-        self.private_comment = instance_dict["private_comment"]
-        self.obfuscate = instance_dict["obfuscate"]
-        self.reject_media = instance_dict["reject_media"]
-        self.reject_reports = instance_dict["reject_reports"]
+    def as_dict(self, private=False):
+        keys = ["domain", "severity", "public_comment", "obfuscate", "reject_media", "reject_reports"]
+        if private:
+            keys.append("private_comment")
+        exportable = {}
+        for key in keys:
+            exportable[key] = getattr(self, key)
+        return exportable
 
-    def parse_local_block(self, instance_dict):
-        try:
-            self.name = instance_dict["name"]
-        except KeyError:
-            pass
-        self.domain = instance_dict["domain"]
-        self.severity = instance_dict["severity"]
-        self.public_comment = instance_dict["public_comment"]
-        self.private_comment = instance_dict["private_comment"]
-        try:
-            self.obfuscate = instance_dict["obfuscate"]
-        except KeyError:
-            pass
-        try:
-            self.reject_media = instance_dict["reject_media"]
-        except KeyError:
-            pass
-        try:
-            self.reject_reports = instance_dict["reject_reports"]
-        except KeyError:
-            pass
+    def parse_block(self, instance_dict):
+        # this specifies possible properties and default values if not found on the remote source. If a default is None
+        # the value is required and the parse will fail
+        properties_and_defaults = [("domain", None), ("severity", "suspend"), ("public_comment", ""),
+                                   ("private_comment", ""), ("obfuscate", False), ("reject_media", True),
+                                   ("reject_reports", True)]
+        for key, default in properties_and_defaults:
+            try:
+                setattr(self, key, instance_dict[key])
+            except KeyError:
+                if default is not None:
+                    setattr(self, key, default)
+                else:
+                    raise KeyError(f"The key {key} was not in the instance_dict response.")
 
     def apply(self, server, token, block_id=None):
         """Applies instance block on the remote server"""
@@ -78,7 +65,7 @@ class Instance:
             response = requests.put(f'https://{server}/api/v1/admin/domain_blocks/{block_id}', data=data,
                                     headers=headers)
         if response.status_code != 200:
-            raise ConnectionError(f"Could not apply block ({response.status_code}: {response.reason})")
+            raise ConnectionError(f"Could not apply block for {self.domain} ({response.status_code}: {response.reason})")
 
     def delete(self, server: str, token: str):
         """Deletes the instance from the blocklist on the remote server"""
@@ -151,3 +138,18 @@ class Instance:
                 table.add_row(diff["local"].domain, diff["local"].status_str(), diff["remote"].status_str())
         console = Console()
         console.print(table)
+
+    @staticmethod
+    def show_diff(instanceA, instanceB, column_names=('Input', 'Original')):
+        from rich.table import Table
+        from rich.console import Console
+        table = Table(title="Differences", expand=True, show_lines=True)
+
+        table.add_column("Attribute", style="cyan")
+        table.add_column(column_names[0], style="green")
+        table.add_column(column_names[1], style="magenta")
+        compare_attributes = ["domain", "severity", "obfuscate", "private_comment", "public_comment", "reject_media", "reject_reports"]
+        for attr in compare_attributes:
+            table.add_row(attr, str(getattr(instanceA, attr)), str(getattr(instanceB, attr)))
+        console = Console()
+        console.print(table)