Liu Zhicong
283 lines
10 KiB
Python
283 lines
10 KiB
Python
from typing import Union
|
|
from fastapi import FastAPI
|
|
from dotenv import load_dotenv
|
|
from googleapiclient.discovery import build
|
|
from llama_index import SimpleDirectoryReader, LLMPredictor, PromptHelper, GPTListIndex, ServiceContext
|
|
from youtube_transcript_api import YouTubeTranscriptApi
|
|
from urllib.parse import urlparse, parse_qs
|
|
import os
|
|
import re
|
|
import requests
|
|
import json
|
|
import logging
|
|
import sys
|
|
from langchain.chat_models import ChatOpenAI
|
|
import tweepy
|
|
import traceback
|
|
import json
|
|
|
|
logging.basicConfig(stream=sys.stdout, level=logging.INFO)
|
|
logging.getLogger().addHandler(logging.StreamHandler(stream=sys.stdout))
|
|
|
|
load_dotenv()
|
|
os.environ['OPENAI_API_KEY'] = os.getenv("JARVIS_OPENAI_API_KEY")
|
|
|
|
youtube_api_key = os.getenv("DEMO_YOUTUBE_API_KEY")
|
|
if youtube_api_key is not None and youtube_api_key != '':
|
|
youtube_client = build('youtube', 'v3', developerKey=youtube_api_key)
|
|
youtube_video_max_result = 3
|
|
youtube_username_set = set()
|
|
|
|
# Configure prompt parameters and initialise helper
|
|
max_input_size = 4096
|
|
num_output = 512
|
|
max_chunk_overlap = 20
|
|
# llama_index
|
|
llm_predictor = LLMPredictor(llm=ChatOpenAI(
|
|
temperature=0, model_name="gpt-3.5-turbo-0301", max_tokens=num_output))
|
|
prompt_helper = PromptHelper(max_input_size, num_output, max_chunk_overlap)
|
|
|
|
# twitter
|
|
twitter_consumer_key = os.getenv("DEMO_TWITTER_CONSUMER_KEY")
|
|
twitter_consumer_secret = os.getenv("DEMO_TWITTER_CONSUMER_SECRET")
|
|
twitter_access_token = os.getenv("DEMO_TWITTER_ACCESS_TOKEN")
|
|
twitter_access_token_secret = os.getenv("DEMO_TWITTER_ACCESS_TOKEN_SECRET")
|
|
twitter_username = os.getenv("DEMO_TWITTER_USERNAME")
|
|
twitter_api = tweepy.Client(
|
|
consumer_key=twitter_consumer_key, consumer_secret=twitter_consumer_secret,
|
|
access_token=twitter_access_token, access_token_secret=twitter_access_token_secret)
|
|
|
|
app = FastAPI()
|
|
|
|
|
|
@app.get(path="/videos/summary",
|
|
summary="retrieve the summary of a youtube video based on its video_id, channel_id, username, and url")
|
|
async def videos_summary(video_id: Union[str, None] = None,
|
|
channel_id: Union[str, None] = None,
|
|
username: Union[str, None] = None,
|
|
url: Union[str, None] = None,
|
|
open_summary: Union[bool, None] = False):
|
|
"""
|
|
- **video_id**: youtube video ID
|
|
- **channel_id**: youtube channel Id
|
|
- **username**: youtube username
|
|
- **url**: youtube url
|
|
"""
|
|
|
|
videos = {}
|
|
if video_id is not None:
|
|
videos = get_video_detail(video_id)
|
|
if channel_id is not None:
|
|
videos = get_channel_videos(channel_id)
|
|
if username is not None:
|
|
videos = get_user_videos(username)
|
|
if url is not None:
|
|
(value_type, value) = get_youtube_value(url)
|
|
if value_type == "username":
|
|
videos = get_user_videos(value)
|
|
if value_type == "video_id":
|
|
videos = get_video_detail(value)
|
|
|
|
return gen_summary_for_videos(videos, open_summary)
|
|
|
|
|
|
@app.post(path="/timer-task/add", summary="add a task to retrieve summary information for the latest video")
|
|
async def timer_task_add(username: Union[str, None] = None,
|
|
url: Union[str, None] = None):
|
|
if username is not None:
|
|
youtube_username_set.add(username)
|
|
if url is not None:
|
|
(value_type, value) = get_youtube_value(url)
|
|
if value_type == "username":
|
|
youtube_username_set.add(value)
|
|
|
|
return youtube_username_set
|
|
|
|
|
|
@app.get(path="/timer-tasks", summary="display the list of current tasks")
|
|
async def timer_task_list():
|
|
return youtube_username_set
|
|
|
|
|
|
@app.post(path="/twitter/tweet_post",
|
|
summary="post a tweet on Twitter: type=3 [success], type=0 [failed due to duplicate tweets], type=4 [other error]")
|
|
def twitter_post_tweet(content: str):
|
|
try:
|
|
response = twitter_api.create_tweet(text=content)
|
|
return {
|
|
"type": 3,
|
|
"msg": "post tweet success",
|
|
"tweet": {
|
|
"id": response.data["id"],
|
|
"text": response.data["text"],
|
|
"url": f"https://twitter.com/{twitter_username}/status/{response.data['id']}"
|
|
}
|
|
}
|
|
except Exception as e:
|
|
if "duplicate content" in str(e):
|
|
return {
|
|
"type": 0,
|
|
"msg": "duplicate tweets"
|
|
}
|
|
else:
|
|
print(e)
|
|
return {
|
|
"type": 4,
|
|
"msg": "other error"
|
|
}
|
|
|
|
|
|
# get a list of youtube videos based on channel_id
|
|
def get_channel_videos(channel_id):
|
|
res = youtube_client.channels().list(
|
|
id=channel_id, part='contentDetails').execute()
|
|
|
|
videos = {}
|
|
|
|
for item in res['items']:
|
|
playlist_id = item["contentDetails"]["relatedPlaylists"]["uploads"]
|
|
res = youtube_client.playlistItems().list(
|
|
playlistId=playlist_id, part='snippet', maxResults=youtube_video_max_result).execute()
|
|
for video in res["items"]:
|
|
video_id = video["snippet"]["resourceId"]["videoId"]
|
|
video["videoId"] = video_id
|
|
videos[video_id] = video
|
|
|
|
return {k: v for k, v in sorted(videos.items(), key=lambda item: item[1]["snippet"]["publishedAt"], reverse=True)}
|
|
|
|
|
|
# retrieve the information of a youtube video based on its video_id
|
|
def get_video_detail(video_id):
|
|
res = youtube_client.videos().list(id=video_id, part='snippet').execute()
|
|
print(res)
|
|
videos = {}
|
|
for video in res["items"]:
|
|
video["videoId"] = video_id
|
|
videos[video_id] = video
|
|
return videos
|
|
|
|
|
|
def get_user_videos(username):
|
|
print(username)
|
|
return get_user_videos_by_channel_id_list(get_channel_id_list_by_url(f"https://www.youtube.com/@{username}"))
|
|
|
|
|
|
# retrieve a list of channel IDs using a youtube url
|
|
def get_channel_id_list_by_url(url):
|
|
try:
|
|
response = requests.get(url)
|
|
response.raise_for_status()
|
|
response.encoding = response.apparent_encoding
|
|
html = response.text
|
|
pattern = r'https://www.youtube.com/channel/(.*?)[?#"\']'
|
|
text_list = re.findall(pattern, html)
|
|
return set(text_list)
|
|
except Exception as e:
|
|
print(f"Error occurred: {e}")
|
|
raise e
|
|
|
|
|
|
# retrieve a list of youtube videos based on a list of channels
|
|
def get_user_videos_by_channel_id_list(channel_id_list: set):
|
|
try:
|
|
user_videos = {}
|
|
for channel_id in channel_id_list:
|
|
print(channel_id)
|
|
user_videos.update(get_channel_videos(channel_id))
|
|
return {k: v for k, v in
|
|
sorted(user_videos.items(), key=lambda item: item[1]["snippet"]["publishedAt"], reverse=True)}
|
|
except Exception as e:
|
|
print(f"Error occurred: {e}")
|
|
raise e
|
|
|
|
|
|
# Generate a summary report using a list of videos
|
|
def gen_summary_for_videos(videos, open_summary):
|
|
return {k: (lambda x: gen_summary_for_video(x, open_summary))(v) for k, v in videos.items()}
|
|
|
|
|
|
# Generate a summary report using a youtube video
|
|
def gen_summary_for_video(video, open_summary):
|
|
video_id = video["videoId"]
|
|
video_title = video["snippet"]["title"]
|
|
video_published_at = video["snippet"]["publishedAt"]
|
|
video_url = f"https://www.youtube.com/watch?v={video_id}"
|
|
summary = ""
|
|
has_caption = None
|
|
if open_summary is not None and open_summary == True:
|
|
try:
|
|
has_caption = False
|
|
(inner_has_caption, dir_name,
|
|
filename) = download_youtube_caption(video_id)
|
|
has_caption = inner_has_caption
|
|
if has_caption:
|
|
summary = gen_summary_for_text(dir_name)
|
|
else:
|
|
summary = ""
|
|
except Exception as e:
|
|
print(e)
|
|
traceback.print_exc()
|
|
print(f"Error occurred: {e}")
|
|
return {"title": video_title, "video_id": video_id, "published_at": video_published_at, "video_url": video_url,
|
|
"has_caption": has_caption, "summary": summary}
|
|
|
|
|
|
# generate a summary report using a directory of documents
|
|
def gen_summary_for_text(dir_name: str):
|
|
documents = SimpleDirectoryReader(dir_name).load_data()
|
|
service_context = ServiceContext.from_defaults(
|
|
llm_predictor=llm_predictor, prompt_helper=prompt_helper, chunk_size_limit=3584)
|
|
index = GPTListIndex.from_documents(
|
|
documents, service_context=service_context)
|
|
query_engine = index.as_query_engine(response_mode="tree_summarize")
|
|
summary = query_engine.query("Summarize the text, using English. ".strip())
|
|
return str(summary)
|
|
|
|
|
|
def download_youtube_caption(video_id: str):
|
|
file_name = "data/" + video_id + "/index.txt"
|
|
if os.path.exists(file_name):
|
|
return True, "data/" + video_id, file_name
|
|
|
|
transcript = YouTubeTranscriptApi.get_transcripts(
|
|
[video_id], languages=['en', 'zh-CN', 'zh', 'zh-Hans', 'zh-Hant'])
|
|
transcript_json_array = json.loads('[]')
|
|
if transcript:
|
|
transcript_json_array = transcript[0][video_id]
|
|
texts = [text['text'] for text in transcript_json_array]
|
|
texts = list(filter(lambda x: len(x.strip()) > 0, texts))
|
|
transcript_text = "\n".join(texts)
|
|
return save_file(transcript_text, video_id)
|
|
|
|
|
|
def save_file(context: str, video_id: str):
|
|
dir_name = "data/" + video_id
|
|
if not os.path.exists(dir_name):
|
|
os.makedirs(dir_name)
|
|
filename = dir_name + "/index.txt"
|
|
if len(context) > 0:
|
|
with open(filename, "w") as file:
|
|
file.write(context)
|
|
return len(context) > 0, dir_name, filename
|
|
|
|
|
|
# retrieve a video ID or channel ID using a url
|
|
def get_youtube_value(url):
|
|
parsed_url = urlparse(url)
|
|
if 'youtube.com' in parsed_url.netloc:
|
|
# check if it contains: youtube.com/@xx
|
|
match = re.search(r'@([\d\w.-]+)\/?', parsed_url.path)
|
|
if match:
|
|
return "username", match.group(1)
|
|
# check if it contains: youtube.com/watch?v=xx
|
|
if '/watch' in parsed_url.path:
|
|
query_params = parse_qs(parsed_url.query)
|
|
if 'v' in query_params:
|
|
return "video_id", query_params['v'][0]
|
|
if 'youtu.be' in parsed_url.netloc:
|
|
# check if it contains: https://youtu.be/xxx
|
|
video_id = parsed_url.path[1:]
|
|
if not "/" in video_id:
|
|
return "video_id", video_id
|
|
return "None", "None"
|