diff --git a/config.py.example b/config.py.example index 9606728..89f9a78 100644 --- a/config.py.example +++ b/config.py.example @@ -2,6 +2,9 @@ admins_bot = ('drq@mastodon.ml',) # Адреса админов бота, кот # Example: ('admin_user', 'another_admin_user2@example.example') or ('admin_user',) bot_acct = 'fmn' # Ник бота на инстансе instance = 'pleroma.dark-alexandr.net' # Инстанс, где будет запущен бот +# Ссылка на live stream FMN (API) +peertube_stream_url = 'https://xxivproduction.video/api/v1/videos/1FZeVVVzWBFShaxQVkYiXd' + # Лимиты limit_movies_per_user = 2 # Ограничение количества фильмов на одного пользователя diff --git a/fmn_bot.py b/fmn_bot.py index 8d5f739..c09cd32 100644 --- a/fmn_bot.py +++ b/fmn_bot.py @@ -1,20 +1,20 @@ -from src import listener_context, listener_mention, imdb_datasets_worker +from src import listener_context, listener_mention from config import logger_default_level from loguru import logger import time import sys - def main(): logger.remove() logger.add(sink=sys.stderr, level=logger_default_level) logger.add(sink='suggestions.log', level='INFO', filter='src.listener_context') - listener_mention.run_scan_notif() # Слушаем упоминания в фоне + listener_mention.run_scan_notif() # Слушаем упоминания в фоне time.sleep(1) - listener_context.scan_context_thread() # Слушаем тред на новые предложения фильмов + listener_context.scan_context_thread() # Слушаем тред на новые предложения фильмов + if __name__ == '__main__': main() diff --git a/src/listener_mention.py b/src/listener_mention.py index db38831..fa82be4 100644 --- a/src/listener_mention.py +++ b/src/listener_mention.py @@ -1,7 +1,7 @@ from src.fedi_api import get_notifications, mark_as_read_notification, post_status, upload_attachment from src.fmn_states_db import write_states, states_stor from src.sheduler import check_stop_thread_scan -from config import admins_bot, limit_movies_per_user, limit_all_movies_poll, hour_poll_posting, fmn_next_watching_hour +from config import admins_bot, limit_movies_per_user, limit_all_movies_poll, hour_poll_posting, fmn_next_watching_hour, peertube_stream_url import threading import time @@ -15,7 +15,7 @@ from loguru import logger @logger.catch def get_peertube_stream_name(): try: - return requests.get("https://xxivproduction.video/api/v1/videos/1FZeVVVzWBFShaxQVkYiXd").json()['name'] + return requests.get(peertube_stream_url).json()['name'] except Exception as E: return f"[не удалось получить название {E}]" diff --git a/src/sheduler.py b/src/sheduler.py index 49acb8a..4edb7b2 100644 --- a/src/sheduler.py +++ b/src/sheduler.py @@ -8,8 +8,16 @@ def check_stop_thread_scan(suggested_date): suggested_year = suggested_date.year tuesdays = [] for i in calendar.Calendar().itermonthdays4(suggested_year, suggested_month): - print(i) if i[3] == 1 and datetime.datetime(year=i[0], month=i[1], day=i[2]) > suggested_date: tuesdays.append(datetime.datetime( year=i[0], month=i[1], day=i[2], hour=hour_poll_posting)) + if tuesdays == []: + shift_for_next_week = suggested_date + datetime.timedelta(days=1) + suggested_month = shift_for_next_week.month + suggested_year = shift_for_next_week.year + for i in calendar.Calendar().itermonthdays4(suggested_year, suggested_month): + if i[3] == 1 and datetime.datetime(year=i[0], month=i[1], day=i[2]) > suggested_date: + tuesdays.append(datetime.datetime( + year=i[0], month=i[1], day=i[2], hour=hour_poll_posting)) + return tuesdays[0] # near tuesday diff --git a/test.py b/test.py new file mode 100644 index 0000000..17b55d7 --- /dev/null +++ b/test.py @@ -0,0 +1,30 @@ +import unittest +import datetime +import time_machine +from src import sheduler + + +class TestDate(unittest.TestCase): + def test_date_tue(self): + """ + test near tue + тестирует корректность расчета даты на след. вторник + условия: расчет начат от воскресенья или понедельника (случайно) + """ + import calendar + import random + for month in range(1, 13): + sundays = [i for i in calendar.Calendar().itermonthdays4(year=2024, month=month) if i[3] == 6] + for sunday in sundays: + fake_date = datetime.datetime(year=sunday[0], month=sunday[1], day=sunday[2], hour=random.randint(21, 23)) + datetime.timedelta(days=random.randint(0, 1)) + with time_machine.travel(fake_date): + try: + result = sheduler.check_stop_thread_scan(fake_date) + print("current fake date: " + fake_date.strftime("%Y.%m.%d %H:%M") + f" near tue: {result}") + except Exception as E: + print("fail fake date: " + fake_date.strftime("%Y.%m.%d %H:%M") + f", err: {E}") + + + +if __name__ == '__main__': + unittest.main()