home/youxiang
2
0
Fork 0
mirror of https://github.com/wahyd4/youxiang.git synced 2026-08-08 21:06:21 +10:00
Code Issues Packages Projects Releases Wiki Activity

first commit

Browse Source
This commit is contained in:
Snow Wang
2020-06-05 11:33:48 +08:00
committed by GitHub
parent 9c64116fef
commit 61095f9401
17 changed files with 1462 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Dockerfile
+12
View File
@@ -0,0 +1,12 @@
# 说明该镜像以哪个镜像为基础
FROM python:3.7-slim
RUN mkdir /youxiang
WORKDIR /youxiang
COPY requirements.txt requirements.txt
RUN pip install -r requirements.txt
COPY . /youxiang
ENTRYPOINT ["python", "/youxiang/main.py"]
__init__.py
+3
View File
@@ -0,0 +1,3 @@
# -*- coding: utf-8 -*-
__author__ = 'why2lyj'
__email__ = 'admin@farseer.vip'
_config.yaml
+95
View File
@@ -0,0 +1,95 @@
# 配置文件请严格遵循 yaml 语法格式，yaml 学习地址:
# https://ansible-tran.readthedocs.io/en/latest/docs/YAMLSyntax.html
# http://einverne.github.io/post/2015/08/yaml.html
wechat_uuid: ''
#---------------------------------------自动回复功能设置--------------------------------------start
auto_reply_info:
# 是否开启自动回复
is_auto_reply: False
# 是否自动回复所有好友(慎重启动！),开启后，仅仅有黑名单的成员才不会自动回复
is_auto_reply_all: True
# 智能闲聊（腾讯）https://ai.qq.com/product/nlpchat.shtml。免费且无限量
qqnlpchat_conf:
app_id: ''
app_key: ''
# 自动回复消息前缀（不需要则设置为空）
auto_reply_prefix: ''
# 自动回复消息后缀（不需要则设置为空）
auto_reply_suffix: ''
# 是否允许通过关键字自动加好友
is_auto_add_friend: True
# 加好友时通过的关键字
auto_add_friend_keywords:
- '优惠群'
- '加群'
- '线报'
# 白名单，is_auto_reply_all: False 生效。此名单的用户才可自动回复。（填：文件传输助手，可回复自己，测试效果）
auto_reply_white_list:
- '文件传输助手'
# 黑名单，is_auto_reply_all:True 生效。此名单不会自动回复。
auto_reply_black_list:
- 'Grace.李'
#---------------------------------------自动回复功能设置--------------------------------------end
##---------------------------群聊助手设置--------------------------start
group_helper_conf:
is_open: True # 开启群助手
is_all: False # 是否对所有群开启。当开启时，只有黑名单的名单才不受影响（慎重开启！）
# 白名单用户。当 is_all：False。只处理这个群里的消息
group_name_white_list:
- '淘宝天猫内部优惠群-综合类①'
- '淘宝内部优惠群-潮流范①'
- '淘宝内部优惠群-零食类①'
- '淘宝内部优惠群-女装类①'
- '口碑KFC必胜客麦当劳优惠券'
# 黑名单用户。当 is_all ：True 。这个群里的用户不受影响。
group_name_black_list:
- ''
is_at: True # 艾特标记。只有当别人艾特机器人，才会处理消息（慎重关闭！）
# 是否重复加入
enter_mult_group: False
#--------------------------- 群聊助手设置 --------------------------end
#--------------------------- 推广客设置 --------------------------start
taobao:
is_open: True
app_key: ''
app_secret: ''
adzone_id: ''
chat_groups:
- {group_name: '淘宝天猫内部优惠群-综合类①', group_material_id: '3756,28026,27446,13366,3786', minute: '15,45', hour: '7-23'}
- {group_name: '淘宝内部优惠群-潮流范①', group_material_id: '4093', minute: '20', hour: '7-23'}
- {group_name: '口碑KFC必胜客麦当劳优惠券', group_material_id: '19810', minute: '30', hour: '7-23'}
- {group_name: '淘宝内部优惠群-女装类①', group_material_id: '3767,27448,13367,3788', minute: '10,40', hour: '7-23'}
- {group_name: '淘宝内部优惠群-零食类①', group_material_id: '13375,3761,27451,3791', minute: '10,40', hour: '7-23'}
jingdong:
is_open: False
app_key: ''
app_secret: ''
site_id: ''
suo_im: ''
chat_groups:
- {group_name: '京东内部优惠群-9.9专区①', group_material_id: '21', minute: '10,40', hour: '7-23'}
#--------------------------- 推广客设置 --------------------------end
chat/itchatHelper.py
+319
View File
@@ -0,0 +1,319 @@
# -*- coding: utf-8 -*-
"""
"""
import itchat
import re
from datetime import datetime
from datetime import timedelta
from importlib import import_module
from untils import config
from untils.common import (
md5_encode,
FILEHELPER_MARK,
FILEHELPER,
)
__all__ = ['init_wechat_config', 'set_system_notice', 'get_group', 'get_friend']
TIME_COMPILE = re.compile(r'^\s*([01]?[0-9]|2[0-3])\s*[：:\-]\s*([0-5]?[0-9])\s*$')
def init_wechat_config():
""" 初始化微信所需数据 """
# print('初始化微信所需数据开始..')
# 从config copy ，用于保存新的接口内容。
myset = config.copy()
print('=' * 80)
base_wechat_info = itchat.search_friends() # 获取此微信号的基础信息
wechat_nick_name = base_wechat_info['NickName'] # 获取此微信号的昵称
wechat_uuid = base_wechat_info['UserName'] # 获取此微信号的uuid
myset['wechat_nick_name'] = wechat_nick_name
myset['wechat_uuid'] = wechat_uuid
# start---------------------------处理自动回复好友---------------------------start
reply = myset.get('auto_reply_info')
if reply is not None and reply.get('is_auto_reply'):
if reply.get('is_auto_reply_all'):
auto_reply_list_key = 'auto_reply_black_list'
auto_reply_list_uuid_name = 'auto_reply_black_uuids'
else:
auto_reply_list_key = 'auto_reply_white_list'
auto_reply_list_uuid_name = 'auto_reply_white_uuids'
auto_reply_uuids_list = []
for name in reply.get(auto_reply_list_key):
if not name.strip():
continue
if name.lower() in FILEHELPER_MARK: # 判断是否文件传输助手
auto_reply_uuids_list.append(FILEHELPER)
continue
friend = get_friend(name)
if friend:
auto_reply_uuids_list.append(friend['UserName'])
else:
print('自动回复中的好友昵称『{}』有误。'.format(name))
reply[auto_reply_list_uuid_name] = set(auto_reply_uuids_list)
# print('已开启图灵自动回复...')
# end---------------------------处理自动回复好友---------------------------end
# start ----------------------------------- 群功能初始化 ----------------------------------- start
helper = myset.get('group_helper_conf')
if helper is not None and helper.get('is_open'):
if helper.get('is_all', False):
group_list_key = 'group_name_black_list'
group_list_uuid_name = 'group_black_uuids'
else:
group_list_key = 'group_name_white_list'
group_list_uuid_name = 'group_white_uuids'
group_uuid_list = []
for name in helper.get(group_list_key):
if not name.strip():
continue
group = get_group(name)
if group:
group_uuid_list.append(group['UserName'])
else:
print('群助手中的群聊名称『{}』有误。'
'(注意：必须要把需要的群聊保存到通讯录)'.format(name))
helper[group_list_uuid_name] = set(group_uuid_list)
# end ----------------------------------- 群功能初始化 ----------------------------------- end
alarm = myset.get('alarm_info')
alarm_dict = {}
if alarm is not None and alarm.get('is_alarm'):
for gi in alarm.get('girlfriend_infos'):
ats = gi.get('alarm_timed')
if not ats:
continue
uuid_list = []
nickname_list = []
# start---------------------------处理好友---------------------------start
friends = gi.get('wechat_name')
if isinstance(friends, str):
friends = [friends]
if isinstance(friends, list):
for name in friends:
if name.lower() in FILEHELPER_MARK: # 判断是否文件传输助手
uuid_list.append(FILEHELPER)
nickname_list.append(name)
continue
name_info = get_friend(name)
if not name_info:
print('定时提醒中的好友昵称『{}』无效'.format(name))
else:
uuid_list.append(name_info['UserName'])
nickname_list.append(name)
# end---------------------------处理好友---------------------------end
# start---------------------------群组处理---------------------------start
group_names = gi.get('group_name')
if isinstance(group_names, str):
group_names = [group_names]
if isinstance(group_names, list):
for name in group_names:
name_info = get_group(name)
if not name_info:
print('定时任务中的群聊名称『{}』有误。'
'(注意：必须要把需要的群聊保存到通讯录)'.format(name))
else:
uuid_list.append(name_info['UserName'])
nickname_list.append(name)
# end---------------------------群组处理---------------------------end
# start---------------------------定时处理---------------------------start
if isinstance(ats, str):
ats = [ats]
if isinstance(ats, list):
for at in ats:
times = TIME_COMPILE.findall(at)
if not times:
print('时间{}格式出错'.format(at))
continue
hour, minute = int(times[0][0]), int(times[0][1])
temp_dict = {'hour': hour, 'minute': minute, 'uuid_list': uuid_list, 'nickname_list': nickname_list}
temp_dict.update(gi)
alarm_dict[md5_encode(str(temp_dict))] = temp_dict
# end---------------------------定时处理---------------------------end
alarm['alarm_dict'] = alarm_dict
# 将解析的数据保存于 config 中。
config.update(myset)
# print(json.dumps(alarm_dict, ensure_ascii=False))
# print('初始化微信所需数据结束..')
# log_all_config()
def set_system_notice(text):
"""
给文件传输助手发送系统日志。
:param text:str 日志内容
"""
if text:
text = '系统通知：' + text
itchat.send(text, toUserName=FILEHELPER)
def get_group(group_name, update=False):
"""
根据群组名获取群组数据
:param group_name:str, 群组名
:param update: bool 强制更新群组数据
:return: obj 单个群组信息
"""
if update: itchat.get_chatrooms(update=True)
if not group_name: return None
groups = itchat.search_chatrooms(name=group_name)
if not groups: return None
return groups[0]
def get_friend(wechat_name, update=False):
"""
根据用户名获取用户数据
:param wechat_name: str 用户名
:param update: bool 强制更新用户数据
:return: obj 单个好友信息
"""
if update: itchat.get_friends(update=True)
if not wechat_name: return None
friends = itchat.search_friends(name=wechat_name)
if not friends: return None
return friends[0]
def get_mps(mp_name, update=False):
"""
根据公众号的名称获取用户数据
:param mp_name: str 用户名
:param update: bool 强制更新用户数据
:return: obj 单个公众号信息
"""
if update: itchat.get_mps(update=True)
if not mp_name: return None
mps = itchat.search_mps(name=mp_name)
if not mps: return None
# mpuuid = mps[0]['UserName'] 公众号的uuid
return mps[0]
# import pysnooper
# @pysnooper.snoop()
def log_all_config():
"""
用于打印设置日志
:return:
"""
print('=' * 80)
channel = config.get('auto_reply_info').get('bot_channel', 7)
source = 'ownthink_robot'
addon = import_module('everyday_wechat.control.bot.' + source, __package__)
bot_name = addon.BOT_NAME
print('自动回复机器人渠道：{}'.format(bot_name))
# start ----------------------------------- 微信好友自动回复的功能日志 ----------------------------------- start
reply = config.get('auto_reply_info', None)
if not reply or not reply.get('is_auto_reply'):
print('未开启微信好友自动回复。')
else:
if reply.get('is_auto_reply_all'):
auto_uuids = reply.get('auto_reply_black_uuids')
nicknames = []
for auid in auto_uuids:
if auid == 'filehelper':
nicknames.append(auid)
else:
friends = itchat.search_friends(userName=auid)
nickname = friends.nickName
nicknames.append(nickname)
nns = '，'.join(nicknames)
print('开启对全部微信好友全部回复，除了：{}'.format(nns))
else:
auto_uuids = reply.get('auto_reply_white_uuids')
nicknames = []
for auid in auto_uuids:
if auid == 'filehelper':
nicknames.append(auid)
else:
friends = itchat.search_friends(userName=auid)
nickname = friends.nickName
nicknames.append(nickname)
nns = '，'.join(nicknames)
print('对微信好友 {}，进行自动回复'.format(nns))
print('=' * 80)
# start ----------------------------------- 群功能日志说明 ----------------------------------- start
helper = config.get('group_helper_conf')
if not helper or not helper.get('is_open'):
print('未开启群助手功能。')
else:
if helper.get('is_all'):
auto_uuids = helper.get('group_black_uuids')
nicknames = []
for auid in auto_uuids:
chatrooms = itchat.search_chatrooms(userName=auid)
nickname = chatrooms['NickName'] # 群聊名称
nicknames.append(nickname)
nns = '，'.join(nicknames)
print('已开启对全部微信群的监听，除了群：{}。'.format(nns))
else:
auto_uuids = helper.get('group_white_uuids')
nicknames = []
for auid in auto_uuids:
chatroom = itchat.search_chatrooms(userName=auid)
nickname = chatroom['NickName'] # 群聊名称
nicknames.append(nickname)
nns = '，'.join(nicknames)
print('已对微信群：{}，开启了群助手功能。'.format(nns))
if helper.get('is_at'):
print('只有群里用户@机器人，才会触发群助手功能。')
if helper.get('is_auto_reply'):
print('已开启对微信群内用户的自动回复。')
if helper.get('is_weather'):
print('已开启天气查询功能，具体使用方法请输入：“help” 查看。')
if helper.get('is_calendar'):
print('已开启日志查询功能，具体使用方法请输入：“help” 查看。')
if helper.get('is_rubbish'):
print('已开启垃圾分类查询功能，具体使用方法请输入：“help” 查看。')
if helper.get('is_moviebox'):
print('已开启票房查询功能，具体使用方法请输入：“help” 查看。')
if helper.get('is_air_quality'):
print('已开启空气质量查询功能，具体使用方法请输入：“help” 查看。')
print('=' * 80)
# start ----------------------------------- 提醒功能的日志说明 ----------------------------------- start
alarm = config.get('alarm_info')
if not alarm or not alarm.get('is_alarm'):
print('未开启每日提醒功能。')
else:
print('已开启定时发送提醒功能。')
alarm_dict = alarm.get('alarm_dict')
for value in alarm_dict.values():
nickname_list = value.get('nickname_list')
nns = '，'.join(nickname_list)
# temp_dict = {'hour': hour, 'minute': minute, 'uuid_list': uuid_list, 'nickname_list': nickname_list}
hour = value.get('hour')
minute = value.get('minute')
alarm_time = "{hour:0>2d}:{minute:0>2d}".format(hour=hour, minute=minute)
# 计算在哪个区间给朋友发送信息
jitter = value.get("alarm_jitter", 0)
if jitter != 0:
set_time = datetime.strptime(alarm_time, '%H:%M')
jitter_time = timedelta(seconds=jitter)
start_time = (set_time - jitter_time).strftime("%H:%M")
end_time = (set_time + jitter_time).strftime("%H:%M")
alarm_time = "{start_time}——{end_time} 期间".format(start_time=start_time, end_time=end_time)
print('定时：{alarm_time}，给：{nicknames}，发送提醒内容一次。'.format(alarm_time=alarm_time, nicknames=nns))
print('=' * 80)
chat/message.py
+129
View File
@@ -0,0 +1,129 @@
# -*- coding: utf-8 -*-
"""
"""
import re
import time
import random
import itchat
from untils import config
from chat.itchatHelper import set_system_notice
from untils.common import FILEHELPER
from untils.qq_nlpchat import get_auto_reply
__all__ = ['handle_friends_message', 'handle_groups_message']
at_compile = r'(@.*?\s{1,}).*?'
help_complie = r'^(?:0|帮忙|帮助|help)\s*$'
common_msg = '@{ated_name}\u2005\n{text} \n ----机器人智能回复 \n 您如有事儿加机器人问话吧'
help_group_content = """@{ated_name}
群助手功能：
有啥事儿问管理员吧，群功能不开放
"""
def handle_friends_message(msg):
""" 处理好友信息 """
try:
# 自己通过手机微信发送给别人的消息(文件传输助手除外)不作处理。
if msg['FromUserName'] == config.get('wechat_uuid') and msg['ToUserName'] != FILEHELPER:
return
conf = config.get('auto_reply_info')
if not conf.get('is_auto_reply'):
return
# 获取发送者的用户id
uuid = FILEHELPER if msg['ToUserName'] == FILEHELPER else msg['FromUserName']
is_all = conf.get('is_auto_reply_all')
auto_uuids = conf.get('auto_reply_black_uuids') if is_all else conf.get('auto_reply_white_uuids')
# 开启回复所有人，当用户是黑名单，不回复消息
if is_all and uuid in auto_uuids:
return
# 关闭回复所有人，当用户不是白名单，不回复消息
if not is_all and uuid not in auto_uuids:
return
receive_text = msg.text # 好友发送来的消息内容
# 好友叫啥，用于打印
nick_name = FILEHELPER if uuid == FILEHELPER else msg.user.nickName
reply_text = get_auto_reply(receive_text, uuid) # 获取自动回复
if reply_text: # 如内容不为空，回复消息
time.sleep(random.randint(1, 2)) # 休眠一秒，保安全。想更快的，可以直接注释。
prefix = conf.get('auto_reply_prefix', '') # 前缀
if prefix:
reply_text = '{}{}'.format(prefix, reply_text)
suffix = conf.get('auto_reply_suffix', '') # 后缀
if suffix:
reply_text = '{}{}'.format(reply_text, suffix)
itchat.send(reply_text, toUserName=uuid)
else:
set_system_notice(f'''自动回复失败：\n『{nick_name}』发来信息：{receive_text} \n''')
except Exception as exception:
print(str(exception))
def handle_groups_message(msg):
"""
处理群消息，
:param msg:
:return:
"""
uuid = msg.fromUserName # 群 uid
ated_uuid = msg.actualUserName # 艾特你的用户的uuid
ated_name = msg.actualNickName # 艾特你的人的群里的名称
text = msg['Text'] # 发送到群里的消息。
conf = config.get('group_helper_conf')
if not conf.get('is_open'):
return
# 自己通过手机端微信发出的消息不作处理
if ated_uuid == config.get('wechat_uuid'):
return
# 如果开启了 『艾特才回复』，而群用户又没有艾特你。走垃圾分类
if conf.get('is_at') and not msg.isAt:
return
is_all = conf.get('is_all', False)
group_uuids = conf.get('group_name_black_list') if is_all else conf.get('group_name_white_list')
# 开启回复所有群，而群组是黑名单，不处理消息
if is_all:
for group_name in group_uuids:
chatrooms = itchat.search_chatrooms(name=f'''{group_name}''')
# print(chatrooms)
for room in chatrooms:
if uuid == room['UserName']:
return
# 未开启回复所有群，而群组不是白名单，不处理消息
in_white_flag = False
if not is_all:
for group_name in group_uuids:
chatrooms = itchat.search_chatrooms(name=f'''{group_name}''')
for room in chatrooms:
if uuid == room['UserName']:
in_white_flag = True
if not in_white_flag:
return
# 去掉 at 标记
text = re.sub(at_compile, '', text)
reply_text = get_auto_reply(text, ated_uuid) # 获取自动回复
if reply_text: # 如内容不为空，回复消息
reply_text = common_msg.format(ated_name=ated_name, text=reply_text)
itchat.send(reply_text, uuid)
# print('回复{}：{}'.format(ated_name, reply_text))
else:
print('自动回复失败\n')
chat/wechat.py
+249
View File
@@ -0,0 +1,249 @@
# coding=utf-8
"""
"""
import time
import platform
import os
import re
import random
import itchat
from itchat.content import (
TEXT,
FRIENDS,
NOTE,
PICTURE
)
from collections import OrderedDict
from untils import config
from untils.scheduler import job_tasks
from chat.itchatHelper import init_wechat_config, set_system_notice
from chat.message import handle_friends_message, handle_groups_message
__all__ = ['run', 'delete_cache']
group_infos_dict = OrderedDict() # 群信息字典
def run():
""" 主运行入口 """
conf = config.init()
# conf = get_yaml()
if not conf: # 如果 conf，表示配置文件出错。
print('程序中止...')
return
# 判断是否登录，如果没有登录则自动登录，返回 False 表示登录失败
print('开始登录...')
if not is_online(auto_login=True):
print('程序已退出...')
return
def is_online(auto_login=False):
"""
判断是否还在线。
:param auto_login: bool,当为 Ture 则自动重连(默认为 False)。
:return: bool,当返回为 True 时，在线；False 已断开连接。
"""
def _online():
"""
通过获取好友信息，判断用户是否还在线。
:return: bool,当返回为 True 时，在线；False 已断开连接。
"""
try:
if itchat.search_friends():
return True
except IndexError:
return False
return True
if _online(): return True # 如果在线，则直接返回 True
if not auto_login: # 不自动登录，则直接返回 False
print('微信已离线..')
return False
hotReload = False #
loginCallback = init_data
exitCallback = exit_msg
try:
for _ in range(2): # 尝试登录 2 次。
if platform.system() in ('Windows', 'Darwin'):
itchat.auto_login(hotReload=hotReload,
loginCallback=loginCallback, exitCallback=exitCallback)
itchat.run(blockThread=True)
else:
# 命令行显示登录二维码。
itchat.auto_login(enableCmdQR=2, hotReload=hotReload, loginCallback=loginCallback,
exitCallback=exitCallback)
itchat.run(blockThread=True)
if _online():
print('登录成功')
return True
except Exception as exception: # 登录失败的错误处理。
sex = str(exception)
if sex == "'User'":
print('此微信号不能登录网页版微信，不能运行此项目。没有任何其它解决办法！可以换个号再试试。')
else:
print(sex)
delete_cache() # 清理缓存数据
print('登录失败。')
return False
def delete_cache():
""" 清除缓存数据，避免下次切换账号时出现 """
file_names = ('QR.png', 'itchat.pkl')
for file_name in file_names:
if os.path.exists(file_name):
os.remove(file_name)
def init_data():
""" 初始化微信所需数据 """
set_system_notice('登录成功')
itchat.get_friends(update=True) # 更新好友数据。
itchat.get_chatrooms(update=True) # 更新群聊数据。
conf = config.get('group_helper_conf')
group_name_list = conf.get('group_name_white_list')
init_chatsroom(group_name_list)
init_wechat_config() # 初始化所有配置内容
init_alarm()
print('初始化完成，开始正常工作。')
def init_chatsroom(group_name_list):
uidlist_compile = re.compile(
r"(?<!'Self': )\<ChatroomMember:.*?'UserName': '(.*?)', 'NickName'.*?") # 筛选出群所有用户的 uid
for group_name in group_name_list:
group_list = itchat.search_chatrooms(name=group_name) # 通过群聊名获取群聊信息
group_info = {}
if group_list:
group_uuid = group_list[0]['UserName']
group = itchat.update_chatroom(group_uuid, detailedMember=True) # 通过群id更新群名单
group_uuid = group['UserName'] # 群聊 id
group_info['group_name'] = group_name # 群聊名称
group_info['group_uuid'] = group_uuid # 群聊 uuid
count = len(group['MemberList']) # 群聊人数
group_info['count'] = count
member_uid_list = uidlist_compile.findall(str(group)) # 根据正则取出群组里所有用户的 uuid。也可以用循环的方式。
if member_uid_list:
group_info['member_uid_list'] = member_uid_list
group_infos_dict[group_uuid] = group_info
def init_alarm():
"""
初始化定时任务
:param alarm_dict: 定时相关内容
"""
# 定时任务
job_tasks()
@itchat.msg_register([TEXT])
def text_reply(msg):
""" 监听用户消息，用于自动回复 """
handle_friends_message(msg)
# @itchat.msg_register([PICTURE], isGroupChat=True)
# def picture_group_reply(msg):
# """ 监听群消息发来的图片消息 """
# pass
# # raise NotImplementedError
@itchat.msg_register([TEXT], isGroupChat=True)
def text_group(msg):
""" 监听群消息，用于自动回复 """
handle_groups_message(msg)
@itchat.msg_register(FRIENDS)
def add_friends_msg(msg):
""" 监听添加好友请求 为了自动同意好友请求
"""
conf = config.get('auto_reply_info')
IS_AUTO_ADD_FRIEND = conf.get('is_auto_add_friend')
add_friend_keys = ''.join(conf.get('auto_add_friend_keywords'))
note_first_meet_text = '''等你等的好辛苦，很高心您的加入！
''' # 好友成功后的第一句话
add_friend_compile = re.compile('|'.join(i.strip() for i in
re.split(r'[,，]+', add_friend_keys) if i), re.I) # 通过关键词同意加好友请求
itchat.get_friends(update=True) # 更新好友数据。
if not IS_AUTO_ADD_FRIEND: # 如果是已关闭添加好友功能，则直接返回
return
userid = msg['RecommendInfo']['UserName']
nickname = msg['RecommendInfo']['NickName']
content = msg['RecommendInfo']['Content'] # 获取验证消息
if add_friend_compile.findall(content):
time.sleep(random.randint(1, 2)) # 随机休眠（1~3）秒，用于防检测机器人
itchat.add_friend(**msg['Text']) # 同意加好友请求
time.sleep(random.randint(1, 2))
itchat.send(note_first_meet_text, userid) # 给刚交的朋友发送欢迎语句
note = '已添加好友：{}'.format(nickname)
set_system_notice(note)
else:
note = '添加好友失败：用户「{}」 发来的验证消息「{}」。'.format(nickname, content)
set_system_notice(note)
@itchat.msg_register([NOTE], isGroupChat=True)
def group_note_msg(msg):
""" 群通知处理
邀请进来的和扫二维码进来的进行提示
"""
invite_compile = re.compile(r'邀请"(.*?)"加入了群聊\s*$')
scan_invite_compile = re.compile(r'"(.*?)"通过扫描"(.*?)"分享的二维码加入群聊\s*$')
scan_my_invite_compile = re.compile(r'"(.*?)"通过扫描你分享的二维码加入群聊\s*$')
group_uuid = msg['FromUserName'] # 获取当前群的 uuid
if group_uuid in group_infos_dict: # 判断是否为同一个群组
text = msg['Text'] # 通知的内容
invite_names = invite_compile.findall(text) or scan_my_invite_compile.findall(text) # 判断是否是加入了新用户
scan_invite = scan_invite_compile.findall(text) ## 判断是否是群内人员邀请而入
if invite_names: # 用于邀请
invite_name = invite_names[0] # 加入者的昵称
time.sleep(random.randint(1, 2))
note_invite_welcome = f'''@{invite_name}\u2005 \n欢迎加入群，请查看群规
① 此群禁止发广告，聊天请随意
② 请勿发涉黄，涉政，辱骂性话语
③ 允许拉小伙伴进来，记得加管理员
请主动维护群内秩序，若有违反，直送飞机票。
如需要帮助请主动联系管理员。
'''
itchat.send(note_invite_welcome, group_uuid) # 向群里发送欢迎语句
if scan_invite:
scan_invite_name = scan_invite[0][0] # 加入者的昵称
scan_inviter = scan_invite[0][1] # 邀请者的昵称
time.sleep(random.randint(1, 2))
note_invite_welcome = f'''@{scan_invite_name}\u2005 \n欢迎加入群，请查看群规
① 此群禁止发广告，聊天请随意
② 请勿发涉黄，涉政，辱骂性话语
③ 允许拉小伙伴进来，记得加管理员
@{scan_inviter}\u2005
请主动约束受邀好友：{scan_invite_name}，若其违反上述约定，送连坐飞机票。
如需要帮助请主动联系管理员。
'''
itchat.send(note_invite_welcome, group_uuid) # 向群里发送欢迎语句
return
def exit_msg():
""" 退出通知 """
print('程序已退出')
if __name__ == '__main__':
run()
coupon/jd.py
+54
View File
@@ -0,0 +1,54 @@
import pprint
import json
from untils.jd_api import JdApiClient
def jingfen_query():
'''
https://union.jd.com/openplatform/api/10421
:return:
'''
client = JdApiClient("", "")
resp = client.call("jd.union.open.goods.jingfen.query",
{"goodsReq":
{"sort": "desc",
"pageSize": "2",
"pageIndex": "1",
"eliteId": "31",
"sortName": "inOrderCount30DaysSku"
}})
# pprint.pprint(json.loads(resp.json()['jd_union_open_goods_jingfen_query_response']['result']))
for data in json.loads(resp.json()['jd_union_open_goods_jingfen_query_response']['result'])['data']:
skuName = data['skuName'] ## 商品全名
skuId = data['skuId'] ## 商品 sku
materialUrl = f'''http://{(data['materialUrl'])}''' ## 商品url
couponInfos = data['couponInfo'] ## 优惠券列表
# 查找最优优惠券
for couponInfo in couponInfos['couponList']:
if int(couponInfo['isBest']) == 1:
discount = couponInfo['discount'] ## 优惠券额度
coupon_link = couponInfo['link'] ## 优惠券领取地址
price = data['priceInfo']['price'] ## 商品价格
print(skuName)
print(skuId)
print(discount)
print(coupon_link)
print(price)
print(materialUrl)
## 获取 images
for image in data['imageInfo']['imageList']:
image_url = image['url']
print(image_url)
print('-----------------------------------')
# def share_text():
def tb_share_text():
raise NotImplementedError
if __name__ == '__main__':
jingfen_query()
coupon/tb.py
+60
View File
@@ -0,0 +1,60 @@
import time
import json
import itchat
import random
from untils.common import save_pic, del_pic
from untils.tb_top_api import TbApiClient
def tb_share_text(group_name: str, material_id: str, app_key, app_secret, adzone_id):
'''
:param group_name:
:param material_id:
:return:
'''
material_id = str(random.choices(material_id.split(','))[0])
print(material_id)
groups = itchat.search_chatrooms(name=f'''{group_name}''')
for room in groups:
group_name = room['UserName']
time.sleep(random.randint(1, 5))
tb_client = TbApiClient(app_key=app_key, secret_key=app_secret, adzone_id=adzone_id)
res = tb_client.taobao_tbk_dg_optimus_material(material_id)
json_data = json.loads(res)['tbk_dg_optimus_material_response']['result_list']['map_data']
count = 0
for item in json_data:
count += 1
if str(item).find("coupon_share_url") > -1:
coupon_share_url = "https:" + item['coupon_share_url']
coupon_amount = item['coupon_amount']
pict_url = "https:" + str(item['pict_url'])
title = item['title']
item_id = item['item_id']
filename = save_pic(pict_url, item_id)
zk_final_price = item['zk_final_price']
# 发送图片
itchat.send('@img@%s' % (f'''{filename}'''), group_name)
itchat.send(f''' {title} \n【在售价】¥{zk_final_price}\n【券后价】¥{round(float(zk_final_price) - float(coupon_amount),
2)}\n-----------------\n復製評论({tb_client.taobao_tbk_tpwd_create(
title, coupon_share_url)})，去【tao寶】下单\n''', group_name)
time.sleep(2)
del_pic(filename)
else:
click_url = "https:" + item['click_url']
title = item['title']
item_id = item['item_id']
pict_url = "https:" + str(item['pict_url'])
zk_final_price = item['zk_final_price']
print(pict_url)
filename = save_pic(pict_url, item_id)
itchat.send('@img@%s' % (f'''{filename}'''), group_name)
itchat.send(
f'''{title} \n【在售价】¥{zk_final_price}\n-----------------\n復製評论({tb_client.taobao_tbk_tpwd_create(
title, click_url)})，去【tao寶】下单\n''', group_name)
time.sleep(2)
del_pic(filename)
if __name__ == '__main__':
print(f'''tb function''')
main.py
+9
View File
@@ -0,0 +1,9 @@
# coding=utf-8
"""
"""
from chat.wechat import *
if __name__ == '__main__':
run()
requirements.txt
+6
View File
@@ -0,0 +1,6 @@
retrying==1.3.3
cn2an==0.4.1
APScheduler==3.6.3
selenium
itchat==1.3.10
pandas==0.24.2
untils/common.py
+80
View File
@@ -0,0 +1,80 @@
# coding=utf-8
"""
工具类
"""
import re
import hashlib
import json
import os
import urllib.request
import datetime
import sys
__all__ = [
'FILEHELPER_MARK', 'FILEHELPER',
'is_json', 'md5_encode', '']
FILEHELPER_MARK = ['文件传输助手', 'filehelper'] # 文件传输助手标识
FILEHELPER = 'filehelper'
def is_json(resp):
"""
判断数据是否能被 Json 化。 True 能，False 否。
:param resp: request.
:return: bool, True 数据可 Json 化；False 不能 JOSN 化。
"""
try:
json.loads(resp.text)
return True
except AttributeError as error:
return False
return False
def md5_encode(text):
""" 把數據 md5 化 """
if not isinstance(text, str):
text = str(text)
md5 = hashlib.md5()
md5.update(text.encode('utf-8'))
encodedStr = md5.hexdigest().upper()
return encodedStr
def _progress(block_num, block_size, total_size):
'''回调函数
@block_num: 已经下载的数据块
@block_size: 数据块的大小
@total_size: 远程文件的大小
'''
sys.stdout.write('\r>> Downloading file %.1f%%' % (
float(block_num * block_size) / float(total_size) * 100.0))
sys.stdout.flush()
def save_pic(img_url, item_id):
'''
:param img_url: 图片url
:param item_id: 物料id，仅限用于图片命名
:return: filename str, 返回一个图片名称，用来定位删除的。
'''
try:
file_suffix = os.path.splitext(img_url)[1]
# print(file_suffix)
# 拼接图片名（包含路径）
filename = f'''tb_{datetime.datetime.now().strftime("%y%m%d-%H%M%S")}_{item_id}{file_suffix}'''
# print(filename)
# 下载图片，并保存到文件夹中
urllib.request.urlretrieve(img_url, filename=filename)
print(f'''图片下载成功：{filename}''')
return filename
except IOError as e:
print(e)
return
except Exception as e:
print(e)
return
def del_pic(filename):
os.remove(filename)
if __name__ == '__main__':
print(md5_encode('aeryou'))
pass
untils/config.py
+56
View File
@@ -0,0 +1,56 @@
# coding=utf-8
"""
用于管理缓存的配置数据
使用前必须先调用 init() 。
"""
import os
import copy as mycopy
import yaml
def init():
"""
将 yaml 里的配置文件导入到 config.py 中
:return: bool ，true 表示数据导入成功。
"""
global opts
opts = get_yaml()
if opts:
return True
return False
def get_yaml():
"""
解析 yaml
:return: s 字典
"""
path = os.path.join(os.path.dirname(os.path.dirname(__file__)), '_config.yaml')
try:
with open(path, 'r', encoding='utf-8') as file:
config = yaml.safe_load(file)
return config
except Exception as exception:
print(str(exception))
print('你的 _config.yaml 文件配置出错...')
return None
def set(key, value):
""" 通过 key 设置某一项值 """
opts[key] = value
def get(key, default=None):
""" 通过 key 获取值 """
return opts.get(key, default)
def copy():
""" 复制配置 """
return mycopy.deepcopy(opts)
def update(new_opts):
""" 全部替换配置 """
opts.update(new_opts)
if __name__ == '__main__':
# init()
# print(copy())
pass
untils/jd_api.py
+41
View File
@@ -0,0 +1,41 @@
import datetime
import requests
import hashlib
import json
JD_API_ROOT = 'https://router.jd.com/api'
class JdApiClient(object):
def __init__(self, app_key, secret_key):
self.app_key = app_key
self.secret_key = secret_key
def get_sign(self, params):
params_list = sorted(list(params.items()), key=lambda x: x[0])
params_bytes = (self.secret_key + ''.join("%s%s" % (k, v) for k, v in params_list) + self.secret_key).encode('utf-8')
sign = hashlib.md5(params_bytes).hexdigest().upper()
return sign
def call(self, method, param_json, **kwargs):
params = {
"v": "1.0",
"method": method,
"app_key": self.app_key,
"timestamp": datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
"format": "json",
"sign_method": "md5"
}
if isinstance(param_json, (dict, list)):
params["param_json"] = json.dumps(param_json)
else:
params["param_json"] = param_json
params['sign'] = self.get_sign(params)
resp = requests.get(JD_API_ROOT, params=params, **kwargs)
return resp
def jd_union_open_goods_jingfen_query(self):
'''
https://union.jd.com/openplatform/api/10421
:return:
'''
return
untils/qq_nlpchat.py
+97
View File
@@ -0,0 +1,97 @@
# -*- coding: utf-8 -*-
"""
Introduction: 智能闲聊（腾讯）
官网：https://ai.qq.com/product/nlpchat.shtml
免费试用，得申请 app_id，app_key。
"""
import hashlib
from urllib import parse
import time
import random
import string
import requests
from untils.common import (
md5_encode
)
from untils import config
__all__ = ['get_auto_reply', 'BOT_INDEX', 'BOT_NAME']
BOT_INDEX = 4
BOT_NAME = '腾讯智能闲聊'
URL = 'https://api.ai.qq.com/fcgi-bin/nlp/nlp_textchat'
def get_nlp_textchat(text, userId):
"""
智能闲聊（腾讯）<https://ai.qq.com/product/nlpchat.shtml>
接口文档：<https://ai.qq.com/doc/nlpchat.shtml>
:param text: 请求的话
:param userId: 用户标识
:return: str
"""
try:
# config.init()
info = config.get('auto_reply_info')['qqnlpchat_conf']
app_id = info['app_id']
app_key = info['app_key']
if not app_id or not app_key:
print('app_id 或 app_key 为空，请求失败')
return
# 产生随机字符串
nonce_str = ''.join(random.sample(
string.ascii_letters + string.digits, random.randint(10, 16)))
time_stamp = int(time.time()) # 时间戳
params = {
'app_id': app_id, # 应用标识
'time_stamp': time_stamp, # 请求时间戳（秒级）
'nonce_str': nonce_str, # 随机字符串
'session': md5_encode(userId), # 会话标识
'question': text # 用户输入的聊天内容
}
# 签名信息
params['sign'] = getReqSign(params, app_key)
resp = requests.get(URL, params=params)
if resp.status_code == 200:
# print(resp.text)
content_dict = resp.json()
if content_dict['ret'] == 0:
data_dict = content_dict['data']
return data_dict['answer']
print('智能闲聊 获取数据失败:{}'.format(content_dict['msg']))
return None
except Exception as exception:
print(str(exception))
def getReqSign(parser, app_key):
'''
获取请求签名，接口鉴权 https://ai.qq.com/doc/auth.shtml
1.将 <key, value> 请求参数对按 key 进行字典升序排序，得到有序的参数对列表 N
2.将列表 N 中的参数对按 URL 键值对的格式拼接成字符串，得到字符串 T（如：key1=value1&key2=value2），
URL 键值拼接过程 value 部分需要 URL 编码，URL 编码算法用大写字母，例如 %E8，而不是小写 %e8
3.将应用密钥以 app_key 为键名，组成 URL 键值拼接到字符串 T 末尾，得到字符串 S（如：key1=value1&key2=value2&app_key = 密钥)
4.对字符串 S 进行 MD5 运算，将得到的 MD5 值所有字符转换成大写，得到接口请求签名
:param parser: dect
:param app_key: str
:return: str,签名
'''
params = sorted(parser.items())
uri_str = parse.urlencode(params, encoding="UTF-8")
sign_str = '{}&app_key={}'.format(uri_str, app_key)
# print('sign =', sign_str.strip())
hash_md5 = hashlib.md5(sign_str.encode("UTF-8"))
return hash_md5.hexdigest().upper()
get_auto_reply = get_nlp_textchat
if __name__ == '__main__':
to_text = '你会爱我吗'
userId = 'userId'
form_text = get_nlp_textchat(to_text, userId)
print("会")
untils/scheduler.py
+76
View File
@@ -0,0 +1,76 @@
# coding=utf-8
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.events import EVENT_JOB_EXECUTED, EVENT_JOB_ERROR
from untils import config
from coupon.tb import tb_share_text
def job_tasks():
scheduler = BackgroundScheduler(timezone="Asia/Shanghai")
tb_job_tasks(scheduler)
# jd_job_task(scheduler)
# 加一个监控
scheduler.add_listener(scheduler_listener, EVENT_JOB_EXECUTED | EVENT_JOB_ERROR)
scheduler.start()
def tb_job_tasks(scheduler):
conf = config.get_yaml()
conf = conf.get('taobao')
if not conf.get('is_open'):
return
if conf.get('app_key') =='' or conf.get('appSecret') =='' or conf.get('adzone_id') =='':
return
app_key = conf.get('app_key')
app_secret = conf.get('app_secret')
adzone_id = conf.get('adzone_id')
chat_groups = conf.get('chat_groups')
for chat_group in chat_groups:
print(chat_group['group_name'])
scheduler.add_job(func=tb_share_text,
kwargs={'group_name': chat_group['group_name'], 'material_id': chat_group['group_material_id'], 'app_key': app_key, 'app_secret': app_secret, 'adzone_id': adzone_id},
trigger='cron', hour=f'''{chat_group['hour']}''', minute=f'''{chat_group['minute']}''', second=0, jitter=300, id=f'''{chat_group['group_name']}''')
def jd_job_task(scheduler):
conf = config.get_yaml()
conf = conf.get('jingdong')
if not conf.get('is_open'):
return
if conf.get('app_key') =='' or conf.get('app_secret') =='' or conf.get('site_id') =='' or conf.get('suo_im') =='':
return
app_key = conf.get('app_key')
appSecret = conf.get('appSecret')
site_id = conf.get('')
suo_im = conf.get('suo_im')
chat_groups = conf.get('chat_groups')
for chat_group in chat_groups:
print(chat_group['group_name'])
scheduler.add_job(func=tb_share_text,
trigger='cron', hour=f'''{chat_group['hour']}''', minute=f'''{chat_group['minute']}''', second=0, jitter=300, id=f'''{chat_group['group_name']}''')
raise NotImplementedError
def scheduler_listener(event):
'''
监听程序，如果发现错误程序终止
:param event:
:return:
'''
if event.exception:
print(f'''Error: JOB_ID: {event.job_id}, 运行时间：{(event.scheduled_run_time).strftime("%Y-%m-%d %H:%M:%S.%f")[0:19]}, 任务出错了！所有程序暂停！''')
# 别闹，不会暂停，就是一轮错误罢了。
else:
print(f'''Success: JOB_ID: {event.job_id}, 运行时间：{(event.scheduled_run_time).strftime("%Y-%m-%d %H:%M:%S.%f")[
:-3]}, 任务运行成功，继续运行...''')
if __name__ == '__main__':
job_tasks()
untils/suo_im.py
+42
View File
@@ -0,0 +1,42 @@
# coding=utf-8
"""
"""
import json
import urllib
import urllib.parse
import urllib.request
import datetime
class Suo_mi(object):
'''
需要注册 http://suo.im/ ， 而后获得 key
'''
def __init__(self, app_key):
self.app_key = app_key
# 我们默认短址一年后过期
self.expireDate = (datetime.date.today() + datetime.timedelta(days=365)).strftime('%Y-%m-%d')
def get_short_url(self, url: str) -> str:
'''
:param url: 长址
:return: 返回suo.im的短址
'''
# 取值地址，接口地址
api_url = f'''http://suo.im/api.htm?format=json&url={urllib.parse.quote(url)}&key={self.app_key}&expireDate={self.expireDate}'''
request = urllib.request.Request(url=api_url)
response = urllib.request.urlopen(request)
data = response.read()
short_url = json.loads(data)['url']
return short_url
if __name__ == '__main__':
# example
url = 'https://union-click.jd.com/jdc?e=&p=AyIGZRtcFAsRAlEfWxQyEg9QGlIQBxAPUhNrUV1KWQorAlBHU0VeBUVOWk1RAk8ECllHGAdFBwtaV1MJBAJQXk8JF0EfGQIaAlQSXhAAGgBdDBsZdmtdPGwoFUJlbilLL0xLRXA8azxhW0dEIkMnRWETb1NsOXJxZnM2WS9KVHV%2BJhscYQBmYSFSMFVhe3MNbBJARWZuMXssTHFFYB18JHV2YkUCTTBecVtOEGwDbVJyZCZbLE12dGQMb15ja0RULH8oVXVNVQVsP2kFcW4maDthcVd%2FLG8%2FYUttWyFiK3d1cGdBGS4le3V5LHsaHUFwbCMdMHF6blwrQyNRch4LZR5aFAMSDlYfWBIyEgZUGFIQBxEGUCtrFQMiRjscXREKEQJlGmsVBhoHVRxYHAMaD1wTaxUKFjcNRgVSVktTBVwPSjIiN1YrayUCETdWKwV7A0EHXRwORgF8XQVTEh0GUQY7GF4RChMCXB1rFwMTBVc%3D'
print(urllib.parse.quote(url))
app_key = ''
c = Suo_mi(app_key).get_short_url(url)
print(c)
untils/tb_top_api.py
+134
View File
@@ -0,0 +1,134 @@
# coding=utf-8
"""
首先要感谢下这篇文章：
https://www.jianshu.com/p/f9b5e3020789
值得看的一篇文章：
http://g.alicdn.com/tmapp/tida-doc/docs/top/00API%E8%B0%83%E7%94%A8%E8%AF%B4%E6%98%8E.html
"""
import hashlib
import json
import random
import time
import urllib
import urllib.parse
import urllib.request
TB_API_ROOT = 'http://gw.api.taobao.com/router/rest?'
class TbApiClient(object):
def __init__(self, app_key, secret_key, adzone_id):
self.app_key = app_key
self.secret_key = secret_key
self.adzone_id = adzone_id
#排序
def ksort(self, d):
return [(k, d[k]) for k in sorted(d.keys())]
#MD5加密
def md5(self, s, raw_output=False):
"""Calculates the md5 hash of a given string"""
res = hashlib.md5(s.encode())
if raw_output:
return res.digest()
return res.hexdigest()
#计算sign
def createSign(self, paramArr):
sign = self.secret_key
paramArr = self.ksort(paramArr)
paramArr = dict(paramArr)
for k, v in paramArr.items():
if k != '' and v != '':
sign += k + v
sign += self.secret_key
sign = self.md5(sign).upper()
return sign
#参数排序
def createStrParam(self, paramArr):
strParam = ''
for k, v in paramArr.items():
if k != '' and v != '':
strParam += k + '=' + urllib.parse.quote_plus(v) + '&'
return strParam
#高效API调用示例
def taobao_tbk_dg_optimus_material(self, material_id: str):
'''
通用物料推荐，传入官方公布的物料id，可获取指定物料
淘宝接口文档：
http://bigdata.taobao.com/api.htm?spm=a219a.7386797.0.0.4ad5669aWaaQFi&source=search&docId=33947&docType=2
:param material_id: 详见https://market.m.taobao.com/app/qn/toutiao-new/index-pc.html#/detail/10628875?_k=gpov9a
:param adzone_id: 广告位
:return:
'''
# 请求参数，根据API文档修改
# TODO
# 把分页现在这里随机有一定考虑
# 原因是：1. 不同 material_id 得到的数据不一，且刷新周期不一
# 2. 微信发送不可太频繁，我仅是怕被封，决定取很小一部分数据
page_no = str(random.choices(['1','2','3','4', '5', '6', '7', '8', '9'])[0])
page_size = str(random.randint(8, 10))
postparm = {
'page_no': page_no,
'page_size': page_size,
'adzone_id': self.adzone_id,
'material_id': material_id,
'method': 'taobao.tbk.dg.optimus.material'
}
# 公共参数，一般不需要修改
paramArr = {'app_key': self.app_key,
'v': '2.0',
'sign_method': 'md5',
'format': 'json',
'timestamp': time.strftime('%Y-%m-%d %H:%M:%S')
}
paramArr = {**paramArr, **postparm}
sign = self.createSign(paramArr)
strParam = self.createStrParam(paramArr)
strParam += 'sign=' + sign
url = TB_API_ROOT + strParam
print(url)
res = urllib.request.urlopen(url).read()
return res
def taobao_tbk_tpwd_create(self, text: str, url: str):
'''
提供淘客生成淘口令接口，淘客提交口令内容、logo、url等参数，生成淘口令关键key如：￥SADadW￥，后续进行文案包装组装用于传播
淘宝接口文档：
http://bigdata.taobao.com/api.htm?spm=a219a.7386797.0.0.494b669atcwg9a&source=search&docId=31127&docType=2
:param text: 口令弹框内容
:param url: 口令跳转目标页
:return: 返回淘口令，如<￥SADadW￥>
'''
postparm = {
'text': text,
'url': url,
'method': 'taobao.tbk.tpwd.create'
}
# 公共参数，一般不需要修改
paramArr = {'app_key': self.app_key,
'v': '2.0',
'sign_method': 'md5',
'format': 'json',
'timestamp': time.strftime('%Y-%m-%d %H:%M:%S')
}
paramArr = {**paramArr, **postparm}
sign = self.createSign(paramArr)
strParam = self.createStrParam(paramArr)
strParam += 'sign=' + sign
url = TB_API_ROOT + strParam
res = urllib.request.urlopen(url).read()
tao_command = json.loads(res)['tbk_tpwd_create_response']['data']['model']
return tao_command