1. ファイルシステムの自動整理

拡張子ベースのファイル分類

指定ディレクトリ内のファイルを拡張子ごとにサブフォルダへ振り分ける処理です。標準ライブラリのpathlibを活用し、ファイルパス操作を安全かつ簡潔に記述しています。

import shutil
from pathlib import Path

def organize_by_extension(target_dir: str) -> None:
    base_path = Path(target_dir)
    if not base_path.exists():
        raise FileNotFoundError(f"パスが見つかりません: {target_dir}")
    
    for file_obj in base_path.iterdir():
        if file_obj.is_file():
            ext_folder = base_path / file_obj.suffix.strip('.')
            ext_folder.mkdir(exist_ok=True)
            shutil.move(str(file_obj), str(ext_folder / file_obj.name))
            print(f"移動完了: {file_obj.name} -> {ext_folder.name}")

空ディレクトリの削除

再帰的にディレクトリ構造を走査し、中身が空のフォルダを自動的に削除します。下から上へ走査(topdown=False)することで、親フォルダの削除判定を正確に行えます。

import os

def purge_empty_dirs(root_path: str) -> None:
    for dirpath, dirnames, filenames in os.walk(root_path, topdown=False):
        if not dirnames and not filenames:
            try:
                os.rmdir(dirpath)
                print(f"削除: {dirpath}")
            except OSError:
                print(f"スキップ: {dirpath} (権限またはシステムフォルダ)")

一括ファイル名変更

特定パターンを含むファイル名を置換するユーティリティです。正規表現を利用すればより高度なパターンマッチングも可能ですが、ここでは簡易な文字列置換を実装しています。

import re
from pathlib import Path

def batch_rename(target_dir: str, pattern: str, replacement: str) -> None:
    dir_path = Path(target_dir)
    regex = re.compile(re.escape(pattern), re.IGNORECASE)
    
    for item in dir_path.iterdir():
        if item.is_file() and pattern.lower() in item.stem.lower():
            new_stem = regex.sub(replacement, item.stem)
            new_name = f"{new_stem}{item.suffix}"
            item.rename(dir_path / new_name)
            print(f"変更: {item.name} -> {new_name}")

2. ウェブデータの取得と解析

HTMLからの要素抽出

requestsとBeautifulSoupを組み合わせ、WebページのDOM構造を解析します。ここでは例としてh2タグとaタグのhref属性を取得するロジックを示します。

import requests
from bs4 import BeautifulSoup

def extract_page_elements(url: str) -> dict:
    headers = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)"}
    resp = requests.get(url, headers=headers, timeout=10)
    resp.raise_for_status()
    
    soup = BeautifulSoup(resp.content, "html.parser")
    titles = [tag.text.strip() for tag in soup.find_all("h2")]
    links = [a["href"] for a in soup.find_all("a", href=True)]
    return {"titles": titles, "links": links}

画像の一括ダウンロード

JSON形式で提供される画像URLリストを順次取得し、ローカルストレージへ保存します。リクエストごとのステータス検証とバイナリ書き込みを行います。

import requests
from pathlib import Path

def fetch_images(api_endpoint: str, save_dir: str) -> None:
    save_path = Path(save_dir)
    save_path.mkdir(parents=True, exist_ok=True)
    
    data = requests.get(api_endpoint).json()
    for idx, img_url in enumerate(data.get("images", [])):
        img_data = requests.get(img_url)
        if img_data.ok:
            ext = Path(img_url).suffix or ".jpg"
            (save_path / f"asset_{idx}{ext}").write_bytes(img_data.content)

フォーム送信の自動化

Webアプリケーションのログインや問い合わせフォームをプログラムから送信します。POSTリクエストに辞書形式のペイロードを付与して実行します。

import requests

def submit_web_form(endpoint: str, payload: dict) -> requests.Response:
    headers = {"Content-Type": "application/x-www-form-urlencoded"}
    resp = requests.post(endpoint, data=payload, headers=headers, allow_redirects=True)
    resp.raise_for_status()
    return resp

3. テキスト処理エンジン

単語数の統計算出

対象のテキストファイルを読み込み、空白文字や改行を区切り文字として単語数をカウントします。記号や特殊文字を除外する前処理を含む実装が実務では一般的です。

import re
from pathlib import Path

def compute_word_count(file_path: str) -> int:
    content = Path(file_path).read_text(encoding="utf-8")
    # 英数字とハイフン、アポストロフィを含む連続文字を単語とみなす
    tokens = re.findall(r"\b[\w'-]+\b", content)
    return len(tokens)

文字列の一括置換

ファイル内の特定のキーワードを別文字列へ置き換えます。メモリ消費を抑えるため、行単位でのストリーム読み書きを行う方式を採用しています。

