设为首页收藏本站
首页 > 论坛 > 掌控板

[求助]mpython 使用smtp发送邮件时ussl库未找到的解决方案

ok0k | 学徒 | 创造力： | 帖子： | 发消息 | 串个门 | 加好友 | 打招呼
2025-7-29 22:45:15 [显示全部楼层]
10浏览
0回复
0收藏
查看: 10|回复: 0

[求助] mpython 使用smtp发送邮件时ussl库未找到的解决方案

[复制链接]
MicroPython新版本中已经将ussl改名为ssl，但不知道为什么官方提供的umail.py没改
把掌控板中的umail替换为以下代码即可:
  1. import usocket

  2. import gc

  3. import ssl  # 替换 ussl 为 ssl

  4. 

  5. DEFAULT_TIMEOUT = 10  # sec

  6. LOCAL_DOMAIN = '127.0.0.1'

  7. CMD_EHLO = 'EHLO'

  8. CMD_STARTTLS = 'STARTTLS'

  9. CMD_AUTH = 'AUTH'

  10. CMD_MAIL = 'MAIL'

  11. AUTH_PLAIN = 'PLAIN'

  12. AUTH_LOGIN = 'LOGIN'

  13. 

  14. # 加密过程

  15. def encrypt_string(message):

  16.     encode_result = ""

  17.     for char in message:

  18.         char_int = ord(char)

  19.         if char.isalpha():  # 判断是否为字母

  20.             if 64 < char_int < 78 or 96 < char_int < 110:  # 针对其中的部分字母进行加密

  21.                 encode_result += "00" + str((char_int + 13) * 2) + "|"

  22.             else:  # 对剩下字母进行加密

  23.                 encode_result += "01" + str(char_int - 23) + "|"

  24. 

  25.         elif '\u4e00' <= char <= '\u9fff':  # 单个汉字可以这么判断

  26.             encode_result += "02" + str(char_int + 24) + "|"

  27.         else:  # 对数字、特殊字符进行加密

  28.             encode_result += "03" + str(char_int) + "|"

  29. 

  30.     return encode_result

  31. 

  32. # 解密过程

  33. def decrypt_string(message):

  34.     decode_result = ""

  35. 

  36.     # 将message转换为list

  37.     message_list = message.split("|")

  38.     message_list.remove("")  # 移除list中的空元素

  39. 

  40.     for i in message_list:

  41.         type_ = i[:2]

  42.         char_number = int(i[2:])

  43.         if type_ == "00":

  44.             char_number = int(char_number / 2 - 13)

  45.         elif type_ == "01":

  46.             char_number = char_number + 23

  47.         elif type_ == "02":

  48.             char_number = char_number - 24

  49.         else:

  50.             char_number = char_number

  51. 

  52.         decode_result += chr(char_number)

  53. 

  54.     return decode_result

  55. 

  56. DEFAULT_EMAIL = 'zhangkongban@163.com'

  57. DEFAULT_PASSWORD = encrypt_string('NTJTTHERKCSJFEXM')

  58. 

  59. class SMTP:

  60.     def cmd(self, cmd_str):

  61.         sock = self._sock

  62.         sock.write('%s\r\n' % cmd_str)

  63.         resp = []

  64.         next = True

  65.         while next:

  66.             code = sock.read(3)

  67.             if not code:  # 添加超时或断开连接检查

  68.                 break

  69.             next = sock.read(1) == b'-'

  70.             line = sock.readline().strip()

  71.             if line:

  72.                 resp.append(line.decode())

  73.         return int(code), resp

  74. 

  75.     def __init__(self, host, port, ssl_mode=False, username=None, password=None):

  76.         self.username = username

  77.         addr = usocket.getaddrinfo(host, port)[0][-1]

  78.         sock = usocket.socket(usocket.AF_INET, usocket.SOCK_STREAM)

  79.         sock.settimeout(DEFAULT_TIMEOUT)

  80.         sock.connect(addr)

  81.         

  82.         # 使用 ssl 替代 ussl

  83.         if ssl_mode:

  84.             sock = ssl.wrap_socket(sock)

  85.             code = int(sock.read(3))

  86.             sock.readline()  # 消耗剩余响应

  87.         

  88.         self._sock = sock

  89.         

  90.         # 读取初始响应（如果不是 SSL 模式）

  91.         if not ssl_mode:

  92.             code = int(sock.read(3))

  93.             sock.readline()  # 消耗剩余响应

  94.             assert code == 220, '无法连接到服务器 %d' % code

  95. 

  96.         code, resp = self.cmd(CMD_EHLO + ' ' + LOCAL_DOMAIN)

  97.         assert code == 250, 'EHLO 失败 %d' % code

  98.         

  99.         # 支持 STARTTLS

  100.         if not ssl_mode and any(CMD_STARTTLS in r for r in resp):

  101.             code, resp = self.cmd(CMD_STARTTLS)

  102.             assert code == 220, 'STARTTLS 失败 %d, %s' % (code, resp)

  103.             self._sock = ssl.wrap_socket(sock)

  104. 

  105.         if username and password:

  106.             self.login(username, password)

  107. 

  108.     def login(self, username, password):

  109.         self.username = username

  110.         code, resp = self.cmd(CMD_EHLO + ' ' + LOCAL_DOMAIN)

  111.         assert code == 250, 'EHLO 错误 %d, %s' % (code, resp)

  112. 

  113.         auths = None

  114.         for feature in resp:

  115.             if feature[:4].upper() == CMD_AUTH:

  116.                 auths = feature[4:].strip('=').upper().split()

  117.         assert auths is not None, "服务器不支持认证方式"

  118. 

  119.         from ubinascii import b2a_base64 as b64

  120.         if AUTH_PLAIN in auths:

  121.             creds = "\0%s\0%s" % (username, password)

  122.             cren = b64(creds.encode()).decode().strip()

  123.             code, resp = self.cmd('%s %s %s' % (CMD_AUTH, AUTH_PLAIN, cren))

  124.         elif AUTH_LOGIN in auths:

  125.             user_b64 = b64(username.encode()).decode().strip()

  126.             code, resp = self.cmd("%s %s %s" % (CMD_AUTH, AUTH_LOGIN, user_b64))

  127.             assert code == 334, '用户名错误 %d, %s' % (code, resp)

  128.             pass_b64 = b64(password.encode()).decode().strip()

  129.             code, resp = self.cmd(pass_b64)

  130.         else:

  131.             raise Exception("认证方式(%s)不被支持" % ', '.join(auths))

  132. 

  133.         assert code in (235, 503), '认证错误 %d, %s' % (code, resp)

  134.         return code, resp

  135. 

  136.     def to(self, addrs, mail_from=None):

  137.         mail_from = self.username if mail_from is None else mail_from

  138.         code, resp = self.cmd('MAIL FROM: <%s>' % mail_from)

  139.         assert code == 250, '发件人拒绝 %d, %s' % (code, resp)

  140. 

  141.         if isinstance(addrs, str):

  142.             addrs = [addrs]

  143.         count = 0

  144.         for addr in addrs:

  145.             code, resp = self.cmd('RCPT TO: <%s>' % addr)

  146.             if code not in (250, 251):

  147.                 print('%s 拒绝, %s' % (addr, resp))

  148.                 count += 1

  149.         assert count != len(addrs), '收件人全部拒绝, %d, %s' % (code, resp)

  150. 

  151.         code, resp = self.cmd('DATA')

  152.         assert code == 354, '数据拒绝, %d, %s' % (code, resp)

  153.         return code, resp

  154. 

  155.     def write(self, content):

  156.         if isinstance(content, str):

  157.             content = content.encode('utf-8')

  158.         self._sock.write(content)

  159. 

  160.     def send(self, content=''):

  161.         if content:

  162.             self.write(content)

  163.         self._sock.write(b'\r\n.\r\n')  # 结束标记

  164.         code = int(self._sock.read(3))

  165.         line = self._sock.readline().decode().strip()

  166.         return code, line

  167. 

  168.     def quit(self):

  169.         try:

  170.             self.cmd("QUIT")

  171.         finally:

  172.             self._sock.close()

  173. 

  174. def send_email(myusername, mypassword, target_email, SMTP_SERVER, subject, text):

  175.     server_map = {

  176.         1: ('smtp.office365.com', 587),

  177.         2: ('smtp.qq.com', 587),

  178.         3: ('smtp.126.com', 25),

  179.         4: ('smtp.163.com', 25)

  180.     }

  181.     

  182.     if SMTP_SERVER in server_map:

  183.         host, port = server_map[SMTP_SERVER]

  184.         ssl_mode = (port == 465)  # 465端口使用SSL

  185.     else:

  186.         raise ValueError("无效的SMTP服务器选择")

  187. 

  188.     if myusername == DEFAULT_EMAIL:

  189.         mypassword = decrypt_string(mypassword)

  190. 

  191.     try:

  192.         gc.collect()

  193.         smtp = SMTP(host, port, ssl_mode=ssl_mode)

  194.         smtp.login(myusername, mypassword)

  195.         smtp.to(target_email)

  196.         

  197.         # 邮件头

  198.         headers = [

  199.             "From: 掌控板 <{}>".format(myusername),

  200.             "To: <{}>".format(target_email),

  201.             "Subject: {}".format(subject),

  202.             ""

  203.         ]

  204.         

  205.         for header in headers:

  206.             smtp.write(header + "\r\n")

  207.         

  208.         smtp.write(text + "\r\n")

  209.         code, resp = smtp.send()

  210.         smtp.quit()

  211.         

  212.         if code == 250:

  213.             print("邮件发送成功!")

  214.         else:

  215.             print("邮件发送失败! 错误代码: {}, 响应: {}".format(code, resp))

  216.     except Exception as e:

  217.         print("邮件发送失败! 错误详情:")

  218.         import sys

  219.         sys.print_exception(e)

  220.         raise e
复制代码
实测有效
回复

使用道具 举报

高级模式
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 | 立即注册

本版积分规则

为本项目制作心愿单
购买心愿单
心愿单 编辑
[[wsData.name]]

硬件清单

  • [[d.name]]
btnicon
我也要做!
点击进入购买页面
撰写新帖
关于楼主

楼主的其它帖子

background 创造力兑换 background 心愿单 background 创客导航
上海智位机器人股份有限公司 沪ICP备09038501号-4 备案 沪公网安备31011502402448

© 2013-2025 Comsenz Inc. Powered by Discuz! X3.4 Licensed

mail