小红薯可以发送大于5m的视频了,但是需要再测试

[yfypyy]

作者用的另一个仓库的代码 那个仓库现在跟新了可以使用分片,来上传!但是仍然需要注意的是对于新号会出现验证码的问题

我是在xhs/core.py文件中的upload_file_with_slice功能加了等待时间2秒! 然后就可以使用了 但是但是我上传到第三个文件的时候 又出现了验证码 我真的没折了 真的很不明白为啥 难道是因为2秒太长了????? 希望能有大佬能解答一下

[yfypyy]

试出来了 我废了两个号 基本上每次post或者get请求一定一定要用随机休眠的方法

# 添加速率限制器，最小请求间隔0.5秒 self.rate_limiter = RateLimiter(min_interval=0.5)

def request(self, method, url, **kwargs):
# 在每次请求前应用速率限制
self.rate_limiter.wait()
upload_xhs.py:
import configparser
from pathlib import Path
from time import sleep
import random
import sys

from xhs import XhsClient
from xhs.exception import NeedVerifyError

from conf import BASE_DIR
from utils.files_times import generate_schedule_time_next_day, get_title_and_hashtags
from uploader.xhs_uploader.main import sign_local, beauty_print

config = configparser.RawConfigParser()
config.read(Path(BASE_DIR / "uploader" / "xhs_uploader" / "accounts.ini"))

def safe_upload_video(xhs_client, file, title, tags, topics, publish_time, max_retries=3):
"""安全地上传视频，包含重试机制"""
tags_str = ' '.join(['#' + tag for tag in tags])
hash_tags_str = ' ' + ' '.join(['#' + tag + '[话题]#' for tag in [t['name'] for t in topics]])

for attempt in range(max_retries):
    try:
        note = xhs_client.create_video_note(
            title=title[:20], 
            video_path=str(file),
            desc=title + tags_str + hash_tags_str,
            topics=topics,
            is_private=False,
            post_time=publish_time
        )
        return note
    except NeedVerifyError as e:
        print(f"\n⚠️  出现验证码：{e}")
        print(f"验证类型: {e.verify_type}")
        print(f"验证UUID: {e.verify_uuid}")
        
        if attempt < max_retries - 1:
            # 出现验证码后等待更长时间 (5-10分钟)
            wait_minutes = random.uniform(5, 10)
            print(f"等待 {wait_minutes:.1f} 分钟后重试 (第 {attempt + 1}/{max_retries} 次)...")
            sleep(wait_minutes * 60)
        else:
            print(f"已达到最大重试次数，跳过此视频")
            raise
    except Exception as e:
        print(f"\n上传出错: {str(e)}")
        if attempt < max_retries - 1:
            wait_time = random.uniform(30, 60)
            print(f"等待 {wait_time:.1f} 秒后重试...")
            sleep(wait_time)
        else:
            raise

return None

if name == 'main':
filepath = Path(BASE_DIR) / "videos"
# 获取视频目录
folder_path = Path(filepath)
# 获取文件夹中的所有文件
files = list(folder_path.glob("*.mp4"))
file_num = len(files)

if file_num == 0:
    print("未找到视频文件！")
    sys.exit(1)

print(f"找到 {file_num} 个视频文件待上传\n")

cookies = config['account1']['cookies']
xhs_client = XhsClient(cookies, sign=sign_local, timeout=60)

# auth cookie
print("正在验证 Cookie...")
try:
    xhs_client.get_video_first_frame_image_id("3214")
    print("✓ Cookie 验证成功\n")
except Exception as e:
    print(f"✗ Cookie 失效: {e}")
    sys.exit(1)

publish_datetimes = generate_schedule_time_next_day(file_num, 1, daily_times=[16])

success_count = 0
failed_count = 0

for index, file in enumerate(files):
    try:
        print(f"\n{'='*60}")
        print(f"处理第 {index + 1}/{file_num} 个视频")
        print(f"{'='*60}")
        
        # 在处理每个视频前添加随机延迟 (5-15秒)，模拟人工准备下一个视频的时间
        if index > 0:
            wait_time = random.uniform(5, 15)
            print(f"等待 {wait_time:.1f} 秒后处理下一个视频...")
            sleep(wait_time)
        
        title, tags = get_title_and_hashtags(str(file))
        # 加入到标题 补充标题（xhs 可以填1000字不写白不写）
        tags_str = ' '.join(['#' + tag for tag in tags])
        hash_tags_str = ''
        hash_tags = []

        # 打印视频文件名、标题和 hashtag
        print(f"\n视频文件名：{file}")
        print(f"标题：{title}")
        print(f"Hashtag：{tags}")

        topics = []
        # 获取hashtag，每次搜索之间添加随机延迟
        print("\n正在搜索话题...")
        for i in tags[:3]:
            # 在搜索话题前添加随机延迟 (1-3秒)
            sleep(random.uniform(1, 3))
            try:
                topic_official = xhs_client.get_suggest_topic(i)
                if topic_official:
                    topic_official[0]['type'] = 'topic'
                    topic_one = topic_official[0]
                    hash_tag_name = topic_one['name']
                    hash_tags.append(hash_tag_name)
                    topics.append(topic_one)
                    print(f"  找到话题: {hash_tag_name}")
            except Exception as e:
                print(f"  搜索话题 '{i}' 失败: {e}")
                continue

        hash_tags_str = ' ' + ' '.join(['#' + tag + '[话题]#' for tag in hash_tags])

        # 在创建笔记前添加随机延迟 (2-5秒)，模拟填写完信息后的检查时间
        print(f"\n准备上传视频...")
        sleep(random.uniform(2, 5))
        
        note = safe_upload_video(
            xhs_client, 
            file, 
            title, 
            tags, 
            topics, 
            publish_datetimes[index].strftime("%Y-%m-%d %H:%M:%S")
        )

        if note:
            beauty_print(note)
            success_count += 1
            print(f"\n✓ 视频发布成功！({success_count}/{index + 1})")
            
            # 发布成功后，添加更长的随机休眠时间 (30-60秒)
            wait_time = random.uniform(30, 60)
            print(f"等待 {wait_time:.1f} 秒后继续...")
            sleep(wait_time)
        
    except NeedVerifyError as e:
        failed_count += 1
        print(f"\n✗ 触发验证码，跳过此视频 ({failed_count} 失败)")
        # 触发验证码后等待更长时间 (10-15分钟)
        error_wait = random.uniform(600, 900)
        print(f"等待 {error_wait/60:.1f} 分钟后继续...")
        sleep(error_wait)
        continue
        
    except Exception as e:
        failed_count += 1
        print(f"\n✗ 处理视频 {file} 时出错: {str(e)}")
        print(f"失败计数: {failed_count}")
        # 出错后等待更长时间 (60-120秒) 再继续
        error_wait = random.uniform(60, 120)
        print(f"等待 {error_wait:.1f} 秒后继续处理下一个视频...")
        sleep(error_wait)
        continue

print(f"\n{'='*60}")
print(f"所有视频处理完成！")
print(f"成功: {success_count} 个")
print(f"失败: {failed_count} 个")
print(f"总计: {file_num} 个")
print(f"{'='*60}")