毎日のTinder自動スワイプが面倒臭いことに加えて、せっかくなら好みの女性芸能人に似た人をLike(=それ以外はPass)したいと思ったことがキッカケである。類似度判定には、AWSの画像認識サービスであるRekognition APIを使用してみた。

先ずは、ブラウザから以下URLへアクセスし、FaceBookのアクセストークンを取得するまでの流れを追ってみよう。

ログインが成功すると、リダイレクト後の/v2.10/dialog/oauth/confirm/のレスポンスを確認すると以下のようなデータが取得できる。

<script type="text/javascript">window.location.href="fb464891386855067:\/\/authorize\/#signed_request=LFt3F8b6JR0QYqlcFQQdaWWJ-xTtK-woHyIuQO1Dkug.eyJ1c2VyX2lkIjoiNTczMjQ4MjU2MTIzMjg3IiwiY29kZSI6IkFRQUc1SDRiZ2I4Z0hxZTVNbEtaLWxpWUhiTTN2cnVCeHRYejZWbXJaZ2ctVVJTdHByWE41NEZ1U3ZrYmpKTXN3TFpIalcxSktUc1FZaW5OTERLZ0FOaGRqdXBycVNxUUh3TE55T1FTeDRKck11c2xfZEpvOUFaQjNGUHJsZlgtanlCR05OQ2NMZk82blJSbXF0SGtjVmRDMG9xeVdDZmIxMW1QLTBPOHdhNnYxTHBuV0ZYWGhDUEgwM1p4WnVBQVFrYUZDQUNUUkpFVjBZdlpXT0dRWklXTG1GaUNROVBRbnV4ZUZPN0t2S045RXN5T3lBR2hzOGltWkhCNE9GMFdtY2VuWS1Gajd0QmhOU1FISFBEMC1wSHg5WE1yNXM2U1BrM0Y1OGR6QmdfekdHY2tGRzdiVE1EV0xWd3pQNXpHUHJ6Q1JDMnZ4X25zS1M3alhadkRpSDc1IiwiYWxnb3JpdGhtIjoiSE1BQy1TSEEyNTYiLCJpc3N1ZWRfYXQiOjE1NzA2MzE2NDV9&access_token=EAAGm0PX4ZCpsBAMGnFxdrE8bga4lW6rWBznvojpUjy8twmP1bTA8ZADATxALwveAKb58SZBj<省略>&data_access_expiration_time=1578407645&expires_in=5155";</script>

上記の「access_token=」の箇所がFaceBookの発行するアクセストークンである。Tinder APIを操作するには、X-Auth-Tokenが必要となるが、このトークンを発行するためにFaceBookのアクセストークンが必要となる。

アクセストークン取得クラス

import requests
import json
from bs4 import BeautifulSoup
import boto3
from botocore.exceptions import ClientError

class GetAccessToken:
    def __init__(self):
        self.client_id = '464891386855067'
        self.fallback_redirect_uri = '221e1158-f2e9-1452-1a05-8983f99f7d6e'
        self.logger_id = '0fb8ab93-fab7-4f2c-b836-483bb9ae6595'
        self.fb_dtsg = 'AQEMLhbRCqkx:AQH2Bj3HImtl'
        self.email = '*****' #FaceBookログインメールアドレス
        self.password = '*****' #FaceBookパスワード
        self.redirect_uri = 'fb464891386855067://authorize/'
        self.jazoest = '22086'
        self.base_url = 'https://www.facebook.com'
        self.dialog_url = self.base_url + '/v2.10/dialog/oauth'
        self.confirm_url = self.base_url + '/v2.10/dialog/oauth/confirm'
        self.session = requests.Session()
        self.default_headers = {
            'Accept-Language': 'ja,en-US;q=0.9,en;q=0.8',
            'Content-Type': 'application/x-www-form-urlencoded, application/json',
            'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3',
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/77.0.3865.90 Safari/537.36',
        }

    def __get_login_page(self):
        params = {
            'redirect_uri': self.redirect_uri,
            'scope': 'user_birthday,user_photos,user_education_history,email,user_relationship_details,user_friends,user_work_history,user_likes',
            'response_type': 'token,signed_request',
            'client_id': self.client_id,
            'ret': 'login',
            'fallback_redirect_uri': self.fallback_redirect_uri,
            'ext': '1556057433',
            'hash': 'Aea6jWwMP_tDMQ9y'
        }

        try:
            res = self.session.get(url=self.dialog_url, headers=self.default_headers, params=params)
            return res
        except Exception as e:
            return 'Caught Exception {}'.format(e)

    def __login(self):
        res = self.__get_login_page()
        login_form_html = BeautifulSoup(res.text, 'html.parser')
        uri = login_form_html.find(attrs={'id': 'login_form'}).get('action')

        data = {
            'email': self.email,
            'pass': self.password
        }

        try:
            res = self.session.post(url=self.base_url + uri, headers=self.default_headers, data=data, cookies=res.cookies.get_dict())
            return res
        except Exception as e:
            return 'Caught Exception {}'.format(e)

    def get_access_token(self):
        res = self.__login()

        data = {
            'jazoest': self.jazoest,
            'fb_dtsg': self.fb_dtsg,
            'from_post': '1',
            'app_id': self.client_id,
            'ret': 'login',
            'return_format': 'signed_request,access_token',
            'logger_id': self.logger_id,
            'sheet_name': 'initial',
            'redirect_uri': self.redirect_uri,
            'fallback_redirect_uri': self.fallback_redirect_uri,
            'display': 'page',
            '__CONFIRM__': '1'
        }

        try:
            response = self.session.post(url=self.confirm_url, headers=self.default_headers, data=data, cookies=res.cookies.get_dict())
            idx = response.text.find('&access_token=')
            end_idx = response.text.find('&data_access_expiration_time=')
            access_token = response.text[idx+14:end_idx]
            return access_token
        except Exception as e:
            return 'Caught Exception {}'.format(e)

