spider_work/ins_img_video_spider/ins_posts_spider.py

# -*- coding: utf-8 -*-
# Author : Charley
# Python : 3.8.10
# Date   : 2025/4/2 18:57
import inspect
import random
import time

import schedule
from loguru import logger
from datetime import datetime
# from tls_client import Session
from curl_cffi import Session
from tenacity import retry, stop_after_attempt, wait_fixed

from mysql_pool import MySQLConnectionPool

USER_NAME_LIST = ['fanatics', 'hobbysbestcards']  # 查询的用户名列表 hobbysbestcards

cookie = r'ig_did=8D2CD910-0CBD-41CD-A5B4-9EB7E2F8BC91; ps_l=1; ps_n=1; datr=0aYZaGecXnDrIALr4HPo5O0h; mid=aBmm0QALAAFOBiNIagQ4prL9V4Zg; dpr=1.5; csrftoken=1Eeolr1d8t3VMjwNQIeMMQx9JTlyUsGu; sessionid=50762414324%3Af7LRzwBjb06Q7U%3A6%3AAYfUreTnqm7V_o3Pvqt0Tej1vwMQDGjOKw_Zm8TOqA; ds_user_id=50762414324; rur="RVA\05450762414324\0541778817145:01f75b26510d73b461bb75b0f907b2ec268507a83f95fb7c5a8571ced3b614c68af0d6b5"; wd=1707x247'

PARAMS = r'("app_id":\s*"[^"]+")|("claim":\s*"[^"]+")|("csrf_token":\s*"[^"]+")|(["LSD",[],{"token":\s*"[^"]+")'

# session = Session(client_identifier="chrome_120", random_tls_extension_order=True)
session = Session()

logger.remove()
logger.add("./logs/{time:YYYYMMDD}.log", encoding='utf-8', rotation="00:00",
           format="[{time:YYYY-MM-DD HH:mm:ss.SSS}] {level} {message}",
           level="DEBUG", retention="7 day")

MAX_PAGE = 3  # 最大页数


def after_log(retry_state):
    """
    retry 回调
    :param retry_state: RetryCallState 对象
    """
    # 检查 args 是否存在且不为空
    if retry_state.args and len(retry_state.args) > 0:
        log = retry_state.args[0]  # 获取传入的 logger
    else:
        log = logger  # 使用全局 logger

    if retry_state.outcome.failed:
        log.warning(
            f"Function '{retry_state.fn.__name__}', Attempt {retry_state.attempt_number} Times")
    else:
        log.info(f"Function '{retry_state.fn.__name__}', Attempt {retry_state.attempt_number} succeeded")


@retry(stop=stop_after_attempt(5), wait=wait_fixed(1), after=after_log)
def get_proxys(log):
    """
    获取代理
    :return: 代理
    """
    tunnel = "x371.kdltps.com:15818"
    kdl_username = "t13753103189895"
    kdl_password = "o0yefv6z"
    try:
        proxies = {
            "http": "http://%(user)s:%(pwd)s@%(proxy)s/" % {"user": kdl_username, "pwd": kdl_password, "proxy": tunnel},
            "https": "http://%(user)s:%(pwd)s@%(proxy)s/" % {"user": kdl_username, "pwd": kdl_password, "proxy": tunnel}
        }
        return proxies
    except Exception as e:
        log.error(f"Error getting proxy: {e}")
        raise e


@retry(stop=stop_after_attempt(5), wait=wait_fixed(15), after=after_log)
def ajax_request(log, url: str, params=None):
    """
    请求封装
    :param log: logger对象
    :param url: api url
    :param params: api params
    :return: json object
    """
    try:
        headers = {
            'sec-fetch-mode': 'cors',
            'referer': 'https://www.instagram.com/',
            'x-ig-app-id': '936619743392459',
            'sec-fetch-site': 'same-site',
            'accept-language': 'zh-CN,zh;q=0.9,en-US;q=0.8,en;q=0.7',
            'x-asbd-id': '198387',
            'accept': '*/*',
            'sec-ch-ua': 'Chromium";v="104", " Not A;Brand";v="99", "Google Chrome";v="104"',
            'sec-ch-ua-mobile': '?0',
            'x-ig-www-claim': 'hmac.AR11qy__GsvLpiS4wKBygLGdxs2DxJB1esTkBw7b2QFaHH2d',
            'authority': 'i.instagram.com',
            'sec-ch-ua-platform': 'Windows"',
            'x-instagram-ajax': '1006400593',
            'sec-fetch-dest': 'empty',
            'user-agent': 'Mozilla/5.0 (Windows NT 6.3; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/104.0.5112.102 Safari/537.36',
            'cookie': cookie
        }
        resp = session.get(url, headers=headers, params=params)
        # print(resp.text)
        resp.raise_for_status()

        return resp.json()
    except Exception as e:
        log.error(f"Request failed: {e}")
        raise


