抖音自动私信脚本，小红书快手微博私信插件，自动采集UID发送私信python版

import time
import random
import requests
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.common.keys import Keys
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from bs4 import BeautifulSoup
import json
import os
import re

class SocialAutoMessenger:
    def __init__(self):
        self.config = self.load_config()
        self.driver = None
        self.platform_handlers = {
            'douyin': self.handle_douyin,
            'xiaohongshu': self.handle_xiaohongshu,
            'kuaishou': self.handle_kuaishou,
            'weibo': self.handle_weibo
        }
        
    def load_config(self):
        """加载配置文件"""
        config_path = 'config.json'
        if not os.path.exists(config_path):
            default_config = {
                "platforms": ["douyin", "xiaohongshu", "kuaishou", "weibo"],
                "message_templates": [
                    "你好，关注你很久了，很喜欢你的内容！",
                    "你的分享很有价值，希望能多多交流~",
                    "看了你的内容很有启发，期待更多精彩分享！"
                ],
                "interval": [5, 10],
                "max_messages": 100,
                "login_info": {
                    "douyin": {"username": "", "password": ""},
                    "xiaohongshu": {"username": "", "password": ""},
                    "kuaishou": {"username": "", "password": ""},
                    "weibo": {"username": "", "password": ""}
                },
                "proxy": None
            }
            with open(config_path, 'w', encoding='utf-8') as f:
                json.dump(default_config, f, indent=4, ensure_ascii=False)
            return default_config
        else:
            with open(config_path, 'r', encoding='utf-8') as f:
                return json.load(f)
    
    def init_driver(self):
        """初始化浏览器驱动"""
        chrome_options = Options()
        chrome_options.add_argument("--disable-notifications")
        chrome_options.add_argument("--disable-popup-blocking")
        chrome_options.add_argument("--disable-infobars")
        
        if self.config.get('proxy'):
            chrome_options.add_argument(f"--proxy-server={self.config['proxy']}")
            
        self.driver = webdriver.Chrome(options=chrome_options)
        self.driver.maximize_window()
    
    def login(self, platform):
        """平台登录"""
        login_info = self.config['login_info'].get(platform)
        if not login_info or not login_info['username'] or not login_info['password']:
            print(f"请先在config.json中配置{platform}的登录信息")
            return False
            
        if platform == 'douyin':
            self.driver.get("https://www.douyin.com/")
            time.sleep(5)
            # 抖音登录逻辑...
            
        elif platform == 'xiaohongshu':
            self.driver.get("https://www.xiaohongshu.com/")
            time.sleep(3)
            # 小红书登录逻辑...
            
        elif platform == 'kuaishou':
            self.driver.get("https://www.kuaishou.com/")
            time.sleep(3)
            # 快手登录逻辑...
            
        elif platform == 'weibo':
            self.driver.get("https://weibo.com/")
            time.sleep(3)
            # 微博登录逻辑...
            
        return True
    
    def handle_douyin(self):
        """处理抖音私信"""
        print("开始处理抖音私信...")
        if not self.login('douyin'):
            return
            
        # 抖音私信逻辑...
        # 这里应该有详细的页面元素定位和交互代码
        # 包括获取用户列表、发送消息等
        
    def handle_xiaohongshu(self):
        """处理小红书私信"""
        print("开始处理小红书私信...")
        if not self.login('xiaohongshu'):
            return
            
        # 小红书私信逻辑...
        
    def handle_kuaishou(self):
        """处理快手私信"""
        print("开始处理快手私信...")
        if not self.login('kuaishou'):
            return
            
        # 快手私信逻辑...
        
    def handle_weibo(self):
        """处理微博私信"""
        print("开始处理微博私信...")
        if not self.login('weibo'):
            return
            
        # 微博私信逻辑...
        
    def get_random_message(self):
        """获取随机消息模板"""
        return random.choice(self.config['message_templates'])
    
    def get_random_interval(self):
        """获取随机间隔时间"""
        min_val, max_val = self.config['interval']
        return random.uniform(min_val, max_val)
    
    def run(self):
        """运行主程序"""
        try:
            self.init_driver()
            
            for platform in self.config['platforms']:
                if platform in self.platform_handlers:
                    handler = self.platform_handlers[platform]
                    handler()
                    time.sleep(self.get_random_interval())
                    
        except Exception as e:
            print(f"发生错误: {str(e)}")
        finally:
            if self.driver:
                self.driver.quit()
            print("程序运行结束")

