发布时间:2026-07-21阅读(1)

def __crawl_user(self, uid):if uid.isdigit(): uid = self.__switch_id(uid)payload = {"operationName": "privateFeedsQuery", "variables": {"principalId": uid, "pcursor": "", "count": 999}, "query": "query privateFeedsQuery($principalId: String, $pcursor: String, $count: Int) {\n privateFeeds(principalId: $principalId, pcursor: $pcursor, count: $count) {\n pcursor\n list {\n id\n thumbnailUrl\n poster\n workType\n type\n useVideoPlayer\n imgUrls\n imgSizes\n magicFace\n musicName\n caption\n location\n liked\n onlyFollowerCanComment\n relativeHeight\n timestamp\n width\n height\n counts {\n displayView\n displayLike\n displayComment\n __typename\n }\n user {\n id\n eid\n name\n avatar\n __typename\n }\n expTag\n __typename\n }\n __typename\n }\n}\n"}res = requests.post(self.__data_url, headers=self.__headers, json=payload)works = json.loads(res.content.decode(encoding=utf-8, errors=strict))[data][privateFeeds][list]if not os.path.exists("../data"): os.makedirs("../data")# 这两行代码将response写入json供分析# with open("data/" uid ".json", "w") as fp:# fp.write(json.dumps(works, indent=2))# 防止该用户在直播,第一个作品默认为直播,导致获取信息为NoneTypeif works[0][id] is None: works.pop(0)name = re.sub(r[\\/:*?"<>|\r\n] , "", works[0][user][name])dir = "data/" name "(" uid ")/"# print(len(works))if not os.path.exists(dir): os.makedirs(dir)# if not os.path.exists(dir ".list"):# print("")print("开始爬取用户 " name ",保存在目录 " dir)print(" 共有" str(len(works)) "个作品")for j in range(len(works)): self.__crawl_work(uid, dir, works[j], j 1) time.sleep(1)print("用户 " name "爬取完成!")print()time.sleep(1)
快手分为五种类型的作品,在作品里面表现为workType属性

def __crawl_work(self, uid, dir, work, wdx): w_type = work[workType] w_caption = re.sub(r"\s ", " ", work[caption]) w_name = re.sub(r[\/:*?"<>|\r\n] , "", w_caption)[0:24] w_time = time.strftime(%Y-%m-%d, time.localtime(work[timestamp] / 1000))if w_type == vertical or w_type == multiple or w_type == "single" or w_type == ksong: w_urls = work[imgUrls] l = len(w_urls) print(" " str(wdx) ")图集作品:" w_caption "," "共有" str(l) "张图片") for i in range(l): p_name = w_time "_" w_name "_" str(i 1) ".jpg" pic = dir p_name if not os.path.exists(pic): r = requests.get(w_urls[i]) r.raise_for_status() with open(pic, "wb") as f: f.write(r.content) print(" " str(i 1) "/" str(l) " 图片 " p_name " 下载成功 √") else: print(" " str(i 1) "/" str(l) " 图片 " p_name " 已存在 √")elif w_type == video: w_url = self.__work_url work[id] res = requests.get(w_url, headers=self.__headers_mobile, params={"fid": 1841409882, "cc": "share_copylink", "shareId": "143108986354"}) html = res.text waitreplace = work[id] ".*?"srcNoMark":"(.*?)" v_url = re.findall(waitreplace, html) # pattern = re.compile(r"playUrl", re.MULTILINE | re.DOTALL) # script = soup.find("script", text=pattern) # s = pattern.search(script.text).string # v_url = s.split(playUrl":")[1].split(.mp4)[0].encode(utf-8).decode(unicode-escape) .mp4 try: print(" " str(wdx) ")视频作品:" w_caption) except: print(" 这里似乎有点小错误,已跳过") v_name = w_time "_" w_name ".mp4" video = dir v_name if v_url: if not os.path.exists(video): r = requests.get(v_url[0]) r.raise_for_status() with open(video, "wb") as f: f.write(r.content) print(" 视频 " v_name " 下载成功 √") else: print(" 视频 " v_name " 已存在 √") else: print("未找到视频")else: print("错误的类型")

注意事项:
项目源码地址 https://github.com/oGsLP/kuaishou-crawler
Copyright © 2024 有趣生活 All Rights Reserve吉ICP备19000289号-5 TXT地图HTML地图XML地图