# CS知識なし: ネストしたループでユーザー検索
def find_common_users(list_a, list_b):
    """2つのリストの共通ユーザーを返す"""
    common = []
    for user_a in list_a:           # O(n)
        for user_b in list_b:       # × O(m) = O(n×m)
            if user_a['id'] == user_b['id']:
                common.append(user_a)
    return common
 
# list_a = 10,000件, list_b = 10,000件
# → 10,000 × 10,000 = 1億回の比較
# → 数十秒〜数分かかる
 
# CS知識あり: ハッシュセットで O(n+m) に改善
def find_common_users(list_a, list_b):
    """ハッシュセットを使い O(n+m) で解決"""
    ids_b = {user['id'] for user in list_b}  # O(m) で構築
    return [user for user in list_a if user['id'] in ids_b]  # O(n) で検索
    # 合計: O(n + m)
 
# 10,000 + 10,000 = 20,000回の操作
# → 数ミリ秒で完了（5,000倍高速）

// CS知識なし: イベントリスナーの解除忘れ
class UserDashboard {
  constructor() {
    // コンポーネント作成のたびにリスナー追加
    window.addEventListener('resize', this.handleResize);
    // → コンポーネントが破棄されてもリスナーは残る
    // → handleResizeがthisを参照し続ける
    // → UserDashboardオブジェクトがGCされない
    // → メモリリーク
  }
 
  handleResize = () => {
    this.updateLayout(); // thisへの参照を保持
  }
}
 
// CS知識あり: ガベージコレクションの仕組みを理解
class UserDashboard {
  #abortController = new AbortController();
 
  constructor() {
    window.addEventListener('resize', this.handleResize, {
      signal: this.#abortController.signal // 自動解除の仕組み
    });
  }
 
  destroy() {
    this.#abortController.abort(); // 全リスナーを一括解除
  }
 
  handleResize = () => {
    this.updateLayout();
  }
}

// CS知識なし: floatで金額計算
const price = 0.1 + 0.2;
console.log(price);           // 0.30000000000000004
console.log(price === 0.3);   // false
 
// 決済システムで1円のずれが発生:
const total = items.reduce((sum, item) => sum + item.price, 0);
// 1000件の小数点計算で誤差が蓄積 → 数円〜数十円のずれ
 
// CS知識あり: IEEE 754の仕組みを理解
// 方法1: 整数で計算（最小単位=「銭」or「セント」で扱う）
const priceInCents = 10 + 20; // 30（正確）
const displayPrice = priceInCents / 100; // 表示時のみ変換
 
// 方法2: Decimalライブラリ使用
// import Decimal from 'decimal.js';
// const price = new Decimal('0.1').plus('0.2');
// price.equals(0.3) → true

# CS知識なし: エンコーディングを意識しない
text = "Hello 世界"
# データベースにLatin-1で保存 → 絵文字と日本語が文字化け
 
# CS知識あり: UTF-8の仕組みを理解
# UTF-8のバイト構造:
# 1バイト: 0xxxxxxx (ASCII互換: 0-127)
# 2バイト: 110xxxxx 10xxxxxx (ラテン拡張: 128-2047)
# 3バイト: 1110xxxx 10xxxxxx 10xxxxxx (日本語等: 2048-65535)
# 4バイト: 11110xxx 10xxxxxx 10xxxxxx 10xxxxxx (絵文字等: 65536-)
 
# 正しい対処:
# 1. DB: CHARACTER SET utf8mb4（MySQLではutf8は3バイトまで！）
# 2. HTTP: Content-Type: text/html; charset=utf-8
# 3. ファイル: BOMなしUTF-8で統一
# 4. Python: open('file.txt', encoding='utf-8')

# CS知識なし: ロックの順序を意識しない
import threading
 
lock_a = threading.Lock()
lock_b = threading.Lock()
 
def transfer_money(from_acc, to_acc, amount):
    # スレッド1: A→B の送金（lock_a → lock_b の順）
    # スレッド2: B→A の送金（lock_b → lock_a の順）
    # → デッドロック！互いに相手のロック解放を永遠に待つ
    with from_acc.lock:
        with to_acc.lock:
            from_acc.balance -= amount
            to_acc.balance += amount
 