from pathlib import Path

def replace_in_file(src: str, dst: str, old_term: str, new_term: str) -> None:
    path = Path(src)
    original_content = path.read_text()
    modified = original_content.replace(old_term, new_term)
    Path(dst).write_text(modified, encoding="utf-8")

テスト用ダミーテキスト生成

指定長さに応じてランダムな英数字を生成します。セキュリティトークンやバリデーションテストの入力データとして利用可能です。

import secrets
import string

def generate_dummy_string(length: int = 32) -> str:
    charset = string.ascii_letters + string.digits
    return "".join(secrets.choice(charset) for _ in range(length))

4. メール配信システム

個別宛先への一斉送信

SMTPプロトコルを直接使用し、宛先リストをループ処理します。MIME構造を構築し、HTML形式またはプレーンテキストの配信に対応します。

import smtplib
from email.mime.text import MIMEText

def dispatch_emails(host: str, port: int, user: str, pwd: str, targets: list, subject: str, body: str) -> None:
    msg = MIMEText(body, "plain", "utf-8")
    msg["Subject"] = subject
    msg["From"] = user
    
    with smtplib.SMTP(host, port) as server:
        server.starttls()
        server.login(user, pwd)
        for addr in targets:
            msg["To"] = addr
            server.sendmail(user, addr, msg.as_string())
            del msg["To"]

添付ファイル付き配信

base64エンコーディングを用いてファイルをMIMEパートに付加します。ファイル形式に応じたMIMEタイプを自動判別することも可能です。

from email.mime.multipart import MIMEMultipart
from email.mime.base import MIMEBase
from email import encoders
import smtplib

def send_with_attachment(host, port, user, pwd, recipient, subject, body, attach_path):
    msg = MIMEMultipart()
    msg["Subject"] = subject
    msg["From"] = user
    msg["To"] = recipient
    msg.attach(MIMEText(body, "plain", "utf-8"))
    
    with open(attach_path, "rb") as f:
        part = MIMEBase("application", "octet-stream")
        part.set_payload(f.read())
        encoders.encode_base64(part)
        part.add_header("Content-Disposition", "attachment", filename=attach_path)
        msg.attach(part)
        
    with smtplib.SMTP(host, port) as srv:
        srv.starttls()
        srv.login(user, pwd)
        srv.send_message(msg)

5. Excelデータの操作と解析

スプレッドシートの入出力

pandasを用いてExcelファイルをDataFrameとして読み込み、加工後に新規ファイルへ書き出します。インデックスの出力を抑制する設定を含みます。

import pandas as pd

def export_excel(data_dict: dict, output_path: str) -> None:
    df = pd.DataFrame(data_dict)
    df.to_excel(output_path, index=False, sheet_name="ProcessedData")
    print(f"保存完了: {output_path}")

複数シートの統合

1つのブック内にある複数のシートを縦方向に結合します。pandasのconcat関数を使用することで、メモリ効率と実行速度を最適化しています。

import pandas as pd

def consolidate_worksheets(xlsx_path: str, output_path: str) -> pd.DataFrame:
    xls = pd.ExcelFile(xlsx_path)
    frames = [pd.read_excel(xls, sheet) for sheet in xls.sheet_names]
    merged_df = pd.concat(frames, ignore_index=True)
    merged_df.to_excel(output_path, index=False)
    return merged_df

6. リレーショナルデータベース連携

クエリ実行と結果取得

SQLiteへの接続確立とSQL文の発行をカプセル化します。パラメータ化クエリを利用することで、SQLインジェクション対策を施しています。

import sqlite3

def query_db(db_path: str, sql: str, params: tuple = ()) -> list:
    conn = sqlite3.connect(db_path)
    try:
        cursor = conn.cursor()
        cursor.execute(sql, params)
        return cursor.fetchall()
    finally:
        conn.close()

データベースのバックアップ

sqlite3のオンラインバックアップ機能を利用し、使用中のデータベースファイルでも安全にコピーを作成します。

import sqlite3

def backup_database(source: str, target: str) -> None:
    with sqlite3.connect(source) as src_conn:
        with sqlite3.connect(target) as dst_conn:
            src_conn.backup(dst_conn)
    print("バックアップ完了")

7. ソーシャルメディア連携

API経由の投稿自動実行

OAuth認証トークンを用いて外部プラットフォームへ投稿します。ここでは抽象化したインターフェースを示しており、実際のSDK実装に合わせて認証情報を設定します。

