TheOne的个人博客

def parse(self, response: HtmlResponse, **kwargs):
    a_list = response.xpath("//div[@class='rank-list']/a")
    for a_temp in a_list:
        rank_number = a_temp.xpath('./div[@class="badge"]/text()').extract_first()  # 排名
        img_url = a_temp.xpath('./img/@src').extract_first()  # 图片地址
        title = a_temp.xpath('./div[@class="content"]/div[@class="title"]/text()').extract_first()  # 标题
        desc = a_temp.xpath('./div[@class="content"]/div[@class="desc"]/text()').extract_first()  # 描述
        play_number = a_temp.xpath('.//div[@class="info-item"][1]/span/text()').extract_first()

        # 需要使用yield关键字将解析好的数据交给管道模块
        yield {
            'type': 'info',
            'rank_number': rank_number,
            'img_url': img_url,
            'title': title,
            'desc': desc,
            'play_number': play_number
        }

def parse(self, response: HtmlResponse, **kwargs):
    a_list = response.xpath("//div[@class='rank-list']/a")
    for a_temp in a_list:
        rank_number = a_temp.xpath('./div[@class="badge"]/text()').extract_first()  # 排名
        img_url = a_temp.xpath('./img/@src').extract_first()  # 图片地址
        title = a_temp.xpath('./div[@class="content"]/div[@class="title"]/text()').extract_first()  # 标题
        desc = a_temp.xpath('./div[@class="content"]/div[@class="desc"]/text()').extract_first()  # 描述
        play_number = a_temp.xpath('.//div[@class="info-item"][1]/span/text()').extract_first()

        # 需要使用yield关键字将解析好的数据交给管道模块
        yield {
            'rank_number': rank_number,
            'img_url': img_url,
            'title': title,
            'desc': desc,
            'play_number': play_number
        }

        # 自行构造新的request请求并交给下载器下载
        """
        callback: 指定解析函数
        cb_kwargs: 如果解析方法中存在形参，则可以通过cb_kwargs传递，传递参数的类型必须是字典，字典中的key必须与解析方法中的形参名称保持一致
        """
        yield scrapy.Request(img_url, callback=self.parse_imgage, cb_kwargs={'image_name': title})

def parse_imgage(self, response, image_name):
    yield {
        'image_name': image_name.replace('|', '') + '.png',
        'image_content': response.body
    }

class Top250Spider(scrapy.Spider):
    """
    如果使用spider类默认的请求方式，则不会对重复请求进行过滤
        会重复请求相同的url
    """
    name = "top250"
    allowed_domains = ["movie.douban.com"]
    start_urls = ["http://movie.douban.com/top250", "http://movie.douban.com/top250"]

    # 需要实现对同一个url去重操作，必须重写start_requests方法
    # 如果请求的地址是api接口，则需要通过start_request方法构造链接
    def start_requests(self):
        url = 'https://movie.douban.com/top250?start={}&filter='
        for page in range(10):
            yield scrapy.Request(url.format(page * 25), dont_filter=False)

class JcInfoSpider(scrapy.Spider):
    name = "jc_info"

    def start_requests(self):
        url = 'http://www.xxx.com.cn/new/disclosure'
        for page in range(1, 11):
            data = {
                "column": "szse_latest",
                "pageNum": str(page),
                "pageSize": "30",
                "sortName": "",
                "sortType": "",
                "clusterFlag": "true"
            }
            yield scrapy.FormRequest(url=url, formdata=data)

from scrapy.http import JsonRequest

class NetEaseInfoSpider(scrapy.Spider):
    name = "netease_info"

    def start_requests(self):
        url = 'https://xxx.com/api/hr163/position/queryPage'
        for page in range(1, 10):
            payload = {
                "currentPage": page,
                "pageSize": 10
            }
            yield JsonRequest(url, data=payload)

