229 lines
7.2 KiB
Plaintext
229 lines
7.2 KiB
Plaintext
# mypy: disable-error-code=attr-defined
|
|
import os
|
|
import sys
|
|
import praw # type: ignore
|
|
import time
|
|
import utils
|
|
import prawcore # type: ignore
|
|
import traceback
|
|
import datetime as dt
|
|
from pathlib import Path
|
|
from logger import Logger
|
|
from jsonwrapper import AutoSaveDict
|
|
from typing import (
|
|
Optional,
|
|
Callable,
|
|
Tuple,
|
|
List,
|
|
Set,
|
|
Any,
|
|
)
|
|
from bot import (
|
|
Datatype,
|
|
Posts,
|
|
Row,
|
|
)
|
|
|
|
|
|
def config_app(path: Path) -> AutoSaveDict:
|
|
config = {
|
|
'client_id': '',
|
|
'client_secret': '',
|
|
'user_agent': '',
|
|
'username': '',
|
|
'password': '',
|
|
'sub_name': '',
|
|
'max_days': '',
|
|
'max_posts': '',
|
|
}
|
|
|
|
configuration: List[List[str]] = []
|
|
|
|
if not os.path.exists(path):
|
|
for key, _ in config.items():
|
|
config_name = ' '.join(key.split('_')).title()
|
|
user_inp = input(f"{config_name}: ")
|
|
configuration.append([key, user_inp])
|
|
|
|
for config_name, value in configuration:
|
|
config[config_name] = value
|
|
|
|
config_handler = AutoSaveDict(
|
|
path,
|
|
**config
|
|
)
|
|
return config_handler
|
|
|
|
|
|
config_file = Path(utils.BASE_DIR, 'config.json')
|
|
handler = config_app(config_file)
|
|
handler.init()
|
|
posts = Posts('post', utils.BASE_DIR)
|
|
logger = Logger(1)
|
|
untracked_flairs = (utils.Flair.SOLVED, utils.Flair.ABANDONED)
|
|
posts.init()
|
|
reddit = praw.Reddit(
|
|
client_id=handler['client_id'],
|
|
client_secret=handler['client_secret'],
|
|
user_agent=handler['user_agent'],
|
|
username=handler['username'],
|
|
password=handler['password'],
|
|
)
|
|
|
|
|
|
def remove_method(submission: praw.reddit.Submission) -> Optional[str]:
|
|
removed = submission.removed_by_category
|
|
if removed is not None:
|
|
# if removed in ('author', 'moderator'):
|
|
# method = 'Removed by moderator'
|
|
if removed in ('author',):
|
|
method = 'Deleted by OP'
|
|
elif removed in ('moderator',):
|
|
method = 'Removed by mod'
|
|
elif removed in ('deleted',):
|
|
method = 'Deleted by user'
|
|
else:
|
|
method = 'Uknown deletion method'
|
|
return method
|
|
|
|
return None
|
|
|
|
|
|
def send_modmail(reddit: praw.Reddit, subreddit: str, subject: str, msg: str) -> None:
|
|
print("Sending modmail...")
|
|
reddit.subreddit(subreddit).message(subject, msg)
|
|
print(msg)
|
|
|
|
|
|
def notify_if_error(func: Callable[..., int]) -> Callable[..., int]:
|
|
def wrapper(*args: Any, **kwargs: Any) -> int:
|
|
try:
|
|
return func(*args, **kwargs)
|
|
except KeyboardInterrupt:
|
|
logger.debug("\nProgram interrupted by user")
|
|
return 0
|
|
except:
|
|
author = 'https://www.reddit.com/user/kaerfkeerg'
|
|
full_error = traceback.format_exc()
|
|
bot_name = utils.BOT_NAME
|
|
msg = f"Error with '{bot_name}':\n\n{full_error}\n\nPlease report to author ({author})"
|
|
send_modmail(
|
|
reddit,
|
|
handler['sub_name'],
|
|
f'An error has occured with {utils.BOT_NAME} msg',
|
|
msg
|
|
)
|
|
return 1
|
|
return wrapper
|
|
|
|
|
|
def should_be_tracked(
|
|
flair: utils.Flair,
|
|
untracked_flairs: Tuple[utils.Flair, ...]) -> bool:
|
|
return flair not in untracked_flairs
|
|
|
|
|
|
def user_is_deleted(submission: praw.reddit.Submission) -> bool:
|
|
return submission.author is None
|
|
|
|
|
|
def check_submission(submission: praw.reddit.Submission, saved_submission_ids: Set[Row]) -> None:
|
|
if not user_is_deleted(submission) and submission.id not in saved_submission_ids:
|
|
flair = utils.get_flair(submission.link_flair_text)
|
|
method = remove_method(submission)
|
|
if should_be_tracked(flair, untracked_flairs):
|
|
if method is None and submission.author is not None:
|
|
original_post = Row(
|
|
username=submission.author.name,
|
|
title=submission.title,
|
|
text=submission.selftext,
|
|
post_id=submission.id,
|
|
deletion_method=Datatype.NULL,
|
|
post_last_edit=Datatype.NULL,
|
|
record_created=str(dt.datetime.now()),
|
|
record_edited=str(dt.datetime.now()),
|
|
)
|
|
posts.save(original_post)
|
|
|
|
|
|
@notify_if_error
|
|
def main() -> int:
|
|
posts_to_delete: Set[Row] = set()
|
|
ignore_methods = ['Removed by mod',]
|
|
|
|
if utils.parse_cmd_line_args(sys.argv, logger, config_file, posts):
|
|
return 0
|
|
|
|
saved_submission_ids = {row.post_id for row in posts.fetch_all()}
|
|
max_posts = handler['max_posts']
|
|
limit = int(max_posts) if max_posts else None
|
|
sub_name = handler['sub_name']
|
|
|
|
for submission in reddit.subreddit(sub_name).new(limit=limit):
|
|
try:
|
|
check_submission(submission, saved_submission_ids)
|
|
except prawcore.exceptions.TooManyRequests:
|
|
time.sleep(60)
|
|
check_submission(submission, saved_submission_ids)
|
|
|
|
for stored_post in posts.fetch_all():
|
|
try:
|
|
submission = reddit.submission(id=stored_post.post_id)
|
|
max_days = int(handler['max_days'])
|
|
created = utils.string_to_dt(stored_post.record_created).date()
|
|
flair = utils.get_flair(submission.link_flair_text)
|
|
|
|
if utils.submission_is_older(created, max_days) or flair in untracked_flairs:
|
|
posts_to_delete.add(stored_post)
|
|
continue
|
|
|
|
submission = reddit.submission(id=stored_post.post_id)
|
|
method = remove_method(submission)
|
|
if user_is_deleted(submission):
|
|
if method not in ignore_methods:
|
|
send_modmail(
|
|
reddit,
|
|
handler['sub_name'],
|
|
"User's account has been deleted",
|
|
utils.modmail_removal_notification(stored_post, 'Account has been deleted')
|
|
)
|
|
posts_to_delete.add(stored_post)
|
|
|
|
elif method is not None and not stored_post.deletion_method:
|
|
if method not in ignore_methods:
|
|
stored_post.deletion_method = method
|
|
stored_post.record_edited = str(dt.datetime.now())
|
|
posts.edit(stored_post)
|
|
msg = utils.modmail_removal_notification(stored_post, method)
|
|
send_modmail(
|
|
reddit,
|
|
handler['sub_name'],
|
|
'A post has been deleted',
|
|
msg
|
|
)
|
|
posts_to_delete.add(stored_post)
|
|
time.sleep(utils.MSG_AWAIT_THRESHOLD)
|
|
|
|
if submission.selftext != stored_post.text\
|
|
or submission.selftext != stored_post.post_last_edit\
|
|
and not stored_post.deletion_method:
|
|
stored_post.post_last_edit = submission.selftext
|
|
stored_post.record_edited = str(dt.datetime.now())
|
|
posts.edit(stored_post)
|
|
except prawcore.exceptions.TooManyRequests:
|
|
time.sleep(60)
|
|
|
|
for row in posts_to_delete:
|
|
posts.delete(post_id=row.post_id)
|
|
|
|
posts_to_delete.clear()
|
|
logger.info("Program finished successfully")
|
|
logger.info(f"Total posts deleted: {len(posts_to_delete)}")
|
|
return 0
|
|
|
|
|
|
if __name__ == '__main__':
|
|
sys.exit(
|
|
main()
|
|
)
|