# CS知識あり: ロック順序を統一（デッドロック防止の基本）
def transfer_money(from_acc, to_acc, amount):
    # 常にID順でロックを取得 → 循環待ちを防止
    first, second = sorted([from_acc, to_acc], key=lambda a: a.id)
    with first.lock:
        with second.lock:
            from_acc.balance -= amount
            to_acc.balance += amount

# デッドロックの4条件（Coffman条件、1971年）
# 4つ全てが揃うとデッドロックが発生する
 
# 1. 相互排除: リソースは同時に1つのスレッドしか使えない
#    → 対策: 可能ならリソースを共有可能にする（読み取りロック）
 
# 2. 保持と待機: リソースを保持したまま別のリソースを待つ
#    → 対策: 全リソースを一括取得（All-or-Nothing）
 
# 3. 非プリエンプション: リソースの強制解放ができない
#    → 対策: タイムアウト付きロック
import threading
 
lock = threading.Lock()
acquired = lock.acquire(timeout=5)  # 5秒でタイムアウト
if not acquired:
    # ロック取得失敗 → リトライまたは別の処理
    handle_timeout()
 
# 4. 循環待ち: A→B→C→A のような循環的な待ち関係
#    → 対策: ロック順序の統一（上述）

# CS知識なし: ORMに任せっきり
users = User.objects.all()  # 1クエリ
for user in users:
    print(user.posts.all())  # ユーザーごとに1クエリ → N回
# 合計: 1 + N クエリ（1000ユーザー → 1001クエリ）
 
# CS知識あり: JOINの仕組みとインデックスを理解
users = User.objects.prefetch_related('posts').all()
# 合計: 2クエリ（ユーザー取得 + 全ポスト取得）
# → 500倍高速
 
# さらに深い理解:
# SELECT * FROM users
# JOIN posts ON posts.user_id = users.id
# WHERE users.active = true
# → B+木インデックスがuser_idにあれば O(n log m)
# → なければフルテーブルスキャン O(n × m)

# CS知識なし: 再帰の深さを考慮しない
def factorial(n):
    return n * factorial(n - 1) if n > 0 else 1
# factorial(10000) → RecursionError: maximum recursion depth exceeded
 
# CS知識あり: 末尾再帰最適化、またはループに変換
def factorial(n):
    result = 1
    for i in range(2, n + 1):
        result *= i
    return result
# factorial(10000) → 正常に計算（巨大な数だがメモリは十分）
 
# さらに深い理解: コールスタックの仕組み
# 再帰呼び出しのたびにスタックフレーム（ローカル変数、戻りアドレス）が積まれる
# Pythonのデフォルト上限: 1000フレーム
# → 大きなNには反復法かメモ化が必須

# ハッシュ衝突の実例と対策
 
# ハッシュDoS攻撃の原理
# 攻撃者が意図的にハッシュ値が衝突するキーを大量に送信
# → ハッシュテーブルが O(n) に退化 → サーバーが応答不能に
 
# Python 3.3以降の対策: ハッシュランダム化
# 起動ごとにハッシュのシードが変わる
# → 攻撃者が衝突するキーを事前計算できない
 
import sys
print(sys.hash_info.hash_bits)     # 64
print(sys.hash_info.algorithm)     # siphash24
 
# 安全なハッシュ: SipHash（暗号学的に安全なハッシュ関数）
# Python, Rust, Ruby が採用

# CS知識なし: MD5やSHA-256でパスワードをハッシュ
import hashlib
hashed = hashlib.sha256(password.encode()).hexdigest()
# レインボーテーブルで一瞬で解読される
 
# CS知識あり: bcryptを使用（意図的に遅いハッシュ関数）
import bcrypt
hashed = bcrypt.hashpw(password.encode(), bcrypt.gensalt(rounds=12))
# ソルト付き + ストレッチング → レインボーテーブル無効化
# rounds=12 → 1回のハッシュに約250ms → ブルートフォース困難

