1from apify import Actor, ApifyClient
2import asyncio
3import httpx
4import json
5import re
6from urllib.parse import quote
7
8async def main():
9 async with Actor() as actor:
10 input_data = await actor.get_input() or {}
11 usernames = input_data.get("usernames", [])
12 max_posts = input_data.get("max_posts", 10)
13
14 if not usernames:
15 Actor.log.info("No usernames provided. Exiting.")
16 return
17
18 results = []
19 headers = {
20 "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
21 "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
22 "Accept-Language": "en-US,en;q=0.5",
23 }
24
25 async with httpx.AsyncClient(timeout=30, follow_redirects=True) as client:
26 for username in usernames:
27 Actor.log.info(f"Scraping profile: {username}")
28 try:
29
30 resp = await client.get(
31 f"https://www.instagram.com/{username}/",
32 headers=headers
33 )
34
35
36 data = {}
37
38
39 scripts = re.findall(r'<script type="text/javascript">window\._sharedData\s*=\s*({.+?});</script>', resp.text)
40 if scripts:
41 json_data = json.loads(scripts[0])
42 user_data = json_data.get("entry_data", {}).get("ProfilePage", [{}])[0].get("graphql", {}).get("user", {})
43 data = {
44 "username": username,
45 "full_name": user_data.get("full_name", ""),
46 "biography": user_data.get("biography", ""),
47 "followers": user_data.get("edge_followed_by", {}).get("count", 0),
48 "following": user_data.get("edge_follow", {}).get("count", 0),
49 "posts_count": user_data.get("edge_owner_to_timeline_media", {}).get("count", 0),
50 "is_private": user_data.get("is_private", False),
51 "is_verified": user_data.get("is_verified", False),
52 "profile_pic_url": user_data.get("profile_pic_url_hd", ""),
53 "external_url": user_data.get("external_url", ""),
54 "business_category": user_data.get("business_category_name", ""),
55 "engagement_rate": 0,
56 }
57
58
59 if not data:
60
61 scripts_jsonl = re.findall(r'<script type="application/ld\+json">(.+?)</script>', resp.text)
62 for s in scripts_jsonl:
63 try:
64 j = json.loads(s)
65 if j.get("@type") == "ProfilePage":
66 data = {
67 "username": username,
68 "full_name": j.get("name", ""),
69 "description": j.get("description", ""),
70 "interactionStatistic": j.get("interactionStatistic", []),
71 }
72 except:
73 pass
74
75 if data:
76 results.append(data)
77 await actor.push_data(data)
78 Actor.log.info(f" Scraped: {data.get('full_name', username)} - {data.get('followers', 'N/A')} followers")
79 else:
80
81 basic = {
82 "username": username,
83 "page_loaded": len(resp.text) > 5000,
84 "status": "partial - may require login or rate limited",
85 }
86 results.append(basic)
87 await actor.push_data(basic)
88
89 except Exception as e:
90 Actor.log.error(f"Error scraping {username}: {e}")
91 await actor.push_data({
92 "username": username,
93 "error": str(e),
94 "status": "failed"
95 })
96
97 await asyncio.sleep(2)
98
99 Actor.log.info(f"Done! Scraped {len(results)} profiles.")
100
101
102if __name__ == "__main__":
103 asyncio.run(main())