ETLの柔軟性向上

[yu23ki14]

Claudeによる仕様
#154 (comment)

現状の整理

ETLで取れるもの

• Community Noteそのもののデータをそのまま一時保存（SQLiteをつかっているが、Postgresqlにいれる方針）
• Community Noteに紐づくPostのデータをX APIをつかって保存（Postgresqlに保存）
• Community Noteのデータを素材に、LLMで言語の推論
• Community Noteのデータを素材に、LLMでトピックの推論
• BirdXplorer API が必要とするデータモデルに変換する

ETLの手順

現在実装されているETLの手順は以下の通りです：

1. Community Notesデータの取得

• Twitter公式のBirdwatch公開データから日次でCommunity Notesデータを取得
• データソース: https://ton.twimg.com/birdwatch-public-data/{日付}/notes/notes-00000.tsv
• 設定可能な日数分遡ってデータを取得（COMMUNITY_NOTE_DAYS_AGOで制御）
• 重複チェックを行い、新しいノートのみをSQLiteに保存
• 同時にNote Status Historyデータも取得・保存

2. 関連Postデータの取得

• Community Noteに紐づくTwitter投稿をX APIで取得
• 対象期間は設定で制御（TARGET_TWITTER_POST_START_UNIX_MILLISECOND）
• 投稿データ、ユーザーデータ、メディアデータ、埋め込みURLデータをPostgreSQLに保存
• 重複チェックにより既存データの再取得を回避

3. データ変換とCSV生成

• Note変換: SQLiteからNoteデータを取得し、AIサービスで言語検出を実行してCSV出力
• Post変換: PostgreSQLからPostデータを取得してCSV形式に変換
• User変換: PostgreSQLからUserデータを取得してCSV形式に変換
• Media変換: メディアデータとPost-Media関連付けデータをCSV出力
• Link変換: 埋め込みURLデータとPost-Link関連付けデータをCSV出力

4. AIによる高度な変換

• 言語検出: ai_service.detect_language()でNoteの言語を推論
• トピック推論: ai_service.detect_topic()でNoteのトピックを分類
• トピックマスターデータはtopic_seed.csvから生成

出力CSVファイル:

• note.csv - ノートデータ（言語情報付き）
• post.csv - 投稿データ
• user.csv - ユーザーデータ
• media.csv - メディアデータ
• post_media_association.csv - 投稿-メディア関連付け
• post_link.csv - リンクデータ
• post_link_association.csv - 投稿-リンク関連付け
• topic.csv - トピックマスターデータ
• note_topic_association.csv - ノート-トピック関連付け

4. Load（読み込み）フェーズ (load.py)

S3への出力:

• 変換されたCSVファイルをAWS S3バケットにアップロード
• タイムスタンプ付きのプレフィックスで整理
• アップロード対象は全9種類のCSVファイル

新しくいれるべきもの

Post取得の効率化

XのPostデータを取得することはコストがかかるので必要なものを優先的に取得するなどが望まれる。例えば、CommunityNoteのデータをもとにAIや条件で先に前処理をしてメタデータをつけて、特定のCommunityNoteのみを対象にしてPostデータを取るなど。

可能性のあるパターン

• CommunityNoteのデータから言語を先に推定し、日本語のみを対象にしてPostデータを取得
• CommunityNoteのデータからキーワードでフィルタし、該当するものを対象にしてPostデータを取得
  • 派生パターンとしてはCommunityNoteの条件（HELPFULのみや、日時期間の指定など）によってフィルタし、該当するものを対象にする
• CommunityNoteのデータから特定のトピックに関連するかをAIで推論し、該当するものを対象にしてPostデータを取得
• 以上のパターンを組み合わせてできるようにすることも重要

Community Noteが紐づいている以外の条件でもPostを取得する

CommunityNoteがついているかかかわらず、特定の条件を満たすPostデータを取得できるようにする。

• ProプランのサーチAPIでの条件検索
• 外部からのデータインポート（協力チームからの提供データをインポートすることを想定。データモデルはBirdXplorerに準拠させるためにコンバーターは別途必要）

元データの充実

現状BirdXplorer APIのためにETLがあるという位置づけで機能開発されているものを、それ以外の需要にも対応するために、取れるデータはアーカイブとしてすべて取る。

• 取れるものはすべて取る方針
• X APIのフルアーカイブをつくる（関連するもののみ）

[yu23ki14]

@claude このIssueは大きなスコープです。これを実装していくために必要な具体的なステップを開発者レベルで教えて。

