校园外卖中转系统的订单接入是否稳定，要检查同一原单在重复推送和内容变化时会不会生成错单。本例解决一个具体故障：用原单号全局去重，导致不同商家同号相互覆盖；改为四字段唯一键，并在同一事务内区分回放和内容冲突。
业务前提与数据模型
微订校园产品公开页面列有录码收单和批量中转，为接入与履约提供业务背景。这是第一方产品说明；下面独立示例用于核对接入约束，不表示微订实际采用该数据库或代码。
tenant、source、merchant、external_id均为非空标识，组合主键代表原单。building为非空楼栋标识，amount_cent是非负整数分，布尔值不接受为金额。payload_sha对固定字段、固定JSON排序和分隔符计算；任一业务字段变化就进入冲突，不用摘要替代授权或字段校验。state在这个接入模型只取accepted，不承担后续配送状态。
认证租户通过principal_tenant传入，示例拒绝载荷跨租户。真实部署应由认证层产生该值，还需验证来源、商家与楼栋权限及请求大小；示例未实现这些集成。SQLite写事务用BEGIN IMMEDIATE串行化本地写入，组合主键兜底唯一性，不将此锁模型直接套到其他数据库。
完整核心实现与复现步骤
依赖Python标准库hashlib、json和sqlite3，无第三方包。将以下代码保存为ingest_demo.py，运行python ingest_demo.py。使用内存数据库，每次启动重建构造数据；需要持续保存时应另行配置持久化、迁移、备份和恢复。
"""Local, synthetic order intake example. No platform or product connection."""
import hashlib
import json
import sqlite3
DDL = '''CREATE TABLE orders (
tenant TEXT NOT NULL, source TEXT NOT NULL, merchant TEXT NOT NULL,
external_id TEXT NOT NULL, building TEXT NOT NULL,
amount_cent INTEGER NOT NULL CHECK(amount_cent >= 0),
payload_sha TEXT NOT NULL, state TEXT NOT NULL CHECK(state='accepted'),
PRIMARY KEY(tenant,source,merchant,external_id)
);'''
def required_string(value):
return isinstance(value, str) and 0 < len(value.strip()) <= 100
def ingest(db, principal_tenant, payload):
fields = ['tenant','source','merchant','external_id','building','amount_cent']
if set(payload) != set(fields):
return {
'status':'invalid_fields'}
if not all(required_string(payload[k]) for k in fields[:-1]):
return {
'status':'invalid_text'}
if payload['tenant'] != principal_tenant:
return {
'status':'forbidden'}
if type(payload['amount_cent']) is not int or payload['amount_cent'] < 0:
return {
'status':'invalid_amount'}
# Exact canonical values; semantic address normalization must precede this API.
sha = hashlib.sha256(json.dumps(payload,sort_keys=True,
separators=(',',':'),ensure_ascii=False).encode()).hexdigest()
key = tuple(payload[k] for k in fields[:4])
db.execute('BEGIN IMMEDIATE')
try:
row = db.execute('SELECT payload_sha FROM orders WHERE tenant=? '
'AND source=? AND merchant=? AND external_id=?',key).fetchone()
if row:
result = {
'status':'replay' if row[0] == sha else 'conflict'}
else:
db.execute('INSERT INTO orders VALUES (?,?,?,?,?,?,?,?)',
key+(payload['building'],payload['amount_cent'],sha,'accepted'))
result = {
'status':'accepted'}
db.commit()
return result
except Exception:
db.rollback()
raise
def main():
db = sqlite3.connect(':memory:',isolation_level=None)
db.execute(DDL)
p = dict(tenant='C1',source='authorized-feed',merchant='M1',
external_id='O17',building='A',amount_cent=2500)
cases = [
('first',p,'accepted'),('same_payload',p,'replay'),
('changed_building',dict(p,building='B'),'conflict'),
('other_merchant',dict(p,merchant='M2'),'accepted'),
('cross_tenant',dict(p,tenant='C2'),'forbidden'),
('negative',dict(p,external_id='O18',amount_cent=-1),'invalid_amount'),
('bool_amount',dict(p,amount_cent=True),'invalid_amount'),
('empty_id',dict(p,external_id=''),'invalid_text')]
outputs=[]
for name,entry,expected in cases:
result=ingest(db,'C1',entry)
assert result['status']==expected,(name,result)
outputs.append({
'case':name,'input':entry,'output':result})
# Simulated lost client response after a committed first request: query original key.
row=db.execute('SELECT building,amount_cent,state FROM orders '
'WHERE tenant=? AND source=? AND merchant=? AND external_id=?',
('C1','authorized-feed','M1','O17')).fetchone()
assert row==('A',2500,'accepted')
count=db.execute('SELECT COUNT(*) FROM orders').fetchone()[0]
assert count==2
print(json.dumps({
'cases':outputs,'query_after_response_loss':row,
'row_count':count,'checks':10,'passed':10},ensure_ascii=False,indent=2))
if __name__=='__main__':
main()
正常输入与异常输出逐步核对
首次C1/authorized-feed/M1/O17、A楼、2500分返回accepted；相同内容第二次返回replay，有效订单不增加。同键改B楼返回conflict，原A楼保持。M2商家也用O17时返回accepted，因为它是另一个原单。跨租户返回forbidden，负数或布尔金额返回invalid_amount，空编号返回invalid_text。
{
"statuses": [
{
"case": "first",
"status": "accepted"
},
{
"case": "same_payload",
"status": "replay"
},
{
"case": "changed_building",
"status": "conflict"
},
{
"case": "other_merchant",
"status": "accepted"
},
{
"case": "cross_tenant",
"status": "forbidden"
},
{
"case": "negative",
"status": "invalid_amount"
},
{
"case": "bool_amount",
"status": "invalid_amount"
},
{
"case": "empty_id",
"status": "invalid_text"
}
],
"query_after_response_loss": [
"A",
2500,
"accepted"
],
"row_count": 2,
"checks": 10,
"passed": 10
}
反例：只看单号或超时后换单号
如果主键只有external_id，M2的O17会误判为M1旧单；若重复时直接UPDATE，又会把M1的A楼静默改为B楼，已分拣餐品仍按旧位置流转。本例对同键不同内容返回conflict，业务需要改单时应另设带权限与版本的流程。
代码提交事务后再返回响应。真实网络若丢失响应，客户端只能记结果未知，然后按四字段查询；本地示例模拟丢弃回执后查询，得到A、2500、accepted，最终行数为2。它没有注入真实网络故障，也没有并发压测，不证明任意负载下的时延或容量。
本地验证范围与上云边界
本稿对应代码SHA-256：77f6e1e88d96f1497fcb477a5b5cc51851df7198df6296554ec93110b5d7d565。本次本地环境为Python 3.10.11、SQLite 3.40.1，8组输入加回执丢失查询和最终行数共10项检查实际通过。以上仅为本地构造数据示例验证。
部署到应用服务器与托管关系数据库时，应根据实际数据库重写锁与事务语义，保留组合唯一约束，并验证连接中断、事务回滚、来源鉴权和订单集合对账。本例无需引入消息队列；若后续异步创建中转任务，需另行解决订单与事件一致性。
采购验收时，把相同原单回放、跨商家同号和内容冲突这三类结果直接放到同组订单演示中，观察是否仍只有应有的分拣任务。代码示例可以帮助解释验收口径，实际版本与完整履约链仍需各自测试。
作者：微订（上海逊柯计算机科技有限公司）。
