251 lines
14 KiB
Python
251 lines
14 KiB
Python
"""Authenticated version-one Hub routes."""
|
|
import hashlib, json, logging
|
|
from datetime import datetime, timezone
|
|
from fastapi import APIRouter, Depends, HTTPException, Request
|
|
from sqlalchemy import select
|
|
from .auth import authenticate, bootstrap
|
|
from .classification import ENGINE_VERSION, Classifier
|
|
from .config import settings
|
|
from .datasets.service import DatasetService
|
|
from .jobs import claim_next_job, complete_job, create_job, fail_job
|
|
from .context_keys import validate_entry
|
|
from .models import FiscalContextEntry, ClassificationCache, ClassificationRequest, ClassificationResult, Dataset, Job
|
|
|
|
logger = logging.getLogger("gti_fiscal_hub")
|
|
|
|
def router(session_factory, bootstrap_token):
|
|
routes = APIRouter(prefix="/v1")
|
|
def principal(request: Request):
|
|
header = request.headers.get("Authorization", "")
|
|
if not header.startswith("Bearer "): raise HTTPException(401, "Bearer token required")
|
|
session = session_factory()
|
|
try:
|
|
if bootstrap_token: bootstrap(session, bootstrap_token)
|
|
agent = authenticate(session, header.removeprefix("Bearer "))
|
|
if not agent: raise HTTPException(401, "Invalid token")
|
|
return session, agent
|
|
except Exception:
|
|
session.close(); raise
|
|
def decide(payload, identity):
|
|
session, agent = identity; service = DatasetService(session)
|
|
supplied = payload.get("tenant_id")
|
|
if supplied is not None and str(supplied) != str(agent.tenant_id): raise HTTPException(403, "Tenant does not match token")
|
|
product = payload.get("product", payload)
|
|
# Cheap key first (engine version + dataset versions + input), then the cache; the heavy classifier runs only on a MISS.
|
|
engine_version, dataset_versions = ENGINE_VERSION, service.dataset_versions()
|
|
canonical = {"tenant": agent.tenant_id, "product": product, "context": payload.get("context", {}), "engine": engine_version, "datasets": dataset_versions}
|
|
fingerprint = hashlib.sha256(json.dumps(canonical, sort_keys=True, default=str, separators=(",", ":")).encode()).hexdigest()
|
|
cached = session.get(ClassificationCache, fingerprint)
|
|
if cached:
|
|
result = dict(cached.result); result["provenance"] = {**result["provenance"], "cache_hit": True}; return result
|
|
decision = Classifier(service).classify(product)
|
|
if decision["engine_version"] != engine_version or decision["dataset_versions"] != dataset_versions: # never store under a key that does not describe the decision
|
|
canonical = {**canonical, "engine": decision["engine_version"], "datasets": decision["dataset_versions"]}
|
|
fingerprint = hashlib.sha256(json.dumps(canonical, sort_keys=True, default=str, separators=(",", ":")).encode()).hexdigest()
|
|
result = {**decision, "request_fingerprint": fingerprint, "provenance": {"cache_hit": False, "dataset_versions": decision["dataset_versions"]}}
|
|
request = ClassificationRequest(tenant_id=agent.tenant_id, agent_id=agent.id, fingerprint=fingerprint, payload=payload); session.add(request); session.flush()
|
|
session.add(ClassificationResult(request_id=request.id, result=result)); session.add(ClassificationCache(fingerprint=fingerprint, tenant_id=agent.tenant_id, result=result)); session.commit(); return result
|
|
@routes.post("/validate/ncm")
|
|
def validate_ncm(payload: dict, identity=Depends(principal)):
|
|
return decide({"product": {"current_ncm": payload.get("ncm"), "description": ""}, "tenant_id": payload.get("tenant_id")}, identity)
|
|
@routes.post("/classify")
|
|
def classify(payload: dict, identity=Depends(principal)):
|
|
try: return decide(payload, identity)
|
|
finally: identity[0].close()
|
|
@routes.post("/classify/batch")
|
|
def classify_batch(payload: dict, identity=Depends(principal)):
|
|
try: return {"results": [decide({**payload, "product": product}, identity) for product in payload.get("products", [])]}
|
|
finally: identity[0].close()
|
|
@routes.get("/datasets/status")
|
|
def status(identity=Depends(principal)):
|
|
session, _ = identity
|
|
try: return [{"id": item.id, "kind": item.kind, "status": item.status, "sha256": item.sha256, "record_count": item.record_count} for item in session.scalars(select(Dataset)).all()]
|
|
finally: session.close()
|
|
|
|
@routes.get("/reference/cfop/{code}")
|
|
def reference_cfop(code: str, identity=Depends(principal)):
|
|
session, _ = identity
|
|
try:
|
|
rows = DatasetService(session).cfop_lookup(code=code)
|
|
return {"found": bool(rows), "entries": [{"code": r.code, "title": r.title, "description": r.description} for r in rows]}
|
|
finally: session.close()
|
|
|
|
@routes.get("/reference/ncm_utrib/{ncm}")
|
|
def reference_ncm_utrib(ncm: str, identity=Depends(principal)):
|
|
session, _ = identity
|
|
try:
|
|
row = DatasetService(session).ncm_utrib_for_ncm(ncm)
|
|
return {"found": row is not None, "utrib_abbreviation": row.utrib_abbreviation if row else None}
|
|
finally: session.close()
|
|
|
|
@routes.get("/reference/anp/{cprod_anp}")
|
|
def reference_anp(cprod_anp: str, identity=Depends(principal)):
|
|
session, _ = identity
|
|
try: return {"found": DatasetService(session).anp_code_exists(cprod_anp)}
|
|
finally: session.close()
|
|
|
|
@routes.get("/reference/fcp/{uf}")
|
|
def reference_fcp(uf: str, identity=Depends(principal)):
|
|
session, _ = identity
|
|
try:
|
|
row = DatasetService(session).fcp_for_uf(uf)
|
|
return {"found": row is not None, "aliq_1": row.aliq_1 if row else None, "observacao": row.observacao if row else None}
|
|
finally: session.close()
|
|
|
|
@routes.get("/reference/rtc_cclasstrib/{code}")
|
|
def reference_rtc_cclasstrib(code: str, identity=Depends(principal)):
|
|
session, _ = identity
|
|
try:
|
|
rows = DatasetService(session).rtc_cclasstrib(code=code)
|
|
return {"found": bool(rows), "entries": [{"cst": r.cst, "cclasstrib": r.cclasstrib, "name": r.name, "description": r.description, "tipo_aliquota": r.tipo_aliquota, "p_red_ibs": r.p_red_ibs, "p_red_cbs": r.p_red_cbs, "d_ini_vig": r.d_ini_vig, "d_fim_vig": r.d_fim_vig} for r in rows]}
|
|
finally: session.close()
|
|
|
|
@routes.get("/reference/rtc_cbs_rate/{year}")
|
|
def reference_rtc_cbs_rate(year: int, identity=Depends(principal)):
|
|
session, _ = identity
|
|
try:
|
|
row = DatasetService(session).rtc_cbs_rate(year)
|
|
return {"found": row is not None, "year": year, "rate": row.rate if row else None, "rate_text": row.rate_text if row else None}
|
|
finally: session.close()
|
|
|
|
@routes.get("/reference/fiscal_codes/{domain}")
|
|
def reference_fiscal_codes(domain: str, identity=Depends(principal)):
|
|
session, _ = identity
|
|
try:
|
|
rows = DatasetService(session).fiscal_codes(domain.upper())
|
|
return {"found": bool(rows), "domain": domain.upper(), "codes": sorted(r.code for r in rows)}
|
|
finally: session.close()
|
|
|
|
@routes.post("/context/import")
|
|
def context_import(payload: dict, identity=Depends(principal)):
|
|
"""Batch upsert of operator fiscal context: {"company_key": "...", "entries": [{scope, scope_key, key, value, reason}]}. All-or-nothing."""
|
|
session, agent = identity
|
|
try:
|
|
company_key = str(payload.get("company_key") or "").strip()
|
|
if not company_key:
|
|
raise HTTPException(400, "company_key is required")
|
|
entries, errors = [], []
|
|
for i, raw in enumerate(payload.get("entries") or []):
|
|
entry, error = validate_entry(raw)
|
|
(errors.append({"index": i, "error": error, "entry": raw}) if error else entries.append(entry))
|
|
if errors:
|
|
raise HTTPException(422, {"errors": errors[:50], "error_count": len(errors)})
|
|
created = updated = 0
|
|
for e in entries:
|
|
row = session.scalar(select(FiscalContextEntry).where(FiscalContextEntry.tenant_id == agent.tenant_id, FiscalContextEntry.company_key == company_key,
|
|
FiscalContextEntry.scope == e["scope"], FiscalContextEntry.scope_key == e["scope_key"], FiscalContextEntry.key == e["key"]))
|
|
if row:
|
|
row.value, row.reason, row.updated_at = e["value"], e["reason"], datetime.now(timezone.utc); updated += 1
|
|
else:
|
|
session.add(FiscalContextEntry(tenant_id=agent.tenant_id, company_key=company_key, source="OPERATOR", **e)); created += 1
|
|
session.commit()
|
|
return {"created": created, "updated": updated}
|
|
finally: session.close()
|
|
|
|
@routes.get("/context")
|
|
def context_list(company_key: str, identity=Depends(principal)):
|
|
session, agent = identity
|
|
try:
|
|
rows = session.scalars(select(FiscalContextEntry).where(FiscalContextEntry.tenant_id == agent.tenant_id, FiscalContextEntry.company_key == company_key)).all()
|
|
return {"company_key": company_key, "entries": [{"scope": r.scope, "scope_key": r.scope_key, "key": r.key, "value": r.value, "source": r.source, "reason": r.reason,
|
|
"created_at": r.created_at.isoformat() if r.created_at else None, "updated_at": r.updated_at.isoformat() if r.updated_at else None} for r in rows]}
|
|
finally: session.close()
|
|
|
|
@routes.post("/agents/heartbeat")
|
|
def heartbeat(payload: dict, identity=Depends(principal)):
|
|
session, agent = identity
|
|
try:
|
|
agent.last_seen_at = datetime.now(timezone.utc)
|
|
agent.last_status = payload.get("status")
|
|
if payload.get("schema_fingerprint"):
|
|
agent.schema_fingerprint = payload["schema_fingerprint"]
|
|
if payload.get("agent_version"):
|
|
agent.agent_version = payload["agent_version"]
|
|
session.add(agent); session.commit()
|
|
logger.info("heartbeat agent_id=%s status=%s last_job_id=%s", agent.id, agent.last_status, payload.get("last_job_id"))
|
|
return {"agent_id": agent.id, "last_seen_at": agent.last_seen_at.isoformat(), "enabled": agent.enabled}
|
|
finally: session.close()
|
|
|
|
@routes.get("/agents/me")
|
|
def agent_status(identity=Depends(principal)):
|
|
session, agent = identity
|
|
try:
|
|
last_seen = agent.last_seen_at
|
|
if last_seen is not None and last_seen.tzinfo is None:
|
|
last_seen = last_seen.replace(tzinfo=timezone.utc) # SQLite (tests) drops tz info
|
|
online = last_seen is not None and (datetime.now(timezone.utc) - last_seen).total_seconds() <= settings.agent_online_threshold_seconds
|
|
return {
|
|
"agent_id": agent.id, "company_id": agent.tenant_id, "name": agent.name, "enabled": agent.enabled,
|
|
"last_seen_at": agent.last_seen_at.isoformat() if agent.last_seen_at else None,
|
|
"last_status": agent.last_status, "schema_fingerprint": agent.schema_fingerprint,
|
|
"agent_version": agent.agent_version, "status": "ONLINE" if online else "OFFLINE",
|
|
}
|
|
finally: session.close()
|
|
|
|
def _job_view(job):
|
|
return {
|
|
"job_id": job.id, "agent_id": job.agent_id, "kind": job.kind, "status": job.status,
|
|
"payload": job.payload, "created_at": job.created_at.isoformat() if job.created_at else None,
|
|
"started_at": job.started_at.isoformat() if job.started_at else None,
|
|
"finished_at": job.finished_at.isoformat() if job.finished_at else None,
|
|
"attempts": job.attempts, "result_summary": job.result_summary, "error": job.error,
|
|
}
|
|
|
|
@routes.post("/jobs")
|
|
def create_job_route(payload: dict, identity=Depends(principal)):
|
|
session, agent = identity
|
|
try:
|
|
kind = payload.get("kind")
|
|
if kind not in ("REPROCESS_ALL", "REPROCESS_PRODUCTS"):
|
|
raise HTTPException(400, "kind must be REPROCESS_ALL or REPROCESS_PRODUCTS")
|
|
job_payload = payload.get("payload", {})
|
|
if kind == "REPROCESS_PRODUCTS" and not job_payload.get("codigos"):
|
|
raise HTTPException(400, "REPROCESS_PRODUCTS requires payload.codigos")
|
|
job = create_job(session, agent, kind, job_payload)
|
|
logger.info("job created job_id=%s agent_id=%s kind=%s", job.id, agent.id, kind)
|
|
return _job_view(job)
|
|
finally: session.close()
|
|
|
|
@routes.get("/jobs/next")
|
|
def next_job(identity=Depends(principal)):
|
|
session, agent = identity
|
|
try:
|
|
job = claim_next_job(session, agent)
|
|
if job is None: return None
|
|
logger.info("job claimed job_id=%s agent_id=%s kind=%s", job.id, agent.id, job.kind)
|
|
return _job_view(job)
|
|
finally: session.close()
|
|
|
|
@routes.get("/jobs/{job_id}")
|
|
def get_job(job_id: int, identity=Depends(principal)):
|
|
session, agent = identity
|
|
try:
|
|
job = session.get(Job, job_id)
|
|
if job is None or job.agent_id != agent.id: raise HTTPException(404, "job not found")
|
|
return _job_view(job)
|
|
finally: session.close()
|
|
|
|
@routes.post("/jobs/{job_id}/complete")
|
|
def complete_job_route(job_id: int, payload: dict, identity=Depends(principal)):
|
|
session, agent = identity
|
|
try:
|
|
job = session.get(Job, job_id)
|
|
if job is None or job.agent_id != agent.id: raise HTTPException(404, "job not found")
|
|
if not complete_job(session, job, payload.get("result_summary", {})): raise HTTPException(409, "job is not RUNNING")
|
|
logger.info("job completed job_id=%s agent_id=%s result=%s", job_id, agent.id, payload.get("result_summary", {}))
|
|
return _job_view(session.get(Job, job_id))
|
|
finally: session.close()
|
|
|
|
@routes.post("/jobs/{job_id}/fail")
|
|
def fail_job_route(job_id: int, payload: dict, identity=Depends(principal)):
|
|
session, agent = identity
|
|
try:
|
|
job = session.get(Job, job_id)
|
|
if job is None or job.agent_id != agent.id: raise HTTPException(404, "job not found")
|
|
if not fail_job(session, job, payload.get("error", "unknown error")): raise HTTPException(409, "job is not RUNNING")
|
|
logger.warning("job failed job_id=%s agent_id=%s error=%s", job_id, agent.id, payload.get("error", "unknown error"))
|
|
return _job_view(session.get(Job, job_id))
|
|
finally: session.close()
|
|
|
|
return routes
|