Python + selenium 获取 Twitter 上画师发布的图片

说明一下至于为什么不使用抓包的方式直接获取推文,是因为推特在获取推文时和服务器之间的数据是加密的,无法成功抓包(抓到的全都是心跳包)。 推特有开发的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使用浏览器插件一键复制粘贴即可使用。

发表评论 (0)

后再参与讨论