[claude]

Claude finished @yu23ki14's task in 1m 26s —— View job

---

ETL柔軟性向上の実装ステップ分析

大きなスコープの改善要件を分析し、開発者レベルの具体的な実装ステップを提供します。

分析タスク

• 現在のETLアーキテクチャを分析
• 改善要件を3つの領域に整理
• 具体的な実装ステップを策定
• 優先順位と依存関係を明確化

---

現在のETLアーキテクチャ分析

現在の処理フロー:

Community Notes TSV → SQLite → X API (Posts) → PostgreSQL → AI処理 → CSV → S3

主要コンポーネント:

• extract.py:19: Community Notes + Post データ取得
• transform.py:27: AI言語検出 + トピック推論 + CSV変換
• load.py:9: S3アップロード
• postlookup.py:53: X API呼び出し (現在は単一Post取得のみ)

---

改善領域別実装ステップ

1. Post取得の効率化 (優先度: 高)

1.1 フィルタリング機能の実装

実装場所: extract.py の Post取得部分（line 99-104）を拡張

# Step 1: フィルタ設定クラスの追加
class PostExtractionFilter:
    language_filter: Optional[List[str]] = None
    keyword_filter: Optional[List[str]] = None  
    topic_filter: Optional[List[int]] = None
    note_status_filter: Optional[List[str]] = None
    date_range_filter: Optional[tuple] = None

具体的な実装ステップ:

1. settings.pyにフィルタ設定を追加
2. extract.pyにapply_note_filters()関数を実装
3. Community Note段階でのAI前処理パイプラインを構築
4. 既存のNote取得クエリ（line 100-104）にフィルタ条件を統合

1.2 AI前処理パイプライン

実装場所: extract.pyとtransform.pyの間に新規モジュール

# Step 2: 前処理AIサービスの実装
def preprocess_notes_for_filtering(notes: List[RowNoteRecord]) -> List[FilteredNote]:
    # 言語検出を先行実行
    # キーワードマッチング
    # 簡易トピック分類

実装ステップ:

1. lib/ai_model/にpreprocessing_service.pyを追加
2. 軽量な言語検出（ローカルライブラリ使用）
3. キーワードベースフィルタリング
4. コスト効率の良いトピック事前分類

2. 多様なPost取得方法 (優先度: 中)

2.1 Search API統合

実装場所: lib/x/に新規ファイルsearch_service.py

実装ステップ:

1. X API v2 Search エンドポイント統合
2. 検索クエリ設定システム（settings.py）
3. 検索結果のPost保存処理
4. 既存Post重複チェック機能

# 新機能実装
class XSearchService:
    def search_posts(self, query: str, max_results: int) -> List[Post]:
        # Pro plan Search API implementation
    
    def search_by_keywords(self, keywords: List[str]) -> List[Post]:
        # Keyword-based search

2.2 外部データインポート機能

実装場所: 新規import/ディレクトリ

実装ステップ:

1. import/data_converter.py: 外部データ → BirdXplorer形式変換
2. import/validation.py: データ形式検証
3. データソース設定システム
4. インポート履歴管理

3. データアーカイブ機能 (優先度: 中)

3.1 完全データ保存

実装場所: extract.pyを拡張、新規archive/モジュール

実装ステップ:

1. X API レスポンス生データの完全保存
2. アーカイブ専用PostgreSQLテーブル設計
3. 段階的データ移行システム
4. データ圧縮・ストレージ最適化

3.2 メタデータ充実

実装場所: birdxplorer_common/models.pyを拡張

追加データ要素:

• 取得時刻、処理履歴
• データ品質メトリクス
• 関連性スコア
• 処理フラグ・ステータス

---

実装優先順位と推奨順序

フェーズ1 (1-2週間): 基盤整備

1. 設定システム拡張: settings.pyにフィルタ設定追加
2. フィルタリング基盤: extract.pyに基本フィルタ機能実装
3. 軽量AI前処理: 言語検出の先行実行

フェーズ2 (2-3週間): コア機能実装

4. Search API統合: X Search API v2の実装
5. フィルタリング高度化: AI+ルールベースの組み合わせ
6. アーカイブ基盤: 生データ保存システム

フェーズ3 (2-4週間): 統合・拡張

7. 外部データインポート: 協力チームデータ対応
8. データ充実: メタデータ・関連性情報追加
9. パフォーマンス最適化: 並列処理・キャッシュ

---

技術的考慮事項

