""" Camoufox MSC tracker — small FastAPI worker. Abre una sesion de Camoufox (Firefox parcheado a nivel C++ para pasar el filtro de huellas digitales de Akamai), hace la consulta al endpoint interno de MSC, y devuelve el JSON aplanado. Pensado para correr detras de Tailscale (bind 127.0.0.1) o expuesto por un reverse proxy. NO exponer a internet publico sin auth. """ import json import logging import re import time from typing import Any, Dict, List from fastapi import FastAPI, HTTPException, Response from pydantic import BaseModel, Field import uvicorn from camoufox.sync_api import Camoufox # --- Config --- HOST = "127.0.0.1" PORT = 9090 CACHE_TTL_SECONDS = 60 # --- Logging --- logging.basicConfig( level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s", ) log = logging.getLogger("worker") # --- App --- app = FastAPI(title="Camoufox MSC tracker", version="1.1.0") _cache: Dict[str, tuple] = {} # --- Modelos --- class TrackRequest(BaseModel): container: str = Field( ..., min_length=4, max_length=20, pattern=r"^[A-Z0-9]+$", description="Numero de container (4-20 chars alfanumericos, mayusculas)", ) # --- Helpers --- def flatten(raw: Dict[str, Any]) -> Dict[str, Any]: """Aplana la respuesta anidada de MSC en algo facil de usar.""" bol = raw["Data"]["BillOfLadings"][0] c = bol["ContainersInfo"][0] gti = bol["GeneralTrackingInfo"] events: List[Dict] = [] for e in sorted(c["Events"], key=lambda x: x.get("Order", 0), reverse=True): detail = e.get("Detail") or [] events.append( { "date": e.get("Date", ""), "location": e.get("Location", ""), "description": e.get("Description", ""), "vessel": detail[0] if len(detail) > 0 else "", "voyage": detail[1] if len(detail) > 1 else "", } ) return { "container": c["ContainerNumber"], "bill_of_lading": bol["BillOfLadingNumber"], "shipped_from": gti["ShippedFrom"], "shipped_to": gti["ShippedTo"], "port_of_load": gti["PortOfLoad"], "port_of_discharge": gti["PortOfDischarge"], "transshipments": gti.get("Transshipments", []), "pod_eta": c.get("PodEtaDate", ""), "latest_move": c.get("LatestMove", ""), "container_type": c.get("ContainerType", ""), "delivered": c.get("Delivered", False), "events": events, "_fetched_at": int(time.time()), } # --- Endpoints --- @app.get("/health") def health() -> Dict[str, Any]: return {"ok": True, "service": "camoufox-msc-worker"} @app.post("/track") def track(req: TrackRequest, response: Response) -> Dict[str, Any]: # Cache hit now = time.time() cached = _cache.get(req.container) if cached and (now - cached[0]) < CACHE_TTL_SECONDS: log.info(f"cache hit for {req.container}") response.headers["X-Cache"] = "HIT" payload = dict(cached[1]) payload["_cached"] = True return payload log.info(f"tracking {req.container}") try: with Camoufox(headless=True, humanize=True) as browser: page = browser.new_page() page.goto( "https://www.msc.com/en/track-a-shipment", wait_until="domcontentloaded", ) page.wait_for_timeout(2500) # Descartar cookie banner (one trust) try: page.get_by_role("button", name=re.compile("accept", re.I)).click( timeout=2000, force=True ) except Exception: pass page.wait_for_timeout(500) page.locator("#trackingNumber").fill(req.container) # Set up waiter ANTES de disparar la busqueda with page.expect_response( lambda r: "TrackingInfo" in r.url, timeout=30000 ) as resp_info: page.locator("#trackingNumber").press("Enter") api_resp = resp_info.value if api_resp.status != 200: raise HTTPException( 502, f"MSC devolvio status={api_resp.status}, probable bloqueo de Akamai", ) raw = json.loads(api_resp.body()) except HTTPException: raise except Exception as e: log.exception("error en la sesion de Camoufox") raise HTTPException(502, f"browser session error: {e}") if not raw.get("IsSuccess"): raise HTTPException(404, f"MSC: {raw.get('ErrorMessage') or raw}") payload = flatten(raw) _cache[req.container] = (now, payload) response.headers["X-Cache"] = "MISS" return payload # --- Track completo: JSON + screenshot (para integraciones tipo Andres) --- class TrackFullRequest(BaseModel): container: str = Field(..., min_length=4, max_length=20, pattern=r"^[A-Z0-9]+$") fresh: bool = False @app.post("/track_full") def track_full(req: TrackFullRequest, response: Response): now = time.time() # Si fresh=True, SIEMPRE fetch nuevo. Si no, usa cache. if not req.fresh: cached = _cache.get(req.container) if cached and (now - cached[0]) < CACHE_TTL_SECONDS: log.info(f"cache hit (full) for {req.container}") response.headers["X-Cache"] = "HIT" payload = dict(cached[1]) payload["_cached"] = True return {"json": payload, "screenshot": None} log.info(f"track_full {req.container} fresh={req.fresh}") import base64 try: with Camoufox(headless=True, humanize=True) as browser: page = browser.new_page() page.goto( "https://www.msc.com/en/track-a-shipment", wait_until="domcontentloaded", ) page.wait_for_timeout(2500) try: page.get_by_role("button", name=re.compile("accept", re.I)).click( timeout=2000, force=True ) except Exception: pass page.wait_for_timeout(500) page.locator("#trackingNumber").fill(req.container) with page.expect_response( lambda r: "TrackingInfo" in r.url, timeout=30000 ) as resp_info: page.locator("#trackingNumber").press("Enter") api_resp = resp_info.value if api_resp.status != 200: raise HTTPException(502, f"MSC status={api_resp.status}") raw = json.loads(api_resp.body()) page.wait_for_timeout(2000) screenshot_bytes = page.screenshot(full_page=True, type="png") screenshot_b64 = base64.b64encode(screenshot_bytes).decode("ascii") except HTTPException: raise except Exception as e: log.exception("error en track_full") raise HTTPException(502, f"browser error: {e}") if not raw.get("IsSuccess"): raise HTTPException(404, f"MSC: {raw.get('ErrorMessage') or raw}") payload = flatten(raw) _cache[req.container] = (now, payload) response.headers["X-Cache"] = "MISS" return {"json": payload, "screenshot": screenshot_b64, "screenshot_format": "png"} # --- Main --- if __name__ == "__main__": log.info(f"arrancando en {HOST}:{PORT}") uvicorn.run(app, host=HOST, port=PORT, log_level="info")