From 2a01c66422314c7db0b297d829aca87ab0a01ffb Mon Sep 17 00:00:00 2001 From: XC-Pro <19840+xc-pro@noreply.example.org> Date: Fri, 14 Aug 2026 13:39:58 +0200 Subject: [PATCH] =?UTF-8?q?=E4=B8=8A=E4=BC=A0=E6=96=87=E4=BB=B6=E8=87=B3?= =?UTF-8?q?=E3=80=8Cxbpq=E3=80=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- xbpq/土豆.py | 224 +++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 224 insertions(+) create mode 100644 xbpq/土豆.py diff --git a/xbpq/土豆.py b/xbpq/土豆.py new file mode 100644 index 0000000..ddcd74a --- /dev/null +++ b/xbpq/土豆.py @@ -0,0 +1,224 @@ +# -*- coding: utf-8 -*- +import base64, html, json, re +from urllib.parse import quote, unquote +import requests +from lxml import etree +try: + from Crypto.Cipher import AES +except Exception: + AES = None +try: + from base.spider import Spider as BaseSpider +except Exception: + BaseSpider = object + +class Spider(BaseSpider): + def getName(self): return "黑料不打烊" + def init(self, extend=""): + self.host = "https://badly.okttbipbu.cc" + self.headers = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0 Safari/537.36", "Referer": self.host + "/", "Accept": "*/*", "Accept-Language": "zh-CN,zh;q=0.9"} + self.session = requests.Session() + self.session.headers.update(self.headers) + self.last_error = "" + self.categories = [{"type_id": "/", "type_name": "首页"}, {"type_id": "/category/24hcg/", "type_name": "今日看料"}, {"type_id": "/category/rgtj/", "type_name": "热门吃瓜"}, {"type_id": "/category/mrrg/", "type_name": "每日热瓜"}, {"type_id": "/category/hlda/", "type_name": "黑料大事"}, {"type_id": "/category/whhl/", "type_name": "网红吃瓜"}, {"type_id": "/category/mxbg/", "type_name": "明星吃瓜"}, {"type_id": "/category/fcns/", "type_name": "反差女神"}, {"type_id": "/category/xyrg/", "type_name": "学院热瓜"}, {"type_id": "/category/mrds/", "type_name": "每日大赛"}, {"type_id": "/category/swdj/", "type_name": "AI短剧"}, {"type_id": "/category/lydt/", "type_name": "撸友看片"}, {"type_id": "/category/avjs/", "type_name": "AV解说"}, {"type_id": "/category/mrst/", "type_name": "禁播动漫"}, {"type_id": "/category/pmv/", "type_name": "PMV混剪"}] + def _client(self): return getattr(self, "session", requests) + def _error(self, msg): + self.last_error = msg + try: self.log(msg) + except Exception: pass + def _get(self, url): + try: + client = self._client() + r = client.get(url, headers=self.headers, timeout=25, verify=False, allow_redirects=True) + if (r.status_code != 200 or not (r.text or "").strip()) and url.rstrip("/") != self.host: + client.get(self.host + "/", headers=self.headers, timeout=25, verify=False, allow_redirects=True) + r = client.get(url, headers=self.headers, timeout=25, verify=False, allow_redirects=True) + if r.status_code != 200: + self._error(f"GET {url} status={r.status_code} body={(r.text or '')[:120]}") + return "" + text = r.content.decode("utf-8", errors="ignore") + if not (text or "").strip(): self._error(f"GET {url} empty body") + return text + except Exception as e: + self._error(f"GET {url} error={type(e).__name__}: {e}") + return "" + def _fix(self, u): + if not u: return "" + u = html.unescape(u.strip()) + return "https:" + u if u.startswith("//") else self.host + u if u.startswith("/") else u + def _clean(self, s): return re.sub(r"\s+", " ", html.unescape(s or "")).strip() + def _player_header(self): return self.headers + def _html(self, text): + if isinstance(text, bytes): text = text.decode("utf-8", errors="ignore") + return etree.HTML((text or "").replace("\x00", "").encode("utf-8", errors="ignore")) + def _ids(self, ids): return ids if isinstance(ids, (list, tuple)) else [ids] if ids else [] + def _detail_url(self, sid): + sid = str(sid or "").strip() + if sid.startswith("http"): return self._fix(sid) + m = re.search(r"(?:archives/)?(\d{3,})(?:\.html)?", sid) + if m: return f"{self.host}/archives/{m.group(1)}.html" + return self._fix(sid) + def _is_encrypted_img(self, u): return any(x in (u or "") for x in ["/xiao/", "/upload_01/xiao/", "/upload/upload/xiao/"]) + def _is_cdn_img(self, u): return any(x in (u or "") for x in ["/xiao/", "/usr/", "/upload_01/", "/uploads/", "/upload/upload/"]) + def _proxy_base(self): + try: + f = getattr(self, "getProxyUrl", None) + return f() if callable(f) else "" + except Exception: + return "" + def _proxy_img_url(self, base, url): + sep = "" if base.endswith(("?", "&")) else "&" if "?" in base else "?" + return base + sep + "type=img&url=" + quote(url, safe="") + def _mime_from_bytes(self, data, ext="jpeg"): + if data.startswith(b"\xff\xd8\xff"): return "jpeg" + if data.startswith(b"\x89PNG\r\n\x1a\n"): return "png" + if data.startswith(b"GIF87a") or data.startswith(b"GIF89a"): return "gif" + if data[:4] == b"RIFF" and data[8:12] == b"WEBP": return "webp" + return "jpeg" if ext in ["jpg", "jpeg"] else ext if ext in ["png", "gif", "webp"] else "jpeg" + def _decrypt_image_bytes(self, data): + if not AES: return "" + try: + raw = AES.new(b"f5d965df75336270", AES.MODE_CBC, b"97b60394abc2fbe1").decrypt(base64.b64decode(data)) + pad = raw[-1] if raw else 0 + return raw[:-pad] if 0 < pad <= 16 and raw.endswith(bytes([pad]) * pad) else raw + except Exception: + return b"" + def _image_data_url(self, url): + try: + data, mime = self._image_bytes(url) + return f"data:image/{mime};base64,{base64.b64encode(data).decode()}" if data else url + except Exception: + return url + def _image_bytes(self, url): + url = unquote(url) + ext = (url.split("?")[0].rsplit(".", 1)[-1] or "jpeg").lower() + r = self._client().get(url, headers=self.headers, timeout=25, verify=False, allow_redirects=True) + if self._is_encrypted_img(url): + raw = self._decrypt_image_bytes(base64.b64encode(r.content).decode()) + if raw: return raw, self._mime_from_bytes(raw, ext) + ctype = (r.headers.get("Content-Type") or "").split(";")[0].lower() + mime = ctype.split("/", 1)[1] if ctype.startswith("image/") else self._mime_from_bytes(r.content, ext) + return r.content, mime + def _pic_url(self, url): + base = self._proxy_base() + if base: return self._proxy_img_url(base, url) + if url.startswith("data:"): return url + return self._image_data_url(url) if self._is_encrypted_img(url) else url + def _cover_url(self, url): + url = self._fix(url) + base = self._proxy_base() + return self._proxy_img_url(base, url) if base and self._is_encrypted_img(url) else url + def proxy(self, params): + url = params.get("url") or params.get("img") or "" + if isinstance(url, list): url = url[0] if url else "" + try: + data, mime = self._image_bytes(url) + return [200, "image/" + mime, data] + except Exception: + return [404, "text/plain", ""] + def localProxy(self, params): return self.proxy(params) + def _page_url(self, tid, pg): + pg = int(pg or 1) + path = tid or "/" + if path.startswith("http"): base = path.rstrip("/") + "/" + else: base = self.host + (path if path.startswith("/") else "/" + path) + base = base.rstrip("/") + "/" + if pg <= 1: return base + return base + f"page/{pg}/" if path.strip("/") == "" else base + f"{pg}/" + def _parse_list(self, html_text): + if not html_text: return [] + tree = self._html(html_text) + if tree is None: return [] + result, seen = [], set() + items = tree.xpath('//div[@id="index" or @id="archive"]//article[.//a[contains(@href,"/archives/")]]') or tree.xpath('//article[.//a[contains(@href,"/archives/")]]') + for item in items: + if self._is_ad_item(item): continue + href = "".join(item.xpath('.//a[contains(@href,"/archives/")]/@href')).strip() + m = re.search(r"/archives/(\d+)\.html", href) + if not m or m.group(1) in seen: continue + seen.add(m.group(1)) + title = self._clean(" ".join(item.xpath('.//*[contains(@class,"post-card-bottom-text")]//text()')) or " ".join(item.xpath('.//h2//text()')) or "".join(item.xpath('.//a[contains(@href,"/archives/")]/@title'))) + pics = item.xpath('.//meta[@itemprop="image" or @itemprop="thumbnailUrl"]/@content') or item.xpath('.//img/@z-image-loader-url') or item.xpath('.//img/@data-xkrkllgl') or item.xpath('.//img/@data-src') or item.xpath('.//img/@src') + pic = pics[0] if pics else "" + result.append({"vod_id": self._fix(href), "vod_name": title or m.group(1), "vod_pic": self._cover_url(pic)}) + return result + def _content_node(self, tree): + nodes = tree.xpath('//div[contains(concat(" ",normalize-space(@class)," ")," post-content ")]') + return nodes[0] if nodes else tree + def _is_ad_item(self, item): + return bool(item.xpath('.//a[contains(concat(" ",normalize-space(@rel)," ")," sponsored ") or @data-event="ad_click" or @data-ad_slot_key or @data-ad_id]')) + def _parse_videos(self, node): + videos, seen = [], set() + for d in node.xpath('.//div[contains(@class,"dplayer") and @data-config]'): + title = f"视频{len(videos)+1}" + conf = html.unescape(d.get("data-config") or "") + url = "" + try: url = ((json.loads(conf).get("video") or {}).get("url") or "").replace("\\/", "/") + except Exception: + m = re.search(r'"video"\s*:\s*\{.*?"url"\s*:\s*"([^"]+)"', conf) + url = m.group(1).replace("\\/", "/") if m else "" + if url and url not in seen: + seen.add(url) + videos.append((title, self._fix(url))) + return videos + def _parse_images(self, node): + imgs, seen = [], set() + for img in node.xpath('.//img'): + src = img.get("z-image-loader-url") or img.get("data-xkrkllgl") or img.get("data-original") or img.get("data-src") or img.get("data-lazyload") or img.get("data-lazy-src") or img.get("src") or "" + src = self._fix(src) + if not src or src in seen or "/usr/themes/" in src or "/usr/plugins/" in src or "/uploads/default/other/" in src: continue + seen.add(src) + imgs.append(src) + return imgs + def homeContent(self, filter): + return {"class": self.categories, "list": self._parse_list(self._get(self.host + "/")), "filters": {}} + def categoryContent(self, tid, pg, filter, extend): + items = self._parse_list(self._get(self._page_url(tid, pg))) + total = 9990 if items else 0 + return {"page": int(pg), "pagecount": 999 if items else int(pg), "limit": 10, "total": total, "count": total, "list": items} + def detailContent(self, ids): + self.last_error = "" + result = {"list": []} + for sid in self._ids(ids): + try: + url = self._detail_url(sid) + html_text = self._get(url) + if not html_text: + if not self.last_error: self._error(f"detail {url} empty html") + continue + tree = self._html(html_text) + if tree is None: + self._error(f"detail {url} etree none body={html_text[:120]}") + continue + node = self._content_node(tree) + name = self._clean("".join(tree.xpath('//meta[@property="og:title"]/@content')) or "".join(tree.xpath('//h1/text()')) or "".join(tree.xpath('//title/text()'))) + videos, imgs = self._parse_videos(node), self._parse_images(node) + pic = self._cover_url(imgs[0] if imgs else "".join(tree.xpath('//meta[@property="og:image"]/@content'))) + text = self._clean(" ".join(node.xpath('.//p[not(ancestor::*[contains(@class,"article-ads-btn")])]/text()')))[:800] + names, plays = [], [] + if videos: + names.append("视频") + plays.append("#".join(f"{t}${u}" for t, u in videos)) + if imgs: + names.append("图集") + plays.append("全部$pics://" + "&&".join(imgs)) + if text: + names.append("正文") + plays.append("正文$text://" + text) + result["list"].append({"vod_id": url, "vod_name": name, "vod_pic": pic, "vod_content": text, "vod_play_from": "$$$".join(names), "vod_play_url": "$$$".join(plays)}) + except Exception as e: + self._error(f"detail sid={sid} error={type(e).__name__}: {e}") + continue + return result if result["list"] else {"list": [], "msg": self.last_error or "detail empty"} + def searchContent(self, key, quick, pg="1"): + url = self.host + "/search/" + quote(key) + "/" + (f"page/{pg}/" if int(pg or 1) > 1 else "") + return {"list": self._parse_list(self._get(url)), "page": int(pg)} + def playerContent(self, flag, id, vipFlags): + if id.startswith("pics://"): + return {"parse": 0, "url": "pics://" + "&&".join(self._pic_url(u) for u in id[7:].split("&&") if u), "header": self._player_header()} + if id.startswith("text://"): return {"parse": 0, "url": id, "header": self._player_header()} + url = self._fix(id) + lower = url.lower() + result = {"parse": 0 if any(x in lower for x in [".m3u8", ".mp4", ".flv"]) else 1, "url": url, "header": self._player_header()} + if ".m3u8" in lower: result["format"] = "application/x-mpegURL" + return result