EveryPublish/myUtils/postVideo.py

import asyncio
from pathlib import Path

from conf import BASE_DIR
from uploader.douyin_uploader.main import DouYinVideo, DouYinNote
from uploader.ks_uploader.main import KSVideo, KSNote
from uploader.tencent_uploader.main import TencentVideo, TencentNote
from uploader.xiaohongshu_uploader.main import XiaoHongShuVideo, XiaoHongShuNote
from uploader.baijiahao_uploader.main import BaiJiaHaoVideo
from uploader.alipay_uploader.main import AlipayVideo
from uploader.weibo_uploader.main import WeiBoVideo
from uploader.hupu_uploader.main import HuPuVideo
from uploader.youtube_uploader.main import YouTubeVideo
from uploader.bilibili_uploader.runtime import run_biliup_command
from uploader.tk_uploader.main_chrome import TiktokVideo
from utils.settings import get_bool
from utils.constant import TencentZoneTypes
from utils.files_times import generate_schedule_time_next_day


def _publish_headless():
    return get_bool("publish_headless", True)


def _resolve_files(files):
    return [Path(BASE_DIR / "videoFile" / file) for file in files]


def _resolve_cookies(account_file):
    return [Path(BASE_DIR / "cookiesFile" / file) for file in account_file]


def _extract_cover(video_path: Path) -> str | None:
    """提取视频第一帧作为封面（百家号发布必须提供横版封面）。"""
    try:
        import cv2
        cap = cv2.VideoCapture(str(video_path))
        ret, frame = cap.read()
        cap.release()
        if ret:
            cover = Path(str(video_path) + ".cover.jpg")
            cv2.imwrite(str(cover), frame)
            return str(cover)
    except Exception:
        pass
    return None


def post_video_tencent(title, files, tags, account_file, category=TencentZoneTypes.LIFESTYLE.value, enableTimer=False, videos_per_day=1, daily_times=None, start_days=0, is_draft=False):
    account_file = _resolve_cookies(account_file)
    files = _resolve_files(files)
    if enableTimer:
        publish_datetimes = generate_schedule_time_next_day(len(files), videos_per_day, daily_times, start_days)
    else:
        publish_datetimes = [0 for i in range(len(files))]
    for index, file in enumerate(files):
        for cookie in account_file:
            print(f"视频文件名：{file}")
            print(f"标题：{title}")
            app = TencentVideo(title, str(file), tags, publish_datetimes[index], cookie, category, is_draft, headless=_publish_headless())
            asyncio.run(app.main(), debug=False)


def post_video_DouYin(title, files, tags, account_file, category=TencentZoneTypes.LIFESTYLE.value, enableTimer=False, videos_per_day=1, daily_times=None, start_days=0,
                      thumbnail_path='',
                      productLink='', productTitle=''):
    account_file = _resolve_cookies(account_file)
    files = _resolve_files(files)
    if enableTimer:
        publish_datetimes = generate_schedule_time_next_day(len(files), videos_per_day, daily_times, start_days)
    else:
        publish_datetimes = [0 for i in range(len(files))]
    for index, file in enumerate(files):
        for cookie in account_file:
            print(f"视频文件名：{file}")
            print(f"标题：{title}")
            app = DouYinVideo(title, str(file), tags, publish_datetimes[index], cookie, thumbnail_path, productLink, productTitle, headless=_publish_headless())
            asyncio.run(app.douyin_upload_video(), debug=False)


def post_video_ks(title, files, tags, account_file, category=TencentZoneTypes.LIFESTYLE.value, enableTimer=False, videos_per_day=1, daily_times=None, start_days=0):
    account_file = _resolve_cookies(account_file)
    files = _resolve_files(files)
    if enableTimer:
        publish_datetimes = generate_schedule_time_next_day(len(files), videos_per_day, daily_times, start_days)
    else:
        publish_datetimes = [0 for i in range(len(files))]
    for index, file in enumerate(files):
        for cookie in account_file:
            print(f"视频文件名：{file}")
            print(f"标题：{title}")
            app = KSVideo(title, str(file), tags, publish_datetimes[index], cookie, headless=_publish_headless())
            asyncio.run(app.main(), debug=False)


