portbot/portbot.py

import os
import sys
import time
from datetime import datetime, timedelta, timezone
import toml
import random
import re
import sched
import math
import string

from mastodon import Mastodon, MastodonNotFoundError
from fedbot.bot import Bot, BotClient

TEST = "test" in sys.argv[1:]
USER_PCT = 35
MIN_MAIN_LEN = 3
MAX_PORT_LEN = 14
MAIN_DICT_PATH = "dicts/main.csv"
USER_DICT_PATH = "dicts/user.dict"
USED_DICT_PATH = "dicts/used.dict"

def next_dt():
    dt = datetime.now(timezone.utc)
    dt -= timedelta(hours        = -1,
                    minutes      = dt.minute,
                    seconds      = dt.second,
                    microseconds = dt.microsecond)
    return dt

config_path = os.path.join(os.path.dirname(sys.argv[0]), "config.toml")
loaded_config = {
    "name": "portmanteaubot",
    **toml.load(config_path)}

AFFIXES = []

def is_affixed(word):
    return any(re.search(suf, word) for suf in AFFIXES)

def overlap_words(left_word, right_word):
    if left_word == right_word:
        return set()

    min_shift = 2

    offset = min_shift
    attempts = set()
    while offset + 2 <= len(left_word[0]):
        if right_word[0].lower().startswith(left_word[0].lower()[offset : offset + 2]):
            word_str = left_word[0][:offset] + right_word[0]
            if len(word_str) >= 6 and not is_affixed(word_str):
                attempts.add(word_str)
        offset += 1

    offset = len(right_word[0]) - 2
    while offset >= 2:
        if left_word[0].lower().endswith(right_word[0].lower()[offset : offset + 2]):
            word_str = left_word[0] + right_word[0][offset + 2:]
            if len(word_str) >= 6 and not is_affixed(word_str):
                attempts.add(word_str)
        offset -= 1

    return attempts

def word_weight(index, length, power):
    a = pow((index + 1) / length, 2)
    return int(350000 * a)

def weights_for(words, power):
    return [word_weight(i, len(words), power = power) for i in range(0, len(words))]

def pick_one_word(words):
    if len(words) == 0:
        return None
   
    return random.choices(list(words.items()), weights = (v[0] for v in words.values()))[0]

def word_diff(a, b):
    seq = difflib.SequenceMatcher(None, a, b)
    return seq.ratio()

class WordMaker:
    def __init__(self):
        print("Loading dictionaries")
        illegal = set(ch for ch in (string.ascii_uppercase + string.punctuation + string.digits + string.whitespace))
        with open (MAIN_DICT_PATH, "r") as f:
            self.main_words = {
                sl[0] :
                    (int(sl[1]), tuple(sl[2].split("="))) for sl in
                        (tuple(l.split(",")) for l in f.read().splitlines()) if 
                    len(sl[0]) >= MIN_MAIN_LEN and
                    not any(c in illegal for c in sl[0])}
        
        with open(USER_DICT_PATH, "r") as f:
            self.user_words = {l : (1, None) for l in f.read().splitlines()}
 
        if os.path.exists(USED_DICT_PATH):
            with open("dicts/used.dict", "r") as f:
                used_words = {l : (1, None) for l in f.read().splitlines()}
        else:
            used_words = dict()

        self.all_words = {k.lower() : v for k, v in {**self.main_words, **self.user_words, **used_words}.items()}

    def extend_word(self, prev_word):
        user_dict = random.randint(0, 100) < USER_PCT
        if user_dict:
            next_dict = self.user_words
        else:
            next_dict = self.main_words

        new_words = dict()
        for w in next_dict.items():
            new_words.update(dict.fromkeys(overlap_words(prev_word, w)))

        if len(new_words) == 0:
            return None

        max_len = max(len(w) for w in new_words)
        for w in new_words:
            new_words[w] = (math.pow(max_len + 1 - len(w), 3), None)

        while len(new_words) > 0:
            new_word = pick_one_word(new_words)
            del new_words[new_word[0]]

            if new_word[0].lower() not in self.all_words:
                return new_word 

        return None

    def get_portmanteau(self, target_times = 1):
        user_dict = random.randint(0, 100) < USER_PCT
        if user_dict:
            words = self.user_words
        else:
            words = self.main_words

        while True:
            while True:
                word = pick_one_word(words)
                
                times = target_times
                while times > 0:
                    ext_word = self.extend_word(word)
                    if ext_word is None:
                        break
                    
                    word = ext_word
                    times -= 1
            
                if times == 0:
                    break

            if len(word[0]) < MAX_PORT_LEN:
                break

        word_str = word[0].lower()

        print(word_str)
        
        return word_str

    def get_portmanteaus(self, count = 10):
        words = set()

        used_words = dict()
        while count > 0:
            word_str = self.get_portmanteau()
            if word_str not in words:
                words.add(word_str)
                used_words[word_str] = (1, None)
                count -= 1
        
        self.all_words.update(used_words)
        
        if not TEST:
            with open("dicts/used.dict", "a") as f:
                f.write("\n".join(used_words.keys()) + "\n")

        return words

class PortBotClient(BotClient):
    def __init__(self, bot, config):
        config = {
            "app_name": "PortmanteuBot",
            "rate_limit": 3,
            "retry_rate": 60,
            "poll_interval": 15,
            **config} 
      
        super().__init__(bot, config)
    
        self.my_id = None

    def on_start(self):
        self.log("Starting")
        self.my_id = self.api.me()["id"]
        pass

    def on_poll(self):
        pass

    def on_status(self, status):
        if status["account"]["id"] != self.my_id:
            return

def post():
    for client_name, client in bot.clients.items():
        words = wm.get_portmanteaus(3)
        print()
        if random.randint(0, 100) <= 100:
            visibility = "public"
        else:
            visibility = "unlisted"
    
        dt = next_dt()
        
        if not TEST:
            client.api.status_post("\n".join(words), visibility = visibility)
            print("Scheduling at", dt)
        
        if TEST:
            scheduler.enter(2, 1, post)
        else:
            scheduler.enterabs(dt.timestamp(), 1, post)

wm = WordMaker()
scheduler = sched.scheduler(timefunc = time.time, delayfunc = time.sleep)

bot = Bot(PortBotClient, loaded_config)
del loaded_config
bot.start()

print("Running")

dt = next_dt()

if TEST:
    scheduler.enter(2, 1, post)
else:
    print("Scheduling at", dt)
    scheduler.enterabs(dt.timestamp(), 1, post)
scheduler.run()