# CAP定理の実務的な適用例
 
# CP（一貫性 + 分断耐性）: 銀行の送金処理
# → 多少の遅延は許容するが、残高の不整合は絶対にNG
# 例: PostgreSQL (単一ノード), ZooKeeper, etcd
 
# AP（可用性 + 分断耐性）: SNSのタイムライン
# → 数秒の遅延は許容するが、サービス停止はNG
# 例: Cassandra, DynamoDB, CouchDB
 
# 実装例: Sagaパターンによる分散トランザクション
class OrderSaga:
    """注文処理のSagaパターン"""
 
    async def execute(self, order):
        try:
            # Step 1: 在庫確保（Inventory Service）
            reservation = await self.inventory.reserve(order.items)
 
            # Step 2: 決済処理（Payment Service）
            payment = await self.payment.charge(order.total)
 
            # Step 3: 配送手配（Shipping Service）
            shipment = await self.shipping.arrange(order)
 
        except PaymentError:
            # 補償トランザクション: 在庫確保を取り消し
            await self.inventory.release(reservation)
            raise
 
        except ShippingError:
            # 補償トランザクション: 決済取消 + 在庫解放
            await self.payment.refund(payment)
            await self.inventory.release(reservation)
            raise

# async/await は「おまじない」ではない
# イベントループとコルーチンの仕組みを理解すると、
# 正しい使い方が見えてくる
 
import asyncio
 
# イベントループの概念図:
#
# ┌───────────────────────────────────┐
# │          イベントループ            │
# │                                   │
# │  ┌─────┐  ┌─────┐  ┌─────┐      │
# │  │Task1│  │Task2│  │Task3│      │
# │  │await│  │ready│  │await│      │
# │  │ I/O │  │     │  │sleep│      │
# │  └─────┘  └─────┘  └─────┘      │
# │                                   │
# │  1. readyなタスクを実行           │
# │  2. awaitでブロックしたら別タスクへ│
# │  3. I/O完了通知で元タスクに戻る   │
# │  → シングルスレッドで並行処理！   │
# └───────────────────────────────────┘
 
async def fetch_data(url: str) -> dict:
    """非同期HTTP リクエスト"""
    # awaitでI/O待ちの間、他のタスクが実行される
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.json()
 
async def main():
    # 3つのリクエストを並行実行（スレッドなし）
    results = await asyncio.gather(
        fetch_data('https://api.example.com/users'),
        fetch_data('https://api.example.com/posts'),
        fetch_data('https://api.example.com/comments'),
    )
    # 3つのI/O待ちが重なるので、
    # 直列(3秒)ではなく並行(1秒程度)で完了

# Before: O(n²) — ネストループ
def remove_duplicates(items):
    unique = []
    for item in items:
        if item not in unique:  # 'not in' は O(n) の線形探索
            unique.append(item)
    return unique
# 10,000件 → 約50,000,000回の比較
 
# After: O(n) — セットの活用
def remove_duplicates(items):
    seen = set()
    unique = []
    for item in items:
        if item not in seen:  # 'not in' は O(1) のハッシュ探索
            seen.add(item)
            unique.append(item)
    return unique
# 10,000件 → 約10,000回のハッシュ計算
 
# さらに簡潔に（順序保持、Python 3.7+）:
def remove_duplicates(items):
    return list(dict.fromkeys(items))

# Before: O(n²) — 文字列の連結は毎回新オブジェクト生成
def build_report(records):
    result = ""
    for record in records:
        result += f"{record['name']}: {record['value']}\n"  # O(n) × n回 = O(n²)
    return result
 
# After: O(n) — joinを使用
def build_report(records):
    lines = [f"{record['name']}: {record['value']}" for record in records]
    return "\n".join(lines)  # 一度にメモリ確保 → O(n)

// Before: O(n × m) — ネストループ
function getUserOrders(users, orders) {
  return users.map(user => ({
    ...user,
    orders: orders.filter(order => order.userId === user.id)
    // filter は全orders を走査 → O(m) × n回 = O(n×m)
  }));
}
 
