225 lines
8.4 KiB
Python
225 lines
8.4 KiB
Python
import os
|
||
import sys
|
||
import requests
|
||
import pyodbc
|
||
import unicodedata
|
||
from datetime import datetime, timedelta
|
||
|
||
# Detectar sistema operacional
|
||
IS_WINDOWS = sys.platform.startswith('win')
|
||
|
||
# Caminho baseado no sistema operacional
|
||
if IS_WINDOWS:
|
||
# Windows - caminho UNC direto
|
||
OUTPUT_DIR_BASE = r"\\10.77.77.11\Contabilidade\AUTOMAÇÃO\NotadeDebito"
|
||
else:
|
||
# Linux/Kubernetes - volume montado pelo PV
|
||
OUTPUT_DIR_BASE = "/mnt/contabilidade/AUTOMAÇÃO/NotadeDebito"
|
||
|
||
DB_CONN = (
|
||
"Driver={ODBC Driver 17 for SQL Server};"
|
||
"Server=10.77.77.10;"
|
||
"Database=GINSENG;"
|
||
"UID=andrey;"
|
||
"PWD=88253332Wa@;"
|
||
)
|
||
|
||
# 🔧 Normalizar caminho (acentuação e encoding)
|
||
def normalize_path(path):
|
||
"""Normaliza acentuação e maiúsculas/minúsculas no caminho"""
|
||
return unicodedata.normalize("NFC", path)
|
||
|
||
# 🔧 Verificar acesso ao compartilhamento
|
||
def check_storage_access():
|
||
"""Verifica se o diretório de armazenamento está acessível"""
|
||
print("📁 Verificando acesso ao armazenamento...")
|
||
|
||
# Normalizar o caminho
|
||
base_dir = normalize_path(OUTPUT_DIR_BASE.rsplit(os.sep, 1)[0])
|
||
out_dir = normalize_path(OUTPUT_DIR_BASE)
|
||
|
||
try:
|
||
os.makedirs(out_dir, exist_ok=True)
|
||
print(f"✅ Diretório de saída pronto: {out_dir}")
|
||
return True
|
||
except Exception as e:
|
||
print(f"❌ Erro ao criar diretório {out_dir}: {e}")
|
||
if not IS_WINDOWS:
|
||
print("⚠️ ATENÇÃO: Verifique permissões no /mnt/contabilidade")
|
||
return False
|
||
|
||
# 1️⃣ Token
|
||
def get_token():
|
||
print("🔐 Obtendo token...")
|
||
r = requests.get("https://api.grupoginseng.com.br/api/tokens")
|
||
r.raise_for_status()
|
||
token = r.json()["data"][0]["token"]
|
||
print("✅ Token obtido com sucesso.")
|
||
return token
|
||
|
||
# 2️⃣ Última data sincronizada
|
||
def get_last_invoice_date(cursor):
|
||
cursor.execute("SELECT MAX(invoiceDate) FROM dbo.DebitNotes WHERE invoiceDate IS NOT NULL")
|
||
result = cursor.fetchone()[0]
|
||
if result:
|
||
print(f"🕓 Última data sincronizada: {result}")
|
||
return result
|
||
else:
|
||
default = datetime.now() - timedelta(days=5)
|
||
print(f"⚙️ Nenhuma data encontrada, buscando últimos 30 dias ({default.date()})")
|
||
return default
|
||
|
||
# 3️⃣ Busca com paginação
|
||
def get_all_documents(token):
|
||
print("📡 Consultando notas fiscais (paginado)...")
|
||
url_base = "https://sf-fiscal-api.grupoboticario.digital/v1/debit-notes/documents-list"
|
||
headers = {
|
||
"accept": "application/json, text/plain, */*",
|
||
"authorization": token,
|
||
"content-type": "application/json",
|
||
}
|
||
payload = {
|
||
"franchiseId": [
|
||
"4494", "12522", "12817", "12818", "12820", "12823", "12824", "12826", "12828", "12829",
|
||
"12830", "12838", "13427", "14617", "14668", "19103", "20005", "20006", "20009", "20056",
|
||
"20057", "20441", "20683", "20712", "20858", "20968", "20969", "20970", "20979", "20986",
|
||
"20988", "20989", "20991", "20992", "20993", "20994", "20995", "20996", "20997", "20998",
|
||
"20999", "21000", "21001", "21007", "21068", "21277", "21278", "21296", "21375", "21381",
|
||
"21383", "21495", "21624", "21647", "21737", "21738", "21739", "21740", "22448", "22475",
|
||
"22501", "22526", "22532", "22533", "22541", "22593", "22632", "23156", "23475", "23665",
|
||
"23701", "23702", "23703", "23704", "23705", "23706", "23707", "23708", "23709", "23711",
|
||
"23712", "23713", "3546", "4560", "5699", "910173", "910291", "911486", "911487", "911488",
|
||
"911489", "911490", "911491", "911492", "911509", "911510", "911511", "911512", "911513",
|
||
"911514", "911515", "911516", "911517", "911518", "911519", "911762", "911766", "911924",
|
||
"911940", "912242", "912273", "912856", "912857", "912858", "912859"
|
||
]
|
||
}
|
||
|
||
skip, take = 0, 2000
|
||
all_docs = []
|
||
|
||
while True:
|
||
url = f"{url_base}?cpId=10269&skip={skip}&take={take}"
|
||
print(f"➡️ Página {skip // take + 1} — buscando registros {skip}–{skip + take}...")
|
||
r = requests.post(url, headers=headers, json=payload)
|
||
if r.status_code != 200:
|
||
print(f"❌ Erro {r.status_code}: {r.text}")
|
||
break
|
||
docs = r.json().get("documentsList", [])
|
||
if not docs:
|
||
print("✅ Nenhum novo documento encontrado.")
|
||
break
|
||
all_docs.extend(docs)
|
||
print(f"📦 Recebidos {len(docs)} registros (total: {len(all_docs)})")
|
||
if len(docs) < take:
|
||
break
|
||
skip += take
|
||
return all_docs
|
||
|
||
# 4️⃣ Filtro local por data
|
||
def filter_documents_by_date(documents, start_date, end_date):
|
||
filtrados = []
|
||
for doc in documents:
|
||
try:
|
||
inv_date = datetime.strptime(doc.get("invoiceDate", ""), "%Y-%m-%d")
|
||
if start_date.date() <= inv_date.date() <= end_date.date():
|
||
filtrados.append(doc)
|
||
except Exception:
|
||
continue
|
||
print(f"📅 {len(filtrados)} documentos dentro do intervalo {start_date}–{end_date}")
|
||
return filtrados
|
||
|
||
# 5️⃣ Inserção e download
|
||
def record_exists(cursor, doc_id):
|
||
cursor.execute("SELECT 1 FROM dbo.DebitNotes WHERE id = ?", doc_id)
|
||
return cursor.fetchone() is not None
|
||
|
||
def insert_record(cursor, doc):
|
||
cursor.execute("""
|
||
INSERT INTO dbo.DebitNotes (
|
||
id, debitNoteOrigin, cnpj, cpId, franchiseId, debitNoteNumber,
|
||
invoiceNumber, invoiceItem, invoiceDate, invoiceDueDate, value,
|
||
serviceDescription, imageId, notificationRead, notificationReadDate,
|
||
acknowledged, acknowledgedDate, UUID, imageName, uploadedFile
|
||
)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||
""", (
|
||
doc.get("id"),
|
||
doc.get("debitNoteOrigin") or "N/D",
|
||
doc.get("cnpj") or "",
|
||
doc.get("cpId"),
|
||
doc.get("franchiseId"),
|
||
doc.get("debitNoteNumber") or "",
|
||
doc.get("invoiceNumber") or "",
|
||
doc.get("invoiceItem") or "",
|
||
doc.get("invoiceDate"),
|
||
doc.get("invoiceDueDate"),
|
||
doc.get("value") or 0,
|
||
doc.get("serviceDescription") or "",
|
||
doc.get("imageId"),
|
||
1 if doc.get("notificationRead") else 0,
|
||
doc.get("notificationReadDate"),
|
||
1 if doc.get("acknowledged") else 0,
|
||
doc.get("acknowledgedDate"),
|
||
doc.get("UUID"),
|
||
doc.get("imageName"),
|
||
1 if doc.get("uploadedFile") else 0
|
||
))
|
||
|
||
def download_pdf(token, franchise_id, doc_id, image_name, invoice_date):
|
||
try:
|
||
url = f"https://sf-fiscal-api.grupoboticario.digital/v1/handle-images/NDEB/{franchise_id}/{doc_id}/{image_name}/download"
|
||
r = requests.get(url, headers={"authorization": token})
|
||
r.raise_for_status()
|
||
s3_url = r.text.strip()
|
||
pasta = os.path.join(OUTPUT_DIR_BASE, str(invoice_date))
|
||
os.makedirs(pasta, exist_ok=True)
|
||
file_path = os.path.join(pasta, image_name)
|
||
pdf = requests.get(s3_url, stream=True)
|
||
if pdf.status_code == 200:
|
||
with open(file_path, "wb") as f:
|
||
for chunk in pdf.iter_content(8192):
|
||
f.write(chunk)
|
||
print(f"📥 PDF salvo: {file_path}")
|
||
except Exception as e:
|
||
print(f"❌ Erro ao baixar {image_name}: {e}")
|
||
|
||
# 🚀 Main
|
||
def main():
|
||
# Verificar acesso ao armazenamento
|
||
if not check_storage_access():
|
||
print("❌ Não foi possível acessar o armazenamento. Abortando.")
|
||
return
|
||
|
||
try:
|
||
token = get_token()
|
||
conn = pyodbc.connect(DB_CONN)
|
||
cursor = conn.cursor()
|
||
|
||
# Filtrar apenas notas do dia 27-10-2025
|
||
data_especifica = datetime(2025, 10, 27)
|
||
todos = get_all_documents(token)
|
||
documentos = filter_documents_by_date(todos, data_especifica, data_especifica)
|
||
|
||
novos, ignorados = 0, 0
|
||
for doc in documentos:
|
||
if record_exists(cursor, doc["id"]):
|
||
ignorados += 1
|
||
continue
|
||
insert_record(cursor, doc)
|
||
novos += 1
|
||
download_pdf(token, doc["franchiseId"], doc["id"], doc["imageName"], doc["invoiceDate"])
|
||
|
||
conn.commit()
|
||
conn.close()
|
||
print(f"\n✅ Finalizado. {novos} novos registros, {ignorados} já existiam.")
|
||
print(f"🕓 Execução concluída em {datetime.now().strftime('%H:%M:%S')}")
|
||
|
||
except Exception as e:
|
||
print(f"❌ Erro durante execução: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
|
||
if __name__ == "__main__":
|
||
main()
|