update

.dockerignore

@@ -0,0 +1,5 @@
+
+.git/
+.idea/
+
+/examples/

.gitignore

@@ -0,0 +1,32 @@
+
+.gradio/
+.git/
+.idea/
+
+**/flagged/
+**/logs/
+**/__pycache__/
+
+#/data/
+/docs/
+/dotenv/
+/hub_datasets/
+/script/
+/thirdparty/
+/trained_models/
+/temp/
+
+/data/live_info
+/data/live_records
+/data/video
+
+/data/.DS_Store
+
+/examples
+
+**/*.flv
+**/*.mp4
+**/*.wav
+**/*.xlsx
+
+**/.DS_store

Dockerfile

@@ -0,0 +1,24 @@
+FROM python:3.12
+
+WORKDIR /code
+
+COPY . /code
+
+RUN apt-get update
+RUN apt-get install -y ffmpeg build-essential
+
+RUN pip install --upgrade pip
+RUN pip install --no-cache-dir --upgrade -r /code/requirements.txt
+
+RUN useradd -m -u 1000 user
+
+USER user
+
+ENV HOME=/home/user \
+    PATH=/home/user/.local/bin:$PATH
+
+WORKDIR $HOME/app
+
+COPY --chown=user . $HOME/app
+
+CMD ["python3", "main.py"]

README.md

@@ -1,6 +1,6 @@
 ---
 title: Video Platform
-emoji: 💻
+emoji: 🎬
 colorFrom: pink
 colorTo: indigo
 sdk: docker

data/douyin_live_info_collect.json

@@ -0,0 +1,9 @@
+{
+  "572033528289": {
+    "nickname": "🪭嘉玉（卷卷）书法",
+    "sec_uid": "MS4wLjABAAAAb0mqEsXmBehDdg2Q9mMA2T6YEWPGbEtYofSzX_bDnz4",
+    "room_id": "572033528289",
+    "status": 2,
+    "title": "🪭嘉玉写字✍️…正在直播"
+  }
+}

data/porter_tasks_dev.json

@@ -0,0 +1,66 @@
+[
+  {
+    "enable": true,
+    "type": "douyin_live_to_bilibili",
+    "room_name": "老陈的退路",
+    "room_id": "330025930592",
+    "check_interval": 10,
+    "key_of_credentials": "bilibili_chenjiesen_credentials"
+  },
+  {
+    "enable": true,
+    "type": "douyin_live_to_bilibili",
+    "room_name": "老陈come_back",
+    "room_id": "78835697536",
+    "check_interval": 10,
+    "key_of_credentials": "bilibili_chenjiesen_credentials"
+  },
+  {
+    "enable": true,
+    "type": "douyin_live_to_bilibili",
+    "room_name": "清源第一帅",
+    "room_id": "654177813521",
+    "check_interval": 10,
+    "key_of_credentials": "bilibili_chenjiesen_credentials"
+  },
+  {
+    "enable": true,
+    "type": "douyin_live_to_bilibili",
+    "room_name": "清源人工智能研究院",
+    "room_id": "81728900292",
+    "check_interval": 10,
+    "key_of_credentials": "bilibili_chenjiesen_credentials"
+  },
+  {
+    "enable": true,
+    "type": "douyin_live_to_bilibili",
+    "room_name": "小熊Bella与老爸",
+    "room_id": "139751520143",
+    "check_interval": 10,
+    "key_of_credentials": "bilibili_chenjiesen_credentials"
+  },
+  {
+    "enable": true,
+    "type": "douyin_live_to_bilibili",
+    "room_name": "清源之虎",
+    "room_id": "998621457719",
+    "check_interval": 10,
+    "key_of_credentials": "bilibili_chenjiesen_credentials"
+  },
+  {
+    "enable": true,
+    "type": "douyin_live_to_bilibili",
+    "room_name": "老陈真是好人",
+    "room_id": "599130203190",
+    "check_interval": 10,
+    "key_of_credentials": "bilibili_chenjiesen_credentials"
+  },
+  {
+    "enable": true,
+    "type": "douyin_live_to_bilibili",
+    "room_name": "老陈小帮手",
+    "room_id": "738682070097",
+    "check_interval": 10,
+    "key_of_credentials": "bilibili_chenjiesen_credentials"
+  }
+]

data/porter_tasks_prd.json

@@ -0,0 +1,66 @@
+[
+  {
+    "enable": true,
+    "type": "douyin_live_to_bilibili",
+    "room_name": "老陈的退路",
+    "room_id": "330025930592",
+    "check_interval": 10,
+    "key_of_credentials": "bilibili_chenjiesen_credentials"
+  },
+  {
+    "enable": true,
+    "type": "douyin_live_to_bilibili",
+    "room_name": "老陈come_back",
+    "room_id": "78835697536",
+    "check_interval": 10,
+    "key_of_credentials": "bilibili_chenjiesen_credentials"
+  },
+  {
+    "enable": true,
+    "type": "douyin_live_to_bilibili",
+    "room_name": "清源第一帅",
+    "room_id": "654177813521",
+    "check_interval": 10,
+    "key_of_credentials": "bilibili_chenjiesen_credentials"
+  },
+  {
+    "enable": true,
+    "type": "douyin_live_to_bilibili",
+    "room_name": "清源人工智能研究院",
+    "room_id": "81728900292",
+    "check_interval": 10,
+    "key_of_credentials": "bilibili_chenjiesen_credentials"
+  },
+  {
+    "enable": true,
+    "type": "douyin_live_to_bilibili",
+    "room_name": "小熊Bella与老爸",
+    "room_id": "139751520143",
+    "check_interval": 10,
+    "key_of_credentials": "bilibili_chenjiesen_credentials"
+  },
+  {
+    "enable": true,
+    "type": "douyin_live_to_bilibili",
+    "room_name": "清源之虎",
+    "room_id": "998621457719",
+    "check_interval": 10,
+    "key_of_credentials": "bilibili_chenjiesen_credentials"
+  },
+  {
+    "enable": true,
+    "type": "douyin_live_to_bilibili",
+    "room_name": "老陈真是好人",
+    "room_id": "599130203190",
+    "check_interval": 10,
+    "key_of_credentials": "bilibili_chenjiesen_credentials",
+  },
+  {
+    "enable": true,
+    "type": "douyin_live_to_bilibili",
+    "room_name": "老陈小帮手",
+    "room_id": "738682070097",
+    "check_interval": 10,
+    "key_of_credentials": "bilibili_chenjiesen_credentials",
+  }
+]

data/porter_tasks_prd2.json

@@ -0,0 +1,273 @@
+[
+  {
+    "enable": false,
+    "type": "douyin_live_info_collect",
+    "check_interval": 10,
+    "key_of_credentials": "douyin_wentao_credentials",
+    "output_file": "data/douyin_live_info_collect.json"
+  },
+  {
+    "enable": true,
+    "type": "douyin_video_download",
+    "user_name": "肖峰说新能源",
+    "sec_user_id": "MS4wLjABAAAAQinRMLyQNYA45OYXoCDrwszhRGaDVirRE1fTNSaGGkc",
+    "check_interval": 900,
+    "key_of_credentials": "douyin_wentao_credentials",
+    "min_date2": "2025-08-06 00:00:00",
+    "output_video_dir": "data/video/douyin/肖峰说新能源",
+    "output_video_info_file": "data/video/douyin/肖峰说新能源/file_info.json"
+  },
+  {
+    "enable": true,
+    "type": "douyin_video_download",
+    "user_name": "陈杰森 资本NewBoombap",
+    "sec_user_id": "MS4wLjABAAAATGoBrO7yiJ3q9go4fxq9JXjrnP1bFpdkgKckC1IpfXA_vrjSmL9ZtjmTju8ApwbT",
+    "check_interval": 900,
+    "min_date2": "2025-09-06 00:00:00",
+    "output_video_dir": "data/video/douyin/陈杰森",
+    "output_video_info_file": "data/video/douyin/陈杰森/file_info.json"
+  },
+  {
+    "enable": true,
+    "type": "douyin_video_download",
+    "user_name": "小熊Bella与老爸",
+    "sec_user_id": "MS4wLjABAAAA49QFP6YhorLIIX9M-FiZeKxmqhqXlttluSsZeaxvxzU",
+    "check_interval": 900,
+    "min_date2": "2025-09-06 00:00:00",
+    "output_video_dir": "data/video/douyin/陈杰森",
+    "output_video_info_file": "data/video/douyin/陈杰森/file_info.json"
+  },
+  {
+    "enable": true,
+    "type": "douyin_video_download",
+    "user_name": "清华陈晶聊商业",
+    "sec_user_id": "MS4wLjABAAAAV5oVsV-RjxHKrcCuqQotWtHvT8_Y7z_aQnTvT61slic",
+    "check_interval": 900,
+    "min_date2": "2025-09-06 00:00:00",
+    "output_video_dir": "data/video/douyin/清华陈晶",
+    "output_video_info_file": "data/video/douyin/清华陈晶/file_info.json"
+  },
+  {
+    "enable": true,
+    "type": "douyin_video_download",
+    "user_name": "清华陈晶聊直播",
+    "sec_user_id": "MS4wLjABAAAARKzB1ApIHOuHlrl8Hqg_0RxIp-2Dz-AW3ipYfCCLr6wX5Y7ewfRce-QQ_19w7R34",
+    "check_interval": 900,
+    "min_date2": "2025-09-06 00:00:00",
+    "output_video_dir": "data/video/douyin/清华陈晶",
+    "output_video_info_file": "data/video/douyin/清华陈晶/file_info.json"
+  },
+  {
+    "enable": true,
+    "type": "douyin_video_download",
+    "user_name": "清华陈晶聊创业",
+    "sec_user_id": "MS4wLjABAAAAwRmjwKuBA0K6VSrBYevRHrG6-c7UFppdICgKqcYhVDWlza3_Xj8f4R8H252e8tiF",
+    "check_interval": 900,
+    "min_date2": "2025-09-06 00:00:00",
+    "output_video_dir": "data/video/douyin/清华陈晶",
+    "output_video_info_file": "data/video/douyin/清华陈晶/file_info.json"
+  },
+  {
+    "enable": true,
+    "type": "douyin_video_download",
+    "user_name": "吕晓彤",
+    "sec_user_id": "MS4wLjABAAAAqejZxZKopDBDEzxcQp-_1b019FfM05C0NzjQNpc5ylU",
+    "check_interval": 900,
+    "min_date2": "2025-09-06 00:00:00",
+    "output_video_dir": "data/video/douyin/吕晓彤",
+    "output_video_info_file": "data/video/douyin/吕晓彤/file_info.json"
+  },
+  {
+    "enable": true,
+    "type": "douyin_video_download",
+    "user_name": "吕晓彤视野",
+    "sec_user_id": "MS4wLjABAAAAenYfLf-t_uRZhgQmEFOLN1iY3l2FTo4ToyC6wGEQVAH3i763q4-QgtUEleMB8n-m",
+    "check_interval": 900,
+    "min_date2": "2025-09-06 00:00:00",
+    "output_video_dir": "data/video/douyin/吕晓彤",
+    "output_video_info_file": "data/video/douyin/吕晓彤/file_info.json"
+  },
+  {
+    "enable": true,
+    "type": "douyin_video_download",
+    "user_name": "吕晓彤说",
+    "sec_user_id": "MS4wLjABAAAAkwn_ZhJXBB5f4qNT4mn_uk5hOmaCFS503C3tSHiSzSQADUpFTPuPHx4ZVScKQ1Yl",
+    "check_interval": 900,
+    "min_date2": "2025-09-06 00:00:00",
+    "output_video_dir": "data/video/douyin/吕晓彤",
+    "output_video_info_file": "data/video/douyin/吕晓彤/file_info.json"
+  },
+  {
+    "enable": true,
+    "type": "douyin_video_download",
+    "user_name": "吕晓彤的诗与远方",
+    "sec_user_id": "MS4wLjABAAAAnDI9XdGvKm9azWhg0qOviLt9xKTVT0E2fu7xOMDiMq_KJp2TdYPXvuhG8leGj-p6",
+    "check_interval": 900,
+    "min_date2": "2025-09-06 00:00:00",
+    "output_video_dir": "data/video/douyin/吕晓彤",
+    "output_video_info_file": "data/video/douyin/吕晓彤/file_info.json"
+  },
+  {
+    "enable": true,
+    "type": "douyin_live_record",
+    "room_name": "老陌",
+    "room_id": "770758107267",
+    "check_interval": 10,
+    "output_video_dir": "data/live_records/douyin/老陌Live",
+    "output_video_info_file": "data/live_records/douyin/老陌Live/file_info.json"
+  },
+  {
+    "enable": true,
+    "type": "douyin_live_record",
+    "room_name": "老陈的退路",
+    "room_id": "330025930592",
+    "check_interval": 10,
+    "output_video_dir": "data/live_records/douyin/陈杰森Live",
+    "output_video_info_file": "data/live_records/douyin/陈杰森Live/file_info.json"
+  },
+  {
+    "enable": true,
+    "type": "douyin_live_record",
+    "room_name": "老陈come_back",
+    "room_id": "78835697536",
+    "check_interval": 10,
+    "output_video_dir": "data/live_records/douyin/陈杰森Live",
+    "output_video_info_file": "data/live_records/douyin/陈杰森Live/file_info.json"
+  },
+  {
+    "enable": true,
+    "type": "douyin_live_record",
+    "room_name": "清源第一帅",
+    "room_id": "654177813521",
+    "check_interval": 10,
+    "output_video_dir": "data/live_records/douyin/陈杰森Live",
+    "output_video_info_file": "data/live_records/douyin/陈杰森Live/file_info.json"
+  },
+  {
+    "enable": true,
+    "type": "douyin_live_record",
+    "room_name": "清源人工智能研究院",
+    "room_id": "81728900292",
+    "check_interval": 10,
+    "output_video_dir": "data/live_records/douyin/陈杰森Live",
+    "output_video_info_file": "data/live_records/douyin/陈杰森Live/file_info.json"
+  },
+  {
+    "enable": true,
+    "type": "douyin_live_record",
+    "room_name": "小熊Bella与老爸",
+    "room_id": "139751520143",
+    "check_interval": 10,
+    "output_video_dir": "data/live_records/douyin/陈杰森Live",
+    "output_video_info_file": "data/live_records/douyin/陈杰森Live/file_info.json"
+  },
+  {
+    "enable": true,
+    "type": "douyin_live_record",
+    "room_name": "清源之虎",
+    "room_id": "998621457719",
+    "check_interval": 10,
+    "output_video_dir": "data/live_records/douyin/陈杰森Live",
+    "output_video_info_file": "data/live_records/douyin/陈杰森Live/file_info.json"
+  },
+  {
+    "enable": true,
+    "type": "douyin_live_record",
+    "room_name": "老陈真是好人",
+    "room_id": "599130203190",
+    "check_interval": 10,
+    "output_video_dir": "data/live_records/douyin/陈杰森Live",
+    "output_video_info_file": "data/live_records/douyin/陈杰森Live/file_info.json"
+  },
+  {
+    "enable": true,
+    "type": "douyin_live_record",
+    "room_name": "老陈小帮手",
+    "room_id": "738682070097",
+    "check_interval": 10,
+    "output_video_dir": "data/live_records/douyin/陈杰森Live",
+    "output_video_info_file": "data/live_records/douyin/陈杰森Live/file_info.json"
+  },
+  {
+    "enable": true,
+    "type": "douyin_live_record",
+    "room_name": "吕晓彤",
+    "room_id": "25132757833",
+    "check_interval": 10,
+    "output_video_dir": "data/live_records/douyin/吕晓彤Live",
+    "output_video_info_file": "data/live_records/douyin/吕晓彤Live/file_info.json"
+  },
+  {
+    "enable": true,
+    "type": "douyin_live_record",
+    "room_name": "文韬武略",
+    "room_id": "1293783051",
+    "check_interval": 10,
+    "output_video_dir": "data/live_records/douyin/文韬武略Live",
+    "output_video_info_file": "data/live_records/douyin/文韬武略Live/file_info.json"
+  },
+  {
+    "enable": true,
+    "type": "douyin_live_to_bilibili",
+    "room_name": "老陈的退路",
+    "room_id": "330025930592",
+    "check_interval": 10,
+    "key_of_credentials": "bilibili_chenjiesen_credentials"
+  },
+  {
+    "enable": true,
+    "type": "douyin_live_to_bilibili",
+    "room_name": "老陈come_back",
+    "room_id": "78835697536",
+    "check_interval": 10,
+    "key_of_credentials": "bilibili_chenjiesen_credentials"
+  },
+  {
+    "enable": true,
+    "type": "douyin_live_to_bilibili",
+    "room_name": "清源第一帅",
+    "room_id": "654177813521",
+    "check_interval": 10,
+    "key_of_credentials": "bilibili_chenjiesen_credentials"
+  },
+  {
+    "enable": true,
+    "type": "douyin_live_to_bilibili",
+    "room_name": "清源人工智能研究院",
+    "room_id": "81728900292",
+    "check_interval": 10,
+    "key_of_credentials": "bilibili_chenjiesen_credentials"
+  },
+  {
+    "enable": true,
+    "type": "douyin_live_to_bilibili",
+    "room_name": "小熊Bella与老爸",
+    "room_id": "139751520143",
+    "check_interval": 10,
+    "key_of_credentials": "bilibili_chenjiesen_credentials"
+  },
+  {
+    "enable": true,
+    "type": "douyin_live_to_bilibili",
+    "room_name": "清源之虎",
+    "room_id": "998621457719",
+    "check_interval": 10,
+    "key_of_credentials": "bilibili_chenjiesen_credentials"
+  },
+  {
+    "enable": true,
+    "type": "douyin_live_to_bilibili",
+    "room_name": "老陈真是好人",
+    "room_id": "599130203190",
+    "check_interval": 10,
+    "key_of_credentials": "bilibili_chenjiesen_credentials",
+  },
+  {
+    "enable": true,
+    "type": "douyin_live_to_bilibili",
+    "room_name": "老陈小帮手",
+    "room_id": "738682070097",
+    "check_interval": 10,
+    "key_of_credentials": "bilibili_chenjiesen_credentials",
+  }
+]

