EmpathemeBot_dev / src /processing /scrape_to_metadata.py
tktm8's picture
Upload 59 files
fb05e78 verified
from pathlib import Path
from typing import List, Dict, Any, Optional
from src.processing.batch_processor import BatchProcessor
class ScrapeToMetadata:
"""スクレイピング→メタデータ生成を一気通貫で処理するファサードクラス"""
def __init__(self, wait_time: float = 1.0):
"""
Args:
wait_time: リクエスト間の待機時間(秒)
"""
self.batch_processor = BatchProcessor(wait_time)
async def process_urls_batch(
self,
urls: List[str],
start_id: int = 1,
mode: str = "memory",
show_progress: bool = True,
save_dir: Optional[Path] = None,
verbose: bool = False
) -> List[Dict[str, Any]]:
"""
複数URLをバッチ処理してメタデータを生成
Args:
urls: 処理するURLのリスト
start_id: 開始ID
mode: "memory" または "save"
show_progress: プログレス表示の有無
save_dir: saveモード時のMarkdownファイル保存先ディレクトリ
verbose: 詳細ログを表示するか
Returns:
生成されたドキュメントメタデータのリスト
"""
return await self.batch_processor.process_urls_batch(
urls=urls,
start_id=start_id,
mode=mode,
show_progress=show_progress,
save_dir=save_dir,
verbose=verbose
)
def save_metadata(self, documents: List[Dict[str, Any]], output_path: Path):
"""メタデータをJSON形式で保存
Args:
documents: 保存するドキュメントメタデータのリスト
output_path: 出力ファイルパス
"""
self.batch_processor.save_metadata(documents, output_path)