[Python] 纯文本查看 复制代码
import os
import time
import random
import requests
import json
import traceback
from pathlib import Path
# ===================== 神图君 =====================
# 全局公用 e_param
COMMON_E_PARAM = "自行抓取"
COMMON_openid = "自行抓取"
# === 签名逻辑 ===
def sign_system(e: str) -> str:
"""
函数的签名生成逻辑
"""
timestamp = str(round(time.time()))
e = e + "###" + timestamp
# __APP__.wxapkg 解密 里面的 app-service.js
t = "PXhw7UT1B0a9kQDKZsjIASmOezxYG4CHo5Jyfg2b8FLpEvRr3WtVnlqMidu6cN"
i = ""
for char in e:
o = t.find(char)
if o == -1:
s_char = char
else:
s_char = t[(o + 3) % 62]
n = int(62 * random.random())
d = int(62 * random.random())
i += t[n] + s_char + t[d]
return i
# === headers ===
def get_common_headers(sign):
return {
"Host": "stapi.youpengw.com",
"Connection": "keep-alive",
"xweb_xhr": "1",
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36 MicroMessenger/7.0.20.1781(0x6700143B) NetType/WIFI MiniProgramEnv/Windows WindowsWechat/WMPF WindowsWechat(0x63090b11)XWEB/14315",
"sign": sign,
"Content-Type": "application/json",
"Accept": "*/*",
"Sec-Fetch-Site": "cross-site",
"Sec-Fetch-Mode": "cors",
"Sec-Fetch-Dest": "empty",
"Referer": "https://servicewechat.com/wxba7de3a24a3ff4d5/105/page-frame.html",
"Accept-Encoding": "gzip, deflate, br",
"Accept-Language": "zh-CN,zh;q=0.9"
}
# === 获取搜索列表 ===
def request_album_photoList(homepage_id):
url = "https://stapi.youpengw.com/miniwechat/v1/shentujun/photoList"
# token
e_param = COMMON_E_PARAM
# 初始化变量
all_data = []
page = 1
limit = 30 # 这里必须和接口保持一致！！！
max_pages = 50
while page <= max_pages:
try:
# 每次重新生成时间戳 + 签名（关键！）
import time
timestamp = int(time.time() * 1000)
sign = sign_system(e_param)
params = {
"homepage_id": homepage_id,
"classify_id": 0,
"sort_type": "default",
"album_id": -2,
"uid": 990439,
"page": page,
"limit": limit, # 统一
"id": 0,
"action": "cover",
"version": 1,
"appid": "wxba7de3a24a3ff4d5",
"openid": COMMON_openid,
"timestamp": timestamp
}
# 请求头
headers = get_common_headers(sign)
response = requests.post(url, headers=headers, json=params, timeout=20)
result = response.json()
# 请求失败直接结束
if result.get("code") != 1 or not result.get("data"):
print(f"❌ 第{page}页请求失败或无数据，结束翻页")
break
current_page_data = result["data"].get("photoList", [])
# 关键：当前页没有数据 → 结束翻页
if not current_page_data:
print(f"✅ 第{page}页无数据，全部爬取完毕")
break
all_data.extend(current_page_data)
print(f"✅ 第{page}页：获取到 {len(current_page_data)} 条，累计 {len(all_data)} 条")
page += 1
time.sleep(0.5)
except Exception as e:
print(f"❌ 第{page}页异常：{str(e)}")
break
# 解析出所有的URL
classify_id = all_data[0].get("classify_id")
# print(classify_id)
# print(all_data)
# exit()
# classify_id=1 视频 classify_id=5 图片 classify_id=7 链接 classify_id=10 集合
if classify_id == 10:
return classify_id
else:
# 解析出所有视频URL
video_urls = []
for item in all_data:
# 用ID作为文件名，避免重复
video_id = item.get("id")
video_url = item.get("video_url") or item.get("url")
if video_url:
video_urls.append({
"id": video_id,
"url": video_url
})
print(f"✅ 成功获取 {len(video_urls)} 个视频下载链接")
return video_urls
# === 获取搜索 ===
def request_album_search(keyword):
url = "https://stapi.youpengw.com/miniwechat/v1/shentujun/search"
# token
e_param = COMMON_E_PARAM
sign = sign_system(e_param)
params = {
"search_code":keyword,
"version":1,
"action":"homepageinfo",
"appid":"wxba7de3a24a3ff4d5",
"openid":COMMON_openid,
"timestamp":1776152477228
}
# 请求头
headers = get_common_headers(sign)
response = requests.post(
url=url,
headers=headers,
json=params,
timeout=20,
)
result = response.json()
# print(result)
# exit()
if result.get("code") == 1 and result.get("data"):
homepage_id = result["data"].get("homepage_id", [])
# print(result)
# exit()
photoList = request_album_photoList(homepage_id)
return photoList
else:
msg = result.get("msg")
print(msg)
exit()
# === 获取集合列表 ===
def request_album_list():
session = requests.Session()
session.trust_env = False
# 禁用代理
url = "https://stapi.youpengw.com/miniwechat/v1/brand/albumList"
# token
e_param = COMMON_E_PARAM
sign = sign_system(e_param)
# 接口参数
params = {
"brand_id":"1",
"page":1,
"limit":20,
"version":1,
"appid":"wxba7de3a24a3ff4d5",
"openid":COMMON_openid,
"timestamp":1776152477228
}
# 请求头
headers = get_common_headers(sign)
response = session.post(
url=url,
headers=headers,
json=params,
timeout=20,
)
result = response.json()
return result
# === 获取详细 ===
def request_album_find(id):
session = requests.Session()
session.trust_env = False
# 禁用代理
url = "https://stapi.youpengw.com/miniwechat/v1/brand/albumImageList"
# token
e_param = COMMON_E_PARAM
sign = sign_system(e_param)
# 接口参数
params = {
"brand_id":2,
"homepage_id":"2550841",
"album_id":id,
"appid":"wxba7de3a24a3ff4d5",
"openid":COMMON_openid,
"page":1,
"limit":24,
"version":1,
"timestamp":1776151154843,
}
# 请求头
headers = get_common_headers(sign)
try:
response = session.post(
url=url,
headers=headers,
json=params,
timeout=20,
)
result = response.json()
# 解析出所有视频URL
video_urls = []
if result.get("code") == 1 and result.get("data"):
photo_list = result["data"].get("photoList", [])
for item in photo_list:
video_url = item.get("url")
video_id = item.get("id") # 用ID作为文件名，避免重复
if video_url and video_url.endswith(".mp4"):
video_urls.append({
"id": video_id,
"url": video_url
})
print(f"✅ 成功获取 {len(video_urls)} 个视频下载链接")
return video_urls
except Exception as e:
return {
"status": "error",
"msg": f"请求失败：{str(e)}",
}
# === 下载文件 ===
def download_video(video_info,title):
"""
下载单个视频
:param video_info: 包含id和url的字典
:param save_dir: 保存目录（自动创建）
"""
session = requests.Session()
session.trust_env = False
# 禁用代理
video_id = video_info["id"]
video_url = video_info["url"]
file_ext = os.path.splitext(video_url)[1]
# 创建保存目录
save_dir = f"download_videos/{title}"
# Path(save_dir).mkdir(exist_ok=True)
Path(save_dir).mkdir(parents=True, exist_ok=True)
# 文件名：用ID命名，避免乱码
filename = f"video_{video_id}{file_ext}"
save_path = os.path.join(save_dir, filename)
# 如果文件已存在，跳过
if os.path.exists(save_path):
print(f"ℹ️ 文件已存在，跳过：{filename}")
return True
try:
# 流式下载（支持大文件）
response = requests.get(
video_url,
stream=True,
timeout=60,
headers={
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/126.0.0.0 Safari/537.36"
}
)
response.raise_for_status() # 检查HTTP错误
# 获取文件大小（显示进度）
total_size = int(response.headers.get("Content-Length", 0))
downloaded_size = 0
# 写入文件
with open(save_path, "wb") as f:
for chunk in response.iter_content(chunk_size=8192):
if chunk:
f.write(chunk)
downloaded_size += len(chunk)
# 显示下载进度
if total_size > 0:
progress = (downloaded_size / total_size) * 100
print(f"\r📥 下载中 {filename}：{progress:.1f}% ({downloaded_size/1024/1024:.2f}MB/{total_size/1024/1024:.2f}MB)", end="")
print(f"\n✅ 下载完成：{filename} → 保存路径：{save_path}")
return True
except Exception as e:
print(f"\n❌ 下载失败 {filename}：{str(e)}")
# 删除不完整的文件
if os.path.exists(save_path):
os.remove(save_path)
return False
# ===================== 运行 =====================
if __name__ == "__main__":
keyword = '雪见0308'
# keyword = 'xx1212'
# keyword = '588688'
# keyword = '8023'
# keyword = '4039'
# keyword = '668866'
print("开始调用接口...\n")
success_countsearch = request_album_search(keyword)
# print(success_countsearch)
# exit()
if success_countsearch == 10:
result1 = request_album_list()
if result1.get("code") == 1 and result1.get("data"):
photo_list = result1["data"].get("albumList", [])
for item in photo_list:
id = item.get("id")
title = item.get("title")
result2 = request_album_find(id)
success_count = 0
fail_count = 0
print("\n============ 开始批量下载 ============")
for idx, video_info in enumerate(result2, 1):
if download_video(video_info,title):
success_count += 1
else:
fail_count += 1
print("\n============ 下载完成 ============")
print(f"✅ 成功：{success_count} 个")
print(f"❌ 失败：{fail_count} 个")
print(f"📂 所有文件保存在：{os.path.abspath('download_videos')}")
else:
success_count = 0
fail_count = 0
print("\n============ 开始批量下载 ============")
for idx, video_info in enumerate(success_countsearch, 1):
if download_video(video_info,keyword):
success_count += 1
else:
fail_count += 1
print("\n============ 下载完成 ============")
print(f"✅ 成功：{success_count} 个")
print(f"❌ 失败：{fail_count} 个")
print(f"📂 所有文件保存在：{os.path.abspath('download_videos')}")
发表于 2026-4-15 12:23
|
发表于 2026-4-16 09:25
