split main.py into multiple files
This commit is contained in:
@@ -3,6 +3,7 @@ target/
|
|||||||
Cargo.lock
|
Cargo.lock
|
||||||
**/*.rs.bk
|
**/*.rs.bk
|
||||||
*.pdb
|
*.pdb
|
||||||
|
__pycache__/
|
||||||
.vscode/
|
.vscode/
|
||||||
|
|
||||||
# program-created data
|
# program-created data
|
||||||
|
|||||||
@@ -2,15 +2,17 @@
|
|||||||
- [x] ~~Embed creation tool~~
|
- [x] ~~Embed creation tool~~
|
||||||
- [ ] Reddit bot that scrapes images with tag "Original Art" and posts them in Discord server
|
- [ ] Reddit bot that scrapes images with tag "Original Art" and posts them in Discord server
|
||||||
- [x] ~~Scrape the data~~
|
- [x] ~~Scrape the data~~
|
||||||
- [ ] Put it in a JSON
|
- [x] ~~Put it in a JSON~~
|
||||||
- [ ] Ship it to Discord using the Rust bot
|
- [ ] Ship it to Discord using the Rust bot
|
||||||
- [ ] Allow updating the data autonomously and via manual commands.
|
- [ ] Allow updating the data autonomously and via manual commands.
|
||||||
- [ ] Manually add posts (via `u/[bot] add` or `/bk_week_add [url]`)
|
- [ ] Manually add posts (via `u/[bot] add` or `/bk_week_add [url]`)
|
||||||
- [ ] Manually remove posts (via `/bk_week_remove [url]`)
|
- [ ] Manually remove posts (via `/bk_week_remove [url]`)
|
||||||
- [ ] Manually approve posts (via `/bk_week_approve [url]`)
|
- [ ] Manually approve posts (via `/bk_week_approve [url]`)
|
||||||
- [ ] Manually un-approve posts (via `/bk_week_disapprove [url]`)
|
- [ ] Manually un-approve posts (via `/bk_week_disapprove [url]`)
|
||||||
- [ ] Automatically add scraped posts to JSON
|
- [x] ~~Automatically add scraped posts to JSON~~
|
||||||
- [ ] Automatically remove posts older than 7 days from JSON
|
- [ ] Automatically remove posts older than 7 days from JSON
|
||||||
|
- [x] ~~Function~~
|
||||||
|
- [ ] Autonomize
|
||||||
- [ ] Automatically approve posts that dont get caught by reverse image search (ris)
|
- [ ] Automatically approve posts that dont get caught by reverse image search (ris)
|
||||||
|
|
||||||
### Medium priority:
|
### Medium priority:
|
||||||
|
|||||||
@@ -0,0 +1,19 @@
|
|||||||
|
from io import TextIOWrapper
|
||||||
|
from praw import models
|
||||||
|
import praw
|
||||||
|
import os
|
||||||
|
|
||||||
|
class Bot:
|
||||||
|
password: str = os.environ["ASSISTANT_R_PASS"]
|
||||||
|
secret: str = os.environ["ASSISTANT_R_TOKEN"]
|
||||||
|
|
||||||
|
r: praw.Reddit = praw.Reddit(
|
||||||
|
client_id = "iCSRWS6PMlTLwmylCJRYmA",
|
||||||
|
client_secret = secret,
|
||||||
|
username = "ByteDiceAssistant",
|
||||||
|
password = password,
|
||||||
|
user_agent = "Byte Dice Assistant by u/RandomPersonDotExe aka u/Byte_Dice"
|
||||||
|
)
|
||||||
|
sr: models.Subreddit = r.subreddit("bytedicetesting") #r.subreddit("boykisser")
|
||||||
|
data_f: TextIOWrapper = None
|
||||||
|
data: dict = {}
|
||||||
@@ -0,0 +1,102 @@
|
|||||||
|
import os
|
||||||
|
import json
|
||||||
|
from typing import Final
|
||||||
|
|
||||||
|
import bot as botPy
|
||||||
|
|
||||||
|
|
||||||
|
BK_WEEKLY: Final[str] = "bk_weekly_art_posts"
|
||||||
|
|
||||||
|
|
||||||
|
class PostData:
|
||||||
|
def __init__(
|
||||||
|
self,
|
||||||
|
url: str,
|
||||||
|
title: str,
|
||||||
|
upvotes: int,
|
||||||
|
date_unix: int,
|
||||||
|
media_type: str,
|
||||||
|
media_urls: list[str],
|
||||||
|
nominated_by_human: bool = False,
|
||||||
|
added_by_human: bool = False,
|
||||||
|
added_by_bot: bool = False,
|
||||||
|
approved_by_human: bool = False,
|
||||||
|
approved_by_ris: bool = False
|
||||||
|
):
|
||||||
|
self.url = url
|
||||||
|
self.title = title
|
||||||
|
self.upvotes = upvotes
|
||||||
|
self.date_unix = date_unix
|
||||||
|
self.media_type = media_type
|
||||||
|
self.media_urls = media_urls
|
||||||
|
self.nominated_by_human = nominated_by_human
|
||||||
|
self.added_by_human = added_by_human
|
||||||
|
self.added_by_bot = added_by_bot
|
||||||
|
self.approved_by_human = approved_by_human
|
||||||
|
self.approved_by_ris = approved_by_ris
|
||||||
|
|
||||||
|
def to_json(self):
|
||||||
|
return {
|
||||||
|
"post_data": {
|
||||||
|
"title": self.title,
|
||||||
|
"upvotes": self.upvotes,
|
||||||
|
"date_unix": self.date_unix,
|
||||||
|
"media_type": self.media_type,
|
||||||
|
"media_urls": self.media_urls
|
||||||
|
},
|
||||||
|
"nominated_by_human": self.nominated_by_human,
|
||||||
|
"added": {
|
||||||
|
"by_human": self.added_by_human,
|
||||||
|
"by_bot": self.added_by_bot
|
||||||
|
},
|
||||||
|
"approved": {
|
||||||
|
"by_human": self.approved_by_human,
|
||||||
|
"by_ris": self.approved_by_ris
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
def read_data(bot: botPy.Bot):
|
||||||
|
# Intentionally unreadable >:]
|
||||||
|
data_path = os.path.abspath(os.path.join(os.path.join(os.getcwd(), "data")))
|
||||||
|
|
||||||
|
try:
|
||||||
|
bot.data_f = open(data_path + "\\reddit_data.json", "r+")
|
||||||
|
|
||||||
|
except FileNotFoundError:
|
||||||
|
print("reddit_data.json not found, creating new from preset...")
|
||||||
|
with open(data_path + "\\reddit_data.json", "w") as f:
|
||||||
|
f.write(open(data_path + "\\reddit_data_preset.json", "r").read())
|
||||||
|
|
||||||
|
bot.data_f = open(data_path + "\\reddit_data.json", "r+")
|
||||||
|
|
||||||
|
data_str = bot.data_f.read()
|
||||||
|
bot.data = json.loads(data_str)
|
||||||
|
|
||||||
|
if not bot.data["file_created_correctly"]:
|
||||||
|
raise Exception("reddit_data.json file wasn't created properly. Delete the file and retry.")
|
||||||
|
|
||||||
|
|
||||||
|
def write_data(bot: botPy.Bot):
|
||||||
|
bot.data_f.seek(0)
|
||||||
|
json.dump(bot.data, bot.data_f, indent=2)
|
||||||
|
bot.data_f.truncate()
|
||||||
|
|
||||||
|
|
||||||
|
def update_post_in_data(bot: botPy.Bot, new_data: PostData):
|
||||||
|
if new_data.url not in bot.data:
|
||||||
|
bot.data[BK_WEEKLY][new_data.url] = new_data.to_json()
|
||||||
|
return
|
||||||
|
|
||||||
|
|
||||||
|
def remove_old_posts(bot: botPy.Bot, max_age_unix: int):
|
||||||
|
for post in bot.data[BK_WEEKLY]:
|
||||||
|
if post["date_unix"] > max_age_unix:
|
||||||
|
dict(bot.data[BK_WEEKLY]).pop(post)
|
||||||
|
|
||||||
|
|
||||||
|
def clear_posts_without_media(bot: botPy.Bot):
|
||||||
|
for post in bot.data[BK_WEEKLY]:
|
||||||
|
post_data = post["post_data"]
|
||||||
|
if post_data["media_type"] == None and len(post_data["media_urls"]) == 0:
|
||||||
|
dict(bot.data[BK_WEEKLY]).pop(post)
|
||||||
+10
-157
@@ -1,96 +1,28 @@
|
|||||||
from io import TextIOWrapper
|
|
||||||
from praw import models
|
from praw import models
|
||||||
import praw
|
|
||||||
import os
|
|
||||||
import json
|
|
||||||
import emoji
|
import emoji
|
||||||
import sys
|
import sys
|
||||||
from typing import Final
|
|
||||||
|
|
||||||
|
import bot as botPy
|
||||||
BK_WEEKLY: Final[str] = "bk_weekly_art_posts"
|
import data
|
||||||
|
import reddit
|
||||||
|
|
||||||
class Bot:
|
|
||||||
password: str = os.environ["ASSISTANT_R_PASS"]
|
|
||||||
secret: str = os.environ["ASSISTANT_R_TOKEN"]
|
|
||||||
|
|
||||||
r: praw.Reddit = praw.Reddit(
|
|
||||||
client_id = "iCSRWS6PMlTLwmylCJRYmA",
|
|
||||||
client_secret = secret,
|
|
||||||
username = "ByteDiceAssistant",
|
|
||||||
password = password,
|
|
||||||
user_agent = "Byte Dice Assistant by u/RandomPersonDotExe aka u/Byte_Dice"
|
|
||||||
)
|
|
||||||
sr: models.Subreddit = r.subreddit("bytedicetesting") #r.subreddit("boykisser")
|
|
||||||
data_f: TextIOWrapper = None
|
|
||||||
data: dict = {}
|
|
||||||
|
|
||||||
|
|
||||||
class PostData:
|
|
||||||
def __init__(
|
|
||||||
self,
|
|
||||||
url: str,
|
|
||||||
title: str,
|
|
||||||
upvotes: int,
|
|
||||||
date_unix: int,
|
|
||||||
media_type: str,
|
|
||||||
media_urls: list[str],
|
|
||||||
nominated_by_human: bool = False,
|
|
||||||
added_by_human: bool = False,
|
|
||||||
added_by_bot: bool = False,
|
|
||||||
approved_by_human: bool = False,
|
|
||||||
approved_by_ris: bool = False
|
|
||||||
):
|
|
||||||
self.url = url
|
|
||||||
self.title = title
|
|
||||||
self.upvotes = upvotes
|
|
||||||
self.date_unix = date_unix
|
|
||||||
self.media_type = media_type
|
|
||||||
self.media_urls = media_urls
|
|
||||||
self.nominated_by_human = nominated_by_human
|
|
||||||
self.added_by_human = added_by_human
|
|
||||||
self.added_by_bot = added_by_bot
|
|
||||||
self.approved_by_human = approved_by_human
|
|
||||||
self.approved_by_ris = approved_by_ris
|
|
||||||
|
|
||||||
def to_json(self):
|
|
||||||
return {
|
|
||||||
"post_data": {
|
|
||||||
"title": self.title,
|
|
||||||
"upvotes": self.upvotes,
|
|
||||||
"date_unix": self.date_unix,
|
|
||||||
"media_type": self.media_type,
|
|
||||||
"media_urls": self.media_urls
|
|
||||||
},
|
|
||||||
"nominated_by_human": self.nominated_by_human,
|
|
||||||
"added": {
|
|
||||||
"by_human": self.added_by_human,
|
|
||||||
"by_bot": self.added_by_bot
|
|
||||||
},
|
|
||||||
"approved": {
|
|
||||||
"by_human": self.approved_by_human,
|
|
||||||
"by_ris": self.approved_by_ris
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
|
|
||||||
def main():
|
def main():
|
||||||
sys.stdout.reconfigure(encoding="utf-8")
|
sys.stdout.reconfigure(encoding="utf-8")
|
||||||
|
|
||||||
bot = Bot()
|
bot = botPy.Bot()
|
||||||
print("Reading data...")
|
print("Reading data...")
|
||||||
read_data(bot)
|
data.read_data(bot)
|
||||||
|
|
||||||
check_emoji = emoji.emojize(":check_mark_button:")
|
check_emoji = emoji.emojize(":check_mark_button:")
|
||||||
cross_emoji = emoji.emojize(":cross_mark:")
|
cross_emoji = emoji.emojize(":cross_mark:")
|
||||||
|
|
||||||
print("Fetching posts...")
|
print("Fetching posts...")
|
||||||
posts = fetch_posts_with_flair(bot, "Original Art")
|
posts = reddit.fetch_posts_with_flair(bot, "Original Art")
|
||||||
|
|
||||||
print("Evaluating posts...\n\n")
|
print("Evaluating posts...\n\n")
|
||||||
for post in posts:
|
for post in posts:
|
||||||
media = has_media(post)
|
media = reddit.has_media(post)
|
||||||
media_urls = "\n ".join(media[3])
|
media_urls = "\n ".join(media[3])
|
||||||
|
|
||||||
print(
|
print(
|
||||||
@@ -100,9 +32,9 @@ def main():
|
|||||||
f"\n {media_urls}\n"
|
f"\n {media_urls}\n"
|
||||||
)
|
)
|
||||||
|
|
||||||
update_post_in_data(
|
data.update_post_in_data(
|
||||||
bot,
|
bot,
|
||||||
PostData(
|
data.PostData(
|
||||||
post.shortlink,
|
post.shortlink,
|
||||||
post.title,
|
post.title,
|
||||||
post.score,
|
post.score,
|
||||||
@@ -113,86 +45,7 @@ def main():
|
|||||||
)
|
)
|
||||||
)
|
)
|
||||||
|
|
||||||
write_data(bot)
|
data.write_data(bot)
|
||||||
|
|
||||||
|
|
||||||
def read_data(bot: Bot):
|
|
||||||
# Intentionally unreadable >:]
|
|
||||||
data_path = os.path.abspath(os.path.join(os.path.join(os.getcwd(), "data")))
|
|
||||||
|
|
||||||
try:
|
|
||||||
bot.data_f = open(data_path + "\\reddit_data.json", "r+")
|
|
||||||
|
|
||||||
except FileNotFoundError:
|
|
||||||
print("reddit_data.json not found, creating new from preset...")
|
|
||||||
with open(data_path + "\\reddit_data.json", "w") as f:
|
|
||||||
f.write(open(data_path + "\\reddit_data_preset.json", "r").read())
|
|
||||||
|
|
||||||
bot.data_f = open(data_path + "\\reddit_data.json", "r+")
|
|
||||||
|
|
||||||
data_str = bot.data_f.read()
|
|
||||||
bot.data = json.loads(data_str)
|
|
||||||
|
|
||||||
if not bot.data["file_created_correctly"]:
|
|
||||||
raise Exception("reddit_data.json file wasn't created properly. Delete the file and retry.")
|
|
||||||
|
|
||||||
|
|
||||||
def fetch_posts_with_flair(bot: Bot, flair_name: str) -> list[models.Submission]:
|
|
||||||
posts: list[models.Submission] = []
|
|
||||||
|
|
||||||
# ~36 OG-art posts per week, round limit to 50 or 75
|
|
||||||
for post in bot.sr.search(f"flair:\"{flair_name}\"", sort="new", limit=10):
|
|
||||||
posts.append(post)
|
|
||||||
|
|
||||||
return posts
|
|
||||||
|
|
||||||
|
|
||||||
def has_media(post: models.Submission) -> tuple[bool, str, int, list[str]]:
|
|
||||||
media_type: str = None
|
|
||||||
media_count = 0
|
|
||||||
media_urls: list[str] = []
|
|
||||||
|
|
||||||
if hasattr(post, "post_hint"):
|
|
||||||
media_type = post.post_hint
|
|
||||||
media_count = 1
|
|
||||||
media_urls.append(post.url)
|
|
||||||
|
|
||||||
elif getattr(post, "is_gallery", False):
|
|
||||||
if not post.is_gallery: pass
|
|
||||||
media_type = "multiple"
|
|
||||||
|
|
||||||
gallery_items = getattr(post, "gallery_data", {}).get("items", [])
|
|
||||||
media_metadata = getattr(post, "media_metadata", {})
|
|
||||||
|
|
||||||
media_count = len(gallery_items)
|
|
||||||
|
|
||||||
for item in gallery_items:
|
|
||||||
media_id = item.get("media_id")
|
|
||||||
image_url = media_metadata.get(media_id, {}).get("s", {}).get("u")
|
|
||||||
|
|
||||||
if image_url:
|
|
||||||
media_urls.append(image_url)
|
|
||||||
|
|
||||||
|
|
||||||
return (media_type != None, media_type, media_count, media_urls)
|
|
||||||
|
|
||||||
|
|
||||||
def write_data(bot: Bot):
|
|
||||||
bot.data_f.seek(0)
|
|
||||||
json.dump(bot.data, bot.data_f, indent=2)
|
|
||||||
bot.data_f.truncate()
|
|
||||||
|
|
||||||
|
|
||||||
def update_post_in_data(bot: Bot, new_data: PostData):
|
|
||||||
if new_data.url not in bot.data:
|
|
||||||
bot.data[BK_WEEKLY][new_data.url] = new_data.to_json()
|
|
||||||
return
|
|
||||||
|
|
||||||
|
|
||||||
def remove_old_posts(bot: Bot, max_age_unix: int):
|
|
||||||
for post in bot.data[BK_WEEKLY]:
|
|
||||||
if post["date_unix"] > max_age_unix:
|
|
||||||
dict(bot.data[BK_WEEKLY]).pop(post)
|
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
|
|||||||
@@ -0,0 +1,44 @@
|
|||||||
|
from praw import models
|
||||||
|
import praw
|
||||||
|
|
||||||
|
import bot as botPy
|
||||||
|
|
||||||
|
|
||||||
|
def fetch_posts_with_flair(bot: botPy.Bot, flair_name: str) -> list[models.Submission]:
|
||||||
|
posts: list[models.Submission] = []
|
||||||
|
|
||||||
|
# ~36 OG-art posts per week, round limit to 50 or 75
|
||||||
|
for post in bot.sr.search(f"flair:\"{flair_name}\"", sort="new", limit=10):
|
||||||
|
posts.append(post)
|
||||||
|
|
||||||
|
return posts
|
||||||
|
|
||||||
|
|
||||||
|
def has_media(post: models.Submission) -> tuple[bool, str, int, list[str]]:
|
||||||
|
media_type: str = None
|
||||||
|
media_count = 0
|
||||||
|
media_urls: list[str] = []
|
||||||
|
|
||||||
|
if hasattr(post, "post_hint"):
|
||||||
|
media_type = post.post_hint
|
||||||
|
media_count = 1
|
||||||
|
media_urls.append(post.url)
|
||||||
|
|
||||||
|
elif getattr(post, "is_gallery", False):
|
||||||
|
if not post.is_gallery: pass
|
||||||
|
media_type = "multiple"
|
||||||
|
|
||||||
|
gallery_items = getattr(post, "gallery_data", {}).get("items", [])
|
||||||
|
media_metadata = getattr(post, "media_metadata", {})
|
||||||
|
|
||||||
|
media_count = len(gallery_items)
|
||||||
|
|
||||||
|
for item in gallery_items:
|
||||||
|
media_id = item.get("media_id")
|
||||||
|
image_url = media_metadata.get(media_id, {}).get("s", {}).get("u")
|
||||||
|
|
||||||
|
if image_url:
|
||||||
|
media_urls.append(image_url)
|
||||||
|
|
||||||
|
|
||||||
|
return (media_type != None, media_type, media_count, media_urls)
|
||||||
Reference in New Issue
Block a user