使用Python实现一个简单的接口服务,可以通过get、post方法请求该接口,拿到响应数据。创建一个api\_server.py文件,添加代码如下: ```python import flask from flask import request, jsonify from rescode.constants import ResponseCode from spider.qiXinSpider import getQiXinCompInfo ''' flask: web框架,通过flask提供的装饰器@server.route()将普通函数转换为服务 登录接口,需要传url、username、passwd ''' # 创建一个服务,把当前这个python文件当做一个服务 server = flask.Flask(__name__) # server.config['JSON_AS_ASCII'] = False # @server.route()可以将普通函数转变为服务 登录接口的路径、请求方式 @server.route('/python-api/getCompTageFromQiXin', methods=['get', 'post']) def getCompTageFromQiXin(): try: # 获取通过url请求传参的数据 httpUrl = request.values.get('httpUrl') if not httpUrl: return jsonify(ResponseCode.PARAM_REQUIRED), 400 if 'www.sdxyjq.com:8080' in httpUrl: httpUrl = httpUrl.replace('www.sdxyjq.com:8080', 'www.sdxyjq.com') # 调用qiXinSpider.py里面的函数,需要传入 # chrome的路径 D:\\APP\\TestChrome2\\Application\\chrome.exe # 信用金桥的http链接的url地址 comp_info = getQiXinCompInfo(httpUrl,'D:\\APP\\TestChrome2\\Application\\chrome.exe') data = { "httpUrl" : httpUrl, "qiXinSpider" : comp_info, "compName":comp_info['baseInfo']['ename'] } return jsonify({**ResponseCode.SUCCESS, "data": data}), 200 except Exception as e: # 统一异常捕获 return jsonify(ResponseCode.INTERNAL_ERROR), 500 @server.errorhandler(404) def not_found(error): return jsonify({ "code": 404, "message": "接口不存在" }), 404 @server.errorhandler(500) def internal_error(error): return jsonify(ResponseCode.INTERNAL_ERROR), 500 if __name__ == '__main__': server.run(debug=True, port=8888, host='0.0.0.0') ``` 爬虫脚本 ```python # -*- encoding:utf-8 -*- import time from selenium.webdriver.support.ui import WebDriverWait from selenium.webdriver.chrome.options import Options from seleniumwire import webdriver as wiredriver from bs4 import BeautifulSoup import requests import gzip import io import json # 初始化selenium def initialize_driver(chromePath: str): # 配置 Chrome 浏览器选项 chrome_options = Options() chrome_options.add_argument('--disable-gpu') # 禁用 GPU 加速,确保拦截请求正常 chrome_options.add_argument('--headless') # 不打开浏览器 chrome_options.add_argument('--ignore-certificate-errors') # 忽略证书错误 # 添加指定的浏览器路径 chrome_path = chromePath chrome_options.binary_location = chrome_path # 初始化 WebDriver,并传入配置 driver = wiredriver.Chrome(options=chrome_options) return driver # 获取启信宝的地址 def get_qixin_url(url): if 'www.sdxyjq.com:8080' in url: url = url.replace('www.sdxyjq.com:8080', 'www.sdxyjq.com') headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/113.0.0.0 Safari/537.36" } response = requests.get(url, headers=headers) html_content = response.text soup = BeautifulSoup(html_content, 'html.parser') iframe = soup.find('iframe') qiXinUrl = '' if iframe: src = iframe.get('src') qiXinUrl = src return qiXinUrl # 格式化请求体 def parse_response_body(response_body_binary): try: # 检查数据是否以 gzip 开头 is_gzip = response_body_binary.startswith(b'\\x1f\\x8b') if is_gzip: with gzip.GzipFile(fileobj=io.BytesIO(response_body_binary), mode='rb') as f: return json.loads(f.read().decode('utf-8')) else: # print('直接解码为 JSON') return json.loads(response_body_binary.decode('utf-8')) except Exception as e: print(f"格式化请求体失败: {e}") return None def extract_response_body(requests, keyword): for request in requests: if keyword in request.url: return request.response.body return None def getQiXinCompInfo(url:str,chromePath:str): try: # 初始化浏览器 driver = initialize_driver(chromePath) # 访问启信宝的网页 driver.get(get_qixin_url(url)) time.sleep(3) # 使用 WebDriverWait 等待页面加载完成之后继续操作,等待时间根据网络情况进行调整 wait = WebDriverWait(driver, 30)# 超时时间30s # 等待页面的 document.readyState 变为 "complete" wait.until(lambda driver: driver.execute_script("return document.readyState") == "complete") # 获取所有拦截的请求 requests = driver.requests # 获取企业的标签信息 getEntLabel res_getEntLabel = extract_response_body(requests, 'getEntLabel') if res_getEntLabel is not None: res_getEntLabel = parse_response_body(res_getEntLabel) else: res_getEntLabel ='' # 获取企业地址信息 getGeocode res_getGeocode = extract_response_body(requests, 'getGeocode') if res_getGeocode is not None: res_getGeocode = parse_response_body(res_getGeocode) else: res_getGeocode = '' # 获取企业的工商信息 getEntBasicInfoNew res_getEntBasicInfoNew = extract_response_body(requests,'getEntBasicInfoNew') if res_getEntBasicInfoNew is not None: res_getEntBasicInfoNew = parse_response_body(res_getEntBasicInfoNew) else: res_getEntBasicInfoNew = '' return { 'baseInfo': res_getEntBasicInfoNew, 'tagInfo': res_getEntLabel, 'addressInfo': res_getGeocode, } finally: # 关闭浏览器 driver.quit() ``` ### **Flask 是什么?** **Flask 是一个用 Python 编写的轻量级 Web 框架** ,它为构建 Web 应用程序和 RESTful API 提供了灵活的基础。Flask 的设计哲学是“简洁和可扩展”,它没有捆绑任何数据库或 ORM(对象关系映射)工具,开发者可以根据需求自由选择技术栈。 ### **Flask 的核心特点** 1. **轻量级与灵活性** : * 没有强制性的数据库或 ORM,开发者可以自由选择技术(如 SQLite、MySQL、MongoDB 等)。 * 不依赖模板引擎,默认提供简单模板,也可以替换为其他引擎(如 Jinja2)。 2. **路由系统** : * 通过装饰器(Decorator)将 URL 路径映射到 Python 函数 3. **扩展性** : * 通过第三方扩展(Extensions)增强功能,例如: * **Flask-SQLAlchemy** :数据库操作。 * **Flask-RESTful** :快速构建 RESTful API。 * **Flask-Login** :用户认证。 * **Flask-JWT** :基于 JWT 的身份验证。 4. **开发友好** : * 内置调试模式(Debug Mode),实时反映代码修改。 * 支持单元测试和集成测试。 5. **社区支持** : * 拥有活跃的开源社区和丰富的文档,适合快速开发和学习。 --- ### **Flask 的典型应用场景** 1. **小型 Web 应用** : * 适合快速开发个人博客、仪表盘、内部工具等。 2. **RESTful API 开发** : * 构建数据接口(如 JSON API),常用于前后端分离项目。 3. **微服务架构** : * 由于轻量级特性,适合构建独立的微服务模块。 4. **学习 Web 开发** : * 简单的 API 和路由设计使其成为学习 Web 开发的理想工具。 --- ### **Flask 与 Django 的对比** | **特性** | **FLASK** | **DJANGO** | | -------------------- | --------------------------------------------- | ----------------------------------------------- | | **设计理念** | **轻量级、灵活,最小功能集** | **重量级、全功能, batteries-included** | | **默认组件** | **无 ORM、模板引擎(可选)** | **内置 ORM(Django ORM)、模板引擎** | | **学习曲线** | **低(简单直接)** | **高(功能丰富但复杂)** | | **适用场景** | **小型项目、API、需要高度控制的场景** | **企业级大型项目、快速全栈开发** | Loading... 使用Python实现一个简单的接口服务,可以通过get、post方法请求该接口,拿到响应数据。创建一个api\_server.py文件,添加代码如下: ```python import flask from flask import request, jsonify from rescode.constants import ResponseCode from spider.qiXinSpider import getQiXinCompInfo ''' flask: web框架,通过flask提供的装饰器@server.route()将普通函数转换为服务 登录接口,需要传url、username、passwd ''' # 创建一个服务,把当前这个python文件当做一个服务 server = flask.Flask(__name__) # server.config['JSON_AS_ASCII'] = False # @server.route()可以将普通函数转变为服务 登录接口的路径、请求方式 @server.route('/python-api/getCompTageFromQiXin', methods=['get', 'post']) def getCompTageFromQiXin(): try: # 获取通过url请求传参的数据 httpUrl = request.values.get('httpUrl') if not httpUrl: return jsonify(ResponseCode.PARAM_REQUIRED), 400 if 'www.sdxyjq.com:8080' in httpUrl: httpUrl = httpUrl.replace('www.sdxyjq.com:8080', 'www.sdxyjq.com') # 调用qiXinSpider.py里面的函数,需要传入 # chrome的路径 D:\\APP\\TestChrome2\\Application\\chrome.exe # 信用金桥的http链接的url地址 comp_info = getQiXinCompInfo(httpUrl,'D:\\APP\\TestChrome2\\Application\\chrome.exe') data = { "httpUrl" : httpUrl, "qiXinSpider" : comp_info, "compName":comp_info['baseInfo']['ename'] } return jsonify({**ResponseCode.SUCCESS, "data": data}), 200 except Exception as e: # 统一异常捕获 return jsonify(ResponseCode.INTERNAL_ERROR), 500 @server.errorhandler(404) def not_found(error): return jsonify({ "code": 404, "message": "接口不存在" }), 404 @server.errorhandler(500) def internal_error(error): return jsonify(ResponseCode.INTERNAL_ERROR), 500 if __name__ == '__main__': server.run(debug=True, port=8888, host='0.0.0.0') ``` 爬虫脚本 ```python # -*- encoding:utf-8 -*- import time from selenium.webdriver.support.ui import WebDriverWait from selenium.webdriver.chrome.options import Options from seleniumwire import webdriver as wiredriver from bs4 import BeautifulSoup import requests import gzip import io import json # 初始化selenium def initialize_driver(chromePath: str): # 配置 Chrome 浏览器选项 chrome_options = Options() chrome_options.add_argument('--disable-gpu') # 禁用 GPU 加速,确保拦截请求正常 chrome_options.add_argument('--headless') # 不打开浏览器 chrome_options.add_argument('--ignore-certificate-errors') # 忽略证书错误 # 添加指定的浏览器路径 chrome_path = chromePath chrome_options.binary_location = chrome_path # 初始化 WebDriver,并传入配置 driver = wiredriver.Chrome(options=chrome_options) return driver # 获取启信宝的地址 def get_qixin_url(url): if 'www.sdxyjq.com:8080' in url: url = url.replace('www.sdxyjq.com:8080', 'www.sdxyjq.com') headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/113.0.0.0 Safari/537.36" } response = requests.get(url, headers=headers) html_content = response.text soup = BeautifulSoup(html_content, 'html.parser') iframe = soup.find('iframe') qiXinUrl = '' if iframe: src = iframe.get('src') qiXinUrl = src return qiXinUrl # 格式化请求体 def parse_response_body(response_body_binary): try: # 检查数据是否以 gzip 开头 is_gzip = response_body_binary.startswith(b'\\x1f\\x8b') if is_gzip: with gzip.GzipFile(fileobj=io.BytesIO(response_body_binary), mode='rb') as f: return json.loads(f.read().decode('utf-8')) else: # print('直接解码为 JSON') return json.loads(response_body_binary.decode('utf-8')) except Exception as e: print(f"格式化请求体失败: {e}") return None def extract_response_body(requests, keyword): for request in requests: if keyword in request.url: return request.response.body return None def getQiXinCompInfo(url:str,chromePath:str): try: # 初始化浏览器 driver = initialize_driver(chromePath) # 访问启信宝的网页 driver.get(get_qixin_url(url)) time.sleep(3) # 使用 WebDriverWait 等待页面加载完成之后继续操作,等待时间根据网络情况进行调整 wait = WebDriverWait(driver, 30)# 超时时间30s # 等待页面的 document.readyState 变为 "complete" wait.until(lambda driver: driver.execute_script("return document.readyState") == "complete") # 获取所有拦截的请求 requests = driver.requests # 获取企业的标签信息 getEntLabel res_getEntLabel = extract_response_body(requests, 'getEntLabel') if res_getEntLabel is not None: res_getEntLabel = parse_response_body(res_getEntLabel) else: res_getEntLabel ='' # 获取企业地址信息 getGeocode res_getGeocode = extract_response_body(requests, 'getGeocode') if res_getGeocode is not None: res_getGeocode = parse_response_body(res_getGeocode) else: res_getGeocode = '' # 获取企业的工商信息 getEntBasicInfoNew res_getEntBasicInfoNew = extract_response_body(requests,'getEntBasicInfoNew') if res_getEntBasicInfoNew is not None: res_getEntBasicInfoNew = parse_response_body(res_getEntBasicInfoNew) else: res_getEntBasicInfoNew = '' return { 'baseInfo': res_getEntBasicInfoNew, 'tagInfo': res_getEntLabel, 'addressInfo': res_getGeocode, } finally: # 关闭浏览器 driver.quit() ``` ### **Flask 是什么?** **Flask 是一个用 Python 编写的轻量级 Web 框架** ,它为构建 Web 应用程序和 RESTful API 提供了灵活的基础。Flask 的设计哲学是“简洁和可扩展”,它没有捆绑任何数据库或 ORM(对象关系映射)工具,开发者可以根据需求自由选择技术栈。 ### **Flask 的核心特点** 1. **轻量级与灵活性** : * 没有强制性的数据库或 ORM,开发者可以自由选择技术(如 SQLite、MySQL、MongoDB 等)。 * 不依赖模板引擎,默认提供简单模板,也可以替换为其他引擎(如 Jinja2)。 2. **路由系统** : * 通过装饰器(Decorator)将 URL 路径映射到 Python 函数 3. **扩展性** : * 通过第三方扩展(Extensions)增强功能,例如: * **Flask-SQLAlchemy** :数据库操作。 * **Flask-RESTful** :快速构建 RESTful API。 * **Flask-Login** :用户认证。 * **Flask-JWT** :基于 JWT 的身份验证。 4. **开发友好** : * 内置调试模式(Debug Mode),实时反映代码修改。 * 支持单元测试和集成测试。 5. **社区支持** : * 拥有活跃的开源社区和丰富的文档,适合快速开发和学习。 --- ### **Flask 的典型应用场景** 1. **小型 Web 应用** : * 适合快速开发个人博客、仪表盘、内部工具等。 2. **RESTful API 开发** : * 构建数据接口(如 JSON API),常用于前后端分离项目。 3. **微服务架构** : * 由于轻量级特性,适合构建独立的微服务模块。 4. **学习 Web 开发** : * 简单的 API 和路由设计使其成为学习 Web 开发的理想工具。 --- ### **Flask 与 Django 的对比** | **特性** | **FLASK** | **DJANGO** | | -------------------- | --------------------------------------------- | ----------------------------------------------- | | **设计理念** | **轻量级、灵活,最小功能集** | **重量级、全功能, batteries-included** | | **默认组件** | **无 ORM、模板引擎(可选)** | **内置 ORM(Django ORM)、模板引擎** | | **学习曲线** | **低(简单直接)** | **高(功能丰富但复杂)** | | **适用场景** | **小型项目、API、需要高度控制的场景** | **企业级大型项目、快速全栈开发** | 最后修改:2025 年 06 月 27 日 © 允许规范转载 打赏 赞赏作者 支付宝微信 赞 如果觉得我的文章对你有用,请随意赞赏