Coding-With-Raptor/Likendin-Agent

1import requests2import os3from dotenv import load_dotenv4from agents import function_tool5 6# Load environment variables7load_dotenv()8 9 10# # Constants11LINKEDIN_ACCESS_TOKEN = os.getenv("LINKEDIN_ACCESS_TOKEN")12 13# Headers for LinkedIn API requests14HEADERS = {15    "Authorization": f"Bearer {LINKEDIN_ACCESS_TOKEN}",16    "X-Restli-Protocol-Version": "2.0.0",17    "Content-Type": "application/json",18}19 20 21def get_user_info():22    """Retrieve the authenticated user's LinkedIn information."""23    url = "https://api.linkedin.com/v2/userinfo"24    response = requests.get(url, headers=HEADERS)25    if response.status_code == 200:26        return response.json()27    else:28        print(f"Error fetching user info: {response.status_code}")29        return None30 31 32def register_image_upload():33    """Register an image upload with LinkedIn."""34    user_info = get_user_info()35    if not user_info:36        print("Failed to retrieve user information.")37        return None, None38    url = "https://api.linkedin.com/v2/assets?action=registerUpload"39    payload = {40        "registerUploadRequest": {41            "recipes": ["urn:li:digitalmediaRecipe:feedshare-image"],42            "owner": f"urn:li:person:{user_info['sub']}",43            "serviceRelationships": [44                {45                    "relationshipType": "OWNER",46                    "identifier": "urn:li:userGeneratedContent",47                }48            ],49        }50    }51    response = requests.post(url, headers=HEADERS, json=payload)52    if response.status_code == 200:53        return response.json().get("value", {}).get("uploadMechanism", {}).get(54            "com.linkedin.digitalmedia.uploading.MediaUploadHttpRequest", {}55        ).get("uploadUrl"), response.json().get("value", {}).get("asset")56    else:57        print(f"Error registering image upload: {response.status_code}")58        return None, None59 60 61def upload_image(upload_url, image_source):62    """63    Upload the image to LinkedIn.64 65    Args:66        upload_url (str): The LinkedIn upload URL67        image_source (str): Either a local file path or a URL to the image68    """69    if image_source.startswith(("http://", "https://")):70        # Handle URL71        response = requests.get(image_source)72        if response.status_code != 200:73            print(f"Error fetching image from URL: {response.status_code}")74            return False75        image_data = response.content76    else:77        # Handle local file78        try:79            with open(image_source, "rb") as image_file:80                image_data = image_file.read()81        except FileNotFoundError:82            print(f"Error: Image file not found at {image_source}")83            return False84 85    response = requests.post(86        upload_url,87        headers={"Authorization": f"Bearer {LINKEDIN_ACCESS_TOKEN}"},88        data=image_data,89    )90 91    if response.status_code == 201:92        print("Image uploaded successfully.")93        return True94    else:95        print(f"Error uploading image: {response.status_code}")96        return False97 98@function_tool99def create_post_links_image(image_path: str, post_text: str):100    """Create a post on LinkedIn with the uploaded image."""101    # Step 1: Register image upload102    upload_url, image_urn = register_image_upload()103    if not (upload_url and image_urn):104        print("Failed to register image upload.")105        return106 107    # Step 2: Upload the image108    if not upload_image(upload_url, image_path):109        print("Failed to upload image.")110        return111 112    # Step 3: Create the post113    user_info = get_user_info()114    if not user_info:115        print("Failed to retrieve user information.")116        return117 118    url = "https://api.linkedin.com/v2/ugcPosts"119    payload = {120        "author": f"urn:li:person:{user_info['sub']}",121        "lifecycleState": "PUBLISHED",122        "specificContent": {123            "com.linkedin.ugc.ShareContent": {124                "shareCommentary": {"text": post_text},125                "shareMediaCategory": "IMAGE",126                "media": [127                    {128                        "status": "READY",129                        "media": image_urn,130                    }131                ],132            }133        },134        "visibility": {"com.linkedin.ugc.MemberNetworkVisibility": "PUBLIC"},135    }136    response = requests.post(url, headers=HEADERS, json=payload)137    if response.status_code == 201:138        print("Post created successfully.")139    else:140        print(f"Error creating post: {response.status_code}")