Crawl_record

#可以通过浏览器复制xml路径或取
html=etree.parse('test.html',etree.HTMLParser()) #指定解析器HTMLParser会根据文件修复HTML文件中缺失的如声明信息
#-----  或者
html=etree.HTML(text) #初始化生成一个XPath解析对象
result=etree.tostring(html,encoding='utf-8')   #解析对象输出代码
print(type(html))
print(type(result))
print(result.decode('utf-8'))
# -------  属性匹配
html=etree.HTML(text,etree.HTMLParser())
#选取class为item-1的li节点
result=html.xpath('//li[@class="item-1"]')
# -------  文本获取
result=html.xpath('//li[@class="item-1"]/a/text()') #获取a节点下的内容
#---------  属性获取
result=html.xpath('//li/a/@href')  #获取a的href属性

from lxml import etree

text1='''
<div>
    <ul>
         <li class="aaa" name="item"><a href="link1.html">第一个</a></li>
         <li class="aaa" name="fore"><a href="link2.html">second item</a></li>
     </ul>
 </div>
'''
html=etree.HTML(text1,etree.HTMLParser())
result=html.xpath('//li[@class="aaa" and @name="fore"]/a/text()')
result1=html.xpath('//li[contains(@class,"aaa") and @name="fore"]/a/text()')
print(result)
print(result1)
#
['second item']
['second item']（11）XPath中的运算符
#---------示例二
from lxml import etree

text1='''
<div>
    <ul>
         <li class="aaa" name="item"><a href="link1.html">第一个</a></li>
         <li class="aaa" name="item"><a href="link1.html">第二个</a></li>
         <li class="aaa" name="item"><a href="link1.html">第三个</a></li>
         <li class="aaa" name="item"><a href="link1.html">第四个</a></li> 
     </ul>
 </div>
'''

html=etree.HTML(text1,etree.HTMLParser())

result=html.xpath('//li[contains(@class,"aaa")]/a/text()') #获取所有li节点下a节点的内容
result1=html.xpath('//li[1][contains(@class,"aaa")]/a/text()') #获取第一个
result2=html.xpath('//li[last()][contains(@class,"aaa")]/a/text()') #获取最后一个
result3=html.xpath('//li[position()>2 and position()<4][contains(@class,"aaa")]/a/text()') #获取第一个
result4=html.xpath('//li[last()-2][contains(@class,"aaa")]/a/text()') #获取倒数第三个

import json
# 创建字典类型Person
person = {
    'name': '知秋小梦',
    'gender': 'male',
    'age': 18
}
# Python字典类型转换为JSON对象
json_person = json.dumps(person)
print(json_person)

# 写入 JSON 数据
with open('data.json', 'w') as f:
    json.dump(data, f)
 
# 读取数据
with open('data.json', 'r') as f:
    data = json.load(f)