コスト管理:

• X API呼び出し最適化（バッチ処理、キャッシュ）
• AI処理の段階的実行（軽量→高精度）

データ一貫性:

• 既存データモデルとの互換性維持
• 段階的移行戦略

拡張性:

• プラグイン型フィルタシステム
• 設定駆動型パイプライン

この実装計画により、ETLの柔軟性を段階的に向上させ、コスト効率と機能性のバランスを保ちながら要件を実現できます。

---

[yu23ki14]

@claude
Extractの段階でどのCommunityNoteに紐づくPostデータをAPIから取得するかをフィルだできるようにします。そうしないと結局すべてのPostデータを取る必要があり、無駄になるからです。

また、場合によっては処理の順番を入れ替えるといったユースケースも考えられます。例えば、言語推定の順番です。
パターン1: CommunityNoteのデータのみで言語推定し、その後該当する言語のPostデータのみを取得。
パターン2: CommunityNoteに紐づくすべてのPostデータを取り、その後CNとPostのデータ両方をつかって言語を推定。

このようにAIを使うタイミングが変化する場合もあります。

これを柔軟に変更しながら実行できるようにするために、各機能をモジュール化して、main.pyもしくはextract.py, transform.py, load.pyの中身を簡単に書き換えるという方法が考えられます。もっとよい方法があれば教えて下さい。
実行にはAWS CDKをつかってAWSへデプロイし、リポジトリをクローンする形で実行することを考えています。スケジューラーなどはCDK側に実装します。

[claude]

Claude finished @yu23ki14's task in 2m 27s —— View job

---

ETL柔軟性向上の設計案分析

ユーザーからの要件を分析し、柔軟でモジュール化されたETLアーキテクチャ設計案を検討します。

分析タスク

• 現在のETLコード構造を詳細分析
• パイプライン設計パターンの検討
• フィルタリングシステムの設計
• 設定駆動型アーキテクチャの提案
• AWS CDK デプロイメント考慮事項
• 具体的な実装例の提示

---

現在のETLアーキテクチャ分析

現在の処理フロー（固定）:

main.py → extract.py → transform.py → load.py

主要な課題:

1. extract.py:99-105: 全Community Noteに対してPost取得（フィルタリングなし）
2. transform.py:75: 言語検出がPost取得後（APIコスト無駄）
3. 処理順序が固定されており、ビジネス要件に応じた変更が困難
4. 設定(settings.py)は静的で、実行時の動的な条件変更ができない

---

推奨アーキテクチャ設計: パイプライン・コンポーネント・パターン

1. コンポーネント分離設計

# pipeline/components/
class PipelineComponent:
    def execute(self, data: Any, context: PipelineContext) -> Any:
        pass

class NoteExtractorComponent(PipelineComponent):
    def execute(self, data, context):
        # Community Notes取得ロジック
        
class NoteFilterComponent(PipelineComponent):
    def execute(self, notes, context):
        # フィルタリングロジック（言語、キーワード、トピック等）
        
class PostExtractorComponent(PipelineComponent):
    def execute(self, filtered_notes, context):
        # フィルタ済みNoteに対するPost取得

2. パイプライン設定駆動型実行

# pipeline_config.yaml
pipelines:
  language_first_pattern:
    components:
      - name: "note_extractor"
      - name: "note_language_detector"  # AI処理（軽量）
      - name: "note_language_filter"    # 日本語のみ
      - name: "post_extractor"          # フィルタ済みに対してAPI
      - name: "data_transformer"
      - name: "data_loader"
      
  post_first_pattern:
    components:
      - name: "note_extractor"
      - name: "post_extractor"          # 全Note対象
      - name: "combined_language_detector" # Note+Post両方でAI処理
      - name: "data_transformer"
      - name: "data_loader"

3. 設定駆動型main.py

# main.py (新設計)
from pipeline.pipeline_executor import PipelineExecutor
from pipeline.config_loader import load_pipeline_config
import sys

def main():
    config_name = sys.argv[1] if len(sys.argv) > 1 else "default"
    config = load_pipeline_config(config_name)
    
    executor = PipelineExecutor(config)
    executor.execute()

if __name__ == "__main__":
    main()

AWS CDK実行例:

# 言語優先パターン
python main.py language_first_pattern

# Post全取得パターン  
python main.py post_first_pattern

# カスタム設定
python main.py custom_filter_config

---

フィルタリングシステム設計

1. 多段階フィルタ・チェーン

