180 lines
7.3 KiB
Python
180 lines
7.3 KiB
Python
"""
|
||
Парсинг публичных групп ВКонтакте без API-ключа.
|
||
Использует мобильную версию m.vk.com — простой HTML без JS-рендеринга.
|
||
"""
|
||
import asyncio
|
||
import re
|
||
from datetime import datetime, timezone, timedelta
|
||
|
||
import httpx
|
||
from bs4 import BeautifulSoup
|
||
|
||
# Имитируем мобильный браузер
|
||
_HEADERS = {
|
||
"User-Agent": (
|
||
"Mozilla/5.0 (Linux; Android 13; Pixel 7) "
|
||
"AppleWebKit/537.36 (KHTML, like Gecko) "
|
||
"Chrome/120.0.0.0 Mobile Safari/537.36"
|
||
),
|
||
"Accept-Language": "ru-RU,ru;q=0.9",
|
||
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
|
||
}
|
||
|
||
_BASE = "https://m.vk.com"
|
||
|
||
# Разбираем русские сокращения дат вида "14 мая в 10:35" или "вчера в 18:00"
|
||
_MONTHS = {
|
||
"янв": 1, "фев": 2, "мар": 3, "апр": 4, "май": 5, "маи": 5,
|
||
"июн": 6, "июл": 7, "авг": 8, "сен": 9, "окт": 10, "ноя": 11, "дек": 12,
|
||
}
|
||
|
||
|
||
def _parse_vk_date(text: str) -> datetime:
|
||
"""Конвертирует строку даты VK в datetime UTC."""
|
||
now = datetime.now(timezone.utc)
|
||
text = text.strip().lower()
|
||
|
||
if "сегодня" in text or "today" in text:
|
||
time_match = re.search(r"(\d{1,2}):(\d{2})", text)
|
||
if time_match:
|
||
return now.replace(hour=int(time_match[1]), minute=int(time_match[2]), second=0, microsecond=0)
|
||
return now
|
||
|
||
if "вчера" in text or "yesterday" in text:
|
||
yesterday = now - timedelta(days=1)
|
||
time_match = re.search(r"(\d{1,2}):(\d{2})", text)
|
||
if time_match:
|
||
return yesterday.replace(hour=int(time_match[1]), minute=int(time_match[2]), second=0, microsecond=0)
|
||
return yesterday
|
||
|
||
# "14 мая в 10:35" или "14 мая 2023 в 10:35"
|
||
match = re.search(r"(\d{1,2})\s+([а-яё]+)(?:\s+(\d{4}))?(?:\s+в\s+(\d{1,2}):(\d{2}))?", text)
|
||
if match:
|
||
day = int(match[1])
|
||
month = _MONTHS.get(match[2][:3], 1)
|
||
year = int(match[3]) if match[3] else now.year
|
||
hour = int(match[4]) if match[4] else 12
|
||
minute= int(match[5]) if match[5] else 0
|
||
try:
|
||
return datetime(year, month, day, hour, minute, tzinfo=timezone.utc)
|
||
except ValueError:
|
||
pass
|
||
|
||
return now
|
||
|
||
|
||
async def fetch_group_posts_html(group_id: str, count: int = 20) -> list[dict]:
|
||
"""
|
||
Скачивает и парсит посты публичной группы ВКонтакте.
|
||
group_id — числовой ID без минуса (например "12345").
|
||
Возвращает список словарей совместимых с форматом vk_parser.
|
||
"""
|
||
url = f"{_BASE}/wall-{group_id}"
|
||
posts = []
|
||
|
||
try:
|
||
async with httpx.AsyncClient(
|
||
headers=_HEADERS,
|
||
timeout=20,
|
||
follow_redirects=True,
|
||
) as client:
|
||
r = await client.get(url)
|
||
|
||
if r.status_code != 200:
|
||
print(f"[VK scraper] HTTP {r.status_code} для группы {group_id}")
|
||
return []
|
||
|
||
soup = BeautifulSoup(r.text, "lxml")
|
||
items = soup.select("div.wall_item, div._post, article.post")
|
||
|
||
if not items:
|
||
# Попробуем запасной селектор
|
||
items = soup.select("[id^='post-']")
|
||
|
||
for item in items[:count]:
|
||
post = _parse_post(item, group_id)
|
||
if post:
|
||
posts.append(post)
|
||
|
||
except Exception as exc:
|
||
print(f"[VK scraper] Ошибка группы {group_id}: {exc}")
|
||
|
||
return posts
|
||
|
||
|
||
def _parse_post(item, group_id: str) -> dict | None:
|
||
"""Парсит один пост из HTML-элемента."""
|
||
# ── ID поста ──────────────────────────────────────────────────────────────
|
||
post_id = None
|
||
elem_id = item.get("id", "")
|
||
id_match = re.search(r"(\d+)_(\d+)$", elem_id)
|
||
if id_match:
|
||
post_id = int(id_match[2])
|
||
else:
|
||
# Ищем ссылку на пост вида /wall-12345_67890
|
||
link = item.select_one("a[href*='wall-']")
|
||
if link:
|
||
lm = re.search(r"wall-?\d+_(\d+)", link["href"])
|
||
if lm:
|
||
post_id = int(lm[1])
|
||
|
||
if not post_id:
|
||
return None
|
||
|
||
owner_id = int(group_id)
|
||
|
||
# ── Текст ─────────────────────────────────────────────────────────────────
|
||
text_el = (
|
||
item.select_one(".pi_text")
|
||
or item.select_one(".wall_post_text")
|
||
or item.select_one("._post_content")
|
||
or item.select_one(".post_text")
|
||
)
|
||
text = text_el.get_text(separator="\n").strip() if text_el else ""
|
||
|
||
# ── Дата ──────────────────────────────────────────────────────────────────
|
||
date_el = item.select_one(".pi_date, time, .post_date, ._date")
|
||
ts = int(datetime.now(timezone.utc).timestamp())
|
||
if date_el:
|
||
dt = _parse_vk_date(date_el.get_text())
|
||
ts = int(dt.timestamp())
|
||
|
||
# ── Изображения ───────────────────────────────────────────────────────────
|
||
attachments = []
|
||
for img in item.select("img.pi_img, .wall_post_img img, img._photo_img"):
|
||
src = img.get("src") or img.get("data-src") or ""
|
||
if src and "userapi.com" in src or "vk.com" in src:
|
||
attachments.append({"type": "photo", "photo": {"sizes": [{"type": "x", "url": src}]}})
|
||
|
||
# ── Видео ─────────────────────────────────────────────────────────────────
|
||
for video_link in item.select("a[href*='/video']"):
|
||
href = video_link.get("href", "")
|
||
vid_match = re.search(r"/video(-?\d+)_(\d+)", href)
|
||
if vid_match:
|
||
oid = vid_match[1]
|
||
vid = vid_match[2]
|
||
player = f"https://vk.com/video_ext.php?oid={oid}&id={vid}"
|
||
title = video_link.get_text(strip=True) or "Видео"
|
||
attachments.append({"type": "video", "video": {"player": player, "title": title}})
|
||
|
||
# ── Ссылки ────────────────────────────────────────────────────────────────
|
||
for a in item.select("a.pi_snippet, a._snippet"):
|
||
href = a.get("href", "")
|
||
label = a.select_one(".pi_snippet_title, .snippet_title")
|
||
if href and not href.startswith("/"):
|
||
attachments.append({
|
||
"type": "link",
|
||
"link": {"url": href, "title": label.get_text(strip=True) if label else href},
|
||
})
|
||
|
||
if not text and not attachments:
|
||
return None
|
||
|
||
return {
|
||
"id": post_id,
|
||
"owner_id": -owner_id,
|
||
"date": ts,
|
||
"text": text,
|
||
"attachments": attachments,
|
||
}
|