Pythonにおけるスクレピング実務で知っておくべきポイント10選

import requests
from bs4 import BeautifulSoup

# ウェブページのURLを指定
url = 'https://example.com'

# ページの内容を取得
response = requests.get(url)

# BeautifulSoupオブジェクトを作成
soup = BeautifulSoup(response.text, 'html.parser')

# ここから実際のスクレイピング処理を書いていきます

import requests
from bs4 import BeautifulSoup

# ウェブページのURLを指定
url = 'https://www.python.org/'

# ページの内容を取得
response = requests.get(url)

# BeautifulSoupオブジェクトを作成
soup = BeautifulSoup(response.text, 'html.parser')

# タイトルを取得
title = soup.title.string

print(f"ページのタイトル: {title}")

import requests
from bs4 import BeautifulSoup

url = 'https://www.python.org/'
response = requests.get(url)
soup = BeautifulSoup(response.text, 'html.parser')

# class名が'news-title'の最初の<a>タグを探す
news_title = soup.find('a', class_='news-title')

if news_title:
    print(f"最新ニュース: {news_title.text.strip()}")
else:
    print("ニュースが見つかりませんでした。")

import requests
from bs4 import BeautifulSoup

url = 'https://www.python.org/'
response = requests.get(url)
soup = BeautifulSoup(response.text, 'html.parser')

# id='top'のul要素内のすべてのli要素を取得
menu_items = soup.select('#top ul li')

print("ナビゲーションメニューの項目:")
for item in menu_items:
    # アンカータグのテキストを取得
    link_text = item.find('a').text.strip()
    print(f"- {link_text}")

ナビゲーションメニューの項目:
- About
- Downloads
- Documentation
- Community
- Success Stories
- News
- Events
- PSF

import scrapy

class NewsItem(scrapy.Item):
    title = scrapy.Field()
    content = scrapy.Field()

scrapy startproject python_blog
cd python_blog
scrapy genspider python_blog_spider blog.python.org

import scrapy

class PythonBlogSpider(scrapy.Spider):
    name = 'python_blog_spider'
    allowed_domains = ['blog.python.org']
    start_urls = ['https://blog.python.org/']

    def parse(self, response):
        for post in response.css('div.date-outer'):
            yield {
                'title': post.css('h3.post-title::text').get().strip(),
                'date': post.css('h2.date-header::text').get().strip()
            }

        next_page = response.css('a.blog-pager-older-link::attr(href)').get()
        if next_page is not None:
            yield response.follow(next_page, self.parse)

[
    {
        "title": "Python 3.11.3 and 3.10.11 are now available",
        "date": "Monday, April 03, 2023"
    },
    {
        "title": "Python 3.11.2, 3.10.10, 3.9.16, 3.8.16, 3.7.16, and 3.12.0 alpha 5 are now available",
        "date": "Tuesday, February 07, 2023"
    },
    ...
]

import sqlite3

class SQLitePipeline:
    def __init__(self):
        self.conn = sqlite3.connect('python_blog.db')
        self.cur = self.conn.cursor()
        self.cur.execute("""
        CREATE TABLE IF NOT EXISTS posts
        (title TEXT, date TEXT)
        """)

    def process_item(self, item, spider):
        self.cur.execute("INSERT INTO posts (title, date) VALUES (?, ?)",
                         (item['title'], item['date']))
        self.conn.commit()
        return item

    def close_spider(self, spider):
        self.conn.close()

ITEM_PIPELINES = {
   'python_blog.pipelines.SQLitePipeline': 300,
}

import sqlite3

conn = sqlite3  .connect('python_blog.db')
cur = conn.cursor()
cur.execute("SELECT * FROM posts")
for row in cur.fetchall():
    print(row)
conn.close()

('Python 3.11.3 and 3.10.11 are now available', 'Monday, April 03, 2023')
('Python 3.11.2, 3.10.10, 3.9.16, 3.8.16, 3.7.16, and 3.12.0 alpha 5 are now available', 'Tuesday, February 07, 2023')
...

from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.common.keys import Keys
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
import time

# ChromeDriverの初期化
driver = webdriver.Chrome()

