feat(linkedin): enhance image handling and improve post publishing

- Add support for both file paths and URLs when publishing LinkedIn posts with images
- Implement proper image format detection in image_utils to handle different input types
- Update database schema to include refresh_token for social network accounts
- Improve error handling and logging in LinkedIn service for token exchange and user info fetching
- Add comprehensive tests for keyword trend analyzer component
- Update dependencies to include pytest-cov for test coverage

These changes improve the reliability of image posting to LinkedIn and provide better error handling throughout the service.

Linkedin_poster_dev

@@ -1 +1 @@
-Subproject commit d84516dcdeb9e85fb0d9c89c19fcc4b02d458168
+Subproject commit 556c23cdbb7a7f9f7a2fe03deb092b7efc521653

backend/services/linkedin_service.py

@@ -8,20 +8,20 @@ import logging
 
 class LinkedInService:
     """Service for LinkedIn API integration."""
-    
+
     def __init__(self):
         self.client_id = current_app.config['CLIENT_ID']
         self.client_secret = current_app.config['CLIENT_SECRET']
         self.redirect_uri = current_app.config['REDIRECT_URL']
         self.scope = ['openid', 'profile', 'email', 'w_member_social']
-        
+
     def get_authorization_url(self, state: str) -> str:
         """
         Get LinkedIn authorization URL.
-        
+
         Args:
             state (str): State parameter for security
-            
+
         Returns:
             str: Authorization URL
         """
@@ -31,28 +31,28 @@ class LinkedInService:
             scope=self.scope,
             state=state
         )
-        
+
         authorization_url, _ = linkedin.authorization_url(
             'https://www.linkedin.com/oauth/v2/authorization'
         )
-        
+
         return authorization_url
-    
+
     def get_access_token(self, code: str) -> dict:
         """
         Exchange authorization code for access token.
-        
+
         Args:
             code (str): Authorization code
-            
+
         Returns:
             dict: Token response
         """
         import logging
         logger = logging.getLogger(__name__)
-        
+
         logger.info(f"🔗 [LinkedIn] Starting token exchange for code: {code[:20]}...")
-        
+
         url = "https://www.linkedin.com/oauth/v2/accessToken"
         headers = {
             "Content-Type": "application/x-www-form-urlencoded"
@@ -64,94 +64,94 @@ class LinkedInService:
             "client_id": self.client_id,
             "client_secret": self.client_secret
         }
-        
+
         logger.info(f"🔗 [LinkedIn] Making request to LinkedIn API...")
         logger.info(f"🔗 [LinkedIn] Request URL: {url}")
         logger.info(f"🔗 [LinkedIn] Request data: {data}")
-        
+
         try:
             response = requests.post(url, headers=headers, data=data)
             logger.info(f"🔗 [LinkedIn] Response status: {response.status_code}")
             logger.info(f"🔗 [LinkedIn] Response headers: {dict(response.headers)}")
-            
+
             response.raise_for_status()
-            
+
             token_data = response.json()
             logger.info(f"🔗 [LinkedIn] Token response: {token_data}")
-            
+
             return token_data
         except requests.exceptions.RequestException as e:
             logger.error(f"🔗 [LinkedIn] Token exchange failed: {str(e)}")
             logger.error(f"🔗 [LinkedIn] Error type: {type(e)}")
             raise e
-    
+
     def get_user_info(self, access_token: str) -> dict:
         """
         Get user information from LinkedIn.
-        
+
         Args:
             access_token (str): LinkedIn access token
-            
+
         Returns:
             dict: User information
         """
         import logging
         logger = logging.getLogger(__name__)
-        
+
         logger.info(f"🔗 [LinkedIn] Fetching user info with token length: {len(access_token)}")
-        
+
         url = "https://api.linkedin.com/v2/userinfo"
         headers = {
             "Authorization": f"Bearer {access_token}"
         }
-        
+
         logger.info(f"🔗 [LinkedIn] Making request to LinkedIn user info API...")
         logger.info(f"🔗 [LinkedIn] Request URL: {url}")
         logger.info(f"🔗 [LinkedIn] Request headers: {headers}")
-        
+
         try:
             response = requests.get(url, headers=headers)
             logger.info(f"🔗 [LinkedIn] Response status: {response.status_code}")
             logger.info(f"🔗 [LinkedIn] Response headers: {dict(response.headers)}")
-            
+
             response.raise_for_status()
-            
+
             user_data = response.json()
             logger.info(f"🔗 [LinkedIn] User info response: {user_data}")
-            
+
             return user_data
         except requests.exceptions.RequestException as e:
             logger.error(f"🔗 [LinkedIn] User info fetch failed: {str(e)}")
             logger.error(f"🔗 [LinkedIn] Error type: {type(e)}")
             raise e
