# -*- coding: utf-8 -*- # @Author: wangkun # @Time: 2023/5/15 import random import time from common.scheduling_db import MysqlHelper class Demo: @classmethod def get_user(cls, log_type, crawler, env): select_user_sql = f"""select * from crawler_user_v3 where task_id=36""" user_list = MysqlHelper.get_values(log_type, crawler, select_user_sql, env, action="") print(user_list) our_uid_list = [] for user in user_list: our_uid_list.append(user["uid"]) print(our_uid_list) our_uid = random.choice(our_uid_list) print(our_uid) @classmethod def test_dict(cls): video_dict = { "play_cnt": 1000, "share_cnt": 1000, "duration": 55, "period": 5, "publish_time_stamp": 1683648000, # 2023-05-10 00:00:00 "video_url": "www.baidu.com" } rule_dict = { "play_cnt": {"min": 0, "max": 0}, "fans_cnt": {"min": 0, "max": 0}, "videos_cnt": {"min": 0, "max": 0}, "like_cnt": {"min": 0, "max": 0}, "video_width": {"min": 0, "max": 0}, "video_height": {"min": 0, "max": 0}, "duration": {"min": 0, "max": 0}, "share_cnt": {"min": 0, "max": 0}, "comment_cnt": {"min": 0, "max": 0}, "favorite_cnt": {"min": 0, "max": 0}, "period": {"min": 10, "max": 0}, "publish_time": {"min": 1673734400000, "max": 0} } # 格式化 video_dict:publish_time_stamp if "publish_time_stamp" in video_dict.keys(): video_dict["publish_time"] = video_dict["publish_time_stamp"]*1000 # 格式化 video_dict:period if "period" not in video_dict.keys() and "publish_time" in video_dict.keys(): video_dict["period"] = int((int(time.time()*1000)-video_dict["publish_time"])/(3600*24*1000)) # 格式化 rule_dict 最大值取值为 0 的问题 for rule_value in rule_dict.values(): if rule_value["max"] == 0: rule_value["max"] = 999999999999999 rule_dict["period"]["max"] = rule_dict["period"]["min"] rule_dict["period"]["min"] = 0 for k, v in rule_dict.items(): print(f"{k}:{v}") # 格式化 rule_dict 有的 key,video_dict 中没有的问题 for rule_key in rule_dict.keys(): if rule_key not in video_dict.keys(): video_dict[rule_key] = int(rule_dict[rule_key]["max"] / 2) # 比较结果,输出结果:True / False for video_key, video_value in video_dict.items(): for rule_key, rule_value in rule_dict.items(): if video_key == rule_key: result = rule_value["min"] <= video_value <= rule_value["max"] print(f'{video_key}: {rule_value["min"]} <= {video_value} <= {rule_value["max"]},{result}') # Common.logger(log_type, crawler).info(f'{video_key}: {rule_value["min"]} <= {video_value} <= {rule_value["max"]},{result}') if result is False: return False else: continue return True @classmethod def save_video_info(cls): video_dict = {'video_title': "测试视频标题", 'video_id': "id_1", 'play_cnt': 199, 'publish_time_stamp': 1683648000, 'publish_time_str': "2023-05-10 00:00:00", 'user_name': "岁岁年年迎福气", 'user_id': "suisuiniannianyingfuqi", 'avatar_url': "https://cdn.jzkksp.com/2022/3/22/lnchd2.jpg?auth_key=1684223012-0-0-2f8ddcf0e5d5f164f792b77c98e1ffde", 'cover_url': "https://cdn.jzkksp.com/2022/3/22/lnchd2.jpg?auth_key=1684223012-0-0-2f8ddcf0e5d5f164f792b77c98e1ffde", 'video_url': "https://cdn.jzkksp.com/2022/3/22/lnchd.mp4", 'session': f"suisuiniannianyingfuqi-{int(time.time())}"} save_dict = { "video_title": "video_title", "video_id": "video_id", "duration": 0, "play_cnt": 0, "comment_cnt": 0, "like_cnt": 0, "share_cnt": 0, "video_width": 1920, "video_height": 1080, "publish_time_stamp": 946656000, # 2000-01-01 00:00:00 "user_name": "crawler", "avatar_url": "http://weapppiccdn.yishihui.com/resources/images/pic_normal.png", "video_url": "video_url", "cover_url": "cover_url", "session": f"session-{int(time.time())}", } for video_key, video_value in video_dict.items(): for save_key, save_value in save_dict.items(): if save_key == video_key: save_dict[save_key] = video_value for k, v in save_dict.items(): print(f"{k}:{v}") with open(f"./info.txt", "w", encoding="UTF-8") as f_a: f_a.write(str(save_dict['video_id']) + "\n" + str(save_dict['video_title']) + "\n" + str(save_dict['duration']) + "\n" + str(save_dict['play_cnt']) + "\n" + str(save_dict['comment_cnt']) + "\n" + str(save_dict['like_cnt']) + "\n" + str(save_dict['share_cnt']) + "\n" + f"{save_dict['video_width']}*{save_dict['video_height']}" + "\n" + str(save_dict['publish_time_stamp']) + "\n" + str(save_dict['user_name']) + "\n" + str(save_dict['avatar_url']) + "\n" + str(save_dict['video_url']) + "\n" + str(save_dict['cover_url']) + "\n" + str(save_dict['session'])) if __name__ == "__main__": # Demo.get_user("demo", "suisuiniannianyingfuqi", "dev") print(Demo.test_dict()) # print(500 <= 1000 <= 100000000) # Demo.save_video_info() pass