Handle JSONDecodeErrors including empty files

robocop_ng/helpers/disabled_ids.py

@@ -2,6 +2,8 @@
 import os
 from typing import Union
 
+from robocop_ng.helpers.notifications import report_critical_error
+
 
 def get_disabled_ids_path(bot) -> str:
  return os.path.join(bot.state_dir, "data/disabled_ids.json")
@@ -22,7 +24,19 @@ def is_ro_section_valid(ro_section: dict[str, str]) -> bool:
 def get_disabled_ids(bot) -> dict[str, dict[str, Union[str, dict[str, str]]]]:
  if os.path.isfile(get_disabled_ids_path(bot)):
  with open(get_disabled_ids_path(bot), "r") as f:
- disabled_ids = json.load(f)
+ try:
+ disabled_ids = json.load(f)
+ except json.JSONDecodeError as e:
+ content = f.read()
+ report_critical_error(
+ bot,
+ e,
+ additional_info={
+ "file": {"length": len(content), "content": content}
+ },
+ )
+ return {}
+
  # Migration code
  if "app_id" in disabled_ids.keys():
  old_disabled_ids = disabled_ids.copy()

robocop_ng/helpers/macros.py

@@ -2,6 +2,8 @@
 import os
 from typing import Optional, Union
 
+from robocop_ng.helpers.notifications import report_critical_error
+
 
 def get_macros_path(bot):
  return os.path.join(bot.state_dir, "data/macros.json")
@@ -10,7 +12,18 @@ def get_macros_path(bot):
 def get_macros_dict(bot) -> dict[str, dict[str, Union[list[str], str]]]:
  if os.path.isfile(get_macros_path(bot)):
  with open(get_macros_path(bot), "r") as f:
- macros = json.load(f)
+ try:
+ macros = json.load(f)
+ except json.JSONDecodeError as e:
+ content = f.read()
+ report_critical_error(
+ bot,
+ e,
+ additional_info={
+ "file": {"length": len(content), "content": content}
+ },
+ )
+ return {}
 
  # Migration code
  if "aliases" not in macros.keys():

robocop_ng/helpers/restrictions.py

@@ -1,14 +1,28 @@
 import json
 import os
 
+from robocop_ng.helpers.notifications import report_critical_error
+
 
 def get_restrictions_path(bot):
  return os.path.join(bot.state_dir, "data/restrictions.json")
 
 
 def get_restrictions(bot):
- with open(get_restrictions_path(bot), "r") as f:
- return json.load(f)
+ if os.path.isfile(get_restrictions_path(bot)):
+ with open(get_restrictions_path(bot), "r") as f:
+ try:
+ return json.load(f)
+ except json.JSONDecodeError as e:
+ content = f.read()
+ report_critical_error(
+ bot,
+ e,
+ additional_info={
+ "file": {"length": len(content), "content": content}
+ },
+ )
+ return {}
 
 
 def set_restrictions(bot, contents):
@@ -18,11 +32,10 @@ def set_restrictions(bot, contents):
 
 def get_user_restrictions(bot, uid):
  uid = str(uid)
- with open(get_restrictions_path(bot), "r") as f:
- rsts = json.load(f)
- if uid in rsts:
- return rsts[uid]
- return []
+ rsts = get_restrictions(bot)
+ if uid in rsts:
+ return rsts[uid]
+ return []
 
 
 def add_restriction(bot, uid, rst):

robocop_ng/helpers/robocronp.py

@@ -2,14 +2,28 @@
 import math
 import os
 
+from robocop_ng.helpers.notifications import report_critical_error
+
 
 def get_crontab_path(bot):
  return os.path.join(bot.state_dir, "data/robocronptab.json")
 
 
 def get_crontab(bot):
- with open(get_crontab_path(bot), "r") as f:
- return json.load(f)
+ if os.path.isfile(get_crontab_path(bot)):
+ with open(get_crontab_path(bot), "r") as f:
+ try:
+ return json.load(f)
+ except json.JSONDecodeError as e:
+ content = f.read()
+ report_critical_error(
+ bot,
+ e,
+ additional_info={
+ "file": {"length": len(content), "content": content}
+ },
+ )
+ return {}
 
 
 def set_crontab(bot, contents):

robocop_ng/helpers/roles.py

@@ -2,6 +2,8 @@
 import os.path
 import os
 
+from robocop_ng.helpers.notifications import report_critical_error
+
 
 def get_persistent_roles_path(bot):
  return os.path.join(bot.state_dir, "data/persistent_roles.json")
@@ -10,7 +12,17 @@ def get_persistent_roles_path(bot):
 def get_persistent_roles(bot) -> dict[str, list[str]]:
  if os.path.isfile(get_persistent_roles_path(bot)):
  with open(get_persistent_roles_path(bot), "r") as f:
- return json.load(f)
+ try:
+ return json.load(f)
+ except json.JSONDecodeError as e:
+ content = f.read()
+ report_critical_error(
+ bot,
+ e,
+ additional_info={
+ "file": {"length": len(content), "content": content}
+ },
+ )
  return {}
 
 

robocop_ng/helpers/userlogs.py

@@ -2,6 +2,8 @@
 import os
 import time
 
+from robocop_ng.helpers.notifications import report_critical_error
+
 userlog_event_types = {
  "warns": "Warn",
  "bans": "Ban",
@@ -16,8 +18,20 @@ def get_userlog_path(bot):
 
 
 def get_userlog(bot):
- with open(get_userlog_path(bot), "r") as f:
- return json.load(f)
+ if os.path.isfile(get_userlog_path(bot)):
+ with open(get_userlog_path(bot), "r") as f:
+ try:
+ return json.load(f)
+ except json.JSONDecodeError as e:
+ content = f.read()
+ report_critical_error(
+ bot,
+ e,
+ additional_info={
+ "file": {"length": len(content), "content": content}
+ },
+ )
+ return {}
 
 
 def set_userlog(bot, contents):