install.sh

@@ -0,0 +1,64 @@
+#!/usr/bin/env bash
+
+# bash install.sh --stage 2 --stop_stage 2 --system_version centos
+
+
+python_version=3.12.1
+system_version="centos";
+
+verbose=true;
+stage=-1
+stop_stage=0
+
+
+# parse options
+while true; do
+  [ -z "${1:-}" ] && break;  # break if there are no arguments
+  case "$1" in
+    --*) name=$(echo "$1" | sed s/^--// | sed s/-/_/g);
+      eval '[ -z "${'"$name"'+xxx}" ]' && echo "$0: invalid option $1" 1>&2 && exit 1;
+      old_value="(eval echo \\$$name)";
+      if [ "${old_value}" == "true" ] || [ "${old_value}" == "false" ]; then
+        was_bool=true;
+      else
+        was_bool=false;
+      fi
+
+      # Set the variable to the right value-- the escaped quotes make it work if
+      # the option had spaces, like --cmd "queue.pl -sync y"
+      eval "${name}=\"$2\"";
+
+      # Check that Boolean-valued arguments are really Boolean.
+      if $was_bool && [[ "$2" != "true" && "$2" != "false" ]]; then
+        echo "$0: expected \"true\" or \"false\": $1 $2" 1>&2
+        exit 1;
+      fi
+      shift 2;
+      ;;
+
+    *) break;
+  esac
+done
+
+work_dir="$(pwd)"
+
+
+if [ ${stage} -le 1 ] && [ ${stop_stage} -ge 1 ]; then
+  $verbose && echo "stage 1: install python"
+  cd "${work_dir}" || exit 1;
+
+  sh ./script/install_python.sh --python_version "${python_version}" --system_version "${system_version}"
+fi
+
+
+if [ ${stage} -le 2 ] && [ ${stop_stage} -ge 2 ]; then
+  $verbose && echo "stage 2: create virtualenv"
+
+  # /usr/local/python-3.9.9/bin/virtualenv LiveRecorder
+  # source /data/local/bin/LiveRecorder/bin/activate
+  /usr/local/python-${python_version}/bin/pip3 install virtualenv
+  mkdir -p /data/local/bin
+  cd /data/local/bin || exit 1;
+  /usr/local/python-${python_version}/bin/virtualenv nx_denoise
+
+fi

log.py

@@ -0,0 +1,252 @@
+#!/usr/bin/python3
+# -*- coding: utf-8 -*-
+from datetime import datetime
+import logging
+from logging.handlers import RotatingFileHandler, TimedRotatingFileHandler
+import os
+from zoneinfo import ZoneInfo  # Python 3.9+ 自带，无需安装
+
+
+def get_converter(tz_info: str = "Asia/Shanghai"):
+    def converter(timestamp):
+        dt = datetime.fromtimestamp(timestamp, ZoneInfo(tz_info))
+        result = dt.timetuple()
+        return result
+    return converter
+
+
+def setup_size_rotating(log_directory: str, tz_info: str = "Asia/Shanghai"):
+    fmt = "%(asctime)s|%(name)s|%(levelname)s|%(filename)s|%(lineno)d|%(message)s"
+
+    formatter = logging.Formatter(
+        fmt=fmt,
+        datefmt="%Y-%m-%d %H:%M:%S %z"
+    )
+    formatter.converter = get_converter(tz_info)
+
+    stream_handler = logging.StreamHandler()
+    stream_handler.setLevel(logging.INFO)
+    stream_handler.setFormatter(formatter)
+
+    # main
+    main_logger = logging.getLogger("main")
+    main_logger.addHandler(stream_handler)
+    main_info_file_handler = RotatingFileHandler(
+        filename=os.path.join(log_directory, "main.log"),
+        maxBytes=100*1024*1024,  # 100MB
+        encoding="utf-8",
+        backupCount=2,
+    )
+    main_info_file_handler.setLevel(logging.INFO)
+    main_info_file_handler.setFormatter(formatter)
+    main_logger.addHandler(main_info_file_handler)
+
+    # http
+    http_logger = logging.getLogger("http")
+    http_file_handler = RotatingFileHandler(
+        filename=os.path.join(log_directory, "http.log"),
+        maxBytes=100*1024*1024,  # 100MB
+        encoding="utf-8",
+        backupCount=2,
+    )
+    http_file_handler.setLevel(logging.DEBUG)
+    http_file_handler.setFormatter(formatter)
+    http_logger.addHandler(http_file_handler)
+
+    # api
+    api_logger = logging.getLogger("api")
+    api_file_handler = RotatingFileHandler(
+        filename=os.path.join(log_directory, "api.log"),
+        maxBytes=10*1024*1024,  # 10MB
+        encoding="utf-8",
+        backupCount=2,
+    )
+    api_file_handler.setLevel(logging.DEBUG)
+    api_file_handler.setFormatter(formatter)
+    api_logger.addHandler(api_file_handler)
+
+    # toolbox
+    toolbox_logger = logging.getLogger("toolbox")
+    toolbox_logger.addHandler(stream_handler)
+    toolbox_file_handler = RotatingFileHandler(
+        filename=os.path.join(log_directory, "toolbox.log"),
+        maxBytes=10*1024*1024,  # 10MB
+        encoding="utf-8",
+        backupCount=2,
+    )
+    toolbox_file_handler.setLevel(logging.DEBUG)
+    toolbox_file_handler.setFormatter(formatter)
+    toolbox_logger.addHandler(toolbox_file_handler)
+
+    # alarm
+    alarm_logger = logging.getLogger("alarm")
+    alarm_file_handler = RotatingFileHandler(
+        filename=os.path.join(log_directory, "alarm.log"),
+        maxBytes=1*1024*1024,  # 1MB
+        encoding="utf-8",
+        backupCount=2,
+    )
+    alarm_file_handler.setLevel(logging.DEBUG)
+    alarm_file_handler.setFormatter(formatter)
+    alarm_logger.addHandler(alarm_file_handler)
+
+    debug_file_handler = RotatingFileHandler(
+        filename=os.path.join(log_directory, "debug.log"),
+        maxBytes=1*1024*1024,  # 1MB
+        encoding="utf-8",
+        backupCount=2,
+    )
+    debug_file_handler.setLevel(logging.DEBUG)
+    debug_file_handler.setFormatter(formatter)
+
+    info_file_handler = RotatingFileHandler(
+        filename=os.path.join(log_directory, "info.log"),
+        maxBytes=1*1024*1024,  # 1MB
+        encoding="utf-8",
+        backupCount=2,
+    )
+    info_file_handler.setLevel(logging.INFO)
+    info_file_handler.setFormatter(formatter)
+
+    error_file_handler = RotatingFileHandler(
+        filename=os.path.join(log_directory, "error.log"),
+        maxBytes=1*1024*1024,  # 1MB
+        encoding="utf-8",
+        backupCount=2,
+    )
+    error_file_handler.setLevel(logging.ERROR)
+    error_file_handler.setFormatter(formatter)
+
+    logging.basicConfig(
+        level=logging.DEBUG,
+        datefmt="%a, %d %b %Y %H:%M:%S",
+        handlers=[
+            debug_file_handler,
+            info_file_handler,
+            error_file_handler,
+        ]
+    )
+
+
+def setup_time_rotating(log_directory: str, tz_info: str = "Asia/Shanghai"):
+    fmt = "%(asctime)s|%(name)s|%(levelname)s|%(filename)s|%(lineno)d|%(message)s"
+
+    formatter = logging.Formatter(
+        fmt=fmt,
+        datefmt="%Y-%m-%d %H:%M:%S %z"
+    )
+    formatter.converter = get_converter(tz_info)
+
+    stream_handler = logging.StreamHandler()
+    stream_handler.setLevel(logging.INFO)
+    stream_handler.setFormatter(formatter)
+
+    # main
+    main_logger = logging.getLogger("main")
+    main_logger.addHandler(stream_handler)
+    main_info_file_handler = TimedRotatingFileHandler(
+        filename=os.path.join(log_directory, "main.log"),
+        encoding="utf-8",
+        when="midnight",
+        interval=1,
+        backupCount=7
+    )
+    main_info_file_handler.setLevel(logging.INFO)
+    main_info_file_handler.setFormatter(formatter)
+    main_logger.addHandler(main_info_file_handler)
+
+    # http
+    http_logger = logging.getLogger("http")
+    http_file_handler = TimedRotatingFileHandler(
+        filename=os.path.join(log_directory, "http.log"),
+        encoding='utf-8',
+        when="midnight",
+        interval=1,
+        backupCount=7
+    )
+    http_file_handler.setLevel(logging.DEBUG)
+    http_file_handler.setFormatter(formatter)
+    http_logger.addHandler(http_file_handler)
+
+    # api
+    api_logger = logging.getLogger("api")
+    api_file_handler = TimedRotatingFileHandler(
+        filename=os.path.join(log_directory, "api.log"),
+        encoding='utf-8',
+        when="midnight",
+        interval=1,
+        backupCount=7
+    )
+    api_file_handler.setLevel(logging.DEBUG)
+    api_file_handler.setFormatter(formatter)
+    api_logger.addHandler(api_file_handler)
+
+    # toolbox
+    toolbox_logger = logging.getLogger("toolbox")
+    toolbox_logger.addHandler(stream_handler)
+    toolbox_file_handler = RotatingFileHandler(
+        filename=os.path.join(log_directory, "toolbox.log"),
+        maxBytes=10*1024*1024,  # 10MB
+        encoding="utf-8",
+        backupCount=2,
+    )
+    toolbox_file_handler.setLevel(logging.DEBUG)
+    toolbox_file_handler.setFormatter(formatter)
+    toolbox_logger.addHandler(toolbox_file_handler)
+
+    # alarm
+    alarm_logger = logging.getLogger("alarm")
+    alarm_file_handler = TimedRotatingFileHandler(
+        filename=os.path.join(log_directory, "alarm.log"),
+        encoding="utf-8",
+        when="midnight",
+        interval=1,
+        backupCount=7
+    )
+    alarm_file_handler.setLevel(logging.DEBUG)
+    alarm_file_handler.setFormatter(formatter)
+    alarm_logger.addHandler(alarm_file_handler)
+
+    debug_file_handler = TimedRotatingFileHandler(
+        filename=os.path.join(log_directory, "debug.log"),
+        encoding="utf-8",
+        when="D",
+        interval=1,
+        backupCount=7
+    )
+    debug_file_handler.setLevel(logging.DEBUG)
+    debug_file_handler.setFormatter(formatter)
+
+    info_file_handler = TimedRotatingFileHandler(
+        filename=os.path.join(log_directory, "info.log"),
+        encoding="utf-8",
+        when="D",
+        interval=1,
+        backupCount=7
+    )
+    info_file_handler.setLevel(logging.INFO)
+    info_file_handler.setFormatter(formatter)
+
+    error_file_handler = TimedRotatingFileHandler(
+        filename=os.path.join(log_directory, "error.log"),
+        encoding="utf-8",
+        when="D",
+        interval=1,
+        backupCount=7
+    )
+    error_file_handler.setLevel(logging.ERROR)
+    error_file_handler.setFormatter(formatter)
+
+    logging.basicConfig(
+        level=logging.DEBUG,
+        datefmt="%a, %d %b %Y %H:%M:%S",
+        handlers=[
+            debug_file_handler,
+            info_file_handler,
+            error_file_handler,
+        ]
+    )
+
+
+if __name__ == "__main__":
+    pass

main.py

@@ -0,0 +1,135 @@
+#!/usr/bin/python3
+# -*- coding: utf-8 -*-
+"""
+docker build -t video_platform:v20250819_2052 .
+docker stop video_platform_7871 && docker rm video_platform_7871
+docker run -itd \
+--name video_platform_7871 \
+--restart=always \
+--network host \
+-e server_port=7870 \
+video_platform:v20250819_2052 /bin/bash
+
+docker run -itd \
+--name video_platform_7871 \
+--restart=always \
+--network host \
+-e server_port=7871 \
+-v /data/tianxing/PycharmProjects/video_platform:/data/tianxing/PycharmProjects/video_platform \
+python:3.12 /bin/bash
+
+nohup python3 main.py --server_port 7871 &
+"""
+import argparse
+import asyncio
+import logging
+import platform
+import time
+import threading
+
+import gradio as gr
+
+import log
+from project_settings import environment, project_path, log_directory, time_zone_info
+
+log.setup_size_rotating(log_directory=log_directory, tz_info=time_zone_info)
+
+from toolbox.os.command import Command
+from toolbox.porter.manager import PorterManager
+from tabs.fs_tab import get_fs_tab
+from tabs.shell_tab import get_shell_tab
+from tabs.youtube_player_tab import get_youtube_player_tab
+
+
+logger = logging.getLogger("main")
+
+
+def get_args():
+    parser = argparse.ArgumentParser()
+    parser.add_argument(
+        "--live_info_collect_tasks_file",
+        default=(project_path / "data/live_info_collect_tasks.json").as_posix(),
+        type=str
+    )
+    parser.add_argument(
+        "--live_recorder_tasks_file",
+        default=(project_path / "data/live_recorder_tasks.json").as_posix(),
+        type=str
+    )
+    parser.add_argument(
+        "--video_download_tasks_file",
+        default=(project_path / "data/video_download_tasks.json").as_posix(),
+        type=str
+    )
+    parser.add_argument(
+        "--youtube_video_upload_tasks_file",
+        default=(project_path / "data/youtube_video_upload_tasks.json").as_posix(),
+        type=str
+    )
+    parser.add_argument(
+        "--bilibili_video_upload_tasks_file",
+        default=(project_path / "data/bilibili_video_upload_tasks.json").as_posix(),
+        type=str
+    )
+    parser.add_argument(
+        "--live_records_dir",
+        default=(project_path / "data/live_records").as_posix(),
+        type=str
+    )
+    parser.add_argument(
+        "--server_port",
+        default=environment.get("server_port", 7860),
+        type=int
+    )
+
+    args = parser.parse_args()
+    return args
+
+
+def shell(cmd: str):
+    return Command.popen(cmd)
+
+
+def async_thread_wrapper(coro_task):
+    logger.info("background_task start")
+
+    loop = asyncio.new_event_loop()
+    asyncio.set_event_loop(loop)
+    loop.run_until_complete(coro_task)
+    return
+
+
+def main():
+    args = get_args()
+
+    porter_task_file = environment.get("porter_tasks_file")
+
+    # porter manager
+    video_download_manager = PorterManager(porter_task_file)
+    video_download_thread = threading.Thread(target=async_thread_wrapper,
+                                             args=(video_download_manager.run(),),
+                                             daemon=True)
+    video_download_thread.start()
+    time.sleep(5)
+
+    # ui
+    with gr.Blocks() as blocks:
+        gr.Markdown(value="live recording.")
+        with gr.Tabs():
+            _ = get_fs_tab()
+            _ = get_shell_tab()
+            _ = get_youtube_player_tab()
+
+    # http://127.0.0.1:7870/
+    # http://10.75.27.247:7870/
+    blocks.queue().launch(
+        # share=True,
+        share=False if platform.system() == "Windows" else False,
+        server_name="127.0.0.1" if platform.system() == "Windows" else "0.0.0.0",
+        server_port=args.server_port
+    )
+    return
+
+
+if __name__ == "__main__":
+    main()

project_settings.py

