diff --git a/backend/app/api/pl/quota_cert.py b/backend/app/api/pl/quota_cert.py index 4d666146b..c9f7b29a4 100644 --- a/backend/app/api/pl/quota_cert.py +++ b/backend/app/api/pl/quota_cert.py @@ -16,12 +16,13 @@ from app.core.oper_log_decorator import oper_log from app.core.tenant_context import get_tenant_no from app.schemas.pl.quota_cert import InputBody, CertConfirmBody, CertVerifyBody, CertTotalAdjustBody from app.schemas.pl.quota_scope import ContractCertBindingsBody -from app.schemas.pl.legal_use_document import LegalDocumentsBody +from app.schemas.pl.legal_use_document import LegalDocumentsBody, StopLegalDocumentsBody from app.schemas.pl.quota_planning import PlanQuotaConfiguration, PlanQuotaPreview from app.services import pl_quota_cert_service as certs from app.services import pl_quota_contract_service as scopes from app.services import pl_quota_contract_query as contract_query from app.services import pl_legal_use_document as documents +from app.services import pl_stop_legal_document as stop_documents from app.services import pl_quota_planning as planning from app.services import pl_quota_adjustment as adjustment from app.services import pl_quota_overview as overview @@ -225,6 +226,28 @@ def list_documents(keyword: Optional[str] = None, status: Optional[str] = None, return documents.list_documents(db, user, keyword, status) +@legal_router.get('/stops') +def list_stop_documents(keyword: Optional[str] = None, status: Optional[str] = None, + contract_id: Optional[str] = Query(None, alias='contractId'), + page: int = Query(1, ge=1), page_size: int = Query(20, alias='pageSize', ge=1, le=100), + db: Session = Depends(get_db), user: DbUserContext = Depends(current_user)): + return _wire(stop_documents.list_documents(db, user, keyword, status, contract_id, page, page_size)) + + +@legal_router.get('/stops/{plan_id}') +def get_stop_documents(plan_id: str, contract_id: Optional[str] = Query(None, alias='contractId'), + db: Session = Depends(get_db), user: DbUserContext = Depends(current_user)): + return _wire(stop_documents.get_documents(db, user, plan_id, contract_id)) + + +@legal_router.patch('/stops/{plan_id}') +@oper_log('逐段合法使用证明', business_type=2, path_param='plan_id') +def save_stop_documents(plan_id: str, body: StopLegalDocumentsBody, + contract_id: Optional[str] = Query(None, alias='contractId'), + db: Session = Depends(get_db), user: DbUserContext = Depends(current_user)): + return _write(db, lambda: stop_documents.save_documents(db, user, plan_id, body, contract_id)) + + @legal_router.get('/{plan_id}') def get_documents(plan_id: str, db: Session = Depends(get_db), user: DbUserContext = Depends(current_user)): return documents.get_documents(db, user, plan_id) diff --git a/backend/app/models/pl_quota_models.py b/backend/app/models/pl_quota_models.py index 2bdb18f57..8e838a04e 100644 --- a/backend/app/models/pl_quota_models.py +++ b/backend/app/models/pl_quota_models.py @@ -299,3 +299,27 @@ class PlCertOperation(Base): __table_args__ = ( UniqueConstraint('tenant_no', 'operation_id', name='uk_cert_request'), ) + + +class PlStopLegalDocument(Base): + """逐段产品证明；稳定身份不随附件更新重建。""" + __tablename__ = 'pl_stop_legal_document' + id = Column(String(64), primary_key=True) + plan_id = Column(String(64), nullable=False) + stop_type = Column(String(16), nullable=False) + stop_product_id = Column(String(64), nullable=False) + legal_use_cert_url = Column(Text, nullable=True) + uploaded_by = Column(String(64), nullable=True) + uploaded_time = Column(DateTime, nullable=True) + tenant_no = Column(String(64), nullable=False) + order_num = Column(Integer, nullable=True) + remark = Column(String(1000), nullable=True) + create_by = Column(String(64), nullable=True) + create_dept_id = Column(String(64), nullable=False, default='') + create_time = Column(DateTime, nullable=False, default=datetime.now) + update_by = Column(String(64), nullable=True) + update_time = Column(DateTime, nullable=False, default=datetime.now, onupdate=datetime.now) + del_flag = Column(SmallInteger, nullable=False, default=0) + __table_args__ = ( + UniqueConstraint('tenant_no', 'plan_id', 'stop_type', 'stop_product_id', name='uk_stop_legal_document'), + ) diff --git a/backend/app/schemas/pl/legal_use_document.py b/backend/app/schemas/pl/legal_use_document.py index c898ed481..5df726986 100644 --- a/backend/app/schemas/pl/legal_use_document.py +++ b/backend/app/schemas/pl/legal_use_document.py @@ -1,4 +1,4 @@ -from typing import List, Optional +from typing import List, Optional, Literal from pydantic import Field from app.schemas.pl.quota_cert import InputBody @@ -10,3 +10,13 @@ class LegalDocumentProduct(InputBody): class LegalDocumentsBody(InputBody): products: Optional[List[LegalDocumentProduct]] = None + + +class StopLegalDocument(InputBody): + stop_type: Literal['load', 'delivery'] + stop_product_id: str = Field(min_length=1, max_length=64) + legal_use_cert_url: Optional[str] = Field(None, max_length=2000) + + +class StopLegalDocumentsBody(InputBody): + items: Optional[List[StopLegalDocument]] = None diff --git a/backend/app/services/pl_child_references.py b/backend/app/services/pl_child_references.py index 46dc768c6..fab64a76b 100644 --- a/backend/app/services/pl_child_references.py +++ b/backend/app/services/pl_child_references.py @@ -15,6 +15,13 @@ def assert_stop_products_removable(db, tenant_no, plan_id, stop_type, stop_id, r if not rows: return ids = {row.id for row in rows} + from app.models.pl_quota_models import PlStopLegalDocument + if inspect(db.connection()).has_table(PlStopLegalDocument.__tablename__): + if _exists(db, PlStopLegalDocument, tenant_no, + PlStopLegalDocument.plan_id == plan_id, PlStopLegalDocument.stop_type == stop_type, + PlStopLegalDocument.stop_product_id.in_(ids), + PlStopLegalDocument.legal_use_cert_url.isnot(None), PlStopLegalDocument.legal_use_cert_url != ''): + raise HTTPException(409, '段产品已有合法使用证明，不能移除或改变身份') if any(any(getattr(row, field, None) is not None for field in ('inbound_qty', 'inbound_price', 'inbound_amount', 'outbound_qty', 'outbound_price', 'outbound_amount')) or bool(getattr(row, 'archive_remark', None)) for row in rows): diff --git a/backend/app/services/pl_plan_stop_service.py b/backend/app/services/pl_plan_stop_service.py index 65a0cade2..6e6211251 100644 --- a/backend/app/services/pl_plan_stop_service.py +++ b/backend/app/services/pl_plan_stop_service.py @@ -1361,7 +1361,10 @@ def prepare_plan_stops_write( if matches: row = matches[0] ids.add(row.id) - if row and (row.plan_product_id != linked or row.product_id != product.product_id): + if row and (row.plan_product_id != linked or row.product_id != product.product_id + or any((getattr(row, field) or '') != (getattr(product, field) or '') + for field in ('contract_id', 'contract_product_id') + if field in product.model_fields_set)): assert_stop_products_removable(db, tenant_no, plan_id, kind, stop_id, [row]) product_jobs.append((row, product, linked)) removed = [row for row in old if row.id not in ids] diff --git a/backend/app/services/pl_stop_legal_document.py b/backend/app/services/pl_stop_legal_document.py new file mode 100644 index 000000000..8e44f1420 --- /dev/null +++ b/backend/app/services/pl_stop_legal_document.py @@ -0,0 +1,185 @@ +"""逐装送段产品维护证明；权限筛选先于汇总、分页，不参与额度或归档。""" +from datetime import datetime +from uuid import uuid4 + +from fastapi import HTTPException +from pydantic import BaseModel +from sqlalchemy import and_, or_, func, case + +from app.crud.crm.data_scope_helpers import apply_crm_owner_scope +from app.models.crm_models import CrmContract +from app.models.pl_models import PlPlan, PlPlanProduct, PlPlanLoad, PlPlanLoadProduct, PlPlanDelivery, PlPlanDeliveryProduct +from app.models.pl_quota_models import PlStopLegalDocument as Document +from app.schemas.pl.legal_use_document import StopLegalDocumentsBody +from app.services.pl_data_scope import apply_pl_scope +from app.services.pl_legal_use_document import _plan +from app.services.pl_quota_contract_service import _contract +from app.services.pl_plan_cert_perm import user_can_upload_plan_cert +from app.services.pl_quota_attachments import authorize_attachment, storage_base_url + +SIDES = (('load', PlPlanLoadProduct, PlPlanLoad, 'load_id'), + ('delivery', PlPlanDeliveryProduct, PlPlanDelivery, 'delivery_id')) + + +def _query(db, user, side, model, parent, field, contract_id=None): + contracts = db.query(CrmContract.id).filter_by(tenant_no=user.tenant_no, del_flag=0) + contracts = apply_crm_owner_scope(contracts, CrmContract, user.to_user_info(), db, user.tenant_no) + query = db.query(model, parent, PlPlanProduct, CrmContract, Document).join( + parent, and_(getattr(model, field) == parent.id, model.plan_id == parent.plan_id, + parent.tenant_no == user.tenant_no, parent.del_flag == 0)).join( + PlPlanProduct, and_(model.plan_product_id == PlPlanProduct.id, model.plan_id == PlPlanProduct.plan_id, + model.product_id == PlPlanProduct.product_id, + PlPlanProduct.tenant_no == user.tenant_no, PlPlanProduct.del_flag == 0, + PlPlanProduct.is_explosive == 1)).outerjoin( + CrmContract, and_(model.contract_id == CrmContract.id, CrmContract.tenant_no == user.tenant_no, + CrmContract.del_flag == 0)).outerjoin( + Document, and_(Document.plan_id == model.plan_id, Document.stop_type == side, + Document.stop_product_id == model.id, Document.tenant_no == user.tenant_no, + Document.del_flag == 0)).filter( + model.tenant_no == user.tenant_no, model.del_flag == 0, + or_(model.contract_id.is_(None), model.contract_id == '', model.contract_id.in_(contracts))) + if contract_id: + query = query.filter(model.contract_id == contract_id) + return query + + +def _unlinked_legacy(db, user): + """无有效段的旧附件只在全局计划范围展示，不猜测合同归属。""" + query = db.query(PlPlanProduct).filter_by(tenant_no=user.tenant_no, del_flag=0, is_explosive=1).filter( + PlPlanProduct.legal_use_cert_url.isnot(None), PlPlanProduct.legal_use_cert_url != '') + for _, model, parent, field in SIDES: + linked = db.query(model.id).join(parent, getattr(model, field) == parent.id).filter( + model.plan_product_id == PlPlanProduct.id, model.plan_id == PlPlanProduct.plan_id, + model.tenant_no == user.tenant_no, model.del_flag == 0, + parent.tenant_no == user.tenant_no, parent.del_flag == 0, parent.plan_id == model.plan_id) + query = query.filter(~linked.exists()) + return query + + +def _legacy_row(product): + return {'id': product.id, 'productId': product.product_id, 'productName': product.product_name, + 'model': product.model, 'legalUseCertUrl': product.legal_use_cert_url} + + +def _details(db, user, plans, contract_id=None, lock=False): + result = {p.id: {'id': p.id, 'planNo': p.plan_no, 'planDate': p.plan_date, + 'licensePlate': p.license_plate, 'transportStatus': p.transport_status, + 'archiveStatus': p.archive_status, 'items': [], 'legacyProducts': []} for p in plans} + legacy = {p.id: {} for p in plans} + for side, model, parent, field in SIDES: + query = _query(db, user, side, model, parent, field, contract_id).filter(model.plan_id.in_(result)) + if lock: + query = query.with_for_update().populate_existing() + for row, stop, product, contract, document in query.order_by(parent.order_num, parent.id, model.order_num, model.id).all(): + url = document.legal_use_cert_url if document else None + result[row.plan_id]['items'].append({ + 'stopType': side, 'stopId': stop.id, 'stopOrder': getattr(stop, 'seq', None) or stop.order_num or 1, + 'stopProductId': row.id, 'productId': row.product_id, 'productName': row.product_name, + 'model': row.model, 'unit': row.unit, 'contractId': row.contract_id, + 'contractNo': contract.contract_no if contract else None, 'contractProductId': row.contract_product_id, + 'partyName': stop.customer_name, 'address': stop.address, 'legalUseCertUrl': url, + 'documentStatus': 'uploaded' if url else 'pending', + 'uploadedBy': document.uploaded_by if document else None, + 'uploadedTime': document.uploaded_time if document else None}) + if product.legal_use_cert_url: + legacy[row.plan_id][product.id] = _legacy_row(product) + if not contract_id: + for product in _unlinked_legacy(db, user).filter(PlPlanProduct.plan_id.in_(result)).all(): + legacy[product.plan_id][product.id] = _legacy_row(product) + for pid, value in result.items(): + value['requiredCount'] = len(value['items']) + value['uploadedCount'] = sum(r['documentStatus'] == 'uploaded' for r in value['items']) + value['legacyProducts'] = list(legacy[pid].values()) + value['documentStatus'] = ('not_required' if not value['requiredCount'] else + 'uploaded' if value['uploadedCount'] == value['requiredCount'] else 'pending') + return result + + +def get_documents(db, user, plan_id, contract_id=None): + plan = _plan(db, user, plan_id) + if contract_id: + _contract(db, user, contract_id) + return _details(db, user, [plan], contract_id)[plan_id] + + +def list_documents(db, user, keyword=None, status=None, contract_id=None, page=1, page_size=20): + if status not in (None, 'pending', 'uploaded') or page < 1 or not 1 <= page_size <= 100: + raise HTTPException(400, '无效资料状态或分页参数') + if contract_id: + _contract(db, user, contract_id) + queries = [_query(db, user, side, model, parent, field, contract_id).with_entities( + model.plan_id.label('plan_id'), case((and_(Document.legal_use_cert_url.isnot(None), + Document.legal_use_cert_url != ''), 1), else_=0).label('uploaded')) + for side, model, parent, field in SIDES] + sources = queries[0].union_all(queries[1]).subquery() + counts = db.query(sources.c.plan_id, func.count().label('required'), + func.sum(sources.c.uploaded).label('uploaded')).group_by(sources.c.plan_id).subquery() + query = db.query(PlPlan).outerjoin(counts, counts.c.plan_id == PlPlan.id).filter( + PlPlan.tenant_no == user.tenant_no, PlPlan.del_flag == 0) + if contract_id: + query = query.filter(counts.c.required > 0) + else: + query = query.filter(or_(counts.c.required > 0, PlPlan.id.in_( + _unlinked_legacy(db, user).with_entities(PlPlanProduct.plan_id)))) + query = apply_pl_scope(query, PlPlan, user, db, user.tenant_no) + if keyword: + query = query.filter(PlPlan.plan_no.contains(keyword)) + if status == 'pending': + query = query.filter(counts.c.uploaded < counts.c.required) + elif status == 'uploaded': + query = query.filter(counts.c.uploaded == counts.c.required) + total = query.count() + plans = query.order_by(PlPlan.create_time.desc(), PlPlan.id).offset((page - 1) * page_size).limit(page_size).all() + details = _details(db, user, plans, contract_id) if plans else {} + return {'list': [details[p.id] for p in plans], 'total': total, 'page': page, 'pageSize': page_size} + + +def save_documents(db, user, plan_id, data, contract_id=None): + if not user_can_upload_plan_cert(db, user): + raise HTTPException(403, '无权维护合法使用证明') + body = StopLegalDocumentsBody(**(data.model_dump(exclude_unset=True) if isinstance(data, BaseModel) else data)) + plan = _plan(db, user, plan_id, lock=True) + if contract_id: + _contract(db, user, contract_id, lock=True) + if 'items' in body.model_fields_set and body.items is None: + raise HTTPException(400, '段产品集合不可为null') + details = _details(db, user, [plan], contract_id, lock=True)[plan_id] + valid = {(r['stopType'], r['stopProductId']) for r in details['items']} + # 包含软删除行以复用稳定唯一键；清空不会删除记录或创建信息。 + records = {(r.stop_type, r.stop_product_id): r for r in db.query(Document).filter_by( + tenant_no=user.tenant_no, plan_id=plan_id).with_for_update().populate_existing().all()} + seen, changes = set(), [] + for item in body.items or []: + key = (item.stop_type, item.stop_product_id) + if key in seen or key not in valid: + raise HTTPException(400, '段产品重复、不适用或不在可访问计划合同范围内') + seen.add(key) + if 'legal_use_cert_url' not in item.model_fields_set: + continue + row = records.get(key) + old = row.legal_use_cert_url if row and row.del_flag == 0 else '' + urls = list(dict.fromkeys(v.strip() for v in (item.legal_use_cert_url or '').split(',') if v.strip())) + for url in urls: + if url not in (old or '').split(','): + authorize_attachment(db, user, url, storage_base_url()) + value = ','.join(urls) + if value != (old or ''): + changes.append((key, row, value)) + for key, row, value in changes: + if row is None: + row = Document(id=str(uuid4()), tenant_no=user.tenant_no, plan_id=plan_id, + stop_type=key[0], stop_product_id=key[1], create_by=user.id, + create_dept_id=user.dept_id or '') + db.add(row) + row.del_flag = 0 + row.legal_use_cert_url = value + row.update_by = user.id + row.uploaded_by = user.id if value else None + row.uploaded_time = datetime.now() if value else None + if changes: + from app.services.pl_plan_service import record_plan_change + record_plan_change(db, tenant_no=user.tenant_no, plan_id=plan_id, change_type='legal_use_document', + change_content='维护逐段合法使用证明：' + '、'.join(k[0] + ':' + k[1] for k, _, _ in changes), + user_id=user.id, user_name=getattr(user, 'name', None), dept_id=user.dept_id or '') + db.flush() + return get_documents(db, user, plan_id, contract_id) diff --git a/backend/tests/test_pl_child_differential_save.py b/backend/tests/test_pl_child_differential_save.py index 00ce19a77..afb0eb23d 100644 --- a/backend/tests/test_pl_child_differential_save.py +++ b/backend/tests/test_pl_child_differential_save.py @@ -317,7 +317,7 @@ def test_update_api_partial_product_edit_retains_plan_total(db,monkeypatch): plan=m.PlPlan(id='p',tenant_no='t',plan_no='P',create_date=date(2026,10,27),belongs_company='shanxia',plan_type='inbound',total_amount=120) db.add(plan);db.flush() monkeypatch.setattr(api,'get_tenant_no',lambda *args:'t') - monkeypatch.setattr(api,'_get_plan',lambda *args:plan) + monkeypatch.setattr(api,'_get_plan',lambda *args,**kwargs:plan) monkeypatch.setattr(api,'collect_hazard_product_ids',lambda *args:set()) monkeypatch.setattr(api,'resolve_hazard_flags_from_crm',lambda *args,**kwargs:(0,0)) monkeypatch.setattr(api,'sync_linked_dibang_dates_for_plan',lambda *args,**kwargs:[]) diff --git a/backend/tests/test_pl_stop_legal_document.py b/backend/tests/test_pl_stop_legal_document.py new file mode 100644 index 000000000..5c9ba133b --- /dev/null +++ b/backend/tests/test_pl_stop_legal_document.py @@ -0,0 +1,214 @@ +from datetime import date +from types import SimpleNamespace + +import pytest +from fastapi import HTTPException +from sqlalchemy import create_engine +from sqlalchemy.orm import Session + + +@pytest.fixture +def case(monkeypatch): + from app.models import pl_models as m + from app.models.crm_models import CrmContract, SysOss + from app.models.pl_quota_models import PlStopLegalDocument + from app.services import pl_stop_legal_document as svc + engine = create_engine('sqlite://') + for model in (m.PlPlan, m.PlPlanProduct, m.PlPlanLoad, m.PlPlanLoadProduct, + m.PlPlanDelivery, m.PlPlanDeliveryProduct, m.PlPlanChange, + m.PlPlanContract, m.PlPlanProductBrandQty, m.PlTransportDriverStopFeedback, + m.PlTransportSiteStopFeedback, m.PlTransportDbLink, CrmContract, SysOss, PlStopLegalDocument): + model.__table__.create(engine) + db = Session(engine) + user = SimpleNamespace(id='u', name='测试', dept_id='d', tenant_no='t', role='super_admin', + to_user_info=lambda: {'role': 'super_admin', 'user_id': 'u'}) + db.add(m.PlPlan(id='p', plan_no='P', create_date=date.today(), belongs_company='shanxia', + plan_type='direct', tenant_no='t', transport_status='arranged', archive_status='archived')) + db.add(m.PlPlanProduct(id='pp', plan_id='p', product_id='sku', product_no='S', product_name='产品', + is_explosive=1, legal_use_cert_url='legacy.jpg', tenant_no='t')) + db.add(CrmContract(id='c', contract_no='C', contract_name='合同', customer_id='customer', contract_type='sale', tenant_no='t')) + for side, rid, cid in [('load', 'l', None), ('delivery', 'd1', 'c'), ('delivery', 'd2', None)]: + parent, product = (m.PlPlanLoad, m.PlPlanLoadProduct) if side == 'load' else (m.PlPlanDelivery, m.PlPlanDeliveryProduct) + db.add(parent(id=rid, plan_id='p', tenant_no='t', customer_name=rid)) + db.add(product(id=rid+'p', plan_id='p', plan_product_id='pp', product_id='sku', product_no='S', + product_name='产品', tenant_no='t', contract_id=cid, **{side+'_id': rid})) + db.add(SysOss(id='o', url='legal/new.jpg', tenant_no='t', create_by='u')) + db.commit() + monkeypatch.setattr(svc, 'storage_base_url', lambda: 'https://files.example/bucket') + yield SimpleNamespace(db=db, user=user, svc=svc, m=m, doc=PlStopLegalDocument) + db.close() + engine.dispose() + + +def patch(c, items, **kwargs): + return c.svc.save_documents(c.db, c.user, 'p', {'items': items}, **kwargs) + + +def item(rid='d1p', url='legal/new.jpg', side='delivery'): + return {'stopType': side, 'stopProductId': rid, 'legalUseCertUrl': url} + + +def test_each_stop_counts_independently_and_legacy_is_separate(case): + c = case + result = patch(c, [item()]) + assert (result['uploadedCount'], result['requiredCount']) == (1, 3) + assert result['legacyProducts'][0]['legalUseCertUrl'] == 'legacy.jpg' + scoped = c.svc.get_documents(c.db, c.user, 'p', contract_id='c') + assert (scoped['uploadedCount'], scoped['requiredCount']) == (1, 1) + assert scoped['items'][0]['contractNo'] == 'C' + assert c.db.get(c.m.PlPlan, 'p').archive_status == 'archived' + + +def test_patch_identity_omissions_and_single_clear(case): + c = case + patch(c, [item()]) + row = c.db.query(c.doc).one() + original = (row.id, row.create_by, row.create_time) + for body in ({}, {'items': []}, {'items': [{'stopType': 'delivery', 'stopProductId': 'd1p'}]}): + assert c.svc.save_documents(c.db, c.user, 'p', body)['uploadedCount'] == 1 + patch(c, [item(url='')]) + assert (row.id, row.create_by, row.create_time) == original + assert c.svc.get_documents(c.db, c.user, 'p')['uploadedCount'] == 0 + + +@pytest.mark.parametrize('bad', [item(), item('missing'), item('lp'), item(url='https://evil/a'), item(url='../a')]) +def test_all_validation_precedes_write(case, bad): + c = case + with pytest.raises(HTTPException): + patch(c, [item(), bad]) + assert c.db.query(c.doc).count() == 0 + assert c.db.query(c.m.PlPlanChange).count() == 0 + + +def test_permissions_tenant_parent_and_contract_scope(case, monkeypatch): + c = case + with pytest.raises(HTTPException): + patch(c, [item('d2p')], contract_id='c') + other = SimpleNamespace(**{**c.user.__dict__, 'tenant_no': 'other'}) + with pytest.raises(HTTPException): + c.svc.get_documents(c.db, other, 'p') + c.db.get(c.m.PlPlanDelivery, 'd1').plan_id = 'other' + c.db.flush() + with pytest.raises(HTTPException): + patch(c, [item()]) + monkeypatch.setattr(c.svc, 'user_can_upload_plan_cert', lambda *a: False) + with pytest.raises(HTTPException) as error: + patch(c, []) + assert error.value.status_code == 403 + + +def test_filtering_precedes_pagination(case): + c = case + patch(c, [item()]) + assert c.svc.list_documents(c.db, c.user, status='uploaded')['total'] == 0 + result = c.svc.list_documents(c.db, c.user, status='uploaded', contract_id='c', page=1, page_size=1) + assert result['total'] == 1 and len(result['list']) == 1 + assert result['pageSize'] == 1 + assert c.svc.list_documents(c.db, c.user, keyword='missing')['total'] == 0 + + +def test_document_protects_stop_product_until_cleared(case): + from app.services.pl_child_references import assert_stop_products_removable + c = case + patch(c, [item()]) + row = c.db.get(c.m.PlPlanDeliveryProduct, 'd1p') + with pytest.raises(HTTPException) as error: + assert_stop_products_removable(c.db, 't', 'p', 'delivery', 'd1', [row]) + assert error.value.status_code == 409 + patch(c, [item(url='')]) + assert_stop_products_removable(c.db, 't', 'p', 'delivery', 'd1', [row]) + + +def test_static_routes_registered_before_legacy_dynamic_route(): + from app.api.pl.quota_cert import legal_router + routes = [(r.path, r.methods) for r in legal_router.routes] + base = '/transport/legal-use-documents' + assert (base + '/stops', {'GET'}) in routes + assert (base + '/stops/{plan_id}', {'PATCH'}) in routes + assert next(i for i, r in enumerate(routes) if r[0] == base+'/stops') < next(i for i, r in enumerate(routes) if r[0] == base+'/{plan_id}') + + +def test_hidden_contract_rows_do_not_leak_and_null_is_rejected(case, monkeypatch): + c = case + monkeypatch.setattr(c.svc, 'apply_crm_owner_scope', lambda query, model, *a: query.filter(model.id != 'c')) + result = c.svc.get_documents(c.db, c.user, 'p') + assert result['requiredCount'] == 2 + assert all(r['contractId'] != 'c' for r in result['items']) + assert c.svc.list_documents(c.db, c.user)['list'][0]['requiredCount'] == 2 + with pytest.raises(HTTPException): + patch(c, [item()]) + with pytest.raises(HTTPException): + patch(c, None) + + +@pytest.mark.parametrize('field,value', [('contractId', None), ('contractProductId', 'new')]) +def test_contract_identity_change_protected(case, field, value): + from app.schemas.pl.plan import PlanDeliveryBody + from app.services.pl_plan_stop_service import prepare_plan_stops_write + c = case + patch(c, [item()]) + body = PlanDeliveryBody(id='d1', products=[{'id': 'd1p', 'planProductId': 'pp', 'productId': 'sku', + 'productNo': 'S', 'productName': '产品', field: value}]) + with pytest.raises(HTTPException) as error: + prepare_plan_stops_write(c.db, 't', 'p', None, [body], [c.db.get(c.m.PlPlanProduct, 'pp')]) + assert error.value.status_code == 409 + + +def test_legacy_only_plan_is_visible_but_not_uploaded(case): + c = case + c.db.query(c.m.PlPlanLoadProduct).update({'del_flag': 1}) + c.db.query(c.m.PlPlanDeliveryProduct).update({'del_flag': 1}) + c.db.flush() + result = c.svc.get_documents(c.db, c.user, 'p') + assert result['legacyProducts'][0]['legalUseCertUrl'] == 'legacy.jpg' + assert result['requiredCount'] == 0 and result['documentStatus'] == 'not_required' + assert c.svc.list_documents(c.db, c.user)['total'] == 1 + assert c.svc.list_documents(c.db, c.user, status='uploaded')['total'] == 0 + + +def test_multiple_plans_filter_then_page_and_route_shape(case): + from app.api.pl.quota_cert import list_stop_documents + c = case + patch(c, [item()]) + for i in range(3): + pid = 'p' + str(i) + c.db.add(c.m.PlPlan(id=pid, plan_no='P'+str(i), create_date=date.today(), belongs_company='shanxia', + plan_type='direct', tenant_no='t')) + c.db.add(c.m.PlPlanProduct(id=pid+'pp', plan_id=pid, product_id='sku', product_no='S', + product_name='产品', is_explosive=1, tenant_no='t')) + c.db.add(c.m.PlPlanDelivery(id=pid+'d', plan_id=pid, tenant_no='t')) + c.db.add(c.m.PlPlanDeliveryProduct(id=pid+'dp', delivery_id=pid+'d', plan_id=pid, + plan_product_id=pid+'pp', product_id='sku', product_no='S', product_name='产品', tenant_no='t')) + c.db.flush() + first = list_stop_documents(keyword=None, status='pending', contract_id=None, page=1, page_size=2, db=c.db, user=c.user) + second = c.svc.list_documents(c.db, c.user, status='pending', page=2, page_size=2) + assert first['total'] == second['total'] == 4 + assert len(first['list']) == len(second['list']) == 2 + assert {r['id'] for r in first['list']}.isdisjoint(r['id'] for r in second['list']) + assert {'uploadedCount', 'requiredCount', 'items', 'legacyProducts', 'planNo'} <= first['list'][0].keys() + + +def test_foreign_and_deleted_stop_products_rejected_before_write(case): + c = case + row = c.db.get(c.m.PlPlanDeliveryProduct, 'd2p') + for field, value in [('tenant_no', 'other'), ('del_flag', 1)]: + row.tenant_no, row.del_flag = 't', 0 + setattr(row, field, value) + c.db.flush() + with pytest.raises(HTTPException): + patch(c, [item(), item('d2p')]) + assert c.db.query(c.doc).count() == 0 + + +def test_soft_deleted_document_not_counted_or_protecting_and_reuses_identity(case): + from app.services.pl_child_references import assert_stop_products_removable + c = case + patch(c, [item()]) + row = c.db.query(c.doc).one() + identity = row.id, row.create_time + row.del_flag = 1 + c.db.flush() + assert c.svc.get_documents(c.db, c.user, 'p')['uploadedCount'] == 0 + assert_stop_products_removable(c.db, 't', 'p', 'delivery', 'd1', [c.db.get(c.m.PlPlanDeliveryProduct, 'd1p')]) + patch(c, [item()]) + assert (row.id, row.create_time) == identity and row.del_flag == 0 diff --git a/docs/sql/2026-11-01-pl-stop-legal-document.sql b/docs/sql/2026-11-01-pl-stop-legal-document.sql new file mode 100644 index 000000000..9c360c66f --- /dev/null +++ b/docs/sql/2026-11-01-pl-stop-legal-document.sql @@ -0,0 +1,21 @@ +-- 逐段产品合法使用证明；仅定义迁移，部署时执行。旧计划产品附件不自动迁移。 +CREATE TABLE IF NOT EXISTS `pl_stop_legal_document` ( + `id` VARCHAR(64) NOT NULL COMMENT '主键UUID', + `plan_id` VARCHAR(64) NOT NULL COMMENT '运输计划ID', + `stop_type` VARCHAR(16) NOT NULL COMMENT 'load装货段/delivery送货段', + `stop_product_id` VARCHAR(64) NOT NULL COMMENT '对应段产品明细ID', + `legal_use_cert_url` TEXT NULL COMMENT 'MinIO附件，多附件逗号分隔', + `uploaded_by` VARCHAR(64) NULL COMMENT '最近上传人', + `uploaded_time` DATETIME NULL COMMENT '最近上传时间', + `tenant_no` VARCHAR(64) NOT NULL COMMENT '租户编号', + `order_num` INT NULL COMMENT '排序号', + `remark` VARCHAR(1000) NULL COMMENT '备注', + `create_by` VARCHAR(64) NULL COMMENT '创建人', + `create_dept_id` VARCHAR(64) NOT NULL DEFAULT '' COMMENT '创建部门', + `create_time` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间', + `update_by` VARCHAR(64) NULL COMMENT '更新人', + `update_time` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间', + `del_flag` SMALLINT NOT NULL DEFAULT 0 COMMENT '0正常1删除', + PRIMARY KEY (`id`), + UNIQUE KEY `uk_stop_legal_document` (`tenant_no`, `plan_id`, `stop_type`, `stop_product_id`) +) ENGINE=InnoDB ROW_FORMAT=DYNAMIC DEFAULT CHARSET=utf8mb4 COMMENT='逐段产品合法使用证明';