Scrape images, video, and post forwarding information for Telegram channel posts

This commit is contained in:
Logan Williams
2020-10-15 08:14:14 -07:00
parent 77bbb9f61f
commit 72b26f2373

View File

@@ -9,7 +9,7 @@ import re
import snscrape.base import snscrape.base
import typing import typing
import urllib.parse import urllib.parse
import base64
_logger = logging.getLogger(__name__) _logger = logging.getLogger(__name__)
_SINGLE_MEDIA_LINK_PATTERN = re.compile(r'^https://t\.me/[^/]+/\d+\?single$') _SINGLE_MEDIA_LINK_PATTERN = re.compile(r'^https://t\.me/[^/]+/\d+\?single$')
@@ -30,6 +30,9 @@ class TelegramPost(snscrape.base.Item):
date: datetime.datetime date: datetime.datetime
content: str content: str
outlinks: list outlinks: list
images: list
video: str
forwarded: str
linkPreview: typing.Optional[LinkPreview] = None linkPreview: typing.Optional[LinkPreview] = None
outlinksss = snscrape.base._DeprecatedProperty('outlinksss', lambda self: ' '.join(self.outlinks), 'outlinks') outlinksss = snscrape.base._DeprecatedProperty('outlinksss', lambda self: ' '.join(self.outlinks), 'outlinks')
@@ -90,17 +93,38 @@ class TelegramChannelScraper(snscrape.base.Scraper):
_logger.warning(f'Possibly incorrect URL: {rawUrl!r}') _logger.warning(f'Possibly incorrect URL: {rawUrl!r}')
url = rawUrl.replace('//t.me/', '//t.me/s/') url = rawUrl.replace('//t.me/', '//t.me/s/')
date = datetime.datetime.strptime(dateDiv.find('time', datetime = True)['datetime'].replace('-', '', 2).replace(':', ''), '%Y%m%dT%H%M%S%z') date = datetime.datetime.strptime(dateDiv.find('time', datetime = True)['datetime'].replace('-', '', 2).replace(':', ''), '%Y%m%dT%H%M%S%z')
images = []
video = None
forwarded = None
if (message := post.find('div', class_ = 'tgme_widget_message_text')): if (message := post.find('div', class_ = 'tgme_widget_message_text')):
content = message.text content = message.get_text(separator="\n")
for video_tag in post.find_all('video'):
video = video_tag['src']
if (forward_tag := post.find('a', class_ = 'tgme_widget_message_forwarded_from_name')):
forwarded = forward_tag['href'].split('t.me/')[1].split('/')[0]
outlinks = [] outlinks = []
for link in post.find_all('a'): for link in post.find_all('a'):
if any(x in link.parent.attrs.get('class', []) for x in ('tgme_widget_message_user', 'tgme_widget_message_author')): if any(x in link.parent.attrs.get('class', []) for x in ('tgme_widget_message_user', 'tgme_widget_message_author')):
# Author links at the top (avatar and name) # Author links at the top (avatar and name)
continue continue
if link['href'] == rawUrl or link['href'] == url: if link['href'] == rawUrl or link['href'] == url:
style = link.attrs.get('style', '')
# Generic filter of links to the post itself, catches videos, photos, and the date link # Generic filter of links to the post itself, catches videos, photos, and the date link
continue if style != '':
image = re.findall('url\(\'(.*?)\'\)', style)
if len(image) == 1:
images.append(image[0])
continue
if _SINGLE_MEDIA_LINK_PATTERN.match(link['href']): if _SINGLE_MEDIA_LINK_PATTERN.match(link['href']):
style = link.attrs.get('style', '')
image = re.findall('url\(\'(.*?)\'\)', style)
if len(image) == 1:
images.append(image[0])
# resp = self._get(image[0])
# encoded_string = base64.b64encode(resp.content)
# Individual photo or video link # Individual photo or video link
continue continue
href = urllib.parse.urljoin(pageUrl, link['href']) href = urllib.parse.urljoin(pageUrl, link['href'])
@@ -109,6 +133,8 @@ class TelegramChannelScraper(snscrape.base.Scraper):
else: else:
content = None content = None
outlinks = [] outlinks = []
images = []
video = None
linkPreview = None linkPreview = None
if (linkPreviewA := post.find('a', class_ = 'tgme_widget_message_link_preview')): if (linkPreviewA := post.find('a', class_ = 'tgme_widget_message_link_preview')):
kwargs = {} kwargs = {}
@@ -125,7 +151,7 @@ class TelegramChannelScraper(snscrape.base.Scraper):
else: else:
_logger.warning(f'Could not process link preview image on {url}') _logger.warning(f'Could not process link preview image on {url}')
linkPreview = LinkPreview(**kwargs) linkPreview = LinkPreview(**kwargs)
yield TelegramPost(url = url, date = date, content = content, outlinks = outlinks, linkPreview = linkPreview) yield TelegramPost(url = url, date = date, content = content, outlinks = outlinks, linkPreview = linkPreview, images = images, video = video, forwarded = forwarded)
def get_items(self): def get_items(self):
r, soup = self._initial_page() r, soup = self._initial_page()