@@ -0,0 +1,27 @@
+#!/usr/bin/python3
+# -*- coding: utf-8 -*-
+import os
+from pathlib import Path
+
+from toolbox.os.environment import EnvironmentManager
+
+
+project_path = os.path.abspath(os.path.dirname(__file__))
+project_path = Path(project_path)
+
+time_zone_info = "Asia/Shanghai"
+
+log_directory = project_path / "logs"
+log_directory.mkdir(parents=True, exist_ok=True)
+
+temp_directory = project_path / "temp"
+temp_directory.mkdir(parents=True, exist_ok=True)
+
+environment = EnvironmentManager(
+    path=os.path.join(project_path, "dotenv"),
+    env=os.environ.get("environment", "dev"),
+)
+
+
+if __name__ == '__main__':
+    pass

requirements.txt

@@ -0,0 +1,20 @@
+gradio==5.33.0
+python-dotenv==1.0.1
+pandas==2.2.3
+openpyxl==3.1.5
+streamlink>=6.8.2
+httpx[http2]>=0.27.0
+ffmpeg-python>=0.2.0
+loguru>=0.7.2
+jsonpath-ng>=1.6.1
+jsengine>=1.0.7.post1
+quickjs>=1.19.4
+httpx-socks[asyncio]>=0.9.1
+google-api-python-client
+google-auth-oauthlib
+google-auth-httplib2
+beautifulsoup4==4.13.4
+aiocache==0.12.3
+biliupload
+cacheout
+tenacity

tabs/fs_tab.py

@@ -0,0 +1,62 @@
+#!/usr/bin/python3
+# -*- coding: utf-8 -*-
+import gradio as gr
+
+from project_settings import project_path
+
+
+def get_fs_tab():
+    with gr.TabItem("fs"):
+        with gr.Row():
+            with gr.Column(scale=3):
+                fs_filename = gr.Textbox(label="filename", max_lines=10)
+                fs_file = gr.File(label="file")
+                fs_file_dir = gr.Textbox(value="data", label="file_dir")
+                fs_query = gr.Button("query", variant="primary")
+            with gr.Column(scale=7):
+                fs_filelist_dataset_state = gr.State(value=[])
+                fs_filelist_dataset = gr.Dataset(
+                    components=[fs_filename, fs_file],
+                    samples=fs_filelist_dataset_state.value,
+                )
+
+        def when_click_query_files(file_dir: str = "data"):
+            file_dir = project_path / file_dir
+            dataset_state = list()
+            for filename in file_dir.glob("**/*.*"):
+                if filename.is_dir():
+                    continue
+                if filename.stem.startswith("."):
+                    continue
+                if filename.name.endswith(".py"):
+                    continue
+                dataset_state.append((
+                    filename.relative_to(file_dir).as_posix(),
+                    filename.as_posix(),
+                ))
+
+            dataset = gr.Dataset(
+                components=[fs_filename, fs_file],
+                samples=dataset_state,
+            )
+            return dataset_state, dataset
+
+        fs_filelist_dataset.click(
+            fn=lambda x: (
+                x[1], x[1]
+            ),
+            inputs=[fs_filelist_dataset],
+            outputs=[fs_filename, fs_file]
+        )
+        fs_query.click(
+            fn=when_click_query_files,
+            inputs=[fs_file_dir],
+            outputs=[fs_filelist_dataset_state, fs_filelist_dataset]
+        )
+    return locals()
+
+
+if __name__ == "__main__":
+    with gr.Blocks() as block:
+        fs_components = get_fs_tab()
+        block.launch()

tabs/shell_tab.py

@@ -0,0 +1,28 @@
+#!/usr/bin/python3
+# -*- coding: utf-8 -*-
+import gradio as gr
+
+from toolbox.os.command import Command
+
+
+def shell(cmd: str):
+    return Command.popen(cmd)
+
+
+def get_shell_tab():
+    with gr.TabItem("shell"):
+        shell_text = gr.Textbox(label="cmd")
+        shell_button = gr.Button("run")
+        shell_output = gr.Textbox(label="output", max_lines=100)
+
+        shell_button.click(
+            shell,
+            inputs=[shell_text, ],
+            outputs=[shell_output],
+        )
+
+    return locals()
+
+
+if __name__ == "__main__":
+    pass

tabs/youtube_player_tab.py

@@ -0,0 +1,195 @@
+#!/usr/bin/python3
+# -*- coding: utf-8 -*-
+import json
+import re
+
+import gradio as gr
+from jinja2 import Template
+import tempfile
+
+
+html_content = """
+<!DOCTYPE html>
+<html lang="en">
+<head>
+    <meta charset="UTF-8">
+    <meta name="viewport" content="width=device-width, initial-scale=1.0">
+    <title>YouTube视频连续播放器</title>
+    <style>
+        body {
+            font-family: Arial, sans-serif;
+            max-width: 1600px;
+            margin: 0 auto;
+            padding: 20px;
+            text-align: center;
+        }
+        #player {
+            width: 1200px;
+            height: 675px;
+            # aspect-ratio: 16/9;
+        }
+        #playlist {
+            text-align: left;
+            margin: 20px 0;
+        }
+        .current {
+            font-weight: bold;
+            color: #ff0000;
+        }
+    </style>
+</head>
+<body>
+    <h1>YouTube视频连续播放器</h1>
+    <div id="player"></div>
+    <div id="playlist"></div>
+    <div id="status">准备播放...</div>
+
+    <script>
+
+        // 获取所有视频ID
+        const videoIds = {{ video_ids }};
+
+        let currentVideoIndex = 0;
+        let player;
+
+        // 加载YouTube IFrame API
+        function onYouTubeIframeAPIReady() {
+            player = new YT.Player('player', {
+                height: '390',
+                width: '640',
+                events: {
+                    'onReady': onPlayerReady,
+                    'onStateChange': onPlayerStateChange
+                }
+            });
+        }
+
+        // 播放器准备就绪
+        function onPlayerReady(event) {
+            loadAndPlayVideo(currentVideoIndex);
+            updatePlaylist();
+        }
+
+        // 播放器状态变化
+        function onPlayerStateChange(event) {
+            if (event.data === YT.PlayerState.ENDED) {
+                // 当前视频结束，播放下一个
+                currentVideoIndex++;
+                if (currentVideoIndex < videoIds.length) {
+                    loadAndPlayVideo(currentVideoIndex);
+                    updatePlaylist();
+                } else {
+                    document.getElementById('status').textContent = '所有视频播放完毕';
+                }
+            }
+        }
+
+        // 加载并播放指定索引的视频
+        function loadAndPlayVideo(index) {
+            if (index >= 0 && index < videoIds.length) {
+                player.loadVideoById(videoIds[index]);
+                document.getElementById('status').textContent = `正在播放: 视频 ${index + 1}/${videoIds.length}`;
+            }
+        }
+
+        // 更新播放列表显示
+        function updatePlaylist() {
+            const playlistElement = document.getElementById('playlist');
+            playlistElement.innerHTML = '<h3>播放列表:</h3>';
+            
+            videoIds.forEach((id, index) => {
+                const item = document.createElement('div');
+                item.textContent = `视频 ${index + 1}: https://youtu.be/${id}`;
+                if (index === currentVideoIndex) {
+                    item.className = 'current';
+                }
+                playlistElement.appendChild(item);
+            });
+        }
+
+        // 动态加载YouTube IFrame API
+        const tag = document.createElement('script');
+        tag.src = "https://www.youtube.com/iframe_api";
+        const firstScriptTag = document.getElementsByTagName('script')[0];
+        firstScriptTag.parentNode.insertBefore(tag, firstScriptTag);
+    </script>
+</body>
+</html>
+"""
+
+
+def make_file(content: str):
+    with tempfile.NamedTemporaryFile(mode='w', delete=False, suffix='.html') as tmp:
+        tmp.write(content)
+        tmp_path = tmp.name
+
+    return tmp_path
+
+
+def get_video_id(url: str):
+    """
+    从各种YouTube URL格式中提取视频ID
+    支持的格式包括：
+    - 普通URL: https://www.youtube.com/watch?v=dQw4w9WgXcQ
+    - 短链接: https://youtu.be/dQw4w9WgXcQ
+    - 嵌入链接: https://www.youtube.com/embed/dQw4w9WgXcQ
+    - 带时间戳: https://youtu.be/dQw4w9WgXcQ?t=120
+    - 带其他参数: https://www.youtube.com/watch?v=dQw4w9WgXcQ&feature=share
+    - 移动端链接: https://m.youtube.com/watch?v=dQw4w9WgXcQ
+    - 无协议链接: youtube.com/watch?v=dQw4w9WgXcQ
+    - 仅视频ID: dQw4w9WgXcQ
+    """
+    pattern = r'(?:youtube\.com\/(?:watch\?v=|embed\/|shorts\/|live\/)|youtu\.be\/)([^"&?\/\s]{11})'
+    match = re.search(pattern, url)
+
+    if match is None:
+        return None
+    video_id = match.group(1)
+    return video_id
+
+
+def when_click_make_html_button(playlist: str):
+    urls = [url.strip() for url in playlist.split("\n") if url.strip()]
+    video_ids = [get_video_id(url) for url in urls]
+    video_ids = [video_id for video_id in video_ids if video_id is not None]
+
+    video_ids_ = json.dumps(video_ids, ensure_ascii=False)
+
+    template = Template(html_content)
+    variables = {
+        "video_ids": video_ids_
+    }
+    html_content_ = template.render(variables)
+
+    filename = make_file(html_content_)
+    return filename
+
+
+def get_youtube_player_tab():
+    with gr.TabItem("youtube_player"):
+        with gr.Row():
+            with gr.Column(scale=5):
+                yp_playlist = gr.Textbox(
+                    label="playlist",
+                    value="https://www.youtube.com/watch?v=tvRNE-ULe94\nhttps://www.youtube.com/watch?v=VQmaDgmyIBc\nhttps://www.youtube.com/watch?v=e_sFLaQUN3k\nhttps://www.youtube.com/watch?v=-VDYnk0jM8Q",
+                    placeholder="每行输入一个YouTube视频链接\n例如:\nhttps://www.youtube.com/watch?v=dQw4w9WgXcQ\nhttps://youtu.be/9bZkp7q19f0",
+                    lines=10,
+                )
+                yp_make_html_button = gr.Button("make_html")
+
+            with gr.Column(scale=5):
+                yp_video_output = gr.File(label="play_html_file")
+
+            yp_make_html_button.click(
+                fn=when_click_make_html_button,
+                inputs=[yp_playlist],
+                outputs=[yp_video_output]
+            )
+
+    return locals()
+
+
+if __name__ == "__main__":
+    with gr.Blocks() as block:
+        fs_components = get_youtube_player_tab()
+        block.launch()

toolbox/__init__.py

@@ -0,0 +1,6 @@
+#!/usr/bin/python3
+# -*- coding: utf-8 -*-
+
+
+if __name__ == "__main__":
+    pass

toolbox/asyncio/__init__.py

@@ -0,0 +1,6 @@
+#!/usr/bin/python3
+# -*- coding: utf-8 -*-
+
+
+if __name__ == "__main__":
+    pass

toolbox/asyncio/cacheout.py

@@ -0,0 +1,194 @@
+#!/usr/bin/python3
+# -*- coding: utf-8 -*-
+import asyncio
+import time
+from functools import wraps
+
+import time
+import hashlib
+import pickle
+import asyncio
+from functools import wraps
+from typing import Any, Callable, Dict, Optional, Tuple
+from dataclasses import dataclass
+from collections import OrderedDict
+
+
+@dataclass
+class CacheEntry:
+    expire_time: float
+    result: Any
+    access_time: float
+
+
+def async_cache_decorator(
+        max_age: int = 10,
+        max_size: Optional[int] = 100,
+        ignore_kwargs: Optional[list] = None
+):
+    """
+    高级缓存装饰器
+
+    Args:
+        max_age: 缓存最大年龄（秒）
+        max_size: 缓存最大大小（None表示无限制）
+        ignore_kwargs: 忽略的kwargs参数名列表
+    """
+    ignore_kwargs = ignore_kwargs or []
+
+    def decorator(func: Callable):
+        cache: Dict[str, CacheEntry] = {}
+        access_order = OrderedDict()  # 用于LRU淘汰
+
+        @wraps(func)
+        async def wrapper(*args, **kwargs):
+            # 清理过期的缓存
+            _clean_expired_cache()
+
+            # 生成缓存键（忽略指定的kwargs）
+            filtered_kwargs = {k: v for k, v in kwargs.items() if k not in ignore_kwargs}
+            cache_key = _generate_cache_key(args, filtered_kwargs)
+            current_time = time.time()
+
+            # 检查缓存
+            if cache_key in cache:
+                entry = cache[cache_key]
+                if current_time < entry.expire_time:
+                    # 更新访问时间和LRU顺序
+                    entry.access_time = current_time
+                    access_order.move_to_end(cache_key)
+                    return entry.result
+                else:
+                    # 缓存过期
+                    _remove_from_cache(cache_key)
+
+            # 调用原函数
+            result = await func(*args, **kwargs)
+
+            # 检查缓存大小并可能淘汰
+            if max_size and len(cache) >= max_size:
+                # 移除最久未使用的缓存
+                oldest_key = next(iter(access_order))
+                _remove_from_cache(oldest_key)
+
+            # 添加新缓存
+            cache[cache_key] = CacheEntry(
+                expire_time=current_time + max_age,
+                result=result,
+                access_time=current_time
+            )
+            access_order[cache_key] = True
+
+            return result
+
+        def _clean_expired_cache():
+            """清理过期缓存"""
+            current_time = time.time()
+            expired_keys = [
+                key for key, entry in cache.items()
+                if current_time >= entry.expire_time
+            ]
+            for key in expired_keys:
+                _remove_from_cache(key)
+
+        def _remove_from_cache(cache_key: str):
+            """从缓存中移除指定键"""
+            if cache_key in cache:
+                del cache[cache_key]
+            if cache_key in access_order:
+                del access_order[cache_key]
+
+        # 缓存管理方法
+        def clear_cache():
+            """清空所有缓存"""
+            cache.clear()
+            access_order.clear()
+
+        def remove_from_cache(*args, **kwargs):
+            """移除特定参数的缓存"""
+            filtered_kwargs = {k: v for k, v in kwargs.items() if k not in ignore_kwargs}
+            cache_key = _generate_cache_key(args, filtered_kwargs)
+            if cache_key in cache:
+                _remove_from_cache(cache_key)
+                return True
+            return False
+
+        def get_cache_info() -> Dict[str, Any]:
+            """获取缓存统计信息"""
+            current_time = time.time()
+            valid_entries = sum(
+                1 for entry in cache.values()
+                if current_time < entry.expire_time
+            )
+            return {
+                'total_entries': len(cache),
+                'valid_entries': valid_entries,
+                'max_size': max_size,
+                'max_age': max_age,
+                'cache_keys': list(cache.keys())
+            }
+
+        def get_cached_result(*args, **kwargs):
+            """获取缓存结果（如果存在且未过期）"""
+            filtered_kwargs = {k: v for k, v in kwargs.items() if k not in ignore_kwargs}
+            cache_key = _generate_cache_key(args, filtered_kwargs)
+            current_time = time.time()
+
+            if cache_key in cache:
+                entry = cache[cache_key]
+                if current_time < entry.expire_time:
+                    return entry.result
+            return None
+
+        # 附加方法
+        wrapper.clear_cache = clear_cache
+        wrapper.remove_from_cache = remove_from_cache
+        wrapper.get_cache_info = get_cache_info
+        wrapper.get_cached_result = get_cached_result
+
+        return wrapper
+
+    return decorator
+
+
+def _generate_cache_key(args: tuple, kwargs: dict) -> str:
+    """生成缓存键"""
+    try:
+        # 尝��使用更高效的序列化
+        key_data = pickle.dumps((args, sorted(kwargs.items())))
+        return hashlib.sha256(key_data).hexdigest()
+    except (TypeError, pickle.PickleError):
+        # 如果无法序列化，使用字符串表示
+        args_str = str(args)
+        kwargs_str = str(sorted(kwargs.items()))
+        return hashlib.sha256(f"{args_str}:{kwargs_str}".encode()).hexdigest()
+
+
+@async_cache_decorator(10)
+async def call_api():
+    await asyncio.sleep(1)
+    return {"data": f"API响应时间: {time.time()}", "status": "success"}
+
+
+async def main():
+    # 第一次调用 - 实际调用API
+    result1 = await call_api()
+    print("第一次结果:", result1)
+
+    # 立即再次调用 - 返回缓存结果
+    result2 = await call_api()
+    print("第二次结果:", result2)
+
+    # 等待5秒后调用 - 返回缓存结果
+    await asyncio.sleep(5)
+    result3 = await call_api()
+    print("第三次结果:", result3)
+
+    # 等待11秒后调用 - 重新调用API
+    await asyncio.sleep(6)  # 总共等待11秒
+    result4 = await call_api()
+    print("第四次结果:", result4)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

toolbox/bilibili/__init__.py

@@ -0,0 +1,6 @@
+#!/usr/bin/python3
+# -*- coding: utf-8 -*-
+
+
+if __name__ == "__main__":
+    pass

