Python，爬虫与深度学习（6）——爬虫详解（二）构建自己的ip池（简易版）
步骤/目录： 1.需求分析 2.mysql建表 3.网页请求分析 4.爬虫编写 （0）安装所需模块 （1）mysql存取 （2）爬虫主体编写 （3）其他爬虫补充 （4）实际运行 （5）检测爬取的ip 5.总结与改进本文首发于个人博客https://lisper517.top/index.php/archives/44/，转载请注明出处。本文的目的是构建自己的ip池。本文写作日期为2022年9月4日。操作的平台为win10，编辑器为VS code。爬虫写的好，牢房进的早。使用爬虫时不要占用服务器太多流量，否则可能惹祸上身。除了python与爬虫，这里实际上还需要了解html格式与mysql命令。可以在 runoob-html 或 runoob-mysql 找到初步的学习资料，或者查看笔者其他的文章。最后，笔者的能力有限，也并非CS从业人员，很多地方难免有纰漏或者不符合代码原则的地方，请在评论中指出。1.需求分析之前说过，在对别人的网页服务器进行爬取时要注意不能占用太多服务器资源。但有时候，我们确实需要在短时间内对某个网页高频访问，又担心自己被封ip，那么此时维护一个ip池就非常有必要，我们需要对网络上的ip（每个ip背后是一个或一群服务器）进行测试，看是否能利用这些ip作为代理服务器访问其他网页。本文先介绍如何爬取ip，后续文章会介绍如何用自己的ip池爬取网页，以及代理失效时更换代理。在爬虫的一些较高级的应用中，一个ipool是非常基础且不可或缺的，所以建议学会scrapy后就应该尽快建立自己的ipool（除非没有高频访问某个网页的需求）。ip池的构建和使用分为两步，第一步是筛选ip。世界上的公网ipv4地址最多有 4,294,967,296 个（256^4。实际上有一些被局域网占用，有一些不会使用），要一个个去验证这些ip耗时太长，何况还需要对已存在的ip进行验证、剔除失效ip。所以，这里使用一些提供免费代理ip的网页缩小ip搜索范围，爬取到的代理ip不做验证，都放入库中（这里使用mysql）。第二步是对mysql中已有的ip进行验证，挑选出对某个网址有效的代理ip。这一步严格来讲和第一步关系不大，它可以在前一个步骤进行时也同步进行，更重要的是这一步是在其他爬虫开始后才进行，下面就讲解一下免费代理ip的使用。免费代理可分为HTTP或HTTPS，高匿名和低匿名。如果爬虫运行时经过了这些免费代理，毕竟还是别人的服务器，你的请求和响应是到了别人那里，所以用代理服务器时不要传递一些私密信息。实在有必要的话，自己买服务器做转发。最后，同一个代理ip，可能你用它代理一个网站可以，代理另一个网站就不行，这也是不确定因素，所以爬到的代理ip在使用时还要对特定的网址进行测试，其实也是比较麻烦的，如果有紧急需求，建议买一些付费的代理ip。2.mysql建表在mysql中存储收集到的ip，需要新建以下列：ip，字符串形式，VARCHAR(50)（为了将来兼容ipv6，稍微长一点）；端口号，数值形式，SMALLINT UNSIGNED（刚好是0~65535）；匿名度，2个bit位，BIT(2)（0表示低匿名，1表示高匿名，3表示未知）；类型（HTTP还是HTTPS），2个bit位，BIT(2)（0表示HTTP，1表示HTTPS，2表示都可，3表示其他）；ip物理位置，字符串，VARCHAR(100)；响应速度，MEDIUMINT，单位是毫秒ms；最后一次验证的时间，日期时间格式。建表的语句在程序中将给出。其实对这种不断更新、添加的数据，redis更加合适，有关redis的内容将在后续的文章中介绍。3.网页请求分析这里使用的提供代理ip的网站为（笔者给出的例子中，只使用了第一个网址）：http://www.ip3366.net/free/ https://www.kuaidaili.com https://ip.jiangxianli.com/ https://www.cool-proxy.net/ https://free-proxy-list.net https://www.ipaddress.com/proxy-list/ http://proxy-list.org/english/ https://www.proxynova.com/proxy-server-list/ http://pubproxy.com/api/proxy?limit=5&format=txt&type=http&level=anonymous&last_check=60&no_country=CN https://www.rmccurdy.com/.scripts/proxy/good.txt http://spys.one/en/anonymous-proxy-list/ https://raw.githubusercontent.com/sunny9577/proxy-scraper/master/proxies.json https://raw.githubusercontent.com/a2u/free-proxy-list/master/free-proxy-list.txt https://raw.githubusercontent.com/clarketm/proxy-list/master/proxy-list.txt https://raw.githubusercontent.com/TheSpeedX/SOCKS-List/master/http.txt这些网页的请求响应都挺简单的，这里就不专门介绍了。由于涉及多个网站，这里将使用多线程的方法，持续对各个ip网站爬取。唯一需要注意的是对每个网站，爬取ip的速度不要太快了，虽然对单个网站爬取速度不快，但毕竟有这么多ip网站，实际上ip入库的速度挺快的。4.爬虫编写（0）安装所需模块需要用到requests（处理网络请求）、bs4（对响应内容进行解析）、pymysql（连接mysql数据库）、lxml（网页解析器）、cryptography（连接mysql时密码加密）库，在cmd中使用 pip install requests bs4 pymysql lxml cryptography 下载（下载失败就一直使用上述命令，直到下载成功）。之后新建一个文件夹，用VS Code打开该文件夹、新建一个py文件，进行下面的操作。（1）mysql存取使用mysql存储ip，其他爬虫需要通过这个类存取ip。自己使用时，记得把mysql连接的地址、端口、密码更改一下。#!/usr/bin/python # -*- coding: UTF-8 -*- import pymysql, time, os, logging from traceback import print_exc if not os.path.isdir(os.path.join(os.getcwd(), 'logs')): os.mkdir(os.path.join(os.getcwd(), 'logs')) class MysqlOperationFailed(Exception): """若更改mysql出现错误则抛出这个异常。""" def __init__(self): message = "Failed to change data in MySQL, please check. " super().__init__(message) class mysql_api(): """通过这个界面实现提交爬取的ip、取出ip的操作。""" conn = pymysql.connect(host='127.0.0.1', port=53306, user='root', passwd='mysqlpasswd', db='mysql', charset='utf8') cur = conn.cursor() def __init__(self, log_name: str): """初始函数，需要提交一个log_name作为进程的log文件名。""" self.cur = mysql_api.conn.cursor() mysql_api.check_table() self.logging = logging.basicConfig( level=logging.INFO, filename=os.path.join( os.getcwd(), 'logs', '{}-log{}.txt'.format(log_name, time.strftime("%Y-%m-%d-%H_%M_%S")))) def __del__(self): self.cur.close() @staticmethod def check_table(): """检查一下mysql中是否已经建好了表，若表不存在则新建表。这里并没有检查：若表存在，表的格式是否正确。""" cur, conn = mysql_api.cur, mysql_api.conn command = ''' CREATE TABLE IF NOT EXISTS ipool ( no BIGINT UNSIGNED AUTO_INCREMENT, ip VARCHAR(50) NOT NULL DEFAULT "0.0.0.0" COMMENT "IP address", port SMALLINT UNSIGNED NOT NULL DEFAULT 0 COMMENT "port", anonymous BIT(2) NOT NULL DEFAULT 3 COMMENT "whether ip is anonymous", type BIT(2) NOT NULL DEFAULT 3 COMMENT "HTTP or HTTPS or both or others", physical_address VARCHAR(100) NOT NULL DEFAULT "unknown" COMMENT "where is the server", response_time MEDIUMINT NOT NULL DEFAULT -1 COMMENT "response_time in microseconds", last_verify DATETIME NOT NULL DEFAULT '1000-01-01 00:00:00' COMMENT "last verify time", created TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, PRIMARY KEY (no));''' command = command.replace('\n', '') command = command.replace(' ', '') try: cur.execute("CREATE DATABASE IF NOT EXISTS ipool;") cur.execute("USE ipool;") cur.execute(command) conn.commit() except: conn.rollback() raise MysqlOperationFailed def save_ips(self, ips_list: list): """存入多个ip（以字典的列表形式提交）。注意这里只是对不存在的数据添加默认值，没有验证数据类型等。 另外注意，响应时间单位为ms。""" cur, conn = self.cur, mysql_api.conn command = '''INSERT INTO ipool (ip,port,anonymous,type,physical_address,response_time,last_verify) VALUES ''' for ip_dict in ips_list: ip, port, anonymous, type_, physical_address, response_time, last_verify = "0.0.0.0", 0, 3, 3, 'unknown', -1, "1000-01-01 00:00:00" if 'ip' in ip_dict.keys(): ip = ip_dict['ip'] if 'port' in ip_dict.keys(): port = ip_dict['port'] if 'anonymous' in ip_dict.keys(): anonymous = ip_dict['anonymous'] if 'type' in ip_dict.keys(): type_ = ip_dict['type'] if 'physical_address' in ip_dict.keys(): physical_address = ip_dict['physical_address'] if 'response_time' in ip_dict.keys(): response_time = ip_dict['response_time'] if 'last_verify' in ip_dict.keys(): last_verify = ip_dict['last_verify'] command += '("{}",{},{},{},"{}",{},"{}"),'.format( ip, port, anonymous, type_, physical_address, response_time, last_verify) command = command[:-1] + ';' try: cur.execute(command) conn.commit() except: conn.rollback() raise MysqlOperationFailed @staticmethod def check_repeat(): """检查爬到的数据里有没有重复的，只检查ip和port都相同的。如果有重复的，保留最新的一条记录，也就是no最大的。""" cur, conn = mysql_api.cur, mysql_api.conn cur.execute('USE ipool;') cur.execute( 'SELECT ip,port FROM ipool GROUP BY ip,port HAVING COUNT(*)>1;') repeated_items = cur.fetchall() if len(repeated_items) == 0: print('\n数据库中无重复数据。') return else: print('\n数据库中检测到重复数据，有{}种。'.format(len(repeated_items))) command = ''' DELETE FROM ipool WHERE no IN (SELECT no FROM (SELECT no FROM ipool WHERE (ip,port) IN (SELECT ip,port FROM ipool GROUP BY ip,port HAVING COUNT(*)>1) AND no NOT IN (SELECT MAX(no) FROM ipool GROUP BY ip,port HAVING COUNT(*)>1)) AS a);''' command = command.replace('\n', '') command = command.replace(' ', '') try: cur.execute(command) conn.commit() except: conn.rollback() raise MysqlOperationFailed @staticmethod def get_ips(anonymous: list = [1], type_: list = [2], response_time: int = 30000, last_verify_interval: str = '48:00:00', limit: int = 1000) -> tuple: """根据要求返回多个ip，只返回ip和port。 :param last_verify_interval 用于指定上次验证的时间距现在不超过多久。""" cur, conn = mysql_api.cur, mysql_api.conn cur.execute('USE ipool;') anonymous, type_ = tuple(anonymous), tuple(type_) command = ''' SELECT ip,port FROM ipool WHERE anonymous IN {} AND type IN {} AND response_time BETWEEN 0 AND {} AND (NOW()-last_verify)<"{}" ORDER BY response_time LIMIT {}; '''.format(anonymous, type_, response_time, last_verify_interval, limit) command = command.replace('\n', '') command = command.replace(' ', '') print(command) cur.execute(command) return cur.fetchall() if __name__ == "__main__": try: a = mysql_api('kuaidaili') ip_list = [] for i in range(1, 256): ip_dict = { 'ip': '192.168.1.{}'.format(i), 'port': i, 'response_time': i * 10, 'last_verify': '2022-09-05 12:00:00' } ip_list.append(ip_dict) a.save_ips(ip_list) a.check_repeat() print(a.get_ips([0, 1, 2, 3], [0, 1, 2, 3], 30000)) except: print_exc()（2）爬虫主体编写这里稍微把前面mysql存取的部分进行了更改，一并写出。单个ip网站的爬虫使用了递归调用的方法，不太适合页码比较多的情况。这里只是写了对 http://www.ip3366.net/free/ 的爬虫、并用线程的方法开启它，其他爬虫可在后面自行补充。最后，自己使用时，记得把mysql连接的地址、端口、密码更改一下。#!/usr/bin/python # -*- coding: UTF-8 -*- import pymysql, os, time, logging from traceback import print_exc if not os.path.isdir(os.path.join(os.getcwd(), 'logs')): os.mkdir(os.path.join(os.getcwd(), 'logs')) logging.basicConfig(level=logging.INFO, filename=os.path.join( os.getcwd(), 'logs', 'log-{}.txt'.format( time.strftime("%Y-%m-%d-%H_%M_%S")))) class MysqlOperationFailed(Exception): """若更改mysql出现错误则抛出这个异常。""" def __init__(self): message = "Failed to change data in MySQL, please check. " super().__init__(message) class mysql_api(): """通过这个界面实现提交爬取的ip、取出ip的操作。""" conn = pymysql.connect(host='127.0.0.1', port=53306, user='root', passwd='mysqlpasswd', db='mysql', charset='utf8') cur = conn.cursor() def __init__(self): self.cur = mysql_api.conn.cursor() mysql_api.check_table() message = 'one of mysql_api object started' logging.info(message) def __del__(self): self.cur.close() message = 'one of mysql_api object stopped' logging.info(message) @staticmethod def check_table(): """检查一下mysql中是否已经建好了表，若表不存在则新建表。这里并没有检查：若表存在，表的格式是否正确。""" cur, conn = mysql_api.cur, mysql_api.conn command = ''' CREATE TABLE IF NOT EXISTS ipool ( no BIGINT UNSIGNED AUTO_INCREMENT, ip VARCHAR(50) NOT NULL DEFAULT "0.0.0.0" COMMENT "IP address", port SMALLINT UNSIGNED NOT NULL DEFAULT 0 COMMENT "port", anonymous BIT(2) NOT NULL DEFAULT 3 COMMENT "whether ip is anonymous", type BIT(2) NOT NULL DEFAULT 3 COMMENT "HTTP or HTTPS or both or others", physical_address VARCHAR(100) NOT NULL DEFAULT "unknown" COMMENT "where is the server", response_time MEDIUMINT NOT NULL DEFAULT -1 COMMENT "response_time in microseconds", last_verify DATETIME NOT NULL DEFAULT '1000-01-01 00:00:00' COMMENT "last verify time", created TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, PRIMARY KEY (no));''' command = command.replace('\n', '') command = command.replace(' ', '') try: cur.execute("CREATE DATABASE IF NOT EXISTS ipool;") cur.execute("USE ipool;") cur.execute(command) conn.commit() message = 'database ipool and table ipool checked' logging.info(message) except: conn.rollback() raise MysqlOperationFailed def save_ips(self, ips_list: list): """存入多个ip（以字典的列表形式提交）。注意这里只是对不存在的数据添加默认值，没有验证数据类型等。 另外注意，响应时间单位为ms。""" cur, conn = self.cur, mysql_api.conn command = '''INSERT INTO ipool (ip,port,anonymous,type,physical_address,response_time,last_verify) VALUES ''' for ip_dict in ips_list: ip, port, anonymous, type_, physical_address, response_time, last_verify = "0.0.0.0", 0, 3, 3, 'unknown', -1, "1000-01-01 00:00:00" if 'ip' in ip_dict.keys(): ip = ip_dict['ip'] if 'port' in ip_dict.keys(): port = ip_dict['port'] if 'anonymous' in ip_dict.keys(): anonymous = ip_dict['anonymous'] if 'type' in ip_dict.keys(): type_ = ip_dict['type'] if 'physical_address' in ip_dict.keys(): physical_address = ip_dict['physical_address'] if 'response_time' in ip_dict.keys(): response_time = ip_dict['response_time'] if 'last_verify' in ip_dict.keys(): last_verify = ip_dict['last_verify'] command += '("{}",{},{},{},"{}",{},"{}"),'.format( ip, port, anonymous, type_, physical_address, response_time, last_verify) message = 'trying to save ip:port={}:{}'.format(ip, port) logging.info(message) command = command[:-1] + ';' try: cur.execute(command) conn.commit() message = 'saving ips successfully' logging.info(message) except: conn.rollback() raise MysqlOperationFailed def check_repeat(self): """检查爬到的数据里有没有重复的，只检查ip和port都相同的。如果有重复的，保留最新的一条记录，也就是no最大的。""" cur, conn = mysql_api.cur, mysql_api.conn cur.execute('USE ipool;') cur.execute( 'SELECT ip,port FROM ipool GROUP BY ip,port HAVING COUNT(*)>1;') repeated_items = cur.fetchall() if len(repeated_items) == 0: message = 'no repeated ip:port in ipool' logging.info(message) return else: message = 'found ip:port repeated {} kinds'.format( len(repeated_items)) logging.info(message) command = ''' DELETE FROM ipool WHERE no IN (SELECT no FROM (SELECT no FROM ipool WHERE (ip,port) IN (SELECT ip,port FROM ipool GROUP BY ip,port HAVING COUNT(*)>1) AND no NOT IN (SELECT MAX(no) FROM ipool GROUP BY ip,port HAVING COUNT(*)>1)) AS a);''' command = command.replace('\n', '') command = command.replace(' ', '') try: cur.execute(command) conn.commit() message = 'repeated ip:port deleted successfully. ' logging.info(message) except: conn.rollback() raise MysqlOperationFailed @staticmethod def get_ips(anonymous: list = [1], type_: list = [2], response_time: int = 30000, last_verify_interval: str = '48:00:00', limit: int = 1000) -> tuple: """根据要求返回多个ip，只返回ip和port。 :param last_verify_interval 用于指定上次验证的时间距现在不超过多久。""" cur, conn = mysql_api.cur, mysql_api.conn cur.execute('USE ipool;') anonymous, type_ = tuple(anonymous), tuple(type_) command = ''' SELECT ip,port FROM ipool WHERE anonymous IN {} AND type IN {} AND response_time BETWEEN 0 AND {} AND (NOW()-last_verify)<"{}" ORDER BY response_time LIMIT {}; '''.format(anonymous, type_, response_time, last_verify_interval, limit) command = command.replace('\n', '') command = command.replace(' ', '') print(command) cur.execute(command) ips = cur.fetchall() return ips import threading, requests, re from lxml import etree headers = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/104.0.0.0 Safari/537.36' } class spider_threads(): """爬虫类，通过多线程的方法同时爬取多个网站。""" websites = [ 'http://www.ip3366.net/free/' ] url_spidered_list = [] for i in range(len(websites)): url_spidered_list.append(set()) def __init__(self): self.threads = len(spider_threads.websites) @staticmethod def get_session(): """返回一个session对象。""" session = requests.Session() session.mount('http://', requests.adapters.HTTPAdapter(max_retries=3)) session.mount('https://', requests.adapters.HTTPAdapter(max_retries=3)) return session class spider_website(): """定义一个网站爬虫的基本操作。""" def __init__(self, no: int): self.session = spider_threads.get_session() self.base_url = spider_threads.websites[no] self.url_spidered = spider_threads.url_spidered_list[no] self.kind = re.search(r'\.{1}.+\.{1}', self.base_url).group().replace('.', '') self.mysql_api = mysql_api() def get_html(self, url): """对一个页面，返回其响应对象。""" global headers session = self.session try: html = session.get(url, timeout=30, headers=headers) except requests.exceptions.TooManyRedirects as e: message = "网页重定向，休息10min（{}）。".format(self.kind) logging.warning(message) print('\r' + message, end='') time.sleep(600) html = session.get(url, timeout=30, headers=headers) html.encoding = html.apparent_encoding return html def get_ip_recursively(self, url): """这个方法用于递归调用爬虫，将在子类中重写。""" pass def start_spider(self): """这个方法用于开启爬虫，将在子类中重写。""" pass class spider_ip3366(spider_website): """ip3366的爬虫类。""" def __init__(self): super().__init__(0) self.start_urls = [ self.base_url, 'http://www.ip3366.net/free/?stype=2' ] def get_ip_recursively(self, url): """这是一个递归调用的爬虫。它将爬取页面上的信息，将爬取过的网页存入字典，对页面上新的网页进行爬取。 重写了父类方法。""" html = self.get_html(url) time.sleep(15) message = '1 page crawled:{}'.format(url) logging.info(message) html = etree.HTML(html.content) ip_nodes = html.xpath(r'//*[@id="list"]/table/tbody/tr') ips_list = [] for ip_node in ip_nodes: ip_dict = {} ip = ip_node.xpath(r'./td[1]/text()')[0].strip() port = int(ip_node.xpath(r'./td[2]/text()')[0].strip()) anonymous = (0, 1)['高匿' in ip_node.xpath(r'./td[3]/text()')[0]] type_ = (0, 1)['HTTPS' in ip_node.xpath(r'./td[4]/text()')[0]] physical_address = ip_node.xpath(r'./td[5]/text()')[0].strip() response_time = 1000 * int( ip_node.xpath(r'./td[6]/text()')[0].strip().replace( '秒', '')) response_time = (response_time, 250)[response_time == 0] last_verify = ip_node.xpath( r'./td[7]/text()')[0].strip().replace('/', '-') ip_dict['ip'] = ip ip_dict['port'] = port ip_dict['anonymous'] = anonymous ip_dict['type'] = type_ ip_dict['physical_address'] = physical_address ip_dict['response_time'] = response_time ip_dict['last_verify'] = last_verify ips_list.append(ip_dict) self.mysql_api.save_ips(ips_list) self.url_spidered.add(url) page_urls = html.xpath(r'//*[@id="listnav"]/ul/a/@href') for i in range(len(page_urls)): page_url = self.base_url + page_urls[i] if page_url not in self.url_spidered: self.get_ip_recursively(page_url) else: return def start_spider(self): """这个方法用于开启爬虫，对父类的同名方法进行重写。""" for start_url in self.start_urls: self.get_ip_recursively(start_url) self.mysql_api.check_repeat() def start_all_spider_process(): process_list = [] process_list.append(crawl_website_ip3366().start_spider) for i in process_list: process_pool.apply_async(i) time.sleep(35) process_pool.close() process_pool.join() if __name__ == "__main__": try: start_all_spider_process() except: traceback.print_exc()spider_ip3366类是一个范例，其他爬虫也照着这样写就行了，所以剩下的只是简单的重复劳动，这里笔者就不进行展示了。如果要继续写的话，注意其他网站的爬虫都和spider_ip3366类处于同一个层级，也在spider_threads内部。添加其他网站的爬虫后，在start_all_spider_threads里添加线程即可。（4）实际运行自行添加其他网站的爬虫后，运行。（5）检测爬取的ip这是相对独立的一个步骤，并且要在使用时同步进行，将在后续的文章中讲解。5.总结与改进（1）这个爬虫使用了递归，在内存小一点的机器上其实不太好运行。有可能的话，可以改用递归外的方法（比如用scrapy），专门买一个更加小巧的机器或云服务器，全天运行这个爬虫和ip检验，以便能及时更新自己的ipool。（2）在上一个例子（ Python，爬虫与深度学习（2）——爬虫详解（一）爬取金融机构成交持仓排名 ）和本例中，都用到了一些重复的方法，比如连接mysql时添加conn、cur，以及log操作。这部分其实可以在自建库里写方法调用，也免得每次都要更改。（3）这里所有爬虫使用的log文件都是同一个，不太容易看清楚。后续可以让每个爬虫使用不同的log文件。（4）除了自己写这种ipool爬虫，其实在github上有其他人做好的成品。笔者在这里推荐两个项目： https://github.com/SpiderClub/haipproxy 和 https://github.com/imWildCat/scylla ，这两个项目都可以用docker安装。更新：docker-scylla可能自带挖矿病毒，详见这个问题： https://github.com/imWildCat/scylla/issues/102 。
Python，爬虫与深度学习（2）——爬虫详解（一）爬取金融机构成交持仓排名
步骤/目录： 1.需求分析 2.网页请求分析 3.mysql数据库的设置 4.爬虫编写 （0）安装所需模块 （1）日期 （2）爬虫主体编写 （3）mysql存储 （4）正式运行 5.总结与后续本文首发于个人博客https://lisper517.top/index.php/archives/39/，转载请注明出处。本文的目的是从实战出发讲解爬虫+python。由于还想讲解一下python的写法，所以爬虫写的稍微多了一点。本文写作日期为2022年8月30日。操作的平台为win10，编辑器为VS code。爬虫写的好，牢房进的早。使用爬虫时不要占用服务器太多流量，否则可能惹祸上身。除了python与爬虫，这里实际上还需要了解html格式与mysql命令。可以在 runoob-html 或 runoob-mysql 找到初步的学习资料，或者查看笔者后续的文章。最后，笔者的能力有限，也并非CS从业人员，很多地方难免有纰漏或者不符合代码原则的地方，请在评论中指出。1.需求分析在中国金融期货交易所（中金期货）的官网上有各个金融机构的成交持仓数据，具体网址为 http://cffex.com.cn/ccpm/ 。搜索期货、期权，选择类目、日期，即可看到各机构的成交持仓排名，这个数据可以一定程度上反映各金融机构对大盘走势的判断。本文将介绍如何爬取这些数据，在之后的文章中会试图用机器学习的方法对这些数据进行学习、预判。需求是爬取这些数据，并存入mysql数据库。这里会用到requests、bs4（BeautifulSoup）库，就不用scrapy这种牛刀了（scrapy一般也不会用于爬这种表格请求，而是用于递归爬取全站）。2.网页请求分析在chrome中打开网页，按F12打开网页开发工具，在右上角竖着的三个点里找到 停靠侧 ，选择 取消停靠至单独的窗口 。在网页窗口中选择期货，选择一种类目（比如IF），选择前一天，点击查询，这时网页里会出现数据，网页开发工具里会出现一条新的项目，点击这个项目，发现请求网址为 http://cffex.com.cn/sj/ccpm/2022XX/XX/IF.xml?id=97 ，其中 2022XX/XX 是当天的日期。复制这个请求网址并打开，发现是XML格式的数据（html），其中一项如下：<positionRank> <instrumentid>IF2212</instrumentid> <tradingday>2022XXXX</tradingday> <datatypeid>0</datatypeid> <rank>13</rank> <shortname>宝城期货</shortname> <volume>343</volume> <varvolume>136</varvolume> <partyid>0110</partyid> <productid>IF</productid> ...与原网页对照，其中 datatypeid 可能有0、1、2，分别表示成交量排名、持买单量排名、持卖单量排名。经过实验，网址里的 ?id=97 没有什么具体的意义，大概就是一个1~99的随机数。那么只用日期和品种生成网址就行了。最后检查一下请求和响应里有没有什么特殊的项目，没有则只设置一下请求的User-Agent（UA）即可。3.mysql数据库的设置这里选择用docker在树莓派上运行mysql服务。当然也可以直接安装mysql，各种平台安装mysql的教程见 runoob教程 。无论何种方法运行mysql，最关键的问题是搞清楚运行mysql机器的ip，和mysql的root用户密码。如果不是在本地运行mysql，还要注意mysql的root用户是否能远程登录。在树莓派上进行如下操作：mkdir -p /docker/mysql/data/mysql mkdir -p /docker/mysql/conf/mysql mkdir -p /docker/mysql/backup/mysql docker pull mysql:8.0.29 nano /docker/mysql/docker-compose.yml写入如下内容：version: "3.9" services: mysql: image: mysql:8.0.29 environment: MYSQL_ROOT_PASSWORD: mysqlpasswd ports: - "53306:3306" command: - mysqld - --character-set-server=utf8mb4 volumes: - /docker/mysql/data/mysql:/var/lib/mysql - /docker/mysql/conf/mysql:/etc/mysql/conf.d - /docker/mysql/backup/mysql:/backup logging: driver: syslog restart: always记得把 MYSQL_ROOT_PASSWORD: mysqlpasswd 这里改成自己想设置的密码，把 53306 也可以改成自己想要的端口（不建议用默认的3306端口）。然后运行：cd /docker/mysql docker-compose config docker-compose up -d接下来进容器服务里看看mysql是否正常运行：docker exec -it mysql-mysql-1 bash #如果是老版本的docker-compose，把这一条命令中的所有 - 换成 _ mysql -uroot -p #输入密码 SHOW DATABASES;到此为止就可以了。在爬虫编写时，会检验需要的数据库、表是否创建，未创建的话会自动创建，这里就不需要手动创建了。如果直接安装mysql并从其他机器访问，可能需要远程登录mysql（docker-mysql无需如下设置），可在mysql命令行中输入如下命令：USE mysql; UPDATE user SET host='%' WHERE user='root';更多关于远程登录mysql的内容可自行搜索，或者查看这篇文章 。4.爬虫编写（0）安装所需模块需要用到requests（处理网络请求）、bs4（对响应内容进行解析）、pymysql（连接mysql数据库）、lxml（网页解析器）、cryptography（连接mysql时密码加密）库，在cmd中使用 pip install requests bs4 pymysql lxml cryptography 下载（下载失败就一直使用上述命令，直到下载成功）。之后新建一个文件夹，用VS Code打开该文件夹、新建一个py文件，进行下面的操作。（1）日期首先编写一个日期类，需要达到的目的主要是给出下一个日期，和计算一个日期是星期几。这里自己创建日期类完全是为了实践，其实可以直接使用python自带的datetime库。#!/usr/bin/python #这一行针对linux系统，指定python解释器的位置 # -*- coding: UTF-8 -*- #这一行指定本文件的内容编码格式为UTF-8 class NotValidateDate(Exception): #(Exception)表示这个类继承自Exception这个父类。python中的类都继承自一个基类，但可以不写出来。 """若日期不合法则抛出这个异常。""" #类、方法/函数、模块的注释都可以这样写。在VS Code中，把鼠标悬停在类、方法/函数、模块的名字或对象上就能看到注释。 def __init__(self): message = "Your input is not a validate date, please check. " super().__init__(message) #super函数用于调用父类的方法。 #如果日期的格式不对，就会抛出这个异常，并提示上述信息 class my_date(): """日期类，最主要的功能是给定一个日期，给出下一个日期；以及查看一个日期是星期几。""" def __init__(self, year: int, month: int, day: int, weekday: int = 0): #类的__init__方法在初始化一个对象时调用。这里对所有参数都指定了类型（int整型），对最后一个参数还指定了默认值。 #year、month、day、weekday 分别表示 年、月、日、星期几，其中星期几可以不提供 if my_date.validate(year, month, day): #验证日期的有效性后再用输入的数据初始化对象 self.year, self.month, self.day, self.weekday = year, month, day, weekday if weekday == 0: #如果weekday是默认值，说明未提供星期几，则进行计算。 self.weekday = self._get_weekday() #为了运行的效率，这里没有验证：若输入了weekday，weekday是否正确。 else: raise NotValidateDate #如果日期有什么地方不对，就抛出无效日期的异常 @staticmethod # @staticmethod 表示静态方法，即使没有实例化对象也可以调用，调用时用 类名.方法名 。另外注意静态方法的参数不要有self def validate(year: int, month: int, day: int) -> bool: # -> bool 指定了这个函数的返回值应该是bool变量 """验证给定的日期是否合法。""" if month < 1 or month > 12: #月份需要在1~12之间 return False elif month in (1, 3, 5, 7, 8, 10, 12): #大月的日期需要在1~31之间 if day < 1 or day > 31: return False elif month in (4, 6, 9, 11): #小月的日期需要在1~30之间 if day < 1 or day > 30: return False elif month == 2: #2月，闰年时日期在1~29，平年在1~28之间 if day < 1 or day > 29: return False elif day == 29: if not my_date.is_leap_year(year): return False return True #以上检测都通过则说明日期有效 # 1,3,5,7,8,10,12 31 # 4,6,9,11 30 # 2 28/29 @staticmethod def is_leap_year(year: int) -> bool: """判断是否闰年，是则返回True。""" return (year % 4 == 0 and year % 100 != 0) or (year % 400 == 0) #闰年的判断：能整除400，或者能整除4且不能整除100 def next_date(self): """以my_date对象形式返回下一天。""" y, m, d, w = self.year, self.month, self.day, self.weekday #python中可以这样进行多个变量赋值。=右边还可以是有序的序列，这种赋值称为解包 w = (w + 1) % 7 if w == 0: #当w=6时，(w + 1) % 7得0，但实际上需要得到7 w = 7 d += 1 if d == 32: #当日期为32，说明月份需要+1 d = 1 m += 1 if m == 13: #当月份为13，说明年份需要+1。反过来，仅当月份为12时年份可能需要+1，而12是大月。 m = 1 y += 1 elif d == 31: #当日期为31且月份为小月，月份也需要+1 if m in (4, 6, 9, 11): d = 1 m += 1 elif d == 30 and m == 2: #当日期为30、月份为2，月份也+1 d = 1 m = 3 elif d == 29 and m == 2: #当日期为29、月份为2、年份为平年，月份要+1 if not my_date.is_leap_year(y): d = 1 m = 3 return my_date(y, m, d, w) # a=my_date(y, m, d, w) 这样的写法赋予了my_date对象一个名字。直接操作 my_date(y, m, d, w) 时，这种对象称为无名对象。 def __lt__(self, another_date): #编写类的 __lt__ （less than）方法可以让对象能使用<符号。这里的目的是比较两个日期对象。 #a < b 时，实际上是a调用了 __lt__ 方法，可以理解为 __lt__(a, b)。 """定义两个日期的 < 操作符。""" #按年、月、日的顺序依次向下比较 if self.year < another_date.year: return True elif self.year > another_date.year: return False elif self.year == another_date.year: if self.month < another_date.month: return True elif self.month > another_date.month: return False elif self.month == another_date.month: if self.day < another_date.day: return True else: return False def __eq__(self, another_date): # __eq__（equal）则是==符号。仅年月日都相同时两个日期才相等。 """定义两个日期的 == 操作符。""" if (another_date.year == self.year) and (another_date.month == self.month) and (another_date.day == self.day): return True else: return False def __gt__(self, another_date): # __gt__ 即 greater than """定义两个日期的 > 操作符。""" if self.year > another_date.year: return True elif self.year < another_date.year: return False elif self.year == another_date.year: if self.month > another_date.month: return True elif self.month < another_date.month: return False elif self.month == another_date.month: if self.day > another_date.day: return True else: return False def __sub__(self, another_date) -> int: # __sub__ 即减法。日期的减法没有太好的方法，这里就从小的日期开始一天一天加。另外注意正负 #按道理来说，还可以定义my_date与整数的加减法，这里用不到，就不定义了。 """定义两个日期对象的减法，返回两者相差的日期数。""" count = 0 if self == another_date: return count elif self > another_date: new_date1 = another_date.next_date() new_date2 = self.next_date() while (new_date1.year != new_date2.year) or ( new_date1.month != new_date2.month) or (new_date1.day != new_date2.day): new_date1 = new_date1.next_date() count += 1 elif self < another_date: new_date1 = another_date.next_date() new_date2 = self.next_date() while (new_date1.year != new_date2.year) or ( new_date1.month != new_date2.month) or (new_date1.day != new_date2.day): new_date2 = new_date2.next_date() count -= 1 return count def _get_weekday(self) -> int: #用前面定义的减法，和一个标准日期，很容易算出星期几 """从标准日期开始，算出星期几。""" standard = my_date(1970, 1, 1, 4) #1970.1.1是一个比较特殊的日期 count = self - standard w = ((4 + count) % 7 + 7) % 7 #考虑到有可能w为负数（要算的年份早于1970），所以再进行一次 + 7) % 7 if w == 0: w = 7 return w def get_date_tuple(self) -> tuple: """以元组的形式返回一个日期，格式为 (年, 月, 日, 星期) """ return (self.year, self.month, self.day, self.weekday) if __name__ == "__main__": a = my_date(2022, 4, 30) b = my_date(2001, 1, 31) print(a.get_date_tuple()) print(b.get_date_tuple()) print(a - b) print(b - a) print(a == b) print(a > b) print(a < b) 可以把 a = my_date(2022, 4, 30) 和 b = my_date(2001, 1, 31) 自行修改数字进行一些测试。（2）爬虫主体编写爬虫的目的是爬取并存储网页信息。如下：base_url = "http://cffex.com.cn/sj/ccpm/" import random, time, requests, os, logging from requests.adapters import HTTPAdapter from traceback import print_exc if not os.path.isdir(os.path.join(os.getcwd(), 'logs')): os.mkdir(os.path.join(os.getcwd(), 'logs')) #如果没有这个文件夹就创建。这是用到某个文件夹时的常规操作，先检查后使用。 logging.basicConfig(level=logging.INFO, filename=os.path.join( os.getcwd(), 'logs', 'log{}.txt'.format( time.strftime("%Y-%m-%d-%H_%M_%S")))) #os.path.join()用于把两个路径连接起来。由于不同系统的目录分隔符不一样，建议用这个函数。 #os.getcwd()可以返回当前脚本运行的路径。VS Code里，最好 打开文件夹 后再运行脚本，否则os.getcwd()返回的当前路径可能不对，但是生产环境中是没问题的。 headers = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/104.0.0.0 Safari/537.36' } #headers是请求头，这里稍微伪装了一下使用的浏览器。 class spider(): """爬虫类，根据开始日期、结束日期及种类爬取相应的网址，并保存爬到的网页。""" session = requests.Session() session.mount('http://', HTTPAdapter(max_retries=3)) session.mount('https://', HTTPAdapter(max_retries=3)) #用session对象可以设置重连次数。包括第一次连接，若出现异常，最多尝试连接4次 time_spent_total = 0.0 #爬虫总用时 time_spent_every_page = 0.0 #每个页面平均用时 save_path = os.path.join(os.getcwd(), 'page_sources') #文件保存路径。 #在 __init__ 外面写的数据成员，不实例化对象也能使用，像静态方法，调用格式也是 类名.数据成员名 def __init__(self, start_date: tuple, end_date: tuple, kinds: tuple, spider_start_time: float = time.time()): #这个爬虫要求是输入开始、结束日期，及要爬取的品种，爬取相应的数据 if start_date > end_date: start_date, end_date = end_date, start_date #开始日期要小于结束日期 self.start_date = my_date(start_date[0], start_date[1], start_date[2]) self.end_date = my_date(end_date[0], end_date[1], end_date[2]) self.kinds = kinds self.total_days = self.end_date - self.start_date self.spider_start_time = spider_start_time #为了查看耗时，添加了爬虫的开始时间 if not os.path.isdir(spider.save_path): os.mkdir(spider.save_path) @staticmethod def get_url(date: my_date, kind: str) -> str: """根据单个日期和单个种类构建一个网址。""" global base_url y, m, d, w = date.get_date_tuple() return base_url + "{}{:02}/{:02}/{}.xml?id={}".format( y, m, d, kind, random.randint(1, 99)) #id就用随机数生成。一般网址里这样 ?id=xx 这种格式是浏览器会提交给服务器的数据 #有时候用户名密码这样的数据也能这样传递，但现在更多使用安全性更高的方法 @staticmethod def get_abs_file_path(date: my_date, kind: str) -> str: #这里把生成文件名的规则单独写一个静态方法，是为了之后存入mysql时用相同的规则。 """根据日期，返回保存文件的绝对路径。""" year, month, day, weekday = date.get_date_tuple() file_name = "{}{:02}{:02}{}".format(year, month, day, kind) + ".html" return os.path.join(spider.save_path, file_name) @staticmethod def page_spider(date: my_date, kind: str): #根据输入的单个日期、单个种类爬取对应的一个页面。注意这里是把页面迅速存储成文件，也是为了减轻别人服务器的负担。 #存下来的页面后续再进行处理，存入数据库 """对一个页面进行爬取，并将页面存成文件。""" url = spider.get_url(date, kind) session = spider.session try: xml = session.get(url, timeout=30, headers=headers) #xml是响应对象，除了响应的内容，还可以看响应的状态码等。 except requests.exceptions.TooManyRedirects as e: #如果短时间内访问太多次，这个网页会重定向（引发这个异常），这时休息10min再爬。 message = "网页重定向，目前进度为{}。".format(date.get_date_tuple()) logging.warning(message) print('\r' + message, end='') time.sleep(600) xml = session.get(url, timeout=30, headers=headers) xml.encoding = xml.apparent_encoding #这次爬取的网页是gb2312编码，有时候可能遇到UTF-8编码，所以要加这句。 if r' 网页错误 ' in xml.text: return 404 #有时候比如国庆放假、没有当天的数据，网页就会有一个title标签，内容是 网页错误 。 abs_file_path = spider.get_abs_file_path(date,kind) with open(abs_file_path, "w", encoding=xml.encoding) as f: f.write(xml.text) f.close() def start_spider(self): #爬虫启动的入口。增加了爬取进度的显示，log日志的写入。 """启动爬虫，在爬取过程中显示进度等信息。""" current_date = self.start_date while not current_date > self.end_date: y, m, d, w = current_date.get_date_tuple() if w == 6 or w == 7: #星期六、七直接跳过，因为这两天是休息日，绝对没有数据，金融市场不会调休。 current_date = current_date.next_date() continue for kind in self.kinds: result = self.page_spider(current_date, kind) time.sleep(1) # 一定注意不要爬的太快了 #大家多体谅一下，世界会变成美好的人间。爬的不要太快，每个页面爬完休息1s。 if result == 404: message = "步骤1/2：{}-{:02}-{:02}-{}".format( y, m, d, kind) + " 页面不存在。" logging.info(message) print('\r' + message, end='') #如果遇到非周末的放假日，直接跳到下一个日期 continue spider.time_spent_total = time.time() - self.spider_start_time spider.time_spent_every_page = spider.time_spent_total / ( current_date - self.start_date + 1) time_left = (self.end_date - current_date) * spider.time_spent_every_page message = "步骤1/2：成功爬取" + "{}-{:02}-{:02}-{}".format( y, m, d, kind) + "页面，共用时{:.0f}分钟，预计还需{:.0f}分钟，当前进度{:.2%}。".format( spider.time_spent_total / 60, time_left / 60, (current_date - self.start_date) / self.total_days) logging.info(message) print('\r' + message, end='') #在命令行界面和日志中打印、保存爬取成功的信息。其他地方也可以加logging。 current_date = current_date.next_date() if __name__ == "__main__": try: base_url = "http://cffex.com.cn/sj/ccpm/" # base_url 作为全局变量已经声明过了，这里是方便以后若网址变动、可及时修改。 headers = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/104.0.0.0 Safari/537.36' } start_date = (2022, 7, 1) end_date = (2022, 7, 30) kinds = ('IF', 'IC', 'IM', 'IH', 'TS', 'TF', 'T', 'IO', 'MO') spider(start_date, end_date, kinds).start_spider() print('\n已完成') input() # input() 会让函数卡在这里，这样运行完后命令行界面不会消失。 input() 本来的作用是让用户输入，返回输入的字符串。 except: print_exc() input() # input() 同理。如果要测试这个爬虫，记得把上面那个日期类最后 if __name__ == "__main__": 及后面的部分删掉，添在爬虫主体代码的前面。一定注意每个页面爬取后休息一下，比如1s，不要占用别人服务器太多带宽，做人留一线，日后好相见。另外，爬取到页面后首先把页面存到本地，以后就不爬了，也是为了减轻别人的服务器负担。（3）mysql存储使用mysql存储，后续查找数据更快。import pymysql from bs4 import BeautifulSoup class MysqlOperationFailed(Exception): """若更改mysql出现错误则抛出这个异常。""" def __init__(self): message = "Failed to change data in MySQL, please check. " super().__init__(message) class mysql_save(): conn = pymysql.connect(host='127.0.0.1', port=53306, user='root', passwd='mysqlpasswd)', db='mysql', charset='utf8') #这里注意要对host，port，passwd进行更改。 cur = conn.cursor() time_spent_total = 0.0 time_spent_every_page = 0.0 def __init__(self, start_date: tuple, end_date: tuple, kinds: list, mysql_save_start_time: float = time.time()): if start_date > end_date: start_date, end_date = end_date, start_date self.start_date = my_date(start_date[0], start_date[1], start_date[2]) self.end_date = my_date(end_date[0], end_date[1], end_date[2]) self.kinds = kinds self.total_days = self.end_date - self.start_date self.mysql_save_start_time = mysql_save_start_time def __del__(self): mysql_save.cur.close() mysql_save.conn.close() @staticmethod def get_bs4_object(date: my_date, kind: str): """根据日期和种类，返回用bs4解析后的BeautifulSoup对象；文件不存在时返回 404 整数。""" abs_file_path = spider.get_abs_file_path(date, kind) if not os.path.exists(abs_file_path): message = '{}文件不存在。'.format(abs_file_path) logging.info(message) return 404 with open(abs_file_path, "rb") as f: #注意这里rb是只读、二进制打开文件。由于是二进制，也不需要指定文件编码。 xml = f.read() f.close() return BeautifulSoup(xml, 'lxml') @staticmethod def analyse_page_source(date: my_date, kind: str): """对单个页面进行解析，得到各种数据，并进行存储。如果文件不存在或文件包含 '网页错误' ，分别返回404、403。""" bs_object = mysql_save.get_bs4_object(date, kind) if bs_object == 404: return 404 if bs_object.find('title', text='网页错误'): abs_file_path = spider.get_abs_file_path(date, kind) message = '在{}文件中发现 "网页错误" ，这可能是爬虫出现了错误，请检查。'.format(abs_file_path) logging.warning(message) print('\n' + message) return 403 for dataitem in bs_object.find_all('data'): # find_all() 方法返回一个列表，其元素为所有符合条件的标签。这里的条件是标签名为data。 instrumentid = dataitem.find('instrumentid').get_text() # get_text() 方法以字符串返回标签的内容。 tradingday = dataitem.find('tradingday').get_text() datatypeid = dataitem.find('datatypeid').get_text() paiming = dataitem.find('rank').get_text() shortname = dataitem.find('shortname').get_text() volume = dataitem.find('volume').get_text() varvolume = dataitem.find('varvolume').get_text() partyid = dataitem.find('partyid').get_text() tradingday = tradingday[0:4] + '-' + tradingday[ 4:6] + '-' + tradingday[6:8] datatypeid = int(datatypeid) paiming = int(paiming) volume = int(volume) varvolume = int(varvolume) partyid = int(partyid) instrumentid = instrumentid.strip() shortname = shortname.strip() #在数据存储的过程中，对数据进行清洗是非常重要的一环。有些时候网站创建者可能更改了一些数据的格式，需要注意。 try: mysql_save.cur.execute( 'INSERT INTO cffex (instrumentid, tradingday, datatypeid, paiming, shortname, volume, varvolume, partyid) VALUES ("%s", "%s", %d, %d, "%s", %d, %d, %d);' % (instrumentid, tradingday, datatypeid, paiming, shortname, volume, varvolume, partyid)) #这里没有检验相同的数据是否存在，这一步可以在数据导入完后进行。注意检验时，对每一条记录，只需要查找比它早的记录里有没有重复的就行了，另外最好用mysql提供的查重来检验。 mysql_save.conn.commit() logging.info('将{}的{}数据写入数据库。'.format(date.get_date_tuple(), kind)) except: mysql_save.conn.rollback() raise MysqlOperationFailed @staticmethod def check_table(): """检查一下mysql中是否已经建好了表，若表不存在则新建表。这里并没有检查：若表存在，表的格式是否正确。""" cur, conn = mysql_save.cur, mysql_save.conn try: cur.execute("CREATE DATABASE IF NOT EXISTS futures;") cur.execute("USE futures;") command = '''CREATE TABLE IF NOT EXISTS cffex ( no INT UNSIGNED NOT NULL AUTO_INCREMENT, instrumentid VARCHAR(7) NOT NULL, tradingday DATE NOT NULL, datatypeid TINYINT NOT NULL, paiming TINYINT NOT NULL, shortname VARCHAR(5) NOT NULL, volume MEDIUMINT NOT NULL, varvolume MEDIUMINT NOT NULL, partyid SMALLINT NOT NULL, created TIMESTAMP DEFAULT CURRENT_TIMESTAMP, PRIMARY KEY (no));''' #创建表的语句。由于 rank 是mysql里的关键字，所以使用 paiming 代替。 command = command.replace('\n', '') command = command.replace(' ', '') cur.execute(command) conn.commit() except: mysql_save.conn.rollback() raise MysqlOperationFailed def start_mysql_save(self): """入口，启动向mysql写入数据的过程，并且显示进度等信息。""" mysql_save.check_table() print('\n步骤1完成') current_date = self.start_date while not current_date > self.end_date: w = current_date.get_date_tuple()[3] if w == 6 or w == 7: current_date = current_date.next_date() continue for kind in self.kinds: mysql_save.analyse_page_source(current_date, kind) mysql_save.time_spent_total = time.time( ) - self.mysql_save_start_time mysql_save.time_spent_every_page = mysql_save.time_spent_total / ( current_date - self.start_date + 1) time_left = (self.end_date - current_date) * mysql_save.time_spent_every_page message = "步骤2/2：存入数据库中，共用时{:.0f}分钟，预计还需{:.0f}分钟，当前进度{:.2%}。".format( mysql_save.time_spent_total / 60, time_left / 60, (current_date - self.start_date) / self.total_days) logging.info(message) print('\r' + message, end='') current_date = current_date.next_date() mysql_save.check_repeat() #进行一次去重 @staticmethod def check_repeat(): """检查爬到的数据里有没有重复的，若有则只保留最早的一条。""" #检查有无重复，若有重复则删除重复数据、只保留最早的一条。 cur, conn = mysql_save.cur, mysql_save.conn cur.execute('USE futures;') cur.execute( 'SELECT instrumentid,tradingday,datatypeid,partyid,volume FROM cffex GROUP BY instrumentid,tradingday,datatypeid,partyid,volume HAVING COUNT(*)>1;' ) repeated_items = cur.fetchall() if len(repeated_items) == 0: print('\n数据库中无重复数据。') return else: print('\n数据库中检测到重复数据，有{}种。'.format(len(repeated_items))) try: #下面的方法是用python循环去重 '''for repeated_item in repeated_items: instrumentid, tradingday, datatypeid, partyid, volume = repeated_item cur.execute( 'SELECT no FROM cffex WHERE instrumentid="{}" AND tradingday="{}" AND datatypeid={} AND partyid={} AND volume={};' .format(instrumentid, tradingday, datatypeid, partyid, volume)) nos = cur.fetchall() for i in range(1, len(nos)): cur.execute('DELETE FROM cffex WHERE no={};'.format( nos[i][0])) conn.commit()''' #上面的方法是用python循环去重 #下面的方法是一条语句去重 command = '''DELETE FROM cffex WHERE no IN (SELECT no FROM (SELECT no FROM cffex WHERE (instrumentid,tradingday,datatypeid,partyid,volume) IN (SELECT instrumentid,tradingday,datatypeid,partyid,volume FROM cffex GROUP BY instrumentid,tradingday,datatypeid,partyid,volume HAVING COUNT(*)>1) AND no NOT IN (SELECT MIN(no) FROM cffex GROUP BY instrumentid,tradingday,datatypeid,partyid,volume HAVING COUNT(*)>1)) AS a);''' command.replace(' ', '') cur.execute(command) conn.commit() #上面的方法是一条语句去重 except: conn.rollback() raise MysqlOperationFailed if __name__ == "__main__": try: base_url = "http://cffex.com.cn/sj/ccpm/" headers = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/104.0.0.0 Safari/537.36' } start_date = (2022, 7, 1) end_date = (2022, 7, 30) kinds = ('IF', 'IC', 'IM', 'IH', 'TS', 'TF', 'T', 'IO', 'MO') #spider(start_date, end_date, kinds).start_spider() #如果只需要从本地文件中导入数据到数据库，就把爬虫运行的这行注释掉。 mysql_save(start_date, end_date, kinds).start_mysql_save() print('\n已完成') input() except: print_exc() input()注意把数据库连接中的host，port，passwd改成自己的。如果要测试去重的mysql_save.check_repeat方法，可以运行多次start_mysql_save方法，最后用这个去重的方法实验去重。（4）正式运行去掉注释的爬虫文件（还是注意把数据库连接中的host，port，passwd改成自己的。并且这个文件有一些小修改）：#!/usr/bin/python # -*- coding: UTF-8 -*- class NotValidateDate(Exception): """若日期不合法则抛出这个异常。""" def __init__(self): message = "Your input is not a validate date, please check. " super().__init__(message) class my_date(): """日期类，最主要的功能是给定一个日期，给出下一个日期；以及查看一个日期是星期几。""" def __init__(self, year: int, month: int, day: int, weekday: int = 0): if my_date.validate(year, month, day): self.year, self.month, self.day, self.weekday = year, month, day, weekday if weekday == 0: self.weekday = self._get_weekday() else: raise NotValidateDate @staticmethod def validate(year: int, month: int, day: int) -> bool: """验证给定的日期是否合法。""" if month < 1 or month > 12: return False elif month in (1, 3, 5, 7, 8, 10, 12): if day < 1 or day > 31: return False elif month in (4, 6, 9, 11): if day < 1 or day > 30: return False elif month == 2: if day < 1 or day > 29: return False elif day == 29: if not my_date.is_leap_year(year): return False return True # 1,3,5,7,8,10,12 31 # 4,6,9,11 30 # 2 28/29 @staticmethod def is_leap_year(year: int) -> bool: """判断是否闰年，是则返回True。""" return (year % 4 == 0 and year % 100 != 0) or (year % 400 == 0) def next_date(self): """以my_date对象形式返回下一天。""" y, m, d, w = self.year, self.month, self.day, self.weekday w = (w + 1) % 7 if w == 0: w = 7 d += 1 if d == 32: d = 1 m += 1 if m == 13: m = 1 y += 1 elif d == 31: if m in (4, 6, 9, 11): d = 1 m += 1 elif d == 30 and m == 2: d = 1 m = 3 elif d == 29 and m == 2: if not my_date.is_leap_year(y): d = 1 m = 3 return my_date(y, m, d, w) def __lt__(self, another_date): """定义两个日期的 < 操作符。""" if self.year < another_date.year: return True elif self.year > another_date.year: return False elif self.year == another_date.year: if self.month < another_date.month: return True elif self.month > another_date.month: return False elif self.month == another_date.month: if self.day < another_date.day: return True else: return False def __eq__(self, another_date): """定义两个日期的 == 操作符。""" if (another_date.year == self.year) and (another_date.month == self.month) and (another_date.day == self.day): return True else: return False def __gt__(self, another_date): """定义两个日期的 > 操作符。""" if self.year > another_date.year: return True elif self.year < another_date.year: return False elif self.year == another_date.year: if self.month > another_date.month: return True elif self.month < another_date.month: return False elif self.month == another_date.month: if self.day > another_date.day: return True else: return False def __sub__(self, another_date) -> int: """定义两个日期对象的减法，返回两者相差的日期数。""" count = 0 if self == another_date: return count elif self > another_date: new_date1 = another_date.next_date() new_date2 = self.next_date() while (new_date1.year != new_date2.year) or ( new_date1.month != new_date2.month) or (new_date1.day != new_date2.day): new_date1 = new_date1.next_date() count += 1 elif self < another_date: new_date1 = another_date.next_date() new_date2 = self.next_date() while (new_date1.year != new_date2.year) or ( new_date1.month != new_date2.month) or (new_date1.day != new_date2.day): new_date2 = new_date2.next_date() count -= 1 return count def _get_weekday(self) -> int: """从标准日期开始，算出星期几。""" standard = my_date(1970, 1, 1, 4) count = self - standard w = ((4 + count) % 7 + 7) % 7 if w == 0: w = 7 return w def get_date_tuple(self) -> tuple: """以元组的形式返回一个日期，格式为 (年, 月, 日, 星期) """ return (self.year, self.month, self.day, self.weekday) base_url = "http://cffex.com.cn/sj/ccpm/" import random, time, requests, os, logging from requests.adapters import HTTPAdapter from traceback import print_exc if not os.path.isdir(os.path.join(os.getcwd(), 'logs')): os.mkdir(os.path.join(os.getcwd(), 'logs')) logging.basicConfig(level=logging.INFO, filename=os.path.join( os.getcwd(), 'logs', 'log{}.txt'.format( time.strftime("%Y-%m-%d-%H_%M_%S")))) headers = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/104.0.0.0 Safari/537.36' } class spider(): """爬虫类，根据开始日期、结束日期及种类爬取相应的网址，并保存爬到的网页。""" session = requests.Session() session.mount('http://', HTTPAdapter(max_retries=3)) session.mount('https://', HTTPAdapter(max_retries=3)) # 设置重连次数。包括第一次连接，若出现异常，最多尝试连接4次 time_spent_total = 0.0 time_spent_every_page = 0.0 save_path = os.path.join(os.getcwd(), 'page_sources') def __init__(self, start_date: tuple, end_date: tuple, kinds: tuple, spider_start_time: float = time.time()): if start_date > end_date: start_date, end_date = end_date, start_date self.start_date = my_date(start_date[0], start_date[1], start_date[2]) self.end_date = my_date(end_date[0], end_date[1], end_date[2]) self.kinds = kinds self.total_days = self.end_date - self.start_date self.spider_start_time = spider_start_time if not os.path.isdir(spider.save_path): os.mkdir(spider.save_path) @staticmethod def get_url(date: my_date, kind: str) -> str: """根据日期和种类构建网址。""" global base_url y, m, d, w = date.get_date_tuple() return base_url + "{}{:02}/{:02}/{}.xml?id={}".format( y, m, d, kind, random.randint(1, 99)) @staticmethod def get_abs_file_path(date: my_date, kind: str) -> str: """根据日期，返回保存文件的绝对路径。""" year, month, day, weekday = date.get_date_tuple() file_name = "{}{:02}{:02}{}".format(year, month, day, kind) + ".html" return os.path.join(spider.save_path, file_name) @staticmethod def page_spider(date: my_date, kind: str): """对一个页面进行爬取，并将页面存成文件。""" url = spider.get_url(date, kind) session = spider.session try: xml = session.get(url, timeout=30, headers=headers) except requests.exceptions.TooManyRedirects as e: message = "网页重定向，目前进度为{}。".format(date.get_date_tuple()) logging.warning(message) print('\r' + message, end='') time.sleep(600) xml = session.get(url, timeout=30, headers=headers) xml.encoding = xml.apparent_encoding if r' 网页错误 ' in xml.text: return 404 abs_file_path = spider.get_abs_file_path(date, kind) with open(abs_file_path, "w", encoding=xml.encoding) as f: f.write(xml.text) f.close() def start_spider(self): """启动爬虫，在爬取过程中显示进度等信息。""" current_date = self.start_date while not current_date > self.end_date: y, m, d, w = current_date.get_date_tuple() if w == 6 or w == 7: current_date = current_date.next_date() continue for kind in self.kinds: result = self.page_spider(current_date, kind) time.sleep(1) # 一定注意不要爬的太快了 if result == 404: message = "步骤1/2：{}-{:02}-{:02}-{}".format( y, m, d, kind) + " 页面不存在。" logging.info(message) print('\r' + message, end='') continue spider.time_spent_total = time.time() - self.spider_start_time spider.time_spent_every_page = spider.time_spent_total / ( current_date - self.start_date + 1) time_left = (self.end_date - current_date) * spider.time_spent_every_page message = "步骤1/2：成功爬取" + "{}-{:02}-{:02}-{}".format( y, m, d, kind) + "页面，共用时{:.0f}分钟，预计还需{:.0f}分钟，当前进度{:.2%}。".format( spider.time_spent_total / 60, time_left / 60, (current_date - self.start_date) / self.total_days) logging.info(message) print('\r' + message, end='') current_date = current_date.next_date() import pymysql from bs4 import BeautifulSoup class MysqlOperationFailed(Exception): """若更改mysql出现错误则抛出这个异常。""" def __init__(self): message = "Failed to change data in MySQL, please check. " super().__init__(message) class mysql_save(): conn = pymysql.connect(host='127.0.0.1', port=53306, user='root', passwd='mysqlpasswd', db='mysql', charset='utf8') cur = conn.cursor() time_spent_total = 0.0 time_spent_every_page = 0.0 def __init__(self, start_date: tuple, end_date: tuple, kinds: list, mysql_save_start_time: float = time.time()): if start_date > end_date: start_date, end_date = end_date, start_date self.start_date = my_date(start_date[0], start_date[1], start_date[2]) self.end_date = my_date(end_date[0], end_date[1], end_date[2]) self.kinds = kinds self.total_days = self.end_date - self.start_date self.mysql_save_start_time = mysql_save_start_time def __del__(self): mysql_save.cur.close() mysql_save.conn.close() @staticmethod def get_bs4_object(date: my_date, kind: str): """根据日期和种类，返回用bs4解析后的BeautifulSoup对象；文件不存在时返回 404 整数。""" abs_file_path = spider.get_abs_file_path(date, kind) if not os.path.exists(abs_file_path): message = '{}文件不存在。'.format(abs_file_path) logging.info(message) return 404 with open(abs_file_path, "rb") as f: xml = f.read() f.close() return BeautifulSoup(xml, 'lxml') @staticmethod def analyse_page_source(date: my_date, kind: str): """对单个页面进行解析，得到各种数据，并进行存储。如果文件不存在或文件包含 '网页错误' ，分别返回404、403。""" bs_object = mysql_save.get_bs4_object(date, kind) if bs_object == 404: return 404 if bs_object.find('title', text='网页错误'): abs_file_path = spider.get_abs_file_path(date, kind) message = '在{}文件中发现 "网页错误" ，这可能是爬虫出现了错误，请检查。'.format(abs_file_path) logging.warning(message) print('\n' + message) return 403 for dataitem in bs_object.find_all('data'): instrumentid = dataitem.find('instrumentid').get_text() tradingday = dataitem.find('tradingday').get_text() datatypeid = dataitem.find('datatypeid').get_text() paiming = dataitem.find('rank').get_text() shortname = dataitem.find('shortname').get_text() volume = dataitem.find('volume').get_text() varvolume = dataitem.find('varvolume').get_text() partyid = dataitem.find('partyid').get_text() tradingday = tradingday[0:4] + '-' + tradingday[ 4:6] + '-' + tradingday[6:8] datatypeid = int(datatypeid) paiming = int(paiming) volume = int(volume) varvolume = int(varvolume) partyid = int(partyid) instrumentid = instrumentid.strip() shortname = shortname.strip() try: mysql_save.cur.execute( 'INSERT INTO cffex (instrumentid, tradingday, datatypeid, paiming, shortname, volume, varvolume, partyid) VALUES ("%s", "%s", %d, %d, "%s", %d, %d, %d);' % (instrumentid, tradingday, datatypeid, paiming, shortname, volume, varvolume, partyid)) mysql_save.conn.commit() logging.info('将{}的{}数据写入数据库。'.format(date.get_date_tuple(), kind)) except: mysql_save.conn.rollback() raise MysqlOperationFailed @staticmethod def check_table(): """检查一下mysql中是否已经建好了表，若表不存在则新建表。这里并没有检查：若表存在，表的格式是否正确。""" cur, conn = mysql_save.cur, mysql_save.conn try: cur.execute("CREATE DATABASE IF NOT EXISTS futures;") cur.execute("USE futures;") command = '''CREATE TABLE IF NOT EXISTS cffex ( no INT UNSIGNED NOT NULL AUTO_INCREMENT, instrumentid VARCHAR(7) NOT NULL, tradingday DATE NOT NULL, datatypeid TINYINT NOT NULL, paiming TINYINT NOT NULL, shortname VARCHAR(5) NOT NULL, volume MEDIUMINT NOT NULL, varvolume MEDIUMINT NOT NULL, partyid SMALLINT NOT NULL, created TIMESTAMP DEFAULT CURRENT_TIMESTAMP, PRIMARY KEY (no));''' command = command.replace('\n', '') command = command.replace(' ', '') cur.execute(command) conn.commit() except: mysql_save.conn.rollback() raise MysqlOperationFailed def start_mysql_save(self): """入口，启动向mysql写入数据的过程，并且显示进度等信息。""" mysql_save.check_table() print('\n步骤1完成') current_date = self.start_date while not current_date > self.end_date: w = current_date.get_date_tuple()[3] if w == 6 or w == 7: current_date = current_date.next_date() continue for kind in self.kinds: mysql_save.analyse_page_source(current_date, kind) mysql_save.time_spent_total = time.time( ) - self.mysql_save_start_time mysql_save.time_spent_every_page = mysql_save.time_spent_total / ( current_date - self.start_date + 1) time_left = (self.end_date - current_date) * mysql_save.time_spent_every_page message = "步骤2/2：存入数据库中，共用时{:.0f}分钟，预计还需{:.0f}分钟，当前进度{:.2%}。".format( mysql_save.time_spent_total / 60, time_left / 60, (current_date - self.start_date) / self.total_days) logging.info(message) print('\r' + message, end='') current_date = current_date.next_date() mysql_save.check_repeat() @staticmethod def check_repeat(): """检查爬到的数据里有没有重复的。""" cur, conn = mysql_save.cur, mysql_save.conn cur.execute('USE futures;') cur.execute( 'SELECT instrumentid,tradingday,datatypeid,partyid,volume FROM cffex GROUP BY instrumentid,tradingday,datatypeid,partyid,volume HAVING COUNT(*)>1;' ) repeated_items = cur.fetchall() if len(repeated_items) == 0: print('\n数据库中无重复数据。') return else: print('\n数据库中检测到重复数据，有{}种。'.format(len(repeated_items))) try: #下面的方法是用python循环去重 '''for repeated_item in repeated_items: instrumentid, tradingday, datatypeid, partyid, volume = repeated_item cur.execute( 'SELECT no FROM cffex WHERE instrumentid="{}" AND tradingday="{}" AND datatypeid={} AND partyid={} AND volume={};' .format(instrumentid, tradingday, datatypeid, partyid, volume)) nos = cur.fetchall() for i in range(1, len(nos)): cur.execute('DELETE FROM cffex WHERE no={};'.format( nos[i][0])) conn.commit()''' #上面的方法是用python循环去重 #下面的方法是一条语句去重 command = '''DELETE FROM cffex WHERE no IN (SELECT no FROM (SELECT no FROM cffex WHERE (instrumentid,tradingday,datatypeid,partyid,volume) IN (SELECT instrumentid,tradingday,datatypeid,partyid,volume FROM cffex GROUP BY instrumentid,tradingday,datatypeid,partyid,volume HAVING COUNT(*)>1) AND no NOT IN (SELECT MIN(no) FROM cffex GROUP BY instrumentid,tradingday,datatypeid,partyid,volume HAVING COUNT(*)>1)) AS a);''' command.replace(' ', '') cur.execute(command) conn.commit() #上面的方法是一条语句去重 except: conn.rollback() raise MysqlOperationFailed if __name__ == "__main__": try: base_url = "http://cffex.com.cn/sj/ccpm/" headers = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/104.0.0.0 Safari/537.36' } start_date = (2010, 1, 1) end_date = (2022, 8, 31) kinds = ('IF', 'IC', 'IM', 'IH', 'TS', 'TF', 'T', 'IO', 'MO') spider(start_date, end_date, kinds).start_spider() #这一行是爬取并存储网页 #mysql_save(start_date, end_date, kinds).start_mysql_save() #这一行是解析存储的文件，存入mysql print('\n已完成') except: print_exc()几个品类的开始日期：IC、IH：2015.4.16 TS：2018.8.17 TF：2013.9.6 T：2015.3.20 IF：2010.4.16其中IF是最早的，为了保险一点，就从2010.1.1开始爬，把结束日期改成运行时的日期，放到合适的机器上（由于运行时间较长、光爬取就要大概11h，建议放到功耗小的机器，如笔记本或树莓派上；存入数据库还要1.5h）即可运行，运行之前在mysql里用 DROP DATABASE futures; 来删除测试过程中产生的数据。用爬虫爬取完后，再把 spider(start_date, end_date, kinds).start_spider() 注释掉、 #mysql_save(start_date, end_date, kinds).start_mysql_save() 去掉注释，重新运行一遍即可。最后，如果通过putty连接树莓派、希望在树莓派上后台运行爬虫，可如下操作：cd 脚本路径 chmod +x 脚本文件名 nohup python3 ./脚本文件名 > ./stdout.log 2>&1 & ps -aux | grep "python"最后一条命令执行后，可看到有一个进程为 python3 ./脚本文件名 ，说明脚本已在后台运行。在当前目录下会生成nohup.out文件用于记录输出。同时，这里把两个 input() 删掉了。如果要停止脚本，用 kill -9 进程号PID 命令，进程号PID就是 ps -aux | grep "python" 输出的第二列内容。5.总结与后续（1）这里编写的日期类实际上很多地方都能用到，可以把它放进自己编写的库里。当然这里的日期类还没有进行运行的优化，有些地方可能运行较慢。另外这个日期类还有可以添加的地方，比如添加 >= 、<= ，添加与int类之间的加减法。当然，datetime库也有类似的功能，在VS Code上新建一个py文件、写入 import datetime ，然后按住ctrl、左键点击 datetime 就能阅读其源码，好好学习一下。（2）爬虫爬取时要注意不要对服务器访问太过频繁。一是会占用别人的服务器资源，二是自己的ip可能会被别人ban掉。主要的方法有两个，增加访问间隔，以及把响应内容即时存储。如果实在想要短时间内爬取全站内容，可以使用一台机器异步爬取或多台机器分布式爬取（但这需要搭配ip池，代理服务器），这部分内容将在后续文章中讲解。（3）对mysql里的内容可定期进行查重，笔者在python中写成了mysql_save.check_repeat静态方法，方便调用。