Adjust the directory structure to prepare for merging into Master.
This commit is contained in:
Liu Zhicong
@@ -0,0 +1 @@
|
||||
data
|
||||
@@ -0,0 +1,282 @@
|
||||
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"
|
||||
Reference in New Issue
Block a user