# pipeline/filters/
class FilterChain:
    def __init__(self, filters: List[Filter]):
        self.filters = filters
    
    def apply(self, notes: List[RowNoteRecord]) -> List[RowNoteRecord]:
        result = notes
        for filter in self.filters:
            result = filter.apply(result)
            logging.info(f"Filter {filter.__class__.__name__}: {len(result)} notes remaining")
        return result

class LanguageFilter(Filter):
    def __init__(self, target_languages: List[str]):
        self.target_languages = target_languages
        self.ai_service = get_ai_service()
    
    def apply(self, notes: List[RowNoteRecord]) -> List[RowNoteRecord]:
        filtered = []
        for note in notes:
            detected_lang = self.ai_service.detect_language(note.summary)
            if detected_lang in self.target_languages:
                filtered.append(note)
        return filtered

class KeywordFilter(Filter):
    def apply(self, notes: List[RowNoteRecord]) -> List[RowNoteRecord]:
        # キーワードマッチング実装
        
class TopicFilter(Filter):
    def apply(self, notes: List[RowNoteRecord]) -> List[RowNoteRecord]:
        # AIトピック分類実装

2. 設定ベース フィルタ構成

# filters_config.yaml
filter_presets:
  japanese_only:
    - type: "language"
      params:
        target_languages: ["ja"]
        
  covid_related:
    - type: "keyword"
      params:
        keywords: ["COVID", "コロナ", "ワクチン"]
    - type: "topic"
      params:
        target_topics: ["health", "medical"]
        
  high_engagement:
    - type: "note_status"
      params:
        statuses: ["HELPFUL"]
    - type: "custom"
      params:
        min_helpful_count: 10

---

設定駆動型アーキテクチャ提案

1. 統合設定管理

# config/etl_config.py
@dataclass
class ETLConfig:
    pipeline_name: str
    components: List[ComponentConfig]
    filters: List[FilterConfig]
    ai_settings: AIConfig
    extraction_settings: ExtractionConfig
    
    @classmethod
    def from_file(cls, config_path: str) -> 'ETLConfig':
        # YAML/JSONからの設定読み込み
        
    def to_env_vars(self) -> Dict[str, str]:
        # AWS CDK用の環境変数変換

2. AWS CDK考慮事項

# cdk_integration/
class ETLTaskDefinition:
    def __init__(self, config: ETLConfig):
        self.config = config
        
    def to_fargate_task(self) -> ecs.TaskDefinition:
        # Fargate用のタスク定義生成
        # 環境変数でパイプライン設定を渡す
        environment = self.config.to_env_vars()
        environment.update({
            "PIPELINE_CONFIG": self.config.pipeline_name,
            "FILTERS_CONFIG": json.dumps(self.config.filters)
        })
        
    def to_lambda_function(self) -> lambda_.Function:
        # Lambda用の関数定義（小規模タスク向け）

3. 実行時設定注入

# pipeline/pipeline_executor.py
class PipelineExecutor:
    def __init__(self, config: ETLConfig):
        self.config = config
        self.components = self._build_components()
        
    def execute(self):
        data = None
        context = PipelineContext(self.config)
        
        for component in self.components:
            data = component.execute(data, context)
            
        return data

---

具体的な実装ステップ

Phase 1: 基盤整備 (1週間)

1. pipeline/ ディレクトリ構造作成
2. PipelineComponent 基底クラス実装
3. 設定ローダー(config_loader.py)実装
4. 既存コードのコンポーネント化開始

Phase 2: フィルタリング実装 (1-2週間)
5. Filter 基底クラスとFilterChain実装
6. 基本フィルタ類（Language, Keyword, Status）実装
7. 設定ファイルフォーマット確定
8. Extract段階でのフィルタ統合

Phase 3: パイプライン統合 (1週間)
9. PipelineExecutor実装
10. main.py改修（設定駆動化）
11. AWS CDK統合コード実装
12. テスト・デバッグ

具体的なファイル構成:

etl/
├── src/birdxplorer_etl/
│   ├── main.py (改修)
│   ├── pipeline/
│   │   ├── components/
│   │   │   ├── note_extractor.py
│   │   │   ├── post_extractor.py  
│   │   │   └── data_transformer.py
│   │   ├── filters/
│   │   │   ├── language_filter.py
│   │   │   └── keyword_filter.py
│   │   ├── pipeline_executor.py
│   │   └── config_loader.py
│   ├── config/
│   │   ├── pipeline_configs/
│   │   │   ├── language_first.yaml
│   │   │   └── post_first.yaml
│   │   └── filter_presets.yaml
│   └── legacy/ (既存ファイル移動)
│       ├── extract.py
│       ├── transform.py
│       └── load.py

