ramforth/pytchat-fork
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Modify termination

Browse Source
This commit is contained in:
taizan-hokuto
2020-05-05 21:16:06 +09:00
parent 84bae4ad2a
commit bd32c75833
1 changed files with 65 additions and 79 deletions
Download Patch File Download Diff File Expand all files Collapse all files

144
pytchat/core_multithread/livechat.py
View File

@@ -11,8 +11,8 @@ from queue import Queue
from .buffer import Buffer from .buffer import Buffer
from ..parser.live import Parser from ..parser.live import Parser
from .. import config from .. import config
from ..exceptions import ChatParseException,IllegalFunctionCall from ..exceptions import ChatParseException, IllegalFunctionCall
from ..paramgen import liveparam, arcparam from ..paramgen import liveparam, arcparam
from ..processors.default.processor import DefaultProcessor from ..processors.default.processor import DefaultProcessor
from ..processors.combinator import Combinator from ..processors.combinator import Combinator
@@ -27,7 +27,7 @@ class LiveChat:
--------- ---------
video_id : str video_id : str
動画ID 動画ID
seektime : int seektime : int
（ライブチャット取得時は無視） （ライブチャット取得時は無視）
取得開始するアーカイブ済みチャットの経過時間(秒） 取得開始するアーカイブ済みチャットの経過時間(秒）
@@ -61,7 +61,7 @@ class LiveChat:
topchat_only : bool topchat_only : bool
Trueの場合、上位チャットのみ取得する。 Trueの場合、上位チャットのみ取得する。
Attributes Attributes
--------- ---------
_executor : ThreadPoolExecutor _executor : ThreadPoolExecutor
@@ -72,22 +72,20 @@ class LiveChat:
''' '''
_setup_finished = False _setup_finished = False
#チャット監視中のListenerのリスト
_listeners = []
def __init__(self, video_id, def __init__(self, video_id,
seektime = 0, seektime=0,
processor = DefaultProcessor(), processor=DefaultProcessor(),
buffer = None, buffer=None,
interruptable = True, interruptable=True,
callback = None, callback=None,
done_callback = None, done_callback=None,
direct_mode = False, direct_mode=False,
force_replay = False, force_replay=False,
topchat_only = False, topchat_only=False,
logger = config.logger(__name__) logger=config.logger(__name__)
): ):
self.video_id = video_id self.video_id = video_id
self.seektime = seektime self.seektime = seektime
if isinstance(processor, tuple): if isinstance(processor, tuple):
self.processor = Combinator(processor) self.processor = Combinator(processor)
@@ -98,57 +96,51 @@ class LiveChat:
self._done_callback = done_callback self._done_callback = done_callback
self._executor = ThreadPoolExecutor(max_workers=2) self._executor = ThreadPoolExecutor(max_workers=2)
self._direct_mode = direct_mode self._direct_mode = direct_mode
self._is_alive = True self._is_alive = True
self._is_replay = force_replay self._is_replay = force_replay
self._parser = Parser(is_replay = self._is_replay) self._parser = Parser(is_replay=self._is_replay)
self._pauser = Queue() self._pauser = Queue()
self._pauser.put_nowait(None) self._pauser.put_nowait(None)
self._setup()
self._first_fetch = True self._first_fetch = True
self._fetch_url = "live_chat/get_live_chat?continuation=" self._fetch_url = "live_chat/get_live_chat?continuation="
self._topchat_only = topchat_only self._topchat_only = topchat_only
self._logger = logger self._logger = logger
LiveChat._logger = logger if interruptable:
if not LiveChat._setup_finished: signal.signal(signal.SIGINT, lambda a, b: self.terminate())
LiveChat._setup_finished = True self._setup()
if interruptable:
signal.signal(signal.SIGINT, (lambda a, b:
(LiveChat.shutdown(None,signal.SIGINT,b))
))
LiveChat._listeners.append(self)
def _setup(self): def _setup(self):
#direct modeがTrueでcallback未設定の場合例外発生。 # direct modeがTrueでcallback未設定の場合例外発生。
if self._direct_mode: if self._direct_mode:
if self._callback is None: if self._callback is None:
raise IllegalFunctionCall( raise IllegalFunctionCall(
"When direct_mode=True, callback parameter is required.") "When direct_mode=True, callback parameter is required.")
else: else:
#direct modeがFalseでbufferが未設定ならばデフォルトのbufferを作成 # direct modeがFalseでbufferが未設定ならばデフォルトのbufferを作成
if self._buffer is None: if self._buffer is None:
self._buffer = Buffer(maxsize = 20) self._buffer = Buffer(maxsize=20)
#callbackが指定されている場合はcallbackを呼ぶループタスクを作成 # callbackが指定されている場合はcallbackを呼ぶループタスクを作成
if self._callback is None: if self._callback is None:
pass pass
else: else:
#callbackを呼ぶループタスクの開始 # callbackを呼ぶループタスクの開始
self._executor.submit(self._callback_loop,self._callback) self._executor.submit(self._callback_loop, self._callback)
#_listenループタスクの開始 # _listenループタスクの開始
listen_task = self._executor.submit(self._startlisten) listen_task = self._executor.submit(self._startlisten)
#add_done_callbackの登録 # add_done_callbackの登録
if self._done_callback is None: if self._done_callback is None:
listen_task.add_done_callback(self.finish) listen_task.add_done_callback(self.finish)
else: else:
listen_task.add_done_callback(self._done_callback) listen_task.add_done_callback(self._done_callback)
def _startlisten(self): def _startlisten(self):
time.sleep(0.1) #sleep shortly to prohibit skipping fetching data time.sleep(0.1) # sleep shortly to prohibit skipping fetching data
"""Fetch first continuation parameter, """Fetch first continuation parameter,
create and start _listen loop. create and start _listen loop.
""" """
initial_continuation = liveparam.getparam(self.video_id,3) initial_continuation = liveparam.getparam(self.video_id, 3)
self._listen(initial_continuation) self._listen(initial_continuation)
def _listen(self, continuation): def _listen(self, continuation):
''' Fetch chat data and store them into buffer, ''' Fetch chat data and store them into buffer,
get next continuaiton parameter and loop. get next continuaiton parameter and loop.
@@ -164,33 +156,34 @@ class LiveChat:
continuation = self._check_pause(continuation) continuation = self._check_pause(continuation)
contents = self._get_contents( contents = self._get_contents(
continuation, session, headers) continuation, session, headers)
metadata, chatdata = self._parser.parse(contents) metadata, chatdata = self._parser.parse(contents)
timeout = metadata['timeoutMs']/1000 timeout = metadata['timeoutMs']/1000
chat_component = { chat_component = {
"video_id" : self.video_id, "video_id": self.video_id,
"timeout" : timeout, "timeout": timeout,
"chatdata" : chatdata "chatdata": chatdata
} }
time_mark =time.time() time_mark = time.time()
if self._direct_mode: if self._direct_mode:
processed_chat = self.processor.process([chat_component]) processed_chat = self.processor.process(
if isinstance(processed_chat,tuple): [chat_component])
if isinstance(processed_chat, tuple):
self._callback(*processed_chat) self._callback(*processed_chat)
else: else:
self._callback(processed_chat) self._callback(processed_chat)
else: else:
self._buffer.put(chat_component) self._buffer.put(chat_component)
diff_time = timeout - (time.time()-time_mark) diff_time = timeout - (time.time()-time_mark)
time.sleep(diff_time if diff_time > 0 else 0) time.sleep(diff_time if diff_time > 0 else 0)
continuation = metadata.get('continuation') continuation = metadata.get('continuation')
except ChatParseException as e: except ChatParseException as e:
self._logger.debug(f"[{self.video_id}]{str(e)}") self._logger.debug(f"[{self.video_id}]{str(e)}")
return return
except (TypeError , json.JSONDecodeError) : except (TypeError, json.JSONDecodeError):
self._logger.error(f"{traceback.format_exc(limit = -1)}") self._logger.error(f"{traceback.format_exc(limit = -1)}")
return return
self._logger.debug(f"[{self.video_id}]finished fetching chat.") self._logger.debug(f"[{self.video_id}]finished fetching chat.")
def _check_pause(self, continuation): def _check_pause(self, continuation):
@@ -202,7 +195,7 @@ class LiveChat:
''' '''
self._pauser.put_nowait(None) self._pauser.put_nowait(None)
if not self._is_replay: if not self._is_replay:
continuation = liveparam.getparam(self.video_id,3) continuation = liveparam.getparam(self.video_id, 3)
return continuation return continuation
def _get_contents(self, continuation, session, headers): def _get_contents(self, continuation, session, headers):
@@ -214,7 +207,7 @@ class LiveChat:
------- -------
'continuationContents' which includes metadata & chat data. 'continuationContents' which includes metadata & chat data.
''' '''
livechat_json = ( livechat_json = (
self._get_livechat_json(continuation, session, headers) self._get_livechat_json(continuation, session, headers)
) )
contents = self._parser.get_contents(livechat_json) contents = self._parser.get_contents(livechat_json)
@@ -225,7 +218,7 @@ class LiveChat:
self._fetch_url = "live_chat_replay/get_live_chat_replay?continuation=" self._fetch_url = "live_chat_replay/get_live_chat_replay?continuation="
continuation = arcparam.getparam( continuation = arcparam.getparam(
self.video_id, self.seektime, self._topchat_only) self.video_id, self.seektime, self._topchat_only)
livechat_json = ( self._get_livechat_json( livechat_json = (self._get_livechat_json(
continuation, session, headers)) continuation, session, headers))
reload_continuation = self._parser.reload_continuation( reload_continuation = self._parser.reload_continuation(
self._parser.get_contents(livechat_json)) self._parser.get_contents(livechat_json))
@@ -244,26 +237,26 @@ class LiveChat:
continuation = urllib.parse.quote(continuation) continuation = urllib.parse.quote(continuation)
livechat_json = None livechat_json = None
status_code = 0 status_code = 0
url =f"https://www.youtube.com/{self._fetch_url}{continuation}&pbj=1" url = f"https://www.youtube.com/{self._fetch_url}{continuation}&pbj=1"
for _ in range(MAX_RETRY + 1): for _ in range(MAX_RETRY + 1):
with session.get(url ,headers = headers) as resp: with session.get(url, headers=headers) as resp:
try: try:
text = resp.text text = resp.text
livechat_json = json.loads(text) livechat_json = json.loads(text)
break break
except json.JSONDecodeError : except json.JSONDecodeError:
time.sleep(1) time.sleep(1)
continue continue
else: else:
self._logger.error(f"[{self.video_id}]" self._logger.error(f"[{self.video_id}]"
f"Exceeded retry count. status_code={status_code}") f"Exceeded retry count. status_code={status_code}")
return None return None
return livechat_json return livechat_json
def _callback_loop(self,callback): def _callback_loop(self, callback):
""" コンストラクタでcallbackを指定している場合、バックグラウンドで """ コンストラクタでcallbackを指定している場合、バックグラウンドで
callbackに指定された関数に一定間隔でチャットデータを投げる。 callbackに指定された関数に一定間隔でチャットデータを投げる。
Parameter Parameter
--------- ---------
callback : func callback : func
@@ -280,13 +273,13 @@ class LiveChat:
def get(self): def get(self):
""" bufferからデータを取り出し、processorに投げ、 """ bufferからデータを取り出し、processorに投げ、
加工済みのチャットデータを返す。 加工済みのチャットデータを返す。
Returns Returns
: Processorによって加工されたチャットデータ : Processorによって加工されたチャットデータ
""" """
if self._callback is None: if self._callback is None:
items = self._buffer.get() items = self._buffer.get()
return self.processor.process(items) return self.processor.process(items)
raise IllegalFunctionCall( raise IllegalFunctionCall(
"既にcallbackを登録済みのため、get()は実行できません。") "既にcallbackを登録済みのため、get()は実行できません。")
@@ -304,13 +297,13 @@ class LiveChat:
return return
if self._pauser.empty(): if self._pauser.empty():
self._pauser.put_nowait(None) self._pauser.put_nowait(None)
def is_alive(self): def is_alive(self):
return self._is_alive return self._is_alive
def finish(self,sender): def finish(self, sender):
'''Listener終了時のコールバック''' '''Listener終了時のコールバック'''
try: try:
self.terminate() self.terminate()
except CancelledError: except CancelledError:
self._logger.debug(f'[{self.video_id}]cancelled:{sender}') self._logger.debug(f'[{self.video_id}]cancelled:{sender}')
@@ -319,14 +312,7 @@ class LiveChat:
''' '''
Listenerを終了する。 Listenerを終了する。
''' '''
self._is_alive = False if self.is_alive():
if self._direct_mode == False: self._is_alive = False
#bufferにダミーオブジェクトを入れてis_alive()を判定させる self._buffer.put({})
self._buffer.put({'chatdata':'','timeout':0}) self._logger.info(f'[{self.video_id}]終了しました')
self._logger.info(f'[{self.video_id}]finished.')
@classmethod
def shutdown(cls, event, sig = None, handler=None):
cls._logger.debug("shutdown...")
for t in LiveChat._listeners:
t._is_alive = False