if __name__ == "__main__":
    messenger = SocialAutoMessenger()
    messenger.run()
 requests
from bs4 import BeautifulSoup
import re
import json
import time
import random
from concurrent.futures import ThreadPoolExecutor

class UIDCollector:
    def __init__(self):
        self.config = self.load_config()
        self.headers = {
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36'
        }
        self.proxy = self.config.get('proxy')
        
    def load_config(self):
        """加载配置文件"""
        config_path = 'collector_config.json'
        if not os.path.exists(config_path):
            default_config = {
                "keywords": ["美食", "旅游", "科技"],
                "max_pages": 10,
                "output_file": "uids.json",
                "platforms": ["douyin", "xiaohongshu", "kuaishou", "weibo"],
                "proxy": None
            }
            with open(config_path, 'w', encoding='utf-8') as f:
                json.dump(default_config, f, indent=4, ensure_ascii=False)
            return default_config
        else:
            with open(config_path, 'r', encoding='utf-8') as f:
                return json.load(f)
    
    def collect_douyin_uids(self, keyword):
        """采集抖音UID"""
        print(f"开始采集抖音关键词'{keyword}'相关UID...")
        uids = []
        base_url = f"https://www.douyin.com/search/{keyword}"
        
        # 这里应该有详细的采集逻辑
        # 包括模拟滚动、解析页面、提取UID等
        
        return uids
    
    def collect_xiaohongshu_uids(self, keyword):
        """采集小红书UID"""
        print(f"开始采集小红书关键词'{keyword}'相关UID...")
        uids = []
        base_url = f"https://www.xiaohongshu.com/search_result/{keyword}"
        
        # 小红书采集逻辑...
        
        return uids
    
    def collect_kuaishou_uids(self, keyword):
        """采集快手UID"""
        print(f"开始采集快手关键词'{keyword}'相关UID...")
        uids = []
        base_url = f"https://www.kuaishou.com/search/video?searchKey={keyword}"
        
        # 快手采集逻辑...
        
        return uids
    
    def collect_weibo_uids(self, keyword):
        """采集微博UID"""
        print(f"开始采集微博关键词'{keyword}'相关UID...")
        uids = []
        base_url = f"https://s.weibo.com/weibo/{keyword}"
        
        # 微博采集逻辑...
        
        return uids
    
    def save_uids(self, uids):
        """保存采集到的UID"""
        with open(self.config['output_file'], 'w', encoding='utf-8') as f:
            json.dump(uids, f, indent=4, ensure_ascii=False)
        print(f"已保存{len(uids)}个UID到{self.config['output_file']}")
    
    def run(self):
        """运行采集程序"""
        all_uids = []
        
        with ThreadPoolExecutor(max_workers=4) as executor:
            futures = []
            for keyword in self.config['keywords']:
                for platform in self.config['platforms']:
                    if platform == 'douyin':
                        futures.append(executor.submit(self.collect_douyin_uids, keyword))
                    elif platform == 'xiaohongshu':
                        futures.append(executor.submit(self.collect_xiaohongshu_uids, keyword))
                    elif platform == 'kuaishou':
                        futures.append(executor.submit(self.collect_kuaishou_uids, keyword))
                    elif platform == 'weibo':
                        futures.append(executor.submit(self.collect_weibo_uids, keyword))
            
            for future in futures:
                try:
                    uids = future.result()
                    all_uids.extend(uids)
                except Exception as e:
                    print(f"采集过程中发生错误: {str(e)}")
        
        # 去重
        unique_uids = list({uid['platform']+uid['id']: uid for uid in all_uids}.values())
        self.save_uids(unique_uids)

if __name__ == "__main__":
    collector = UIDCollector()
    collector.run()