import logging import datetime import asyncio from db.constants import TIMELINE_ORDER_CONFIRMED, TIMELINE_ORDER_ISSUED, TIMELINE_ORDER_NOT_ISSUED, TIMELINE_ORDER_NOT_PUNISHED, TIMELINE_ORDER_PUNISHED from util import make_session from generate import generate_order, generate_punishment from db.queries import domsubusers_doms, order_status_by_id, order_status_put, order_status_confirm, timeline_event_put from mastodon import Mastodon from telegram.telegram import Telegram from settings import ENV from util import timezone logger = logging.getLogger(__name__) async def order_mastodon_post(session, orders_pool, orders_str, repeats, due_at, verify_at=None): user = orders_pool.user post = "Here are today's orders for @%s -\n\n" % user.mastodon_account() post += orders_str + "\n\n" if repeats > 1: post += f"These are the same orders from the last {repeats} days\n\n" post += "Proof of compliance is due by " + due_at.strftime("%I:%M %p") + "\n" if verify_at is not None: post += "Verification due by " + verify_at.strftime("%I:%M %p") + "\n" post += "\n" if ENV == 'dev': post += "⚠️ DEV" elif user.mastodon_attn_list: post += f"ATTN - {user.mastodon_attn_list}\n" m = Mastodon(session) return await m.statusPost(post, user) async def order_telegram_post(session, orders_pool, orders_str, repeats, due_at, m_url, verify_at=None): post = "Here are your orders -\n\n" post += orders_str + "\n\n" if repeats > 1: post += f"These are the same orders from the last {repeats} days\n\n" post += "Proof of compliance is due by " + due_at.strftime("%I:%M %p") + "\n" if verify_at is not None: post += "Verification due by " + verify_at.strftime("%I:%M %p") + "\n" post += "\n" post += m_url if ENV == 'dev': post += "\n⚠️ DEV" t = Telegram(session) await t.message_send(orders_pool.user.telegram_chat_id, post) async def order_telegram_post_need_mastodon(session, orders_pool): post = "Cannot issue an order without a mastodon username" if ENV == 'dev': post += "\n⚠️ DEV" t = Telegram(session) await t.message_send(orders_pool.user.telegram_chat_id, post) async def order_telegram_post_none(session, orders_pool): post = "No orders for today" if ENV == 'dev': post += "\n⚠️ DEV" t = Telegram(session) await t.message_send(orders_pool.user.telegram_chat_id, post) async def order_issue(orders_pool): async with make_session() as session: user = orders_pool.user if user.mastodon_username is None: logger.info(f"{orders_pool} - Cannot issue order without mastodon username") await order_telegram_post_need_mastodon(session, orders_pool) timeline_event_put( TIMELINE_ORDER_NOT_ISSUED, "Cannot issue order without mastodon username", user=orders_pool.user, orders_pool=orders_pool ) return { "reason": "Cannot issue order without mastodon username" } orders_info = generate_order(orders_pool) if 'orders' not in orders_info: logger.info(f"{orders_pool} - {orders_info['reason']}") await order_telegram_post_none(session, orders_pool) timeline_event_put( TIMELINE_ORDER_NOT_ISSUED, orders_info['reason'], user=orders_pool.user, orders_pool=orders_pool ) return { "reason": orders_info['reason'] } orders_str = "\n".join(orders_info['orders']) created_at = datetime.datetime.now(tz=timezone()) due_at = None verify_at = None if orders_pool.confirm_delay is not None: due_at = created_at + datetime.timedelta(hours=orders_pool.confirm_delay) if user.verify_mastodon_favorite: verify_at = due_at + datetime.timedelta(hours=user.verify_delay) repeats_count = orders_info.get('count', 0) m_status = await order_mastodon_post( session, orders_pool, orders_str, repeats_count, due_at, verify_at=verify_at ) await order_telegram_post( session, orders_pool, orders_str, repeats_count, due_at, m_status['url'], verify_at=verify_at ) order_status = order_status_put( orders_pool, orders_pool.user, m_status['id'], created_at, due_at, orders_str, verify_at=verify_at ) timeline_event_put( TIMELINE_ORDER_ISSUED, order_status.text.split("\n")[0], user=orders_pool.user, orders_pool=orders_pool, order_status=order_status, extra={ "mastodon_status_url": m_status['url'] } ) logger.info(f"Issued {order_status}") return { "order_status" : order_status, "mastodon_status": m_status } async def punishment_mastodon_post(session, orders_pool, punishment_str, reply_id=None): user = orders_pool.user post = "@%s has failed to post proof of compliance. Here is the punishment -\n\n" % user.mastodon_account() post += punishment_str + "\n\n" if ENV == 'dev': post += "⚠️ DEV" elif user.mastodon_attn_list: post += f"ATTN - {user.mastodon_attn_list}\n" m = Mastodon(session) return await m.statusPost( post, user, in_reply_to_id=reply_id ) async def punishment_telegram_post(session, orders_pool, punishment_str, m_url): post = "You failed to show proof of compliance. Here is your punishment -\n\n" post += punishment_str + "\n\n" post += m_url if ENV == 'dev': post += "\n\n⚠️ DEV" t = Telegram(session) await t.message_send(orders_pool.user.telegram_chat_id, post) async def punishment_issue(session, order_status): if order_status.pool is None or order_status.pool.punishment_pool is None: logger.info(f'Unable to issue a punishment for {order_status}, no punishment pool for order pool {order_status.pool.name}') return { "reason": "No punishment pool"} punishment_pool = order_status.pool.punishment_pool punishment = generate_punishment(punishment_pool) punishment_str = "\n".join(punishment['orders']) punishment_status = await punishment_mastodon_post( session, punishment_pool, punishment_str, order_status.mastodon_id, ) await punishment_telegram_post( session, punishment_pool, punishment_str, punishment_status['url'] ) return { "order_status": order_status_put( punishment_pool, order_status.user, punishment_status['id'], punishment_status['created_at'], None, punishment_str, punishment_for=order_status ), "mastodon_status": punishment_status } async def status_has_favorites(m, status_id, user): favorites = await m.statusFavorites(status_id) user_mastodon_account = user.mastodon_account() accts = [f['acct'] for f in favorites] for dom in domsubusers_doms(user): dom_account = dom.mastodon_account() if dom_account is not None and dom_account in accts: return True for attn_user in user.mastodon_attn_list.split(" "): if (attn_user[1:] != user_mastodon_account and attn_user[1:] in accts): return True return False order_check_lock = asyncio.Lock() async def order_check(order_status_id): async with order_check_lock: async with make_session() as session: order_status = order_status_by_id(order_status_id) user = order_status.user if order_status.punishment.count() > 0: logger.info(f'Punishment already issued for {order_status}') return m = Mastodon(session) context = await m.statusContext(order_status.mastodon_id) confirmed_at = None had_replies = False had_reply_on_time = False had_alt_text = False had_media_attachment = False had_favorites = False for d in context['descendants']: if ( d['in_reply_to_id'] == order_status.mastodon_id and d['account']['acct'] == order_status.user.mastodon_account() ): had_replies = True had_reply_on_time = False had_alt_text = False had_media_attachment = False had_favorites = False if (datetime.datetime.fromisoformat(d['created_at']) < datetime.datetime.fromisoformat(order_status.due_at)): had_reply_on_time = True else: continue if len(d['media_attachments']) > 0: had_media_attachment = True else: continue if user.verify_mastodon_alt_text: if all([( 'description' in a and a['description'] is not None ) for a in d['media_attachments']]): had_alt_text = True else: continue if user.verify_mastodon_favorite: if await status_has_favorites(m, d['id'], order_status.user): had_favorites = True else: continue confirmed_at = d['created_at'] order_status_confirm(order_status.id, confirmed_at) logger.info(f"Confirmed order {order_status}") timeline_event_put( TIMELINE_ORDER_CONFIRMED, order_status.text.split("\n")[0], order_status.user, order_status.pool, order_status, extra={ "mastodon_status_url": d['url'] } ) break if confirmed_at is None: logger.info(f"Order {order_status} remains unconfirmed") due_at = datetime.datetime.fromisoformat(order_status.due_at) if(due_at < datetime.datetime.now(datetime.UTC)): reason = None if had_replies is False: reason = "No replies were found" elif had_reply_on_time is False: reason = "Reply was after due date" elif had_media_attachment is False: reason = "No replies had a media attachment" elif user.verify_mastodon_alt_text and had_alt_text is False: reason = "Not all media attachments had alt text" elif user.verify_mastodon_favorite and had_favorites is False: reason = "No replies had a favorite from a dom or tagged account" logger.info(f"Time to issue a punishment for {order_status}") issue_result = await punishment_issue(session, order_status) if 'order_status' in issue_result: punishment_status = issue_result['order_status'] log_text = (reason + "\n") if reason is not None else '' log_text += punishment_status.text.split("\n")[0] timeline_event_put( TIMELINE_ORDER_PUNISHED, log_text, punishment_status.user, punishment_status.pool, punishment_status, extra={ "mastodon_status_url": issue_result['mastodon_status']['url'] } ) elif 'reason' in issue_result: timeline_event_put( TIMELINE_ORDER_NOT_PUNISHED, issue_result['reason'], order_status.user, order_status.pool, order_status )