mirror of
https://codeberg.org/vlw/misskey-microblogger.git
synced 2025-09-13 19:03:41 +02:00
wip: 2024-11-12T05:32:34+0100 (1731385954)
This commit is contained in:
parent
439f970a31
commit
e265431ad2
12 changed files with 189 additions and 98 deletions
2
.gitignore
vendored
2
.gitignore
vendored
|
@ -1,6 +1,6 @@
|
|||
# Config
|
||||
data/users
|
||||
config.json
|
||||
.config.json
|
||||
|
||||
# Bootstrapping
|
||||
__pycache__
|
|
@ -74,6 +74,12 @@
|
|||
}
|
||||
}
|
||||
},
|
||||
"note": {
|
||||
"text_length": {
|
||||
"flux": 0,
|
||||
"average": 0
|
||||
}
|
||||
},
|
||||
"relationships": {
|
||||
"friends": [],
|
||||
"enemies": [],
|
||||
|
|
28
generate.py
28
generate.py
|
@ -1,30 +1,46 @@
|
|||
import sys
|
||||
import json
|
||||
import typing
|
||||
from os import system
|
||||
|
||||
from src.Config import Config
|
||||
from src.User.User import User
|
||||
from src.Generator.GenerateUser import GenerateUser
|
||||
from src.Generator.GenerateRelationships import GenerateRelationships
|
||||
from src.Generate.GenerateUser import GenerateUser
|
||||
from src.Generate.GenerateRelationships import GenerateRelationships
|
||||
|
||||
DEFAULT_GENERATE_USER_COUNT = 5
|
||||
DEFAULT_GENERATE_USER_COUNT = 10
|
||||
|
||||
# Generate a user
|
||||
def generate_user() -> User:
|
||||
while True:
|
||||
user = GenerateUser()
|
||||
print(f"1. Create username: {user.username}")
|
||||
#user.set_api_key(input("2. Paste API key:"))
|
||||
# We need the human's help here since I haven't found a way to get API keys from Misskey automatically
|
||||
user.set_api_key(input("2. Paste API key:"))
|
||||
|
||||
yield user.autorun()
|
||||
|
||||
def main():
|
||||
config = Config()
|
||||
users = []
|
||||
|
||||
for i in (range(sys.argv[1] if len(sys.argv) > 1 else DEFAULT_GENERATE_USER_COUNT)):
|
||||
print(f"\nGenerating user: {i + 1}")
|
||||
# Generate n amount of users
|
||||
for i in (range(int(sys.argv[1]) if len(sys.argv) > 1 else DEFAULT_GENERATE_USER_COUNT)):
|
||||
system("clear")
|
||||
print(f"Generating user: {i + 1}")
|
||||
users.append(next(generate_user()))
|
||||
i += 1
|
||||
|
||||
# Create random relationships for generated users
|
||||
print("Generating random user relationships")
|
||||
GenerateRelationships(users).autorun()
|
||||
|
||||
print("Activating users")
|
||||
# Add generated users to active users in config
|
||||
for user in users:
|
||||
config.add_active_user(user.username)
|
||||
|
||||
config.save_config()
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
18
run.py
18
run.py
|
@ -2,26 +2,18 @@ import typing
|
|||
import random
|
||||
import json
|
||||
|
||||
from src.Config import Config
|
||||
from src.Poster import Poster
|
||||
from src.User.User import User
|
||||
from src.Misskey import Misskey
|
||||
|
||||
def main():
|
||||
with open("config.json", "r") as f:
|
||||
config = json.load(f)
|
||||
config = Config()
|
||||
|
||||
# Don't do ANYTHING this time if the roll against the global activity percentage failed
|
||||
if (random.randint(0, 100) >= config["global"]["activity"]):
|
||||
if (not config.get_global_activity_roll()):
|
||||
return False
|
||||
|
||||
for username in config["active_users"]:
|
||||
user = User(username)
|
||||
|
||||
# Don't do anything for this user if they're not active right now
|
||||
if (not user.is_online()):
|
||||
continue
|
||||
|
||||
Poster(user.username, Misskey(config["server"]["url"], user.config["key"])).autorun()
|
||||
for username in config.get_active_users():
|
||||
Poster(username).autorun()
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
|
28
src/Config.py
Normal file
28
src/Config.py
Normal file
|
@ -0,0 +1,28 @@
|
|||
import json
|
||||
import random
|
||||
import typing
|
||||
from pathlib import Path
|
||||
|
||||
CONFIG_FILEPATH = Path.cwd() / ".config.json"
|
||||
|
||||
class Config():
|
||||
def __init__(self):
|
||||
with open(CONFIG_FILEPATH, "r") as f:
|
||||
self.config = json.load(f)
|
||||
|
||||
# Overwrite modified config
|
||||
def save_config(self) -> bool:
|
||||
with open(CONFIG_FILEPATH, "w") as f:
|
||||
json.dump(self.config, f, indent=4, ensure_ascii=False)
|
||||
|
||||
def add_active_user(self, username: str) -> None:
|
||||
self.config["active_users"].append(username)
|
||||
|
||||
def get_global_activity_roll(self) -> bool:
|
||||
return random.randint(0, 100) < self.config["global"]["activity"]
|
||||
|
||||
def get_misskey_server(self) -> str:
|
||||
return self.config["server"]["url"]
|
||||
|
||||
def get_active_users(self) -> list:
|
||||
return self.config["active_users"]
|
101
src/Generate/GenerateRelationships.py
Normal file
101
src/Generate/GenerateRelationships.py
Normal file
|
@ -0,0 +1,101 @@
|
|||
import typing
|
||||
import random
|
||||
|
||||
from ..Misskey import Misskey
|
||||
from ..Enums import RelationshipType
|
||||
from ..User.User import USER_CONFIG_DIR, User
|
||||
|
||||
CHANCE_FRIEND = 70
|
||||
CHANCE_PARTNER = 30
|
||||
|
||||
MIN_RELATIONSHIPS = 3
|
||||
MAX_RELATIONSHIPS = 7
|
||||
|
||||
RECURSE_LIMIT = 10
|
||||
|
||||
class GenerateRelationships():
|
||||
def __init__(self, users: list = None):
|
||||
self.users = users
|
||||
|
||||
self.partners = []
|
||||
|
||||
# Compute diff between a provided target list of users against list of all users
|
||||
def users_diff(self, target: list) -> list:
|
||||
return list(set(self.users) - set(target))
|
||||
|
||||
def pick_random_user_from_list(self, target: list, ignore: User) -> User | None:
|
||||
available = self.users_diff(target)
|
||||
available.remove(ignore)
|
||||
|
||||
return random.choice(available) if available else None
|
||||
|
||||
# Pick a random partner from available partners
|
||||
def set_random_partner(self, user: User, i: int = 0) -> None:
|
||||
# Pick a random user
|
||||
partner = self.pick_random_user_from_list(self.partners, user)
|
||||
|
||||
# Give up trying to find a partner if none found or we've reached the recurse depth limit
|
||||
if (not partner or i >= RECURSE_LIMIT):
|
||||
return None
|
||||
|
||||
# Don't partner up with this user if they're already a friend or enemy
|
||||
if (user.get_relationship_with_user(partner.username) != RelationshipType.NEUTRAL):
|
||||
return self.set_random_partner(user, i + 1)
|
||||
|
||||
# Set partner's usernames in each other's configs
|
||||
user.config["relationships"][RelationshipType.PARTNER.value] = partner.username
|
||||
partner.config["relationships"][RelationshipType.PARTNER.value] = user.username
|
||||
|
||||
# Mark users as partners
|
||||
self.partners.extend((user, partner))
|
||||
|
||||
# Set a another random user as friend or foe
|
||||
def set_random_user_relationship(self, user: User, i: int = 0) -> None:
|
||||
# Roll if ranomd user should be a friend or enemy
|
||||
relationship = RelationshipType.FRIEND if random.randint(0, 100) < CHANCE_FRIEND else RelationshipType.ENEMY
|
||||
|
||||
# Get available friends (not already a friend, enemy, or self)
|
||||
target = self.pick_random_user_from_list(user.get_friends() + user.get_enemies(), user)
|
||||
|
||||
# Give up trying to find a friend if none found or if we've reached the recurse depth limit
|
||||
if (not target or i >= RECURSE_LIMIT):
|
||||
return None
|
||||
|
||||
# Try again if target user is not neutral with user
|
||||
if (user.get_relationship_with_user(target.username) != RelationshipType.NEUTRAL):
|
||||
return self.set_random_user_relationship(user, i + 1)
|
||||
|
||||
# Set relationship on both users
|
||||
user.config["relationships"][relationship.value].append(target.username)
|
||||
target.config["relationships"][relationship.value].append(user.username)
|
||||
|
||||
return None
|
||||
|
||||
# Save all modified user configs and follow users with new relationships
|
||||
def save_all(self) -> None:
|
||||
for user in self.users:
|
||||
# Save modified config
|
||||
user.save_config()
|
||||
|
||||
# Create a Misskey instance for user
|
||||
mk = Misskey(user.get_api_key())
|
||||
|
||||
# Place partner string in a list if set
|
||||
partner = [user.get_partner()] if user.get_partner() else []
|
||||
|
||||
# Follow all users on Misskey we've added relationships for. It's required for the home timeline
|
||||
for username in user.get_friends() + user.get_enemies() + partner:
|
||||
mk.follow_user(username)
|
||||
|
||||
def autorun(self) -> None:
|
||||
for user in self.users:
|
||||
# Find a random partner for user if they don't have one already and if roll is in bounds
|
||||
if (random.randint(0, 100) < CHANCE_PARTNER and not user.get_partner()):
|
||||
self.set_random_partner(user)
|
||||
|
||||
# Set random relationships for user
|
||||
for i in range(random.randint(MIN_RELATIONSHIPS, MAX_RELATIONSHIPS)):
|
||||
self.set_random_user_relationship(user)
|
||||
|
||||
return self.save_all()
|
||||
|
|
@ -64,7 +64,7 @@ class GenerateUser():
|
|||
def set_api_key(self, key: str) -> None:
|
||||
self.config["key"] = key
|
||||
|
||||
def save(self) -> bool:
|
||||
def save_config(self) -> bool:
|
||||
USER_CONFIG_DIR.mkdir(exist_ok=True)
|
||||
|
||||
with open(USER_CONFIG_DIR / f"{self.username}.json", "w") as f:
|
||||
|
@ -103,5 +103,5 @@ class GenerateUser():
|
|||
for x in NoteTypes:
|
||||
self.config["personality"]["type"]["probability"][x.value] = random.randint(0, 100)
|
||||
|
||||
self.save()
|
||||
self.save_config()
|
||||
return User(self.username)
|
|
@ -1,70 +0,0 @@
|
|||
import typing
|
||||
import random
|
||||
|
||||
from ..User.User import USER_CONFIG_DIR, User
|
||||
|
||||
CHANCE_PARTNER = 10
|
||||
|
||||
RECURSE_LIMIT = 10
|
||||
|
||||
class GenerateRelationships():
|
||||
def __init__(self, users: list = None):
|
||||
self.users = users
|
||||
|
||||
self.partners = []
|
||||
|
||||
# Compute diff between a provided target list of users against list of all users
|
||||
def users_diff(self, target: list) -> list:
|
||||
return list(set(self.users) - set(target))
|
||||
|
||||
def pick_random_user_from_list(self, target: list, ignore: User) -> User:
|
||||
available = self.users_diff(target)
|
||||
available.remove(ignore)
|
||||
|
||||
return random.choice(available)
|
||||
|
||||
def set_random_partner(self, user: User, i: int = 0) -> None:
|
||||
# Give up trying to find a partner if we've reached the recurse depth limit
|
||||
if (i >= RECURSE_LIMIT):
|
||||
return None
|
||||
|
||||
# Pick a random user
|
||||
partner = self.pick_random_user_from_list(self.partners, user)
|
||||
|
||||
# Don't partner up with this user if they're already a friend or enemy
|
||||
if (partner in user.get_enemies() or partner in user.get_friends()):
|
||||
return self.set_random_partner(user, i + 1)
|
||||
|
||||
# Set partner's usernames in each other's configs
|
||||
user.config["relationships"]["partner"] = partner.username
|
||||
partner.config["relationships"]["partner"] = user.username
|
||||
|
||||
# Mark users as partners
|
||||
self.partners.extend((user, partner))
|
||||
|
||||
def set_random_friend(self, user: User, i: int = 0) -> None:
|
||||
# Give up trying to find a friend if we've reached the recurse depth limit
|
||||
if (i >= RECURSE_LIMIT):
|
||||
return None
|
||||
|
||||
# Get available friends (not already a friend, enemy, or self)
|
||||
friend = self.pick_random_user_from_list(user.get_friends() + user.get_enemies(), user)
|
||||
|
||||
if (friend == user.get_partner()):
|
||||
return set_random_friend(user, i + 1)
|
||||
|
||||
user.config["relationships"]["friends"].append(friend.username)
|
||||
friend.config["relationships"]["friends"].append(user.username)
|
||||
|
||||
return None
|
||||
|
||||
def autorun(self) -> None:
|
||||
for user in self.users:
|
||||
# Find a random partner for user if they don't have one already and if roll is in bounds
|
||||
if (not user.get_partner()):
|
||||
self.set_random_partner(user)
|
||||
|
||||
self.set_random_friend(user)
|
||||
|
||||
return None
|
||||
|
|
@ -1,6 +1,8 @@
|
|||
import typing
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
from .Config import Config
|
||||
|
||||
from misskey import Misskey as lib_Misskey
|
||||
from misskey.enum import NotificationsType, NoteVisibility
|
||||
|
||||
|
@ -12,8 +14,8 @@ TIMELINE_FETCH_LIMIT = 5
|
|||
TIMELINE_THROTTLE_SECONDS = 300
|
||||
|
||||
class Misskey(lib_Misskey):
|
||||
def __init__(self, server: str, key: str):
|
||||
super().__init__(server, i=key)
|
||||
def __init__(self, key: str):
|
||||
super().__init__(Config().get_misskey_server(), i=key)
|
||||
|
||||
# Reverse lists returned by Misskey.py because for some stupid reason they're ordered oldest-to-newest
|
||||
@staticmethod
|
||||
|
@ -81,4 +83,10 @@ class Misskey(lib_Misskey):
|
|||
return self.notes_reactions_create(
|
||||
note_id=note["id"],
|
||||
reaction=reaction
|
||||
)
|
||||
|
||||
# Send a follow request to username
|
||||
def follow_user(self, username: str) -> dict:
|
||||
return self.following_create(
|
||||
user_id=self.resolve_user_id(username)
|
||||
)
|
|
@ -2,17 +2,17 @@ import typing
|
|||
from datetime import datetime
|
||||
|
||||
from .Note import Note
|
||||
from .Enums import Intent
|
||||
from .Misskey import Misskey
|
||||
from .User.UserIntent import UserIntent
|
||||
|
||||
from misskey.enum import NoteVisibility
|
||||
|
||||
class Poster():
|
||||
def __init__(self, username: str, mk: Misskey):
|
||||
self.mk = mk
|
||||
self.note = Note(username)
|
||||
def __init__(self, username: str):
|
||||
self.user = UserIntent(username)
|
||||
self.note = Note(username)
|
||||
|
||||
self.mk = Misskey(self.user.get_api_key())
|
||||
|
||||
def note_is_older_than_cooldown(self, note: dict) -> bool:
|
||||
date_now = datetime.now()
|
||||
|
|
|
@ -19,6 +19,13 @@ class User():
|
|||
with open(USER_CONFIG_DIR / f"{self.username}.json", "r") as f:
|
||||
self.config = json.load(f)
|
||||
|
||||
# Overwrite current config to user config file
|
||||
def save_config(self) -> None:
|
||||
USER_CONFIG_DIR.mkdir(exist_ok=True)
|
||||
|
||||
with open(USER_CONFIG_DIR / f"{self.username}.json", "w") as f:
|
||||
json.dump(self.config, f, indent=4, ensure_ascii=False)
|
||||
|
||||
# Check if the user is currently online given their time intervals
|
||||
def is_online(self) -> bool:
|
||||
# Find the first time interval that is within the current time
|
||||
|
@ -38,6 +45,9 @@ class User():
|
|||
# Current time was not in range of any configured intervals
|
||||
return False
|
||||
|
||||
def get_api_key(self) -> str:
|
||||
return self.config["key"]
|
||||
|
||||
def get_post_cooldown(self, visibility: NoteVisibility = NoteVisibility.PUBLIC) -> int:
|
||||
return self.config["actions"][Intent.POST.value][visibility.value]["cooldown"]
|
||||
|
||||
|
|
Loading…
Add table
Reference in a new issue