→コード: v4.1.9 --search-unarchivedモードでツイートの整形も行う。ちゃんとテストはできていない。v4.1.8ではNitterでのページングのためにCookieを追加した。
>Fet-Fe (→コード: v4.1.8 Nitterd) |
>Fet-Fe (→コード: v4.1.9 --search-unarchivedモードでツイートの整形も行う。ちゃんとテストはできていない。v4.1.8ではNitterでのページングのためにCookieを追加した。) |
||
11行目: | 11行目: | ||
"""Twitter自動収集スクリプト | """Twitter自動収集スクリプト | ||
ver4.1. | ver4.1.9 2023/11/27恒心 | ||
当コードは恒心停止してしまった https://rentry.co/7298g の降臨ショーツイート自動収集スクリプトの復刻改善版です | 当コードは恒心停止してしまった https://rentry.co/7298g の降臨ショーツイート自動収集スクリプトの復刻改善版です | ||
123行目: | 123行目: | ||
""" | """ | ||
nitter_instance: Final[str] = ' | nitter_instance: Final[str] = 'https://nitter.moomoo.me/' | ||
"""Final[str]: Nitterのインスタンス。 | """Final[str]: Nitterのインスタンス。 | ||
153行目: | 153行目: | ||
""" | """ | ||
incremented_num_default: Final[int] = | incremented_num_default: Final[int] = 6 | ||
"""Final[int]: ツイートURLの数字部分うち、インクリメントする桁のデフォルト値。 | """Final[int]: ツイートURLの数字部分うち、インクリメントする桁のデフォルト値。 | ||
1,887行目: | 1,887行目: | ||
class ArchiveCrawler(TwitterArchiver): | class ArchiveCrawler(TwitterArchiver): | ||
"""archive.todayに記録された尊師のツイートのうち、Wiki未掲載のものを収集する。 | """archive.todayに記録された尊師のツイートのうち、Wiki未掲載のものを収集する。 | ||
Todo: | |||
*ちゃんとテストする。 | |||
*ツイートのテーブル生成でこけてもURL一覧ファイルだけはダンプするようにする。 | |||
""" | """ | ||
1,908行目: | 1,912行目: | ||
""" | """ | ||
URL_LIST_FILENAME: Final[str] = \ | |||
UserProperties.ArchiveCrawler.url_list_filename | |||
"""Final[str]: URLのリストをダンプするファイル名。 | """Final[str]: URLのリストをダンプするファイル名。 | ||
""" | """ | ||
1,981行目: | 1,986行目: | ||
soup: Final[BeautifulSoup] = BeautifulSoup(page, 'html.parser') | soup: Final[BeautifulSoup] = BeautifulSoup(page, 'html.parser') | ||
url_as = soup.select('tr > th a') | url_as = soup.select('tr > th a') | ||
assert len(url_as) > 0 | |||
for url_a in url_as: | for url_a in url_as: | ||
href: str | list[str] | None = url_a.get('href') | href: str | list[str] | None = url_a.get('href') | ||
2,088行目: | 2,094行目: | ||
* :const:`~INCREMENTED_NUM_DEFAULT`: `incremented_num` のデフォルト値。 | * :const:`~INCREMENTED_NUM_DEFAULT`: `incremented_num` のデフォルト値。 | ||
""" | """ | ||
assert 0 <= | assert 0 <= incremented_num <= 9, \ | ||
f'incremented_numが{incremented_num}でふ' | f'incremented_numが{incremented_num}でふ' | ||
logger.info(self.TWITTER_URL + self._name + '/status/' | logger.info(self.TWITTER_URL + self._name + '/status/' | ||
2,132行目: | 2,138行目: | ||
@override | @override | ||
def _tweet_date(self, tweet: Tag) -> datetime: | def _tweet_date(self, tweet: Tag) -> datetime: | ||
datetime_tag: Final[Tag | None] = tweet. | # TODO: 日付が合わないのを修正する | ||
datetime_tag: Final[Tag | None] = tweet.select('time[datetime]')[-1] | |||
assert datetime_tag is not None | assert datetime_tag is not None | ||
datetime_str: Final[str | list[str] | None] = ( | datetime_str: Final[str | list[str] | None] = ( | ||
2,138行目: | 2,145行目: | ||
assert isinstance(datetime_str, str) | assert isinstance(datetime_str, str) | ||
raw_time: Final[datetime] = datetime.strptime( | raw_time: Final[datetime] = datetime.strptime( | ||
datetime_str, '%Y-%m-%dT%H:%M:% | datetime_str, '%Y-%m-%dT%H:%M:%S.000Z') | ||
return raw_time.replace(tzinfo=ZoneInfo('Asia/Tokyo')) | return raw_time.replace(tzinfo=ZoneInfo('Asia/Tokyo')) | ||
def | def _parse_images(self, soup: Tag, | ||
accessor: AccessorHandler) -> tuple[str, ...]: | |||
"""ツイートの魚拓から画像をダウンロードし、ファイル名のタプルを返す。 | |||
引用リツイート内の画像や、外部リンクの画像は除く。 | |||
Args: | |||
soup (Tag): ツイートの魚拓のタグ。 | |||
accessor (AccessorHandler): アクセスハンドラ。 | |||
Returns: | |||
tuple[str, ...]: ファイル名のタプル。 | |||
""" | |||
image_list: list[str] = [] | |||
image_tags: ResultSet[Tag] = soup.select( | |||
'img[alt="Image"]:not(div[role="link"] img[alt="Image"])') | |||
for image_tag in image_tags: | |||
media_url_path: str | list[str] | None = ( | |||
image_tag.get('src')) | |||
assert isinstance(media_url_path, str) | |||
original_image_url: str | list[str] | None = ( | |||
image_tag.get('new-cursrc')) | |||
assert isinstance(original_image_url, str) | |||
parse_result: ParseResult = urlparse(original_image_url) | |||
original_image_name: str = parse_result.path.split('/')[-1] | |||
original_extension: str = parse_qs(parse_result.query)['format'][0] | |||
original_image_name += '.' + original_extension | |||
self._download_media( | |||
urljoin(self.ARCHIVE_TODAY, media_url_path), | |||
original_image_name, | |||
accessor) | |||
image_list.append(original_image_name) | |||
return tuple(image_list) | |||
def _replace_links(self, tag: Tag, accessor: AccessorHandler) -> Tag: | |||
"""ツイートの魚拓の、本文内のaタグを整形する。 | |||
Args: | |||
tag (Tag): aタグを置き換えるべきタグ。 | |||
accessor (AccessorHandler): アクセスハンドラ。 | |||
Returns: | |||
Tag: テキスト内のaタグが置き換えられたタグ。 | |||
""" | |||
internal_a_tags: Final[ResultSet[Tag]] = tag.select( | |||
'div[data-testid="tweetText"]:not(div[role="link"] ' | |||
f'div[data-testid="tweetText"]) a[href*="{self.TWITTER_URL}"]') | |||
# Twitter内部のリンク | |||
if len(internal_a_tags) > 0: | |||
for a_tag in internal_a_tags: | |||
account_name: Final[str] = a_tag.text | |||
if account_name.startswith('#'): | |||
a_tag.replace_with(a_tag.text) | |||
else: | |||
url: Final[str] = urljoin( | |||
self.TWITTER_URL, account_name[1:]) | |||
a_tag.replace_with(TableBuilder.archive_url( | |||
url, | |||
self._archive(url, accessor), | |||
account_name)) | |||
# 通常のリンク | |||
a_tags: Final[ResultSet[Tag]] = tag.select( | |||
'div[data-testid="tweetText"]:not(' | |||
'div[role="link"] div[data-testid="tweetText"]) > a') | |||
for a_tag in a_tags: | |||
a_tag.replace_with( | |||
'{{Archive|1=リンクがあります。重複に注意して下さい|2=リンクがあります}}') | |||
return tag | |||
@staticmethod | |||
def _retrieve_emojis(tag: Tag) -> Tag: | |||
"""絵文字を画像タグからUnicodeに戻す。 | |||
Args: | |||
tag (Tag): 絵文字の含まれる可能性のあるタグ。 | |||
Returns: | |||
Tag: 絵文字をUnicodeに戻したタグ。 | |||
""" | |||
img_tags: Final[ResultSet[Tag]] = tag.select( | |||
'img[new-cursrc^="https://abs-0.twimg.com/emoji"]') | |||
for img_tag in img_tags: | |||
alt_text: str | list[str] | None = img_tag.get('alt') | |||
assert isinstance(alt_text, str) | |||
img_tag.replace_with(alt_text) | |||
return tag | |||
@staticmethod | |||
def _concat_texts(*texts: str) -> str: | |||
"""2つ以上のテキストを改行タグを挟んで結合する。 | |||
Args: | |||
*texts: 入力テキスト。 | |||
Returns: | |||
str: 結合されたテキスト。 | |||
""" | |||
return '\n'.join(filter(None, texts)) | |||
def _get_tweet_from_archive( | |||
self, | self, | ||
url_pairs: list[UrlTuple], | url_pairs: list[UrlTuple], | ||
accessor: AccessorHandler) -> | accessor: AccessorHandler) -> tuple[str, ...]: | ||
""" | """魚拓からツイート本文を取得する。 | ||
リツイートは除く。 | |||
Args: | Args: | ||
2,152行目: | 2,264行目: | ||
Returns: | Returns: | ||
tuple[str]: リツイートを除いたURLのタプル。 | |||
""" | """ | ||
filtered_urls: Final[list[ | filtered_urls: Final[list[str]] = [] | ||
table_builder: Final[TableBuilder] = TableBuilder() | |||
for url_pair in url_pairs: | for url_pair in url_pairs: | ||
page: Final[str | None] = accessor.request(url_pair.archive_url) | page: Final[str | None] = accessor.request(url_pair.archive_url) | ||
assert page is not None | assert page is not None | ||
article: Final[Tag | None] = BeautifulSoup( | |||
if | page, 'html.parser' | ||
filtered_urls.append(url_pair) | ).select_one('article[tabindex="-1"]') | ||
# リツイート以外のURLを保存する | |||
if article is not None and article.select_one( | |||
'span[data-testid="socialContext"]') is None: | |||
filtered_urls.append(url_pair.url) | |||
logger.debug(url_pair.url + 'を整形しますを') | |||
tweet_date: Final[datetime] = self._tweet_date(article) | |||
table_builder.next_day_if_necessary(tweet_date) | |||
tweet_callinshow_template: Final[str] = ( | |||
self._callinshowlink_url(url_pair.url, accessor)) | |||
# 本文 | |||
text_tag: Tag | None = article.select_one( | |||
'div[data-testid="tweetText"]:not(' | |||
'div[role="link"] div[data-testid="tweetText"])') | |||
if text_tag is not None: | |||
text_tag = self._retrieve_emojis(text_tag) | |||
text: str = self._replace_links(text_tag, accessor).text | |||
else: | |||
text: str = '' | |||
# YouTube等のリンク | |||
card_tag: Final[Tag | None] = article.select_one( | |||
'div[data-testid="card.layoutSmall.media"]:not(' | |||
'div[role="link"] ' | |||
'div[data-testid="card.layoutSmall.media"])') | |||
if card_tag is not None: | |||
text = self._concat_texts( | |||
text, '{{Archive|1=リンクがあります。重複に注意して下さい|2=リンクがあります}}') | |||
# 画像に埋め込まれた外部サイトへのリンク | |||
article_tag: Final[Tag | None] = article.select_one( | |||
'article div[data-testid="card.layoutLarge.media"] a') | |||
if article_tag is not None: | |||
text = self._concat_texts( | |||
text, '{{Archive|1=リンクがあります。重複に注意して下さい|2=リンクがあります}}') | |||
# 引用の有無のチェック | |||
retweet_tag: Final[Tag | None] = article.select_one( | |||
'article[data-testid="tweet"] div[role="link"]') | |||
if retweet_tag is not None: | |||
account_name_tag: Final[Tag | None] = ( | |||
retweet_tag.select_one( | |||
'div[data-testid="User-Name"] div[tabindex="-1"]')) | |||
assert account_name_tag is not None | |||
text = self._concat_texts( | |||
text, '{{Archive|1=' | |||
+ account_name_tag.text | |||
+ 'のリツイートがあります|2=リツイートがあります}}') | |||
# 画像の取得 | |||
image_list: tuple[str, ...] = self._parse_images(article, | |||
accessor) | |||
image_txt: str = ' '.join(map( | |||
lambda t: f'[[ファイル:{t}|240px]]', image_list)) | |||
text = self._concat_texts(text, image_txt) | |||
# TODO: 投票の処理 | |||
table_builder.append( | |||
tweet_callinshow_template, | |||
TableBuilder.escape_wiki_reserved_words(text) | |||
) | |||
else: | |||
logger.debug(url_pair.url + 'はリツイートなのでポア') | |||
return filtered_urls | table_builder.dump_file() | ||
return tuple(filtered_urls) | |||
@override | @override | ||
2,184行目: | 2,364行目: | ||
self.INCREMENTED_NUM_DEFAULT) | self.INCREMENTED_NUM_DEFAULT) | ||
# | # ツイート本文を取得する | ||
filtered_url_tuple: Final[tuple[str, ...]] = ( | |||
self. | self._get_tweet_from_archive(self._url_list, accessor)) | ||
# URL一覧ファイルのダンプ | |||
# TODO: _get_tweet_from_archiveが失敗しても実行できるようにする。 | |||
with codecs.open(self.URL_LIST_FILENAME, 'w', 'utf-8') as f: | |||
for url in filtered_url_tuple: | |||
f.write(url + '\n') | |||
logger.info('テキストファイル手に入ったやで〜') | |||