#!/usr/bin/env python3 """ Synchronisation des posts LinkedIn vers les actualités. - Lit les tokens depuis .env (LINKEDIN_*) - Récupère tous les posts de l'organisation - Si post avec 1 image : télécharge l'image et la sauvegarde - Si post vidéo ou carousel : pas d'image (le frontend utilisera generique.jpg) - Sortie JSON sur stdout pour que Laravel puisse importer """ import os import sys import json import urllib.parse import re from datetime import datetime try: import requests except ImportError: print('{"error": "pip install requests"}', file=sys.stderr) sys.exit(2) # Charger .env si présent (depuis le répertoire backend) SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__)) BACKEND_ROOT = os.path.dirname(SCRIPT_DIR) ENV_PATH = os.path.join(BACKEND_ROOT, ".env") if os.path.isfile(ENV_PATH): with open(ENV_PATH, "r", encoding="utf-8") as f: for line in f: line = line.strip() if line and not line.startswith("#") and "=" in line: key, _, value = line.partition("=") key, value = key.strip(), value.strip().strip('"').strip("'") if key.startswith("LINKEDIN_"): os.environ.setdefault(key, value) CLIENT_ID = os.environ.get("LINKEDIN_CLIENT_ID") CLIENT_SECRET = os.environ.get("LINKEDIN_CLIENT_SECRET") REFRESH_TOKEN = os.environ.get("LINKEDIN_REFRESH_TOKEN") ACCESS_TOKEN = os.environ.get("LINKEDIN_ACCESS_TOKEN") TOKEN_FILE = os.path.join(BACKEND_ROOT, "storage", "app", ".linkedin_tokens.json") MEDIA_ACTUALITES = os.path.join(BACKEND_ROOT, "public", "media", "actualites") def get_access_token(): """Obtient un access_token via refresh_token ou fichier.""" if REFRESH_TOKEN and CLIENT_ID and CLIENT_SECRET: redirect_uri = os.environ.get("LINKEDIN_REDIRECT_URI", "http://localhost:8080/callback") r = requests.post( "https://www.linkedin.com/oauth/v2/accessToken", data={ "grant_type": "refresh_token", "refresh_token": REFRESH_TOKEN, "client_id": CLIENT_ID, "client_secret": CLIENT_SECRET, "redirect_uri": redirect_uri, }, headers={"Content-Type": "application/x-www-form-urlencoded"}, ) if r.status_code == 200: data = r.json() token = data.get("access_token") if token: os.makedirs(os.path.dirname(TOKEN_FILE), exist_ok=True) with open(TOKEN_FILE, "w") as f: json.dump( { "access_token": token, "expires_in": data.get("expires_in", 5184000), "refresh_token": data.get("refresh_token", REFRESH_TOKEN), }, f, ) return token print(json.dumps({"error": "refresh_failed", "body": r.text}), file=sys.stderr) return None if os.path.isfile(TOKEN_FILE): try: with open(TOKEN_FILE) as f: data = json.load(f) return data.get("access_token") except Exception: pass if ACCESS_TOKEN: return ACCESS_TOKEN print( json.dumps( {"error": "no_token", "hint": "Set LINKEDIN_REFRESH_TOKEN or LINKEDIN_ACCESS_TOKEN in .env"} ), file=sys.stderr, ) return None def get_organization_urn(access_token): headers = { "Authorization": f"Bearer {access_token}", "X-Restli-Protocol-Version": "2.0.0", } r = requests.get( "https://api.linkedin.com/v2/organizationalEntityAcls?q=roleAssignee", headers=headers, ) if r.status_code != 200 or not r.json().get("elements"): return None return r.json()["elements"][0]["organizationalTarget"] def extract_single_image_url(post): """Retourne l'URL de l'image si le post a exactement une image (pas vidéo, pas carousel).""" try: specific = post.get("specificContent", {}) share = specific.get("com.linkedin.ugc.ShareContent", {}) media = share.get("media", []) if len(media) != 1: return None item = media[0] if item.get("status") == "READY": thumbnails = item.get("thumbnails", []) if thumbnails: return thumbnails[-1].get("url") return item.get("originalUrl") return None except Exception: return None def download_image(url, filepath): try: r = requests.get(url, stream=True, timeout=30) if r.status_code == 200: os.makedirs(os.path.dirname(filepath), exist_ok=True) with open(filepath, "wb") as f: for chunk in r.iter_content(8192): f.write(chunk) return True except Exception: pass return False def fetch_all_posts(access_token, org_urn): headers = { "Authorization": f"Bearer {access_token}", "X-Restli-Protocol-Version": "2.0.0", } encoded = urllib.parse.quote(org_urn) all_posts = [] start = 0 count = 50 while True: url = ( f"https://api.linkedin.com/v2/ugcPosts?q=authors&authors=List({encoded})" f"&sortBy=LAST_MODIFIED&start={start}&count={count}" ) r = requests.get(url, headers=headers) if r.status_code != 200: break data = r.json() elements = data.get("elements", []) if not elements: break all_posts.extend(elements) if len(elements) < count: break start += count return all_posts def parse_linkedin_date(value): """Parse une date ISO LinkedIn (ex: 2025-02-06T12:00:00.000Z).""" if not value: return datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S") try: if "T" in value: dt = datetime.fromisoformat(value.replace("Z", "+00:00")) else: dt = datetime.strptime(value[:10], "%Y-%m-%d") return dt.strftime("%Y-%m-%d %H:%M:%S") except Exception: return datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S") def main(): token = get_access_token() if not token: sys.exit(1) org_urn = get_organization_urn(token) if not org_urn: print(json.dumps({"error": "no_organization"}), file=sys.stderr) sys.exit(1) posts = fetch_all_posts(token, org_urn) results = [] for post in posts: try: share = post.get("specificContent", {}).get("com.linkedin.ugc.ShareContent", {}) text = (share.get("shareCommentary", {}).get("text") or "").strip() post_id = post.get("id", "") linkedin_url = f"https://www.linkedin.com/feed/update/{post_id}" if post_id else "" created = post.get("createdAt") or post.get("lastModified") or "" date_str = parse_linkedin_date(created) image_filename = None image_url = extract_single_image_url(post) if image_url: safe_id = re.sub(r"[^\w\-]", "_", post_id)[:80] filename = f"linkedin_{safe_id}.jpg" filepath = os.path.join(MEDIA_ACTUALITES, filename) if download_image(image_url, filepath): image_filename = f"actualites/{filename}" results.append({ "linkedin_post_id": post_id, "title": (text[:200] if text else "Sans titre").split("\n")[0], "content": text, "date": date_str, "image": image_filename, "linkedin_url": linkedin_url, }) except Exception as e: print(json.dumps({"error": "post_error", "id": post.get("id"), "message": str(e)}), file=sys.stderr) continue print(json.dumps(results, ensure_ascii=False)) if __name__ == "__main__": main()