def post_video_xhs(title, files, tags, account_file, category=TencentZoneTypes.LIFESTYLE.value, enableTimer=False, videos_per_day=1, daily_times=None, start_days=0):
    account_file = _resolve_cookies(account_file)
    files = _resolve_files(files)
    file_num = len(files)
    if enableTimer:
        publish_datetimes = generate_schedule_time_next_day(file_num, videos_per_day, daily_times, start_days)
    else:
        publish_datetimes = 0
    for index, file in enumerate(files):
        for cookie in account_file:
            print(f"视频文件名：{file}")
            print(f"标题：{title}")
            app = XiaoHongShuVideo(title, file, tags, publish_datetimes, cookie, headless=_publish_headless())
            asyncio.run(app.main(), debug=False)


# ===== 新增：CLI 主线平台接入 Web 后端 =====

def post_video_bilibili(title, files, tags, account_file, category=None, enableTimer=False, videos_per_day=1, daily_times=None, start_days=0, desc='', tid=249):
    """B站视频发布（封装 biliup）。Web 端暂不支持定时（立即发布），定时请用 CLI。"""
    account_file = _resolve_cookies(account_file)
    files = _resolve_files(files)
    for file in files:
        for cookie in account_file:
            args = ["-u", str(cookie), "upload", str(file), "--title", title, "--desc", desc or "", "--tid", str(tid)]
            if tags:
                args.extend(["--tag", ",".join(tags)])
            print(f"biliup 命令: {' '.join(args)}")
            result = run_biliup_command(args)
            if result.returncode != 0:
                raise RuntimeError((result.stderr or result.stdout or "").strip() or "B站上传失败")


def post_video_baijiahao(title, files, tags, account_file, category=None, enableTimer=False, videos_per_day=1, daily_times=None, start_days=0, thumbnail_path='', desc=''):
    account_file = _resolve_cookies(account_file)
    files = _resolve_files(files)
    for file in files:
        cover = thumbnail_path or _extract_cover(file)
        for cookie in account_file:
            app = BaiJiaHaoVideo(
                title=title, file_path=str(file), tags=tags, account_file=str(cookie),
                desc=desc or "", thumbnail_path=cover, collection_name=None, debug=False, headless=_publish_headless(),
            )
            asyncio.run(app.main(), debug=False)


def post_video_alipay(title, files, tags, account_file, category=None, enableTimer=False, videos_per_day=1, daily_times=None, start_days=0, thumbnail_path='', desc=''):
    account_file = _resolve_cookies(account_file)
    files = _resolve_files(files)
    for file in files:
        for cookie in account_file:
            app = AlipayVideo(
                title=title, file_path=str(file), tags=tags, account_file=str(cookie),
                desc=desc or "", thumbnail_path=thumbnail_path or None, collection_name=None, debug=False, headless=_publish_headless(),
            )
            asyncio.run(app.main(), debug=False)


def post_video_weibo(title, files, tags, account_file, category=None, enableTimer=False, videos_per_day=1, daily_times=None, start_days=0, thumbnail_path='', desc=''):
    account_file = _resolve_cookies(account_file)
    files = _resolve_files(files)
    for file in files:
        for cookie in account_file:
            app = WeiBoVideo(
                title=title, file_path=str(file), tags=tags, account_file=str(cookie),
                desc=desc or "", thumbnail_path=thumbnail_path or None, collection_name=None, debug=False, headless=_publish_headless(),
            )
            asyncio.run(app.main(), debug=False)