toolbox/bilibili/bilibili_client.py

@@ -0,0 +1,241 @@
+#!/usr/bin/python3
+# -*- coding: utf-8 -*-
+"""
+https://pypi.org/project/biliupload/
+
+https://github.com/SocialSisterYi/bilibili-API-collect
+
+"""
+import argparse
+import hashlib
+import json
+import logging
+import subprocess
+import time
+from urllib.parse import urlencode, urlparse
+
+import qrcode
+import requests
+import requests.utils
+
+from project_settings import project_path
+from toolbox.design_patterns.singleton import ParamsSingleton
+
+
+logger = logging.getLogger("toolbox")
+
+
+class BilibiliUtils(object):
+    app_key = "4409e2ce8ffd12b8"
+    app_sec = "59b43e04ad6965f34319062b478f83dd"
+
+    @classmethod
+    def signature(cls, params):
+        params["appkey"] = cls.app_key
+        keys = sorted(params.keys())
+        query = "&".join(f"{k}={params[k]}" for k in keys)
+        query += cls.app_sec
+        md5_hash = hashlib.md5(query.encode("utf-8")).hexdigest()
+        params["sign"] = md5_hash
+
+    @staticmethod
+    def map_to_string(params):
+        return urlencode(params)
+
+    @classmethod
+    def execute_curl_command(cls, api, data):
+        data_string = cls.map_to_string(data)
+        headers = "Content-Type: application/x-www-form-urlencoded"
+        curl_command = f"curl -X POST -H \"{headers}\" -d \"{data_string}\" {api}"
+        result = subprocess.run(
+            curl_command, shell=True, capture_output=True, text=True, encoding="utf-8"
+        )
+        if result.returncode != 0:
+            raise Exception(f"curl command failed: {result.stderr}")
+        return json.loads(result.stdout)
+
+
+class BilibiliClient(BilibiliUtils, ParamsSingleton):
+    headers = {
+        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36"
+    }
+
+    def __init__(self):
+        if not self._initialized:
+            self.credentials = None
+            self.cookies = None
+
+            self._session = requests.Session()
+
+            self._initialized = True
+
+    @property
+    def session(self):
+        if not self._session.cookies:
+            self._session.headers = self.headers
+            self._session.cookies = requests.utils.cookiejar_from_dict(self.cookies)
+        return self._session
+
+    @classmethod
+    def get_tv_qrcode_url_and_auth_code(cls):
+        api = "https://passport.bilibili.com/x/passport-tv-login/qrcode/auth_code"
+        data = {
+            "local_id": "0",
+            "ts": str(int(time.time()))
+        }
+        cls.signature(data)
+        body = cls.execute_curl_command(api, data)
+        if body["code"] == 0:
+            qrcode_url = body["data"]["url"]
+            auth_code = body["data"]["auth_code"]
+            return qrcode_url, auth_code
+        else:
+            raise Exception("get_tv_qrcode_url_and_auth_code error")
+
+    @classmethod
+    def verify_login(cls, auth_code):
+        api = "https://passport.bilibili.com/x/passport-tv-login/qrcode/poll"
+        data = {
+            "auth_code": auth_code,
+            "local_id": "0",
+            "ts": str(int(time.time()))
+        }
+        cls.signature(data)
+        while True:
+            body = cls.execute_curl_command(api, data)
+            if body["code"] == 0:
+                print("Login success!")
+                return body
+            else:
+                time.sleep(3)
+
+    def set_cookies(self, credentials: dict):
+        access_token = credentials["data"]["access_token"]
+        sessdata_value = credentials["data"]["cookie_info"]["cookies"][0]["value"]
+        bili_jct_value = credentials["data"]["cookie_info"]["cookies"][1]["value"]
+        dede_user_id_value = credentials["data"]["cookie_info"]["cookies"][2]["value"]
+        dede_user_id_ckmd5_value = credentials["data"]["cookie_info"]["cookies"][3]["value"]
+        sid_value = credentials["data"]["cookie_info"]["cookies"][4]["value"]
+        cookies = {
+            # "access_token": access_token,
+            "SESSDATA": sessdata_value,
+            "bili_jct": bili_jct_value,
+            "DedeUserID": dede_user_id_value,
+            "DedeUserID__ckMd5": dede_user_id_ckmd5_value,
+            # "sid": sid_value,
+        }
+        self.cookies = cookies
+        return self.cookies
+
+    def login_with_qrcode_url(self):
+        input("Please maximize the window to ensure the QR code is fully displayed, press Enter to continue: ")
+        login_url, auth_code = self.get_tv_qrcode_url_and_auth_code()
+        qr = qrcode.QRCode()
+        qr.add_data(login_url)
+        qr.print_ascii()
+        print("Or copy this link to your phone Bilibili:", login_url)
+        credentials = self.verify_login(auth_code)
+        self.credentials = credentials
+        self.set_cookies(credentials)
+        return True
+
+    def login_with_credentials_file(self, credentials_file: str):
+        with open(credentials_file, "r", encoding="utf-8") as f:
+            credentials = json.load(f)
+        self.credentials = credentials
+        self.set_cookies(credentials)
+        return True
+
+    def login_with_credentials_info(self, credentials_info: dict):
+        self.credentials = credentials_info
+        self.set_cookies(credentials_info)
+        return True
+
+    def check_login(self):
+        url = "https://api.bilibili.com/x/web-interface/nav"
+        with requests.Session() as session:
+            session.headers = self.headers
+            session.cookies = requests.utils.cookiejar_from_dict(self.cookies)
+            response = session.get(url)
+            if response.status_code == 200:
+                response_data = json.loads(response.text)
+                if response_data["data"]["isLogin"] is True:
+                    return True
+                else:
+                    return False
+            else:
+                logger.error(f"Check failed, please check the info; status_code: {response.status_code}, text: {response.text}")
+                return False
+
+    def get_now(self):
+        url = "https://api.bilibili.com/x/report/click/now"
+
+        response = self.session.get(
+            url,
+            headers=self.headers,
+        )
+        if response.status_code != 200:
+            raise AssertionError(f"request failed; status_code: {response.status_code}, text: {response.text}")
+        js = response.json()
+        return js
+
+    def get_version(self):
+        url = "https://api.live.bilibili.com/xlive/app-blink/v1/liveVersionInfo/getHomePageLiveVersion"
+
+        js = self.get_now()
+        ts = js["data"]["now"]
+
+        params = {
+            "system_version": 2,
+            "ts": ts,
+        }
+        self.signature(params)
+
+        response = self.session.get(
+            url,
+            headers=self.headers,
+            params=params,
+        )
+        if response.status_code != 200:
+            raise AssertionError(f"request failed; status_code: {response.status_code}, text: {response.text}")
+        js = response.json()
+        return js
+
+
+def get_args():
+    parser = argparse.ArgumentParser()
+    parser.add_argument(
+        "--credentials_file",
+        default=(project_path / "dotenv/bilibili_chenjiesen_credentials.json").as_posix(),
+        type=str
+    )
+    args = parser.parse_args()
+    return args
+
+
+def main():
+    import log
+    from project_settings import project_path, log_directory
+
+    log.setup_size_rotating(log_directory=log_directory)
+
+    args = get_args()
+
+    client = BilibiliClient()
+
+    flag = client.check_login()
+    print(f"flag: {flag}")
+    client.login_with_credentials_file(args.credentials_file)
+    # client.login_with_qrcode_url()
+    flag = client.check_login()
+    print(f"flag: {flag}")
+
+    # result = client.get_room_id()
+    # result = client.get_now()
+    result = client.get_version()
+    print(result)
+    return
+
+
+if __name__ == "__main__":
+    main()

toolbox/bilibili/live/__init__.py

@@ -0,0 +1,6 @@
+#!/usr/bin/python3
+# -*- coding: utf-8 -*-
+
+
+if __name__ == "__main__":
+    pass

toolbox/bilibili/live/live_manager.py

@@ -0,0 +1,301 @@
+#!/usr/bin/python3
+# -*- coding: utf-8 -*-
+"""
+https://github.com/lateautumn233/Bilibili-Live-Stream/blob/main/bilibili-live.py
+
+https://github.com/ChaceQC/bilibili_live_stream_code/blob/master/main/bilibili_live_stream_code.py
+
+"""
+import argparse
+import logging
+import math
+from pathlib import Path
+import re
+import json
+import urllib
+import urllib.parse
+import hashlib
+
+import requests
+import requests.utils
+from tenacity import before_sleep_log, retry, retry_if_exception_type, stop_after_attempt, wait_fixed
+
+logger = logging.getLogger("toolbox")
+
+from project_settings import project_path
+from toolbox.bilibili.bilibili_client import BilibiliClient
+
+
+def appsign(params, appkey, appsec):
+    """
+    为请求参数进行app签名
+    :param params: 原参数
+    :param appkey: key
+    :param appsec: key对应的secret
+    :return:
+    """
+    params.update({'appkey': appkey})
+    params = dict(sorted(params.items()))  # 按照 key 重排参数
+    query = urllib.parse.urlencode(params)  # 序列化参数
+    sign = hashlib.md5((query + appsec).encode()).hexdigest()  # 计算 api 签名
+    params.update({'sign': sign})
+    return params
+
+
+class RtmpPublishCMD(object):
+    def __init__(self, cmd_list: list, cmd_str: str):
+        self.cmd_list = cmd_list
+        self.cmd_str = cmd_str
+
+
+class BilibiliLiveManager(BilibiliClient):
+    def __init__(self):
+        super().__init__()
+
+    def set_live_title(self, title: str):
+        """
+        :return
+        {
+            "code": 0,
+            "msg": "ok",
+            "message": "ok",
+            "data": {
+                "sub_session_key": "",
+                "audit_info": {
+                    "audit_title_reason": "进入审核",
+                    "update_title": "",
+                    "audit_title_status": 0,
+                    "audit_title": "设置直播标题"
+                }
+            }
+        }
+        """
+        url = "https://api.live.bilibili.com/room/v1/Room/update"
+
+        js = self.get_room_id_by_uid()
+        room_id = js["data"]["room_id"]
+
+        bili_jct = self.cookies["bili_jct"]
+
+        data = {
+            "room_id": room_id,
+            "platform": "pc_link",
+            "title": title,
+            "csrf_token": bili_jct,
+            "csrf": bili_jct,
+        }
+        response = self.session.post(
+            url,
+            headers=self.headers,
+            data=data,
+        )
+        if response.status_code != 200:
+            raise AssertionError(f"request failed; status_code: {response.status_code}, text: {response.text}")
+        js = response.json()
+        # print(json.dumps(js, ensure_ascii=False, indent=4))
+        return js
+
+    def get_room_id_by_uid(self):
+        dede_user_id = self.cookies.get("DedeUserID")
+
+        url = f"https://api.live.bilibili.com/room/v2/Room/room_id_by_uid?uid={dede_user_id}"
+        response = self.session.get(
+            url,
+            headers=self.headers
+        )
+        if response.status_code != 200:
+            raise AssertionError(f"request failed; status_code: {response.status_code}, text: {response.text}")
+        js = response.json()
+        return js
+
+    def get_rtmp_publish_cmd_by_flv_file(self, input_source: str):
+        js = self.get_rtmp_info_by_room_id()
+        # print(json.dumps(js, ensure_ascii=False, indent=4))
+        rtmp_code = js["data"]["rtmp"]["code"]
+        rtmp_addr = js["data"]["rtmp"]["addr"]
+
+        rtmp_url = f"{rtmp_addr}/{rtmp_code}"
+
+        # cmd = f'ffmpeg -re -i "{input_source}" -c copy -f flv -flvflags no_duration_filesize "{rtmp_url}"'
+
+        # cmd_list = [
+        #     'ffmpeg',
+        #     '-i', input_source,
+        #     '-c', 'copy',
+        #     '-f', 'flv',
+        #     rtmp_url
+        # ]
+        cmd_list = [
+            "ffmpeg",
+            "-i", input_source,
+            "-c", "copy",
+            "-f", "flv",
+            rtmp_url
+        ]
+        cmd_str = [
+            "ffmpeg",
+            '-i', f'"{input_source}"',
+            "-c", "copy",
+            "-f", "flv",
+            f'"{rtmp_url}"'
+        ]
+        cmd_str = " ".join(cmd_str)
+        cmd = RtmpPublishCMD(
+            cmd_list=cmd_list,
+            cmd_str=cmd_str
+        )
+        return cmd
+
+    def get_rtmp_info_by_room_id(self):
+        """
+        :return:
+        {
+            "code": 0,
+            "data": {
+                "change": 1,
+                "status": "LIVE",
+                "try_time": "0000-00-00 00:00:00",
+                "room_type": 0,
+                "live_key": "630309426298156613",
+                "sub_session_key": "630309426298156613sub_time:1757133948",
+                "rtmp": {
+                    "type": 1,
+                    "addr": "rtmp://txy2.live-push.bilivideo.com/live-bvc/",
+                    "code": "?streamname=live_442286660_79175069&key=fd263bd4cad752a05c2e284adda16302&schedule=rtmp&pflag=2",
+                    "new_link": "",
+                    "provider": "txy2"
+                },
+                "protocols": [
+                    {
+                        "protocol": "rtmp",
+                        "addr": "rtmp://txy2.live-push.bilivideo.com/live-bvc/",
+                        "code": "?streamname=live_442286660_79175069&key=fd263bd4cad752a05c2e284adda16302&schedule=rtmp&pflag=2",
+                        "new_link": "",
+                        "provider": "txy"
+                    }
+                ],
+                "notice": {
+                    "type": 1,
+                    "status": 0,
+                    "title": "",
+                    "msg": "",
+                    "button_text": "",
+                    "button_url": ""
+                },
+                "qr": "",
+                "need_face_auth": false,
+                "service_source": "live-streaming",
+                "rtmp_backup": null,
+                "up_stream_extra": {
+                    "isp": "小运营商"
+                },
+                "collaboration_live_extra": null
+            },
+            "message": "",
+            "msg": ""
+        }
+        """
+        url = "https://api.live.bilibili.com/room/v1/Room/startLive"
+
+        js = self.get_room_id_by_uid()
+        room_id = js["data"]["room_id"]
+
+        js = self.get_version()
+        build = js["data"]["build"]
+        curr_version = js["data"]["curr_version"]
+
+        js = self.get_now()
+        ts = js["data"]["now"]
+
+        bili_jct = self.cookies["bili_jct"]
+        data = {
+            "room_id": room_id,
+            "platform": "pc_link",
+            "area_v2": "624",
+            "backup_stream": "0",
+            "csrf_token": bili_jct,
+            "csrf": bili_jct,
+            "build": build,
+            "version": curr_version,
+            "ts": ts,
+        }
+        self.signature(data)
+
+        response = self.session.post(
+            url,
+            headers=self.headers,
+            data=data,
+        )
+        if response.status_code != 200:
+            raise AssertionError(f"request failed; status_code: {response.status_code}, text: {response.text}")
+        js = response.json()
+        return js
+
+    @retry(
+        wait=wait_fixed(10),
+        stop=stop_after_attempt(3),
+        before_sleep=before_sleep_log(logger, logging.ERROR),
+    )
+    def stop_live(self):
+        url = "https://api.live.bilibili.com/room/v1/Room/stopLive"
+
+        js = self.get_room_id_by_uid()
+        room_id = js["data"]["room_id"]
+
+        bili_jct = self.cookies["bili_jct"]
+
+        data = {
+            "room_id": room_id,
+            "platform": "pc_link",
+            "csrf_token": bili_jct,
+            "csrf": bili_jct,
+        }
+        response = self.session.post(
+            url,
+            headers=self.headers,
+            data=data,
+        )
+        if response.status_code != 200:
+            raise AssertionError(f"request failed; status_code: {response.status_code}, text: {response.text}")
+        js = response.json()
+        return js
+
+
+def get_args():
+    parser = argparse.ArgumentParser()
+    parser.add_argument(
+        "--credentials_file",
+        default=(project_path / "dotenv/bilibili_chenjiesen_credentials.json").as_posix(),
+        type=str
+    )
+    args = parser.parse_args()
+    return args
+
+
+def main():
+    import log
+    from project_settings import project_path, log_directory
+
+    log.setup_size_rotating(log_directory=log_directory)
+
+    args = get_args()
+
+    client = BilibiliLiveManager()
+
+    flag = client.check_login()
+    print(f"flag: {flag}")
+    client.login_with_credentials_file(args.credentials_file)
+    # client.login_with_qrcode_url()
+    flag = client.check_login()
+    print(f"flag: {flag}")
+
+    # result = client.get_room_id()
+    # result = client.stop_live()
+    result = client.start_live_by_flv_file(args.room_id)
+    # result = client.set_live_title("设置直播标题")
+    print(result)
+    return
+
+
+if __name__ == "__main__":
+    main()

toolbox/bilibili/video/__init__.py