def post_to_platform(client_lib, message: str, media_path: str = None) -> dict:
    params = {"status": message}
    if media_path:
        with open(media_path, "rb") as media:
            resp = client_lib.media_upload(media=media, **params)
            return client_lib.post_update(media_ids=resp.media_id_string)
    return client_lib.post_update(**params)

8. オペレーティングシステム制御

プロセス監視と強制終了

psutilモジュールを介してシステムプロセスをフィルタリングし、特定の実行ファイル名を持つプロセスを安全に終了させます。

import psutil

def terminate_process_by_name(target_name: str) -> list:
    terminated = []
    for proc in psutil.process_iter(["pid", "name"]):
        if proc.info.get("name") and target_name.lower() in proc.info["name"].lower():
            try:
                proc.terminate()
                proc.wait(timeout=3)
                terminated.append(proc.info["pid"])
            except (psutil.NoSuchProcess, psutil.AccessDenied):
                continue
    return terminated

ストレージ容量監視

ディスク使用率を監視し、閾値を下回った場合に警告を出力します。CI/CDパイプラインやバッチジョブの前検証として有効です。

import psutil

def check_storage_threshold(mount_point: str, min_gb: float) -> bool:
    usage = psutil.disk_usage(mount_point)
    available_gb = usage.free / (1024 ** 3)
    if available_gb < min_gb:
        raise SystemExit(f"警告: {mount_point} の空き容量が不足 ({available_gb:.2f}GB)")
    return True

9. 画像処理パイプライン

リサイズとトリミング

Pillowライブラリを用いて画像の寸法を変更し、特定領域を切り抜きます。アスペクト比を維持するオプションも提供可能です。

from PIL import Image

def adjust_dimensions(src: str, dst: str, target_w: int, target_h: int) -> None:
    with Image.open(src) as img:
        img = img.convert("RGB")
        resized = img.resize((target_w, target_h), Image.Resampling.LANCZOS)
        resized.save(dst, optimize=True)
        
def crop_region(src: str, dst: str, box: tuple) -> None:
    with Image.open(src) as img:
        img.crop(box).save(dst)

透かしの埋め込み

既存画像の上に半透明のテキストやロゴを重ねます。Alphaチャネルを操作して不透明度を制御しています。

from PIL import Image, ImageDraw, ImageFont

def apply_watermark(src: str, dst: str, text: str, font_path: str, size: int = 36) -> None:
    base = Image.open(src).convert("RGBA")
    overlay = Image.new("RGBA", base.size, (255, 255, 255, 0))
    draw = ImageDraw.Draw(overlay)
    font = ImageFont.truetype(font_path, size)
    w, h = draw.textsize(text, font=font)
    pos = (base.width - w - 20, base.height - h - 20)
    draw.text(pos, text, font=font, fill=(255, 255, 255, 180))
    Image.alpha_composite(base, overlay).convert("RGB").save(dst)

10. ネットワークインフラ自動化

エンドポイント死活監視

HTTPステータスコードとレスポンスタイムを測定し、サービス可用性を評価します。

import requests

def probe_endpoint(url: str) -> dict:
    try:
        resp = requests.get(url, timeout=5)
        return {"url": url, "status": resp.status_code, "ok": resp.ok, "latency": resp.elapsed.total_seconds()}
    except requests.RequestException as e:
        return {"url": url, "status": None, "ok": False, "error": str(e)}

FTPファイル転送

コンテキストマネージャを用いてFTPセッションを安全に管理し、バイナリモードでアップロードを実行します。

from ftplib import FTP

def upload_via_ftp(host: str, user: str, pwd: str, local_file: str, remote_dir: str) -> None:
    with FTP(host) as client:
        client.login(user=user, passwd=pwd)
        client.cwd(remote_dir)
        with open(local_file, "rb") as f:
            client.storbinary(f"STOR {remote_dir}/{f.name}", f)

11. データクレンジング

重複レコードの除外

pandasのdrop_duplicatesメソッドを活用し、特定列または全行の重複を削除してデータ品質を確保します。

import pandas as pd

def deduplicate_dataset(df: pd.DataFrame, subset_cols: list = None) -> pd.DataFrame:
    return df.drop_duplicates(subset=subset_cols, keep="first")

欠損値の補完

時系列データや連続データに対して前方補間または線形補間を適用し、分析前のデータ整形を自動化します。

def impute_missing_values(df: pd.DataFrame, strategy: str = "ffill") -> pd.DataFrame:
    if strategy == "ffill":
        return df.ffill()
    elif strategy == "interpolate":
        return df.interpolate()
    return df.fillna(0)

12. PDFドキュメント処理

テキスト抽出と結合

PyPDF2を使用して複数ページのテキストを抽出し、あるいは複数のPDFファイルを単一のドキュメントに統合します。

