記事取得python
pythnで自動取得
出来てるフロー
- urlからhtml取得
- 必要な情報をテキストファイルに保存
まだ目標 3) geminiのAPIで送る
スレログをまとめるプロンプト と組み合わせる
https://rentry.org/oyeg7tr8
windows の anadonda の python を想定
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 | #! python
TARGET_URL = 'https://mercury.bbspink.com//test/read.cgi/onatech/1756914267/1-999'
import requests
## pip install requests
import os
import re
from bs4 import BeautifulSoup
import chardet
import json
def get_file_encoding(file_path: str) -> tuple[str | None, float]:
"""
指定されたファイルの文字コードと確信度を推定して返します。
Args: file_path (str): 調査するファイルのパス。
Returns: tuple[str | None, float]: (推定された文字コード, 確信度) のタプル。
ファイルが存在しない、または判定できなかった場合は (None, 0.0) を返します。
"""
if not os.path.exists(file_path):
print(f"エラー: ファイルが見つかりません。パス: {file_path}")
return (None, 0.0)
try:
# ファイルを必ずバイナリモード('rb')で開く
with open(file_path, 'rb') as f:
# ファイル全体を読み込む (巨大なファイルの場合は一部だけ読み込むなどの工夫が必要)
raw_data = f.read()
# chardetで解析を実行
result = chardet.detect(raw_data)
encoding = result['encoding']
confidence = result['confidence']
return (encoding, confidence)
except Exception as e:
print(f"エラー: ファイルの解析中に問題が発生しました。 {e}")
return (None, 0.0)
#enddef
#enddef
def get_clean_text_from_html_block(htmltext: str) -> str:
"""
Args: TMLtext (str): 読み込むHTMLtext。
Returns: str: 指定ブロックから抽出された、整形済みの生テキストデータ。
"""
print("=== HTML TEXT to json ===")
try:
soup = BeautifulSoup(htmltext, 'html.parser')
# 指定したIDとクラスを持つブロックを検索
# find()は最初にマッチした要素を返します。
# class_ はPythonの予約語 'class' との衝突を避けるためのものです。
###
title_block = soup.find("div",id="threadtitle")
thread_title = title_block.get_text(separator='\n', strip=True).strip()
json_o = {
"thread_title" : thread_title ,
"post_list" : {
1 : {
"id" : 1 ,
"username" : 1 ,
"date" : 1 ,
"content" : 1
}
}
}
###
getresobjlist = soup.find_all('article', id=re.compile('^[0-9]+'))
for resobj in getresobjlist :
# print( resobj.prettify() ) # debug
postid = resobj.find( class_="postid").get_text()
postusername = resobj.find( class_="postusername").get_text() + " " + resobj.find( class_="uid").get_text()
postdate = resobj.find( class_="date").get_text()
content = resobj.find( class_="post-content").get_text(separator='\n', strip=True)
json_o["post_list"][int(postid)] = {
"id" : postid ,
"username" : postusername ,
"date" : postdate ,
"content" : content
}
#endfor
return json_o
#print(json.dumps(json_o, indent=4, ensure_ascii=False))
except Exception as e:
print(f"エラー: HTMLファイルの読み込みまたは解析中に問題が発生しました。 {e}")
return ""
#enddef
# --- requests.get() を直接使用する関数 ---
def get_html_simple(url: str):
"""
指定されたURLのHTMLを取得。
Args: url (str): 取得したいページのURL。
"""
print(f"--- 開始: {url} ---")
# ブラウザからのアクセスを偽装するためのヘッダー
headers = { 'User-Agent': 'AppleWebKit/537.36 (KHTML, like Gecko)' }
# Cookieが必要な場合は、辞書形式で用意する
cookies = { 'your_cookie_name': 'your_cookie_value' }
try:
# verify=False はSSL証明書のエラーを無視しますが、セキュリティリスクがあるため非推奨です。
# 安全な接続が確認されている場合や、自己署名証明書の場合に限定して使用してください。
response = requests.get(url, headers=headers, cookies=cookies, timeout=10, verify=True)
# ステータスコードが200番台でない場合、例外を発生させる
response.raise_for_status()
# 文字化け対策: レスポンスヘッダーからエンコーディングを取得
response.encoding = response.apparent_encoding
"""
# アプローチ2: requests.Session() を使用
# セッションオブジェクトを作成
session = requests.Session()
# セッション全体で共通のヘッダーを設定
session.headers.update(headers)
# セッションに初期Cookieを設定 (ログイン後のページ取得などに利用)
session.cookies.update(cookies)
# Sessionオブジェクトを使ってリクエストを送信
# verify=True がデフォルトです。証明書エラーの場合は False にしますが非推奨です。
response = session.get(url, timeout=10, verify=True)
"""
return response.text , response.encoding
except requests.exceptions.HTTPError as e:
print(f"エラー: HTTPエラーが発生しました。 Status Code: {e.response.status_code}")
except requests.exceptions.ConnectionError as e:
print(f"エラー: 接続エラーが発生しました。URLを確認してください。")
except requests.exceptions.Timeout as e:
print(f"エラー: リクエストがタイムアウトしました。")
except requests.exceptions.SSLError as e:
print(f"エラー: SSL証明書エラーが発生しました。'verify=False'を試すこともできますが、注意が必要です。")
except requests.exceptions.RequestException as e:
print(f"エラー: 不明なリクエストエラーが発生しました。 {e}")
except Exception as e:
print(f"エラー: 予期せぬエラーが発生しました。 {e}")
print(f"失敗: HTMLの取得に失敗しました。")
return False , False
#enddef
def batch_main() :
# 保存したいURLとファイルパスを設定
#target_url = 'https://mercury.bbspink.com//test/read.cgi/onatech/1756914267/1-999'
# html
htmltext , enc = get_html_simple(TARGET_URL)
if htmltext is False :
return
print(enc)
# html to json
res_data = get_clean_text_from_html_block(htmltext)
if res_data == "" :
return
# fname
thread_title = res_data["thread_title"]
post_size = len(res_data["post_list"])
print(thread_title)
print(post_size)
# os.makedirsでディレクトリが存在しない場合に作成
# os.makedirs(os.path.dirname(file_path), exist_ok=True)
html_file_name = thread_title + f"_{post_size}.html"
# ファイルへ保存
with open( html_file_name , 'w', encoding=enc) as f:
f.write(htmltext)
json_text = json.dumps(res_data, indent=4, ensure_ascii=False)
print( json_text )
json_file_name = thread_title + f"_{post_size}.json.txt"
with open( json_file_name , 'w', encoding="utf-8") as f:
f.write(json_text)
#print(f"成功: HTMLを {file_path} に保存しました。")
# --- メインの実行部分 ---
if __name__ == '__main__':
batch_main()
|