253 lines
8.9 KiB
Python
253 lines
8.9 KiB
Python
import os
|
||
import re
|
||
import requests
|
||
from requests.auth import HTTPBasicAuth
|
||
import tkinter as tk
|
||
from tkinter import filedialog, messagebox
|
||
import time
|
||
import tempfile
|
||
import uuid
|
||
import shutil
|
||
|
||
# 配置信息
|
||
local_image_folder = 'downloaded_images'
|
||
lsky_pro_url = 'http://192.168.107.248:18089/api/v1'
|
||
upload_endpoint = '/upload'
|
||
token = '1|QJP2YEr9GIN52VBgmm5hCqV5DwBSvJLUKjnwcKB8'
|
||
|
||
# 确保本地图片文件夹存在
|
||
#os.makedirs(local_image_folder, exist_ok=True)
|
||
def get_image_folder(markdown_path):
|
||
# 获取markdown文件所在目录
|
||
markdown_dir = os.path.dirname(markdown_path)
|
||
# 在markdown文件同目录下创建images文件夹
|
||
image_folder = os.path.join(markdown_dir, 'images')
|
||
return image_folder
|
||
def choose_input_file():
|
||
file_path = filedialog.askopenfilename(
|
||
title="选择Markdown文件",
|
||
filetypes=(("Markdown files", "*.md"), ("All files", "*.*"))
|
||
)
|
||
if file_path:
|
||
input_file_entry.delete(0, tk.END)
|
||
input_file_entry.insert(0, file_path)
|
||
|
||
def clean_filename(url):
|
||
try:
|
||
# 提取fdId参数
|
||
fd_id = re.search(r'fdId=([0-9a-f]+)', url)
|
||
if fd_id:
|
||
return f"{fd_id.group(1)}.jpg"
|
||
|
||
# 如果没有fdId，使用时间戳
|
||
return f"image_{int(time.time())}.jpg"
|
||
except:
|
||
return f"image_{int(time.time())}.jpg"
|
||
|
||
def download_images():
|
||
markdown_file_path = input_file_entry.get()
|
||
if not markdown_file_path or not os.path.isfile(markdown_file_path):
|
||
messagebox.showwarning("警告", "请选择有效的Markdown文件。")
|
||
return
|
||
# 根据markdown文件位置设置图片文件夹
|
||
global local_image_folder
|
||
local_image_folder = get_image_folder(markdown_file_path)
|
||
os.makedirs(local_image_folder, exist_ok=True)
|
||
with open(markdown_file_path, 'r', encoding='utf-8') as file:
|
||
markdown_content = file.read()
|
||
|
||
image_urls = re.findall(r'!\[.*?\]\((.*?)\)', markdown_content)
|
||
|
||
success_count = 0
|
||
url_mapping = {} # 存储URL到本地路径的映射
|
||
|
||
for url in image_urls:
|
||
try:
|
||
response = requests.get(url)
|
||
if response.status_code == 200:
|
||
image_name = clean_filename(url)
|
||
image_path = os.path.join(local_image_folder, image_name)
|
||
|
||
# 保存图片
|
||
with open(image_path, 'wb') as img_file:
|
||
img_file.write(response.content)
|
||
|
||
# 使用相对路径
|
||
relative_path = os.path.join('.', local_image_folder, image_name)
|
||
relative_path = relative_path.replace('\\', '/') # 统一使用正斜杠
|
||
|
||
url_mapping[url] = relative_path
|
||
success_count += 1
|
||
|
||
except Exception as e:
|
||
print(f"下载失败 {url}: {str(e)}")
|
||
continue
|
||
|
||
# 更新Markdown内容
|
||
if url_mapping:
|
||
new_content = markdown_content
|
||
for old_url, new_path in url_mapping.items():
|
||
new_content = new_content.replace(old_url, new_path)
|
||
|
||
# 保存更新后的Markdown文件
|
||
with open(markdown_file_path, 'w', encoding='utf-8') as file:
|
||
file.write(new_content)
|
||
|
||
messagebox.showinfo("完成", f"成功下载 {success_count} 张图片并更新Markdown文件")
|
||
else:
|
||
messagebox.showwarning("警告", "没有图片被下载")
|
||
|
||
def upload_image_to_lsky_pro(image_path):
|
||
"""上传图片到Lsky Pro图床"""
|
||
with open(image_path, 'rb') as img_file:
|
||
files = {'file': img_file}
|
||
headers = {
|
||
'Authorization': f'Bearer {token}',
|
||
'Accept': 'application/json'
|
||
}
|
||
response = requests.post(f'{lsky_pro_url}{upload_endpoint}', files=files, headers=headers)
|
||
if response.status_code == 200:
|
||
try:
|
||
data = response.json()
|
||
if data['status']:
|
||
return data['data']['links']['url']
|
||
except:
|
||
pass
|
||
return None
|
||
|
||
def upload_images():
|
||
markdown_file_path = input_file_entry.get()
|
||
if not markdown_file_path or not os.path.isfile(markdown_file_path):
|
||
messagebox.showwarning("警告", "请选择有效的Markdown文件。")
|
||
return
|
||
|
||
with open(markdown_file_path, 'r', encoding='utf-8') as file:
|
||
markdown_content = file.read()
|
||
|
||
# 提取所有图片URL
|
||
image_urls = re.findall(r'!\[.*?\]\((.*?)\)', markdown_content)
|
||
url_mapping = {} # 存储原始URL到新URL的映射
|
||
|
||
for url in image_urls:
|
||
try:
|
||
# 下载图片
|
||
|
||
|
||
# 上传到图床
|
||
new_url = upload_image_to_lsky_pro(url)
|
||
if new_url:
|
||
url_mapping[url] = new_url
|
||
print(f"处理成功: {url} -> {new_url}")
|
||
else:
|
||
print(f"上传失败: {url}")
|
||
|
||
except Exception as e:
|
||
print(f"处理出错 {url}: {str(e)}")
|
||
continue
|
||
|
||
# 更新Markdown内容
|
||
if url_mapping:
|
||
new_content = markdown_content
|
||
for old_url, new_url in url_mapping.items():
|
||
new_content = new_content.replace(old_url, new_url)
|
||
|
||
# 保存更新后的文件
|
||
with open(markdown_file_path, 'w', encoding='utf-8') as file:
|
||
file.write(new_content)
|
||
|
||
messagebox.showinfo("完成", f"成功处理 {len(url_mapping)} 张图片")
|
||
else:
|
||
messagebox.showwarning("警告", "没有图片被处理")
|
||
def download_and_upload_images():
|
||
markdown_file_path = input_file_entry.get()
|
||
if not markdown_file_path:
|
||
messagebox.showerror("错误", "请选择Markdown文件")
|
||
return
|
||
|
||
try:
|
||
# 读取Markdown文件
|
||
with open(markdown_file_path, 'r', encoding='utf-8') as file:
|
||
markdown_content = file.read()
|
||
|
||
# 提取所有图片URL
|
||
image_urls = re.findall(r'!\[.*?\]\((.*?)\)', markdown_content)
|
||
if not image_urls:
|
||
messagebox.showinfo("提示", "文件中未找到图片链接")
|
||
return
|
||
|
||
url_mapping = {}
|
||
temp_dir = tempfile.mkdtemp()
|
||
|
||
for img_url in image_urls:
|
||
try:
|
||
# 下载图片
|
||
response = requests.get(img_url, timeout=10)
|
||
if response.status_code == 200:
|
||
# 生成临时文件名
|
||
file_extension = os.path.splitext(img_url.split('/')[-1])[-1]
|
||
if not file_extension:
|
||
file_extension = '.png'
|
||
temp_file = os.path.join(temp_dir, f"temp_{uuid.uuid4()}{file_extension}")
|
||
|
||
# 保存图片
|
||
with open(temp_file, 'wb') as f:
|
||
f.write(response.content)
|
||
|
||
# 使用现有函数上传到lsky
|
||
new_url = upload_image_to_lsky_pro(temp_file)
|
||
if new_url:
|
||
url_mapping[img_url] = new_url
|
||
print(f"成功上传: {img_url} -> {new_url}")
|
||
else:
|
||
print(f"上传失败: {img_url}")
|
||
|
||
except Exception as e:
|
||
print(f"处理图片失败: {img_url}, 错误: {str(e)}")
|
||
|
||
# 更新Markdown内容
|
||
if url_mapping:
|
||
new_content = markdown_content
|
||
for old_url, new_url in url_mapping.items():
|
||
new_content = new_content.replace(old_url, new_url)
|
||
|
||
# 保存更新后的文件
|
||
with open(markdown_file_path, 'w', encoding='utf-8') as file:
|
||
file.write(new_content)
|
||
|
||
messagebox.showinfo("完成", f"成功处理 {len(url_mapping)} 张图片")
|
||
else:
|
||
messagebox.showwarning("警告", "没有图片被处理")
|
||
|
||
# 清理临时目录
|
||
shutil.rmtree(temp_dir, ignore_errors=True)
|
||
|
||
except Exception as e:
|
||
messagebox.showerror("错误", f"处理过程中出错: {str(e)}")
|
||
|
||
|
||
|
||
|
||
# 创建主窗口
|
||
root = tk.Tk()
|
||
root.title("Markdown图片处理工具")
|
||
|
||
# 文件路径显示
|
||
tk.Label(root, text="Input File:").grid(row=0, column=0, padx=10, pady=10)
|
||
input_file_entry = tk.Entry(root, width=50)
|
||
input_file_entry.grid(row=0, column=1, padx=10, pady=10)
|
||
tk.Button(root, text="Browse", command=choose_input_file).grid(row=0, column=2, padx=10, pady=10)
|
||
|
||
# 下载图片按钮
|
||
download_button = tk.Button(root, text="下载图片", command=download_images)
|
||
download_button.grid(row=1, column=0, columnspan=3, pady=5)
|
||
|
||
# 上传图片按钮
|
||
upload_button = tk.Button(root, text="上传图片", command=upload_images)
|
||
upload_button.grid(row=2, column=0, columnspan=3, pady=5)
|
||
|
||
# 下载并上传图片按钮
|
||
download_upload_button = tk.Button(root, text="下载并上传图片", command=download_and_upload_images)
|
||
download_upload_button.grid(row=3, column=0, columnspan=3, pady=5)
|
||
|
||
# 运行主循环
|
||
root.mainloop() |