unzip/db.py

import pymysql
from pymysql.cursors import DictCursor
from data_dict import special_symbols, escape_list

db_config = {
    "host": "192.168.66.101",
    "user": "root",
    "passwd": "Sxzgx1209",
    "port": 3306,
    'database': 'scrapy'
}


class DbAction:
    def __init__(self):
        self.conn = pymysql.Connect(**db_config)
        self.cursor = self.conn.cursor(cursor=DictCursor)

    @staticmethod
    def decode_pwd(_pwd):
        for k, v in special_symbols.items():
            if k in _pwd:
                _pwd = _pwd.replace(k, v)
        for item in escape_list:
            if item[0] in _pwd:
                _pwd = _pwd.replace(item[0], item[1])
        return _pwd

    def get_data_by_id(self, _id):
        SELECT_SQL = "SELECT * FROM scrapyh s WHERE s.id = %s;"
        self.cursor.execute(SELECT_SQL, (_id,))
        result = self.cursor.fetchone()
        if result:
            result['unzip_pwd'] = self.decode_pwd(result['unzip_pwd'])
        return result

    def get_available_pwd(self):
        SELECT_SQL = "SELECT * FROM scrapyh_pwd;"
        self.cursor.execute(SELECT_SQL)
        result = self.cursor.fetchall()
        return [r['pwd'] for r in result]

    def insert_pwd(self, _pwd):
        SELECT_SQL = "SELECT * FROM scrapyh_pwd sp WHERE sp.pwd = %s;"
        self.cursor.execute(SELECT_SQL, (_pwd,))
        result = self.cursor.fetchone()
        if not result:
            try:
                INSERT_SQL = "INSERT INTO scrapyh_pwd (pwd) VALUES (%s);"
                self.cursor.execute(INSERT_SQL, (_pwd,))
                self.conn.commit()
            except Exception as e:
                print(e)
                self.conn.rollback()