---

メリット・効果

1. APIコスト削減: Extract段階でのフィルタリングによりX API呼び出し削減
2. 柔軟性向上: 設定変更のみで処理順序・フィルタ条件変更可能
3. 運用性向上: AWS CDKでのスケジューラ設定時に、パイプライン種別を環境変数で指定
4. テスト容易性: コンポーネント単位でのユニットテスト可能
5. 拡張性: 新しいフィルタ・コンポーネントの追加が容易

この設計により、単一ファイル修正ではなく設定変更で柔軟なETL実行が可能になり、AWS CDK環境での運用に最適化されます。

---

[yu23ki14]

@claude
この方針で問題なさそうです。以下について具体的なIssueをつくってください。各IssueにはこのIssue#154をメンションして全体像がどこに書いてあるかがわかるようにしてください。

Phase 1: 基盤整備

• pipeline/ ディレクトリ構造作成
• PipelineComponent 基底クラス実装
• 設定ローダー(config_loader.py)実装
• 既存コードのコンポーネント化開始

Phase 2: フィルタリング実装

• Filter 基底クラスとFilterChain実装
• 基本フィルタ類（Language, Keyword, Status）実装
• 設定ファイルフォーマット確定
• Extract段階でのフィルタ統合

Phase 3: パイプライン統合

• PipelineExecutor実装
• main.py改修

[claude]

Claude finished @yu23ki14's task in 2m 18s —— View job

---

ETL Pipeline Enhancement - Issue Creation Request

Analysis & Planning

• Analyze the request for creating GitHub issues
• Review the 3-phase breakdown structure
• Create Phase 1 issue templates
• Create Phase 2 issue templates
• Create Phase 3 issue templates

---

Unfortunately, I don't have access to GitHub issue creation tools with my current permissions. However, I can provide you with detailed issue templates that you can copy-paste to create these issues manually.

Below are the formatted issue templates for each phase:

Phase 1 Issues

Issue 1: Pipeline Directory Structure and Base Classes

**Title:** ETL Pipeline基盤整備: ディレクトリ構造とPipelineComponent基底クラス実装

**Body:**
# ETL Pipeline基盤整備: ディレクトリ構造とPipelineComponent基底クラス実装

関連Issue: #154 (ETLの柔軟性向上)

## 概要
ETLパイプラインの柔軟性向上のための基盤整備として、モジュール化されたアーキテクチャの基礎となるディレクトリ構造とPipelineComponent基底クラスを実装します。

## 実装内容

### 1. Pipeline ディレクトリ構造作成

etl/src/birdxplorer_etl/
├── pipeline/
│ ├── init.py
│ ├── components/
│ │ └── init.py
│ ├── filters/
│ │ └── init.py
│ ├── config/
│ │ └── init.py
│ └── base/
│ └── init.py

### 2. PipelineComponent 基底クラス実装
- `pipeline/base/component.py`: 基底クラス定義
- `pipeline/base/context.py`: パイプライン実行コンテキスト
- 型定義とインターフェース仕様

## 技術要件
- Python 3.10+
- 型ヒント必須
- 抽象基底クラス使用
- ログ機能統合

## 受け入れ条件
- [ ] ディレクトリ構造が正しく作成されている
- [ ] PipelineComponent基底クラスが実装されている
- [ ] PipelineContextクラスが実装されている
- [ ] 型チェックが通る
- [ ] 基本的な単体テストが実装されている

**Priority:** High
**Labels:** enhancement, etl, phase-1

Issue 2: Configuration Loader Implementation

**Title:** ETL Pipeline基盤整備: 設定ローダー(config_loader.py)実装

**Body:**
# ETL Pipeline基盤整備: 設定ローダー実装

関連Issue: #154 (ETLの柔軟性向上)

## 概要
設定駆動型ETLパイプラインの実現のため、YAML/JSON設定ファイルを読み込み、パイプライン設定オブジェクトに変換する設定ローダーを実装します。

## 実装内容

### 1. 設定ローダー実装
- `pipeline/config/config_loader.py`: メイン実装
- `pipeline/config/models.py`: 設定データクラス定義
- YAML/JSON両対応の設定ファイル読み込み

