195 lines
8.4 KiB
Python
195 lines
8.4 KiB
Python
import os
|
||
from openai import OpenAI
|
||
import tweepy
|
||
import logging
|
||
import requests
|
||
|
||
# 配置日志
|
||
logging.basicConfig(
|
||
level=logging.INFO,
|
||
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
|
||
handlers=[logging.FileHandler("../../../../twitter_user_bot.log"), logging.StreamHandler()]
|
||
)
|
||
logger = logging.getLogger("TwitterUserBot")
|
||
|
||
class TwitterUserBot:
|
||
def __init__(self):
|
||
"""初始化Twitter和用户推文总结机器人"""
|
||
try:
|
||
# 初始化OpenAI客户端
|
||
self.openai_client = OpenAI(
|
||
api_key="sk-8a121704a9bc4ec6a5ab0ae16e0bc0ba", # 替换为DeepSeek的API密钥
|
||
base_url="https://api.deepseek.com"
|
||
)
|
||
|
||
# Twitter API凭证
|
||
self.api_key = "3nt1jN4VvqUaaXGHv9AN5VsTV"
|
||
self.api_secret = "M2io73S7TzitFiBw825QIq8atyZRljbIDQuTpH39uFZanQ4XFh"
|
||
self.access_token = "1944636908-prxfjL6OIb56BQjuFTdChrUPh81OjmBbV7pfnWw"
|
||
self.access_secret = "D5AdCVRvIhGEmTmXA8hL5ciAUxIqNMZ3K3B3YejpqqNKj"
|
||
self.bearer_token = "AAAAAAAAAAAAAAAAAAAAAGOd0gEAAAAALtv%2BzLsfGLLa5ydUt60ci6J5ce0%3DMBXViJ1NLY4XYdeuMq1xZQ98kHbeGK5lAJoV2j7Ssmcafk8Skn"
|
||
|
||
# 初始化Twitter客户端（用于发送）
|
||
self.twitter_client = tweepy.Client(
|
||
bearer_token=self.bearer_token,
|
||
consumer_key=self.api_key,
|
||
consumer_secret=self.api_secret,
|
||
access_token=self.access_token,
|
||
access_token_secret=self.access_secret
|
||
)
|
||
|
||
# 调试信息
|
||
logger.info("TwitterUserBot 初始化成功")
|
||
logger.info(f"Twitter API密钥长度: {len(self.api_key)}")
|
||
logger.info(f"Twitter API秘钥长度: {len(self.api_secret)}")
|
||
logger.info(f"Twitter Access Token长度: {len(self.access_token)}")
|
||
logger.info(f"Twitter Access Secret长度: {len(self.access_secret)}")
|
||
except Exception as e:
|
||
logger.error(f"初始化失败: {str(e)}")
|
||
raise
|
||
|
||
def get_user_tweets(self, username):
|
||
"""获取指定用户的最近5条推文，使用 twitterapi.io 的 advanced_search 端点"""
|
||
username = username.lstrip('@')
|
||
|
||
url = "https://api.twitterapi.io/twitter/tweet/advanced_search"
|
||
headers = {"X-API-Key": "e3dad005b0e54bdc88c6178a89adec13"}
|
||
params = {
|
||
"queryType": "Latest",
|
||
"query": f"from:{username}",
|
||
"count": 5
|
||
}
|
||
try:
|
||
logger.info(f"请求URL: {url}")
|
||
logger.info(f"请求头: {headers}")
|
||
logger.info(f"请求参数: {params}")
|
||
|
||
response = requests.get(url, headers=headers, params=params)
|
||
logger.info(f"响应状态码: {response.status_code}")
|
||
|
||
if response.status_code == 200:
|
||
data = response.json()
|
||
# 优化日志输出，只打印关键信息
|
||
tweets = data.get("tweets", [])
|
||
if tweets:
|
||
for tweet in tweets:
|
||
tweet_id = tweet.get("id")
|
||
created_at = tweet.get("createdAt")
|
||
text = tweet.get("text", "")[:100]
|
||
logger.info(f"推文ID: {tweet_id}，时间: {created_at}，内容: {text}...")
|
||
tweet_texts = [tweet["text"] for tweet in tweets]
|
||
logger.info(f"成功获取用户 '{username}' 的 {len(tweet_texts)} 条推文")
|
||
return "\n".join(tweet_texts)
|
||
else:
|
||
logger.warning(f"没有找到用户 '{username}' 的推文")
|
||
return None
|
||
else:
|
||
logger.error(f"API 请求失败，状态码: {response.status_code}")
|
||
try:
|
||
error_content = response.text
|
||
logger.error(f"错误内容: {error_content}")
|
||
except:
|
||
pass
|
||
return None
|
||
except Exception as e:
|
||
logger.error(f"获取用户推文失败: {str(e)}")
|
||
return None
|
||
|
||
def generate_summary(self, tweet_content, username):
|
||
"""根据推文内容生成总结，以序号分隔并控制长度，删除不完整的要点，使用简体字"""
|
||
try:
|
||
# 计算最大摘要长度
|
||
prefix = f"{username} 的最近推文摘要："
|
||
max_tweet_length = 280
|
||
max_summary_length = max_tweet_length - len(prefix) - 1
|
||
|
||
# 要求 AI 以序号分隔格式生成摘要，使用简体字，并控制长度
|
||
completion = self.openai_client.chat.completions.create(
|
||
model="deepseek-chat",
|
||
messages=[
|
||
{"role": "system", "content": "You are a helpful assistant"},
|
||
{"role": "user",
|
||
"content": f"请总结用户 @{username} 以下5条推文的主要内容，必须使用简体中文，总结必须以序号（1.、2.、...）分隔每条要点，每条要点简洁明了且以句号或感叹号结尾，总长度不得超过 {max_summary_length} 字符（包括标点、空格和换行符），不要添加字数统计：\n{tweet_content}"}
|
||
],
|
||
stream=False
|
||
)
|
||
|
||
# 直接获取完整内容
|
||
summary_content = completion.choices[0].message.content if hasattr(completion.choices[0], 'message') else ''
|
||
|
||
# 确保返回的内容不超过限制
|
||
summary = summary_content.strip()
|
||
if len(summary) > max_summary_length:
|
||
summary = summary[:max_summary_length]
|
||
|
||
# 检查并删除不完整的最后一条要点
|
||
lines = summary.split('\n')
|
||
if lines:
|
||
last_line = lines[-1]
|
||
# 检查最后一条是否以句号、感叹号或问号结尾
|
||
if not last_line.endswith(('。', '！', '？', '.')):
|
||
# 如果不完整，删除最后一条
|
||
lines.pop()
|
||
summary = '\n'.join(lines)
|
||
|
||
return summary
|
||
except Exception as e:
|
||
logger.error(f"生成总结失败: {str(e)}")
|
||
return None
|
||
|
||
def send_tweet(self, summary, username):
|
||
"""发送推文"""
|
||
try:
|
||
prefix = f"{username} 的最近推文摘要："
|
||
text = f"{prefix}\n{summary}"
|
||
|
||
logger.info(f"准备发送推文: {text}")
|
||
logger.info(f"推文长度: {len(text)} 字符")
|
||
|
||
response = self.twitter_client.create_tweet(text=text)
|
||
tweet_id = response.data['id'] if hasattr(response, 'data') and 'id' in response.data else "未知"
|
||
logger.info(f"推文发送成功，ID: {tweet_id}")
|
||
return True
|
||
except Exception as e:
|
||
logger.error(f"发送推文失败: {str(e)}")
|
||
return False
|
||
|
||
def run(self):
|
||
"""运行机器人：获取用户推文，生成总结并发送"""
|
||
try:
|
||
username = input("请输入用户名（不含@符号）：")
|
||
logger.info(f"接收到用户名: {username}")
|
||
|
||
tweet_content = self.get_user_tweets(username)
|
||
if not tweet_content:
|
||
logger.error("无法获取用户推文，退出")
|
||
print(f"未找到用户 @{username} 的推文，程序退出。")
|
||
return
|
||
|
||
logger.info(f"获取到的用户推文内容: \n{tweet_content}")
|
||
print(f"成功获取 @{username} 的推文内容")
|
||
|
||
print(f"正在使用AI对推文内容进行总结...")
|
||
summary = self.generate_summary(tweet_content, username)
|
||
if not summary:
|
||
logger.error("无法生成总结，退出")
|
||
print("生成总结失败，程序退出。")
|
||
return
|
||
|
||
logger.info(f"生成总结: {summary}")
|
||
print(f"总结生成成功: {summary}")
|
||
|
||
print("正在发送总结到Twitter...")
|
||
success = self.send_tweet(summary, username)
|
||
if success:
|
||
print(f"关于 @{username} 的推文摘要已成功发送！")
|
||
else:
|
||
print("发送推文失败，请检查日志。")
|
||
except Exception as e:
|
||
logger.error(f"运行时出错: {str(e)}")
|
||
print(f"发生错误: {str(e)}")
|
||
print("更多详细信息请查看日志。")
|
||
|
||
if __name__ == "__main__":
|
||
bot = TwitterUserBot()
|
||
bot.run() |