faynot/qwork
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add get_project_details, scane more pages, scan pages delay; docs: create main.py docs

Browse Source
This commit is contained in:
Faynot
2026-03-28 14:38:05 +03:00
parent 4eeb622e37
commit b52990acc6
2 changed files with 367 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files

174
DOCS.md Normal file
View File

@@ -0,0 +1,174 @@
## Типы данных
```
Project = dict[str, str]
```
## Функции
### get_kwork_projects(max_pages: int = 1) → list[Project]
Собирает проекты со списка страниц категории.
**Назначение**: Автоматический обход страниц проектов с извлечением данных из карточек.
**Вход**:
- `max_page`: `int` - количество страниц для обработки (по умолчанию 1)
**Выход**:
```
[
{
"title": String,
"price": String,
"url": String,
"description": String
}
]
```
---
### get_project_details(url: str) → Optional[dict]
Парсит детальную страницу одного проекта.
**Назначение**: Извлечение полной информации о проекте по прямой ссылке.
**Вход**:
- `url`: `str` - полная ссылка на страницу проекта
**Выход**:
```
{
"url": String,
"title": String,
"description": String,
"budget": {
"desired": String,
"maximum": String
},
"buyer": {
"name": String,
"total_projects": String,
"hired_percent": String
},
"stats": {
"offers": String,
"time_left": String
}
}
```
или None при ошибке
---
### normalize_text(text: str) → str
Очищает текст от множественных пробелов и переносов.
**Выход**: `String`
---
### first_line(text: str) → str
Возвращает первую строку текста.
**Выход**: `String`
---
### normalize_url(href: str) → str
Преобразует относительную ссылку в абсолютную для **Kwork.ru**.
**Выход**: `String`
---
### safe_inner_text(locator: Locator, default: str = "") → str
Безопасно извлекает текст элемента с таймаутом.
**Выход**: `String`
---
### first_text(root: Locator, selectors: list[str], default: str = "") → str
Пробует селекторы по очереди, возвращает текст первого найденного.
**Выход**: `String`
---
### get_card_root(page: Page, href: str) → Locator
Находит корневой контейнер карточки по ссылке проекта.
**Выход**: `Locator`
---
### extract_price(card: Locator) → str
Извлекает информацию о бюджете из карточки.
**Выход**: `String`
**Пример:**
```
("до 5000 ₽ | Допустимый: до 10000 ₽" или "По договоренности")
```
---
### extract_description(card: Locator) → str
Извлекает краткое описание проекта из карточки.
**Выход**: `String`
---
### extract_kwork_project(page: Page, title_block: Locator) → Optional[Project]
Экстрактор данных одной карточки по блоку заголовка.
**Выход**: `Project` или `None`
---
### clean(text: str) → str
Очищает текст от неразрывных пробелов и лишних символов.
**Выход**: `String`
---
### get_text(page: Page, selector: str) → str
Извлекает текст по CSS-селектору.
**Выход**: `String`
## Использование
```python
import asyncio
from main import get_kwork_projects, get_project_details
# Список проектов
projects = asyncio.run(get_kwork_projects(max_pages=3))
# Детали проекта
details = asyncio.run(get_project_details("https://kwork.ru/projects/123"))
```

241
main.py
View File

@@ -1,11 +1,13 @@
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"
@@ -16,15 +18,21 @@ USER_AGENT = (
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 ""
@@ -32,10 +40,15 @@ def first_line(text: str) -> str:
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", " ")
@@ -45,6 +58,10 @@ async def safe_inner_text(locator: Locator, default: str = "") -> str:
async def first_text(root: Locator, selectors: list[str], default: str = "") -> str:
"""
Пробует по очереди список CSS-селекторов внутри корневого элемента.
Возвращает очищенный текст первого найденного элемента.
"""
for selector in selectors:
try:
loc = root.locator(selector).first
@@ -58,6 +75,10 @@ async def first_text(root: Locator, selectors: list[str], default: str = "") ->
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]'
)
@@ -76,6 +97,10 @@ async def get_card_root(page: Page, href: str) -> Locator:
async def extract_price(card: Locator) -> str:
"""
Извлекает стоимость проекта из карточки.
Сначала ищет текст по ключевым фразам (бюджет), затем по CSS-классам цен.
"""
card_text = await safe_inner_text(card, "")
card_text = normalize_text(card_text)
@@ -124,6 +149,10 @@ async def extract_price(card: Locator) -> str:
async def extract_description(card: Locator) -> str:
"""
Извлекает описание проекта из карточки, удаляя лишние элементы управления
('Показать полностью', 'Скрыть').
"""
description = await first_text(
card,
[
@@ -143,46 +172,11 @@ async def extract_description(card: Locator) -> str:
return first_line(description)
async def scrape_items(
*,
url: str,
item_selector: str,
extractor: Extractor,
wait_until: str = "networkidle",
render_delay: float = 3.0,
) -> list[Project]:
async with Stealth().use_async(async_playwright()) as p:
browser = await p.chromium.launch(headless=True)
try:
context = await browser.new_context(
user_agent=USER_AGENT,
viewport=VIEWPORT,
)
page = await context.new_page()
await page.goto(url, wait_until=wait_until)
await asyncio.sleep(render_delay)
result: list[Project] = []
items = await page.locator(item_selector).all()
for item in items:
try:
data = await extractor(page, item)
if data is not None:
result.append(data)
except Exception:
continue
return result
except Exception as e:
print(f"Ошибка: {e}")
return []
finally:
await browser.close()
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
@@ -205,16 +199,167 @@ async def extract_kwork_project(page: Page, title_block: Locator) -> Optional[Pr
}
async def get_kwork_projects() -> list[Project]:
print("Загружаем проекты...")
# --- Основная логика скрапинга ---
return await scrape_items(
url=PROJECTS_URL,
item_selector=".wants-card__header-title",
extractor=extract_kwork_project,
)
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__":
data = asyncio.run(get_kwork_projects())
print(json.dumps(data, ensure_ascii=False, indent=4))
#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))