1from apify import Actor
2import asyncio
3import httpx
4import json
5import re
6
7async def main():
8 async with Actor() as actor:
9 input_data = await actor.get_input() or {}
10 usernames = input_data.get("usernames", [])
11 keywords = input_data.get("keywords", [])
12 mode = input_data.get("mode", "profiles")
13 max_results = input_data.get("max_results", 50)
14
15 results = []
16 headers = {
17 "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",
18 "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
19 "Accept-Language": "en-US,en;q=0.5",
20 }
21
22 async with httpx.AsyncClient(timeout=30, follow_redirects=True) as client:
23 if mode == "profiles" and usernames:
24 for username in usernames:
25 Actor.log.info(f"Scraping X/Twitter profile: @{username}")
26 try:
27 resp = await client.get(
28 f"https://x.com/{username}",
29 headers=headers
30 )
31
32 data = {"username": username, "url": f"https://x.com/{username}"}
33
34
35
36 scripts = re.findall(r'<script[^>]*>(.*?)</script>', resp.text)
37
38 for script in scripts:
39 if "profile_user" in script.lower() or "followers" in script.lower():
40
41 json_matches = re.findall(r'({[\s\S]*?"followers"[\s\S]*?})', script)
42 for jm in json_matches[:3]:
43 try:
44 j = json.loads(jm)
45 if "followers" in j:
46 data.update({
47 "full_name": j.get("full_name", j.get("name", "")),
48 "followers": j.get("followers", 0),
49 "following": j.get("following", 0),
50 "tweets": j.get("tweets", j.get("statuses_count", 0)),
51 "likes": j.get("likes", j.get("favourites_count", 0)),
52 "bio": j.get("description", j.get("bio", "")),
53 "location": j.get("location", ""),
54 "website": j.get("website", j.get("url", "")),
55 "joined": j.get("joined", j.get("created_at", "")),
56 "is_verified": j.get("verified", False),
57 })
58 except:
59 pass
60
61 results.append(data)
62 await actor.push_data(data)
63 Actor.log.info(f" Scraped: @{username}")
64
65 except Exception as e:
66 Actor.log.error(f"Error scraping @{username}: {e}")
67 await actor.push_data({"username": username, "error": str(e)})
68
69 await asyncio.sleep(2)
70
71 elif mode == "tweets" and keywords:
72 for keyword in keywords:
73 Actor.log.info(f"Searching tweets: {keyword}")
74 try:
75 search_url = f"https://x.com/search?q={keyword}&src=typed_query&f=top"
76 resp = await client.get(search_url, headers=headers)
77
78 data = {
79 "keyword": keyword,
80 "url": search_url,
81 "tweets_found": [],
82 "page_loaded": len(resp.text) > 5000,
83 }
84
85
86 tweet_pattern = re.findall(
87 r'data-testid="tweetText">([^<]+)',
88 resp.text
89 )
90 for tweet_text in tweet_pattern[:max_results]:
91 data["tweets_found"].append({"text": tweet_text.strip()})
92
93 results.append(data)
94 await actor.push_data(data)
95 Actor.log.info(f" Found {len(data.get('tweets_found', []))} tweets")
96
97 except Exception as e:
98 Actor.log.error(f"Error searching '{keyword}': {e}")
99 await actor.push_data({"keyword": keyword, "error": str(e)})
100
101 await asyncio.sleep(2)
102
103 Actor.log.info(f"Done! Processed {len(results)} items.")
104
105
106if __name__ == "__main__":
107 asyncio.run(main())