爬虫实战：利用代理ip爬取推特网站数据

引言

亮数据-网络IP代理及全网数据一站式服务商屡获殊荣的代理网络、强大的数据挖掘工具和现成可用的数据集。亮数据：网络数据平台领航者https://www.bright.cn/?promo=RESIYEAR50/?utm_source=brand&utm_campaign=brnd-mkt_cn_csdn_yingjie202502

在跨境电商、社交媒体运营以及数据采集的过程中，我们经常会遇到被平台拦截的问题。这是因为各大平台为了保护其正常业务，防止恶意攻击或数据滥用，会限制非人类用户的访问频率。如果我们使用多账号登录或通过自动化工具频繁访问平台，系统很容易识别出异常行为，进而将我们的账号或IP地址列入黑名单，导致访问受限或直接被封禁。这种情况不仅影响了业务的正常开展，还可能导致数据采集中断，甚至丢失重要信息。面对这一问题，目前最有效的解决方案之一是使用动态IP。动态IP的核心优势在于，每次访问时都可以切换到一个新的IP地址，从而降低被平台识别和拦截的概率。与传统的商业数据中心IP不同，动态IP中有一类特殊的动态住宅IP，它们来自真实的家庭网络，能够更好地模拟普通用户的上网行为。

通过使用动态住宅IP服务，我们可以在进行社交媒体数据采集时，大幅减少被平台拦截的风险。无论是跨境电商的竞品分析，还是社交媒体运营的数据监控，动态住宅IP都能为我们提供更加稳定和高效的访问环境。接下来，我们将详细介绍如何配置和使用动态住宅IP服务，并完成社交平台的数据采集工作。

准备

首先我们需要配置动态住宅IP，我问了一下AI，推荐了几家常见的服务商，意义了解之后，我发现亮数据平台正好在促销，性价比是几家中最高的，这次我们就来试用一下。只需要注册账号即可开始使用。登录以后会跳转到工作台，在这里点击获取代理。

之后要进行简单的配置，这里我们只需要填写名称就可以了，其他选项大家可以根据需求选择。

动态住宅代理的计费模式是按照流量计算，使用期间可以随时切换IP地址。

配置好以后建议安装一下，这样可以提高安全性。当然不安装也不会太大的影响，后面需要的时候还可以再安装。

这样就配置好了。在这个页面可以找到我们的主机地址、用户名和密码，旁边有一个样例程序，可以修改它称为我们的爬虫程序。

配置好就可以开始配置社交平台API接口了。在平台中注册为开发者就可以拿到自己的token，在后面的访问需要使用它才能正常接入。

采集社交媒体数据

接下来就可以制作爬虫程序。首先我们需要将服务的参数设置好。

proxies = {'http': 'http://brd-customer-hl_a0a48734-zone-residential proxy1:[e-mail]:33335''https': 'https://brd-customer-hl_a0a48734-zone-residential proxy1:[e-mail]:33335'
}

我们本次的任务是抓取下图账号的所有帖子和每个帖子点赞数、转发数等指标。在开始之前，我们需要配置一下请求头参数，这个可以直接在网站中获得。在控制台中找到header和cookie的值复制出来就可以了。

找到之后把它们打包在一个类中方便后面使用。

class CsxqTwitterKeywordSearch:def __init__(self,saveFileName,cookie_str):self.saveFileName = saveFileNameself.searchCondition = Noneself.headers = {'headers对应的参数'}self.cookies = self.cookie_str_to_dict(cookie_str)

接下来我们需要配置一下请求参数，这些需要通过cursor去网站获取。

def get_params(self,cursor):if cursor == "":variables = {"rawQuery": self.searchCondition, "count": 20,"querySource": "typed_query", "product": "Latest"}params = {"variables": json.dumps(variables,separators=(",",":")),"features": ""}else:variables = {"rawQuery": self.searchCondition, "count": 20, "cursor": cursor, "querySource": "typed_query", "product": "Latest"}params = {"variables":  json.dumps(variables,separators=(",",":")),"features": ""}return params

之后使用获取到的参数访问并获取元数据，这里需要注意元数据是一个json表单。这里url部分需要将中间替换为自己的token 才能使用。

def get(self,cursor):self.headers["x-csrf-token"] = self.cookies['ct0']url = "https://x.com/i/api/graphql/6uoFezW1o4e-n-VI5vfksA/SearchTimeline"params = self.get_params(cursor)while True:try:response = requests.get(url,headers=self.headers,cookies=self.cookies,params=params,timeout=(3,10),proxies=proxies)if response.status_code == 429:time.sleep(60*20)if response.status_code == 200:data = response.json()return dataexcept Exception as e:print("搜索接口发生错误：%s" % e)

