掘金 后端 ( ) • 2024-04-26 14:07

1、问题&困扰

下边这个问题困扰我有段时间了:

因为文章是在掘金,知乎这类技术网站发表的,在发表时文章中的图片都是上传到了掘金 or 知乎的图床上了,不管是用的哪家哪个大厂的图床,我总是担心哪一天网站崩掉或者其他什么不可控的事情发生,那我博客中的所有的图床文件(大部分都是图片)那岂不是都访问不到了? 像这样: image.png 这就尴尬了,我一般写文章很重视图解,没有图片的文章那没法串联起来,所以我总是想的把文章复制到本机上,但是复制到本地上的文章中的图片不用说也仍然是访问的三方的图床呀,如下: image.png

2、我的目标

  1. 将这类三方图床文件下载到本地
  2. 将markdown中的图传连接替换为指向本机的图片路径,如下:image.png

我不可能一个个手动去下载替换,那样士可忍孰不可忍。所以我决定使用python批量替换,基本逻辑是:

  1. 找到替换前的markdown文章集合
  2. 依次读取文章内容,并使用正则找到图床文件,并下载到本地(会在当前目录创建个专门放图片的文件夹)
  3. 替换markdown中的图床链接为本地的图片路径。
  4. 另外我还生成了word文件,以作备份。

3、使用python实现

读取指定文件夹中的文章集合,并解析,正则匹配与下载替换:


from functools import partial

from util.download_util import download_pic
from util.file_util import is_dir_existed, search_all_file, read_file_text_content, write_text_to_file
from util.logger_util import default_logger
import os
import re
import time

logger = default_logger()
origin_md_dir = os.path.join("/Users/hzz/Documents/黄壮壮_写作/已发表的博客_掘金存储图片/")  # 原始md文件目录
local_md_dir = os.path.join("/Users/hzz/Documents/黄壮壮_写作/已发表博客_本地存储图片/")  # 本地md文件目录
local_doc_dir = os.path.join("/Users/hzz/Documents/黄壮壮_写作/已发表博客_doc/")  # 本地doc文件目录
server_md_dir = os.path.join(os.getcwd(), "server_md")  # 转换成自己的图片服务器md文件目录
pic_match_pattern = re.compile(r'(]: |()+(http.*?.(png|PNG|jpg|JPG|gif|GIF|svg|SVG|webp|awebp|image))??()?)',
                               re.M)  # 匹配图片的正则
order_set = {i for i in range(1, 500000)}  # 避免图片名重复后缀
generate_server_md = False  # 是否在本地化后生成为自己图床的md文件
generate_doc = True  # 是否在本地化后生成doc文件


# 检索md文件
def retrieve_md(file_dir):
    logger.info("检索路径 → %s" % file_dir)
    md_file_list = search_all_file(file_dir, target_suffix_tuple=('md', "MD"))
    file_name = input('请输入指定的文件: ')

    if len(file_name)!=0:
        filtered_files = []
        for filename in md_file_list:
            if  file_name in filename:
                # 这里可以添加更多逻辑处理
                filtered_files.append(filename)
        md_file_list=filtered_files

    if len(md_file_list) == 0:
        logger.info("未检测到Markdown文件,请检查后重试!")
        exit(-1)
    else:
        logger.info("检测到Markdown文件 → %d个" % len(md_file_list))
        logger.info("=" * 64)
        for pos, md_file in enumerate(md_file_list):
            logger.info("%d、%s" % (pos + 1, md_file))
        logger.info("=" * 64)
        logger.info("执行批处理操作")
        process_md(md_file_list)


# 处理文件列表
def process_md(file_list):
    for file in file_list:
        to_local_md(file)


# 转换成本地MD
def to_local_md(md_file):
    logger.info("处理文件:【%s】" % md_file)
    # 获取md文件名
    md_file_name = os.path.basename(md_file)
    # 生成md文件的目录、图片目录,doc目录
    new_md_dir = os.path.join(local_md_dir, md_file_name[:-3])
    new_picture_dir = os.path.join(new_md_dir, "images")
    is_dir_existed(new_md_dir)
    is_dir_existed(new_picture_dir)
    # 生成md文件路径
    new_md_file_path = os.path.join(new_md_dir, md_file_name)
    new_doc_file_path = os.path.join(local_doc_dir, md_file_name[:-3].replace("知乎盐选 ", "") + ".docx")
    # 读取md文件内容
    old_content = read_file_text_content(md_file)
    # 替换原内容
    new_content = pic_match_pattern.sub(partial(pic_to_local, pic_save_dir=new_picture_dir), old_content)
    # 生成新的md文件
    write_text_to_file(new_content, new_md_file_path)
    logger.info("新md文件已生成 → {}".format(new_md_file_path))
    # 生成新的doc文件
    if generate_doc:
        os.chdir(new_md_dir)
        logger.info("新doc文件已生成 → {}".format(new_doc_file_path))
    logger.info("=" * 64)
    if generate_server_md:
        to_server_md(new_md_file_path)


# 远程图片转换为本地图片
def pic_to_local(match_result, pic_save_dir):
    logger.info("替换前的图片路径:{}".format(match_result[2]))
    # 生成新的图片名
    img_file_name = "{}_{}.{}".format(int(round(time.time())), order_set.pop(), "png")# match_result[3]
    # 拼接图片相对路径(Markdown用到的)
    relative_path = 'images/{}'.format(img_file_name)
    # 拼接图片绝对路径,下载到本地
    absolute_path = os.path.join(pic_save_dir, img_file_name)
    logger.info("替换后的图片路径:{}".format(relative_path))
    # 下载图片
    download_pic(absolute_path, match_result[2])
    # 还需要拼接前后括号()
    return "{}{}{}".format(match_result[1], relative_path, match_result[4])

