爬虫实战：利用代理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，在后面的访问需要使用它才能正常接入。

采集社交媒体数据

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

  1. proxies = {
  2.   'http': 'http://brd-customer-hl_a0a48734-zone-residential proxy1:[email protected]:33335'
  3.   'https': 'https://brd-customer-hl_a0a48734-zone-residential proxy1:[email protected]:33335'
  4. }

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

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

  1. class CsxqTwitterKeywordSearch:
  2.     def __init__(self,saveFileName,cookie_str):
  3.         self.saveFileName = saveFileName
  4.         self.searchCondition = None
  5.         self.headers = {
  6.         'headers对应的参数'
  7.         }
  8.         self.cookies = self.cookie_str_to_dict(cookie_str)

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

  1. def get_params(self,cursor):
  2.     if cursor == "":
  3.         variables = {"rawQuery": self.searchCondition, "count": 20,"querySource": "typed_query", "product": "Latest"}
  4.         params = {
  5.             "variables": json.dumps(variables,separators=(",",":")),
  6.             "features": ""
  7.         }
  8.     else:
  9.         variables = {"rawQuery": self.searchCondition, "count": 20, "cursor": cursor, "querySource": "typed_query", "product": "Latest"}
  10.         params = {
  11.             "variables":  json.dumps(variables,separators=(",",":")),
  12.             "features": ""
  13.         }
  14.     return params

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

  1. def get(self,cursor):
  2.     self.headers["x-csrf-token"] = self.cookies['ct0']
  3.     url = "https://x.com/i/api/graphql/6uoFezW1o4e-n-VI5vfksA/SearchTimeline"
  4.     params = self.get_params(cursor)
  5.     while True:
  6.         try:
  7.             response = requests.get(url,
  8.                                     headers=self.headers,
  9.                                     cookies=self.cookies,
  10.                                     params=params,
  11.                                     timeout=(3,10),
  12. 			proxies=proxies)
  13.             if response.status_code == 429:
  14.                 time.sleep(60*20)
  15.             if response.status_code == 200:
  16.                 data = response.json()
  17.                 return data
  18.         except Exception as e:
  19.             print("搜索接口发生错误：%s" % e)

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

  1. def parse_data(self,entries):
  2.     resultList = []
  3.     def transTime(dd):
  4.         GMT_FORMAT = '%a %b %d %H:%M:%S +0000 %Y'
  5.         timeArray = datetime.datetime.strptime(dd, GMT_FORMAT)
  6.         return timeArray.strftime("%Y-%m-%d %H:%M:%S")
  7.  
  8.     contentList = []
  9.     for index, ent in enumerate(entries):
  10.         try:
  11.             entryId = ent.get('entryId', "")
  12.  
  13.             if 'tweet' in entryId:
  14.                 l_result = ent['content']['itemContent']['tweet_results']['result'] if ent['content'].get(
  15.                     'itemContent') else None
  16.                 if l_result:
  17.                     contentList.append(l_result)
  18.             elif "profile-conversation" in entryId:
  19.                 items = ent['content']['items']
  20.                 for i in items:
  21.                     l_result = i['item']['itemContent']['tweet_results']['result'] if i['item'].get(
  22.                         'itemContent') else None
  23.                     if l_result:
  24.                         contentList.append(l_result)
  25.         except:
  26.             pass
  27.     for l in contentList:
  28.         try:
  29.             result = l.get('tweet') if l.get('tweet') else l
  30.             legacy = result['legacy']
  31.             core = result['core']
  32.             created_at = transTime(legacy.get('created_at'))
  33.             full_text = legacy.get('full_text')
  34.             note_tweet = result.get('note_tweet')
  35.             favorite_count = legacy.get('favorite_count')  # 点赞
  36.             reply_count = legacy.get('reply_count')  # 回复
  37.             retweet_count = legacy.get('retweet_count', 0)
  38.             quote_count = legacy.get('quote_count', 0)
  39.             retweet_count = retweet_count + quote_count
  40.             if note_tweet:
  41.                 try:
  42.                     full_text = note_tweet['note_tweet_results']['result']['text']
  43.                 except:
  44.                     pass
  45.  
  46.             u_legacy = core['user_results']['result']['legacy']
  47.             hash_uname = u_legacy.get('screen_name')
  48.             description = u_legacy['description']
  49.             friends_count = u_legacy['friends_count']
  50.             followers_count = u_legacy.get('followers_count')
  51.  
  52.             item = {"内容":full_text,"时间": created_at,"点赞":favorite_count,"评论":reply_count,"转发":retweet_count,"用户名": hash_uname,"简介": description,\
  53.                     "粉丝量":followers_count,"关注量":friends_count}
  54.             print("数据->",item)
  55.             resultList.append(item)
  56.         except:
  57.             pass
  58.     self.save_data(resultList)

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

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

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

  1. def run(self,word):
  2.     cursor = ""
  3.     page = 1
  4.     while True:
  5.         # if page > 2:
  6.         #     break
  7.         print("正在爬取的页数：%s，cursor：%s"%(page,cursor))
  8.         resqJson = self.get(cursor)
  9.         if not resqJson:
  10.             break
  11.         cursor,entries = self.get_cursor(resqJson)
  12.         if entries:
  13.             self.parse_data(entries)
  14.             page += 1
  15.         else:
  16.             break
  17. def main(self,fromDate,endDate):
  18.  
  19.     wordList = ["climate change"]
  20.  
  21.     start = 0
  22.     for index,word in enumerate(wordList[start:],start):
  23.         self.searchCondition = f"{word} lang:en until:{endDate} since:{fromDate}"
  24.         print("搜索条件：",self.searchCondition)
  25.         self.run(word)
  26.  

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

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

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

总结

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

粉丝福利

英杰代码编程
微信公众号
C/C++、python领域知识分享
注：本文转载自blog.csdn.net的Yan-英杰的文章"https://blog.csdn.net/m0_73367097/article/details/145639379"。版权归原作者所有，此博客不拥有其著作权，亦不承担相应法律责任。如有侵权，请联系我们删除。
复制链接
相关推荐
发表评论

评论记录：

未查询到任何数据！