增加通知

common/sql_help.py

@@ -83,6 +83,18 @@ class sqlCollect():
         count = MysqlHelper.get_values(sql, (str(channel)))
         return count
 
+    @classmethod
+    def get_channel_count(cls, channel, name):
+        sql = """SELECT count(0) FROM machine_making_data where channel = %s and name = %s and  DATE(data_time) = CURRENT_DATE;"""
+        count = MysqlHelper.get_values(sql, (str(channel),name))
+        return count
+
+    @classmethod
+    def get_name_count(cls, name):
+        sql = """SELECT count(0) FROM machine_making_data where name = %s and  DATE(data_time) = CURRENT_DATE;"""
+        count = MysqlHelper.get_values(sql, (str(name)))
+        return count
+
     """
     判断该任务id是否用过
     """

video_count_notification.py

@@ -0,0 +1,61 @@
+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])}**条**\n"
+            f"**抖音搜索**: {int(dy_sou_count[0][0])}**条**\n"
+            f"**视频号品类账号**: {int(sph_count[0][0])}**条**\n"
+            f"**视频号推荐流**: {int(sph_feed_count[0][0])}**条**\n"
+            f"**视频号搜索**: {int(sph_sou_count[0][0])}**条**\n"
+            f"**快手品类账号**: {int(ks_count[0][0])}**条**\n"
+            f"**快手推荐流**: {int(ks_feed_count[0][0])}**条**\n"
+            f"**快手小程序**: {int(ks_xcx_count[0][0])}**条**\n"
+            f"**快手搜索**: {int(ks_sou_count[0][0])}**条**\n"
+            f"**相似溯源（xx）**: {int(xx_count[0][0])}**条**\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()
+