類似度判定クラス (※折角なので、マッチのユーザが見つかった際に、SNSでメール通知を行うようにした)

S3バケットの「src」フォルダに候補女性の画像一式を自動でダウンロードしアップロードするようにしている。
「target」フォルダには好みの女性の写真を格納してね。

class AWSAPI:
    def __init__(self):
        self.sns = boto3.client('sns')
        self.rekognition = boto3.client('rekognition')
        self.src_bucket_name = '*****' #画像データを格納するS3バケット名
        self.s3 = boto3.client('s3')
        self.topicArn = '*****' #トピックARN

    def compare_images(self, bucketName, srcKeyName, targetKeyName):
        try:
            response = self.rekognition.compare_faces(
                SourceImage={
                    'S3Object': {
                        'Bucket': bucketName,
                        'Name': srcKeyName,
                    }
                },
                TargetImage={
                    'S3Object': {
                        'Bucket': bucketName,
                        'Name': targetKeyName,
                    }
                },
                SimilarityThreshold=0
            )
        except ClientError as e:
            return 'Caught ClientError {}'.format(e)

        return response['FaceMatches'][0]['Similarity']

    def create_obj_list(self, bucketName, targetfolderName):
        try:
            response = self.s3.list_objects(
                Bucket=bucketName,
                )
            target_key_list = []
            for i in range(len(response['Contents'])):
                if targetfolderName in response['Contents'][i]['Key']:
                    target_key_list.append(response['Contents'][i]['Key'])
            target_key_list.remove(targetfolderName + '/')
            return target_key_list
        except ClientError as e:
            return 'Caught ClientError {}'.format(e)

    def upload_images(self, bucketName, localfile, keyName):
        with open(localfile, 'rb') as data:
            self.s3.upload_fileobj(data, self.src_bucket_name, keyName)

    def delete_images(self, bucketName, keyName):
        try:
            response = self.s3.delete_object(
                Bucket=bucketName,
                Key=keyName
            )
            return response
        except ClientError as e:
            return 'Caught ClientError {}'.format(e)

    def publish_sns(self):
        try:
            response = self.sns.publish(
                TopicArn=self.topicArn,
                Message='マッチしたユーザが見つかりました。',
                Subject='Tinder Match Notification.'
            )
        except ClientError as e:
            print('Caught exception: %s' % e)

Tinder API実行クラス

"<類似度の閾値>"に閾値を入れる。可愛い女性芸能人で閾値を70(%)とかにすると、殆どライク判定がされないので注意。w
まぁ、そりゃそうだ。

class TinderAPI(AWSAPI):
    def __init__(self, access_token):
        super().__init__()
        facebook_id = '**********'
        self.base_url = 'https://api.gotinder.com'
        auth_url = self.base_url + '/v2/auth/login/facebook'
        data = {
            'facebook_id': facebook_id,
            'token': access_token
        }

        self.session = requests.Session()
        res = self.session.post(url=auth_url, data=json.dumps(data))
        self.x_auth_token = res.json()['data']['api_token']

        self.api_headers = {
            'X-Auth-Token': self.x_auth_token,
            'Content-type': 'application/json',
            'User-agent': 'Tinder/3.0.4 (iPhone; iOS 7.1; Scale/2.00)'
        }

    def swipe_like_user(self):
        uri = self.base_url + '/user/recs'
        response = self.session.get(url=uri, headers=self.api_headers)
        res = json.loads(response.text)

        for i in range(len(res['results'])):
            for j in range(len(res['results'][i]['photos'][0]['processedFiles'])):
                img_url = res['results'][i]['photos'][0]['processedFiles'][j]['url']
                self.__get_images(img_url, j)
                self.upload_images(bucketName=self.src_bucket_name, localfile=str(j)+'.jpg', keyName='src/'+str(j)+'.jpg')
                target_key_list = self.create_obj_list(self.src_bucket_name, 'target')
                for k in range(len(target_key_list)):
                    response = self.compare_images(self.src_bucket_name, 'src/' + str(j) + '.jpg', target_key_list[k])
                    like_res = self.__like_user(res['results'][i]['_id']) if response > <類似度の閾値> else self.__pass_user(res['results'][i]['_id'])
                    self.publish_sns() if (self.__like_user(res['results'][i]['_id'])) else print('Not matched.')
            src_key_list = self.create_obj_list(self.src_bucket_name, 'src')
            for delete_object in src_key_list:
                self.delete_images(self.src_bucket_name, delete_object)

    def __get_images(self, url, prefix):
        img = requests.get(url, stream=True)
        with open(str(prefix) + '.jpg', 'wb') as f:
            f.write(img.content)

    def __like_user(self, uid):
        uri = self.base_url + '/like/' + uid
        res = self.session.get(url=uri, headers=self.api_headers)
        if json.loads(res.text)['match'] == True:
            return True
        else:
            return False

    def __pass_user(self, uid):
        uri = self.base_url + '/pass/' + uid
        res = self.session.get(url=uri, headers=self.api_headers)

        return res