// After: O(n + m) — グルーピング
function getUserOrders(users, orders) {
  // 前処理: ordersをuserIdでグルーピング O(m)
  const ordersByUser = new Map();
  for (const order of orders) {
    if (!ordersByUser.has(order.userId)) {
      ordersByUser.set(order.userId, []);
    }
    ordersByUser.get(order.userId).push(order);
  }
 
  // 結合: O(n)
  return users.map(user => ({
    ...user,
    orders: ordersByUser.get(user.id) || []
  }));
}

# Before: 同じ計算を何度も実行
def get_user_stats(user_id):
    user = db.query(f"SELECT * FROM users WHERE id = {user_id}")  # 毎回DB
    posts = db.query(f"SELECT * FROM posts WHERE user_id = {user_id}")  # 毎回DB
    return {"user": user, "post_count": len(posts)}
 
# After: LRUキャッシュで頻繁なクエリを高速化
from functools import lru_cache
 
@lru_cache(maxsize=1000)
def get_user_stats(user_id):
    user = db.query(f"SELECT * FROM users WHERE id = %s", (user_id,))
    posts = db.query(f"SELECT * FROM posts WHERE user_id = %s", (user_id,))
    return {"user": user, "post_count": len(posts)}
 
# CS知識: LRUキャッシュの内部実装
# - ハッシュマップ（O(1)検索）+ 双方向リンクリスト（O(1)更新）
# - 最も古い（Least Recently Used）エントリを自動削除

# Before: 「ソートは sort() を呼ぶだけ」（正しい、だが理解が浅い）
items.sort()  # Python の TimSort: O(n log n) — 常に最適
 
# CS知識があると分かること:
# 1. sort() は安定ソート（同じキーの相対順序を保持）
# 2. ほぼソート済みなら O(n) に近い（TimSortの特性）
# 3. key= でカスタムキーを指定すると比較コストを制御できる
 
# 実務応用: 複合ソートの最適化
users.sort(key=lambda u: (u['department'], -u['salary']))
# CS知識: Pythonのsortはタプルの辞書順比較を利用
# → 部門名の昇順、同じ部門内は給与の降順
 
# さらに: 100万件を超える場合
# - メモリに乗る → TimSort O(n log n)
# - メモリに乗らない → 外部ソート（マージソート）
# - 特定の範囲の整数 → 計数ソート O(n + k) [非比較ソート]

import re
 
# Before: バックトラッキングが爆発する正規表現
# ReDoS（Regular Expression Denial of Service）
pattern_bad = r'^(a+)+$'
text = 'a' * 30 + 'b'
# re.match(pattern_bad, text)  # 数十秒〜フリーズ！
 
# 原因: (a+)+ のネストにより、指数的なバックトラッキングが発生
# 'aaaa...b' に対して:
# (a)(a)(a)...b → 失敗
# (aa)(a)(a)...b → 失敗
# (a)(aa)(a)...b → 失敗
# → 2^n 通りの分割を試行 → 指数時間
 
# After: CS知識でバックトラッキングを回避
pattern_good = r'^a+$'  # ネストを排除
# または原子グループ/所有量指定子を使用
 
# CS知識: NFAとDFAの違い
# Python の re はNFA（バックトラッキングあり）
# → 悪い正規表現で指数時間になりうる
# Google RE2 はDFA（バックトラッキングなし）
# → 常に線形時間で動作
 
# 対策:
# 1. ネストした量指定子を避ける: (a+)+ → a+
# 2. 入力長を制限する
# 3. タイムアウトを設定する
# 4. RE2などDFAベースのエンジンを使う

# 問題: 配列の中で合計がtargetになるペアを見つける
def two_sum(nums, target):
    for i in range(len(nums)):
        for j in range(i + 1, len(nums)):
            if nums[i] + nums[j] == target:
                return [i, j]
    return []

# デバッグ用ユーティリティ
import logging
import traceback
from functools import wraps
 
