社交媒体爬虫

import tweepy

# 创建 Tweepy 客户端
client = tweepy.Client(
    bearer_token=account.bearer_token,
    consumer_key=account.api_key,
    consumer_secret=account.api_key_secret,
    access_token=account.access_token,
    access_token_secret=account.access_token_secret,
    wait_on_rate_limit=True
)

try:
    # 发布推文
    response = client.create_tweet(text=request.content)
    tweet_id = response.data['id']
    return {
        "success": True,
        "tweet_id": tweet_id,
        "message": "推文发布成功"
    }
except tweepy.TweepyException as e:
    return f"发布推文失败：{e}"


# 使用代理发推文，待验证
# tweepy.API 支持代理参数
auth = tweepy.OAuth1UserHandler(
    account.api_key,
    account.api_key_secret,
    account.access_token,
    account.access_token_secret
)

api = tweepy.API(
    auth,
    proxy=f'http://{PROXY_USERNAME}:{PROXY_PASSWORD}@{PROXY_HOST}:{PROXY_PORT}',
    wait_on_rate_limit=True
)
# 使用 API 发推
api.update_status(request.content)

from telethon import TelegramClient, events                   # 导入客户端类与事件装饰器
import asyncio                                                # 标准库：异步事件循环

# ----------------- 账户配置 -----------------
api_id   = 123456                                            # 替换为你的 api_id
api_hash = 'xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx'                # 替换为你的 api_hash
session  = 'session'                                         # 会生成本地会话文件
GROUP_NAME = '测试群'                                         # 需要监听的群组名称或 chat_id
TARGET_USER = '@target_username'                             # 启动时要主动私聊的对象
INIT_TEXT  = '您好，这是一条测试消息'                          # 启动时发送的问候语
# ------------------------------------------------------------

client = TelegramClient(session, api_id, api_hash)            # 创建客户端实例

# 1️⃣ 监听群聊消息并回复
@client.on(events.NewMessage(chats=GROUP_NAME))               # 只监控指定群
async def group_handler(event):                               # 异步事件回调
    msg = event.message.message                               # 获取文本内容
    if 'ping' in msg.lower():                                 # 简单匹配关键词
        await event.reply('pong（来自机器人）')                # 群内回复

# 2️⃣ 监听所有私聊消息并回复
@client.on(events.NewMessage(incoming=True))                  # 监听所有入站消息
async def private_handler(event):
    if not event.is_private:                                  # 过滤掉群消息
        return                                                # 不是私聊则忽略
    text = event.raw_text                                     # 获取原始文本
    if '你好' in text:                                        # 判断关键字
        await event.respond('你好！有什么能帮您？')             # 私聊回复

# 3️⃣ 启动后给指定对象发送一条消息
async def send_initial_message():                             # 定义异步发送函数
    await client.send_message(TARGET_USER, INIT_TEXT)         # 主动发送私聊/群消息

# -------------------- 入口 --------------------
async def main():
    await send_initial_message()                              # 启动即发送问候
    await client.run_until_disconnected()                     # 保持长连接

with client:                                                  # 自动完成登录流程
    asyncio.run(main())                                       # 运行主协程

from neonize.client import NewClient
from neonize.events import MessageEv, GroupMessageEv, event

BOT = NewClient("session")                 # 会在当前目录生成 session.db

@BOT.event
def on_ready(c, _):                        # 连接成功事件
    print("✅ 已连接 WhatsApp!")

@BOT.event
def on_message(c: NewClient, ev: MessageEv):          # 私聊
    if ev.message.conversation.lower() == "ping":
        c.reply_message("pong", ev.message)

@BOT.event
def on_group_message(c: NewClient, ev: GroupMessageEv):  # 群聊
    txt = ev.message.conversation or ""
    if txt.lower() == "ping":
        c.reply_message("pong (群)", ev.message)

def main():
    # 启动后先给自己发一句话作演示
    BOT.connect()                                   # 首次会显示二维码，手机扫码登录
    me = BOT.get_self()                             # 自己的 JID
    BOT.send_message(me, "机器人已上线~")             # 主动发私信
    event.wait()                                    # 阻塞主线程

if __name__ == "__main__":
    main()

# 中文注释：
# 1) 扫码后即获完整个人号权限；监听私聊/群消息，收到“ping”就回“pong”
# 2) Neonize 内部会捕获 Flood-Wait，如需大批量发消息请自行 await asyncio.sleep(err.retry_after)