73 lines
2.3 KiB
Python
73 lines
2.3 KiB
Python
from __future__ import annotations
|
||
|
||
import html
|
||
from typing import Final
|
||
from urllib.parse import urlparse
|
||
from xml.etree import ElementTree
|
||
|
||
import requests
|
||
|
||
from app.models import Offer
|
||
from app.sources.base import OfferSource
|
||
|
||
|
||
class FlSource(OfferSource):
|
||
source_name = "fl"
|
||
_default_category_id: Final[int] = 5
|
||
|
||
def __init__(self, rss_url: str, timeout_seconds: int) -> None:
|
||
self._rss_url = rss_url
|
||
self._timeout_seconds = timeout_seconds
|
||
self._session = requests.Session()
|
||
self._session.headers.update(
|
||
{
|
||
"User-Agent": (
|
||
"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
|
||
"(KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36"
|
||
)
|
||
}
|
||
)
|
||
|
||
def fetch_offers(self) -> list[Offer]:
|
||
response = self._session.get(self._rss_url, timeout=self._timeout_seconds)
|
||
response.raise_for_status()
|
||
|
||
root = ElementTree.fromstring(response.content)
|
||
offers: list[Offer] = []
|
||
|
||
for item in root.findall("./channel/item"):
|
||
url = _get_item_text(item, "link")
|
||
external_id = _extract_external_id(url)
|
||
if external_id is None:
|
||
continue
|
||
|
||
offers.append(
|
||
Offer(
|
||
source=self.source_name,
|
||
external_id=external_id,
|
||
title=_get_item_text(item, "title", "Без названия"),
|
||
description=_get_item_text(item, "description"),
|
||
price="Не указан",
|
||
price_amount=None,
|
||
url=url,
|
||
category_id=self._default_category_id,
|
||
published_at=_get_item_text(item, "pubDate"),
|
||
)
|
||
)
|
||
|
||
return offers
|
||
|
||
|
||
def _get_item_text(item: ElementTree.Element, tag_name: str, default: str = "") -> str:
|
||
element = item.find(tag_name)
|
||
if element is None or element.text is None:
|
||
return default
|
||
return html.unescape(element.text.strip())
|
||
|
||
|
||
def _extract_external_id(url: str) -> str | None:
|
||
path_parts = [part for part in urlparse(url).path.split("/") if part]
|
||
if len(path_parts) < 2 or path_parts[0] != "projects":
|
||
return None
|
||
return path_parts[1]
|