1246 lines
44 KiB
Python
1246 lines
44 KiB
Python
import requests
|
||
import pdfplumber
|
||
import base64
|
||
import json
|
||
import time
|
||
import re
|
||
import unicodedata
|
||
import threading
|
||
from concurrent.futures import ThreadPoolExecutor, as_completed
|
||
from datetime import datetime
|
||
from io import BytesIO
|
||
from dataclasses import dataclass
|
||
from typing import Any, Dict, List, Optional, Set, Tuple
|
||
|
||
TOKENS_URL = "https://api.grupoginseng.com.br/api/tokens"
|
||
STORES_URL = "https://api-extranet.grupoboticario.digital/api/person-logged/stores"
|
||
DOCS_URL = "https://sf-relatorios-api.grupoboticario.digital/v1/reports/documents-list"
|
||
HANDLE_URL = "https://sf-relatorios-api.grupoboticario.digital/v1/handle-images"
|
||
|
||
|
||
# -----------------------------
|
||
# TOKEN (com cache simples)
|
||
# -----------------------------
|
||
def _jwt_payload(jwt_token: str) -> Dict[str, Any]:
|
||
parts = jwt_token.split(".")
|
||
if len(parts) != 3:
|
||
return {}
|
||
payload_b64 = parts[1] + "=" * (-len(parts[1]) % 4)
|
||
raw = base64.urlsafe_b64decode(payload_b64.encode("utf-8"))
|
||
return json.loads(raw.decode("utf-8"))
|
||
|
||
|
||
@dataclass
|
||
class TokenCache:
|
||
bearer: Optional[str] = None
|
||
exp_epoch: int = 0
|
||
|
||
def valid(self, skew_seconds: int = 30) -> bool:
|
||
return bool(self.bearer) and (time.time() < (self.exp_epoch - skew_seconds))
|
||
|
||
|
||
class Auth:
|
||
def __init__(self, s: requests.Session):
|
||
self.s = s
|
||
self.cache = TokenCache()
|
||
|
||
def get_bearer(self) -> str:
|
||
if self.cache.valid():
|
||
return self.cache.bearer # type: ignore
|
||
|
||
r = self.s.get(TOKENS_URL, timeout=30)
|
||
r.raise_for_status()
|
||
body = r.json()
|
||
|
||
if not body.get("success"):
|
||
raise RuntimeError(f"Token API retornou success=false: {body}")
|
||
|
||
bearer = body["data"][0]["token"] # já vem "Bearer ..."
|
||
jwt = bearer.split(" ", 1)[1] if bearer.lower().startswith("bearer ") else bearer
|
||
exp = int(_jwt_payload(jwt).get("exp") or 0)
|
||
|
||
self.cache = TokenCache(bearer=bearer, exp_epoch=exp)
|
||
return bearer
|
||
|
||
def invalidate(self) -> None:
|
||
self.cache = TokenCache()
|
||
|
||
|
||
# -----------------------------
|
||
# PDF: download + leitura texto (fallback)
|
||
# -----------------------------
|
||
def baixar_pdf_bytes(url_pdf: str, session: requests.Session) -> BytesIO:
|
||
r = session.get(url_pdf, timeout=120)
|
||
r.raise_for_status()
|
||
return BytesIO(r.content)
|
||
|
||
|
||
def ler_texto_pdf(pdf_bytes: BytesIO) -> str:
|
||
texto_total = ""
|
||
with pdfplumber.open(pdf_bytes) as pdf:
|
||
for pagina in pdf.pages:
|
||
texto = pagina.extract_text(layout=True) or ""
|
||
if texto:
|
||
texto_total += texto + "\n"
|
||
return texto_total
|
||
|
||
|
||
# -----------------------------
|
||
# EXTRAÇÃO: Remuneração Franquia
|
||
# 1) tenta por TABELA (mais confiável)
|
||
# 2) se falhar, cai no TEXTO (como antes)
|
||
# -----------------------------
|
||
def _norm(s: str) -> str:
|
||
base = _strip_accents(s or "")
|
||
return re.sub(r"\s+", " ", base).strip().lower()
|
||
|
||
|
||
def _parse_money_pt(s: str) -> float:
|
||
s = (s or "").strip()
|
||
# "12.769,43" -> "12769.43"
|
||
if "," in s:
|
||
s = s.replace(".", "").replace(",", ".")
|
||
return float(s)
|
||
|
||
|
||
def _money_from_text(s: str) -> Optional[float]:
|
||
if not s:
|
||
return None
|
||
m = re.search(r"\d{1,3}(?:\.\d{3})*,\d{2}|\d+\.\d{2}", s)
|
||
if not m:
|
||
return None
|
||
try:
|
||
return _parse_money_pt(m.group(0))
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
def _float5_from_text(s: str) -> Optional[float]:
|
||
if not s:
|
||
return None
|
||
m = re.search(r"\d+(?:[.,])\d{5}", s) # 1,00000 / 1.00000
|
||
if not m:
|
||
return None
|
||
x = m.group(0)
|
||
x = x.replace(".", "").replace(",", ".") if "," in x else x
|
||
try:
|
||
return float(x)
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
def _pick_first_number(s: str) -> Optional[str]:
|
||
if not s:
|
||
return None
|
||
chunks = re.findall(r"\d+", s)
|
||
if not chunks:
|
||
return None
|
||
|
||
# Em alguns PDFs a NF vem quebrada em linhas, ex: "12282" + "58".
|
||
# Se houver um sufixo curto após um bloco maior, recompõe.
|
||
if len(chunks) >= 2 and len(chunks[-2]) >= 3 and len(chunks[-1]) <= 3:
|
||
return chunks[-2] + chunks[-1]
|
||
|
||
for c in chunks:
|
||
if len(c) >= 3:
|
||
return c
|
||
return None
|
||
|
||
|
||
def _extract_date_pt(s: str) -> Optional[str]:
|
||
if not s:
|
||
return None
|
||
s = re.sub(r"(?<=\d)\s+(?=\d)", "", s)
|
||
m = re.search(r"\d{2}\.\d{2}\.\d{4}", s)
|
||
return m.group(0) if m else None
|
||
|
||
|
||
def _nota_from_link_text(s: str) -> Optional[str]:
|
||
if not s:
|
||
return None
|
||
# querystring da URL da NFS-e
|
||
m = re.search(r"(?:nrnfs|nnfse)\s*=\s*(\d+)", s, flags=re.IGNORECASE)
|
||
return m.group(1) if m else None
|
||
|
||
|
||
def _nota_from_cell_text(s: str) -> Optional[str]:
|
||
if not s:
|
||
return None
|
||
txt = " ".join(str(s).split())
|
||
# Evita confundir com valor monetario (ex: 1315.42) e aceita sufixo "-1".
|
||
m = re.search(r"(?<![\d])(\d{3,}(?:-\d{1,4})?)(?![.,]\d)", txt)
|
||
if not m:
|
||
return None
|
||
|
||
base = m.group(1)
|
||
# Em alguns PDFs a Nota Fiscal quebra em 2 linhas (ex: "2025000012854" + "90").
|
||
if "-" not in base:
|
||
mt = re.search(re.escape(base) + r"\s+(\d{1,4})(?!\d)", txt)
|
||
if mt:
|
||
return base + mt.group(1)
|
||
return base
|
||
|
||
|
||
def _nota_digits_len(v: Optional[str]) -> int:
|
||
if not v:
|
||
return 0
|
||
return len(re.sub(r"\D", "", str(v)))
|
||
|
||
|
||
def _normalizar_nota_fiscal(v: Optional[str]) -> Optional[str]:
|
||
if not v:
|
||
return None
|
||
digits = re.sub(r"\D", "", str(v))
|
||
if not digits:
|
||
return None
|
||
# Regra de negocio: manter somente os ultimos 9 digitos.
|
||
return digits[-9:] if len(digits) > 9 else digits
|
||
|
||
|
||
def _nota_ou_dps(nota: Optional[str], nota_link: Optional[str], dps_num: Optional[str]) -> Optional[str]:
|
||
cands = [_normalizar_nota_fiscal(nota), _normalizar_nota_fiscal(nota_link), _normalizar_nota_fiscal(dps_num)]
|
||
valid = [c for c in cands if _nota_fiscal_valida(c)]
|
||
if valid:
|
||
# Se houver mais de uma opção, prefere a mais completa (mais dígitos).
|
||
return max(valid, key=_nota_digits_len)
|
||
for c in cands:
|
||
if c:
|
||
return c
|
||
return None
|
||
|
||
|
||
def _dps_num_from_text(s: str) -> Optional[str]:
|
||
if not s:
|
||
return None
|
||
m = re.search(
|
||
r"(?:dados\s*dps|dps).{0,80}?\bnum(?:ero|\.?)\s*[:=]\s*(\d{3,})\b",
|
||
s,
|
||
flags=re.IGNORECASE | re.DOTALL,
|
||
)
|
||
return m.group(1) if m else None
|
||
|
||
def _nota_fiscal_valida(nota: Optional[str]) -> bool:
|
||
if not nota:
|
||
return False
|
||
digits = re.sub(r"\D", "", str(nota))
|
||
# Evita ano (ex.: "2026") e outros falsos positivos curtos.
|
||
return len(digits) >= 5
|
||
|
||
|
||
def _extrair_parcelas(dp_text: str, emissao: Optional[str] = None) -> List[Dict[str, Any]]:
|
||
pares = re.findall(
|
||
r"\b(\d{2}\.\d{2}\.\d{4})\s+(\d{1,3}(?:\.\d{3})*,\d{2}|\d+\.\d{2})\b",
|
||
dp_text or "",
|
||
)
|
||
parcelas: List[Dict[str, Any]] = []
|
||
seen = set()
|
||
for data, v in pares:
|
||
try:
|
||
val = _parse_money_pt(v)
|
||
except Exception:
|
||
continue
|
||
k = (data, val)
|
||
if k not in seen:
|
||
seen.add(k)
|
||
parcelas.append({"data": data, "valor": val})
|
||
|
||
# Quando a coluna vem mesclada, pode capturar "emissao + desconto 0,00" como 1a parcela.
|
||
if emissao and len(parcelas) > 1:
|
||
parcelas = [
|
||
p for p in parcelas
|
||
if not (p.get("data") == emissao and abs(float(p.get("valor") or 0.0)) < 0.00001)
|
||
]
|
||
|
||
return _ordenar_parcelas_por_data(parcelas)
|
||
|
||
|
||
def _date_from_iso(s: Optional[str]):
|
||
if not s:
|
||
return None
|
||
try:
|
||
return datetime.strptime(s, "%Y-%m-%d").date()
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
def _date_from_br(s: Optional[str]):
|
||
if not s:
|
||
return None
|
||
try:
|
||
return datetime.strptime(s, "%d.%m.%Y").date()
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
def _ordenar_parcelas_por_data(parcelas: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
|
||
"""
|
||
Ordena parcelas por data (dd.mm.aaaa) e, em empate, por valor.
|
||
Parcelas sem data válida ficam no final.
|
||
"""
|
||
def _k(p: Dict[str, Any]):
|
||
dt = _date_from_br(p.get("data"))
|
||
val = p.get("valor")
|
||
val_num = float(val) if isinstance(val, (int, float)) else float("inf")
|
||
invalida = dt is None
|
||
return (invalida, dt or datetime.max.date(), val_num)
|
||
|
||
return sorted(parcelas or [], key=_k)
|
||
|
||
|
||
def _isolar_bloco(texto: str, titulo_regex: str, proximos_titulos_regex: List[str]) -> Optional[str]:
|
||
t = " ".join((texto or "").split())
|
||
m0 = re.search(titulo_regex, t, flags=re.IGNORECASE)
|
||
if not m0:
|
||
return None
|
||
after = t[m0.start():]
|
||
|
||
end_pos = None
|
||
for rx in proximos_titulos_regex:
|
||
m = re.search(rx, after, flags=re.IGNORECASE)
|
||
if m:
|
||
end_pos = m.start()
|
||
break
|
||
|
||
return after[:end_pos] if end_pos is not None else after
|
||
|
||
|
||
def extrair_remuneracao_franquia_por_tabela(pdf: pdfplumber.PDF) -> Optional[Dict[str, Any]]:
|
||
"""
|
||
Extrai APENAS a tabela do bloco:
|
||
"Notas Fiscais de Serviço de taxa de remuneração de Franquia"
|
||
|
||
Regra para não confundir com "Venda de Produtos":
|
||
- o header TEM que ter "valor taxa" e "fat.conv"
|
||
"""
|
||
for page in pdf.pages:
|
||
tables = page.extract_tables() or []
|
||
for tbl in tables:
|
||
if not tbl or len(tbl) < 2:
|
||
continue
|
||
|
||
# procura um header que tenha as colunas especÃÂÂÂficas do bloco de remuneração
|
||
header_idx = None
|
||
header_join = ""
|
||
for i, row in enumerate(tbl[:8]):
|
||
header_join = " | ".join(_norm(c) for c in row if c)
|
||
# precisa ter essas chaves (venda não tem "valor taxa")
|
||
if ("valor taxa" in header_join) and ("fat" in header_join) and ("dados pagamento" in header_join):
|
||
header_idx = i
|
||
break
|
||
|
||
if header_idx is None:
|
||
continue
|
||
|
||
nota_link_tabela = _nota_from_link_text(_table_text(tbl))
|
||
header = [_norm(c) for c in tbl[header_idx]]
|
||
|
||
def idx_like(keys: List[str]) -> Optional[int]:
|
||
for j, h in enumerate(header):
|
||
for k in keys:
|
||
if k in h:
|
||
return j
|
||
return None
|
||
|
||
i_emissao = idx_like(["emissao"])
|
||
i_nota = idx_like(["nota fiscal"])
|
||
i_valor_taxa = idx_like(["valor taxa"])
|
||
i_valor_desc = idx_like(["valor desconto"])
|
||
i_fat = idx_like(["fat.conv", "fat conv", "fat"])
|
||
i_valor_nf = idx_like(["valor nf"])
|
||
i_enc = idx_like(["encargos"])
|
||
i_dp = idx_like(["dados pagamento"])
|
||
|
||
# essenciais do bloco
|
||
if i_emissao is None or i_nota is None or i_valor_taxa is None or i_valor_nf is None or i_dp is None:
|
||
continue
|
||
|
||
# lê a primeira linha "real" de dados (normalmente só tem 1)
|
||
for row in tbl[header_idx + 1:]:
|
||
if not any((c or "").strip() for c in row):
|
||
continue
|
||
|
||
emissao_cell = row[i_emissao] if i_emissao < len(row) else ""
|
||
nota_cell = row[i_nota] if i_nota < len(row) else ""
|
||
dp_cell = _cell(row[i_dp]) if i_dp < len(row) else ""
|
||
|
||
emissao = _extract_date_pt(_cell(emissao_cell))
|
||
nota = _nota_from_cell_text(_cell(nota_cell))
|
||
nota_link = _nota_from_link_text(dp_cell)
|
||
dps_num = _dps_num_from_text(dp_cell)
|
||
nota_fiscal = _nota_ou_dps(nota, nota_link or nota_link_tabela, dps_num)
|
||
|
||
if not emissao:
|
||
continue
|
||
|
||
valor_taxa = _money_from_text(row[i_valor_taxa]) if i_valor_taxa < len(row) else None
|
||
valor_desconto = _money_from_text(row[i_valor_desc]) if i_valor_desc is not None and i_valor_desc < len(row) else None
|
||
fat_conv = _float5_from_text(row[i_fat]) if i_fat is not None and i_fat < len(row) else None
|
||
valor_nf = _money_from_text(row[i_valor_nf]) if i_valor_nf < len(row) else None
|
||
encargos = _money_from_text(row[i_enc]) if i_enc is not None and i_enc < len(row) else None
|
||
|
||
# Dados pagamento: várias linhas data + valor
|
||
dp = dp_cell
|
||
parcelas = _extrair_parcelas(dp, emissao=emissao)
|
||
tem_valor_bloco = any(v is not None for v in (valor_taxa, valor_nf, encargos)) or bool(parcelas)
|
||
|
||
# Só considera TRF quando o bloco tem preenchimento real.
|
||
if (not _nota_fiscal_valida(nota_fiscal)) or (not tem_valor_bloco):
|
||
continue
|
||
|
||
return {
|
||
"tipo_bloco": "TAXA_REMUNERACAO",
|
||
"emissao": emissao,
|
||
"nota_fiscal": nota_fiscal,
|
||
"valor_taxa": valor_taxa,
|
||
"valor_desconto": valor_desconto,
|
||
"fat_conv": fat_conv,
|
||
"valor_nf": valor_nf,
|
||
"encargos": encargos,
|
||
"parcelas": parcelas,
|
||
"error": None,
|
||
}
|
||
|
||
return None
|
||
|
||
|
||
def _cell(c: Any) -> str:
|
||
return " ".join(str(c or "").split())
|
||
|
||
|
||
def _table_text(tbl: List[List[Any]]) -> str:
|
||
return " | ".join(_norm(_cell(c)) for row in tbl for c in (row or []))
|
||
|
||
|
||
def _strip_accents(s: str) -> str:
|
||
return "".join(ch for ch in unicodedata.normalize("NFD", s or "") if unicodedata.category(ch) != "Mn")
|
||
|
||
|
||
def extrair_remuneracao_franquia_por_texto(texto: str) -> Dict[str, Any]:
|
||
"""
|
||
Fallback (como você já fazia): acha a seção por regex e tenta extrair
|
||
emissão, nota e parcelas. (Os valores podem falhar dependendo do PDF.)
|
||
"""
|
||
bloco = _isolar_bloco(
|
||
texto,
|
||
titulo_regex=r"Notas\s+Fiscais\s+de\s+Servi.{0,3}o\s+de\s+taxa\s+de\s+remunera.{0,5}o\s+de\s+Franquia",
|
||
proximos_titulos_regex=[
|
||
r"\bLink\s*:",
|
||
r"Notas\s+de\s+Despesas\s+de\s+Propaganda",
|
||
r"Outras\s+Notas\s+de\s+D[eé]bito",
|
||
r"Outras\s+Notas\s+de\s+Cr[eé]dito",
|
||
],
|
||
)
|
||
if not bloco:
|
||
return {
|
||
"tipo_bloco": "TAXA_REMUNERACAO",
|
||
"emissao": None,
|
||
"nota_fiscal": None,
|
||
"valor_taxa": None,
|
||
"valor_desconto": None,
|
||
"fat_conv": None,
|
||
"valor_nf": None,
|
||
"encargos": None,
|
||
"parcelas": [],
|
||
"error": "Seção não encontrada",
|
||
}
|
||
|
||
b = " ".join(bloco.split())
|
||
|
||
m_em = re.search(r"\b(\d{2}\.\d{2}\.\d{4})\b", b)
|
||
if not m_em:
|
||
return {
|
||
"tipo_bloco": "TAXA_REMUNERACAO",
|
||
"emissao": None,
|
||
"nota_fiscal": None,
|
||
"valor_taxa": None,
|
||
"valor_desconto": None,
|
||
"fat_conv": None,
|
||
"valor_nf": None,
|
||
"encargos": None,
|
||
"parcelas": [],
|
||
"error": "Não achei emissão + nota",
|
||
}
|
||
|
||
emissao = m_em.group(1)
|
||
m_nota = re.search(r"\bnota\s+fiscal\b.{0,220}?\b(\d{3,}(?:-\d{1,4})?)\b", b, flags=re.IGNORECASE)
|
||
nota = m_nota.group(1) if m_nota else None
|
||
if not nota:
|
||
m_nota2 = re.search(r"\b\d{2}\.\d{2}\.\d{4}\b\s+(\d{3,}(?:-\d{1,4})?)\b", b, flags=re.IGNORECASE)
|
||
nota = m_nota2.group(1) if m_nota2 else None
|
||
nota_link = _nota_from_link_text(texto)
|
||
dps_num = _dps_num_from_text(texto)
|
||
nota_fiscal = _nota_ou_dps(nota, nota_link, dps_num)
|
||
|
||
# parcelas depois de "Dados Pagamento"
|
||
idx_dp = b.lower().find("dados pagamento")
|
||
dp = b[idx_dp:] if idx_dp != -1 else b
|
||
|
||
parcelas = _extrair_parcelas(dp, emissao=emissao)
|
||
valor_taxa = _money_from_text(re.search(r"valor\s+taxa.{0,80}", b, flags=re.IGNORECASE).group(0)) if re.search(r"valor\s+taxa.{0,80}", b, flags=re.IGNORECASE) else None
|
||
valor_nf = _money_from_text(re.search(r"valor\s+nf.{0,80}", b, flags=re.IGNORECASE).group(0)) if re.search(r"valor\s+nf.{0,80}", b, flags=re.IGNORECASE) else None
|
||
tem_valor_bloco = any(v is not None for v in (valor_taxa, valor_nf)) or bool(parcelas)
|
||
if (not _nota_fiscal_valida(nota_fiscal)) or (not tem_valor_bloco):
|
||
return {
|
||
"tipo_bloco": "TAXA_REMUNERACAO",
|
||
"emissao": None,
|
||
"nota_fiscal": None,
|
||
"valor_taxa": None,
|
||
"valor_desconto": None,
|
||
"fat_conv": None,
|
||
"valor_nf": None,
|
||
"encargos": None,
|
||
"parcelas": [],
|
||
"error": "Bloco TRF sem preenchimento valido",
|
||
}
|
||
|
||
# aqui você pode manter os valores como None no fallback
|
||
return {
|
||
"tipo_bloco": "TAXA_REMUNERACAO",
|
||
"emissao": emissao,
|
||
"nota_fiscal": nota_fiscal,
|
||
"valor_taxa": None,
|
||
"valor_desconto": None,
|
||
"fat_conv": None,
|
||
"valor_nf": None,
|
||
"encargos": None,
|
||
"parcelas": parcelas,
|
||
"error": None,
|
||
}
|
||
|
||
|
||
def extrair_remuneracao_franquia(pdf_bytes: BytesIO) -> Dict[str, Any]:
|
||
"""
|
||
Função única que você chama no seu fluxo:
|
||
- tenta tabela (melhor)
|
||
- se não conseguir, cai no texto (como era)
|
||
"""
|
||
# tenta bloco principal por tabela (melhor cenário)
|
||
with pdfplumber.open(pdf_bytes) as pdf:
|
||
via_tabela = extrair_remuneracao_franquia_por_tabela(pdf)
|
||
if via_tabela:
|
||
return via_tabela
|
||
|
||
# fallback texto do bloco principal
|
||
pdf_bytes.seek(0)
|
||
texto = ler_texto_pdf(pdf_bytes)
|
||
via_texto = extrair_remuneracao_franquia_por_texto(texto)
|
||
if via_texto.get("emissao"):
|
||
return via_texto
|
||
|
||
return via_texto
|
||
|
||
|
||
def _tem_emissao_nota(d: Optional[Dict[str, Any]]) -> bool:
|
||
if not d:
|
||
return False
|
||
return bool(d.get("emissao")) and bool(d.get("nota_fiscal"))
|
||
|
||
|
||
def _dados_bloco(d: Dict[str, Any]) -> Dict[str, Any]:
|
||
return {
|
||
"emissao_nf": d.get("emissao"),
|
||
"nota_fiscal": d.get("nota_fiscal"),
|
||
"valor_taxa": d.get("valor_taxa"),
|
||
"valor_desconto": d.get("valor_desconto"),
|
||
"fat_conv": d.get("fat_conv"),
|
||
"valor_nf": d.get("valor_nf"),
|
||
"encargos": d.get("encargos"),
|
||
"parcelas": d.get("parcelas") or [],
|
||
}
|
||
|
||
|
||
def extrair_remuneracao_franquia_detalhado(pdf_bytes: BytesIO) -> Dict[str, Any]:
|
||
principal: Optional[Dict[str, Any]] = None
|
||
|
||
pdf_bytes.seek(0)
|
||
with pdfplumber.open(pdf_bytes) as pdf:
|
||
via_tabela = extrair_remuneracao_franquia_por_tabela(pdf)
|
||
if via_tabela:
|
||
principal = via_tabela
|
||
|
||
if not principal:
|
||
pdf_bytes.seek(0)
|
||
texto = ler_texto_pdf(pdf_bytes)
|
||
principal = extrair_remuneracao_franquia_por_texto(texto)
|
||
|
||
principal_ok = _tem_emissao_nota(principal)
|
||
escolhido = principal
|
||
if not escolhido:
|
||
escolhido = {
|
||
"tipo_bloco": "TAXA_REMUNERACAO",
|
||
"emissao": None,
|
||
"nota_fiscal": None,
|
||
"valor_taxa": None,
|
||
"valor_desconto": None,
|
||
"fat_conv": None,
|
||
"valor_nf": None,
|
||
"encargos": None,
|
||
"parcelas": [],
|
||
"error": "Falha na extração",
|
||
}
|
||
|
||
warnings: List[str] = []
|
||
if principal and principal.get("error"):
|
||
warnings.append(str(principal.get("error")))
|
||
|
||
blocos: List[Dict[str, Any]] = []
|
||
if principal:
|
||
blocos.append({
|
||
"tipo": "TAXA_REMUNERACAO",
|
||
"encontrado": principal_ok,
|
||
"dados": _dados_bloco(principal) if principal_ok else None,
|
||
"erro": None if principal_ok else principal.get("error"),
|
||
})
|
||
|
||
return {
|
||
"bloco_principal": "TAXA_REMUNERACAO",
|
||
"status_principal": "ok" if principal_ok else "sem_dados",
|
||
"bloco_utilizado": escolhido.get("tipo_bloco"),
|
||
"resumo": {
|
||
"emissao_nf": escolhido.get("emissao"),
|
||
"nota_fiscal": escolhido.get("nota_fiscal"),
|
||
"valor_nf": escolhido.get("valor_nf"),
|
||
"encargos": escolhido.get("encargos"),
|
||
},
|
||
"blocos": blocos,
|
||
"warnings": warnings,
|
||
"escolhido": escolhido, # compatibilidade com estrutura atual
|
||
}
|
||
|
||
|
||
# -----------------------------
|
||
# CLIENTE PRINCIPAL
|
||
# -----------------------------
|
||
class Client:
|
||
def __init__(self):
|
||
self.s = requests.Session()
|
||
self.auth = Auth(self.s)
|
||
self._worker_local = threading.local()
|
||
|
||
def _headers_json(self) -> Dict[str, str]:
|
||
return {
|
||
"Authorization": self.auth.get_bearer(),
|
||
"Accept": "application/json",
|
||
"Content-Type": "application/json",
|
||
"Origin": "https://extranet.grupoboticario.com.br",
|
||
"Referer": "https://extranet.grupoboticario.com.br/",
|
||
"User-Agent": "Mozilla/5.0",
|
||
}
|
||
|
||
def get_franchises(self, only_channels: Optional[Set[str]] = None) -> List[str]:
|
||
r = self.s.get(STORES_URL, headers=self._headers_json(), timeout=30)
|
||
if r.status_code in (401, 403):
|
||
self.auth.invalidate()
|
||
r = self.s.get(STORES_URL, headers=self._headers_json(), timeout=30)
|
||
r.raise_for_status()
|
||
|
||
body = r.json()
|
||
seen = set()
|
||
codes = []
|
||
|
||
for item in body.get("data", []):
|
||
if only_channels is not None and item.get("channel") not in only_channels:
|
||
continue
|
||
code = item.get("code")
|
||
if code is None:
|
||
continue
|
||
code = str(code).strip()
|
||
if code and code not in seen:
|
||
seen.add(code)
|
||
codes.append(code)
|
||
|
||
return codes
|
||
|
||
def get_documents_page(self, cp_id: int, document_type: str, franchises: List[str], offset: int, limit: int):
|
||
params = {"cpId": cp_id, "documentType": document_type, "offset": offset, "limit": limit}
|
||
payload = {"franchises": franchises}
|
||
|
||
r = self.s.post(DOCS_URL, headers=self._headers_json(), params=params, json=payload, timeout=60)
|
||
if r.status_code in (401, 403):
|
||
self.auth.invalidate()
|
||
r = self.s.post(DOCS_URL, headers=self._headers_json(), params=params, json=payload, timeout=60)
|
||
|
||
r.raise_for_status()
|
||
return r.json() # {"total":..., "documents":[...]}
|
||
|
||
def get_presigned_pdf_url(self, document_type: str, franchise_id: str, image_name: str) -> str:
|
||
url = f"{HANDLE_URL}/{document_type}/{franchise_id}/{image_name}/download"
|
||
|
||
headers = {
|
||
"Authorization": self.auth.get_bearer(),
|
||
"Accept": "application/json",
|
||
"Origin": "https://extranet.grupoboticario.com.br",
|
||
"Referer": "https://extranet.grupoboticario.com.br/",
|
||
"User-Agent": "Mozilla/5.0",
|
||
}
|
||
|
||
r = self.s.get(url, headers=headers, timeout=60)
|
||
if r.status_code in (401, 403):
|
||
self.auth.invalidate()
|
||
headers["Authorization"] = self.auth.get_bearer()
|
||
r = self.s.get(url, headers=headers, timeout=60)
|
||
|
||
r.raise_for_status()
|
||
|
||
# pode vir JSON com {"url": "..."} ou texto puro
|
||
try:
|
||
body = r.json()
|
||
if isinstance(body, dict) and "url" in body and isinstance(body["url"], str):
|
||
return body["url"]
|
||
except Exception:
|
||
pass
|
||
|
||
return r.text.strip()
|
||
|
||
@staticmethod
|
||
def _headers_with_auth(auth: Auth) -> Dict[str, str]:
|
||
return {
|
||
"Authorization": auth.get_bearer(),
|
||
"Accept": "application/json",
|
||
"Origin": "https://extranet.grupoboticario.com.br",
|
||
"Referer": "https://extranet.grupoboticario.com.br/",
|
||
"User-Agent": "Mozilla/5.0",
|
||
}
|
||
|
||
def _get_presigned_pdf_url_with_client(
|
||
self,
|
||
session: requests.Session,
|
||
auth: Auth,
|
||
document_type: str,
|
||
franchise_id: str,
|
||
image_name: str,
|
||
) -> str:
|
||
url = f"{HANDLE_URL}/{document_type}/{franchise_id}/{image_name}/download"
|
||
headers = self._headers_with_auth(auth)
|
||
|
||
r = session.get(url, headers=headers, timeout=60)
|
||
if r.status_code in (401, 403):
|
||
auth.invalidate()
|
||
headers["Authorization"] = auth.get_bearer()
|
||
r = session.get(url, headers=headers, timeout=60)
|
||
r.raise_for_status()
|
||
|
||
try:
|
||
body = r.json()
|
||
if isinstance(body, dict) and "url" in body and isinstance(body["url"], str):
|
||
return body["url"]
|
||
except Exception:
|
||
pass
|
||
return r.text.strip()
|
||
|
||
def _get_worker_http(self) -> Tuple[requests.Session, Auth]:
|
||
session = getattr(self._worker_local, "session", None)
|
||
auth = getattr(self._worker_local, "auth", None)
|
||
if session is None or auth is None:
|
||
session = requests.Session()
|
||
auth = Auth(session)
|
||
self._worker_local.session = session
|
||
self._worker_local.auth = auth
|
||
return session, auth
|
||
|
||
def _processar_documento(self, d: Dict[str, Any], document_type: str) -> Dict[str, Any]:
|
||
franchise_id = str(d.get("franchiseId") or "").strip()
|
||
image_name = str(d.get("imageName") or "").strip()
|
||
|
||
base_item = {
|
||
"id": d.get("id"),
|
||
"UUID": d.get("UUID"),
|
||
"franchiseId": franchise_id,
|
||
"imageName": image_name,
|
||
"emissionDate": d.get("emissionDate"),
|
||
}
|
||
|
||
if not franchise_id or not image_name:
|
||
return {**base_item, "error": "sem franchiseId/imageName"}
|
||
|
||
session, auth = self._get_worker_http()
|
||
try:
|
||
presigned = self._get_presigned_pdf_url_with_client(
|
||
session, auth, document_type, franchise_id, image_name
|
||
)
|
||
pdf_bytes = baixar_pdf_bytes(presigned, session)
|
||
extra_detalhado = extrair_remuneracao_franquia_detalhado(pdf_bytes)
|
||
extra = extra_detalhado.get("escolhido", {})
|
||
status_principal = extra_detalhado.get("status_principal")
|
||
resumo = extra_detalhado.get("resumo") or {}
|
||
emissao_nf = resumo.get("emissao_nf")
|
||
nota_fiscal = resumo.get("nota_fiscal")
|
||
bloco_ok = bool(status_principal == "ok" and emissao_nf and nota_fiscal)
|
||
|
||
return {
|
||
**base_item,
|
||
"bloco_principal": extra_detalhado.get("bloco_principal"),
|
||
"status_principal": status_principal,
|
||
"bloco_utilizado": extra_detalhado.get("bloco_utilizado"),
|
||
"resumo": resumo,
|
||
"blocos": extra_detalhado.get("blocos"),
|
||
"warnings": extra_detalhado.get("warnings"),
|
||
"tipo_bloco": extra.get("tipo_bloco"),
|
||
"emissao_nf": extra.get("emissao"),
|
||
"nota_fiscal": extra.get("nota_fiscal"),
|
||
"valor_taxa": extra.get("valor_taxa"),
|
||
"valor_desconto": extra.get("valor_desconto"),
|
||
"fat_conv": extra.get("fat_conv"),
|
||
"valor_nf": extra.get("valor_nf"),
|
||
"encargos": extra.get("encargos"),
|
||
"parcelas": extra.get("parcelas"),
|
||
"error": None if bloco_ok else "skip_sem_bloco_taxa_remuneracao",
|
||
}
|
||
except Exception as e:
|
||
return {**base_item, "error": f"falha_processar_pdf: {e}"}
|
||
def processar_pagina(
|
||
self,
|
||
cp_id: int = 10269,
|
||
document_type: str = "EFAT",
|
||
offset: int = 0,
|
||
limit: int = 25,
|
||
only_channels: Optional[Set[str]] = None,
|
||
max_workers: int = 6,
|
||
) -> Dict[str, Any]:
|
||
franchises = self.get_franchises(only_channels=only_channels)
|
||
|
||
page = self.get_documents_page(cp_id, document_type, franchises, offset, limit)
|
||
docs = page.get("documents", [])
|
||
total = int(page.get("total") or 0)
|
||
|
||
out: List[Dict[str, Any]] = []
|
||
if docs:
|
||
workers = max(1, min(int(max_workers or 1), len(docs)))
|
||
progress_step = max(1, min(10, len(docs) // 10 if len(docs) > 10 else 1))
|
||
if workers == 1:
|
||
out = []
|
||
for i, d in enumerate(docs, start=1):
|
||
out.append(self._processar_documento(d, document_type))
|
||
if i % progress_step == 0 or i == len(docs):
|
||
print(f"[proc] offset={offset} processados={i}/{len(docs)}")
|
||
else:
|
||
ordered: List[Optional[Dict[str, Any]]] = [None] * len(docs)
|
||
with ThreadPoolExecutor(max_workers=workers) as ex:
|
||
futures = {
|
||
ex.submit(self._processar_documento, d, document_type): i
|
||
for i, d in enumerate(docs)
|
||
}
|
||
done = 0
|
||
for fut in as_completed(futures):
|
||
i = futures[fut]
|
||
try:
|
||
ordered[i] = fut.result()
|
||
except Exception as e:
|
||
ordered[i] = {
|
||
"id": docs[i].get("id"),
|
||
"UUID": docs[i].get("UUID"),
|
||
"franchiseId": str(docs[i].get("franchiseId") or "").strip(),
|
||
"imageName": str(docs[i].get("imageName") or "").strip(),
|
||
"emissionDate": docs[i].get("emissionDate"),
|
||
"error": f"falha_processar_pdf: {e}",
|
||
}
|
||
done += 1
|
||
if done % progress_step == 0 or done == len(docs):
|
||
print(f"[proc] offset={offset} processados={done}/{len(docs)}")
|
||
out = [x for x in ordered if x is not None]
|
||
|
||
return {
|
||
"total": total,
|
||
"offset": offset,
|
||
"limit": limit,
|
||
"count": len(docs),
|
||
"hasNext": (offset + limit) < total,
|
||
"items": out
|
||
}
|
||
|
||
|
||
# -----------------------------
|
||
# SQL SERVER: persistência
|
||
# -----------------------------
|
||
class SqlServerSink:
|
||
def __init__(self, connection_string: str):
|
||
self.connection_string = connection_string
|
||
self.cn = None
|
||
self.cur = None
|
||
self.doc_table = "dbo.TrfRemDocumento"
|
||
self.parcela_table = "dbo.TrfRemParcela"
|
||
|
||
def __enter__(self):
|
||
try:
|
||
import pyodbc # type: ignore
|
||
except Exception as e:
|
||
raise RuntimeError("pyodbc não encontrado. Instale com: pip install pyodbc") from e
|
||
|
||
self.cn = pyodbc.connect(self.connection_string, timeout=30)
|
||
self.cn.autocommit = False
|
||
self.cur = self.cn.cursor()
|
||
return self
|
||
|
||
def __exit__(self, exc_type, exc, tb):
|
||
if self.cur is not None:
|
||
try:
|
||
self.cur.close()
|
||
except Exception:
|
||
pass
|
||
if self.cn is not None:
|
||
try:
|
||
if exc_type is None:
|
||
self.cn.commit()
|
||
else:
|
||
self.cn.rollback()
|
||
except Exception:
|
||
pass
|
||
try:
|
||
self.cn.close()
|
||
except Exception:
|
||
pass
|
||
|
||
def reconnect(self) -> None:
|
||
try:
|
||
if self.cur is not None:
|
||
self.cur.close()
|
||
except Exception:
|
||
pass
|
||
try:
|
||
if self.cn is not None:
|
||
self.cn.close()
|
||
except Exception:
|
||
pass
|
||
|
||
import pyodbc # type: ignore
|
||
self.cn = pyodbc.connect(self.connection_string, timeout=30)
|
||
self.cn.autocommit = False
|
||
self.cur = self.cn.cursor()
|
||
|
||
@staticmethod
|
||
def _is_comm_error(e: Exception) -> bool:
|
||
msg = str(e)
|
||
return ("08S01" in msg) or ("10054" in msg) or ("communication link failure" in msg.lower())
|
||
|
||
def ensure_schema(self) -> None:
|
||
assert self.cur is not None
|
||
self.cur.execute(
|
||
f"""
|
||
IF OBJECT_ID('{self.doc_table}', 'U') IS NULL
|
||
BEGIN
|
||
CREATE TABLE {self.doc_table} (
|
||
id BIGINT IDENTITY(1,1) PRIMARY KEY,
|
||
UUID VARCHAR(80) NOT NULL UNIQUE,
|
||
IdExterno BIGINT NULL,
|
||
FranchiseId VARCHAR(20) NULL,
|
||
ImageName VARCHAR(150) NULL,
|
||
EmissionDate DATE NULL,
|
||
EmissaoNF DATE NULL,
|
||
NotaFiscal VARCHAR(40) NULL,
|
||
ValorNF DECIMAL(18,2) NULL,
|
||
Encargos DECIMAL(18,2) NULL,
|
||
AtualizadoEm DATETIME2 NOT NULL DEFAULT SYSUTCDATETIME()
|
||
);
|
||
END;
|
||
"""
|
||
)
|
||
self.cur.execute(
|
||
f"""
|
||
IF OBJECT_ID('{self.parcela_table}', 'U') IS NULL
|
||
BEGIN
|
||
CREATE TABLE {self.parcela_table} (
|
||
id BIGINT IDENTITY(1,1) PRIMARY KEY,
|
||
DocumentoId BIGINT NOT NULL,
|
||
NumeroParcela INT NOT NULL,
|
||
DataVencimento DATE NOT NULL,
|
||
ValorParcela DECIMAL(18,2) NOT NULL,
|
||
CONSTRAINT FK_TrfRemParcela_Documento
|
||
FOREIGN KEY (DocumentoId) REFERENCES {self.doc_table}(id),
|
||
CONSTRAINT UQ_TrfRemParcela UNIQUE (DocumentoId, NumeroParcela)
|
||
);
|
||
END;
|
||
"""
|
||
)
|
||
self.cur.execute(
|
||
f"""
|
||
IF NOT EXISTS (SELECT 1 FROM sys.indexes WHERE name='IX_TrfRemDocumento_NotaFiscal' AND object_id=OBJECT_ID('{self.doc_table}'))
|
||
CREATE INDEX IX_TrfRemDocumento_NotaFiscal ON {self.doc_table}(NotaFiscal);
|
||
IF NOT EXISTS (SELECT 1 FROM sys.indexes WHERE name='IX_TrfRemDocumento_UUID' AND object_id=OBJECT_ID('{self.doc_table}'))
|
||
CREATE UNIQUE INDEX IX_TrfRemDocumento_UUID ON {self.doc_table}(UUID);
|
||
IF NOT EXISTS (SELECT 1 FROM sys.indexes WHERE name='IX_TrfRemDocumento_Franchise_Emission' AND object_id=OBJECT_ID('{self.doc_table}'))
|
||
CREATE INDEX IX_TrfRemDocumento_Franchise_Emission ON {self.doc_table}(FranchiseId, EmissionDate);
|
||
IF NOT EXISTS (SELECT 1 FROM sys.indexes WHERE name='IX_TrfRemParcela_DocumentoId_DataVenc' AND object_id=OBJECT_ID('{self.parcela_table}'))
|
||
CREATE INDEX IX_TrfRemParcela_DocumentoId_DataVenc ON {self.parcela_table}(DocumentoId, DataVencimento);
|
||
"""
|
||
)
|
||
self.cn.commit()
|
||
|
||
def upsert_documento(self, item: Dict[str, Any]) -> Tuple[int, bool]:
|
||
assert self.cur is not None
|
||
uuid = item.get("UUID")
|
||
if not uuid:
|
||
raise ValueError("item sem UUID")
|
||
|
||
resumo = item.get("resumo") or {}
|
||
self.cur.execute(f"SELECT id FROM {self.doc_table} WHERE UUID = ?", uuid)
|
||
row = self.cur.fetchone()
|
||
|
||
if row:
|
||
documento_id = int(row[0])
|
||
self.cur.execute(
|
||
f"""
|
||
UPDATE {self.doc_table}
|
||
SET IdExterno=?, FranchiseId=?, ImageName=?, EmissionDate=?,
|
||
EmissaoNF=?, NotaFiscal=?, ValorNF=?, Encargos=?,
|
||
AtualizadoEm=SYSUTCDATETIME()
|
||
WHERE id=?
|
||
""",
|
||
item.get("id"),
|
||
item.get("franchiseId"),
|
||
item.get("imageName"),
|
||
_date_from_iso(item.get("emissionDate")),
|
||
_date_from_br(resumo.get("emissao_nf")),
|
||
resumo.get("nota_fiscal"),
|
||
resumo.get("valor_nf"),
|
||
resumo.get("encargos"),
|
||
documento_id,
|
||
)
|
||
return documento_id, False
|
||
|
||
self.cur.execute(
|
||
f"""
|
||
INSERT INTO {self.doc_table} (
|
||
UUID, IdExterno, FranchiseId, ImageName, EmissionDate,
|
||
EmissaoNF, NotaFiscal, ValorNF, Encargos
|
||
)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?);
|
||
""",
|
||
uuid,
|
||
item.get("id"),
|
||
item.get("franchiseId"),
|
||
item.get("imageName"),
|
||
_date_from_iso(item.get("emissionDate")),
|
||
_date_from_br(resumo.get("emissao_nf")),
|
||
resumo.get("nota_fiscal"),
|
||
resumo.get("valor_nf"),
|
||
resumo.get("encargos"),
|
||
)
|
||
self.cur.execute(f"SELECT id FROM {self.doc_table} WHERE UUID = ?", uuid)
|
||
new_id = self.cur.fetchone()
|
||
if not new_id:
|
||
raise RuntimeError("Falha ao recuperar id após inserir TrfRemDocumento")
|
||
return int(new_id[0]), True
|
||
|
||
def replace_parcelas(self, documento_id: int, item: Dict[str, Any]) -> None:
|
||
assert self.cur is not None
|
||
self.cur.execute(f"DELETE FROM {self.parcela_table} WHERE DocumentoId = ?", documento_id)
|
||
|
||
parcelas = _ordenar_parcelas_por_data(item.get("parcelas") or [])
|
||
if not parcelas:
|
||
return
|
||
|
||
rows = []
|
||
for idx, p in enumerate(parcelas, start=1):
|
||
dt = _date_from_br(p.get("data"))
|
||
val = p.get("valor")
|
||
if dt is None or val is None:
|
||
continue
|
||
rows.append((documento_id, idx, dt, val))
|
||
|
||
if rows:
|
||
self.cur.fast_executemany = True
|
||
self.cur.executemany(
|
||
f"INSERT INTO {self.parcela_table} (DocumentoId, NumeroParcela, DataVencimento, ValorParcela) VALUES (?, ?, ?, ?)",
|
||
rows,
|
||
)
|
||
|
||
def persist_items(self, items: List[Dict[str, Any]]) -> Tuple[int, int]:
|
||
count = 0
|
||
novos = 0
|
||
for item in items:
|
||
err = item.get("error")
|
||
err_txt = str(err) if err is not None else ""
|
||
if err_txt:
|
||
continue
|
||
if not item.get("UUID"):
|
||
continue
|
||
resumo = item.get("resumo") or {}
|
||
if item.get("status_principal") != "ok":
|
||
continue
|
||
if not resumo.get("emissao_nf") or not resumo.get("nota_fiscal"):
|
||
continue
|
||
if not _nota_fiscal_valida(resumo.get("nota_fiscal")):
|
||
continue
|
||
if (resumo.get("valor_nf") is None) and (item.get("valor_taxa") is None) and not (item.get("parcelas") or []):
|
||
continue
|
||
try:
|
||
doc_id, is_new = self.upsert_documento(item)
|
||
self.replace_parcelas(doc_id, item)
|
||
except Exception as e:
|
||
if not self._is_comm_error(e):
|
||
raise
|
||
# reconecta e tenta uma única vez o item atual
|
||
self.reconnect()
|
||
doc_id, is_new = self.upsert_documento(item)
|
||
self.replace_parcelas(doc_id, item)
|
||
count += 1
|
||
if is_new:
|
||
novos += 1
|
||
return count, novos
|
||
|
||
|
||
def sincronizar_paginas_sqlserver(
|
||
connection_string: str,
|
||
cp_id: int = 10269,
|
||
document_type: str = "EFAT",
|
||
limit: int = 100,
|
||
start_offset: int = 0,
|
||
only_channels: Optional[Set[str]] = None,
|
||
commit_cada_paginas: int = 1,
|
||
) -> Dict[str, Any]:
|
||
cli = Client()
|
||
offset = start_offset
|
||
paginas = 0
|
||
docs_persistidos = 0
|
||
total = None
|
||
|
||
with SqlServerSink(connection_string) as sink:
|
||
sink.ensure_schema()
|
||
while True:
|
||
pagina = cli.processar_pagina(
|
||
cp_id=cp_id,
|
||
document_type=document_type,
|
||
offset=offset,
|
||
limit=limit,
|
||
only_channels=only_channels,
|
||
)
|
||
if total is None:
|
||
total = int(pagina.get("total") or 0)
|
||
|
||
itens = pagina.get("items") or []
|
||
persistidos_pag, novos_pag = sink.persist_items(itens)
|
||
docs_persistidos += persistidos_pag
|
||
paginas += 1
|
||
|
||
if paginas % commit_cada_paginas == 0:
|
||
sink.cn.commit()
|
||
|
||
print(f"[sync] offset={offset} count={len(itens)} novos_pag={novos_pag} persistidos={docs_persistidos} total={total}")
|
||
if not pagina.get("hasNext"):
|
||
break
|
||
offset += limit
|
||
|
||
sink.cn.commit()
|
||
|
||
return {
|
||
"total": total,
|
||
"paginas_processadas": paginas,
|
||
"documentos_persistidos": docs_persistidos,
|
||
"offset_final": offset,
|
||
}
|
||
|
||
|
||
def sincronizar_incremental_sqlserver(
|
||
connection_string: str,
|
||
cp_id: int = 10269,
|
||
document_type: str = "EFAT",
|
||
limit: int = 100,
|
||
only_channels: Optional[Set[str]] = None,
|
||
max_paginas_sem_novidade: int = 3,
|
||
max_paginas: int = 20,
|
||
) -> Dict[str, Any]:
|
||
cli = Client()
|
||
offset = 0
|
||
paginas = 0
|
||
docs_persistidos = 0
|
||
docs_novos = 0
|
||
sem_novidade = 0
|
||
total = None
|
||
|
||
with SqlServerSink(connection_string) as sink:
|
||
sink.ensure_schema()
|
||
while True:
|
||
pagina = cli.processar_pagina(
|
||
cp_id=cp_id,
|
||
document_type=document_type,
|
||
offset=offset,
|
||
limit=limit,
|
||
only_channels=only_channels,
|
||
)
|
||
if total is None:
|
||
total = int(pagina.get("total") or 0)
|
||
|
||
itens = pagina.get("items") or []
|
||
persistidos_pag, novos_pag = sink.persist_items(itens)
|
||
sink.cn.commit()
|
||
|
||
docs_persistidos += persistidos_pag
|
||
docs_novos += novos_pag
|
||
paginas += 1
|
||
sem_novidade = 0 if novos_pag > 0 else (sem_novidade + 1)
|
||
|
||
print(
|
||
f"[inc] offset={offset} count={len(itens)} novos_pag={novos_pag} "
|
||
f"sem_novidade={sem_novidade}/{max_paginas_sem_novidade} total_novos={docs_novos}"
|
||
)
|
||
|
||
if sem_novidade >= max_paginas_sem_novidade:
|
||
break
|
||
if paginas >= max_paginas:
|
||
break
|
||
if not pagina.get("hasNext"):
|
||
break
|
||
offset += limit
|
||
|
||
return {
|
||
"total": total,
|
||
"paginas_processadas": paginas,
|
||
"documentos_persistidos": docs_persistidos,
|
||
"documentos_novos": docs_novos,
|
||
"offset_final": offset,
|
||
"parada_por_sem_novidade": sem_novidade >= max_paginas_sem_novidade,
|
||
}
|
||
|
||
|
||
# -----------------------------
|
||
# RUN
|
||
# -----------------------------
|
||
if __name__ == "__main__":
|
||
# Modos:
|
||
# - "full": carga completa paginada
|
||
# - "incremental": para após X páginas sem novidades
|
||
# - "json": só imprime uma página em JSON
|
||
RUN_MODE = "incremental" # "full" | "incremental" | "json"
|
||
|
||
if RUN_MODE in ("full", "incremental"):
|
||
SQLSERVER_CONN = (
|
||
"DRIVER={ODBC Driver 17 for SQL Server};"
|
||
"SERVER=10.77.77.10;"
|
||
"DATABASE=GINSENG;"
|
||
"UID=andrey;"
|
||
"PWD=88253332;"
|
||
"TrustServerCertificate=yes;"
|
||
)
|
||
if RUN_MODE == "full":
|
||
resultado = sincronizar_paginas_sqlserver(
|
||
connection_string=SQLSERVER_CONN,
|
||
cp_id=10269,
|
||
document_type="EFAT",
|
||
limit=100,
|
||
start_offset=0,
|
||
only_channels=None,
|
||
commit_cada_paginas=1,
|
||
)
|
||
else:
|
||
resultado = sincronizar_incremental_sqlserver(
|
||
connection_string=SQLSERVER_CONN,
|
||
cp_id=10269,
|
||
document_type="EFAT",
|
||
limit=100,
|
||
only_channels=None,
|
||
max_paginas_sem_novidade=5,
|
||
max_paginas=15,
|
||
)
|
||
else:
|
||
c = Client()
|
||
resultado = c.processar_pagina(
|
||
cp_id=10269,
|
||
document_type="EFAT",
|
||
offset=0,
|
||
limit=25,
|
||
only_channels=None,
|
||
)
|
||
|
||
print(json.dumps(resultado, ensure_ascii=False, indent=2))
|
||
|