Celery Task¶

celery_task.celery_main ¶

Celery

on_failure ¶

on_failure(**kwargs: Any) -> None

on failure

Source code in celery_task/celery_main.py

@task_failure.connect
def on_failure(**kwargs: Any) -> None:
    ''' on failure '''
    ses = AWSSES(setting.AWS_ID, setting.AWS_KEY, setting.AWS_SES_FROM)
    raw_mail = ses.raw_mail(
        to_addresses=[setting.ADMIN_To, ],
        subject=f"[COSCUP-SECRETARY] {kwargs['sender'].name} [{kwargs['sender'].request.id}]",
        body=f"""kwargs: <pre>{kwargs['kwargs']}</pre><br>
einfo: <pre>{kwargs['einfo'].traceback}</pre><br>
request: <pre>{kwargs['sender'].request}</pre>""",
    )
    ses.send_raw_email(data=raw_mail)

celery_task.task_expense ¶

Task expense

expense_create ¶

expense_create(
    sender: Any, **kwargs: dict[str, Any]
) -> None

Expense create

Source code in celery_task/task_expense.py

@app.task(bind=True, name='expense.create',
          autoretry_for=(Exception, ), retry_backoff=True, max_retries=2,
          routing_key='cs.expense.create', exchange='COSCUP-SECRETARY')
def expense_create(sender: Any, **kwargs: dict[str, Any]) -> None:  # pylint: disable=too-many-locals
    ''' Expense create '''
    pid = kwargs['expense']['pid']
    buid = kwargs['expense']['request']['buid']

    budget = None
    for data in Budget.get(buids=[buid, ], pid=pid):
        budget = data

    if not budget:
        return

    uids = set()

    for team in TeamDB(pid=None, tid=None).find({'pid': pid, 'tid': 'finance'}):
        uids.update(team['chiefs'])
        uids.update(team['members'])

    project = Project.get(pid=pid)
    if project:
        uids.update(project.owners)

    uids.add(kwargs['expense']['create_by'])

    logger.info(uids)

    users = User.get_info(uids=list(uids))

    mmt = MattermostTools(token=setting.MATTERMOST_BOT_TOKEN,
                          base_url=setting.MATTERMOST_BASEURL)
    for uid in uids:
        mid = mmt.find_possible_mid(uid=uid)
        if mid:
            channel_info = mmt.create_a_direct_message(
                users=(mid, setting.MATTERMOST_BOT_ID)).json()
            message = f"收到 **{users[kwargs['expense']['create_by']]['profile']['badge_name']}**" +\
                f"申請費用 - **[{kwargs['expense']['code']}] / {budget['name']}**" +\
                f"，前往 [管理費用](https://volunteer.coscup.org/expense/{pid})"

            resp = mmt.posts(
                channel_id=channel_info['id'],
                message=message,
            )
            logger.info(resp.json())

celery_task.task_ipinfo ¶

Task ipinfo

ipinfo_update_usession ¶

ipinfo_update_usession(sender: Any) -> None

IPInfo update usession

Source code in celery_task/task_ipinfo.py

@app.task(bind=True, name='ipinfo.update.usession',
          autoretry_for=(Exception, ), retry_backoff=True, max_retries=5,
          routing_key='cs.ipinfo.update.usession', exchange='COSCUP-SECRETARY')
def ipinfo_update_usession(sender: Any) -> None:
    ''' IPInfo update usession '''
    for user in USession.get_no_ipinfo():
        ipinfo_update_usession_one.apply_async(
            kwargs={'ip': user['header']['X-Real-Ip'], 'sid': user['_id']})

ipinfo_update_usession_one ¶

ipinfo_update_usession_one(
    sender: Any, **kwargs: str
) -> None

update session ipinfo

Source code in celery_task/task_ipinfo.py

@app.task(bind=True, name='ipinfo.update.usession.one',
          autoretry_for=(Exception, ), retry_backoff=True, max_retries=5,
          routing_key='cs.ipinfo.update.usession.one', exchange='COSCUP-SECRETARY')
def ipinfo_update_usession_one(sender: Any, **kwargs: str) -> None:
    ''' update session ipinfo '''
    logger.info(kwargs)

    resp = IPInfo(setting.IPINFO_TOKEN).get_info(ip_address=kwargs['ip'])
    USession.update_ipinfo(sid=kwargs['sid'], data=resp.json())

session_daily_clean ¶

session_daily_clean(sender: Any) -> None

Daily clean session

Source code in celery_task/task_ipinfo.py

@app.task(bind=True, name='session.daily.clean',
          autoretry_for=(Exception, ), retry_backoff=True, max_retries=5,
          routing_key='cs.session.daily.clean', exchange='COSCUP-SECRETARY')
def session_daily_clean(sender: Any) -> None:
    ''' Daily clean session '''
    clean = USession.clean()
    logger.info('matched: %s, modified: %s, raw_result: %s',
                clean.matched_count, clean.modified_count, clean.raw_result)

celery_task.task_mail_sys ¶

Task mail sys