class UserAgentDownloaderMiddleware:
    USER_AGENTS_LIST = [
        "Mozilla/5.0 (compatible; MSIE 9.0; Windows NT 6.1; Win64; x64; Trident/5.0; .NET CLR 3.5.30729; .NET CLR 3.0.30729; .NET CLR 2.0.50727; Media Center PC 6.0)",
        "Mozilla/5.0 (compatible; MSIE 8.0; Windows NT 6.0; Trident/4.0; WOW64; Trident/4.0; SLCC2; .NET CLR 2.0.50727; .NET CLR 3.5.30729; .NET CLR 3.0.30729; .NET CLR 1.0.3705; .NET CLR 1.1.4322)",
        "Mozilla/4.0 (compatible; MSIE 7.0b; Windows NT 5.2; .NET CLR 1.1.4322; .NET CLR 2.0.50727; InfoPath.2; .NET CLR 3.0.04506.30)",
        "Mozilla/5.0 (Windows; U; Windows NT 5.1; zh-CN) AppleWebKit/523.15 (KHTML, like Gecko, Safari/419.3) Arora/0.3 (Change: 287 c9dfb30)",
        "Mozilla/5.0 (X11; U; Linux; en-US) AppleWebKit/527+ (KHTML, like Gecko, Safari/419.3) Arora/0.6",
        "Mozilla/5.0 (Windows; U; Windows NT 5.1; en-US; rv:1.8.1.2pre) Gecko/20070215 K-Ninja/2.1.1",
        "Mozilla/5.0 (Windows; U; Windows NT 5.1; zh-CN; rv:1.9) Gecko/20080705 Firefox/3.0 Kapiko/3.0",
        "Mozilla/5.0 (X11; Linux i686; U;) Gecko/20070322 Kazehakase/0.4.5"
    ]

    def process_request(self, request, spider):
        print('下载中间件的process_request方法被调用...')
        user_agent = random.choice(self.USER_AGENTS_LIST)
        request.headers['User-Agent'] = user_agent

        """
        如果返回None, 表示当前的request提交下一个权重低的process_request。
        如果传递到最后一个process_request,则传递给下载器进行下载。

        如果返回的是response对象则不再请求，把response对象直接交给引擎
        """

class ProxyDownloaderMiddleware:
    def process_request(self, request, spider):
        print('免费代理中间件 - 代理设置...')

        # 元数据
        request.meta['proxy'] = 'http://117.0.0.1:7890'
        print(request.meta['proxy'])

    def process_response(self, request, response, spider):
        print('免费中间件 - 代理检测...')
        if response.status != 200:
            request.dont_filter = True  # 关闭过滤
            return request  # 把请求失败的请求对象重新提交给调度器
        return response

class TxWorkFilePipeline:
    # 在爬虫启动时创建一个文件对象
    def open_spider(self, spider):  # spider: 接收的是spider类的一个实例
        if spider.name == 'tx_work_info':
            self.file_obj = open('tx_work_info.txt', 'a', encoding='utf-8')

    def process_item(self, item, spider):
        """
        在当前方法中可以对item进行数据判断，如果不符合数据要求，一般有两种方式来处理:
            1. 扔掉
                抛出一个异常中断当前管道，阻止item通过return传递给下一个管道
            2. 修复
                在当前方法中编辑修复代码逻辑并使用return将修复的数据传递给下一个item

            注意:
                当前方法如果存在return item则将item数据传递给下一个item
                如果return不存在则将None传递给下一个item
        """
        # raise DropItem  # 丢掉不符合要求的数据
        if spider.name == 'tx_work_info':
            self.file_obj.write(json.dumps(item, ensure_ascii=False, indent=4) + '\n')
            print('数据写入成功:', item)
        return item

    def close_spider(self, spider):
        if spider.name == 'tx_work_info':
            self.file_obj.close()


class TxWorkMongoPipeline:
    # 在爬虫启动时创建一个数据库链接对象对象
    def open_spider(self, spider):  # spider: 接收的是spider类的一个实例
        if spider.name == 'tx_work_info':
            self.mongo_client = pymongo.MongoClient(host='192.168.198.130')
            self.collection = self.mongo_client['py_spider']['tx_work_info']

    def process_item(self, item, spider):
        if spider.name == 'tx_work_info':
            self.collection.insert_one(item)
            print('数据写入mongodb成功:', item)
        return item

    def close_spider(self, spider):
        if spider.name == 'tx_work_info':
            self.mongo_client.close()

ROBOTSTXT_OBEY = True

DOWNLOAD_DELAY = 3

DEFAULT_REQUEST_HEADERS = {
   "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
   "Accept-Language": "en",
}

SPIDER_MIDDLEWARES = {
    "TxWork.middlewares.SeleniumDownloaderMiddleware": 543,
}

DOWNLOADER_MIDDLEWARES = {
    "TxWork.middlewares.SeleniumDownloaderMiddleware": 543,
}

ITEM_PIPELINES = {
    "TxWork.pipelines.TxWorkFilePipeline": 300,
    "TxWork.pipelines.TxWorkMongoPipeline": 301
}

EXTENSIONS = {
   "scrapy.extensions.telnet.TelnetConsole": None,
}

