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

Fix JsonfileArchiveProcessor:

Browse Source 
Rename to `JsonfileArchiver`
Add tests
This commit is contained in:
taizan-hokuto
2020-02-26 21:42:27 +09:00
parent 3c95242ddf
commit 61d4e06470
8 changed files with 710 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
pytchat/__init__.py
View File

@@ -20,7 +20,7 @@ from .api import (
CompatibleProcessor, CompatibleProcessor,
DefaultProcessor, DefaultProcessor,
SimpleDisplayProcessor, SimpleDisplayProcessor,
JsonfileArchiveProcessor, JsonfileArchiver,
SpeedCalculator, SpeedCalculator,
DummyProcessor DummyProcessor
) )

2
pytchat/api.py
View File

@@ -6,7 +6,7 @@ from .processors.chat_processor import ChatProcessor
from .processors.default.processor import DefaultProcessor from .processors.default.processor import DefaultProcessor
from .processors.compatible.processor import CompatibleProcessor from .processors.compatible.processor import CompatibleProcessor
from .processors.simple_display_processor import SimpleDisplayProcessor from .processors.simple_display_processor import SimpleDisplayProcessor
from .processors.jsonfile_archive_processor import JsonfileArchiveProcessor from .processors.jsonfile_archiver import JsonfileArchiver
from .processors.speed_calculator import SpeedCalculator from .processors.speed_calculator import SpeedCalculator
from .processors.dummy_processor import DummyProcessor from .processors.dummy_processor import DummyProcessor
from . import config from . import config

46
pytchat/processors/jsonfile_archive_processor.py
View File

@@ -1,46 +0,0 @@
import json
import os
import datetime
from .chat_processor import ChatProcessor
class JsonfileArchiveProcessor(ChatProcessor):
def __init__(self,filepath):
super().__init__()
if os.path.exists(filepath):
print('filepath is already exists!: ')
print(' '+filepath)
newpath=os.path.dirname(filepath) + \
'/'+datetime.datetime.now() \
.strftime('%Y-%m-%d %H-%M-%S')+'.data'
print('created alternate filename:')
print(' '+newpath)
self.filepath = newpath
else:
print('filepath: '+filepath)
self.filepath = filepath
def process(self,chat_components: list):
if chat_components:
with open(self.filepath, mode='a', encoding = 'utf-8') as f:
for component in chat_components:
if component:
chatdata = component.get('chatdata')
for action in chatdata:
if action:
if action.get("addChatItemAction"):
if action["addChatItemAction"]["item"].get(
"liveChatViewerEngagementMessageRenderer"):
continue
s = json.dumps(action,ensure_ascii = False)
#print(s[:200])
f.writelines(s+'\n')
def _parsedir(self,_dir):
if _dir[-1]=='\\' or _dir[-1]=='/':
separator =''
else:
separator ='/'
os.makedirs(_dir + separator, exist_ok=True)
return _dir + separator

66
pytchat/processors/jsonfile_archiver.py Normal file
View File

@@ -0,0 +1,66 @@
import datetime
import json
import os
import re
from .chat_processor import ChatProcessor
PATTERN = re.compile(r"(.*)\(([0-9]+)\)$")
class JsonfileArchiver(ChatProcessor):
"""
JsonfileArchiver saves chat data as text of JSON lines.
Parameter:
----------
save_path : str :
save path of file.If a file with the same name exists,
it is automatically saved under a different name
with suffix '(number)'
"""
def __init__(self,save_path):
super().__init__()
self.save_path = self._checkpath(save_path)
self.line_counter = 0
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: return
with open(self.save_path, mode='a', encoding = 'utf-8') as f:
for component in chat_components:
if component is None: continue
chatdata = component.get('chatdata')
if chatdata is None: continue
for action in chatdata:
if action is None: continue
json_line = json.dumps(action, ensure_ascii = False)
f.writelines(json_line+'\n')
self.line_counter+=1
return { "save_path" : self.save_path,
"total_lines": self.line_counter }
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

48
tests/test_jsonfile_archiver.py Normal file
View File

