added more config options & made python type-safe(r).
This commit is contained in:
+13
-12
@@ -1,7 +1,7 @@
|
||||
from io import TextIOWrapper
|
||||
import asyncpraw as praw
|
||||
import asyncpraw as praw # type: ignore
|
||||
import os
|
||||
from typing import Final
|
||||
from typing import Final, Any
|
||||
from macros import *
|
||||
import toml
|
||||
|
||||
@@ -11,11 +11,11 @@ CFG_DATA_RE: Final[str] = "reddit"
|
||||
|
||||
|
||||
class Bot:
|
||||
args: dict = {"NO_RUST": True, "dev": True, "py": True, "port": 2920}
|
||||
r_id: str = os.environ.get("ASSISTANT_R_ID")
|
||||
secret: str = os.environ.get("ASSISTANT_R_TOKEN")
|
||||
username: str = os.environ.get("ASSISTANT_R_NAME")
|
||||
password: str = os.environ.get("ASSISTANT_R_PASS")
|
||||
args: dict[str, Any] = {"NO_RUST": True, "dev": True, "py": True, "port": 2920}
|
||||
r_id: str | None = os.environ.get("ASSISTANT_R_ID")
|
||||
secret: str | None = os.environ.get("ASSISTANT_R_TOKEN")
|
||||
username: str | None = os.environ.get("ASSISTANT_R_NAME")
|
||||
password: str | None = os.environ.get("ASSISTANT_R_PASS")
|
||||
|
||||
fetch_limit = 0
|
||||
|
||||
@@ -39,14 +39,14 @@ class Bot:
|
||||
)
|
||||
self.sr_list: list[str] = ["bytedicetesting"]
|
||||
self.sr = None
|
||||
self.data_f: TextIOWrapper = None
|
||||
self.data: dict = {}
|
||||
self.data_f: TextIOWrapper | None = None
|
||||
self.data: dict[str, Any] = {}
|
||||
self.flairs: list[str] = []
|
||||
|
||||
async def initialize(self):
|
||||
self.sr = await self.r.subreddit("+".join(self.sr_list))
|
||||
|
||||
async def set_args(self, args: dict):
|
||||
async def set_args(self, args: dict[str, Any]):
|
||||
self.args = args
|
||||
|
||||
async def stop(self) -> bool:
|
||||
@@ -59,12 +59,13 @@ class Bot:
|
||||
|
||||
async def update_cfg_str(self, new_cfg: str) -> bool:
|
||||
json_cfg = toml.loads(new_cfg)
|
||||
self.update_cfg(json_cfg)
|
||||
await self.update_cfg(json_cfg)
|
||||
return True
|
||||
|
||||
async def update_cfg(self, new_cfg: dict) -> bool:
|
||||
async def update_cfg(self, new_cfg: dict[str, Any]) -> bool:
|
||||
self.sr_list = new_cfg[CFG_DATA_RE]["subreddits"].split("+")
|
||||
self.sr = await self.r.subreddit("+".join(self.sr_list))
|
||||
self.fetch_limit = new_cfg[CFG_DATA_RE]["fetch_limit"]
|
||||
self.flairs = new_cfg[CFG_DATA_RE]["search_flairs"]
|
||||
init_lang(new_cfg["general"]["lang"])
|
||||
return True
|
||||
@@ -1,5 +1,6 @@
|
||||
import json
|
||||
import os
|
||||
import json
|
||||
|
||||
from printColors import PrintColors
|
||||
|
||||
|
||||
@@ -8,14 +9,15 @@ G_LANG: dict[str, str] = {}
|
||||
DATA_PATH_LANG: str = "./data/lang/"
|
||||
|
||||
|
||||
def py_print(*args):
|
||||
def py_print(*args: str):
|
||||
print(
|
||||
PrintColors.FG.blue + "Py",
|
||||
"-",
|
||||
" ".join(args) + PrintColors.Special.reset
|
||||
)
|
||||
|
||||
def py_error(*args):
|
||||
|
||||
def py_error(*args: str):
|
||||
print(
|
||||
PrintColors.BG.red + "ERROR" + PrintColors.Special.reset,
|
||||
PrintColors.FG.blue + "Py",
|
||||
@@ -29,13 +31,13 @@ def lang(k: str) -> str:
|
||||
if G_LANG == {}:
|
||||
py_error("Language must be initialized before use!")
|
||||
t = G_LANG.get(k)
|
||||
if k is None: py_error(f"Key not found in language \"{G_LANG_NAME}\": {k}")
|
||||
if t is None: py_error(f"Key not found in language \"{G_LANG_NAME}\": {k}")
|
||||
return str(t)
|
||||
|
||||
|
||||
def init_lang(lang_name: str):
|
||||
global G_LANG, G_LANG_NAME
|
||||
G_LANG_NAME = lang_name
|
||||
G_LANG_NAME = lang_name # type: ignore
|
||||
|
||||
full_path = f"{DATA_PATH_LANG}{lang_name}.json"
|
||||
|
||||
@@ -50,4 +52,4 @@ def init_lang(lang_name: str):
|
||||
except json.JSONDecodeError as e:
|
||||
py_error(f"Failed to parse JSON for language \"{lang_name}\":\n{e}")
|
||||
|
||||
G_LANG = json_data
|
||||
G_LANG = json_data # type: ignore
|
||||
+9
-7
@@ -4,12 +4,12 @@ import time
|
||||
|
||||
from macros import *
|
||||
import bot as botPy
|
||||
import data
|
||||
import py_data
|
||||
import py_websocket
|
||||
|
||||
|
||||
async def main():
|
||||
sys.stdout.reconfigure(encoding="utf-8")
|
||||
sys.stdout.reconfigure(encoding="utf-8") # type: ignore
|
||||
|
||||
py_print("Creating Reddit bot...")
|
||||
bot = botPy.Bot()
|
||||
@@ -20,8 +20,10 @@ async def main():
|
||||
# args is supposed to be undefined.
|
||||
# It gets defined in Rust.
|
||||
try:
|
||||
await bot.set_args(args)
|
||||
init_lang(lang_name)
|
||||
await bot.set_args(args) # type: ignore
|
||||
py_print("Fetching language file...")
|
||||
init_lang(lang_name) # type: ignore
|
||||
py_print(f"[IMPORTANT] The below message is a test message, it should be written in the language you've selected\nTest message: {lang('log_lang_load_success')}")
|
||||
except NameError:
|
||||
py_print("No command args or language name found from Rust. Don't worry though, we have backup in place.")
|
||||
init_lang("en")
|
||||
@@ -30,17 +32,17 @@ async def main():
|
||||
py_print("ARGS:", str(bot.args))
|
||||
|
||||
py_print("Reading config file...")
|
||||
await data.read_cfg(bot)
|
||||
await py_data.read_cfg(bot)
|
||||
|
||||
py_print("Reading Reddit data...")
|
||||
rd = data.read_data(bot)
|
||||
rd = py_data.read_data(bot)
|
||||
data_retries = 0
|
||||
|
||||
while not rd :
|
||||
data_retries += 1
|
||||
time.sleep(1)
|
||||
py_print(f"Failed to read data: File doesn't exist yet. Retrying (#{data_retries}/5)...")
|
||||
rd = data.read_data(bot)
|
||||
rd = py_data.read_data(bot)
|
||||
|
||||
if data_retries == 5 and not rd:
|
||||
raise Exception("Couldn't read re_data.json: File doesn't exist")
|
||||
|
||||
+19
-16
@@ -1,10 +1,10 @@
|
||||
import emoji
|
||||
from asyncpraw import models
|
||||
import asyncprawcore as prawcore
|
||||
import asyncpraw.exceptions as exc
|
||||
from asyncpraw import models # type: ignore
|
||||
import asyncprawcore as prawcore # type: ignore
|
||||
import asyncpraw.exceptions as exc # type: ignore
|
||||
import time
|
||||
|
||||
import data
|
||||
import py_data
|
||||
import bot as botPy
|
||||
from macros import *
|
||||
|
||||
@@ -46,7 +46,7 @@ async def add_new_posts(bot: botPy.Bot, max_age: int) -> bool:
|
||||
without_media += 1
|
||||
continue
|
||||
|
||||
post_added = data.add_post_to_data(
|
||||
post_added = py_data.add_post_to_data(
|
||||
bot,
|
||||
details
|
||||
)
|
||||
@@ -60,7 +60,7 @@ async def add_new_posts(bot: botPy.Bot, max_age: int) -> bool:
|
||||
f"{not_added} are removed or already existed, " +
|
||||
f"and {old_posts} were older than the max age threshold.")
|
||||
|
||||
data.write_data(bot)
|
||||
py_data.write_data(bot)
|
||||
|
||||
return True
|
||||
|
||||
@@ -72,6 +72,8 @@ async def fetch_posts_with_flair(bot: botPy.Bot, flair_names: list[str]) -> list
|
||||
f"flair:{flair_names[0]}" if len(flair_names) == 1\
|
||||
else " OR ".join(f"flair:{flair}" for flair in flair_names)
|
||||
|
||||
if bot.sr is None: return []
|
||||
|
||||
# ~36 OG-art posts per week, round limit to 50, 75 or 100
|
||||
async for post in bot.sr.search(f"{flair_names_str}", sort="new", limit=bot.fetch_limit):
|
||||
posts.append(post)
|
||||
@@ -79,8 +81,8 @@ async def fetch_posts_with_flair(bot: botPy.Bot, flair_names: list[str]) -> list
|
||||
return posts
|
||||
|
||||
|
||||
def has_media(post: models.Submission) -> tuple[bool, str, int, list[str]]:
|
||||
media_type: str = None
|
||||
def has_media(post: models.Submission) -> tuple[bool, str | None, int, list[str]]:
|
||||
media_type: str | None = None
|
||||
media_count = 0
|
||||
media_urls: list[str] = []
|
||||
|
||||
@@ -109,7 +111,7 @@ def has_media(post: models.Submission) -> tuple[bool, str, int, list[str]]:
|
||||
return (media_type != None, media_type, media_count, media_urls)
|
||||
|
||||
|
||||
async def from_url(bot: botPy.Bot, url: str) -> tuple[bool, models.Submission]:
|
||||
async def from_url(bot: botPy.Bot, url: str) -> tuple[bool, models.Submission | None]:
|
||||
try:
|
||||
post: models.Submission = await bot.r.submission(url=url)
|
||||
return True, post
|
||||
@@ -120,11 +122,12 @@ async def from_url(bot: botPy.Bot, url: str) -> tuple[bool, models.Submission]:
|
||||
return False, None
|
||||
|
||||
|
||||
def get_post_details(post: models.Submission, added_by_h: bool = False) -> data.PostData:
|
||||
def get_post_details(post: models.Submission, added_by_h: bool = False) -> py_data.PostData:
|
||||
media = has_media(post)
|
||||
|
||||
return data.PostData(
|
||||
return py_data.PostData(
|
||||
post.shortlink,
|
||||
post.subreddit,
|
||||
post.title,
|
||||
post.score,
|
||||
int(post.created_utc),
|
||||
@@ -135,12 +138,12 @@ def get_post_details(post: models.Submission, added_by_h: bool = False) -> data.
|
||||
)
|
||||
|
||||
|
||||
async def add_post_url(bot, url: str, approve: bool = False, added_by_h: bool = False) -> bool:
|
||||
async def add_post_url(bot: botPy.Bot, url: str, approve: bool = False, added_by_h: bool = False) -> bool:
|
||||
result, post = await from_url(bot, url)
|
||||
|
||||
if not result:
|
||||
return False
|
||||
|
||||
if not result: return False
|
||||
if post is None: return False
|
||||
|
||||
post_data = get_post_details(post, added_by_h)
|
||||
post_data.approved_by_human = approve
|
||||
return data.add_post_to_data(bot, post_data, True)
|
||||
return py_data.add_post_to_data(bot, post_data, True)
|
||||
@@ -1,4 +1,4 @@
|
||||
import asyncpraw.models as models
|
||||
import asyncpraw.models as models # type: ignore
|
||||
|
||||
from macros import *
|
||||
import bot as botPy
|
||||
@@ -2,6 +2,7 @@ import os
|
||||
import toml
|
||||
import json
|
||||
import time
|
||||
from typing import Any
|
||||
|
||||
import bot as botPy
|
||||
from macros import *
|
||||
@@ -17,10 +18,11 @@ class PostData:
|
||||
def __init__(
|
||||
self,
|
||||
url: str,
|
||||
subreddit: str,
|
||||
title: str,
|
||||
upvotes: int,
|
||||
date_unix: int,
|
||||
media_type: str,
|
||||
media_type: str | None,
|
||||
media_urls: list[str],
|
||||
removed: bool = False,
|
||||
removed_by: str | None = None,
|
||||
@@ -33,6 +35,7 @@ class PostData:
|
||||
approved_by_human: bool = False,
|
||||
approved_by_ris: bool = False
|
||||
):
|
||||
self.subreddit = subreddit
|
||||
self.removed = removed
|
||||
self.removed_by = removed_by
|
||||
self.removed_reason = removed_reason
|
||||
@@ -50,7 +53,7 @@ class PostData:
|
||||
self.approved_by_human = approved_by_human
|
||||
self.approved_by_ris = approved_by_ris
|
||||
|
||||
def to_json(self):
|
||||
def to_json(self) -> dict[str, Any]:
|
||||
return {
|
||||
"removed": {
|
||||
"removed": self.removed,
|
||||
@@ -58,6 +61,7 @@ class PostData:
|
||||
"reason": self.removed_reason
|
||||
},
|
||||
"post_data": {
|
||||
"subreddit": self.subreddit,
|
||||
"title": self.title,
|
||||
"upvotes": self.upvotes,
|
||||
"date_unix": self.date_unix,
|
||||
@@ -92,10 +96,9 @@ def read_data(bot: botPy.Bot) -> bool:
|
||||
|
||||
py_print("re_data.json not found, creating new from preset...")
|
||||
with open(os.path.join(DEFAULT_PATH, "re_data_preset.json"), "r") as f:
|
||||
data_preset_json = json.load(f)
|
||||
data_preset_json: dict[str, Any] = json.load(f)
|
||||
|
||||
data_preset_json[botPy.RE_DATA_POSTS].pop("EXAMPLE VALUE", None)
|
||||
data_preset_json[botPy.RE_DATA_POSTS].pop("EXAMPLE VALUE DELETED", None)
|
||||
|
||||
with open(r_path, "w") as f:
|
||||
json.dump(data_preset_json, f, indent = 2)
|
||||
@@ -110,6 +113,8 @@ def read_data(bot: botPy.Bot) -> bool:
|
||||
|
||||
|
||||
def write_data(bot: botPy.Bot) -> bool:
|
||||
if bot.data_f is None: return False
|
||||
|
||||
bot.data_f.seek(0)
|
||||
json.dump(bot.data, bot.data_f, indent=2)
|
||||
bot.data_f.truncate()
|
||||
@@ -128,7 +133,7 @@ async def read_cfg(bot: botPy.Bot) -> bool:
|
||||
data_preset_json = toml.load(f)
|
||||
|
||||
with open(r_path, "w") as f:
|
||||
toml.dump(data_preset_json, f, indent = 2)
|
||||
toml.dump(data_preset_json, f, indent = 2) # type: ignore
|
||||
|
||||
bot.data_f = open(r_path, "r+")
|
||||
|
||||
@@ -224,12 +229,12 @@ def set_vote_post(
|
||||
if remove_vote:
|
||||
if user not in target_voters:
|
||||
return False
|
||||
target_voters.remove(user)
|
||||
target_voters.remove(user) # type: ignore
|
||||
|
||||
else:
|
||||
if user in target_voters:
|
||||
return False
|
||||
target_voters.add(user)
|
||||
target_voters.add(user) # type: ignore
|
||||
|
||||
bot.data[botPy.RE_DATA_POSTS][url]["votes"]["voters_re"] = list(re_voters)
|
||||
bot.data[botPy.RE_DATA_POSTS][url]["votes"]["voters_dc"] = list(dc_voters)
|
||||
+22
-21
@@ -1,13 +1,13 @@
|
||||
import websockets
|
||||
import asyncio
|
||||
import json
|
||||
from typing import Any
|
||||
|
||||
from macros import *
|
||||
import bot as botPy
|
||||
import data
|
||||
import py_data
|
||||
import posts
|
||||
import cmds
|
||||
import macros
|
||||
import py_cmds
|
||||
|
||||
ws_global = None
|
||||
is_connected = False
|
||||
@@ -36,9 +36,10 @@ async def websocket_client(bot: botPy.Bot):
|
||||
|
||||
while True:
|
||||
response = await ws.recv()
|
||||
if not response.startswith("json:"):
|
||||
str_response = str(response)
|
||||
if not str_response.startswith("json:"): # type: ignore
|
||||
py_print(f"Received from Rust: {response}")
|
||||
await parse_json(response, bot)
|
||||
await parse_json(str_response, bot)
|
||||
|
||||
|
||||
async def parse_json(response: str, bot: botPy.Bot):
|
||||
@@ -50,7 +51,7 @@ async def parse_json(response: str, bot: botPy.Bot):
|
||||
if json_response["print"]: py_print(f"Received from Rust: {response}")
|
||||
|
||||
result = await json_to_func(json_response, bot)
|
||||
await ws_global.ping()
|
||||
if ws_global is not None: await ws_global.ping()
|
||||
await send_message(f"json:{json.dumps(result)}")
|
||||
except json.JSONDecodeError as e:
|
||||
if bot.args["dev"]: py_print(f"failed to parse json: {json_str}\n reason: {e}")
|
||||
@@ -62,29 +63,29 @@ def run_thread(bot: botPy.Bot):
|
||||
loop.run_until_complete(websocket_client(bot))
|
||||
|
||||
|
||||
async def json_to_func(v: dict, bot: botPy.Bot) -> dict:
|
||||
if "type" not in v or "value" not in v or not isinstance(v, dict):
|
||||
async def json_to_func(v: dict[str, Any], bot: botPy.Bot) -> dict[str, Any]:
|
||||
if "type" not in v or "value" not in v:
|
||||
if bot.args["dev"]: py_print("JSON is not a dictionary or does not include \"type\" and \"value\" keys.")
|
||||
return
|
||||
return result_json(False, True)
|
||||
if v["type"] != "function":
|
||||
v_type = v["type"]
|
||||
if bot.args["dev"]: py_print(f"Type \"{v_type}\" is not supported.")
|
||||
return
|
||||
return result_json(False, True)
|
||||
|
||||
value_supported = True
|
||||
r = False
|
||||
|
||||
match v["value"]:
|
||||
case "update_data_file": r = data .write_data (bot)
|
||||
case "respond_mentions": r = await cmds .respond_to_mention(bot)
|
||||
case "add_new_posts": r = await posts .add_new_posts (bot, *v["args"])
|
||||
case "add_post_url": r = await posts .add_post_url (bot, *v["args"])
|
||||
case "remove_post_url": r = data .remove_post (bot, *v["args"])
|
||||
case "set_approve_post": r = data .set_approve_post (bot, *v["args"])
|
||||
case "set_vote_post": r = data .set_vote_post (bot, *v["args"])
|
||||
case "remove_old_posts": r = data .remove_old_posts (bot, *v["args"])
|
||||
case "update_cfg": r = await bot .update_cfg_str (*v["args"])
|
||||
case "stop_praw": r = await bot .stop ()
|
||||
case "update_data_file": r = py_data.write_data (bot)
|
||||
case "respond_mentions": r = await py_cmds.respond_to_mention(bot)
|
||||
case "add_new_posts": r = await posts .add_new_posts (bot, *v["args"])
|
||||
case "add_post_url": r = await posts .add_post_url (bot, *v["args"])
|
||||
case "remove_post_url": r = py_data.remove_post (bot, *v["args"])
|
||||
case "set_approve_post": r = py_data.set_approve_post (bot, *v["args"])
|
||||
case "set_vote_post": r = py_data.set_vote_post (bot, *v["args"])
|
||||
case "remove_old_posts": r = py_data.remove_old_posts (bot, *v["args"])
|
||||
case "update_cfg": r = await bot .update_cfg_str (*v["args"])
|
||||
case "stop_praw": r = await bot .stop ()
|
||||
case _: value_supported = False
|
||||
|
||||
print_result = v["print"]
|
||||
@@ -97,5 +98,5 @@ async def json_to_func(v: dict, bot: botPy.Bot) -> dict:
|
||||
return result_json(r, print_result)
|
||||
|
||||
|
||||
def result_json(bool: bool, print_result: bool) -> dict:
|
||||
def result_json(bool: bool, print_result: bool) -> dict[str, Any]:
|
||||
return {"type": "result", "value": bool, "print": print_result}
|
||||
Reference in New Issue
Block a user