胶水代码：分析、策略与利弊

基于 O2O 平台政策汇总项目的真实解剖

---

一、什么是胶水代码

胶水代码 指主要作用是连接不同系统/API/库、自身几乎不产生核心业务价值的代码。典型特征：

特征	说明
调用外部服务	核心能力来自第三方 API，项目只负责传参数和收结果
格式转换	A → B → C 的数据搬运，没有注入新的业务价值
无领域建模	数据从头到尾以原始形态传递（dict → dict → dict）
模块间复制	同样的解析逻辑重复出现在多个文件中
意图隐匿	代码只表述”怎么做”，不表述”为什么这么做”

本项目的胶水成分解剖

整个数据链路：

assets/*.png ──→ OCR API ──→ 解析 ──→ CSV ──┬──→ Excel (openpyxl)
                                              ├──→ JSON  (json.dumps)
                                              └──→ HTML  (模板拼接)

胶水集中在：OCR 调用、CSV 读写、多格式导出。
非胶水在于：校验规则、审计逻辑、OCR 后处理的业务归一化。

量化结论：约 55% 胶水代码。

模块	胶水程度	定性
ocr_o2o.py	75%	核心动作是调腾讯云 API。非胶水部分：表头别名归一化、平台列消歧、产品白名单过滤
watcher_o2o.py	95%	标准 watchdog 模式，零业务逻辑
config.py	95%	通用 YAML + 环境变量加载
export_json.py	70%	CSV → JSON 映射，含商品分类和日期格式化
export_html.py	50%	前半段数据转换，后半段生成完整交互式 SPA 页面
export_excel.py	90%	CSV → Excel 加样式
verify.py	20%	纯领域逻辑：数值一致性校验、DG 联动检查
audit.py	15%	纯业务分析：跨平台最低价发现、风险提示
o2o_cli.py	65%	CLI 命令路由和流程编排

---

二、胶水代码的根源

胶水代码的产生不是技术能力问题，而是开发顺序问题：

典型的胶水项目生长过程：

需求1："帮我识别这张图片里的表格"
  → 查腾讯云文档，复制示例代码，调 API，拿到 JSON ✓

需求2："数据太多了，存到 CSV 方便编辑"
  → 把 JSON 字段映射成 CSV 列名，写文件 ✓

需求3："业务想看 Excel"
  → CSV 读到 openpyxl，加个边框和表头颜色 ✓

需求4："还需要一个网页版本方便分享"
  → 再读一遍 CSV，拼 HTML 字符串 ✓

需求5："再加个校验，免得数据出错"
  → 再读一遍 CSV，写校验逻辑 ✓

需求6："跨平台价格对比"
  → 再读一遍 CSV，写审计逻辑 ✓

问题在哪？ 每个步骤都直接把 CSV 当作数据模型。Dict[str, str] 在管道里从头流到尾，没有任何一个环节说清楚”这份数据本质上是什么”。结果是：

• _parse_price() 在 3 个文件里重复定义
• _get_category() 在 2 个文件里重复定义
• _extract_period() 在 2 个文件里重复定义
• 改一个字段名要改 5 个文件
• 新人要读完 6 个脚本才能理解”一行数据到底有哪些字段”

本质原因：没有在代码中显式定义领域模型。

---

三、避免胶水代码的五个策略

策略 1：先建模，后写管道

在任何 IO 代码之前，先定义数据是什么：

# o2o/domain.py —— 领域核心，零外部依赖

from dataclasses import dataclass
from typing import Optional
from enum import Enum

class Platform(Enum):
    JD = "京东到家"
    ELEME = "饿了么"
    MEITUAN = "美团"
    DOUYIN = "抖音"
    TMALL = "天猫"

class ProductCategory(Enum):
    IPHONE = "iPhone"
    IPAD = "iPad"
    MAC = "Mac"
    WATCH = "Watch"
    AIRPODS = "AirPods"
    ACCESSORY = "配件"
    BEATS = "Beats"
    OTHER = "其他"

@dataclass
class Offer:
    """一条 O2O 优惠记录——领域核心实体，不依赖任何外部库"""
    platform: Platform
    product: str
    spec: str
    alp: int                      # ALP 最低 SKU 价
    final_price: Optional[int]    # 到手价，None 表示免息分期无具体价格
    discount_amount: Optional[int] # 优惠金额（纯数字部分）
    is_installment: bool
    dealer_burden: Optional[int]  # 经销商承担
    platform_burden: Optional[int] # 平台承担
    limit_policy: str
    dg_related: bool
    start_date: str
    end_date: str
    note: str = ""

    # ── 业务规则内聚为方法 ──

    def discount_via_price(self) -> Optional[int]:
        """ALP - 到手价 = 预期优惠金额"""
        if self.alp and self.final_price:
            return self.alp - self.final_price
        return None

    def discount_is_consistent(self) -> bool:
        """优惠金额 是否与 ALP - 到手价 一致（容差 ±1）"""
        expected = self.discount_via_price()
        if expected is None or self.discount_amount is None:
            return True  # 无法校验则放过
        return abs(self.discount_amount - expected) <= 1

    def burden_total(self) -> Optional[int]:
        """经销商承担 + 平台承担"""
        if self.dealer_burden is not None and self.platform_burden is not None:
            return self.dealer_burden + self.platform_burden
        return None

    def burden_is_consistent(self) -> bool:
        """承担合计 是否与优惠金额一致"""
        total = self.burden_total()
        if total is None or self.discount_amount is None:
            return True
        return abs(total - self.discount_amount) <= 1

    def is_price_valid(self) -> bool:
        """到手价不应高于 ALP"""
        if self.alp and self.final_price:
            return self.final_price <= self.alp
        return True

    def category(self) -> ProductCategory:
        """根据产品名自动归类"""
        product = self.product
        if product.startswith("iPhone"):  return ProductCategory.IPHONE
        if product.startswith("iPad"):    return ProductCategory.IPAD
        if product.startswith("MacBook"): return ProductCategory.MAC
        if product.startswith("Apple Watch"): return ProductCategory.WATCH
        if product.startswith("AirPods"): return ProductCategory.AIRPODS
        if product.startswith("Beats"):   return ProductCategory.BEATS
        if any(k in product for k in ["充电器","保护壳","充电线","EarPods","AirTag"]):
            return ProductCategory.ACCESSORY
        return ProductCategory.OTHER

    @property
    def is_dg(self) -> bool:
        return self.dg_related

    @property
    def needs_limit_policy(self) -> bool:
        """DG 商品必须有对应的限购政策"""
        return not self.is_dg or bool(self.limit_policy)

对比改造前：verify.py 中散落的 _extract_number + abs(discount_num - expected_discount) > 1，全部收敛为 offer.discount_is_consistent() 一行调用。

---

策略 2：端口-适配器架构

核心规则：领域核心不 import 任何外部库。 所有文件读写、API 调用、格式导出都是外围适配器。

                   ┌──────────────────────┐
OCR API ──────────→│  o2o/adapters/        │
                    │    ocr_adapter.py     │──────┐
                    └──────────────────────┘      │
                                                   ↓
                    ┌──────────────────────┐  ┌──────────┐  ┌──────────────────────┐
手动录入 ──────────→│  o2o/adapters/        │→│ 核心领域  │→│  o2o/adapters/        │──────→ CSV
                    │    manual_adapter.py  │  │ Offer   │  │    csv_adapter.py     │
                    └──────────────────────┘  │ 规则    │  └──────────────────────┘
                                               │ Auditor │
                    ┌──────────────────────┐  │          │  ┌──────────────────────┐
                    │  o2o/adapters/        │←│          │←│  o2o/adapters/        │──────→ Excel
                    │    excel_adapter.py   │  └──────────┘  │    html_adapter.py    │
                    └──────────────────────┘                  └──────────────────────┘
                                                                         │
                                                                         ↓──────→ HTML

抽象的端口定义：

# o2o/ports.py

from abc import ABC, abstractmethod
from typing import List
from o2o.domain import Offer

class OfferSource(ABC):
    """数据来源的抽象——不关心是 OCR 还是人工录入"""
    @abstractmethod
    def fetch(self) -> List[Offer]:
        ...

class OfferSink(ABC):
    """数据输出的抽象——不关心导出格式"""
    @abstractmethod
    def write(self, offers: List[Offer]) -> None:
        ...

class OfferValidator(ABC):
    """校验规则的抽象"""
    @abstractmethod
    def validate(self, offer: Offer) -> List[str]:
        """返回错误信息列表，空列表表示通过"""
        ...

适配器负责脏活：

# o2o/adapters/ocr_adapter.py

class TencentOCROfferSource(OfferSource):
    """腾讯云表格识别 V3 适配器——所有 OCR 特有的脏活局限于此"""

    def fetch(self) -> List[Offer]:
        raw_tables = self._call_api()
        records = parse_table(raw_tables[0])
        return [self._to_offer(r) for r in records
                if self._is_valid_product(r)]

    def _to_offer(self, raw: dict) -> Offer:
        return Offer(
            platform=Platform(resolve_platform(raw.get("平台", ""))),
            product=raw.get("活动商品", "").strip(),
            spec=raw.get("商品规格", "").strip(),
            alp=parse_int(raw.get("ALP（最低sku价）", "")),
            final_price=parse_int_or_none(raw.get("到手价（最低sku价）", "")),
            discount_amount=parse_int_or_none(raw.get("优惠金额", "")),
            is_installment="免息" in str(raw.get("优惠金额", "")),
            dealer_burden=parse_int_or_none(raw.get("经销商承担", "")),
            platform_burden=parse_int_or_none(raw.get("平台承担", "")),
            limit_policy=raw.get("限政策", "").strip(),
            dg_related=raw.get("DG Related", "").strip().upper() == "Y",
            start_date=raw.get("开始时间", "").strip(),
            end_date=raw.get("结束时间", "").strip(),
        )

效果：

• 换 OCR 服务商（百度 OCR、阿里 OCR）→ 只改这一个适配器
• 测试时用 FakeOfferSource 返回固定数据，不调云 API
• 核心逻辑的单元测试不需要任何外部依赖

---

策略 3：管道声明式配置

改造前——命令式硬编码（当前 o2o_cli.py）：

# 现在的 build 命令：流水账式调用
def cmd_build(args):
    result = run_verify(csv_path)
    audit_report = run_audit(csv_path)
    out1 = export_excel(csv_path)
    out2 = export_json(csv_path)
    out3 = export_html(csv_path)

改造后——声明式管道：

# o2o/pipeline.py

from dataclasses import dataclass, field
from typing import List, Callable

@dataclass
class PipelineResult:
    step_name: str
    ok: bool
    message: str = ""
    data: any = None

@dataclass
class Pipeline:
    """声明式构建管道——步骤可配置、可测试、可复用"""
    steps: List[Callable] = field(default_factory=list)
    abort_on_error: bool = True

    def run(self, ctx: dict) -> List[PipelineResult]:
        results = []
        for step in self.steps:
            result = step(ctx)
            results.append(result)
            if self.abort_on_error and not result.ok:
                break
        return results

# 管道定义——组合哪些步骤、按什么顺序
BUILD_PIPELINE = Pipeline(steps=[
    load_offers_step,    # CSV → List[Offer]
    verify_step,         # 校验 → 不通过则终止
    audit_step,          # 审计 → 生成报告但不阻断
    export_excel_step,
    export_json_step,
    export_html_step,
])

---

策略 4：消除模块间复制

当前项目中最典型的胶水气味——完全相同的函数出现在两个文件中：

函数	出现位置
_parse_price	export_json.py:26 + export_html.py:26
_parse_discount	export_json.py:38 + export_html.py:36
_extract_platform_id	export_json.py:51 + export_html.py:48
_get_category	export_json.py:62 + export_html.py:53
_format_date	export_json.py:80 + export_html.py:71
_extract_period	export_json.py:90 + export_html.py:81
_format_period	export_json.py:105 + export_html.py:97

这些函数应该只存在于一个地方。如果采用了策略 1（Offer 数据类），导出器接收的是已经解析好的 Offer 对象列表，这些解析函数自然就消失了——它们会变成 Offer 的类方法或构造函数。

改造前（每个导出器各自解析）：

# export_json.py
def export(csv_path):
    rows = load_csv_data(csv_path)
    for row in rows:
        alp = _parse_price(row.get("ALP（最低sku价）", ""))  # 自己解析
        ...

# export_html.py
def _build_data(csv_path):
    rows = load_csv_data(csv_path)
    for row in rows:
        alp = _parse_price(row.get("ALP（最低sku价）", ""))  # 又解析一遍
        ...

改造后（统一解析，导出器只负责格式化）：

# o2o/adapters/csv_adapter.py —— 唯一的 CSV 解析入口
class CSVOfferSource(OfferSource):
    def fetch(self) -> List[Offer]:
        ...
        return [Offer(alp=parse_alp(row), ...) for row in raw_rows]

# export_json.py —— 不再解析，只格式化
def export(offers: List[Offer]):
    data = {"platforms": group_by_platform(offers)}
    json.dump(data, f)

# export_html.py —— 不再解析，只渲染
def export(offers: List[Offer]):
    data = {"platforms": group_by_platform(offers)}
    render_html(data)

---

策略 5：让代码说”为什么”

胶水写法	改善写法
if "免息" in s: return float('nan')	if self.is_installment: return None # 免息分期不以降价为优惠手段，不参与数值比较
m = re.search(r'(\d+)', s)	extract_first_integer(s) # OCR 可能混入中文描述，只取第一个连续数字
platform_cols.sort(); for col in platform_cols[1:]	# 表头中 col 0 是"平台名"，col 10+ 是"平台承担金额"，按位置消歧
limit_map.get(pname, "")	PLATFORM_LIMIT_POLICIES: dict[Platform, str] = {...}

胶水代码的通用气味：代码只表述了操作（how），没有表述意图（why）。 改善的方法是把意图写成命名或注释，让后续维护者不需要读完整段代码就能理解为什么这么做。

---

四、利弊权衡：什么时候不该避免胶水代码

维度	建模优先（避免胶水）	管道优先（接受胶水）
初期速度	❌ 慢 — 先设计实体、接口、适配器，100 行抽象才产出第一份 Excel	✅ 快 — 直接调 API、写 CSV、导 Excel，半小时出活
中期维护	✅ 低 — 改规则只改核心，换 OCR 只换适配器	❌ 高 — 改一个字段要改 5 个文件
可测试性	✅ 核心逻辑可脱离云 API 单测	❌ 几乎所有逻辑和外部依赖耦合
学习曲线	✅ 新人先看 domain.py 就理解全貌	❌ 需在 6 个文件间跳跃拼凑
复用性	✅ 下次做类似项目直接复用实体	❌ 下次只能复制粘贴改字段名
代码量 < 1000 行	⚠️ 过度设计风险高	✅ 性价比最高
一人维护	⚠️ 构架成本由一人承担	✅ 心智负担可接受
生命周期 < 3 月	⚠️ 抽象投资收不回	✅ 快速交付优先
依赖不会变	⚠️ 适配器抽象是浪费	✅ 直接耦合即可

判断拐点

如果你的项目满足以下两条以上，胶水代码反而是正确选择：

1. 总代码量预计 < 1000 行
2. 只有你一个人维护
3. 预期生命周期 < 3 个月
4. 外部依赖不会更换（公司统一采购了腾讯云 OCR）
5. 目的是快速验证想法，而非长期运营

当前 O2O 项目恰好部分满足以上条件（一人维护、依赖固定），所以它目前的胶水程度并非完全不合理。但如果它要长期维护、或者”预分货””促销追踪”等类似项目要复用，就应该重构。

---

五、渐进式改造路线

不需要一次到位。每个步骤独立发布、独立验证：

第 1 步：创建 o2o/domain.py
  → 定义 Offer 数据类
  → 把 verify.py 的校验逻辑改为 Offer 方法
  → 不改动任何其他文件

第 2 步：创建 o2o/parsers.py
  → 集中 _parse_price / _get_category / _extract_period 等函数
  → export_json.py 和 export_html.py 从 parsers 导入
  → 删除各自文件中的重复定义

第 3 步：统一下游接口
  → export_json / export_html / verify / audit 接收 List[Offer]
  → 不再各自从 CSV 解析

第 4 步：统一上游接口
  → ocr_o2o.py 的 process_image 返回 List[Offer]
  → CSV 读写成为 Offer 的序列化/反序列化层

第 5 步：引入端口抽象（可选）
  → OfferSource / OfferSink 接口
  → 适配器模式，为"预分货"项目复用做准备

核心原则只有一个：让”数据是什么”的定义只出现在一个地方，而不是每个文件各自重新理解一遍。

---

常见问题

胶水代码和工具函数库有什么区别？

工具函数库是通用、可复用的，一个函数被多个独立模块调用。胶水代码是专用、粘连的，它只在特定管道中连接 A 和 B——换掉 A 或 B，这层代码就得重写。

一定要用 ports-adapters 架构才能避免胶水代码吗？

不一定。对于代码量小于 1000 行的小项目，一个 domain.py + 简单的函数拆分就足够了。ports-adapters 的抽象层只有在外部依赖可能替换时才值得引入。

怎么判断当前项目的胶水偏离了合理范围？

改一个字段需要改动超过 3 个文件，或新人需要阅读 3 个以上文件才能理解一行数据的完整结构时，就已经偏离了合理范围。此时最简单的做法是创建一个 domain.py 数据类作为统一的数据描述。

小团队项目也需要严格避免胶水代码吗？

不需要。一人维护、生命周期短的「验证型」项目大量使用胶水代码是合理的权衡。关键在于设置一个「信号」：当改一个字段要改 4 个文件时，就是建模的时机到了。

胶水代码会影响项目可测试性吗？

会。如果所有逻辑和外部 API 耦合在一起，单测就必须 mock 云服务。通过将数据转换逻辑从 IO 中剥离出来（如策略 1 的 Offer 数据类），核心验证逻辑就不再依赖任何外部服务，可以直接跑单元测试。

相关文章

• Agent 技能蒸馏：从领域专家到可复用技能包 - 从业务逻辑中提炼可复用模块的方法论，与胶水代码的消除思路一脉相承
• AI 内容自动化生产线：从飞书到博客的完整工作流 - 完整管道式架构的实践案例，展示端口适配器模式在内容生产场景中的应用
• OpenClaw 高级技能开发指南 - OpenClaw Skill 系统的中间件、钩子和生命周期机制，和本文的声明式管道思路相呼应