闲鱼批量发布工具，一键批量发布商品软件，python实现

import os import time import csv import random 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 XianyuAutoPublisher: def init(self): self.driver = None self.wait_timeout = 30 self.retry_max = 3 self.cookie_dir = "cookies/" self.data_dir = "data/"

    # 初始化目录
    os.makedirs(self.cookie_dir, exist_ok=True)
    os.makedirs(self.data_dir, exist_ok=True)

def init_driver(self, headless=False):
    """初始化浏览器驱动"""
    options = webdriver.ChromeOptions()
    if headless:
        options.add_argument('--headless')
    options.add_argument("--disable-gpu")
    options.add_argument("--window-size=1920,1080")
    options.add_argument("--log-level=3")
    
    self.driver = webdriver.Chrome(options=options)
    self.driver.implicitly_wait(10)

def login_with_cookie(self, username):
    """使用Cookie登录"""
    cookie_file = os.path.join(self.cookie_dir, f"{username}.cookie")
    if not os.path.exists(cookie_file):
        raise FileNotFoundError(f"Cookie文件不存在: {cookie_file}")
        
    self.driver.get("https://login.taobao.com")
    with open(cookie_file, 'r') as f:
        cookies = eval(f.read())
    
    for cookie in cookies:
        if 'expiry' in cookie:
            del cookie['expiry']
        self.driver.add_cookie(cookie)
    
    # 验证登录状态
    self.driver.get("https://2.taobao.com")
    try:
        WebDriverWait(self.driver, self.wait_timeout).until(
            EC.presence_of_element_located((By.CLASS_NAME, "nickname"))
        )
        return True
    except TimeoutException:
        return False

def load_product_data(self, csv_file):
    """加载商品数据"""
    products = []
    with open(os.path.join(self.data_dir, csv_file), 'r', encoding='utf-8') as f:
        reader = csv.DictReader(f)
        for row in reader:
            products.append({
                'title': row['title'],
                'desc': row['description'],
                'price': float(row['price']),
                'images': row['images'].split('|'),
                'category': row['category'],
                'location': row['location']
            })
    return products

def upload_images(self, image_paths):
    """批量上传图片"""
    upload_btn = self.driver.find_element(By.CSS_SELECTOR, ".uploader-input input")
    for img in image_paths:
        upload_btn.send_keys(os.path.abspath(img))
        time.sleep(random.uniform(0.5, 1.5))
    
    # 等待所有图片上传完成
    WebDriverWait(self.driver, self.wait_timeout).until(
        lambda d: len(d.find_elements(By.CSS_SELECTOR, ".image-item")) == len(image_paths)
    )

def fill_product_info(self, product):
    """填写商品信息"""
    # 填写标题
    title_input = self.driver.find_element(By.NAME, "title")
    title_input.clear()
    title_input.send_keys(product['title'])
    
    # 填写描述
    desc_textarea = self.driver.find_element(By.CSS_SELECTOR, "textarea.lego-textarea")
    desc_textarea.clear()
    desc_textarea.send_keys(product['desc'])
    
    # 设置价格
    price_input = self.driver.find_element(By.NAME, "price")
    price_input.clear()
    price_input.send_keys(str(product['price']))
    
    # 选择分类
    self.select_category(product['category'])
    
    # 设置位置
    if product['location']:
        self.set_location(product['location'])

def select_category(self, category_path):
    """选择商品分类"""
    # 实现三级分类选择的完整代码
    pass

def set_location(self, location):
    """设置发货地"""
    # 实现地理位置选择的完整代码
    pass

def publish_product(self, retry=0):
    """发布商品"""
    try:
        publish_btn = WebDriverWait(self.driver, self.wait_timeout).until(
            EC.element_to_be_clickable((By.CSS_SELECTOR, ".submit-btn"))
        )
        publish_btn.click()
        
        # 验证发布成功
        WebDriverWait(self.driver, self.wait_timeout).until(
            EC.url_contains("item.htm")
        )
        return True
    except Exception as e:
        if retry < self.retry_max:
            time.sleep(2)
            return self.publish_product(retry + 1)
        raise Exception(f"发布失败: {str(e)}")

def batch_publish(self, username, products_csv):
    """批量发布主流程"""
    try:
        self.init_driver()
        if not self.login_with_cookie(username):
            raise Exception("登录失败")
        
        products = self.load_product_data(products_csv)
        
        for idx, product in enumerate(products, 1):
            print(f"正在发布第 {idx}/{len(products)} 个商品: {product['title']}")
            
            self.driver.get("https://2.taobao.com/publish/post.htm")
            self.upload_images(product['images'])
            self.fill_product_info(product)
            
            if self.publish_product():
                print(f"✅ 商品 [{product['title']}] 发布成功")
            else:
                print(f"❌ 商品 [{product['title']}] 发布失败")
            
            # 随机间隔防止被封
            time.sleep(random.uniform(5, 15))
        
        print(f"🎉 批量发布完成，共成功发布 {len(products)} 个商品")
        return True
        
    except Exception as e:
        print(f"⚠️ 发布过程中出现异常: {str(e)}")
        return False
    finally:
        if self.driver:
            self.driver.quit()

if name == "main": publisher = XianyuAutoPublisher()

# 示例用法
publisher.batch_publish(
    username="your_xianyu_account",
    products_csv="products_list.csv"
)