@@ -0,0 +1,5 @@
+#!/usr/bin/python3
+# -*- coding: utf-8 -*-
+
+if __name__ == "__main__":
+    pass

toolbox/bilibili/video/draft_manager.py

@@ -0,0 +1,279 @@
+#!/usr/bin/python3
+# -*- coding: utf-8 -*-
+"""
+https://pypi.org/project/biliupload/
+
+https://github.com/SocialSisterYi/bilibili-API-collect
+
+"""
+import argparse
+import logging
+
+from tenacity import before_sleep_log, retry, retry_if_exception_type, stop_after_attempt, wait_fixed
+
+logger = logging.getLogger("toolbox")
+
+from project_settings import project_path
+from toolbox.bilibili.video.video_manager import BilibiliVideoUploader
+
+
+class BilibiliVideoDraftUploader(BilibiliVideoUploader):
+    def __init__(self):
+        super().__init__()
+
+    @retry(
+        wait=wait_fixed(10),
+        stop=stop_after_attempt(3),
+        before_sleep=before_sleep_log(logger, logging.ERROR),
+    )
+    def list_draft(self):
+        url = f"https://member.bilibili.com/x/vupre/web/draft/list"
+
+        response = self.session.get(url)
+        if response.status_code != 200:
+            raise AssertionError(f"request failed; status_code: {response.status_code}, text: {response.text}")
+        js = response.json()
+        return js
+
+    @retry(
+        wait=wait_fixed(10),
+        stop=stop_after_attempt(3),
+        before_sleep=before_sleep_log(logger, logging.ERROR),
+    )
+    def add_draft(self, biz_id: int, bilibili_filename, metadata: dict):
+        csrf = self.cookies["bili_jct"]
+        url = "https://member.bilibili.com/x/vupre/web/draft/add"
+
+        params = {
+            "csrf": csrf,
+        }
+
+        data = {
+            "copyright": metadata["copyright"],
+            "videos": [
+                {
+                    "filename": bilibili_filename,
+                    "title": metadata["title"],
+                    "desc": metadata["desc"],
+                    "cid": biz_id,
+                }
+            ],
+
+            "source": metadata["source"],
+            "tid": metadata["tid"],
+            "title": metadata["title"],
+            "cover": metadata["cover"],
+            "tag": metadata["tag"],
+            "desc_format_id": 0,
+            "desc": metadata["desc"],
+            "dynamic": metadata["dynamic"],
+            "subtitle": {"open": 0, "lan": ""},
+
+        }
+
+        response = self.session.post(
+            url,
+            params=params,
+            json=data,
+            headers=self.headers,
+        )
+        if response.status_code != 200:
+            raise AssertionError(f"request failed; status_code: {response.status_code}, text: {response.text}")
+        js = response.json()
+        return js
+
+    @retry(
+        wait=wait_fixed(10),
+        stop=stop_after_attempt(3),
+        before_sleep=before_sleep_log(logger, logging.ERROR),
+    )
+    def update_draft(self, draft_id: int, biz_id: int, bilibili_filename, metadata: dict):
+        csrf = self.cookies["bili_jct"]
+        url = "https://member.bilibili.com/x/vupre/web/draft/update"
+
+        params = {
+            "csrf": csrf,
+        }
+
+        data = {
+            "id": draft_id,
+
+            "copyright": metadata["copyright"],
+            "videos": [
+                {
+                    "filename": bilibili_filename,
+                    "title": metadata["title"],
+                    "desc": metadata["desc"],
+                    "cid": biz_id,
+                }
+            ],
+
+            "source": metadata["source"],
+            "tid": metadata["tid"],
+            "title": metadata["title"],
+            "cover": metadata["cover"],
+            "tag": metadata["tag"],
+            "desc_format_id": 0,
+            "desc": metadata["desc"],
+            "dynamic": metadata["dynamic"],
+            "subtitle": {"open": 0, "lan": ""},
+
+        }
+
+        response = self.session.post(
+            url,
+            params=params,
+            json=data,
+            headers=self.headers,
+        )
+        if response.status_code != 200:
+            raise AssertionError(f"request failed; status_code: {response.status_code}, text: {response.text}")
+        js = response.json()
+        return js
+
+    @retry(
+        wait=wait_fixed(10),
+        stop=stop_after_attempt(3),
+        before_sleep=before_sleep_log(logger, logging.ERROR),
+    )
+    def publish_draft(self, draft_id: int, bilibili_filename: str, metadata: dict):
+        csrf = self.cookies["bili_jct"]
+        url = "https://member.bilibili.com/x/vu/web/add/v3"
+
+        params = {
+            "csrf": csrf,
+        }
+
+        data = {
+            "draft_id": draft_id,
+            "copyright": metadata["copyright"],
+            "videos":[
+                {
+                    "filename": bilibili_filename,
+                    "title": metadata["title"],
+                    "desc": metadata["desc"],
+                }
+            ],
+            "source": metadata["source"],
+            "tid": metadata["tid"],
+            "title": metadata["title"],
+            "cover": metadata["cover"],
+            "tag": metadata["tag"],
+            "desc_format_id": 0,
+            "desc": metadata["desc"],
+            "dynamic": metadata["dynamic"],
+            "subtitle": {"open": 0, "lan": ""},
+        }
+
+        if metadata["copyright"] != 2:
+            del data["source"]
+            # copyright: 1 original 2 reprint
+            data["copyright"] = 1
+
+        response = self.session.post(
+            url,
+            params=params,
+            json=data,
+            headers=self.headers,
+        )
+        if response.status_code != 200:
+            raise AssertionError(f"request failed; status_code: {response.status_code}, text: {response.text}")
+        js = response.json()
+        return js
+
+    def upload_video_draft_and_publish(self, filename: str, metadata: dict):
+        biz_id, bilibili_filename = self.upload_video_file(filename)
+        js = self.add_draft(biz_id, bilibili_filename, metadata)
+        if js["code"] != 0:
+            raise AssertionError(f"add draft; status_code: {js["code"]}, text: {js["message"]}")
+        js = self.list_draft()
+        if js["code"] != 0:
+            raise AssertionError(f"add draft; status_code: {js["code"]}, text: {js["message"]}")
+        draft_list = js["data"]
+
+        draft_id = None
+        for draft in draft_list:
+            draft_id_ = draft["id"]
+            draft_cid_ = draft["cid"]
+            if draft_cid_ != biz_id:
+                draft_id = draft_id_
+                break
+
+        if draft_id is None:
+            raise AssertionError(f"add draft failed; biz_id not found, biz_id: {biz_id}")
+
+        js = self.publish_draft(draft_id, bilibili_filename, metadata)
+
+        bvid = None
+        status_code = js["code"]
+        if status_code == 137022:
+            message = js["message"]
+            logger.info(f"publish_video_draft failed; code: {status_code}, message: {message}")
+        else:
+            bvid = js["data"]["bvid"]
+        return bvid
+
+
+def get_args():
+    parser = argparse.ArgumentParser()
+    parser.add_argument(
+        "--credentials_file",
+        default=(project_path / "dotenv/bilibili_chenjiesen_credentials.json").as_posix(),
+        type=str
+    )
+    parser.add_argument(
+        "--filename",
+        default=(project_path / "data/video/douyin/陈杰森/[7546905431348612362][20250906_172733]资本市场重大事件 。某行市值第一的隐秘故事。#易会满.mp4").as_posix(),
+        type=str
+    )
+    args = parser.parse_args()
+    return args
+
+
+def main():
+    import log
+    from project_settings import project_path, log_directory
+
+    log.setup_size_rotating(log_directory=log_directory)
+
+    args = get_args()
+
+    client = BilibiliVideoDraftUploader()
+
+    flag = client.check_login()
+    print(f"flag: {flag}")
+    client.login_with_credentials_file(args.credentials_file)
+    flag = client.check_login()
+    print(f"flag: {flag}")
+
+    js = client.list_draft()
+    print(f"js: {js}")
+
+    tags = [
+      "阅兵",
+      "商机",
+      "干货分享",
+      "金融",
+      "商业"
+    ]
+    metadata = {
+        "title": "九三阅兵的意义有多重大。 为什么说这次阅兵是“雷军式BOSS直销”？来的是客户，更是未来的“合伙人”！",
+        "desc": "九三阅兵的意义有多重大。 为什么说这次阅兵是“雷军式BOSS直销”？来的是客户，更是未来的“合伙人”！#阅兵 #金融 #商业 #商机 #干货分享",
+        "tag": ",".join(tags),
+
+        "copyright": 1,
+        "source": None,
+        "tid": 138,
+        "cover": "https://archive.biliimg.com/bfs/archive/124bf16affdfc2260f9fa7e1794bf946f1ad4997.jpg",
+        "dynamic": "",
+    }
+    js = client.upload_video_draft_and_publish(
+        filename=args.filename,
+        metadata=metadata
+    )
+    print(f"js: {js}")
+    return
+
+
+if __name__ == "__main__":
+    main()

toolbox/bilibili/video/video_manager.py

@@ -0,0 +1,332 @@
+#!/usr/bin/python3
+# -*- coding: utf-8 -*-
+"""
+https://pypi.org/project/biliupload/
+
+https://github.com/SocialSisterYi/bilibili-API-collect
+
+"""
+import argparse
+import logging
+import math
+from pathlib import Path
+import re
+
+logger = logging.getLogger("toolbox")
+
+from project_settings import project_path
+from toolbox.bilibili.bilibili_client import BilibiliClient
+
+
+class BilibiliVideoUploader(BilibiliClient):
+    def __init__(self):
+        super().__init__()
+
+    def preupload(self, filename, filesize):
+        """
+        The preupload process to get `upos_uri` and `auth` information.
+
+        [Easter egg] Sometimes I'm also confused why it is called `upos`
+        So I ask a question on the V2EX: https://v2ex.com/t/1103152
+        Finally, the netizens reckon that may be the translation style of bilibili.
+        :param filename: str, the name of the video to be uploaded
+        :param filesize: str, the size of the video to be uploaded
+        :param biz_id: int, the business id
+        :return:
+        upos_uri, str, the uri of the video will be stored in server
+        auth, str, the auth information
+        """
+        url = "https://member.bilibili.com/preupload"
+        params = {
+            "name":	filename,
+            "size":	filesize,
+            # The parameters below are fixed
+            "r": "upos",
+            "profile": "ugcupos/bup",
+            "ssl":	0,
+            "version":	"2.8.9",
+            "build": "2080900",
+            "upcdn": "bda2",
+            "probe_version": "20200810"
+        }
+        response = self.session.get(
+            url,
+            params=params,
+            headers={"TE": "Trailers"}
+        )
+
+        if response.status_code != 200:
+            raise AssertionError(f"request failed; status_code: {response.status_code}, text: {response.text}")
+        js = response.json()
+        if js["OK"] != 1:
+            raise AssertionError(f"request failed;")
+        return js
+
+    def get_upload_video_id(self, *, upos_uri, auth):
+        """
+        Get the `upload_id` of video.
+        :param upos_uri: str, get from `preupload`
+        :param auth: str, get from `preupload`
+        :return: upload_id, str, the id of the video to be uploaded
+        """
+        url = f"https://upos-sz-upcdnbda2.bilivideo.com/{upos_uri}?uploads&output=json"
+        response = self.session.post(
+            url,
+            headers={"X-Upos-Auth": auth}
+        )
+
+        if response.status_code != 200:
+            raise AssertionError(f"request failed; status_code: {response.status_code}, text: {response.text}")
+        js = response.json()
+        if js["OK"] != 1:
+            raise AssertionError(f"request failed;")
+        return js
+
+    def upload_video_in_chunks(self, *, upos_uri, auth, upload_id, fileio, filesize, chunk_size, chunks):
+        """
+        Upload the video in chunks.
+        :param upos_uri: str, get from `preupload`
+        :param auth: str, get from `preupload`
+        :param upload_id: str, get from `get_upload_video_id`
+        :param fileio: io.BufferedReader, the io stream of the video to be uploaded
+        :param filesize: int, the size of the video to be uploaded
+        :param chunk_size: int, the size of each chunk to be uploaded
+        :param chunks: int, the number of chunks to be uploaded
+        :return:
+        """
+        url = f"https://upos-sz-upcdnbda2.bilivideo.com/{upos_uri}"
+        params = {
+            "partNumber": None,  # start from 1
+            "uploadId":	upload_id,
+            "chunk": None,  # start from 0
+            "chunks": chunks,
+            "size":	None,  # current batch size
+            "start": None,
+            "end":	None,
+            "total": filesize,
+        }
+        # Single thread upload
+        for chunknum in range(chunks):
+            start = fileio.tell()
+            batchbytes = fileio.read(chunk_size)
+            params["partNumber"] = chunknum + 1
+            params["chunk"] = chunknum
+            params["size"] = len(batchbytes)
+            params["start"] = start
+            params["end"] = fileio.tell()
+            response = self.session.put(
+                url,
+                params=params,
+                data=batchbytes,
+                headers={"X-Upos-Auth": auth}
+            )
+            assert response.status_code == 200
+            # logger.debug(f'Completed chunk{chunknum+1} uploading')
+            # print(res)
+
+    def finish_upload(self, *, upos_uri, auth, filename, upload_id, biz_id, chunks):
+        """
+        Notify the all chunks have been uploaded.
+        :param upos_uri: str, get from `preupload`
+        :param auth: str, get from `preupload`
+        :param filename: str, the name of the video to be uploaded
+        :param upload_id: str, get from `get_upload_video_id`
+        :param biz_id: int, get from `preupload`
+        :param chunks: int, the number of chunks to be uploaded
+        :return:
+        """
+        url = f"https://upos-sz-upcdnbda2.bilivideo.com/{upos_uri}"
+        params = {
+            "output": "json",
+            "name":	filename,
+            "profile" : "ugcupos/bup",
+            "uploadId": upload_id,
+            "biz_id": biz_id
+        }
+        data = {
+            "parts": [
+                {"partNumber": i, "eTag": "etag"}
+                for i in range(chunks, 1)
+            ]
+        }
+        response = self.session.post(
+            url,
+            params=params,
+            json=data,
+            headers={"X-Upos-Auth": auth}
+        )
+        if response.status_code != 200:
+            raise AssertionError(f"request failed; status_code: {response.status_code}, text: {response.text}")
+        js = response.json()
+        if js["OK"] != 1:
+            raise AssertionError(f"request failed;")
+        return js
+
+    def publish_video(self, bilibili_filename, metadata: dict):
+        """
+        publish the uploaded video
+
+        https://github.com/SocialSisterYi/bilibili-API-collect/blob/master/docs/creativecenter/upload.md
+        :param bilibili_filename:
+        :param metadata:
+        :return:
+        """
+        url = f'https://member.bilibili.com/x/vu/web/add?csrf={self.cookies["bili_jct"]}'
+
+        data = {
+            "copyright": metadata["copyright"],
+            "videos": [
+                {
+                    "filename": bilibili_filename,
+                    "title": metadata["title"],
+                    "desc": metadata["desc"]
+                }
+            ],
+            "source": metadata["source"],
+            "tid": metadata["tid"],
+            "title": metadata["title"],
+            "cover": metadata["cover"],
+            "tag": metadata["tag"],
+            "desc_format_id": 0,
+            "desc": metadata["desc"],
+            "dynamic": metadata["dynamic"],
+            "subtitle": {"open": 0, "lan": ""},
+
+        }
+
+        if metadata["copyright"] != 2:
+            del data["source"]
+            # copyright: 1 original 2 reprint
+            data["copyright"] = 1
+
+        response = self.session.post(
+            url,
+            json=data,
+            headers={"TE": "Trailers"}
+        )
+        if response.status_code != 200:
+            raise AssertionError(f"request failed; status_code: {response.status_code}, text: {response.text}")
+        js = response.json()
+        return js
+
+    def upload_video_file(self, filename):
+        filename = Path(filename)
+        filesize = filename.stat().st_size
+
+        js = self.preupload(filename=filename, filesize=filesize)
+        upos_uri = js["upos_uri"].split("//")[-1]
+        auth = js["auth"]
+        biz_id = js["biz_id"]
+        chunk_size = js["chunk_size"]
+        chunks = math.ceil(filesize / chunk_size)
+
+        upload_video_id_response = self.get_upload_video_id(upos_uri=upos_uri, auth=auth)
+        upload_id = upload_video_id_response["upload_id"]
+        key = upload_video_id_response["key"]
+
+        bilibili_filename = re.search(r"/(.*)\.", key).group(1)
+        fileio = filename.open(mode="rb")
+        self.upload_video_in_chunks(
+            upos_uri=upos_uri,
+            auth=auth,
+            upload_id=upload_id,
+            fileio=fileio,
+            filesize=filesize,
+            chunk_size=chunk_size,
+            chunks=chunks
+        )
+        fileio.close()
+
+        # notify the all chunks have been uploaded
+        self.finish_upload(
+            upos_uri=upos_uri, auth=auth, filename=filename,
+            upload_id=upload_id, biz_id=biz_id, chunks=chunks
+        )
+
+        return biz_id, bilibili_filename
+
+    def upload_video_and_publish(self, filename: str, metadata: dict):
+        _, bilibili_filename = self.upload_video_file(filename)
+        publish_video_response = self.publish_video(bilibili_filename=bilibili_filename, metadata=metadata)
+
+        bvid = None
+        status_code = publish_video_response["code"]
+        if status_code == 137022:
+            message = publish_video_response["message"]
+            logger.info(f"publish_video failed; code: {status_code}, message: {message}")
+        else:
+            bvid = publish_video_response["data"]["bvid"]
+        return bvid
+
+
+def get_args():
+    parser = argparse.ArgumentParser()
+    parser.add_argument(
+        "--credentials_file",
+        default=(project_path / "dotenv/bilibili_chenjiesen_credentials.json").as_posix(),
+        type=str
+    )
+    parser.add_argument(
+        "--filename",
+        default=(project_path / "data/video/douyin/陈杰森/[7546905431348612362][20250906_172733]资本市场重大事件 。某行市值第一的隐秘故事。#易会满.mp4").as_posix(),
+        type=str
+    )
+    args = parser.parse_args()
+    return args
+
+
+def main():
+    import log
+    from project_settings import project_path, log_directory
+
+    log.setup_size_rotating(log_directory=log_directory)
+
+    args = get_args()
+
+    client = BilibiliVideoUploader()
+
+    flag = client.check_login()
+    print(f"flag: {flag}")
+    client.login_with_credentials_file(args.credentials_file)
+    flag = client.check_login()
+    print(f"flag: {flag}")
+
+    tags = [
+      "阅兵",
+      "商机",
+      "干货分享",
+      "金融",
+      "商业"
+    ]
+    metadata = {
+        "title": "九三阅兵的意义有多重大。 为什么说这次阅兵是“雷军式BOSS直销”？来的是客户，更是未来的“合伙人”！",
+        "desc": "九三阅兵的意义有多重大。 为什么说这次阅兵是“雷军式BOSS直销”？来的是客户，更是未来的“合伙人”！#阅兵 #金融 #商业 #商机 #干货分享",
+        "tag": ",".join(tags),
+
+        "copyright": 1,
+        "source": None,
+        "tid": 138,
+        "cover": "",
+        "dynamic": "",
+    }
+
+    bvid = client.upload_video_and_publish(
+        filename=args.filename,
+        metadata={
+            "title": "九三阅兵的意义有多重大。 为什么说这次阅兵是“雷军式BOSS直销”？来的是客户，更是未来的“合伙人”！",
+            "desc": "九三阅兵的意义有多重大。 为什么说这次阅兵是“雷军式BOSS直销”？来的是客户，更是未来的“合伙人”！#阅兵 #金融 #商业 #商机 #干货分享",
+            "tag": ",".join(tags),
+
+            "copyright": 1,
+            "source": None,
+            "tid": 138,
+            "cover": "",
+            "dynamic": "",
+        }
+    )
+    print(f"bvid: {bvid}")
+    return
+
+
+if __name__ == "__main__":
+    main()