import requests
from lxml import etree
import re
import pymysql
class GovementSpider(object):
    def __init__(self):
        self.url = 'http://www.mca.gov.cn/article/sj/xzqh/2019/'
        self.headers = {'User-Agent': 'Mozilla/5.0'}
        # 创建2个对象
        self.db = pymysql.connect('127.0.0.1', 'root', '123456', 'govdb', charset='utf8')
        self.cursor = self.db.cursor()

    # 获取假链接
    def get_false_link(self):
        html = requests.get(url=self.url, headers=self.headers).text
        # 此处隐藏了真实的二级页面的url链接，真实的在假的响应网页中，通过js脚本生成，
        # 假的链接在网页中可以访问，但是爬取到的内容却不是我们想要的
        parse_html = etree.HTML(html)
        a_list = parse_html.xpath('//a[@class="artitlelist"]')
        for a in a_list:
            # get()方法:获取某个属性的值
            title = a.get('title')
            if title.endswith('代码'):
                # 获取到第1个就停止即可，第1个永远是最新的链接
                false_link = 'http://www.mca.gov.cn' + a.get('href')
                print("二级“假”链接的网址为", false_link)
                break
        # 提取真链接
        self.incr_spider(false_link)
    # 增量爬取函数
    def incr_spider(self, false_link):   #false_link: http://www.mca.gov.cn/article/sj/xzqh/2019/202002/20200200024708.shtml
        self.cursor.execute('select url from version where url=%s', [false_link])
        # fetchall: (('http://xxxx.html',),)
        result = self.cursor.fetchall()
        # not result:代表数据库version表中无数据
        if not result:
            self.get_true_link(false_link)
            # 可选操作: 数据库version表中只保留最新1条数据
            self.cursor.execute("delete from version")
            # 把爬取后的url插入到version表中
            self.cursor.execute('insert into version values(%s)', [false_link])
            self.db.commit()
        else:
            print('数据已是最新,无须爬取')

    # 获取真链接
    def get_true_link(self, false_link):
        # 先获取假链接的响应,然后根据响应获取真链接
        html = requests.get(url=false_link, headers=self.headers).text
        # 从二级页面的响应中提取真实的链接（此处为JS动态加载跳转的地址）
        re_bds = r'window.location.href="(.*?)"'
        pattern = re.compile(re_bds, re.S)
        true_link = pattern.findall(html)[0]
        self.save_data(true_link)  # 提取真链接的数据
    # 用xpath直接提取数据
    def save_data(self, true_link):
        html = requests.get(url=true_link, headers=self.headers).text
        # 基准xpath,提取每个信息的节点列表对象
        parse_html = etree.HTML(html)
        tr_list = parse_html.xpath('//tr[@height="19"]')
        for tr in tr_list:
            code = tr.xpath('./td[2]/text()')[0].strip()  # 行政区划代码
            name = tr.xpath('./td[3]/text()')[0].strip()  # 单位名称
            print(name, code)
    # 主函数
    def main(self):
        self.get_false_link()
if __name__ == '__main__':
    spider = GovementSpider()
    spider.main()

import requests
import time
from fake_useragent import UserAgent


class DoubanSpider(object):
    def __init__(self):
        self.base_url = 'https://movie.douban.com/j/chart/top_list?'
        self.i = 0
    def get_html(self, params):
        headers = {'User-Agent': UserAgent().random}
        res = requests.get(url=self.base_url, params=params, headers=headers)
        res.encoding = 'utf-8'
        html = res.json()  # 将json格式的字符串转为python数据类型
        self.parse_html(html)  # 直接调用解析函数
    def parse_html(self, html):
        # html: [{电影1信息},{电影2信息},{}]
        item = {}   #python 列表的使用
        for one in html:
            item['name'] = one['title']  # 电影名
            item['score'] = one['score']  # 评分
            item['time'] = one['release_date']  # 打印测试
            # 打印显示
            print(item)
            self.i += 1
    # 获取电影总数
    def get_total(self, typ):
        # 异步动态加载的数据 都可以在XHR数据抓包
        url = 'https://movie.douban.com/j/chart/top_list_count?type={}&interval_id=100%3A90'.format(typ)
        ua = UserAgent()
        html = requests.get(url=url, headers={'User-Agent': ua.random}).json()
        total = html['total']
        return total
    def main(self):
        typ = input('请输入电影类型(剧情|喜剧|动作):')
        typ_dict = {'剧情': '11', '喜剧': '24', '动作': '5'}
        typ = typ_dict[typ]
        total = self.get_total(typ)  # 获取该类型电影总数量
        for page in range(0, int(total), 20):
            params = {
                'type': typ,
                'interval_id': '100:90',
                'action': '',
                'start': str(page),
                'limit': '20'}
            self.get_html(params)
            time.sleep(1)
        print('爬取的电影的数量:', self.i)
if __name__ == '__main__':
    spider = DoubanSpider()
    spider.main()

import time
import json
import random
import requests
from useragents import ua_list


