为什么要做这个

看自己的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)

支持的直播平台

平台名称请求说明
棉花糖直播房间号
九秀直播房间号
17直播房间号
羚萌直播房间号
95秀直播房间号
艺气山直播房间号
AcFun直播房间号
哔哩哔哩直播房间号
网易CC直播房间号
畅秀阁直播房间号
抖音抖音直播间room_id或分享链接
斗鱼直播间号
企鹅电竞房间号
新浪疯播直播房间号
红人直播房间号
花椒直播房间号
火猫直播房间号
虎牙直播房间号
艾米直播房间号
陌陌直播房间号
映客直播房间号
爱奇艺直播房间号
西瓜直播房间号
京东直播房间号
腾讯体育直播间地址
KK直播房间号
快手直播房间ID
酷狗直播房间号
酷我聚星直播房间号
来疯直播房间号
乐嗨直播房间号
龙珠直播房间号
Look直播房间号
猫耳直播房间号
NOW直播间号
PPS奇秀直播房间号
PP体育直播间地址
千帆直播房间号
企鹅体育直播房间号
人人直播房间号
秀色直播房间号
爱奇艺体育直播间完整地址
TikTok分享链接
星光直播房间号
twitch房间名
六间房直播房间号
小米直播房间号
我秀直播房间号
迅雷直播房间号
央视频央视频地址
一直播房间地址
优酷轮播台房间号
热猫直播房间号
YY直播房间号
战旗直播房间号
中国体育房间号
Last modification:August 2nd, 2021 at 09:10 am
如果觉得我的文章对你有用,请作者喝杯咖啡把~