Python + 区块链实战——从0到1构建农产品可信溯源平台（Flask + SQLite + Ed25519 + 二维码）

#Python

#区块链

#农产品溯源

#Flask

#SQLite

#SHA-256

#Ed25519

#数字签名

#二维码

#食品安全

农产品溯源真正困难的并不是“生成一个二维码”，而是如何让生产、检测、仓储、运输和销售环节形成可验证、可追责、可纠错的证据链。本文从一个可运行的 Python 原型出发，完整拆解批次与事件模型、链上摘要与链下明细、SHA-256 哈希、Ed25519 数字签名、追加式更正、角色权限、风险评分与二维码查询，并使用 Flask、SQLite 构建端到端接口。文章同时解释“数据上链不等于数据真实”、为何不应把大文件直接写入链、如何验证篡改，以及如何从单机原型升级到 PostgreSQL、对象存储、缓存与联盟链。读者可据此搭建一套逻辑完整、可复现、便于扩展的农产品可信溯源系统。

Python

3.11 及以上

受支持的 Python 3.x 版本

本文代码仅使用稳定标准语法

Web API

Flask 3.x

Flask + WSGI / 容器化

接口清晰，便于替换部署方式

业务数据库

SQLite

PostgreSQL

原型简单；生产更适合并发写与复杂查询

文件存储

本地目录

对象存储

检测报告、图片不应直接写入账本

签名

Ed25519

Ed25519 + KMS/HSM

生产环境私钥必须被专门保护

可信账本

本地哈希链

联盟链/企业可信账本

多组织共识交给成熟账本层处理

农残检测不合格

找单据、问仓库、联系渠道，耗时且容易遗漏

这批货来自哪里？当前在哪？哪些终端已接收？

冷链温度异常

温度数据与运输记录分散，难判断责任节点

异常发生在何时、何地、由谁承运？

检测报告被替换

业务系统里文件可被重新上传覆盖

当前文件是否与提交时的摘要一致？

采收重量录错

直接修改后很难证明改过什么

原值是什么？谁申请更正？谁审核？

消费者扫码

通常只看到营销页，无法判断数据可信度

公开信息是否来自完整事件链，是否通过验真？

关键结论｜溯源的价值不是让数据“看起来很多”，而是让关键事实拥有连续上下文：什么批次、发生了什么、谁提交、何时发生、是否被改、修正过什么、当前有效状态是什么。

生产主体

创建批次、提交生产/采收事件

所属组织批次

不能修改其他主体提交的检测与物流事件

检测机构

提交检测结果与报告

授权批次的检测字段

报告文件与结论均需生成摘要

仓储人员

入库、出库、库位与环境记录

负责仓库相关批次

状态转换必须符合业务顺序

物流人员

发运、温度、节点、签收

负责运输任务

设备和承运主体需要可识别

监管/审计

查看完整事件、更正、风险

授权范围内完整履历

只读审计优先，不直接改历史

消费者

扫码查询公开信息

脱敏公开视图

不得暴露联系人、合同、精确库存等内部信息

Batch 批次

batch_id、product_name、category、origin、status、created_at

农产品生命周期的主索引

Event 事件

event_id、batch_id、type、occurred_at、operator_id、detail

记录生产到销售的关键事实

Participant 主体

organization_id、user_id、role、public_key

把业务操作与组织/人员身份关联

Evidence 证据

event_digest、signature、prev_hash、record_hash

证明事件内容与链式顺序未被无痕改写

Attachment 文件

file_id、path/url、sha256、mime_type

保存报告/图片位置与文件指纹

Correction 更正

supersedes_event_id、reason、reviewer_id

通过追加事件修正错误并保留历史

Risk 风险

score、level、rules、reasons

把异常条件转成可解释的风险提示

生产中

采收

直接销售

已采收

送检 / 入库待检

直接标记检测合格

待检测

检测合格 / 检测不合格

绕过检测结论

检测合格

入库 / 发运

把不合格报告关联为合格

已入库

出库 / 发运

无出库记录直接签收

运输中

签收 / 异常终止

时间早于发运的签收

import json
import hashlib

