andrey.cunha/trm_rem_saggezza
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

att faturamento pedido

Browse Source
This commit is contained in:
Cunha 2026-03-06 17:07:26 -03:00
parent 5ddfda737a
commit b57fae393e
2 changed files with 18 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

BIN
__pycache__/installments_reader.cpython-313.pyc Normal file
View File

Binary file not shown.

25
installments_reader.py
View File

@ -5,7 +5,7 @@ import os
import re
import time
from dataclasses import dataclass
from datetime import date
from datetime import date, timedelta
from typing import Any, Dict, List, Optional
from urllib.parse import urlencode
@ -453,6 +453,9 @@ WHERE DocPedidoId = ? AND InstallmentCode NOT IN ({placeholders})
def main() -> None:
today = date.today()
last_days_env = os.getenv("LAST_N_DAYS", "5").strip()
last_n_days = int(last_days_env) if last_days_env else None
rolling_start = (today - timedelta(days=last_n_days)).isoformat() if last_n_days is not None else None
default_start = date(today.year, 1, 1).isoformat()
default_end = today.isoformat()
start_date_env = os.getenv("START_INSTALLMENT_CHANGE_DATE")
@ -506,6 +509,8 @@ def main() -> None:
print(f"[info] incremental ativo com watermark em {watermark_file}")
else:
print("[info] incremental desativado")
if last_n_days is not None:
print(f"[info] janela movel ativa: ultimos {last_n_days} dias ({rolling_start}..{end_date})")
authorized: list[int] = []
unauthorized: list[int] = []
@ -513,16 +518,20 @@ def main() -> None:
failed: Dict[int, str] = {}
def process_store(mediator_code: int) -> Dict[str, Any]:
store_start_date = start_date_fixed
if not store_start_date:
if incremental_mode:
store_start_date = watermark.get(str(mediator_code), default_start)
else:
store_start_date = default_start
if rolling_start:
store_start_date = rolling_start
else:
store_start_date = start_date_fixed
if not store_start_date:
if incremental_mode:
store_start_date = watermark.get(str(mediator_code), default_start)
else:
store_start_date = default_start
local_session = requests.Session()
local_session.trust_env = False
local_auth = Auth(local_session)
store_bearer = local_auth.get_bearer()
try:
print(
@ -608,6 +617,8 @@ def main() -> None:
group_session = requests.Session()
group_session.trust_env = False
group_auth = Auth(group_session)
# Reutiliza o token da loja para evitar nova ida ao endpoint /api/tokens por grupo.
group_auth.override_bearer = store_bearer
group_page = first_page
all_group_installments: List[Dict[str, Any]] = []
group_total = 0