238 lines
9.6 KiB
Python
238 lines
9.6 KiB
Python
"""Reads F95zone metadata with an optional personal session cookie."""
|
||
import re
|
||
import urllib.error
|
||
import urllib.request
|
||
from html import unescape
|
||
from html.parser import HTMLParser
|
||
from urllib.parse import urljoin, urlparse
|
||
|
||
THREAD_RE = re.compile(r"^/threads/[^/]+\.\d+/?|^/threads/\d+/?")
|
||
UA = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124 Safari/537.36"
|
||
MAX_BYTES = 3 * 1024 * 1024
|
||
MAX_IMAGE_BYTES = 20 * 1024 * 1024
|
||
# first-post fields that are not part of the description itself
|
||
STOP_LABELS = ("thread updated", "release date", "developer", "censored", "version", "os",
|
||
"language", "store", "genre", "installation", "changelog", "developer notes",
|
||
"download", "animations")
|
||
|
||
|
||
def _request(url: str, referer: str | None = None, cookie: str = "", user_agent: str = "") -> urllib.request.Request:
|
||
headers = {"User-Agent": user_agent or UA, "Accept-Language": "en-US,en;q=0.8"}
|
||
if referer:
|
||
headers["Referer"] = referer
|
||
request = urllib.request.Request(url, headers=headers)
|
||
cookie = cookie.strip()
|
||
if "\r" in cookie or "\n" in cookie:
|
||
raise F95Error("Некорректные cookie F95zone")
|
||
if cookie:
|
||
# Redirect handlers explicitly reapply the cookie after validating the next host.
|
||
request.add_unredirected_header("Cookie", cookie)
|
||
return request
|
||
|
||
|
||
def image_url(value: str) -> str | None:
|
||
"""Only download images hosted by F95; never follow arbitrary post links."""
|
||
try:
|
||
url = urljoin("https://f95zone.to/", value.strip())
|
||
p = urlparse(url)
|
||
port = p.port
|
||
except ValueError:
|
||
return None
|
||
host = (p.hostname or "").lower()
|
||
if (value.strip() and p.scheme in ("http", "https")
|
||
and (host == "f95zone.to" or host.endswith(".f95zone.to"))
|
||
and not p.username and not p.password and port in (None, 80, 443)
|
||
and not p.path.startswith("/styles/") and "logo" not in p.path.lower()):
|
||
return p._replace(scheme="https", fragment="").geturl()
|
||
return None
|
||
|
||
|
||
class _F95Redirect(urllib.request.HTTPRedirectHandler):
|
||
images = False
|
||
|
||
def redirect_request(self, req, fp, code, msg, headers, newurl):
|
||
target = image_url(newurl)
|
||
if not target or (not self.images and urlparse(target).hostname not in ("f95zone.to", "www.f95zone.to")):
|
||
raise F95Error("Неожиданный редирект F95zone")
|
||
redirected = super().redirect_request(req, fp, code, msg, headers, target)
|
||
if redirected is not None:
|
||
cookie = req.get_header("Cookie", "")
|
||
if cookie:
|
||
redirected.add_unredirected_header("Cookie", cookie)
|
||
return redirected
|
||
|
||
|
||
class _ImageRedirect(_F95Redirect):
|
||
images = True
|
||
|
||
|
||
def fetch_image(url: str, thread_url: str, cookie: str = "", user_agent: str = "") -> bytes:
|
||
target = image_url(url)
|
||
if not target:
|
||
raise F95Error("Недопустимая ссылка на превью F95zone")
|
||
request = _request(target, thread_url, cookie, user_agent)
|
||
try:
|
||
with urllib.request.build_opener(_ImageRedirect()).open(request, timeout=15) as response:
|
||
if response.headers.get_content_type() == "text/html":
|
||
raise F95AuthError("F95zone запрашивает авторизацию для превью")
|
||
data = response.read(MAX_IMAGE_BYTES + 1)
|
||
if len(data) > MAX_IMAGE_BYTES:
|
||
raise F95Error("Превью F95zone слишком большое")
|
||
return data
|
||
except urllib.error.HTTPError as e:
|
||
if e.code in (401, 403):
|
||
raise F95AuthError("F95zone не принял cookie или ограничил доступ к превью") from e
|
||
raise F95Error("Не удалось скачать превью F95zone") from e
|
||
except OSError as e:
|
||
raise F95Error("Не удалось скачать превью F95zone") from e
|
||
|
||
|
||
class F95Error(Exception):
|
||
pass
|
||
|
||
|
||
class F95AuthError(F95Error):
|
||
pass
|
||
|
||
|
||
def validate_url(url: str) -> str:
|
||
p = urlparse(url.strip())
|
||
if p.scheme not in ("http", "https") or (p.hostname or "").lower() not in ("f95zone.to", "www.f95zone.to"):
|
||
raise F95Error("Ссылка должна вести на тему f95zone.to")
|
||
if not THREAD_RE.match(p.path):
|
||
raise F95Error("Это не ссылка на тему (нужен вид f95zone.to/threads/…)")
|
||
return f"https://f95zone.to{p.path}"
|
||
|
||
|
||
def fetch(url: str, cookie: str = "", user_agent: str = "") -> str:
|
||
req = _request(url, cookie=cookie, user_agent=user_agent)
|
||
try:
|
||
with urllib.request.build_opener(_F95Redirect()).open(req, timeout=15) as r:
|
||
final = urlparse(r.geturl())
|
||
if (final.hostname or "") not in ("f95zone.to", "www.f95zone.to"):
|
||
raise F95Error("Неожиданный редирект")
|
||
return r.read(MAX_BYTES).decode("utf-8", "replace")
|
||
except F95Error:
|
||
raise
|
||
except urllib.error.HTTPError as e:
|
||
if e.code in (401, 403):
|
||
raise F95AuthError("F95zone не принял cookie или ограничил доступ к теме") from e
|
||
if e.code >= 500:
|
||
raise F95Error(f"F95zone сейчас недоступен (ответил {e.code}), попробуйте позже")
|
||
raise F95Error(f"F95zone ответил {e.code} (тема недоступна или сайт блокирует запрос)")
|
||
except OSError:
|
||
raise F95Error("Не удалось связаться с F95zone")
|
||
|
||
|
||
class _Parser(HTMLParser):
|
||
"""Collects <title>-less data: og:title, tag items and the text of the first .bbWrapper."""
|
||
|
||
def __init__(self) -> None:
|
||
super().__init__(convert_charrefs=True)
|
||
self.og_title = ""
|
||
self.og_image = None
|
||
self.post_image = None
|
||
self.tags: list[str] = []
|
||
self.post: list[str] = []
|
||
self.template = ""
|
||
self.logged_in = None
|
||
self._in_tag = False
|
||
self._tag_buf = ""
|
||
self._depth = 0 # div depth inside the first bbWrapper; 0 = outside
|
||
self._done = False
|
||
self._skip = 0 # inside hidden/spoiler blocks
|
||
|
||
def handle_starttag(self, tag, attrs):
|
||
a = dict(attrs)
|
||
cls = (a.get("class") or "").split()
|
||
if tag == "html":
|
||
self.template = a.get("data-template") or ""
|
||
self.logged_in = a.get("data-logged-in")
|
||
elif tag == "meta" and a.get("property") == "og:title":
|
||
self.og_title = a.get("content") or ""
|
||
elif tag == "meta" and a.get("property") == "og:image":
|
||
self.og_image = image_url(a.get("content") or "")
|
||
elif tag == "a" and "tagItem" in cls:
|
||
self._in_tag, self._tag_buf = True, ""
|
||
if self._done:
|
||
return
|
||
if self._depth == 0:
|
||
if tag == "div" and "bbWrapper" in cls and not self.post:
|
||
self._depth = 1
|
||
return
|
||
if tag == "div":
|
||
self._depth += 1
|
||
if self._skip or {"bbCodeSpoiler", "messageHide"} & set(cls):
|
||
self._skip += 1
|
||
elif tag == "img" and "bbImage" in cls and not self._skip and not self.post_image:
|
||
for key in ("data-src", "src"):
|
||
if candidate := image_url(a.get(key) or ""):
|
||
self.post_image = candidate
|
||
break
|
||
elif tag == "br" and not self._skip:
|
||
self.post.append("\n")
|
||
elif tag in ("script", "style"):
|
||
self._skip += 1
|
||
|
||
def handle_endtag(self, tag):
|
||
if tag == "a" and self._in_tag:
|
||
self._in_tag = False
|
||
if t := self._tag_buf.strip().lower():
|
||
self.tags.append(t)
|
||
if self._depth and tag == "div":
|
||
if self._skip:
|
||
self._skip -= 1
|
||
self._depth -= 1
|
||
if self._depth == 0:
|
||
self._done = True
|
||
elif self._depth and tag in ("script", "style") and self._skip:
|
||
self._skip -= 1
|
||
|
||
def handle_data(self, data):
|
||
if self._in_tag:
|
||
self._tag_buf += data
|
||
if self._depth and not self._skip and not self._done:
|
||
self.post.append(data)
|
||
|
||
|
||
def parse(html: str) -> dict:
|
||
p = _Parser()
|
||
p.feed(html)
|
||
if p.template == "login":
|
||
raise F95AuthError("F95zone запрашивает авторизацию")
|
||
if not p.post:
|
||
raise F95Error("Тема не найдена или недоступна")
|
||
title = unescape(p.og_title)
|
||
# "VN - Ren'Py - Hooked! [v0.6.0] [BlenderKnight]" -> "Hooked!"
|
||
title = re.sub(r"\s*\[[^\]]*\]", "", title)
|
||
title = re.sub(r"^(?:[A-Za-z0-9'’ ]{1,20} - )+", "", title).strip()
|
||
return {"title": title, "description": _description("".join(p.post)),
|
||
"tags": sorted(set(p.tags)), "image_url": p.og_image or p.post_image,
|
||
"logged_in": p.logged_in}
|
||
|
||
|
||
def _description(text: str) -> str:
|
||
text = text.replace("", "").replace("\xa0", " ")
|
||
lines = [ln.strip() for ln in text.splitlines()]
|
||
out: list[str] = []
|
||
started = False
|
||
for ln in lines:
|
||
label = ln.split(":", 1)[0].strip().lower() if ":" in ln else ""
|
||
if label in STOP_LABELS and started:
|
||
break
|
||
if label == "overview":
|
||
started = True
|
||
ln = ln.split(":", 1)[1].strip()
|
||
if not ln:
|
||
continue
|
||
elif not started and label in STOP_LABELS:
|
||
continue
|
||
started = started or bool(ln)
|
||
out.append(ln)
|
||
return re.sub(r"\n{3,}", "\n\n", "\n".join(out)).strip()
|
||
|
||
|
||
def load(url: str, cookie: str = "", user_agent: str = "") -> dict:
|
||
url = validate_url(url)
|
||
return {**parse(fetch(url, cookie, user_agent)), "url": url}
|