try:
    # GitHubのログインページにアクセス
    driver.get("https://github.com/login")

    # ユーザー名とパスワードを入力
    username = driver.find_element(By.ID, "login_field")
    password = driver.find_element(By.ID, "password")

    username.send_keys("your_username")
    password.send_keys("your_password")

    # ログインボタンをクリック
    login_button = driver.find_element(By.NAME, "commit")
    login_button.click()

    # プロフィールページに移動
    driver.get("https://github.com/your_username")

    # 名前を取得
    name = WebDriverWait(driver, 10).until(
        EC.presence_of_element_located((By.CLASS_NAME, "p-name"))
    )
    print(f"名前: {name.text}")

    # バイオを取得
    bio = driver.find_element(By.CLASS_NAME, "p-note")
    print(f"バイオ: {bio.text}")

except Exception as e:
    print(f"エラーが発生しました: {e}")

finally:
    # ブラウザを閉じる
    driver.quit()

名前: John Doe
バイオ: Python developer and open source enthusiast.

from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.common.action_chains import ActionChains
import time

driver = webdriver.Chrome()

try:
    # GitHubのトレンディングページにアクセス
    driver.get("https://github.com/trending")

    # スクロール回数
    scroll_count = 5
    repos = []

    for _ in range(scroll_count):
        # 新しいリポジトリ情報を取得
        repo_elements = driver.find_elements(By.CSS_SELECTOR, ".Box-row")

        for repo in repo_elements:
            name = repo.find_element(By.CSS_SELECTOR, "h3").text.strip()
            stars = repo.find_element(By.CSS_SELECTOR, "span.d-inline-block.float-sm-right").text.strip()
            repos.append((name, stars))

        # ページの最下部までスクロール
        driver.execute_script("window.scrollTo(0, document.body.scrollHeight);")
        time.sleep(2)  # コンテンツの読み込みを待つ

    # 結果を表示
    for name, stars in repos:
        print(f"リポジトリ名: {name}, スター数: {stars}")

except Exception as e:
    print(f"エラーが発生しました: {e}")

finally:
    driver.quit()

リポジトリ名: openai/chatgpt-retrieval-plugin, スター数: 2,971
リポジトリ名: yt-dlp/yt-dlp, スター数: 120
リポジトリ名: microsoft/DeepSpeed, スター数: 321
...

import requests

headers = {
    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36'
}

response = requests.get('https://example.com', headers=headers)
print(response.status_code)

import requests
import time

for url in urls:
    response = requests.get(url)
    print(response.status_code)
    time.sleep(1)  # 1秒間待機

import requests
import time
import random

for url in urls:
    response = requests.get(url)
    print(response.status_code)
    time.sleep(random.uniform(1, 3))  # 1~3秒のランダムな待機時間

import requests
import time

def get_with_retry(url, max_retries=5):
    for i in range(max_retries):
        try:
            response = requests.get(url)
            response.raise_for_status()
            return response
        except requests.exceptions.RequestException:
            wait_time = 2 ** i  # 指数関数的に待機時間を増やす
            print(f"エラーが発生しました。{wait_time}秒後にリトライします。")
            time.sleep(wait_time)

    raise Exception("最大リトライ回数を超えました。")

response = get_with_retry('https://example.com')
print(response.status_code)

import requests

proxies = {
    'http': 'http://10.10.1.10:3128',
    'https': 'http://10.10.1.10:1080',
}

response = requests.get('https://example.com', proxies=proxies)
print(response.status_code)

import pandas as pd

# スクレイピングで取得したデータ（仮定）
data = [
    {"name": "商品A", "price": "1000円", "category": "電化製品"},
    {"name": "商品B", "price": "2000円", "category": "家具"},
    {"name": "商品C", "price": "1500円", "category": "電化製品"}
]

# DataFrameの作成
df = pd.DataFrame(data)

# 価格の'円'を取り除き、数値型に変換
df['price'] = df['price'].str.replace('円', '').astype(int)

# カテゴリごとの平均価格を計算
category_avg = df.groupby('category')['price'].mean()

print(df)
print("\nカテゴリごとの平均価格:")
print(category_avg)

     name  price category
0    商品A   1000    電化製品
1    商品B   2000      家具
2    商品C   1500    電化製品

カテゴリごとの平均価格:
category
家具      2000.0
電化製品    1250.0
Name: price, dtype: float64

import pandas as pd
import matplotlib.pyplot as plt

# 前回のデータを再利用
data = [
    {"name": "商品A", "price": "1000円", "category": "電化製品"},
    {"name": "商品B", "price": "2000円", "category": "家具"},
    {"name": "商品C", "price": "1500円", "category": "電化製品"}
]

df = pd.DataFrame(data)
df['price'] = df['price'].str.replace('円', '').astype(int)

