comprehensive_video_rewriting/utils/sql_help.py

import os
import sys
from datetime import datetime

sys.path.append(os.getcwd())
from utils.mysql_db import MysqlHelper


class sqlCollect:

    @classmethod
    def get_name_count(cls, name):
        sql = """SELECT count(0) FROM machine_making_data where name = %s and  DATE(data_time) = CURRENT_DATE AND HOUR(data_time) <= HOUR(NOW());"""
        count = MysqlHelper.get_values(sql, (str(name)))
        return count

    @classmethod
    def get_ytd_name_count(cls, name):
        sql = """SELECT count(0) FROM machine_making_data where name = %s and  data_time >= CURRENT_DATE - INTERVAL 1 DAY AND data_time <= CURRENT_DATE - INTERVAL 1 DAY + INTERVAL HOUR(NOW()) HOUR;"""
        count = MysqlHelper.get_values(sql, (str(name)))
        return count

    @classmethod
    def get_crawler_count(cls, platform):
        sql = """SELECT count(0) FROM crawler_video where platform = %s and  DATE(create_time) = CURRENT_DATE AND HOUR(create_time) < HOUR(NOW());"""
        count = MysqlHelper.get_values(sql, (str(platform)))
        return count

    @classmethod
    def get_ytd_crawler_count(cls, platform):
        sql = """SELECT count(0) FROM crawler_video where platform = %s and create_time >= CURRENT_DATE - INTERVAL 1 DAY AND create_time < CURRENT_DATE - INTERVAL 1 DAY + INTERVAL HOUR(NOW()) HOUR;"""
        count = MysqlHelper.get_values(sql, (str(platform)))
        return count

    @classmethod
    def get_bygj_count(cls, name):
        sql = """SELECT count(0) FROM machine_making_data where name = %s and task_mark = "搬运工具" and  DATE(data_time) = CURRENT_DATE AND HOUR(data_time) <= HOUR(NOW());"""
        count = MysqlHelper.get_values(sql, (str(name)))
        return count

    @classmethod
    def get_bygj_all_count(cls):
        sql = """SELECT count(0) FROM machine_making_data where task_mark = %s  and  DATE(data_time) = CURRENT_DATE AND HOUR(data_time) <= HOUR(NOW());"""
        count = MysqlHelper.get_values(sql, (str("搬运工具")))
        return count

    @classmethod
    def get_ytd_bygj_count(cls, name):
        sql = """SELECT count(0) FROM machine_making_data where name = %s and task_mark = "搬运工具" and data_time >= CURRENT_DATE - INTERVAL 1 DAY AND data_time <= CURRENT_DATE - INTERVAL 1 DAY + INTERVAL HOUR(NOW()) HOUR;"""
        count = MysqlHelper.get_values(sql, (str(name)))
        return count

    """
    视频信息写入库中
    """
    @classmethod
    def insert_task(cls, task_mark, video_id, mark, channel):
        current_time = datetime.now()
        formatted_time = current_time.strftime("%Y-%m-%d %H:%M:%S")
        insert_sql = f"""INSERT INTO pj_video_data (task_name, used_video_id, mark_name, data_time, channel) values ('{task_mark}' ,'{video_id}','{mark}', '{formatted_time}', '{channel}')"""
        MysqlHelper.update_values(
            sql=insert_sql
        )

    """写入数据库"""
    @classmethod
    def insert_machine_making_data(cls, name: str, task_mark: str, channel_id: str, url: str, v_id: str, piaoquan_id: str, new_title: str, code: str, formatted_time, old_title: str, oss_object_key: str):
        insert_sql = f"""INSERT INTO machine_making_data (name, task_mark, channel, user, v_id, pq_uid, title, pq_vid, data_time, old_title, oss_object_key) values ("{name}", "{task_mark}", "{channel_id}", "{url}", "{v_id}" , "{piaoquan_id}", "{new_title}", "{code}", "{formatted_time}", "{old_title}", "{oss_object_key}")"""
        MysqlHelper.update_values(
            sql=insert_sql
        )
        return

    @classmethod
    def get_mark_count(cls, mark: str):
        sql = """SELECT count(0) FROM machine_making_data where task_mark = %s  and  DATE(data_time) = CURRENT_DATE AND HOUR(data_time) <= HOUR(NOW());"""
        count = MysqlHelper.get_values(sql, (str(mark)))
        return count

    """
    根据时间判断该任务id是否用过
    """
    @classmethod
    def is_used_days(cls, video_id, channel, day_count):
        sql = """
            SELECT used_video_id
            FROM pj_video_data
            WHERE used_video_id = %s 
              AND channel = %s 
              AND data_time >= DATE_SUB(NOW(), INTERVAL %s DAY)
            ORDER BY data_time DESC
            LIMIT 1;
        """
        data = MysqlHelper.get_values(sql, (str(video_id), channel, int(day_count)))
        if len(data) == 0 or data == ():
            return False
        return True

    """
    判断该任务id是否用过
    """
    @classmethod
    def is_used(cls, video_id, channel):
        sql = """
                SELECT used_video_id
                FROM pj_video_data
                WHERE used_video_id = %s AND  channel = %s 
                ORDER BY data_time DESC
                LIMIT 1
            """
        data = MysqlHelper.get_values(sql, (str(video_id), channel))
        if len(data) == 0 or data == ():
            return False
        return True