import sqlite3 import os import time import requests import json import datetime DB_FILE=os.path.join(os.path.expanduser("~"),".ssh/files.db") url_api_cnpj="https://brasilapi.com.br/api/cnpj/v1/" class SQLConn: def __init__(self): self.conn=sqlite3.connect(DB_FILE, timeout=10.0) self.cursor=self.conn.cursor() self.cursor.execute( ''' CREATE TABLE IF NOT EXISTS company( id INTEGER PRIMARY KEY AUTOINCREMENT, codigo INTEGER UNIQUE NOT NULL, razao TEXT NOT NULL, nome_fantasia TEXT Null, cnpj TEXT NOT NULL UNIQUE, cnae TEXT DEFAULT Null, cidade TEXT DEFAULT Null, cep TEXT DEFAULT Null, uf TEXT DEFAULT Null, bairro TEXT DEFAULT Null, logradouro TEXT DEFAULT Null, numero TEXT DEFAULT Null, complemento TEXT DEFAULT Null, telefone TEXT DEFAULT Null, email TEXT DEFAULT Null, modalidade TEXT NOT NULL, ie TEXT DEFAULT Null, im TEXT DEFAULT Null, porte TEXT DEFAULT Null, codigo_municipio TEXT DEFAULT Null ) ''' ) time.sleep(0.5) mode=self.cursor.execute("PRAGMA journal_mode=WAL;") mode=mode.fetchone()[0] if str(mode).upper()!="WAL": raise RuntimeError(f"Não foi habilitado o Wal, journal_mode retornou: {mode}") else: print("WAL Habilitado!") def parametro_create(self, lista=[]): """ lista[0] -> cnpj -- OBRIGATORIO lista[1] -> codigo dominio -- OBRIGATORIO lista[2] -> modalidade -- OBRIGATORIO lista[3] -> inscricao estadual -- OPCIONAL lista[4] -> inscricao municipal -- OPCIONAL """ try: print("Retornando requisição de criar dados") modalidade=ie=im=None cnpj=lista[0] codigo_dominio=lista[1] ie=lista[3] im=lista[4] match str(lista[2]).replace(" ","").upper(): case "LP": modalidade="LUCRO PRESUMIDO" case "SN": modalidade="SIMPLES NACIONAL" case "ME": modalidade="MEI" case "LR": modalidade="LUCRO REAL" if lista[0]!=None and lista[1]!=None and modalidade!=None: requeste=requests.get(url_api_cnpj+cnpj) if requeste.status_code==200: dados=requeste.json() razao=dados['razao_social'] if dados['razao_social'] else None nome_fantasia=dados['nome_fantasia'] if dados['nome_fantasia'] else None cnae=dados['cnae_fiscal'] if dados['cnae_fiscal'] else None cidade=dados['municipio'] if dados['municipio'] else None cep=dados['cep'] if dados['cep'] else None uf=dados['uf'] if dados['uf'] else None bairro=dados['bairro'] if dados['bairro'] else None logradouro=dados['logradouro'] if dados['logradouro'] else None numero=dados['numero'] if dados['numero'] else None complemento=dados['complemento'] if dados['complemento'] else None telefone=dados['ddd_telefone_1'] if dados['ddd_telefone_1'] else None email=dados['email'] if dados['email'] else None porte=dados['porte'] if dados['porte'] else None codigo_municipio=dados['codigo_municipio_ibge'] self.cursor.execute('INSERT INTO company (codigo, razao, nome_fantasia, cnpj, cnae, cidade, cep, uf, bairro, logradouro, numero, complemento, telefone, email, modalidade, ie, im, porte, codigo_municipio) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)',(codigo_dominio,razao,nome_fantasia,cnpj,cnae,cidade,cep,uf,bairro,logradouro,numero,complemento,telefone,email,modalidade,ie,im,porte,codigo_municipio)) self.conn.commit() else: print("Erro ao retornar informações da empresa") else: print("Erro, informações não inseridas") except Exception as e: print(f"Erro -> {e}") raise e finally: self.conn.close() def parametro_read(self): try: sql="SELECT * FROM company" dados=[i for i in self.cursor.execute(sql).fetchall()] dados_=[] for ind,v in enumerate(dados): dados_.append({ f"{self.cursor.description[0][0]}":v[0], f"{self.cursor.description[1][0]}":v[1], f"{self.cursor.description[2][0]}":v[2], f"{self.cursor.description[3][0]}":v[3], f"{self.cursor.description[4][0]}":v[4], f"{self.cursor.description[5][0]}":v[5], f"{self.cursor.description[6][0]}":v[6],f"{self.cursor.description[7][0]}":v[7], f"{self.cursor.description[8][0]}":v[8],f"{self.cursor.description[9][0]}":v[9], f"{self.cursor.description[10][0]}":v[10],f"{self.cursor.description[11][0]}":v[11], f"{self.cursor.description[12][0]}":v[12],f"{self.cursor.description[13][0]}":v[13], f"{self.cursor.description[14][0]}":v[14],f"{self.cursor.description[15][0]}":v[15], f"{self.cursor.description[16][0]}":v[16],f"{self.cursor.description[17][0]}":v[17], f"{self.cursor.description[18][0]}":v[18],f"{self.cursor.description[19][0]}":v[19] }) except Exception as e: print(f"Erro ao efetuar leitura do banco -> {e}") finally: self.conn.close() return dados_ def parametro_readline(self, param=[]): try: if param[0]!="id": sql=f"SELECT * FROM company WHERE {param[0]}='{param[1]}'" else: sql=f"SELECT * FROM company WHERE {param[0]}={param[1]}" dados=[i for i in self.cursor.execute(sql).fetchall()] a={} for i,v in enumerate(self.cursor.description): a[f"{v[0]}"]=dados[0][i] dados=a except Exception as e: print(f"Erro ao retornar empresa -> {e}") raise e finally: self.conn.close() return dados def parametro_update(self, id, coluna, valor): try: print(self.cursor.execute(f"SELECT * FROM company WHERE id={id}").fetchall()) self.cursor.execute(f"UPDATE company SET {coluna}='{valor}' WHERE id={id}") self.conn.commit() except Exception as e: print(f"Erro ao efetuar alteração -> {e}") raise e finally: self.conn.close() def parametro_delete(self, param=[]): try: if param[0]!="id": sql=f"DELETE FROM company WHERE {param[0]}='{param[1]}'" else: sql=f"DELETE FROM company WHERE {param[0]}='{param[1]}'" self.cursor.execute(sql) self.conn.commit() except Exception as e: print(f"Erro ao excluir empresa -> {e}") raise e finally: self.conn.close() pasta_bots=os.path.join(os.path.expanduser("~"), "api/estrutura_api/api/utils/bots") with open(f"{pasta_bots}/controle/controle.json", 'r', encoding='utf-8') as f:json_=json.load(f) # print(SQLConn().parametro_readline(param=["cnpj", "54337931000109"])) SQLConn().parametro_delete(param=["cnpj","54337931000109"])