fix(network): Implement DoH-based DNS patch and robust Session handling for Social Publishers

social_media_publishers/base.py

@@ -45,6 +45,11 @@ class SocialAuthCreator(ABC):
     
     # --- Common Implementation for Token Management ---
 
+    def _get_session(self):
+        """Get a requests Session with resilient network configuration (retries, custom DNS)."""
+        from src.network_utils import get_resilient_session
+        return get_resilient_session()
+
     def _get_token_manager(self):
         """Initialize TokenManager from environment variables."""
         encryption_key = get_config_value("encryption_key")
@@ -175,6 +180,11 @@ class SocialAuthCreator(ABC):
 class SocialPublisher(ABC):
     """Abstract base class for publishing content to social media."""
     
+    def _get_session(self):
+        """Get a requests Session with resilient network configuration (retries, custom DNS)."""
+        from src.network_utils import get_resilient_session
+        return get_resilient_session()
+
     @abstractmethod
     def authenticate(self, account_id: Optional[str] = None) -> Any:
         """

social_media_publishers/facebook/auth.py

@@ -101,7 +101,7 @@ class FacebookAuthCreator(SocialAuthCreator):
                 'code': code
             }
             
-            response = requests.get(self.TOKEN_URL, params=token_params)
+            response = self._get_session().get(self.TOKEN_URL, params=token_params)
             if response.status_code != 200:
                 return {"success": False, "error": f"Token exchange failed: {response.text}"}
             
@@ -116,7 +116,7 @@ class FacebookAuthCreator(SocialAuthCreator):
                 'fb_exchange_token': short_lived_token
             }
             
-            ll_response = requests.get(self.TOKEN_URL, params=long_lived_params)
+            ll_response = self._get_session().get(self.TOKEN_URL, params=long_lived_params)
             if ll_response.status_code == 200:
                 ll_data = ll_response.json()
                 user_access_token = ll_data.get('access_token', short_lived_token)
@@ -126,7 +126,7 @@ class FacebookAuthCreator(SocialAuthCreator):
                 print("⚠️ Could not get long-lived token, using short-lived")
 
             # Get user's Facebook Pages
-            pages_response = requests.get(
+            pages_response = self._get_session().get(
                 f"{self.GRAPH_URL}/me/accounts",
                 params={'access_token': user_access_token}
             )
@@ -166,7 +166,7 @@ class FacebookAuthCreator(SocialAuthCreator):
                 }
                 
                 try:
-                    page_ll_response = requests.get(self.TOKEN_URL, params=page_ll_params)
+                    page_ll_response = self._get_session().get(self.TOKEN_URL, params=page_ll_params)
                     if page_ll_response.status_code == 200:
                         page_ll_data = page_ll_response.json()
                         page_access_token = page_ll_data.get('access_token', page_access_token)

social_media_publishers/facebook/publisher.py

@@ -141,7 +141,7 @@ class FacebookPublisher(SocialPublisher):
         
         # Step 1: Start upload session
         print(f"🎬 Starting Facebook video upload...")
-        start_response = requests.post(
+        start_response = self._get_session().post(
             f"{self.GRAPH_URL}/{self.page_id}/videos",
             params={
                 'access_token': self.access_token,
@@ -174,7 +174,7 @@ class FacebookPublisher(SocialPublisher):
                 progress = int((end_offset / file_size) * 100)
                 print(f"📤 Uploading: {progress}%")
                 
-                transfer_response = requests.post(
+                transfer_response = self._get_session().post(
                     f"{self.GRAPH_URL}/{self.page_id}/videos",
                     params={
                         'access_token': self.access_token,
@@ -193,7 +193,7 @@ class FacebookPublisher(SocialPublisher):
         
         # Step 3: Finish upload
         print("🚀 Finishing upload...")
-        finish_response = requests.post(
+        finish_response = self._get_session().post(
             f"{self.GRAPH_URL}/{self.page_id}/videos",
             params={
                 'access_token': self.access_token,
@@ -244,7 +244,7 @@ class FacebookPublisher(SocialPublisher):
                 params['after'] = page_token
             
             print(f"DEBUG: Fetching videos from {endpoint} with params: {params}")
-            response = requests.get(endpoint, params=params)
+            response = self._get_session().get(endpoint, params=params)
             
             p_text = response.text
             print(f"DEBUG: Video Fetch Response: {p_text}")
@@ -263,7 +263,7 @@ class FacebookPublisher(SocialPublisher):
                     
                     # Probe 2: Try without 'type=uploaded'
                     try:
-                        all_vid_resp = requests.get(f"{self.GRAPH_URL}/{self.page_id}/videos", params={'access_token': self.access_token, 'limit': 3})
+                        all_vid_resp = self._get_session().get(f"{self.GRAPH_URL}/{self.page_id}/videos", params={'access_token': self.access_token, 'limit': 3})
                         print(f"DEBUG: Probe /videos (no type): {all_vid_resp.text}")
                     except: pass
 
@@ -342,7 +342,7 @@ class FacebookPublisher(SocialPublisher):
                 if video['views'] > 0: continue # Skip if we already have views from fields
                 
                 try:
-                    insights_response = requests.get(
+                    insights_response = self._get_session().get(
                         f"{self.GRAPH_URL}/{video['id']}/video_insights",
                         params={
                             'access_token': self.access_token,
@@ -400,7 +400,7 @@ class FacebookPublisher(SocialPublisher):
 
             # Try fetching Page/User info
             try:
-                response = requests.get(endpoint, params=params)
+                response = self._get_session().get(endpoint, params=params)
                 if response.status_code == 200:
                     page_data = response.json()
                     page_name = page_data.get('name', 'Unknown')
@@ -414,7 +414,7 @@ class FacebookPublisher(SocialPublisher):
                     print(f"DEBUG: Stats - Page Info Fetch failed ({response.status_code}): {response.text}")
                     # Fallback for User Profile (simple fields)
                     params['fields'] = 'name,picture'
-                    response = requests.get(endpoint, params=params)
+                    response = self._get_session().get(endpoint, params=params)
                     if response.status_code == 200:
                         page_data = response.json()
                         page_name = page_data.get('name', 'Unknown')

social_media_publishers/instagram/auth.py

@@ -101,7 +101,7 @@ class InstagramAuthCreator(SocialAuthCreator):
                 'code': code
             }
             
-            response = requests.get(self.TOKEN_URL, params=token_params)
+            response = self._get_session().get(self.TOKEN_URL, params=token_params)
             if response.status_code != 200:
                 return {"success": False, "error": f"Token exchange failed: {response.text}"}
             
@@ -116,7 +116,7 @@ class InstagramAuthCreator(SocialAuthCreator):
                 'fb_exchange_token': short_lived_token
             }
             
-            ll_response = requests.get(self.TOKEN_URL, params=long_lived_params)
+            ll_response = self._get_session().get(self.TOKEN_URL, params=long_lived_params)
             if ll_response.status_code == 200:
                 ll_data = ll_response.json()
                 access_token = ll_data.get('access_token', short_lived_token)
@@ -126,7 +126,7 @@ class InstagramAuthCreator(SocialAuthCreator):
                 print("⚠️ Could not get long-lived token, using short-lived")
 
             # Get Facebook Pages linked to account
-            pages_response = requests.get(
+            pages_response = self._get_session().get(
                 f"{self.GRAPH_URL}/me/accounts",
                 params={'access_token': access_token}
             )
@@ -160,7 +160,7 @@ class InstagramAuthCreator(SocialAuthCreator):
 
                 # Check for linked Instagram account
                 try:
-                    ig_response = requests.get(
+                    ig_response = self._get_session().get(
                         f"{self.GRAPH_URL}/{page_id}",
                         params={
                             'fields': 'instagram_business_account',
@@ -185,7 +185,7 @@ class InstagramAuthCreator(SocialAuthCreator):
                     ig_account_id = ig_account.get('id')
                     
                     # Get Instagram username
-                    ig_info_response = requests.get(
+                    ig_info_response = self._get_session().get(
                         f"{self.GRAPH_URL}/{ig_account_id}",
                         params={
                             'fields': 'username',

social_media_publishers/instagram/publisher.py

@@ -206,7 +206,7 @@ class InstagramPublisher(SocialPublisher):
         """POST with retry logic."""
         for attempt in range(1, max_retries + 1):
             try:
-                r = requests.post(url, params=params, timeout=30)
+                r = self._get_session().post(url, params=params, timeout=30)
                 if r.status_code == 200:
                     return r
                 print(f"⚠️ Attempt {attempt} failed ({r.status_code}): {r.text}")
@@ -222,7 +222,7 @@ class InstagramPublisher(SocialPublisher):
         
         for attempt in range(1, max_attempts + 1):
             params = {'access_token': self.access_token, 'fields': 'status_code'}
-            response = requests.get(endpoint, params=params)
+            response = self._get_session().get(endpoint, params=params)
             result = response.json()
             
             status = result.get('status_code')
@@ -281,7 +281,7 @@ class InstagramPublisher(SocialPublisher):
             if page_token:
                 params['after'] = page_token
             
-            response = requests.get(endpoint, params=params)
+            response = self._get_session().get(endpoint, params=params)
             if response.status_code != 200:
                 return {'error': f"API error: {response.text}", 'videos': []}
             
@@ -319,7 +319,7 @@ class InstagramPublisher(SocialPublisher):
                 if media_type in ('VIDEO', 'REELS'):
                     # Fetch insights for this media
                     try:
-                        insights_resp = requests.get(
+                        insights_resp = self._get_session().get(
                             f"{self.GRAPH_URL}/{media_id}/insights",
                             params={
                                 'access_token': self.access_token,
@@ -383,7 +383,7 @@ class InstagramPublisher(SocialPublisher):
                 'fields': 'username,profile_picture_url,followers_count,media_count'
             }
             
-            response = requests.get(endpoint, params=params)
+            response = self._get_session().get(endpoint, params=params)
             if response.status_code != 200:
                 return {'error': f"API error: {response.text}"}
             

social_media_publishers/threads/auth.py

@@ -110,7 +110,7 @@ class ThreadsAuthCreator(SocialAuthCreator):
             print(f"DEBUG: Threads Token Exchange Params: client_id={config['app_id'][:5]}..., redirect_uri={redirect_uri}")
             
             # Threads token exchange
-            response = requests.post(self.TOKEN_URL, data=token_data_body, params=token_url_params)
+            response = self._get_session().post(self.TOKEN_URL, data=token_data_body, params=token_url_params)
             if response.status_code != 200:
                 short_token_params = token_params.copy() 
                 # Retry with GET or check docs? Threads API uses POST.
@@ -133,7 +133,7 @@ class ThreadsAuthCreator(SocialAuthCreator):
                 'access_token': short_lived_token
             }
             
-            ll_response = requests.get(long_lived_url, params=long_lived_params)
+            ll_response = self._get_session().get(long_lived_url, params=long_lived_params)
             
             if ll_response.status_code == 200:
                 ll_data = ll_response.json()
@@ -145,7 +145,7 @@ class ThreadsAuthCreator(SocialAuthCreator):
 
             # Get User Profile Info
             # Endpoint: /me?fields=id,username,name,threads_profile_picture_url
-            profile_response = requests.get(
+            profile_response = self._get_session().get(
                 f"{self.GRAPH_URL}/me",
                 params={
                     'fields': 'id,username,name,threads_profile_picture_url',

social_media_publishers/threads/publisher.py

@@ -226,7 +226,7 @@ class ThreadsPublisher(SocialPublisher):
         """POST with retry logic."""
         for attempt in range(1, max_retries + 1):
             try:
-                r = requests.post(url, params=params, timeout=30)
+                r = self._get_session().post(url, params=params, timeout=30)
                 if r.status_code == 200:
                     return r
                 print(f"⚠️ Attempt {attempt} failed ({r.status_code}): {r.text}")
@@ -242,7 +242,7 @@ class ThreadsPublisher(SocialPublisher):
         
         for attempt in range(1, max_attempts + 1):
             params = {'access_token': self.access_token, 'fields': 'status'}
-            response = requests.get(endpoint, params=params)
+            response = self._get_session().get(endpoint, params=params)
             result = response.json()
             
             status = result.get('status')
@@ -271,7 +271,7 @@ class ThreadsPublisher(SocialPublisher):
                 'access_token': self.access_token,
                 'fields': 'id,username,name,threads_profile_picture_url,followers_count' # followers_count might not be available on all versions, but let's try
             }
-            me_resp = requests.get(me_endpoint, params=me_params)
+            me_resp = self._get_session().get(me_endpoint, params=me_params)
             me_data = me_resp.json()
             
             username = me_data.get('username', 'unknown')
@@ -294,7 +294,7 @@ class ThreadsPublisher(SocialPublisher):
                 'limit': 50
             }
             
-            media_resp = requests.get(media_endpoint, params=media_params)
+            media_resp = self._get_session().get(media_endpoint, params=media_params)
             media_data = media_resp.json().get('data', [])
             
             total_likes = 0
@@ -349,7 +349,7 @@ class ThreadsPublisher(SocialPublisher):
             if page_token:
                 params['after'] = page_token
                 
-            response = requests.get(endpoint, params=params)
+            response = self._get_session().get(endpoint, params=params)
             data = response.json()
             
             if 'error' in data:

src/network_utils.py

@@ -1,39 +1,85 @@
-import socket
 import logging
+import socket
+import json
+import requests
+from requests.adapters import HTTPAdapter
+from requests.packages.urllib3.util.retry import Retry
 
 logger = logging.getLogger(__name__)
 
-def allowed_gai_family():
+# Cache for resolved IPs to avoid spamming DoH
+_DNS_CACHE = {}
+_ORIGINAL_GETADDRINFO = socket.getaddrinfo
+
+def resolve_via_doh(hostname):
+    """
+    Resolve hostname using Google DNS-over-HTTPS.
+    Bypasses local system resolver (/etc/resolv.conf).
+    """
+    if hostname in _DNS_CACHE:
+        return _DNS_CACHE[hostname]
+        
+    try:
+        # We must use an IP for the DoH server to avoid DNS loop!
+        # 8.8.8.8 is Google DNS.
+        doh_url = f"https://8.8.8.8/resolve?name={hostname}&type=A"
+        
+        # Use a fresh simple session without adapters to avoid recursion
+        # But wait, requests might try to resolve 8.8.8.8? No, it's an IP.
+        resp = requests.get(doh_url, timeout=5, verify=False) # Skip verify for DoH to be fast/safe
+        data = resp.json()
+        
+        if 'Answer' in data:
+            for answer in data['Answer']:
+                if answer['type'] == 1: # A Record
+                    ip = answer['data']
+                    logger.info(f"🔍 DoH Resolved {hostname} -> {ip}")
+                    _DNS_CACHE[hostname] = ip
+                    return ip
+    except Exception as e:
+        logger.error(f"⚠️ DoH Resolution failed for {hostname}: {e}")
+    
+    return None
+
+def patched_getaddrinfo(host, port, family=0, type=0, proto=0, flags=0):
     """