def get_userPosts(log, userName: str, sql_uid_list: list):
    """
    从用户名获取所有帖子
    :param log: logger对象
    :param userName:  用户名
    :param sql_uid_list: sql_uid_list 列表
    :return: generator
    """
    page = 1
    continuations = [{
        'count': '12',
    }]
    temp = userName + '/username/'
    while continuations:
        continuation = continuations.pop()
        log.info(f"The page number currently requested is: {page}.........")
        # Url将在第二次请求时更改
        url = 'https://i.instagram.com/api/v1/feed/user' + f'/{temp}'
        resp = ajax_request(log, url, params=continuation)

        if not resp:
            log.error("API请求失败，跳过当前分页")

        time.sleep(random.uniform(5, 8))
        page += 1

        if page > MAX_PAGE:
            log.info(f"The page number currently requested is: {page}.........")
            break

        # 没有这样的用户
        if not resp.get('user'):
            log.warning(f"checking cookie or unknown/private User: {userName}")
            yield 'checking cookie or unknown/private User: {}'.format(userName)
        else:
            _items = resp.get('items', [])
            # 模拟鼠标按下

            if resp.get('more_available'):
                continuations.append({'count': '12', 'max_id': resp.get('next_max_id')})
                user = resp.get('user')
                temp = user.get('pk_id') if user.get('pk_id') else user.get('pk')
            yield from extract_post(log, _items, userName, sql_uid_list)


def extract_post(log, posts, user_name: str, sql_uid_list: list):
    """
    从帖子列表中提取一个帖子
    :param log: logger对象
    :param posts: original instagram posts
    :param user_name: user_name
    :param sql_uid_list: sql_uid_list 列表
    :return: dict of posts
    """
    # print("extract_post")
    if not posts:  # 处理 None 或空列表
        log.debug("No posts found.")
        return {}
    for post in posts:
        # print('post:',post)
        caption = post.get('caption')

        created_at_stamp = caption.get('created_at') if caption else post.get('taken_at')
        created_at = datetime.fromtimestamp(created_at_stamp).strftime("%Y-%m-%d %H:%M:%S")

        uid = post.get('code')
        if uid in sql_uid_list:
            log.info(f"uid:{uid} has been processed, skipping................")
            continue
        item = {
            'user_name': user_name,
            'uid': uid,
            'pid': post.get('pk'),
            'pk_id': post.get('id'),
            'comment_count': post.get('comment_count'),
            'like_count': post.get('like_count'),
            'title': caption.get('text') if caption else None,
            'created_at': created_at
        }
        # 其他类型可再添加
        types = post.get('media_type')

        if types == 8:
            try:
                imgs_list = [post.get('image_versions2', {}).get('candidates', [{}])[0].get('url') for _ in
                             post.get('carousel_media')]
            except Exception as e:
                log.warning(f"imgs_list processing post: {e}")
                # imgs_list = [_.get('url') for _ in post.get('image_versions2', {}).get('candidates', [{}])[0]]
                imgs_list = [post.get('image_versions2', {}).get('candidates', [{}])[0].get('url')]
            imgs = ','.join(imgs_list) if imgs_list else None

            item.update({
                'imgs_url': imgs,
                'video_url': post.get('carousel_media', [{}])[0].get('video_versions', [{}])[0].get('url')
            })
        elif types == 2:
            item.update({
                'imgs_url': None,
                'video_url': post.get('video_versions', [{}])[0].get('url')
            })

        elif types == 1:
            item.update({
                'imgs_url': post.get('image_versions2', {}).get('candidates', [{}])[0].get('url'),
                'video_url': None
            })
        yield item


@retry(stop=stop_after_attempt(50), wait=wait_fixed(1800), after=after_log)
def ins_posts_main(log):
    """
    主函数
    :param log: logger对象
    """
    log.info(
        f'开始运行 {inspect.currentframe().f_code.co_name} 爬虫任务....................................................')

    # 配置 MySQL 连接池
    sql_pool = MySQLConnectionPool(log=log)
    if not sql_pool.check_pool_health():
        log.error("数据库连接池异常")
        raise RuntimeError("数据库连接池异常")

    try:
        for user_name in USER_NAME_LIST:
            log.info(
                f'-------------------------------- 开始爬取用户 {user_name} 的所有帖子 --------------------------------')
            sql_uid_list = sql_pool.select_all('select uid from instagram_posts_record where user_name = %s',
                                               (user_name,))
            sql_uid_list = [_[0] for _ in sql_uid_list]
            log.debug(f'查询到 uid 列表sql_uid_list的长度为: {len(sql_uid_list)}')
            items_ = get_userPosts(log, user_name, sql_uid_list)
            for item_ in items_:
                # print(item_)
                sql_pool.insert_one_or_dict('instagram_posts_record', item_)
            sql_uid_list.clear()
    except Exception as e:
        log.error(f'{inspect.currentframe().f_code.co_name} error: {e}')
    finally:
        log.info(f'爬虫程序 {inspect.currentframe().f_code.co_name} 运行结束,等待下一轮的采集任务............')


def schedule_task():
    """
    设置定时任务
    """
    # ins_posts_main(log=logger)

    # schedule.every().day.at("05:00").do(ins_posts_main, log=logger)
    schedule.every().monday.at("06:00").do(ins_posts_main, log=logger)

    while True:
        schedule.run_pending()
        time.sleep(1)


if __name__ == '__main__':
    schedule_task()