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]