社群营销 - 程序自动发淘宝商品消息到微信群 案例展示
作者：互联网
一、背景：
有一个小伙伴儿找到我，希望能帮他实现TB商品自动群发到VX，减少工作量（毕竟都是复制转发的活，又繁琐又辛苦），解放双手才是硬道理。
二、设计思路如下：
1.python flask 服务器，HTTP API形式接收任意TB、TM、JD、PDD等商品信息。
2.将商品信息转发到指定的微信群（使用JSON配置），实现自动化社群营销。
3.目前网页版VX各种限制登录，所以本方案采用的是微信PC版，无登录限制。
三、功能实现：
上代码 ， web服务器模块：
import sys
import threading
import time
import traceback
from flask import Flask, json
from flask import request
import wechat_qq
from mylog import logger
import click
import wx
import wx.adv
global g_group
app_flask = Flask(__name__)
# {"data":{"id":"559802945645","imgUrl":"https://img.alicdn.com/bao/uploaded/i3/2726809950/TB2dLIybNTpK1RjSZFMXXbG_VXa_!!2726809950.jpg_80x80.jpg","price":"当前价格:￥900.00","title":"全新正品天能12v150ah蓄电池奇瑞宝雅电动车免维护电池6-EVF-150"},"time":1576508829011,"type":"market","config":{"hz":"300","title":"新页签"}}
@app_flask.route('/pushToWechat/<wechat_group>', methods=['GET', 'POST'])
def pushToWechat(wechat_group):
# @app.route('/pushToWechat', methods=['GET', 'POST'])
# def pushToWechat():
result = {}
msg = {}
try:
# global g_group
# wechat_group = g_group
logger.debug("wechat_group %s", wechat_group)
req_data = request.get_data().decode("utf-8")
json_data = json.loads(req_data)
data = json_data.get("data")
title = data["title"]
# msg.append(title)
msg["title"] = "%s" % title
logger.debug(title)
tp = json_data.get("type")
tp_str = ""
if tp.find("market") >= 0:
tp_str = "【购物车】"
elif tp.find("ft") >= 0:
tp_str = "【足迹】"
logger.debug("type %s %s" % (tp, tp_str))
# id = data.get("id")
nickname = json_data.get("nickname")
logger.debug("nickname %s " % nickname)
# msg.append("%s%s"% (tp_str, nickname))
msg["type_nickname"] = "%s%s"% (tp_str, nickname)
imgUrl = data.get("imgUrl")
logger.debug("imgUrl %s", imgUrl)
msg["imgUrl"] = "%s" % imgUrl
price_ori = data.get("price")
price = price_ori
index = price_ori.find("/")
if index > 0:
price = price_ori[0: index]
# msg.append("【在售价】%s" % price)
msg["price"] = "【在售价】%s" % price
tm = json_data.get("time")
# msg.append("【时间】%s" % time.strftime("%H:%M:%S", time.localtime(time/1000)))
msg["time"] = "【时间】%s" % time.strftime("%H:%M:%S", time.localtime(tm/1000))
url = data["url"]
msg["url"] = "%s" % url
# msg.append(url)
logger.debug(url)
wechat_qq.wechat_send(windowTitle=wechat_group, msgList=msg)
except Exception as e:
logger.error(traceback.format_exc())
print(e)
return {"ret": 0}
@click.command()
# @click.option('--group', '-g', prompt='Please enter wechat group name ', help='')
@click.option('--port', '-p', default= "5000", prompt='Please enter listen port, default 5000', help='')
@click.option('--debug', '-d', default = 0 , help='For example: 1 or 0 ')
# def run( port , group, debug):
def run( port , debug):
print(port, debug)
# global g_group
# g_group = group
app_flask.run(host="0.0.0.0", port=port, debug = debug)
#
# if __name__ == '__main__':
# run()
class MyTaskBarIcon(wx.adv.TaskBarIcon):
ICON = "logo.ico" # 图标地址
ID_ABOUT = wx.NewId() # 菜单选项“关于”的ID
# ID_EXIT = wx.NewId() # 菜单选项“退出”的ID
# ID_SHOW_WEB = wx.NewId() # 菜单选项“显示页面”的ID
TITLE = "推送消息到微信" # 鼠标移动到图标上显示的文字
def __init__(self):
wx.adv.TaskBarIcon.__init__(self)
self.SetIcon(wx.Icon(self.ICON), self.TITLE) # 设置图标和标题
self.Bind(wx.EVT_MENU, self.onAbout, id=self.ID_ABOUT) # 绑定“关于”选项的点击事件
# self.Bind(wx.EVT_MENU, self.onExit, id=self.ID_EXIT) # 绑定“退出”选项的点击事件
# self.Bind(wx.EVT_MENU, self.onShowWeb, id=self.ID_SHOW_WEB) # 绑定“显示页面”选项的点击事件
# “关于”选项的事件处理器
def onAbout(self, event):
wx.MessageBox('最后更新日期：2019-12-21', "关于")
# “退出”选项的事件处理器
def onExit(self, event):
wx.Exit()
# “显示页面”选项的事件处理器
def onShowWeb(self, event):
pass
# 创建菜单选项
def CreatePopupMenu(self):
menu = wx.Menu()
for mentAttr in self.getMenuAttrs():
menu.Append(mentAttr[1], mentAttr[0])
return menu
# 获取菜单的属性元组
def getMenuAttrs(self):
return [#('进入程序', self.ID_SHOW_WEB),
('关于', self.ID_ABOUT)]
# ('退出', self.ID_EXIT)]
class MyFrame(wx.Frame):
def __init__(self):
wx.Frame.__init__(self)
MyTaskBarIcon() # 显示系统托盘图标
# run()
t = threading.Thread(target=run)
t.start()
#
# def __init__(self):
# wx.Frame.__init__(self, None)
# self.panel = wx.Panel(self, -1)
# self.button = wx.Button(self.panel, wx.ID_OK)
# self.Bind(wx.EVT_BUTTON, self.onClick, self.button)
# print(app_flask)
# t = threading.Thread(target=run)
# t.start()
# run()
#
# def onClick(self, evt):
# dialog = wx.Dialog(self.panel)
# rec = dialog.ShowModal()
class MyApp(wx.App):
def OnInit(self):
MyFrame()
return True
if __name__ == "__main__":
# app = MyApp()
# app.MainLoop()
app = wx.App()
frame = MyFrame()
frame.Show()
app.MainLoop()
群发消息模块：
# coding = utf-8
import win32gui
import win32api
from threading import Lock
import win32con
import win32clipboard as clipboard
import time
from pynput.mouse import Button, Controller as mController
from pynput.keyboard import Key, Controller as kController
from mylog import *
from PIL import ImageGrab
import image_clipboard
import traceback, threading
mouse = mController()
keyboard = kController()
import win32api,win32gui,win32con
import os
import time
from enum import Enum
class Lan(Enum):
"""
语言代码值参考：https://msdn.microsoft.com/en-us/library/cc233982.aspx
"""
EN = 0x4090409
ZH = 0x8040804
def change_lan(lan :Lan):
"""
修改当前激活窗口输入法
:param lan: 语言类型
:return: True 修改成功，False 修改失败
"""
# 获取系统输入法列表
hwnd = win32gui.GetForegroundWindow()
im_list = win32api.GetKeyboardLayoutList()
im_list = list(map(hex, im_list))
# 加载输入法
if hex(lan.value) not in im_list:
win32api.LoadKeyboardLayout('0000' + hex(lan.value)[-4:], 1)
result = win32api.SendMessage(
hwnd,
win32con.WM_INPUTLANGCHANGEREQUEST,
0,
lan.value)
if result == 0:
logger.info('设置 %s 键盘 %d (0 == success)！' % (lan.name, result))
return result
def get_keyboard_status():
# 获取键盘布局列表
im_list = win32api.GetKeyboardLayoutList()
im_list = list(map(hex, im_list))
print(im_list)
# 传入类名，标题，返回tuple(句柄，坐标左，坐标左顶，坐标左右，坐标左底)
def findWindow(classname, titlename):
hwnd = win32gui.FindWindow(classname, titlename)
if (hwnd != 0):
left, top, right, bottom = win32gui.GetWindowRect(hwnd)
return {'hwnd': hwnd, 'left': left, 'top': top, 'right': right, 'bottom': bottom}
else:
return 0
def send_line(message):
keyboard.type(message) # 程序运行时候，这里一定要是英文输入状态，要不然可能无法发送消息
with keyboard.pressed(Key.ctrl):
keyboard.press(Key.enter)
keyboard.release(Key.enter)
# 发送消息,需要窗口标题，消息内容两个参数
def wechat_send(windowTitle, msgList):
# 微信pc端的输入框都没有句柄，所以需要通过模拟点击来获得焦点.虽然QQ有句柄，但是为了统一，也用模拟点击吧
# 定位QQ(tim)窗口输入框位置，模拟鼠标点击来获得焦点。
winClass = "ChatWnd" # 默认是微信消息
win = findWindow(winClass, windowTitle)
if 0 == win:
logger.error("发送消息给[%s]失败：找不到窗口" % windowTitle)
return
hwnd = win['hwnd']
try:
# shell = win32com.client.Dispatch("WScript.Shell")
# shell.SendKeys('%') #发送ALT键，ALT键使用%号表示
win32gui.SetForegroundWindow(hwnd) # bool
except Exception as e:
logger.error(traceback.format_exc())
time.sleep(0.002) # 这里要缓一下电脑才能反应过来，要不然可能找不到焦点
change_lan(Lan.EN)
mutex.acquire()
inputPos = [win['right'] - 50, win['bottom'] - 50]
win32api.SetCursorPos(inputPos) # 定位鼠标到输入位置
# 执行左单键击，若需要双击则延时几毫秒再点击一次即可
mouse.press(Button.left)
mouse.release(Button.left)
send_line(msgList.get("title"))
send_line(msgList.get("type_nickname"))
send_line(msgList.get("price"))
send_line(msgList.get("time"))
send_line("")
send_line(msgList.get("url"))
# 获取网络图片粘贴到剪贴板南极
try:
image_clipboard.setImageClipboard(image_clipboard.set_image_txt(msgList))
except Exception as e:
pass
# image_clipboard.set_image_clipboard(msgList["imgUrl"])
# image_clipboard.setImageClipboard(image_clipboard.set_image(msgList.get("imgUrl")))
# 粘贴到微信输入框
with keyboard.pressed(Key.ctrl):
keyboard.press('v')
keyboard.release('v')
# 发送消息的快捷键是 Alt+s
with keyboard.pressed(Key.alt):
keyboard.press('s')
keyboard.release('s')
mutex.release()
logger.info("发送消息给[%s]成功" % windowTitle)
# 发送QQ消息,这里默认使用 TIM
def qqsend(windowTitle, message):
win = findWindow("TXGuiFoundation", windowTitle)
if (win):
clipboard.OpenClipboard()
clipboard.EmptyClipboard()
clipboard.SetClipboardData(win32con.CF_UNICODETEXT, message)
clipboard.CloseClipboard()
# 填充消息
win32gui.SendMessage(win['hand'], 770, 0, 0)
# 回车发送消息
win32gui.SendMessage(win['hand'], win32con.WM_KEYDOWN, win32con.VK_RETURN, 0)
else:
print("发送消息给[%s]失败" % windowTitle)
# 发送消息不能太频繁，会被微信封锁 1秒一个，发1000条被限制
# i = 0
# while True:
# i = i + 1
# wechat_send(windowTitle=u"test", message="%d 我是 robot , 我在测试自动发消息 %d" % (i, time.time()))
# time.sleep(1)
# 英文必须末尾加上\n ，才把备选字符串放入了输入框\
mutex = threading.Lock()
msgList = list()
msgList.append("迪士尼儿童帽子时尚男女孩加厚加绒宝宝针织帽可爱保暖护耳冬帽")
msgList.append("【用户id】560184843123")
# wechat_send(windowTitle=u"test", msgList=msgList, imgUrl="" )
四、总结
两页代码实现了这个功能，让小伙伴儿有更多的时间精力去做更有创造力的事情。
标签：__,微信群,data,self,淘宝,import,社群,def,wx 来源： https://blog.51cto.com/u_15294888/2990196