toolbox/design_patterns/__init__.py

@@ -0,0 +1,6 @@
+#!/usr/bin/python3
+# -*- coding: utf-8 -*-
+
+
+if __name__ == '__main__':
+    pass

toolbox/design_patterns/singleton.py

@@ -0,0 +1,124 @@
+#!/usr/bin/python3
+# -*- coding: utf-8 -*-
+import inspect
+
+
+class ParamsSingleton(object):
+    """根据传入的参数不同而创建单例.
+    由于参数中可能包含字典, 如果转字符串的话, 字典的 key 是无序的.
+    所以用了列表而不是字典来存实例. """
+    __instance = list()
+    _initialized = False
+
+    def __new__(cls, *args, **kwargs):
+        kwargs = cls.to_kwargs(*args, **kwargs)
+        kwargs['cls'] = cls
+
+        for obj, params in cls.__instance:
+            if params == kwargs:
+                return obj
+
+        obj = super().__new__(cls)
+        # 让每个类实例, 可以拿到自己的 kwargs
+        # setattr(obj, 'kwargs', kwargs)
+        obj.kwargs = kwargs
+        cls.__instance.append((obj, kwargs))
+        return obj
+
+    @classmethod
+    def get_all_instance(cls) -> list:
+        return cls.__instance
+
+    @classmethod
+    def to_kwargs(cls, *args, **kwargs):
+        """将传入 __init__ 的参数全部转为 key-value 字典的关键字参数"""
+
+        # 获取当前传入参数值.
+        argvalues = inspect.getargvalues(inspect.currentframe())
+        args = list(argvalues.locals['args'])
+        kwargs = argvalues.locals['kwargs']
+        for k, v in argvalues.locals.items():
+            if k in ('cls', 'args', 'kwargs'):
+                continue
+            else:
+                kwargs[k] = v
+
+        # 获取函数接受哪些参数.
+        fullargspec = inspect.getfullargspec(cls.__init__)
+        # 函数的参数分为已知的位置参数, 未知的位置参数, 已知的关键字参数, 未知的关键字参数.
+        # 在 `未知的位置参数` 之前的参数都是 `已知的位置参数`. 它们可能有默认值
+        # 有默认值的参数并不都是关键字参数. 关键字参数也可以没有默认值.
+
+        # fullargspec.args: `已知的位置参数` 的名称的列表.
+        # fullargspec.defaults: 元组或None. `已知的位置参数` 中最后几项的默认值.
+        # fullargspec.kwonlyargs: `已知的关键字参数` 的名称列表 (没有默认值的关键字参数, 是必须要传入的).
+        # fullargspec.kwonlydefaults: `已知的关键字参数` 的默认值.
+
+        arg_name_list = fullargspec.args
+
+        # 将未被赋值 `已知的位置参数` 的默认值写入 kwargs.
+        if fullargspec.defaults is not None:
+            l = len(fullargspec.defaults)
+            default_args = fullargspec.args[-l:]
+            for k, v in zip(default_args, fullargspec.defaults):
+                if k in kwargs:
+                    continue
+                else:
+                    kwargs[k] = v
+
+        # 将 `已知关键字参数` 的默认值写入 kwargs.
+        if fullargspec.kwonlydefaults is not None:
+            for k, v in fullargspec.kwonlydefaults.items():
+                if k in kwargs:
+                    continue
+                else:
+                    kwargs[k] = v
+
+        # if fullargspec.kwonlyargs is not None:
+        #     arg_name_list.extend(fullargspec.kwonlyargs)
+                kwargs = dict()
+        for arg_name in arg_name_list:
+            if arg_name == 'self':
+                continue
+            try:
+                value = args.pop(0)
+            except IndexError:
+                break
+            kwargs[arg_name] = value
+
+        if fullargspec.varargs is not None:
+            kwargs[fullargspec.varargs] = tuple(args)
+
+        return kwargs
+
+    @classmethod
+    def flush(cls):
+        cls.__instance = list()
+        return
+
+
+def demo1():
+    class A(ParamsSingleton):
+        pass
+
+    class B(A):
+        # def __init__(self, name, *args1, age, **kwargs):
+        def __init__(self, name, age=27, **kwargs):
+
+            pass
+
+    b1 = B('jack')
+    print('-' * 25)
+    # b2 = B('jack', 1, 2, age=25, **{'high': 165})
+    # print('-' * 25)
+    b3 = B(name='jack', **{'age': 25, 'high': 165})
+    # b3 = B(name='jack', **{'high': 165})
+
+    print('-' * 25)
+
+    # print(b1)
+    return
+
+
+if __name__ == '__main__':
+    demo1()

toolbox/douyin/__init__.py

@@ -0,0 +1,6 @@
+#!/usr/bin/python3
+# -*- coding: utf-8 -*-
+
+
+if __name__ == "__main__":
+    pass

toolbox/douyin/douyin_client.py

@@ -0,0 +1,218 @@
+#!/usr/bin/python3
+# -*- coding: utf-8 -*-
+"""
+https://blog.csdn.net/qq_18303993/article/details/114281349
+"""
+import argparse
+import json
+import logging
+import time
+import httpx
+
+import qrcode
+import requests
+import requests.utils
+
+from project_settings import project_path
+from toolbox.design_patterns.singleton import ParamsSingleton
+
+
+logger = logging.getLogger("toolbox")
+
+
+class DouyinClient(ParamsSingleton):
+    headers = {
+        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36",
+    }
+
+    def __init__(self):
+        if not self._initialized:
+            self.credentials = None
+            self.cookies = None
+
+            self._session = requests.Session()
+            self._async_session = httpx.AsyncClient(
+                http2=True,
+                limits=httpx.Limits(max_keepalive_connections=100, keepalive_expiry=100),
+                headers=self.headers,
+                cookies=self.cookies,
+            )
+            self._initialized = True
+
+    @property
+    def session(self):
+        if not self._session.cookies:
+            self._session.headers = self.headers
+            self._session.cookies = requests.utils.cookiejar_from_dict(self.cookies)
+        return self._session
+
+    @property
+    def async_session(self):
+        if not self._async_session.cookies:
+            self._async_session.headers = self.headers
+            self._async_session.cookies = httpx.Cookies(self.cookies)
+        return self._async_session
+
+    @classmethod
+    def get_qrcode(cls):
+        url = "https://login.douyin.com/passport/web/get_qrcode/"
+        params = {
+            "aid": "6383",
+            "next": "https://www.douyin.com",
+        }
+        response = requests.get(url, params=params, headers=cls.headers)
+        if response.status_code != 200:
+            raise AssertionError(f"request failed; status_code: {response.status_code}, text: {response.text}")
+        js = response.json()
+        return js
+
+    @classmethod
+    def check_qr_connect(cls, token: str):
+        url = "https://login.douyin.com/passport/web/check_qrconnect"
+
+        params = {
+            "aid": "6383",
+            "next": "https://www.douyin.com",
+        }
+        data = {
+            "need_logo": False,
+            "need_short_url": False,
+            "is_frontier": True,
+            "token": token,
+            "is_new_login": 1,
+            "next": "https://www.douyin.com",
+        }
+        response = requests.post(
+            url,
+            headers=cls.headers,
+            params=params,
+            data=data,
+        )
+        if response.status_code != 200:
+            raise AssertionError(f"request failed; status_code: {response.status_code}, text: {response.text}")
+        js = response.json()
+        print(response.cookies)
+        return js
+
+    @classmethod
+    def get_cookies(cls, url: str):
+        response = requests.get(url, headers=cls.headers)
+        if response.status_code != 200:
+            raise AssertionError(f"request failed; status_code: {response.status_code}, text: {response.text}")
+        js = response.json()
+        return js
+
+    def check_login(self):
+        url = "https://creator.douyin.com/web/api/media/user/info"
+        response = self.session.get(url)
+        if response.status_code != 200:
+            raise AssertionError(f"request failed; status_code: {response.status_code}, text: {response.text}")
+        js = response.json()
+        # print(f"js: {json.dumps(js, ensure_ascii=False)}")
+
+        status_code = js["status_code"]
+        if status_code == 8:
+            # js: {"extra": {"logid": "20250908143856B85220348FE83A18AA76", "now": 1757313536000}, "status_code": 8, "status_msg": "用户未登录"}
+            return False
+        elif status_code == 0:
+            return True
+        else:
+            raise AssertionError(f"js: {json.dumps(js, ensure_ascii=False)}")
+
+    def login_with_qrcode(self):
+        js = self.get_qrcode()
+        # print(f"js: {json.dumps(js, ensure_ascii=False, indent=4)}")
+
+        qrcode_index_url = js["data"]["qrcode_index_url"]
+        token = js["data"]["token"]
+        # qr = qrcode.QRCode()
+        qr = qrcode.QRCode(
+            version=1,
+            error_correction=qrcode.constants.ERROR_CORRECT_L,
+            box_size=1,
+            border=1,
+        )
+        qr.add_data(qrcode_index_url)
+        qr.make(fit=True)
+        qr.print_ascii()
+        print("Or copy this link to your phone Douyin:", qrcode_index_url)
+
+        cookies = None
+        while True:
+            time.sleep(3)
+            js = self.check_qr_connect(token=token)
+            message = js["message"]
+            if message == "error":
+                error_code = js["data"]["error_code"]
+                description = js["data"]["description"]
+                raise AssertionError(f"check qr connect error; error_code: {error_code}, description: {description}")
+            status = js["data"]["status"]
+            if status == "new":
+                pass
+            elif status == "expired":
+                qrcode_index_url = js["data"]["qrcode_index_url"]
+                token = js["data"]["token"]
+                qr = qrcode.QRCode()
+                qr.add_data(qrcode_index_url)
+                qr.print_ascii()
+                print("Or copy this link to your phone Douyin:", qrcode_index_url)
+            elif status == "scanned":
+                print('已扫码，请确认登录！')
+                pass
+
+        # js = self.check_login()
+        # print(f"js: {json.dumps(js, ensure_ascii=False)}")
+        return js
+
+    def login_with_credentials_file(self, credentials_file: str):
+        with open(credentials_file, "r", encoding="utf-8") as f:
+            credentials = json.load(f)
+        self.credentials = credentials
+        self.set_cookies(credentials)
+        return True
+
+    def login_with_credentials_info(self, credentials_info: dict):
+        self.credentials = credentials_info
+        self.set_cookies(credentials_info)
+        return True
+
+    def set_cookies(self, cookies: dict):
+        self.cookies = cookies
+
+
+def get_args():
+    parser = argparse.ArgumentParser()
+    parser.add_argument(
+        "--credentials_file",
+        default=(project_path / "dotenv/douyin_login_credentials.json").as_posix(),
+        type=str
+    )
+    args = parser.parse_args()
+    return args
+
+
+def main():
+    import log
+    from project_settings import project_path, log_directory
+
+    log.setup_size_rotating(log_directory=log_directory)
+
+    args = get_args()
+
+    client = DouyinClient()
+
+    flag = client.check_login()
+    print(f"flag: {flag}")
+    client.login_with_credentials_file(args.credentials_file)
+    # client.login_with_qrcode_url()
+    flag = client.check_login()
+    print(f"flag: {flag}")
+
+    # js = client.login_with_qrcode()
+    # js = client.check_login()
+    # print(js)
+    return
+
+
+if __name__ == "__main__":
+    main()

toolbox/douyin/homepage/__init__.py

@@ -0,0 +1,6 @@
+#!/usr/bin/python3
+# -*- coding: utf-8 -*-
+
+
+if __name__ == "__main__":
+    pass

toolbox/douyin/homepage/follow.py

@@ -0,0 +1,111 @@
+#!/usr/bin/python3
+# -*- coding: utf-8 -*-
+import argparse
+import asyncio
+import json
+import logging
+
+from project_settings import project_path
+from toolbox.douyin.douyin_client import DouyinClient
+from toolbox.asyncio.cacheout import async_cache_decorator
+
+logger = logging.getLogger("toolbox")
+
+
+class FollowManager(DouyinClient):
+    def __init__(self):
+        super().__init__()
+
+    @async_cache_decorator(60)
+    async def get_living_list(self):
+        url = "https://www.douyin.com/webcast/web/feed/follow/"
+
+        params = {
+            "device_platform": "webapp",
+            "aid": 6383,
+            "channel": "channel_pc_web",
+            "scene": "aweme_pc_follow_top",
+            "update_version_code": 170400,
+            "pc_client_type": 1,
+            "pc_libra_divert": "Mac",
+            "version_code": "170400",
+            "version_name": "17.4.0",
+            "cookie_enabled": "true",
+
+        }
+
+        response = await self.async_session.request("GET", url, params=params)
+        if response.status_code != 200:
+            raise AssertionError(f"request failed, status_code: {response.status_code}, text: {response.text}")
+
+        js = response.json()
+
+        status_code = js["status_code"]
+        if status_code == 20003:
+            # 请登录后进入直播间
+            raise AssertionError(f"request failed, status_code: {response.status_code}, text: {response.text}")
+
+        data = js["data"]["data"]
+
+        result = list()
+        for row in data:
+            room = row["room"]
+            title = room["title"]
+            stream_url = room["stream_url"]
+            owner = room["owner"]
+            sec_uid = owner["sec_uid"]
+            nickname = owner["nickname"]
+            room_id = row["web_rid"]
+
+            stream_data = json.loads(stream_url["live_core_sdk_data"]["pull_data"]["stream_data"])
+            stream_data = stream_data["data"]
+
+            row_ = {
+                "nickname": nickname,
+                "sec_uid": sec_uid,
+                "room_id": room_id,
+
+                "status": 2,
+                "title": title,
+                "stream_data": stream_data,
+            }
+            result.append(row_)
+
+        return result
+
+
+def get_args():
+    parser = argparse.ArgumentParser()
+    parser.add_argument(
+        "--credentials_file",
+        default=(project_path / "dotenv/douyin_wentao_credentials.json").as_posix(),
+        type=str
+    )
+    args = parser.parse_args()
+    return args
+
+
+async def main():
+    import log
+    from project_settings import project_path, log_directory
+
+    log.setup_size_rotating(log_directory=log_directory)
+
+    args = get_args()
+
+    client = FollowManager()
+
+    flag = client.check_login()
+    print(f"flag: {flag}")
+    client.login_with_credentials_file(args.credentials_file)
+    # client.login_with_qrcode_url()
+    flag = client.check_login()
+    print(f"flag: {flag}")
+
+    js = await client.get_living_list()
+    print(f"js: {json.dumps(js, ensure_ascii=False, indent=4)}")
+    return
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

