getting it running for the first time
HeyMoon
parent
9b3d96fb7e
commit
e26e7056cd
23
BotModels.py
23
BotModels.py
|
|
@ -139,5 +139,24 @@ class CommentString(Base):
|
|||
'''
|
||||
Mostly for keeping track of how long it's been since the last run, so we can run things without worrying about set number of pages
|
||||
'''
|
||||
class ScriptCalls(Base):
|
||||
pass
|
||||
class ScriptCall(Base):
|
||||
__tablename__ = "script_call"
|
||||
|
||||
id = Column(Integer, primary_key = True)
|
||||
time = Column(DateTime)
|
||||
|
||||
def register(session : Session):
|
||||
sc = ScriptCall(time = datetime.now())
|
||||
session.add(sc)
|
||||
|
||||
def get_time_of_last_run(session) -> datetime:
|
||||
stmt = sqlalchemy.select(ScriptCall).order_by(ScriptCall.time.desc()).limit(1)
|
||||
sc = session.execute(stmt).scalar_one_or_none()
|
||||
|
||||
if (sc == None):
|
||||
return datetime.now()
|
||||
else:
|
||||
return sc.time
|
||||
|
||||
|
||||
|
||||
121
automeme.py
121
automeme.py
|
|
@ -1,16 +1,15 @@
|
|||
import base64
|
||||
import io
|
||||
import json
|
||||
from random import random
|
||||
from random import choice, choices, random
|
||||
import re
|
||||
import traceback
|
||||
from typing import Callable, TypeVar
|
||||
import meme_generator
|
||||
from meme_generator import WebcomicPanel, OneCharacterWebcomicPanel, TwoCharacterWebcomicPanel, TitleCardWebcomicPanel, add_watermark, create_webcomic
|
||||
from RDramaAPIInterface import RDramaAPIInterface
|
||||
from datetime import datetime
|
||||
from datetime import datetime, timedelta
|
||||
from os.path import exists, join, realpath, split
|
||||
from threading import Timer
|
||||
from BotModels import Base, Comment, CommentString, Post, User
|
||||
from BotModels import Base, Comment, CommentString, Post, ScriptCall, User
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy.orm import Session
|
||||
import os
|
||||
|
|
@ -18,25 +17,50 @@ from markdown import markdown
|
|||
from bs4 import BeautifulSoup
|
||||
from utils import get_real_filename
|
||||
|
||||
TEST_MODE = True
|
||||
ANTISPAM_MESSAGES = [
|
||||
'i have a meme for u :marseyshy:',
|
||||
'anon -- can you suck my penis please?',
|
||||
'uwu i love memes',
|
||||
'~ oh... dramatardchan -- i wish ud stop talking about trains',
|
||||
'hey dramatardchan, its ok if u have autism',
|
||||
'its okay silly ~ i know u are retarded',
|
||||
'on rdrama someone told me to sneed, what does that mean?',
|
||||
'what if no-one likes my shitposts :marseycry:',
|
||||
'a ha ha ha, that\'s very funny dramatardchan',
|
||||
'im going to a party ~ no you dont have to come ~ love u',
|
||||
'oh dramatardchan, your such a special boy',
|
||||
'omg hes talking about "trannys" again, im so wet down there',
|
||||
'idk what that means but i still love u',
|
||||
'i love u but pls dont send me penis pictures, its kinda gross >.<',
|
||||
'aha thats very funny, thank u',
|
||||
'hey dramatardchan... i was thinking... maybe u could take a shower... if u want',
|
||||
'i love u becuz ur so sweet, even if you are kind of a pushover, dramatardchan',
|
||||
'maybe u should delete ur account - not trying to be mean tho',
|
||||
'look at this cute penis picture i found on twitter... no, its ok, u dont have to show me urs',
|
||||
'FUCK YOU. I HAVE HAD IT UP TO HERE WITH YOUR FUCKING DRAMA BULLSHIT. FUCK. OFF.',
|
||||
'idk what u mean by mommy milkers 😭 im not ur mommy pls stop saying that',
|
||||
'goldstein sama is so cute ~ what a cute nose... urs is fine to tho'
|
||||
]
|
||||
|
||||
TEST_MODE = False
|
||||
DRY_MODE = False
|
||||
TEST_AUTH_TOKEN = "ED3eURMKP9FKBFbi-JUxo8MPGWkEihuyIlAUGtVL7xwx0NEy4Nf6J_mxWYTPgAQx1iy1X91hx7PPHyEBS79hvKVIy5DMEzOyAe9PAc5pmqSJlLGq_-ROewMwFzGrqer4"
|
||||
BETA_MODE = False
|
||||
TEST_AUTH_TOKEN = "lawoSNzuNeBRJBld0boApOceCNSEqBhiRu0aoUWh9kTK7AV37NECZoAK-mEUJ1PM1SsTfTY4f3t_LjooMq4QgPLqPC-F7LNGHQ6_0RFacZmIvC2ixOHPKM821RroJexn"
|
||||
MINUTES_BEFORE_FORCED_SHUTDOWN = 10
|
||||
DB_FILENAME = "automeme_database.db"
|
||||
PAGES_TO_SCAN = 5
|
||||
AUTOMEME_ID = 13427
|
||||
ALLOWED_COMMENTS_PER_POST = 20
|
||||
ALLOWED_COMMENTS_PER_USER_PER_DAY = 20
|
||||
MESSAGE = "i have a meme for u :marseyshy: \nPROTIP: I reply randomly. Put !!meme in your message to make sure I reply"
|
||||
|
||||
SOY_VS_CHAD_TRIGGER_CHANGE = 1.0
|
||||
MODERN_MEME_WITH_MARSEY_TRIGGER_CHANGE = 1.0 #0.1
|
||||
MODERN_MEME_WITH_IMAGE_TRIGGER_CHANGE = 1.0
|
||||
CLASSIC_MEME_WITH_MARSEY_TRIGGER_CHANGE = 1.0 #0.5
|
||||
CLASSIC_MEME_WITH_IMAGE_TRIGGER_CHANGE = 1.0
|
||||
MODERN_MEME_WITH_MARSEY_TRIGGER_CHANGE = 0.01
|
||||
MODERN_MEME_WITH_IMAGE_TRIGGER_CHANGE = 0.05
|
||||
CLASSIC_MEME_WITH_MARSEY_TRIGGER_CHANGE = 0.02
|
||||
CLASSIC_MEME_WITH_IMAGE_TRIGGER_CHANGE = 0.1
|
||||
WEBCOMIC_TRIGGER_CHANCE = 1.0
|
||||
|
||||
FREE_POSTS = [6]
|
||||
FREE_POSTS = [6, 97416, 98286]
|
||||
RESTRICTED_POSTS = [5, 16583, 75878, 35835]
|
||||
|
||||
EMOJI_REGEX = r":[#!a-zA-Z0-9]*:"
|
||||
|
|
@ -133,7 +157,7 @@ class Emoji(TextElement):
|
|||
def __repr__(self) -> str:
|
||||
return f"Emoji({self.emoji}, big={self.big})"
|
||||
|
||||
def get_text_only(text_elements : list[TextElement]) -> str:
|
||||
def get_text_only(text_elements : 'list[TextElement]') -> str:
|
||||
text = [i.text for i in text_elements if isinstance(i, Text)]
|
||||
return " ".join(text)
|
||||
|
||||
|
|
@ -195,15 +219,16 @@ def remove_duplicates(list):
|
|||
return [json.loads(j) for j in set([json.dumps(i) for i in list])]
|
||||
|
||||
def get_eligible_comments(rdrama : RDramaAPIInterface, session : Session):
|
||||
comments = rdrama.get_comments(number_of_pages=PAGES_TO_SCAN)['data']
|
||||
comments = [comment for comment in comments if not comment['is_bot']] #No bots
|
||||
comments = [comment for comment in comments if comment['author'] == '👻' or not comment['author']['id'] == AUTOMEME_ID] #Don't reply to self
|
||||
comments = [comment for comment in comments if Post.get_number_of_replies(comment['post_id'], session) < ALLOWED_COMMENTS_PER_POST] #Don't spam posts
|
||||
comments = [comment for comment in comments if User.get_number_of_comments(0 if comment['author'] == '👻' else comment['author']['id'], session) < ALLOWED_COMMENTS_PER_USER_PER_DAY] #Don't spam users
|
||||
comments = [comment for comment in comments if Comment.get_comment(comment['id'], session) is None] #Double check that we haven't replied to the comment
|
||||
comments = remove_duplicates(comments) #Remove the duplicates
|
||||
last_time = ScriptCall.get_time_of_last_run(session)
|
||||
current_time = int(datetime.now().timestamp())
|
||||
|
||||
begin = datetime.now()
|
||||
comments = comment_chunk(last_time, rdrama) #rdrama.get_comments(number_of_pages=3, upper_bound=current_time, lower_bound=last_time, sort="old")['data']
|
||||
end = datetime.now()
|
||||
print(end-begin)
|
||||
#comments = [comment for comment in comments if comment['created_utc'] > last_time]
|
||||
for comment in comments:
|
||||
|
||||
user_id = 0 if comment['author'] == '👻' else comment['author']['id']
|
||||
comment_id = comment['id']
|
||||
text = comment['body']
|
||||
|
|
@ -211,10 +236,29 @@ def get_eligible_comments(rdrama : RDramaAPIInterface, session : Session):
|
|||
|
||||
CommentString.add_comment(comment_id, user_id, time, text, session)
|
||||
|
||||
print([comment['id'] for comment in comments])
|
||||
comments = [comment for comment in comments if not comment['is_bot']] #No bots
|
||||
print([comment['id'] for comment in comments])
|
||||
comments = [comment for comment in comments if comment['author'] == '👻' or not comment['author']['id'] == AUTOMEME_ID] #Don't reply to self
|
||||
print([comment['id'] for comment in comments])
|
||||
comments = [comment for comment in comments if Post.get_number_of_replies(comment['post_id'], session) < ALLOWED_COMMENTS_PER_POST] #Don't spam posts
|
||||
print([comment['id'] for comment in comments])
|
||||
comments = [comment for comment in comments if User.get_number_of_comments(0 if comment['author'] == '👻' else comment['author']['id'], session) < ALLOWED_COMMENTS_PER_USER_PER_DAY] #Don't spam users
|
||||
print([comment['id'] for comment in comments])
|
||||
comments = [comment for comment in comments if Comment.get_comment(comment['id'], session) is None] #Double check that we haven't replied to the comment
|
||||
|
||||
if BETA_MODE:
|
||||
comments = [comment for comment in comments if comment['post_id'] in FREE_POSTS]
|
||||
print([comment['id'] for comment in comments])
|
||||
|
||||
comments = remove_duplicates(comments) #Remove the duplicates
|
||||
|
||||
print([comment['id'] for comment in comments])
|
||||
|
||||
return comments
|
||||
|
||||
T = TypeVar('T')
|
||||
def lambda_count(list : list[T], predicate : 'Callable[[T], bool]' ):
|
||||
def lambda_count(list : 'list[T]', predicate : 'Callable[[T], bool]' ):
|
||||
return sum(1 for i in list if predicate(i))
|
||||
|
||||
def get_full_rdrama_image_url(partial_url) -> str:
|
||||
|
|
@ -232,7 +276,11 @@ def create_comment_message(chud: bool, pizza: bool, bird : bool, marsey : bool):
|
|||
if (marsey):
|
||||
return ":marseyshy:"
|
||||
|
||||
message = MESSAGE
|
||||
|
||||
|
||||
message = choice(ANTISPAM_MESSAGES) +"\n"
|
||||
|
||||
message += f"<source src=\"{''.join(choices(['a','b','c','d','f','g','h','j','k','l','m','n','p','q','r','s','t','v','w','x','z'], k=500))}.mp3\">"
|
||||
|
||||
if (chud):
|
||||
message += "\ntrans lives matter"
|
||||
|
|
@ -270,9 +318,13 @@ def main_processing_task(rdrama : RDramaAPIInterface, session : Session):
|
|||
print("We are banned. STOP.")
|
||||
can_communicate = False
|
||||
|
||||
if not can_communicate:
|
||||
print("I can't communicate. Why????")
|
||||
|
||||
if can_communicate:
|
||||
eligible_comments = get_eligible_comments(rdrama, session)
|
||||
for eligible_comment in eligible_comments:
|
||||
try:
|
||||
begin = datetime.now()
|
||||
under_post_limit = Post.get_number_of_replies(eligible_comment['post_id'], session) < ALLOWED_COMMENTS_PER_POST
|
||||
under_user_limit = User.get_number_of_comments(0 if eligible_comment['author'] == '👻' else eligible_comment['author']['id'], session) < ALLOWED_COMMENTS_PER_USER_PER_DAY
|
||||
|
|
@ -286,7 +338,7 @@ def main_processing_task(rdrama : RDramaAPIInterface, session : Session):
|
|||
cleaned_comment_text = strip_markdown(comment_text)
|
||||
|
||||
|
||||
if ("meme" in directives):
|
||||
if ("meme" in directives or eligible_comment['post_id'] in FREE_POSTS):
|
||||
random_float = 0.0
|
||||
else:
|
||||
random_float = random()
|
||||
|
|
@ -414,7 +466,7 @@ def main_processing_task(rdrama : RDramaAPIInterface, session : Session):
|
|||
automeme_comment_id = comment_with_image(message, image, eligible_comment['id'], eligible_comment['post_id'])
|
||||
else:
|
||||
automeme_comment_id = None
|
||||
image.save(f"dry/{eligible_comment['id']}.webp")
|
||||
image.save(get_real_filename(f"dry/{eligible_comment['id']}.webp"))
|
||||
|
||||
Comment.create_new_comment(parent_comment_id, automeme_comment_id, session)
|
||||
if post_id not in FREE_POSTS:
|
||||
|
|
@ -426,16 +478,31 @@ def main_processing_task(rdrama : RDramaAPIInterface, session : Session):
|
|||
print(end-begin)
|
||||
else:
|
||||
Comment.create_new_comment(eligible_comment['id'], None, session)
|
||||
except BaseException as e:
|
||||
print(f"YIKERINOS! GOT AN EXCEPTION: {e}")
|
||||
traceback.print_exc()
|
||||
|
||||
ScriptCall.register(session)
|
||||
|
||||
def comment_chunk(time : datetime, api: RDramaAPIInterface):
|
||||
current_time_cur = int(time.timestamp()) # int(time.time() - 60*60*8)
|
||||
comments = []
|
||||
while True:
|
||||
res = api.get_comments(sort="old", lower_bound=current_time_cur)
|
||||
if len(res['data']) == 0:
|
||||
break
|
||||
comments = comments + res['data']
|
||||
current_time_cur = res['data'][-1]['created_utc']
|
||||
|
||||
return comments
|
||||
if __name__ == "__main__":
|
||||
TEST_AUTH_TOKEN = "ED3eURMKP9FKBFbi-JUxo8MPGWkEihuyIlAUGtVL7xwx0NEy4Nf6J_mxWYTPgAQx1iy1X91hx7PPHyEBS79hvKVIy5DMEzOyAe9PAc5pmqSJlLGq_-ROewMwFzGrqer4"
|
||||
print(f"======= RUNNING AT {datetime.now().hour}:{datetime.now().minute} ======= ")
|
||||
if TEST_MODE:
|
||||
website = "localhost"
|
||||
auth = TEST_AUTH_TOKEN
|
||||
https = False
|
||||
timeout = 1
|
||||
AUTOMEME_ID = 7
|
||||
AUTOMEME_ID = 6
|
||||
OPERATOR_ID = 9
|
||||
ACTUALLY_CALL_OPEN_AI = False
|
||||
else:
|
||||
|
|
|
|||
|
|
@ -1,5 +1,4 @@
|
|||
import math
|
||||
from tkinter import UNITS
|
||||
from typing import Tuple, Union
|
||||
from PIL import Image
|
||||
from PIL import ImageDraw, ImageSequence
|
||||
|
|
@ -118,7 +117,7 @@ def create_modern_meme_from_emoji(emoji: str, caption: str):
|
|||
|
||||
class WebcomicPanel():
|
||||
PANEL_SIZE = 400
|
||||
FONT = "Impact.ttf"
|
||||
FONT = "impact.ttf"
|
||||
COLOR = ColorScheme.WHITE_WITH_BLACK_BORDER
|
||||
|
||||
def __init__(self):
|
||||
|
|
@ -127,7 +126,7 @@ class WebcomicPanel():
|
|||
def create_image(self) -> Image:
|
||||
return AnimatedImage.new((self.PANEL_SIZE,self.PANEL_SIZE))
|
||||
|
||||
def add_text_box(self, base : Image, caption : str, region_size : tuple[int, int], coordinates : tuple[int, int], align="") -> 'AnimatedImage':
|
||||
def add_text_box(self, base : Image, caption : str, region_size : 'tuple[int, int]', coordinates : 'tuple[int, int]', align="") -> 'AnimatedImage':
|
||||
return add_text_box(base, caption, region_size, coordinates, align=align, font=self.FONT, color=self.COLOR, init_font_size=int(self.PANEL_SIZE/10))
|
||||
|
||||
class OneCharacterWebcomicPanel(WebcomicPanel):
|
||||
|
|
@ -228,7 +227,9 @@ def create_webcomic(layout : 'list[WebcomicPanel]'):
|
|||
image = image.paste(panel.create_image(), (x*assumed_panel_x_size, y*assumed_panel_y_size))
|
||||
return image
|
||||
|
||||
def add_text_box(base : Image, caption : str, region_size : tuple[int, int], coordinates : tuple[int, int], font : str= "arial.ttf", init_font_size = 45, align :str = "", color = ColorScheme.BLACK):
|
||||
def add_text_box(base : Image, caption : str, region_size : 'tuple[int, int]', coordinates : 'tuple[int, int]', font : str= "arial.ttf", init_font_size = 45, align :str = "", color = ColorScheme.BLACK):
|
||||
font = get_real_filename(font)
|
||||
|
||||
if caption == "":
|
||||
return base
|
||||
if init_font_size == 0:
|
||||
|
|
@ -244,6 +245,7 @@ def add_text_box(base : Image, caption : str, region_size : tuple[int, int], coo
|
|||
fill_color = (255,255,255)
|
||||
stroke = (0,0,0)
|
||||
stroke_size = 3
|
||||
caption = caption.upper()
|
||||
line_image = ImageText((region_x_size+stroke_size, region_y_size))
|
||||
place = "left"
|
||||
if "ch" in align:
|
||||
|
|
@ -285,7 +287,9 @@ def add_text_box(base : Image, caption : str, region_size : tuple[int, int], coo
|
|||
else:
|
||||
return add_text_box(base, caption, region_size, coordinates, font=font, init_font_size=init_font_size-1, align=align, color=color)
|
||||
|
||||
def add_text(base : Image, caption : str, region_size : tuple[int, int], coordinates : tuple[int, int], font : str= "arial.ttf"):
|
||||
def add_text(base : Image, caption : str, region_size : 'tuple[int, int]', coordinates : 'tuple[int, int]', font : str= "arial.ttf"):
|
||||
font = get_real_filename(font)
|
||||
print(font)
|
||||
if caption == "":
|
||||
return
|
||||
region_x_size, region_y_size = region_size
|
||||
|
|
@ -312,7 +316,7 @@ def add_watermark(image : Image, name_of_other_creator):
|
|||
|
||||
return base
|
||||
|
||||
def center_and_paste(base : Image, to_paste : Image, coordinates: tuple[int, int], box_size : tuple[int, int]) -> 'AnimatedImage':
|
||||
def center_and_paste(base : Image, to_paste : Image, coordinates: 'tuple[int, int]', box_size : 'tuple[int, int]') -> 'AnimatedImage':
|
||||
to_paste = to_paste.fit(box_size)
|
||||
|
||||
image_size_x, image_size_y = to_paste.size
|
||||
|
|
@ -346,7 +350,7 @@ def get_emoji_from_rdrama(emoji_name) -> 'AnimatedImage':
|
|||
should_flip = True
|
||||
|
||||
if (exists(get_real_filename(f"emoji_cache/{cleaned_emoji_name}.webp"))):
|
||||
image = AnimatedImage.from_image(Image.open(f"emoji_cache/{cleaned_emoji_name}.webp"))
|
||||
image = AnimatedImage.from_image(Image.open(get_real_filename(f"emoji_cache/{cleaned_emoji_name}.webp")))
|
||||
|
||||
else:
|
||||
image = get_image_file_from_url(f"https://www.rdrama.net/e/{cleaned_emoji_name}.webp")
|
||||
|
|
@ -375,11 +379,14 @@ def parse_caption_file(filename):
|
|||
watermark_captions = parse_caption_file(get_real_filename(CAPTION_FILENAME))
|
||||
|
||||
class AnimatedImage():
|
||||
def __init__(self, frames : list[Image.Image]):
|
||||
def __init__(self, frames : 'list[Image.Image]'):
|
||||
if len(frames) > 100:
|
||||
self.frames = frames[0:100]
|
||||
else:
|
||||
self.frames = frames
|
||||
|
||||
@property
|
||||
def size(self) -> Tuple[int, int]:
|
||||
def size(self) -> 'Tuple[int, int]':
|
||||
return self.frames[0].size
|
||||
|
||||
@property
|
||||
|
|
@ -431,11 +438,11 @@ class AnimatedImage():
|
|||
new_frames.append(new_frame)
|
||||
return AnimatedImage(new_frames)
|
||||
|
||||
def new(size : Tuple[int, int], color=(255,255,255)) -> 'AnimatedImage':
|
||||
def new(size : 'Tuple[int, int]', color=(255,255,255)) -> 'AnimatedImage':
|
||||
base = Image.new(mode="RGB", size=size, color=color)
|
||||
return AnimatedImage.from_image(base)
|
||||
|
||||
def fit(self, region : Tuple[int, int]) -> 'AnimatedImage':
|
||||
def fit(self, region : 'Tuple[int, int]') -> 'AnimatedImage':
|
||||
new_frames = []
|
||||
for frame in self.frames:
|
||||
new_frames.append(ImageOps.contain(frame, region))
|
||||
|
|
|
|||
Loading…
Reference in New Issue