Delete unnecessary file
This commit is contained in:
taizan-hokuto
@@ -1,297 +0,0 @@
|
|||||||
import aiohttp, asyncio
|
|
||||||
import datetime
|
|
||||||
import json
|
|
||||||
import random
|
|
||||||
import signal
|
|
||||||
import time
|
|
||||||
import traceback
|
|
||||||
import urllib.parse
|
|
||||||
from aiohttp.client_exceptions import ClientConnectorError
|
|
||||||
from concurrent.futures import CancelledError
|
|
||||||
from asyncio import Queue
|
|
||||||
from .buffer import Buffer
|
|
||||||
from ..parser.live import Parser
|
|
||||||
from .. import config
|
|
||||||
from ..exceptions import ChatParseException,IllegalFunctionCall
|
|
||||||
from ..paramgen import liveparam
|
|
||||||
from ..processors.default.processor import DefaultProcessor
|
|
||||||
from ..processors.combinator import Combinator
|
|
||||||
|
|
||||||
logger = config.logger(__name__)
|
|
||||||
headers = config.headers
|
|
||||||
MAX_RETRY = 10
|
|
||||||
|
|
||||||
|
|
||||||
class LiveChatAsync:
|
|
||||||
'''asyncio(aiohttp)を利用してYouTubeのライブ配信のチャットデータを取得する。
|
|
||||||
|
|
||||||
Parameter
|
|
||||||
---------
|
|
||||||
video_id : str
|
|
||||||
動画ID
|
|
||||||
|
|
||||||
processor : ChatProcessor
|
|
||||||
チャットデータを加工するオブジェクト
|
|
||||||
|
|
||||||
buffer : Buffer(maxsize:20[default])
|
|
||||||
チャットデータchat_componentを格納するバッファ。
|
|
||||||
maxsize : 格納できるchat_componentの個数
|
|
||||||
default値20個。1個で約5~10秒分。
|
|
||||||
|
|
||||||
interruptable : bool
|
|
||||||
Ctrl+Cによる処理中断を行うかどうか。
|
|
||||||
|
|
||||||
callback : func
|
|
||||||
_listen()関数から一定間隔で自動的に呼びだす関数。
|
|
||||||
|
|
||||||
done_callback : func
|
|
||||||
listener終了時に呼び出すコールバック。
|
|
||||||
|
|
||||||
exception_handler : func
|
|
||||||
例外を処理する関数
|
|
||||||
|
|
||||||
direct_mode : bool
|
|
||||||
Trueの場合、bufferを使わずにcallbackを呼ぶ。
|
|
||||||
Trueの場合、callbackの設定が必須
|
|
||||||
(設定していない場合IllegalFunctionCall例外を発生させる）
|
|
||||||
|
|
||||||
Attributes
|
|
||||||
---------
|
|
||||||
_is_alive : bool
|
|
||||||
チャット取得を停止するためのフラグ
|
|
||||||
'''
|
|
||||||
|
|
||||||
_setup_finished = False
|
|
||||||
|
|
||||||
def __init__(self, video_id,
|
|
||||||
seektime = 0,
|
|
||||||
processor = DefaultProcessor(),
|
|
||||||
buffer = None,
|
|
||||||
interruptable = True,
|
|
||||||
callback = None,
|
|
||||||
done_callback = None,
|
|
||||||
exception_handler = None,
|
|
||||||
direct_mode = False):
|
|
||||||
self.video_id = video_id
|
|
||||||
self.seektime = seektime
|
|
||||||
if isinstance(processor, tuple):
|
|
||||||
self.processor = Combinator(processor)
|
|
||||||
else:
|
|
||||||
self.processor = processor
|
|
||||||
self._buffer = buffer
|
|
||||||
self._callback = callback
|
|
||||||
self._done_callback = done_callback
|
|
||||||
self._exception_handler = exception_handler
|
|
||||||
self._direct_mode = direct_mode
|
|
||||||
self._is_alive = True
|
|
||||||
self._parser = Parser()
|
|
||||||
self._pauser = Queue()
|
|
||||||
self._pauser.put_nowait(None)
|
|
||||||
self._setup()
|
|
||||||
|
|
||||||
if not LiveChatAsync._setup_finished:
|
|
||||||
LiveChatAsync._setup_finished = True
|
|
||||||
if exception_handler == None:
|
|
||||||
self._set_exception_handler(self._handle_exception)
|
|
||||||
else:
|
|
||||||
self._set_exception_handler(exception_handler)
|
|
||||||
if interruptable:
|
|
||||||
signal.signal(signal.SIGINT,
|
|
||||||
(lambda a, b:asyncio.create_task(
|
|
||||||
LiveChatAsync.shutdown(None,signal.SIGINT,b))
|
|
||||||
))
|
|
||||||
|
|
||||||
def _setup(self):
|
|
||||||
#direct modeがTrueでcallback未設定の場合例外発生。
|
|
||||||
if self._direct_mode:
|
|
||||||
if self._callback is None:
|
|
||||||
raise IllegalFunctionCall(
|
|
||||||
"direct_mode=Trueの場合callbackの設定が必須です。")
|
|
||||||
else:
|
|
||||||
#direct modeがFalseでbufferが未設定ならばデフォルトのbufferを作成
|
|
||||||
if self._buffer is None:
|
|
||||||
self._buffer = Buffer(maxsize = 20)
|
|
||||||
#callbackが指定されている場合はcallbackを呼ぶループタスクを作成
|
|
||||||
if self._callback is None:
|
|
||||||
pass
|
|
||||||
else:
|
|
||||||
#callbackを呼ぶループタスクの開始
|
|
||||||
loop = asyncio.get_event_loop()
|
|
||||||
loop.create_task(self._callback_loop(self._callback))
|
|
||||||
#_listenループタスクの開始
|
|
||||||
loop = asyncio.get_event_loop()
|
|
||||||
listen_task = loop.create_task(self._startlisten())
|
|
||||||
#add_done_callbackの登録
|
|
||||||
if self._done_callback is None:
|
|
||||||
listen_task.add_done_callback(self.finish)
|
|
||||||
else:
|
|
||||||
listen_task.add_done_callback(self._done_callback)
|
|
||||||
|
|
||||||
async def _startlisten(self):
|
|
||||||
"""最初のcontinuationパラメータを取得し、
|
|
||||||
_listenループのタスクを作成し開始する
|
|
||||||
"""
|
|
||||||
initial_continuation = liveparam.getparam(self.video_id,3)
|
|
||||||
await self._listen(initial_continuation)
|
|
||||||
|
|
||||||
async def _listen(self, continuation):
|
|
||||||
''' continuationに紐付いたチャットデータを取得し
|
|
||||||
Bufferにチャットデータを格納、
|
|
||||||
次のcontinuaitonを取得してループする。
|
|
||||||
|
|
||||||
Parameter
|
|
||||||
---------
|
|
||||||
continuation : str
|
|
||||||
次のチャットデータ取得に必要なパラメータ
|
|
||||||
'''
|
|
||||||
try:
|
|
||||||
async with aiohttp.ClientSession() as session:
|
|
||||||
while(continuation and self._is_alive):
|
|
||||||
if self._pauser.empty():
|
|
||||||
'''pause'''
|
|
||||||
await self._pauser.get()
|
|
||||||
'''resume:
|
|
||||||
prohibit from blocking by putting None into _pauser.
|
|
||||||
'''
|
|
||||||
self._pauser.put_nowait(None)
|
|
||||||
continuation= liveparam.getparam(self.video_id,3)
|
|
||||||
livechat_json = (await
|
|
||||||
self._get_livechat_json(continuation, session, headers)
|
|
||||||
)
|
|
||||||
metadata, chatdata = self._parser.parse( livechat_json )
|
|
||||||
timeout = metadata['timeoutMs']/1000
|
|
||||||
chat_component = {
|
|
||||||
"video_id" : self.video_id,
|
|
||||||
"timeout" : timeout,
|
|
||||||
"chatdata" : chatdata
|
|
||||||
}
|
|
||||||
time_mark =time.time()
|
|
||||||
if self._direct_mode:
|
|
||||||
await self._callback(
|
|
||||||
self.processor.process([chat_component])
|
|
||||||
)
|
|
||||||
else:
|
|
||||||
await self._buffer.put(chat_component)
|
|
||||||
diff_time = timeout - (time.time()-time_mark)
|
|
||||||
await asyncio.sleep(diff_time)
|
|
||||||
continuation = metadata.get('continuation')
|
|
||||||
except ChatParseException as e:
|
|
||||||
self.terminate()
|
|
||||||
logger.error(f"{str(e)}（video_id:\"{self.video_id}\"）")
|
|
||||||
return
|
|
||||||
except (TypeError , json.JSONDecodeError) :
|
|
||||||
self.terminate()
|
|
||||||
logger.error(f"{traceback.format_exc(limit = -1)}")
|
|
||||||
return
|
|
||||||
|
|
||||||
logger.debug(f"[{self.video_id}]チャット取得を終了しました。")
|
|
||||||
self.terminate()
|
|
||||||
|
|
||||||
async def _get_livechat_json(self, continuation, session, headers):
|
|
||||||
'''
|
|
||||||
チャットデータが格納されたjsonデータを取得する。
|
|
||||||
'''
|
|
||||||
continuation = urllib.parse.quote(continuation)
|
|
||||||
livechat_json = None
|
|
||||||
status_code = 0
|
|
||||||
url =(
|
|
||||||
f"https://www.youtube.com/live_chat/get_live_chat?"
|
|
||||||
f"continuation={continuation}&pbj=1")
|
|
||||||
for _ in range(MAX_RETRY + 1):
|
|
||||||
async with session.get(url ,headers = headers) as resp:
|
|
||||||
try:
|
|
||||||
text = await resp.text()
|
|
||||||
status_code = resp.status
|
|
||||||
livechat_json = json.loads(text)
|
|
||||||
break
|
|
||||||
except (ClientConnectorError,json.JSONDecodeError) :
|
|
||||||
await asyncio.sleep(1)
|
|
||||||
continue
|
|
||||||
else:
|
|
||||||
logger.error(f"[{self.video_id}]"
|
|
||||||
f"Exceeded retry count. status_code={status_code}")
|
|
||||||
return None
|
|
||||||
return livechat_json
|
|
||||||
|
|
||||||
async def _callback_loop(self,callback):
|
|
||||||
""" コンストラクタでcallbackを指定している場合、バックグラウンドで
|
|
||||||
callbackに指定された関数に一定間隔でチャットデータを投げる。
|
|
||||||
|
|
||||||
Parameter
|
|
||||||
---------
|
|
||||||
callback : func
|
|
||||||
加工済みのチャットデータを渡す先の関数。
|
|
||||||
"""
|
|
||||||
while self.is_alive():
|
|
||||||
items = await self._buffer.get()
|
|
||||||
data = self.processor.process(items)
|
|
||||||
await callback(data)
|
|
||||||
|
|
||||||
async def get(self):
|
|
||||||
""" bufferからデータを取り出し、processorに投げ、
|
|
||||||
加工済みのチャットデータを返す。
|
|
||||||
|
|
||||||
Returns
|
|
||||||
: Processorによって加工されたチャットデータ
|
|
||||||
"""
|
|
||||||
if self._callback is None:
|
|
||||||
items = await self._buffer.get()
|
|
||||||
return self.processor.process(items)
|
|
||||||
raise IllegalFunctionCall(
|
|
||||||
"既にcallbackを登録済みのため、get()は実行できません。")
|
|
||||||
|
|
||||||
def pause(self):
|
|
||||||
if self._callback is None:
|
|
||||||
return
|
|
||||||
if not self._pauser.empty():
|
|
||||||
self._pauser.get_nowait()
|
|
||||||
|
|
||||||
def resume(self):
|
|
||||||
if self._callback is None:
|
|
||||||
return
|
|
||||||
if self._pauser.empty():
|
|
||||||
self._pauser.put_nowait(None)
|
|
||||||
|
|
||||||
def is_alive(self):
|
|
||||||
return self._is_alive
|
|
||||||
|
|
||||||
def finish(self,sender):
|
|
||||||
'''Listener終了時のコールバック'''
|
|
||||||
try:
|
|
||||||
self.terminate()
|
|
||||||
except CancelledError:
|
|
||||||
logger.debug(f'[{self.video_id}]cancelled:{sender}')
|
|
||||||
|
|
||||||
def terminate(self):
|
|
||||||
'''
|
|
||||||
Listenerを終了する。
|
|
||||||
'''
|
|
||||||
self._is_alive = False
|
|
||||||
if self._direct_mode == False:
|
|
||||||
#bufferにダミーオブジェクトを入れてis_alive()を判定させる
|
|
||||||
self._buffer.put_nowait({'chatdata':'','timeout':1})
|
|
||||||
logger.info(f'[{self.video_id}]終了しました')
|
|
||||||
|
|
||||||
@classmethod
|
|
||||||
def _set_exception_handler(cls, handler):
|
|
||||||
loop = asyncio.get_event_loop()
|
|
||||||
loop.set_exception_handler(handler)
|
|
||||||
|
|
||||||
@classmethod
|
|
||||||
def _handle_exception(cls, loop, context):
|
|
||||||
if not isinstance(context["exception"],CancelledError):
|
|
||||||
logger.error(f"Caught exception: {context}")
|
|
||||||
loop= asyncio.get_event_loop()
|
|
||||||
loop.create_task(cls.shutdown(None,None,None))
|
|
||||||
|
|
||||||
@classmethod
|
|
||||||
async def shutdown(cls, event, sig = None, handler=None):
|
|
||||||
logger.debug("シャットダウンしています")
|
|
||||||
tasks = [t for t in asyncio.all_tasks() if t is not
|
|
||||||
asyncio.current_task()]
|
|
||||||
[task.cancel() for task in tasks]
|
|
||||||
|
|
||||||
logger.debug(f"残っているタスクを終了しています")
|
|
||||||
await asyncio.gather(*tasks,return_exceptions=True)
|
|
||||||
loop = asyncio.get_event_loop()
|
|
||||||
loop.stop()
|
|
||||||
Reference in New Issue
Block a user