-    
+
     def _create_temp_image_file(self, image_bytes: bytes) -> str:
         """
         Create a temporary file from image bytes.
-        
+
         Args:
             image_bytes: Image data as bytes
-            
+
         Returns:
             Path to the temporary file
         """
         # Create a temporary file
         temp_file = tempfile.NamedTemporaryFile(delete=False, suffix='.jpg')
         temp_file_path = temp_file.name
-        
+
         try:
             # Write image bytes to the temporary file
             temp_file.write(image_bytes)
             temp_file.flush()
         finally:
             temp_file.close()
-        
+
         return temp_file_path
-    
+
     def _cleanup_temp_file(self, file_path: str) -> None:
         """
         Safely remove a temporary file.
-        
+
         Args:
             file_path: Path to the temporary file to remove
         """
@@ -161,17 +161,17 @@ class LinkedInService:
         except Exception as e:
             # Log the error but don't fail the operation
             logging.error(f"Failed to cleanup temporary file {file_path}: {str(e)}")
-    
+
     def publish_post(self, access_token: str, user_id: str, text_content: str, image_url: str = None) -> dict:
         """
         Publish a post to LinkedIn.
-        
+
         Args:
             access_token (str): LinkedIn access token
             user_id (str): LinkedIn user ID
             text_content (str): Post content
             image_url (str or bytes, optional): Image URL or image bytes
-            
+
         Returns:
             dict: Publish response
         """
@@ -182,12 +182,12 @@ class LinkedInService:
             "X-Restli-Protocol-Version": "2.0.0",
             "Content-Type": "application/json"
         }
-        
+
         try:
             if image_url and isinstance(image_url, bytes):
                 # Handle bytes data - create temporary file and upload
                 temp_file_path = self._create_temp_image_file(image_url)
-                
+
                 # Register upload
                 register_body = {
                     "registerUploadRequest": {
@@ -199,33 +199,33 @@ class LinkedInService:
                         }]
                     }
                 }
-                
+
                 r = requests.post(
                     "https://api.linkedin.com/v2/assets?action=registerUpload",
                     headers=headers,
                     json=register_body
                 )
-                
+
                 if r.status_code not in (200, 201):
                     raise Exception(f"Failed to register upload: {r.status_code} {r.text}")
-                
+
                 datar = r.json()["value"]
                 upload_url = datar["uploadMechanism"]["com.linkedin.digitalmedia.uploading.MediaUploadHttpRequest"]["uploadUrl"]
                 asset_urn = datar["asset"]
-                
+
                 # Upload image from temporary file
                 upload_headers = {
                     "Authorization": f"Bearer {access_token}",
                     "X-Restli-Protocol-Version": "2.0.0",
                     "Content-Type": "application/octet-stream"
                 }
-                
+
                 with open(temp_file_path, 'rb') as f:
                     image_data = f.read()
                     upload_response = requests.put(upload_url, headers=upload_headers, data=image_data)
                     if upload_response.status_code not in (200, 201):
                         raise Exception(f"Failed to upload image: {upload_response.status_code} {upload_response.text}")
-                
+
                 # Create post with image
                 post_body = {
                     "author": f"urn:li:person:{user_id}",
@@ -245,7 +245,7 @@ class LinkedInService:
                     "visibility": {"com.linkedin.ugc.MemberNetworkVisibility": "PUBLIC"}
                 }
             elif image_url and isinstance(image_url, str):
-                # Handle image upload for URL-based images
+                # Handle image upload - check if it's a file path or URL
                 register_body = {
                     "registerUploadRequest": {
                         "recipes": ["urn:li:digitalmediaRecipe:feedshare-image"],
@@ -256,34 +256,45 @@ class LinkedInService:
                         }]
                     }
                 }
-                
+
                 r = requests.post(
                     "https://api.linkedin.com/v2/assets?action=registerUpload",
                     headers=headers,
                     json=register_body
                 )
-                
+
                 if r.status_code not in (200, 201):
                     raise Exception(f"Failed to register upload: {r.status_code} {r.text}")
-                
+
                 datar = r.json()["value"]
                 upload_url = datar["uploadMechanism"]["com.linkedin.digitalmedia.uploading.MediaUploadHttpRequest"]["uploadUrl"]
                 asset_urn = datar["asset"]
-                
+
                 # Upload image
                 upload_headers = {
                     "Authorization": f"Bearer {access_token}",
                     "X-Restli-Protocol-Version": "2.0.0",
                     "Content-Type": "application/octet-stream"
                 }
