update

data/porter_tasks/porter_task_chenjieshen_douyin_video_to_bilibili.json

@@ -31,6 +31,6 @@
     "target_user_id": "442286660",
     "key_of_credentials": "bilibili_chenjiesen_credentials",
     "remove_after_upload": true,
-    "min_date2": "2025-09-09 00:00:00"
+    "min_date": "2025-09-09 00:00:00"
   }
 ]

toolbox/bilibili/video/video_manager.py

@@ -2,9 +2,7 @@
 # -*- coding: utf-8 -*-
 """
 https://pypi.org/project/biliupload/
-
 https://github.com/SocialSisterYi/bilibili-API-collect
-
 """
 import argparse
 import logging
@@ -31,19 +29,6 @@ class BilibiliVideoUploader(BilibiliClient):
         before_sleep=before_sleep_log(logger, logging.ERROR),
     )
     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,
@@ -76,12 +61,6 @@ class BilibiliVideoUploader(BilibiliClient):
         before_sleep=before_sleep_log(logger, logging.ERROR),
     )
     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,
@@ -189,7 +168,6 @@ class BilibiliVideoUploader(BilibiliClient):
     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:

toolbox/bilibili/video/video_manager2.py

@@ -0,0 +1,475 @@
+#!/usr/bin/python3
+# -*- coding: utf-8 -*-
+"""
+https://pypi.org/project/biliupload/
+
+https://github.com/SocialSisterYi/bilibili-API-collect
+
+"""
+import argparse
+import json
+import logging
+import math
+from pathlib import Path
+import re
+
+logger = logging.getLogger("toolbox")
+
+from tenacity import before_sleep_log, retry, retry_if_exception_type, stop_after_attempt, wait_fixed
+from tqdm import tqdm
+
+from project_settings import project_path
+from toolbox.bilibili.bilibili_client import BilibiliClient
+
+
+class BilibiliVideoUploader(BilibiliClient):
+    def __init__(self):
+        super().__init__()
+
+    @retry(
+        wait=wait_fixed(10),
+        stop=stop_after_attempt(3),
+        before_sleep=before_sleep_log(logger, logging.ERROR),
+    )
+    def preupload_mp4(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 = {
+            "probe_version": "20221109",
+            "upcdn": "bldsa",
+            "zone": "cs",
+            "name": filename,
+            "r": "upos",
+            "profile": "ugcfx/bup",
+            "ssl": 0,
+            "version": "2.14.0.0",
+            "build": "2140000",
+            "size": filesize,
+        }
+        # 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
+
+    @retry(
+        wait=wait_fixed(10),
+        stop=stop_after_attempt(3),
+        before_sleep=before_sleep_log(logger, logging.ERROR),
+    )
+    def preupload_txt(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": "file_meta.txt",
+            "size": 2000,
+            "r": "upos",
+            "profile": "fxmeta/bup",
+            "ssl": 0,
+            "version": "2.14.0.0",
+            "build": "2140000",
+            "webVersion": "2.14.0",
+        }
+        response = self.session.get(
+            url,
+            params=params,
+            headers=self.headers,
+        )
+        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
+
+    @retry(
+        wait=wait_fixed(10),
+        stop=stop_after_attempt(3),
+        before_sleep=before_sleep_log(logger, logging.ERROR),
+    )
+    def get_upload_video_id(self, *, endpoint, upos_uri, auth,
+                            filesize, chunk_size, biz_id,
+                            # txt_upos_uri
+                            ):
+        # endpoint = "//upos-sz-upcdnbda2.bilivideo.com"
+        url = f"https:{endpoint}/{upos_uri}?uploads&output=json&"
+        # url = f"https://upos-sz-upcdnbda2.bilivideo.com/{upos_uri}?uploads&output=json"
+
+        params = {
+            "profile": "ugcfx/bup",
+            "filesize": filesize,
+            "partsize": chunk_size,
+            "biz_id": biz_id,
+            # "meta_upos_uri": txt_upos_uri,
+
+        }
+        response = self.session.post(
+            url,
+            headers={"X-Upos-Auth": auth},
+            params=params,
+        )
+        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, *, endpoint, 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:
+        """
+        # endpoint = "//upos-sz-upcdnbda2.bilivideo.com"
+        url = f"https:{endpoint}/{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,
+        }
+        process_bar = tqdm(desc="bilibili_upload_video", total=chunks)
+        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.options(
+                url,
+                params=params,
+                data=batchbytes,
+                headers={"X-Upos-Auth": auth}
+            )
+            if response.status_code != 200:
+                raise AssertionError(f"request failed; status_code: {response.status_code}, text: {response.text}")
+            process_bar.update(1)
+
+    @retry(
+        wait=wait_fixed(10),
+        stop=stop_after_attempt(3),
+        before_sleep=before_sleep_log(logger, logging.ERROR),
+    )
+    def finish_upload(self, *, endpoint, upos_uri, auth, filename, upload_id, biz_id, chunks):
+        # endpoint = "//upos-sz-upcdnbda2.bilivideo.com"
+        url = f"https:{endpoint}/{upos_uri}"
+
+        params = {
+            "output": "json",
+            "name":	filename,
+            "profile" : "ugcfx/bup",
+            "uploadId": upload_id,
+            "biz_id": biz_id
+        }
+        print(params)
+        data = {
+            "parts": [
+                {"partNumber": i+1, "eTag": "etag"}
+                for i in range(chunks)
+            ]
+        }
+        print(data)
+        response = self.session.post(
+            url,
+            params=params,
+            data=json.dumps(data),
+            headers={
+                "X-Upos-Auth": auth,
+                "Content-Type": "application/json",
+                "referer": "https://member.bilibili.com/",
+                "user-agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/139.0.0.0 Safari/537.36"
+            }
+        )
+        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
+
+    @retry(
+        wait=wait_fixed(10),
+        stop=stop_after_attempt(3),
+        before_sleep=before_sleep_log(logger, logging.ERROR),
+    )
+    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_mp4(filename=filename, filesize=filesize)
+        endpoint = js["endpoint"]
+        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)
+
+        # js = self.preupload_txt(filename=filename, filesize=filesize)
+        # endpoint = js["endpoint"]
+        # txt_upos_uri = js["upos_uri"]
+        # 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(
+            endpoint=endpoint,
+            upos_uri=upos_uri,
+            auth=auth,
+            filesize=filesize,
+            chunk_size=chunk_size,
+            biz_id=biz_id,
+            # txt_upos_uri=txt_upos_uri,
+
+        )
+        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(
+            endpoint=endpoint,
+            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(
+            endpoint=endpoint,
+            upos_uri=upos_uri, auth=auth, filename=filename.name,
+            upload_id=upload_id, biz_id=biz_id, chunks=chunks
+        )
+
+        return biz_id, upload_id, bilibili_filename
+
+    def upload_video_and_publish(self, filename: str, metadata: dict):
+        _, upload_id, bilibili_filename = self.upload_video_file(filename)
+        logger.info(f"finish uploading, upload_id: {upload_id}, bilibili_filename: {bilibili_filename}, filename: {filename}")
+        tid = self.predict_tid(
+            title=metadata["title"],
+            upload_id=upload_id,
+            bilibili_filename=bilibili_filename,
+        )
+        metadata["tid"] = tid
+        logger.info(f"predict tid, tid: {tid}, filename: {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
+
+    @retry(
+        wait=wait_fixed(10),
+        stop=stop_after_attempt(3),
+        before_sleep=before_sleep_log(logger, logging.ERROR),
+    )
+    def predict_tid(self, title: str, upload_id: str, bilibili_filename: str):
+        csrf = self.cookies["bili_jct"]
+        url = f"https://member.bilibili.com/x/vupre/web/archive/types/predict"
+        params = {
+            "csrf": csrf,
+        }
+        data = {
+            "title": title,
+            "upload_id": upload_id,
+            "filename": bilibili_filename,
+        }
+        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 get_args():
+    parser = argparse.ArgumentParser()
+    parser.add_argument(
+        "--key_of_credentials",
+        default="bilibili_chenjiesen_credentials",
+        type=str
+    )
+    parser.add_argument(
+        "--filename",
+        default=(project_path / "data/tasks/chenjieshen_douyin_video_to_bilibili/video/douyin/陈杰森/7548332163859533083_20250910T170000_千万不要帮朋友担保！否则你可能背上一身债！.mp4").as_posix(),
+        type=str
+    )
+    args = parser.parse_args()
+    return args
+
+
+def main():
+    import log
+    from project_settings import project_path, log_directory, environment
+
+    log.setup_size_rotating(log_directory=log_directory)
+
+    args = get_args()
+
+    client = BilibiliVideoUploader()
+
+    flag = client.check_login()
+    print(f"flag: {flag}")
+    credentials = environment.get(args.key_of_credentials, dtype=json.loads)
+    client.login_with_credentials_info(credentials)
+    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()