Scrapyでクロールし、S3へアップロードしたhtmlファイルを本文抽出して、Elasticsearchのインデックスへ保存したい。

###環境:
Mac OS 10.13.6, Python 3.8.5, Scrapy 2.2.1, botocore/2.0.0dev38,
scrapy-s3pipeline 0.3.0, readability-lxml 0.8.1

前提・実現したいこと

クローリングフレームワークのScrapyを使用してAWS S3のバケットにアップロードしたクロール結果htmlファイルを
Pythonプログラムから参照し、htmlから本文抽出して検索エンジンのElasticsearchにインデックスする正しい方法を教えていただきたいです。

今回は以下の書籍の内容を組み合わせて、実験を行なっています。
「Python クローリング&スクレイピングデータ収集・解析のための実践開発ガイド」
https://scraping-book.com/

【クロール & S3へアップロード】
はてなブックマークの新着エントリーのクロール結果をスクレイピング処理を行わないでS3にアップロードしました。
S3へアップロードするためのライブラリはs3pipelineを使用しています。

python
1import scrapy
2
3from myproject.items import Page
4# from myproject.utils import get_content
5
6# S3 lib
7from s3pipeline import Page
8
9class BroadSpider(scrapy.Spider):
10    name = 'broad_s3'
11
12    start_urls = ['http://b.hatena.ne.jp/entrylist/all']
13
14    def parse(self, response):
15
16        for url in response.css('.entrylist-contents-title > a::attr("href")').getall():
17            # parse_page() 
18            yield scrapy.Request(url, callback=self.parse_page)
19
20        url_more = response.css('.entrylist-readmore > a::attr("href")').re_first(r'.*?page=\d{1}$')
21        if url_more:
22            yield response.follow(url_more)
23
24    def parse_page(self, response):
25        """
26        スクレイピング処理を行わず、単にs3pipeline.Pageオブジェクトをyieldする
27        """
28        yield Page.from_response(response)
29

S3へのアップロードは同じく、書籍の著書様のブログを参考にしました。
https://orangain.hatenablog.com/entry/serverless-crawler
https://github.com/orangain/serverless-crawler

S3へのアップロードと、ダウンロードは成功しましたが、

同書籍、クロール実行時にSpiderからreadability-lxmlを利用して本文抽出を行なっていたように、

S3からダウンロードしたクロール済みhtmlファイルに対してreadability-lxmlで本文抽出した結果をElasticsearchへインデックス出来れば...と考えています。

【クロール実行時に本文抽出】

python
1    # Spider callback関数
2    def parse_page(self, response):
3        """
4        個別のWebページをパースする。
5        """
6        # utils.pyに定義したget_content()関数でタイトルと本文を抽出する。
7        title, content = get_content(response.text)
8        # Pageオブジェクトを作成してyieldする。
9        yield Page(url=response.url, title=title, content=content)
10
11     def get_content(html: str) -> Tuple[str, str]:
12         """
13         HTMLの文字列から (タイトル, 本文) のタプルを取得する。
14         """
15         document = readability.Document(html)
16         content_html = document.summary()
17         # HTMLタグを除去して本文のテキストのみを取得する。
18         content_text = lxml.html.fromstring(content_html).text_content().strip()
19         short_title = document.short_title()
20
21         return short_title, content_text

【ダウンロードしたhtmlから本文抽出したい】

python
1def main():
2    """
3    メインとなる処理。
4    はてブ クロール結果をESへインデックス
5    """
6
7    # Elasticsearchのクライアントを作成する。
8    es = Elasticsearch(['localhost:9200'])
9    create_pages_index(es)  # pagesインデックスを作成。
10
11    # S3のバケットを参照し、保存したhtmlを取得する
12    process_object('repo-XXX', 'broad_s3/XXX/XXX.jl.gz')
13
14def process_object(bucket_name, object_key):
15    """
16    1つのS3オブジェクトを処理する。
17    """
18    s3 = boto3.client('s3')
19
20    for page in read_pages(s3, bucket_name, object_key):
21        page = scrape_from_page(page)
22
23        # パース結果をElasticsearchへインデックスしたい
24        doc_id = hashlib.sha1(page['url'].encode('utf-8')).hexdigest()
25        # pagesインデックスにインデックス化（保存）する。
26        es.index(index='pages', doc_type='_doc', id=doc_id, body=page)
27
28def read_pages(s3, bucket_name, object_key):
29    """
30    S3オブジェクトからページをyieldするジェネレーター
31    """
32    use_gzip = object_key.endswith('.gz')
33
34    bio = io.BytesIO()
35    s3.download_fileobj(bucket_name, object_key, bio)
36    bio.seek(0)
37    f = gzip.GzipFile(mode='rb', fileobj=bio) if use_gzip else bio
38
39    for line in f:
40        page = json.loads(line)
41        yield page
42
43def scrape_from_page(page):
44
45    # ??? page -> Spiderのresponse.textと同じ形式に復元したい
46    
47    document = readability.Document(page)
48    content_html = document.summary()
49    # HTMLタグを除去して本文のテキストのみを取得する。
50    return lxml.html.fromstring(content_html).text_content().strip()