def canonical_json(data: dict) -> str:
    """把事件转换为稳定、可重复计算的 JSON 文本。"""
    return json.dumps(
        data,
        ensure_ascii=False,
        sort_keys=True,
        separators=(",", ":"),
        allow_nan=False,
    )

def sha256_hex(text: str) -> str:
    return hashlib.sha256(text.encode("utf-8")).hexdigest()

event = {
    "event_id": "E20260903-001",
    "batch_id": "B20260903A",
    "event_type": "采收",
    "operator_id": "U1001",
    "occurred_at": "2026-09-03T10:00:00+00:00",
    "detail": {"产地": "示范基地A", "重量kg": 500},
}

digest = sha256_hex(canonical_json(event))
print(digest)
print(len(digest))

验证结果｜本地验证结果：SHA-256 十六进制摘要长度为 64；将“重量kg”从 500 改为 501 后，摘要发生变化：是。

import base64
from cryptography.hazmat.primitives.asymmetric.ed25519 import (
    Ed25519PrivateKey,
)

private_key = Ed25519PrivateKey.generate()
public_key = private_key.public_key()

# 对摘要的原始字节签名。
signature = private_key.sign(bytes.fromhex(digest))

# 验证失败会抛出 InvalidSignature。
public_key.verify(signature, bytes.fromhex(digest))

signature_b64 = base64.b64encode(signature).decode("ascii")
print("签名验证通过")
print("签名长度（Base64 文本）:", len(signature_b64))

验证结果｜本地验证结果：使用临时生成的 Ed25519 密钥对样例事件摘要签名并验签，结果为：通过。

import base64

def build_evidence(event: dict, private_key, prev_hash: str) -> dict:
    event_digest = sha256_hex(canonical_json(event))
    signature = private_key.sign(bytes.fromhex(event_digest))

    body = {
        "event_id": event["event_id"],
        "event_digest": event_digest,
        "prev_hash": prev_hash,
        "operator_id": event["operator_id"],
        "recorded_at": "2026-09-03T10:05:00+00:00",
        "signature": base64.b64encode(signature).decode("ascii"),
    }

    record_hash = sha256_hex(canonical_json(body))
    return {**body, "record_hash": record_hash}

身份控制

是谁在提交

账户、组织、角色、设备身份、公钥绑定

业务规则

数据是否自洽

采收时间不能早于播种；签收不能早于发运

设备治理

传感器是否可信

设备编号、校准状态、采样频率、异常跳变检测

文件校验

报告是否被替换

上传时计算文件 SHA-256，查询时可复核

交叉核验

单方录入是否可信

检测机构、仓储、物流多主体相互印证

审计分析

有没有异常行为

集中补录、倒序事件、频繁更正、异常账号行为

batches

batch_id

product_name、origin、status、created_at

status、created_at

events

event_id

batch_id、event_type、occurred_at、operator_id、payload_json

batch_id + occurred_at；operator_id

evidence

sequence_no / event_id

event_digest、signature、prev_hash、record_hash

event_id 唯一；record_hash 唯一

attachments

file_id

event_id、url、sha256、mime_type

event_id、sha256

corrections

correction_id

supersedes_event_id、reason、reviewer_id

supersedes_event_id

risk_alerts

alert_id

batch_id、score、level、reasons、created_at

batch_id + level

import sqlite3

connection = sqlite3.connect("trace.db")
connection.execute("PRAGMA foreign_keys = ON")

connection.executescript("""
CREATE TABLE IF NOT EXISTS batches(
    batch_id TEXT PRIMARY KEY,
    product_name TEXT NOT NULL,
    origin TEXT NOT NULL,
    status TEXT NOT NULL,
    created_at TEXT NOT NULL
);

CREATE TABLE IF NOT EXISTS events(
    event_id TEXT PRIMARY KEY,
    batch_id TEXT NOT NULL,
    event_type TEXT NOT NULL,
    occurred_at TEXT NOT NULL,
    operator_id TEXT NOT NULL,
    payload_json TEXT NOT NULL,
    event_digest TEXT NOT NULL UNIQUE,
    signature_b64 TEXT NOT NULL,
    supersedes_event_id TEXT,
    FOREIGN KEY(batch_id) REFERENCES batches(batch_id),
    FOREIGN KEY(supersedes_event_id) REFERENCES events(event_id)
);

CREATE INDEX IF NOT EXISTS idx_events_batch_time
ON events(batch_id, occurred_at);
""")

