Python自动化处理之文件批量下载与自动压缩

import requests
import os
from concurrent.futures import ThreadPoolExecutor, as_completed
import time

class BatchDownloader:
    """批量下载器"""

    def __init__(self, save_dir="downloads", max_workers=5):
        self.save_dir = save_dir
        self.max_workers = max_workers
        os.makedirs(save_dir, exist_ok=True)

    def download_one(self, url, filename=None):
        """下载单个文件"""
        try:
            if not filename:
                filename = url.split("/")[-1] or "index.html"

            filepath = os.path.join(self.save_dir, filename)

            # 断点续传
            headers = {}
            if os.path.exists(filepath):
                headers["Range"] = f"bytes={os.path.getsize(filepath)}-"

            resp = requests.get(url, headers=headers, timeout=30, stream=True)
            resp.raise_for_status()

            mode = "ab" if os.path.exists(filepath) else "wb"

            with open(filepath, mode) as f:
                for chunk in resp.iter_content(chunk_size=8192):
                    if chunk:
                        f.write(chunk)

            size = os.path.getsize(filepath)
            print(f"✅ {filename} ({size/1024:.1f}KB)")
            return filepath

        except Exception as e:
            print(f"❌ {filename or url}: {e}")
            return None

    def download_many(self, urls, filenames=None):
        """多线程批量下载"""
        if filenames is None:
            filenames = [None] * len(urls)

        start = time.time()
        success = 0

        with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
            futures = {
                executor.submit(self.download_one, url, name): url
                for url, name in zip(urls, filenames)
            }

            for future in as_completed(futures):
                if future.result():
                    success += 1

        elapsed = time.time() - start
        print(f"\n下载完成：{success}/{len(urls)} 成功，耗时 {elapsed:.1f}s")

# 使用
downloader = BatchDownloader("教程附件", max_workers=10)

# 从文本文件读取 URL 列表
with open("urls.txt", "r") as f:
    urls = [line.strip() for line in f if line.strip()]

downloader.download_many(urls)

import requests
from bs4 import BeautifulSoup
import os
from urllib.parse import urljoin

def download_images(page_url, save_dir="images"):
    """下载网页中的所有图片"""
    os.makedirs(save_dir, exist_ok=True)

    # 获取页面内容
    resp = requests.get(page_url)
    soup = BeautifulSoup(resp.text, "html.parser")

    # 找到所有图片
    img_tags = soup.find_all("img")
    img_urls = []

    for img in img_tags:
        src = img.get("src")
        if src:
            full_url = urljoin(page_url, src)
            # 只下载常见图片格式
            if any(ext in full_url.lower() for ext in [".jpg", ".jpeg", ".png", ".gif", ".webp"]):
                img_urls.append(full_url)

    print(f"找到 {len(img_urls)} 张图片")

    # 下载
    downloader = BatchDownloader(save_dir)
    downloader.download_many(img_urls)

# 使用
download_images("https://example.com/article/1")

import requests

def download_github_dir(repo, dir_path, save_dir="github_files"):
    """下载 GitHub 仓库中的指定目录"""
    api_url = f"https://api.github.com/repos/{repo}/contents/{dir_path}"
    resp = requests.get(api_url)
    files = resp.json()

    urls = []
    names = []
    for f in files:
        if f["type"] == "file":
            urls.append(f["download_url"])
            names.append(f["name"])

    downloader = BatchDownloader(save_dir)
    downloader.download_many(urls, names)

# 使用
download_github_dir("spring-projects/spring-boot", "starters")

import shutil
import os
from datetime import datetime

def zip_folder(folder_path, output_name=None):
    """压缩整个文件夹"""
    if not output_name:
        folder_name = os.path.basename(folder_path)
        today = datetime.now().strftime("%Y%m%d")
        output_name = f"{folder_name}_{today}"

    shutil.make_archive(output_name, "zip", folder_path)
    print(f"已压缩: {output_name}.zip ({os.path.getsize(output_name+'.zip')/1024/1024:.1f}MB)")
    return f"{output_name}.zip"

# 使用
zip_folder("downloads", "课件包_20260630")

import zipfile

def zip_selected_files(directory, extensions, output_name="selected"):
    """只压缩指定后缀名的文件"""
    with zipfile.ZipFile(f"{output_name}.zip", "w", zipfile.ZIP_DEFLATED) as zf:
        for root, _, files in os.walk(directory):
            for f in files:
                if any(f.endswith(ext) for ext in extensions):
                    filepath = os.path.join(root, f)
                    arcname = os.path.relpath(filepath, directory)
                    zf.write(filepath, arcname)
                    print(f"已添加: {arcname}")

    print(f"压缩完成: {output_name}.zip")

# 使用：只压缩 PDF 和 Word 文件
zip_selected_files("课件", [".pdf", ".docx"], "课件PDF版")

# 使用 pyminizip 库
# pip install pyminizip

import pyminizip

def zip_with_password(folder_path, password, output_name="protected"):
    """带密码压缩"""
    pyminizip.compress_dir(folder_path, f"{output_name}.zip", password, 5)
    print(f"已加密压缩: {output_name}.zip")

# 使用
zip_with_password("机密文件", "mypassword123", "机密文件_加密")

def clean_source(source_path):
    """清理源文件"""
    if os.path.isfile(source_path):
        os.remove(source_path)
        print(f"已删除: {source_path}")
    elif os.path.isdir(source_path):
        shutil.rmtree(source_path)
        print(f"已删除目录: {source_path}")

def download_and_pack(urls_file, output_name="打包文件"):
    """下载 → 压缩 → 清理 一条龙"""
    base_dir = "temp_download"

    # 1. 读取 URL
    with open(urls_file, "r") as f:
        urls = [line.strip() for line in f if line.strip()]
    print(f"共 {len(urls)} 个文件待下载")

    # 2. 批量下载（10 线程）
    downloader = BatchDownloader(base_dir, max_workers=10)
    results = downloader.download_many(urls)

    # 3. 检查是否全部成功
    success_count = sum(1 for r in results if r)
    print(f"成功下载: {success_count}/{len(results)}")

    if success_count > 0:
        # 4. 压缩
        zip_file = zip_folder(base_dir, output_name)

        # 5. 清理临时文件
        shutil.rmtree(base_dir)
        print(f"临时文件已清理")

        return zip_file

# 使用
download_and_pack("urls.txt", "课程资料包_202607")

import schedule
import time

def daily_download_task():
    """每日定时下载任务"""
    print(f"开始定时下载任务: {datetime.now()}")

    # 1. 从 API 获取今日需要下载的文件列表
    resp = requests.get("https://api.example.com/today/files")
    files = resp.json()

    urls = [f["url"] for f in files]
    names = [f["name"] for f in files]

    # 2. 下载
    downloader = BatchDownloader("每日更新")
    downloader.download_many(urls, names)

    # 3. 压缩
    zip_folder("每日更新", f"更新包_{datetime.now().strftime('%Y%m%d')}")

    print("任务完成")

# 每天早上 8 点执行
schedule.every().day.at("08:00").do(daily_download_task)

while True:
    schedule.run_pending()
    time.sleep(60)