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

之后使用获取到的参数访问并获取元数据，这里需要注意元数据是一个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)

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)

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)

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)