Service for learning from scraping patterns and improving suggestions.
| 76 | |
| 77 | |
| 78 | class PatternLearner: |
| 79 | """Service for learning from scraping patterns and improving suggestions.""" |
| 80 | |
| 81 | def __init__(self): |
| 82 | self.redis_client: Optional[redis.Redis] = None |
| 83 | self.pattern_cache = {} |
| 84 | self.domain_cache = {} |
| 85 | self.learning_threshold = 5 # Minimum executions before learning |
| 86 | |
| 87 | async def initialize(self): |
| 88 | """Initialize the pattern learner.""" |
| 89 | # Connect to Redis for caching |
| 90 | self.redis_client = await redis.from_url( |
| 91 | settings.REDIS_URL, |
| 92 | encoding="utf-8", |
| 93 | decode_responses=True |
| 94 | ) |
| 95 | |
| 96 | # Load frequently used patterns into cache |
| 97 | await self._load_pattern_cache() |
| 98 | |
| 99 | async def _load_pattern_cache(self): |
| 100 | """Load frequently used patterns into memory cache.""" |
| 101 | # This would load from database in production |
| 102 | # For now, initialize with common patterns |
| 103 | self.pattern_cache = { |
| 104 | "ecommerce": { |
| 105 | "common_fields": ["title", "price", "description", "availability", "rating", "image"], |
| 106 | "selectors": { |
| 107 | "title": ["h1", ".product-title", "[itemprop='name']"], |
| 108 | "price": [".price", "[itemprop='price']", ".product-price"], |
| 109 | "description": [".description", "[itemprop='description']", ".product-description"] |
| 110 | } |
| 111 | }, |
| 112 | "news": { |
| 113 | "common_fields": ["title", "content", "author", "date", "category", "tags"], |
| 114 | "selectors": { |
| 115 | "title": ["h1", ".article-title", "[itemprop='headline']"], |
| 116 | "content": ["article", ".article-content", "[itemprop='articleBody']"], |
| 117 | "author": [".author", "[itemprop='author']", ".by-line"] |
| 118 | } |
| 119 | }, |
| 120 | "weather": { |
| 121 | "common_fields": ["temperature", "humidity", "wind_speed", "conditions", "forecast"], |
| 122 | "selectors": { |
| 123 | "temperature": [".temperature", ".temp", "[data-testid='temperature']"], |
| 124 | "humidity": [".humidity", "[data-testid='humidity']"], |
| 125 | "wind_speed": [".wind", "[data-testid='wind']"] |
| 126 | } |
| 127 | } |
| 128 | } |
| 129 | |
| 130 | def _extract_domain(self, url: str) -> str: |
| 131 | """Extract domain from URL.""" |
| 132 | from urllib.parse import urlparse |
| 133 | parsed = urlparse(url) |
| 134 | return parsed.netloc or "unknown" |
| 135 |