-                
-                # Download image and upload to LinkedIn
-                image_response = requests.get(image_url)
-                if image_response.status_code == 200:
-                    upload_response = requests.put(upload_url, headers=upload_headers, data=image_response.content)
-                    if upload_response.status_code not in (200, 201):
-                        raise Exception(f"Failed to upload image: {upload_response.status_code} {upload_response.text}")
-                
+
+                # Check if image_url is a file path or a URL
+                import os
+                if os.path.exists(image_url) and not image_url.startswith(('http://', 'https://', 'ftp://', 'file://')):
+                    # It's a local file path, read the file
+                    with open(image_url, 'rb') as f:
+                        image_content = f.read()
+                else:
+                    # It's a URL, download the image
+                    image_response = requests.get(image_url)
+                    if image_response.status_code != 200:
+                        raise Exception(f"Failed to download image from URL: {image_response.status_code} {image_response.text}")
+                    image_content = image_response.content
+
+                # Upload to LinkedIn
+                upload_response = requests.put(upload_url, headers=upload_headers, data=image_content)
+                if upload_response.status_code not in (200, 201):
+                    raise Exception(f"Failed to upload image: {upload_response.status_code} {upload_response.text}")
+
                 # Create post with image
                 post_body = {
                     "author": f"urn:li:person:{user_id}",
@@ -319,7 +330,7 @@ class LinkedInService:
                         "com.linkedin.ugc.MemberNetworkVisibility": "PUBLIC"
                     }
                 }
-            
+
             response = requests.post(url, headers=headers, json=post_body)
             response.raise_for_status()
             return response.json()

backend/utils/image_utils.py

@@ -1,39 +1,50 @@
 """Utility functions for image handling and processing."""
 
 import base64
+import re
 
 
 def ensure_bytes_format(image_data):
     """
     Ensure image data is in the proper format for storage.
-    
+    File paths and URLs should remain as strings, only actual image data should be bytes.
+
     Args:
-        image_data: Image data that could be bytes, base64 string, or URL string
-        
+        image_data: Image data that could be bytes, base64 string, file path, or URL string
+
     Returns:
-        Properly formatted data for database storage
+        Properly formatted data for database storage (bytes for actual image data, string for paths/URLs)
     """
     if image_data is None:
         return None
-        
-    # If it's already bytes, return as is
+
+    # If it's already bytes, return as is (for actual image data)
     if isinstance(image_data, bytes):
         return image_data
-        
-    # If it's a string, check if it's base64 encoded
+
+    # If it's a string, determine if it's a file path, URL, or base64 data
     if isinstance(image_data, str):
-        # Check if it's a data URL
+        # Check if it's a data URL (base64 encoded image)
         if image_data.startswith('data:image/'):
             try:
                 # Extract base64 part and decode to bytes
                 base64_part = image_data.split(',')[1]
                 return base64.b64decode(base64_part)
             except Exception:
-                # If decoding fails, store as string (URL)
+                # If decoding fails, return the original string
                 return image_data
+        # Check if it looks like a file path
+        # This regex matches common file path patterns like /tmp/..., ./path/, ../path/, C:\path\, etc.
+        elif re.match(r'^[/\\]|[.][/\\]|[\w]:[/\\]|.*\.(jpg|jpeg|png|gif|webp|bmp|tiff|svg)$', image_data.lower()):
+            # This appears to be a file path, return as string - don't convert to bytes
+            return image_data
+        # Check if it looks like a URL (has a scheme like http:// or https://)
+        elif image_data.startswith(('http://', 'https://', 'ftp://', 'file://')):
+            # This is a URL, return as string
+            return image_data
         else:
-            # Assume it's a URL, store as string
+            # Assume it's a string that should remain as is
             return image_data
-            
+
     # For any other type, return as is
     return image_data

docu_code/My_data_base_schema_.txt

@@ -5,7 +5,7 @@ CREATE TABLE public.Post_content (
   id bigint GENERATED ALWAYS AS IDENTITY NOT NULL,
   id_social bigint,
   Text_content text,
-  image_content_url bytea,
+  image_content_url character varying,
   Video_content text,
   post_time time without time zone,
   created_at timestamp with time zone NOT NULL DEFAULT now(),
@@ -36,6 +36,7 @@ CREATE TABLE public.Social_network (
   family_name character varying NOT NULL,
   account_name text NOT NULL UNIQUE,
   expiration date,
+  refresh_token character varying,
   CONSTRAINT Social_network_pkey PRIMARY KEY (id),
   CONSTRAINT Social_network_id_utilisateur_fkey FOREIGN KEY (id_utilisateur) REFERENCES auth.users(id)
 );