Django+Celery学习笔记5——定时推送消息

from __future__ import absolute_import, unicode_literals

from djangocelerydemo.celery import app
import requests,json

@app.task
def task_send_dd_text(url, msg, atMoblies, atAll="flase"):
    """
    发送钉钉提醒
    :param url:
    :param msg:
    :param atMoblies:
    :param atAll:
    :return:
    """
    body = {
        "msgtype": "text",
        "text": {
            "content": msg
        },
        "at": {
            "atMobiles": atMoblies, # 被艾特人的手机号码
            "isAtAll": atAll
        }


    }
    headers = {'content-type': 'application/json',
               }
    r = requests.post(url, headers=headers, data=json.dumps(body))

    print(r.text)

@app.task
def dd_task():
    url = 'https://oapi.dingtalk.com/robot/send?access_token=****************************'

    task_send_dd_text.delay(url=url,msg='【钉钉消息播报】：\n test-services服务正常\n test-manager服务正常\n',atMoblies=["18612345678"],atAll='true')

# django发送邮件
EMAIL_BACKEND = 'django.core.mail.backends.smtp.EmailBackend'
EMAIL_USE_TLS = False   #是否使用TLS安全传输协议(用于在两个通信应用程序之间提供保密性和数据完整性。)
EMAIL_USE_SSL = False    #是否使用SSL加密，qq企业邮箱要求使用
EMAIL_HOST = 'smtp.126.com'  # 如果是 163 改成 smtp.163.com
EMAIL_PORT = 25
EMAIL_HOST_USER = '####@126.com' # 帐号
EMAIL_HOST_PASSWORD = '####'  # 密码
DEFAULT_FROM_EMAIL = EMAIL_HOST_USER

from django.core.mail import send_mail,send_mass_mail
from __future__ import absolute_import, unicode_literals

@app.task
def task_send_mail(*args, **kwargs):
    """
    django 的 发送邮件，支持 html，html_message="html 内容"
    :param args:
    :param kwargs:
    :return:
    """
    # message = ('Subject1 here', 'Here is the message', '', ['136633063@qq.com'])
    try:
        send_mail('自动发送邮件测试', '这里是邮件消息内容', '', ['136633063@qq.com'],fail_silently=False)
    except Exception as e:
        print("失败信息:%s"%e)
        return {"msg":"发送失败"}
    else:
        return {"msg":"发送成功"}

from django.core.mail import send_mail,send_mass_mail
import django
import os
# 防止报错
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "djangocelerydemo.settings") # project_name 项目名称
django.setup()

message1 = ('Subject1 here', 'Here is the message', '', ['136633063@qq.com'])
message2 = ('Subject2 here', 'Here is the message', '', ['136633063@qq.com'])
from email.header import make_header
# 单个邮件
# send_mail('Subject1 here', 'Here is the message', '', ['136633063@qq.com'],fail_silently=False)

# 多个邮件
send_mass_mail((message1,message2), fail_silently=False)

from django.core.mail import EmailMultiAlternatives
from email.header import make_header

subject = '定时任务测试邮件附件主题'

text_content = '这是一封重要的邮件.'

html_content = '<p>这是一封<strong>重要的</strong>邮件.</p>'

msg = EmailMultiAlternatives(subject, text_content, '', ['136633063@qq.com'])

msg.attach_alternative(html_content, "text/html")
# 发送附件
file_path = r"E:\djangocelerydemo\test附件.html"   # 发送附件的文件路径
text = open(file_path, 'rb').read()
file_name = os.path.basename(file_path)
# 对文件进行编码处理
b = make_header([(file_name, 'utf-8')]).encode('utf-8')
msg.attach(b, text)   # 传入文件名和附件
try:
    msg.send()
except Exception as  e:
    print(e)
else:
    print("发送成功！")
finally:
    print("程序结束！")