使用Scrapy调用GraphQL API爬取Ouedkniss房产店铺数据报错排查

我尝试从Ouedkniss的房产店铺页面获取数据，发现该网站采用GraphQL API。调用API时拉取数据和分页均失败，出现错误：AttributeError: 'list' object has no attribute 'get'，不确定是否遗漏了配置，以下是当前代码：

import scrapy
import json
from ..items import OuedknissItem
from scrapy.loader import ItemLoader

class StoresSpider(scrapy.Spider):
    name = 'stores'
    allowed_domains = ['www.ouedkniss.com']
  

    def start_requests(self):
        payload = json.dumps([
        {
        "operationName": "SearchStore",
        "query": "query Campaign($slug: String!) {\n  project(slug: $slug) {\n    id\n    isSharingProjectBudget\n    risks\n    story(assetWidth: 680)\n    currency\n    spreadsheet {\n      displayMode\n      public\n      url\n      data {\n        name\n        value\n        phase\n        rowNum\n        __typename\n      }\n      dataLastUpdatedAt\n      __typename\n    }\n    environmentalCommitments {\n      id\n      commitmentCategory\n      description\n      __typename\n    }\n    __typename\n  }\n}\n",
        "variables": {
            "q": "", "filter": {
            "categorySlug": "immobilier", 
            "count": 12, "page": 1},
            "categorySlug": "immobilier",
            "count": 12,
            "page": 1
        },
        
        }
        ])
        headers= {
            "Content-Type": "application/json",
            # "X-Requested-With": "XMLHttpRequest",
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/103.0.0.0 Safari/537.36"
            }
        yield scrapy.Request(
            url='https://api.ouedkniss.com/graphql',
            method="POST",
            headers=headers,
            body=payload,
            callback=self.parse
            )
        return super().start_requests()

    def parse(self, response):
        json_resp = json.loads(response.body)
        # print(json_resp)
        stores = json_resp.get('data')[0].get('stores').get('data')
        for store in stores:
            loader = ItemLoader(item=OuedknissItem())
            loader.add_value('name', store.get('name'))
            yield loader.load_item()

问题根源分析 #

1. GraphQL查询不匹配：operationName是SearchStore，但查询语句是Campaign，请求的是project字段，和预期获取stores的需求完全不符，导致返回数据结构错误。
2. Payload格式错误：把请求体包装成数组，多数GraphQL服务仅接受单个对象，这种格式会导致返回数据结构异常。
3. 变量冗余：variables里同时存在filter对象和单独的categorySlug、count、page，参数结构混乱。
4. 解析逻辑脆弱：假设data是数组，且stores字段一定存在，一旦返回数据不符合预期，调用get方法就会触发AttributeError。

修正后的代码 #

import scrapy
import json
from ..items import OuedknissItem
from scrapy.loader import ItemLoader

class StoresSpider(scrapy.Spider):
    name = 'stores'
    allowed_domains = ['www.ouedkniss.com']
    # 提取公共headers避免重复代码
    headers = {
        "Content-Type": "application/json",
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/103.0.0.0 Safari/537.36"
    }

    def start_requests(self):
        # 匹配operationName的正确查询语句，获取stores数据
        query = """
        query SearchStore($q: String, $filter: StoreFilterInput!, $count: Int, $page: Int) {
          stores(q: $q, filter: $filter, count: $count, page: $page) {
            data {
              id
              name
              # 可添加其他需要的字段
            }
            pagination {
              total
              page
              count
            }
          }
        }
        """
        payload = json.dumps({
            "operationName": "SearchStore",
            "query": query.strip(),
            "variables": {
                "q": "",
                "filter": {"categorySlug": "immobilier"},
                "count": 12,
                "page": 1
            }
        })
        yield scrapy.Request(
            url='https://api.ouedkniss.com/graphql',
            method="POST",
            headers=self.headers,
            body=payload,
            callback=self.parse
        )

    def parse(self, response):
        json_resp = json.loads(response.body)
        
        # 先检查GraphQL返回的错误信息
        if 'errors' in json_resp:
            self.logger.error(f"GraphQL请求错误: {json_resp['errors']}")
            return
        
        # 多级get加默认值，避免字段不存在报错
        stores_data = json_resp.get('data', {}).get('stores', {}).get('data', [])
        for store in stores_data:
            loader = ItemLoader(item=OuedknissItem())
            loader.add_value('name', store.get('name'))
            yield loader.load_item()
        
        # 处理分页逻辑
        pagination = json_resp.get('data', {}).get('stores', {}).get('pagination', {})
        current_page = pagination.get('page', 1)
        total_items = pagination.get('total', 0)
        items_per_page = pagination.get('count', 12)
        total_pages = (total_items + items_per_page - 1) // items_per_page

        if current_page < total_pages:
            next_page = current_page + 1
            # 构造下一页请求
            next_payload = json.dumps({
                "operationName": "SearchStore",
                "query": """
                query SearchStore($q: String, $filter: StoreFilterInput!, $count: Int, $page: Int) {
                  stores(q: $q, filter: $filter, count: $count, page: $page) {
                    data {
                      id
                      name
                    }
                    pagination {
                      total
                      page
                      count
                    }
                  }
                }
                """.strip(),
                "variables": {
                    "q": "",
                    "filter": {"categorySlug": "immobilier"},
                    "count": 12,
                    "page": next_page
                }
            })
            yield scrapy.Request(
                url='https://api.ouedkniss.com/graphql',
                method="POST",
                headers=self.headers,
                body=next_payload,
                callback=self.parse
            )

关键修正点 #

• 替换GraphQL查询语句：使用和operationName匹配的SearchStore查询，直接请求stores字段。
• 调整Payload格式：将数组改为单个对象，符合GraphQL API常规接收格式。
• 优化参数结构：去掉冗余变量，统一按API要求组织参数。
• 增强解析健壮性：使用多级get并设置默认值，避免字段缺失引发错误；增加错误日志便于排查。
• 实现分页功能：根据返回的分页信息自动构造下一页请求。

内容的提问来源于stack exchange，提问作者Raisul Islam