awaioi/EveryPublish
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add bilibili cli commands

Browse Source
This commit is contained in:
home-dev-pookz
2026-03-25 03:44:56 +08:00
parent 7c3e002429
commit 1bc8ae7698
3 changed files with 149 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
sau_cli.py
+106
View File
@@ -9,6 +9,7 @@ from pathlib import Path
from typing import Iterable, Sequence
from conf import BASE_DIR
from uploader.bilibili_uploader.runtime import run_biliup_command
from uploader.douyin_uploader.main import (
DOUYIN_PUBLISH_STRATEGY_IMMEDIATE,
DOUYIN_PUBLISH_STRATEGY_SCHEDULED,
@@ -81,6 +82,17 @@ class KuaishouNoteUploadRequest:
headless: bool = True
@dataclass(slots=True)
class BilibiliVideoUploadRequest:
account_name: str
video_file: Path
title: str
description: str
tid: int
tags: list[str]
publish_date: datetime | int
def resolve_runtime_home() -> Path:
return Path(BASE_DIR)
@@ -137,6 +149,25 @@ async def check_kuaishou_account(account_name: str) -> bool:
return await kuaishou_cookie_auth(str(account_file))
async def login_bilibili_account(account_name: str) -> dict:
account_file = resolve_account_file("bilibili", account_name)
result = run_biliup_command(["-u", str(account_file), "login"])
success = result.returncode == 0
return {
"success": success,
"message": (result.stderr or result.stdout or "").strip() or "Bilibili login completed" if success else (result.stderr or result.stdout or "").strip() or "Bilibili login failed",
"account_file": str(account_file),
}
async def check_bilibili_account(account_name: str) -> bool:
account_file = resolve_account_file("bilibili", account_name)
if not account_file.exists():
return False
result = run_biliup_command(["-u", str(account_file), "renew"])
return result.returncode == 0
async def upload_video(request: DouyinVideoUploadRequest) -> Path:
account_file = resolve_account_file("douyin", request.account_name)
is_ready = await douyin_setup(str(account_file), handle=False)
@@ -229,6 +260,36 @@ async def upload_kuaishou_note(request: KuaishouNoteUploadRequest) -> Path:
return account_file
async def upload_bilibili_video(request: BilibiliVideoUploadRequest) -> Path:
account_file = resolve_account_file("bilibili", request.account_name)
if not account_file.exists():
raise RuntimeError(
f"Bilibili account file is missing: {account_file}. Run `sau bilibili login --account {request.account_name}` first."
)
arguments = [
"-u",
str(account_file),
"upload",
str(request.video_file),
"--title",
request.title,
"--desc",
request.description,
"--tid",
str(request.tid),
]
if request.tags:
arguments.extend(["--tag", ",".join(request.tags)])
if isinstance(request.publish_date, datetime):
arguments.extend(["--dtime", str(int(request.publish_date.timestamp()))])
result = run_biliup_command(arguments)
if result.returncode != 0:
raise RuntimeError((result.stderr or result.stdout or "").strip() or "Bilibili upload failed")
return account_file
def existing_file_path(value: str) -> Path:
path = Path(value)
if not path.is_file():
@@ -314,6 +375,22 @@ def build_parser() -> argparse.ArgumentParser:
kuaishou_upload_note_parser.add_argument("--tags", default="", help="Comma-separated tags, such as tag1,tag2")
kuaishou_upload_note_parser.add_argument("--schedule", type=schedule_value, help=f"Schedule time in {schedule_help}")
add_runtime_flags(kuaishou_upload_note_parser)
bilibili_parser = platform_parsers.add_parser("bilibili", help="Bilibili operations")
bilibili_actions = bilibili_parser.add_subparsers(dest="action", required=True)
for action_name in ("login", "check"):
action_parser = bilibili_actions.add_parser(action_name, help=f"Bilibili {action_name}")
action_parser.add_argument("--account", required=True, help="Bilibili account alias")
bilibili_upload_video_parser = bilibili_actions.add_parser("upload-video", help="Upload one video to Bilibili")
bilibili_upload_video_parser.add_argument("--account", required=True, help="Bilibili account alias")
bilibili_upload_video_parser.add_argument("--file", required=True, type=existing_file_path, help="Video file path")
bilibili_upload_video_parser.add_argument("--title", required=True, help="Video title")
bilibili_upload_video_parser.add_argument("--desc", required=True, help="Video description")
bilibili_upload_video_parser.add_argument("--tid", required=True, type=int, help="Bilibili category id")
bilibili_upload_video_parser.add_argument("--tags", default="", help="Comma-separated tags, such as tag1,tag2")
bilibili_upload_video_parser.add_argument("--schedule", type=schedule_value, help=f"Schedule time in {schedule_help}")
return parser
@@ -416,6 +493,35 @@ async def dispatch(args: argparse.Namespace) -> int:
raise RuntimeError(f"Unsupported Kuaishou action: {args.action}")
if args.platform == "bilibili":
if args.action == "login":
result = await login_bilibili_account(args.account)
if not result["success"]:
raise RuntimeError(result["message"])
print(f"Bilibili login flow completed: {result['account_file']}")
return 0
if args.action == "check":
is_valid = await check_bilibili_account(args.account)
print("valid" if is_valid else "invalid")
return 0 if is_valid else 1
if args.action == "upload-video":
request = BilibiliVideoUploadRequest(
account_name=args.account,
video_file=args.file,
title=args.title,
description=args.desc,
tid=args.tid,
tags=parse_tags(args.tags),
publish_date=args.schedule or 0,
)
await upload_bilibili_video(request)
print(f"Bilibili video upload submitted: {request.video_file}")
return 0
raise RuntimeError(f"Unsupported Bilibili action: {args.action}")
raise RuntimeError(f"Unsupported platform: {args.platform}")
tests/test_sau_bilibili_cli.py
+38
View File
@@ -0,0 +1,38 @@
import asyncio
import unittest
from argparse import Namespace
from unittest.mock import AsyncMock, patch
import sau_cli
class BilibiliCliTests(unittest.TestCase):
def test_build_parser_accepts_bilibili_login(self):
parser = sau_cli.build_parser()
args = parser.parse_args(["bilibili", "login", "--account", "creator"])
self.assertEqual(args.platform, "bilibili")
self.assertEqual(args.action, "login")
def test_build_parser_requires_tid_for_upload_video(self):
parser = sau_cli.build_parser()
with self.assertRaises(SystemExit):
parser.parse_args(
[
"bilibili",
"upload-video",
"--account",
"creator",
"--file",
"demo.mp4",
"--title",
"hello",
"--desc",
"hello",
]
)
def test_dispatch_bilibili_check_prints_valid(self):
args = Namespace(platform="bilibili", action="check", account="creator")
with patch("sau_cli.check_bilibili_account", new=AsyncMock(return_value=True)):
code = asyncio.run(sau_cli.dispatch(args))
self.assertEqual(code, 0)
uploader/bilibili_uploader/runtime.py
+5 -1
View File
@@ -142,7 +142,9 @@ def download_biliup_asset(release: dict, destination: Path) -> Path:
tar_file.extractall(extract_root)
extracted_binary = _pick_executable(extract_root)
shutil.copy2(extracted_binary, destination)
temp_binary = destination.with_suffix(f"{destination.suffix}.tmp")
shutil.copy2(extracted_binary, temp_binary)
temp_binary.replace(destination)
if _normalize_system() != "windows":
destination.chmod(destination.stat().st_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH)
return destination
@@ -171,4 +173,6 @@ def run_biliup_command(arguments: list[str]) -> subprocess.CompletedProcess[str]
check=False,
capture_output=True,
text=True,
encoding="utf-8",
errors="replace",
)