### 2. 設定データ構造定義
```python
@dataclass
class ETLConfig:
    pipeline_name: str
    components: List[ComponentConfig]
    filters: List[FilterConfig]
    ai_settings: AIConfig
    extraction_settings: ExtractionConfig

3. 設定ファイルフォーマット例

pipeline_name: "language_first_pattern"
components:
  - name: "note_extractor"
    type: "NoteExtractorComponent"
  - name: "note_language_filter"
    type: "LanguageFilterComponent"

技術要件

• PyYAML使用
• 型安全な設定読み込み
• バリデーション機能
• デフォルト設定対応

受け入れ条件

• 設定ローダーが実装されている
• YAML/JSON両方の形式に対応
• 設定バリデーション機能がある
• エラーハンドリングが適切
• 単体テストが実装されている

Priority: High
Labels: enhancement, etl, phase-1

#### Issue 3: Existing Code Componentization
```markdown
**Title:** ETL Pipeline基盤整備: 既存コードのコンポーネント化開始

**Body:**
# ETL Pipeline基盤整備: 既存コードのコンポーネント化

関連Issue: #154 (ETLの柔軟性向上)

## 概要
既存のextract.py, transform.py, load.pyの機能をPipelineComponentベースのモジュール化されたコンポーネントに変換します。

## 実装内容

### 1. 既存コードのコンポーネント分割
- `NoteExtractorComponent`: Community Notes取得機能
- `PostExtractorComponent`: X API Post取得機能
- `DataTransformerComponent`: データ変換・AI処理機能
- `DataLoaderComponent`: S3アップロード機能

### 2. 段階的移行戦略
- 既存ファイルをlegacy/ディレクトリに移動
- 新コンポーネントで既存機能を再実装
- 互換性レイヤーの実装

### 3. 設定対応
- 各コンポーネントの設定パラメータ定義
- 既存設定値との互換性確保

## 技術要件
- PipelineComponent基底クラス継承
- 既存機能の完全互換性
- 段階的移行可能な設計
- テストカバレッジ維持

## 受け入れ条件
- [ ] 主要コンポーネントクラスが実装されている
- [ ] 既存機能が正常に動作する
- [ ] 段階的移行が可能
- [ ] テストが実装されている
- [ ] ドキュメントが更新されている

**Priority:** Medium
**Labels:** enhancement, etl, phase-1, refactoring

Phase 2 Issues

Issue 4: Filter System Implementation

**Title:** ETL Pipeline フィルタリング実装: Filter基底クラスとFilterChain

**Body:**
# ETL Pipeline フィルタリング実装: Filter基底クラスとFilterChain

関連Issue: #154 (ETLの柔軟性向上)

## 概要
Community Noteデータを効率的にフィルタリングするための基盤となるFilter基底クラスとFilterChainを実装します。

## 実装内容

### 1. Filter基底クラス実装
```python
class Filter(ABC):
    @abstractmethod
    def apply(self, notes: List[RowNoteRecord]) -> List[RowNoteRecord]:
        pass

2. FilterChain実装

• 複数フィルタの連鎖実行
• フィルタ実行順序制御
• 各段階での結果ログ出力

3. 設定ベースフィルタ構成

• 設定ファイルからのフィルタ生成
• フィルタパラメータの動的設定

技術要件

• 抽象基底クラス使用
• ジェネリック型対応
• ログ機能統合
• 型安全性確保

受け入れ条件

• Filter基底クラスが実装されている
• FilterChainクラスが実装されている
• 設定ベースでのフィルタ生成ができる
• フィルタ実行ログが出力される
• 単体テストが実装されている

Priority: High
Labels: enhancement, etl, phase-2

#### Issue 5: Basic Filter Implementations
```markdown
**Title:** ETL Pipeline 基本フィルタ実装: Language, Keyword, Status フィルタ

**Body:**
# ETL Pipeline 基本フィルタ実装

関連Issue: #154 (ETLの柔軟性向上)

## 概要
Post取得コスト削減のための基本的なフィルタクラス（Language, Keyword, Status）を実装します。

## 実装内容

### 1. LanguageFilter実装
- AIサービスを使用した言語検出
- 対象言語での絞り込み
- バッチ処理対応

### 2. KeywordFilter実装
- テキストマッチング機能
- 正規表現対応
- 大文字小文字の区別設定

### 3. StatusFilter実装
- Community Note ステータスでの絞り込み
- HELPFUL, NOT_HELPFUL, NEEDS_MORE_RATINGS対応