# ロガーの設定
logging.basicConfig(
    level=logging.DEBUG,
    format='%(asctime)s [%(levelname)s] %(name)s: %(message)s'
)
logger = logging.getLogger(__name__)
 
def debug_decorator(func):
    """関数の入出力をログ出力するデコレータ"""
    @wraps(func)
    def wrapper(*args, **kwargs):
        logger.debug(f"呼び出し: {func.__name__}(args={args}, kwargs={kwargs})")
        try:
            result = func(*args, **kwargs)
            logger.debug(f"戻り値: {func.__name__} -> {result}")
            return result
        except Exception as e:
            logger.error(f"例外発生: {func.__name__}: {e}")
            logger.error(traceback.format_exc())
            raise
    return wrapper
 
@debug_decorator
def process_data(items):
    """データ処理（デバッグ対象）"""
    if not items:
        raise ValueError("空のデータ")
    return [item * 2 for item in items]

# 設計判断の記録テンプレート
class ArchitectureDecisionRecord:
    """ADR (Architecture Decision Record) の作成"""
 
    def __init__(self, title: str):
        self.title = title
        self.context = ""
        self.decision = ""
        self.consequences = []
        self.alternatives = []
 
    def set_context(self, context: str):
        """背景と課題の記述"""
        self.context = context
        return self
 
    def set_decision(self, decision: str):
        """決定内容の記述"""
        self.decision = decision
        return self
 
    def add_consequence(self, consequence: str, positive: bool = True):
        """結果の追加"""
        self.consequences.append({
            'description': consequence,
            'type': 'positive' if positive else 'negative'
        })
        return self
 
    def add_alternative(self, name: str, reason_rejected: str):
        """却下した代替案の追加"""
        self.alternatives.append({
            'name': name,
            'reason_rejected': reason_rejected
        })
        return self
 
    def to_markdown(self) -> str:
        """Markdown形式で出力"""
        md = f"# ADR: {self.title}\n\n"
        md += f"## 背景\n{self.context}\n\n"
        md += f"## 決定\n{self.decision}\n\n"
        md += "## 結果\n"
        for c in self.consequences:
            icon = "✅" if c['type'] == 'positive' else "⚠️"
            md += f"- {icon} {c['description']}\n"
        md += "\n## 却下した代替案\n"
        for a in self.alternatives:
            md += f"- **{a['name']}**: {a['reason_rejected']}\n"
        return md

# チーム間のAPI契約定義
from dataclasses import dataclass
from typing import List, Optional
from enum import Enum
 
