为什么要做这个

看自己的github上的star的时候发现了wbt5/real-url，感觉一个个的脚本找起来挺麻烦的，就想着能不能整合成一个服务去调用，抽空摸了几个小时鱼就有了现在这个项目。

如何实现

首先先看看完整的项目结构：

root.
│  config.py    # 配置文件
│  flask_server.py    # flask app
│  README.md    # 说明文档
│  requirements.txt    # 依赖包
├─apps    # 注册的app
│    live_url_api.py    # 获取直播链接的api文件
│   __init__.py
│
├─Logs    # 日志模块
│      Error.log    # 错误日志
│      Runtime.log    # 运行日志
│
├─routings    # 直播平台区分路由
│    live_api_routing.py    # 直播平台区分路由
│    __init__.py
│
├─scripts
│  │  base.py    # 基础类
│  │  __init__.py    # 读取所有继承基础类的类对象
│  │
│  └─lives    # 各直播平台脚本
│    │  173.py
│    │  17live.py
│    │  2cq.py
│    │  51lm.py
│    │  95xiu.py
│    │  9xiu.py
│    │  acfun.py
│    │  bilibili.py
│    │  cc.py
│    │  changyou.py
│    │  douyin.py
│    │  douyu.py
│    │  egame.py
│    │  fengbolive.py
│    │  hongle.py
│    │  huajiao.py
│    │  huomao.py
│    │  huya.py
│    │  imifun.py
│    │  immomo.py
│    │  inke.py
│    │  iqiyi.js
│    │  iqiyi.py
│    │  ixigua.py
│    │  jd.py
│    │  kbs.py
│    │  kk.py
│    │  kuaishou.py
│    │  kugou.py
│    │  kuwo.py
│    │  laifeng.py
│    │  lehai.py
│    │  longzhu.py
│    │  look.py
│    │  maoer.py
│    │  now.py
│    │  pps.py
│    │  ppsport.py
│    │  qf.py
│    │  qie.py
│    │  renren.py
│    │  showself.py
│    │  sports_iqiyi.py
│    │  tiktok.py
│    │  tuho.py
│    │  twitch.py
│    │  v6cn.py
│    │  wali.py
│    │  woxiu.py
│    │  xunlei.py
│    │  yangshipin.py
│    │  yizhibo.py
│    │  youku.py
│    │  yuanbobo.py
│    │  yy.py
│    │  zhanqi.py
│    │  zhibotv.py
│    └─ __init__.py
└─templates    # 模板文件夹
      LiveUrlDocs.html    # 接口文档HTML
      LiveUrlDocs.md    # 接口文档

我目前比较顺手的web框架就是flask了，简单易上手。

首先使用flask搭建一个web服务：

# flask_server.py
# -*- coding: utf-8 -*-
from flask import Flask
from flask_restful import Api

from loguru import logger
from apps.live_url_api import live_api
from config import *

# 注册flask
app = Flask(__name__)
# restful
api = Api(app)
# encode
app.config.update(RESTFUL_JSON=dict(ensure_ascii=False))
app.config['JSON_AS_ASCII'] = False
# registered
app.register_blueprint(live_api, url_prefix='/live-api')

if __name__ == '__main__':
    logger.info(f"server run on {API_HOST}:{API_PORT}. threaded is {'on' if API_THREADED else 'off'}.")
    app.run(host=API_HOST, port=API_PORT, threaded=API_THREADED)

端口 HOST 等再创建一个config配置方便管理，因为本项目没有用的数据库，如果要添加数据库，请使用环境变量配置，如何配置请看这篇文章：flask配置文件之python-dotenv的使用 - PY·DUCK (pyduck.com)

# config.py
# -*- coding: utf-8 -*-
# 日志模块
from loguru import logger

logger.add('./Logs/Runtime.log', level='DEBUG', retention='7 days', encoding='utf-8')
logger.add('./Logs/Error.log', level='ERROR', retention='7 days', encoding='utf-8')

# api配置
API_HOST = '0.0.0.0'
API_PORT = 7000
API_THREADED = True    # 是否开启多线程

如果用其他服务部署的话API配置可以不必计较，如何部署我们后边会讲。

接下来我们在live_url_api.py里面配置接口：

# live_url_api.py
# -*- coding: utf-8 -*-
from flask import Blueprint, render_template
from flask_restful import Resource
from flask_restful import reqparse
from loguru import logger
from routings.live_api_routing import RunScripts

live_api = Blueprint('live_api', __name__)

# 响应模板
return_model = {'state': 0, 'data': None}


