データ収集の前提と対象ドメインの選定

ウェブ上の公開情報を自動収集する際、最初の工程は抽出が容易な対象URLの特定です。単純なHTTPリクエストを送信し、レスポンスとして返されるHTML構造がブラウザでのレンダリング結果と一致しているか確認することで、初期段階のアクセス制限やクライアントサイドレンダリングの有無を判定できます。学習目的では、複雑な認証フローや高度なボット対策が施されていないドメインを選択するのが効率的です。

本記事では、地域別の気象予報と関連する飲食店情報を紐づけて取得するフローを実装します。

都道府県と市区町村のマッピング取得

気象データは地理情報に依存するため、まずは階層構造を持つ都市一覧を解析します。以下は、指定されたURLから都道府県名と配下の都市リストを抽出する実装例です。

import urllib.request
from bs4 import BeautifulSoup

def fetch_region_mapping(source_url: str) -> dict:
    req = urllib.request.Request(
        source_url,
        headers={'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36'}
    )
    with urllib.request.urlopen(req) as response:
        raw_bytes = response.read()
        doc = BeautifulSoup(raw_bytes, 'html.parser')

    region_map = {}
    for header in doc.find_all('h2'):
        province = header.get_text(strip=True)
        cities = []
        sibling = header.find_next_sibling()
        if sibling:
            for link in sibling.find_all('a'):
                city_name = link.get_text(strip=True)
                if city_name:
                    cities.append(city_name)
        region_map[province] = cities
    return region_map

if __name__ == '__main__':
    data = fetch_region_mapping('https://www.tianqi.com/chinacity.html')
    for prov, cts in data.items():
        print(f"[{prov}] {', '.join(cts[:3])}...")

気象情報と地域グルメの並列抽出

都市の階層構造が把握できたら、次は特定エリアのリアルタイム天気予報と関連する飲食店リンクを並列で取得します。HTMLツリー内で別々のクラス名を持つ要素を対象に、必要な属性を抽出します。

import urllib.request
from bs4 import BeautifulSoup

def parse_city_metrics(target_url: str) -> dict:
    req = urllib.request.Request(
        target_url,
        headers={'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36'}
    )
    with urllib.request.urlopen(req) as res:
        soup = BeautifulSoup(res.read(), 'html.parser')

    results = {'main_weather': [], 'food_links': [], 'detailed_conditions': {}}

    for tag in soup.select('div.mainWeather a'):
        if 'd15' not in tag.get('class', []):
            results['main_weather'].append(tag.get('title', '').strip())

    for food_list in soup.select('ul.paihang_good_food'):
        for link in food_list.find_all('a'):
            results['food_links'].append({
                'url': link.get('href', ''),
                'name': link.get('title', '')
            })

    for block in soup.select('dl.weather_info'):
        loc = block.find('h1').get_text(strip=True)
        results['detailed_conditions'][loc] = {
            'date': block.find('dd', class_='week').get_text(strip=True),
            'temp': block.find('p', class_='now').get_text(strip=True),
            'humidity': block.find('dd', class_='shidu').get_text(strip=True),
            'air': block.find('dd', class_='kongqi').h5.get_text(strip=True)
        }
    return results

グルメ詳細ページのテキスト解析

気象条件に適した移動先が決まったら、次に現地で推奨される料理の特徴や選定理由を収集します。対象ページ内の特定のテキストノードをクリーンアップして抽出します。

import urllib.request
from bs4 import BeautifulSoup

def extract_cuisine_details(detail_url: str) -> str:
    req = urllib.request.Request(
        detail_url,
        headers={'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36'}
    )
    with urllib.request.urlopen(req) as resp:
        soup = BeautifulSoup(resp.read(), 'html.parser')

    target_span = soup.find('span', class_='traffic')
    if target_span:
        return ''.join(target_span.stripped_strings)
    return "データが見つかりません"

CLIインタフェースへの統合

個別の抽出処理を独立したスクリプトとして実行する代わりに、コマンドライン上で対話型のナビゲーションを提供する統合環境を構築します。外部モジュールを活用してターミナル出力を装飾し、キーボード入力による分岐処理を実装します。

