youtube-comments/comments.py

import itertools
import os
import qwarc
import qwarc.utils


responseHandler = qwarc.utils.handle_response_limit_error_retries(5)


class Comments(qwarc.Item):
	itemType = 'comments'
	# itemValue = '{videoId}'

	@classmethod
	def generate(cls):
		yield os.environ['YOUTUBE_VIDEOID']

	def get_json_obj_from_pos(self, content, startPos):
		# Given a startPos in content, extracts the content until braces or brackets are matching (requiring at least one set of parentheses)
		openParens = None
		for pos in itertools.count(start = startPos):
			char = content[pos:pos+1]
			if char in (b'{', b'['):
				if openParens is None: # First {[ in the string
					openParens = 0
				openParens += 1
			elif char in (b'}', b']'):
				openParens -= 1
			if openParens == 0:
				break
		return content[startPos:pos]

	def get_continuation_parameters(self, content):
		continuationToken = qwarc.utils.str_get_between(content, b'"continuation":"', b'"')
		if continuationToken.lstrip(b'0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ_-') not in (b'', b'=', b'==', b'%3D', b'%3D%3D'):
			self.logger.error(f'Unexpected continuation token value: {continuationToken!r}')
			return
		continuationToken = continuationToken.decode('ascii')
		itct = qwarc.utils.str_get_between(content, b'"clickTrackingParams":"', b'"')
		if itct.lstrip(b'0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ_-') not in (b'', b'=', b'==', b'%3D', b'%3D%3D'):
			self.logger.error(f'Unexpected itct value: {itct!r}')
			return
		itct = itct.decode('ascii')
		return continuationToken, itct

	async def continue_recursively(self, videoPageUrl, sessionToken, continuationToken, itct, nested = False, initial = False):
		'''
		Fetch the comments, recursively, including other sort orders.
		nested indicates whether the continuationToken is for nested comments (which require a different URL parameter), i.e. "View N replies" or "Show more replies".
		'''
		response, _ = await self.fetch(
			f'https://www.youtube.com/comment_service_ajax?action_get_{"comments" if not nested else "comment_replies"}=1&pbj=1&ctoken={continuationToken}&continuation={continuationToken}&itct={itct}',
			method = 'POST',
			data = {'session_token': sessionToken},
			headers = [('X-YouTube-Client-Name', '1'), ('X-YouTube-Client-Version', '2.20191212.06.02'), ('X-SPF-Referer', videoPageUrl), ('X-SPF-Previous', videoPageUrl)],
			responseHandler = responseHandler,
		  )
		if not response or response.status != 200 or (await response.read(16) == b'{"reload":"now"}'):
			self.logger.error('Could not fetch initial comments')
			return
		content = await response.read()
		# Yes, the response is JSON and I could parse that into an object, but where's the fun in that?

		for continuationsPos in qwarc.utils.find_all(content, b'"continuations":'):
			continuations = self.get_json_obj_from_pos(content, continuationsPos)
			subContinuationToken, subItct = self.get_continuation_parameters(continuations)
			await self.continue_recursively(videoPageUrl, sessionToken, subContinuationToken, subItct, nested = b'"label":' in continuations)

		if initial:
			sortMenuPos = content.find(b'"sortMenu":')
			if sortMenuPos < 0:
				self.logger.error('Could not find sort menu')
				return
			sortMenu = self.get_json_obj_from_pos(content, sortMenuPos)
			for continuationPos in qwarc.utils.find_all(sortMenu, b'"continuation":{'):
				continuation = self.get_json_obj_from_pos(sortMenu, continuationPos)
				subContinuationToken, subItct = self.get_continuation_parameters(continuation)
				await self.continue_recursively(videoPageUrl, sessionToken, subContinuationToken, subItct)

	async def process(self):
		videoPageUrl = f'https://www.youtube.com/watch?v={self.itemValue}'
		response, _ = await self.fetch(videoPageUrl, responseHandler = responseHandler)
		if not response or response.status != 200:
			self.logger.error('Could not fetch video page')
			return
		contents = await response.read()

		sessionToken = qwarc.utils.str_get_between(contents, b'"XSRF_TOKEN":"', b'"')
		if not sessionToken:
			self.logger.error('Could not find session token')
			return
		if sessionToken.lstrip(b'0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ_-') not in (b'', b'=', b'=='):
			self.logger.error('Unexpected session token value: {sessionToken!r}')
			return
		sessionToken = sessionToken.decode('ascii')

		sectionIdentifierPos = contents.find(b'"comment-item-section"')
		if sectionIdentifierPos < 0:
			self.logger.error('Could not find comment section identifier')
			return
		continuationStartPos = contents.rfind(b'"continuation":', 0, sectionIdentifierPos)
		if continuationStartPos < 0:
			self.logger.error('Could not find continuation start position')
			return
		section = contents[continuationStartPos:sectionIdentifierPos]
		continuationToken = qwarc.utils.str_get_between(section, b'"continuation":"', b'"')
		if continuationToken.lstrip(b'0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ_-') not in (b'', b'=', b'=='):
			self.logger.error('Unexpected continuation token value: {continuationToken!r}')
			return
		continuationToken = continuationToken.decode('ascii')
		itct = qwarc.utils.str_get_between(section, b'"clickTrackingParams":"', b'"')
		if itct.lstrip(b'0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ_-') not in (b'', b'=', b'=='):
			self.logger.error('Unexpected itct value: {itct!r}')
			return
		itct = itct.decode('ascii')

		await self.continue_recursively(videoPageUrl, sessionToken, continuationToken, itct, initial = True)


specDependencies = qwarc.utils.SpecDependencies(extra = (('videoId', os.environ['YOUTUBE_VIDEOID']),))