### 4. 設定ファイル対応
```yaml
filter_presets:
  japanese_only:
    - type: "language"
      params:
        target_languages: ["ja"]

技術要件

• Filter基底クラス継承
• 設定ファイル対応
• AI連携機能
• エラーハンドリング

受け入れ条件

• LanguageFilterが実装されている
• KeywordFilterが実装されている
• StatusFilterが実装されている
• 設定ファイルからの設定読み込みができる
• 各フィルタの単体テストが実装されている

Priority: High
Labels: enhancement, etl, phase-2

#### Issue 6: Configuration File Format Definition
```markdown
**Title:** ETL Pipeline 設定ファイルフォーマット確定

**Body:**
# ETL Pipeline 設定ファイルフォーマット確定

関連Issue: #154 (ETLの柔軟性向上)

## 概要
パイプライン実行とフィルタリングの設定ファイルフォーマットを確定し、実用的な設定例を作成します。

## 実装内容

### 1. パイプライン設定フォーマット定義
```yaml
pipeline_name: "language_first_pattern"
components:
  - name: "note_extractor"
    type: "NoteExtractorComponent"
    config:
      days_ago: 7
  - name: "language_filter"
    type: "LanguageFilterComponent"
    config:
      target_languages: ["ja"]

2. フィルタ設定フォーマット定義

filter_presets:
  japanese_healthcare:
    - type: "language"
      params:
        target_languages: ["ja"]
    - type: "keyword"
      params:
        keywords: ["医療", "健康", "ワクチン"]

3. 実用的な設定例作成

• 言語優先パターン設定
• 全取得パターン設定
• カスタムフィルタ組み合わせ例

技術要件

• YAMLスキーマ定義
• 設定検証ルール
• デフォルト値設定
• 環境変数置換対応

受け入れ条件

• 設定ファイルスキーマが定義されている
• 実用的な設定例が作成されている
• 設定検証機能が実装されている
• ドキュメントが整備されている

Priority: Medium
Labels: enhancement, etl, phase-2, documentation

#### Issue 7: Extract Stage Filter Integration
```markdown
**Title:** ETL Pipeline Extract段階でのフィルタ統合

**Body:**
# ETL Pipeline Extract段階でのフィルタ統合

関連Issue: #154 (ETLの柔軟性向上)

## 概要
Extract段階でCommunity Noteのフィルタリングを行い、フィルタ済みのNoteに対してのみPost取得を実行する機能を実装します。

## 実装内容

### 1. NoteExtractorComponentの拡張
- フィルタ統合機能
- フィルタ結果のログ出力
- フィルタ前後の件数比較

### 2. PostExtractorComponentの改修
- フィルタ済みNote IDリストでの取得
- API呼び出し最適化
- コスト効率の改善

### 3. フィルタ実行フロー

Notes取得 → フィルタ適用 → フィルタ済みNotes → Post取得

## 技術要件
- 既存Extract機能との互換性
- フィルタチェーンの統合
- API呼び出し削減効果の測定
- ログ機能の充実

## 受け入れ条件
- [ ] Extract段階でのフィルタリングが実装されている
- [ ] フィルタ済みNoteに対するPost取得が動作する
- [ ] API呼び出し削減効果が確認できる
- [ ] 詳細なログが出力される
- [ ] 統合テストが実装されている

**Priority:** High
**Labels:** enhancement, etl, phase-2, api-optimization

Phase 3 Issues

Issue 8: Pipeline Executor Implementation

**Title:** ETL Pipeline統合: PipelineExecutor実装

**Body:**
# ETL Pipeline統合: PipelineExecutor実装

関連Issue: #154 (ETLの柔軟性向上)

## 概要
設定ファイルベースでパイプラインコンポーネントを動的に実行するPipelineExecutorを実装します。

## 実装内容

### 1. PipelineExecutor実装
- 設定ファイルからのコンポーネント読み込み
- コンポーネントの順次実行
- 実行コンテキストの管理
- エラーハンドリングと復旧

### 2. コンポーネント生成機能
- ファクトリーパターンによるコンポーネント生成
- 設定パラメータの注入
- 依存関係の解決

### 3. 実行監視・ログ機能
- 各コンポーネントの実行時間測定
- 進捗状況の報告
- エラー詳細情報の出力

## 技術要件
- ファクトリーパターン使用
- 型安全な実行環境
- 構造化ログ出力
- 例外処理の統一