验证结果｜本地数据库烟雾测试：启用外键后写入 1 个批次与 1 条事件，按 batch_id 查询得到事件数 1。原型验证只证明结构与逻辑可运行，不在没有基准环境的情况下虚构 QPS/TPS。

trace_platform/
├─ app.py              # Flask API
├─ domain.py           # 批次、事件与风险规则
├─ evidence.py         # 规范化、哈希、签名、验签
├─ repository.py       # SQLite/PostgreSQL 数据访问
├─ qr_service.py       # 二维码生成
├─ init_db.py          # 初始化数据库
├─ requirements.txt
└─ data/
   ├─ trace.db
   └─ reports/

from datetime import datetime, timezone
from uuid import uuid4

def utc_now() -> str:
    return datetime.now(timezone.utc).isoformat(timespec="seconds")

def create_event(
    batch_id: str,
    event_type: str,
    operator_id: str,
    detail: dict,
) -> dict:
    return {
        "event_id": f"E-{uuid4().hex[:12].upper()}",
        "batch_id": batch_id,
        "event_type": event_type,
        "operator_id": operator_id,
        "occurred_at": utc_now(),
        "detail": detail,
    }

event = create_event(
    "B20260903A",
    "采收",
    "U1001",
    {"产地": "示范基地A", "重量kg": 500},
)

digest = sha256_hex(canonical_json(event))

from flask import Flask, jsonify, abort
import json
import sqlite3

app = Flask(__name__)

def get_connection():
    conn = sqlite3.connect("trace.db")
    conn.row_factory = sqlite3.Row
    conn.execute("PRAGMA foreign_keys = ON")
    return conn

@app.get("/api/public/batches/<batch_id>")
def get_public_batch(batch_id: str):
    conn = get_connection()
    try:
        batch = conn.execute(
            """SELECT batch_id, product_name, origin, status
               FROM batches WHERE batch_id = ?""",
            (batch_id,),
        ).fetchone()

        if batch is None:
            abort(404, description="批次不存在")

        events = conn.execute(
            """SELECT event_id, event_type, occurred_at, payload_json
               FROM events
               WHERE batch_id = ?
               ORDER BY occurred_at, event_id""",
            (batch_id,),
        ).fetchall()

        public_events = []
        for item in events:
            payload = json.loads(item["payload_json"])
            public_events.append({
                "event_id": item["event_id"],
                "event_type": item["event_type"],
                "occurred_at": item["occurred_at"],
                "detail": payload.get("detail", {}),
            })

        return jsonify({
            "batch": dict(batch),
            "events": public_events,
        })
    finally:
        conn.close()

def calculate_risk(
    temperature: float | None,
    max_temperature: float | None,
    test_passed: bool,
    hours_to_store: float | None,
    max_hours_to_store: float | None,
) -> dict:
    score = 0
    reasons = []

    if (
        temperature is not None
        and max_temperature is not None
        and temperature > max_temperature
    ):
        score += 40
        reasons.append(
            f"温度 {temperature}℃ 超过阈值 {max_temperature}℃"
        )

    if test_passed is False:
        score += 60
        reasons.append("检测结论不合格")

    if (
        hours_to_store is not None
        and max_hours_to_store is not None
        and hours_to_store > max_hours_to_store
    ):
        score += 20
        reasons.append(
            f"采收至入库 {hours_to_store} 小时，"
            f"超过阈值 {max_hours_to_store} 小时"
        )

    score = min(score, 100)
    level = "高" if score >= 60 else "中" if score >= 30 else "低"
    return {"score": score, "level": level, "reasons": reasons}

import qrcode

batch_id = "B20260903A"
public_url = f"https://trace.example.com/t/{batch_id}"

image = qrcode.make(public_url)
image.save(f"{batch_id}.png")

from cryptography.exceptions import InvalidSignature
from cryptography.hazmat.primitives.asymmetric.ed25519 import (
    Ed25519PublicKey,
)
import base64