toolbox/douyin/live/__init__.py

@@ -0,0 +1,6 @@
+#!/usr/bin/python3
+# -*- coding: utf-8 -*-
+
+
+if __name__ == "__main__":
+    pass

toolbox/douyin/live/live_recording.py

@@ -0,0 +1,213 @@
+#!/usr/bin/python3
+# -*- coding: utf-8 -*-
+import argparse
+import asyncio
+import httpx
+import json
+import logging
+import random
+import re
+import string
+from typing import List
+
+from bs4 import BeautifulSoup
+from tenacity import before_sleep_log, retry, retry_if_exception_type, stop_after_attempt, wait_fixed
+
+from project_settings import project_path
+from toolbox.douyin.homepage.follow import FollowManager
+from toolbox.asyncio.cacheout import async_cache_decorator
+
+logger = logging.getLogger("toolbox")
+
+
+class LiveRecording(FollowManager):
+    def __init__(self):
+        super().__init__()
+
+    @async_cache_decorator(5)
+    @retry(
+        wait=wait_fixed(10),
+        stop=stop_after_attempt(3),
+        before_sleep=before_sleep_log(logger, logging.ERROR),
+    )
+    async def get_live_info_by_web_enter(self, room_id: str):
+        if not self.async_session.cookies:
+            await self.async_session.request(
+                method="GET",
+                url="https://live.douyin.com/"
+            )
+        response = await self.async_session.request(
+            method="GET",
+            url="https://live.douyin.com/webcast/room/web/enter/",
+            params={
+                "aid": "6383",
+                "device_platform": "web",
+                "browser_language": "zh-CN",
+                "browser_platform": "Win32",
+                "browser_name": "Chrome",
+                "browser_version": "100.0.0.0",
+                "web_rid": room_id
+            },
+        )
+        if response.status_code != 200:
+            raise AssertionError(f"live enter request failed with status code {response.status_code}")
+        if len(response.text) == 0:
+            self.async_session.cookies = httpx.Cookies()
+            return None
+
+        js = response.json()
+        return js
+
+    @async_cache_decorator(5)
+    @retry(
+        wait=wait_fixed(10),
+        stop=stop_after_attempt(3),
+        before_sleep=before_sleep_log(logger, logging.ERROR),
+    )
+    async def get_live_info_by_room_url(self, room_id: str):
+        if not self.async_session.cookies:
+            await self.async_session.request(
+                method="GET",
+                url="https://live.douyin.com/"
+            )
+        room_url = f"https://live.douyin.com/{room_id}"
+
+        __ac_nonce = "".join(random.choices(string.ascii_letters + string.digits, k=16))
+        headers= {
+            "Cookie": f"__ac_nonce={__ac_nonce}; ",
+            **self.headers,
+        }
+
+        response = await self.async_session.request(
+            method="GET",
+            url=room_url,
+            headers=headers,
+        )
+        html_text = response.text
+        soup = BeautifulSoup(html_text, "html.parser")
+
+        result = None
+        for script in soup.find_all("script"):
+            content = str(script)
+            match = re.search(
+                pattern=r"<script nonce=\"(?:.*?)\">self.__pace_f.push\(\[1,(.*?)\]\)</script>",
+                string=content,
+                flags=re.IGNORECASE
+            )
+            if match is not None:
+                text = match.group(1)
+
+                try:
+                    text = json.loads(text)
+                except json.JSONDecodeError:
+                    continue
+
+                if text[1:3] != ":[":
+                    continue
+                text = text[2:]
+
+                try:
+                    text = json.loads(text)
+                except json.JSONDecodeError:
+                    continue
+
+                if not isinstance(text, list):
+                    continue
+                js = text
+                if len(js) != 4:
+                    continue
+                js = js[3]
+                if not isinstance(js, dict):
+                    continue
+                js = js.get("state")
+                if js is None:
+                    continue
+
+                try:
+                    room_info = js["roomStore"]["roomInfo"]
+                    # 2 表示正在直播
+                    room = room_info["room"]
+                    alive_status = room["status"]
+                    title = room["title"]
+                    web_rid = room_info["web_rid"]
+                    anchor = room_info["anchor"]
+                    sec_uid = anchor["sec_uid"]
+                    nickname = anchor["nickname"]
+                    web_stream_url = room_info["web_stream_url"]
+
+                    stream_store = js["streamStore"]
+                    stream_data = stream_store["streamData"]
+                    h265_stream_data = stream_data["H265_streamData"]["stream"]
+                    h264_stream_data = stream_data["H264_streamData"]["stream"]
+
+                    camera_store = js["cameraStore"]
+                except KeyError:
+                    continue
+                result = {
+                    "status": alive_status,
+                    "title": title,
+                    "web_rid": web_rid,
+                    "sec_uid": sec_uid,
+                    "nickname": nickname,
+                    # "web_stream_url": web_stream_url,
+                    # "h265_stream_data": h265_stream_data,
+                    # "h264_stream_data": h264_stream_data,
+                    "stream_data": h264_stream_data,
+                    # "camera_store": camera_store,
+                }
+                break
+        return result
+
+    @async_cache_decorator(60)
+    async def get_live_info_by_follow(self, room_id: str):
+        result = None
+        live_info_list: List[dict] = await self.get_living_list()
+
+        if live_info_list is None:
+            return None
+        for live_info in live_info_list:
+            room_id_ = live_info["room_id"]
+            if room_id_ == room_id:
+                result = live_info
+                break
+        return result
+
+
+def get_args():
+    parser = argparse.ArgumentParser()
+    parser.add_argument(
+        "--credentials_file",
+        default=(project_path / "dotenv/douyin_login_credentials.json").as_posix(),
+        type=str
+    )
+    args = parser.parse_args()
+    return args
+
+
+async def main():
+    import log
+    from project_settings import project_path, log_directory
+
+    log.setup_size_rotating(log_directory=log_directory)
+
+    args = get_args()
+
+    client = LiveRecording()
+
+    flag = client.check_login()
+    print(f"flag: {flag}")
+    client.login_with_credentials_file(args.credentials_file)
+    # client.login_with_qrcode_url()
+    flag = client.check_login()
+    print(f"flag: {flag}")
+
+    # js = await client.get_live_info_by_web_enter(room_id="770758107267")
+    js = await client.get_live_info_by_room_url(room_id="572033528289")
+    # js = await client.get_live_info_by_follow(room_id="572033528289")
+    # js = await client.get_live_info_by_follow(room_id="572033528289")
+    print(f"js: {json.dumps(js, ensure_ascii=False, indent=4)}")
+    return
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

toolbox/douyin/video/__init__.py

@@ -0,0 +1,6 @@
+#!/usr/bin/python3
+# -*- coding: utf-8 -*-
+
+
+if __name__ == "__main__":
+    pass

toolbox/douyin/video/download.py

@@ -0,0 +1,220 @@
+#!/usr/bin/python3
+# -*- coding: utf-8 -*-
+import argparse
+import asyncio
+from datetime import datetime
+import json
+import logging
+from pathlib import Path
+
+from tenacity import before_sleep_log, retry, retry_if_exception_type, stop_after_attempt, wait_fixed
+
+from project_settings import project_path
+from toolbox.douyin.douyin_client import DouyinClient
+from toolbox.exception import ExpectedError
+
+logger = logging.getLogger("toolbox")
+
+
+class VideoDownload(DouyinClient):
+    def __init__(self):
+        super().__init__()
+
+    @retry(
+        wait=wait_fixed(10),
+        stop=stop_after_attempt(3),
+        before_sleep=before_sleep_log(logger, logging.ERROR),
+    )
+    async def download_video_by_url(self, filename: Path, url: str):
+        filename = Path(filename)
+        filename.parent.mkdir(parents=True, exist_ok=True)
+
+        response = await self.async_session.request(
+            method="GET",
+            url=url,
+            headers={
+                **self.headers,
+                "referer": "https://www.douyin.com/",
+            },
+        )
+        # 302 重定向
+        if response.status_code == 302:
+            url = response.headers["Location"]
+            return await self.download_video_by_url(filename, url)
+        elif response.status_code == 200:
+            with open(filename, "wb") as f:
+                f.write(response.content)
+            return filename
+        else:
+            raise AssertionError(f"Got status code {response.status_code}")
+
+    # @retry(
+    #     wait=wait_fixed(10),
+    #     stop=stop_after_attempt(3),
+    #     before_sleep=before_sleep_log(logger, logging.ERROR),
+    # )
+    async def get_video_list_by_user_id(self, sec_user_id: str, max_cursor: int = 0, count: int = 18):
+        url = "https://www.douyin.com/aweme/v1/web/aweme/post/"
+
+        params = {
+            "device_platform": "webapp",
+            "aid": "6383",
+            "channel": "channel_pc_web",
+            "sec_user_id": sec_user_id,
+            "max_cursor": max_cursor,
+            "count": count,
+            "publish_video_strategy_type": "2",
+            "version_code": "290100",
+            "version_name": "29.1.0",
+        }
+
+        response = await self.async_session.request(
+            method="GET",
+            url=url,
+            headers={
+                **self.headers,
+                "referer": "https://www.douyin.com/",
+            },
+            params=params,
+        )
+        if response.status_code == 444:
+            # Access Denied
+            raise ExpectedError(status_code=60444, message=f"request failed, status_code: {response.status_code}, text: {response.text}")
+        elif response.status_code == 200 and len(response.text) == 0:
+            # Maybe Access Denied
+            raise ExpectedError(status_code=60444, message=f"request failed, status_code: {response.status_code}, text: {response.text}")
+        elif response.status_code != 200:
+            raise AssertionError(f"request failed, status_code: {response.status_code}, text: {response.text}")
+        elif response.text == "blocked":
+            raise AssertionError(f"request failed, status_code: {response.status_code}, text: {response.text}")
+        js = response.json()
+        aweme_list = js["aweme_list"]
+
+        result = list()
+        for aweme in aweme_list:
+            # aweme_ = json.dumps(aweme, ensure_ascii=False, indent=4)
+            # print(aweme_)
+
+            aweme_id = aweme["aweme_id"]
+            desc = aweme["desc"]
+            create_time = aweme["create_time"]
+            create_time_ = datetime.fromtimestamp(create_time)
+            create_time_str = create_time_.strftime("%Y%m%dT%H%M%S")
+
+            # video
+            video = aweme["video"]
+            video_url_list = video["play_addr"]["url_list"]
+
+            # tags
+            text_extra = aweme["text_extra"]
+            tags = set()
+            for t in text_extra:
+                tag = t.get("hashtag_name")
+                if tag is None:
+                    tag = t.get("search_text")
+                if tag is None:
+                    # print(t)
+                    continue
+                tags.add(tag)
+            tags = list(tags)
+
+            # title
+            title: str = desc
+            for tag in tags:
+                title = title.replace(f"#{tag}", "")
+                # title = title.replace(f"# {tag}", "")
+            title = title.strip()
+
+            row = {
+                "aweme_id": aweme_id,
+                "create_time": create_time,
+                "create_time_str": create_time_str,
+                "title": title,
+                "desc": desc,
+                "video_url_list": video_url_list,
+                "tags": tags,
+            }
+            result.append(row)
+        return result
+
+    async def get_video_list_by_min_date(self, sec_user_id: str, min_date: str = "2025-06-10 00:00:00"):
+        min_date_ = datetime.strptime(min_date, "%Y-%m-%d %H:%M:%S")
+
+        result = list()
+
+        stop_flag = False
+        max_cursor = 0
+        for i in range(100):
+            if stop_flag:
+                break
+            rows = await self.get_video_list_by_user_id(sec_user_id=sec_user_id, max_cursor=max_cursor, count=18)
+            this_min_date_ = [datetime.fromtimestamp(row["create_time"]) < min_date_ for row in rows]
+            if all(this_min_date_):
+                break
+            for row in rows:
+                create_time = row["create_time"]
+                aweme_id = row["aweme_id"]
+                create_time_str = row["create_time_str"]
+                title = row["title"]
+                desc = row["desc"]
+                video_url_list = row["video_url_list"]
+                tags = row["tags"]
+
+                max_cursor_ = int(create_time * 1000)
+                if max_cursor == 0 or max_cursor_ < max_cursor:
+                    max_cursor = max_cursor_
+
+                create_time_ = datetime.fromtimestamp(create_time)
+
+                if create_time_ > min_date_:
+                    task = {
+                        "aweme_id": aweme_id,
+                        "create_time_str": create_time_str,
+                        "title": title,
+                        "desc": desc,
+                        "video_url_list": video_url_list,
+                        "tags": tags,
+                    }
+                    result.append(task)
+        return result
+
+
+def get_args():
+    parser = argparse.ArgumentParser()
+    parser.add_argument(
+        "--credentials_file",
+        default=(project_path / "dotenv/douyin_wentao_credentials.json").as_posix(),
+        type=str
+    )
+    args = parser.parse_args()
+    return args
+
+
+async def main():
+    import log
+    from project_settings import project_path, log_directory
+
+    log.setup_size_rotating(log_directory=log_directory)
+
+    args = get_args()
+
+    client = VideoDownload()
+
+    flag = client.check_login()
+    print(f"flag: {flag}")
+    client.login_with_credentials_file(args.credentials_file)
+    # client.login_with_qrcode_url()
+    flag = client.check_login()
+    print(f"flag: {flag}")
+
+    js = await client.get_video_list_by_min_date(
+        # sec_user_id="MS4wLjABAAAAb0mqEsXmBehDdg2Q9mMA2T6YEWPGbEtYofSzX_bDnz4",
+        sec_user_id="MS4wLjABAAAAQinRMLyQNYA45OYXoCDrwszhRGaDVirRE1fTNSaGGkc",
+        min_date="2025-08-06 00:00:00"
+    )
+    print(f"js: {json.dumps(js, ensure_ascii=False, indent=4)}")
+    return
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

toolbox/exception.py

@@ -0,0 +1,8 @@
+
+
+class ExpectedError(Exception):
+    def __init__(self, status_code, message, traceback="", detail=""):
+        self.status_code = status_code
+        self.message = message
+        self.traceback = traceback
+        self.detail = detail

toolbox/json/__init__.py

@@ -0,0 +1,6 @@
+#!/usr/bin/python3
+# -*- coding: utf-8 -*-
+
+
+if __name__ == '__main__':
+    pass

toolbox/json/misc.py

@@ -0,0 +1,63 @@
+#!/usr/bin/python3
+# -*- coding: utf-8 -*-
+from typing import Callable
+
+
+def traverse(js, callback: Callable, *args, **kwargs):
+    if isinstance(js, list):
+        result = list()
+        for l in js:
+            l = traverse(l, callback, *args, **kwargs)
+            result.append(l)
+        return result
+    elif isinstance(js, tuple):
+        result = list()
+        for l in js:
+            l = traverse(l, callback, *args, **kwargs)
+            result.append(l)
+        return tuple(result)
+    elif isinstance(js, dict):
+        result = dict()
+        for k, v in js.items():
+            k = traverse(k, callback, *args, **kwargs)
+            v = traverse(v, callback, *args, **kwargs)
+            result[k] = v
+        return result
+    elif isinstance(js, int):
+        return callback(js, *args, **kwargs)
+    elif isinstance(js, str):
+        return callback(js, *args, **kwargs)
+    else:
+        return js
+
+
+def demo1():
+    d = {
+        "env": "ppe",
+        "mysql_connect": {
+            "host": "$mysql_connect_host",
+            "port": 3306,
+            "user": "callbot",
+            "password": "NxcloudAI2021!",
+            "database": "callbot_ppe",
+            "charset": "utf8"
+        },
+        "es_connect": {
+            "hosts": ["10.20.251.8"],
+            "http_auth": ["elastic", "ElasticAI2021!"],
+            "port": 9200
+        }
+    }
+
+    def callback(s):
+        if isinstance(s, str) and s.startswith('$'):
+            return s[1:]
+        return s
+
+    result = traverse(d, callback=callback)
+    print(result)
+    return
+
+
+if __name__ == '__main__':
+    demo1()

toolbox/os/__init__.py

@@ -0,0 +1,6 @@
+#!/usr/bin/python3
+# -*- coding: utf-8 -*-
+
+
+if __name__ == '__main__':
+    pass

