常用小工具记录

自用记录文

图片横竖区分

import os
import shutil
import time
import random
from PIL import Image
Image.MAX_IMAGE_PIXELS = None
def is_horizontal(image_path):
    with Image.open(image_path) as img:
        width, height = img.size
        return width > height

def move_and_rename_images(source_dir, horizontal_dir, vertical_dir):
    for root, dirs, files in os.walk(source_dir):
        for filename in files:
            if filename.endswith(('.jpg', '.jpeg', '.png')):
                file_path = os.path.join(root, filename)

                if is_horizontal(file_path):
                    destination_dir = horizontal_dir
                else:
                    destination_dir = vertical_dir

                timestamp = int(time.time())
                random_suffix = str(random.randint(1000, 9999))

                new_filename = f"{timestamp}_{random_suffix}.jpg"
                destination_path = os.path.join(destination_dir, new_filename)

                shutil.move(file_path, destination_path)
                print(f"移动 {filename} 到 {destination_path}")

if __name__ == "__main__":
    source_directory = "/hdd/tmp"
    horizontal_directory = "/hdd/p/h"
    vertical_directory = "/hdd/p/v"

    os.makedirs(horizontal_directory, exist_ok=True)
    os.makedirs(vertical_directory, exist_ok=True)

    move_and_rename_images(source_directory, horizontal_directory, vertical_directory)

图片质量压缩

import os
from PIL import Image
from concurrent.futures import ThreadPoolExecutor
Image.MAX_IMAGE_PIXELS = None
def compress_image(file_path, quality=85):
    try:
        with Image.open(file_path) as img:
            img.save(file_path, quality=quality)
        print(f"压缩 {file_path}")
    except Exception as e:
        print(f"处理 {file_path} 时发生错误: {e}")

def compress_images(directory, quality=85, num_threads=4):
    with ThreadPoolExecutor(max_workers=num_threads) as executor:
        for root, dirs, files in os.walk(directory):
            for filename in files:
                if filename.endswith(('.jpg', '.jpeg', '.png')):
                    file_path = os.path.join(root, filename)
                    executor.submit(compress_image, file_path, quality)

if __name__ == "__main__":
    images_directory = "/root/p/p"

    # 压缩图像，默认压缩质量为85，使用4个线程
    compress_images(images_directory, quality=85, num_threads=16)

TG瑟瑟视频下载(未完善)

import os
import logging
import time
from telethon import TelegramClient, sync
from telethon.tl.functions.messages import GetHistoryRequest 
from telethon.tl.types import PeerChannel

# Telegram API配置
api_id = '2xxxx50'  
api_hash = 'f8daxxxxxxxxf20a28fe8'
phone_number = '+6xxxxxxx0'  

# 频道配置
channel_id = -10xxxxxx99

# Client配置  
client = TelegramClient('bot', api_id, api_hash)

def main():
    client.connect()

    if not client.is_user_authorized():
        client.send_code_request(phone_number)
        client.sign_in(phone=phone_number, code=input('Enter code: '))

    os.makedirs('downloads', exist_ok=True)

    messages = client(GetHistoryRequest(
        peer=PeerChannel(channel_id=channel_id), 
        limit=0,
        offset_date=None,
        offset_id=0,
        max_id=0,
        min_id=0,
        add_offset=0,
        hash=0
    ))

    for msg in messages.messages:
        time.sleep(1)
        if msg.document:
            print(msg.document)
            print("AAAAAAAAAAAAAAAAAAAAAAAAAAA")
            print(msg.document.attributes[1].file_name)
            print("AAAAAAAAAAAAAAAAAAAAAAAAAAA")
            if msg.message:
                print(msg)
                print("AAAAAAAAAAAAAAAAAAAAAAAAAAA")
                print(msg.message)
            client.download_media(msg, f'downloads/{msg.id}')
        time.sleep(60)

    client.disconnect()

with client:
    client.loop.run_until_complete(main())