Add ShopifySovereignBridge for Robert biometric cart sync and inventory adjustment

api/shopify_bridge.py

@@ -21,6 +21,9 @@
 import urllib.error
 import urllib.parse
 import urllib.request
+from typing import Any
+
+import requests as _requests
 
 SIREN_SELL = "943 610 196"
 PATENTE = "PCT/EP2025/067317"
@@ -130,3 +133,97 @@ def resolve_shopify_checkout_url(lead_id: int, fabric_sensation: str) -> str | N
     if inv:
         return inv
     return build_shopify_perfect_selection_url(lead_id, fabric_sensation)
+
+
+class ShopifySovereignBridge:
+    """
+    Puente soberano Shopify — inyecta métricas biométricas de Robert en el carrito
+    y sincroniza el stock del Armario Solidario contra el Admin API de Shopify.
+
+    Patente: PCT/EP2025/067317
+    """
+
+    def __init__(self, api_key: str, shop_url: str) -> None:
+        host = shop_url.replace("https://", "").replace("http://", "").split("/")[0]
+        ver = os.environ.get("SHOPIFY_ADMIN_API_VERSION", "2026-04").strip() or "2026-04"
+        self.api_base = f"https://{host}/admin/api/{ver}"
+        self.headers: dict[str, str] = {
+            "X-Shopify-Access-Token": api_key,
+            "Content-Type": "application/json",
+        }
+
+    def sync_robert_to_shopify(
+        self, look_id: int, engine_metrics: dict[str, Any]
+    ) -> str | None:
+        """
+        Inyecta las métricas de Robert (talla recomendada por biometría)
+        en el carrito de Shopify para evitar devoluciones.
+
+        Returns the draft order invoice_url on success, or None on failure.
+        """
+        fit_score = engine_metrics.get("fitScore", 0)
+        payload: dict[str, Any] = {
+            "draft_order": {
+                "line_items": [
+                    {
+                        "variant_id": look_id,
+                        "quantity": 1,
+                        "properties": [
+                            {
+                                "name": "Robert_Fit_Score",
+                                "value": f"{fit_score}%",
+                            },
+                            {
+                                "name": "Biometric_Validation",
+                                "value": "Sovereign_V10",
+                            },
+                        ],
+                    }
+                ],
+                "note": "Venta realizada a través de Espejo Digital TryOnYou.",
+                "tags": f"TryOnYou,Robert,PCT_EP2025_067317,FitScore_{fit_score}",
+            }
+        }
+        try:
+            response = _requests.post(
+                f"{self.api_base}/draft_orders.json",
+                json=payload,
+                headers=self.headers,
+                timeout=15,
+            )
+        except _requests.RequestException:
+            return None
+        if response.status_code not in (200, 201):
+            return None
+        inv = response.json().get("draft_order", {}).get("invoice_url")
+        if isinstance(inv, str) and inv.startswith("http"):
+            return inv
+        return f"Checkout de Shopify listo para Look {look_id}"
+
+    def update_inventory_physics(self, fabric_key: str, stock_change: int) -> bool:
+        """
+        Actualiza stock cuando el Armario Solidario retira una prenda.
+
+        Calls Shopify Inventory API (inventory_levels/adjust) for the given
+        inventory item (fabric_key = inventory_item_id) at the configured
+        SHOPIFY_LOCATION_ID location.  Returns True on success, False on failure.
+        """
+        print(f"Sincronizando stock en Shopify para {fabric_key}: {stock_change}")
+        location_id_raw = os.environ.get("SHOPIFY_LOCATION_ID", "").strip()
+        if not location_id_raw.isdigit():
+            return False
+        payload: dict[str, Any] = {
+            "location_id": int(location_id_raw),
+            "inventory_item_id": fabric_key,
+            "available_adjustment": stock_change,
+        }
+        try:
+            response = _requests.post(
+                f"{self.api_base}/inventory_levels/adjust.json",
+                json=payload,
+                headers=self.headers,
+                timeout=15,
+            )
+        except _requests.RequestException:
+            return False
+        return response.status_code in (200, 201)

tests/test_shopify_bridge.py

