说明一下至于为什么不使用抓包的方式直接获取推文,是因为推特在获取推文时和服务器之间的数据是加密的,无法成功抓包(抓到的全都是心跳包)。 推特有开发的api文档,但是需要企业什么各种证明来申请,还要审核非常麻烦(加上推特开发者api全是英文的更让我头疼)。 所以直接通过“模拟用户操作”(控制浏览器)来进行爬虫了!
程序做的比较草率,程序主要的流程有:
使用获取COOKIE登录Twitter、跳转到画师发布媒体的页面、寻找推文、获取推文(时间 内容 图片等)、点赞推文、使用 requests 模块下载图片。
获取推文图片主要思路为:
①推文会有 article 的标签名(TAG_NAME 就是会放在 <article></article> 内)涵盖,先寻找标签名,在获取其中的div(<div></div>)的内容(一般会有时间和推文内容)。
②在上述的article中寻找img标签名,这个就是图片了。图片的格式一般会以 https://xxx.xx/xxxxx?format=jpg&name=small 的形式出现,代码中将其处理成 https://xxx.xx/xxxxx?format=png 获取原图(后面发现还要再加个 &name=large 才是原图)。并传入下载函数下载。然后删除这个article(推文就会消失),自动加载下一个推文重复步骤。
③找到有“like”表示的div点赞(这一步在第②步之前就执行了,因为是后面加上去的,所以写在后面会协调一点)
已知问题:
①视频中有时下载时会出现ERROR,是因为保存文件名时不能包含换行符(文件名是我把画师名、时间、推文内容一同合成的),在前面忘记用替换文本的方法去掉了。
②程序没有停止的信号,会一直运行下去,需要手动停止。
代码总览:
# -*- coding:utf-8=-*- import json, time import requests from selenium.webdriver import Edge from selenium.webdriver.common.by import By from selenium.common.exceptions import TimeoutException from selenium.webdriver.chrome.service import Service from selenium import webdriver def download(src, pTime, context, who): """下载图片 图片地址 发布时间 推特原话 作者""" try: # 生成保存文件名 saveFileName = src[src.rfind("/") + 1:src.rfind("?")].replace('/', '').replace('\\', '').replace(':', ''). \ replace('*', '').replace('?', '').replace('"', '').replace('>', '').replace('<', '').replace('|', '') print(f"准备下载图片:{saveFileName}.png,图片网址:{src},作者:{who},推特原文:{context},发布时间:{pTime}") saveFileName = f"({who})({saveFileName})({pTime}){context}.png" response = requests.get(src, stream=True, proxies={"http": "http://127.0.0.1:10809", "https": "http://127.0.0.1:10809"}) content_size = int(response.headers['content-length']) # 内容体总大小 file_size = 0 with open(f"image/{saveFileName}", "wb") as f: for data in response.iter_content(1024 * 5): f.write(data) file_size += len(data) print( f"\r正在下载图片 {saveFileName} ,进度:{file_size}B / {content_size}B - {round(file_size / content_size * 100, 2)}%", end="") print(f"\n下载图片 {saveFileName} 完成!") response.close() except BaseException as error: print(str(error)) if __name__ == "__main__": # 初始化 service = Service("C:\\Program Files (x86)\\Microsoft\\Edge\\Application\\msedgedriver.exe") driver = Edge(service=service) driver.set_page_load_timeout(5) try: driver.get('https://twitter.com/i/flow/login') except TimeoutException: driver.execute_script('window.stop()') driver.set_page_load_timeout(60) # 设置cookie print("设置 cookie 中......") cookies = json.load(open("cookie.json", "r")) for cookie in cookies: driver.add_cookie(cookie) print("访问画师图片页面中......") driver.get("https://twitter.com/mafumuffin/media") print("准备获取图片中......") time.sleep(5) while True: print("获取图片中......") try: for article in driver.find_elements(By.TAG_NAME, "article"): pTime = article.find_element(By.TAG_NAME, "time").text info = [] for div in article.find_elements(By.CLASS_NAME, "css-1dbjc4n"): if div.get_attribute("data-testid") == "videoPlayer": driver.execute_script( 'tagElements = document.getElementsByTagName("article");for( var m = 0 ; m < tagElements.length ; m++ ){tagElements[m].parentElement.parentElement.remove();}') continue info.append(div.text) if div.get_attribute("data-testid") == "like": webdriver.ActionChains(driver).click(div).perform() for img in article.find_elements(By.TAG_NAME, "img"): src = img.get_attribute('src') if "profile_images" not in src: download(src[:src.find('?')] + "?format=png&size=large", pTime, info[-32], info[0][:info[0].find("\n")]) driver.execute_script( 'tagElements = document.getElementsByTagName("article");for( var m = 0 ; m < tagElements.length ; m++ ){tagElements[m].parentElement.parentElement.remove();}') print("获取了一个推文。") except BaseException as error: # print(str(error)) continue
需要手动创建image文件夹,cookie.json使用浏览器插件一键复制粘贴即可使用。