114 lines
3.9 KiB
Python
114 lines
3.9 KiB
Python
import os
|
|
import sys
|
|
import logging
|
|
import argparse
|
|
import asyncio
|
|
import aiohttp
|
|
import datetime
|
|
|
|
from generate import generate_order, generate_punishment
|
|
from mastodon import Mastodon
|
|
from db import Database
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
ORDER_TIMEOUT = datetime.timedelta(
|
|
hours=os.getenv('ORDER_TIMEOUT', default=3)
|
|
)
|
|
MASTODON_USERNAME = os.getenv('MASTODON_USERNAME')
|
|
|
|
def make_session():
|
|
return aiohttp.ClientSession()
|
|
|
|
async def immediate():
|
|
orders_info = generate_order()
|
|
|
|
if 'orders' not in orders_info:
|
|
logger.info("No orders for today")
|
|
return
|
|
|
|
post = "Here are today's orders for @%s - \n\n" % MASTODON_USERNAME
|
|
if "count" in orders_info and orders_info['count'] > 1:
|
|
post += f"These are the same orders from the last {orders_info['count']} days\n\n"
|
|
post += "\n".join(orders_info['orders'])
|
|
|
|
async with make_session() as session:
|
|
m = Mastodon(session)
|
|
status = await m.statusPost(post)
|
|
db = Database()
|
|
|
|
db.order_status_put(
|
|
status['id'],
|
|
status['created_at'],
|
|
post
|
|
)
|
|
|
|
async def check():
|
|
async with make_session() as session:
|
|
m = Mastodon(session)
|
|
db = Database()
|
|
|
|
outstanding_orders = db.order_status_outstanding()
|
|
for outstanding_order in outstanding_orders:
|
|
context = await m.statusContext(outstanding_order['mastodon_id'])
|
|
|
|
confirmed_at = None
|
|
for d in context['descendants']:
|
|
if (
|
|
d['in_reply_to_id'] == outstanding_order['mastodon_id'] and
|
|
d['account']['username'] == MASTODON_USERNAME and
|
|
len(d['media_attachments']) > 0
|
|
):
|
|
confirmed_at = d['created_at']
|
|
db.order_status_confirm(outstanding_order['id'], confirmed_at)
|
|
logger.info('Confirmed order %s' % (outstanding_order['created_at']))
|
|
break
|
|
|
|
if confirmed_at is None:
|
|
logger.info('Order %s remains unconfirmed' % (outstanding_order['created_at']))
|
|
|
|
oo_created_at = datetime.datetime.fromisoformat(outstanding_order['created_at'])
|
|
if(oo_created_at + ORDER_TIMEOUT < datetime.datetime.now(datetime.UTC)):
|
|
logger.info('Time to issue a punishment for %s' % outstanding_order['created_at'])
|
|
|
|
punishment = generate_punishment()
|
|
|
|
post = "@%s has failed to post proof of compliance. Punishment is as follows -\n\n" % MASTODON_USERNAME
|
|
post += "\n".join(punishment)
|
|
|
|
punishment_status = await m.statusPost(
|
|
post,
|
|
in_reply_to_id=outstanding_order['mastodon_id']
|
|
)
|
|
db.punishment_status_put(
|
|
outstanding_order['id'],
|
|
punishment_status['id'],
|
|
punishment_status['created_at'],
|
|
post
|
|
)
|
|
|
|
if __name__=='__main__':
|
|
logging.basicConfig(
|
|
format="%(asctime)s %(module)s [%(levelname)-4.4s] %(message)s",
|
|
handlers=[
|
|
logging.StreamHandler(sys.stdout)
|
|
],
|
|
level=logging.INFO)
|
|
|
|
parser = argparse.ArgumentParser(description="Generate gear orders", prog='PROG')
|
|
|
|
subparsers = parser.add_subparsers(help="Sub-command help", dest="command")
|
|
|
|
parser_immediate = subparsers.add_parser('immediate', help='Immediately generate a command')
|
|
parser_check = subparsers.add_parser('check', help="Checks if any orders are outstanding")
|
|
|
|
args = parser.parse_args()
|
|
|
|
if args.command == 'immediate':
|
|
loop = asyncio.new_event_loop()
|
|
loop.run_until_complete(immediate())
|
|
loop.close()
|
|
elif args.command == 'check':
|
|
loop = asyncio.new_event_loop()
|
|
loop.run_until_complete(check())
|
|
loop.close()
|