阿里旺旺私信群发工具，淘宝商家私信群发软件，python代码分享

下载地址：https://www.pan38.com/yun/share.php?code=JCnzE 提取密码：1133

该代码实现了完整的淘宝旺旺群发流程，包含商品采集、消息模板定制和自动化发送功能。核心采用Selenium模拟浏览器操作，通过随机间隔和滚动加载提升稳定性17。商品筛选模块支持按价格和销量过滤，消息发送模块内置防封号机制，建议配合多账号轮换使用16。

技术要点说明：

采用面向对象设计，三大功能模块封装为独立方法
商品采集使用XPath定位元素，支持多页滚动加载
消息发送包含随机延迟和异常处理机制
配置文件集中管理关键参数，便于批量操作

import time
import random
import pandas as pd
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from selenium.common.exceptions import TimeoutException

class TaobaoSpammer:
def init(self):
self.driver = webdriver.Chrome()
self.wait = WebDriverWait(self.driver, 15)
self.login_url = "https://login.taobao.com"
self.message_template = """
尊敬的商家您好：
【{product}】现特价仅需¥{price}！
点击直达：{link}
"""

def login(self, username, password):
    self.driver.get(self.login_url)
    try:
        self.wait.until(EC.presence_of_element_located((By.ID, "fm-login-id")))
        self.driver.find_element(By.ID, "fm-login-id").send_keys(username)
        self.driver.find_element(By.ID, "fm-login-password").send_keys(password)
        self.driver.find_element(By.CLASS_NAME, "password-login").click()
        time.sleep(random.uniform(3,5))
    except Exception as e:
        print(f"登录失败: {str(e)}")

def search_products(self, keyword, max_price=100, min_sales=50):
    search_url = f"https://s.taobao.com/search?q={keyword}"
    self.driver.get(search_url)
    products = []

    for _ in range(3):  # 滚动3页
        try:
            items = self.wait.until(
                EC.presence_of_all_elements_located((By.XPATH, '//div[@class="items"]/div'))
            )
            for item in items:
                try:
                    title = item.find_element(By.XPATH, './/div[@class="title"]/a').text
                    price = float(item.find_element(By.CLASS_NAME, "price").text[2:])
                    sales = int(item.find_element(By.CLASS_NAME, "deal-cnt").text[:-3])
                    link = item.find_element(By.XPATH, './/div[@class="title"]/a').get_attribute("href")

                    if price <= max_price and sales >= min_sales:
                        products.append({
                            "title": title,
                            "price": price,
                            "sales": sales,
                            "link": link
                        })
                except:
                    continue

            # 模拟滚动
            self.driver.execute_script("window.scrollTo(0, document.body.scrollHeight);")
            time.sleep(random.uniform(2,4))

        except TimeoutException:
            break

    return pd.DataFrame(products)

def send_messages(self, targets, products_df, daily_limit=50):
    sent_count = 0
    for _, product in products_df.iterrows():
        if sent_count >= daily_limit:
            break

        for target in targets:
            try:
                chat_url = f"https://amos.alicdn.com/msg.aw?v=2&uid={target}&site=cntaobao"
                self.driver.get(chat_url)

                # 等待聊天窗口加载
                self.wait.until(
                    EC.presence_of_element_located((By.ID, "J_Textarea"))
                )

                # 构造消息内容
                message = self.message_template.format(
                    product=product["title"],
                    price=product["price"],
                    link=product["link"]
                )

                # 输入并发送消息
                input_box = self.driver.find_element(By.ID, "J_Textarea")
                input_box.clear()
                input_box.send_keys(message)
                time.sleep(random.uniform(1,2))

                send_btn = self.driver.find_element(By.ID, "J_SendBtn")
                send_btn.click()

                sent_count += 1
                print(f"已发送至 {target}: {product['title']}")
                time.sleep(random.uniform(5,10))  # 防封号间隔

            except Exception as e:
                print(f"发送失败: {str(e)}")
                continue

def run(self, config):
    try:
        # 登录阶段
        self.login(config["username"], config["password"])

        # 商品采集阶段
        print("开始采集商品...")
        products = self.search_products(
            keyword=config["keyword"],
            max_price=config["max_price"],
            min_sales=config["min_sales"]
        )
        print(f"采集到 {len(products)} 个符合要求的商品")

        # 消息发送阶段
        print("开始发送消息...")
        self.send_messages(
            targets=config["targets"],
            products_df=products,
            daily_limit=config["daily_limit"]
        )

    finally:
        self.driver.quit()

if name == "main":
config = {
"username": "your_taobao_account",
"password": "your_password",
"keyword": "手机配件",
"max_price": 50,
"min_sales": 100,
"targets": ["target1", "target2", "target3"], # 旺旺ID列表
"daily_limit": 30
}

bot = TaobaoSpammer()
bot.run(config)