forked from zovos/bot_tg
319 lines
11 KiB
Python
319 lines
11 KiB
Python
import asyncio
|
||
import json
|
||
import logging
|
||
import os
|
||
import random
|
||
from pathlib import Path, PurePosixPath
|
||
from urllib.parse import urlparse
|
||
|
||
import aiohttp
|
||
from aiogram import types
|
||
|
||
import config
|
||
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
CAPTIONS = [
|
||
"Шо ты блять псюнчик решил свой подергать?",
|
||
"Решил подрочить сука?",
|
||
"гнойный писюн у тебя кнч, да",
|
||
]
|
||
|
||
LESAINT_POSTS_API = "https://lesaintdesseins.fr/wp-json/wp/v2/posts"
|
||
LESAINT_REFERER = "https://lesaintdesseins.fr/"
|
||
REQUEST_HEADERS = {
|
||
"User-Agent": (
|
||
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
|
||
"AppleWebKit/537.36 (KHTML, like Gecko) "
|
||
"Chrome/124.0.0.0 Safari/537.36"
|
||
),
|
||
"Accept": "application/json,text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
|
||
"Accept-Language": "en-US,en;q=0.9",
|
||
"Referer": LESAINT_REFERER,
|
||
}
|
||
|
||
_NUDE_HISTORY_LOCK = asyncio.Lock()
|
||
|
||
|
||
def _resolve_int_setting(name: str, default: int) -> int:
|
||
value = getattr(config, name, None)
|
||
if value is None:
|
||
value = os.getenv(name, str(default))
|
||
|
||
try:
|
||
return int(value)
|
||
except (TypeError, ValueError):
|
||
logger.warning("Invalid %s=%r, using default %s", name, value, default)
|
||
return default
|
||
|
||
|
||
def _resolve_path_setting(name: str, default: str) -> Path:
|
||
value = getattr(config, name, None)
|
||
if value is None:
|
||
value = os.getenv(name, default)
|
||
|
||
if isinstance(value, Path):
|
||
return value
|
||
|
||
return Path(str(value))
|
||
|
||
|
||
NUDE_HISTORY_PATH = _resolve_path_setting("NUDE_HISTORY_PATH", "/db/nude_history.json")
|
||
NUDE_POSTS_PER_PAGE = _resolve_int_setting("NUDE_POSTS_PER_PAGE", 100)
|
||
NUDE_POSTS_POOL_TARGET = _resolve_int_setting("NUDE_POSTS_POOL_TARGET", 1200)
|
||
NUDE_MAX_POSTS_TO_SCAN = _resolve_int_setting("NUDE_MAX_POSTS_TO_SCAN", 2000)
|
||
NUDE_MIN_UNSEEN_POOL = _resolve_int_setting("NUDE_MIN_UNSEEN_POOL", 100)
|
||
NUDE_HISTORY_MAX_SIZE = _resolve_int_setting("NUDE_HISTORY_MAX_SIZE", 5000)
|
||
|
||
|
||
def _extract_photo_candidates(posts: list[dict]) -> list[dict]:
|
||
candidates = []
|
||
seen_urls = set()
|
||
|
||
for post in posts:
|
||
embedded = post.get("_embedded") or {}
|
||
media_items = embedded.get("wp:featuredmedia") or []
|
||
|
||
for media in media_items:
|
||
if media.get("media_type") != "image":
|
||
continue
|
||
|
||
media_details = media.get("media_details") or {}
|
||
sizes = media_details.get("sizes") or {}
|
||
image_url = (
|
||
(sizes.get("full") or {}).get("source_url")
|
||
or media.get("source_url")
|
||
)
|
||
|
||
if not image_url or image_url in seen_urls:
|
||
continue
|
||
|
||
seen_urls.add(image_url)
|
||
candidates.append(
|
||
{
|
||
"image_url": image_url,
|
||
"post_url": post.get("link", LESAINT_REFERER),
|
||
"title": (post.get("title") or {}).get("rendered", "").strip(),
|
||
}
|
||
)
|
||
break
|
||
|
||
return candidates
|
||
|
||
|
||
def _guess_filename(image_url: str) -> str:
|
||
parsed = urlparse(image_url)
|
||
filename = PurePosixPath(parsed.path).name or "nude.jpg"
|
||
|
||
if "." not in filename:
|
||
filename = f"{filename}.jpg"
|
||
|
||
return filename
|
||
|
||
|
||
def _load_nude_history() -> dict:
|
||
path = NUDE_HISTORY_PATH
|
||
if not path.exists():
|
||
return {"sent_images": []}
|
||
|
||
try:
|
||
with path.open("r", encoding="utf-8") as fh:
|
||
data = json.load(fh)
|
||
except (OSError, json.JSONDecodeError):
|
||
logger.exception("Failed to load nude history")
|
||
return {"sent_images": []}
|
||
|
||
if not isinstance(data, dict):
|
||
return {"sent_images": []}
|
||
|
||
sent_images = data.get("sent_images")
|
||
if not isinstance(sent_images, list):
|
||
sent_images = []
|
||
|
||
cleaned = [item for item in sent_images if isinstance(item, str) and item.strip()]
|
||
return {"sent_images": cleaned[-NUDE_HISTORY_MAX_SIZE :]}
|
||
|
||
|
||
def _save_nude_history(data: dict) -> None:
|
||
path = NUDE_HISTORY_PATH
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
tmp_path = path.with_name(f"{path.name}.tmp")
|
||
with tmp_path.open("w", encoding="utf-8") as fh:
|
||
json.dump(data, fh, ensure_ascii=False, indent=2)
|
||
tmp_path.replace(path)
|
||
|
||
|
||
def _remember_sent_image(history: dict, image_url: str) -> None:
|
||
sent_images = history.get("sent_images")
|
||
if not isinstance(sent_images, list):
|
||
sent_images = []
|
||
|
||
sent_images = [item for item in sent_images if item != image_url]
|
||
sent_images.append(image_url)
|
||
history["sent_images"] = sent_images[-NUDE_HISTORY_MAX_SIZE :]
|
||
|
||
|
||
async def _fetch_posts_page(
|
||
session: aiohttp.ClientSession, page: int, per_page: int
|
||
) -> tuple[list[dict], int | None]:
|
||
params = {
|
||
"page": str(page),
|
||
"per_page": str(per_page),
|
||
"orderby": "date",
|
||
"order": "desc",
|
||
"_embed": "wp:featuredmedia",
|
||
}
|
||
|
||
async with session.get(LESAINT_POSTS_API, params=params, headers=REQUEST_HEADERS) as resp:
|
||
if resp.status == 400 and page > 1:
|
||
body = await resp.text()
|
||
if "rest_post_invalid_page_number" in body:
|
||
return [], None
|
||
raise RuntimeError(f"API returned 400: {body[:300]}")
|
||
|
||
if resp.status != 200:
|
||
body = await resp.text()
|
||
raise RuntimeError(f"API returned {resp.status}: {body[:300]}")
|
||
|
||
total_pages_header = resp.headers.get("X-WP-TotalPages")
|
||
total_pages = (
|
||
int(total_pages_header)
|
||
if total_pages_header and total_pages_header.isdigit()
|
||
else None
|
||
)
|
||
posts = await resp.json(content_type=None)
|
||
|
||
if not isinstance(posts, list):
|
||
raise RuntimeError("Unexpected API response format")
|
||
|
||
return posts, total_pages
|
||
|
||
|
||
async def _fetch_latest_photo_candidates(
|
||
session: aiohttp.ClientSession, sent_urls: set[str]
|
||
) -> list[dict]:
|
||
per_page = max(1, min(NUDE_POSTS_PER_PAGE, 100))
|
||
max_scan = max(NUDE_POSTS_POOL_TARGET, NUDE_MAX_POSTS_TO_SCAN)
|
||
max_pages = max(1, (max_scan + per_page - 1) // per_page)
|
||
|
||
candidates = []
|
||
seen_urls = set()
|
||
unseen_count = 0
|
||
total_pages = None
|
||
|
||
for page in range(1, max_pages + 1):
|
||
if total_pages is not None and page > total_pages:
|
||
break
|
||
|
||
posts, reported_total_pages = await _fetch_posts_page(session, page, per_page)
|
||
if reported_total_pages is not None:
|
||
total_pages = reported_total_pages
|
||
if not posts:
|
||
break
|
||
|
||
for candidate in _extract_photo_candidates(posts):
|
||
image_url = candidate["image_url"]
|
||
if image_url in seen_urls:
|
||
continue
|
||
|
||
seen_urls.add(image_url)
|
||
candidates.append(candidate)
|
||
|
||
if image_url not in sent_urls:
|
||
unseen_count += 1
|
||
|
||
if len(candidates) >= NUDE_POSTS_POOL_TARGET and unseen_count >= NUDE_MIN_UNSEEN_POOL:
|
||
break
|
||
if len(candidates) >= max_scan:
|
||
break
|
||
|
||
return candidates
|
||
|
||
|
||
def _pick_unsent_candidate(candidates: list[dict], sent_urls: set[str]) -> dict | None:
|
||
unseen = [candidate for candidate in candidates if candidate["image_url"] not in sent_urls]
|
||
if not unseen:
|
||
return None
|
||
return random.choice(unseen)
|
||
|
||
|
||
async def _download_photo(
|
||
session: aiohttp.ClientSession, image_url: str
|
||
) -> tuple[bytes | None, str]:
|
||
async with session.get(image_url, headers=REQUEST_HEADERS) as resp:
|
||
if resp.status != 200:
|
||
logger.warning("Image download failed for %s: %s", image_url, resp.status)
|
||
return None, _guess_filename(image_url)
|
||
|
||
content_type = resp.headers.get("Content-Type", "")
|
||
if "image" not in content_type.lower():
|
||
logger.warning(
|
||
"Unexpected content type for %s: %s", image_url, content_type
|
||
)
|
||
return None, _guess_filename(image_url)
|
||
|
||
return await resp.read(), _guess_filename(image_url)
|
||
|
||
|
||
async def handle_nude_cmd(message: types.Message):
|
||
args = message.text.split(maxsplit=1)
|
||
|
||
if len(args) > 1 and args[1].strip() == "-h":
|
||
help_msg = (
|
||
"🔥 <b>Справка по команде /nude:</b>\n\n"
|
||
"Берет случайную фотку из последних постов сайта le saint des seins.\n"
|
||
"• Источник: <code>https://lesaintdesseins.fr/</code>\n"
|
||
"• Пагинацией тянет 1000+ последних постов через WordPress REST API\n"
|
||
"• Хранит историю отправок и старается не повторять фотки\n"
|
||
"• Выбирает случайную картинку и отправляет ее прямо в Telegram\n\n"
|
||
"<i>Пример:</i> <code>/nude</code>"
|
||
)
|
||
await message.reply(help_msg, parse_mode="HTML")
|
||
return
|
||
|
||
timeout = aiohttp.ClientTimeout(total=60)
|
||
history_snapshot = _load_nude_history()
|
||
sent_urls_snapshot = set(history_snapshot["sent_images"])
|
||
|
||
try:
|
||
async with aiohttp.ClientSession(timeout=timeout) as session:
|
||
candidates = await _fetch_latest_photo_candidates(session, sent_urls_snapshot)
|
||
|
||
if not candidates:
|
||
await message.reply("Не удалось найти свежие фотки :(")
|
||
return
|
||
|
||
async with _NUDE_HISTORY_LOCK:
|
||
history = _load_nude_history()
|
||
sent_urls = set(history["sent_images"])
|
||
selected = _pick_unsent_candidate(candidates, sent_urls)
|
||
|
||
if selected is None:
|
||
logger.info("Nude history exhausted current pool, resetting history")
|
||
history = {"sent_images": []}
|
||
selected = random.choice(candidates)
|
||
|
||
_remember_sent_image(history, selected["image_url"])
|
||
_save_nude_history(history)
|
||
|
||
caption = random.choice(CAPTIONS)
|
||
image_bytes, filename = await _download_photo(session, selected["image_url"])
|
||
|
||
if image_bytes:
|
||
photo = types.BufferedInputFile(image_bytes, filename=filename)
|
||
await message.answer_photo(photo=photo, caption=caption, has_spoiler=True)
|
||
return
|
||
|
||
try:
|
||
photo = types.URLInputFile(selected["image_url"])
|
||
await message.answer_photo(photo=photo, caption=caption, has_spoiler=True)
|
||
except Exception:
|
||
logger.exception("Failed to send photo by URL")
|
||
await message.answer(
|
||
f"{caption}\n{selected['image_url']}\n{selected['post_url']}"
|
||
)
|
||
|
||
except Exception:
|
||
logger.exception("Error in nude command")
|
||
await message.reply("Произошла ошибка при получении фото :(")
|