# カテゴリごとの平均価格を計算
category_avg = df.groupby('category')['price'].mean()

# グラフの作成
plt.figure(figsize=(10, 6))
category_avg.plot(kind='bar')
plt.title('カテゴリ別平均価格')
plt.xlabel('カテゴリ')
plt.ylabel('平均価格 (円)')
plt.xticks(rotation=0)
plt.tight_layout()
plt.show()

import pandas as pd
from sklearn.model_selection import train_test_split
from sklearn.linear_model import LinearRegression
from sklearn.preprocessing import OneHotEncoder
from sklearn.compose import ColumnTransformer
from sklearn.pipeline import Pipeline

# より多くのサンプルデータを用意
data = [
    {"name": "商品A", "price": 1000, "category": "電化製品", "rating": 4.5},
    {"name": "商品B", "price": 2000, "category": "家具", "rating": 4.2},
    {"name": "商品C", "price": 1500, "category": "電化製品", "rating": 3.8},
    {"name": "商品D", "price": 3000, "category": "家具", "rating": 4.7},
    {"name": "商品E", "price": 1200, "category": "電化製品", "rating": 4.0}
]

df = pd.DataFrame(data)

# 特徴量とターゲットを分離
X = df[['category', 'rating']]
y = df['price']

# データを訓練セットとテストセットに分割
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)

# カテゴリ変数をワンホットエンコーディング
preprocessor = ColumnTransformer(
    transformers=[
        ('cat', OneHotEncoder(drop='first'), ['category'])
    ])

# モデルの作成
model = Pipeline([
    ('preprocessor', preprocessor),
    ('regressor', LinearRegression())
])

# モデルの訓練
model.fit(X_train, y_train)

# テストデータで予測
predictions = model.predict(X_test)

# 結果の表示
for true, pred in zip(y_test, predictions):
    print(f"実際の価格: {true}, 予測価格: {pred:.2f}")

# モデルの性能評価
score = model.score(X_test, y_test)
print(f"\nモデルの決定係数 (R^2): {score:.2f}")

実際の価格: 1200, 予測価格: 1200.00
モデルの決定係数 (R^2): 1.00

#!/usr/bin/env python3
import os
import sys
import logging
from datetime import datetime

# 環境変数の設定
os.environ['PATH'] = '/usr/local/bin:/usr/bin:/bin'

# ログの設定
logging.basicConfig(filename='/path/to/your/scraping.log', level=logging.INFO)

try:
    # スクレイピングのメイン処理
    # ここに実際のスクレイピングコードを記述します
    logging.info(f"スクレイピング成功: {datetime.now()}")
except Exception as e:
    logging.error(f"エラーが発生しました: {e}")
    sys.exit(1)

import json
import requests
from bs4 import BeautifulSoup

def lambda_handler(event, context):
    url = "https://example.com"
    response = requests.get(url)
    soup = BeautifulSoup(response.text, 'html.parser')

    # ここでスクレイピングの処理を行います
    title = soup.title.string

    return {
        'statusCode': 200,
        'body': json.dumps(f'Scraped title: {title}')
    }

import requests
import logging
from slack_sdk import WebClient
from slack_sdk.errors import SlackApiError

# Slackの設定
SLACK_TOKEN = "your-slack-token"
CHANNEL_ID = "your-channel-id"

client = WebClient(token=SLACK_TOKEN)

def send_slack_notification(message):
    try:
        response = client.chat_postMessage(
            channel=CHANNEL_ID,
            text=message
        )
    except SlackApiError as e:
        logging.error(f"Slackへの通知に失敗しました: {e}")

# スクレイピングの処理
try:
    # ここにスクレイピングのコードを記述します
    # エラーが発生した場合は例外が発生します
    pass
except Exception as e:
    error_message = f"スクレイピングでエラーが発生しました: {str(e)}"
    send_slack_notification(error_message)
    logging.error(error_message)

import requests
from bs4 import BeautifulSoup
import sqlite3
from datetime import datetime

# データベース接続
conn = sqlite3.connect('news_articles.db')
c = conn.cursor()

# テーブル作成（存在しない場合）
c.execute('''CREATE TABLE IF NOT EXISTS articles
             (title TEXT, url TEXT UNIQUE, date TEXT)''')

