| """ |
| Instagram via instaloader. Public data needs no auth; pass an optional |
| session cookie in tool_keys (tool='instagram') for stories/highlights/saved. |
| """ |
| import instaloader |
| from supabase_client import get_tool_keys |
|
|
|
|
| def _get_loader() -> instaloader.Instaloader: |
| L = instaloader.Instaloader() |
| creds = get_tool_keys("instagram") |
| if creds and creds.get("api_key"): |
| try: |
| L.context._session.cookies.set("sessionid", creds["api_key"]) |
| except Exception: |
| pass |
| return L |
|
|
|
|
| def run(params: dict) -> dict: |
| mode = params.get("mode") |
| target = params.get("target") |
| if mode not in ("profile", "hashtag") or not target: |
| raise ValueError("mode must be 'profile' or 'hashtag', and target is required") |
| max_posts = int(params.get("max_posts", 20)) |
|
|
| L = _get_loader() |
|
|
| if mode == "profile": |
| profile = instaloader.Profile.from_username(L.context, target) |
| posts = [] |
| for i, post in enumerate(profile.get_posts()): |
| if i >= max_posts: |
| break |
| posts.append( |
| { |
| "shortcode": post.shortcode, |
| "caption": (post.caption or "")[:200], |
| "likes": post.likes, |
| "comments": post.comments, |
| "date": str(post.date_utc), |
| "url": f"https://www.instagram.com/p/{post.shortcode}/", |
| } |
| ) |
| return { |
| "profile": target, |
| "followers": profile.followers, |
| "bio": profile.biography, |
| "post_count": profile.mediacount, |
| "posts": posts, |
| } |
|
|
| |
| hashtag = instaloader.Hashtag.from_name(L.context, target) |
| posts = [] |
| for i, post in enumerate(hashtag.get_posts()): |
| if i >= max_posts: |
| break |
| posts.append( |
| { |
| "shortcode": post.shortcode, |
| "caption": (post.caption or "")[:200], |
| "likes": post.likes, |
| "comments": post.comments, |
| "date": str(post.date_utc), |
| "url": f"https://www.instagram.com/p/{post.shortcode}/", |
| } |
| ) |
| return {"hashtag": target, "count": len(posts), "results": posts} |
|
|
