451 lines
29 KiB
Python
451 lines
29 KiB
Python
"""Authenticated version-one Hub routes."""
|
|
import hashlib, json, logging
|
|
import uuid
|
|
from datetime import datetime, timezone
|
|
from fastapi import APIRouter, Depends, HTTPException, Request, Response
|
|
import csv
|
|
import io
|
|
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 SYNC_KINDS, claim_next_job, complete_job, create_job, fail_job, schedule_automatic_sync
|
|
from .blocked_products import list_blockers, resolve_blocker, resolve_product_blockers, upsert_blocker
|
|
from .context_keys import validate_entry
|
|
from .models import AgentSnapshot, BlockedProduct, FiscalContextEntry, ClassificationCache, ClassificationRequest, ClassificationResult, Dataset, Installation, Job
|
|
from .events import record_event
|
|
from .installation_identity import compute_agent_health, compute_version_drift, reference_agent_version, sanitize_registration
|
|
from .control_plane import list_jobs
|
|
from .human_auth import require_human
|
|
|
|
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:
|
|
try:
|
|
supplied = sanitize_registration(payload)
|
|
except (TypeError, ValueError) as error:
|
|
raise HTTPException(422, f"invalid registration metadata: {error}") from error
|
|
now = datetime.now(timezone.utc)
|
|
agent.last_seen_at = now
|
|
agent.last_status = supplied.get("status") or 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"]
|
|
installation_key = supplied.get("installation_id") or str(uuid.uuid4())
|
|
installation = session.scalar(select(Installation).where(Installation.agent_id == agent.id))
|
|
if installation is not None and installation.installation_id != installation_key:
|
|
raise HTTPException(409, "installation_id conflicts with registered Agent")
|
|
if installation is None:
|
|
installation = Installation(agent_id=agent.id, tenant_id=agent.tenant_id, installation_id=installation_key, display_name=supplied.get("display_name") or agent.name)
|
|
session.add(installation); session.flush()
|
|
fields = {"display_name": supplied.get("display_name"), "customer_code": supplied.get("customer_code"), "notes": supplied.get("notes"), "hostname": supplied.get("hostname"), "windows_version": supplied.get("windows_version"), "machine_fingerprint": supplied.get("machine_fingerprint"), "company_code": supplied.get("company_code"), "company_name": supplied.get("company_name"), "company_cnpj_masked": supplied.get("company_cnpj_masked"), "company_uf": supplied.get("company_uf"), "company_crt": supplied.get("company_crt"), "firebird_version": supplied.get("firebird_version"), "firebird_ods": supplied.get("firebird_ods"), "database_identifier": supplied.get("database_identifier"), "capabilities": supplied.get("capabilities"), "fiscal_engine_version": supplied.get("fiscal_engine_version"), "dataset_versions": supplied.get("dataset_versions"), "last_status": supplied.get("status") or payload.get("status"), "last_seen_at": now}
|
|
for key, value in fields.items():
|
|
if value is not None: setattr(installation, key, value)
|
|
if supplied.get("started_at"):
|
|
try:
|
|
installation.started_at = datetime.fromisoformat(str(supplied["started_at"]).replace("Z", "+00:00"))
|
|
except ValueError:
|
|
raise HTTPException(422, "started_at must be ISO-8601")
|
|
snapshot = session.scalar(select(AgentSnapshot).where(AgentSnapshot.installation_id == installation.id))
|
|
if snapshot is None:
|
|
snapshot = AgentSnapshot(installation_id=installation.id)
|
|
session.add(snapshot)
|
|
if payload.get("coverage"):
|
|
coverage = payload["coverage"]
|
|
snapshot.total_products = int(coverage.get("total_products", coverage.get("total", snapshot.total_products)) or 0)
|
|
snapshot.fiscally_complete = int(coverage.get("fiscally_complete", coverage.get("complete", snapshot.fiscally_complete)) or 0)
|
|
snapshot.safe_blocked = int(coverage.get("safe_blocked", snapshot.safe_blocked) or 0)
|
|
snapshot.failed_products = int(coverage.get("failed_products", snapshot.failed_products) or 0)
|
|
snapshot.coverage_percent = float(coverage.get("coverage_percent", snapshot.coverage_percent) or 0)
|
|
snapshot.blockers = coverage.get("blockers", snapshot.blockers) or {}
|
|
snapshot.version_drift = compute_version_drift(agent.agent_version, reference_agent_version(session))
|
|
snapshot.agent_health, snapshot.health_reason = compute_agent_health(installation, supplemental=snapshot)
|
|
session.add(agent); session.flush()
|
|
record_event(session, tenant_id=agent.tenant_id, installation_id=installation.id, agent_id=agent.id, severity="INFO", event_type="HEARTBEAT", message="Agent heartbeat", context={"status": agent.last_status, "agent_version": agent.agent_version})
|
|
session.commit()
|
|
# Central scheduler: exactly one INITIAL_SYNC after the first heartbeat, then
|
|
# one SYNC_INCREMENTAL per interval — idempotent, no local cron/toggle needed.
|
|
if agent.enabled:
|
|
schedule_automatic_sync(session, agent, installation, snapshot)
|
|
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, "installation_id": installation.installation_id, "last_seen_at": agent.last_seen_at.isoformat(), "enabled": agent.enabled,
|
|
"processing_enabled": bool(installation.processing_enabled), "sync_interval_seconds": installation.sync_interval_seconds}
|
|
finally: session.close()
|
|
|
|
@routes.post("/agents/register")
|
|
def register_agent(payload: dict, identity=Depends(principal)):
|
|
return heartbeat({**payload, "status": payload.get("status", "IDLE")}, identity)
|
|
|
|
@routes.post("/agents/events")
|
|
def agent_event(payload: dict, identity=Depends(principal)):
|
|
session, agent = identity
|
|
try:
|
|
installation = session.scalar(select(Installation).where(Installation.agent_id == agent.id))
|
|
if installation is None:
|
|
raise HTTPException(409, "Agent must heartbeat/register before sending events")
|
|
event = record_event(session, tenant_id=agent.tenant_id, installation_id=installation.id, agent_id=agent.id, severity=payload.get("severity", "INFO"), event_type=payload.get("event_type", "HEARTBEAT"), message=payload.get("message", ""), context=payload.get("context") or {}, job_id=payload.get("job_id"), product_code=payload.get("product_code"), blocker_id=payload.get("blocker_id"))
|
|
session.commit()
|
|
return {"event_id": event.id}
|
|
finally: session.close()
|
|
|
|
@routes.get("/agents/me/identity")
|
|
def agent_identity(identity=Depends(principal)):
|
|
"""Return only the authenticated Agent's existing installation identity."""
|
|
session, agent = identity
|
|
try:
|
|
installation = session.scalar(select(Installation).where(Installation.agent_id == agent.id))
|
|
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)
|
|
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,
|
|
"installation_id": installation.installation_id if installation else None,
|
|
"hostname": installation.hostname if installation else None,
|
|
"version": agent.agent_version,
|
|
"status": "ONLINE" if online else "OFFLINE",
|
|
}
|
|
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 _blocked_view(row):
|
|
return {
|
|
"id": row.id, "tenant_id": row.tenant_id, "company": row.company, "product_code": row.product_code,
|
|
"installation_db_id": row.installation_id,
|
|
"gtin": row.gtin, "description": row.description, "blocker_type": row.blocker_type, "field": row.field,
|
|
"reason_code": row.reason_code, "evidence_summary": row.evidence_summary, "first_seen_at": row.first_seen_at.isoformat() if row.first_seen_at else None,
|
|
"last_seen_at": row.last_seen_at.isoformat() if row.last_seen_at else None, "retry_count": row.retry_count,
|
|
"next_retry_at": row.next_retry_at.isoformat() if row.next_retry_at else None, "resolved_at": row.resolved_at.isoformat() if row.resolved_at else None,
|
|
"resolution_source": row.resolution_source,
|
|
}
|
|
|
|
@routes.get("/blockers")
|
|
def blockers_list(blocker_type: str | None = None, field: str | None = None, identity=Depends(principal)):
|
|
session, agent = identity
|
|
try:
|
|
installation = session.scalar(select(Installation).where(Installation.agent_id == agent.id))
|
|
return {"blockers": [_blocked_view(row) for row in list_blockers(session, agent.tenant_id, blocker_type=blocker_type, field=field, installation_id=installation.id if installation else None)]}
|
|
finally:
|
|
session.close()
|
|
|
|
@routes.get("/blockers/export.csv")
|
|
def blockers_export(identity=Depends(principal)):
|
|
session, agent = identity
|
|
try:
|
|
installation = session.scalar(select(Installation).where(Installation.agent_id == agent.id))
|
|
output = io.StringIO()
|
|
writer = csv.writer(output)
|
|
writer.writerow(["id", "company", "product_code", "gtin", "blocker_type", "field", "reason_code", "retry_count", "first_seen_at", "last_seen_at"])
|
|
for row in list_blockers(session, agent.tenant_id, installation_id=installation.id if installation else None):
|
|
writer.writerow([row.id, row.company, row.product_code, row.gtin or "", row.blocker_type, row.field, row.reason_code, row.retry_count, row.first_seen_at.isoformat() if row.first_seen_at else "", row.last_seen_at.isoformat() if row.last_seen_at else ""])
|
|
return Response(output.getvalue(), media_type="text/csv", headers={"Content-Disposition": "attachment; filename=blocked-products.csv"})
|
|
finally:
|
|
session.close()
|
|
|
|
@routes.post("/blockers/reprocess")
|
|
def blockers_reprocess(payload: dict, identity=Depends(principal)):
|
|
session, agent = identity
|
|
try:
|
|
installation = session.scalar(select(Installation).where(Installation.agent_id == agent.id))
|
|
rows = list_blockers(session, agent.tenant_id, blocker_type=payload.get("blocker_type"), field=payload.get("field"), installation_id=installation.id if installation else None)
|
|
codes = [row.product_code for row in rows if row.product_code]
|
|
if not codes:
|
|
return {"created": False, "reason": "NO_DUE_BLOCKERS"}
|
|
job = create_job(session, agent, "REPROCESS_PRODUCTS", {"codigos": codes, "reason": "BLOCKER_REPROCESS"})
|
|
return {"created": True, "job_id": job.id, "products": len(codes)}
|
|
finally:
|
|
session.close()
|
|
|
|
@routes.post("/blockers/{blocker_id}/resolve")
|
|
def blockers_resolve(blocker_id: int, payload: dict, identity=Depends(principal)):
|
|
session, agent = identity
|
|
try:
|
|
row = session.get(BlockedProduct, blocker_id)
|
|
if row is None or row.tenant_id != agent.tenant_id:
|
|
raise HTTPException(404, "blocker not found")
|
|
if not resolve_blocker(session, blocker_id, payload.get("resolution_source", "OPERATOR")):
|
|
raise HTTPException(404, "blocker not found")
|
|
return {"resolved": True, "blocker_id": blocker_id}
|
|
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,
|
|
}
|
|
|
|
def _apply_job_outcome(session, agent, installation, job, result_summary, *, failed):
|
|
"""After a job finishes: coverage snapshot (from the Agent's operational
|
|
summary), sync bookkeeping and an audit-friendly event. Never stores secrets."""
|
|
now = datetime.now(timezone.utc)
|
|
snapshot = session.scalar(select(AgentSnapshot).where(AgentSnapshot.installation_id == installation.id))
|
|
if snapshot is None:
|
|
snapshot = AgentSnapshot(installation_id=installation.id); session.add(snapshot)
|
|
snapshot.last_job_status = "FAILED" if failed else "COMPLETED"
|
|
snapshot.last_job_at = now
|
|
context = {"kind": job.kind}
|
|
operational = result_summary.get("operational") if not failed else None
|
|
if operational and job.kind in SYNC_KINDS:
|
|
total = int(operational.get("TOTAL_PRODUCTS", 0))
|
|
complete = int(operational.get("COMPLETE", 0)) + int(operational.get("CORRECTED", 0))
|
|
snapshot.total_products = total
|
|
snapshot.fiscally_complete = complete
|
|
snapshot.safe_blocked = int(operational.get("SAFE_BLOCKED", 0))
|
|
snapshot.failed_products = int(operational.get("FAILED_TECHNICAL", 0))
|
|
snapshot.coverage_percent = round(100.0 * complete / total, 2) if total else 0.0
|
|
snapshot.blockers = {state: int(count) for state, count in operational.items() if state.startswith("BLOCKED_") and count}
|
|
snapshot.last_successful_job_at = now
|
|
installation.last_sync_at = now
|
|
installation.last_sync_changed = int(result_summary.get("products_changed", result_summary.get("products_processed", 0)) or 0)
|
|
context.update({"total_products": total, "products_changed": installation.last_sync_changed, "products_processed": result_summary.get("products_processed")})
|
|
if failed:
|
|
context["error"] = result_summary.get("error")
|
|
record_event(session, tenant_id=agent.tenant_id, installation_id=installation.id, agent_id=agent.id, severity="ERROR" if failed else "INFO",
|
|
event_type="JOB_FAILED" if failed else "JOB_COMPLETED", message=f"Job {job.kind} {'failed' if failed else 'completed'}", context=context, job_id=job.id)
|
|
session.commit()
|
|
|
|
@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
|
|
installation = session.scalar(select(Installation).where(Installation.agent_id == agent.id))
|
|
if installation is not None:
|
|
record_event(session, tenant_id=agent.tenant_id, installation_id=installation.id, agent_id=agent.id, severity="INFO", event_type="JOB_STARTED", message=f"Job {job.kind} started", context={"kind": job.kind}, job_id=job.id)
|
|
session.commit()
|
|
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/global")
|
|
def global_jobs(request: Request, page: int = 1, page_size: int = 50, status: str | None = None, kind: str | None = None):
|
|
session = session_factory()
|
|
try:
|
|
require_human(request, session)
|
|
return list_jobs(session, page=page, page_size=page_size, status=status, kind=kind)
|
|
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")
|
|
result_summary = payload.get("result_summary", {})
|
|
installation = session.scalar(select(Installation).where(Installation.agent_id == agent.id))
|
|
for blocker in result_summary.get("blocked_products", []):
|
|
upsert_blocker(session, agent.tenant_id, blocker, installation_id=installation.id if installation else None)
|
|
for product_code in result_summary.get("resolved_products", []):
|
|
resolve_product_blockers(session, agent.tenant_id, product_code, installation_id=installation.id if installation else None)
|
|
if not complete_job(session, job, result_summary): raise HTTPException(409, "job is not RUNNING")
|
|
if installation is not None:
|
|
_apply_job_outcome(session, agent, installation, job, result_summary, failed=False)
|
|
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")
|
|
installation = session.scalar(select(Installation).where(Installation.agent_id == agent.id))
|
|
if installation is not None:
|
|
_apply_job_outcome(session, agent, installation, job, {"error": str(payload.get("error", "unknown error"))[:500]}, failed=True)
|
|
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
|