def scrape_news():
    url = "https://example-news-site.com"
    response = requests.get(url)
    soup = BeautifulSoup(response.text, 'html.parser')

    articles = soup.find_all('article', class_='news-item')

    for article in articles:
        title = article.find('h2').text.strip()
        link = article.find('a')['href']
        date = datetime.now().strftime("%Y-%m-%d %H:%M:%S")

        # データベースに挿入（重複する場合はスキップ）
        try:
            c.execute("INSERT INTO articles (title, url, date) VALUES (?, ?, ?)",
                      (title, link, date))
            print(f"新しい記事を追加しました: {title}")
        except sqlite3.IntegrityError:
            print(f"既存の記事をスキップしました: {title}")

    conn.commit()

# スクレイピングの実行
scrape_news()

# データベース接続のクローズ
conn.close()

新しい記事を追加しました: 新型スマートフォンの発売日が決定
新しい記事を追加しました: AIが作曲した交響曲が話題に
既存の記事をスキップしました: 東京オリンピック開催まであと100日

from selenium import webdriver
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
import sqlite3
import smtplib
from email.mime.text import MIMEText
from datetime import datetime

# データベース接続
conn = sqlite3.connect('price_monitor.db')
c = conn.cursor()

# テーブル作成（存在しない場合）
c.execute('''CREATE TABLE IF NOT EXISTS prices
             (product_id TEXT, price REAL, date TEXT)''')

def scrape_price(url, product_id):
    chrome_options = Options()
    chrome_options.add_argument("--headless")
    driver = webdriver.Chrome(options=chrome_options)

    driver.get(url)

    try:
        price_element = WebDriverWait(driver, 10).until(
            EC.presence_of_element_located((By.CLASS_NAME, "product-price"))
        )
        price = float(price_element.text.replace('￥', '').replace(',', ''))

        # 最新の価格を取得
        c.execute("SELECT price FROM prices WHERE product_id = ? ORDER BY date DESC LIMIT 1", (product_id,))
        last_price = c.fetchone()

        if last_price and abs(price - last_price[0]) / last_price[0] > 0.1:
            send_notification(product_id, last_price[0], price)

        # 新しい価格をデータベースに挿入
        date = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        c.execute("INSERT INTO prices (product_id, price, date) VALUES (?, ?, ?)",
                  (product_id, price, date))
        conn.commit()

        print(f"製品 {product_id} の価格を更新しました: {price}円")

    finally:
        driver.quit()

def send_notification(product_id, old_price, new_price):
    subject = f"価格変動アラート: 製品 {product_id}"
    body = f"製品 {product_id} の価格が大幅に変動しました。\n旧価格: {old_price}円\n新価格: {new_price}円"

    msg = MIMEText(body)
    msg['Subject'] = subject
    msg['From'] = "your_email@example.com"
    msg['To'] = "recipient@example.com"

    smtp_server = smtplib.SMTP('smtp.gmail.com', 587)
    smtp_server.starttls()
    smtp_server.login("your_email@example.com", "your_password")
    smtp_server.send_message(msg)
    smtp_server.quit()

# 価格モニタリングの実行
scrape_price("https://example-ecommerce.com/product/123", "PROD123")

# データベース接続のクローズ
conn.close()

import tweepy
from textblob import TextBlob
import matplotlib.pyplot as plt
import pandas as pd

# Twitter API認証情報
consumer_key = "your_consumer_key"
consumer_secret = "your_consumer_secret"
access_token = "your_access_token"
access_token_secret = "your_access_token_secret"

# Twitter API認証
auth = tweepy.OAuthHandler(consumer_key, consumer_secret)
auth.set_access_token(access_token, access_token_secret)
api = tweepy.API(auth)

def get_tweets(query, count=100):
    tweets = []
    try:
        fetched_tweets = api.search_tweets(q=query, count=count)
        for tweet in fetched_tweets:
            parsed_tweet = {}
            parsed_tweet['text'] = tweet.text
            parsed_tweet['sentiment'] = get_tweet_sentiment(tweet.text)
            if tweet.retweet_count > 0:
                if parsed_tweet not in tweets:
                    tweets.append(parsed_tweet)
            else:
                tweets.append(parsed_tweet)
        return tweets
    except tweepy.TweepError as e:
        print(f"エラーが発生しました : {str(e)}")

def get_tweet_sentiment(tweet):
    analysis = TextBlob(tweet)
    if analysis.sentiment.polarity > 0:
        return 'positive'
    elif analysis.sentiment.polarity == 0:
        return 'neutral'
    else:
        return 'negative'

