Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
jobily
Nightingale
提交
93f88296
N
Nightingale
项目概览
jobily
/
Nightingale
通知
1
Star
0
Fork
0
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
0
列表
看板
标记
里程碑
合并请求
0
DevOps
流水线
流水线任务
计划
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
N
Nightingale
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
0
Issue
0
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
DevOps
DevOps
流水线
流水线任务
计划
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
流水线任务
提交
Issue看板
前往新版Gitcode,体验更适合开发者的 AI 搜索 >>
提交
93f88296
编写于
3月 01, 2022
作者:
U
Ulric Qin
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
update notify.py in docker dir
上级
1f4e8e75
变更
2
隐藏空白更改
内联
并排
Showing
2 changed file
with
215 addition
and
147 deletion
+215
-147
docker/n9eetc/script/notify.bak.py
docker/n9eetc/script/notify.bak.py
+206
-0
docker/n9eetc/script/notify.py
docker/n9eetc/script/notify.py
+9
-147
未找到文件。
docker/n9eetc/script/notify.bak.py
0 → 100755
浏览文件 @
93f88296
#!/usr/bin/env python
# -*- coding: UTF-8 -*-
import
sys
import
json
import
urllib2
import
smtplib
from
email.mime.text
import
MIMEText
notify_channel_funcs
=
{
"email"
:
"email"
,
"sms"
:
"sms"
,
"voice"
:
"voice"
,
"dingtalk"
:
"dingtalk"
,
"wecom"
:
"wecom"
,
"feishu"
:
"feishu"
}
mail_host
=
"smtp.163.com"
mail_port
=
994
mail_user
=
"ulricqin"
mail_pass
=
"password"
mail_from
=
"ulricqin@163.com"
class
Sender
(
object
):
@
classmethod
def
send_email
(
cls
,
payload
):
if
mail_user
==
"ulricqin"
and
mail_pass
==
"password"
:
print
(
"invalid smtp configuration"
)
return
users
=
payload
.
get
(
'event'
).
get
(
"notify_users_obj"
)
emails
=
{}
for
u
in
users
:
if
u
.
get
(
"email"
):
emails
[
u
.
get
(
"email"
)]
=
1
if
not
emails
:
return
recipients
=
emails
.
keys
()
mail_body
=
payload
.
get
(
'tpls'
).
get
(
"mailbody.tpl"
,
"mailbody.tpl not found"
)
message
=
MIMEText
(
mail_body
,
'html'
,
'utf-8'
)
message
[
'From'
]
=
mail_from
message
[
'To'
]
=
", "
.
join
(
recipients
)
message
[
"Subject"
]
=
payload
.
get
(
'tpls'
).
get
(
"subject.tpl"
,
"subject.tpl not found"
)
try
:
smtp
=
smtplib
.
SMTP_SSL
(
mail_host
,
mail_port
)
smtp
.
login
(
mail_user
,
mail_pass
)
smtp
.
sendmail
(
mail_from
,
recipients
,
message
.
as_string
())
smtp
.
close
()
except
smtplib
.
SMTPException
,
error
:
print
(
error
)
@
classmethod
def
send_wecom
(
cls
,
payload
):
users
=
payload
.
get
(
'event'
).
get
(
"notify_users_obj"
)
tokens
=
{}
for
u
in
users
:
contacts
=
u
.
get
(
"contacts"
)
if
contacts
.
get
(
"wecom_robot_token"
,
""
):
tokens
[
contacts
.
get
(
"wecom_robot_token"
,
""
)]
=
1
opener
=
urllib2
.
build_opener
(
urllib2
.
HTTPHandler
())
method
=
"POST"
for
t
in
tokens
:
url
=
"https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key={}"
.
format
(
t
)
body
=
{
"msgtype"
:
"markdown"
,
"markdown"
:
{
"content"
:
payload
.
get
(
'tpls'
).
get
(
"wecom.tpl"
,
"wecom.tpl not found"
)
}
}
request
=
urllib2
.
Request
(
url
,
data
=
json
.
dumps
(
body
))
request
.
add_header
(
"Content-Type"
,
'application/json;charset=utf-8'
)
request
.
get_method
=
lambda
:
method
try
:
connection
=
opener
.
open
(
request
)
print
(
connection
.
read
())
except
urllib2
.
HTTPError
,
error
:
print
(
error
)
@
classmethod
def
send_dingtalk
(
cls
,
payload
):
users
=
payload
.
get
(
'event'
).
get
(
"notify_users_obj"
)
tokens
=
{}
phones
=
{}
for
u
in
users
:
if
u
.
get
(
"phone"
):
phones
[
u
.
get
(
"phone"
)]
=
1
contacts
=
u
.
get
(
"contacts"
)
if
contacts
.
get
(
"dingtalk_robot_token"
,
""
):
tokens
[
contacts
.
get
(
"dingtalk_robot_token"
,
""
)]
=
1
opener
=
urllib2
.
build_opener
(
urllib2
.
HTTPHandler
())
method
=
"POST"
for
t
in
tokens
:
url
=
"https://oapi.dingtalk.com/robot/send?access_token={}"
.
format
(
t
)
body
=
{
"msgtype"
:
"text"
,
"text"
:
{
"content"
:
payload
.
get
(
'tpls'
).
get
(
"dingtalk.tpl"
,
"dingtalk.tpl not found"
)
},
"at"
:
{
"atMobiles"
:
phones
.
keys
(),
"isAtAll"
:
False
}
}
request
=
urllib2
.
Request
(
url
,
data
=
json
.
dumps
(
body
))
request
.
add_header
(
"Content-Type"
,
'application/json;charset=utf-8'
)
request
.
get_method
=
lambda
:
method
try
:
connection
=
opener
.
open
(
request
)
print
(
connection
.
read
())
except
urllib2
.
HTTPError
,
error
:
print
(
error
)
@
classmethod
def
send_feishu
(
cls
,
payload
):
users
=
payload
.
get
(
'event'
).
get
(
"notify_users_obj"
)
tokens
=
{}
phones
=
{}
for
u
in
users
:
if
u
.
get
(
"phone"
):
phones
[
u
.
get
(
"phone"
)]
=
1
contacts
=
u
.
get
(
"contacts"
)
if
contacts
.
get
(
"feishu_robot_token"
,
""
):
tokens
[
contacts
.
get
(
"feishu_robot_token"
,
""
)]
=
1
opener
=
urllib2
.
build_opener
(
urllib2
.
HTTPHandler
())
method
=
"POST"
for
t
in
tokens
:
url
=
"https://open.feishu.cn/open-apis/bot/v2/hook/{}"
.
format
(
t
)
body
=
{
"msg_type"
:
"text"
,
"content"
:
{
"text"
:
payload
.
get
(
'tpls'
).
get
(
"feishu.tpl"
,
"feishu.tpl not found"
)
},
"at"
:
{
"atMobiles"
:
phones
.
keys
(),
"isAtAll"
:
False
}
}
request
=
urllib2
.
Request
(
url
,
data
=
json
.
dumps
(
body
))
request
.
add_header
(
"Content-Type"
,
'application/json;charset=utf-8'
)
request
.
get_method
=
lambda
:
method
try
:
connection
=
opener
.
open
(
request
)
print
(
connection
.
read
())
except
urllib2
.
HTTPError
,
error
:
print
(
error
)
@
classmethod
def
send_sms
(
cls
,
payload
):
users
=
payload
.
get
(
'event'
).
get
(
"notify_users_obj"
)
phones
=
{}
for
u
in
users
:
if
u
.
get
(
"phone"
):
phones
[
u
.
get
(
"phone"
)]
=
1
if
phones
:
print
(
"send_sms not implemented, phones: {}"
.
format
(
phones
.
keys
()))
@
classmethod
def
send_voice
(
cls
,
payload
):
users
=
payload
.
get
(
'event'
).
get
(
"notify_users_obj"
)
phones
=
{}
for
u
in
users
:
if
u
.
get
(
"phone"
):
phones
[
u
.
get
(
"phone"
)]
=
1
if
phones
:
print
(
"send_voice not implemented, phones: {}"
.
format
(
phones
.
keys
()))
def
main
():
payload
=
json
.
load
(
sys
.
stdin
)
with
open
(
".payload"
,
'w'
)
as
f
:
f
.
write
(
json
.
dumps
(
payload
,
indent
=
4
))
for
ch
in
payload
.
get
(
'event'
).
get
(
'notify_channels'
):
send_func_name
=
"send_{}"
.
format
(
notify_channel_funcs
.
get
(
ch
.
strip
()))
if
not
hasattr
(
Sender
,
send_func_name
):
print
(
"function: {} not found"
,
send_func_name
)
continue
send_func
=
getattr
(
Sender
,
send_func_name
)
send_func
(
payload
)
def
hello
():
print
(
"hello nightingale"
)
if
__name__
==
"__main__"
:
if
len
(
sys
.
argv
)
==
1
:
main
()
elif
sys
.
argv
[
1
]
==
"hello"
:
hello
()
else
:
print
(
"I am confused"
)
\ No newline at end of file
docker/n9eetc/script/notify.py
浏览文件 @
93f88296
...
...
@@ -2,165 +2,27 @@
# -*- coding: UTF-8 -*-
import
sys
import
json
import
urllib2
import
smtplib
from
email.mime.text
import
MIMEText
notify_channel_funcs
=
{
"email"
:
"email"
,
"sms"
:
"sms"
,
"voice"
:
"voice"
,
"dingtalk"
:
"dingtalk"
,
"wecom"
:
"wecom"
,
"feishu"
:
"feishu"
}
mail_host
=
"smtp.163.com"
mail_port
=
994
mail_user
=
"ulricqin"
mail_pass
=
"password"
mail_from
=
"ulricqin@163.com"
class
Sender
(
object
):
@
classmethod
def
send_email
(
cls
,
payload
):
if
mail_user
==
"ulricqin"
and
mail_pass
==
"password"
:
print
(
"invalid smtp configuration"
)
return
users
=
payload
.
get
(
'event'
).
get
(
"notify_users_obj"
)
emails
=
{}
for
u
in
users
:
if
u
.
get
(
"email"
):
emails
[
u
.
get
(
"email"
)]
=
1
if
not
emails
:
return
recipients
=
emails
.
keys
()
mail_body
=
payload
.
get
(
'tpls'
).
get
(
"mailbody.tpl"
,
"mailbody.tpl not found"
)
message
=
MIMEText
(
mail_body
,
'html'
,
'utf-8'
)
message
[
'From'
]
=
mail_from
message
[
'To'
]
=
", "
.
join
(
recipients
)
message
[
"Subject"
]
=
payload
.
get
(
'tpls'
).
get
(
"subject.tpl"
,
"subject.tpl not found"
)
try
:
smtp
=
smtplib
.
SMTP_SSL
(
mail_host
,
mail_port
)
smtp
.
login
(
mail_user
,
mail_pass
)
smtp
.
sendmail
(
mail_from
,
recipients
,
message
.
as_string
())
smtp
.
close
()
except
smtplib
.
SMTPException
,
error
:
print
(
error
)
# already done in go code
pass
@
classmethod
def
send_wecom
(
cls
,
payload
):
users
=
payload
.
get
(
'event'
).
get
(
"notify_users_obj"
)
tokens
=
{}
for
u
in
users
:
contacts
=
u
.
get
(
"contacts"
)
if
contacts
.
get
(
"wecom_robot_token"
,
""
):
tokens
[
contacts
.
get
(
"wecom_robot_token"
,
""
)]
=
1
opener
=
urllib2
.
build_opener
(
urllib2
.
HTTPHandler
())
method
=
"POST"
for
t
in
tokens
:
url
=
"https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key={}"
.
format
(
t
)
body
=
{
"msgtype"
:
"markdown"
,
"markdown"
:
{
"content"
:
payload
.
get
(
'tpls'
).
get
(
"wecom.tpl"
,
"wecom.tpl not found"
)
}
}
request
=
urllib2
.
Request
(
url
,
data
=
json
.
dumps
(
body
))
request
.
add_header
(
"Content-Type"
,
'application/json;charset=utf-8'
)
request
.
get_method
=
lambda
:
method
try
:
connection
=
opener
.
open
(
request
)
print
(
connection
.
read
())
except
urllib2
.
HTTPError
,
error
:
print
(
error
)
# already done in go code
pass
@
classmethod
def
send_dingtalk
(
cls
,
payload
):
users
=
payload
.
get
(
'event'
).
get
(
"notify_users_obj"
)
tokens
=
{}
phones
=
{}
for
u
in
users
:
if
u
.
get
(
"phone"
):
phones
[
u
.
get
(
"phone"
)]
=
1
contacts
=
u
.
get
(
"contacts"
)
if
contacts
.
get
(
"dingtalk_robot_token"
,
""
):
tokens
[
contacts
.
get
(
"dingtalk_robot_token"
,
""
)]
=
1
opener
=
urllib2
.
build_opener
(
urllib2
.
HTTPHandler
())
method
=
"POST"
for
t
in
tokens
:
url
=
"https://oapi.dingtalk.com/robot/send?access_token={}"
.
format
(
t
)
body
=
{
"msgtype"
:
"text"
,
"text"
:
{
"content"
:
payload
.
get
(
'tpls'
).
get
(
"dingtalk.tpl"
,
"dingtalk.tpl not found"
)
},
"at"
:
{
"atMobiles"
:
phones
.
keys
(),
"isAtAll"
:
False
}
}
request
=
urllib2
.
Request
(
url
,
data
=
json
.
dumps
(
body
))
request
.
add_header
(
"Content-Type"
,
'application/json;charset=utf-8'
)
request
.
get_method
=
lambda
:
method
try
:
connection
=
opener
.
open
(
request
)
print
(
connection
.
read
())
except
urllib2
.
HTTPError
,
error
:
print
(
error
)
# already done in go code
pass
@
classmethod
def
send_feishu
(
cls
,
payload
):
users
=
payload
.
get
(
'event'
).
get
(
"notify_users_obj"
)
tokens
=
{}
phones
=
{}
for
u
in
users
:
if
u
.
get
(
"phone"
):
phones
[
u
.
get
(
"phone"
)]
=
1
contacts
=
u
.
get
(
"contacts"
)
if
contacts
.
get
(
"feishu_robot_token"
,
""
):
tokens
[
contacts
.
get
(
"feishu_robot_token"
,
""
)]
=
1
opener
=
urllib2
.
build_opener
(
urllib2
.
HTTPHandler
())
method
=
"POST"
for
t
in
tokens
:
url
=
"https://open.feishu.cn/open-apis/bot/v2/hook/{}"
.
format
(
t
)
body
=
{
"msg_type"
:
"text"
,
"content"
:
{
"text"
:
payload
.
get
(
'tpls'
).
get
(
"feishu.tpl"
,
"feishu.tpl not found"
)
},
"at"
:
{
"atMobiles"
:
phones
.
keys
(),
"isAtAll"
:
False
}
}
request
=
urllib2
.
Request
(
url
,
data
=
json
.
dumps
(
body
))
request
.
add_header
(
"Content-Type"
,
'application/json;charset=utf-8'
)
request
.
get_method
=
lambda
:
method
try
:
connection
=
opener
.
open
(
request
)
print
(
connection
.
read
())
except
urllib2
.
HTTPError
,
error
:
print
(
error
)
# already done in go code
pass
@
classmethod
def
send_sms
(
cls
,
payload
):
...
...
@@ -187,7 +49,7 @@ def main():
with
open
(
".payload"
,
'w'
)
as
f
:
f
.
write
(
json
.
dumps
(
payload
,
indent
=
4
))
for
ch
in
payload
.
get
(
'event'
).
get
(
'notify_channels'
):
send_func_name
=
"send_{}"
.
format
(
notify_channel_funcs
.
get
(
ch
.
strip
()
))
send_func_name
=
"send_{}"
.
format
(
ch
.
strip
(
))
if
not
hasattr
(
Sender
,
send_func_name
):
print
(
"function: {} not found"
,
send_func_name
)
continue
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录