class TencentSpider(object):
    def __init__(self):
        self.one_url = 'https://careers.tencent.com/tencentcareer/api/post/Query?timestamp=1563912271089&countryId=&cityId=&bgIds=&productId=&categoryId=&parentCategoryId=&attrId=&keyword=&pageIndex={}&pageSize=10&language=zh-cn&area=cn'
        self.two_url = 'https://careers.tencent.com/tencentcareer/api/post/ByPostId?timestamp=1563912374645&postId={}&language=zh-cn'
        self.f = open('tencent.json', 'a')  # 打开文件
        self.item_list = []  # 存放抓取的item字典数据

    # 获取响应内容函数
    def get_page(self, url):
        headers = {'User-Agent': random.choice(ua_list)}
        html = requests.get(url=url, headers=headers).text
        html = json.loads(html)  # json格式字符串转为Python数据类型

        return html

    # 主线函数: 获取所有数据
    def parse_page(self, one_url):
        html = self.get_page(one_url)
        item = {}
        for job in html['Data']['Posts']:
            item['name'] = job['RecruitPostName']  # 名称
            post_id = job['PostId']  # postId，拿postid为了拼接二级页面地址
            # 拼接二级地址,获取职责和要求
            two_url = self.two_url.format(post_id)
            item['duty'], item['require'] = self.parse_two_page(two_url)
            print(item)
            self.item_list.append(item)  # 添加到大列表中

    # 解析二级页面函数
    def parse_two_page(self, two_url):
        html = self.get_page(two_url)
        duty = html['Data']['Responsibility']  # 工作责任
        duty = duty.replace('\r\n', '').replace('\n', '')  # 去掉换行
        require = html['Data']['Requirement']  # 工作要求
        require = require.replace('\r\n', '').replace('\n', '')  # 去掉换行

        return duty, require

    # 获取总页数
    def get_numbers(self):
        url = self.one_url.format(1)
        html = self.get_page(url)
        numbers = int(html['Data']['Count']) // 10 + 1  # 每页有10个推荐

        return numbers

    def main(self):
        number = self.get_numbers()
        for page in range(1, 3):
            one_url = self.one_url.format(page)
            self.parse_page(one_url)

        # 保存到本地json文件:json.dump
        json.dump(self.item_list, self.f, ensure_ascii=False)
        self.f.close()


if __name__ == '__main__':
    start = time.time()
    spider = TencentSpider()
    spider.main()
    end = time.time()
    print('执行时间:%.2f' % (end - start))

