# ============================================================================== # Author: Simon-Pierre Boucher # File: src/main.py (ka-instagram) # Desc: Profils Instagram publics ULTRA DÉTAILLÉS via l'endpoint web officiel # web_profile_info (JSON du web public) : identité complète, compteurs, # 12 derniers posts (likes/commentaires/vues/légende/musique/lieu/ # co-auteurs/épinglés), agrégats d'engagement et de cadence, profils # reliés — + recherche d'utilisateurs (topsearch) en découverte. # ============================================================================== from __future__ import annotations import asyncio import json import urllib.parse from datetime import datetime, timezone from apify import Actor from .net import Fetcher PROFILE_API = ("https://i.instagram.com/api/v1/users/web_profile_info/" "?username={u}") SEARCH_API = ("https://www.instagram.com/web/search/topsearch/" "?context=blended&query={q}&count=30") SEARCH_API_ALT = ("https://i.instagram.com/api/v1/web/search/topsearch/" "?context=blended&query={q}&count=30") HDRS = {"x-ig-app-id": "936619743392459", "Accept": "application/json"} def _iso(ts) -> str | None: try: return datetime.fromtimestamp(int(ts), tz=timezone.utc) \ .strftime("%Y-%m-%dT%H:%M:%SZ") except (TypeError, ValueError, OSError): return None def _avg(vals: list) -> int | None: vals = [v for v in vals if isinstance(v, (int, float))] return round(sum(vals) / len(vals)) if vals else None def parse_post(node: dict) -> dict: """Nœud média du fil → post détaillé (métriques, musique, lieu, épinglé).""" cap = (node.get("edge_media_to_caption") or {}).get("edges") or [] music = node.get("clips_music_attribution_info") or {} dims = node.get("dimensions") or {} return { "id": node.get("id"), "shortcode": node.get("shortcode"), "url": f"https://www.instagram.com/p/{node.get('shortcode')}/", "type": ("reel" if node.get("product_type") == "clips" else "video" if node.get("is_video") else "carousel" if node.get("edge_sidecar_to_children") else "image"), "caption": ((cap[0].get("node") or {}).get("text", "") if cap else "")[:500], "likes": (node.get("edge_liked_by") or {}).get("count"), "comments": (node.get("edge_media_to_comment") or {}).get("count"), "video_views": node.get("video_view_count"), "video_duration_s": node.get("video_duration"), "timestamp": node.get("taken_at_timestamp"), "posted_at": _iso(node.get("taken_at_timestamp")), "display_url": node.get("display_url"), "thumbnail_url": node.get("thumbnail_src"), "width": dims.get("width"), "height": dims.get("height"), "is_pinned": bool(node.get("pinned_for_users")), "comments_disabled": node.get("comments_disabled"), "location": (node.get("location") or {}).get("name"), "coauthors": [c.get("username") for c in (node.get("coauthor_producers") or []) if c.get("username")], "music": ({"artist": music.get("artist_name"), "song": music.get("song_name"), "uses_original_audio": music.get("uses_original_audio")} if music.get("song_name") or music.get("artist_name") else None), "accessibility_caption": (node.get("accessibility_caption") or "")[:200] or None, "tagged_users": [((t.get("node") or {}).get("user") or {}) .get("username") for t in ((node.get("edge_media_to_tagged_user") or {}).get("edges") or [])][:10], } def parse_user(user: dict) -> dict: """Objet user web_profile_info → enregistrement riche standardisé.""" posts = [parse_post((e.get("node") or {})) for e in ((user.get("edge_owner_to_timeline_media") or {}) .get("edges") or [])] bio_ent = user.get("biography_with_entities") or {} bio_mentions, bio_hashtags = [], [] for ent in (bio_ent.get("entities") or []): eu = ((ent or {}).get("user") or {}).get("username") eh = ((ent or {}).get("hashtag") or {}).get("name") if eu: bio_mentions.append(eu) if eh: bio_hashtags.append(eh) followers = (user.get("edge_followed_by") or {}).get("count") avg_likes = _avg([p["likes"] for p in posts]) avg_comments = _avg([p["comments"] for p in posts]) videos = [p for p in posts if p["type"] in ("video", "reel")] avg_video_views = _avg([p["video_views"] for p in videos]) engagement = None if followers and avg_likes is not None: engagement = round( (avg_likes + (avg_comments or 0)) / followers * 100, 3) timestamps = sorted(p["timestamp"] for p in posts if isinstance(p["timestamp"], int)) posts_per_week = None if len(timestamps) >= 3 and timestamps[-1] > timestamps[0]: span_weeks = (timestamps[-1] - timestamps[0]) / 604_800 posts_per_week = round((len(timestamps) - 1) / span_weeks, 2) top_post = max((p for p in posts if isinstance(p["likes"], int)), key=lambda p: p["likes"], default=None) related = [] for edge in ((user.get("edge_related_profiles") or {}).get("edges") or []): n = edge.get("node") or {} if n.get("username"): related.append({"username": n["username"], "full_name": n.get("full_name"), "is_verified": n.get("is_verified"), "is_private": n.get("is_private")}) return { "kind": "profile", "platform": "instagram", "found": True, "id": user.get("id"), "username": user.get("username"), "full_name": user.get("full_name"), "biography": user.get("biography"), "bio_mentions": bio_mentions[:10], # @comptes cités dans la bio "bio_hashtags": bio_hashtags[:10], "pronouns": user.get("pronouns") or None, "external_url": user.get("external_url"), "bio_links": [b.get("url") for b in (user.get("bio_links") or []) if b.get("url")], "followers": followers, "following": (user.get("edge_follow") or {}).get("count"), "posts_count": (user.get("edge_owner_to_timeline_media") or {}).get("count"), "igtv_videos_count": (user.get("edge_felix_video_timeline") or {}).get("count"), "mutual_followers": (user.get("edge_mutual_followed_by") or {}).get("count"), "highlight_reels": user.get("highlight_reel_count"), "is_verified": user.get("is_verified"), "is_private": user.get("is_private"), "is_business": user.get("is_business_account"), "is_professional": user.get("is_professional_account"), "category": user.get("category_name"), "business_category": user.get("business_category_name"), "business_address": user.get("business_address_json"), # courriel PRO affiché publiquement par le créateur (§15 : public only) "business_email": user.get("business_email") or None, # présent = compte Threads relié (cross-link fort, même handle) "has_threads": (bool(user.get("has_onboarded_to_text_post_app")) if "has_onboarded_to_text_post_app" in user else None), "has_clips": user.get("has_clips"), "has_channel": user.get("has_channel"), "has_ar_effects": user.get("has_ar_effects"), "recently_joined": user.get("is_joined_recently"), "avatar": user.get("profile_pic_url_hd") or user.get("profile_pic_url"), # agrégats calculés sur les 12 derniers posts "avg_likes": avg_likes, "avg_comments": avg_comments, "avg_video_views": avg_video_views, "engagement_rate_pct": engagement, "posts_per_week": posts_per_week, "last_post_at": _iso(timestamps[-1]) if timestamps else None, "video_share_pct": (round(len(videos) / len(posts) * 100) if posts else None), "top_post": ({"url": top_post["url"], "likes": top_post["likes"], "comments": top_post["comments"], "type": top_post["type"]} if top_post else None), "recent_posts": posts, "related_profiles": related, } async def main() -> None: async with Actor: inp = await Actor.get_input() or {} usernames = [u.strip().lstrip("@").lower() for u in (inp.get("usernames") or []) if u and u.strip()] queries = [q.strip() for q in (inp.get("queries") or []) if q and q.strip()] proxy = await Actor.create_proxy_configuration( actor_proxy_input=inp.get("proxyConfiguration")) fetcher = Fetcher(proxy, delay=float(inp.get("requestDelay") or 1)) sem = asyncio.Semaphore(int(inp.get("concurrency") or 3)) async def one_profile(u: str) -> None: async with sem: miss = {"kind": "profile", "platform": "instagram", "username": u, "found": False} try: resp = await fetcher.get(PROFILE_API.format(u=u), headers=HDRS, session_id=u) except Exception as exc: await Actor.push_data({**miss, "error": str(exc)[:200]}) return if resp.status_code == 404: await Actor.push_data({**miss, "error": "http_404"}) return user = None if resp.status_code == 200: try: user = ((json.loads(resp.text).get("data") or {}) .get("user")) except Exception: user = None if not user: await Actor.push_data( {**miss, "error": f"http_{resp.status_code}"}) return await Actor.push_data(parse_user(user)) async def one_search(q: str) -> None: async with sem: data = {} for endpoint in (SEARCH_API, SEARCH_API_ALT): u2 = endpoint.format(q=urllib.parse.quote(q)) try: resp = await fetcher.get(u2, headers=HDRS, session_id=f"srch{q[:4]}") except Exception as exc: Actor.log.warning(f"recherche '{q}' : {exc}") continue if resp.status_code == 200: try: data = json.loads(resp.text) except Exception: data = {} if data.get("users"): break if not data.get("users"): Actor.log.warning( f"recherche '{q}' : 0 user " f"(http {resp.status_code}) {resp.text[:200]!r}") for item in data.get("users") or []: u = item.get("user") or {} if not u.get("username") or u.get("is_private"): continue await Actor.push_data({ "kind": "search_user", "platform": "instagram", "query": q, "username": u.get("username"), "full_name": u.get("full_name"), "is_verified": u.get("is_verified"), "followers": u.get("follower_count"), "avatar": u.get("profile_pic_url"), }) await asyncio.gather(*[one_profile(u) for u in usernames], *[one_search(q) for q in queries]) Actor.log.info(f"terminé : {len(usernames)} profils, " f"{len(queries)} recherches")