File indexing completed on 2024-04-21 16:10:58
0001 import json 0002 import timeago, datetime 0003 import dateutil.parser 0004 import peertube 0005 from youtube_dl import YoutubeDL 0006 from peertube.rest import ApiException 0007 from mycroft.skills.core import MycroftSkill, intent_handler, intent_file_handler 0008 from mycroft.messagebus.message import Message 0009 from mycroft.util.log import LOG 0010 from xdg import XDG_CACHE_HOME, XDG_DATA_HOME 0011 from json_database import JsonStorage 0012 0013 __author__ = 'aix' 0014 0015 class PeerTubeSkill(MycroftSkill): 0016 def __init__(self): 0017 super(PeerTubeSkill, self).__init__(name="PeerTubeSkill") 0018 self.default_host = "https://peertube.co.uk/api/v1" 0019 self.configured_host = None 0020 cache_path = str(XDG_CACHE_HOME.absolute()) 0021 if cache_path: 0022 self.peer_instance_config = JsonStorage(cache_path + "/peertube-instance-config.conf") 0023 else: 0024 self.peer_instance_config = JsonStorage(str(self.root_dir).rsplit("/", 1)[0] + "peertube-instance-config.conf") 0025 0026 0027 if "model" in self.peer_instance_config: 0028 self.peer_instance_host = self.peer_instance_config["model"] 0029 self.configured_host = self.peer_instance_config["model"]["hosturl"] 0030 else: 0031 self.peer_instance_host = None 0032 0033 if self.configured_host is not None: 0034 self.peer_config = peertube.Configuration( 0035 host = self.configured_host) 0036 else: 0037 self.peer_config = peertube.Configuration( 0038 host = self.default_host) 0039 0040 self.api_client = peertube.ApiClient(configuration=self.peer_config) 0041 self.search_instance = peertube.SearchApi(self.api_client) 0042 self.video_instance = peertube.VideoApi(self.api_client) 0043 0044 def initialize(self): 0045 self.bus.on('peertube-skill.aiix.home', self.display_homepage) 0046 self.gui.register_handler('PeerTube.SearchQuery', self.search_page_query) 0047 self.gui.register_handler('PeerTube.WatchVideo', self.watch_video_via_query) 0048 self.gui.register_handler('PeerTube.SettingsPage', self.peer_settings_page) 0049 self.gui.register_handler('PeerTube.ConfigureHost', self.peer_configure_host) 0050 0051 @intent_file_handler("PtOpenApp.intent") 0052 def display_homepage(self, message): 0053 self.gui.clear() 0054 self.gui.show_page("PeerTubeLoading.qml", override_idle=True, override_animations=True) 0055 self.build_categories() 0056 0057 def search_videos(self, query): 0058 # Global search for queries with PT api 0059 # JSONIFY return results 0060 try: 0061 search_pt = self.search_instance.search_videos_get(query) 0062 search_response = json.loads(json.dumps(search_pt.to_dict(), default=self.dt_converter)) 0063 return search_response 0064 0065 except ApiException as e: 0066 self.speak_dialog("Search.Failed", data={'query': message.data.get("search_query")}) 0067 0068 def search_page_query(self, message): 0069 # Receive query from GUI and return results for search 0070 user_query = message.data.get("search_query") 0071 get_response = self.search_videos(user_query) 0072 self.gui["search_results"] = get_response 0073 self.gui["search_completed"] = True 0074 0075 def build_categories(self): 0076 # 1: Music, 2: Films, 3: Vehicles, 4: Art, 5: Sports, 6: Travels, 7: Gaming, 8: People, 0077 # 9: Comedy, 10: Entertainment, 11: News & Politics, 12: How To, 13: Education, 14: Activism, 0078 # 15: Science & Technology 0079 0080 try: 0081 # Build only five categories 0082 news_category_list = self.video_instance.videos_get(category_one_of=11, count=10) 0083 music_category_list = self.video_instance.videos_get(category_one_of=1, count=10) 0084 entertainment_category_list = self.video_instance.videos_get(category_one_of=10, count=10) 0085 technology_category_list = self.video_instance.videos_get(category_one_of=15, count=10) 0086 gaming_category_list = self.video_instance.videos_get(category_one_of=7, count=10) 0087 0088 LOG.info(json.dumps(news_category_list.to_dict(), default=self.dt_converter)) 0089 0090 self.gui["news_category_list"] = json.loads(json.dumps(news_category_list.to_dict(), default=self.dt_converter)) 0091 self.gui["music_category_list"] = json.loads(json.dumps(music_category_list.to_dict(), default=self.dt_converter)) 0092 self.gui["entertainment_category_list"] = json.loads(json.dumps(entertainment_category_list.to_dict(), default=self.dt_converter)) 0093 self.gui["technology_category_list"] = json.loads(json.dumps(technology_category_list.to_dict(), default=self.dt_converter)) 0094 self.gui["gaming_category_list"] = json.loads(json.dumps(gaming_category_list.to_dict(), default=self.dt_converter)) 0095 self.gui["search_results"] = "" 0096 self.gui["search_completed"] = True 0097 0098 self.gui.remove_page("PeerTubeLoading.qml") 0099 self.gui.show_page("PeerTubeHomePage.qml", override_idle=True, override_animations=True) 0100 0101 except ApiException as e: 0102 self.gui.show_page("PeerTubeErrorPage.qml") 0103 self.speak_dialog("build_category_failed") 0104 0105 def watch_video_via_query(self, message): 0106 # Event invoked by interaction on homepage video tiles 0107 # We already have all the video listing data required 0108 video_idx0_host = message.data["account"]["host"] 0109 video_idx0_embed = message.data["embed_path"] 0110 video_embed_link = "https://" + video_idx0_host + video_idx0_embed 0111 try: 0112 # First run get static video url via youtube-dl as pt video api is unstable 0113 video_stream_link = self.process_stream(video_embed_link) 0114 if video_stream_link == False: 0115 # Try getting the static link manually from embedded object 0116 video_stream_link = str("https://" + video_idx0_host + "/static/webseed/" 0117 + message.data["uuid"] + "-480.mp4") 0118 0119 except ApiException as e: 0120 self.speak_dialog("search_failed") 0121 0122 self.gui["video_meta"] = message.data 0123 self.gui["video_stream"] = video_stream_link 0124 self.gui["video_status"] = "play" 0125 self.gui.show_page("PeerTubePlayer.qml", override_idle=True, override_animations=True) 0126 0127 @intent_file_handler("PtVoiceQuery.intent") 0128 def watch_video_via_voice(self, message): 0129 # Event invoked via voice query, triggered from anywhere 0130 # We don't have the video listing data required 0131 user_query = message.data['query'] 0132 print(user_query) 0133 videos_matched = self.search_videos(user_query) 0134 print(videos_matched) 0135 video_idx0_host = videos_matched['data'][0]["account"]["host"] 0136 video_idx0_embed = videos_matched['data'][0]["embed_path"] 0137 video_embed_link = "https://" + video_idx0_host + video_idx0_embed 0138 try: 0139 # First run get static video url via youtube-dl as pt video api is unstable 0140 video_stream_link = self.process_stream(video_embed_link) 0141 if video_stream_link == False: 0142 # Try getting the static link manually from embedded object 0143 video_stream_link = str("https://" + video_idx0_host + "/static/webseed/" 0144 + videos_matched['data'][0]["uuid"] + "-480.mp4") 0145 0146 except ApiException as e: 0147 self.speak_dialog("search_failed") 0148 0149 self.gui["video_meta"] = videos_matched['data'][0] 0150 self.gui["video_stream"] = video_stream_link 0151 self.gui["video_status"] = "play" 0152 self.gui.show_page("PeerTubePlayer.qml", override_idle=True, override_animations=True) 0153 0154 def peer_settings_page(self, message): 0155 if message.data.get("settings_open") == True: 0156 if self.configured_host is not None: 0157 self.gui["current_instance"] = self.configured_host 0158 else: 0159 self.gui["current_instance"] = self.default_host 0160 0161 instance_local_model = self.list_instances() 0162 self.gui["instances_model"] = instance_local_model 0163 gen_idx = [] 0164 for x in range(len(instance_local_model)): 0165 gen_idx.append(instance_local_model[x]["hosturl"]) 0166 0167 self.gui["instance_string_model"] = gen_idx 0168 self.gui.show_page("PeerTubeSettings.qml", override_idle=True, override_animations=True) 0169 elif message.data.get("settings_open") == False: 0170 self.gui.remove_page("PeerTubeSettings.qml") 0171 0172 def peer_configure_host(self, message): 0173 new_instance = message.data.get('selected_instance') 0174 self.peer_instance_config["model"] = {"hosturl": new_instance} 0175 self.peer_instance_config.store() 0176 self.configured_host = new_instance 0177 self.peer_config = peertube.Configuration( 0178 host = self.configured_host) 0179 self.api_client = peertube.ApiClient(configuration=self.peer_config) 0180 self.search_instance = peertube.SearchApi(self.api_client) 0181 self.video_instance = peertube.VideoApi(self.api_client) 0182 self.clean_built_models() 0183 0184 def process_stream(self, embedded_url): 0185 # First round of processing to extract video stream 0186 ydl = YoutubeDL() 0187 try: 0188 ydl_results = ydl.extract_info(embedded_url, download=False) 0189 return ydl_results['url'] 0190 except: 0191 return False 0192 0193 def dt_converter(self, o): 0194 if isinstance(o, datetime.datetime): 0195 return o.__str__() 0196 0197 def list_instances(self): 0198 instance_listing = [ 0199 {"hostname": "Peertube UK Instance", "hosturl": "https://peertube.co.uk/api/v1"}, 0200 {"hostname": "Peertube Austrian Instance", "hosturl": "https://peertube.at/api/v1"}, 0201 {"hostname": "Peertube Scandinavian Instance", "hosturl": "https://peertube.dk/api/v1"}, 0202 {"hostname": "Peertube European Instance", "hosturl": "https://peertube.be/api/v1"}, 0203 {"hostname": "Peertube Italian Instance", "hosturl": "https://peertube.uno/api/v1"} 0204 ] 0205 return instance_listing 0206 0207 def clean_built_models(self): 0208 self.gui["news_category_list"] = "" 0209 self.gui["music_category_list"] = "" 0210 self.gui["entertainment_category_list"] = "" 0211 self.gui["technology_category_list"] = "" 0212 self.gui["gaming_category_list"] = "" 0213 self.gui["search_results"] = "" 0214 self.gui["search_completed"] = True 0215 self.display_homepage({}) 0216 0217 def create_skill(): 0218 return PeerTubeSkill()