class LiveUrl(Resource):
    """
    直播流接口
    """

    def __init__(self):
        self.return_data = return_model
        self.reqparse = reqparse.RequestParser()
        self.reqparse.add_argument('live_platform', type=str, required=True, help='直播平台')
        self.reqparse.add_argument('parameter', type=str, required=True, help='需要解析的内容')
        super(LiveUrl, self).__init__()

    def get(self):
        # 定义消息体
        return_data = self.return_data.copy()
        # 接收front参数
        args = self.reqparse.parse_args()
        live_platform = args.get('live_platform')
        parameter = args.get('parameter')
        logger.info(f"req: live_platform-{live_platform} parameter-{parameter}")
        live_url = RunScripts(live_platform, parameter).choice()
        logger.info(f"res: live_platform-{live_platform} parameter-{parameter} res-{live_url}")
        if not live_url:
            return return_data
        return_data['state'] = 1
        return_data['data'] = live_url
        return return_data


class LiveUrlDocs(Resource):
    """
    接口文档
    """
    def get(self):
        """
        获取接口文档
        :return:
        """
        return render_template('LiveUrlDocs.html')


live_api.add_url_rule(rule='/api-docs', view_func=LiveUrlDocs.as_view('api-docs'))
live_api.add_url_rule(rule='/get-url', view_func=LiveUrl.as_view('get-live-url'))

因为要对接很多直播平台的脚本文件，我们不可能一个一个去判断再去调用相应的脚本，这里我之前看过崔庆才的代理池项目：[[Python3 网络爬虫开发实战] 9.2 - 代理池的维护 | 静觅 (cuiqingcai.com)](https://cuiqingcai.com/7048.html)，里面用到了如何灵活调用类方法。

基类文件：

# base.py
class Base(object):
    _name = 'Base'

    def __init__(self):
        self.rid = ''

    def get_real_url(self):
        return

读取所有继承类：

# __init__.py
# -*- coding: utf-8 -*-
import pkgutil
from .base import Base
import inspect


# load classes subclass of BaseCrawler
classes = []
for loader, name, is_pkg in pkgutil.walk_packages(__path__):
    module = loader.find_module(name).load_module(name)
    for name, value in inspect.getmembers(module):
        globals()[name] = value
        if inspect.isclass(value) and issubclass(value, Base) and value is not Base \
                and not getattr(value, 'ignore', False):
            classes.append(value)
__all__ = __ALL__ = classes

然后将脚本文件统一继承基类，统一调用函数：

# 2cq.py
# 棉花糖直播：https://www.2cq.com/rank

import requests

from scripts.base import Base


class MHT(Base):    # 继承基类

    _name = '棉花糖'    # 定义名称，方便判断

    def __init__(self, rid):
        super(Base, self).__init__()
        self.rid = rid

    def get_real_url(self):
        with requests.Session() as s:
            res = s.get('https://www.2cq.com/proxy/room/room/info?roomId={}&appId=1004'.format(self.rid))
        res = res.json()
        if res['status'] == 1:
            result = res['result']
            if result['liveState'] == 1:
                real_url = result['pullUrl']
                return real_url
            else:
                return '未开播'
        else:
            return '直播间可能不存在'


if __name__ == '__main__':
    r = input('输入棉花糖直播房间号：\n')
    print(MHT(r).get_real_url())

最后把判断调用的脚本写好就OK了：

# live_api_routing.py
# -*- coding: utf-8 -*-
from loguru import logger

from scripts import __all__ as scripts_cls


class RunScripts:

    def __init__(self, live_platform, parameter):
        self.scripts_cls = scripts_cls
        self.live_platform = live_platform
        self.parameter = parameter
        self.scripts = [scripts(self.parameter) for scripts in self.scripts_cls]

    def choice(self):
        """
        选择脚本
        :return:
        """
        for live_platform in self.scripts:
            if self.live_platform.upper() == live_platform._name:
                logger.info(f'live_platform {live_platform} to get live url.')
                data = live_platform.get_real_url()
                return data
        else:
            return


if __name__ == '__main__':
    print(RunScripts('羚萌', '24003').choice())

如何部署

运行环境：

python3.6及以上

安装依赖

在项目路径下运行此命令

pip install -r rquirements.txt

服务部署

1. mac
   暂未测试

2. windows
   直接运行

   python flask_server.py

3. linux
   可以用gunicorn部署

   运行命令
   gunicorn -w 4 -b 127.0.0.1:8080 flask_server:app
   配置说明
   -c CONFIG, --config=CONFIG 设定配置文件。
   -b BIND, --bind=BIND 设定服务需要绑定的端口。建议使用HOST:PORT。 
   -w WORKERS, --workers=WORKERS 设置工作进程数。建议服务器每一个核心可以设置2-4个。 
   -k MODULE 选定异步工作方式使用的模块。

接口说明

接口地址

http://api.pyduck.com/live-api/get-url

接口免费，服务器性能有限，勿压请爱惜。

接口文档请访问

获取直播流接口文档

项目说明

本项目开源，由于本身有工作，再加上下班就躺，所以有问题请在github上发issue，我会用上班摸鱼的时间去看。-> XGSClear7 (26℃) (github.com)

支持的直播平台