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

1. 为什么需要自定义Markdown图片适配器写技术文档最头疼的事情之一就是处理Markdown里的图片引用。我遇到过无数次这样的场景在本地用Typora写完文档图片都存在./images文件夹下等到要发布到公司Wiki或者博客平台时发现所有图片路径都要重新调整。更崩溃的是不同平台对图片存储的要求各不相同——有的用OSS有的用自建文件服务还有的要求图片必须带CDN前缀。这时候imarkdown的适配器模式就派上用场了。它的核心设计理念很像手机充电器Type-C接口是统一的就像Markdown的语法但不同厂商的充电协议各种图床API可以自由替换。通过继承MdAdapter基类我们就能实现自己的充电协议。举个例子我们团队内部用MinIO搭建了私有文件存储。原本需要手动把图片拖到管理后台上传再复制URL回填到Markdown。现在只需要写个20行的适配器class MinIOAdapter(BaseMdAdapter): def upload(self, key, file): response requests.put( fhttps://minio.example.com/{key}, datafile, headers{Authorization: Bearer xxxx} ) return response.json()[object_url]这个设计最妙的地方在于转换逻辑和存储逻辑完全解耦。就像你不需要知道手机充电器用的是PD协议还是QC协议MdImageConverter也只需要关心把A格式的URL转成B格式这件事具体怎么上传、怎么生成URL都由适配器处理。2. 适配器开发实战从零实现七牛云支持虽然官方已经提供了阿里云OSS适配器但国内很多团队在用七牛云。下面我带大家完整实现一个七牛云适配器你会看到扩展新图床有多简单。2.1 准备工作首先安装七牛SDKpip install qiniu然后去七牛控制台拿到这几个关键参数Access KeySecret Key存储空间名称Bucket域名比如xxx.clouddn.com2.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}关键点说明name属性是适配器标识符转换器会用到upload方法处理文件二进制流的上传逻辑get_replaced_url生成最终展示在Markdown中的URL2.3 实际使用示例假设我们要把本地docs目录下的Markdown文件全部迁移到七牛云from imarkdown import MdImageConverter, MdFolder from qiniu_adapter import QiniuAdapter converter MdImageConverter( adapterQiniuAdapter( access_keyyour_ak, secret_keyyour_sk, bucket_nametech-docs, domainhttps://img.example.com ) ) converter.convert( MdFolder(docs, image_typelocal), output_directoryconverted_docs )转换完成后原本的会自动变成并且图片已经上传到七牛云。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, formatJPEG, quality85) 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: fBearer {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 fpending://{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( adapterself.name, statussuccess ).inc() return result except Exception as e: UPLOAD_COUNTER.labels( adapterself.name, statusfailed ).inc() raise