@@ -0,0 +1,211 @@
+"""Tests for ShopifySovereignBridge — Protocolo Robert / Biometría V10."""
+
+from __future__ import annotations
+
+import os
+import sys
+import unittest
+from unittest.mock import MagicMock, patch
+
+_ROOT = os.path.normpath(os.path.join(os.path.dirname(__file__), ".."))
+_API = os.path.join(_ROOT, "api")
+for _p in (_ROOT, _API):
+    if _p not in sys.path:
+        sys.path.insert(0, _p)
+
+from shopify_bridge import ShopifySovereignBridge  # noqa: E402
+
+
+class TestShopifySovereignBridgeInit(unittest.TestCase):
+    def test_api_base_built_from_shop_url(self) -> None:
+        bridge = ShopifySovereignBridge("tok", "mystore.myshopify.com")
+        self.assertIn("mystore.myshopify.com", bridge.api_base)
+        self.assertTrue(bridge.api_base.startswith("https://"))
+
+    def test_api_base_strips_https_prefix(self) -> None:
+        bridge = ShopifySovereignBridge("tok", "https://mystore.myshopify.com")
+        self.assertNotIn("https://https://", bridge.api_base)
+        self.assertIn("mystore.myshopify.com", bridge.api_base)
+
+    def test_headers_contain_access_token(self) -> None:
+        bridge = ShopifySovereignBridge("secret-key", "mystore.myshopify.com")
+        self.assertEqual(bridge.headers["X-Shopify-Access-Token"], "secret-key")
+
+    def test_headers_content_type_json(self) -> None:
+        bridge = ShopifySovereignBridge("tok", "mystore.myshopify.com")
+        self.assertEqual(bridge.headers["Content-Type"], "application/json")
+
+    def test_api_base_contains_admin_api(self) -> None:
+        bridge = ShopifySovereignBridge("tok", "mystore.myshopify.com")
+        self.assertIn("/admin/api/", bridge.api_base)
+
+
+class TestSyncRobertToShopify(unittest.TestCase):
+    def _make_bridge(self) -> ShopifySovereignBridge:
+        return ShopifySovereignBridge("tok", "mystore.myshopify.com")
+
+    def test_returns_invoice_url_on_success(self) -> None:
+        bridge = self._make_bridge()
+        mock_resp = MagicMock()
+        mock_resp.status_code = 201
+        mock_resp.json.return_value = {
+            "draft_order": {"invoice_url": "https://mystore.myshopify.com/invoices/abc"}
+        }
+        with patch("shopify_bridge._requests.post", return_value=mock_resp):
+            result = bridge.sync_robert_to_shopify(42, {"fitScore": 95})
+        self.assertEqual(result, "https://mystore.myshopify.com/invoices/abc")
+
+    def test_returns_fallback_string_when_no_invoice_url(self) -> None:
+        bridge = self._make_bridge()
+        mock_resp = MagicMock()
+        mock_resp.status_code = 201
+        mock_resp.json.return_value = {"draft_order": {}}
+        with patch("shopify_bridge._requests.post", return_value=mock_resp):
+            result = bridge.sync_robert_to_shopify(7, {"fitScore": 88})
+        self.assertIsNotNone(result)
+        self.assertIn("7", str(result))
+
+    def test_returns_none_on_http_error(self) -> None:
+        bridge = self._make_bridge()
+        mock_resp = MagicMock()
+        mock_resp.status_code = 401
+        mock_resp.json.return_value = {}
+        with patch("shopify_bridge._requests.post", return_value=mock_resp):
+            result = bridge.sync_robert_to_shopify(1, {"fitScore": 80})
+        self.assertIsNone(result)
+
+    def test_returns_none_on_request_exception(self) -> None:
+        import requests as real_requests
+
+        bridge = self._make_bridge()
+        with patch(
+            "shopify_bridge._requests.post",
+            side_effect=real_requests.RequestException("timeout"),
+        ):
+            result = bridge.sync_robert_to_shopify(1, {"fitScore": 80})
+        self.assertIsNone(result)
+
+    def test_payload_includes_fit_score(self) -> None:
+        bridge = self._make_bridge()
+        mock_resp = MagicMock()
+        mock_resp.status_code = 201
+        mock_resp.json.return_value = {"draft_order": {}}
+        with patch("shopify_bridge._requests.post", return_value=mock_resp) as mock_post:
+            bridge.sync_robert_to_shopify(99, {"fitScore": 77})
+        _, kwargs = mock_post.call_args
+        line = kwargs["json"]["draft_order"]["line_items"][0]
+        props = {p["name"]: p["value"] for p in line["properties"]}
+        self.assertEqual(props["Robert_Fit_Score"], "77%")
+        self.assertEqual(props["Biometric_Validation"], "Sovereign_V10")
+
+    def test_payload_includes_look_id_as_variant(self) -> None:
+        bridge = self._make_bridge()
+        mock_resp = MagicMock()
+        mock_resp.status_code = 201
+        mock_resp.json.return_value = {"draft_order": {}}
+        with patch("shopify_bridge._requests.post", return_value=mock_resp) as mock_post:
+            bridge.sync_robert_to_shopify(55, {"fitScore": 90})
+        _, kwargs = mock_post.call_args
+        line = kwargs["json"]["draft_order"]["line_items"][0]
+        self.assertEqual(line["variant_id"], 55)
+        self.assertEqual(line["quantity"], 1)
+
+    def test_note_mentions_tryonyou(self) -> None:
+        bridge = self._make_bridge()
+        mock_resp = MagicMock()
+        mock_resp.status_code = 201
+        mock_resp.json.return_value = {"draft_order": {}}
+        with patch("shopify_bridge._requests.post", return_value=mock_resp) as mock_post:
+            bridge.sync_robert_to_shopify(1, {"fitScore": 85})
+        _, kwargs = mock_post.call_args
+        note = kwargs["json"]["draft_order"]["note"]
+        self.assertIn("TryOnYou", note)
+
+
+class TestUpdateInventoryPhysics(unittest.TestCase):
+    def _make_bridge(self) -> ShopifySovereignBridge:
+        return ShopifySovereignBridge("tok", "mystore.myshopify.com")
+
+    def _set_location(self, loc: str = "98765") -> None:
+        os.environ["SHOPIFY_LOCATION_ID"] = loc
+
+    def tearDown(self) -> None:
+        os.environ.pop("SHOPIFY_LOCATION_ID", None)
+
+    def test_returns_true_on_success(self) -> None:
+        self._set_location()
+        bridge = self._make_bridge()
+        mock_resp = MagicMock()
+        mock_resp.status_code = 200
+        with patch("shopify_bridge._requests.post", return_value=mock_resp):
+            result = bridge.update_inventory_physics("LOC-123", -1)
+        self.assertTrue(result)
+
+    def test_returns_false_without_location_id_env(self) -> None:
+        os.environ.pop("SHOPIFY_LOCATION_ID", None)
+        bridge = self._make_bridge()
+        mock_resp = MagicMock()
+        mock_resp.status_code = 200
+        with patch("shopify_bridge._requests.post", return_value=mock_resp):
+            result = bridge.update_inventory_physics("LOC-123", -1)
+        self.assertFalse(result)
+
+    def test_returns_false_on_http_error(self) -> None:
+        self._set_location()
+        bridge = self._make_bridge()
+        mock_resp = MagicMock()
+        mock_resp.status_code = 422
+        with patch("shopify_bridge._requests.post", return_value=mock_resp):
+            result = bridge.update_inventory_physics("LOC-123", -1)
+        self.assertFalse(result)
+
+    def test_returns_false_on_request_exception(self) -> None:
+        import requests as real_requests
+
+        self._set_location()
+        bridge = self._make_bridge()
+        with patch(
+            "shopify_bridge._requests.post",
+            side_effect=real_requests.RequestException("conn error"),
+        ):
+            result = bridge.update_inventory_physics("LOC-999", -2)
+        self.assertFalse(result)
+
+    def test_prints_sync_message(self) -> None:
+        self._set_location()
+        bridge = self._make_bridge()
+        mock_resp = MagicMock()
+        mock_resp.status_code = 200
+        with patch("shopify_bridge._requests.post", return_value=mock_resp):
+            with patch("builtins.print") as mock_print:
+                bridge.update_inventory_physics("FABRIC-7", -3)
+        mock_print.assert_called_once()
+        args = mock_print.call_args[0][0]
+        self.assertIn("FABRIC-7", args)
+        self.assertIn("-3", args)
+
+    def test_calls_inventory_adjust_endpoint(self) -> None:
+        self._set_location()
+        bridge = self._make_bridge()
+        mock_resp = MagicMock()
+        mock_resp.status_code = 200
+        with patch("shopify_bridge._requests.post", return_value=mock_resp) as mock_post:
+            bridge.update_inventory_physics("LOC-42", 5)
+        url = mock_post.call_args[0][0]
+        self.assertIn("inventory_levels/adjust", url)
+
+    def test_payload_uses_fabric_key_as_inventory_item_id(self) -> None:
+        self._set_location("12345")
+        bridge = self._make_bridge()
+        mock_resp = MagicMock()
+        mock_resp.status_code = 200
+        with patch("shopify_bridge._requests.post", return_value=mock_resp) as mock_post:
+            bridge.update_inventory_physics("ITEM-99", -2)
+        _, kwargs = mock_post.call_args
+        self.assertEqual(kwargs["json"]["inventory_item_id"], "ITEM-99")
+        self.assertEqual(kwargs["json"]["location_id"], 12345)
+        self.assertEqual(kwargs["json"]["available_adjustment"], -2)
+
+
+if __name__ == "__main__":
+    unittest.main()