325 lines
13 KiB
Python
325 lines
13 KiB
Python
from fastapi import APIRouter, HTTPException, status
|
||
from typing import Optional
|
||
import os
|
||
import logging
|
||
import requests
|
||
import tweepy
|
||
import json
|
||
from pydantic import BaseModel
|
||
|
||
# 配置日志
|
||
logging.basicConfig(
|
||
level=logging.INFO,
|
||
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
|
||
handlers=[logging.FileHandler("twitter_post_api.log"), logging.StreamHandler()]
|
||
)
|
||
logger = logging.getLogger("TwitterPostAPI")
|
||
|
||
router = APIRouter(
|
||
tags=["twitter_post"],
|
||
responses={404: {"description": "Not found"}},
|
||
)
|
||
|
||
# DeepSeek API配置
|
||
API_KEY = "sk-8a121704a9bc4ec6a5ab0ae16e0bc0ba"
|
||
BASE_URL = "https://api.deepseek.com"
|
||
|
||
# 请求模型
|
||
class TwitterPostRequest(BaseModel):
|
||
username: str
|
||
post_to_twitter: bool = False
|
||
|
||
# 帮助调试问题的简单测试路由
|
||
@router.get("/test")
|
||
async def test_twitter_post_api():
|
||
"""测试Twitter发推API路由是否正常工作"""
|
||
return {"status": "ok", "message": "Twitter发推路由正常工作"}
|
||
|
||
class TwitterPostService:
|
||
def __init__(self):
|
||
"""初始化Twitter发推服务"""
|
||
try:
|
||
# 使用HTTP请求方式调用DeepSeek API
|
||
self.api_key = API_KEY
|
||
self.base_url = BASE_URL
|
||
|
||
# Twitter API凭证
|
||
self.api_key_twitter = "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_twitter,
|
||
consumer_secret=self.api_secret,
|
||
access_token=self.access_token,
|
||
access_token_secret=self.access_secret
|
||
)
|
||
|
||
# 用于获取推文的API
|
||
self.twitter_api_key = "e3dad005b0e54bdc88c6178a89adec13"
|
||
self.twitter_api_url = "https://api.twitterapi.io/twitter/tweet/advanced_search"
|
||
|
||
logger.info("TwitterPostService 初始化成功")
|
||
except Exception as e:
|
||
logger.error(f"初始化TwitterPostService失败: {str(e)}")
|
||
raise
|
||
|
||
def call_deepseek_api(self, messages, model="deepseek-chat"):
|
||
"""使用HTTP请求调用DeepSeek API"""
|
||
try:
|
||
headers = {
|
||
"Authorization": f"Bearer {self.api_key}",
|
||
"Content-Type": "application/json"
|
||
}
|
||
|
||
data = {
|
||
"model": model,
|
||
"messages": messages,
|
||
"temperature": 0.1,
|
||
"stream": False
|
||
}
|
||
|
||
response = requests.post(
|
||
f"{self.base_url}/chat/completions",
|
||
headers=headers,
|
||
json=data,
|
||
timeout=60
|
||
)
|
||
|
||
if response.status_code == 200:
|
||
result = response.json()
|
||
return result["choices"][0]["message"]["content"]
|
||
else:
|
||
logger.error(f"DeepSeek API请求失败: {response.status_code}, {response.text}")
|
||
return None
|
||
|
||
except Exception as e:
|
||
logger.error(f"调用DeepSeek API失败: {e}")
|
||
return None
|
||
|
||
def get_user_tweets(self, username):
|
||
"""获取指定用户的最近推文"""
|
||
username = username.lstrip('@')
|
||
|
||
try:
|
||
logger.info(f"请求Twitter API获取用户 {username} 的推文")
|
||
|
||
# 使用与原始代码相同的请求参数
|
||
url = self.twitter_api_url
|
||
headers = {"X-API-Key": self.twitter_api_key}
|
||
params = {
|
||
"queryType": "Latest",
|
||
"query": f"from:{username}",
|
||
"count": 5 # 获取5条推文
|
||
}
|
||
|
||
# 记录请求详情
|
||
logger.info(f"API请求详情: URL={url}, 参数={params}")
|
||
|
||
# 发送请求
|
||
response = requests.get(url, headers=headers, params=params)
|
||
|
||
# 记录响应状态
|
||
logger.info(f"API响应状态码: {response.status_code}")
|
||
|
||
if response.status_code == 200:
|
||
try:
|
||
data = response.json()
|
||
# 记录API返回的JSON数据以便调试
|
||
logger.info(f"API响应数据: {json.dumps(data)[:500]}")
|
||
|
||
tweets = data.get("tweets", [])
|
||
if tweets:
|
||
tweet_texts = [tweet["text"] for tweet in tweets]
|
||
logger.info(f"成功获取用户 '{username}' 的 {len(tweet_texts)} 条推文")
|
||
return {
|
||
"success": True,
|
||
"tweets": tweet_texts,
|
||
"content": "\n".join(tweet_texts)
|
||
}
|
||
else:
|
||
logger.warning(f"没有找到用户 '{username}' 的推文")
|
||
return {
|
||
"success": False,
|
||
"detail": f"没有找到用户 @{username} 的推文"
|
||
}
|
||
except Exception as e:
|
||
logger.error(f"解析API响应失败: {str(e)}")
|
||
logger.error(f"响应内容: {response.text[:500]}")
|
||
return {
|
||
"success": False,
|
||
"detail": f"解析Twitter API响应失败: {str(e)}"
|
||
}
|
||
else:
|
||
# 记录错误响应内容以便调试
|
||
logger.error(f"Twitter API 请求失败，状态码: {response.status_code}")
|
||
logger.error(f"响应内容: {response.text[:500]}")
|
||
return {
|
||
"success": False,
|
||
"detail": f"Twitter API请求失败，状态码: {response.status_code}"
|
||
}
|
||
|
||
except Exception as e:
|
||
logger.error(f"获取用户推文失败: {str(e)}")
|
||
logger.exception("详细错误信息:")
|
||
return {
|
||
"success": False,
|
||
"detail": f"获取用户推文失败: {str(e)}"
|
||
}
|
||
|
||
def generate_summary(self, tweet_content, username):
|
||
"""根据推文内容生成总结 - 使用HTTP请求方式"""
|
||
try:
|
||
# 记录推文内容长度
|
||
content_length = len(tweet_content) if tweet_content else 0
|
||
logger.info(f"生成摘要: 用户={username}, 推文长度={content_length}")
|
||
|
||
# 如果推文内容为空，返回错误
|
||
if not tweet_content:
|
||
logger.warning(f"推文内容为空，无法生成摘要")
|
||
return f"无法获取用户 @{username} 的推文数据，请稍后再试。"
|
||
|
||
# 计算最大摘要长度
|
||
prefix = f"{username} 的最近推文摘要："
|
||
max_tweet_length = 280
|
||
max_summary_length = max_tweet_length - len(prefix) - 1
|
||
|
||
# 使用HTTP请求调用DeepSeek API生成摘要
|
||
try:
|
||
logger.info("调用DeepSeek API生成摘要")
|
||
|
||
messages = [
|
||
{"role": "system", "content": "You are a helpful assistant that summarizes tweets accurately"},
|
||
{"role": "user", "content": f"请总结用户 @{username} 以下推文的主要内容，必须使用简体中文，总结必须以序号（1.、2.、...）分隔每条要点，每条要点简洁明了且以句号或感叹号结尾，总长度不得超过 {max_summary_length} 字符：\n{tweet_content}"}
|
||
]
|
||
|
||
summary_content = self.call_deepseek_api(messages)
|
||
|
||
if summary_content:
|
||
logger.info(f"成功生成摘要，长度: {len(summary_content)}")
|
||
else:
|
||
logger.error("DeepSeek API调用失败")
|
||
return f"生成摘要失败，请稍后再试。"
|
||
|
||
except Exception as e:
|
||
logger.error(f"DeepSeek API调用失败: {str(e)}")
|
||
logger.exception("详细错误信息:")
|
||
return f"生成摘要失败，请稍后再试。错误: {str(e)[:100]}"
|
||
|
||
# 确保返回的内容不超过限制
|
||
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)
|
||
|
||
logger.info(f"成功为用户 '{username}' 生成摘要")
|
||
return summary
|
||
|
||
except Exception as e:
|
||
logger.error(f"生成摘要失败: {str(e)}")
|
||
logger.exception("详细错误信息:")
|
||
return f"生成摘要失败，请稍后再试。"
|
||
|
||
def post_to_twitter(self, summary, username):
|
||
"""将摘要发布到Twitter"""
|
||
try:
|
||
prefix = f"{username} 的最近推文摘要："
|
||
tweet_text = f"{prefix}\n{summary}"
|
||
|
||
logger.info(f"准备发送推文: {tweet_text[:100]}...")
|
||
logger.info(f"推文长度: {len(tweet_text)} 字符")
|
||
|
||
# 发送推文
|
||
response = self.twitter_client.create_tweet(text=tweet_text)
|
||
|
||
# 获取发送的推文ID
|
||
tweet_id = response.data['id'] if hasattr(response, 'data') and 'id' in response.data else "未知"
|
||
logger.info(f"推文发送成功，ID: {tweet_id}")
|
||
|
||
return {
|
||
"success": True,
|
||
"tweet_id": tweet_id,
|
||
"tweet_url": f"https://twitter.com/user/status/{tweet_id}" if tweet_id != "未知" else None
|
||
}
|
||
except Exception as e:
|
||
logger.error(f"发送推文失败: {str(e)}")
|
||
logger.exception("详细错误信息:")
|
||
return {
|
||
"success": False,
|
||
"detail": f"发送推文失败: {str(e)}"
|
||
}
|
||
|
||
|
||
# 创建TwitterPostService实例
|
||
twitter_post_service = TwitterPostService()
|
||
|
||
|
||
@router.post("/create")
|
||
async def create_twitter_post(request: TwitterPostRequest):
|
||
"""获取Twitter用户最近推文的摘要并可选择发布到Twitter"""
|
||
try:
|
||
username = request.username
|
||
post_to_twitter = request.post_to_twitter
|
||
|
||
logger.info(f"收到Twitter发推请求，用户名: {username}, 是否发送: {post_to_twitter}")
|
||
|
||
# 获取用户推文
|
||
tweets_result = twitter_post_service.get_user_tweets(username)
|
||
logger.info(f"获取推文结果: {json.dumps(tweets_result)[:200] if isinstance(tweets_result, dict) else '未知结果'}")
|
||
|
||
# 如果获取推文失败，直接返回错误信息
|
||
if not tweets_result.get("success", False):
|
||
logger.warning(f"获取推文失败: {tweets_result.get('detail', '未知错误')}")
|
||
return {
|
||
"detail": tweets_result.get("detail", "获取推文失败")
|
||
}
|
||
|
||
# 生成摘要
|
||
tweet_content = tweets_result.get("content", "")
|
||
summary = twitter_post_service.generate_summary(tweet_content, username)
|
||
logger.info(f"生成摘要: 长度={len(summary) if summary else 0}")
|
||
|
||
# 返回结果对象
|
||
result = {
|
||
"username": username,
|
||
"summary": summary,
|
||
"status": "success"
|
||
}
|
||
|
||
# 如果需要发送到Twitter
|
||
if post_to_twitter:
|
||
logger.info(f"准备发送摘要到Twitter...")
|
||
post_result = twitter_post_service.post_to_twitter(summary, username)
|
||
|
||
if post_result.get("success", False):
|
||
result["tweet_posted"] = True
|
||
result["tweet_id"] = post_result.get("tweet_id")
|
||
result["tweet_url"] = post_result.get("tweet_url")
|
||
else:
|
||
result["tweet_posted"] = False
|
||
result["tweet_error"] = post_result.get("detail", "发送推文失败")
|
||
else:
|
||
result["tweet_posted"] = False
|
||
|
||
return result
|
||
|
||
except Exception as e:
|
||
logger.error(f"处理Twitter发推请求失败: {str(e)}")
|
||
logger.exception("详细错误信息:")
|
||
# 与原始main.py中的错误处理一致
|
||
return {
|
||
"detail": f"处理请求失败: {str(e)}"
|
||
} |