https://pastein.ru/t/ctM
Загрузка данных
"""
services/nemezida.py — поиск в базе данных Немезида (nemez1da.ru).
Источник: https://nemez1da.ru — сайт группы RaHDit, содержащий персональные данные
украинских военных, сотрудников спецслужб и силовых структур.
Доступен только с российских IP (vpn_location: "Moscow").
Алгоритм:
1. Переключение VPN на российскую локацию (если vpn_location задан)
2. Поиск через WordPress-поиск /?s=NAME (первая страница, до 20 результатов)
3. Парсинг rel="bookmark" ссылок в зоне posts-wrapper
4. Дополнительная загрузка деталей для совпавших записей (до MAX_DETAIL)
5. Опциональная фильтрация по дате рождения
Поля карточки:
full_name — ФИО (русское + украинское название из заголовка)
category — категория (ГУР, СБУ, СВР и т.д.)
birth_date — дата рождения (DD.MM.YYYY)
position — должность
address — адрес регистрации
drfo — ДРФО (украинский налоговый номер)
passport — номер паспорта
phone — телефон
email — электронная почта
vk — ВКонтакте
activity — деятельность/подразделение
extra — дополнительно
photo_url — прямая ссылка на фото
profile_url — ссылка на страницу профиля
"""
from __future__ import annotations
import asyncio
import logging
import re
import unicodedata
from typing import Any
from urllib.parse import quote
import aiohttp
logger = logging.getLogger(__name__)
_BASE_URL = "https://nemez1da.ru"
_TIMEOUT = aiohttp.ClientTimeout(total=20)
_MAX_DETAIL = 5 # максимум загрузок детальных страниц за один поиск
_HEADERS = {
"User-Agent": (
"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
"(KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
),
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
"Accept-Language": "ru-RU,ru;q=0.9,en;q=0.8",
}
# Перевод URL-слагов категорий в читаемые названия
_CATEGORY_SLUGS: dict[str, str] = {
"gur": "ГУР",
"sotrudniki-sbu": "СБУ",
"svr-ukrainy": "СВР Украины",
"inostrannye-naemniki": "Иностранные наёмники",
"inostrannye-posobniki-i-naemniki": "Иностранные пособники и наёмники",
"vsu": "ВСУ",
"uchastniki-ato": "Участники АТО",
"stratkom": "Страткомом ВСУ",
"czipso": "ЦИПСО",
"voennoplennye": "Военнопленные",
"bez-sroka-davnosti": "Без срока давности",
"batalon-revansh": "Батальон Реванш",
}
# Маппинг заголовков полей на ключи результата
_FIELD_MAP: dict[str, str] = {
"Дата рождения": "birth_date",
"Категория": "category",
"Должность": "position",
"Зарегистрирован по адресу": "address",
"ДРФО": "drfo",
"Паспорт": "passport",
"Телефон": "phone",
"Почта": "email",
"Вконтакте": "vk",
"Деятельность": "activity",
"Дополнительно": "extra",
}
# ── Вспомогательные функции ────────────────────────────────────────────────────
def _normalize(text: str) -> set[str]:
"""Возвращает множество слов в верхнем регистре для сравнения имён."""
text = unicodedata.normalize("NFC", text)
return set(re.sub(r"[^А-ЯЁA-Z0-9\s]", "", text.upper()).split())
def _parse_search_results(html: str) -> list[dict]:
"""
Извлекает список записей из страницы поиска WordPress.
Ищет ссылки с rel="bookmark" в зоне posts-wrapper.
Каждый результат: {title, url, category_slug, category}.
"""
start = html.find("simple-grid-posts-wrapper")
end = html.find("simple-grid-featured-posts-area")
zone = html[start:end] if 0 < start < end else html
entries: list[dict] = []
for m in re.finditer(
r'href="(https://nemez1da\.ru/([^"]+))"[^>]*rel="bookmark">([^<]+)</a>',
zone,
):
url, path, title = m.group(1), m.group(2), m.group(3).strip()
if not title:
continue
parts = path.rstrip("/").split("/")
slug = parts[-2] if len(parts) >= 2 else ""
entries.append({
"title": title,
"url": url,
"category_slug": slug,
"category": _CATEGORY_SLUGS.get(slug, slug),
})
return entries
def _parse_person_page(html: str, url: str) -> dict:
"""
Извлекает все поля из страницы профиля человека.
Поля: <div><b>FIELD_NAME</b> VALUE</div>
Фото: href='...' внутри div.photos_single_place
"""
person: dict[str, Any] = {"profile_url": url}
# Фото
photo_m = re.search(
r'class="photos_single_place".*?href=\'([^\']+)\'',
html, re.DOTALL,
)
if photo_m:
person["photo_url"] = photo_m.group(1)
# Зона entry-content
content_m = re.search(
r'<div[^>]*class="entry-content[^"]*"[^>]*>(.*?)<!-- \.entry-content -->',
html, re.DOTALL,
)
if not content_m:
return person
raw = content_m.group(1)
# Все пары <b>FIELD</b> VALUE
for fm in re.finditer(r"<b>([^<]+)</b>(.*?)(?=<b>|<!--)", raw, re.DOTALL):
field_name = fm.group(1).strip()
field_val = re.sub(r"<[^>]+>", " ", fm.group(2))
field_val = re.sub(r"\s+", " ", field_val).strip()
key = _FIELD_MAP.get(field_name)
if key and field_val:
person[key] = field_val
return person
def _match_date(bd_query: str | None, bd_found: str | None) -> bool:
"""
Проверяет совпадение дат рождения.
Принимает DD.MM.YYYY или YYYY-MM-DD в bd_query; bd_found — DD.MM.YYYY.
Возвращает True если совпадение есть или bd_query не задан.
"""
if not bd_query:
return True
if not bd_found:
return False
bd_q = bd_query.strip()
if re.match(r"^\d{4}-\d{2}-\d{2}", bd_q):
p = bd_q[:10].split("-")
bd_q = f"{p[2]}.{p[1]}.{p[0]}"
return bd_q in bd_found
# ── Основная функция поиска ────────────────────────────────────────────────────
async def search_nemezida(
name: str | None,
birth_date: str | None = None,
vpn_location: str | None = None,
) -> dict[str, Any]:
"""
Ищет человека на nemez1da.ru по ФИО и опциональной дате рождения.
Параметры
---------
name : ФИО для поиска (обязательно)
birth_date : дата рождения (DD.MM.YYYY или YYYY-MM-DD), опционально
vpn_location : локация VPN перед запросами (например "Moscow")
Возвращает
----------
{
"status" : "found" | "not_found" | "skipped" | "error",
"matched" : list[dict],
"total" : int, — кол-во совпадений имени на первой странице
"error" : str | None,
}
"""
if not name or not name.strip():
return {"status": "skipped", "matched": [], "total": 0, "error": None}
if vpn_location:
from services.vpn import ensure_location
ok = await ensure_location(vpn_location)
if not ok:
return {
"status": "error", "matched": [], "total": 0,
"error": f"Не удалось переключить VPN на {vpn_location}",
}
query_words = _normalize(name)
try:
async with aiohttp.ClientSession(headers=_HEADERS, timeout=_TIMEOUT) as session:
# ── Шаг 1: поиск ──────────────────────────────────────────────────
search_url = f"{_BASE_URL}/?s={quote(name.strip())}"
async with session.get(search_url) as resp:
resp.raise_for_status()
search_html = await resp.text()
entries = _parse_search_results(search_html)
logger.info("nemezida: поиск '%s' → %d результатов", name, len(entries))
# Все слова запроса должны присутствовать в заголовке записи
candidates = [
e for e in entries
if query_words.issubset(_normalize(e["title"]))
]
if not candidates:
return {"status": "not_found", "matched": [], "total": 0, "error": None}
# ── Шаг 2: загрузка деталей ───────────────────────────────────────
to_fetch = candidates[:_MAX_DETAIL]
tasks = [_fetch_person(session, e) for e in to_fetch]
results = await asyncio.gather(*tasks, return_exceptions=True)
matched: list[dict] = []
for i, result in enumerate(results):
entry = to_fetch[i]
if isinstance(result, Exception):
logger.warning("nemezida: ошибка загрузки %s: %s", entry["url"], result)
person: dict = {
"full_name": entry["title"],
"category": entry["category"],
"profile_url": entry["url"],
}
else:
person = {
"full_name": entry["title"],
# Категория из детальной страницы приоритетнее (полное название)
"category": result.get("category") or entry["category"],
**{k: v for k, v in result.items() if k != "category"},
}
if not _match_date(birth_date, person.get("birth_date")):
continue
matched.append(person)
return {
"status": "found" if matched else "not_found",
"matched": matched,
"total": len(candidates),
"error": None,
}
except Exception as exc:
logger.warning("nemezida: ошибка поиска '%s': %s", name, exc)
return {"status": "error", "matched": [], "total": 0, "error": str(exc)}
async def _fetch_person(session: aiohttp.ClientSession, entry: dict) -> dict:
"""Загружает и парсит детальную страницу профиля."""
async with session.get(entry["url"]) as resp:
resp.raise_for_status()
html = await resp.text()
return _parse_person_page(html, entry["url"])