from PyPDF2 import PdfReader, PdfWriter

def extract_pdf_text(path: str) -> str:
    reader = PdfReader(path)
    return "\n".join(page.extract_text() for page in reader.pages)

def concatenate_pdfs(input_list: list, output_path: str) -> None:
    merger = PdfWriter()
    for file in input_list:
        merger.append(file)
    with open(output_path, "wb") as out:
        merger.write(out)

13. GUI操作とイベント制御

マウス・キーボードのシミュレーション

pyautoguiを用いて座標ベースのクリック操作やテキスト入力をプログラム化します。GUIテストやレガシーアプリケーションの自動化に利用されます。

import pyautogui
import time

def execute_gui_sequence(coords: list, delay: float = 1.0) -> None:
    pyautogui.FAILSAFE = True
    for x, y in coords:
        pyautogui.moveTo(x, y, duration=0.5)
        pyautogui.click()
        time.sleep(delay)

14. テスト自動化フレームワーク

ユニットテストの実行

unittestモジュールを継承し、境界値や例外ケースを含むテストスイートを定義します。

import unittest

def calculate_discount(price: float, rate: float) -> float:
    if not (0 <= rate <= 1):
        raise ValueError("Rate must be between 0 and 1")
    return price * (1 - rate)

class TestDiscountLogic(unittest.TestCase):
    def test_valid_discount(self):
        self.assertAlmostEqual(calculate_discount(100, 0.2), 80.0)
        
    def test_invalid_rate(self):
        with self.assertRaises(ValueError):
            calculate_discount(100, 1.5)

if __name__ == "__main__":
    unittest.main()

Seleniumによるブラウザ検証

ヘッドレスモードでブラウザを起動し、DOM要素の存在確認やフォーム操作を自動実行します。

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

def validate_login_page(url: str) -> bool:
    options = webdriver.ChromeOptions()
    options.add_argument("--headless")
    with webdriver.Chrome(options=options) as driver:
        driver.get(url)
        try:
            WebDriverWait(driver, 10).until(EC.presence_of_element_located((By.ID, "username")))
            return True
        except Exception:
            return False

15. クラウドリソース管理

AWS EC2インスタンスの自動起動

Boto3 SDKを使用してインフラストラクチャのプロビジョニングを実行します。タグ付けとセキュリティグループの適用を自動化します。

import boto3

def provision_ec2_instance(ami_id: str, instance_type: str, key_pair: str) -> str:
    ec2 = boto3.client("ec2", region_name="us-east-1")
    response = ec2.run_instances(
        ImageId=ami_id,
        MinCount=1,
        MaxCount=1,
        InstanceType=instance_type,
        KeyName=key_pair,
        TagSpecifications=[{"ResourceType": "instance", "Tags": [{"Key": "Name", "Value": "AutoManaged"}]}]
    )
    return response["Instances"][0]["InstanceId"]

16. 財務データ解析

為替レートの取得

公開APIからJSON形式の為替データを取得し、指定通貨ペアの換算レートを計算します。

import requests

def fetch_exchange_rate(base: str, target: str) -> float:
    api_url = f"https://api.exchangerate-api.com/v4/latest/{base}"
    data = requests.get(api_url).json()
    return data["rates"].get(target, 0.0)

収支レポートの生成

CSVファイルから取引履歴を読み込み、月次ごとの収入・支出・純利益を集計します。

import pandas as pd

def generate_budget_report(csv_path: str) -> pd.DataFrame:
    df = pd.read_csv(csv_path, parse_dates=["date"])
    df["month"] = df["date"].dt.to_period("M")
    return df.groupby(["month", "category"])["amount"].sum().unstack().fillna(0)

17. 自然言語処理(NLP)

感情分析の実装

transformersライブラリのPipeline機能を利用し、テキストのポジティブ/ネガティブスコアを高速に取得します。

from transformers import pipeline

classifier = pipeline("sentiment-analysis", model="distilbert-base-uncased-finetuned-sst-2-english")

def analyze_sentiment(text: str) -> str:
    result = classifier(text)[0]
    return f"{result['label']} (confidence: {result['score']:.4f})"

多言語翻訳パイプライン

ローカルモデルまたは外部APIを呼び出し、入力テキストを指定言語へ変換します。バッチ処理に対応した実装パターンを示します。

import requests

def translate_batch(texts: list, target_lang: str, api_key: str) -> list:
    endpoint = "https://api.mymemory.translated.net/get"
    translations = []
    for txt in texts:
        resp = requests.get(endpoint, params={"q": txt, "langpair": f"en|{target_lang}", "key": api_key})
        translations.append(resp.json()["responseData"]["translatedText"])
    return translations