因为经常有发图-传图-公开的需求,比如说贴图到博客里,或者论坛水贴的时候贴图。
以前我的做法是,我并没有搭图床(懒),我只搭了一个类似于文件服务器的玩意儿用于直链分享文件和图片,这样我直接把链接贴到博客里就可以看了。但我要做的操作有:
- 截图,把本地的文件ssh或者scp到服务器上
- 把已上传的图片或文件再复制到我文件服务器对应的目录下
- 打开文件服务器的网页,把这张图片的链接粘贴出来
- 再把链接丢到我想传的地方
以上操作搞完不说头晕目眩,但凡手边的电脑没有服务器的ssh我就彻底歇b了 (╯‵□′)╯︵┴─┴
于是今天闲着就搓了个telegram bot,把图片直接丢给他,他就会自动帮我把文件传到正确的位置,然后把链接回给我,这样就方便多了~ 如图
其实实现原理也很简单,由于我的文件服务器和这只 telegram bot 是在同一个服务器上的,所以本质上也省去了再套一层ssh/scp的步骤,接收到之后直接把文件复制过去就好了。
# 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}...")
try:
# Create destination path
server_path = os.path.join(upload_dir, file_name)
# Download the file directly to the dufs directory
await file.download_to_drive(server_path)
# Create download link
download_url = await get_public_url(user_id, file_name)
# Log which user uploaded the file and where
logger.info(f"User {user_id} uploaded file: {file_name} to {upload_dir}")
await update.message.reply_text(f"File uploaded successfully!\nDownload link: {download_url}")
except Exception as e:
logger.error(f"Error: {str(e)}")
await update.message.reply_text(f"An error occurred: {str(e)}")
else:
await update.message.reply_text("Please send a file to upload.")
因为我有些文件是想当图床上传,而有些单纯是想分享文件,于是还写了个逻辑可以设定不同的文件夹路径,这样可以把不同的文件传到不同的目录下了:
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分钟不到 ⌇●﹏●⌇