class BookItem(scrapy.Item):
    """
    items文件主要用于字段校验
    """
    title = scrapy.Field()
    price = scrapy.Field()
    author = scrapy.Field()
    date_data = scrapy.Field()
    detail = scrapy.Field()
    producer = scrapy.Field()

class BookInfoSpider(scrapy.Spider):
    name = "book_info"
    start_urls = ["http://xxx.com/"]

    def parse(self, response: HtmlResponse, **kwargs):
        li_list = response.xpath('//ul[@class="bigimg"]/li')
        for li in li_list:
            item = BookItem()
            item['title'] = li.xpath('./a/@title').extract_first()
            item['price'] = li.xpath('./p[@class="price"]/span[1]/text()').extract_first()
            item['author'] = li.xpath('./p[@class="search_book_author"]/span[1]/a[1]/@title').extract_first()
            item['date_data'] = li.xpath('./p[@class="search_book_author"]/span[last()-1]/text()').extract_first()
            item['detail'] = li.xpath('./p[@class="detail"]/text()').extract_first() if li.xpath(
                './p[@class="detail"]/text()') else '空'
            item['producer'] = li.xpath(
                './p[@class="search_book_author"]/span[last()]/a/text()').extract_first() if li.xpath(
                './p[@class="search_book_author"]/span[last()]/a/text()') else '空'

            yield item

def say_hello():
    print("hello!")


def say_goodbye():
    print("hello!")  # 此处应打印goodbye


if __name__ == '__main__':
    say_hello()
    say_goodbye()

def say_hello():
    print("[DEBUG]: enter say_hello()")
    print("hello!") 
 
def say_goodbye():
    print("[DEBUG]: enter say_goodbye()")
    print("hello!")
 
if __name__ == '__main__':
    say_hello()
    say_goodbye()

def debug(func):
    def wrapper():
        print("[DEBUG]: enter {}()".format(func.__name__))
        return func()
    return wrapper


def say_hello():
    print("hello!")


say_hello = debug(say_hello)

say_hello()

def debug(func):
    def wrapper():
        print("[DEBUG]: enter {}()".format(func.__name__))
        return func()

    return wrapper

@debug
def say_hello():
    print("hello!")


say_hello()

def debug(func):
    def wrapper(something):
        print("[DEBUG]: enter {}()".format(func.__name__))
        return func(something)

    return wrapper

@debug
def say_hello(something):
    print("hello {}!".format(something))


say_hello('theone')

def debug(func):
    def wrapper(*args, **kwargs):
        print("[DEBUG]: enter {}()".format(func.__name__))
        return func(*args, **kwargs)
    return wrapper


@debug
def say(something):
    print("hello {}!".format(something))


say('theone')

def logging(level):
    def wrapper(func):
        def inner_wrapper(*args, **kwargs):
            print("[{level}]: enter function {func}()".format(
                level=level,
                func=func.__name__))
            return func(*args, **kwargs)
        return inner_wrapper
    return wrapper


@logging(level='INFO')
def say(something):
    print("say {}!".format(something))


# 如果没有使用@语法，等同于
# say = logging(level='INFO')(say)

@logging(level='DEBUG')
def do(something):
    print("do {}...".format(something))


if __name__ == '__main__':
    say('hello')
    do("my work")

class Demo:
    def __call__(*args, **kwargs):
        print('实例方法被执行')

demo = Demo()
demo()  # 实例方法被执行

class Logging:
    def __init__(self, func):
        self.func = func

    def __call__(self, *args, **kwargs):
        print("[DEBUG]: enter function {func}()".format(
            func=self.func.__name__))
        return self.func(*args, **kwargs)


@Logging
def say(something):
    print("say {}!".format(something))


say('木木')

class Logging:
    def __init__(self, level='INFO'):
        self.level = level

    def __call__(self, func):  # 接收函数
        def wrapper(*args, **kwargs):
            print("[{level}]: enter function {func}()".format(
                level=self.level,
                func=func.__name__))
            func(*args, **kwargs)

        return wrapper  # 返回函数


@Logging(level='INFO')
def say(something):
    print("say {}!".format(something))


say('木木')

def send_message(message):
    def print_message():
        print(message)  # 获取的是外部函数的形式参数

    # 在外部函数中返回内部函数的地址
    return print_message


# 创建一个变量去接收内部函数地址
func_obj = send_message('python天下第一...')
func_obj()

