# -*- coding: utf-8 -*- # !/usr/bin/env python """ 为了让任务模块不影响基本业务，需要做统一的处理 """ import logging import datetime import os import time from django.conf import settings from typing import TYPE_CHECKING, Optional import simplejson as json from celery.local import PromiseProxy from apps.thirdparties.dingding import DingDingRobot from apps.web.core.utils import async_operation # noinspection PyUnresolvedReferences from taskmanager.tasks import ( report_feedback_to_dealer_via_wechat, report_new_payment_to_dealer_via_wechat, report_to_user_via_wechat, send_msg_to_user_via_wechat, report_to_dealer_via_wechat, whale_withdraw_order_alert, withdraw_error_alert, send_topic_command, poll_dealer_recharge_record, poll_user_recharge_record, report_to_user_low_power, turn_on_power_huan_dian_gui, test_sync, check_withdraw_via_bank, generate_ad_excel_report, generate_business_stats_report_by_dealer, export_consume_order_excel_from_db, export_API_order_excel_from_db, export_on_points_order_excel_from_db, export_send_coins_to_card_order_excel_from_db, export_charge_order_excel_from_db, export_group_stat_excel_from_db, export_vcard_info_excel_from_db, export_group_user_account_excel_form_db, generate_simCharge_excel_report, generate_dealerWithDraw_excel_report, generate_biz_stats_for_manager, import_simcard_excel_to_db, export_simcard_excel_from_db, export_device_excel_from_db, manager_export_charge_order_excel_from_db, manager_export_consume_order_excel_from_db, manager_export_dealer_info_excel_from_db, export_aggregate_dealer_income, device_offline_notify, batch_set_device_params, set_device_params, batch_set_server_settings, set_server_settings, send_to_xf_all_dev_info, send_to_xf_falut, send_to_xf_fault_handle, notify_insurance_order_subscribe, notify_insurance_order_cancel, push_shanghai_platform_heatbeat, export_modify_customer_balance_record_excel_from_db, ) logger = logging.getLogger(__name__) if TYPE_CHECKING: from taskmanager import TraceTask _task_registry = {k: v for k, v in globals().items() if isinstance(v, PromiseProxy)} # type: dict def task_caller(func_name, delay = None, expires = None, offline_task_id = None, **kwargs): """ 任务中继器，为了不影响主线任务运作 注意做了保护, 如果rabbitmq锁死, 会暂时停止 所以只能提交可以允许忽略的任务 :param func_name: :param delay: :param expires: :param offline_task_id: :param kwargs: :return: """ if settings.NO_CELERY_TASK: logger.info('[task_caller] NO_CELERY_TASK') return if settings.DEBUG_CELERY_TASK_ROUTINE: try: f = _task_registry.get(func_name) # type: TraceTask logger.info('[task_caller] function(name=%s)(kwargs=%s)' % (f.__name__, json.dumps(kwargs))) async_operation(f, **kwargs) except Exception as e: logger.exception('[task_caller] error=%s' % (e,)) else: from apps.web.core.models import SystemSettings, OfflineTask if SystemSettings.get_system_setting("DISABLE_CELERY", False): logger.info('[task_caller] function(name={}) result=no publish.'.format(func_name)) if offline_task_id: offline_task = OfflineTask.objects(id = offline_task_id).first() # type: Optional[OfflineTask] if not offline_task: logger.error('no such task id({})'.format(offline_task_id)) else: updated = offline_task.update(celery_task_id = offline_task_id, status = 'FAILURE', finishedTime = datetime.datetime.now()) if not updated: logger.error(u'update offline task(id=%s) failed.' % (offline_task_id,)) else: logger.debug(u'update offline task(id=%s) success.' % (offline_task_id,)) else: if settings.CELERY_PUBLISH_METHOD == 'gevent': try: import gevent gevent.spawn(celery_task_impl, func_name, delay, expires, offline_task_id, **kwargs) except Exception as e: logger.exception(e) elif settings.CELERY_PUBLISH_METHOD == 'thread': async_operation(celery_task_impl, func_name, delay, expires, offline_task_id, **kwargs) else: celery_task_impl(func_name, delay, expires, offline_task_id, **kwargs) def celery_task_impl(func_name, delay = None, expires = None, offline_task_id = None, **kwargs): start = time.time() try: f = _task_registry.get(func_name) # type: TraceTask logger.info('[task_caller] function(name=%s)(kwargs=%s)' % (f.__name__, json.dumps(kwargs))) now_time = datetime.datetime.utcnow() if delay: eta = now_time + datetime.timedelta(seconds = delay) else: eta = None if expires: expires = now_time + datetime.timedelta(seconds = expires) else: expires = None if offline_task_id: headers = {'offline_task_id': offline_task_id} result = f.s(**kwargs).apply_async(eta = eta, expires = expires, ignore_result = True, headers = headers) # from apps.web.core.models import SystemSettings, OfflineTask # offline_task = OfflineTask.objects(id = offline_task_id).first() # type: Optional[OfflineTask] # offline_task.update(celery_task_id = result) else: result = f.s(**kwargs).apply_async(eta = eta, expires = expires, ignore_result = True) logger.info('[task_caller] function(name=%s) result=%s' % (f.__name__, result)) except Exception as e: logger.exception('[task_caller] error=%s' % (e,)) finally: end = time.time() over_time = (end - start) if over_time > 5: logger.warning('[task_caller] function(name=%s) result=exceed %s seconds.' % (func_name, over_time)) if over_time > 30: logger.warning('[task_caller] function(name=%s) result=exceed 30 seconds to disable.' % func_name) from apps.web.core.models import SystemSettings SystemSettings.set_mem_setting("DISABLE_CELERY", True) DingDingRobot().send_msg( msg = u'[CELERY告警({})] 发布任务时间超过30秒，暂时屏蔽所有发布。请检查RABBITMQ是否正常。'.format(os.environ.get('MY_ENV'))) def task_caller_once(func_name, routing_key, queue, **kwargs): """ 任务中继器，为了不影响主线任务运作 :param queue: :param routing_key: :param func_name: :param kwargs: :return: """ try: f = _task_registry.get(func_name) # type: TraceTask logger.info('[task_caller_once] function(name=%s)(kwargs=%s)' % (f.__name__, json.dumps(kwargs))) result = f.s(**kwargs).apply_async(routing_key = routing_key, queue = queue) logger.info('[task_caller_once] function(name=%s) result=%s' % (f.__name__, result)) except Exception as e: logger.error('[task_caller_once] error=%s' % (e,))