Evan：/* pre */

2025-05-31T07:46:21Z

pre

新页面

=利用python检测远程 IP和端口是否可连接并dingtalk 报警=
=pre=

如果不是很重要的场景,记得要配 ding关键字关键词一般建议英文我这里用 anomaly

create 一个dingding 群 --> 在群里添加机器人 --> 选择类型为自定义（通过webhook），并记下你的 token url 下面要用

=code=
<pre>

cat /home/evan/scr/chk-port.py
import socket
import yaml
import requests
from datetime import datetime

DING_WEBHOOK ="https://oapi.dingtalk.com/robot/send?access_token=4578bcd63ad05f3a315ef4f971c7defe9990ef0adc47f804c3273371a7355555"

def load_from_yaml(file_path):
with open(file_path,'r') as f:
data = yaml.safe_load(f)
return data.get("targets",[])

def check_port(host,port,timeout=4):
try:
with socket.create_connection((host,port),timeout=timeout):
return True
except:
return False

def send_ding_alert(message):
headers = {'Content-Type': 'application/json;charset=utf-8'}
data = {
"msgtype": "text",
"text": {
"content": message
}
}
try:
requests.post(DING_WEBHOOK, headers=headers, json=data,timeout=4)
except Exception as e:
print(f"Failed to send dingtalk alert: {str(e)}")

def main():
ips = load_from_yaml("/home/evan/scr/ips.yaml")
down_list = []

for item in ips:
host = item["host"]
port = item["port"]
if not check_port(host,port):
down_list.append(f"{host}:{port}")

if down_list:
now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
msg = f"[{now}] Port detection anomaly端口检测异常:\n" + "\n".join(down_list)
#print(msg)
#print(down_list)
send_ding_alert(msg)
else:
print("All ports are up.")
send_ding_alert("All ports are up.")

if __name__== "__main__":
main()
#send_ding_alert("这是测试消息")

cat ips.yaml
targets:
- host: 45.192.18.733
port: 7100
- host: 45.192.18.733
port: 7101
- host: 45.192.18.733
port: 7148
- host: 45.192.18.733
port: 7149

Usage

*/4 * * * * /usr/bin/python3 /home/evan/scr/chk-port.py >> /home/evan/port_checker.log 2>&1

</pre>

[[category:python]] [[category:devops]]

利用python检测远程 IP和端口是否可连接并钉钉报警 - 版本历史

Evan：​/* pre */

Evan：/* pre */