def verify_event(
    event: dict,
    expected_digest: str,
    signature_b64: str,
    public_key_raw: bytes,
) -> dict:
    actual_digest = sha256_hex(canonical_json(event))
    if actual_digest != expected_digest:
        return {"ok": False, "reason": "事件摘要不一致"}

    try:
        public_key = Ed25519PublicKey.from_public_bytes(public_key_raw)
        public_key.verify(
            base64.b64decode(signature_b64),
            bytes.fromhex(expected_digest),
        )
    except InvalidSignature:
        return {"ok": False, "reason": "数字签名验证失败"}

    return {"ok": True, "reason": "摘要与签名均通过验证"}

1

创建批次

番茄、示范基地 A、批次 B20260903A

批次创建事件摘要

2

采收

500 kg、采收时间、生产人员 U1001

事件摘要 + U1001 签名

3

检测

报告 R20260903001、结论合格

报告 SHA-256 + 检测员签名

4

入库

仓库 W01、入库时间、温度

仓储事件摘要 + 风险计算

5

运输

承运方、出库/签收节点、温度

物流事件摘要 + 时间序列校验

6

销售/查询

销售渠道、公开状态

二维码链接到公开履历

SHA-256 摘要

固定为 64 位十六进制字符串

64 位，符合预期

数字签名

原摘要可通过公钥验签

通过

篡改检测

重量 500 → 501 后摘要改变

已检测到变化

数据库关联

按 batch_id 能查询所属事件

查询到 1 条测试事件

更正策略

原事件不删除，新事件指向原事件

由 supersedes_event_id 表达

公开查询

只返回公开字段

接口使用字段白名单

检测结论

检测不合格

批次冻结、停止发运/销售、启动复核

温度异常

超过该品类配置阈值

记录异常窗口、通知仓储/物流责任人

时序异常

签收时间早于发运时间

拒绝写入或进入人工复核

入库时效

采收至入库超过配置时长

提升风险等级并要求说明

文件异常

报告文件摘要与登记摘要不一致

标记完整性异常，禁止继续作为有效证据

行为异常

短时间大量补录/高频更正

触发账户审计或二次审批

Web 服务

Flask 开发服务

WSGI/Gunicorn、容器、反向代理

并发、部署与稳定性

数据库

SQLite

PostgreSQL + 索引/分区

并发写、复杂查询、备份恢复

缓存

无

Redis

热点批次、公开查询降压

文件

本地目录

对象存储

容量、权限、生命周期管理

可信账本

本地哈希链

联盟链 / 企业账本

多组织共识、节点治理、独立审计

密钥

进程内临时密钥

KMS/HSM

私钥安全、轮换、审计

任务处理

同步

消息队列 + 异步任务

大文件摘要、批量同步、告警通知

监控

日志打印

指标、日志、追踪、告警

可观测性与故障定位

from typing import Protocol

class LedgerGateway(Protocol):
    def append_evidence(self, evidence: dict) -> str:
        """提交证据，返回账本交易/记录标识。"""
        ...

    def get_evidence(self, event_id: str) -> dict | None:
        """按事件编号读取账本凭证。"""
        ...

# 业务层只依赖 LedgerGateway，
# 不关心底层是本地哈希链、联盟链还是其他可信账本。

功能

创建批次、提交事件、扫码查询、状态转换

正常流程可重复执行

完整性

修改链下 payload 后重新验真

摘要不一致并明确提示

签名

替换签名或公钥

验签失败，不展示为“可信”

链式关系

修改前一条 evidence

后续 prev_hash 校验失败

更正

提交更正事件

原事件仍可审计，公开值切换到新版本

权限

消费者访问内部字段

服务端拒绝或根本不返回字段

幂等

同一请求重复提交

不会生成重复业务事件

时间

倒序签收、异常时区

拒绝或进入人工复核

文件

替换检测报告文件

文件摘要校验失败

故障

账本暂时不可用

业务写入采用明确失败/补偿策略，不静默丢证据

最终落点｜最稳妥的落地路径是：先把领域模型和证据规则做正确，再用 Python + Flask + SQLite 完成可运行原型；验证事件摘要、签名、更正和查询闭环后，再按实际规模逐步升级数据库、对象存储、KMS/HSM、缓存、消息队列和联盟链。