@@ -0,0 +1,48 @@
import json
from pytchat.processors.jsonfile_archiver import JsonfileArchiver
from unittest.mock import patch, mock_open
from tests.testdata.jsonfile_archiver.chat_component import chat_component
def _open_file(path):
with open(path,mode ='r',encoding = 'utf-8') as f:
return f.read()
def test_checkpath(mocker):
processor = JsonfileArchiver("path")
mocker.patch('os.path.exists').side_effect = exists_file
'''Test no duplicate file.'''
assert processor._checkpath("z:/other.txt") == "z:/other.txt"
'''Test duplicate filename.
The case the name first renamed ('test.txt -> test(0).txt')
is also duplicated.
'''
assert processor._checkpath("z:/test.txt") == "z:/test(1).txt"
'''Test no extention file (duplicate).'''
assert processor._checkpath("z:/test") == "z:/test(0)"
def test_read_write():
'''Test read and write chatdata'''
mock = mock_open(read_data = "")
with patch('builtins.open',mock):
processor = JsonfileArchiver("path")
save_path = processor.process([chat_component])
fh = mock()
actuals = [args[0] for (args, kwargs) in fh.writelines.call_args_list]
'''write format is json dump string with 0x0A'''
to_be_written = [json.dumps(action, ensure_ascii=False)+'\n'
for action in chat_component["chatdata"]]
for i in range(len(actuals)):
assert actuals[i] == to_be_written[i]
assert save_path == {'save_path': 'path', 'total_lines': 7}
def exists_file(path):
if path == "z:/test.txt":
return True
if path == "z:/test(0).txt":
return True
if path == "z:/test":
return True

487
tests/testdata/jsonfile_archiver/chat_component.py vendored Normal file
View File

