112 lines
3.8 KiB
Python
112 lines
3.8 KiB
Python
from __future__ import annotations
|
||
|
||
import html
|
||
from typing import Any
|
||
|
||
import requests
|
||
|
||
from app.models import Offer
|
||
from app.sources.base import OfferSource
|
||
|
||
|
||
class KworkSource(OfferSource):
|
||
source_name = "kwork"
|
||
_url = "https://kwork.ru/projects"
|
||
|
||
def __init__(self, category_ids: list[int], timeout_seconds: int) -> None:
|
||
self._category_ids = category_ids
|
||
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"
|
||
),
|
||
"X-Requested-With": "XMLHttpRequest",
|
||
}
|
||
)
|
||
|
||
def fetch_offers(self) -> list[Offer]:
|
||
offers: list[Offer] = []
|
||
|
||
for category_id in self._category_ids:
|
||
first_page_payload = self._fetch_page_payload(category_id, page=1)
|
||
offers.extend(self._parse_payload(first_page_payload, category_id))
|
||
|
||
last_page = _extract_last_page(first_page_payload)
|
||
for page in range(2, last_page + 1):
|
||
payload = self._fetch_page_payload(category_id, page=page)
|
||
offers.extend(self._parse_payload(payload, category_id))
|
||
|
||
return offers
|
||
|
||
def _fetch_page_payload(self, category_id: int, page: int) -> dict[str, Any]:
|
||
response = self._session.post(
|
||
self._url,
|
||
params={"fc": category_id, "page": page},
|
||
timeout=self._timeout_seconds,
|
||
)
|
||
response.raise_for_status()
|
||
return response.json()
|
||
|
||
def _parse_payload(self, payload: dict[str, Any], category_id: int) -> list[Offer]:
|
||
items = payload.get("data", {}).get("pagination", {}).get("data", [])
|
||
parsed_offers: list[Offer] = []
|
||
|
||
for item in items:
|
||
offer_id = item.get("id")
|
||
if offer_id is None:
|
||
continue
|
||
|
||
parsed_offers.append(
|
||
Offer(
|
||
source=self.source_name,
|
||
external_id=str(offer_id),
|
||
title=_decode_html_entities(item.get("name") or "Без названия"),
|
||
description=_decode_html_entities(item.get("description") or ""),
|
||
price=_format_price(item.get("priceLimit"), item.get("possiblePriceLimit")),
|
||
price_amount=_extract_price_amount(item.get("priceLimit"), item.get("possiblePriceLimit")),
|
||
url=f"https://kwork.ru/projects/{offer_id}/view",
|
||
category_id=category_id,
|
||
published_at=str(item.get("wantDates", {}).get("dateCreate") or ""),
|
||
seller_username=_extract_username(item.get("user") or {}),
|
||
)
|
||
)
|
||
|
||
return parsed_offers
|
||
|
||
|
||
def _format_price(price_limit: Any, possible_price_limit: Any) -> str:
|
||
if price_limit:
|
||
return f"{price_limit} RUB"
|
||
if possible_price_limit:
|
||
return f"до {possible_price_limit} RUB"
|
||
return "Не указан"
|
||
|
||
|
||
def _extract_price_amount(price_limit: Any, possible_price_limit: Any) -> int | None:
|
||
if price_limit:
|
||
return int(float(price_limit))
|
||
if possible_price_limit:
|
||
return int(float(possible_price_limit))
|
||
return None
|
||
|
||
|
||
def _extract_username(user_data: dict[str, Any]) -> str | None:
|
||
username = user_data.get("username")
|
||
if not username:
|
||
return None
|
||
return str(username)
|
||
|
||
|
||
def _extract_last_page(payload: dict[str, Any]) -> int:
|
||
last_page = payload.get("data", {}).get("pagination", {}).get("last_page")
|
||
if last_page is None:
|
||
return 1
|
||
return max(1, int(last_page))
|
||
|
||
|
||
def _decode_html_entities(value: Any) -> str:
|
||
return html.unescape(str(value))
|