import requests
import time
from fake_useragent import UserAgent
import re
import json
from urllib import parse
from lxml import etree
import xlwt     
import pandas as pd
import random
class JobSpider(object):
    def __init__(self):
        self.base_url = 'https://movie.douban.com/j/chart/top_list?'
        self.i = 0   #记录计数器
        #self.wirte_sheet=self.init_xkwt()  

    def init_xkwt(self):
        '''
            初始化csv文件表头
        '''
        #新建表格空间
        excel1 = xlwt.Workbook()
        #新建一个sheet,设置单元格格式,cell_overwrite_ok=True防止对一个单元格重复操作引发的错误
        sheet1 = excel1.add_sheet('Job', cell_overwrite_ok=True)
        sheet1.write(0, 0, 'jobid')
        sheet1.write(0, 1, 'job_href')
        sheet1.write(0, 2, 'job_name')
        sheet1.write(0, 3, 'jobwelf')
        sheet1.write(0, 4, 'attribute_text')
        sheet1.write(0, 5, 'providesalary_text')
        sheet1.write(0, 6, 'company_name')
        sheet1.write(0, 7, 'companysize_text')
        sheet1.write(0, 8, 'companyind_text')
        sheet1.write(0, 9, 'company_href')
        sheet1.write(0, 10,'degreefrom')
        sheet1.write(0, 11,'companytype_text')
        sheet1.write(0, 12,'workarea_text')
        sheet1.write(0, 13,'companytype_text')
        sheet1.write(0, 14,'updatedate')
        sheet1.write(0, 15,'job_info')
        sheet1.write(0, 16,'job_work')
        sheet1.write(0, 17,'company_info')
        return sheet1
        #sheet1.write(number,0,number)
    def parse_html(self, html):
        '''
            html: [{jobid},{job_href},{job_name},{jobwelf},{attribute_text},{providesalary_text},{company_name},{companysize_text},{companyind_text},{company_href},{degreefrom},{companytype_text},{workarea_text},{updatedate}]
            解析一页中html文件，获得如上文件个数数据
        '''
        items = []   #python 列表的使用
        for one in html:
            item={}
            item['jobid']=one['jobid']
            item['job_href']=one['job_href']
            item['job_name']=one['job_name']
            item['job_welf']=one['jobwelf']
            #print(item['job_href'])
            # 根据job介绍url地址 获取 职位要求，工作地点，公司信息
            a,b,c=self.get_jobdetail(item['job_href'])
            item['job_info']=a
            item['job_work']=b
            item['company_info']=c
            item['attribute_text']=one['attribute_text']
            item['providesalary_text']=one['providesalary_text']
            item['company_name']=one['company_name']
            item['companysize_text']=one['companysize_text']
            item['companyind_text']=one['companyind_text']
            item['company_href']=one['company_href']
            item['degreefrom']=one['degreefrom']
            item['companytype_text']=one['companytype_text']
            item['workarea_text']=one['workarea_text']
            item['updatedate']=one['updatedate']
            items.append(item)
        print("parse_one_page finished: size=",len(html),"开始存储")
        df=pd.DataFrame(items)
        df.to_csv("51job_data.csv",encoding="utf_8_sig",mode = 'a',columns=['jobid','job_href','job_name','job_welf','job_info','job_work','company_info',
            'attribute_text','providesalary_text','company_name','companysize_text','companyind_text','company_href',
            'degreefrom','companytype_text','workarea_text','updatedate'])



    def get_html(self, key,page):
        '''
            根据key，page获取对应的html文件中对应的有效数据
        '''
        #伪装爬取头部，以防止被网站禁止
        headers={'Host':'search.51job.com',
                'Upgrade-Insecure-Requests':'1',
                'User-Agent':'Mozilla/5.0 (Windows NT 6.1) AppleWebKit/537.36 (KHTML, like Gecko)\
                Chrome/63.0.3239.132 Safari/537.36'}
        headers = {'User-Agent': UserAgent().random}
        url='https://search.51job.com/list/050000%252c010000,000000,0100,00,9,99,{},2,{}.html'.format(key,page)
        html = requests.get(url=url, headers=headers).content.decode('gbk')
        pattern=re.compile(r'window.__SEARCH_RESULT__ =(.*?)</script>')
        data=pattern.findall(html)[0]
        data=json.loads(data)["engine_search_result"]
        #print("长度",len(data))
        self.parse_html(data)  # 直接调用解析函数

    def get_jobdetail(self,url):
        '''
            https://jobs.51job.com/beijing-hdq/126533792.html?s=01&t=0
        '''
        headers = {'User-Agent': UserAgent().random}
        html = requests.get(url=url, headers=headers).content.decode('gbk')
        parse_html = etree.HTML(html)
        job_info=parse_html.xpath('/html/body/div[3]/div[2]/div[3]/div[1]/div/p/text()')  #/html/body/div[3]/div[2]/div[3]/div[1]/div/p[1]
        job_work=parse_html.xpath('/html/body/div[3]/div[2]/div[3]/div[2]/div/p/text()')
        company_info=parse_html.xpath('/html/body/div[3]/div[2]/div[3]/div[3]/div/text()')
        #print(job_info,job_work,company_info)
        return job_info,job_work,company_info

    # 获取page 总数   #"total_page":"195"
    def get_total(self,key,page=1):
        url='https://search.51job.com/list/000000,000000,0100,00,9,99,{},2,{}.html'.format(key,page)
        ua = UserAgent()
        html = requests.get(url=url, headers={'User-Agent': ua.random}).content.decode('gbk')
        pattern=re.compile(r'window.__SEARCH_RESULT__ =(.*?)</script>')
        data=pattern.findall(html)[0]
        data=json.loads(data)
        total =data['total_page']
        #print("url:",url,"该职位总页数为：",total)
        return total
    def main(self):
        #key = input('请输入职业类型(数据挖掘|喜剧|动作):')
        key=['数据挖掘','人工智能','语音识别','物联网','嵌入式','java','python','c++','android','运维','销售','制造']
        #key=parse.quote(parse.quote(key))
        for item in key:
            key=parse.quote(parse.quote(item))
            total = self.get_total(key)  # 获取该类型电影总数量
            print("当前存储记录：",self.i,"\t 即将爬取关键词：",item,"\t 该关键词个数: ",total)
            for page in range(1, int(total)):
                self.get_html(key,page)
                time.sleep(random.random()*4)
        #print('爬取的电影的数量:', self.i)