# 转换成自己的图片服务器md文件
def to_server_md(md_file):
    pass


if __name__ == '__main__':
    is_dir_existed(origin_md_dir)
    is_dir_existed(local_md_dir, is_recreate=True)
    is_dir_existed(local_doc_dir, is_recreate=True)
    is_dir_existed(server_md_dir, is_recreate=True)
    # loop = asyncio.get_event_loop()
    retrieve_md(origin_md_dir)

下载图片 download_util 工具类:



import os
import time

import requests


from util.logger_util import default_logger

logger = default_logger()

default_headers = {
    'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
    'Accept-Encoding': 'gzip, deflate, br',
    'Sec-Ch-Ua': '"Not_A Brand";v="8", "Chromium";v="120", "Google Chrome";v="120"',
    'Sec-Ch-Ua-Mobile': '?0',
    'Sec-Ch-Ua-Platform': '"macOS"',
    'Sec-Fetch-Dest': 'document',
    'Sec-Fetch-Mode': 'navigate',
    'Sec-Fetch-User': '?0',
    'Upgrade-Insecure-Requests': '1'
}


def download_pic(pic_path, url, headers=None):
    try:
        if headers is None:
            headers = default_headers
        if url.startswith("http") | url.startswith("https"):
            if os.path.exists(pic_path):
                logger.info("图片已存在,跳过下载:%s" % pic_path)
            else:
                res1 = requests.get(url, headers=headers,verify=False)

                # res1.encoding = "utf8"
                with open(pic_path, "wb+") as f:
                    f.write(res1.content)
        else:
            logger.info("图片链接格式不正确:%s - %s" % (pic_path, url))
        time.sleep(1)
    except Exception as e:
        logger.info("下载异常:{}\n{}".format(url, e))

logger 工具类:


import logging


# 默认日志工具
def default_logger():
    custom_logger = logging.getLogger("CpPythonBox")
    if not custom_logger.hasHandlers():
        handler = logging.StreamHandler()
        handler.setFormatter(logging.Formatter(
            fmt='%(asctime)s %(process)d:%(processName)s- %(levelname)s === %(message)s',
            datefmt="%Y-%m-%d %H:%M:%S %p"))
        custom_logger.addHandler(handler)
        custom_logger.setLevel(logging.INFO)
    return custom_logger

file_util 工具类:


def search_all_file(file_dir=os.getcwd(), target_suffix_tuple=()):
    """ 递归遍历文夹与子文件夹中的特定后缀文件

    Args:
        file_dir (str): 文件目录
        target_suffix_tuple (Tuple(Str)): 文件目录

    Returns:
        list : 文件路径列表
    """
    file_list = []
    # 切换到目录下
    os.chdir(file_dir)
    file_name_list = os.listdir(os.curdir)
    for file_name in file_name_list:
        # 获取文件绝对路径
        file_path = "{}{}{}".format(os.getcwd(), os.path.sep, file_name)
        # 判断是否为目录,是往下递归
        if os.path.isdir(file_path):
            # print("[-]", file_path)
            file_list.extend(search_all_file(file_path, target_suffix_tuple))
            os.chdir(os.pardir)
        elif target_suffix_tuple is not None and file_name.endswith(target_suffix_tuple):
            # print("[!]", file_path)
            file_list.append(file_path)
        else:
            pass
            # print("[+]", file_path)
    return file_list



def write_text_to_file(content, file_path, mode="w+"):
    """ 将文字写入到文件中

    Args:
        content (str): 文字内容
        file_path (str): 写入文件路径
        mode (str): 文件写入模式,w写入、a追加、+可读写

    Returns:
        None
    """
    with lock:
        try:
            with open(file_path, mode, encoding='utf-8') as f:
                f.write(content + "\n", )
        except OSError as reason:
            print(str(reason))



def read_file_text_content(file_path):
    """ 以文本形式读取文件内容

    Args:
        file_path (str): 文件路径

    Returns:
        str: 文件内容
    """
    if not os.path.exists(file_path):
        return None
    else:
        with open(file_path, 'r+', encoding='utf-8') as f:
            return f.read()


def is_dir_existed(file_path, mkdir=True, is_recreate=False):
    """ 判断目录是否存在,不存在则创建

    Args:
        file_path (str): 文件路径
        mkdir (bool): 不存在是否新建
        is_recreate (bool): 存在是否删掉重建

    Returns:
        默认返回None,如果mkdir为False返回文件是否存在
    """
    if mkdir:
        if not os.path.exists(file_path):
            os.makedirs(file_path)
        else:
            if is_recreate:
                delete_file(file_path)
                if not os.path.exists(file_path):
                    os.makedirs(file_path)
    else:
        return os.path.exists(file_path)

4、效果

执行过程: image.png image.png 本机图片: image.png image.png markdown中展示: image.png

所有文章都替换完成: image.png

同样的,知乎等其他网站上的文章只要是markdown格式的 图床文件能请求成功的 都可以使用此工具实现下载到本地并替换到文章中去实现本机存储图片永久保留。如果帮助到了你,麻烦点赞收藏评论三连!😋😋😋😋😋😋😋😋😋😋😋😋😋😋😋😋😋😋😋😋😋😋