import urllib.request
from bs4 import BeautifulSoup
from random import choice
from colorama import init, Fore
from termcolor import colored
from readchar import readkey
from xpinyin import Pinyin
import os

class WeatherFoodApp:
    def __init__(self):
        self.headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36'}
        self.pinyin_conv = Pinyin()
        self.city_db = {}
        self.district_weather = []
        self.cuisine_list = []
        self.colors = [Fore.GREEN, Fore.YELLOW, Fore.BLUE, Fore.CYAN, Fore.MAGENTA, Fore.RED]
        init(autoreset=True)

    def _fetch_html(self, url):
        req = urllib.request.Request(url, headers=self.headers)
        with urllib.request.urlopen(req) as r:
            return BeautifulSoup(r.read(), 'html.parser')

    def load_regions(self):
        doc = self._fetch_html('https://www.tianqi.com/chinacity.html')
        for h2 in doc.find_all('h2'):
            prov = h2.get_text(strip=True)
            cts = [a.get_text(strip=True) for a in h2.find_next_sibling().find_all('a')]
            self.city_db[prov] = cts

    def get_metrics(self, prov_pinyin):
        doc = self._fetch_html(f'https://www.tianqi.com/{prov_pinyin}/')
        self.district_weather.clear()
        self.cuisine_list.clear()

        for a in doc.select('div.mainWeather a'):
            if 'd15' not in a.get('class', []):
                self.district_weather.append(a.get('title', '').strip())

        for ul in doc.select('ul.paihang_good_food'):
            for a in ul.find_all('a'):
                self.cuisine_list.append({'url': a.get('href'), 'name': a.get('title')})

        for dl in doc.select('dl.weather_info'):
            print(colored(f"■ {dl.find('h1').get_text(strip=True)}", choice(self.colors)))
            print(colored(f"  日付: {dl.find('dd', class_='week').get_text(strip=True)}", choice(self.colors)))
            print(colored(f"  気温: {dl.find('p', class_='now').get_text(strip=True)}", choice(self.colors)))
            print(colored(f"  湿度: {dl.find('dd', class_='shidu').get_text(strip=True)}", choice(self.colors)))
            print(colored(f"  空気質: {dl.find('dd', class_='kongqi').h5.get_text(strip=True)}", choice(self.colors)))

    def show_cuisine(self, url):
        doc = self._fetch_html(f'https://www.tianqi.com{url}')
        span = doc.find('span', class_='traffic')
        if span:
            print(colored(''.join(span.stripped_strings), choice(self.colors)))

    def run(self):
        self.load_regions()
        cmds = {'q': '終了', 'm': '地域選択', 'h': 'エリア天気', 'f': 'グルメ検索', 'c': '画面クリア'}
        while True:
            print(f"\n[操作コマンド] {' | '.join([f'{k}({v})' for k, v in cmds.items()])}")
            key = readkey()
            if key == 'q': break
            elif key == 'c': os.system('cls' if os.name=='nt' else 'clear')
            elif key == 'h':
                for i, w in enumerate(self.district_weather): print(f"{i}: {w}")
            elif key == 'f':
                for i, food in enumerate(self.cuisine_list): print(f"{i}: {food['name']}")
                try:
                    idx = int(input("対象番号を入力: "))
                    if 0 <= idx < len(self.cuisine_list): self.show_cuisine(self.cuisine_list[idx]['url'])
                except ValueError: pass
            elif key == 'm':
                for i, (p, c) in enumerate(self.city_db.items()): print(f"{i}: {p}")
                try:
                    idx = int(input("地域番号を入力: "))
                    if idx in self.city_db:
                        pinyin = self.pinyin_conv.get_pinyin(list(self.city_db.keys())[idx], '')
                        self.get_metrics(pinyin)
                except (ValueError, IndexError): pass

if __name__ == '__main__':
    app = WeatherFoodApp()
    app.run()

個別の抽出ロジックをクラスベースのアーキテクチャに再編成することで、状態管理と画面出力の分離が容易になります。キー入力イベントを監視するループ構造により、外部依存の少ない軽量なデータ参照環境が実現します。取得したメタデータを外部ファイルやデータベースへ逐次保存するモジュールを追加することで、スクリプトは定常的な情報収集パイプラインへと拡張可能です。