Upload core/cdn.py with huggingface_hub

core/cdn.py

@@ -59,13 +59,15 @@ class CDNManager:
                 with open(self.private_key_path, "rb") as f:
                     self.private_key = rsa.PrivateKey.load_pkcs1(f.read())
 
-            print(f"✅ Loaded private key for {self.provider} signing")
+            print(f"Loaded private key for {self.provider} signing")
 
         except Exception as e:
-            print(f"❌ Failed to load private key: {e}")
+            logger.error(f"Failed to load private key: {e}")
             self.enable_signing = False
 
-    def get_asset_url(self, asset_path: str, signed: bool = False, expiry_seconds: int = 3600) -> str:
+    def get_asset_url(
+        self, asset_path: str, signed: bool = False, expiry_seconds: int = 3600
+    ) -> str:
         """
         Get the full CDN URL for an asset with optional signing.
         Falls back to local path if CDN is not configured.
@@ -107,7 +109,14 @@ class CDNManager:
             # Create the policy
             expires = int(time.time()) + expiry_seconds
 
-            policy = {"Statement": [{"Resource": url, "Condition": {"DateLessThan": {"AWS:EpochTime": expires}}}]}
+            policy = {
+                "Statement": [
+                    {
+                        "Resource": url,
+                        "Condition": {"DateLessThan": {"AWS:EpochTime": expires}},
+                    }
+                ]
+            }
 
             # Convert policy to JSON and base64
             policy_json = json.dumps(policy, separators=(",", ":"))
@@ -118,13 +127,17 @@ class CDNManager:
             signature_b64 = base64.b64encode(signature).decode("utf-8")
 
             # Construct signed URL
-            params = {"Policy": policy_b64, "Signature": signature_b64, "Key-Pair-Id": self.key_pair_id}
+            params = {
+                "Policy": policy_b64,
+                "Signature": signature_b64,
+                "Key-Pair-Id": self.key_pair_id,
+            }
 
             signed_url = f"{base_url}?{urlencode(params)}"
             return signed_url
 
         except Exception as e:
-            print(f"❌ CloudFront signing failed: {e}")
+            logger.error(f"CloudFront signing failed: {e}")
             return url
 
     def _sign_cloudflare_url(self, url: str, expiry_seconds: int) -> str:
@@ -136,14 +149,16 @@ class CDNManager:
             token_data = f"{url}{expires}"
 
             # Create HMAC signature
-            signature = hmac.new(self.private_key, token_data.encode("utf-8"), hashlib.sha256).hexdigest()
+            signature = hmac.new(
+                self.private_key, token_data.encode("utf-8"), hashlib.sha256
+            ).hexdigest()
 
             # Construct signed URL
             signed_url = f"{url}?expires={expires}&signature={signature}"
             return signed_url
 
         except Exception as e:
-            print(f"❌ Cloudflare signing failed: {e}")
+            logger.error(f"Cloudflare signing failed: {e}")
             return url
 
     def _sign_generic_url(self, url: str, expiry_seconds: int) -> str:
@@ -156,7 +171,11 @@ class CDNManager:
 
             # Create HMAC signature
             signature = hmac.new(
-                self.private_key if isinstance(self.private_key, bytes) else str(self.private_key).encode(),
+                (
+                    self.private_key
+                    if isinstance(self.private_key, bytes)
+                    else str(self.private_key).encode()
+                ),
                 message.encode("utf-8"),
                 hashlib.sha256,
             ).hexdigest()
@@ -166,14 +185,16 @@ class CDNManager:
             return signed_url
 
         except Exception as e:
-            print(f"❌ Generic signing failed: {e}")
+            logger.error(f"Generic signing failed: {e}")
             return url
 
     def validate_signed_url(self, signed_url: str) -> bool:
         """Validate that a signed URL is properly signed and not expired"""
         try:
             parsed = urlparse(signed_url)
-            query_params = dict(param.split("=") for param in parsed.query.split("&") if "=" in param)
+            query_params = dict(
+                param.split("=") for param in parsed.query.split("&") if "=" in param
+            )
 
             expires = int(query_params.get("expires", 0))
             signature = query_params.get("signature", "")
@@ -188,7 +209,11 @@ class CDNManager:
 
             # Verify signature
             expected_signature = hmac.new(
-                self.private_key if isinstance(self.private_key, bytes) else str(self.private_key).encode(),
+                (
+                    self.private_key
+                    if isinstance(self.private_key, bytes)
+                    else str(self.private_key).encode()
+                ),
                 message.encode("utf-8"),
                 hashlib.sha256,
             ).hexdigest()
@@ -204,7 +229,9 @@ class CDNManager:
             "signing_enabled": self.enable_signing,
             "provider": self.provider,
             "private_key_loaded": self.private_key is not None,
-            "key_pair_id": self.key_pair_id is not None if self.provider == "cloudfront" else None,
+            "key_pair_id": (
+                self.key_pair_id is not None if self.provider == "cloudfront" else None
+            ),
             "cdn_url": self.cdn_url,
         }
 
@@ -233,7 +260,9 @@ cdn_service = create_cdn_manager()
 # Utility functions for CDN operations
 def generate_signed_asset_url(asset_path: str, expiry_seconds: int = 3600) -> str:
     """Generate a signed URL for a protected asset"""
-    return cdn_service.get_asset_url(asset_path, signed=True, expiry_seconds=expiry_seconds)
+    return cdn_service.get_asset_url(
+        asset_path, signed=True, expiry_seconds=expiry_seconds
+    )
 
 
 def validate_asset_access(signed_url: str) -> bool:
@@ -244,7 +273,12 @@ def validate_asset_access(signed_url: str) -> bool:
 def get_cdn_health_status() -> dict[str, Any]:
     """Get comprehensive CDN health and configuration status"""
     status = cdn_service.get_signing_status()
-    status.update({"cdn_reachable": _test_cdn_connectivity(), "signing_functional": _test_signing_functionality()})
+    status.update(
+        {
+            "cdn_reachable": _test_cdn_connectivity(),
+            "signing_functional": _test_signing_functionality(),
+        }
+    )
     return status