feat: bot start, database
This commit is contained in:
Faynot
365
kwork.py
Normal file
365
kwork.py
Normal file
@@ -0,0 +1,365 @@
|
||||
import asyncio
|
||||
import json
|
||||
import re
|
||||
import random
|
||||
from typing import Awaitable, Callable, Optional
|
||||
|
||||
from playwright.async_api import Locator, Page, async_playwright
|
||||
from playwright_stealth import Stealth
|
||||
|
||||
# --- Константы ---
|
||||
BASE_URL = "https://kwork.ru"
|
||||
PROJECTS_URL = f"{BASE_URL}/projects?c=11"
|
||||
|
||||
USER_AGENT = (
|
||||
"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
|
||||
"(KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36"
|
||||
)
|
||||
|
||||
VIEWPORT = {"width": 1920, "height": 1080}
|
||||
|
||||
# Задержка между запросами к страницам (в секундах), чтобы избежать блокировок
|
||||
PAGE_LOAD_DELAY = (3, 6)
|
||||
|
||||
Project = dict[str, str]
|
||||
Extractor = Callable[[Page, Locator], Awaitable[Optional[Project]]]
|
||||
|
||||
# --- Вспомогательные функции ---
|
||||
|
||||
def normalize_text(text: str) -> str:
|
||||
"""Очищает текст от лишних пробельных символов и переносов строк."""
|
||||
return re.sub(r"\s+", " ", text).strip()
|
||||
|
||||
|
||||
def first_line(text: str) -> str:
|
||||
"""Извлекает только первую строку из переданного текста."""
|
||||
text = text.strip()
|
||||
if not text:
|
||||
return ""
|
||||
return text.splitlines()[0].strip()
|
||||
|
||||
|
||||
def normalize_url(href: str) -> str:
|
||||
"""Превращает относительную ссылку (напр. /projects/1) в полную (https://kwork.ru/...)."""
|
||||
return f"{BASE_URL}{href}" if href.startswith("/") else href
|
||||
|
||||
|
||||
async def safe_inner_text(locator: Locator, default: str = "") -> str:
|
||||
"""
|
||||
Безопасно извлекает внутренний текст элемента.
|
||||
Если элемент не найден или произошла ошибка, возвращает значение по умолчанию.
|
||||
"""
|
||||
try:
|
||||
text = await locator.inner_text(timeout=1500)
|
||||
text = text.replace("\xa0", " ")
|
||||
return text.strip()
|
||||
except Exception:
|
||||
return default
|
||||
|
||||
|
||||
async def first_text(root: Locator, selectors: list[str], default: str = "") -> str:
|
||||
"""
|
||||
Пробует по очереди список CSS-селекторов внутри корневого элемента.
|
||||
Возвращает очищенный текст первого найденного элемента.
|
||||
"""
|
||||
for selector in selectors:
|
||||
try:
|
||||
loc = root.locator(selector).first
|
||||
text = await safe_inner_text(loc, "")
|
||||
text = normalize_text(text)
|
||||
if text:
|
||||
return text
|
||||
except Exception:
|
||||
continue
|
||||
return default
|
||||
|
||||
|
||||
async def get_card_root(page: Page, href: str) -> Locator:
|
||||
"""
|
||||
Находит родительский контейнер (карточку проекта) на основе ссылки на проект.
|
||||
Использует поиск по XPath 'ancestor', чтобы подняться вверх по DOM-дереву.
|
||||
"""
|
||||
card = page.locator(
|
||||
f'xpath=//a[@href="{href}"]/ancestor::div[contains(@class, "wants-card__top")][1]'
|
||||
)
|
||||
|
||||
if await card.count() > 0:
|
||||
return card.first
|
||||
|
||||
card = page.locator(
|
||||
f'xpath=//a[@href="{href}"]/ancestor::div[contains(@class, "wants-card")][1]'
|
||||
)
|
||||
|
||||
if await card.count() > 0:
|
||||
return card.first
|
||||
|
||||
return page.locator(f'xpath=//a[@href="{href}"]/ancestor::div[1]').first
|
||||
|
||||
|
||||
async def extract_price(card: Locator) -> str:
|
||||
"""
|
||||
Извлекает стоимость проекта из карточки.
|
||||
Сначала ищет текст по ключевым фразам (бюджет), затем по CSS-классам цен.
|
||||
"""
|
||||
card_text = await safe_inner_text(card, "")
|
||||
card_text = normalize_text(card_text)
|
||||
|
||||
patterns = [
|
||||
r"(Желаемый бюджет:\s*до\s*[\d\s]+₽)",
|
||||
r"(Цена до:\s*[\d\s]+₽)",
|
||||
r"(Допустимый:\s*до\s*[\d\s]+₽)",
|
||||
]
|
||||
|
||||
found: list[str] = []
|
||||
for pattern in patterns:
|
||||
match = re.search(pattern, card_text, flags=re.IGNORECASE)
|
||||
if match:
|
||||
value = normalize_text(match.group(1))
|
||||
if value and value not in found:
|
||||
found.append(value)
|
||||
|
||||
if found:
|
||||
return " | ".join(found)
|
||||
|
||||
primary = await first_text(
|
||||
card,
|
||||
[
|
||||
".wants-card__price",
|
||||
".wants-card__header-right-block .wants-card__price",
|
||||
"[class*='wants-card__price']",
|
||||
"[class*='price']",
|
||||
],
|
||||
"",
|
||||
)
|
||||
|
||||
higher = await first_text(
|
||||
card,
|
||||
[
|
||||
".wants-card__description-higher-price",
|
||||
"[class*='description-higher-price']",
|
||||
],
|
||||
"",
|
||||
)
|
||||
|
||||
parts = [part for part in [primary, higher] if part]
|
||||
if parts:
|
||||
return " | ".join(parts)
|
||||
|
||||
return "По договоренности"
|
||||
|
||||
|
||||
async def extract_description(card: Locator) -> str:
|
||||
"""
|
||||
Извлекает описание проекта из карточки, удаляя лишние элементы управления
|
||||
('Показать полностью', 'Скрыть').
|
||||
"""
|
||||
description = await first_text(
|
||||
card,
|
||||
[
|
||||
".wants-card__description-text .overflow-hidden .d-inline",
|
||||
".wants-card__description-text .overflow-hidden",
|
||||
".wants-card__description-text",
|
||||
"[class*='description-text']",
|
||||
],
|
||||
"",
|
||||
)
|
||||
|
||||
description = description.replace("Показать полностью", "")
|
||||
description = description.replace("Скрыть", "")
|
||||
description = description.replace("\xa0", " ")
|
||||
description = normalize_text(description)
|
||||
|
||||
return first_line(description)
|
||||
|
||||
|
||||
async def extract_kwork_project(page: Page, title_block: Locator) -> Optional[Project]:
|
||||
"""
|
||||
Функция-экстрактор: собирает все поля (заголовок, цена, ссылка, описание)
|
||||
для одного конкретного блока заголовка.
|
||||
"""
|
||||
link = title_block.locator("a").first
|
||||
if await link.count() == 0:
|
||||
return None
|
||||
|
||||
title_text = normalize_text(await safe_inner_text(link, ""))
|
||||
href = await link.get_attribute("href")
|
||||
if not href:
|
||||
return None
|
||||
|
||||
card = await get_card_root(page, href)
|
||||
|
||||
price = await extract_price(card)
|
||||
description = await extract_description(card)
|
||||
|
||||
return {
|
||||
"title": title_text,
|
||||
"price": price,
|
||||
"url": normalize_url(href),
|
||||
"description": description,
|
||||
}
|
||||
|
||||
|
||||
# --- Основная логика скрапинга ---
|
||||
|
||||
async def get_kwork_projects(max_pages: int = 1) -> list[Project]:
|
||||
"""
|
||||
Основная функция запуска браузера и обхода страниц.
|
||||
:param max_pages: Сколько страниц нужно просмотреть.
|
||||
"""
|
||||
all_results: list[Project] = []
|
||||
|
||||
async with Stealth().use_async(async_playwright()) as p:
|
||||
browser = await p.chromium.launch(headless=True)
|
||||
context = await browser.new_context(
|
||||
user_agent=USER_AGENT,
|
||||
viewport=VIEWPORT,
|
||||
)
|
||||
page = await context.new_page()
|
||||
|
||||
try:
|
||||
for current_page in range(1, max_pages + 1):
|
||||
# Формируем URL с учетом номера страницы
|
||||
url = f"{PROJECTS_URL}&page={current_page}"
|
||||
print(f"Загрузка страницы {current_page}: {url}")
|
||||
|
||||
await page.goto(url, wait_until="networkidle")
|
||||
|
||||
# Дополнительная задержка для рендеринга JS элементов
|
||||
await asyncio.sleep(2)
|
||||
|
||||
items = await page.locator(".wants-card__header-title").all()
|
||||
print(f"Найдено проектов на странице: {len(items)}")
|
||||
|
||||
for item in items:
|
||||
try:
|
||||
data = await extract_kwork_project(page, item)
|
||||
if data:
|
||||
all_results.append(data)
|
||||
except Exception as e:
|
||||
print(f"Ошибка при парсинге карточки: {e}")
|
||||
continue
|
||||
|
||||
# Ограничение частоты запросов (Rate Limiting)
|
||||
if current_page < max_pages:
|
||||
delay = random.uniform(*PAGE_LOAD_DELAY)
|
||||
print(f"Ожидаем {delay:.2f} сек. перед следующей страницей...")
|
||||
await asyncio.sleep(delay)
|
||||
|
||||
return all_results
|
||||
|
||||
except Exception as e:
|
||||
print(f"Произошла критическая ошибка: {e}")
|
||||
return all_results
|
||||
finally:
|
||||
await browser.close()
|
||||
|
||||
|
||||
def clean(text: str) -> str:
|
||||
"""Очищает текст от мусора, неразрывных пробелов и лишних пустот."""
|
||||
if not text: return ""
|
||||
text = text.replace("\xa0", " ")
|
||||
return re.sub(r"\s+", " ", text).strip()
|
||||
|
||||
async def get_text(page: Page, selector: str) -> str:
|
||||
"""Извлекает текст из элемента, если он существует."""
|
||||
try:
|
||||
element = page.locator(selector).first
|
||||
if await element.count() > 0:
|
||||
return await element.inner_text(timeout=3000)
|
||||
return ""
|
||||
except:
|
||||
return ""
|
||||
|
||||
|
||||
async def get_project_details(url: str) -> Optional[dict]:
|
||||
"""
|
||||
Открывает страницу проекта, парсит HTML и возвращает структурированные данные.
|
||||
"""
|
||||
async with Stealth().use_async(async_playwright()) as p:
|
||||
# Запускаем браузер
|
||||
browser = await p.chromium.launch(headless=True)
|
||||
context = await browser.new_context(user_agent=USER_AGENT)
|
||||
page = await context.new_page()
|
||||
|
||||
try:
|
||||
print(f"Парсим проект: {url}")
|
||||
await page.goto(url, wait_until="networkidle")
|
||||
# Ждем немного, чтобы JS отработал до конца
|
||||
await asyncio.sleep(2)
|
||||
|
||||
# 1. Извлекаем заголовок
|
||||
title = await get_text(page, "h1.wants-card__header-title")
|
||||
|
||||
# 2. Извлекаем описание (сохраняем структуру)
|
||||
description_raw = await get_text(page, ".wants-card__description-text")
|
||||
|
||||
# 3. Бюджет (Желаемый и Допустимый)
|
||||
# Извлекаем только цифры через регулярку
|
||||
price_desired_raw = await get_text(page, ".wants-card__price")
|
||||
price_max_raw = await get_text(page, ".wants-card__description-higher-price")
|
||||
|
||||
# 4. Информация о заказчике
|
||||
buyer_block = await get_text(page, ".want-payer-statistic")
|
||||
buyer_name = await get_text(page, ".want-payer-statistic a")
|
||||
|
||||
# 5. Статистика проекта (предложения и время)
|
||||
informers_block = await get_text(page, ".want-card__informers")
|
||||
|
||||
# --- Парсинг данных через регулярные выражения для точности ---
|
||||
|
||||
# Чистим цену: оставляем только цифры
|
||||
def extract_digits(text):
|
||||
digits = "".join(re.findall(r'\d', text))
|
||||
return f"{digits} ₽" if digits else "По договоренности"
|
||||
|
||||
# Вырезаем статистику из текста блоков
|
||||
total_projects = re.search(r"Размещено проектов на бирже: (\d+)", buyer_block)
|
||||
hired_percent = re.search(r"Нанято: (\d+%)", buyer_block)
|
||||
offers_count = re.search(r"Предложений:\s*(\d+)", informers_block)
|
||||
time_left = re.search(r"Осталось:\s*(.*?)(?:\n|$)", informers_block)
|
||||
|
||||
# Формируем итоговый объект
|
||||
data = {
|
||||
"url": url,
|
||||
"title": clean(title),
|
||||
"description": description_raw.strip(),
|
||||
"budget": {
|
||||
"desired": extract_digits(price_desired_raw),
|
||||
"maximum": extract_digits(price_max_raw)
|
||||
},
|
||||
"buyer": {
|
||||
"name": clean(buyer_name),
|
||||
"total_projects": total_projects.group(1) if total_projects else "0",
|
||||
"hired_percent": hired_percent.group(1) if hired_percent else "н/д"
|
||||
},
|
||||
"stats": {
|
||||
"offers": offers_count.group(1) if offers_count else "0",
|
||||
"time_left": clean(time_left.group(1)) if time_left else "н/д"
|
||||
}
|
||||
}
|
||||
|
||||
return data
|
||||
|
||||
except Exception as e:
|
||||
print(f"Ошибка при парсинге {url}: {e}")
|
||||
return None
|
||||
finally:
|
||||
await browser.close()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
#pages_to_scan = 2
|
||||
|
||||
#data = asyncio.run(get_kwork_projects(max_pages=pages_to_scan))
|
||||
|
||||
target_url = "https://kwork.ru/projects/3134701"
|
||||
|
||||
# Запуск парсера для одной ссылки
|
||||
result = asyncio.run(get_project_details(target_url))
|
||||
|
||||
if result:
|
||||
print("\n--- Данные проекта в формате JSON ---")
|
||||
print(json.dumps(result, ensure_ascii=False, indent=4))
|
||||
|
||||
#print(f"\nВсего собрано проектов: {len(data)}")
|
||||
#print(json.dumps(data, ensure_ascii=False, indent=4))
|
||||
Reference in New Issue
Block a user