# Process the file if update.message.document: file = await context.bot.get_file(update.message.document.file_id) file_name = update.message.document.file_name
# Get the appropriate upload directory upload_dir = await get_upload_directory(user_id)
# Ensure the directory exists os.makedirs(upload_dir, exist_ok=True)
# Notify user that upload is starting custom_path_info = f" to {user_paths[user_id]}" if user_id in user_paths else "" await update.message.reply_text(f"Downloading {file_name} and preparing to save{custom_path_info}...")
async def set_path(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: """Set a custom path for file uploads.""" if update.effective_user.id not in AUTHORIZED_USER_IDS: await update.message.reply_text("Sorry, you are not authorized to use this bot.") return
# Get the path from arguments if not context.args: await update.message.reply_text( "Please specify a path: /path your_custom_path\n" "For example: /path public" ) return
# Get the custom path and clean it custom_path = context.args[0].strip()
# Sanitize the path to prevent directory traversal if '..' in custom_path or custom_path.startswith('/'): await update.message.reply_text("Invalid path. Path cannot contain '..' or start with '/'.") return
# Store the custom path for this user user_id = update.effective_user.id user_paths[user_id] = custom_path
# Make sure the directory exists full_path = os.path.join(DUFS_BASE_DIRECTORY, custom_path) try: os.makedirs(full_path, exist_ok=True) await update.message.reply_text(f"Upload path set to: {custom_path}\nFiles will be uploaded to this directory.") except Exception as e: logger.error(f"Error creating directory: {str(e)}") await update.message.reply_text(f"Error setting path: {str(e)}")
偶尔写点这种能让自己变得更懒的小工具真的很爽 (/ω\) 顺便以上代码大部分都是由 Claude 3.7 生成的,整个小东西搓完只花了10分钟不到 ⌇●﹏●⌇