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

Merge branch 'hotfix/fix_param'

Browse Source
This commit is contained in:
taizan-hokouto
2021-02-11 02:02:14 +09:00
parent af4c2fe4b9 f4bf30c0e9
commit 0a9837adca
8 changed files with 33 additions and 141 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
pytchat/__init__.py
View File

@@ -2,7 +2,7 @@
pytchat is a lightweight python library to browse youtube livechat without Selenium or BeautifulSoup.
"""
__copyright__ = 'Copyright (C) 2019, 2020 taizan-hokuto'
__version__ = '0.5.2'
__version__ = '0.5.3'
__license__ = 'MIT'
__author__ = 'taizan-hokuto'
__author_email__ = '55448286+taizan-hokuto@users.noreply.github.com'

5
pytchat/core/pytchat.py
View File

@@ -95,7 +95,10 @@ class PytchatCore:
"""Fetch first continuation parameter,
create and start _listen loop.
"""
self.continuation = liveparam.getparam(self._video_id, past_sec=3)
self.continuation = liveparam.getparam(
self._video_id,
channel_id=util.get_channelid(httpx.Client(http2=True), self._video_id),
past_sec=3)
def _get_chat_component(self):
''' Fetch chat data and store them into buffer,

13
pytchat/core_async/livechat.py
View File

@@ -152,7 +152,11 @@ class LiveChatAsync:
create and start _listen loop.
"""
if not self.continuation:
self.continuation = liveparam.getparam(self._video_id, 3)
self.continuation = liveparam.getparam(
self._video_id,
channel_id=util.get_channelid(httpx.Client(http2=True), self._video_id),
past_sec=3)
await self._listen(self.continuation)
async def _listen(self, continuation):
@@ -210,8 +214,11 @@ class LiveChatAsync:
'''
self._pauser.put_nowait(None)
if not self._is_replay:
continuation = liveparam.getparam(
self._video_id, 3, self._topchat_only)
async with httpx.AsyncClient(http2=True) as client:
continuation = await liveparam.getparam(self._video_id,
channel_id=util.get_channelid_async(client, self.video_id),
past_sec=3)
return continuation
async def _get_contents(self, continuation, client, headers):

9
pytchat/core_multithread/livechat.py
View File

@@ -148,7 +148,10 @@ class LiveChat:
create and start _listen loop.
"""
if not self.continuation:
self.continuation = liveparam.getparam(self._video_id, 3)
self.continuation = liveparam.getparam(
self._video_id,
channel_id=util.get_channelid(httpx.Client(http2=True), self._video_id),
past_sec=3)
self._listen(self.continuation)
def _listen(self, continuation):
@@ -207,7 +210,9 @@ class LiveChat:
self._pauser.put_nowait(None)
if not self._is_replay:
continuation = liveparam.getparam(
self._video_id, 3, self._topchat_only)
self._video_id, channel_id=util.get_channelid(httpx.Client(http2=True), self._video_id),
past_sec=3, topchat_only=self._topchat_only)
return continuation
def _get_contents(self, continuation, client, headers):

17
pytchat/paramgen/liveparam.py
View File

@@ -5,11 +5,16 @@ from base64 import urlsafe_b64encode as b64enc
from urllib.parse import quote
def _header(video_id) -> str:
return b64enc(enc.rs(1, enc.rs(1, enc.rs(1, video_id))) + enc.nm(4, 1))
def _header(video_id, channel_id) -> str:
S1_3 = enc.rs(1, video_id)
S1_5 = enc.rs(1, channel_id) + enc.rs(2, video_id)
S1 = enc.rs(3, S1_3) + enc.rs(5, S1_5)
S3 = enc.rs(48687757, enc.rs(1, video_id))
header_replay = enc.rs(1, S1) + enc.rs(3, S3) + enc.nm(4, 1)
return b64enc(header_replay)
def _build(video_id, ts1, ts2, ts3, ts4, ts5, topchat_only) -> str:
def _build(video_id, channel_id, ts1, ts2, ts3, ts4, ts5, topchat_only) -> str:
chattype = 4 if topchat_only else 1
b1 = enc.nm(1, 0)
@@ -23,7 +28,7 @@ def _build(video_id, ts1, ts2, ts3, ts4, ts5, topchat_only) -> str:
b11 = enc.nm(11, 3)
b15 = enc.nm(15, 0)
header = enc.rs(3, _header(video_id))
header = enc.rs(3, _header(video_id, channel_id))
timestamp1 = enc.nm(5, ts1)
s6 = enc.nm(6, 0)
s7 = enc.nm(7, 0)
@@ -53,7 +58,7 @@ def _times(past_sec):
return list(map(lambda x: int(x * 1000000), [_ts1, _ts2, _ts3, _ts4, _ts5]))
def getparam(video_id, past_sec=0, topchat_only=False) -> str:
def getparam(video_id, channel_id, past_sec=0, topchat_only=False) -> str:
'''
Parameter
---------
@@ -62,4 +67,4 @@ def getparam(video_id, past_sec=0, topchat_only=False) -> str:
topchat_only : bool
if True, fetch only 'top chat'
'''
return _build(video_id, *_times(past_sec), topchat_only)
return _build(video_id, channel_id, *_times(past_sec), topchat_only)

48
tests/test_livechat.py
View File

@@ -1,48 +0,0 @@
import asyncio
import json
from pytest_httpx import HTTPXMock
from concurrent.futures import CancelledError
from pytchat.core_multithread.livechat import LiveChat
from pytchat.core_async.livechat import LiveChatAsync
from pytchat.exceptions import ResponseContextError
def _open_file(path):
with open(path, mode='r', encoding='utf-8') as f:
return f.read()
def add_response_file(httpx_mock: HTTPXMock, jsonfile_path: str):
testdata = json.loads(_open_file(jsonfile_path))
httpx_mock.add_response(json=testdata)
def test_async(httpx_mock: HTTPXMock):
add_response_file(httpx_mock, 'tests/testdata/paramgen_firstread.json')
async def test_loop():
try:
chat = LiveChatAsync(video_id='__test_id__')
_ = await chat.get()
assert chat.is_alive()
chat.terminate()
assert not chat.is_alive()
except ResponseContextError:
assert False
loop = asyncio.get_event_loop()
try:
loop.run_until_complete(test_loop())
except CancelledError:
assert True
def test_multithread(httpx_mock: HTTPXMock):
add_response_file(httpx_mock, 'tests/testdata/paramgen_firstread.json')
try:
chat = LiveChat(video_id='__test_id__')
_ = chat.get()
assert chat.is_alive()
chat.terminate()
assert not chat.is_alive()
except ResponseContextError:
assert False

71
tests/test_livechat_2.py
View File

@@ -1,71 +0,0 @@
import asyncio
import json
from pytest_httpx import HTTPXMock
from concurrent.futures import CancelledError
from pytchat.core_multithread.livechat import LiveChat
from pytchat.core_async.livechat import LiveChatAsync
from pytchat.processors.dummy_processor import DummyProcessor
def _open_file(path):
with open(path, mode='r', encoding='utf-8') as f:
return f.read()
def add_response_file(httpx_mock: HTTPXMock, jsonfile_path: str):
testdata = json.loads(_open_file(jsonfile_path))
httpx_mock.add_response(json=testdata)
def test_async_live_stream(httpx_mock: HTTPXMock):
add_response_file(httpx_mock, 'tests/testdata/test_stream.json')
async def test_loop():
chat = LiveChatAsync(video_id='__test_id__', processor=DummyProcessor())
chats = await chat.get()
rawdata = chats[0]["chatdata"]
assert list(rawdata[0]["addChatItemAction"]["item"].keys())[
0] == "liveChatTextMessageRenderer"
assert list(rawdata[1]["addChatItemAction"]["item"].keys())[
0] == "liveChatTextMessageRenderer"
assert list(rawdata[2]["addChatItemAction"]["item"].keys())[
0] == "liveChatPlaceholderItemRenderer"
assert list(rawdata[3]["addLiveChatTickerItemAction"]["item"].keys())[
0] == "liveChatTickerPaidMessageItemRenderer"
assert list(rawdata[4]["addChatItemAction"]["item"].keys())[
0] == "liveChatPaidMessageRenderer"
assert list(rawdata[5]["addChatItemAction"]["item"].keys())[
0] == "liveChatPaidStickerRenderer"
assert list(rawdata[6]["addLiveChatTickerItemAction"]["item"].keys())[
0] == "liveChatTickerSponsorItemRenderer"
loop = asyncio.get_event_loop()
try:
loop.run_until_complete(test_loop())
except CancelledError:
assert True
def test_multithread_live_stream(httpx_mock: HTTPXMock):
add_response_file(httpx_mock, 'tests/testdata/test_stream.json')
chat = LiveChat(video_id='__test_id__', processor=DummyProcessor())
chats = chat.get()
rawdata = chats[0]["chatdata"]
# assert fetching livachat data
assert list(rawdata[0]["addChatItemAction"]["item"].keys())[
0] == "liveChatTextMessageRenderer"
assert list(rawdata[1]["addChatItemAction"]["item"].keys())[
0] == "liveChatTextMessageRenderer"
assert list(rawdata[2]["addChatItemAction"]["item"].keys())[
0] == "liveChatPlaceholderItemRenderer"
assert list(rawdata[3]["addLiveChatTickerItemAction"]["item"].keys())[
0] == "liveChatTickerPaidMessageItemRenderer"
assert list(rawdata[4]["addChatItemAction"]["item"].keys())[
0] == "liveChatPaidMessageRenderer"
assert list(rawdata[5]["addChatItemAction"]["item"].keys())[
0] == "liveChatPaidStickerRenderer"
assert list(rawdata[6]["addLiveChatTickerItemAction"]["item"].keys())[
0] == "liveChatTickerSponsorItemRenderer"
chat.terminate()

9
tests/test_liveparam.py
View File

@@ -1,9 +0,0 @@
import pytest
from pytchat.paramgen import liveparam
def test_liveparam_0(mocker):
_ts1= 1546268400
param = liveparam._build("01234567890",
*([_ts1*1000000 for i in range(5)]), topchat_only=False)
test_param="0ofMyAN1GhxDZzhLRFFvTE1ERXlNelExTmpjNE9UQWdBUT09KIC41tWqyt8CMAA4AEABShsIABAAGAAgADoAQABKAFCAuNbVqsrfAlgDeABQgLjW1arK3wJYgLjW1arK3wJoAYIBAggBiAEAmgECCACgAYC41tWqyt8C"
assert test_param == param