# -*- coding: utf-8 -*- # @Author: wangkun # @Time: 2022/5/16 import os import sys import time import requests import urllib3 sys.path.append(os.getcwd()) from main.common import Common from main.feishu_lib import Feishu from main.xiaoniangao_publish import Publish proxies = {"http": None, "https": None} class Demo: # 查询视频详情的各项数据 @classmethod def video_detail_info(cls, user_id, user_mid, video_id): url = "https://kapi.xiaoniangao.cn/profile/get_profile_by_id" headers = { "x-b3-traceid": "bd267349bf41b", "X-Token-Id": "86f6d7cc2b2b6870004df5d16c82aaf3-1185665701", "uid": "8fde3c6c-c070-4379-bfc4-15c7e85139c9", "content-type": "application/json", "Accept-Encoding": "gzip,compress,br,deflate", "User-Agent": 'Mozilla/5.0 (iPhone; CPU iPhone OS 14_7_1 like Mac OS X)' ' AppleWebKit/605.1.15 (KHTML, like Gecko) Mobile/15E148 ' 'MicroMessenger/8.0.20(0x18001432) NetType/WIFI Language/zh_CN', "Referer": "https://servicewechat.com/wxd7911e4c177690e4/617/page-frame.html" } data = { "play_src": "1", "profile_id": int(user_id), "profile_mid": int(user_mid), "qs": "imageMogr2/gravity/center/rotate/$/thumbnail/" "!400x400r/crop/400x400/interlace/1/format/jpg", "h_qs": "imageMogr2/gravity/center/rotate/$/thumbnail" "/!80x80r/crop/80x80/interlace/1/format/jpg", "share_width": 625, "share_height": 500, "no_comments": True, "no_follow": True, "vid": video_id, "hot_l1_comment": True, "token": "90747742180aeb22c0fe3a3c6a38f3d9", "uid": "8fde3c6c-c070-4379-bfc4-15c7e85139c9", "proj": "ma", "wx_ver": "8.0.20", "code_ver": "3.62.0", "log_common_params": { "e": [{ "data": { "page": "dynamicSharePage" } }], "ext": { "brand": "iPhone", "device": "iPhone 11", "os": "iOS 14.7.1", "weixinver": "8.0.20", "srcver": "2.24.3", "net": "wifi", "scene": "1089" }, "pj": "1", "pf": "2", "session_id": "7bcce313-b57d-4305-8d14-6ebd9a1bad29" } } try: urllib3.disable_warnings() r = requests.post(headers=headers, url=url, json=data, proxies=proxies, verify=False) print(r.text) video_title = r.json()["data"]["title"] video_id = r.json()["data"]["vid"] video_play_cnt = r.json()["data"]["play_pv"] video_duration = int(r.json()["data"]["du"]) / 1000 video_url = r.json()["data"]["v_url"] video_send_time = r.json()["data"]["t"] print(f"video_title:{video_title}") print(f"video_id:{video_id}") print(f"video_play_cnt:{video_play_cnt}") print(f"video_duration:{video_duration}") print(f'video_send_time:{time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(int(video_send_time) / 1000))}') print(f"video_url:{video_url}") except Exception as e: print(e) @classmethod def download_publish(cls, log_type, env): # try: play_feeds_sheet = Feishu.get_values_batch(log_type, 'xiaoniangao', 'CtXmf8') for i in range(1, len(play_feeds_sheet)): download_video_title = play_feeds_sheet[i][5] download_video_play_cnt = play_feeds_sheet[i][7] download_video_resolution = play_feeds_sheet[i][8] download_user_name = play_feeds_sheet[i][9] download_user_id = play_feeds_sheet[i][10] download_user_mid = play_feeds_sheet[i][11] download_head_url = play_feeds_sheet[i][12] download_cover_url = play_feeds_sheet[i][12] download_video_url = play_feeds_sheet[i][13] Common.logger(log_type).info("正在判断第{}行，视频:{}", i + 1, download_video_title) # 过滤空行及空标题视频 if download_video_title is None or download_video_title == "": # 删除行或列，可选 ROWS、COLUMNS Feishu.dimension_range(log_type, "xiaoniangao", "CtXmf8", "ROWS", i + 1, i + 1) Common.logger(log_type).warning("标题为空或空行，删除成功\n") return else: # 下载封面 Common.download_method(log_type=log_type, text="cover", d_name=str(download_video_title), d_url=str(download_cover_url)) # 下载视频 Common.download_method(log_type=log_type, text="video", d_name=str(download_video_title), d_url=str(download_video_url)) # 保存视频信息至 "./videos/{download_video_title}/info.txt" with open("./videos/" + download_video_title + "/" + "info.txt", "a", encoding="UTF-8") as f_a: f_a.write(str(int(time.time())) + "\n" + str(download_video_title) + "\n" + '120' + "\n" + str(download_video_play_cnt) + "\n" + '0' + "\n" + '120' + "\n" + '120' + "\n" + str(download_video_resolution) + "\n" + str(int(time.time())) + "\n" + str(download_user_name) + "\n" + str(download_head_url) + "\n" + str(download_video_url) + "\n" + str(download_cover_url) + "\n" + "xiaoniangao"+str(int(time.time()))) Common.logger(log_type).info("==========视频信息已保存至info.txt==========") # 上传视频 Common.logger(log_type).info("开始上传视频:{}".format(download_video_title)) our_video_id = Publish.upload_and_publish(log_type, env, "play") our_video_link = "https://testadmin.piaoquantv.com/cms/post-detail/" + str(our_video_id) + "/info" Common.logger(log_type).info("视频上传完成:{}", download_video_title) # 保存视频: 播放量_已下载表 Common.logger(log_type).info("保存视频ID至云文档:{}", download_video_title) # 视频ID工作表，插入首行 Feishu.insert_columns(log_type, "xiaoniangao", "67U1WF", "ROWS", 1, 2) # 视频ID工作表，首行写入数据 upload_time = int(time.time()) values = [[ str(download_video_title), our_video_link, download_video_play_cnt, str(download_video_resolution), str(download_user_name), str(download_user_id), str(download_user_mid), str(download_cover_url), str(download_video_url)]] time.sleep(1) Feishu.update_values(log_type, "xiaoniangao", "67U1WF", "F2:Z2", values) # 删除行或列，可选 ROWS、COLUMNS Feishu.dimension_range(log_type, "xiaoniangao", "CtXmf8", "ROWS", i + 1, i + 1) Common.logger(log_type).info("视频:{}，下载/上传成功\n", download_video_title) return # except Exception as e: # Common.logger(log_type).error('download_publish异常:{}', e) @classmethod def run_download_publish(cls, log_type, env): # try: while True: if len(Feishu.get_values_batch(log_type, 'xiaoniangao', 'CtXmf8')) == 1: Common.logger(log_type).info('上传/下载完成\n') break else: cls.download_publish(log_type, env) # except Exception as e: # Common.logger(log_type).error('run_download_publish异常:{}', e) @classmethod def test_sheet(cls): sheet = Feishu.get_values_batch("test", "xiaoniangao", "zI9QJz") print(sheet) if __name__ == "__main__": # Demo.run_download_publish('test', 'dev') Demo.test_sheet() pass