| 12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061 |
- import datetime
- import time
- from datetime import datetime
- import schedule
- from loguru import logger
- from common import Feishu
- from common.sql_help import sqlCollect
- def bot_carry_data():
- try:
- sph_feed_count = sqlCollect.get_channel_count("视频号推荐流", "视频号推荐流")
- ks_feed_count = sqlCollect.get_channel_count("快手推荐流", "快手推荐流")
- ks_xcx_count = sqlCollect.get_channel_count("快手小程序", "快手小程序")
- ks_sou_count = sqlCollect.get_channel_count("快手搜索", "品类关键词搜索")
- dy_sou_count = sqlCollect.get_channel_count("抖音搜索", "品类关键词搜索")
- sph_sou_count = sqlCollect.get_channel_count("视频号搜索", "品类关键词搜索")
- dy_count = sqlCollect.get_name_count("抖音品类账号")
- ks_count = sqlCollect.get_name_count("快手品类账号")
- sph_count = sqlCollect.get_name_count("视频号品类账号")
- xx_count = sqlCollect.get_name_count("信欣")
- dd_count = sqlCollect.get_name_count("单点视频")
- text = (
- f"**抖音品类账号**: {int(dy_count[0][0])}**/1500条**\n"
- f"**抖音搜索**: {int(dy_sou_count[0][0])}**/1200条**\n"
- f"**视频号品类账号**: {int(sph_count[0][0])}**/350条**\n"
- f"**视频号推荐流**: {int(sph_feed_count[0][0])}**/200条**\n"
- f"**视频号搜索**: {int(sph_sou_count[0][0])}**/100条**\n"
- f"**快手品类账号**: {int(ks_count[0][0])}**/900条**\n"
- f"**快手推荐流**: {int(ks_feed_count[0][0])}**/100条**\n"
- f"**快手小程序**: {int(ks_xcx_count[0][0])}**/?条**\n"
- f"**快手搜索**: {int(ks_sou_count[0][0])}**/650条**\n"
- f"**相似溯源（xx）**: {int(xx_count[0][0])}**/200条**\n"
- f"**单点视频**: {int(dd_count[0][0])}**条**\n"
- )
- Feishu.finish_bot(text,
- "https://open.feishu.cn/open-apis/bot/v2/hook/805ab4fe-0246-409c-b383-ea25831148e6",
- f"【 自制视频{datetime.now().strftime('%Y年%m月%d日 %H时')}数据通知 】")
- except Exception as e:
- logger.error(f"[+] 报警失败{e}")
- def schedule_tasks():
- schedule.every().hour.at(":05").do(bot_carry_data)
- if __name__ == "__main__":
- schedule_tasks() # 调用任务调度函数
- while True:
- schedule.run_pending()
- time.sleep(1) # 每秒钟检查一次
- # bot_carry_data()
|
