top_automatic_traceability/common/sql_help.py

import os
import sys
from datetime import datetime

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


class sqlCollect():


    @classmethod
    def get_channle_id(cls, pq_id):
        """
        从数据库表中读取 id
        """
        sql = f"""select v_id,channel from machine_making_data where pq_vid = %s limit 1"""
        data = MysqlHelper.get_values(sql, (pq_id,))
        if data:
            return data[0][0],data[0][1]
        else:
            return None, None

    @classmethod
    def insert_pj_video_data(cls, user_video_id: str, channel: str):
        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, mark_name,user_video_id,channel,data_time) values ("{user_video_id}", "{channel}", "{user_video_id}", "{channel}", '{formatted_time}')"""
        MysqlHelper.update_values(
            sql=insert_sql
        )

    @classmethod
    def select_pj_video_data(cls, user_video_id):
        """
        从数据库表中读取 id
        """
        sql = f"""select user_video_id from pj_video_data where user_video_id = %s limit 1"""
        data = MysqlHelper.get_values(sql, (user_video_id,))
        if data:
            return True
        else:
            return False