# メイン関数
def main():
    tweets = get_tweets(query="Python programming", count=200)

    positive_tweets = [tweet for tweet in tweets if tweet['sentiment'] == 'positive']
    negative_tweets = [tweet for tweet in tweets if tweet['sentiment'] == 'negative']
    neutral_tweets = [tweet for tweet in tweets if tweet['sentiment'] == 'neutral']

    print(f"ポジティブなツイート: {len(positive_tweets)}")
    print(f"ネガティブなツイート: {len(negative_tweets)}")
    print(f"中立的なツイート: {len(neutral_tweets)}")

    # 結果の可視化
    sentiments = ['Positive', 'Negative', 'Neutral']
    counts = [len(positive_tweets), len(negative_tweets), len(neutral_tweets)]

    plt.bar(sentiments, counts)
    plt.title('Sentiment Analysis of Tweets about Python Programming')
    plt.xlabel('Sentiment')
    plt.ylabel('Number of Tweets')
    plt.show()

if __name__ == "__main__":
    main()

ポジティブなツイート: 120
ネガティブなツイート: 30
中立的なツイート: 50

import requests
from bs4 import BeautifulSoup
from concurrent.futures import ThreadPoolExecutor
import time

def fetch_url(url):
    response = requests.get(url)
    return BeautifulSoup(response.text, 'html.parser')

def extract_data(soup):
    # ここでデータ抽出のロジックを実装
    title = soup.find('h1').text
    return title

def process_url(url):
    soup = fetch_url(url)
    return extract_data(soup)

urls = [f'https://example.com/page/{i}' for i in range(1, 101)]

start_time = time.time()

# シングルスレッドでの実行
single_thread_results = []
for url in urls:
    single_thread_results.append(process_url(url))

single_thread_time = time.time() - start_time
print(f"シングルスレッド実行時間: {single_thread_time:.2f}秒")

# マルチスレッドでの実行
start_time = time.time()
with ThreadPoolExecutor(max_workers=10) as executor:
    multi_thread_results = list(executor.map(process_url, urls))

multi_thread_time = time.time() - start_time
print(f"マルチスレッド実行時間: {multi_thread_time:.2f}秒")

print(f"速度向上率: {single_thread_time / multi_thread_time:.2f}倍")

シングルスレッド実行時間: 25.34秒
マルチスレッド実行時間: 3.67秒
速度向上率: 6.91倍

import requests
from requests.exceptions import RequestException
import time
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def fetch_url_with_retry(url, max_retries=3, backoff_factor=0.3):
    for attempt in range(max_retries):
        try:
            response = requests.get(url, timeout=10)
            response.raise_for_status()
            return response.text
        except RequestException as e:
            wait_time = backoff_factor * (2 ** attempt)
            logger.warning(f"アクセス失敗 ({url}): {e}. {wait_time}秒後に再試行します。")
            time.sleep(wait_time)

    logger.error(f"最大再試行回数に達しました ({url})")
    return None

# 使用例
url = "https://example.com"
content = fetch_url_with_retry(url)

if content:
    logger.info(f"コンテンツの取得に成功しました: {len(content)} バイト")
else:
    logger.error("コンテンツの取得に失敗しました")

WARNING:__main__:アクセス失敗 (https://example.com): HTTPSConnectionPool(host='example.com', port=443): Read timed out. (read timeout=10). 0.3秒後に再試行します。
WARNING:__main__:アクセス失敗 (https://example.com): HTTPSConnectionPool(host='example.com', port=443): Read timed out. (read timeout=10). 0.6秒後に再試行します。
INFO:__main__:コンテンツの取得に成功しました: 1256 バイト

import csv
from itertools import islice

def process_large_csv(file_path, batch_size=1000):
    with open(file_path, 'r') as f:
        csv_reader = csv.DictReader(f)
        while True:
            batch = list(islice(csv_reader, batch_size))
            if not batch:
                break
            yield batch

def process_batch(batch):
    # ここでバッチ処理のロジックを実装
    processed_data = [row['column_name'].upper() for row in batch]
    return processed_data

# 大規模CSVファイルの処理
file_path = 'large_data.csv'
for i, batch in enumerate(process_large_csv(file_path)):
    processed_batch = process_batch(batch)
    print(f"バッチ {i+1} を処理しました: {len(processed_batch)} 行")

    # ここで処理結果を保存するなどの操作を行う

バッチ 1 を処理しました: 1000 行
バッチ 2 を処理しました: 1000 行
バッチ 3 を処理しました: 1000 行
バッチ 4 を処理しました: 1000 行
バッチ 5 を処理しました: 1000 行
...