spider_work/zhongjian_spider/zj_new_daily_spider.py

# -*- coding: utf-8 -*-
# Author : Charley
# Python : 3.10.8
# Date   : 2025/6/9 15:56
import sys
import time
import json
import requests
import schedule
import hashlib
import base64
import user_agent
from loguru import logger
from datetime import datetime
from Crypto.Cipher import AES
from Crypto.Util.Padding import unpad
from mysql_pool import MySQLConnectionPool
from tenacity import retry, stop_after_attempt, wait_fixed

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")

RESPONSE_KEY = b"3bd48ea5e910b195843941351be7cbae"  # 16字节AES密钥(UTF8编码)
REQUEST_KEY = "1ba48ea2e910b666843941351be7cbad"

"""
https://www.zhongjianjiantong.com/web/index.html#/pages/check/details?rating_no=519149616
"""

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


def make_sign():
    """生成sign: MD5(REQUEST_KEY + 当前时间戳秒)"""
    now = datetime.now()
    dt_str = now.strftime("%Y-%m-%d %H:%M:%S")
    # JS: new Date(o).getTime() / 1e3 — 取秒级时间戳
    timestamp = int(datetime.strptime(dt_str, "%Y-%m-%d %H:%M:%S").timestamp())
    raw = REQUEST_KEY + str(timestamp)
    return hashlib.md5(raw.encode()).hexdigest(), dt_str


def decrypt_response(data_b64, iv_hex):
    """AES-CBC解密响应数据"""
    iv = iv_hex.encode("utf-8")  # JS: CryptoJS.enc.Utf8.parse(iv) — 直接UTF8编码
    cipher = AES.new(RESPONSE_KEY, AES.MODE_CBC, iv)
    ciphertext = base64.b64decode(data_b64)
    plaintext = unpad(cipher.decrypt(ciphertext), AES.block_size)
    return json.loads(plaintext.decode("utf-8"))


@retry(stop=stop_after_attempt(5), wait=wait_fixed(1), after=after_log)
def get_request_one_page(log, rating_no) -> dict:
    """
    获取单页数据
    :param log: logger
    :param rating_no: 证书编号(即 tag_no)
    :return: dict, 外层结构 {'code':..., 'msg':..., 'data':...}。
             code==200 时 data 为解密后的业务字典(含 obj_order_rating_goods);
             code==400 表示该评级单不存在;其他为异常。
    """
    sign, dt_str = make_sign()

    headers = {
        "accept": "*/*",
        "accept-language": "en,zh-CN;q=0.9,zh;q=0.8",
        "content-type": "application/json;charset=UTF-8",
        "datetime": dt_str,
        "origin": "https://www.zhongjianjiantong.com",
        "referer": "https://www.zhongjianjiantong.com/web/index.html",
        "sign": sign,
        "user-agent": user_agent.generate_user_agent()
    }
    # 20260903  接口变化
    # url = "https://www.zhongjianjiantong.com/Api/OrderRatingGoods/detail"
    url = "https://www.zhongjianjiantong.com/Api/OrderRatingGoods/webDetail"
    #      https://www.zhongjianjiantong.com/Api/OrderRatingGoods/webDetail
    data = {
        "rating_no": rating_no
    }

    with requests.Session() as session:
        response = session.post(url, headers=headers, json=data, proxies=get_proxys(log), timeout=5)
    # print(response.text)
    response.raise_for_status()
    result = response.json()
    # print(result)
    # 新版 webDetail:成功(code==200 且带 iv)时对 data 做 AES-CBC 解密,
    # 把解密后的业务数据(含 obj_order_rating_goods)回填到 result['data'],
    # code 始终保留在外层,供上层 loop_rating_no 判断状态(不存在时 code==400)。
    if result.get("code") == 200 and result.get("iv"):
        result["data"] = decrypt_response(result["data"], result["iv"])
    return result