def post_video_hupu(title, files, tags, account_file, category=None, enableTimer=False, videos_per_day=1, daily_times=None, start_days=0, thumbnail_path='', desc=''):
    account_file = _resolve_cookies(account_file)
    files = _resolve_files(files)
    for file in files:
        for cookie in account_file:
            app = HuPuVideo(
                title=title, file_path=str(file), tags=tags, account_file=str(cookie),
                desc=desc or "", thumbnail_path=thumbnail_path or None, debug=False, headless=_publish_headless(),
            )
            asyncio.run(app.main(), debug=False)


def post_video_youtube(title, files, tags, account_file, category=None, enableTimer=False, videos_per_day=1, daily_times=None, start_days=0, thumbnail_path='', desc=''):
    account_file = _resolve_cookies(account_file)
    files = _resolve_files(files)
    for file in files:
        for cookie in account_file:
            app = YouTubeVideo(
                title, str(file), tags, str(cookie),
                description=desc or "", thumbnail_path=thumbnail_path or None,
                playlist=None, visibility="public", debug=False, headless=_publish_headless(),
            )
            asyncio.run(app.main(), debug=False)

def post_video_tiktok(title, files, tags, account_file, category=None, enableTimer=False, videos_per_day=1, daily_times=None, start_days=0, thumbnail_path="", desc=""):
    """TikTok 视频发布（Chrome 版实现）。支持定时发布与封面（thumbnail）。"""
    account_file = _resolve_cookies(account_file)
    files = _resolve_files(files)
    if enableTimer:
        publish_datetimes = generate_schedule_time_next_day(len(files), videos_per_day, daily_times, start_days)
    else:
        publish_datetimes = [0 for i in range(len(files))]
    for index, file in enumerate(files):
        for cookie in account_file:
            app = TiktokVideo(
                title, str(file), tags, publish_datetimes[index], str(cookie),
                thumbnail_path if thumbnail_path else None,
            )
            app.headless = _publish_headless()
            asyncio.run(app.main(), debug=False)


# ===== 图文（note）发布 —— 小红书/抖音/快手 =====

def post_note_xiaohongshu(title, files, tags, account_file, desc=""):
    account_files = _resolve_cookies(account_file)
    image_files = _resolve_files(files)
    for cookie in account_files:
        app = XiaoHongShuNote(
            image_paths=[str(f) for f in image_files],
            note=desc or "",
            tags=tags,
            publish_date=0,
            account_file=str(cookie),
            title=title,
            desc=desc or "",
            debug=False,
            headless=_publish_headless(),
        )
        asyncio.run(app.main(), debug=False)


def post_note_douyin(title, files, tags, account_file, desc=""):
    account_files = _resolve_cookies(account_file)
    image_files = _resolve_files(files)
    for cookie in account_files:
        app = DouYinNote(
            image_paths=[str(f) for f in image_files],
            note=desc or "",
            tags=tags,
            publish_date=0,
            account_file=str(cookie),
            title=title,
            debug=False,
            headless=_publish_headless(),
        )
        asyncio.run(app.douyin_upload_note(), debug=False)


def post_note_kuaishou(title, files, tags, account_file, desc=""):
    account_files = _resolve_cookies(account_file)
    image_files = _resolve_files(files)
    for cookie in account_files:
        app = KSNote(
            image_paths=[str(f) for f in image_files],
            note=desc or "",
            tags=tags,
            publish_date=0,
            account_file=str(cookie),
            title=title,
            debug=False,
            headless=_publish_headless(),
        )
        asyncio.run(app.main(), debug=False)


def post_note_tencent(title, files, tags, account_file, desc="", is_draft=False):
    account_files = _resolve_cookies(account_file)
    image_files = _resolve_files(files)
    for cookie in account_files:
        app = TencentNote(
            image_paths=[str(f) for f in image_files],
            note=desc or "",
            tags=tags,
            publish_date=0,
            account_file=str(cookie),
            title=title,
            debug=False,
            headless=_publish_headless(),
            is_draft=is_draft,
        )
        asyncio.run(app.main(), debug=False)