## 受け入れ条件
- [ ] PipelineExecutorが実装されている
- [ ] 設定ファイルベースでの実行ができる
- [ ] コンポーネントファクトリーが実装されている
- [ ] 実行ログが適切に出力される
- [ ] エラーハンドリングが実装されている
- [ ] 統合テストが実装されている

**Priority:** High
**Labels:** enhancement, etl, phase-3

Issue 9: Main.py Refactoring for Configuration-Driven Execution

**Title:** ETL Pipeline統合: main.py設定駆動実行への改修

**Body:**
# ETL Pipeline統合: main.py設定駆動実行への改修

関連Issue: #154 (ETLの柔軟性向上)

## 概要
現在の固定処理フローから設定ファイルベースの動的パイプライン実行に変更するため、main.pyを改修します。

## 実装内容

### 1. main.py改修
```python
def main():
    config_name = sys.argv[1] if len(sys.argv) > 1 else "default"
    config = load_pipeline_config(config_name)
    
    executor = PipelineExecutor(config)
    executor.execute()

2. 設定ファイル例作成

• config/pipeline_configs/language_first.yaml
• config/pipeline_configs/post_first.yaml
• config/pipeline_configs/default.yaml

3. 既存コードの移行完了

• legacy/ディレクトリへの移動完了
• 新システムでの動作確認

4. AWS CDK対応

• 環境変数での設定指定対応
• コンテナ実行時の設定注入

技術要件

• 既存機能の互換性維持
• AWS環境での実行対応
• 設定検証機能
• 詳細なログ出力

受け入れ条件

• main.pyが設定駆動型に改修されている
• 設定ファイル例が作成されている
• 既存機能が正常に動作する
• AWS CDK環境での実行ができる
• ドキュメントが更新されている
• 統合テストが実装されている

Priority: High
Labels: enhancement, etl, phase-3

---

### 次のステップ

1. 上記のテンプレートを使用して9つのIssueを作成してください
2. 各Issueに適切なLabelとAssigneeを設定してください
3. Milestoneを作成して各フェーズをグループ化することをお勧めします

### 推奨実装順序

**Phase 1 (基盤整備):**
1. Issue 1: ディレクトリ構造とPipelineComponent基底クラス
2. Issue 2: 設定ローダー実装
3. Issue 3: 既存コードのコンポーネント化

**Phase 2 (フィルタリング実装):**
4. Issue 4: Filter基底クラスとFilterChain
5. Issue 5: 基本フィルタ実装
6. Issue 6: 設定ファイルフォーマット確定
7. Issue 7: Extract段階でのフィルタ統合

**Phase 3 (パイプライン統合):**
8. Issue 8: PipelineExecutor実装
9. Issue 9: main.py改修

これらのIssueは依存関係があるため、順序通りの実装をお勧めします。

[yu23ki14]

@claude
あなたはGitHubにissueを新しく作ることができますか？できるならしてください。

[claude]

Claude finished @yu23ki14's task in 34s —— View job

---

GitHub Issue Creation Request

Tasks

• Check available tools for GitHub issue creation capabilities
• Explain limitations and provide alternative solution

---

Issue Creation Capability Analysis

Unfortunately, I cannot create GitHub issues directly. My available tools are limited to:

• File operations (commit, delete, edit)
• Comment updates
• Code analysis and execution

I do not have access to GitHub's issue creation API.

Alternative Solution

Since I cannot create the issues directly, I recommend using the detailed issue templates I provided in my previous response. Here are the 9 issues you can create manually:

Phase 1: 基盤整備 (3 Issues)

1. Pipeline Directory Structure and Base Classes - Priority: High
2. Configuration Loader Implementation - Priority: High
3. Existing Code Componentization - Priority: Medium

Phase 2: フィルタリング実装 (4 Issues)

4. Filter System Implementation - Priority: High
5. Basic Filter Implementations - Priority: High
6. Configuration File Format Definition - Priority: Medium
7. Extract Stage Filter Integration - Priority: High

Phase 3: パイプライン統合 (2 Issues)

8. Pipeline Executor Implementation - Priority: High
9. Main.py Refactoring - Priority: High

Quick Creation Process

1. Copy each issue template from my previous response
2. Create new issue in GitHub
3. Paste the template content
4. Add labels: enhancement, etl, phase-1/2/3
5. Add reference to Issue ETLの柔軟性向上 #154 in each issue body

Each template is ready-to-use with:

• Detailed implementation requirements
• Technical specifications
• Acceptance criteria
• Priority levels

Would you like me to help with any specific aspect of these implementations once the issues are created?

---