最后我们需要将目标数据从获取到的元数据中提取出来。这里我们需要提取的是内容、时间、点赞、评论、转发、用户名、简介、粉丝量和关注量，由于元数据是json表单所以只需要简单转换为字典就可以轻松获取。

def parse_data(self,entries):resultList = []def transTime(dd):GMT_FORMAT = '%a %b %d %H:%M:%S +0000 %Y'timeArray = datetime.datetime.strptime(dd, GMT_FORMAT)return timeArray.strftime("%Y-%m-%d %H:%M:%S")contentList = []for index, ent in enumerate(entries):try:entryId = ent.get('entryId', "")if 'tweet' in entryId:l_result = ent['content']['itemContent']['tweet_results']['result'] if ent['content'].get('itemContent') else Noneif l_result:contentList.append(l_result)elif "profile-conversation" in entryId:items = ent['content']['items']for i in items:l_result = i['item']['itemContent']['tweet_results']['result'] if i['item'].get('itemContent') else Noneif l_result:contentList.append(l_result)except:passfor l in contentList:try:result = l.get('tweet') if l.get('tweet') else llegacy = result['legacy']core = result['core']created_at = transTime(legacy.get('created_at'))full_text = legacy.get('full_text')note_tweet = result.get('note_tweet')favorite_count = legacy.get('favorite_count')  # 点赞reply_count = legacy.get('reply_count')  # 回复retweet_count = legacy.get('retweet_count', 0)quote_count = legacy.get('quote_count', 0)retweet_count = retweet_count + quote_countif note_tweet:try:full_text = note_tweet['note_tweet_results']['result']['text']except:passu_legacy = core['user_results']['result']['legacy']hash_uname = u_legacy.get('screen_name')description = u_legacy['description']friends_count = u_legacy['friends_count']followers_count = u_legacy.get('followers_count')item = {"内容":full_text,"时间": created_at,"点赞":favorite_count,"评论":reply_count,"转发":retweet_count,"用户名": hash_uname,"简介": description,\"粉丝量":followers_count,"关注量":friends_count}print("数据->",item)resultList.append(item)except:passself.save_data(resultList)

最后我们要将数据保存为一个本地csv文件。

def save_data(self, resultList):if resultList:df = pd.DataFrame(resultList)if not os.path.exists(f'./{self.saveFileName}.csv'):df.to_csv(f'./{self.saveFileName}.csv', index=False, mode='a', sep=",", encoding="utf_8_sig")else:df.to_csv(f'./{self.saveFileName}.csv', index=False, mode='a', sep=",", encoding="utf_8_sig",header=False)self.resultList = []print("保存成功")

整个流程通过一个入口函数控制，在运行的同时打印一些状态信息。

def run(self,word):cursor = ""page = 1while True:# if page > 2:#     breakprint("正在爬取的页数：%s，cursor：%s"%(page,cursor))resqJson = self.get(cursor)if not resqJson:breakcursor,entries = self.get_cursor(resqJson)if entries:self.parse_data(entries)page += 1else:break
def main(self,fromDate,endDate):wordList = ["climate change"]start = 0for index,word in enumerate(wordList[start:],start):self.searchCondition = f"{word} lang:en until:{endDate} since:{fromDate}"print("搜索条件：",self.searchCondition)self.run(word)

通过一个主函数执行整个程序，这里需要用户粘贴自己的cookie。

if __name__ == '__main__':cookie_str = '改成你自己的cookies'fromDate = "2024-08-10"endDate = "2024-10-13"saveFileName= "Tim_Cook"ctks = CsxqTwitterKeywordSearch(saveFileName,cookie_str)ctks.main(fromDate,endDate)

运行一下就可以获得结果，可以看到程序运行正常。

总结

上面的实战演示展示了配置动态住宅IP和社交平台API接口的方法，并演示了如何制作爬虫程序进行数据采集，包括设置参数、配置请求头和请求参数、获取元数据、提取目标数据以及将数据保存为本地CSV文件，通过粘贴自己的cookie即可运行程序。在这一过程中我们也看到了动态住宅IP的作用，在大量采集数据的时候，通过使用动态住宅IP服务，可以有效减少被平台识别和拦截的概率，不仅提高了数据采集的效率，还增强了隐匿性，从而更好地规避平台的限制。于此同时我们也测试了亮数据产品的可靠性，不论是在易用性和产品的丰富性上都有独到之处。不仅如此，亮数据还有一些爬虫工具可供使用，

粉丝福利