前書き
chatGPT が登場して以来、テキスト処理の能力が一段と向上しました。以前はウェブ上のテキストコンテンツをクロールした後、テキストをクリーニングするためのプログラムを書く必要がありましたが、chatGPT のおかげで、すべてのタイプや形式のテキストコンテンツを簡単にクリーニングすることができるようになりました。
クリーニングされた記事には、核心的な視点の抽出、記事の書き換えなど、さまざまな操作を行うことができます。chatGPT を使用すれば、これらの操作は非常に簡単に解決できます。
2019 年には、公式アカウントの記事をクロールする方法についての記事を書いたことがありますが、今でも有効です。ただし、以前はクロールした記事に対してほとんど処理を行っていませんでしたが、今は異なります。ただし、法律や規制を遵守する必要があります。
記事の URL を取得する
ウェブリソースをクロールするためには、まず対象ウェブサイトの URL を見つけることが最も重要です。そして、パターンを見つけて、アドレスを一つずつまたはマルチスレッドでクロールします。一般的なアドレスの取得方法は、ウェブページのページングを通じて、pageNum = num などのパラメータのパターンを推測することです。num を累積させるだけです。もう一つは HTML に隠されているもので、現在のウェブページのハイパーリンクを解析し、のようなものを取り出して、後続のクロールアドレスとして使用します。
残念ながら、WeChat 公式アカウントではこれらの方法のいずれも使用するのが難しいです。ウェブページで WeChat 公式アカウントのアドレスを開くと、その URL の形式を確認できます。
https://mp.weixin.qq.com/s?__biz=MzIxMTgyODczNg==&mid=2247483660&idx=1&sn=2c14b9b416e2d8eeed0cbbc6f44444e9&chksm=974e2863a039a1752605e6e76610eb39e5855c473edab16ab2c7c8aa2f624b4d892d26130110&token=20884314&lang=zh_CN#rd
前半部分のドメイン名以外のパラメータはまったく規則性がなく、また、一つの記事内で次の記事にリンクすることもできません。したがって、一つの記事からこの公式アカウントのすべての記事をクロールすることはできません。
しかし、幸いなことに、特定の公式アカウントのすべての記事のアドレスを一括で取得する方法があります。それを保存しておけば、後続のクロールははるかに簡単になります。
1. まず、公式アカウントを持っている必要があります。持っていない場合は、登録することができます。これは前提条件です。公式アカウントの登録手順は簡単ですので、自分で操作してください。
2. 登録後、WeChat 公式プラットフォームにログインし、左側の下書きボックスをクリックして、新しい記事を作成します。
3. 新しい記事作成ページに入り、上部のハイパーリンクをクリックすると、ポップアップウィンドウが表示されます。公式アカウントの記事を選択し、クロールしたい公式アカウントの名前を入力します。
4. 選択すると、この公式アカウントのすべての記事のリストが表示されます。この時点で、F12 を開いて、ウェブページのネットワークリクエストを確認します。
次のページをクリックすると、リクエストの URL とビジネスパラメータ、およびリクエストヘッダパラメータが表示されます。
上記はいくつかのビジネスパラメータです。これらのパラメータは非常に理解しやすく、重要なのは begin は何番目からクエリを開始するか、count は一度に何個のリストをクエリするか、fakeid は公式アカウントの一意の識別子であり、各公式アカウントが異なることです。他の公式アカウントをクロールする場合は、このパラメータを変更するだけです。random は省略できます。応答結果も確認できます。
コードの作成
上記の情報を持っているので、コードを書くことができます。私は Python3.8 を使用しています。まず、URL、ヘッダー、および必要なパラメータを定義します。
# ターゲットURL
url = "https://mp.weixin.qq.com/cgi-bin/appmsg"
# ヘッダーパラメータ
headers = {
"Cookie": "ua_id=YF6RyP41YQa2QyQHAAAAAGXPy_he8M8KkNCUbRx0cVU=; pgv_pvi=2045358080; pgv_si=s4132856832; uuid=48da56b488e5c697909a13dfac91a819; bizuin=3231163757; ticket=5bd41c51e53cfce785e5c188f94240aac8fad8e3; ticket_id=gh_d5e73af61440; cert=bVSKoAHHVIldcRZp10_fd7p2aTEXrTi6; noticeLoginFlag=1; remember_acct=mf1832192%40smail.nju.edu.cn; data_bizuin=3231163757; data_ticket=XKgzAcTceBFDNN6cFXa4TZAVMlMlxhorD7A0r3vzCDkS++pgSpr55NFkQIN3N+/v; slave_sid=bU0yeTNOS2VxcEg5RktUQlZhd2xheVc5bjhoQTVhOHdhMnN2SlVIZGRtU3hvVXJpTWdWakVqcHowd3RuVF9HY19Udm1PbVpQMGVfcnhHVGJQQTVzckpQY042QlZZbnJzel9oam5SdjRFR0tGc0c1eExKQU9ybjgxVnZVZVBtSmVnc29ZcUJWVmNWWEFEaGtk; slave_user=gh_d5e73af61440; xid=93074c5a87a2e98ddb9e527aa204d0c7; openid2ticket_obaWXwJGb9VV9FiHPMcNq7OZzlzY=lw6SBHGUDQf1lFHqOeShfg39SU7awJMxhDVb4AbVXJM=; mm_lang=zh_CN",
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/62.0.3202.62 Safari/537.36",
}
# ビジネスパラメータ
data = {
"token": "1378111188",
"lang": "zh_CN",
"f": "json",
"ajax": "1",
"action": "list_ex",
"begin": "0",
"count": "5",
"query": "",
"fakeid": "MzU5MDUzMTk5Nw==",
"type": "9",
}
cookie と token は、自分のリクエスト URL に基づいて変更する必要があります。そして、リクエストを送信してレスポンスを取得し、レスポンス結果を解析します。
content_list = []
for i in range(20):
data["begin"] = i*5
time.sleep(5)
# getメソッドを使用して送信する
content_json = requests.get(url, headers=headers, params=data).json()
# 各ページのデータが含まれるjsonが返されます
for item in content_json["app_msg_list"]:
# 各ページの記事のタイトルと対応するURLを抽出します
items = []
items.append(item["title"])
items.append(item["link"])
t = time.localtime(item["create_time"])
items.append(time.strftime("%Y-%m-%d %H:%M:%S", t))
content_list.append(items)
最初の for ループはクロールするページ数です。ここでは 20 ページを一度にクロールすることをお勧めします。各ページに 5 つのデータがありますので、100 の記事になります。公式アカウントの履歴記事リストの総ページ数を把握しておく必要があります。この数はページ数より小さくする必要があります。data ["begin"] を変更することで、何番目からクエリを開始するかを示します。クロールが多すぎたり頻繁すぎると、IP や Cookie がブロックされる可能性があるため、数秒待機する必要があります。さもないと、公式アカウント自体がブロックされる可能性があります。
最後に、タイトルと URL を保存するだけです。保存した後、一つずつクロールすることができます。
name = ['title', 'link', 'create_time']
test = pd.DataFrame(columns=name, data=content_list)
test.to_csv("url.csv", mode='a', encoding='utf-8')
すべての履歴記事を取得するためには、app_msg_cnt から記事の総数を取得し、ページ数を計算して、すべての記事を一度にクロールすることができます。
content_json = requests.get(url, headers=headers, params=data).json()
count = int(content_json["app_msg_cnt"])
print(count)
page = int(math.ceil(count / 5))
print(page)
頻度が高すぎないようにするために、一定の時間をスリープする必要があります。10 回クロールした後、プログラムを数秒間スリープさせることができます。
if (i > 0) and (i % 10 == 0):
name = ['title', 'link', 'create_time']
test = pd.DataFrame(columns=name, data=content_list)
test.to_csv("url.csv", mode='a', encoding='utf-8')
print("第" + str(i) + "回保存成功")
content_list = []
time.sleep(random.randint(60,90))
else:
time.sleep(random.randint(15,25))
完全なコードは GitHub にありますので、直接ダウンロードして使用することができます。
https://github.com/cxyxl66/WeChatCrawler