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
  • joyjoy@lemmy.world
    3·
    3 years ago

    I recommend either checking response.status_code < 300 or calling response.raise_for_status() in case of server errors. In these cases, the server might return a html page.

  • MoreCoffee@lemm.eeEnglish
    3·
    3 years ago

    Your get_admin_ids() needs some try/except. I would check the requests call and you can specifically check for requests.exceptions.JSONDecodeError exceptions.

  • CoderSupreme@programming.devOP
    21·
    3 years ago
    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
  • o11c@programming.dev
    1·
    3 years ago

    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.