toolbox/os/command.py

@@ -0,0 +1,59 @@
+#!/usr/bin/python3
+# -*- coding: utf-8 -*-
+import os
+
+
+class Command(object):
+    custom_command = [
+        "cd"
+    ]
+
+    @staticmethod
+    def _get_cmd(command):
+        command = str(command).strip()
+        if command == "":
+            return None
+        cmd_and_args = command.split(sep=" ")
+        cmd = cmd_and_args[0]
+        args = " ".join(cmd_and_args[1:])
+        return cmd, args
+
+    @classmethod
+    def popen(cls, command):
+        cmd, args = cls._get_cmd(command)
+        if cmd in cls.custom_command:
+            method = getattr(cls, cmd)
+            return method(args)
+        else:
+            resp = os.popen(command)
+            result = resp.read()
+            resp.close()
+            return result
+
+    @classmethod
+    def cd(cls, args):
+        if args.startswith("/"):
+            os.chdir(args)
+        else:
+            pwd = os.getcwd()
+            path = os.path.join(pwd, args)
+            os.chdir(path)
+
+    @classmethod
+    def system(cls, command):
+        return os.system(command)
+
+    def __init__(self):
+        pass
+
+
+def ps_ef_grep(keyword: str):
+    cmd = "ps -ef | grep {}".format(keyword)
+    rows = Command.popen(cmd)
+    rows = str(rows).split("\n")
+    rows = [row for row in rows if row.__contains__(keyword) and not row.__contains__("grep")]
+    return rows
+
+
+if __name__ == "__main__":
+    pass

toolbox/os/environment.py

@@ -0,0 +1,114 @@
+#!/usr/bin/python3
+# -*- coding: utf-8 -*-
+import json
+import os
+
+from dotenv import load_dotenv
+from dotenv.main import DotEnv
+
+from toolbox.json.misc import traverse
+
+
+class EnvironmentManager(object):
+    def __init__(self, path, env, override=False):
+        filename = os.path.join(path, '{}.env'.format(env))
+        self.filename = filename
+
+        load_dotenv(
+            dotenv_path=filename,
+            override=override
+        )
+
+        self._environ = dict()
+
+    def open_dotenv(self, filename: str = None):
+        filename = filename or self.filename
+        dotenv = DotEnv(
+            dotenv_path=filename,
+            stream=None,
+            verbose=False,
+            interpolate=False,
+            override=False,
+            encoding="utf-8",
+        )
+        result = dotenv.dict()
+        return result
+
+    def get(self, key, default=None, dtype=str):
+        result = os.environ.get(key)
+        if result is None:
+            if default is None:
+                result = None
+            else:
+                result = default
+        else:
+            result = dtype(result)
+        self._environ[key] = result
+        return result
+
+
+_DEFAULT_DTYPE_MAP = {
+    'int': int,
+    'float': float,
+    'str': str,
+    'json.loads': json.loads
+}
+
+
+class JsonConfig(object):
+    """
+    将 json 中, 形如 `$float:threshold` 的值, 处理为:
+    从环境变量中查到 threshold, 再将其转换为 float 类型.
+    """
+    def __init__(self, dtype_map: dict = None, environment: EnvironmentManager = None):
+        self.dtype_map = dtype_map or _DEFAULT_DTYPE_MAP
+        self.environment = environment or os.environ
+
+    def sanitize_by_filename(self, filename: str):
+        with open(filename, 'r', encoding='utf-8') as f:
+            js = json.load(f)
+
+        return self.sanitize_by_json(js)
+
+    def sanitize_by_json(self, js):
+        js = traverse(
+            js,
+            callback=self.sanitize,
+            environment=self.environment
+        )
+        return js
+
+    def sanitize(self, string, environment):
+        """支持 $ 符开始的, 环境变量配置"""
+        if isinstance(string, str) and string.startswith('$'):
+            dtype, key = string[1:].split(':')
+            dtype = self.dtype_map[dtype]
+
+            value = environment.get(key)
+            if value is None:
+                raise AssertionError('environment not exist. key: {}'.format(key))
+
+            value = dtype(value)
+            result = value
+        else:
+            result = string
+        return result
+
+
+def demo1():
+    import json
+
+    from project_settings import project_path
+
+    environment = EnvironmentManager(
+        path=os.path.join(project_path, 'server/callbot_server/dotenv'),
+        env='dev',
+    )
+    init_scenes = environment.get(key='init_scenes', dtype=json.loads)
+    print(init_scenes)
+    print(environment._environ)
+    return
+
+
+if __name__ == '__main__':
+    demo1()

toolbox/os/other.py

@@ -0,0 +1,9 @@
+import os
+import inspect
+
+
+def pwd():
+    """你在哪个文件调用此函数, 它就会返回那个文件所在的 dir 目标"""
+    frame = inspect.stack()[1]
+    module = inspect.getmodule(frame[0])
+    return os.path.dirname(os.path.abspath(module.__file__))

toolbox/porter/__init__.py

@@ -0,0 +1,6 @@
+#!/usr/bin/python3
+# -*- coding: utf-8 -*-
+
+
+if __name__ == "__main__":
+    pass

toolbox/porter/common/__init__.py

@@ -0,0 +1,6 @@
+#!/usr/bin/python3
+# -*- coding: utf-8 -*-
+
+
+if __name__ == '__main__':
+    pass

toolbox/porter/common/params.py

@@ -0,0 +1,197 @@
+#!/usr/bin/python3
+# -*- coding: utf-8 -*-
+import logging
+import inspect
+import typing
+
+from toolbox.porter.common.registrable import Registrable
+
+logger = logging.getLogger("toolbox")
+
+
+class Params(Registrable):
+    """
+    仿照 AllenNLP 框架, Params 与 Registrable 配合使用, 通过子类的 Annotation 类型标注使得子类可以通过
+
+    .from_json({
+        **parameters
+    })
+
+    方式实例化.
+
+    注意事项:
+    1. 子类除了 self, 每个参数都必须有类型标注.
+    2. 当子类没有 __init__ 方法时, 会调用到基类的该方法 (基类的默认实现没有类型标注),
+    因此, 都应实现 __init__ 方法.
+
+    """
+    # def __init__(self):
+    #     # Subclasses should override this method, even if it is def __init__(self): pass
+    #     pass
+
+    @classmethod
+    def from_json(cls, params: dict = None, global_params: dict = None):
+        """
+        :param params:
+        :param global_params: 当缺少某参数时, 尝试从 global_params 中查找.
+        :return:
+        """
+        if params is None:
+            params = dict()
+        if global_params is None:
+            global_params = dict()
+
+        if "type" in params:
+            cls = cls.by_name(params["type"])
+
+        signature = inspect.signature(cls.__init__)
+
+        kwargs = dict()
+        for k, v in signature.parameters.items():
+            if k in ("self",):
+                continue
+            if k in ("args", "kwargs"):
+                msg = (
+                    f"parameter: args or kwargs is not expected. "
+                    f"you may need to override the __init__ method of cls: {cls.__name__}."
+                )
+                logger.warning(msg)
+                print(msg)
+                continue
+
+            if v.annotation is inspect._empty:
+                raise NotImplementedError(
+                    "all parameter should have a annotation. "
+                    "parameter `{}` of {} have not annotation".format(k, cls)
+                )
+
+            if v.name in params:
+                sub_params = params[v.name]
+            elif v.name in global_params:
+                sub_params = global_params[v.name]
+            else:
+                continue
+
+            if isinstance(v.annotation, str):
+                raise NotImplementedError("string annotation not supported.")
+
+            if hasattr(v.annotation, "_subs_tree"):
+                # typing 标注类型.
+                subs_tree = v.annotation._subs_tree()
+                kwargs[v.name] = cls.from_annotation(sub_params, global_params, subs_tree)
+            elif isinstance(v.annotation, typing._GenericAlias):
+                # typing 标注类型.
+                subs_tree = (v.annotation.__origin__, *v.annotation.__args__)
+                kwargs[v.name] = cls.from_annotation(sub_params, global_params, subs_tree)
+            elif issubclass(v.annotation, Params):
+                # Params 子类.
+                kwargs[v.name] = v.annotation.from_json(
+                    sub_params, global_params
+                )
+            elif isinstance(sub_params, v.annotation):
+                # 传入的是已实例化好的值.
+                kwargs[v.name] = sub_params
+            else:
+                # str, int, list, dict 等基本类型.
+                value = sub_params
+                if isinstance(value, dict):
+                    value = v.annotation(**value)
+                else:
+                    value = v.annotation(value)
+
+                kwargs[v.name] = value
+
+        obj = cls.__new__(cls, **kwargs)
+        try:
+            obj.__init__(**kwargs)
+        except TypeError as e:
+            print(e)
+            print("cls: {}, obj: {}, kwargs: {}".format(cls, obj, kwargs))
+            logger.error(e)
+            logger.error("cls: {}, obj: {}, kwargs: {}".format(cls, obj, kwargs))
+            raise e
+
+        return obj
+
+    @classmethod
+    def from_annotation(cls, params, global_params: dict, subs_tree=None):
+        if params is None:
+            return params
+        if subs_tree is None:
+            return params
+
+        if isinstance(subs_tree, tuple) and len(subs_tree) > 1:
+            # such as: (Dict, str, int) in List[Dict[str, int]]
+            args_type = subs_tree[0]
+            annotation = subs_tree[1:]
+        elif isinstance(subs_tree, tuple) and len(subs_tree) == 1:
+            args_type = subs_tree[0]
+            annotation = None
+        else:
+            args_type = subs_tree
+            annotation = None
+
+        if args_type is typing.List or args_type is list:
+            result = list()
+            for param in params:
+                result.append(cls.from_annotation(param, global_params, annotation))
+            return result
+        elif args_type is typing.Dict or args_type is list:
+            result = dict()
+            for k, v in params.items():
+                key = cls.from_annotation(k, global_params, annotation[0])
+                value = cls.from_annotation(v, global_params, annotation[1])
+                result[key] = value
+            return result
+        elif args_type is typing.Tuple or args_type is tuple:
+            if len(annotation) != len(params):
+                raise AssertionError(
+                    "number of params not match the annotation. "
+                    "{}, annotation: {}, params: {}".format(cls, annotation, params)
+                )
+            result = list()
+            for param, sub_annotation in zip(params, annotation):
+                result.append(cls.from_annotation(param, global_params, sub_annotation))
+            return tuple(result)
+        elif args_type is typing.Union:
+            for option in annotation:
+                try:
+                    result = cls.from_annotation(params, global_params, option)
+                    break
+                except Exception:
+                    continue
+            else:
+                raise ValueError("no type of Union match the params {}".format(params))
+            return result
+        elif args_type is typing.Any:
+            result = params
+            return result
+
+        if hasattr(typing, "GenericMeta"):
+            built_in_type = typing.GenericMeta
+        elif hasattr(typing, "GenericAlias"):
+            built_in_type = typing.GenericAlias
+        else:
+            raise NotImplementedError
+
+        if not isinstance(args_type, built_in_type):
+            if hasattr(args_type, "from_json"):
+                result = args_type.from_json(params, global_params)
+            elif isinstance(args_type, tuple) and len(args_type) > 0 and isinstance(args_type[0], built_in_type):
+                # List[Dict[str, List[str]]]
+                result = cls.from_annotation(params, global_params, args_type)
+            else:
+                if isinstance(params, dict):
+                    result = args_type(**params)
+                else:
+                    result = args_type(params)
+
+            return result
+
+        raise NotImplementedError(
+            "{}, params: {}, subs_tree: {}".format(cls, params, subs_tree)
+        )
+
+
+if __name__ == "__main__":
+    pass

toolbox/porter/common/registrable.py

@@ -0,0 +1,58 @@
+from collections import defaultdict
+from typing import TypeVar, Type, Dict, List
+import importlib
+import logging
+
+logger = logging.getLogger("toolbox")
+
+T = TypeVar("T")
+
+
+class Registrable(object):
+    _registry: Dict[Type, Dict[str, Type]] = defaultdict(dict)
+    default_implementation: str = None
+    register_name: str = "unknown"
+
+    @classmethod
+    def register(cls: Type[T], name: str, exist_ok=False):
+        registry = Registrable._registry[cls]
+        def add_subclass_to_registry(subclass: Type[T]):
+            # set a name on the subclass
+            setattr(subclass, "register_name", name)
+            if name in registry:
+                if exist_ok:
+                    message = (f"{name} has already been registered as {registry[name].__name__}, but "
+                               f"exist_ok=True, so overwriting with {cls.__name__}")
+                    # logger.info(message)
+                else:
+                    message = (f"Cannot register {name} as {cls.__name__}; "
+                               f"name already in use for {registry[name].__name__}")
+                    raise ValueError(message)
+            registry[name] = subclass
+            return subclass
+        return add_subclass_to_registry
+
+    @classmethod
+    def by_name(cls: Type[T], name: str) -> Type[T]:
+        # logger.info(f"instantiating registered subclass {name} of {cls}")
+        if name in Registrable._registry[cls]:
+            return Registrable._registry[cls].get(name)
+        else:
+            raise ValueError(
+                f"{name} is not a registered name for {cls.__name__}. "
+                f"the available is: [{Registrable._registry[cls].keys()}]"
+            )
+
+
+    @classmethod
+    def list_available(cls) -> List[str]:
+        keys = list(Registrable._registry[cls].keys())
+        default = cls.default_implementation
+
+        if default is None:
+            return keys
+        elif default not in keys:
+            message = "Default implementation %s is not registered" % default
+            raise ValueError(message)
+        else:
+            return [default] + [k for k in keys if k != default]

toolbox/porter/manager.py

@@ -0,0 +1,67 @@
+#!/usr/bin/python3
+# -*- coding: utf-8 -*-
+import asyncio
+import json
+
+from toolbox.porter.tasks.base_task import BaseTask
+
+
+class PorterManager(object):
+    def __init__(self,
+                 tasks_file: str,
+                 ):
+        self.tasks_file = tasks_file
+
+        # state
+        self.coro_task_set = set()
+
+    def get_init_tasks(self):
+        with open(self.tasks_file, "r", encoding="utf-8") as f:
+            tasks = json.load(f)
+
+        for task in tasks:
+            enable = task.pop("enable")
+            task_type = task.pop("type")
+
+            if not enable:
+                continue
+            task_cls = BaseTask.by_name(task_type)
+            task_obj = task_cls(**task)
+
+            self.coro_task_set.add(task_obj.start())
+        return self.coro_task_set
+
+    async def run(self):
+        coro_task_set = self.get_init_tasks()
+
+        future_tasks = list()
+        for task in coro_task_set:
+            task = asyncio.ensure_future(task)
+            # task = asyncio.create_task(task)
+            future_tasks.append(task)
+            await asyncio.sleep(3)
+
+        await asyncio.wait(future_tasks)
+
+    async def run2(self):
+        future_tasks = self.get_init_tasks()
+        await asyncio.wait(future_tasks)
+
+
+async def main():
+    import log
+    from project_settings import environment, project_path, log_directory, time_zone_info
+
+    log.setup_size_rotating(log_directory=log_directory, tz_info=time_zone_info)
+
+    porter_task_file = environment.get("porter_tasks_file")
+
+    youtube_video_upload_tasks_file = project_path / porter_task_file
+
+    manager = PorterManager(youtube_video_upload_tasks_file)
+    await manager.run()
+    return
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

toolbox/porter/tasks/__init__.py

@@ -0,0 +1,10 @@
+#!/usr/bin/python3
+# -*- coding: utf-8 -*-
+from .douyin_live_info_collect_task import DouyinLiveInfoCollectTask
+from .douyin_video_download_task import DouyinVideoDownloadTask
+from .douyin_live_record_task import DouyinLiveRecordTask
+from .douyin_live_to_bilibili_task import DouyinLiveToBilibiliTask
+
+
+if __name__ == "__main__":
+    pass

toolbox/porter/tasks/base_task.py

@@ -0,0 +1,37 @@
+#!/usr/bin/python3
+# -*- coding: utf-8 -*-
+import asyncio
+import logging
+import traceback
+
+logger = logging.getLogger("toolbox")
+
+from toolbox.porter.common.params import Params
+
+
+class BaseTask(Params):
+    def __init__(self,
+                 flag: str,
+                 check_interval: int,
+                 ):
+        super().__init__()
+        self.flag = flag
+        self.check_interval = check_interval
+
+    async def do_task(self):
+        raise NotImplementedError
+
+    async def start(self):
+        while True:
+            try:
+                await self.do_task()
+                logger.info(f"{self.flag}任务检测... 刷新间隔 {self.check_interval}s")
+                await asyncio.sleep(self.check_interval)
+            except Exception as error:
+                logger.error(f"{self.flag}任务检测出错\nerror type: {type(error)}, error text: {error}, traceback: {traceback.format_exc()}")
+                await asyncio.sleep(self.check_interval)
+                continue
+
+
+if __name__ == "__main__":
+    pass