logoChat01.ai

import os import subprocess from concurrent.future...

作成日: 2025年5月8日

使用モデル O3 by Chat01

質問

import os
import subprocess
from concurrent.futures import ThreadPoolExecutor, as_completed
import sys

def process_folder(folder_path):
final_folder = os.path.join(folder_path, "FINAL")
os.makedirs(final_folder, exist_ok=True)
print(f"🛠 Начинаем обработку папки: {folder_path}")

text
def render_video(video_path):
    filename = os.path.basename(video_path)
    output_filename = os.path.splitext(filename)[0] + ".mp4"
    output_path = os.path.join(final_folder, output_filename)

    cmd = [
        'ffmpeg', '-y', '-i', video_path,
        '-vf', 'scale=1920:1080',
        '-r', '30',
        '-c:v', 'libx264',
        '-pix_fmt', 'yuv420p',
        output_path
    ]

    try:
        subprocess.run(cmd, check=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
        os.remove(video_path)
        print(f"✅ Успешно: {filename}")
        return True
    except Exception as e:
        print(f"🔥 Сожгли файл {filename}: {str(e)}")
        return False

# Собираем все видео рекурсивно
video_files = []
for root, _, files in os.walk(folder_path):
    if "FINAL" in root:
        continue
    for file in files:
        if file.lower().endswith(('.mp4', '.mov', '.mkv', '.qt')):
            video_files.append(os.path.join(root, file))

print(f"📦 Найдено видеофайлов: {len(video_files)}")

# Параллельная обработка
with ThreadPoolExecutor(max_workers=2) as executor:
    futures = [executor.submit(render_video, f) for f in video_files]
    for i, future in enumerate(as_completed(futures), 1):
        print(f"📊 Прогресс: {i}/{len(video_files)}")
        future.result()

# Чистка пустых папок
for root, dirs, _ in os.walk(folder_path, topdown=False):
    for dir in dirs:
        dir_path = os.path.join(root, dir)
        try:
            if not os.listdir(dir_path):
                os.rmdir(dir_path)
                print(f"🧹 Удалили пустую папку: {dir}")
        except Exception:
            pass

if name == "main":
if len(sys.argv) < 2:
print("Укажите путь к папке в аргументах!")
sys.exit(1)

text
folder = sys.argv[1]
if os.path.exists(folder):
    process_folder(folder)
    print("🎉 Всё готово! Можно проверять папку FINAL")
else:
    print(f"❌ Папка {folder} не найдена!")

помоги изменить код. мне нужно чтобы когда я ему передавал folder - он заходил в эту папку и работал с её дочерними папками. и СОЗДАВАЛ В КАЖДОЙ ДОЧЕРНЕЙ ПАПКИ - FINAL ПАПКУ - ТУДА И СОХРАНЯТЬ ОБРАБОТАННЫЕ ПАПКИ. ПОСЛЕ ТОГО КАК ОДНА ДОЧЕРНЯЯ ПАПКА ОБРАБОТАЛАСЬ - НУЖНО ИЗ ПАПКИ FINAL ПЕРЕНЕСТИ ВСЕ ФАЙЛЫ В САМУ ЭТУ ДОЧЕРНЮЮ ПАПКУ ( САМУ FINAL НУЖНО УДАЛИТЬ - НО УБЕДИТЬСЯ ЧТОБЫ ОНА БЫЛА ПУСТАЯ - ИНАЧЕ УДАЛЯТЬ НЕ НУЖНО )
ПОТОМ НУЖНО ПЕРЕХОДИТЬ ПО ВСЕМ ПАПКАМ ТАК ВНУТРИ folder до конца
ПИШИ МНЕ ПОЛНОСТЬЮ КОД.

思考

Modifying the Python script

The user wants to adjust the way the Python script processes video files. Instead of placing all processed videos in a single "FINAL" folder, each child directory should have its own subfolder named FINAL. After processing, the videos should be moved from the FINAL folder back to the child directory’s root, ensuring no conflicts (like overwriting files). The FINAL folder should be deleted afterward, but only if it’s empty. I’ll make sure to handle potential duplicates and ensure data isn’t lost.

Structuring the solution

The user wants to walk through all subdirectories in a folder recursively, processing videos in each without going into deeper subfolders. For each subdirectory, a "FINAL" folder will hold processed files, which will later be moved to the subdirectory's root. I'll use os.walk for traversal, skipping any directories named FINAL. Files will be moved after processing, with duplicates handled simply by renaming. To keep things efficient, I'll process each directory one by one but parallelize video file processing within each directory. The function process_directory() will be responsible for the given task.

Clarifying script changes

The user wants to process all child directories in a folder by creating a "FINAL" subfolder within each one, converting video files there, and then moving them back to the original directory after processing. They don’t want to process the root folder itself. To ensure this behavior, I’ll use os.walk, skipping the root and the "FINAL" folder, and process one directory at a time. I'll use threading inside the directories to handle videos concurrently. After processing, I’ll move the files and delete the "FINAL" folder if empty. The code will be enclosed in a markdown block.

Thought for 38 seconds

回答

python
import os
import shutil
import subprocess
import sys
from concurrent.futures import ThreadPoolExecutor, as_completed

VIDEO_EXT = ('.mp4', '.mov', '.mkv', '.qt')


def render_video(src: str, dst: str) -> None:
    """
    Конвертирует видео в 1920&times;1080 / 30 fps / H.264 и удаляет исходник.
    """
    cmd = [
        "ffmpeg", "-y", "-i", src,
        "-vf", "scale=1920:1080",
        "-r", "30",
        "-c:v", "libx264",
        "-pix_fmt", "yuv420p",
        dst
    ]
    subprocess.run(cmd, check=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
    os.remove(src)


def process_directory(dir_path: str, workers: int = 2) -> None:
    """
    Обрабатывает все видео **только в данной папке** (не заглядывая в под‑папки):
      1. Создаёт dir_path/FINAL
      2. Конвертирует видео параллельно
      3. Перемещает результаты из FINAL в dir_path
      4. Удаляет FINAL, если она опустела
    """
    print(f"🛠  Обрабатываем: {dir_path}")

    video_files = [
        os.path.join(dir_path, f)
        for f in os.listdir(dir_path)
        if f.lower().endswith(VIDEO_EXT) and os.path.isfile(os.path.join(dir_path, f))
    ]
    if not video_files:
        print("   ⏭  Видео не найдено, пропускаем")
        return

    final_dir = os.path.join(dir_path, "FINAL")
    os.makedirs(final_dir, exist_ok=True)

    # Параллельный рендеринг
    with ThreadPoolExecutor(max_workers=workers) as ex:
        futures = []
        for v in video_files:
            out_name = os.path.splitext(os.path.basename(v))[0] + ".mp4"
            out_path = os.path.join(final_dir, out_name)
            futures.append(ex.submit(render_video, v, out_path))

        for i, fut in enumerate(as_completed(futures), 1):
            try:
                fut.result()
                print(f"   ✅ {i}/{len(futures)}")
            except Exception as e:
                print(f"   🔥 Ошибка: {e}")

    # Перемещаем конвертированные файлы в родительскую папку
    for f in os.listdir(final_dir):
        src = os.path.join(final_dir, f)
        dst = os.path.join(dir_path, f)

        # Если файл уже есть, добавляем суффикс _n
        if os.path.exists(dst):
            base, ext = os.path.splitext(f)
            n = 1
            while True:
                new_dst = os.path.join(dir_path, f"{base}_{n}{ext}")
                if not os.path.exists(new_dst):
                    dst = new_dst
                    break
                n += 1
        shutil.move(src, dst)

    # Удаляем FINAL, если пустая
    try:
        if not os.listdir(final_dir):
            os.rmdir(final_dir)
            print("   🗑  Пустая FINAL удалена")
    except OSError:
        pass


def walk_and_process(root_folder: str) -> None:
    """
    Проходит по всем под‑папкам root_folder (рекурсивно) и обрабатывает каждую.
    Сам root_folder не трогаем.
    """
    for current_dir, _, _ in os.walk(root_folder):
        # пропускаем сам корень и любые папки FINAL
        if current_dir == root_folder or os.path.basename(current_dir).upper() == "FINAL":
            continue
        process_directory(current_dir)


if __name__ == "__main__":
    if len(sys.argv) < 2:
        print("Укажите путь к папке в аргументах!")
        sys.exit(1)

    root = sys.argv[1]
    if not os.path.isdir(root):
        print(f"❌ Папка {root} не найдена или не является директорией!")
        sys.exit(1)

    walk_and_process(root)
    print("🎉 Обработка завершена!")

このQ&Aを共有