Compare commits

...

6 commits

5 changed files with 48 additions and 7 deletions

View file

@ -2,6 +2,9 @@ admins_bot = ('drq@mastodon.ml',) # Адреса админов бота, кот
# Example: ('admin_user', 'another_admin_user2@example.example') or ('admin_user',) # Example: ('admin_user', 'another_admin_user2@example.example') or ('admin_user',)
bot_acct = 'fmn' # Ник бота на инстансе bot_acct = 'fmn' # Ник бота на инстансе
instance = 'pleroma.dark-alexandr.net' # Инстанс, где будет запущен бот instance = 'pleroma.dark-alexandr.net' # Инстанс, где будет запущен бот
# Ссылка на live stream FMN (API)
peertube_stream_url = 'https://xxivproduction.video/api/v1/videos/1FZeVVVzWBFShaxQVkYiXd'
# Лимиты # Лимиты
limit_movies_per_user = 2 # Ограничение количества фильмов на одного пользователя limit_movies_per_user = 2 # Ограничение количества фильмов на одного пользователя

View file

@ -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 config import logger_default_level
from loguru import logger from loguru import logger
import time import time
import sys import sys
def main(): def main():
logger.remove() logger.remove()
logger.add(sink=sys.stderr, level=logger_default_level) logger.add(sink=sys.stderr, level=logger_default_level)
logger.add(sink='suggestions.log', level='INFO', filter='src.listener_context') logger.add(sink='suggestions.log', level='INFO', filter='src.listener_context')
listener_mention.run_scan_notif() # Слушаем упоминания в фоне listener_mention.run_scan_notif() # Слушаем упоминания в фоне
time.sleep(1) time.sleep(1)
listener_context.scan_context_thread() # Слушаем тред на новые предложения фильмов listener_context.scan_context_thread() # Слушаем тред на новые предложения фильмов
if __name__ == '__main__': if __name__ == '__main__':
main() main()

View file

@ -1,7 +1,7 @@
from src.fedi_api import get_notifications, mark_as_read_notification, post_status, upload_attachment 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.fmn_states_db import write_states, states_stor
from src.sheduler import check_stop_thread_scan 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 threading
import time import time
@ -15,7 +15,7 @@ from loguru import logger
@logger.catch @logger.catch
def get_peertube_stream_name(): def get_peertube_stream_name():
try: 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: except Exception as E:
return f"[не удалось получить название {E}]" return f"[не удалось получить название {E}]"

View file

@ -8,8 +8,16 @@ def check_stop_thread_scan(suggested_date):
suggested_year = suggested_date.year suggested_year = suggested_date.year
tuesdays = [] tuesdays = []
for i in calendar.Calendar().itermonthdays4(suggested_year, suggested_month): 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: if i[3] == 1 and datetime.datetime(year=i[0], month=i[1], day=i[2]) > suggested_date:
tuesdays.append(datetime.datetime( tuesdays.append(datetime.datetime(
year=i[0], month=i[1], day=i[2], hour=hour_poll_posting)) 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 return tuesdays[0] # near tuesday

30
test.py Normal file
View file

@ -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()