使用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 的核心特点

  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 的对比

特性FLASKDJANGO
设计理念轻量级、灵活,最小功能集重量级、全功能, batteries-included
默认组件无 ORM、模板引擎(可选)内置 ORM(Django ORM)、模板引擎
学习曲线低(简单直接)高(功能丰富但复杂)
适用场景小型项目、API、需要高度控制的场景企业级大型项目、快速全栈开发
最后修改:2025 年 04 月 18 日
如果觉得我的文章对你有用,请随意赞赏