步骤/目录： 1.运行mysql数据库与准备工作 2.自建库 3.创建、修改、运行scrapy项目 （1）其他配置 （2）爬虫配置 a.a2u.py b.clarketm.py c.cool_proxy.py d.free_proxy_list.py e.ip3366.py f.ipaddress.py g.jiangxianli.py h.kuaidaili.py i.pubproxy.py j.rmccurdy.py k.sunny9577.py l.TheSpeedX.py （3）运行爬虫 4.验证ip 5.日常使用本文首发于个人博客https://lisper517.top/index.php/archives/48/，转载请注明出处。本文的目的是使用 scrapy+mysql 搭建一个自己的代理ip池，即ipool。由于本文是对前文进行阶段性总结+拓展，所以内容较多，笔者希望本文单独拿出来也能成功跟着一步步操作。本文写作日期为2022年9月11日。mysql数据库运行在树莓派上，其他部分运行在win10机器；使用的编辑器为VS Code。爬虫写的好，牢房进的早。使用爬虫时不要占用服务器太多流量，否则可能惹祸上身。笔者的能力有限，也并非CS从业人员，很多地方难免有纰漏或者不符合代码原则的地方，请在评论中指出。1.运行mysql数据库与准备工作笔者在树莓派上使用docker安装mysql，需要预先安装docker、docker-compose，在树莓派上安装docker、docker-compose可查看 Docker入门（一） 。也不一定非要在树莓派上运行mysql，或者也不一定要用docker，或者也可以用redis等其他数据库代替mysql（使用其他数据库时后续的py文件可能需要自行调整），反正最终的目的是运行一个数据库，可以参考 runoob教程 使用其他方法安装运行mysql数据库。在树莓派上进行如下操作：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在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端口）。这里的 /docker/mysql/backup/mysql 目录是为了方便docker-mysql导出、备份数据。然后运行：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正常运行即可。为了保证创建database和table的命令能记录下来，这些命令将写在python文件中。然后在win10机器上准备一下环境。需要安装python3，安装scrapy和其他库，创建自建库文件夹，创建scrapy项目文件夹。win10机器安装python3的过程这里就不讲了。安装好后，在cmd中运行 pip install Scrapy ，其他可能需要的库还有requests（处理网络请求）、bs4（对响应内容进行解析）、pymysql（连接mysql数据库）、lxml（网页解析器）、cryptography（连接mysql时密码加密）、selenium（操控浏览器），在cmd中使用 pip install requests bs4 pymysql lxml cryptography selenium 下载（这几个库在本文中不一定都会用到，但学习爬虫迟早会用），注意若pip下载失败就一直使用上述命令直到成功，或者bing搜索一下pip换源。最后创建文件夹 D:\PythonLib\MyPythonLib 和 D:\spiders\scrapy ，或者你也可以自己选择其他路径。2.自建库在 D:\PythonLib\MyPythonLib 中新建一个空白的 __init__.py 文件，然后到python的安装目录里存放第三方库的文件夹（win10上一般是site-packages目录），比如 C:\Program Files\Python310\Lib\site-packages 中，新建一个 PythonLib.pth 文件，写入一行内容： D:\PythonLib 。接下来创建自建库文件，本次需要用到的自建库主要与log、mysql相关。新建 D:\PythonLib\MyPythonLib\log.py ，写入：#!/usr/bin/python # -*- coding: UTF-8 -*- import logging, time, os def get_logger(abs_path: str, log_file_name: str = '', getLogger_name: str = ''): '''传入绝对路径和log文件名，返回log对象。 :param abs_path: log文件存储的绝对路径 :param log_file_name: log文件名，若log文件名为空，则使用默认的文件名 :param getLogger_name: logger对象名，如果在同一个程序中需要同时写多个log文件，建议这个参数不要相同 :return log对象''' try: formatter = logging.Formatter( '%(lineno)d : %(asctime)s : %(levelname)s : %(funcName)s : %(message)s' ) if log_file_name == '': log_file_name = 'log#{}.txt'.format(time.strftime('-%Y-%m-%d')) fileHandler = logging.FileHandler( (os.path.join(abs_path, log_file_name)), mode='w', encoding='utf-8') fileHandler.setFormatter(formatter) if getLogger_name == '': getLogger_name = 'logger' log = logging.getLogger(getLogger_name) log.setLevel(logging.DEBUG) log.addHandler(fileHandler) return log except: raise Exception('error: MyPythonLib.log.get_logger failed ')新建 D:\PythonLib\MyPythonLib\mysql.py ，写入（注意把get_conn函数中对应的host、port、密码改成自己的mysql配置）：#!/usr/bin/python # -*- coding: UTF-8 -*- import pymysql, os, logging, time from MyPythonLib import log as mylog def get_conn(database_name: str = 'mysql'): '''指定数据库名，返回mysql连接对象。 :param database_name: 数据库名，不指定时默认用mysql库 :return Connection''' try: conn = pymysql.connect(host='127.0.0.1', port=53306, user='root', passwd='mysqlpasswd', db=database_name, charset='utf8') return conn except: raise Exception('MyPythonLib.mysql.get_conn failed ') def str_into_mysql(string: str) -> str: '''输入一个字符串，将其转换为可以在mysql中储存的形式。 这个字符串的使用位置是列值处，比如 "SELECT * FROM 表名 WHERE 列名='字符串';" 中的字符串，若包含 \\ 、 ' 、 " 这三个字符，就需要用本函数处理。 :param string: 目标字符串 :return 可以存入mysql的字符串 ''' try: string = string.replace('\\', '\\\\') string = string.replace("'", "\\'") string = string.replace('"', '\\"') return string except: raise Exception('MyPythonLib.mysql.str_into_mysql failed ') def table_exists(cur, table_name: str) -> bool: '''判断表在当前库中是否存在。 :param cur: 游标对象 :param table_name: 表名 :return 存在则返回True，否则返回False ''' try: cur.execute("SHOW TABLES;") results = cur.fetchall() for table in results: table = table[0] if table == table_name: return True return False except: raise Exception('MyPythonLib.mysql.table_exists failed ') class mysql_api_ipool(): """通过这个接口实现提交爬取的ip、取出ip的操作。""" conn = get_conn() cur = conn.cursor() #这个cursor是类公用的，用于不需要实例化对象的mysql操作 def __init__(self, log: logging.Logger = None): """:param log: 可选的log对象，若不提供，将在当前文件夹下新建logs目录、创建log文件""" if log == None: log_dir = os.path.join(os.getcwd(), 'logs') if not os.path.isdir(log_dir): os.mkdir(log_dir) log_file_name = 'mysql_api_ipool_log#{}.txt'.format( time.strftime('-%Y-%m-%d_%H_%M_%S')) log = mylog.get_logger(log_dir, log_file_name, 'mysql_api_ipool') self.log = log message = 'mysql_api_ipool object initialized' self.log.info(message) self.cur = mysql_api_ipool.conn.cursor() #对象自己的cur mysql_api_ipool.check_table() message = 'database ipool and table ipool checked' self.log.info(message) def __del__(self): self.cur.close() message = 'mysql_api_ipool object deleted' self.log.info(message) @staticmethod def check_table(): """检查一下mysql中是否已经建好了表，若表不存在则新建表。 :这里并没有检查：若表存在，表的格式是否正确。 数据库的各列含义与默认值： :no，自增主键； :ip，代理ip（"0.0.0.0"）； :port，代理ip端口（0）； :anonymous，是否为高匿，0为否，1为是，3为其他（3）； :type，0为HTTP，1为HTTPS，2为都可，3为其他（3）； :physical_address，代理ip的地理位置（"unknown"）； :response_time，响应时间，单位为ms（-1）； :last_verify，上次验证该代理ip的时间（"1000-01-01 00:00:00"）； :grade，该代理ip的分数。每验证成功一次会加分，失败则扣分，扣到0会被删除，在爬取ip时不会使用到（100）； :created，该行创建时间。""" cur, conn = mysql_api_ipool.cur, mysql_api_ipool.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", grade TINYINT UNSIGNED NOT NULL DEFAULT 100 COMMENT "grade of ip, used for validation", 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 Exception( 'MyPythonLib.mysql.mysql_api_ipool.check_table failed') def save_ips(self, ips_dict_list: list): """存入多个ip（以字典的列表形式提交）。注意这里只是对不存在的数据添加默认值，没有验证数据类型等。 :param ips_dict_list: ip字典的列表，每个ip是一个字典。""" cur, conn, log = self.cur, mysql_api_ipool.conn, self.log if len(ips_dict_list) == 0: message = 'no ip submitted' log.info(message) return command = '''INSERT INTO ipool (ip,port,anonymous,type,physical_address,response_time,last_verify) VALUES ''' for ip_dict in ips_dict_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'] physical_address = str_into_mysql(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) log.info(message) command = command[:-1] + ';' try: cur.execute(command) conn.commit() message = 'saving ips successfully' log.info(message) except: conn.rollback() log.info('错误命令为：' + command) raise Exception( 'MyPythonLib.mysql.mysql_api_ipool.save_ips failed') def check_repeat(self): """检查爬到的数据里有没有重复的，只检查ip和port都相同的。 如果有重复的，保留no最小的一条记录。""" cur, conn, log = mysql_api_ipool.cur, mysql_api_ipool.conn, self.log 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' log.info(message) return else: message = 'found repeated ip:port {} kinds'.format( len(repeated_items)) log.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 MIN(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. ' log.info(message) except: conn.rollback() raise Exception( 'MyPythonLib.mysql.mysql_api_ipool.check_repeat failed') @staticmethod def get_ips(random: bool = False, anonymous: list = [0, 1, 2, 3], type_: list = [0, 1, 2, 3], response_time: int = 30000, last_verify_interval: str = '48:00:00', grade: int = 100, limit: int = 1000) -> tuple: """根据要求返回多个ip，只返回ip和port。 :param random: 是否随机返回ip :param anonymous: 以列表形式指定匿名性 :param type_: 对HTTP还是HTTPS或其他代理 :param response_time: 响应时间在多少ms之内 :param last_verify_interval: 用于指定上次验证的时间距现在不超过多久，若为None，则是不限定验证时间 :param grade: ip的分数必须 >= grade :param limit: 最多返回几条代理ip :return tuple('ip:port','ip:port', ...)""" cur = mysql_api_ipool.cur cur.execute('USE ipool;') anonymous, type_ = tuple(anonymous), tuple(type_) if last_verify_interval == None: last_verify_interval = '99999999:00:00' command = ''' SELECT ip,port FROM ipool WHERE anonymous IN {} AND type IN {} AND response_time BETWEEN 0 AND {} AND (NOW()-last_verify)<"{}" AND grade >= {} ORDER BY response_time,last_verify DESC LIMIT {}; '''.format(anonymous, type_, response_time, last_verify_interval, grade, limit) if random: command = ''' SELECT ip,port FROM (SELECT ip,port FROM ipool WHERE anonymous IN {} AND type IN {} AND response_time BETWEEN 0 AND {} AND (NOW()-last_verify)<"{}" AND grade >= {} ) AS a ORDER BY RAND() LIMIT {}; '''.format(anonymous, type_, response_time, last_verify_interval, grade, limit) command = command.replace('\n', '') command = command.replace(' ', '') try: cur.execute(command) except: raise Exception('MyPythonLib.mysql.mysql_api_ipool.get_ips failed') ips = cur.fetchall() ips = tuple( [ips[i][0] + ':' + str(ips[i][1]) for i in range(len(ips))]) return ips def get_to_be_validated_ips(self, last_verify_interval: str = '48:00:00', limit: int = -1): """返回需要验证的ip，只返回no、ip和port。 :param last_verify_interval: 用于指定上次验证的时间距现在大于多久 :param limit: 最多返回几条代理ip，为-1时全部取出 :return tuple((no,ip,port,garde), (no,ip,port,garde), ...)""" cur, log = self.cur, self.log cur.execute('USE ipool;') command = ''' SELECT no,ip,port,grade FROM ipool WHERE (NOW()-last_verify)>"{}" ORDER BY last_verify,no LIMIT {}; '''.format(last_verify_interval, limit) command = command.replace('\n', '') command = command.replace(' ', '') if limit == -1: command = command.replace(' LIMIT -1', '') try: cur.execute(command) except: raise Exception( 'MyPythonLib.mysql.mysql_api_ipool.get_to_be_validated_ips failed' ) message = 'get {} to_be_validated ips'.format(limit) log.info(message) return cur.fetchall() def update_ip(self, ip_dict: dict): '''根据ip_dict更新数据，ip_dict中必须包含no、ip、port，另外还应该有至少一个需要更改的列，未提到的列不更改。''' cur, conn, log = self.cur, mysql_api_ipool.conn, self.log try: no, ip, port = ip_dict['no'], ip_dict['ip'], ip_dict['port'] except: raise Exception( 'MyPythonLib.mysql.mysql_api_ipool.update_ip failed: ip_dict must include no,ip,port' ) if len(ip_dict) <= 3: raise Exception( 'MyPythonLib.mysql.mysql_api_ipool.update_ip failed: at least 1 column must be updated' ) cur.execute('USE ipool;') command = 'SELECT no,ip,port FROM ipool WHERE no={};'.format(no) cur.execute(command) results = cur.fetchall() if len(results) == 0: message = 'ip no.{} not exists'.format(no) log.warning(message) return if ip != results[0][1] or port != results[0][2]: message = 'ip no.{} ip:port not right: {}:{}, recieved {}:{}, update given up'.format( no, results[0][1], results[0][2], ip, port) log.error(message) return command = '''UPDATE ipool SET ''' if 'anonymous' in ip_dict.keys(): command += 'anonymous={},'.format(ip_dict['anonymous']) if 'type' in ip_dict.keys(): command += 'type={},'.format(ip_dict['type']) if 'physical_address' in ip_dict.keys(): command += 'physical_address="{}",'.format( str_into_mysql(ip_dict['physical_address'])) if 'response_time' in ip_dict.keys(): command += 'response_time={},'.format(ip_dict['response_time']) if 'last_verify' in ip_dict.keys(): command += 'last_verify="{}",'.format(ip_dict['last_verify']) if 'grade' in ip_dict.keys(): command += 'grade={},'.format(ip_dict['grade']) command = command[:-1] + ' WHERE no={};'.format(no) try: cur.execute(command) conn.commit() message = 'update ip no.{} successfully'.format(no) log.info(message) except: conn.rollback() log.info('错误命令为：' + command) raise Exception( 'MyPythonLib.mysql.mysql_api_ipool.update_ip failed') 3.创建并修改scrapy项目本次使用到的提供代理ip的网站有12个，如下：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/ 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 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其中 https://www.kuaidaili.com 页数很多（最早的数据是2014年的，有4000多页），但用scrapy会封ip，所以只爬前2页。另外，还有3个可提供代理ip的网站：http://proxy-list.org/english/ https://www.proxynova.com/proxy-server-list/ http://spys.one/en/anonymous-proxy-list/这三个网站使用js生成代理ip表格，爬取它们最简单的方法是使用selenium，但它们提供的代理ip不多，且proxynova可用 https://raw.githubusercontent.com/sunny9577/proxy-scraper/master/proxies.json 替代，所以这里就不爬了。（1）其他配置使用cmd，进行如下操作：D: cd D:\spiders\scrapy scrapy startproject ipool在资源管理器中打开 D:\spiders\scrapy\ipool\ipool ，修改一下4个文件，修改settings.py为：# Scrapy settings for ipool project # # For simplicity, this file contains only settings considered important or # commonly used. You can find more settings consulting the documentation: # # https://docs.scrapy.org/en/latest/topics/settings.html # https://docs.scrapy.org/en/latest/topics/downloader-middleware.html # https://docs.scrapy.org/en/latest/topics/spider-middleware.html LOG_LEVEL = 'ERROR' # 只打印错误信息 BOT_NAME = 'ipool' SPIDER_MODULES = ['ipool.spiders'] NEWSPIDER_MODULE = 'ipool.spiders' # Crawl responsibly by identifying yourself (and your website) on the user-agent #USER_AGENT = 'ipool (+http://www.yourdomain.com)' # Obey robots.txt rules ROBOTSTXT_OBEY = False #不遵守robots.txt # Configure maximum concurrent requests performed by Scrapy (default: 16) #CONCURRENT_REQUESTS = 32 # Configure a delay for requests for the same website (default: 0) # See https://docs.scrapy.org/en/latest/topics/settings.html#download-delay # See also autothrottle settings and docs #DOWNLOAD_DELAY = 3 # The download delay setting will honor only one of: #CONCURRENT_REQUESTS_PER_DOMAIN = 16 #CONCURRENT_REQUESTS_PER_IP = 16 # Disable cookies (enabled by default) #COOKIES_ENABLED = False # Disable Telnet Console (enabled by default) #TELNETCONSOLE_ENABLED = False # Override the default request headers: #DEFAULT_REQUEST_HEADERS = { # 'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8', # 'Accept-Language': 'en', #} # Enable or disable spider middlewares # See https://docs.scrapy.org/en/latest/topics/spider-middleware.html #SPIDER_MIDDLEWARES = { # 'ipool.middlewares.IpoolSpiderMiddleware': 543, #} # Enable or disable downloader middlewares # See https://docs.scrapy.org/en/latest/topics/downloader-middleware.html DOWNLOADER_MIDDLEWARES = { 'ipool.middlewares.IpoolDownloaderMiddleware': 543, } # Enable or disable extensions # See https://docs.scrapy.org/en/latest/topics/extensions.html #EXTENSIONS = { # 'scrapy.extensions.telnet.TelnetConsole': None, #} # Configure item pipelines # See https://docs.scrapy.org/en/latest/topics/item-pipeline.html ITEM_PIPELINES = {'ipool.pipelines.MysqlPipeline': 1} # Enable and configure the AutoThrottle extension (disabled by default) # See https://docs.scrapy.org/en/latest/topics/autothrottle.html #AUTOTHROTTLE_ENABLED = True # The initial download delay #AUTOTHROTTLE_START_DELAY = 5 # The maximum download delay to be set in case of high latencies #AUTOTHROTTLE_MAX_DELAY = 60 # The average number of requests Scrapy should be sending in parallel to # each remote server #AUTOTHROTTLE_TARGET_CONCURRENCY = 1.0 # Enable showing throttling stats for every response received: #AUTOTHROTTLE_DEBUG = False # Enable and configure HTTP caching (disabled by default) # See https://docs.scrapy.org/en/latest/topics/downloader-middleware.html#httpcache-middleware-settings #HTTPCACHE_ENABLED = True #HTTPCACHE_EXPIRATION_SECS = 0 #HTTPCACHE_DIR = 'httpcache' #HTTPCACHE_IGNORE_HTTP_CODES = [] #HTTPCACHE_STORAGE = 'scrapy.extensions.httpcache.FilesystemCacheStorage' pipelines.py为：# Define your item pipelines here # # Don't forget to add your pipeline to the ITEM_PIPELINES setting # See: https://docs.scrapy.org/en/latest/topics/item-pipeline.html # useful for handling different item types with a single interface from itemadapter import ItemAdapter from MyPythonLib.mysql import mysql_api_ipool import os class MysqlPipeline: def open_spider(self, spider): self.mysql_api_ipool = mysql_api_ipool() def process_item(self, item, spider): self.mysql_api_ipool.save_ips(item['ips_dict_list']) return item def close_spider(self, spider): self.mysql_api_ipool.check_repeat()middlewares.py为：# Define here the models for your spider middleware # # See documentation in: # https://docs.scrapy.org/en/latest/topics/spider-middleware.html from scrapy import signals # useful for handling different item types with a single interface from itemadapter import is_item, ItemAdapter # 爬虫中间件 class IpoolSpiderMiddleware: # Not all methods need to be defined. If a method is not defined, # scrapy acts as if the spider middleware does not modify the # passed objects. @classmethod def from_crawler(cls, crawler): # This method is used by Scrapy to create your spiders. s = cls() crawler.signals.connect(s.spider_opened, signal=signals.spider_opened) return s def process_spider_input(self, response, spider): # Called for each response that goes through the spider # middleware and into the spider. # Should return None or raise an exception. return None def process_spider_output(self, response, result, spider): # Called with the results returned from the Spider, after # it has processed the response. # Must return an iterable of Request, or item objects. for i in result: yield i def process_spider_exception(self, response, exception, spider): # Called when a spider or process_spider_input() method # (from other spider middleware) raises an exception. # Should return either None or an iterable of Request or item objects. pass def process_start_requests(self, start_requests, spider): # Called with the start requests of the spider, and works # similarly to the process_spider_output() method, except # that it doesn’t have a response associated. # Must return only requests (not items). for r in start_requests: yield r def spider_opened(self, spider): spider.logger.info('Spider opened: %s' % spider.name) import random # 下载中间件 class IpoolDownloaderMiddleware: # Not all methods need to be defined. If a method is not defined, # scrapy acts as if the downloader middleware does not modify the # passed objects. UA_list = [ 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/105.0.0.0 Safari/537.36', 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/105.0.0.0 Safari/537.36 Edg/105.0.1343.33', 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:104.0) Gecko/20100101 Firefox/104.0' ] @classmethod def from_crawler(cls, crawler): # This method is used by Scrapy to create your spiders. s = cls() crawler.signals.connect(s.spider_opened, signal=signals.spider_opened) return s # 拦截无异常的请求，可进行UA伪装 def process_request(self, request, spider): # Called for each request that goes through the downloader # middleware. # Must either: # - return None: continue processing this request # - or return a Response object # - or return a Request object # - or raise IgnoreRequest: process_exception() methods of # installed downloader middleware will be called request.headers['User-Agent'] = random.choice(self.UA_list) return None # 拦截所有响应，可用于爬动态加载的数据 def process_response(self, request, response, spider): # Called with the response returned from the downloader. # Must either; # - return a Response object # - return a Request object # - or raise IgnoreRequest return response # 拦截发生异常的请求，可进行代理ip设置 def process_exception(self, request, exception, spider): # Called when a download handler or a process_request() # (from other downloader middleware) raises an exception. # Must either: # - return None: continue processing this exception # - return a Response object: stops process_exception() chain # - return a Request object: stops process_exception() chain pass def spider_opened(self, spider): spider.logger.info('Spider opened: %s' % spider.name) items.py为：# Define here the models for your scraped items # # See documentation in: # https://docs.scrapy.org/en/latest/topics/items.html import scrapy class IpoolItem(scrapy.Item): # define the fields for your item here like: # name = scrapy.Field() ips_dict_list = scrapy.Field() （2）爬虫配置要爬12个网站，所以需要写12个爬虫。其实有几个网站的数据格式差不多，这些爬虫还能再精简合并一下。这些爬虫都放在 D:\spiders\scrapy\ipool\ipool\spiders 目录下。a.a2u.pyimport scrapy, re from ipool.items import IpoolItem class A2uSpider(scrapy.Spider): name = 'a2u' start_urls = [ 'https://raw.githubusercontent.com/a2u/free-proxy-list/master/free-proxy-list.txt' ] def parse(self, response): ip_nodes = re.findall( r'[1-2]{0,1}\d{0,2}\.[1-2]{0,1}\d{0,2}\.[1-2]{0,1}\d{0,2}\.[1-2]{0,1}\d{0,2}:\d{1,5}', response.text) ips_dict_list = [] #这是一个字典的列表，和mysql_api.save_ips里的格式匹配 for ip_node in ip_nodes: ip_dict = {} ip, port = ip_node.split(':') port = int(port) ip_dict['ip'] = ip ip_dict['port'] = port ips_dict_list.append(ip_dict) item = IpoolItem() item['ips_dict_list'] = ips_dict_list yield itemb.clarketm.pyimport scrapy, re from ipool.items import IpoolItem class ClarketmSpider(scrapy.Spider): name = 'clarketm' start_urls = [ 'https://raw.githubusercontent.com/clarketm/proxy-list/master/proxy-list.txt' ] def parse(self, response): ip_nodes = re.findall( r'[1-2]{0,1}\d{0,2}\.[1-2]{0,1}\d{0,2}\.[1-2]{0,1}\d{0,2}\.[1-2]{0,1}\d{0,2}:\d{1,5}', response.text) ips_dict_list = [] #这是一个字典的列表，和mysql_api.save_ips里的格式匹配 for ip_node in ip_nodes: ip_dict = {} ip, port = ip_node.split(':') ip = ip.strip() port = int(port.strip()) ip_dict['ip'] = ip ip_dict['port'] = port ips_dict_list.append(ip_dict) item = IpoolItem() item['ips_dict_list'] = ips_dict_list yield item c.cool_proxy.pyimport scrapy, json, time from ipool.items import IpoolItem class CoolProxySpider(scrapy.Spider): name = 'cool_proxy' start_urls = ['https://www.cool-proxy.net/proxies.json'] def parse(self, response): ip_nodes = json.loads(response.text.replace('\'', '\"')) ips_dict_list = [] #这是一个字典的列表，和mysql_api.save_ips里的格式匹配 for ip_node in ip_nodes: ip_dict = {} ip = ip_node['ip'].strip() port = int(ip_node['port']) anonymous = (0, 1)[1 == ip_node['anonymous']] type_ = 3 physical_address = ip_node['country_name'].strip() response_time = int(1000 * float(ip_node['response_time_average'])) last_verify = float(ip_node['update_time']) last_verify = time.localtime(last_verify) last_verify = time.strftime('%Y-%m-%d %H:%M:%S', last_verify) 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_dict_list.append(ip_dict) item = IpoolItem() item['ips_dict_list'] = ips_dict_list yield item d.free_proxy_list.pyimport scrapy, time from ipool.items import IpoolItem class FreeProxyListSpider(scrapy.Spider): name = 'free_proxy_list' start_urls = ['https://free-proxy-list.net/'] def parse(self, response): ip_nodes = response.xpath( r'//*[@id="list"]/div/div[2]/div/table/tbody/tr') ips_dict_list = [] #这是一个字典的列表，和mysql_api.save_ips里的格式匹配 for ip_node in ip_nodes: ip_dict = {} ip = ip_node.xpath('./td[1]/text()').extract()[0].strip() port = int(ip_node.xpath('./td[2]/text()').extract()[0].strip()) anonymous = ( 0, 1)['anonymous' in ip_node.xpath('./td[5]/text()').extract()[0]] # (a,b)[条件] 也是python三元表达式的一种写法。注意条件为真时，表达式整体值为b type_ = (0, 1)['yes' in ip_node.xpath('./td[7]/text()').extract()[0]] physical_address = ip_node.xpath( './td[4]/text()').extract()[0].strip() response_time = 1000 last_verify = time.strftime('%Y-%m-%d %H:%M:%S') 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_dict_list.append(ip_dict) item = IpoolItem() item['ips_dict_list'] = ips_dict_list yield iteme.ip3366.pyimport scrapy from ipool.items import IpoolItem class Ip3366Spider(scrapy.Spider): name = 'ip3366' #allowed_domains = ['www.ip3366.net'] # 允许的域名。如果域名在此之外，则不会爬取 start_urls = [ 'http://www.ip3366.net/free/', 'http://www.ip3366.net/free/?stype=2' ] # 起始url的列表 base_url = 'http://www.ip3366.net/free/' def parse(self, response): ip_nodes = response.xpath(r'//*[@id="list"]/table/tbody/tr') ips_dict_list = [] #这是一个字典的列表，和mysql_api.save_ips里的格式匹配 for ip_node in ip_nodes: ip_dict = {} ip = ip_node.xpath(r'./td[1]/text()').extract()[0].strip() port = int(ip_node.xpath(r'./td[2]/text()').extract()[0].strip()) anonymous = ( 0, 1)['高匿' in ip_node.xpath(r'./td[3]/text()').extract()[0]] # (a,b)[条件] 也是python三元表达式的一种写法。注意条件为真时，表达式整体值为b type_ = ( 0, 1)['HTTPS' in ip_node.xpath(r'./td[4]/text()').extract()[0]] physical_address = ip_node.xpath( r'./td[5]/text()').extract()[0].strip() response_time = 1000 * int( ip_node.xpath(r'./td[6]/text()').extract()[0].strip().replace( '秒', '')) response_time = (response_time, 250)[response_time == 0] #这个网站中有很多代理ip的延迟为0，不太准确，一律改成250即0.25s last_verify = ip_node.xpath( r'./td[7]/text()').extract()[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_dict_list.append(ip_dict) item = IpoolItem() item['ips_dict_list'] = ips_dict_list yield item next_url = response.xpath( r'//*[@id="listnav"]/ul/a[text()="下一页"]/@href').extract() if next_url != None and len(next_url) != 0: next_url = Ip3366Spider.base_url + next_url[0] yield scrapy.Request(url=next_url, callback=self.parse) # 回调函数是自己，即递归解析网页 # 本方法在爬虫结束时被调用 def closed(self, spider): passf.ipaddress.pyimport scrapy from ipool.items import IpoolItem class IpaddressSpider(scrapy.Spider): name = 'ipaddress' start_urls = ['https://www.ipaddress.com/proxy-list/'] def parse(self, response): ip_nodes = response.xpath(r'/html/body/div[1]/main/table/tbody/tr') ips_dict_list = [] #这是一个字典的列表，和mysql_api.save_ips里的格式匹配 for ip_node in ip_nodes: ip_dict = {} ip = ip_node.xpath('./td[1]/a[1]/text()').extract()[0].strip() port = int( ip_node.xpath('./td[1]/text()').extract()[0].strip().replace( '"', '').replace(':', '')) anonymous = ( 0, 1)['anonymous' in ip_node.xpath('./td[2]/text()').extract()[0]] # (a,b)[条件] 也是python三元表达式的一种写法。注意条件为真时，表达式整体值为b type_ = 3 physical_address = ip_node.xpath( './td[3]/text()').extract()[0].strip()[3:] response_time = 1000 last_verify = ip_node.xpath( './td[4]/text()').extract()[0].strip() + ':00' 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_dict_list.append(ip_dict) item = IpoolItem() item['ips_dict_list'] = ips_dict_list yield itemg.jiangxianli.pyimport scrapy, json from ipool.items import IpoolItem class JiangxianliSpider(scrapy.Spider): name = 'jiangxianli' start_urls = ['https://ip.jiangxianli.com/api/proxy_ips?page=1'] def parse(self, response): ip_nodes = json.loads(response.text)['data']['data'] ips_dict_list = [] #这是一个字典的列表，和mysql_api.save_ips里的格式匹配 for i in range(len(ip_nodes)): ip_dict = {} ip = ip_nodes[i]['ip'].strip() port = int(ip_nodes[i]['port'].strip()) anonymous = (0, 1)[2 == ip_nodes[i]['anonymity']] type_ = (0, 1)['https' == ip_nodes[i]['protocol']] physical_address = ip_nodes[i]['ip_address'].strip() response_time = ip_nodes[i]['speed'] last_verify = ip_nodes[i]['validated_at'].strip() 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_dict_list.append(ip_dict) item = IpoolItem() item['ips_dict_list'] = ips_dict_list yield item next_url = json.loads(response.text)['data']['next_page_url'] if next_url != None: yield scrapy.Request(url=next_url, callback=self.parse) h.kuaidaili.pyimport scrapy from ipool.items import IpoolItem class KuaidailiSpider(scrapy.Spider): name = 'kuaidaili' #allowed_domains = ['www.kuaidaili.com'] start_urls = [ 'https://free.kuaidaili.com/free/inha/1/', 'https://free.kuaidaili.com/free/intr/1/' ] base_url = 'https://free.kuaidaili.com' custom_settings = { 'CONCURRENT_REQUESTS': 1, 'DOWNLOAD_DELAY': 10, 'CONCURRENT_REQUESTS_PER_DOMAIN': 1 } def parse(self, response): ip_nodes = response.xpath(r'//*[@id="list"]/table/tbody/tr') ips_dict_list = [] #这是一个字典的列表，和mysql_api.save_ips里的格式匹配 for ip_node in ip_nodes: ip_dict = {} ip = ip_node.xpath(r'./td[1]/text()').extract()[0].strip() port = int(ip_node.xpath(r'./td[2]/text()').extract()[0].strip()) anonymous = ( 0, 1)['高匿' in ip_node.xpath(r'./td[3]/text()').extract()[0]] # (a,b)[条件] 也是python三元表达式的一种写法。注意条件为真时，表达式整体值为b type_ = ( 0, 1)['HTTPS' in ip_node.xpath(r'./td[4]/text()').extract()[0]] physical_address = ip_node.xpath( r'./td[5]/text()').extract()[0].strip() response_time = 1000 * float( ip_node.xpath(r'./td[6]/text()').extract()[0].strip().replace( '秒', '')) last_verify = ip_node.xpath(r'./td[7]/text()').extract()[0].strip() 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_dict_list.append(ip_dict) item = IpoolItem() item['ips_dict_list'] = ips_dict_list yield item page_num = response.url.split('/')[-2] next_url = response.url.replace(page_num, str(int(page_num) + 1)) other_urls_list = response.xpath( r'//*[@id="listnav"]/ul/li/a/@href').extract() other_urls_list = [ KuaidailiSpider.base_url + other_urls_list[i] for i in range(len(other_urls_list)) ] if int(page_num) < 5 and next_url in other_urls_list: yield scrapy.Request(url=next_url, callback=self.parse) custom_settings 用于指定这个爬虫的个性化设置，本例中限制了爬取的频率；并且只爬前5页。i.pubproxy.pyimport scrapy, re from ipool.items import IpoolItem class PubproxySpider(scrapy.Spider): name = 'pubproxy' start_urls = [ 'http://pubproxy.com/api/proxy?limit=5&format=txt&type=http&level=anonymous&last_check=60&no_country=CN' ] def parse(self, response): ip_nodes = re.findall( r'[1-2]{0,1}\d{0,2}\.[1-2]{0,1}\d{0,2}\.[1-2]{0,1}\d{0,2}\.[1-2]{0,1}\d{0,2}:\d{1,5}', response.text) ips_dict_list = [] #这是一个字典的列表，和mysql_api.save_ips里的格式匹配 for ip_node in ip_nodes: ip_dict = {} ip, port = ip_node.split(':') port = int(port) ip_dict['ip'] = ip ip_dict['port'] = port ips_dict_list.append(ip_dict) item = IpoolItem() item['ips_dict_list'] = ips_dict_list yield itemj.rmccurdy.pyimport scrapy, re from ipool.items import IpoolItem class RmccurdySpider(scrapy.Spider): name = 'rmccurdy' start_urls = ['https://www.rmccurdy.com/.scripts/proxy/good.txt'] def parse(self, response): ip_nodes = re.findall( r'[1-2]{0,1}\d{0,2}\.[1-2]{0,1}\d{0,2}\.[1-2]{0,1}\d{0,2}\.[1-2]{0,1}\d{0,2}:\d{1,5}', response.text) ips_dict_list = [] #这是一个字典的列表，和mysql_api.save_ips里的格式匹配 for ip_node in ip_nodes: ip_dict = {} ip, port = ip_node.split(':') port = int(port) ip_dict['ip'] = ip ip_dict['port'] = port ips_dict_list.append(ip_dict) item = IpoolItem() item['ips_dict_list'] = ips_dict_list yield item k.sunny9577.pyimport scrapy, json from ipool.items import IpoolItem class Sunny9577Spider(scrapy.Spider): name = 'sunny9577' start_urls = [ 'https://raw.githubusercontent.com/sunny9577/proxy-scraper/master/proxies.json' ] def parse(self, response): ip_nodes = json.loads(response.text)['proxynova'] ips_dict_list = [] #这是一个字典的列表，和mysql_api.save_ips里的格式匹配 for ip_node in ip_nodes: ip_dict = {} ip = ip_node['ip'].strip() port = int(ip_node['port'].strip()) anonymous = (0, 1)['Anonymous' in ip_node['anonymity']] # (a,b)[条件] 也是python三元表达式的一种写法。注意条件为真时，表达式整体值为b type_ = 3 physical_address = ip_node['country'] ip_dict['ip'] = ip ip_dict['port'] = port ip_dict['anonymous'] = anonymous ip_dict['type'] = type_ ip_dict['physical_address'] = physical_address ips_dict_list.append(ip_dict) item = IpoolItem() item['ips_dict_list'] = ips_dict_list yield item l.TheSpeedX.pyimport scrapy, re from ipool.items import IpoolItem class ThespeedxSpider(scrapy.Spider): name = 'TheSpeedX' start_urls = [ 'https://raw.githubusercontent.com/TheSpeedX/SOCKS-List/master/http.txt', 'https://raw.githubusercontent.com/TheSpeedX/SOCKS-List/master/socks5.txt', 'https://raw.githubusercontent.com/TheSpeedX/SOCKS-List/master/socks4.txt' ] def parse(self, response): ip_nodes = re.findall( r'[1-2]{0,1}\d{0,2}\.[1-2]{0,1}\d{0,2}\.[1-2]{0,1}\d{0,2}\.[1-2]{0,1}\d{0,2}:\d{1,5}', response.text) ips_dict_list = [] #这是一个字典的列表，和mysql_api.save_ips里的格式匹配 for ip_node in ip_nodes: ip_dict = {} ip, port = ip_node.split(':') ip = ip.strip() port = int(port.strip()) ip_dict['ip'] = ip ip_dict['port'] = port ips_dict_list.append(ip_dict) item = IpoolItem() item['ips_dict_list'] = ips_dict_list yield item （3）运行爬虫这里涉及到了多个爬虫，运行的时候一个个使用 scrapy crawl 爬虫名 比较麻烦。scrapy提供了其他方法，可同时运行一个项目中的多个爬虫或全部爬虫。以下内容参考了 这篇文章 。在 D:\spiders\scrapy\ipool\ipool 下新建一个py文件，比如 crawl_all.py ，内容为：from scrapy.crawler import CrawlerProcess from scrapy.utils.project import get_project_settings from scrapy.spiderloader import SpiderLoader # 根据项目配置获取 CrawlerProcess 实例 process = CrawlerProcess(get_project_settings()) # 获取 spiderloader 对象，以进一步获取项目下所有爬虫名称 spider_loader = SpiderLoader(get_project_settings()) # 添加全部爬虫 for spidername in spider_loader.list(): process.crawl(spidername) # 添加一个爬虫 #process.crawl('爬虫名') # 执行 process.start()如上所述，添加一个爬虫就用 process.crawl('爬虫名') ，而 spider_loader.list() 方法可获取项目中的所有爬虫名，两者搭配就可运行全部爬虫。另外需要提醒，笔者爬取的几个网站中有几个不能直接访问，比如 raw.githubusercontent.com 那几个，你可以把它们去掉，或者使用一些设置。如果需要使用V2rayN并使用requests库或scrapy，可查看 Docker实践——如何查阅wiki百科及V2Ray ，其中提到了如何使用V2rayN并运行爬虫。最后，在cmd中进入 D:\spiders\scrapy\ipool\ipool ，运行 python crawl_all.py 即可。4.验证ip原本想使用树莓派多进程验证ip，但占用树莓派资源太高，改在win10电脑上验证。在win10电脑的自建库目录中，除了 log.py 、 mysql.py ，还需要添加一个spider.py，文件内容如下：#!/usr/bin/python # -*- coding: UTF-8 -*- import requests, random 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' } def get_session(max_retries: int = 3) -> requests.Session: '''接受最大重连数，返回一个requests会话对象 :param max_retries: 最大重连次数 :return 会话对象''' # 设置重连次数。包括开头，若出现异常，最多尝试连接4次 try: session = requests.Session() session.mount('http://', requests.adapters.HTTPAdapter(max_retries=max_retries)) session.mount('https://', requests.adapters.HTTPAdapter(max_retries=max_retries)) return session except: raise Exception('error: MyPythonLib.spider.get_session failed ') def multiprocess_job_spliter(process_num: int, to_do_list, random_shuffle: bool = True) -> dict: '''给出进程数、总任务序列，返回分好的任务列表。 :param process_num: 进程数（需要将总任务序列分为多少份） :param to_do_list: 总任务序列（元组或列表），比如待爬取的所有url字符串 :param random_shuffle: 是否打乱总任务序列的顺序 :return {0: todo_list1, 1: todo_list2, ... }''' try: to_do_list = list(to_do_list) if random_shuffle: random.shuffle(to_do_list) total_num = len(to_do_list) one_share = total_num // process_num share_list = {} for i in range(0, process_num): if i != process_num - 1: share_list[i] = to_do_list[(i * one_share):((i + 1) * one_share)] elif i == process_num - 1: share_list[i] = to_do_list[(i * one_share):] return share_list except: raise Exception( 'error: MyPythonLib.spider.multiprocess_job_spliter failed ')这个文件后面还有很多内容，但实际上只会用到 multiprocess_job_spliter 这个函数。然后在一个合适的地方放python脚本，比如就放在scrapy的ipool项目中： D:\spiders\scrapy\ipool\ipool\update_ips.py ，文件内容如下：#!/usr/bin/python # -*- coding: UTF-8 -*- from MyPythonLib import mysql, spider from MyPythonLib import log as mylog import logging, os, time, multiprocessing headers = spider.headers class update_ips(): test_http_url = 'http://example.org/' test_https_url = 'https://example.org/' def __init__(self, log: logging.Logger = None): """:param log: 可选的log对象，若不提供，将在当前文件夹下新建logs目录、创建log文件""" if log == None: log_dir = os.path.join(os.getcwd(), 'logs') if not os.path.isdir(log_dir): os.mkdir(log_dir) log_file_name = 'mysql_api_ipool_log#{}.txt'.format( time.strftime('-%Y-%m-%d_%H_%M_%S')) log = mylog.get_logger(log_dir, log_file_name, 'mysql_api_ipool') self.log = log message = 'ready to update ips' self.log.info(message) self.mysql_api_ipool = mysql.mysql_api_ipool(self.log) self.conn = self.mysql_api_ipool.conn self.cur = self.conn.cursor() self.mysql_api_ipool.check_table() message = 'database ipool and table ipool checked' self.log.info(message) self.cur.execute('USE ipool;') self.conn.commit() def __del__(self): self.cur.close() message = 'update ips complete' self.log.info(message) @staticmethod def validate_ips_single_process(ips_list: list, process_no: int = 0, grade_minus: int = 50): '''验证ip，只涉及type、last_verify、grade的改变。若grade<=0，会将这个ip删除。 :这个方法是单进程的。 :param ips_list: 待验证ip的列表，格式为 [(no,ip,port,grade), (no,ip,port,grade), ...] :param process_no: 进程编号，主要用于log文件取名 :param grade_minus: 若代理ip无法使用，则扣除该分数''' global headers log = mylog.get_logger( abs_path=os.path.join(os.getcwd(), 'logs'), log_file_name='validate_ips-{}'.format(process_no), getLogger_name=str(process_no)) message = 'received {} ips, validating...'.format(len(ips_list)) log.info(message) mysql_api_ipool = mysql.mysql_api_ipool(log) session = spider.get_session(max_retries=1) for no, ip, port, grade in ips_list: ip_port = '{}:{}'.format(ip, port) proxies = {'http': ip_port, 'https': ip_port} http_success, https_success = True, True try: session.get(url=update_ips.test_http_url, timeout=10, headers=headers, proxies=proxies) except: http_success = False try: session.get(url=update_ips.test_https_url, timeout=10, headers=headers, proxies=proxies) except: https_success = False if http_success and not https_success: type_ = 0 elif https_success and not http_success: type_ = 1 elif http_success and https_success: type_ = 2 elif not http_success and not https_success: type_ = 3 ip_dict = {} ip_dict['no'] = no ip_dict['ip'] = ip ip_dict['port'] = port ip_dict['type'] = type_ ip_dict['last_verify'] = time.strftime('%Y-%m-%d %H:%M:%S') if type_ == 3: grade -= grade_minus else: grade = 100 ip_dict['grade'] = grade mysql_api_ipool.update_ip(ip_dict=ip_dict) def validate_ips_multiprocess(self, process_num: int = 5, limit: int = -1, grade_minus: int = 50): '''验证ip，只涉及type、last_verify、grade的改变。若grade<=0，会将这个ip删除。 :这个方法是多进程的，将会调用 validate_ips_single_process 。 :param process_num: 验证ip的进程数 :param limit: 限制待验证ip的条数，为-1时表示全部取出 :param grade_minus: 若代理ip无法使用，则扣除该分数''' cur, conn, log = self.cur, self.conn, self.log command = 'DELETE FROM ipool WHERE grade<=0;' try: cur.execute(command) conn.commit() message = 'delete ips where grade<=0 successfully' log.info(message) except: conn.rollback() log.info('错误命令为：' + command) raise Exception( 'update_ips.validate_ips_multiprocess failed: delete error') ips_list = self.mysql_api_ipool.get_to_be_validated_ips(limit=limit) ips_list_dict = spider.multiprocess_job_spliter(process_num, ips_list) process_pool = multiprocessing.Pool(processes=process_num) for i in range(process_num): process_pool.apply_async(update_ips.validate_ips_single_process, args=(ips_list_dict[i], i, grade_minus)) process_pool.close() process_pool.join() try: cur.execute(command) conn.commit() message = 'delete ips where grade<=0 successfully' log.info(message) except: conn.rollback() log.info('错误命令为：' + command) raise Exception( 'update_ips.validate_ips_multiprocess failed: delete error') if __name__ == '__main__': update_ips().validate_ips_multiprocess(process_num=60, limit=-1, grade_minus=50) update_ips().validate_ips_multiprocess(process_num=60, limit=-1, grade_minus=50) 中， process_num 和 limit 可以自行修改，主要与win10机器的配置有关，验证ip的过程mysql读写频率不算高。另外，这里的验证ip只涉及了type（http还是https）、last_verify（最后验证时间）、grade（ip分数）的改变，如果想一同更新代理ip的匿名性、物理地址、响应时间，可以在这里的基础上自行更改。最后，在cmd中，使用：cd D:\spiders\scrapy\ipool\ipool python update_ips.py即可运行验证ip的py脚本。5.日常使用爬取ip和验证ip的脚本可以每天运行一次，顺序如下：（1）若需要使用V2rayN，先开启V2rayN。（2）在cmd中使用如下命令：cd D:\spiders\scrapy\ipool\ipool python crawl_all.py python update_ips.py即可。如果想把命令写成脚本、开机自动运行，可以参考 不同系统下的开机自动运行 。另外，mysql记得时常备份。使用代理ip时，用 MyPythonLib.mysql.mysql_api_ipool.get_ips 方法可取出ip。
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 。
步骤/目录： 1.再谈数据库基础概念 2.mysql基础命令 （1）SHOW与USE （2）HELP 3.检索数据 （1）SELECT （2）DISTINCT （3）LIMIT （4）ORDER BY （5）WHERE （6）IN （7）LIKE与通配符% （8）正则表达式 4.检索数据的进阶操作 （1）计算字段 （2）数据处理函数 （3）聚集函数 （4）数据分组 （5）子查询 （6）联结表 （7）其他联结概念 （8）组合查询 （9）全文本搜索 5.插入数据 6.更新删除数据 7.创建和操作表 8.视图 9.存储过程及其他 10.数据库维护 11.mysql使用的原则 12.python中使用mysql本文首发于个人博客https://lisper517.top/index.php/archives/43/，转载请注明出处。本文的目的是对MySQL进行进阶学习、总结。本文写作日期为2022年9月3日。本文主要整理自《MySQL必知必会》。1.再谈数据库基础概念数据库，从字面意思上理解就是存储数据的仓库。其实把爬虫爬取到的页面存成html放到一个文件夹里也算数据库，只是存取效率较慢。实际上，现在说的数据库一般指数据库管理系统（DBMS，database management system），相比于简单地把数据存成文件，DBMS的存取效率更高，占用空间更小。前一篇文章中提到，在数据库中的每一行数据应该有一项独特的数据区别于其他行，称为主键。实际上，主键也可以是几项数据的组合，比如说有一群人，存储他们的身高、体重两项数据，单独一项拿出来可能很容易有重复，但两项都相同的情况就会大大减少，大概就是这个意思。一般来说，主键确定以后就不要更新，即使某一行被删除、它的主键也不应该复用，主键的值在未来不应该更改。SQL即structured query language，与程序设计的C++、Python语言等不同，SQL的构成词少、相对简单，主要是为了操作数据而存在。很多数据库都支持SQL，比如MySQL就使用标准SQL语言，但是不同数据库软件间的语法可能有一些区别，mysql的语句就有可能无法在其他数据库中正常运行（除了mysql的同卵双胞胎mariadb）。MySQL的运行模式是C/S，即Client/Server，数据库存储在服务器上，其他机器通过mysql客户端访问（即使一台机器访问在本机运行的mysql服务器也要通过客户端）。mysql的客户端和服务器端使用同一套软件，就是说安装了mysql就同时安装了服务器端和客户端；或者也可以使用编程语言，如python的pymysql库当作客户端访问服务器端；或者有一些图形化的管理工具，可以一定程度上简化mysql操作。目前（2022年9月）MySQL的版本已经到8.0.29（指docker-mysql），但很多语句还是通用的。一般使用的mysql客户端就是mysql命令行，即你在安装了mysql后用 mysql -h 服务器ip -P 端口号 -uroot -p 进入的界面。mysql命令行中使用命令有两个注意事项：每行命令用 ; 或 \g 结尾后再按 enter键 才会执行（除了少部分简单的语句）；输入 exit 或 quit 可退出。另外还有一些图形化的界面，笔者在这里推荐 MySQL Workbench ，其下载页面为 https://dev.mysql.com/downloads/workbench/ ，它是mysql官方推出的工具（之前mysql被Oracle拿下了），功能强大。另外mysql现在还推出了MySQL Installer，可以一键管理各种mysql官方软件。最后，如果要在mysql命令行里清屏，使用 system clear 或 system cls ，前者适用于linux客户端、后者适用于cmd。2.mysql基础命令（1）SHOW与USESHOW DATABASES 显示所有数据库；USE 数据库名 使用某个数据库，后续操作都在该库中进行；SHOW TABLES 展示库中的表名；SHOW COLUMNS FROM 表名 或 DESCRIBE 表名 来查看一张表有哪些列，列的数据类型等信息，返回的表格式为：+--------------+--------------+------+-----+-------------------+-------------------+ | Field | Type | Null | Key | Default | Extra | +--------------+--------------+------+-----+-------------------+-------------------+分别表示 列名 ， 数据类型 ， 可否为NULL ， 是否主键 ， 默认值 ， 其他（自增，默认生成等）。SHOW CREATE TABLE 表名 可以查看创建表时的完整语句。SHOW 的其他命令还有 SHOW STATUS（查看mysql服务器状态） 、 SHOW GRANTS（查看所有mysql用户的权限） 、 SHOW ERRORS 和 SHOW WARNINGS（查看mysql服务器的错误和警告）。（2）HELP使用 HELP 其他命令名 可以查看mysql里的命令帮助手册，比如在忘记 SELECT 各种从句顺序时就可以使用 HELP SELECT。最后，以上所有的命令都使用大写，但实际上mysql不区分大小写。标准的规范中，应当尽量只把列、表名、值等小写；同一条命令可以拆成多行输入mysql命令行，便于阅读调试，最后加 ; 即可。3.检索数据（1）SELECT检索数据使用 SELECT 语句，一个简单的例子是：SELECT 列名 FROM 表名由于没有指定输出顺序，这里的返回的数据顺序可能有变化。列名也可以写多个，用 , 分隔：SELECT 列名1,列名2,列名3... FROM 表名或者用 * 表示输出所有列，即 SELECT * FROM 表名 。最后， SELECT 中的列名和表名还可以写成完全限定的形式：SELECT 表名.列名 FROM 库名.表名这在一些复杂的 SELECT 语句中比较有用，同时注意即使用 表名.列名 已经指定了表，还是需要 FROM 表名 来指定表，即 SELECT 表名.列名; 这样的写法无效。（2）DISTINCT用 DISTINCT 可修饰列名，只输出该列有多少种不同的值：SELECT DISTINCT 列名 FROM 表名DISTINCT 用于单独一列时可以理解为将相同的行合并后输出。 DISTINCT 修饰多个列名时， DISTINC 放在哪里都等效于放在最前面，即 DISTINCT 列名1,列名2... ，这时的一行数据，只有列出的 列名1,列名2... 都相同才会被当成同一条数据合并。（3）LIMIT用 LIMIT 可限制输出行数、从第几行后开始输出（不包括）。它的第一种用法为：SELECT 列名 FROM 表名 LIMIT n这里n为一个整数，表示最后返回的结果应该 <= n 行。如果写成 LIMIT m,n 表示从第 m 行后开始查找，输出 <= n 行。所以 LIMIT n 也可以理解为 LIMIT 0,n 。由于 LIMIT m,n 容易让人搞不清从第 m 行还是从第 n 行开始，mysql 5 之后的版本支持 LIMIT n OFFSET m 的写法， OFFSET 表示偏移量。（4）ORDER BY前面已经提到，不指定顺序时mysql返回的数据排序比较随意，这种顺序一般是表中数据在底层的存储顺序，尤其是在表中删除、新增过行后，mysql有一些重新利用空间的操作，底层顺序可能改变。要指定顺序，可以用 ORDER BY 从句：SELECT 列名1 FROM 表名 ORDER BY 条件这个条件一般可以用检索的列名，即 ORDER BY 列名1 ，但使用其他列对输出列排序也可以，即 ORDER BY 列名2 。如果该列是数字，就按数字大小排序（日期时间也是）；字符串则按字母表顺序。 ORDER BY 默认是升序输出的（大的数字或靠后的字母在下面）。ORDER BY 后面也可以接多个条件，比如 ORDER BY 列名1,列名2 ，那么mysql会先根据 列名1 排序，当两行的 列名1 相同时则按 列名2 排序。另外， ORDER BY 条件 DESC 可以指定降序输出。对于多个条件，如果写成 ORDER BY 列名1 DESC,列名2 则按 列名1 降序，按 列名2 升序，即 DESC 只对它紧跟着的前面的那个条件生效。升序也可以通过 ASC 显式指定，但一般不用写。如果聪明的读者，可能已经想到，用 ORDER BY 和 LIMIT 可以选出最大值或最小值。其实，mysql中有专门实现这项功能的函数，即 MAX() 和 MIN() ，这将在后面提及。最后， ORDER BY 可能区分大小写，也可能不区分（在ASCII表中小写字母排在大写字母前面），取决于数据库的设置（collate）。（5）WHERE用于指定条件。比如在 SELECT 中使用：SELECT 列名 FROM 表名 WHERE 条件WHERE 也可以和前面介绍的 ORDER BY 和 LIMIT 联用，如果不确定这三个关键字的顺序，可以用 HELP SELECT 查看。WHERE 条件中有以下的操作符：= 表示 等于， <> 和 != 都可以表示 不等于 ，其他的还有 < 、 > 、 <= 、 >= ，另外还有 BETWEEN 表示在两个值之间（包括两端）。WHERE 条件可以为多个，用 AND （ BETWEEN 也要用到 AND ）、 OR 、 NOT 连接。使用例子：WHERE 列名="字符串1" WHERE 列名!=123 WHERE 列名 BETWEEN 123 AND 234 WHERE NOT 列名=123 WHERE ((列名1="字符串1") AND (列名2!=123)) OR (列名3 BETWEEN 123 AND 234)在最后一条语句中，使用 () 明确指定三项条件的组合顺序。在默认情况下， AND 比 OR 优先。另外，如果要检查列值是否为NULL，使用 IS ：WHERE 列名 IS NULL WHERE 列名 IS NOT NULL WHERE NOT 列名 IS NULL这里提醒一下，NULL和数值0、空格（字符串）、空字符串（""）都不同，在创建表时请尽量禁止列的值可以为NULL，而是用0、空字符串等来代替，可提高检索性能。由于NULL值比较特殊，对于 != 等刚才提到的操作符的检索，它是不会出现的：WHERE 列名!=1 WHERE 列名!=1 OR 列名 IS NULL所以上面两行命令中，前者不会返回该列值为NULL的行，后者会。（6）IN刚才已经提到了 WHERE 子句中 AND 、 OR 和 NOT ，现在介绍一下 IN 。 IN 后面跟的可以理解为集合，比如：SELECT 列名 FROM 表名 WHERE 列名 IN (1,2,3,4)上面的例子也可以用4个 OR 来代替。实际上， IN 后面更多情况下跟的是其他 SELECT 语句（嵌套 SELECT ，即子查询，见后），来实现复杂功能。在（5）中提到过 NOT ，实际上 NOT 在mysql中用途广泛，比如：NOT IN IS NOT NOT BETWEEN CREATE TABLE IF NOT EXISTS 表名NOT 可以对各种条件取反。（7）LIKE与通配符%前面的 = 只适用于完全匹配、精确相等，如果查找字符串时只知道其中某一段，就可以用 LIKE 和通配符 % 、 _ 。如：SELECT 列名 FROM 表名 WHERE 列名 LIKE "%jet%"这里的 LIKE 替换了 = ， "%jet%" 被称为模式（pattern）。本例中，任何字符串只要包含连续的 jet 三个字母，就算是满足了 WHERE 从句中的条件。所以 % 可以替代任意个字符（包括0个）。如果写成 %jet 或 jet% ，则指定字符串必须以 jet 结尾或开头；若写成 je%t ，则指定字符串必须以 je 开头、以 t 结尾，中间可以有0~n个字符。另外，NULL值仍然是一个特例，不会被 LIKE "%" 匹配到。另一个匹配符是 _ ，它与 % 的唯一不同是只能匹配一个任意字符。由于通配符搜索较慢，使用时请谨慎，能用其他方法替代的就用其他方法。（8）正则表达式REGEXP 在mysql中用来支持re（regular expression）：SELECT 列名 FROM 表名 WEHRE 列名 REGEXP "模式"REGEXP 在这里和 LIKE 类似，也替换了 = 。另外，如果要让mysql的re区分大小写，用 REGEXP BINARY "模式" 。在mysql中写正则，用转义字符要写 \\ ，比如匹配 \ 就用 \\\ ，因为mysql和mysql的re库各需要用一个 \ ，这和python的re库类似。另外需要注意的是 LIKE 和 REGEXP 的小区别， REGEXP 默认只匹配字符串的一部分，除非用 ^ 和 $ 指定了开头结尾。关于re的更多内容，请参考 runoob教程 ，或查看笔者后续的文章。4.检索数据的进阶操作前面介绍的是一些基础操作，没有涉及 SELECT 嵌套或其他高级内容。下面介绍的进阶操作中，最关键也是最难的是子查询，联结和 UNION 。（1）计算字段如果mysql表中记录的内容不是直接需要的，比如要对表中数值做一定计算，或者把字符串进行大小写转换，然后把处理后的数据返回客户端，就叫做创建了计算字段。当然，也可以把表中的数据都传给客户端后再处理，比如用python时，使用pymysql取得数据、在python中进行计算，但这种方法的缺点是传输的数据更多（占用带宽），并且mysql服务器端处理起来一般效率更高（DBMS从设计时就考虑到了这些情况。即使服务器端的机器配置有时可能比客户端差，能在服务器端完成的操作也应该尽量在服务器端完成）。计算字段的特点是并没有实际存储在表中，它是由mysql服务器从已有的数据推导、计算出来的。当然，从客户端的角度，并没有办法区分返回的数据是实际存储的，还是计算字段。第一个计算字段的例子是拼接函数 CONCAT() （即 concatenate 。其他数据库软件可能使用 + 或者 || 操作符，mysql中使用函数）：SELECT CONCAT(列名1,"123",列名2,114.514) FROM ...在这个例子中，返回的值为 列值1 、 "123" 、 列值2 、 "114.514" 拼接后的结果（如果读者进行实践的话，注意这一列的名字，列名不存在于原来的表中）。使用 HELP CONCAT ，可看到mysql对这个函数行为的说明，主要特点有：如果所有参数都不是字符串的二进制表示，则返回的结果也是一个非二进制形式表示的字符串；只要参数中有一个是二进制字符串，返回的结果就也会是二进制字符串；如果参数是数值，它将被转换为非二进制字符串，字面意思和原来一样；只要其中一个参数为NULL，则返回的结果也为NULL。第二个函数的例子是 RTRIM() ，它用于删除字符串末尾的空格， LTRIM() 、 TRIM() 同理。刚才提到，计算字段得到的列，其列名不存在于原来的表中，有时可能显得有点丑陋。这个问题可以用别名解决，别名需要用到 AS 关键字：SELECT CONCAT(列名1,"123",列名2,114.514) AS 列的别名 FROM ...这时列名会变为指定的 列的别名 。当然， AS 不只是为了美观而存在，别名有其他的用处。别名（alias）又可称为导出列（derived column），因为它一般和原来的表中列名不同。计算字段中另外一种重要的计算是对数值的处理。如进行简单算数（乘法）：SELECT 列名1*列名2 FROM ...列名1 和 列名2 中的数值应该是数值型的。实际上， + 、 - 、 * 、 / 都是支持的。除了 CONCAT() 、3个 TRIM 函数，还有很多其他函数，将在后面一一介绍。（2）数据处理函数mysql虽然使用标准的SQL语言，但各个DBMS在函数上的区别比较大，有些函数可能功能相同，但在某些情况下行为不同，或者压根就没有一些函数。为了可移植性，在编写SQL时应该少用函数，或者用了函数以后写上详细的注释。函数可以分为4类，处理数字、字符串、日期时间的3类，以及返回DBMS信息的函数。最后，如果要对函数进行一些实验，可以使用 SELECT 函数(参数) ，就不需要到表里取数据计算了。最后，以下所有的函数，如果有不清楚的地方，可以用 HLEP 函数名 获得帮助。在字符串处理函数中，有如下函数与功能：UPPER() ，将字符串中的字母全部转换为大写； LOWER() 则相反。要注意这两个函数对二进制字符串无效（BINARY、VARBINARY和各种BLOB）。LEFT(str, n) ，返回str中最靠左的n个字符。 RIGHT() 同理。LENGTH() ，返回一个字符串占的字节数； CHAR_LENGTH() 则返回字符串中的字符个数。这两个函数在处理中文时有区别。LOCATE(substr, str, pos) ，其中 pos参数 可选，用于返回 str 中从左边开始数，第一次出现 substr 是第几个字符，如果不包含则返回0。若任意一个参数为NULL，则会返回NULL。SOUNDEX() ，用于返回一个字符串的SOUNDEX值（也是字符串），这个值用于描述字符串的发音。要注意这个功能主要是针对英语的，其他非英文字符会被当成元音处理。另外，对于utf-8等双字节的字符集，这个函数可能会有一些bug。SUBSTRING() ，用于返回字符串的子字符串，它有两种共四个形态： SUBSTRING(str, pos) （同 SUBSTRING(str FROM pos) ）和 SUBSTRING(str, pos, len) （同 SUBSTRING(str FROM pos FOR len) ），具体请用 HELP SUBSTRING 查看。在日期时间函数中，有下列函数与功能：ADDDATE() ，用于返回日期加某个时间后的日期，有两种形式： ADDDATE(原日期, INTERVAL 数字 单位) 和 ADDDATE(原日期，天数) ，前者如 ADDDATE('2020-01-01',INTERVAL 30 DAY) （DAY可以换成MIN、YEAR等其他单位，适用性更强），而且前者的功能和 DATE_ADD() 相同；后者如 ADDDATE('2020-01-01', 30) ，把整数解读为天数。除了 ADDDATE() 、 DATE_ADD() ，还有 SUBDATE() 和 DATE_SUB() 。ADDTIME(表达式1, 表达式2) ，返回两个时间表达式相加后的值。前一个表达式需要为时间或日期时间表达式，后者为时间表达式。CURDATE() 返回当前日期（以 'YYYY-MM-DD' 的字符串形式，或 YYYYMMDD 的数值形式，取决于本函数是否在数值环境中使用，比如 CURDATE()+0）， CURTIME([fsp]) 返回当前时间，fsp（fractional seconds precision）可选项用于指定秒的精度，同样有 'hh:mm:ss' 和 hhmmss 两种形式。类似的还有 NOW() ，返回当前日期时间。DATE(日期时间表达式) ，用于返回日期时间表达式的日期部分。 TIME() 同理。类似的还有 YEAR() 、 MONTH() 、 DAY()（ DAY() 是 DAYOFMONTH() 的同义函数） 、 DAYOFWEEK() 、 HOUR() 、 MINUTE() 、 SECONDS() ，分别返回年份、月份、日期、星期几、小时数、分钟数、秒数。DATEDIFF(表达式1, 表达式2) 用于对两个日期时间表达式做差，返回相差天数，分正负号。注意，虽然两个日期时间表达式中包含了时间，但只有日期的部分被使用。DATE_FORMAT(表达式, 格式) 用于将日期时间表达式格式化输出，类似python中的 time.strftime 方法。另外补充一点日期时间类型的知识。日期时间在mysql中也要用单或双引号括起来，可以进行 = 、 != 等比较，也可以用 BETWEEN 。需要注意的是只写日期时，表示时间为 '00:00:00' ，可能对大小比较有影响，可用 DATE() 函数处理。在数值处理函数中，有如下函数与功能：ABS() ，取绝对值。COS(X) ，返回角度的余弦（这里的X单位为弧度），类似的有 SIN() 、 TAN() 等三角函数。EXP(X) ，返回e的X次方，e是自然常数。相反的函数为 LN() 和只有一个参数的 LOG() 。MOD(N, M) ，返回N除以M后的余数，也可以写成 N MOD M 或 N % M 。PI() ，返回圆周率，默认精度为小数点后6位，但实际上它在mysql中是一个双精度浮点数。RAND() ，返回0~1之间的随机单精度浮点数。如果要获得一个随机整数R，比如 i<=R<j （i、j都是整数），需要搭配 FLOOR() ： FLOOR(i + (RAND() * (j-i) )) 。FLOOR(X) ，返回小于等于 X 的最大整数。SQRT(X) ，返回X的平方根。在4种函数中，数值计算函数在各大DBMS中几乎都相同。（3）聚集函数有时常需要将表中的结果汇总后输出，比如求某列的均值、最值，或者总和、行数，就可以用聚集函数，常用的有：AVG(列名) ，返回某列的均值（忽略为NULL的行）。这里的列名可用 DISTINCT 修饰，即 AVG(DISTINCT 列名) 。COUNT(列名) ，返回某列值不为NULL的行数。如果写作 COUNT(*) ，则返回所有行数，不管是否为NULL（可能因为主键一般不会是NULL）。MAX() 和 MIN() 用于寻找最值，注意它们对字符串、数值和日期时间都能使用，因为mysql中这三种类型在内部都能比较大小。SUM() 用于求和。另外还有求标准差、方差的函数，就不介绍了。最后，所有聚集函数的参数中都可以带 + 、 - 、 * 、 / ，或者其他函数。（4）数据分组GROUP BY 从句，可以对数据进行分组，经常使用的一种格式是：SELECT 列名1,COUNT(*) FROM 表名 GROUP BY 列名1 这将会输出两列，左边的列名就是列名1，右边为 COUNT(*) ；左边类似用了 DISTINCT 限定，不会输出重复的值，右边则会输出该值出现过多少次。这个例子看起来用 DISTINCT 也能实现，但由于 DISTINCT 需要作用于所有列、所以不能有聚集函数 COUNT() ，于是这里只能用 GROUP BY 实现。如果对 GROUP BY 的效果有一定了解，应该可以理解其以下设定：GROUP BY 可以包含任意数目的列，效果和 DISTINCT 类似，也作用于所有列，就是说如果 GROUP BY 包含有两行，这两行所有选定的列值都相同，就会被算作同一行数据；GROUP BY 中，出现在 SELECT 后的列名，都必须出现在 GROUP BY 后面，除了聚集函数（如果不使用联结的话。在后文中介绍联结时，将打破这项规则）；如果列值中有NULL，它也会被当成分组（对NULL生效）。如果要对 GROUP BY 设置条件，用 HAVING 代替 WHERE ，语法相同。如：SELECT 列名1,COUNT(*) FROM 表名 GROUP BY 列名1 HAVING COUNT(*)>2这里可以理解为， WHERE 在数据汇总（ GROUP BY ）之前对数据过滤， HAVING 则针对分组后的数据过滤。另外， GROUP BY 经常和 ORDER BY 搭配使用。到这里，各种关键字的顺序已经有点复杂了，忘记时记得用 HELP SELECT 查看。最后，在 GROUP BY 里还可以添加 WITH ROLLUP ：SELECT 列名1,COUNT(*) FROM 表名 GROUP BY 列名1 WITH ROLLUP它可以对某一列（如果它是数值类型）进行求和，但不能和 ORDER BY 一起出现。（5）子查询子查询即嵌套的 SELECT 。对于多张表或单张表，都经常会用到子查询。一般使用的格式为：SELECT 列名1 FROM 表名1 WHERE 列名2 IN (SELECT 列名3 FROM 表名2 WHERE ...)注意这里的 IN ，之前说过，可以把 IN 后面接的当成一个集合，这里就比较好理解了。有时， IN 也可以替换成 = 、 <> 。如果需要的话，可以嵌套很多层，但从性能上考虑，还是尽量少嵌套太多，有时候，子查询可以被其他方法替代。从编写SQL和mysql执行的角度上看，都应该从内到外看待子查询。在涉及多张表的子查询中，两张表可能会有相同的列名，这时可以用之前提到的完全限定列名，即 表名.列名 格式来指定要使用哪张表中的列。子查询比较复杂，这里举一个例子，需要对表中数据进行去重。构造以下的表：CREATE TABLE IF NOT EXISTS test( no INT UNSIGNED NOT NULL AUTO_INCREMENT, name VARCHAR(20) NOT NULL DEFAULT "unknown" COMMENT "姓名", `id` INT UNSIGNED NOT NULL DEFAULT "1" COMMENT "学号", PRIMARY KEY (no))COMMENT="学生姓名与学号";这个例子中，创建了一张表，name列存储学生的姓名，id列存储学生的学号（ COMMENT 是对列或表的注释，用 SHOW FULL COLUMNS FROM 表名 可查看）。然后插入值：INSERT INTO test (name,`id`) VALUES ("a",1001),("a",1001),("b",1002),("c",1003),("c",1003),("c",1003),("d",1004);由于工作人员在录入时粗心大意，本来只有4个学生，却录入了7条记录，有3条需要删除。按子查询的构造原则，从内层构造，先构建一个查询出现次数超过1的查询：SELECT name,`id`,COUNT(*) AS times FROM test GROUP BY name,`id` HAVING COUNT(*)>1;然后构造一个查询重复的行的主键的查询：SELECT no,name,`id` FROM test WHERE (name,`id`) IN ( SELECT name,`id` FROM test GROUP BY name,`id` HAVING COUNT(*)>1);注意，这里对里面嵌套的子查询返回的结果进行了修改。然后，由于每种重复的数据需要保留一条最初的记录，因此要排除no最小的记录。这里我们构造一个查询每种重复数据最小的no的查询：SELECT MIN(no),name,`id` FROM test GROUP BY name,`id` HAVING COUNT(*)>1;最后，把这两个查询组合到一起：SELECT no,name,`id` FROM test WHERE (name,`id`) IN ( SELECT name,`id` FROM test GROUP BY name,`id` HAVING COUNT(*)>1) AND no NOT IN ( SELECT MIN(no) FROM test GROUP BY name,`id` HAVING COUNT(*)>1);这样就返回了需要删除的行的主键，name和id（实际上只需要主键），结合后面会讲解的 DELETE ，就可删除重复行。（6）联结表在mysql中，常需要用很多表来存储信息，比如下面的例子：有很多导演，每一名导演执导过不同的电影（导演还需要存储其身高体重年龄等其他信息）；有很多电影，每部电影由一名导演执导（电影还要存储其票房等其他信息），那么要存储这些信息，最好是用两张表，一张表存储导演的信息，另一张表存储电影的信息。那么此时这两张表就产生了关系，因为导演表中有电影列，电影表中有导演列。如下创建这两张表：CREATE TABLE IF NOT EXISTS director( name VARCHAR(30) NOT NULL, height SMALLINT UNSIGNED NOT NULL DEFAULT 0 COMMENT "身高，单位cm", age TINYINT UNSIGNED NOT NULL DEFAULT 0 COMMENT "年龄", PRIMARY KEY (name))COMMENT="导演"; CREATE TABLE IF NOT EXISTS movie( name VARCHAR(30) NOT NULL, director VARCHAR(30) NOT NULL DEFAULT "unknown" COMMENT "导演", box_office BIGINT NOT NULL DEFAULT -9223372036854775808 COMMENT "票房", PRIMARY KEY (name))COMMENT="电影";需要注意，导演对电影是一对多的关系（假设一名导演不只有一部电影），电影对导演则是一对一（假设一部电影有且仅有一名导演）；在创建表时，导演表里没有其执导过的电影，电影表里有其对应的导演。然后我们给两张表插入数据：INSERT INTO director (name,age) VALUES ("a",20),("b",25),("c",30),("d",40); INSERT INTO movie (name,director,box_office) VALUES ("movie1","a",100),("movie2","a",150),("movie3","a",50),("movie4","b",55),("movie5","c",65),("movie6","c",80),("movie7","b",25),("movie8","d",35),("movie9","d",95);现在，我需要输出一张表，其列分别为电影名、票房、导演、导演年龄，那么就需要联结两张表。为了使两张表产生联结，可以将导演表的主键关联到电影表里，即把电影表里的director列作为外键（一个外键是另一张表的主键），也就是说本例中电影表里的director列是导演表的主键（这里进行演示时没有创建外键，后续将会讲解如何创建外键。创建外键后，更改另一张表中的信息，改动会同步到本张表，比如本例中更改导演表中的导演名，改动会同步到电影表中的导演名这一列）。但是要注意，这种联结只在查询时临时创建，在mysql中它并不物理存在。从多张相互关联的表里创建联结，只要使用完全限定列名即可：SELECT movie.name AS movie,movie.box_office AS box,movie.director AS director,director.age AS director_age FROM director,movie WHERE movie.director=director.name;关键之处是它的 FROM 接了两张表， WHERE 里指定两张表的某个列值相同，另外它所有的列名都用了完全限定（有些列名可以不用，但笔者建议所有列名都用）和 AS 。最后，这里的 movie.director AS director 也可以改成 director.name AS director实际上，这里的运行逻辑是这样的，mysql先执行前两行语句（这两句你也可以试试执行，最好把下面语句中 SELECT 后面跟的列名改成 * ）：SELECT movie.name AS movie,movie.box_office AS box,movie.director AS director,director.age AS director_age FROM director,movie;这将会联结每一行，就是说最终的返回的结果总行数为笛卡尔积，即两张表各自总行数的乘积（本例中为4*9即36行）。然后mysql再用 WHERE 中给出的条件对结果进行筛选，最后只剩下符合要求的行（这里限定的条件使得最终行数和movie表中的行数相同，即9行）。最后，既然说到这里，不如试试 WITH ROLLUP 的效果。输入以下查询语句：SELECT director,sum(box_office) FROM movie GROUP BY director WITH ROLLUP;这将输出每位导演执导电影的总票房，最后一行则是这几位导演的总票房之和。刚才的例子被称为等值联结，因为 WHERE 里的条件是两张表的某个值相等。但其实等值联结还有一种写法，称为内部联结：SELECT movie.name AS movie,movie.box_office AS box,director.name AS director,director.age AS director_age FROM director INNER JOIN movie ON movie.director=director.name;在这种写法中，明确了两张表是使用内部联结，并且把 WHERE 换成了 ON 。实际上，这种方法是更推荐的，因为它的联结表示得更加显眼，且有时内部联结的效率更高。最后，除了两张表，也可以联结多张表，但性能下降的很快（因为此时的笛卡尔积是所有表的总行数的乘积）。之前曾经介绍过子查询，举了一个例子（在一张表里查重），如果子查询涉及多张表时，常常可以用表联结的方法替代，针对不同的情况，子查询和联结的性能表现也不同，需要自己实际测试。现在，在脑海中回想一下子查询与联结：子查询是 SELECT 的嵌套，需要用 IN ；等值联结或内部联结是在表的笛卡尔积中返回符合条件的行，需要用 INNER JOIN 和 ON （这两个关键字用于内部联结写法，比等值联结写法更推荐），最好完全限定列名（有时候不需要，但涉及多张表时都推荐写成完全限定的形式）。另外要注意，现在并没有举在一张表中使用联结的例子，但这种联结也是存在的。（7）其他联结概念除了等值联结、内部联结（这两种联结是一种方法的不同写法，但后者有时效率更高），还有其他联结概念，包括自联结、自然联结和外部联结。要了解这些联结，需要先了解如何给表起别名。在前面的例子中，我们对列和计算字段起过别名，格式是在 SELECT 后跟的列里，写： SELECT 列名1 AS 别名1, 计算字段1 AS 别名2, ... 。其实，还可以用 AS 对表起别名：SELECT m.name AS movie,m.box_office AS box,d.name AS director,d.age AS director_age FROM director AS d INNER JOIN movie AS m ON m.director=d.name;这里沿用了上一个例子中的两张表。注意，只是在 FROM 从句中对两张表指定了别名，后续和前面所有用到这两张表名字的地方都使用了别名（这里是缩写，所以大大减少了SQL的长度），包括给两张表指定别名之前，即 SELECT 中。实际上，mysql会先读完语句，最后再执行。表别名的主要作用是缩短SQL，和多次复用表别名（包括临时生成的表）。下面讲解自联结、自然联结和外部联结。首先介绍一下自联结，仍然使用导演表、电影表的例子。现在我们发现一部电影（ "movie1" ）可能涉嫌严重的偷税漏税，需要查找该电影导演的其他电影作品（用电影查找导演，再查找电影）。我们可以用两条语句，即：SELECT director FROM movie WHERE name="movie1"; #返回结果为a SELECT name FROM movie WHERE director="a"; 但是这样逼格明显太低。下面试试用一条语句，首先用子查询：SELECT name FROM movie WHERE director IN (SELECT director FROM movie WHERE name="movie1");接下来，我们使用自联结：SELECT m1.name FROM movie AS m1, movie AS m2 WHERE m2.name="movie1" AND m1.director=m2.director;这一条语句可以这样理解：mysql用同一张表创建两个相同的临时表，对这两个临时表求笛卡尔积，最后返回符合条件的行。由于是对一张表使用联结，这种联结称为自联结。然后是自然联结。这里的自然联结更像是一种概念、一种哲学，比如在导演表和电影表的例子中，我会把电影表的导演名这一列，和导演表的姓名这一列用来比较（等值联结），这是自然而然的事情（虽然它们在两张表中的列名不同，但表达的是同一个事物，即导演的名字）；如果我把电影表的票房这一列和导演表的年龄这两列拿去比较，虽然它们都是INT，但却没有任何意义，所以有悖自然、有悖人伦。实际上，非自然联结没有任何意义。最后是外部联结。比如我要输出电影名，导演名，导演年龄这三列，可以使用：SELECT m.name,d.name,d.age FROM movie AS m LEFT OUTER JOIN director AS d ON m.director=d.name;这个功能使用前面的内部联结、等值联结也能轻易做到，这里需要关注的重点是导演年龄其实和电影名关系很小（所以在电影表中没有存储导演的年龄）。所以，使用 LEFT OUTER JOIN 或 RIGHT OUTER JOIN ，将两个表中非关联列（非外键列）也输出的行为，就称为外部联结。 LEFT 和 RIGHT 用于指定以哪张表为主，影响的只有输出顺序，如果你用我的电影表和导演表实验，会发现输出的电影名顺序不同；在其他方面二者没有区别。最后，简单讲一下带聚集函数的联结。比如我要统计每位导演有多少部电影作品（这当然可以只在电影表中用 GROUP BY 完成，但这里使用联结+聚集函数）：SELECT d.name,COUNT(*) AS num_of_movies FROM movie AS m INNER JOIN director AS d ON m.director=d.name GROUP BY d.name;在这里要注意，前文曾说过， SELECT 后的列名在 GROUP BY 后必须出现，上例也遵守了这项规则；但在联结中用 GROUP BY ，有时可以不遵循该规则。最后，再提醒一下，有些情况下可用子查询，也可用联结，二者查询性能如何需要做实验来确定。（8）组合查询一个 SELECT 返回一个结果集，有时候我们需要将结果集求并集一起返回，就可使用组合查询（一条语句查询所有结果）。细分下来，组合查询可分为两种情况：对单个表进行多次查询（查询的返回结果相同），将结果集求并后返回；或者有时一张表能存储的数据达到上限，不得已将一张表分成多张表，这也可以算是对单张表进行多次查询（但从操作过程上看是对多张表多次查询）。对多张表进行多次查询（查询结果的数据结构类似），求并后返回结果集合。比如电影表和导演表中，有些小众电影也有导演，但在导演表中没有记录这些导演的名字，如果我想得到所有导演的名字，就可以用组合查询。其实 WHERE 条件中的 OR 也可以理解为求并集，所以这种带 OR 的 SELECT 语句也可以改写成组合查询的形式。如同子查询和联结，这两种等效写法的性能也互有高低，需要测试，但大部分情况下 OR 的效率更低，尤其是有很多个 OR 时。组合查询只需要简单地在 SELECT 语句间加一个 UNION 就完事。这里简单举一个例子，我想得到导演"a"执导的所有电影名，以及票房超过80的所有电影名，可以执行下面的SQL:SELECT name FROM movie WHERE director="a" OR box_office>80;把它改写成组合查询的形式如下：SELECT name FROM movie WHERE director="a" UNION ALL SELECT name FROM movie WHERE box_office>80;UNION 的规则也很好理解，它在每个 SELECT 间插入一个 UNION （比如4个 SELECT 就需要3个 UNION ），每条 SELECT 后跟的列、表达式或聚集函数必须有相似的数据类型（比如TINYINT和INT这种可以转换的类型就算相似），而且如果把每条 SELECT 后跟的列、表达式或聚集函数看作一个集合，那么各个集合的元素个数相同，且可以建立一一对应的关系。组合查询不要求所有 SELECT 后跟的列、表达式或聚集函数的顺序相同，但笔者建议最好写成相同的。另外，由于这是求并集，所以mysql会把重复的部分去除（ A并B=A+B-(A交B) ）。如果想把重复的部分也输出，用 UNION ALL 。最后，组合查询也能排序，但 ORDER BY 仅需要在最后的 SELECT 后面写一次。毕竟每条 SELECT 后面都写的话太麻烦了，而且一不留神还可能写错顺序。（9）全文本搜索全文本搜索，即 FULL TEXT ，是一种空间换时间的做法，用于对mysql中的字符串进行快速高效的模糊匹配。前文提到用 LIKE 做模糊匹配，但FULLTEXT是对所有的文本建立索引，查起来效率比 LIKE 高到不知道哪里去了；可惜由于需要对全文本建立索引，FULLTEXT的硬盘占用比较高。mysql中有数据库引擎这一说，常用的引擎有MyISAM（现在基本不用）和InnoDB（现在的默认引擎），在mysql 6之后的版本，二者都支持FULLTEXT。据说mysql 5的版本对中文用FULLTEXT有bug，不知道现在如何。使用FULLTEXT的场景一般是大量的文本数据，比如你用爬虫爬取了一家报纸媒体的所有文章。要使用FULLTEXT，从建表时就要有相关的设计：CREATE TABLE IF NOT EXISTS articles( no int NOT NULL AUTO_INCREMENT, author VARCHAR(40) NOT NULL DEFAULT "unknown", content TEXT NULL, PRIMARY KEY (no), FULLTEXT (content) )ENGINE=InnoDB;之后，如果content列里的数据发生改变，mysql会自动更新索引。由于更新索引也需要一定的时间（虽然很少），如果需要把一个FULLTEXT列导入到一张新表，最好的做法是先创建正常的表，导入数据完成后再指定这一列为FULLTEXT。更改列属性的方法将在后面介绍。创建完成后，使用 MATCH() 和 AGAINST() 函数搜索：SELECT content FROM articles WHERE MATCH(content) AGAINST("搜索表达式");全文本搜索也不区分大小写，需要的话用 BINARY 指定。如果在 SELECT 后指定了多个FULLTEXT列，在 MATCH() 中必须以相同的顺序列出这些列，就是说次序和个数都要正确。在 AGAINST() 的使用上，比如想搜索含有 'rabbit' 的行，就写 AGAINST("rabbit") 。注意，FULLTEXT返回的顺序是按匹配程度排序的，比如 'rabbit' 越靠前则匹配程度越高，在返回结果中的排名也越靠前。另外， AGAINST() 中还可以使用查询扩展，如 AGAINST('rabbit' WITH QUERY EXPANSION) ，它的工作原理是先找到含有 'rabbit' 的行；对不含有 'rabbit' 的行，如果它们含有一些词语，这些词语在能匹配到 'rabbit' 的行里也出现过，则这些不含 'rabbit' 的行也可以被输出，只是排序靠后。最后，还有布尔搜索，它不需要使用FULLTEXT的索引，但可能比较慢。你可以给它提供一些参数，包括：要匹配的词；要排除的词。当某个值含有要排除的词，它就绝对不会出现在查询结果中，即使它含有要匹配的词；对要匹配的词的重要性排序；其他内容。用 AGAINST(... IN BOOLEAN MODE) 即可指定布尔搜索，前面同样是 MATCH() 。比如，想要包含abc，但不以xyz开头的行： AGAINST('abc-xyz*' IN BOOLEAN MODE) 。其他操作符也能使用，含义如下：布尔操作符 说 明 + 包含，词必须存在 - 排除，词必须不出现 > 包含，而且增加等级值 < 包含，且减少等级值 () 把词组成子表达式（允许这些子表达式作为一个组被包含、排除、排列等） ~ 取消一个词的排序值 * 词尾的通配符 "" 定义一个短语（与单个词的列表不一样，它匹配整个短语以便包含或排除这个短语）另外一个注意的地方是，布尔搜索中不会对返回的结果进行匹配程度的排序。5.插入数据到此为止，检索数据的操作讲完了。正常的数据库要实现的功能有增删查改（用的最多的就是查，所以讲的篇幅最长），接下来讲其他的，首先就是增，即插入数据。插入数据使用 INSERT ，一般格式为：INSERT INTO 表名 (列名1, 列名2, ...) VALUES (值1, 值2, ...);如果对一张表内的所有值都进行指定（除了一些特殊列，比如 AUTO_INCREMENT 的列），则可以忽略列名：INSERT INTO 表名 VALUES (NULL, 值1, 值2, ...);这里使用NULL，是假设第一列为 AUTO_INCREMENT 的列，由于不能跳过它，就给了一个NULL值，不过这个NULL值会被mysql自动忽略。在生产环境中，不指定列名的写法尽量不要使用，因为它高度依赖列的顺序；在第一种写法中，对于有默认值或自增的列，可以不写出来。如果插入多行数据，用：INSERT INTO 表名 (列名1, 列名2, ...) VALUES (值1, 值2, ...), (值1, 值2, ...), (值1, 值2, ...), (值1, 值2, ...);这样的格式。如果需要插入多条数据，尽量用这种形式，它的执行效率更高。最后，有一种 INSERT SELECT 的写法，它一般用于从旧表中导出新表，一个比较简单的写法为：INSERT INTO 新表 (列名1, 列名2, ...) SELECT 列名1, 列名2, ... FROM 旧表;新旧表的列名不需要相同，但次序是固定的。这里列出的 SELECT 比较简单，其实也可以写复杂一点。6.更新删除数据UPDATE 可以用于更新表中的数据。它的写法为：UPDATE 表名 SET 列名1=新值1,列名2=新值2, ... WHERE 条件;一定要注意条件的设置，否则所有行都会被更新。如果确实要更新所有行的话，那就不写 WHERE 从句。默认情况下， UPDATE 会有all or nothing的特性，即如果每一行都成功更新，则最后所有更新作为一个整体提交；一旦某一行更新时出现错误，则整个 UPDATE 操作都会被取消，之前更新的行也会恢复到原来的数据。所以 UPDATE 要么不更新，要么全部更新，没有中间状态，称为all or nothing（或者称为粒子性、事务管理）。DELETE 用于删除数据。和 UPDATE 相同，如果不用 WHERE 从句限制，它会删除所有行。其一般格式为：DELETE FROM 表名 WHERE 条件;但如果确实要删除表中所有数据，更推荐使用 TRUNCATE TABLE 表名 ，它实际上是把原来的表删除并重建一张表，速度比 DELETE 快。经过笔者实验，如果表中有自增列，使用 DELETE 删除整张表后新建列不是从1开始，而 TRUNCATE 和 DROP 是从1开始的。由于 UPDATE 和 DELETE 会破坏原来的数据，在更新或删除之前最好仔细检查 WHERE 指定的更新或删除范围是否正确，不要最后从删库到跑路。7.创建和操作表用 CREATE TABLE 创建表，其一般格式为：CREATE TABLE 表名( 列名1 类型 是否可为NULL [默认值] [其他特殊项], 列名2 类型 是否可为NULL [默认值] [其他特殊项], 列名3 类型 是否可为NULL [默认值] [其他特殊项], ... PRIMARY KEY (主键列名) )[ENGINE=引擎名] [CHAR_SET=];笔者更推荐的创建语句需要加上 IF NOT EXISTS 和一个自增的主键，如下：CREATE TABLE IF NOT EXISTS 表名( no INT UNSIGNED NOT NULL AUTO_INCREMENT, #这行的 NOT NULL 可以省略 列名1 类型 是否可为NULL [默认值] [其他特殊项], 列名2 类型 是否可为NULL [默认值] [其他特殊项], 列名3 类型 是否可为NULL [默认值] [其他特殊项], ... PRIMARY KEY (no));对于数据类型，可查看 Python，爬虫与深度学习（4）——番外篇（二）MySQL数据库基础 ；是否可为NULL，只需要写 NULL 或 NOT NULL ，后者表示不可为NULL（后者更推荐，因为NULL在很多操作中被自动忽略，且可能影响性能，一般的做法是用一些不可能的值代替NULL，比如空字符串 "" ）；默认值用 DEFAULT 设置，比如 DEFAULT "unknown" ；其他特殊项，比如 AUTO_INCREMENT 。笔者使用的是一个单独的自增列做主键，实际上也可以用其他列，只要这个列的值在每行都是唯一的，而且确定后就不会变化（笔者的导演表中使用导演名字做主键，但其实并不稳妥，因为人是可以改名的）；或者也可以用多个列的组合做主键，只要这些列值的组合在每个列也是唯一的就行，这时就写成 PRIMARY KEY (列名1, 列名2, ...) 即可。需要注意的是，主键列不允许可以为NULL值，即主键列必须是 NOT NULL 。之前说过，如果表中有自增列，使用 DELETE 删除整张表后新建列不是从1开始，而 TRUNCATE 和 DROP 是从1开始的。用 SELECT LAST_INSERT_ID() 可获取最后使用的一个自增列值，把它+1后就是下一个即将使用的自增值。要注意的是，这个函数返回的是最近一次成功的 INSERT 操作插入的自增值，所以这个函数最好在插入操作后立即使用。最后，mysql现在支持用函数做默认值，至少笔者知道datetime类型的列可以设置默认值为 DEFAULT NOW() 。对表的定义进行更改称为操作表，从数据库维护的角度看，创建表的时候就要考虑各种情况，尽量在后期运行时不要更改表。使用 ALTER TABLE 可以操作表，常用操作如下：ALTER TABLE 表名 ADD 列名 类型 是否可为NULL [默认值] [其他特殊项]; #在所有列的最后新增一个列 ALTER TABLE 表名 DROP COLUMN 列名; #删除一个列 ALTER TABLE 表1名 ADD CONSTRAINT 外键名（可用fk_表1_表2，这里fk表示foreign key） FOREIGN KEY (表1的列名) REFERENCES 表2名 (表2的列名); #添加外键。这里 表2的列名 为表2的主键列名另外，使用 RENAME 可以重命名表，格式为 RENAME TABLE 表名 TO 新的表名 。如果要更改复杂、庞大的表，建议用数据迁移的方法而不用 ALTER ，步骤为：用新的列布局创建新表；用 INSERT SELECT 将旧表中的数据迁移到新表；检验测试新表；重命名或删除旧表，把新表改名为旧表名；重新创建外链、触发器、索引等。8.视图SELECT 返回的是一个结果集，前面也提到从客户端的角度根本分辨不了哪些列是库中物理存在的，哪列数据是mysql服务器从已有的数据计算、导出的。其实， SELECT 返回的结果很像一张表，有时我们需要多次重复使用一个 SELECT 返回的结果（多次重复使用一条 SELECT 语句），那么就可以把该SQL存成一个视图。视图本质上是一个虚拟表，要遵守表的规则，比如：不能与数据库中的其他表或视图重名；和真实的表一样，视图也有访问权限；视图可以当成真实的表使用，比如视图查询语句可以嵌套。视图也有不同于表的地方，比如不能增删改（实际上可以，对视图操作就是对基表操作。但推荐不要通过视图更新基表，直接到基表里更新。视图只应该用于检索），不能索引，也不能设置默认值等。更重要的是每次调用视图时其实都是执行了原来的SQL，这可能对性能有影响。最后，要注意的是，如果创建视图的SQL中含有 ORDER BY ，那么调用视图的查询中再使用 ORDER BY 会被覆盖。下面是视图的基本操作：CREATE VIEW 视图名 AS SELECT语句 #创建视图 SHOW CREATE VIEW 视图名 #查看创建视图的语句 DROP VIEW 视图名 #删除视图 CREATE OR REPLACE VIEW 视图名 AS SELECT语句 #创建或更新视图。推荐用这条语句更新视图。视图最主要的作用就是复用SQL，简化新创建的SQL语句。9.存储过程及其他存储过程是为了满足更复杂的需要而创建的SQL，有时可能涉及到条件判断等高级操作，它实际上是把SQL当高级语言用。这实际上已经一定程度上超出了使用mysql的日常需要了，而且使用python的pymysql库等高级语言进行处理有时更为方便，所以笔者这里就不进行介绍了。有兴趣的请自行搜索，或者查看 runoob教程 。其他类似的mysql进阶概念还有：游标（用于交互式应用，比如用户滚动网页、更新加载查询结果）；触发器（满足一些条件后就执行的SQL）；事务管理和回退（就是各种操作的all or nothing。虽然pymysql使用时会涉及，但只是很少一点，并且是范式）；字符集（charset）和校对（collate）；安全管理（即用户管理，比如不要经常用mysql的root登录，严格控制用户权限等）。对这些内容有需要的可查看《MySQL必知必会》原书。10.数据库维护主要是检查mysql是否正常运行，备份、还原数据库。ANALYZE TABLE 表名 #检查表键是否正确 CHECK TABLE 表名 EXTENDED #彻底检查表的一些常见问题 OPTIMIZE TABLE 表名 #在表中删除大量数据后，用来优化其存储空间的利用在mysql中导出数据时，其实导出的是创建库、表所需要的SQL。导出、导入数据的命令如下：mysqldump -uroot -p 库名 表名 > /backup/库名.表名.sql #导出一张表。注意这里是在mysql命令行外面，即cmd或terminal里执行的。后面的路径或文件名可自行更改 mysqldump -uroot -p 库名 > /backup/库名.sql #导出一个库 mysqldump -uroot -p --all-databases > /backup/all_database_dump.sql #导出mysql中的所有库 mysql -u root -p 库名 < /backup/库名.表名.sql #在一个库中导入一张备份的表 mysql -u root -p < /backup/all_database_dump.sql #导入所有库导入导出文件的后缀名没有什么特殊要求，但一般用.sql以示区分。11.mysql使用的原则在《MySQL必知必会》一书的最后，作者给出了使用MySQL的忠告：（1）首先，MySQL（与所有DBMS一样）具有一定的硬件要求。在学习和研究MySQL时，使用任何旧的计算机作为服务器都可以。但对于生产环境中的服务器来说，应该整点好的配置。一般来说，关键的生产DBMS应该运行在自己的专用服务器上。 （2）MySQL是用一系列的默认设置预先配置的，从这些设置开始通常是很好的。但过一段时间后你可能需要调整内存分配、缓冲区大小等。（为查看当前设置，可使用 SHOW VARIABLES; 和 SHOW STATUS; 。） （3）MySQL是一个多用户多线程的DBMS，换言之，它经常同时执行多个任务。如果这些任务中的某一个执行缓慢，则所有请求都会执行缓慢。如果你遇到显著的性能不良，可使用 SHOW PROCESSLIST 显示所有活动进程（以及它们的线程ID和执行时间）。你还可以用 KILL 命令终结某个特定的进程（使用这个命令需要作为管理员登录）。 （4）总是有不止一种方法编写同一条查询语句，建议试验联结、并、子查询等，找出最佳的方法。 （5）使用 EXPLAIN 语句可以让MySQL解释它将如何执行一条 SELECT 语句。 （6）一般来说，存储过程执行得比一条一条地执行其中的各条MySQL语句快。 （7）应该总是使用正确的数据类型，在创建表时要仔细考虑。 （8）不要检索需求以外的数据。换言之，不要用 SELECT * ，除非你真的需要所有列。 （9）有的操作（包括 INSERT ）支持一个可选的 DELAYED 关键字，如果使用它，将在有可能时（服务器有空闲资源时）才实际执行该操作，但是它会立即给客户端返回一个信号，假装这个操作已经执行了，客户端就可以继续进行其他操作，如查询等。 （10）在导入数据时，应该关闭自动提交。你可能还要记住不要在新表建立时（指把旧表数据导入新表时）就创建索引（包括 FULLTEXT索引 ），而是应该在导入完成后再重建它们。 （11）建议对表创建索引以改善检索的性能。确定对哪些列创建索引是一件比较繁琐的任务，需要分析使用的 SELECT 语句以找出重复的 WHERE 和 ORDER BY 子句。如果一个简单的 WHERE 子句返回结果所花的时间太长，则可以断定其中使用的列（或几个列）就是需要索引的对象。但是有些时候不要创建索引，见下一条。 （12）索引可以改善数据检索的性能，但损害数据插入、删除和更新的性能。如果你有一些表，它们经常更改但不经常被搜索，则在有必要之前不要索引它们。（索引可根据需要添加和删除。） （13）你的 SELECT 语句中有一系列复杂的 OR 条件吗？改用 UNION ，你能看到极大的性能提升。 （14） LIKE 很慢。一般来说，最好是使用 FULLTEXT 而不是 LIKE （空间换时间）。 （15）数据库是不断变化的，一组优化良好的表一会儿后可能就面目全非了。由于表的使用和内容的更改，理想的优化和配置也会改变。 （16）最重要的规则就是，每条规则在某些条件下都会被打破。12.python中使用mysql首先，用pip安装pymysql库：pip install pymysql然后，在.py文件里使用pymysql：import pymysql conn = pymysql.connect(host='127.0.0.1', port=3306, user='root', passwd='mysqlpasswd', db='mysql', charset='utf8') cur = conn.cursor() try: cur.execute('会对mysql进行更改的操作') conn.commit() except: conn.rollback() cur.execute('查询操作') results=cur.fetchall() #注意results是一个二维元组注意更改host，port，user，passwd的值。另外，注意 conn.commit() 和 conn.rollback() 的使用，它们只在对mysql进行更改操作（增删改）时才使用。
步骤/目录： 1.mysql特性与基础概念，及安装 2.进入mysql命令行与基本设置 3.mysql命令行的基础命令 4.增删库、表及注意事项 5.表内数据的基础增删查改 6.LIKE 与模糊查找本文首发于个人博客https://lisper517.top/index.php/archives/42/，转载请注明出处。本文的目的是对MySQL进行简单介绍，便于用mysql存储爬虫爬取到的数据。本文写作日期为2022年9月2日。本文主要整理自 runoob教程 。进阶学习mysql可以参考《MySQL必知必会》、《高性能MySQL》等其他书籍，github上有收集这些CS书的仓库；或者查看笔者后续的文章。1.mysql特性与基础概念，及安装除了mysql，还有redis等各种数据库（或者mysql的亲兄弟mariadb等各种sql），它们是为了高速存储数据而出现的。mysql是一种关系性数据库，从上到下，它的构成是：各数据库（database） - 各张表格（table） - 每张表格由行和列组成，列有列名，每一行存储一个条目/记录，每一列的数据类型相同 。另外，mysql中还有主键和索引的概念，每一行记录的主键都不同（其他列可能相同），所以主键可以用于区分每一行记录；索引，类似书籍的目录，用于快速访问数据。mysql支持大型数据库，单张表的上限主要和操作系统允许的最大文件有关，一般来说，一张表占4GB就差不多了，再高可能有一些性能上的问题。最后，mysql是开源的，目前是被Oracle买下。在安装mysql上，笔者建议尽量用docker-mysql的方式安装，可参考 docker-mysql官网 ，或笔者之前的文章。一个基础的docker-compose模板如下：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" 这里的端口（不建议用默认的3306端口）。2.进入mysql命令行与基本设置使用命令： mysql -h 127.0.0.1 -P 3306 -uroot -p 可以mysql里的root用户身份登录mysql， -h 参数指定mysql服务器端的ip， -P 指定端口， -u 指定登录用户（如果用root以外的其他用户登，可用 -u 其他用户名 的格式。注意登录的用户不是linux系统里的用户，而是mysql数据库里的用户）， -p 指定密码登录，之后输入密码即可。如果是本机、默认的3306端口（mysql默认使用3306端口），则前两个参数可以省略。最后，登录其他机器上的mysql服务器时要注意登录的用户是否允许远程登录（docker-mysql则没有这个问题）。如果使用docker-mysql，根据 docker-mysql官网 的描述，将自己的mysql配置文件夹挂载到容器的 /etc/mysql/conf.d 文件夹，如果该文件夹里有任何cnf结尾的文件都会覆盖docker-mysql的默认配置。另外，docker-mysql的默认配置文件为容器里的 /etc/mysql/my.cnf 。关于配置方面，笔者几乎没有更改过，所以这方面了解的不多。用 mysql -h 127.0.0.1 -P 3306 -uroot -p 登录到mysql命令行后，可以新建其他用户。比较推荐的是用GRANT命令：GRANT SELECT,INSERT,UPDATE,DELETE,CREATE,DROP ON TUTORIALS.* TO 'zara'@'localhost' IDENTIFIED BY 'zara123';这条命令的第一行是为新建的用户赋予权限。这里赋予了该用户查询、插入、更新、删除数据，及新建、删除数据库/表的权限。也可以自行增加或删除权限，或者用 ALL PRIVILEGES 指代所有权限。第二行限定该用户只对TUTORIALS这个database下的所有表有这些权限。第三行指定用户名为zara、登录地点限于从本机登录（当然，对docker-mysql无效）。最后一行指定zara用户的密码为zara123。REVOKE 可剥夺用户的选项，语法和 GRANT 是一样的，只是把 TO 换成 FROM 。3.mysql命令行的基础命令mysql命令对大小写不敏感，但笔者建议，自己随便查查的时候可以用全小写，但写到python里的还是建议把所有mysql的关键字大写，以后阅读的时候更容易区分。另外，所有的命令最后都以 ; 结尾，只有少数简单的命令（如 USE 数据库名 ）等可以不加 ; ，笔者写的时候有时会省略，自己写的时候记得尽量加上。SHOW DATABASES 显示所有数据库的名字；用 USE 数据库名 选择要操作的数据库（database）；SHOW TABLES 显示该数据库下所有表名；SHOW COLUMNS FROM 数据表名 或 DESCRIBE 数据表名 显示表的各列是什么数据类型等信息；SHOW INDEX FROM 数据表名 显示数据表的详细索引；SHOW TABLE STATUS [FROM 数据库名] [LIKE "模式"] [\G] 可输出数据库的性能、统计信息， [] 里括起来的是可选项。注意 \G 也可以起 ; 的作用，添加这个选项就不加 ; 。4.增删库、表及注意事项CREATE DATABASE 数据库名 来创建一个新的库；更保险的做法是 CREATE DATABASE IF NOT EXISTS 数据库名 。DROP DATABASE 数据库名 删除数据库。创建表则更为复杂，因为要指定每一列的变量类型。mysql中大致有数值、时间、字符串这三种类型的变量，具体见下：数值（分有符号和无符号）名称 占用字节 表示范围 TINYINT 1 Bytes (-128，127)或(0，255) SMALLINT 2 Bytes (-32 768，32 767)或(0，65 535) MEDIUMINT 3 Bytes (-8 388 608，8 388 607)或(0，16 777 215) INT或INTEGER 4 Bytes (-2 147 483 648，2 147 483 647)或(0，4 294 967 295) BIGINT 8 Bytes (-9,223,372,036,854,775,808，9 223 372 036 854 775 807)或(0，18 446 744 073 709 551 615) FLOAT 4 Bytes 10的+-38次方级 DOUBLE 8 Bytes 10的+-308次方级 DECIMAL 对DECIMAL(M,D) ，如果M>D，为M+2否则为D+2 依赖于M和D的值另外还有 BIT位 、 REAL实数 等数值。时间（分为日期和时间）名称 占用字节 范围 格式 DATE 3 1000-01-01/9999-12-31 YYYY-MM-DD TIME 3 '-838:59:59'/'838:59:59' HH:MM:SS YEAR 1 1901/2155 YYYY DATETIME 8 '1000-01-01 00:00:00' 到 '9999-12-31 23:59:59' YYYY-MM-DD hh:mm:ss TIMESTAMP 4 '1970-01-01 00:00:01'UTC 到 '2038-01-19 03:14:07'UTC YYYY-MM-DD hh:mm:ss字符串（分二进制形式和字符形式）名称 占用字节 用途 CHAR 0-255 bytes 定长字符串 VARCHAR 0-65535 bytes 变长字符串 TINYTEXT 0-255 bytes 短文本字符串 TEXT 0-65 535 bytes 长文本数据 MEDIUMTEXT 0-16 777 215 bytes 中等长度文本数据 LONGTEXT 0-4 294 967 295 bytes 极大文本数据 TINYBLOB 0-255 bytes 不超过 255 个字符的二进制字符串 BLOB 0-65 535 bytes 二进制形式的长文本数据 MEDIUMBLOB 0-16 777 215 bytes 二进制形式的中等长度文本数据 LONGBLOB 0-4 294 967 295 bytes 二进制形式的极大文本数据另外还有BINARY、VARBINARY等。对于汉字，用UTF-8编码时占3个字节、用GBK时占2个字节；VARCHAR(n)表示无论汉字或英文，都能存入n个字符（所以占用字节长度会变化，称为 VAR CHAR ）。一般来说，经常变化的字段用VARCHAR，定长、不超过255字节的用CHAR（不够长度的部分mysql在存储时会用空格补齐），偶尔用TEXT（TEXT不能有默认值）。最常用的就是VARCHAR。创建数据表示需要指定每列的数据类型，还要指定主键。例句如下：CREATE TABLE IF NOT EXISTS `test`( `no` INT UNSIGNED AUTO_INCREMENT, `title` VARCHAR(100) NOT NULL DEFAULT "unknown", `created_time` TIMESTAMP DEFAULT CURRENT_TIMESTAMP, PRIMARY KEY (`no`) )ENGINE=InooDB DEFAULT CHARSET=utf8;反引号（就是键盘上ESC下面那个键）起到引号的作用（为了避免有时可能这些名字和mysql关键字冲突，但是尽量勿用关键字做表名、列名）； UNSIGNED 为无符号； AUTO_INCREMENT 说明这列的数据是自动增加的（要注意，如果有些行被删除了后再新建行，后面的行 no 值是从被删除的行开始算的，也就是说一个 no 被用过一次后就不会再用，即使对应的行被删除）； NOT NULL 指定这一列不能为空（应该用"0"或其他值代替NULL，可提高查询性能，这里用的是"unknown"）； DEFAULT 指定默认值； CURRENT_TIMESTAMP 则是该行被创建时的时间戳。 ENGINE=InooDB 指定存储引擎， DEFAULT CHARSET=utf8 指定默认编码（这两个也可以不指定，现在mysql默认引擎就是InooDB，默认编码则可输入 STATUS; 查看）。一般创建表的时候就指定默认编码，笔者使用的docker-compose中有： command: - mysqld - --character-set-server=utf8mb4mysql默认编码为Latin1，要换成utf8或GBK才能存储中文。删除表就用 DROP TABLE 表名 即可。5.表内数据的基础增删查改用 INSERT 向表内插入数据：INSERT INTO 表名( 列名1, 列名2, 列名3...) VALUES (值1, 值2, 值3...);可向表内插入一行数据。插入的值是字符串时，使用单引号或双引号括起来。如果某一列有默认值或者是自动生成的，可以忽略、不用写出列名和值。下面则是添加多行数据：INSERT INTO 表名( 列名1, 列名2, 列名3...) VALUES (值1, 值2, 值3...), (值1, 值2, 值3...), (值1, 值2, 值3...), ...;查找数据用 SELECT ，最基本的是 SELECT * FROM 表名 ，返回表内的所有行，其中 * 表示匹配所有列。一般格式为：SELECT 列名1, 列名2, ... FROM 表名 [WHERE Clause] [LIMIT n] [OFFSET m];中括号内的都是可选项。 WHERE 从句用来限定条件， LIMIT 限制返回的记录数， OFFSET 指定开始查询的行数（偏移量）， LIMIT m,n 是 LIMIT n OFFSET m 的简写（从第几条开始，返回多少条结果）。在高级查询中，可以从不止一个表中查询数据，表名间用 , 分隔，搭配 WHERE 使用。实际上，删查改一般都要用 WHERE 限定范围。mysql中的比较符： = 为等于， <> 或 != 为不等，此外还有 > 、 < 、 >= 、 <= 。另外， BINARY 可以区分大小写：SELECT * FROM 表名 WHERE BINARY 列名="字符串";改数据用 UPDATE ，一般格式如下：UPDATE 表名 SET 列名1=值1, 列名2=值2 [WHERE Clause]比如，把符合条件的行的列值+1：UPDATE test SET int_data=int_data+1又或者用mysql函数：UPDATE test SET str_data=REPLACE(str_data,"旧字符串","新字符串")上述命令将该列所有数据内的旧字符串替换为新字符串（部分替换）。删除记录用 DELETE ，一般格式为：DELETE FROM 表名 [WHERE Clause]不加 WHERE 时，会删除所有行。但是，如同之前提到的，如果列中有 AUTO_INCREMENT ，下次再加新行时会从断点继续，就是说新行的自增列不会从1开始。另外，mysql中的删除其实有三个关键字： DELETE 、 DROP 、 TRUNCATE ，第三个的使用方式：TRUNCATE 表名可以把 TRUNCATE 看作是不带 WHERE 的 DELETE FROM ，只用于删除表中的全部数据，优点是执行更快一些；DROP 是把表也给删了，速度最快。6.LIKE 与模糊查找前面已经介绍过了， WHERE 从句中可以使用 = 等符号。有时要只想匹配部分字符串，可以使用 LIKE ，或 % 、 _ 通配符。另外mysql也可以使用re，见后续文章。LIKE 可以替代 = ：WHERE 列名1 LIKE 条件1比如：WHERE field1 LIKE "%com%"只要某行的field1列中含连续的 com 三个字符，就符合条件。如果改成 _com_ ，就是指必须是5个字符，中间三个为 com 。也就是说， % 匹配任意个数（包括0个）的任意字符， _ 匹配单个任意字符。另外， LIKE 里的条件类似re，也有 [] （括号内所列的任意一个字符）、 [^] （非括号内的任意一个字符），如：LIKE "[abc]"表示匹配单个字符，这个字符为 a 或 b 或 c 。 LIKE "[^abc]" 则表示不能是 a 或 b 或 c 。在中括号内，用 [] 括起来表示原本的通配符，即 [[%]abc] 、 [[_]abc] 、 [[[]]abc]（匹配 [） 。到这里为止，mysql的基础已经讲解完毕，其实本文的内容对于爬虫中使用mysql几乎够用了，在下一篇文章中将会讲解进阶的内容。
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静态方法，方便调用。
步骤/目录： 1.在windows笔记本上下载并配置redis-server 2.购买阿里云服务器，安装frp并配置frps服务器端 3.配置frpc客户端 4.开启服务并验证 5.附录：mysql+frp （1）修改配置 （2）创建远程连接用户 （3）赋予用户权限并刷新退出 （4）打开端口 （5）修改frpc.ini并开启frp本文首发于个人博客https://lisper517.top/index.php/archives/3/，转载请注明出处。本文实验日期2022年1月31日。采用windows笔记本（跑redis服务器）+阿里云服务器（系统镜像Ubuntu 20.04）1.在windows笔记本上下载并配置redis-serverredis官方不建议在windows下运行redis，所以并未提供windows版redis下载。有一些其他大神更新了windows版redis，以下推荐2个：首先是微软版redis。由微软方提供，于2016年7月1日更新了3.0.504版后停止更新。其次是tporadowski版redis（推荐）。由github用户tporadowski提供，目前最新版本为2021年10月18日更新的Redis 5.0.14 for Windows。（若遇到github网页无法打开，可在gitee上注册并导入仓库后下载）（根据runoob网站，Redis5.0新增了Stream数据结构，所以还是下新版的好）对于tporadowski版redis，下载.zip包，解压到安装目录即可。解压完毕后，打开Redis文件夹，编辑redis.windows.conf文件，需要修改3处：将bind 127.0.0.1行注释掉。这一行是仅允许从本地访问redis；用frp应该不用改，但是笔者需要在局域网中访问，所以还是打开了。 将protected-mode yes改为protected-mode no。老版本的redis可能没有这一行，如果没有则建议重新下载新版redis； 新增一行：requirepass 你的密码。由于redis的处理速度较快，建议密码设置的复杂一些。配置完成后，打开cmd命令行，cd到redis文件夹（也可以将redis文件夹添加到系统路径），输入redis-server redis.windows.conf，正常显示redis服务器端界面，即可。在linux下redis的配置也差不多，不同的是linux下redis的配置文件是redis.conf。2.购买阿里云服务器，安装frp并配置frps服务器端首先介绍一下frp。frp分为frps（server）和frpc（client），其中frps运行在有公网ip的云服务器上，frpc运行在内网中的机器上。frp主要用于把内网机器的端口映射到云服务器端口，比如把内网机器的22端口（一般用于ssh）映射到云服务器的6000端口，那么访问云服务器的6000端口即可接入内网机器的ssh服务。如果有其他方法搞到公网ip，也可以跳过这一步。随便选择一个便宜的服务器，购买后安装镜像，这里选择的系统镜像是Ubuntu 20.04。在左侧 服务器运维-远程连接 中设置一下root账户密码，然后下载putty，通过putty连接：连接类型为SSH，端口为默认的22，主机地址填云服务器的ip地址。输入root和root密码后即登录到云服务器。目前frp最新版本为0.38.0（2021年10月25更新）。这里安装frp到/bin目录下（wget github速度慢可以换成gitee上的仓库）：cd /bin wget https://github.com/fatedier/frp/releases/download/v0.38.0/frp_0.38.0_linux_amd64.tar.gz tar -zxvf frp_0.38.0_linux_amd64.tar.gz mv frp_0.38.0_linux_amd64 frp cd frp nano frps.ini然后配置一下frps.ini：[common] bind_port = 7000 #frpc上也要有相同的token值 token = 你的密码 #配置log log_file = /var/logs/frps.log log_level = info log_max_days = 3 #frps统计板 dashboard_port = 7500 dashboard_user = wuhuqifei dashboard_pwd = wuhuqifei kcp_bind_port = 7000解读一下配置选项：bind_port是frp使用的端口；token是密码令牌；log是日志相关；dashboard是统计板，设置好以后，网页访问 服务器ip:7500 即可看到frps的统计情况，user和pwd分别设置统计板的用户名密码（不用太复杂，简单易记即可）。如果后面运行frps时显示没有frps.log文件，可以自己新建一个并chmod改一下权限（或者直接删掉log相关设置）。最后的kcp_bind_port是多消耗15%左右流量、降低tcp延迟的，不要也可以不加。详细配置可以参考frps_full.ini。配置完成后，需要打开云服务器相应端口。在 安全-防火墙 中新增规则，打开7000和7500端口（云服务器在此处打开，非云服务器则使用ufw、iptables等。云服务器用ufw会出问题）。附带一些github上常见问题：流量是否经过服务器中转？ 目前除了 xtcp 外，其他类型的内网穿透模式的流量都需要经过 frps 所在服务器中转。 网络传输速度慢是什么原因？ 由于流量需要经过服务器转发，所以传输速度的快慢取决于服务器的下行带宽和客户端的上行带宽，通常家用宽带的上行带宽较低，限制了出口的速度。 另外一种情况是服务器部署在国外的 VPS 上，丢包率较高，也会影响到传输速度。这种情况下可以考虑开启 kcp 传输模式。 开启加密和压缩会消耗一定的 cpu 资源，且影响到传输速度，具体影响情况取决于机器性能。 客户端连接失败，提示 authorization failed 出现这种情况说明鉴权失败，检查 frps 和 frpc 的配置文件中的 token 是否一致。 客户端连接失败，提示 authorization timeout 出现这种情况是因为 frps 所在服务器和 frpc 所在服务器的系统时间相差较大。如果不希望在身份校验时加入系统时间，可以将 frps 配置文件中的 authentication_timeout 设置为 0 来解决这个问题。 frpc 能否在系统启动阶段无网络时一直等待而不是直接退出？ 在 frpc 的配置文件中将 login_fail_exit 设置为 false，则 frpc 启动后会不断尝试连接 frps，直到连接成功，而不是直接退出。最后使用./frps -c ./frps.ini即可启动frps。启动后不能关闭putty，这时可以访问7500端口查看是否成功开启。3.配置frpc客户端linux安装frp同上。对于windows，在frp下载页面下载frp_0.38.0_windows_amd64.zip，解压到目标文件夹即可。windows安全中心可能会删掉frps.exe，因为用不到所以无妨。配置一下frpc.ini：[common] server_addr = 云服务器ip地址 server_port = 7000 token = 你的密码 login_fail_exit = false [frp-redis] type = tcp local_port = 6379 local_ip = localhost remote_port = 16380解读frpc.ini：[common]是基础设置，下面的[frp-redis]是一个映射。以后要添加新的映射就在下面加，不用改动[common]，具体可以参考frpc_full.ini。要注意的是，如果有多个内网机器的多个端口需要映射，每个映射名要不同。比如多个redis要开放到外网，就可以写[frp-redis-1]、[frp-redis-2]、[frp-redis-3]等等。然后打开云服务器的16380端口（同前），本地的6379端口不用打开（因为frp穿透时，被映射的本地端口接收的操作都会被frp改成从本地发起）。Ubuntu系统可以安装ufw，然后ufw allow 6379（其他发行版请自行搜索）。另外，windows打开端口，是在 控制面板-Windows防火墙-高级设置-入站规则-新建规则 ，新建一个tcp类型、6379端口的规则，之后有程序使用6379端口时即可使用netstat -na查看所有开放的端口。以上都配置完后，使用frpc -c frpc.ini即可开启服务。这时访问dashboard地址，就能在 Proxies-TCP 中看到名为 frp-redis 的服务，其status为online。4.开启服务并验证验证时需要在云服务器上也安装redis。本例中云服务器的系统镜像是Ubuntu20.04，可以直接用apt下载：apt install redis-server开启frps：使用putty连接至云服务器，输入frp路径\frps -c frp路径\frps.ini开启后勿关闭putty。开启frpc：在windows下，cmd中输入frp路径\frpc -c frp路径\frpc.ini开启后勿关闭cmd窗口。开启redis-server：在windows下，cmd中输入redis路径\redis-server redis路径\redis.windows.conf开启后勿关闭cmd窗口。这时用另一个putty窗口连接到云服务器，输入redis-cli -h 云服务器ip -p 16380 -a redis密码（也可省略-a，在redis命令行中使用auth命令）其中redis密码即requirepass中自己设置的密码。登录上后使用ping命令，服务器端会返回PONG。之后为了方便，可以设置开机自动运行frps、frpc、redis-server。之后即可将redis用于scrapy-redis等其他用途。5.附录：mysql+frp这里假设windows10机器上已安装好mysql。（1）修改配置在mysql安装目录下找到配置文件my.ini，添加一行：bind-address=*允许从其他ip地址访问mysql数据库。用frp应该不用改，但是笔者需要在局域网中访问，所以还是打开了。（2）创建远程连接用户在mysql命令行中输入以下命令，可查看mysql用户及允许登录的地点：select distinct concat('User: ''',user,'''@''',host,''';') as query from mysql.user;如果想将root登录的地点改为任意ip，可使用：RENAME USER 'root'@'localhost' TO 'root'@'%';但允许root用户远程登录mysql比较危险，不如创建一个新用户专门用于远程连接：CREATE USER 'mysql-remote1'@'%' IDENTIFIED BY '你的密码';mysql 8.0后用户认证默认使用sha2，这里将该用户改为用原生密码认证：ALTER USER 'mysql-remote1'@'%' IDENTIFIED WITH mysql_native_password BY '你的密码';创建用原生密码认证的用户，只需将上述命令的ALTER改为CREATE。（3）赋予用户权限并刷新退出接着赋予该用户一定的权限。输入下面的命令可以给账号全局权限，包括创建（CREATE）、修改（ALTER）、删除（DROP） 数据库、表、用户，任意表的插入（INSERT）、更新（UPDATE）、删除（DELETE）操作权限。可以使用 SELECT 查询数据，使用 REFERENCES 建立外键关系权限，以及使用 RELOAD 权限执行 FLUSH 操作的权限。当然，你也可以根据自己都需求，对账号权限进行调整。GRANT CREATE, ALTER, DROP, INSERT, UPDATE, DELETE, SELECT, REFERENCES, RELOAD on *.* TO 'mysql-remote1'@'%' WITH GRANT OPTION;这里我需要能远程修改数据库，所以只赋予了部分权限：GRANT ALTER, INSERT, DELETE, SELECT on *.* TO 'mysql-remote1'@'%' WITH GRANT OPTION;然后刷新权限并退出:FLUSH PRIVILEGES; quit;（4）打开端口mysql默认用3306端口，但是本地端口不用打开。在云服务器上打开frp要用的端口（以16381为例）。（5）修改frpc.ini并开启frp在frpc.ini中添加：[frp-mysql] type = tcp local_port = 3306 local_ip = localhost remote_port = 16381之后重启win10机器的frpc即可。