tvbox/upload_to_webdav.py

import os
import requests
from requests.auth import HTTPBasicAuth

def upload_folder_to_webdav(local_folder_path, remote_webdav_path, webdav_url, username, password, exclude_patterns=None):
    """
    将本地文件夹中的所有文件复制到指定的WebDAV路径，支持排除指定目录和文件
    
    Args:
        local_folder_path (str): 本地文件夹路径
        remote_webdav_path (str): WebDAV上的目标路径
        webdav_url (str): WebDAV服务器URL
        username (str): WebDAV用户名
        password (str): WebDAV密码
        exclude_patterns (list): 要排除的目录或文件名列表，默认排除 ['.git', '.DS_Store']
    
    Returns:
        bool: 上传成功返回True，否则返回False
    """
    
    # 默认排除列表
    if exclude_patterns is None:
        exclude_patterns = ['.git', '.DS_Store', '__pycache__', '.svn', '.hg']
    
    # 检查本地文件夹是否存在
    if not os.path.exists(local_folder_path):
        print(f"本地文件夹 {local_folder_path} 不存在")
        return False
        
    if not os.path.isdir(local_folder_path):
        print(f"{local_folder_path} 不是一个文件夹")
        return False
    
    print(f"开始上传文件夹 {local_folder_path} 到 {webdav_url}{remote_webdav_path}")
    print(f"排除模式: {exclude_patterns}")
    
    # 遍历本地文件夹中的所有文件和子文件夹
    for root, dirs, files in os.walk(local_folder_path):
        # 过滤目录，移除需要排除的目录
        dirs[:] = [d for d in dirs if d not in exclude_patterns]
        
        # 过滤文件，移除需要排除的文件
        files = [f for f in files if f not in exclude_patterns]
        
        # 计算相对路径
        relative_path = os.path.relpath(root, local_folder_path)
        if relative_path == ".":
            relative_path = ""
            
        # 创建远程路径
        if relative_path:
            remote_dir_path = os.path.join(remote_webdav_path, relative_path).replace("\\", "/")
        else:
            remote_dir_path = remote_webdav_path
            
        # 确保远程目录存在
        if not create_webdav_directory(remote_dir_path, webdav_url, username, password):
            print(f"创建远程目录 {remote_dir_path} 失败")
            continue
            
        # 上传文件
        for file in files:
            local_file_path = os.path.join(root, file)
            # 计算远程文件路径
            if relative_path:
                remote_file_path = os.path.join(remote_webdav_path, relative_path, file).replace("\\", "/")
            else:
                remote_file_path = os.path.join(remote_webdav_path, file).replace("\\", "/")
                
            if not upload_file_to_webdav(local_file_path, remote_file_path, webdav_url, username, password):
                print(f"上传文件 {local_file_path} 失败")
                continue
                
    print(f"文件夹 {local_folder_path} 上传完成")
    return True

def create_webdav_directory(remote_dir_path, webdav_url, username, password):
    """
    在WebDAV上创建目录
    
    Args:
        remote_dir_path (str): WebDAV上的目录路径
        webdav_url (str): WebDAV服务器URL
        username (str): WebDAV用户名
        password (str): WebDAV密码
    
    Returns:
        bool: 创建成功返回True，否则返回False
    """
    try:
        response = requests.request("MKCOL", f"{webdav_url}{remote_dir_path}", 
                                  auth=HTTPBasicAuth(username, password))
        # 201表示创建成功，405表示目录已存在
        if response.status_code in [201, 405]:
            return True
        else:
            # 忽略目录已存在的错误
            if response.status_code == 405:
                return True
            print(f"创建目录 {remote_dir_path} 失败: {response.status_code}")
            return False
    except Exception as e:
        print(f"创建目录 {remote_dir_path} 时出错: {e}")
        return False

def upload_file_to_webdav(local_file_path, remote_file_path, webdav_url, username, password):
    """
    上传单个文件到WebDAV
    
    Args:
        local_file_path (str): 本地文件路径
        remote_file_path (str): WebDAV上的文件路径
        webdav_url (str): WebDAV服务器URL
        username (str): WebDAV用户名
        password (str): WebDAV密码
    
    Returns:
        bool: 上传成功返回True，否则返回False
    """
    try:
        # 先删除WebDAV上同名文件（如果存在）
        delete_from_webdav(remote_file_path, webdav_url, username, password)
        
        with open(local_file_path, 'rb') as f:
            response = requests.put(
                f"{webdav_url}{remote_file_path}",
                data=f,
                auth=HTTPBasicAuth(username, password)
            )
            if response.status_code in [201, 204]:
                print(f"文件 {local_file_path} 上传成功到 {remote_file_path}")
                return True
            else:
                print(f"文件 {local_file_path} 上传失败: {response.status_code} {response.text}")
                return False
    except Exception as e:
        print(f"上传文件 {local_file_path} 时出错: {e}")
        return False

def delete_from_webdav(remote_path, webdav_url, username, password):
    """
    从WebDAV删除文件
    
    Args:
        remote_path (str): WebDAV上的文件路径
        webdav_url (str): WebDAV服务器URL
        username (str): WebDAV用户名
        password (str): WebDAV密码
    
    Returns:
        bool: 删除成功或文件不存在返回True，否则返回False
    """
    try:
        response = requests.delete(
            f"{webdav_url}{remote_path}",
            auth=HTTPBasicAuth(username, password)
        )
        # 204表示删除成功，404表示文件不存在（也认为是成功）
        if response.status_code in [204, 404]:
            return True
        else:
            # 忽略删除失败（可能文件不存在）
            return True
    except Exception as e:
        print(f"删除文件 {remote_path} 时出错: {e}")
        return False

# 使用示例
if __name__ == "__main__":
    # 配置参数
    local_folder = "../tvbox"  # 本地文件夹路径
    remote_path = "/home/TVBox/Private/tvbox"  # WebDAV目标路径
    webdav_url = "http://47.106.254.96:9120/dav"  # WebDAV服务器URL
    username = "lwang"  # 用户名
    password = "lw19971017"  # 密码
    
    # 排除列表
    exclude_list = ['.git', '.DS_Store', '__pycache__', '.svn', '.hg', 'node_modules']
    
    # 执行上传
    upload_folder_to_webdav(local_folder, remote_path, webdav_url, username, password, exclude_list)