class Priority(Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    CRITICAL = "critical"
 
@dataclass
class APIContract:
    """チーム間のAPI契約"""
    endpoint: str
    method: str
    owner_team: str
    consumers: List[str]
    sla_ms: int  # レスポンスタイムSLA
    priority: Priority
 
    def validate_sla(self, actual_ms: int) -> bool:
        """SLA準拠の確認"""
        return actual_ms <= self.sla_ms
 
    def to_openapi(self) -> dict:
        """OpenAPI形式で出力"""
        return {
            'path': self.endpoint,
            'method': self.method,
            'x-owner': self.owner_team,
            'x-consumers': self.consumers,
            'x-sla-ms': self.sla_ms
        }
 
# 使用例
contracts = [
    APIContract(
        endpoint="/api/v1/users",
        method="GET",
        owner_team="user-team",
        consumers=["order-team", "notification-team"],
        sla_ms=200,
        priority=Priority.HIGH
    ),
    APIContract(
        endpoint="/api/v1/orders",
        method="POST",
        owner_team="order-team",
        consumers=["payment-team", "inventory-team"],
        sla_ms=500,
        priority=Priority.CRITICAL
    )
]

# セキュアコーディング例
import hashlib
import secrets
import hmac
from typing import Optional
 
class SecurityUtils:
    """セキュリティユーティリティ"""
 
    @staticmethod
    def generate_token(length: int = 32) -> str:
        """暗号学的に安全なトークン生成"""
        return secrets.token_urlsafe(length)
 
    @staticmethod
    def hash_password(password: str, salt: Optional[str] = None) -> tuple:
        """パスワードのハッシュ化"""
        if salt is None:
            salt = secrets.token_hex(16)
        hashed = hashlib.pbkdf2_hmac(
            'sha256',
            password.encode('utf-8'),
            salt.encode('utf-8'),
            iterations=100000
        )
        return hashed.hex(), salt
 
    @staticmethod
    def verify_password(password: str, hashed: str, salt: str) -> bool:
        """パスワードの検証"""
        new_hash, _ = SecurityUtils.hash_password(password, salt)
        return hmac.compare_digest(new_hash, hashed)
 
    @staticmethod
    def sanitize_input(value: str) -> str:
        """入力値のサニタイズ"""
        dangerous_chars = ['<', '>', '"', "'", '&', '\\']
        result = value
        for char in dangerous_chars:
            result = result.replace(char, '')
        return result.strip()
 
# 使用例
token = SecurityUtils.generate_token()
hashed, salt = SecurityUtils.hash_password("my_password")
is_valid = SecurityUtils.verify_password("my_password", hashed, salt)

# マイグレーションスクリプトのテンプレート
import json
import logging
from pathlib import Path
from datetime import datetime
from typing import List, Dict, Callable
 
logger = logging.getLogger(__name__)
 
class MigrationRunner:
    """段階的マイグレーション実行エンジン"""
 
    def __init__(self, migration_dir: str):
        self.migration_dir = Path(migration_dir)
        self.migrations: List[Dict] = []
        self.completed: List[str] = []
 
    def register(self, version: str, description: str,
                 up: Callable, down: Callable):
        """マイグレーションの登録"""
        self.migrations.append({
            'version': version,
            'description': description,
            'up': up,
            'down': down,
            'registered_at': datetime.now().isoformat()
        })
 
    def run_up(self, target_version: str = None):
        """マイグレーションの実行（アップグレード）"""
        for migration in self.migrations:
            if migration['version'] in self.completed:
                continue
            logger.info(f"実行中: {migration['version']} - "
                       f"{migration['description']}")
            try:
                migration['up']()
                self.completed.append(migration['version'])
                logger.info(f"完了: {migration['version']}")
            except Exception as e:
                logger.error(f"失敗: {migration['version']}: {e}")
                raise
            if target_version and migration['version'] == target_version:
                break
 
    def run_down(self, target_version: str):
        """マイグレーションのロールバック"""
        for migration in reversed(self.migrations):
            if migration['version'] not in self.completed:
                continue
            if migration['version'] == target_version:
                break
            logger.info(f"ロールバック: {migration['version']}")
            migration['down']()
            self.completed.remove(migration['version'])
 
    def status(self) -> Dict:
        """マイグレーション状態の確認"""
        return {
            'total': len(self.migrations),
            'completed': len(self.completed),
            'pending': len(self.migrations) - len(self.completed),
            'versions': {
                m['version']: 'completed'
                if m['version'] in self.completed else 'pending'
                for m in self.migrations
            }
        }

# テストの ROI（投資対効果）を示す例
class TestROICalculator:
    """テスト投資対効果の計算"""
 
    def __init__(self):
        self.test_writing_hours = 0
        self.bugs_prevented = 0
        self.debug_hours_saved = 0
 
    def add_test_investment(self, hours: float):
        """テスト作成にかかった時間"""
        self.test_writing_hours += hours
 
    def add_bug_prevention(self, count: int, avg_debug_hours: float = 2.0):
        """テストにより防いだバグ"""
        self.bugs_prevented += count
        self.debug_hours_saved += count * avg_debug_hours
 
    def calculate_roi(self) -> dict:
        """ROIの計算"""
        net_benefit = self.debug_hours_saved - self.test_writing_hours
        roi_percent = (net_benefit / self.test_writing_hours * 100
                      if self.test_writing_hours > 0 else 0)
        return {
            'test_hours': self.test_writing_hours,
            'bugs_prevented': self.bugs_prevented,
            'hours_saved': self.debug_hours_saved,
            'net_benefit_hours': net_benefit,
            'roi_percent': f'{roi_percent:.1f}%'
        }