站在巨人的肩膀上开发 — Toutatis 批量获取 Instagram 数据更新数据库

对爬虫不是很熟悉，电商平台数据库里沉淀了一批博主的基本信息（包括主页链接） 分别来自Instagram，TikTok和YouTube ，但博主的粉丝量会增长，如何保持博主粉丝的实时更新就是一个问题，TikTok和YouTube 的博主比较好解决，直接访问该博主的主页链接即可获取该博主的粉丝数量，但Instagram比较特殊，最简单的爬虫爬不出来，直到在github上找到了 Toutatis --https://github.com/megadose/toutatis?tab=readme-ov-file

TikTok和YouTube爬粉丝数：

Python 根据TikTok，YouTube 主页链接批量修改粉丝数

一.项目简介

 Toutatis is a tool that allows you to extract information from instagrams accounts such as e-mails, phone numbers and more

 Toutatis 是一款可让您从 Instagram 帐户中提取信息（例如电子邮件、电话号码等）的工具

二.项目部署

项目拉取下来之后

pip install toutatis

python setup.py install

 根据用户名查找信息（其中sessionid一段时间内不会改变，不刷新google即可）

toutatis -u username -s instagramsessionid

 例如：

可以看到，能够爬取到该博主的平台id，简介，粉丝数，email，phone等等

这样在源代码上进行开发即可满足现有的需求

三.站在巨人的肩膀上开发

上代码：
 

import subprocess
import re  # 导入正则表达式模块
from datetime import datetime
import mysql.connector
from mysql.connector import Error

# 数据库配置
DB_CONFIG = {
    "host": "xxxxxx",  # 数据库地址
    "database": "xxxx",  # 数据库名称
    "user": "xxxx",  # 用户名
    "password": "xxxx",  # 密码
    "port": xxxx  # 端口
}


def get_instagram_usernames_to_update(connection):
    """
    从数据库中获取需要更新的 Instagram 博主用户名
    条件：profile_url 包含 %insta%，且 update_time 不在今天
    :param connection: 数据库连接
    :return: Instagram 博主用户名列表
    """
    try:
        cursor = connection.cursor()
        today = datetime.now().strftime('%Y-%m-%d')  # 获取今天的日期
        sql_query = """
        SELECT name FROM kol_base_info
        WHERE profile_url LIKE '%insta%'
        AND (update_time IS NULL OR DATE(update_time) < %s)
        """
        cursor.execute(sql_query, (today,))  # 将日期作为参数传入
        rows = cursor.fetchall()  # 获取所有结果
        return [row[0] for row in rows]  # 提取用户名列表
    except Error as e:
        print(f"查询失败: {e}")
        return []
    finally:
        cursor.close()

def get_user_info(username):
    """
    使用 toutatis 命令获取 Instagram 博主的 userID 和粉丝数
    :param username: Instagram 博主的用户名
    :return: 包含 userID 和粉丝数的字典 (如果能成功获取), 否则返回 None
    """
    try:
        # 构造 toutatis 命令，用户名转换为小写
        command = f"toutatis -u {username.lower()} -s 71337797720%3AjmkMPB6tx25q4P%3A24%3AAYdyyWUVum_W-er6o8RyRL3aw49VDwIqe5i4Swr2RA"

        # 执行命令，指定输出编码为 utf-8
        print(f"正在执行命令: {command}")  # 添加日志输出，方便调试
        result = subprocess.run(command, shell=True, capture_output=True, text=True, encoding='utf-8')

        # 如果命令成功执行
        if result.returncode == 0:
            # 提取 userID
            user_id_match = re.search(r"userID\s*:\s*(\d+)", result.stdout)
            user_id = int(user_id_match.group(1)) if user_id_match else None

            # 提取粉丝数
            follower_match = re.search(r"Follower\s*:\s*(\d+)", result.stdout)
            followers = int(follower_match.group(1)) if follower_match else None

            # 检查是否成功提取到所需数据
            if user_id is not None and followers is not None:
                print(f"获取到 userID: {user_id}, 粉丝数: {followers}")
                return {"userID": user_id, "followers": followers}
            else:
                print(f"未找到 userID 或粉丝数，输出内容为: {result.stdout}")
                return None
        else:
            # 如果命令执行失败，打印错误信息
            print(f"执行 toutatis 命令失败: {result.stderr}")
            return None
    except subprocess.CalledProcessError as e:
        # 处理 subprocess 执行错误
        print(f"命令执行失败: {e}")
        return None
    except Exception as e:
        # 捕获其他类型的错误
        print(f"获取信息失败: {e}")
        return None

def update_follower_count_and_userid(connection, name, fans_num, user_id):
    """
    更新数据库中对应博主的粉丝数和 platform_id（即 userID）
    :param connection: 数据库连接
    :param name: 博主的用户名
    :param fans_num: 博主的粉丝数
    :param user_id: 博主的 userID
    """
    try:
        cursor = connection.cursor()
        sql_update = """
        UPDATE kol_base_info
        SET fans_num = %s, platform_id = %s
        WHERE name = %s
        """
        # 参数与占位符匹配
        cursor.execute(sql_update, (fans_num, user_id, name))
        connection.commit()
        print(f"成功更新博主 {name} 的粉丝数为 {fans_num}，platform_id 为 {user_id}")
    except Error as e:
        print(f"更新粉丝数和 platform_id 失败: {e}")
    finally:
        cursor.close()

def main():
    try:
        # 连接数据库
        connection = mysql.connector.connect(**DB_CONFIG)
        if connection.is_connected():
            print("成功连接到数据库")

            # 从数据库获取需要更新的 Instagram 博主用户名
            instagram_usernames = get_instagram_usernames_to_update(connection)
            print(f"找到 {len(instagram_usernames)} 个需要更新的 Instagram 用户名")

            # 遍历所有博主
            for username in instagram_usernames:
                print(f"正在查询 {username} 的信息...")

                # 获取 userID 和粉丝数
                user_info = get_user_info(username)

                # 如果成功获取到信息，则更新数据库
                if user_info:
                    user_id = user_info.get("userID")
                    fans_num = user_info.get("followers")

                    if user_id is not None and fans_num is not None:
                        update_follower_count_and_userid(connection, username, fans_num, user_id)
                    else:
                        print(f"未能获取 {username} 的完整信息：{user_info}")
                else:
                    print(f"未能获取 {username} 的信息，跳过更新。")

    except Error as e:
        print(f"数据库连接失败: {e}")
    finally:
        if 'connection' in locals() and connection.is_connected():
            connection.close()
            print("数据库连接已关闭")


if __name__ == "__main__":
    main()

 回头把这三个代码写到一块，做一个定时任务，就完成啦。