if __name__ == '__main__':
    spider = JobSpider()
    spider.main()

#https://search.51job.com/list/050000,000000,0100,00,9,99,%25E6%2595%25B0%25E6%258D%25AE%25E6%258C%2596%25E6%258E%2598,2,1.html?lang=c&postchannel=0000&workyear=99&cotype=99&degreefrom=99&jobterm=99&companysize=99&ord_field=0&dibiaoid=0&line=&welfare=
#https://search.51job.com/list/010000,000000,0100,00,9,99,%25E6%2595%25B0%25E6%258D%25AE%25E6%258C%2596%25E6%258E%2598,2,1.html?lang=c&postchannel=0000&workyear=99&cotype=99&degreefrom=99&jobterm=99&companysize=99&ord_field=0&dibiaoid=0&line=&welfare=
#010000-330000: 代表不同的省份，其中000000： 代表全国  ；%25E6%2595%25B0%25E6%258D%25AE%25E6%258C%2596%25E6%258E%2598 对应关键词编码； ,2: 代表页数

# common imports
import requests
from lxml import etree
import time
import random
import pymongo
from retrying import retry
from scrapy import Selector 
from lxml import etree
import urllib.request
import urllib.parse
import xlwt

# ---------------------
# 连接到MongoDB
MONGO_URL = 'localhost'
MONGO_DB = 'Graduation_project'
MONGO_COLLECTION = 'shanghai_discovery'
client = pymongo.MongoClient(MONGO_URL, port=27017)
db = client[MONGO_DB]
# 页面获取函数
def get_page(page, keyword):
    header = {    
        'user-agent': "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/86.0.4240.111 Safari/537.36"
    }
    print('正在爬取第', page, '页')
    url = 'https://www.zhipin.com/c101020100/?query={k}&page={page}&ka=page-{page}'.format(page=page, k=keyword)
    response = requests.get(url, headers=header)
    return response.decode("utf-8")


# --------------
@retry(wait_fixed=8000)
def job_detail(link):
    header = {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) '
                      'Chrome/69.0.3497.12 Safari/537.36 '
    }
    response = requests.get(link, headers=header)
    data = etree.HTML(response.text)

    # ---检验是否出现验证码
    tips = data.xpath('/html/head/title/text()')
    tips_title = 'BOSS直聘验证码'
    if tips[0] == tips_title:
        print('检查是否弹出验证码')
        # 弹出验证码则引发IOError来进行循环
        raise IOError
    # ----------------------
    job_desc = data.xpath('//*[@id="main"]/div[3]/div/div[2]/div[3]/div[@class="job-sec"][1]/div/text()')

    jd = "".join(job_desc).strip()
    return jd


def parse_page(html, keyword, page):
    # 观察数据结构可得
    data = etree.HTML(html)
    items = data.xpath('//*[@id="main"]/div/div[2]/ul/li')
    for item in items:
        district = item.xpath('./div/div[1]/p/text()[1]')[0]
        job_links = item.xpath('./div/div[1]/h3/a/@href')[0]
        job_title = item.xpath('./div/div[1]/h3/a/div[1]/text()')[0]
        job_salary = item.xpath('./div/div[1]/h3/a/span/text()')[0]
        job_company = item.xpath('./div/div[2]/div/h3/a/text()')[0]
        job_experience = item.xpath('./div/div[1]/p/text()[2]')[0]
        job_degree = item.xpath('./div/div[1]/p/text()[3]')[0]
        fin_status = item.xpath('./div/div[2]/div/p/text()[2]')[0]
        try:
            company_scale = item.xpath('./div/div[2]/div/p/text()[3]')[0]
        except Exception:
            company_scale = item.xpath('./div/div[2]/div/p/text()[2]')[0]
        job_link = host + job_links
        print(job_link)
        # 获取职位描述
        detail = job_detail(job_link)
        # ---------------
        job = {
            'Keyword': keyword,
            '地区': district,
            '职位名称': job_title,
            '职位薪资': job_salary,
            '公司名称': job_company,
            '工作经验': job_experience,
            '学历要求': job_degree,
            '公司规模': company_scale,
            '融资情况': fin_status,
            '职位描述': detail,
        }
        print(job)
        save_to_mongo(job)
        time.sleep(random.randint(6, 9))
        # ---------------------------------------





