125 lines
4.9 KiB
Python
125 lines
4.9 KiB
Python
from dotenv import load_dotenv
|
||
load_dotenv() # 加载 .env 文件中的环境变量
|
||
import tweepy
|
||
import schedule
|
||
import time
|
||
import os
|
||
import logging
|
||
|
||
# 配置日志
|
||
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# 从环境变量读取密钥（安全实践）
|
||
consumer_key = os.environ.get('TWITTER_CONSUMER_KEY')
|
||
consumer_secret = os.environ.get('TWITTER_CONSUMER_SECRET')
|
||
access_token = os.environ.get('TWITTER_ACCESS_TOKEN')
|
||
access_token_secret = os.environ.get('TWITTER_ACCESS_TOKEN_SECRET')
|
||
bearer_token = os.environ.get('TWITTER_BEARER_TOKEN')
|
||
|
||
if not all([consumer_key, consumer_secret, access_token, access_token_secret, bearer_token]):
|
||
raise ValueError("Missing Twitter API keys in environment variables.")
|
||
|
||
# 创建 OAuth1 用户认证（用于 v1.1 端点，如媒体上传）
|
||
auth = tweepy.OAuth1UserHandler(consumer_key, consumer_secret, access_token, access_token_secret)
|
||
api_v1 = tweepy.API(auth, wait_on_rate_limit=True)
|
||
|
||
# 创建 Client（用于 v2 端点，如发帖）
|
||
client = tweepy.Client(
|
||
bearer_token=bearer_token,
|
||
consumer_key=consumer_key,
|
||
consumer_secret=consumer_secret,
|
||
access_token=access_token,
|
||
access_token_secret=access_token_secret,
|
||
wait_on_rate_limit=True
|
||
)
|
||
|
||
def translate_text(text, target_lang='en'):
|
||
"""
|
||
简单文本自动翻译模拟（从中文到英语）。
|
||
这里使用字符串替换作为示例；实际可替换为 API 调用（如 Google Translate）或库（如 transformers）。
|
||
"""
|
||
# 示例翻译映射（扩展这个字典以支持更多）
|
||
translations = {
|
||
'每日自动问候！当前时间:': 'Daily auto greeting! Current time:',
|
||
'每日自动带图问候！当前时间:': 'Daily auto greeting with image! Current time:',
|
||
'这是一条自动发帖测试。': 'This is an auto-post test.'
|
||
}
|
||
for cn, en in translations.items():
|
||
if cn in text:
|
||
text = text.replace(cn, en)
|
||
return text
|
||
|
||
def upload_media(image_paths, max_retries=3):
|
||
"""上传媒体文件，支持重试"""
|
||
media_ids = []
|
||
for image_path in image_paths[:4]: # 限制最多4张
|
||
if not os.path.exists(image_path):
|
||
logger.error(f"图片文件不存在: {image_path}")
|
||
continue
|
||
for attempt in range(max_retries):
|
||
try:
|
||
media = api_v1.media_upload(filename=image_path)
|
||
media_ids.append(media.media_id)
|
||
logger.info(f"图片上传成功: {image_path} (Media ID: {media.media_id})")
|
||
break
|
||
except Exception as e:
|
||
logger.warning(f"图片上传失败 {image_path} (尝试 {attempt+1}/{max_retries}): {e}")
|
||
if attempt == max_retries - 1:
|
||
return [] # 失败后返回空列表
|
||
return media_ids
|
||
|
||
def post_tweet(text, image_paths=None):
|
||
"""
|
||
通用发帖函数：支持纯文本或带图片。
|
||
自动翻译文本。
|
||
"""
|
||
text = translate_text(text) # 自动翻译
|
||
media_ids = []
|
||
|
||
if image_paths:
|
||
media_ids = upload_media(image_paths)
|
||
if not media_ids:
|
||
logger.error("没有图片上传成功，无法发帖")
|
||
return
|
||
|
||
try:
|
||
response = client.create_tweet(text=text, media_ids=media_ids if media_ids else None)
|
||
tweet_id = response.data['id']
|
||
logger.info(f"发帖成功！帖子 ID: {tweet_id}")
|
||
logger.info(f"链接: https://x.com/user/status/{tweet_id}")
|
||
except tweepy.TweepyException as e:
|
||
logger.error(f"发帖失败: {e}")
|
||
|
||
def scheduled_post(with_images=False):
|
||
"""定时任务函数"""
|
||
timestamp = time.strftime('%Y-%m-%d %H:%M:%S')
|
||
if with_images:
|
||
message = f"每日自动带图问候！当前时间: {timestamp}"
|
||
image_paths = ["path/to/your/daily_image.jpg"] # 可动态替换（如从文件夹随机选）
|
||
post_tweet(message, image_paths)
|
||
else:
|
||
message = f"每日自动问候！当前时间: {timestamp}"
|
||
post_tweet(message)
|
||
|
||
if __name__ == "__main__":
|
||
import argparse
|
||
parser = argparse.ArgumentParser(description="Twitter Auto Poster")
|
||
parser.add_argument('--schedule', action='store_true', help="Run in scheduled mode")
|
||
args = parser.parse_args()
|
||
|
||
if args.schedule:
|
||
# 设置定时任务：每天9:00发纯文本帖（可添加带图任务）
|
||
schedule.every().day.at("09:00").do(scheduled_post, with_images=False)
|
||
# 示例：每天10:00发带图帖
|
||
# schedule.every().day.at("10:00").do(scheduled_post, with_images=True)
|
||
|
||
logger.info("定时发帖程序已启动...")
|
||
while True:
|
||
schedule.run_pending()
|
||
time.sleep(60) # 每分钟检查一次
|
||
else:
|
||
# 立即执行示例
|
||
message = "Hello from Tweepy! 😊 date: 2026-01-05"
|
||
image_paths = ["path/to/your/image1.jpg", "path/to/your/image2.png"]
|
||
post_tweet(message, image_paths) |