diff --git a/games/nude.py b/games/nude.py
index 0e0a0fd..4c37767 100644
--- a/games/nude.py
+++ b/games/nude.py
@@ -1,3 +1,5 @@
+import asyncio
+import json
import logging
import random
from pathlib import PurePosixPath
@@ -6,6 +8,8 @@ from urllib.parse import urlparse
import aiohttp
from aiogram import types
+import config
+
logger = logging.getLogger(__name__)
@@ -17,7 +21,6 @@ CAPTIONS = [
LESAINT_POSTS_API = "https://lesaintdesseins.fr/wp-json/wp/v2/posts"
LESAINT_REFERER = "https://lesaintdesseins.fr/"
-LATEST_POSTS_LIMIT = 30
REQUEST_HEADERS = {
"User-Agent": (
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
@@ -29,9 +32,12 @@ REQUEST_HEADERS = {
"Referer": LESAINT_REFERER,
}
+_NUDE_HISTORY_LOCK = asyncio.Lock()
+
def _extract_photo_candidates(posts: list[dict]) -> list[dict]:
candidates = []
+ seen_urls = set()
for post in posts:
embedded = post.get("_embedded") or {}
@@ -48,9 +54,10 @@ def _extract_photo_candidates(posts: list[dict]) -> list[dict]:
or media.get("source_url")
)
- if not image_url:
+ if not image_url or image_url in seen_urls:
continue
+ seen_urls.add(image_url)
candidates.append(
{
"image_url": image_url,
@@ -73,25 +80,130 @@ def _guess_filename(image_url: str) -> str:
return filename
-async def _fetch_latest_photo_candidates(session: aiohttp.ClientSession) -> list[dict]:
+def _load_nude_history() -> dict:
+ path = config.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[-config.NUDE_HISTORY_MAX_SIZE :]}
+
+
+def _save_nude_history(data: dict) -> None:
+ path = config.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[-config.NUDE_HISTORY_MAX_SIZE :]
+
+
+async def _fetch_posts_page(
+ session: aiohttp.ClientSession, page: int, per_page: int
+) -> tuple[list[dict], int | None]:
params = {
- "per_page": str(LATEST_POSTS_LIMIT),
+ "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 _extract_photo_candidates(posts)
+ return posts, total_pages
+
+
+async def _fetch_latest_photo_candidates(
+ session: aiohttp.ClientSession, sent_urls: set[str]
+) -> list[dict]:
+ per_page = max(1, min(config.NUDE_POSTS_PER_PAGE, 100))
+ max_scan = max(config.NUDE_POSTS_POOL_TARGET, config.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) >= config.NUDE_POSTS_POOL_TARGET and unseen_count >= config.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(
@@ -120,24 +232,39 @@ async def handle_nude_cmd(message: types.Message):
"🔥 Справка по команде /nude:\n\n"
"Берет случайную фотку из последних постов сайта le saint des seins.\n"
"• Источник: https://lesaintdesseins.fr/\n"
- "• Достает последние посты через WordPress REST API\n"
+ "• Пагинацией тянет 1000+ последних постов через WordPress REST API\n"
+ "• Хранит историю отправок и старается не повторять фотки\n"
"• Выбирает случайную картинку и отправляет ее прямо в Telegram\n\n"
"Пример: /nude"
)
await message.reply(help_msg, parse_mode="HTML")
return
- timeout = aiohttp.ClientTimeout(total=30)
+ 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)
+ candidates = await _fetch_latest_photo_candidates(session, sent_urls_snapshot)
if not candidates:
await message.reply("Не удалось найти свежие фотки :(")
return
- selected = random.choice(candidates)
+ 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"])