@@ -0,0 +1,487 @@
chat_component = {
"video_id" : "video_id",
"timeout" : 10,
"chatdata": [
{
"addChatItemAction": {
"item": {
"liveChatTextMessageRenderer": {
"message": {
"runs": [
{
"text": "This is normal message."
}
]
},
"authorName": {
"simpleText": "author_name"
},
"authorPhoto": {
"thumbnails": [
{
"url": "https://yt3.ggpht.com/------------/AAAAAAAAAAA/AAAAAAAAAAA/xxxxxxxxxxxx/s32-x-x-xx-xx-xx-c0xffffff/photo.jpg",
"width": 32,
"height": 32
},
{
"url": "https://yt3.ggpht.com/------------/AAAAAAAAAAA/AAAAAAAAAAA/xxxxxxxxxxxx/s32-x-x-xx-xx-xx-c0xffffff/photo.jpg",
"width": 64,
"height": 64
}
]
},
"contextMenuEndpoint": {
"commandMetadata": {
"webCommandMetadata": {
"ignoreNavigation": True
}
},
"liveChatItemContextMenuEndpoint": {
"params": "___params___"
}
},
"id": "dummy_id",
"timestampUsec": 0,
"authorExternalChannelId": "http://www.youtube.com/channel/author_channel_url",
"contextMenuAccessibility": {
"accessibilityData": {
"label": "コメントの操作"
}
}
}
},
"clientId": "dummy_client_id"
}
},
{
"addChatItemAction": {
"item": {
"liveChatTextMessageRenderer": {
"message": {
"runs": [
{
"text": "This is members's message"
}
]
},
"authorName": {
"simpleText": "author_name"
},
"authorPhoto": {
"thumbnails": [
{
"url": "https://yt3.ggpht.com/------------/AAAAAAAAAAA/AAAAAAAAAAA/xxxxxxxxxxxx/s32-x-x-xx-xx-xx-c0xffffff/photo.jpg",
"width": 32,
"height": 32
},
{
"url": "https://yt3.ggpht.com/------------/AAAAAAAAAAA/AAAAAAAAAAA/xxxxxxxxxxxx/s32-x-x-xx-xx-xx-c0xffffff/photo.jpg",
"width": 64,
"height": 64
}
]
},
"contextMenuEndpoint": {
"commandMetadata": {
"webCommandMetadata": {
"ignoreNavigation": True
}
},
"liveChatItemContextMenuEndpoint": {
"params": "___params___"
}
},
"id": "dummy_id",
"timestampUsec": 0,
"authorBadges": [
{
"liveChatAuthorBadgeRenderer": {
"customThumbnail": {
"thumbnails": [
{
"url": "https://yt3.ggpht.com/X=s32-c-k"
},
{
"url": "https://yt3.ggpht.com/X=s32-c-k"
}
]
},
"tooltip": "メンバー（2 か月）",
"accessibility": {
"accessibilityData": {
"label": "メンバー（2 か月）"
}
}
}
}
],
"authorExternalChannelId": "http://www.youtube.com/channel/author_channel_url",
"contextMenuAccessibility": {
"accessibilityData": {
"label": "コメントの操作"
}
}
}
},
"clientId": "dummy_client_id"
}
},
{
"addChatItemAction": {
"item": {
"liveChatPlaceholderItemRenderer": {
"id": "dummy_id",
"timestampUsec": 0
}
},
"clientId": "dummy_client_id"
}
},
{
"addLiveChatTickerItemAction": {
"item": {
"liveChatTickerPaidMessageItemRenderer": {
"id": "dummy_id",
"amount": {
"simpleText": "￥10,000"
},
"amountTextColor": 4294967295,
"startBackgroundColor": 4293271831,
"endBackgroundColor": 4291821568,
"authorPhoto": {
"thumbnails": [
{
"url": "https://yt3.ggpht.com/------------/AAAAAAAAAAA/AAAAAAAAAAA/xxxxxxxxxxxx/s32-x-x-xx-xx-xx-c0xffffff/photo.jpg",
"width": 32,
"height": 32
},
{
"url": "https://yt3.ggpht.com/------------/AAAAAAAAAAA/AAAAAAAAAAA/xxxxxxxxxxxx/s32-x-x-xx-xx-xx-c0xffffff/photo.jpg",
"width": 64,
"height": 64
}
]
},
"durationSec": 3600,
"showItemEndpoint": {
"commandMetadata": {
"webCommandMetadata": {
"ignoreNavigation": True
}
},
"showLiveChatItemEndpoint": {
"renderer": {
"liveChatPaidMessageRenderer": {
"id": "dummy_id",
"timestampUsec": 0,
"authorName": {
"simpleText": "author_name"
},
"authorPhoto": {
"thumbnails": [
{
"url": "https://yt3.ggpht.com/------------/AAAAAAAAAAA/AAAAAAAAAAA/xxxxxxxxxxxx/s32-x-x-xx-xx-xx-c0xffffff/photo.jpg",
"width": 32,
"height": 32
},
{
"url": "https://yt3.ggpht.com/------------/AAAAAAAAAAA/AAAAAAAAAAA/xxxxxxxxxxxx/s32-x-x-xx-xx-xx-c0xffffff/photo.jpg",
"width": 64,
"height": 64
}
]
},
"purchaseAmountText": {
"simpleText": "￥10,000"
},
"message": {
"runs": [
{
"text": "This is superchat message."
}
]
},
"headerBackgroundColor": 4291821568,
"headerTextColor": 4294967295,
"bodyBackgroundColor": 4293271831,
"bodyTextColor": 4294967295,
"authorExternalChannelId": "http://www.youtube.com/channel/author_channel_url",
"authorNameTextColor": 3019898879,
"contextMenuEndpoint": {
"commandMetadata": {
"webCommandMetadata": {
"ignoreNavigation": True
}
},
"liveChatItemContextMenuEndpoint": {
"params": "___params___"
}
},
"timestampColor": 2164260863,
"contextMenuAccessibility": {
"accessibilityData": {
"label": "コメントの操作"
}
}
}
}
}
},
"authorExternalChannelId": "http://www.youtube.com/channel/author_channel_url",
"fullDurationSec": 3600
}
},
"durationSec": "3600"
}
},
{
"addChatItemAction": {
"item": {
"liveChatPaidMessageRenderer": {
"id": "dummy_id",
"timestampUsec": 0,
"authorName": {
"simpleText": "author_name"
},
"authorPhoto": {
"thumbnails": [
{
"url": "https://yt3.ggpht.com/------------/AAAAAAAAAAA/AAAAAAAAAAA/xxxxxxxxxxxx/s32-x-x-xx-xx-xx-c0xffffff/photo.jpg",
"width": 32,
"height": 32
},
{
"url": "https://yt3.ggpht.com/------------/AAAAAAAAAAA/AAAAAAAAAAA/xxxxxxxxxxxx/s32-x-x-xx-xx-xx-c0xffffff/photo.jpg",
"width": 64,
"height": 64
}
]
},
"purchaseAmountText": {
"simpleText": "￥10,800"
},
"message": {
"runs": [
{
"text": "This is superchat message."
}
]
},
"headerBackgroundColor": 4291821568,
"headerTextColor": 4294967295,
"bodyBackgroundColor": 4293271831,
"bodyTextColor": 4294967295,
"authorExternalChannelId": "http://www.youtube.com/channel/author_channel_url",
"authorNameTextColor": 3019898879,
"contextMenuEndpoint": {
"commandMetadata": {
"webCommandMetadata": {
"ignoreNavigation": True
}
},
"liveChatItemContextMenuEndpoint": {
"params": "___params___"
}
},
"timestampColor": 2164260863,
"contextMenuAccessibility": {
"accessibilityData": {
"label": "コメントの操作"
}
}
}
}
}
},
{
"addChatItemAction": {
"item": {
"liveChatPaidStickerRenderer": {
"id": "dummy_id",
"contextMenuEndpoint": {
"clickTrackingParams": "___clickTrackingParams___",
"commandMetadata": {
"webCommandMetadata": {
"ignoreNavigation": True
}
},
"liveChatItemContextMenuEndpoint": {
"params": "___params___"
}
},
"contextMenuAccessibility": {
"accessibilityData": {
"label": "コメントの操作"
}
},
"timestampUsec": 0,
"authorPhoto": {
"thumbnails": [
{
"url": "https://yt3.ggpht.com/------------/AAAAAAAAAAA/AAAAAAAAAAA/xxxxxxxxxxxx/s32-x-x-xx-xx-xx-c0xffffff/photo.jpg",
"width": 32,
"height": 32
},
{
"url": "https://yt3.ggpht.com/------------/AAAAAAAAAAA/AAAAAAAAAAA/xxxxxxxxxxxx/s32-x-x-xx-xx-xx-c0xffffff/photo.jpg",
"width": 64,
"height": 64
}
]
},
"authorName": {
"simpleText": "author_name"
},
"authorExternalChannelId": "http://www.youtube.com/channel/author_channel_url",
"sticker": {
"thumbnails": [
{
"url": "//lh3.googleusercontent.com/param_s=s40-rp",
"width": 40,
"height": 40
},
{
"url": "//lh3.googleusercontent.com/param_s=s80-rp",
"width": 80,
"height": 80
}
],
"accessibility": {
"accessibilityData": {
"label": "___sticker_label___"
}
}
},
"moneyChipBackgroundColor": 4280191205,
"moneyChipTextColor": 4294967295,
"purchaseAmountText": {
"simpleText": "￥150"
},
"stickerDisplayWidth": 40,
"stickerDisplayHeight": 40,
"backgroundColor": 4279592384,
"authorNameTextColor": 3019898879,
"trackingParams": "___trackingParams___"
}
}
}
},
{
"addLiveChatTickerItemAction": {
"item": {
"liveChatTickerSponsorItemRenderer": {
"id": "dummy_id",
"detailText": {
"runs": [
{
"text": "メンバー"
}
]
},
"detailTextColor": 4294967295,
"startBackgroundColor": 4279213400,
"endBackgroundColor": 4278943811,
"sponsorPhoto": {
"thumbnails": [
{
"url": "https://yt3.ggpht.com/------------/AAAAAAAAAAA/AAAAAAAAAAA/xxxxxxxxxxxx/s32-x-x-xx-xx-xx-c0xffffff/photo.jpg",
"width": 32,
"height": 32
},
{
"url": "https://yt3.ggpht.com/------------/AAAAAAAAAAA/AAAAAAAAAAA/xxxxxxxxxxxx/s32-x-x-xx-xx-xx-c0xffffff/photo.jpg",
"width": 64,
"height": 64
}
]
},
"durationSec": 300,
"showItemEndpoint": {
"commandMetadata": {
"webCommandMetadata": {
"ignoreNavigation": True
}
},
"showLiveChatItemEndpoint": {
"renderer": {
"liveChatMembershipItemRenderer": {
"id": "dummy_id",
"timestampUsec": 0,
"authorExternalChannelId": "http://www.youtube.com/channel/author_channel_url",
"headerSubtext": {
"runs": [
{
"text": "メンバーシップ"
},
{
"text": " へようこそ！"
}
]
},
"authorName": {
"simpleText": "author_name"
},
"authorPhoto": {
"thumbnails": [
{
"url": "https://yt3.ggpht.com/------------/AAAAAAAAAAA/AAAAAAAAAAA/xxxxxxxxxxxx/s32-x-x-xx-xx-xx-c0xffffff/photo.jpg",
"width": 32,
"height": 32
},
{
"url": "https://yt3.ggpht.com/------------/AAAAAAAAAAA/AAAAAAAAAAA/xxxxxxxxxxxx/s32-x-x-xx-xx-xx-c0xffffff/photo.jpg",
"width": 64,
"height": 64
}
]
},
"authorBadges": [
{
"liveChatAuthorBadgeRenderer": {
"customThumbnail": {
"thumbnails": [
{
"url": "https://yt3.ggpht.com/X=s32-c-k"
},
{
"url": "https://yt3.ggpht.com/X=s32-c-k"
}
]
},
"tooltip": "新規メンバー",
"accessibility": {
"accessibilityData": {
"label": "新規メンバー"
}
}
}
}
],
"contextMenuEndpoint": {
"commandMetadata": {
"webCommandMetadata": {
"ignoreNavigation": True
}
},
"liveChatItemContextMenuEndpoint": {
"params": "___params___"
}
},
"contextMenuAccessibility": {
"accessibilityData": {
"label": "コメントの操作"
}
}
}
}
}
},
"authorExternalChannelId": "http://www.youtube.com/channel/author_channel_url",
"fullDurationSec": 300
}
},
"durationSec": "300"
}
}
]
}

