50 lines
1.7 KiB
Python
50 lines
1.7 KiB
Python
from abc import ABC, abstractmethod
|
|
|
|
from DrissionPage import Chromium, SessionPage, ChromiumOptions
|
|
|
|
from database.database import get_session
|
|
from database.tnews.crud import update_news_by_id
|
|
from database.tnews.model import TNews
|
|
from log.log_manager import log
|
|
|
|
|
|
class ContentBase(ABC):
|
|
def __init__(self, news: TNews):
|
|
self.news = news
|
|
self.session = None # 初始化为 None
|
|
self.browser = None # 初始化为 None
|
|
if news.is_static:
|
|
self.session = SessionPage()
|
|
self.session.get(news.url)
|
|
else:
|
|
co = ChromiumOptions()
|
|
self.browser = Chromium(addr_or_opts=co)
|
|
# self.tab = self.browser.latest_tab
|
|
self.tab = self.browser.new_tab()
|
|
self.tab.get(news.url)
|
|
|
|
@abstractmethod
|
|
def get_content(self):
|
|
"""Abstract method to fetch news from a specific source."""
|
|
pass
|
|
|
|
def get_occurrence_date(self):
|
|
return None
|
|
|
|
def do_seek_task(self):
|
|
"""Saves the list of news to the database if the URL does not already exist."""
|
|
self.news.content = self.get_content()
|
|
if self.news.occurrence_date is None:
|
|
self.news.occurrence_date = self.get_occurrence_date()
|
|
with get_session() as db:
|
|
update_news_by_id(db, self.news)
|
|
log(f'successful fetch {self.news.title} news content into the database.')
|
|
|
|
def finish(self):
|
|
"""Closes the browser and session."""
|
|
if self.tab:
|
|
self.tab.close()
|
|
# if self.browser:
|
|
# self.browser.quit()
|
|
if self.session:
|
|
self.session.close() |