You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

Tweepy程序无报错但未实际发推文问题排查求助

问题描述

使用Python 3.11(也曾尝试Python 3.7+Anaconda)与PyCharm社区版开发推特机器人,当前处于测试阶段。代码编译运行无报错,控制台显示已完成推文操作,但实际推特平台无对应推文发出。

控制台输出示例:

Connecting to Twitter...
Connected!

Tweeted Slide1.JPG
Waiting for XYZ secs....

附原代码:

import tweepy
import configparser
import os
import threading
import time
import sys

WAIT_TIME = 300000
done = False

def get_tokens():
    api_key = 'key'
    api_secret_key = 'key secret'
    access_token_key = 'token key'
    access_token_secret = 'token secret'
    return api_key, api_secret_key, access_token_key, access_token_secret


def animate():
    for c in range(WAIT_TIME, -1, -1):
        if done:
            break
        sys.stdout.write('\rWaiting for ' + str(c) + ' secs.... ')
        sys.stdout.flush()
        time.sleep(1)
    sys.stdout.write('\rTweeting\n')


def write_tweeted_img(file_name):
    file = open(r"new.txt", 'a') #file path
    file.write('{}\n'.format(file_name))
    file.close()


def check_img_tweeted(file_name):
    with open(r"new.txt") as f: #file path
        if file_name in f.read():
            return True
    return False


if __name__ == '__main__':
    consumer_key, consumer_secret, access_token, access_secret = get_tokens()
    count = 0

    print('Connecting to Twitter...')
    auth = tweepy.OAuthHandler(consumer_key, consumer_secret)
    auth.set_access_token(access_token, access_secret)
    api = tweepy.API(auth)
    print('Connected!\n')

    os.chdir(r"") #imge directory path
    for image in os.listdir(r""): #imge directory path

        if check_img_tweeted(image):
            print('Already Tweeted {} before'.format(image))
            os.remove(image)
            continue

        try:
            api.search_tweets(image)
            print('Tweeted {}'.format(image))
            write_tweeted_img(image)
            count += 1
            os.remove(image)
            done = False

        except TweepError:
            continue

        t = threading.Thread(target=animate)
        t.start()
        time.sleep(WAIT_TIME)
        done = True

    print('\nSuccessfully Tweeted {} Images'.format(count))
问题原因分析
  1. 核心错误:调用了错误的API方法
    代码中使用api.search_tweets(image),这是Tweepy的搜索推文接口,作用是在推特平台搜索包含指定关键词的内容,而非发布推文。这是控制台显示“已推文”但平台无内容的根本原因。
  2. 错误处理失效
    原代码未正确导入tweepy.errors.TweepError,且异常块直接continue,没有打印错误信息,无法排查认证、权限或API调用问题。
  3. 等待时间设置不合理
    WAIT_TIME = 300000表示等待300000秒(约5000分钟),远超推特API的调用限制逻辑,属于无效设置。
  4. 线程逻辑冗余
    使用线程实现倒计时动画,但主线程仍调用time.sleep(WAIT_TIME),导致等待逻辑重复,且done变量未做线程安全处理。
解决方法

1. 替换为正确的发推方法

对于带图片的推文,使用api.update_status_with_media()方法,该方法支持上传图片并发布文本内容。

2. 修复错误类导入与异常处理

从tweepy.errors导入TweepError,同时在异常块中打印错误信息,方便排查问题。

3. 调整等待时间

将WAIT_TIME改为符合推特API限制的数值,例如300秒(5分钟),避免触发限流。

4. 完善路径配置

填写正确的图片目录路径和new.txt文件路径,确保程序能读取图片和记录已推文文件。

5. 优化线程逻辑

移除冗余的线程等待逻辑,或使用更简洁的倒计时实现。

修正后的代码示例

import tweepy
from tweepy.errors import TweepError
import os
import time
import sys

# 调整为合理的等待时间(5分钟,300秒)
WAIT_TIME = 300
done = False

def get_tokens():
    # 替换为你的真实密钥
    api_key = 'your_api_key'
    api_secret_key = 'your_api_secret_key'
    access_token_key = 'your_access_token'
    access_token_secret = 'your_access_token_secret'
    return api_key, api_secret_key, access_token_key, access_token_secret

def animate_wait():
    for c in range(WAIT_TIME, -1, -1):
        if done:
            break
        sys.stdout.write(f'\rWaiting for {c} secs.... ')
        sys.stdout.flush()
        time.sleep(1)
    sys.stdout.write('\rReady to tweet next\n')

def write_tweeted_img(file_name):
    # 替换为你的new.txt绝对路径
    with open(r"/path/to/your/new.txt", 'a') as file:
        file.write(f'{file_name}\n')

def check_img_tweeted(file_name):
    # 替换为你的new.txt绝对路径
    try:
        with open(r"/path/to/your/new.txt") as f:
            return file_name in f.read()
    except FileNotFoundError:
        # 文件不存在则创建并返回False
        open(r"/path/to/your/new.txt", 'w').close()
        return False

if __name__ == '__main__':
    consumer_key, consumer_secret, access_token, access_secret = get_tokens()
    count = 0

    print('Connecting to Twitter...')
    auth = tweepy.OAuthHandler(consumer_key, consumer_secret)
    auth.set_access_token(access_token, access_secret)
    # 添加wait_on_rate_limit=True自动处理限流
    api = tweepy.API(auth, wait_on_rate_limit=True)
    
    # 验证认证是否成功
    try:
        api.verify_credentials()
        print('Connected successfully!\n')
    except TweepError as e:
        print(f'Authentication failed: {e}')
        sys.exit(1)

    # 替换为你的图片目录绝对路径
    img_dir = r"/path/to/your/images"
    os.chdir(img_dir)

    for image in os.listdir(img_dir):
        # 过滤非图片文件(可根据需要调整后缀)
        if not image.lower().endswith(('.jpg', '.jpeg', '.png', '.gif')):
            continue

        if check_img_tweeted(image):
            print(f'Already Tweeted {image} before, deleting file')
            os.remove(image)
            continue

        try:
            # 发布带图片的推文,可自定义文本内容
            api.update_status_with_media(status=f'Sharing {image}', filename=image)
            print(f'Successfully tweeted {image}')
            write_tweeted_img(image)
            count += 1
            os.remove(image)

        except TweepError as e:
            print(f'Failed to tweet {image}: {e.response.text}')
            continue

        # 执行等待倒计时
        done = False
        animate_wait()
        done = True

    print(f'\nSuccessfully Tweeted {count} Images')

额外注意事项

  • 确保推特开发者账号已启用API权限,且密钥(API Key、API Secret、Access Token等)正确无误。
  • 遵守推特API的调用频率限制,避免因频繁调用导致账号被限流或封禁。
  • 测试阶段可先使用小号进行测试,避免影响主账号。

内容的提问来源于stack exchange,提问作者LoneWalkerNS

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.08.03 13:01:05