知乎高顔值圖檔抓取到本地(Python3 爬蟲.人臉檢測.顔值檢測)

1.代碼在vscode和centos下均可成功執行

2.安裝好python3和pip3

3.安裝好依賴庫(pip3 install requests lxml baidu-aip requests)

4.在百度雲注冊登入賬号.開通人臉檢查服務(

https://cloud.baidu.com/product/face).

必須在代碼中填寫appid和ak資訊

5.image目錄必須和代碼檔案在同一個目錄下

#!/usr/bin/python3
#coding: utf-8

import time
import os
import re

import requests
# shell pip install requests lxml baidu-aip
from lxml import etree

from aip import AipFace

#百度雲 人臉檢測 申請資訊




#唯一必須填的資訊就這三行
APP_ID = ""
API_KEY = ""
SECRET_KEY = ""




# 檔案存放目錄名，相對于目前目錄
DIR = "image"
# 過濾顔值門檻值，存儲空間大的請随意
BEAUTY_THRESHOLD = 45

#浏覽器中打開知乎，在開發者工具複制一個，無需登入
#如何替換該值下文有講述
AUTHORIZATION = "oauth c3cef7c66a1843f8b3a9e6a1e3160e20"

#以下皆無需改動

#每次請求知乎的讨論清單長度，不建議設定太長，注意節操
LIMIT = 5

#這是話題『美女』的 ID，其是『顔值』（20013528）的父話題
SOURCE = "19552207"

#爬蟲假裝下正常浏覽器請求
USER_AGENT = "Mozilla/5.0 (Windows NT 5.1) AppleWebKit/534.55.3 (KHTML, like Gecko) Version/5.1.5 Safari/534.55.3"
#爬蟲假裝下正常浏覽器請求
REFERER = "https://www.zhihu.com/topic/%s/newest" % SOURCE
#某話題下讨論清單請求 url
BASE_URL = "https://www.zhihu.com/api/v4/topics/%s/feeds/timeline_activity"
#初始請求 url 附帶的請求參數
URL_QUERY = "?include=data%5B%3F%28target.type%3Dtopic_sticky_module%29%5D.target.data%5B%3F%28target.type%3Danswer%29%5D.target.content%2Crelationship.is_authorized%2Cis_author%2Cvoting%2Cis_thanked%2Cis_nothelp%3Bdata%5B%3F%28target.type%3Dtopic_sticky_module%29%5D.target.data%5B%3F%28target.type%3Danswer%29%5D.target.is_normal%2Ccomment_count%2Cvoteup_count%2Ccontent%2Crelevant_info%2Cexcerpt.author.badge%5B%3F%28type%3Dbest_answerer%29%5D.topics%3Bdata%5B%3F%28target.type%3Dtopic_sticky_module%29%5D.target.data%5B%3F%28target.type%3Darticle%29%5D.target.content%2Cvoteup_count%2Ccomment_count%2Cvoting%2Cauthor.badge%5B%3F%28type%3Dbest_answerer%29%5D.topics%3Bdata%5B%3F%28target.type%3Dtopic_sticky_module%29%5D.target.data%5B%3F%28target.type%3Dpeople%29%5D.target.answer_count%2Carticles_count%2Cgender%2Cfollower_count%2Cis_followed%2Cis_following%2Cbadge%5B%3F%28type%3Dbest_answerer%29%5D.topics%3Bdata%5B%3F%28target.type%3Danswer%29%5D.target.content%2Crelationship.is_authorized%2Cis_author%2Cvoting%2Cis_thanked%2Cis_nothelp%3Bdata%5B%3F%28target.type%3Danswer%29%5D.target.author.badge%5B%3F%28type%3Dbest_answerer%29%5D.topics%3Bdata%5B%3F%28target.type%3Darticle%29%5D.target.content%2Cauthor.badge%5B%3F%28type%3Dbest_answerer%29%5D.topics%3Bdata%5B%3F%28target.type%3Dquestion%29%5D.target.comment_count&limit=" + str(LIMIT)

#指定 url，擷取對應原始内容 / 圖檔
def fetch_image(url):
    try:
        headers = {
                "User-Agent": USER_AGENT,
                "Referer": REFERER,
                "authorization": AUTHORIZATION
                }
        s = requests.get(url, headers=headers)
    except Exception as e:
        print("fetch last activities fail. " + url)
        raise e

    return s.content

#指定 url，擷取對應 JSON 傳回 / 話題清單
def fetch_activities(url):
    try:
        headers = {
                "User-Agent": USER_AGENT,
                "Referer": REFERER,
                "authorization": AUTHORIZATION
                }
        s = requests.get(url, headers=headers)
    except Exception as e:
        print("fetch last activities fail. " + url)
        raise e

    return s.json()

#處理傳回的話題清單
def process_activities(datums, face_detective):
    for data in datums["data"]:

        target = data["target"]
        if "content" not in target or "question" not in target or "author" not in target:
            continue

        #解析清單中每一個元素的内容
        html = etree.HTML(target["content"])

        seq = 0

        #question_url = target["question"]["url"]
        question_title = target["question"]["title"]

        author_name = target["author"]["name"]
        #author_id = target["author"]["url_token"]

        print("current answer: " + question_title + " author: " + author_name)

        #擷取所有圖檔位址
        images = html.xpath("//img/@src")
        for image in images:
            if not image.startswith("http"):
                continue
            s = fetch_image(image)
            
            #請求人臉檢測服務
            scores = face_detective(s)

            for score in scores:
                filename = ("%d--" % score) + author_name + "--" + question_title + ("--%d" % seq) + ".jpg"
                filename = re.sub(r'(?u)[^-\w.]', '_', filename)
                #注意檔案名的處理，不同平台的非法字元不一樣，這裡隻做了簡單處理，特别是 author_name / question_title 中的内容
                seq = seq + 1
                with open(os.path.join(DIR, filename), "wb") as fd:
                    fd.write(s)

            #人臉檢測 免費，但有 QPS 限制
            time.sleep(2)

    if not datums["paging"]["is_end"]:
        #擷取後續讨論清單的請求 url
        return datums["paging"]["next"]
    else:
        return None

def get_valid_filename(s):
    s = str(s).strip().replace(' ', '_')
    return re.sub(r'(?u)[^-\w.]', '_', s)

import base64
def detect_face(image, token):
    try:
        URL = "https://aip.baidubce.com/rest/2.0/face/v3/detect"
        params = {
                "access_token": token
                }
        data = {
                "face_field": "age,gender,beauty,qualities",
                "image_type": "BASE64",
                "image": base64.b64encode(image)
                }
        s = requests.post(URL, params=params, data=data)
        return s.json()["result"]
    except Exception as e:
        print("detect face fail. " + url)
        raise e

def fetch_auth_token(api_key, secret_key):
    try:
        URL = "https://aip.baidubce.com/oauth/2.0/token"
        params = {
                "grant_type": "client_credentials",
                "client_id": api_key,
                "client_secret": secret_key
                }
        s = requests.post(URL, params=params)
        return s.json()["access_token"]
    except Exception as e:
        print("fetch baidu auth token fail. " + url)
        raise e

def init_face_detective(app_id, api_key, secret_key):
    # client = AipFace(app_id, api_key, secret_key)
    # 百度雲 V3 版本接口，需要先擷取 access token   
    token = fetch_auth_token(api_key, secret_key)
    def detective(image):
        #r = client.detect(image, options)
        # 直接使用 HTTP 請求
        r = detect_face(image, token)
        #如果沒有檢測到人臉
        if r is None or r["face_num"] == 0:
            return []

        scores = []
        for face in r["face_list"]:
            #人臉置信度太低
            if face["face_probability"] < 0.6:
                continue
            #顔值低于門檻值
            if face["beauty"] < BEAUTY_THRESHOLD:
                continue
            #性别非女性
            if face["gender"]["type"] != "female":
                continue
            scores.append(face["beauty"])

        return scores

    return detective

def init_env():
    if not os.path.exists(DIR):
        os.makedirs(DIR)

init_env()
face_detective = init_face_detective(APP_ID, API_KEY, SECRET_KEY)

url = BASE_URL % SOURCE + URL_QUERY
while url is not None:
    print("current url: " + url)
    datums = fetch_activities(url)
    url = process_activities(datums, face_detective)
    #注意節操，爬蟲休息間隔不要調小
    time.sleep(5)


# vim: set ts=4 sw=4 sts=4 tw=100 et: