Игорь Бандурист
ef453f661a
All checks were successful
continuous-integration/drone/push Build is passing
124 lines
4.5 KiB
Python
124 lines
4.5 KiB
Python
"""
|
||
Парсер любого источника - универсальный парсер
|
||
"""
|
||
import requests
|
||
from bs4 import BeautifulSoup
|
||
from datetime import datetime
|
||
from newspaper import Article
|
||
from urllib.parse import urljoin, urlparse, urldefrag
|
||
from typing import Set
|
||
|
||
from .base import BaseParser
|
||
from utils import logger
|
||
from services import gpt_response_message, update_bd_and_create_document
|
||
import work_parser as wp
|
||
|
||
|
||
def check_url(url: str) -> bool:
|
||
"""
|
||
Проверяет, существует ли URL в базе данных
|
||
"""
|
||
try:
|
||
response = wp.check_url_exists(url)
|
||
if response.status_code == 200:
|
||
result = response.json()
|
||
# print(result["exists"])
|
||
return result["exists"]
|
||
else:
|
||
return False
|
||
except Exception:
|
||
return False
|
||
|
||
|
||
class UniversalParser(BaseParser):
|
||
"""
|
||
Универсальный парсер для любого источника
|
||
"""
|
||
|
||
def __init__(self, url: str, promt: str):
|
||
super().__init__("universal")
|
||
self.url = url
|
||
self.promt = promt
|
||
|
||
def parse(self) -> None:
|
||
"""
|
||
Основной метод парсинга любого источника
|
||
"""
|
||
print(f"Начало парсинга: {self.url} с промтом: {self.promt}")
|
||
self.start_task(self.url)
|
||
num = 0
|
||
try:
|
||
response = requests.get(self.url)
|
||
# print(response.text)
|
||
response.raise_for_status()
|
||
except requests.RequestException:
|
||
print(f"Ошибка при запросе к {self.url}")
|
||
self.fail_task()
|
||
return
|
||
|
||
soup = BeautifulSoup(response.text, 'html.parser')
|
||
base_domain = urlparse(self.url).netloc
|
||
# print(base_domain)
|
||
|
||
for a_tag in soup.find_all('a', href=True):
|
||
href = a_tag['href'].strip()
|
||
if not href or href.startswith('mailto:') or href.startswith('javascript:'):
|
||
continue
|
||
|
||
# Приведение к абсолютному URL и удаление якорей (#...)
|
||
abs_url = urljoin(self.url, href)
|
||
abs_url, _ = urldefrag(abs_url)
|
||
parsed = urlparse(abs_url)
|
||
|
||
# Фильтр: ссылка должна быть на тот же домен
|
||
if parsed.netloc != base_domain:
|
||
continue
|
||
|
||
|
||
|
||
if not check_url(abs_url) and wp.check_error_url(abs_url):
|
||
try:
|
||
article = Article(abs_url)
|
||
article.download()
|
||
article.parse()
|
||
|
||
print("URL:", abs_url)
|
||
|
||
if len(article.text) > 200 and article.publish_date:
|
||
num += 1
|
||
# Если дата публикации отсутствует - используем текущую
|
||
if article.publish_date:
|
||
time_text = article.publish_date.strftime("%Y/%m/%d %H:%M:%S")
|
||
else:
|
||
time_text = datetime.now().strftime("%Y/%m/%d %H:%M:%S")
|
||
print(f"Дата публикации отсутствует, используем текущую: {time_text}")
|
||
|
||
response_text = gpt_response_message(str(article.text), self.promt)
|
||
# print(response_text)
|
||
if response_text:
|
||
update_bd_and_create_document(
|
||
response_text=response_text,
|
||
article_date=time_text,
|
||
url=abs_url,
|
||
parsed_at=str(datetime.now()),
|
||
original_text=article.text,
|
||
other=self.promt
|
||
)
|
||
else:
|
||
wp.add_error_url(self.url, abs_url)
|
||
except Exception as e:
|
||
print(f"Ошибка при обработке статьи {abs_url}: {e}")
|
||
logger.info(f"Ошибка при обработке статьи {abs_url}: {e}")
|
||
continue
|
||
if num:
|
||
wp.update_source_status(url)
|
||
self.complete_task()
|
||
|
||
|
||
def start_pars_all_istochnik(url: str, promt: str) -> None:
|
||
"""
|
||
Точка входа для парсинга любого источника
|
||
"""
|
||
parser = UniversalParser(url, promt)
|
||
parser.parse()
|