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

Merge branch 'release/v0.0.6.4'

Browse Source
This commit is contained in:
taizan-hokuto
2020-03-08 23:43:00 +09:00
parent 36e0fd5c54 44dffc7650
commit 316fc5594a
16 changed files with 342 additions and 651 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
README.md
View File

@@ -22,10 +22,23 @@ For more detailed information, see [wiki](https://github.com/taizan-hokuto/pytch
```python
pip install pytchat
```
## Demo
![demo](https://taizan-hokuto.github.io/statics/demo.gif "demo")
## Examples
### CLI
One-liner command.
Save chat data to html.
```bash
$ pytchat -v ZJ6Q4U_Vg6s -o "c:/temp/"
# options:
# -v : video_id
# -o : output directory (default path: './')
# saved filename is [video_id].html
```
### on-demand mode
```python
from pytchat import LiveChat
@@ -263,6 +276,15 @@ Structure of author object.
[![MIT License](http://img.shields.io/badge/license-MIT-blue.svg?style=flat)](LICENSE)
## Contributes
Great thanks:
Most of source code of CLI refer to:
[PetterKraabol / Twitch-Chat-Downloader](https://github.com/PetterKraabol/Twitch-Chat-Downloader)
## Author
[taizan-hokuto](https://github.com/taizan-hokuto)

5
pytchat/__init__.py
View File

@@ -2,7 +2,7 @@
pytchat is a python library for fetching youtube live chat without using yt api, Selenium, or BeautifulSoup.
"""
__copyright__ = 'Copyright (C) 2019 taizan-hokuto'
__version__ = '0.0.6.3'
__version__ = '0.0.6.4'
__license__ = 'MIT'
__author__ = 'taizan-hokuto'
__author_email__ = '55448286+taizan-hokuto@users.noreply.github.com'
@@ -11,6 +11,7 @@ __url__ = 'https://github.com/taizan-hokuto/pytchat'
__all__ = ["core_async","core_multithread","processors"]
from .api import (
cli,
config,
LiveChat,
LiveChatAsync,
@@ -19,6 +20,8 @@ from .api import (
DummyProcessor,
DefaultProcessor,
Extractor,
HTMLArchiver,
TSVArchiver,
JsonfileArchiver,
SimpleDisplayProcessor,
SpeedCalculator,

3
pytchat/api.py
View File

@@ -1,3 +1,4 @@
from . import cli
from . import config
from .core_multithread.livechat import LiveChat
from .core_async.livechat import LiveChatAsync
@@ -5,6 +6,8 @@ from .processors.chat_processor import ChatProcessor
from .processors.compatible.processor import CompatibleProcessor
from .processors.default.processor import DefaultProcessor
from .processors.dummy_processor import DummyProcessor
from .processors.html_archiver import HTMLArchiver
from .processors.tsv_archiver import TSVArchiver
from .processors.jsonfile_archiver import JsonfileArchiver
from .processors.simple_display_processor import SimpleDisplayProcessor
from .processors.speed.calculator import SpeedCalculator

51
pytchat/cli/__init__.py Normal file
View File

@@ -0,0 +1,51 @@
import argparse
import os
from pathlib import Path
from typing import List, Callable
from .arguments import Arguments
from .. exceptions import InvalidVideoIdException, NoContentsException
from .. processors.tsv_archiver import TSVArchiver
from .. processors.html_archiver import HTMLArchiver
from .. tool.extract.extractor import Extractor
from .. tool.videoinfo import VideoInfo
from .. import __version__
'''
Most of CLI modules refer to
Petter Kraabøl's Twitch-Chat-Downloader
https://github.com/PetterKraabol/Twitch-Chat-Downloader
(MIT License)
'''
def main():
# Arguments
parser = argparse.ArgumentParser(description=f'pytchat v{__version__}')
parser.add_argument('-v', f'--{Arguments.Name.VIDEO}', type=str,
help='Video IDs separated by commas without space')
parser.add_argument('-o', f'--{Arguments.Name.OUTPUT}', type=str,
help='Output directory (end with "/")', default='./')
parser.add_argument(f'--{Arguments.Name.VERSION}', action='store_true',
help='Settings version')
Arguments(parser.parse_args().__dict__)
if Arguments().print_version:
print(f'pytchat v{__version__}')
return
# Extractor
if Arguments().video_ids:
for video_id in Arguments().video_ids:
try:
info = VideoInfo(video_id)
print(f"Extracting...\n"
f" video_id: {video_id}\n"
f" channel: {info.get_channel_name()}\n"
f" title: {info.get_title()}")
Extractor(video_id,
processor = HTMLArchiver(Arguments().output+video_id+'.html')
).extract()
print("Extraction end.\n")
except (InvalidVideoIdException, NoContentsException) as e:
print(e)
return
parser.print_help()

39
pytchat/cli/arguments.py Normal file
View File

@@ -0,0 +1,39 @@
from typing import Optional, Dict, Union, List
from .singleton import Singleton
'''
This modules refer to
Petter Kraabøl's Twitch-Chat-Downloader
https://github.com/PetterKraabol/Twitch-Chat-Downloader
(MIT License)
'''
class Arguments(metaclass=Singleton):
"""
Arguments singleton
"""
class Name:
VERSION: str = 'version'
OUTPUT: str = 'output'
VIDEO: str = 'video'
def __init__(self,
arguments: Optional[Dict[str, Union[str, bool, int]]] = None):
"""
Initialize arguments
:param arguments: Arguments from cli
(Optional to call singleton instance without parameters)
"""
if arguments is None:
print('Error: arguments were not provided')
exit()
self.print_version: bool = arguments[Arguments.Name.VERSION]
self.output: str = arguments[Arguments.Name.OUTPUT]
self.video_ids: List[int] = []
# Videos
if arguments[Arguments.Name.VIDEO]:
self.video_ids = [video_id
for video_id in arguments[Arguments.Name.VIDEO].split(',')]

19
pytchat/cli/singleton.py Normal file
View File

@@ -0,0 +1,19 @@
'''
This modules refer to
Petter Kraabøl's Twitch-Chat-Downloader
https://github.com/PetterKraabol/Twitch-Chat-Downloader
(MIT License)
'''
class Singleton(type):
"""
Abstract class for singletons
"""
_instances = {}
def __call__(cls, *args, **kwargs):
if cls not in cls._instances:
cls._instances[cls] = super().__call__(*args, **kwargs)
return cls._instances[cls]
def get_instance(cls, *args, **kwargs):
cls.__call__(*args, **kwargs)

12
pytchat/config/mylogger.py
View File

@@ -1,6 +1,6 @@
from logging import NullHandler, getLogger, StreamHandler, FileHandler, Formatter
import logging
import datetime
from datetime import datetime
def get_logger(modname,loglevel=logging.DEBUG):
@@ -28,5 +28,11 @@ def get_logger(modname,loglevel=logging.DEBUG):
class MyFormatter(logging.Formatter):
def format(self, record):
s =(datetime.datetime.fromtimestamp(record.created)).strftime("%m-%d %H:%M:%S")+'| '+ (record.module).ljust(15)+(' { '+record.funcName).ljust(20) +":"+str(record.lineno).rjust(4)+'} - '+record.getMessage()
return s
timestamp = (
datetime.fromtimestamp(record.created)).strftime("%m-%d %H:%M:%S")
module = (record.module).ljust(15)
funcname = (record.funcName).ljust(18)
lineno = str(record.lineno).rjust(4)
message = record.getMessage()
return timestamp+'| '+module+' { '+funcname+':'+lineno+'} - '+message

317
pytchat/core_async/replaychat.py
View File

@@ -1,317 +0,0 @@
import aiohttp, asyncio
import datetime
import json
import random
import signal
import time
import traceback
import urllib.parse
import warnings
from aiohttp.client_exceptions import ClientConnectorError
from concurrent.futures import CancelledError
from asyncio import Queue
from .buffer import Buffer
from ..parser.replay import Parser
from .. import config
from ..exceptions import ChatParseException,IllegalFunctionCall
from ..paramgen import arcparam
from ..processors.default.processor import DefaultProcessor
from ..processors.combinator import Combinator
logger = config.logger(__name__)
headers = config.headers
MAX_RETRY = 10
class ReplayChatAsync:
'''
### -----------------------------------------------------------
### [Warning] ReplayChatAsync is integrated into LiveChatAsync.
### This class is deprecated and will be removed at v0.0.5.0.
### ReplayChatAsyncはLiveChatAsyncに統合しました。
### このクラスはv0.0.5.0で廃止予定です。
### -----------------------------------------------------------
asyncio(aiohttp)を利用してYouTubeのチャットデータを取得する。
Parameter
---------
video_id : str
動画ID
seektime : int
リプレイするチャットデータの開始時間（秒）
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):
warnings.warn(""
f"\n{'-'*60}\n[WARNING] ReplayChatAsync is integrated "
f"into LiveChatAsync.\n{' '*5} This is deprecated and will"
f" be removed at v0.0.5.0.\n{'-'*60}\n"
)
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 ReplayChatAsync._setup_finished:
ReplayChatAsync._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(
ReplayChatAsync.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 = arcparam.getparam(self.video_id, self.seektime)
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)
#when replay, not reacquire continuation param
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_replay/get_live_chat_replay?"
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()

309
pytchat/core_multithread/replaychat.py
View File

@@ -1,309 +0,0 @@
import requests
import datetime
import json
import random
import signal
import time
import traceback
import urllib.parse
import warnings
from concurrent.futures import CancelledError, ThreadPoolExecutor
from queue import Queue
from .buffer import Buffer
from ..parser.replay import Parser
from .. import config
from ..exceptions import ChatParseException,IllegalFunctionCall
from ..paramgen import arcparam
from ..processors.default.processor import DefaultProcessor
from ..processors.combinator import Combinator
logger = config.logger(__name__)
headers = config.headers
MAX_RETRY = 10
class ReplayChat:
'''
### -----------------------------------------------------------
### [Warning] ReplayChat is integrated into LiveChat.
### This class is deprecated and will be removed at v0.0.5.0.
### ReplayChatはLiveChatに統合しました。
### このクラスはv0.0.5.0で廃止予定です。
### -----------------------------------------------------------
スレッドプールを利用してYouTubeのライブ配信のチャットデータを取得する
Parameter
---------
video_id : str
動画ID
seektime : int
リプレイするチャットデータの開始時間（秒）
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終了時に呼び出すコールバック。
direct_mode : bool
Trueの場合、bufferを使わずにcallbackを呼ぶ。
Trueの場合、callbackの設定が必須
(設定していない場合IllegalFunctionCall例外を発生させる）
Attributes
---------
_executor : ThreadPoolExecutor
チャットデータ取得ループ（_listen）用のスレッド
_is_alive : bool
チャット取得を停止するためのフラグ
'''
_setup_finished = False
#チャット監視中のListenerのリスト
_listeners= []
def __init__(self, video_id,
seektime = 0,
processor = DefaultProcessor(),
buffer = None,
interruptable = True,
callback = None,
done_callback = None,
direct_mode = False
):
warnings.warn(""
f"\n{'-'*60}\n[WARNING] ReplayChat is integrated into LiveChat.\n"
f"{' '*5}This is deprecated and will be removed at v0.0.5.0.\n"
f"{'-'*60}\n"
)
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._executor = ThreadPoolExecutor(max_workers=2)
self._direct_mode = direct_mode
self._is_alive = True
self._parser = Parser()
self._pauser = Queue()
self._pauser.put_nowait(None)
self._setup()
if not ReplayChat._setup_finished:
ReplayChat._setup_finished = True
if interruptable:
signal.signal(signal.SIGINT, (lambda a, b:
(ReplayChat.shutdown(None,signal.SIGINT,b))
))
ReplayChat._listeners.append(self)
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を呼ぶループタスクの開始
self._executor.submit(self._callback_loop,self._callback)
#_listenループタスクの開始
listen_task = self._executor.submit(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)
def _startlisten(self):
"""最初のcontinuationパラメータを取得し、
_listenループのタスクを作成し開始する
"""
initial_continuation = self._get_initial_continuation()
if initial_continuation is None:
self.terminate()
logger.debug(f"[{self.video_id}]No initial continuation.")
return
self._listen(initial_continuation)
def _get_initial_continuation(self):
''' チャットデータ取得に必要な最初のcontinuationを取得する。'''
try:
initial_continuation = arcparam.getparam(self.video_id,self.seektime)
except ChatParseException as e:
self.terminate()
logger.debug(f"[{self.video_id}]Error:{str(e)}")
return
except KeyError:
logger.debug(f"[{self.video_id}]KeyError:"
f"{traceback.format_exc(limit = -1)}")
self.terminate()
return
return initial_continuation
def _listen(self, continuation):
''' continuationに紐付いたチャットデータを取得し
BUfferにチャットデータを格納、
次のcontinuaitonを取得してループする
Parameter
---------
continuation : str
次のチャットデータ取得に必要なパラメータ
'''
try:
with requests.Session() as session:
while(continuation and self._is_alive):
if self._pauser.empty():
#pause
self._pauser.get()
#resume
#prohibit from blocking by putting None into _pauser.
self._pauser.put_nowait(None)
livechat_json = (
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:
self._callback(
self.processor.process([chat_component])
)
else:
self._buffer.put(chat_component)
diff_time = timeout - (time.time()-time_mark)
if diff_time < 0 : diff_time=0
time.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}]チャット取得を終了しました。")
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_replay/get_live_chat_replay?"
f"continuation={continuation}&pbj=1")
for _ in range(MAX_RETRY + 1):
with session.get(url ,headers = headers) as resp:
try:
text = resp.text
status_code = resp.status_code
livechat_json = json.loads(text)
break
except json.JSONDecodeError :
time.sleep(1)
continue
else:
logger.error(f"[{self.video_id}]"
f"Exceeded retry count. status_code={status_code}")
self.terminate()
return None
return livechat_json
def _callback_loop(self,callback):
""" コンストラクタでcallbackを指定している場合、バックグラウンドで
callbackに指定された関数に一定間隔でチャットデータを投げる。
Parameter
---------
callback : func
加工済みのチャットデータを渡す先の関数。
"""
while self.is_alive():
items = self._buffer.get()
data = self.processor.process(items)
callback(data)
def get(self):
""" bufferからデータを取り出し、processorに投げ、
加工済みのチャットデータを返す。
Returns
: Processorによって加工されたチャットデータ
"""
if self._callback is None:
items = self._buffer.get()
return self.processor.process(items)
raise IllegalFunctionCall(
"既にcallbackを登録済みのため、get()は実行できません。")
def pause(self):
if not self._pauser.empty():
self._pauser.get()
def resume(self):
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 RuntimeError:
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({'chatdata':'','timeout':1})
logger.info(f'[{self.video_id}]終了しました')
@classmethod
def shutdown(cls, event, sig = None, handler=None):
logger.debug("シャットダウンしています")
for t in ReplayChat._listeners:
t._is_alive = False

3
pytchat/processors/default/renderer/base.py
View File

@@ -59,6 +59,7 @@ class BaseRenderer:
def get_badges(self,renderer):
self.author.type = ''
isVerified = False
isChatOwner = False
isChatSponsor = False
@@ -68,6 +69,7 @@ class BaseRenderer:
for badge in badges:
if badge["liveChatAuthorBadgeRenderer"].get("icon"):
author_type = badge["liveChatAuthorBadgeRenderer"]["icon"]["iconType"]
self.author.type = author_type
if author_type == 'VERIFIED':
isVerified = True
if author_type == 'OWNER':
@@ -76,6 +78,7 @@ class BaseRenderer:
isChatModerator = True
if badge["liveChatAuthorBadgeRenderer"].get("customThumbnail"):
isChatSponsor = True
self.author.type = 'MEMBER'
self.get_badgeurl(badge)
return isVerified, isChatOwner, isChatSponsor, isChatModerator

92
pytchat/processors/html_archiver.py Normal file
View File

@@ -0,0 +1,92 @@
import csv
import os
import re
from .chat_processor import ChatProcessor
from .default.processor import DefaultProcessor
PATTERN = re.compile(r"(.*)\(([0-9]+)\)$")
fmt_headers = ['datetime','elapsed','authorName','message','superchat'
,'type','authorChannel']
class HTMLArchiver(ChatProcessor):
'''
HtmlArchiver saves chat data as HTML table format.
'''
def __init__(self, save_path):
super().__init__()
self.save_path = self._checkpath(save_path)
with open(self.save_path, mode='a', encoding = 'utf-8') as f:
f.write('<table border="1" style="border-collapse: collapse">')
f.writelines(self._parse_html_header(fmt_headers))
self.processor = DefaultProcessor()
def _checkpath(self, filepath):
splitter = os.path.splitext(os.path.basename(filepath))
body = splitter[0]
extention = splitter[1]
newpath = filepath
counter = 0
while os.path.exists(newpath):
match = re.search(PATTERN,body)
if match:
counter=int(match[2])+1
num_with_bracket = f'({str(counter)})'
body = f'{match[1]}{num_with_bracket}'
else:
body = f'{body}({str(counter)})'
newpath = os.path.join(os.path.dirname(filepath),body+extention)
return newpath
def process(self, chat_components: list):
"""
Returns
----------
dict :
save_path : str :
Actual save path of file.
total_lines : int :
count of total lines written to the file.
"""
if chat_components is None or len (chat_components) == 0:
return
with open(self.save_path, mode='a', encoding = 'utf-8') as f:
chats = self.processor.process(chat_components).items
for c in chats:
f.writelines(
self._parse_html_line([
c.datetime,
c.elapsedTime,
c.author.name,
c.message,
c.amountString,
c.author.type,
c.author.channelId]
)
)
'''
#Palliative treatment#
Comment out below line to prevent the table
display from collapsing.
'''
#f.write('</table>')
def _parse_html_line(self, raw_line):
html = ''
html+=' <tr>'
for cell in raw_line:
html+='<td>'+cell+'</td>'
html+='</tr>\n'
return html
def _parse_html_header(self,raw_line):
html = ''
html+='<thead>\n'
html+=' <tr>'
for cell in raw_line:
html+='<th>'+cell+'</th>'
html+='</tr>\n'
html+='</thead>\n'
return html

70
pytchat/processors/tsv_archiver.py Normal file
View File

@@ -0,0 +1,70 @@
import csv
import os
import re
from .chat_processor import ChatProcessor
from .default.processor import DefaultProcessor
PATTERN = re.compile(r"(.*)\(([0-9]+)\)$")
fmt_headers = ['datetime','elapsed','authorName','message','superchatAmount'
,'authorType','authorChannel']
class TSVArchiver(ChatProcessor):
'''
TsvArchiver saves chat data as Tab Separated Values format text.
'''
def __init__(self, save_path):
super().__init__()
self.save_path = self._checkpath(save_path)
with open(self.save_path, mode='a', encoding = 'utf-8') as f:
writer = csv.writer(f, delimiter='\t')
writer.writerow(fmt_headers)
self.processor = DefaultProcessor()
def _checkpath(self, filepath):
splitter = os.path.splitext(os.path.basename(filepath))
body = splitter[0]
extention = splitter[1]
newpath = filepath
counter = 0
while os.path.exists(newpath):
match = re.search(PATTERN,body)
if match:
counter=int(match[2])+1
num_with_bracket = f'({str(counter)})'
body = f'{match[1]}{num_with_bracket}'
else:
body = f'{body}({str(counter)})'
newpath = os.path.join(os.path.dirname(filepath),body+extention)
return newpath
def process(self, chat_components: list):
"""
Returns
----------
dict :
save_path : str :
Actual save path of file.
total_lines : int :
count of total lines written to the file.
"""
if chat_components is None or len (chat_components) == 0:
return
with open(self.save_path, mode='a', encoding = 'utf-8') as f:
writer = csv.writer(f, delimiter='\t')
chats = self.processor.process(chat_components).items
for c in chats:
writer.writerow([
c.datetime,
c.elapsedTime,
c.author.name,
c.message,
c.amountString,
c.author.type,
c.author.channelId
])

3
pytchat/tool/extract/worker.py
View File

@@ -68,6 +68,9 @@ def _search_new_block(worker) -> Block:
continuation = continuation,
during_split = True,
is_last = worker.parent_block.is_last)
'''swap last block'''
if worker.parent_block.is_last:
worker.parent_block.is_last = False
worker.blocks.insert(index+1, new_block)
return new_block

13
pytchat/tool/videoinfo.py
View File

@@ -65,8 +65,7 @@ item_moving_thumbnail = [
class VideoInfo:
'''
VideoInfo object retrieves YouTube video informations
from the video page.
VideoInfo object retrieves YouTube video information.
Parameter
---------
@@ -93,13 +92,19 @@ class VideoInfo:
res= json.loads(result.group(1))
response = self._get_item(res, item_response)
if response is None:
raise InvalidVideoIdException(
f"Specified video_id [{self.video_id}] is invalid.")
self._check_video_is_private(res.get("args"))
self._renderer = self._get_item(json.loads(response), item_renderer)
if self._renderer is None:
raise InvalidVideoIdException(
f"No renderer found in video_id: [{self.video_id}].")
def _check_video_is_private(self,args):
if args and args.get("video_id"):
raise InvalidVideoIdException(
f"video_id [{self.video_id}] is private or deleted.")
raise InvalidVideoIdException(
f"video_id [{self.video_id}] is invalid.")
def _get_item(self, dict_body, items: list):
for item in items:
if dict_body is None:

1
requirements.txt
View File

@@ -1,4 +1,5 @@
aiohttp
argumentparser
pytz
requests
urllib3

28
setup.py
View File

@@ -44,29 +44,29 @@ with open('README.md', encoding='utf-8') as f:
setup(
name=package_name,
packages=find_packages(exclude=['*log.txt','*tests','*testrun']),
version=version,
url=url,
author=author,
author_email=author_email,
long_description=long_description,
long_description_content_type='text/markdown',
license=license,
install_requires=_requirements(),
description="a python library for fetching youtube live chat.",
classifiers=[
'Natural Language :: Japanese',
'Development Status :: 4 - Beta',
'Programming Language :: Python',
'Programming Language :: Python :: 3',
'Programming Language :: Python :: 3.4',
'Programming Language :: Python :: 3.5',
'Programming Language :: Python :: 3.6',
'Programming Language :: Python :: 3.7',
'Programming Language :: Python :: 3.8',
'License :: OSI Approved :: MIT License',
],
description="a python library for fetching youtube live chat.",
entry_points=
'''
[console_scripts]
pytchat=pytchat.cli:main
''',
install_requires=_requirements(),
keywords='youtube livechat asyncio',
license=license,
long_description=long_description,
long_description_content_type='text/markdown',
name=package_name,
packages=find_packages(exclude=['*log.txt','*tests','*testrun']),
url=url,
version=version,
)