wechat_message.py 4.2KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192
  1. import re
  2. from bridge.context import ContextType
  3. from channel.chat_message import ChatMessage
  4. from common.log import logger
  5. from common.tmp_dir import TmpDir
  6. from lib import itchat
  7. from lib.itchat.content import *
  8. class WechatMessage(ChatMessage):
  9. def __init__(self, itchat_msg, is_group=False):
  10. super().__init__(itchat_msg)
  11. self.msg_id = itchat_msg["MsgId"]
  12. self.create_time = itchat_msg["CreateTime"]
  13. self.is_group = is_group
  14. if itchat_msg["Type"] == TEXT:
  15. self.ctype = ContextType.TEXT
  16. self.content = itchat_msg["Text"]
  17. elif itchat_msg["Type"] == VOICE:
  18. self.ctype = ContextType.VOICE
  19. self.content = TmpDir().path() + itchat_msg["FileName"] # content直接存临时目录路径
  20. self._prepare_fn = lambda: itchat_msg.download(self.content)
  21. elif itchat_msg["Type"] == PICTURE and itchat_msg["MsgType"] == 3:
  22. self.ctype = ContextType.IMAGE
  23. self.content = TmpDir().path() + itchat_msg["FileName"] # content直接存临时目录路径
  24. self._prepare_fn = lambda: itchat_msg.download(self.content)
  25. elif itchat_msg["Type"] == NOTE and itchat_msg["MsgType"] == 10000:
  26. if is_group and (
  27. "加入群聊" in itchat_msg["Content"] or "加入了群聊" in itchat_msg["Content"]
  28. ):
  29. self.ctype = ContextType.JOIN_GROUP
  30. self.content = itchat_msg["Content"]
  31. # 这里只能得到nickname, actual_user_id还是机器人的id
  32. if "加入了群聊" in itchat_msg["Content"]:
  33. self.actual_user_nickname = re.findall(
  34. r"\"(.*?)\"", itchat_msg["Content"]
  35. )[-1]
  36. elif "加入群聊" in itchat_msg["Content"]:
  37. self.actual_user_nickname = re.findall(
  38. r"\"(.*?)\"", itchat_msg["Content"]
  39. )[0]
  40. elif "拍了拍我" in itchat_msg["Content"]:
  41. self.ctype = ContextType.PATPAT
  42. self.content = itchat_msg["Content"]
  43. if is_group:
  44. self.actual_user_nickname = re.findall(
  45. r"\"(.*?)\"", itchat_msg["Content"]
  46. )[0]
  47. else:
  48. raise NotImplementedError(
  49. "Unsupported note message: " + itchat_msg["Content"]
  50. )
  51. else:
  52. raise NotImplementedError(
  53. "Unsupported message type: Type:{} MsgType:{}".format(
  54. itchat_msg["Type"], itchat_msg["MsgType"]
  55. )
  56. )
  57. self.from_user_id = itchat_msg["FromUserName"]
  58. self.to_user_id = itchat_msg["ToUserName"]
  59. user_id = itchat.instance.storageClass.userName
  60. nickname = itchat.instance.storageClass.nickName
  61. # 虽然from_user_id和to_user_id用的少,但是为了保持一致性,还是要填充一下
  62. # 以下很繁琐,一句话总结:能填的都填了。
  63. if self.from_user_id == user_id:
  64. self.from_user_nickname = nickname
  65. if self.to_user_id == user_id:
  66. self.to_user_nickname = nickname
  67. try: # 陌生人时候, 'User'字段可能不存在
  68. self.other_user_id = itchat_msg["User"]["UserName"]
  69. self.other_user_nickname = itchat_msg["User"]["NickName"]
  70. if self.other_user_id == self.from_user_id:
  71. self.from_user_nickname = self.other_user_nickname
  72. if self.other_user_id == self.to_user_id:
  73. self.to_user_nickname = self.other_user_nickname
  74. except KeyError as e: # 处理偶尔没有对方信息的情况
  75. logger.warn("[WX]get other_user_id failed: " + str(e))
  76. if self.from_user_id == user_id:
  77. self.other_user_id = self.to_user_id
  78. else:
  79. self.other_user_id = self.from_user_id
  80. if self.is_group:
  81. self.is_at = itchat_msg["IsAt"]
  82. self.actual_user_id = itchat_msg["ActualUserName"]
  83. if self.ctype not in [ContextType.JOIN_GROUP, ContextType.PATPAT]:
  84. self.actual_user_nickname = itchat_msg["ActualNickName"]