def save_to_mongo(data):
    # 保存到MongoDB中
    try:
        if db[MONGO_COLLECTION].insert(data):
            print('存储到 MongoDB 成功')
    except Exception:
        print('存储到 MongoDB 失败')

#  header 通过浏览器检查， 网络， 点击对应的name在headers 属性下面；  选择某一个属性后可以通过 鼠标右击copy 复制xpath 路径。
if __name__ == '__main__':
    #url = r'https://list.jd.com/list.html?cat=670%2C671%2C2694&ev=exbrand_%E5%BE%AE%E8%BD%AF%EF%BC%88Microsoft%EF%BC%89%5E1107_90246%5E244_116227%5E3753_76033%5E&cid3=2694'
    MAX_PAGE = 10
    host = 'https://www.zhipin.com'
    keywords = ['数据分析', '数据挖掘', '商业分析', '机器学习']
    for keyword in keywords:
        for i in range(1, MAX_PAGE + 1):
            html = get_page(i, keyword)
            # ------------ 解析数据 ---------------
            parse_page(html, keyword, i)
            print('-' * 100)
            # -----------------
            timewait = random.randint(15, 18)
            time.sleep(timewait)
            print('等待', timewait, '秒')

pip install scrapy
# projectname为项目名称，可自定义
scrapy startproject projectname
#文件结构
projectname/
    scrapy.cfg            # 配置文件
    projectname/             # 爬虫模块文件夹
        __init__.py
        items.py          # items定义文件,设置数据存储模板，用于结构化数据，如：Django的Model
        middlewares.py    # 中间件middlewares文件
        pipelines.py      # 项目管道pipelines文件,数据处理行为，如：一般结构化的数据持久化
        settings.py       # 设置settings文件
        spiders/          # 爬虫文件夹 爬虫目录，如：创建文件，编写爬虫规则
            __init__.py

# -*- coding: utf-8 -*-
import scrapy
'''
    Spiders are classes that you define and that Scrapy uses to scrape information from a website (or a group of websites). They must subclass Spider and define the initial requests to make, optionally how to follow links in the pages, and how to parse the downloaded page content to extract data.
    start_requests(): must return an iterable of Requests (you can return a list of requests or write a generator function) which the Spider will begin to crawl from. Subsequent requests will be generated successively from these initial requests.
    parse(): a method that will be called to handle the response downloaded for each of the requests made. The response parameter is an instance of TextResponse that holds the page content and has further helpful methods to handle it.
    The parse() method usually parses the response, extracting the scraped data as dicts and also finding new URLs to follow and creating new requests (Request) from them.
'''
# 注意一定要继承 scrapy.Spider 类

class ToScrapeSpiderXPath(scrapy.Spider):
    name = 'toscrape-xpath'
    start_urls = [
        'http://quotes.toscrape.com/',
    ]

    def parse(self, response):
        for quote in response.xpath('//div[@class="quote"]'):
            yield {
                'text': quote.xpath('./span[@class="text"]/text()').extract_first(),
                'author': quote.xpath('.//small[@class="author"]/text()').extract_first(),
                'tags': quote.xpath('.//div[@class="tags"]/a[@class="tag"]/text()').extract()
            }

        next_page_url = response.xpath('//li[@class="next"]/a/@href').extract_first()
        if next_page_url is not None:
            yield scrapy.Request(response.urljoin(next_page_url))

