@@ -1,12 +1,8 @@
|
||||
import unittest
|
||||
from base64 import b64encode
|
||||
from importlib.machinery import SourceFileLoader
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
from Cryptodome.Cipher import AES
|
||||
from Cryptodome.Util.Padding import pad
|
||||
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
MODULE = SourceFileLoader("fktv_spider", str(ROOT / "凡客TV.py")).load_module()
|
||||
@@ -106,49 +102,6 @@ class TestFKTVSpider(unittest.TestCase):
|
||||
self.assertEqual(result["page"], 3)
|
||||
self.assertEqual(result["list"][0]["vod_id"], "xyz789")
|
||||
|
||||
@patch.object(Spider, "_request_html")
|
||||
def test_category_content_uses_img_src_when_data_src_missing(self, mock_request_html):
|
||||
mock_request_html.return_value = """
|
||||
<div class="card-wrap">
|
||||
<div class="meta-wrap">
|
||||
<a class="normal-title" href="/movie/detail/src001" title="封面走 src">封面走 src</a>
|
||||
<img class="lazy-load" src="/poster-src.jpg" />
|
||||
<span class="tag">电影</span>
|
||||
</div>
|
||||
</div>
|
||||
"""
|
||||
result = self.spider.categoryContent("1", "1", False, {})
|
||||
self.assertEqual(result["list"][0]["vod_pic"], "https://fktv.me/poster-src.jpg")
|
||||
|
||||
@patch.object(Spider, "fetch")
|
||||
@patch.object(Spider, "_request_html")
|
||||
def test_category_content_decodes_bnc_cover_to_data_url(self, mock_request_html, mock_fetch):
|
||||
mock_request_html.return_value = """
|
||||
<div class="card-wrap">
|
||||
<div class="meta-wrap">
|
||||
<a class="normal-title" href="/movie/detail/bnc001" title="加密封面">加密封面</a>
|
||||
<div class="lazy-load" data-src="https://img.example/cover.bnc"></div>
|
||||
<span class="tag">电影</span>
|
||||
</div>
|
||||
</div>
|
||||
"""
|
||||
png_bytes = (
|
||||
b"\x89PNG\r\n\x1a\n"
|
||||
b"\x00\x00\x00\rIHDR"
|
||||
b"\x00\x00\x00\x01\x00\x00\x00\x01"
|
||||
b"\x08\x06\x00\x00\x00"
|
||||
)
|
||||
key = b"525202f9149e061d"
|
||||
encrypted = AES.new(key, AES.MODE_ECB).encrypt(pad(png_bytes, AES.block_size))
|
||||
mock_fetch.return_value.status_code = 200
|
||||
mock_fetch.return_value.content = encrypted
|
||||
|
||||
result = self.spider.categoryContent("1", "1", False, {})
|
||||
|
||||
expected_prefix = "data:image/png;base64,"
|
||||
expected_body = b64encode(png_bytes).decode("ascii")
|
||||
self.assertEqual(result["list"][0]["vod_pic"], expected_prefix + expected_body)
|
||||
|
||||
@patch.object(Spider, "_request_html")
|
||||
def test_detail_content_extracts_state_and_builds_multiline_playlist(self, mock_request_html):
|
||||
mock_request_html.return_value = """
|
||||
|
||||
+2
-30
@@ -1,13 +1,9 @@
|
||||
# coding=utf-8
|
||||
import base64
|
||||
import json
|
||||
import re
|
||||
import sys
|
||||
from urllib.parse import quote, urljoin
|
||||
|
||||
from Cryptodome.Cipher import AES
|
||||
from Cryptodome.Util.Padding import unpad
|
||||
|
||||
from base.spider import Spider as BaseSpider
|
||||
|
||||
sys.path.append("..")
|
||||
@@ -23,8 +19,6 @@ class Spider(BaseSpider):
|
||||
"AppleWebKit/537.36 (KHTML, like Gecko) "
|
||||
"Chrome/146.0.0.0 Safari/537.36 Edg/146.0.0.0"
|
||||
)
|
||||
self._img_key = b"525202f9149e061d"
|
||||
self._img_cache = {}
|
||||
self.classes = [
|
||||
{"type_id": "1", "type_name": "电影"},
|
||||
{"type_id": "2", "type_name": "剧集"},
|
||||
@@ -61,26 +55,6 @@ class Spider(BaseSpider):
|
||||
return "https:" + raw
|
||||
return urljoin(self.host + "/", raw)
|
||||
|
||||
def _decode_bnc_cover(self, url):
|
||||
cover_url = self._abs_url(url)
|
||||
if not cover_url.endswith(".bnc"):
|
||||
return cover_url
|
||||
if cover_url in self._img_cache:
|
||||
return self._img_cache[cover_url]
|
||||
try:
|
||||
response = self.fetch(cover_url, headers=self._page_headers(self.host + "/"), timeout=10, verify=False)
|
||||
if response.status_code != 200:
|
||||
return cover_url
|
||||
encrypted = getattr(response, "content", b"") or b""
|
||||
if not encrypted:
|
||||
return cover_url
|
||||
plain = unpad(AES.new(self._img_key, AES.MODE_ECB).decrypt(encrypted), AES.block_size)
|
||||
decoded = "data:image/png;base64," + base64.b64encode(plain).decode("ascii")
|
||||
self._img_cache[cover_url] = decoded
|
||||
return decoded
|
||||
except Exception:
|
||||
return cover_url
|
||||
|
||||
def _page_headers(self, referer=""):
|
||||
headers = {"User-Agent": self.user_agent, "Referer": referer or self.host + "/"}
|
||||
if self.cookie:
|
||||
@@ -156,8 +130,6 @@ class Spider(BaseSpider):
|
||||
)
|
||||
)
|
||||
pic = self._clean_text("".join(node.xpath(".//*[contains(@class,'lazy-load')][1]/@data-src")))
|
||||
if not pic:
|
||||
pic = self._clean_text("".join(node.xpath(".//*[contains(@class,'lazy-load')][1]/@src")))
|
||||
tags = [
|
||||
self._clean_text("".join(tag.xpath(".//text()")))
|
||||
for tag in node.xpath(".//*[contains(@class,'tag')]")
|
||||
@@ -169,7 +141,7 @@ class Spider(BaseSpider):
|
||||
{
|
||||
"vod_id": vod_id,
|
||||
"vod_name": title,
|
||||
"vod_pic": self._decode_bnc_cover(pic),
|
||||
"vod_pic": self._abs_url(pic),
|
||||
"vod_remarks": " | ".join(tags),
|
||||
"type_name": tags[0] if tags else "",
|
||||
}
|
||||
@@ -306,7 +278,7 @@ class Spider(BaseSpider):
|
||||
{
|
||||
"vod_id": state["movieId"] or vod_id,
|
||||
"vod_name": title or vod_id,
|
||||
"vod_pic": self._decode_bnc_cover(pic),
|
||||
"vod_pic": self._abs_url(pic),
|
||||
"vod_content": content,
|
||||
"vod_remarks": " | ".join(remarks),
|
||||
"vod_play_from": "$$$".join(name for name, _ in groups),
|
||||
|
||||
Reference in New Issue
Block a user