Compare commits

...
Author SHA1 Message Date
wuzilongandCursor be0ba622a0 fix(pl): 证件额度切点改为按首次安排时间并冻结版本。
切点前已安排的计划归档跳过购买证/运输证校验;创建较早但安排在生效后的计划走客户+品类新规则。

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-11-15 19:11:02 +08:00
12 changed files with 257 additions and 21 deletions

No files matched your search

+3 -6
View File
@@ -2206,8 +2206,7 @@ def create_plan(
create_by=user.id,
create_dept_id=user.dept_id,
)
from app.services.pl_cert_quota_rule_config_service import stamp_plan_rule_version
stamp_plan_rule_version(db, user, plan)
# 证件额度规则版本在首次安排时按生效时间打标,创建不打标。
db.add(plan)
# 添加产品明细
@@ -2379,8 +2378,7 @@ def create_plan_full(
create_by=user.id,
create_dept_id=user.dept_id,
)
from app.services.pl_cert_quota_rule_config_service import stamp_plan_rule_version
stamp_plan_rule_version(db, user, plan)
# 证件额度规则版本在首次安排时按生效时间打标,创建不打标。
db.add(plan)
db.flush()
@@ -2531,8 +2529,7 @@ def create_from_contract(
create_by=user.id,
create_dept_id=user.dept_id,
)
from app.services.pl_cert_quota_rule_config_service import stamp_plan_rule_version
stamp_plan_rule_version(db, user, plan)
# 证件额度规则版本在首次安排时按生效时间打标,创建不打标。
db.add(plan)
# 查询合同产品明细
+5
View File
@@ -888,7 +888,10 @@ def arrange_transport(
)
apply_resource_snapshot_json(plan=plan, transport=transport, snapshot=snapshot)
from app.services.pl_quota_planning import pin_measurement_version
from app.services.pl_cert_quota_rule_config_service import stamp_plan_rule_on_arrange
pin_measurement_version(db, user, plan)
# 首次安排按生效时间打标并冻结;改派不重打。
stamp_plan_rule_on_arrange(db, user, plan)
plan.transport_status = "arranged"
plan.follow_status = "arranged"
plan.freight_amount = freight_amount or 0
@@ -1304,11 +1307,13 @@ def process_approval(
transport.carrier_name = vehicle_info['carrier_name']
snapshot = build_transport_resource_snapshot(db, tenant_no, transport.vehicle_id,
selected_driver_id=transport.driver_id, selected_escort_id=transport.escort_id)
from app.services.pl_cert_quota_rule_config_service import stamp_plan_rule_on_arrange
for plan in plans:
assert_pl_row_in_scope(db, plan, PlPlan, user, tenant_no)
plan.vehicle_id = transport.vehicle_id
plan.license_plate = transport.license_plate
apply_resource_snapshot_json(plan=plan, transport=transport, snapshot=snapshot)
stamp_plan_rule_on_arrange(db, user, plan)
apply_plan_quota(db, user, plan, "reserve", payload)
plan.transport_status = "arranged"
plan.follow_status = "arranged"
+2 -1
View File
@@ -370,7 +370,8 @@ class PlPlan(Base):
quota_cert_waived_by = Column(String(64), nullable=True, comment="证书齐备确认人ID")
quota_cert_waived_by_name = Column(String(100), nullable=True, comment="证书齐备确认人姓名")
quota_cert_waived_time = Column(DateTime, nullable=True, comment="证书齐备确认时间")
cert_quota_rule_version = Column(String(32), nullable=True, comment="证件额度规则版本 legacy|customer_category;创建时打标")
cert_quota_rule_version = Column(String(32), nullable=True, comment="证件额度规则版本 legacy|customer_category;首次安排时打标并冻结")
cert_quota_arranged_at = Column(DateTime, nullable=True, comment="首次安排时间(证件额度切点用);再次安排不改")
quota_cert_pending_fulfill = Column(SmallInteger, nullable=False, default=0, comment="新规则证书齐备待完善 0-否 1-是")
erp_entered = Column(SmallInteger, nullable=False, default=0, comment="是否已录入ERP 0-否 1-是")
cancel_reason = Column(String(500), nullable=True, comment="作废原因")
+2 -1
View File
@@ -417,7 +417,8 @@ class PlanRead(PlanBase, PlanResourceSnapshotMixin):
quota_cert_waived_by: Optional[str] = Field(default=None, alias="quotaCertWaivedBy", description="证书齐备确认人ID")
quota_cert_waived_by_name: Optional[str] = Field(default=None, alias="quotaCertWaivedByName", description="证书齐备确认人姓名")
quota_cert_waived_time: Optional[datetime] = Field(default=None, alias="quotaCertWaivedTime", description="证书齐备确认时间")
cert_quota_rule_version: Optional[str] = Field(default=None, alias="certQuotaRuleVersion", description="legacy|customer_category")
cert_quota_rule_version: Optional[str] = Field(default=None, alias="certQuotaRuleVersion", description="legacy|customer_category;首次安排打标")
cert_quota_arranged_at: Optional[datetime] = Field(default=None, alias="certQuotaArrangedAt", description="首次安排时间(证件额度切点)")
quota_cert_pending_fulfill: int = Field(default=0, alias="quotaCertPendingFulfill", description="新规则齐备待完善 0否 1是")
erp_entered: int = Field(default=0, alias="erpEntered", description="是否已录入ERP 0-否 1-是")
cancel_reason: Optional[str] = Field(default=None, alias="cancelReason", description="作废原因")
@@ -1,4 +1,4 @@
"""客户品类额度规则切点:系统信息配置;未配置则拦住安排。"""
"""客户品类额度规则切点:按首次安排时间打标;切点前已安排归档跳过校证。"""
from datetime import datetime
from typing import Optional
from uuid import uuid4
@@ -58,20 +58,77 @@ def assert_cert_quota_rule_configured(db: Session, tenant_no: str) -> datetime:
return effective_at
def resolve_rule_version_for_create(db: Session, tenant_no: str, create_time: Optional[datetime] = None) -> str:
"""计划创建时打标;未配置生效时间时仍标 legacy(安排会被拦住)。"""
def resolve_rule_version_for_arrange(
db: Session, tenant_no: str, arrange_time: Optional[datetime] = None
) -> str:
"""首次安排时打标:安排时刻 >= 生效时间 → 新规则,否则祖父 legacy。"""
effective_at = get_effective_at(db, tenant_no)
when = create_time or datetime.now()
when = arrange_time or datetime.now()
if effective_at is not None and when >= effective_at:
return VERSION_CUSTOMER_CATEGORY
return VERSION_LEGACY
def stamp_plan_rule_version(db: Session, user, plan) -> str:
version = resolve_rule_version_for_create(db, user.tenant_no, getattr(plan, 'create_time', None) or datetime.now())
# 兼容旧测试/调用名;语义已改为安排时刻。
resolve_rule_version_for_create = resolve_rule_version_for_arrange
def stamp_plan_rule_on_arrange(
db: Session, user, plan, *, arrange_time: Optional[datetime] = None
) -> str:
"""首次安排写入版本与安排时刻;再次安排冻结,不改版本。"""
if getattr(plan, 'cert_quota_arranged_at', None) is not None:
return getattr(plan, 'cert_quota_rule_version', None) or VERSION_LEGACY
when = arrange_time or datetime.now()
version = resolve_rule_version_for_arrange(db, user.tenant_no, when)
plan.cert_quota_arranged_at = when
plan.cert_quota_rule_version = version
return version
def stamp_plan_rule_version(db: Session, user, plan) -> str:
"""兼容旧名:等价于首次安排打标(创建路径不应再调用)。"""
return stamp_plan_rule_on_arrange(db, user, plan)
def plan_uses_customer_category(plan) -> bool:
return (getattr(plan, 'cert_quota_rule_version', None) or VERSION_LEGACY) == VERSION_CUSTOMER_CATEGORY
def infer_arranged_at(db: Session, plan) -> Optional[datetime]:
"""历史缺字段时用关联运输单创建时间推断首次安排。"""
stamped = getattr(plan, 'cert_quota_arranged_at', None)
if stamped is not None:
return stamped
transport_id = getattr(plan, 'transport_id', None)
tenant_no = getattr(plan, 'tenant_no', None)
if not transport_id or not tenant_no or db is None:
return None
from app.models.pl_models import PlTransport
transport = (
db.query(PlTransport)
.filter_by(id=transport_id, tenant_no=tenant_no, del_flag=0)
.first()
)
return getattr(transport, 'create_time', None) if transport else None
def plan_skips_archive_cert_quota(plan, db: Session = None, tenant_no: str = None) -> bool:
"""首次安排早于生效时间则归档跳过校证。
优先比较安排时刻与生效时间;若无生效配置或无法得到安排时刻,则不因空版本自动跳过
(避免误伤仍走合同绑证结转的历史路径;迁移/安排打标会补齐 arranged_at)。
"""
tn = tenant_no or getattr(plan, 'tenant_no', None)
arranged_at = getattr(plan, 'cert_quota_arranged_at', None)
if arranged_at is None and db is not None:
arranged_at = infer_arranged_at(db, plan)
if arranged_at is None:
return False
if db is not None and tn:
effective_at = get_effective_at(db, tn)
if effective_at is not None:
return arranged_at < effective_at
# 已记录首次安排且版本为 legacy(安排时未达生效点)→ 跳过
return not plan_uses_customer_category(plan)
+2
View File
@@ -1166,7 +1166,9 @@ def create_transport_for_plan(
escort_id=escort_id,
)
from app.services.pl_quota_planning import pin_measurement_version
from app.services.pl_cert_quota_rule_config_service import stamp_plan_rule_on_arrange
pin_measurement_version(db, user, plan)
stamp_plan_rule_on_arrange(db, user, plan)
plan.transport_status = "arranged"
plan.follow_status = "arranged"
plan.freight_amount = freight_amount or 0
+11 -2
View File
@@ -120,12 +120,21 @@ def apply_plan_quota(db, user, plan, action, payload=None):
evidenceUrls=getattr(payload, 'quota_evidence_urls', None),
sourceNote=getattr(payload, 'quota_source_note', None))
from app.services.pl_quota_waiver import assert_waived_sources, quota_cert_waived
from app.services.pl_cert_quota_rule_config_service import plan_uses_customer_category
from app.services.pl_cert_quota_rule_config_service import (
plan_skips_archive_cert_quota,
plan_uses_customer_category,
)
# 新规则:齐备仅放行安排/改派预占;归档与实际调整必须真证结转。
# 旧规则(legacy):齐备仍跳过安排与归档(祖父兼容)。
# 切点前首次安排:归档/实调不校证件(无需齐备);齐备仍可跳过安排预占。
skip_actions = ('reserve', 'resize')
if not plan_uses_customer_category(plan):
skip_actions = ('reserve', 'resize', 'settle', 'actual_adjust')
grandfather_archive_skip = (
action in ('settle', 'actual_adjust')
and plan_skips_archive_cert_quota(plan, db, getattr(user, 'tenant_no', None))
)
if grandfather_archive_skip:
return {'operationId': ledger_id, 'usageIds': [], 'waived': True, 'skippedCertQuota': True}
if quota_cert_waived(plan) and action in skip_actions:
if action in ('reserve', 'resize'):
assert_waived_sources(db, user, plan, lock=True)
+82 -4
View File
@@ -19,16 +19,94 @@ def test_assert_rule_not_configured_blocks_arrange():
assert '系统信息' in str(exc.value.detail)
def test_resolve_rule_version_by_effective_at():
def test_resolve_rule_version_by_arrange_effective_at():
from app.services import pl_cert_quota_rule_config_service as cfg
db = MagicMock()
row = SimpleNamespace(effective_at=datetime(2026, 11, 14, 0, 0, 0))
db.query.return_value.filter_by.return_value.first.return_value = row
assert cfg.resolve_rule_version_for_create(db, 't', datetime(2026, 11, 13)) == 'legacy'
assert cfg.resolve_rule_version_for_create(db, 't', datetime(2026, 11, 14)) == 'customer_category'
assert cfg.resolve_rule_version_for_arrange(db, 't', datetime(2026, 11, 13)) == 'legacy'
assert cfg.resolve_rule_version_for_arrange(db, 't', datetime(2026, 11, 14)) == 'customer_category'
db.query.return_value.filter_by.return_value.first.return_value = None
assert cfg.resolve_rule_version_for_create(db, 't', datetime(2026, 11, 20)) == 'legacy'
assert cfg.resolve_rule_version_for_arrange(db, 't', datetime(2026, 11, 20)) == 'legacy'
def test_stamp_plan_rule_version_freezes_first_arrange():
from app.services import pl_cert_quota_rule_config_service as cfg
db = MagicMock()
row = SimpleNamespace(effective_at=datetime(2026, 11, 14, 0, 0, 0))
db.query.return_value.filter_by.return_value.first.return_value = row
user = SimpleNamespace(tenant_no='t', id='u')
plan = SimpleNamespace(cert_quota_rule_version=None, cert_quota_arranged_at=None)
first = cfg.stamp_plan_rule_on_arrange(db, user, plan, arrange_time=datetime(2026, 11, 13, 10, 0, 0))
assert first == 'legacy'
assert plan.cert_quota_rule_version == 'legacy'
assert plan.cert_quota_arranged_at == datetime(2026, 11, 13, 10, 0, 0)
second = cfg.stamp_plan_rule_on_arrange(db, user, plan, arrange_time=datetime(2026, 11, 20, 10, 0, 0))
assert second == 'legacy'
assert plan.cert_quota_rule_version == 'legacy'
assert plan.cert_quota_arranged_at == datetime(2026, 11, 13, 10, 0, 0)
def test_stamp_after_cutoff_uses_customer_category_even_if_created_earlier():
from app.services import pl_cert_quota_rule_config_service as cfg
db = MagicMock()
row = SimpleNamespace(effective_at=datetime(2026, 11, 14, 0, 0, 0))
db.query.return_value.filter_by.return_value.first.return_value = row
user = SimpleNamespace(tenant_no='t', id='u')
plan = SimpleNamespace(
create_time=datetime(2026, 11, 1),
cert_quota_rule_version=None,
cert_quota_arranged_at=None,
)
version = cfg.stamp_plan_rule_on_arrange(db, user, plan, arrange_time=datetime(2026, 11, 15))
assert version == 'customer_category'
assert plan.cert_quota_rule_version == 'customer_category'
def test_plan_skips_archive_when_arranged_before_effective_at():
from app.services import pl_cert_quota_rule_config_service as cfg
db = MagicMock()
db.query.return_value.filter_by.return_value.first.return_value = SimpleNamespace(
effective_at=datetime(2026, 11, 14, 0, 0, 0)
)
before = SimpleNamespace(
tenant_no='t',
cert_quota_arranged_at=datetime(2026, 11, 10),
cert_quota_rule_version='legacy',
transport_id=None,
)
after = SimpleNamespace(
tenant_no='t',
cert_quota_arranged_at=datetime(2026, 11, 15),
cert_quota_rule_version='customer_category',
transport_id=None,
)
assert cfg.plan_skips_archive_cert_quota(before, db, 't') is True
assert cfg.plan_skips_archive_cert_quota(after, db, 't') is False
def test_legacy_arranged_before_cutoff_skips_archive_cert_without_waive(monkeypatch):
"""切点前首次安排的计划:归档不校证件,无需证书齐备。"""
from app.services import pl_quota_business as biz
plan = SimpleNamespace(
id='p1',
quota_cert_waived=0,
quota_cert_pending_fulfill=0,
cert_quota_rule_version='legacy',
cert_quota_arranged_at=datetime(2026, 11, 10),
cert_configuration={},
)
monkeypatch.setattr(
'app.services.pl_cert_quota_rule_config_service.plan_skips_archive_cert_quota',
lambda p, db=None, tenant_no=None: True,
)
result = biz.apply_plan_quota(MagicMock(), SimpleNamespace(id='u', tenant_no='t'), plan, 'settle', None)
assert result.get('waived') is True or result.get('skippedCertQuota') is True
def test_apply_plan_quota_new_rule_waive_skips_reserve_not_settle(monkeypatch):
@@ -0,0 +1,28 @@
-- 证件额度切点改为首次安排时间:补 arranged_at 列,并按运输单创建时间回填已安排计划。
-- MySQL 5.7。可重复执行。
SET NAMES utf8mb4;
SET @schema_name = DATABASE();
SET @col = (SELECT COUNT(*) FROM information_schema.COLUMNS
WHERE TABLE_SCHEMA=@schema_name AND TABLE_NAME='pl_plan' AND COLUMN_NAME='cert_quota_arranged_at');
SET @sql = IF(@col=0,
'ALTER TABLE pl_plan ADD COLUMN cert_quota_arranged_at datetime DEFAULT NULL COMMENT ''首次安排时间(证件额度切点用);再次安排不改'' AFTER cert_quota_rule_version',
'SELECT 1');
PREPARE s FROM @sql; EXECUTE s; DEALLOCATE PREPARE s;
-- 已安排:回填首次安排时刻,并按「安排时刻 vs 生效时间」校正规则版本(修正历史按创建打标)。
UPDATE pl_plan p
INNER JOIN pl_transport t
ON t.id = p.transport_id AND t.tenant_no = p.tenant_no AND t.del_flag = 0
LEFT JOIN pl_cert_quota_rule_config cfg
ON cfg.tenant_no = p.tenant_no AND cfg.del_flag = 0
SET
p.cert_quota_arranged_at = COALESCE(p.cert_quota_arranged_at, t.create_time),
p.cert_quota_rule_version = CASE
WHEN cfg.effective_at IS NOT NULL
AND COALESCE(p.cert_quota_arranged_at, t.create_time) >= cfg.effective_at
THEN 'customer_category'
ELSE 'legacy'
END
WHERE p.del_flag = 0
AND p.transport_status = 'arranged';
@@ -0,0 +1,28 @@
-- 证件额度切点改为首次安排时间:补 arranged_at 列,并按运输单创建时间回填已安排计划。
-- MySQL 5.7。可重复执行。
SET NAMES utf8mb4;
SET @schema_name = DATABASE();
SET @col = (SELECT COUNT(*) FROM information_schema.COLUMNS
WHERE TABLE_SCHEMA=@schema_name AND TABLE_NAME='pl_plan' AND COLUMN_NAME='cert_quota_arranged_at');
SET @sql = IF(@col=0,
'ALTER TABLE pl_plan ADD COLUMN cert_quota_arranged_at datetime DEFAULT NULL COMMENT ''首次安排时间(证件额度切点用);再次安排不改'' AFTER cert_quota_rule_version',
'SELECT 1');
PREPARE s FROM @sql; EXECUTE s; DEALLOCATE PREPARE s;
-- 已安排:回填首次安排时刻,并按「安排时刻 vs 生效时间」校正规则版本(修正历史按创建打标)。
UPDATE pl_plan p
INNER JOIN pl_transport t
ON t.id = p.transport_id AND t.tenant_no = p.tenant_no AND t.del_flag = 0
LEFT JOIN pl_cert_quota_rule_config cfg
ON cfg.tenant_no = p.tenant_no AND cfg.del_flag = 0
SET
p.cert_quota_arranged_at = COALESCE(p.cert_quota_arranged_at, t.create_time),
p.cert_quota_rule_version = CASE
WHEN cfg.effective_at IS NOT NULL
AND COALESCE(p.cert_quota_arranged_at, t.create_time) >= cfg.effective_at
THEN 'customer_category'
ELSE 'legacy'
END
WHERE p.del_flag = 0
AND p.transport_status = 'arranged';
@@ -0,0 +1,30 @@
# 证件额度切点:按首次安排时间(设计)
工单:[#40](https://gitea.aichengyan.com/cykj/shanxia2.0/issues/40)
## 背景
原设计按计划 `create_time` 与 `effective_at` 比较打 `cert_quota_rule_version`。业务要求改为按**首次安排运输**时刻切点;切点前已安排的计划归档不再校购买证/运输证。
## 定稿规则
1. **未配置 `effective_at`**:安排仍 409 拦住(不变)。
2. **首次安排**时写入:
- `cert_quota_arranged_at = 安排时刻`
- `cert_quota_rule_version = customer_category` 若 `安排时刻 >= effective_at`,否则 `legacy`
3. **再次调整运输**:不重打 `cert_quota_arranged_at` / `cert_quota_rule_version`(冻结首次安排)。
4. **创建计划**:不再按创建时间打标(可空,待安排时写入)。
5. **归档**:
- `legacy`(切点前首次安排):不校验、不扣购买证/运输证,额度结转按跳过处理,允许直接归档。
- `customer_category`:保持现有最新业务逻辑(选证、齐备仅放行安排、归档须真证等)。
6. **历史已安排且缺 `cert_quota_arranged_at`**:归档/读规则时用关联 `pl_transport.create_time` 推断;若推断时刻 `< effective_at` 视为 `legacy` 祖父。
## 范围
- backend:切点服务、安排打标、归档跳过、迁移列
- web-admin:系统信息「运输证件规则」说明文案
## 非目标
- 不批量改历史 `quota_cert_waived`
- 不改选证算法本身
@@ -7,7 +7,7 @@
</div>
<div class="p-6 max-w-xl">
<p class="text-sm text-slate-500 mb-4">
配置「客户+品类」额度生效时间。未配置时无法安排运输。计划创建时按该时间打规则版本;修改后只影响之后新建的计划。
配置「客户+品类」额度生效时间。未配置时无法安排运输。以计划首次安排时刻与该时间比较:之前已安排的归档不再校验购买证/运输证;之后安排的走客户+品类最新规则。首次安排后版本冻结,改派不重打;修改生效时间只影响尚未安排的计划。
</p>
<el-form label-position="top" class="dialog-form-layout">
<el-form-item label="客户品类额度生效时间">