class QuotesSpider(scrapy.Spider):
    name = "quotes"

    def start_requests(self):
        urls = [
            'http://quotes.toscrape.com/page/1/',
            'http://quotes.toscrape.com/page/2/',
        ]
        for url in urls:
            yield scrapy.Request(url=url, callback=self.parse)

    def parse(self, response):
        page = response.url.split("/")[-2]
        filename = f'quotes-{page}.html'
        with open(filename, 'wb') as f:
            f.write(response.body)
        self.log(f'Saved file {filename}')
        
class AuthorSpider(scrapy.Spider):
    name = 'author'

    start_urls = ['http://quotes.toscrape.com/']

    def parse(self, response):
        author_page_links = response.css('.author + a')
        yield from response.follow_all(author_page_links, self.parse_author)

        pagination_links = response.css('li.next a')
        yield from response.follow_all(pagination_links, self.parse)

    def parse_author(self, response):
        def extract_with_css(query):
            return response.css(query).get(default='').strip()

        yield {
            'name': extract_with_css('h3.author-title::text'),
            'birthdate': extract_with_css('.author-born-date::text'),
            'bio': extract_with_css('.author-description::text'),
        }

#item 类型
from scrapy.item import Item, Field
class CustomItem(Item):
    one_field = Field()
    another_field = Field()
# database 类型
from dataclasses import dataclass
@dataclass
class CustomItem:
    one_field: str
    another_field: int
#class 类型
import scrapy
class Product(scrapy.Item):
    name = scrapy.Field()
    price = scrapy.Field()
    stock = scrapy.Field()
    tags = scrapy.Field()
    last_updated = scrapy.Field(serializer=str)

from scrapy.loader import ItemLoader
from myproject.items import Product

def parse(self, response):
    l = ItemLoader(item=Product(), response=response)
    l.add_xpath('name', '//div[@class="product_name"]')
    l.add_xpath('name', '//div[@class="product_title"]')
    l.add_xpath('price', '//p[@id="price"]')
    l.add_css('stock', 'p#stock]')
    l.add_value('last_updated', 'today') # you can also use literal values
    return l.load_item()
#when all data is collected, the ItemLoader.load_item() method is called which actually returns the item populated with the data previously extracted and collected with the add_xpath(), add_css(), and add_value() calls.

#vaidation
from itemadapter import ItemAdapter
from scrapy.exceptions import DropItem
class PricePipeline:
    vat_factor = 1.15
    def process_item(self, item, spider):
        adapter = ItemAdapter(item)
        if adapter.get('price'):
            if adapter.get('price_excludes_vat'):
                adapter['price'] = adapter['price'] * self.vat_factor
            return item
        else:
            raise DropItem(f"Missing price in {item}")
#write to json
import json
from itemadapter import ItemAdapter
class JsonWriterPipeline:
    def open_spider(self, spider):
        self.file = open('items.jl', 'w')
    def close_spider(self, spider):
        self.file.close()
    def process_item(self, item, spider):
        line = json.dumps(ItemAdapter(item).asdict()) + "\n"
        self.file.write(line)
        return item
#write to Mongodb
import pymongo
from itemadapter import ItemAdapter
class MongoPipeline:
    collection_name = 'scrapy_items'
    def __init__(self, mongo_uri, mongo_db):
        self.mongo_uri = mongo_uri
        self.mongo_db = mongo_db
    @classmethod
    def from_crawler(cls, crawler):
        return cls(
            mongo_uri=crawler.settings.get('MONGO_URI'),
            mongo_db=crawler.settings.get('MONGO_DATABASE', 'items')
        )
    def open_spider(self, spider):
        self.client = pymongo.MongoClient(self.mongo_uri)
        self.db = self.client[self.mongo_db]
    def close_spider(self, spider):
        self.client.close()
    def process_item(self, item, spider):
        self.db[self.collection_name].insert_one(ItemAdapter(item).asdict())
        return item
#duplicate
from itemadapter import ItemAdapter
from scrapy.exceptions import DropItem
class DuplicatesPipeline:
    def __init__(self):
        self.ids_seen = set()
    def process_item(self, item, spider):
        adapter = ItemAdapter(item)
        if adapter['id'] in self.ids_seen:
            raise DropItem(f"Duplicate item found: {item!r}")
        else:
            self.ids_seen.add(adapter['id'])
            return item