58 lines
1.3 KiB
Python
58 lines
1.3 KiB
Python
from abc import ABC, abstractmethod
|
|
from typing import Dict, Any, Optional
|
|
import logging
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.models import Species, ApiKey
|
|
|
|
|
|
class BaseScraper(ABC):
|
|
"""Base class for all image scrapers."""
|
|
|
|
name: str = "base"
|
|
requires_api_key: bool = True
|
|
|
|
@abstractmethod
|
|
def scrape_species(
|
|
self,
|
|
species: Species,
|
|
db: Session,
|
|
logger: Optional[logging.Logger] = None
|
|
) -> Dict[str, int]:
|
|
"""
|
|
Scrape images for a species.
|
|
|
|
Args:
|
|
species: The species to scrape images for
|
|
db: Database session
|
|
logger: Optional logger for debugging
|
|
|
|
Returns:
|
|
Dict with 'downloaded' and 'rejected' counts
|
|
"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def test_connection(self, api_key: ApiKey) -> str:
|
|
"""
|
|
Test API connection.
|
|
|
|
Args:
|
|
api_key: The API key configuration
|
|
|
|
Returns:
|
|
Success message
|
|
|
|
Raises:
|
|
Exception if connection fails
|
|
"""
|
|
pass
|
|
|
|
def get_api_key(self, db: Session) -> ApiKey:
|
|
"""Get API key for this scraper."""
|
|
return db.query(ApiKey).filter(
|
|
ApiKey.source == self.name,
|
|
ApiKey.enabled == True
|
|
).first()
|