diff --git a/py/tests/test_凡客TV.py b/py/tests/test_凡客TV.py index c3a6ce9..61bee37 100644 --- a/py/tests/test_凡客TV.py +++ b/py/tests/test_凡客TV.py @@ -156,6 +156,116 @@ class TestFKTVSpider(unittest.TestCase): self.assertIn("ep-9", vod["vod_play_url"]) self.assertIn("VIP", vod["vod_remarks"]) + def test_player_content_returns_direct_media_without_request(self): + result = self.spider.playerContent("fktv", "https://cdn.example.com/demo.m3u8", {}) + self.assertEqual(result["parse"], 0) + self.assertEqual(result["url"], "https://cdn.example.com/demo.m3u8") + + @patch.object(Spider, "_request_json") + @patch.object(Spider, "_request_html") + def test_player_content_uses_switch_api_and_filters_by_line_id( + self, mock_request_html, mock_request_json + ): + play_id = self.spider._encode_play_id( + { + "movie_id": "9001", + "link_id": "ep-2", + "line_id": "line-b", + "line_name": "线路B", + "episode_name": "第2集", + "type": "switch", + "page": "https://fktv.me/movie/detail/9001", + } + ) + mock_request_html.return_value = """ + + """ + mock_request_json.return_value = { + "status": True, + "data": { + "play_links": [ + {"id": "line-a", "name": "线路A", "m3u8_url": "/media/a.m3u8"}, + {"id": "line-b", "name": "线路B", "m3u8_url": "https://cdn.example.com/b.m3u8"}, + ] + }, + } + result = self.spider.playerContent("fktv", play_id, {}) + self.assertEqual(result["parse"], 0) + self.assertEqual(result["url"], "https://cdn.example.com/b.m3u8") + self.assertEqual(mock_request_json.call_args.args[0], "https://fktv.me/movie/detail/9001") + self.assertEqual( + mock_request_json.call_args.kwargs["data"], + {"link_id": "ep-2", "is_switch": 1}, + ) + + @patch.object(Spider, "_request_json") + @patch.object(Spider, "_request_html") + def test_player_content_returns_all_lines_when_line_not_specified( + self, mock_request_html, mock_request_json + ): + play_id = self.spider._encode_play_id( + { + "movie_id": "9003", + "link_id": "ep-8", + "line_id": "", + "line_name": "", + "episode_name": "正片", + "type": "switch", + "page": "https://fktv.me/movie/detail/9003", + } + ) + mock_request_html.return_value = ( + "" + ) + mock_request_json.return_value = { + "status": True, + "data": { + "play_links": [ + {"id": "line-a", "name": "线路A", "m3u8_url": "/media/a.m3u8"}, + {"id": "line-b", "name": "线路B", "preview_m3u8_url": "/media/b.m3u8"}, + ] + }, + } + result = self.spider.playerContent("fktv", play_id, {}) + self.assertEqual(result["parse"], 0) + self.assertEqual(len(result["urls"]), 2) + self.assertEqual(result["url"], "https://fktv.me/media/a.m3u8") + + @patch.object(Spider, "_request_json") + @patch.object(Spider, "_request_html") + def test_player_content_falls_back_to_detail_page_when_api_has_no_url( + self, mock_request_html, mock_request_json + ): + play_id = self.spider._encode_play_id( + { + "movie_id": "9004", + "link_id": "ep-3", + "line_id": "line-z", + "line_name": "线路Z", + "episode_name": "第3集", + "type": "switch", + "page": "https://fktv.me/movie/detail/9004", + } + ) + mock_request_html.return_value = ( + "" + ) + mock_request_json.return_value = { + "status": True, + "data": {"play_links": [], "play_error_type": "need_vip"}, + } + result = self.spider.playerContent("fktv", play_id, {}) + self.assertEqual(result["parse"], 1) + self.assertEqual(result["url"], "https://fktv.me/movie/detail/9004") + if __name__ == "__main__": unittest.main() diff --git a/py/凡客TV.py b/py/凡客TV.py index 6ca65fa..42ef623 100644 --- a/py/凡客TV.py +++ b/py/凡客TV.py @@ -67,6 +67,33 @@ class Spider(BaseSpider): return "" return str(response.text or "") + def _ajax_headers(self, referer): + headers = { + "User-Agent": self.user_agent, + "Referer": referer, + "Origin": self.host, + "X-Requested-With": "XMLHttpRequest", + "Content-Type": "application/x-www-form-urlencoded; charset=UTF-8", + } + if self.cookie: + headers["Cookie"] = self.cookie + return headers + + def _request_json(self, url, data=None, headers=None): + response = self.post( + url, + data=data, + headers=headers or self._ajax_headers(self.host + "/"), + timeout=10, + verify=False, + ) + if response.status_code != 200: + return {} + try: + return json.loads(str(response.text or "")) + except Exception: + return {} + def _extract_cards(self, html): root = self.html(html or "") if root is None: @@ -260,6 +287,83 @@ class Spider(BaseSpider): ] } + def _normalize_play_links(self, items): + results = [] + for item in items or []: + raw_url = str((item or {}).get("m3u8_url") or (item or {}).get("preview_m3u8_url") or "").strip() + if not raw_url: + continue + results.append( + { + "line_id": str((item or {}).get("id") or "").strip(), + "name": self._clean_text((item or {}).get("name")) or str((item or {}).get("id") or "线路"), + "url": urljoin(self.host + "/", raw_url), + } + ) + return results + + def _build_player_headers(self, referer, with_origin=False): + headers = {"User-Agent": self.user_agent, "Referer": referer} + if with_origin: + headers["Origin"] = self.host + return headers + + def _build_direct_result(self, entries, referer): + return { + "parse": 0, + "playUrl": "", + "url": entries[0]["url"] if entries else "", + "urls": [{"name": item["name"], "url": item["url"]} for item in entries], + "header": self._build_player_headers(referer, with_origin=True), + } + + def _build_parse_result(self, url): + return { + "parse": 1, + "playUrl": "", + "url": url, + "header": self._build_player_headers(url), + } + + def playerContent(self, flag, id, vipFlags): + raw = str(id or "").strip() + if not raw: + return {"parse": 0, "playUrl": "", "url": "", "header": {}} + if re.search(r"\.(m3u8|mp4|flv)(?:\?|#|$)", raw, re.I): + return { + "parse": 0, + "playUrl": "", + "url": raw, + "header": self._build_player_headers(self.host + "/", with_origin=True), + } + + meta = self._decode_play_id(raw) + page_url = str(meta.get("page") or "").strip() + if not page_url: + movie_id = str(meta.get("movie_id") or "").strip() + page_url = self.host + "/movie/detail/" + movie_id if movie_id else self.host + "/" + + html = self._request_html(page_url, self._page_headers(self.host + "/")) + state = self._extract_page_state(html) + movie_id = str(meta.get("movie_id") or state.get("movieId") or "").strip() + link_id = str(meta.get("link_id") or state.get("linkId") or "").strip() + line_id = str(meta.get("line_id") or "").strip() + + if movie_id and link_id: + payload = self._request_json( + self.host + "/movie/detail/" + movie_id, + data={"link_id": link_id, "is_switch": 1}, + headers=self._ajax_headers(page_url), + ) + data = (payload or {}).get("data") or {} + urls = self._normalize_play_links(data.get("play_links") or []) + if line_id: + urls = [item for item in urls if item["line_id"] == line_id] + if urls: + return self._build_direct_result(urls, page_url) + + return self._build_parse_result(page_url) + def _encode_play_id(self, payload): return json.dumps(payload, ensure_ascii=False, separators=(",", ":"))