57 lines
1.4 KiB
Python
57 lines
1.4 KiB
Python
import hashlib
|
|
from abc import ABC, abstractmethod
|
|
from dataclasses import dataclass, field
|
|
from datetime import datetime
|
|
|
|
|
|
@dataclass
|
|
class PipelineResult:
|
|
stored: int = 0
|
|
filtered: int = 0
|
|
notified: int = 0
|
|
|
|
|
|
@dataclass
|
|
class CrawlResult:
|
|
source_code: str
|
|
source_name: str
|
|
total_count: int = 0
|
|
new_count: int = 0
|
|
announcements: list = field(default_factory=list)
|
|
error_message: str | None = None
|
|
crawled_at: datetime = field(default_factory=datetime.now)
|
|
duration: float = 0.0
|
|
pipeline_result: PipelineResult | None = None
|
|
|
|
@property
|
|
def success(self) -> bool:
|
|
return self.error_message is None
|
|
|
|
|
|
@dataclass
|
|
class PipelineConfig:
|
|
filter_enabled: bool = True
|
|
keywords: list[str] = field(default_factory=list)
|
|
dedup_enabled: bool = True
|
|
notify_mode: str = "filtered"
|
|
mark_sent: bool = False
|
|
|
|
|
|
class BaseSpider(ABC):
|
|
name: str
|
|
source_code: str
|
|
source_name: str
|
|
|
|
@abstractmethod
|
|
async def crawl(self) -> CrawlResult:
|
|
...
|
|
|
|
def get_pipeline_config(self) -> PipelineConfig:
|
|
return PipelineConfig()
|
|
|
|
@staticmethod
|
|
def generate_content_hash(title: str, publish_date: str, purchase_name: str,
|
|
content_url: str, source_code: str) -> str:
|
|
content = f"{title}|{publish_date}|{purchase_name}|{content_url}|{source_code}"
|
|
return hashlib.sha256(content.encode("utf-8")).hexdigest()
|