使用Python实现一个简单的接口服务,可以通过get、post方法请求该接口,拿到响应数据。创建一个api\_server.py文件,添加代码如下:
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')
爬虫脚本
# -*- 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 的核心特点
轻量级与灵活性 :
- 没有强制性的数据库或 ORM,开发者可以自由选择技术(如 SQLite、MySQL、MongoDB 等)。
- 不依赖模板引擎,默认提供简单模板,也可以替换为其他引擎(如 Jinja2)。
路由系统 :
- 通过装饰器(Decorator)将 URL 路径映射到 Python 函数
扩展性 :
通过第三方扩展(Extensions)增强功能,例如:
- Flask-SQLAlchemy :数据库操作。
- Flask-RESTful :快速构建 RESTful API。
- Flask-Login :用户认证。
- Flask-JWT :基于 JWT 的身份验证。
开发友好 :
- 内置调试模式(Debug Mode),实时反映代码修改。
- 支持单元测试和集成测试。
社区支持 :
- 拥有活跃的开源社区和丰富的文档,适合快速开发和学习。
Flask 的典型应用场景
小型 Web 应用 :
- 适合快速开发个人博客、仪表盘、内部工具等。
RESTful API 开发 :
- 构建数据接口(如 JSON API),常用于前后端分离项目。
微服务架构 :
- 由于轻量级特性,适合构建独立的微服务模块。
学习 Web 开发 :
- 简单的 API 和路由设计使其成为学习 Web 开发的理想工具。
Flask 与 Django 的对比
特性 | FLASK | DJANGO |
---|---|---|
设计理念 | 轻量级、灵活,最小功能集 | 重量级、全功能, batteries-included |
默认组件 | 无 ORM、模板引擎(可选) | 内置 ORM(Django ORM)、模板引擎 |
学习曲线 | 低(简单直接) | 高(功能丰富但复杂) |
适用场景 | 小型项目、API、需要高度控制的场景 | 企业级大型项目、快速全栈开发 |