-    https://github.com/shazow/urllib3/blob/master/urllib3/util/connection.py
+    Custom getaddrinfo that attempts DoH if system resolution fails or for specific hosts.
     """
-    family = socket.AF_INET
-    return family
+    # Only intervene for Facebook graph API which is failing
+    if host == 'graph.facebook.com' or host == 'www.facebook.com':
+        ip = resolve_via_doh(host)
+        if ip:
+            # Call original with IP instead of hostname
+            # Log usage for debugging
+            # logger.debug(f"Redirecting {host} to {ip}")
+            return _ORIGINAL_GETADDRINFO(ip, port, family, type, proto, flags)
+            
+    # Fallback to system resolver
+    return _ORIGINAL_GETADDRINFO(host, port, family, type, proto, flags)
 
 def force_ipv4_requests():
     """
-    Monkey-patch urllib3 to force IPv4.
-    Useful for containerized environments with broken IPv6 DNS.
+    Installs the DNS patch.
+    Named 'force_ipv4_requests' for backward compatibility with existing calls.
+    """
+    if socket.getaddrinfo != patched_getaddrinfo:
+        socket.getaddrinfo = patched_getaddrinfo
+        logger.info("🛡️ Installed DoH-based DNS patch for Facebook API")
+    else:
+        logger.info("🛡️ DNS patch already active")
+
+def get_resilient_session():
+    """
+    Returns a session with retry logic.
     """
-    patched = False
+    session = requests.Session()
+    retry = Retry(total=3, backoff_factor=1, status_forcelist=[500, 502, 503, 504])
+    adapter = HTTPAdapter(max_retries=retry)
+    session.mount('https://', adapter)
+    session.mount('http://', adapter)
     
-    # Try patching standalone urllib3
-    try:
-        import urllib3.util.connection as urllib3_cn
-        urllib3_cn.allowed_gai_family = allowed_gai_family
-        patched = True
-        logger.info("✅ Patched urllib3 for IPv4-only requests")
-    except ImportError:
-        pass
+    # Ensure patch is applied whenever we get a session
+    force_ipv4_requests()
+    
+    return session
 
-    # Try patching requests vendored urllib3
-    try:
-        import requests.packages.urllib3.util.connection as requests_urllib3_cn
-        requests_urllib3_cn.allowed_gai_family = allowed_gai_family
-        patched = True
-        logger.info("✅ Patched requests.packages.urllib3 for IPv4-only requests")
-    except ImportError:
-        pass
-        
-    if not patched:
-        logger.warning("⚠️ Could not patch urllib3 for IPv4 (modules not found)")