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()