【imarkdown】如何通过自定义适配器扩展你的Markdown图片管理能力

1. 为什么需要自定义Markdown图片适配器

写技术文档最头疼的事情之一，就是处理Markdown里的图片引用。我遇到过无数次这样的场景：在本地用Typora写完文档，图片都存在./images文件夹下，等到要发布到公司Wiki或者博客平台时，发现所有图片路径都要重新调整。更崩溃的是不同平台对图片存储的要求各不相同——有的用OSS，有的用自建文件服务，还有的要求图片必须带CDN前缀。

这时候imarkdown的适配器模式就派上用场了。它的核心设计理念很像手机充电器：Type-C接口是统一的（就像Markdown的![alt](url)语法），但不同厂商的充电协议（各种图床API）可以自由替换。通过继承MdAdapter基类，我们就能实现自己的"充电协议"。

举个例子，我们团队内部用MinIO搭建了私有文件存储。原本需要手动把图片拖到管理后台上传，再复制URL回填到Markdown。现在只需要写个20行的适配器：

class MinIOAdapter(BaseMdAdapter): def upload(self, key, file): response = requests.put( f"https://minio.example.com/{key}", data=file, headers={"Authorization": "Bearer xxxx"} ) return response.json()["object_url"]

这个设计最妙的地方在于，转换逻辑和存储逻辑完全解耦。就像你不需要知道手机充电器用的是PD协议还是QC协议，MdImageConverter也只需要关心"把A格式的URL转成B格式"这件事，具体怎么上传、怎么生成URL都由适配器处理。

2. 适配器开发实战：从零实现七牛云支持

虽然官方已经提供了阿里云OSS适配器，但国内很多团队在用七牛云。下面我带大家完整实现一个七牛云适配器，你会看到扩展新图床有多简单。

2.1 准备工作

首先安装七牛SDK：

pip install qiniu

然后去七牛控制台拿到这几个关键参数：

• Access Key
• Secret Key
• 存储空间名称（Bucket）
• 域名（比如xxx.clouddn.com）

2.2 核心代码实现

新建qiniu_adapter.py文件：

from qiniu import Auth, put_file from imarkdown import BaseMdAdapter import os class QiniuAdapter(BaseMdAdapter): name = "qiniu" def __init__(self, access_key, secret_key, bucket_name, domain): self.auth = Auth(access_key, secret_key) self.bucket = bucket_name self.domain = domain.rstrip('/') def upload(self, key, file): # 生成上传token token = self.auth.upload_token(self.bucket, key) # 七牛SDK要求文件保存到临时目录 temp_path = f"/tmp/{key}" with open(temp_path, 'wb') as f: f.write(file.read()) # 执行上传 ret, _ = put_file(token, key, temp_path) os.remove(temp_path) return ret['key'] def get_replaced_url(self, key): return f"{self.domain}/{key}"

关键点说明：

1. name属性是适配器标识符，转换器会用到
2. upload方法处理文件二进制流的上传逻辑
3. get_replaced_url生成最终展示在Markdown中的URL

2.3 实际使用示例

假设我们要把本地docs目录下的Markdown文件全部迁移到七牛云：

from imarkdown import MdImageConverter, MdFolder from qiniu_adapter import QiniuAdapter converter = MdImageConverter( adapter=QiniuAdapter( access_key="your_ak", secret_key="your_sk", bucket_name="tech-docs", domain="https://img.example.com" ) ) converter.convert( MdFolder("docs", image_type="local"), output_directory="converted_docs" )

转换完成后，原本的![流程图](./images/flow.png)会自动变成![流程图](https://img.example.com/flow.png)，并且图片已经上传到七牛云。

3. 高级技巧：适配器组合与预处理

实际项目中我们经常需要处理复杂场景，比如：

• 上传前压缩图片
• 根据文件类型选择不同图床
• 添加水印等后处理

3.1 装饰器模式增强适配器

我们可以用Python的装饰器给适配器添加额外功能。下面实现一个自动压缩图片的装饰器：

from PIL import Image import io def compress_image(adapter_cls): class WrappedAdapter(adapter_cls): def upload(self, key, file): # 只处理图片文件 if key.split('.')[-1].lower() in ('jpg', 'jpeg', 'png'): img = Image.open(file) # 长边不超过2000px if max(img.size) > 2000: ratio = 2000 / max(img.size) new_size = (int(img.size[0]*ratio), int(img.size[1]*ratio)) img = img.resize(new_size, Image.LANCZOS) # 转换为JPEG格式 output = io.BytesIO() img.save(output, format='JPEG', quality=85) output.seek(0) return super().upload(key, output) return super().upload(key, file) return WrappedAdapter

使用时只需要装饰原有适配器：

@compress_image class QiniuAdapter(BaseMdAdapter): ...

3.2 多适配器路由

对于混合使用多个图床的场景，可以设计路由适配器：

class RouterAdapter(BaseMdAdapter): def __init__(self): self.oss_adapter = AliyunAdapter(...) self.qiniu_adapter = QiniuAdapter(...) def upload(self, key, file): # 设计你的路由逻辑，例如： if key.startswith("screenshots/"): return self.oss_adapter.upload(key, file) else: return self.qiniu_adapter.upload(key, file) def get_replaced_url(self, key): if key.startswith("screenshots/"): return self.oss_adapter.get_replaced_url(key) else: return self.qiniu_adapter.get_replaced_url(key)

4. 企业级实践：私有化部署方案

很多公司出于安全考虑会自建文件存储服务。我曾帮一个金融客户实现过对接内部系统的适配器，这里分享几个关键经验：

4.1 认证与安全

企业内部系统通常需要复杂的认证，比如JWT令牌刷新：

class InternalStorageAdapter(BaseMdAdapter): def __init__(self): self.token = self._refresh_token() def _refresh_token(self): resp = requests.post( "https://auth.internal.com/token", json={"app_id": "markdown", "secret": "xxx"} ) return resp.json()["access_token"] def upload(self, key, file): headers = { "Authorization": f"Bearer {self.token}", "X-File-Meta": json.dumps({ "uploader": "markdown-system", "expire_days": 365 }) } ...

4.2 分布式文件处理

当处理大量文件时，可以考虑使用Celery等任务队列：

from celery import Celery app = Celery('markdown') @app.task def async_upload(adapter_config, key, file_path): adapter = load_adapter(adapter_config) with open(file_path, 'rb') as f: adapter.upload(key, f) class BatchAdapter(BaseMdAdapter): def upload(self, key, file): temp_path = f"/tmp/{key}" with open(temp_path, 'wb') as f: f.write(file.read()) async_upload.delay( self._export_config(), key, temp_path ) return f"pending://{key}"

4.3 监控与日志

生产环境需要添加完善的监控：

from prometheus_client import Counter UPLOAD_COUNTER = Counter( 'markdown_upload_total', 'Total file uploads', ['adapter', 'status'] ) class MonitoredAdapter(BaseMdAdapter): def upload(self, key, file): try: result = super().upload(key, file) UPLOAD_COUNTER.labels( adapter=self.name, status='success' ).inc() return result except Exception as e: UPLOAD_COUNTER.labels( adapter=self.name, status='failed' ).inc() raise