def create_calculator(operation):
    def add(x, y):
        return x + y

    def subtract(x, y):
        return x - y

    def multiply(x, y):
        return x * y

    def divide(x, y):
        if y != 0:
            return x / y
        else:
            return "Error: Division by zero!"

    if operation == "add":
        return add
    elif operation == "subtract":
        return subtract
    elif operation == "multiply":
        return multiply
    elif operation == "divide":
        return divide
    else:
        return None

# 创建加法器
add_calculator = create_calculator("add")
result = add_calculator(5, 3)
print(result)  # 输出 8

# 创建减法器
subtract_calculator = create_calculator("subtract")
result = subtract_calculator(10, 4)
print(result)  # 输出 6

# 创建乘法器
multiply_calculator = create_calculator("multiply")
result = multiply_calculator(7, 2)
print(result)  # 输出 14

# 创建除法器
divide_calculator = create_calculator("divide")
result = divide_calculator(8, 2)
print(result)  # 输出 4.0

def counter(start=0):
    def add():
        nonlocal start  # 针对函数参数
        start += 1
        return start

    return add


obj = counter(10)
print(obj())

"""
global: 作用于不可变对象且这个对象是一个全局的
nonlocal: 作用于不可变对象且这个对象具有作用域的
"""

def wrapper(name):
    def message(content):
        print(f'{name}: {content}')

    print(f'内部函数引用的地址为:', id(message))
    return message


func_1 = wrapper('安娜')
func_2 = wrapper('双双')

# 闭包多次运行返回的内部函数的地址是不一样的，所以闭包具有内存隔离的特性，与面向对象中的实例对象类似
print(id(func_1))  # 2610767972528
print(id(func_2))  # 2610767970944

def get_num(number):  # 生成器函数
    index = 0
    while index < number:
        yield index  # yield可以返回一个结果并且不会终止一个函数的运行
        index += 1


# 生成器函数不能直接使用小括号的方式获取结果
# 生成器函数调用后返回的是一个生成器对象
print(get_num(5).__next__())

# 生成器对象中的结果可以使用next函数进行获取，和迭代器类似
obj = get_num(5)
print(next(obj))
print(next(obj))
print(next(obj))

def get_num(number):
    print('开始迭代...')
    i = 0
    while i < number:
        print('迭代中...')
        yield i
        print('迭代结束')
        i += 1


obj = get_num(5)
print(next(obj))
print(next(obj))

"""
打印结果为:
开始迭代...
迭代中...
0
迭代结束
迭代中...
1
"""

nums = (i for i in range(1, 11))

def get_num(number):
    i = 0
    while i < number:
        data = yield i
        if data == '这是测试时传递的一个值':
            print(123)
        print('data变量的值:', data)
        i += 1


obj = get_num(5)
# print(next(obj))
print(obj.send(None))  # 第一次执行生成器时不能使用send方法去传递任何对象，除了None对象
print(obj.send('这是测试时传递的一个值'))  # 可以在运行生成器对象的过程中传递一个值

obj.close()  # 关闭生成器对象，后续代码无法运行生成器对象
print(next(obj))

"""
运行结果为:
0
123
data变量的值: 这是测试时传递的一个值
1
"""

from collections.abc import Iterable

nums = [1, 2, 3]

for num in nums:
    print(num)

print(isinstance(nums, Iterable))  # True

class A:
    pass


class B(A):
    pass


a = A()
b = B()

print(isinstance(a, A))  # True
print(isinstance(b, A))  # True
print(isinstance(a, B))  # False

from collections.abc import Iterator

nums = [1, 2, 3]
print(isinstance(nums, Iterator))  # False,列表对象不是迭代器对象
print(isinstance(iter(nums), Iterator))  # True

class ListNode:
    def __init__(self):
        self.person_list = list()

    # 如果一个类内部定义了__iter__方法，那么创建出来的实例一定是一个可迭代对象
    def __iter__(self):
        pass

    def __next__(self):
        pass


list_node = ListNode()
print(isinstance(list_node, Iterable))  # True
print(isinstance(list_node, Iterator))  # True

class ListNode:
    def __init__(self):
        self.index = 0
        self.stu_list = list()

    def __iter__(self):
        return self

    def __next__(self):
        if self.index < len(self.stu_list):
            item = self.stu_list[self.index]
            self.index += 1
            return item
        else:
            print('元素被取完了')
            raise StopIteration

    def add(self, name):
        self.stu_list.append(name)


list_node = ListNode()
list_node.add('安娜')
list_node.add('双双')

for stu in list_node:
    print(stu)