def parse_data(resp_json, sql_pool):
    """
    解析数据
    :param resp_json: 响应数据
    :param sql_pool: 数据库连接池
    """
    card_id = resp_json.get('obj_order_rating_goods', {}).get('id')
    order_no = resp_json.get('obj_order_rating_goods', {}).get('order_no')
    tag_no = resp_json.get('obj_order_rating_goods', {}).get('tag_no')  # 标签号/查询的号码

    images = resp_json.get('obj_order_rating_goods', {}).get('images')
    card_create_time = resp_json.get('obj_order_rating_goods', {}).get('create_time')
    card_update_time = resp_json.get('obj_order_rating_goods', {}).get('update_time')
    score = resp_json.get('obj_order_rating_goods', {}).get('score')  # 中检评分(总分)
    # 注意:2026/09 起 webDetail 不再返回以下分项评分(四角/边缘/表面/居中/颜色/修复),
    #      obj_detail 仅保留 spxl/spmc/fxnf/yy/spbh,故下列字段现均为 None,入库即 NULL。
    corners = resp_json.get('obj_order_rating_goods', {}).get('obj_detail', {}).get('corners')  # 四角
    eoges = resp_json.get('obj_order_rating_goods', {}).get('obj_detail', {}).get('eoges')  # 边缘
    surface = resp_json.get('obj_order_rating_goods', {}).get('obj_detail', {}).get('surface')  # 表面
    centering = resp_json.get('obj_order_rating_goods', {}).get('obj_detail', {}).get('centering')  # 居中
    colour = resp_json.get('obj_order_rating_goods', {}).get('obj_detail', {}).get('colour')  # 颜色
    repair = resp_json.get('obj_order_rating_goods', {}).get('obj_detail', {}).get('repair')  # 修复

    rating_no = resp_json.get('obj_order_rating_goods', {}).get('rating_no')  # 证书编号
    obj_brand_title = resp_json.get('obj_order_rating_goods', {}).get('obj_brand', {}).get(
        'title')  # 商品品牌
    obj_detail_spxl = resp_json.get('obj_order_rating_goods', {}).get('obj_detail', {}).get(
        'spxl')  # 商品系列
    obj_detail_spmc = resp_json.get('obj_order_rating_goods', {}).get('obj_detail', {}).get(
        'spmc')  # 商品名称
    obj_detail_fxnf = resp_json.get('obj_order_rating_goods', {}).get('obj_detail', {}).get(
        'fxnf')  # 发行年份
    obj_detail_yy = resp_json.get('obj_order_rating_goods', {}).get('obj_detail', {}).get('yy')  # 语言
    obj_detail_spbh = resp_json.get('obj_order_rating_goods', {}).get('obj_detail', {}).get(
        'spbh')  # 商品编号

    info = (
        card_id, order_no, tag_no, images, card_create_time, card_update_time, score, corners, eoges, surface,
        centering,
        colour, repair, rating_no, obj_brand_title, obj_detail_spxl, obj_detail_spmc, obj_detail_fxnf, obj_detail_yy,
        obj_detail_spbh)
    # print(info)
    sql = """
    INSERT IGNORE INTO zhongjian_record (card_id, order_no, tag_no, images, card_create_time, card_update_time, score, corners, eoges, surface, centering, colour, repair, rating_no, obj_brand_title, obj_detail_spxl, obj_detail_spmc, obj_detail_fxnf, obj_detail_yy, obj_detail_spbh)
    VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
    """
    sql_pool.insert_one(sql, info)


def loop_rating_no(log, sql_pool, sql_ra_no_list):
    """
    循环处理每个证书编号
    :param log: logger
    :param sql_pool: 数据库连接池
    :param sql_ra_no_list: 证书编号列表
    """
    # sql_ra_no_list = sql_pool.select_all('select tag_no from zhongjian_task where state = 0 limit 10000')
    # sql_ra_no_list = [i[0] for i in sql_ra_no_list]
    for rating_no_ in sql_ra_no_list:
        log.info(f"{rating_no_} is start ......................................")
        try:
            resp_json = get_request_one_page(log, rating_no_)
            code = resp_json.get('code') if isinstance(resp_json, dict) else None
            # 新版 webDetail:成功时 code==200 且 data 内含 obj_order_rating_goods;不存在时 code==400
            biz = resp_json.get('data') if isinstance(resp_json, dict) else None

            if code == 200 and isinstance(biz, dict) and biz.get('obj_order_rating_goods'):
                # print(resp_json)
                parse_data(biz, sql_pool)
                sql_pool.update_one('update zhongjian_task set state = 1 where tag_no = %s', (rating_no_,))
            elif code == 400:
                log.warning(f"{rating_no_} is not exist ......................................")
                sql_pool.update_one('update zhongjian_task set state = 2 where tag_no = %s', (rating_no_,))
            else:
                log.warning(f"{rating_no_} other warning code={code}, please check ......")
                sql_pool.update_one('update zhongjian_task set state = 3 where tag_no = %s', (rating_no_,))
        except Exception as e:
            log.warning(f"{sys._getframe().f_code.co_name} error: {e}")
            sql_pool.update_one('update zhongjian_task set state = 3 where tag_no = %s', (rating_no_,))
            continue


@retry(stop=stop_after_attempt(100), wait=wait_fixed(3600), after=after_log)
def zhongjian_main(log):
    """
    主函数
    :param log:
    """
    log.info(
        f'开始运行 {sys._getframe().f_code.co_name} 爬虫任务....................................................')

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

    try:
        # 单轮内循环:每批取 5 万 state=0,处理完再取下一批,直到本轮把 state=0 抓空为止。
        # loop_rating_no 对每条都会把 state 置为 1/2/3(必然离开 state=0),故循环单调收敛、不会死循环。
        batch_no = 0
        while True:
            # 519/529/539/549 新入队的号码段均在此,走 index_state 索引
            sql_ra_no_list = sql_pool.select_all(
                "select tag_no from zhongjian_task where state = 0 limit 50000")
            sql_ra_no_list = [i[0] for i in sql_ra_no_list]
            if not sql_ra_no_list:
                log.info(f'state=0 任务已全部处理完,本轮结束,等待下一轮........................................')
                break

            batch_no += 1
            log.info(f'第 {batch_no} 批:取到 {len(sql_ra_no_list)} 条待处理,开始采集............')
            try:
                loop_rating_no(log, sql_pool, sql_ra_no_list)
            except Exception as e:
                log.error(f'{sys._getframe().f_code.co_name} 第 {batch_no} 批 error: {e}')
    except Exception as e:
        log.error(f'{sys._getframe().f_code.co_name} error: {e}')
    finally:
        log.info(f'爬虫程序 {sys._getframe().f_code.co_name} 运行结束,等待下一轮的采集任务............')


def schedule_task():
    """
    爬虫模块 定时任务 的启动文件
    """
    # 立即运行一次任务
    zhongjian_main(log=logger)

    # 设置定时任务
    # schedule.every(30).days.at("00:01").do(zhongjian_main, log=logger)
    schedule.every().days.at("05:00").do(zhongjian_main, log=logger)

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


if __name__ == '__main__':
    # schedule_task()
    zhongjian_main(log=logger)