gustavosantos/GTI_GERENCIADOR_FISCAL
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: OSCBR role is IDENTITY (V3 contract has no NCM/CEST): v3 endpoint, provides_ncm=false, leading-zero NCM guard, backfill refuses until NCM is proven

Browse Source 
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
gti
2026-09-26 12:00:05 -03:00
co-authored by Claude Sonnet 5
parent c68c6c67ed
commit ab65ddd679
4 changed files with 56 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
central/app/enrichment/providers.py
+9 -3
View File
@@ -231,7 +231,10 @@ class Oscbr(Provider):
name = "oscbr"
rate_per_minute = 5 # free plan as currently documented: 50 calls/day, 5/min (conservative: the lower figure wins)
daily_limit = 50
provides_ncm = True
# ROLE = IDENTITY. The operator-provided V3 contract (GET /api/v3/gtin/{ean}) documents ean, ean_tipo, nome, marca, categoria, link_foto - NO ncm/cest. The public landing page mentions NCM/CEST
# and the legacy open-source client shows an `ncm` integer (leading zero lost), but the V3 documentation is behind the login, so NCM is UNPROVEN: the provider does not count as an NCM family
# (provides_ncm=False). If a real authenticated V3 answer ever carries `ncm`, it is stored as evidence but the role only changes after that proof is reviewed.
provides_ncm = False
base = "https://gtin.rscsistemas.com.br"
def __init__(self, client=None, token=None, user=None, password=None):
@@ -272,7 +275,7 @@ class Oscbr(Provider):
def lookup_gtin(self, gtin: str) -> ProviderProduct:
if not self.configured:
raise ProviderError("AUTH_ERROR", "NOT_CONFIGURED")
response = self._get(f"{self.base}/api/gtin/infor/{gtin}", headers={"Authorization": f"Bearer {self._auth()}", "User-Agent": USER_AGENT})
response = self._get(f"{self.base}/api/v3/gtin/{gtin}", headers={"Authorization": f"Bearer {self._auth()}", "User-Agent": USER_AGENT})
headers = {k.lower(): v for k, v in response.headers.items()}
if response.status_code == 404:
return ProviderProduct(self.name, gtin, False, latency_ms=getattr(response, "latency_ms", 0), headers=headers)
@@ -286,7 +289,10 @@ class Oscbr(Provider):
body = body[0] if body else {}
if not body or not (body.get("nome") or body.get("ncm")):
return ProviderProduct(self.name, gtin, False, latency_ms=getattr(response, "latency_ms", 0), headers=headers)
return ProviderProduct(self.name, gtin, True, title=body.get("nome"), brand=body.get("marca"), category=body.get("categoria"), ncm=_digits(body.get("ncm")), cest=_digits(body.get("cest")),
ncm = _digits(body.get("ncm"))
if ncm and len(ncm) == 7: # numeric JSON drops the leading zero (7133399 is NCM 07133399)
ncm = ncm.zfill(8)
return ProviderProduct(self.name, gtin, True, title=body.get("nome"), brand=body.get("marca"), category=body.get("categoria"), ncm=ncm if len(ncm or "") == 8 else None, cest=_digits(body.get("cest")) or None,
raw_hash=hashlib.sha256(json.dumps(body, sort_keys=True, ensure_ascii=False).encode()).hexdigest(), latency_ms=getattr(response, "latency_ms", 0), headers=headers)
central/scripts/oscbr_backfill.py
+3
View File
@@ -23,6 +23,9 @@ def main(argv=None):
args = ap.parse_args(argv)
providers = build_providers(settings)
oscbr = next(p for p in providers if p.name == "oscbr")
if not oscbr.provides_ncm and not args.dry_run:
print(json.dumps({"OSCBR_ROLE": "IDENTITY", "reason": "NCM not documented in the V3 contract; backfill for NCM verification is meaningless until proven", "requests": 0}))
return 3
if not oscbr.configured and not args.dry_run:
print(json.dumps({"OSCBR_STATUS": "NOT_CONFIGURED", "requests": 0}))
return 2
central/tests/test_oscbr_backfill.py
+17 -1
View File
@@ -31,7 +31,9 @@ def _oscbr(ncm_by_gtin, calls):
if g in ncm_by_gtin:
return httpx.Response(200, json={"ean": g, "nome": "PRODUTO X", "marca": "M", "ncm": ncm_by_gtin[g], "cest": "0300700"})
return httpx.Response(404, json={})
return Oscbr(httpx.Client(transport=httpx.MockTransport(handler)), None, "u", "p")
prov = Oscbr(httpx.Client(transport=httpx.MockTransport(handler)), None, "u", "p")
prov.provides_ncm = True # simulates a plan/answer that DOES carry NCM: exercises the backfill/promotion logic (the real V3 role is IDENTITY)
return prov
def _seed(env, n=6):
@@ -108,3 +110,17 @@ def test_provider_query_canonical_forms():
assert backfill.provider_query("00000078936683") == "78936683" # GTIN-8
assert backfill.provider_query("07891000100103") == "7891000100103" # EAN-13
assert backfill.provider_query("17891000100100") == "17891000100100" # GTIN-14 with indicator
def test_oscbr_real_role_is_identity_and_uses_the_v3_endpoint():
seen = []
def handler(request):
seen.append(request.url.path)
return httpx.Response(200, json={"ean": "7891000100103", "ean_tipo": "EAN13", "nome": "COLA 350ML", "marca": "X", "categoria": "Bebidas", "link_foto": "u"})
p = Oscbr(httpx.Client(transport=httpx.MockTransport(handler)), "tok")
r = p.lookup_gtin("7891000100103")
assert seen == ["/api/v3/gtin/7891000100103"] and r.found and r.title == "COLA 350ML" and r.ncm is None and r.cest is None
assert Oscbr.provides_ncm is False
legacy = Oscbr(httpx.Client(transport=httpx.MockTransport(lambda q: httpx.Response(200, json={"nome": "FEIJAO", "ncm": 7133399}))), "tok").lookup_gtin("7896116900029")
assert legacy.ncm == "07133399" # 7-digit integer restored to the 8-digit NCM
docs/runs/sandbox_1_3_4.json
+27
View File
@@ -0,0 +1,27 @@
{
"INSTALLED_SHA256_MATCHES_PUBLISHED": true,
"SERVICE_RUNNING": true,
"VERSION_1_3_4": true,
"HEARTBEAT_S": "8",
"WAITING_UNTIL_CONFIGURED": true,
"PRODUCT_DETECTED": true,
"PLACEHOLDER_RECOGNIZED": true,
"ENRICHMENT_ATTEMPTED": true,
"DRY_RUN_WRITES_0": true,
"enrichment_row": "RESOLVED|EXACT_GTIN_VERIFIED|AUTO_FILL_SAFE|00000000|DRY_RUN",
"row_after_dry_run": [
[
"SN001",
"00000000",
"<null>",
"<null>"
]
],
"SERVICE_BACK_AFTER_REBOOT_S": 153,
"HEARTBEAT_AFTER_REBOOT_S": 2,
"SAME_IDENTITY_AFTER_REBOOT": true,
"NO_MANUAL_ENV_EDIT": true,
"NO_DUPLICATE_AGENT": true,
"AGENT_ID": "617",
"INSTALLATION_ID": "7ac81323-034d-47ba-bcef-b070e88bf08c"
}