message_admins.py
import csv
import requests
from typing import Any, Dict, List
from pythorhead import Lemmy
from config import *
def get_admin_ids(domain: str) -> List[str]:
url = f"https://{domain}/api/v3/site"
response = requests.get(url)
data = response.json()
admin_ids = [
item["person"]["id"] for item in data["admins"] if item["person"].get("admin")
]
return admin_ids
def send_private_message(lemmy: Lemmy, user_id: str, content: str) -> bool:
pm = lemmy.private_message(content, user_id)
if not pm:
print("Sending private message failed")
return False
return True
def get_domains_from_csv(filename: str) -> List[str]:
domains = []
with open(filename, "r") as csvfile:
reader = csv.DictReader(csvfile)
for row in reader:
domains.append(row["Domain"])
return domains
def main() -> None:
domains = get_domains_from_csv("output.csv")
lemmy = Lemmy(LEMMY_INSTANCE_URL)
lemmy.log_in(LEMMY_USERNAME, LEMMY_PASSWORD)
for domain in domains:
admin_ids = get_admin_ids(domain)
content = f'''
Dear admins,
I hope this message finds you well. I wanted to bring to your attention a recent change in stance regarding bots on the lemm.ee instance. As stated in the post lemm.ee/post/1847525, lemm.ee has decided to introduce new rules to limit repost bots. I am part of the !youtube_feed@lemm.ee reposting community, and manually posting everything has become quite burdensome. Therefore, I am kindly requesting if it would be possible to move our community to the {domain} instance. Thank you for your consideration.
Best regards,
{LEMMY_USERNAME}
'''
for admin_id in admin_ids:
pm = lemmy.private_message(content, admin_id)
if not pm:
print("Sending private message failed")
else:
print(f"Sent private message to {admin_id}")
if __name__ == "__main__":
main()
Traceback
Traceback (most recent call last):
File "/home/user/code/python/lemmy/venv/lib/python3.11/site-packages/requests/models.py", line 971, in json
return complexjson.loads(self.text, **kwargs)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/usr/lib/python3.11/json/__init__.py", line 346, in loads
return _default_decoder.decode(s)
^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/usr/lib/python3.11/json/decoder.py", line 337, in decode
obj, end = self.raw_decode(s, idx=_w(s, 0).end())
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/usr/lib/python3.11/json/decoder.py", line 355, in raw_decode
raise JSONDecodeError("Expecting value", s, err.value) from None
json.decoder.JSONDecodeError: Expecting value: line 1 column 1 (char 0)
During handling of the above exception, another exception occurred:
Traceback (most recent call last):
File "/home/user/code/python/lemmy/spam_admins.py", line 60, in
main()
File "/home/user/code/python/lemmy/spam_admins.py", line 42, in main
admin_ids = get_admin_ids(domain)
^^^^^^^^^^^^^^^^^^^^^
File "/home/user/code/python/lemmy/spam_admins.py", line 13, in get_admin_ids
data = response.json()
^^^^^^^^^^^^^^^
File "/home/user/code/python/lemmy/venv/lib/python3.11/site-packages/requests/models.py", line 975, in json
raise RequestsJSONDecodeError(e.msg, e.doc, e.pos)
requests.exceptions.JSONDecodeError: Expecting value: line 1 column 1 (char 0)
output.csv
Domain,Latency
lemmy.ml,459.306001663208
lemm.ee,467.4856662750244
midwest.social,514.9338245391846
lemmy.one,550.7447719573975
slrpnk.net,586.4944458007812
Aside: Note that requests
is sloppy there, it should use either raise ... from e
to make the cause explicit, or from None
to hide it. Default propagation is supposed to imply that the second exception was unexpected.
This is not StackOverflow
I recommend either checking response.status_code
or calling response.raise_for_status()
in case of server errors. In these cases, the server might return a html page.
def get_admin_ids(domain: str) -> List[str]:
url = f"https://{domain}/api/v3/site"
response = requests.get(url)
try:
data = response.json()
except json.JSONDecodeError:
print(f"Error: Invalid or empty JSON response for domain {domain}")
return []
admin_ids = [
item["person"]["id"] for item in data["admins"] if item["person"].get("admin")
]
return admin_ids
Your get_admin_ids() needs some try/except. I would check the requests call and you can specifically check for requests.exceptions.JSONDecodeError exceptions.