18
tests/testdata/jsonfile_archiver/replay_end.json vendored Normal file
View File

@@ -0,0 +1,18 @@
{
"response": {
"responseContext": {
"webResponseContextExtensionData": ""
},
"continuationContents": {
"liveChatContinuation": {
"continuations": [
{
"playerSeekContinuationData": {
"continuation": "___reload_continuation___"
}
}
]
}
}
}
}

89
tests/testdata/jsonfile_archiver/text_only.json vendored Normal file
View File

@@ -0,0 +1,89 @@
{
"response": {
"responseContext": {
"webResponseContextExtensionData": ""
},
"continuationContents": {
"liveChatContinuation": {
"continuations": [
{
"invalidationContinuationData": {
"invalidationId": {
"objectSource": 1000,
"objectId": "___objectId___",
"topic": "chat~00000000000~0000000",
"subscribeToGcmTopics": true,
"protoCreationTimestampMs": "1577804400000"
},
"timeoutMs": 10000,
"continuation": "___continuation___"
}
}
],
"actions": [
{
"replayChatItemAction": {
"actions": [
{
"addChatItemAction": {
"item": {
"liveChatTextMessageRenderer": {
"message": {
"runs": [
{
"text": "dummy_message"
}
]
},
"authorName": {
"simpleText": "author_name"
},
"authorPhoto": {
"thumbnails": [
{
"url": "https://yt3.ggpht.com/------------/AAAAAAAAAAA/AAAAAAAAAAA/xxxxxxxxxxxx/s32-x-x-xx-xx-xx-c0xffffff/photo.jpg",
"width": 32,
"height": 32
},
{
"url": "https://yt3.ggpht.com/------------/AAAAAAAAAAA/AAAAAAAAAAA/xxxxxxxxxxxx/s32-x-x-xx-xx-xx-c0xffffff/photo.jpg",
"width": 64,
"height": 64
}
]
},
"contextMenuEndpoint": {
"commandMetadata": {
"webCommandMetadata": {
"ignoreNavigation": true
}
},
"liveChatItemContextMenuEndpoint": {
"params": "___params___"
}
},
"id": "dummy_id",
"timestampUsec": 0,
"authorExternalChannelId": "http://www.youtube.com/channel/author_channel_url",
"contextMenuAccessibility": {
"accessibilityData": {
"label": "コメントの操作"
}
},
"timestampText": {
"simpleText": "0:00"
}
}
},
"clientId": "dummy_client_id"
}
}
],
"videoOffsetTimeMsec": "10000"
}
}
]
}
}
}
}