# _*_ coding: utf-8 _*_
"""
python_wechat.py by xianhu
主è¦åŒ…括如下功能：
(1) 自动æ醒群红包
(2) 自动监测被撤回消æ¯
(3) 群关键å—æ醒，群被@æ醒
"""
import time
import itchat
import logging
from itchat.content import *
# åˆå§‹åŒ–
my = itchat.new_instance()
my.auto_login(hotReload=False, enableCmdQR=2)
# my还包括的以下属性，注æ„用点.查看：
# (1) alive 是å¦è¿˜æ´»ç€ï¼ŒisLogging 是å¦å·²ç™»é™†
# (2) loginInfo 登陆信æ¯ï¼Œå…¶ä¸çš„User属性为自己的信æ¯Userå—典类，包括UserName, NickName, RemarkName, Sex(1 or 2), Signature, Province, Cityç‰
# (3) memberList 通讯录列表，æ¯ä¸€é¡¹ä¸ºä¸€ä¸ªUserå—典类，包括UserName, NickName, RemarkName, Sex(1 or 2), Signature, Province, Cityç‰
# (4) chatroomList 群èŠåˆ—表，æ¯ä¸€é¡¹ä¸ºä¸€ä¸ªChatroomå—典类，包括UserName, NickName, RemarkName, MemberCount, MemberList, Selfç‰
# (5) mpList 订阅å·åˆ—表，æ¯ä¸€é¡¹ä¸ºä¸€ä¸ªMassivePlatformå—典类，包括UserName, NickNameç‰
my.global_keys = ["创业", "人工智能", "ä¼ä¸šæœåŠ¡"]
my.to_user_name = "filehelper" # 消æ¯æŽ¥å—者
my.update_time = time.time() # ä¿¡æ¯æ›´æ–°æ—¶é—´
my.msg_store = {} # 消æ¯å˜å‚¨é˜Ÿåˆ—
my.friends = {} # 好å‹å—典列表
my.groups = {} # 群èŠå—典列表
def update_my_infos():
"""
æ›´æ–°ä¿¡æ¯
"""
# 获å–并更新通讯录: {UserName: UserInstance}
my.friends = {user["UserName"]: user for user in my.get_friends(update=True)}
# 获å–并更新群列表: {UserName: UserInstance}
my.groups = {group["UserName"]: group for group in my.get_chatrooms(update=True)}
return
update_my_infos()
class Message(object):
"""
消æ¯ç±»
"""
def __init__(self, msg):
"""
æž„é€ å‡½æ•°ï¼šæå–消æ¯å†…容
消æ¯æ¥æºåˆ†ç±»ï¼š
（1）æ¥è‡ªå¥½å‹çš„消æ¯
（2）æ¥è‡ªç¾¤çš„消æ¯
æå–消æ¯å†…容，消æ¯ç±»åž‹åˆ†ç±»ï¼š
（1）文å—（2）图片（3）è¯éŸ³ï¼ˆ4）视频（5）地å€ï¼ˆ6）å片（7）æ醒（8）分享（9）附件
"""
# æ›´æ–°ä¿¡æ¯ï¼Œå分钟更新一次
# logging.warning("message: %s", msg)
if time.time() - my.update_time > 600:
update_my_infos()
my.update_time = time.time()
self.msg_id = msg["MsgId"] # 消æ¯ID
self.from_user_name = msg["FromUserName"] # 消æ¯å‘é€è€…ID，如果为群消æ¯ï¼Œåˆ™ä¸ºç¾¤ID
self.msg_type = msg["MsgType"] # 消æ¯ç±»åž‹ï¼Œè¿™é‡Œå‚考下边的we_type
self.msg_content = msg["Content"] # 消æ¯å†…容，这里å‚考下边的we_text
self.msg_time = msg["CreateTime"] # 消æ¯å‘é€æ—¶é—´ï¼Œæ—¶é—´æˆ³æ ¼å¼
self.msg_file = msg["FileName"] # 消æ¯ä¸æ‰€å¸¦æ–‡ä»¶çš„å称
self.msg_file_length = msg["FileSize"] # 消æ¯ä¸æ‰€å¸¦æ–‡ä»¶çš„大å°ï¼Œå—符串类型
self.msg_voice_length = msg["VoiceLength"] # 消æ¯ä¸æ‰€å¸¦è¯éŸ³çš„长度（毫秒）
self.msg_play_length = msg["PlayLength"] # 消æ¯ä¸æ‰€å¸¦è§†é¢‘的长度（秒）
self.msg_url = msg["Url"] # 消æ¯ä¸æ‰€å¸¦é“¾æŽ¥çš„地å€
self.user_user_name = msg["User"].get("UserName", "") # 消æ¯å‘é€è€…ID，如果为群消æ¯ï¼Œåˆ™ä¸ºç¾¤ID
self.user_nick_name = msg["User"].get("NickName", "") # 消æ¯å‘é€è€…昵称，如果为群消æ¯ï¼Œåˆ™ä¸ºç¾¤å
self.user_remark_name = msg["User"].get("RemarkName", "") # 消æ¯å‘é€è€…备注å称，如果为群消æ¯ï¼Œåˆ™ä¸ºç¾¤å¤‡æ³¨å称
self.wind_name = self.user_remark_name if self.user_remark_name else (
self.user_nick_name if self.user_nick_name else (
my.friends[self.user_user_name]["NickName"] if self.user_user_name in my.friends else (
my.groups[self.user_user_name]["NickName"] if self.user_user_name in my.groups else "未知窗å£"
)
)
)
self.actual_user_name = msg.get("ActualUserName", "") # 群消æ¯ä¸ï¼Œæ¶ˆæ¯å‘é€è€…çš„ID
self.actual_nick_name = msg.get("ActualNickName", "") # 群消æ¯ä¸ï¼Œæ¶ˆæ¯å‘é€è€…的群昵称
self.actual_remark_name = self.actual_nick_name \
if (self.actual_user_name not in my.friends) or (not my.friends[self.actual_user_name]["RemarkName"]) \
else my.friends[self.actual_user_name]["RemarkName"]
self.is_at = msg.get("IsAt", None) # 是å¦åœ¨ç¾¤å†…被@
self.we_type = msg["Type"] # 消æ¯ç±»åž‹
self.we_text = msg["Text"] # 消æ¯å†…容
logging.warning("wind_name=%s, send_name=%s, we_type=%s, we_text=%s", self.wind_name, self.actual_remark_name, self.we_type, self.we_text)
return
def process_message_group(msg):
"""
处ç†ç¾¤æ¶ˆæ¯
"""
# ==== 处ç†çº¢åŒ…æ¶ˆæ¯ ====
if msg.we_type == "Note" and msg.we_text.find("收到红包，请在手机上查看") >= 0:
my.send("ã€%s】ä¸æœ‰äººå‘红包啦，快抢ï¼" % msg.wind_name, toUserName=my.to_user_name)
# ==== 处ç†å…³é”®è¯æ¶ˆæ¯ ====
for key in my.global_keys:
if msg.we_type == "Text" and msg.we_text.find(key) >= 0:
my.send("ã€%s】ä¸ã€%s】æåŠäº†å…³é”®å—：%s" % (msg.wind_name, msg.actual_remark_name, key), toUserName=my.to_user_name)
my.send(msg.we_text, toUserName=my.to_user_name)
break
# ==== 群内是å¦è¢«@ ====
if msg.we_type == "Text" and msg.is_at:
my.send("ã€%s】ä¸ã€%s】@äº†ä½ " % (msg.wind_name, msg.actual_remark_name), toUserName=my.to_user_name)
my.send(msg.we_text, toUserName=my.to_user_name)
return
def process_message_revoke(msg):
"""
处ç†æ’¤å›žæ¶ˆæ¯
"""
# 消æ¯å˜å‚¨ï¼Œåˆ 除过期消æ¯
my.msg_store[msg.msg_id] = msg
for _id in [_id for _id in my.msg_store if time.time() - my.msg_store[_id].msg_time > 120]:
my.msg_store.pop(_id)
# ä¿å˜æ¶ˆæ¯ä¸çš„内容（图片ã€è¯éŸ³ç‰ï¼‰
if msg.we_type in ["Picture", "Recording"]:
try:
msg.we_text(".Cache/" + msg.msg_file)
logging.warning("process_message_revoke: download %s to .Cache/", msg.msg_file)
except Exception as excep:
logging.error("process_message_revoke: download %s to .Cache/ error: %s", msg.msg_file, excep)
# ==== 撤回消æ¯å¤„ç†ï¼ˆå¿…须为最åŽä¸€æ¥ï¼‰ ====
if msg.we_type == "Note" and msg.we_text.find("撤回了一æ¡æ¶ˆæ¯") >= 0:
old_msg = my.msg_store.get(msg.msg_content[msg.msg_content.find("