diff --git a/backend/app/api/v1/business/businesses.py b/backend/app/api/v1/business/businesses.py index 915cdb0..7bdaeb0 100644 --- a/backend/app/api/v1/business/businesses.py +++ b/backend/app/api/v1/business/businesses.py @@ -25,6 +25,7 @@ from app.schemas.business import ( OwnershipChangeLogOut, ) from app.schemas.business_source import ( + GuidanceConfirmIn, SalesRelationConfirmIn, SalesRelationQuery, SalesRelationReleaseIn, @@ -72,6 +73,17 @@ def confirm_source_price(session: DbSession, actor: PayrollWriter, business_id: return ok(business_source.confirm_missing_source_price(session, business_id, document_id, payload, actor, meta)) +@router.put("/{business_id}/guidance") +def confirm_guidance(session: DbSession, actor: PayrollWriter, business_id: Annotated[int, Path(ge=1)], + payload: GuidanceConfirmIn, meta: RequestMetaDep): + """从本地发货明细登记来源产品的发货月份指导成本。""" + from app.services.authorization_service import require_permission + + require_permission(session, actor, "payroll:prepare") + require_permission(session, actor, "tplus:facts") + return ok(business_source.confirm_guidance(session, business_id, payload, actor, meta)) + + @router.get("/{business_id}/sales-relation-candidates") def sales_relation_candidates(session: DbSession, _: CurrentUser, business_id: Annotated[int, Path(ge=1)], query: Annotated[SalesRelationQuery, Query()]): diff --git a/backend/app/core/config.py b/backend/app/core/config.py index 281f184..d4f2a95 100644 --- a/backend/app/core/config.py +++ b/backend/app/core/config.py @@ -89,6 +89,7 @@ class Settings(BaseSettings): t_plus_user: str = "" t_plus_password: str = Field(default="", repr=False) t_plus_allow_dev_sa: bool = False + t_plus_allow_admin_account: bool = False t_plus_connection_file: str = "" # 部署前必须指定已在 SQL Server 2008 R2 实测的驱动，不自动选择最新版。 t_plus_odbc_driver: str = "" @@ -102,6 +103,7 @@ class Settings(BaseSettings): t_plus_schedule_weekday: int = Field(default=0, ge=0, le=6) t_plus_schedule_time: str = Field(default="02:00", pattern=r"^([01][0-9]|2[0-3]):[0-5][0-9]$") t_plus_stop_file: str = "" + t_plus_supervisor_health_url: str = "http://127.0.0.1:8000/api/health" @field_validator("app_secret_key") @classmethod diff --git a/backend/app/core/tplus_db.py b/backend/app/core/tplus_db.py index b7770be..28238ce 100644 --- a/backend/app/core/tplus_db.py +++ b/backend/app/core/tplus_db.py @@ -56,8 +56,10 @@ def resolve_tplus_settings(settings: Settings) -> Settings: ) resolved = settings.model_copy(update=values) allow_dev_sa = settings.app_env.strip().casefold() == "dev" and settings.t_plus_allow_dev_sa - if resolved.t_plus_user.strip().casefold() == "sa" and not allow_dev_sa: - raise RuntimeError("应用必须使用 T+ 专用只读账号，不能使用 sa") + if resolved.t_plus_user.strip().casefold() == "sa" and not ( + allow_dev_sa or settings.t_plus_allow_admin_account + ): + raise RuntimeError("T+ 管理员账号未获授权，请检查连接配置") if not resolved.t_plus_database or not resolved.t_plus_user or not resolved.t_plus_odbc_driver: raise RuntimeError("T+ 只读连接缺少数据库、账号或已验证的 ODBC 驱动") return resolved diff --git a/backend/app/models/__init__.py b/backend/app/models/__init__.py index 3788802..6f79907 100644 --- a/backend/app/models/__init__.py +++ b/backend/app/models/__init__.py @@ -31,7 +31,12 @@ from app.models.role import Permission, Role, RolePermission from app.models.settlement import AdjustmentOrder, MonthlySettlement, SettlementItem from app.models.shipment import Shipment, ShipmentItem from app.models.system_setting import SystemSetting -from app.models.tplus_automation import TPlusAutomationConfig, TPlusAutomationState +from app.models.tplus_automation import ( + TPlusAutomationConfig, + TPlusAutomationState, + TPlusRuntimeControl, +) +from app.models.tplus_business_sync import TPlusBusinessDocument, TPlusBusinessSyncConfig from app.models.tplus_sync import ( TPlusSyncBatch, TPlusSyncError, @@ -47,11 +52,7 @@ from app.models.tplus_workbench import ( ) from app.models.user import User, UserRole -from app.models.tplus_business_sync import TPlusBusinessDocument, TPlusBusinessSyncConfig - __all__ = [ - "TPlusBusinessDocument", - "TPlusBusinessSyncConfig", "AdjustmentOrder", "ApprovalOrder", "ApprovalStep", @@ -95,7 +96,10 @@ __all__ = [ "SystemSetting", "TPlusAutomationConfig", "TPlusAutomationState", + "TPlusBusinessDocument", + "TPlusBusinessSyncConfig", "TPlusFact", + "TPlusRuntimeControl", "TPlusSourceHead", "TPlusSourceLink", "TPlusSyncBatch", diff --git a/backend/app/schemas/business_source.py b/backend/app/schemas/business_source.py index 94b8f61..eee856d 100644 --- a/backend/app/schemas/business_source.py +++ b/backend/app/schemas/business_source.py @@ -1,5 +1,6 @@ """来源业务的人工分类资料。""" +from decimal import Decimal, InvalidOperation from typing import Literal from pydantic import BaseModel, ConfigDict, Field, model_validator @@ -29,6 +30,28 @@ class SourcePriceConfirmIn(BaseModel): return self +class GuidanceConfirmIn(BaseModel): + """从业务发货明细登记对应发货月份的单位指导成本。""" + + model_config = ConfigDict(extra="forbid") + shipment_item_id: int = Field(ge=1) + amount: str = Field(min_length=1, max_length=32) + evidence: str = Field(min_length=1, max_length=500) + + @model_validator(mode="after") + def validate_values(self): + self.evidence = self.evidence.strip() + if not self.evidence: + raise ValueError("核对依据不能为空") + try: + value = Decimal(self.amount) + except (InvalidOperation, TypeError, ValueError): + raise ValueError("指导成本必须为有效数字") from None + if not value.is_finite() or value < 0 or value.as_tuple().exponent < -4: + raise ValueError("指导成本最多保留4位小数且不能为负数") + return self + + class SourceProductCategory(BaseModel): product_id: int = Field(ge=1) category: Literal["municipal_water", "rail_transit"] diff --git a/backend/app/services/business_source.py b/backend/app/services/business_source.py index 1191597..7e821b4 100644 --- a/backend/app/services/business_source.py +++ b/backend/app/services/business_source.py @@ -23,6 +23,7 @@ from app.models import ( ShipmentItem, TPlusBusinessDocument, TPlusBusinessSyncConfig, + TPlusFact, TPlusSourceHead, TPlusSyncRecord, TPlusWorkRun, @@ -631,7 +632,7 @@ def confirm_missing_source_price(session, business_id, document_id, payload, ope raise BadRequestError("来源单价状态已变化，请刷新后核对") target.sale_price = decimal_value(raw["price"], 4) session.flush() - ctx = Context(session, source) + ctx = Context(session, source, operator) action, checked = _apply_document(ctx, parent_type, head, details(ctx, parent_type, parent.source_id), "MANUAL", START) if action not in ("UNCHANGED", "CONFLICT") or (action == "CONFLICT" and checked.reason != "SOURCE_PRICE_REVIEW_REQUIRED"): raise BadRequestError("来源单据已变化，请刷新后核对") @@ -642,6 +643,156 @@ def confirm_missing_source_price(session, business_id, document_id, payload, ope return result +def confirm_guidance(session, business_id, payload, operator, meta=None): + """登记并封存一个来源产品发货月份的单位指导成本。""" + from app.services.audit_service import AuditRecord, audit + from app.services.tplus.business_sync import statement + from app.services.tplus.guidance import ( + check_unit_guidance_conflicts, + frozen_guidance_shipments, + ) + + with source_calculation_lock(session): + audit_record = AuditRecord( + module="业务管理", + action="登记月度指导成本", + summary=f"业务 #{business_id} 发货明细 #{payload.shipment_item_id}", + request_data=payload.model_dump(mode="json"), + ) + with audit(session, operator=operator, meta=meta, record=audit_record): + row = session.get(Business, business_id, with_for_update=True) + if row is None or not row.integration_json: + raise BadRequestError("仅可为自动接入业务登记指导成本") + source = row.integration_json.get("source_database") + if not source or not source_snapshot_ready(session, source): + raise BadRequestError("来源快照或业务接入未完成，请稍后重试") + if _latest(session, business_id) is not None: + raise BadRequestError("业务已有封存核算资料，请通过差异核对处理") + + item = session.scalar( + select(ShipmentItem) + .join(Shipment, Shipment.id == ShipmentItem.shipment_id) + .where(ShipmentItem.id == payload.shipment_item_id, Shipment.business_id == business_id) + .with_for_update() + ) + if item is None or not item.unit: + raise BadRequestError("发货明细不存在或缺少计量单位") + shipment = session.get(Shipment, item.shipment_id, with_for_update=True) + if shipment is None or shipment.business_id != business_id: + raise BadRequestError("发货单不存在或不属于该业务") + documents = list(session.scalars( + select(TPlusBusinessDocument) + .where( + TPlusBusinessDocument.business_id == business_id, + TPlusBusinessDocument.source_database == source, + TPlusBusinessDocument.entity_type == "shipment_items", + TPlusBusinessDocument.target_id == item.id, + TPlusBusinessDocument.status == "APPLIED", + TPlusBusinessDocument.applied_record_id.is_not(None), + ) + )) + if not documents: + raise BadRequestError("发货明细尚未建立来源关联") + if len(documents) != 1: + raise BadRequestError("发货明细来源关联不唯一，请先核对") + document = documents[0] + parent_id = document.data_json.get("source_parent_id") + parents = list(session.scalars( + select(TPlusBusinessDocument).where( + TPlusBusinessDocument.business_id == business_id, + TPlusBusinessDocument.source_database == source, + TPlusBusinessDocument.entity_type == "shipments", + TPlusBusinessDocument.source_id == str(parent_id), + TPlusBusinessDocument.status == "APPLIED", + TPlusBusinessDocument.applied_record_id.is_not(None), + ) + )) + if not parents: + raise BadRequestError("发货单尚未建立来源关联") + if len(parents) != 1: + raise BadRequestError("发货单来源关联不唯一，请先核对") + parent = parents[0] + current_line = session.scalar( + statement(session, source, "shipment_items").where( + TPlusSyncRecord.source_id == document.source_id + ) + ) + current_parent = session.scalar( + statement(session, source, "shipments").where( + TPlusSyncRecord.source_id == parent.source_id + ) + ) + if ( + current_line is None + or current_parent is None + or current_line.id != document.applied_record_id + or current_parent.id != parent.applied_record_id + ): + raise BadRequestError("来源单据已变化，请刷新后核对") + source_product_id = current_line.payload_json.get("idinventory") + if source_product_id is None: + raise BadRequestError("来源明细缺少产品身份，请先核对") + source_product_id = str(source_product_id).strip() + if not source_product_id.isascii() or not source_product_id.isdigit() or len(source_product_id) > 64: + raise BadRequestError("来源产品身份格式异常，请先核对") + source_date = current_parent.payload_json.get("voucherdate") or current_parent.payload_json.get("voucherDate") + if source_date is None or str(source_date)[:10] != shipment.ship_date.isoformat(): + raise BadRequestError("来源发货日期已变化，请刷新后核对") + month = shipment.ship_date.strftime("%Y-%m") + amount = format(Decimal(payload.amount), "f") + facts = latest_facts(session, source) + fact_key = f"GUIDANCE:{source_product_id}:{month}:UNIT" + existing = facts.get(("GUIDANCE", fact_key)) + if existing is not None: + existing_data = existing.data_json + if ( + existing_data.get("is_finalized") + and existing_data.get("unit") == item.unit + and Decimal(str(existing_data.get("amount"))) == Decimal(amount) + ): + return { + "status": "REUSED", + "fact_id": existing.id, + "product_id": str(source_product_id), + "month": month, + "unit": item.unit, + "amount": existing_data["amount"], + } + raise BadRequestError("同一产品同月已有不同指导成本，请先核对") + frozen = frozen_guidance_shipments(session, source, str(source_product_id), month) + check_unit_guidance_conflicts( + facts, str(source_product_id), month, item.unit, amount, frozen + ) + fact = TPlusFact( + source_database=source, + kind="GUIDANCE", + fact_key=fact_key, + version=1, + data_json={ + "product_id": str(source_product_id), + "month": month, + "basis": "UNIT", + "amount": amount, + "unit": item.unit, + "is_finalized": True, + }, + evidence=payload.evidence, + requested_by=operator.id, + ) + session.add(fact) + session.flush() + result = { + "status": "CREATED", + "fact_id": fact.id, + "product_id": str(source_product_id), + "month": month, + "unit": item.unit, + "amount": amount, + } + session.commit() + return result + + def save_materials(session, business_id, payload, actor, meta=None): """同一业务引用的档案才可批量维护；不接受名称、金额或来源字段。""" from app.services.audit_service import AuditRecord, audit @@ -780,7 +931,11 @@ def _prepare_locked(session, business_id, payload, operator, meta=None): submitted_cost = Decimal(entry.unit_cost) if guidance["basis"] == "UNIT" else cost_total if submitted_cost != Decimal(guidance["amount"]): raise BadRequestError("指导成本与已封存依据不一致，请先核对") - ship = {"source_id": str(head.id), "line_id": str(item.id), "source_no": head.source_no, + ship = { + "source_id": str(head.id), + "line_id": str(item.id), + "product_id": str(product_source_id) if product_source_id is not None else None, + "source_no": head.source_no, "date": head.ship_date.isoformat(), "group_key": key, "business_key": key, "personnel_id": str(entry.personnel_id), "manager_id": str(entry.manager_id) if entry.manager_id else None, "amount": str(item.amount - freight), "gross_tax_amount": str(item.amount), "freight_amount": str(freight), @@ -821,7 +976,8 @@ def _prepare_locked(session, business_id, payload, operator, meta=None): "amount": str(receipt.amount), "voucher_confirmed": confirmed, "exclusion": exclusion, "pending": [], "voucher_evidence": evidence}) from app.services.receipt_confirmation import ensure_confirmed - ensure_confirmed(session, [receipt.id for receipt, snapshot in zip(cash, receipts, strict=True) if not snapshot["exclusion"]]) + ensure_confirmed(session, [receipt.id for receipt, snapshot in zip(cash, receipts, strict=True) + if not snapshot["exclusion"] and receipt.receipt_date.strftime("%Y-%m") <= payload.month]) values = {"input_schema": "TPLUS_SYNC_V1", "group": {"key": key, "scope_confirmed": True, "opening_confirmed": True, "opening_balance": payload.opening_balance, "historical_settlement_confirmed": True}, "shipments": ships, "receipts": receipts, "facts": {"personnel": people_facts, "handovers": handovers}, diff --git a/backend/app/services/payroll_months.py b/backend/app/services/payroll_months.py index dcfb32b..1ca849d 100644 --- a/backend/app/services/payroll_months.py +++ b/backend/app/services/payroll_months.py @@ -29,10 +29,8 @@ def month_state(session, month): } -def lock_month(session, month, operator, *, require_maker=True): - """INSERT 的唯一键串行化首次建行，避免两个请求同时认为月份不存在。""" - if require_maker: - ensure_payroll_writer(operator) +def lock_month_row(session, month, operator): + """以月份唯一键串行化首次建行及后续操作，不改变已有制作人。""" validate_month(month) values = {"month": month, "maker_id": operator.id, "maker_name": operator.username} if session.bind.dialect.name == "mysql": @@ -43,7 +41,15 @@ def lock_month(session, month, operator, *, require_maker=True): session.flush() control = session.scalar( select(PayrollMonth).where(PayrollMonth.month == month).with_for_update() + .execution_options(populate_existing=True) ) + return control + + +def lock_month(session, month, operator, *, require_maker=True): + if require_maker: + ensure_payroll_writer(operator) + control = lock_month_row(session, month, operator) if control.generated_at is not None and control.maker_id != operator.id: raise ForbiddenError("该月工资由其他制作人生成，不可代为修改或重新生成") if control.generated_at is None: diff --git a/backend/app/services/tplus/business_sync.py b/backend/app/services/tplus/business_sync.py index 875625e..c0a544a 100644 --- a/backend/app/services/tplus/business_sync.py +++ b/backend/app/services/tplus/business_sync.py @@ -234,8 +234,9 @@ def preview_product_scope(session, source, start): class Context: - def __init__(self, session, source): + def __init__(self, session, source, operator=None): self.session, self.source = session, source + self.operator = operator self.facts = latest_facts(session, source) self.records, self.targets, self.recalculated = {}, {}, set() self.people = {r.source_id: r for r in session.scalars(statement(session, source, "personnel"))} @@ -556,16 +557,18 @@ def _apply_document(ctx, entity, record, children, run_no, start): customer_id = raw.get("idpartner") if entity == "receipts" else raw.get("idcustomer") if customer_id is None: raise WorkbenchError("GROUP_CUSTOMER_MISMATCH") + # 新月份可追加，但已生成工资月份的迟到单据必须另行核对。 + from app.models import MonthlySettlement + from app.services.payroll_months import lock_month_row + month = day.strftime("%Y-%m") + control = lock_month_row(ctx.session, month, ctx.operator) + locked = ctx.session.scalar(select(MonthlySettlement.id).where( + MonthlySettlement.settlement_month == month).limit(1).with_for_update()) + if locked or (control and control.generated_at): + raise WorkbenchError("PAYROLL_MONTH_LOCKED") business = ctx.business(customer_id, day, run_no) if business.status != 1: raise WorkbenchError("BUSINESS_TERMINAL_REVIEW_REQUIRED") - # 新月份可追加，但已生成工资月份的迟到单据必须另行核对。 - from app.models import MonthlySettlement, PayrollMonth - month = day.strftime("%Y-%m") - locked = ctx.session.scalar(select(MonthlySettlement.id).where(MonthlySettlement.settlement_month == month).limit(1)) - control = ctx.session.scalar(select(PayrollMonth).where(PayrollMonth.month == month)) - if locked or (control and control.generated_at): - raise WorkbenchError("PAYROLL_MONTH_LOCKED") person = ctx.person(raw) summary = dict(business.integration_json) for key, value in (("personnel_names", person["source_personnel_name"]), ("personnel_codes", person["source_personnel_no"])): @@ -621,11 +624,12 @@ def execute_sync(session, run): """按单据提交；中断后幂等重跑，不生成 PayrollInput 或工资。""" from app.models import User from app.services.authorization_service import require_permission - require_permission(session, session.get(User, run.requested_by), "tplus:import") + operator = session.get(User, run.requested_by) + require_permission(session, operator, "tplus:import") start = date.fromisoformat(run.input_json.get("start_date", START.isoformat())) manifest = source_manifest(session, run.source_database) rebuild_projections(session, run.source_database) - ctx, counts, processed = Context(session, run.source_database), Counter(), 0 + ctx, counts, processed = Context(session, run.source_database, operator), Counter(), 0 for entity in ENTITIES: # 先固定所有依赖的发布状态；失败的全量采集不得泄露半批数据。 if entity in DETAILS: diff --git a/backend/app/services/tplus/workbench.py b/backend/app/services/tplus/workbench.py index 6d1d669..0b28905 100644 --- a/backend/app/services/tplus/workbench.py +++ b/backend/app/services/tplus/workbench.py @@ -64,9 +64,21 @@ def save_fact(session, payload, actor_id): try: with sync_lock(session): session.rollback() - current = latest_facts(session, source).get((payload.kind, payload.fact_key)) + facts = latest_facts(session, source) + current = facts.get((payload.kind, payload.fact_key)) if payload.kind == "GUIDANCE" and current and current.data_json.get("is_finalized"): raise BadRequestError("该发货月份指导成本已封存，后续回款不得修改原指导成本") + if payload.kind == "GUIDANCE": + if payload.data.get("is_finalized"): + raise BadRequestError("请在业务详情中核对来源并封存指导成本") + if any( + kind == "GUIDANCE" + and fact.data_json.get("is_finalized") + and fact.data_json.get("product_id") == payload.data.get("product_id") + and fact.data_json.get("month") == payload.data.get("month") + for (kind, _), fact in facts.items() + ): + raise BadRequestError("该产品发货月份已有封存指导成本，请先核对") if (current.version if current else 0) != payload.expected_version: raise BadRequestError("补充资料已经变化，请刷新后再保存") fact = TPlusFact( @@ -452,11 +464,14 @@ def report_csv(session, run_no): return rows() -def execute_next(session): +def execute_next(session, *, may_claim=None): """调用方持有原有全局锁；中断任务保留明细，禁止不明状态自动应用。""" for old in session.scalars(select(TPlusWorkRun).where(TPlusWorkRun.status == "RUNNING")): old.status, old.error_code, old.finished_at = "FAILED", "INTERRUPTED", datetime.now() session.commit() + if may_claim is not None and not may_claim(session): + session.rollback() + return None run = session.scalar( select(TPlusWorkRun) .where(TPlusWorkRun.status == "QUEUED") diff --git a/backend/tests/integration/test_tplus_sync_console.py b/backend/tests/integration/test_tplus_sync_console.py index 5359325..936f32e 100644 --- a/backend/tests/integration/test_tplus_sync_console.py +++ b/backend/tests/integration/test_tplus_sync_console.py @@ -13,9 +13,10 @@ pytestmark = pytest.mark.integration BASE = "/api/v1/system/tplus" -def test_新增配置表迁移可往返且不修改同步批次(): +def test_不可逆迁移拒绝回退且不修改同步批次(): from alembic import command from alembic.config import Config + from alembic.script import ScriptDirectory from sqlalchemy import inspect, text from app.core.db import get_engine @@ -25,12 +26,12 @@ def test_新增配置表迁移可往返且不修改同步批次(): with engine.connect() as connection: before = connection.exec_driver_sql('SELECT COUNT(*) FROM tplus_sync_batches').scalar_one() config = Config(str(Path(__file__).resolve().parents[2] / "alembic.ini")) - command.downgrade(config, "26e7f8a9b0c1") - assert not inspect(engine).has_table("tplus_automation_configs") + with pytest.raises(RuntimeError, match="不可直接回退"): + command.downgrade(config, "26e7f8a9b0c1") command.upgrade(config, "head") assert inspect(engine).has_table("tplus_automation_configs") with engine.connect() as connection: - assert connection.execute(text('SELECT version_num FROM alembic_version')).scalar_one() == '28a9b0c1d2e3' + assert connection.execute(text('SELECT version_num FROM alembic_version')).scalar_one() == ScriptDirectory.from_config(config).get_current_head() assert connection.exec_driver_sql('SELECT COUNT(*) FROM tplus_sync_batches').scalar_one() == before diff --git a/backend/tests/unit/test_tplus_compatibility_script.py b/backend/tests/unit/test_tplus_compatibility_script.py index cc35ed9..9dd4e1c 100644 --- a/backend/tests/unit/test_tplus_compatibility_script.py +++ b/backend/tests/unit/test_tplus_compatibility_script.py @@ -21,7 +21,10 @@ def script(monkeypatch): spec = importlib.util.spec_from_file_location("compatibility_script", path) module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) - settings = Settings(_env_file=None, app_secret_key="x" * 32, t_plus_encrypt=True) + settings = Settings( + _env_file=None, app_secret_key="x" * 32, t_plus_encrypt=True, + t_plus_allow_dev_sa=False, t_plus_allow_admin_account=False, + ) monkeypatch.setattr(module, "get_settings", lambda: settings) return module @@ -44,7 +47,7 @@ def test_脚本复用连接文件校验与加密设置(script, tmp_path): assert script.get_settings().t_plus_enabled is False payload["user"] = "sa" path.write_text(json.dumps(payload), encoding="utf-8") - with pytest.raises(RuntimeError, match="专用只读"): + with pytest.raises(RuntimeError, match="管理员账号未获授权"): script.compatibility_settings(path) diff --git a/backend/tests/unit/test_tplus_connection_file.py b/backend/tests/unit/test_tplus_connection_file.py index fc2220a..2f8e34f 100644 --- a/backend/tests/unit/test_tplus_connection_file.py +++ b/backend/tests/unit/test_tplus_connection_file.py @@ -72,11 +72,37 @@ def test_连接文件不接受sa或应用配置覆盖(tmp_path): resolve_tplus_settings(settings) +def test_生产连接文件允许显式授权的管理员账号(tmp_path): + path = tmp_path / "reader.json" + path.write_text( + json.dumps({ + "host": "127.0.0.1", "port": 1433, "database": "test_source", + "user": "sa", "password": "test-only-value", "driver": "test_driver", + }), + encoding="utf-8", + ) + settings = Settings( + _env_file=None, + app_secret_key="x" * 32, + app_env="prod", + t_plus_enabled=True, + t_plus_allow_admin_account=True, + t_plus_connection_file=str(path), + ) + assert resolve_tplus_settings(settings).t_plus_user == "sa" + + @pytest.mark.parametrize( - ("environment", "allowed", "accepted"), - [("dev", False, False), ("prod", True, False), ("dev", True, True)], + ("environment", "allowed", "admin_allowed", "accepted"), + [ + ("dev", False, False, False), + ("prod", True, False, False), + ("dev", True, False, True), + ("prod", False, True, True), + ("prod", False, False, False), + ], ) -def test_sa仅允许显式启用的开发环境(environment, allowed, accepted): +def test_sa仅允许显式授权(environment, allowed, admin_allowed, accepted): settings = Settings( _env_file=None, app_secret_key="x" * 32, @@ -86,9 +112,10 @@ def test_sa仅允许显式启用的开发环境(environment, allowed, accepted): t_plus_database="test_source", t_plus_odbc_driver="test_driver", t_plus_allow_dev_sa=allowed, + t_plus_allow_admin_account=admin_allowed, ) if accepted: assert resolve_tplus_settings(settings).t_plus_user == "sa" else: - with pytest.raises(RuntimeError, match="不能使用 sa"): + with pytest.raises(RuntimeError, match="管理员账号未获授权"): resolve_tplus_settings(settings) diff --git a/backend/tests/unit/test_tplus_workbench.py b/backend/tests/unit/test_tplus_workbench.py index d7f9164..6d95e04 100644 --- a/backend/tests/unit/test_tplus_workbench.py +++ b/backend/tests/unit/test_tplus_workbench.py @@ -433,6 +433,21 @@ def test_资料版本冲突不覆盖_历史保留(work_session): assert s.scalar(select(func.count()).select_from(TPlusFact)) == 2 +def test_旧工作台不能绕过指导成本封存检查(work_session): + from app.core.exceptions import BadRequestError + + data = {"product_id": "9", "month": "2026-03", "basis": "UNIT", "amount": "25", "unit": "件"} + with pytest.raises(BadRequestError, match="业务详情"): + fact(work_session, "GUIDANCE", "legacy", {**data, "is_finalized": True}) + work_session.add(TPlusFact( + source_database="test_source", kind="GUIDANCE", fact_key="sealed", version=1, + data_json={**data, "is_finalized": True}, evidence="已核对", requested_by=1, + )) + work_session.commit() + with pytest.raises(BadRequestError, match="已有封存"): + fact(work_session, "GUIDANCE", "another-key", data) + + def test_更正事实要求排除或原业务归属(): with pytest.raises(ValidationError): FactIn( diff --git a/backend/tests/unit/test_tplus_worker.py b/backend/tests/unit/test_tplus_worker.py index f135bbd..de11e49 100644 --- a/backend/tests/unit/test_tplus_worker.py +++ b/backend/tests/unit/test_tplus_worker.py @@ -57,6 +57,13 @@ def test_忙锁时不领取任务也不恢复其他运行记录(tplus_session, m assert batch.status == "QUEUED" +def test_停止请求阻止领取排队任务(tplus_session, monkeypatch): + source = _worker_setup(monkeypatch) + batch = _queued(tplus_session) + assert worker.run_once(tplus_session, source_factory=source, may_claim=lambda _session: False) is None + assert batch.status == "QUEUED" + + def test_源连接失败写入固定错误不泄漏凭据(tplus_session, monkeypatch): _worker_setup(monkeypatch) batch = _queued(tplus_session) diff --git a/frontend/src/api/tplus.ts b/frontend/src/api/tplus.ts index 8c110ff..3dd412f 100644 --- a/frontend/src/api/tplus.ts +++ b/frontend/src/api/tplus.ts @@ -90,6 +90,18 @@ export interface AutomationStatus { next_retry_at: string | null alerts: {code: string; message: string; batch_no?: string | null}[] } +export interface RuntimeStatus { + desired_state: 'RUNNING' | 'STOPPED' + actual_state: 'RUNNING' | 'STOPPED' | 'STOPPING' | 'DEGRADED' | 'OFFLINE' + worker_phase: string + heartbeat_at: string | null + last_error: string | null + host: string | null + platform: string | null + version: number + updated_by: number | null + editable: boolean +} export interface BatchQuery { status?: TaskStatus; task_type?: TaskType; pageNum?: number; pageSize?: number } export interface ErrorQuery { entity_type?: EntityType; pageNum?: number; pageSize?: number } export interface CreateTask { task_type: TaskType; entity_types: EntityType[]; request_key: string } @@ -105,6 +117,12 @@ export function listBatches(params: BatchQuery, options: ReadOptions = {}): Prom export function updateAutomation(data: { enabled: boolean; expected_version: number }): Promise { return request({ url: `${base}/automation`, method: 'put', data }) } +export function getRuntime(options: ReadOptions = {}): Promise { + return request({ url: `${base}/runtime`, ...options }) +} +export function updateRuntime(data: { desired_state: 'RUNNING' | 'STOPPED'; expected_version: number }): Promise { + return request({ url: `${base}/runtime`, method: 'put', data }) +} /** 按需查询报告。 */ export function getBatch(number: string, options: ReadOptions = {}): Promise { return request({ url: `${base}/batches/${encodeURIComponent(number)}`, ...options }) diff --git a/frontend/src/views/system/tplus/index.vue b/frontend/src/views/system/tplus/index.vue index 516e6c4..2cb97c7 100644 --- a/frontend/src/views/system/tplus/index.vue +++ b/frontend/src/views/system/tplus/index.vue @@ -20,8 +20,8 @@ import { useSyncPage } from './useSyncPage' const user = useUserStore() const route = useRoute() -const {overview, records, total, pageNum, pageSize, status, taskType, loading, failed, - saving, refreshedAt, refresh, changeAutomation, search, onPageChange, onSizeChange} = useSyncPage() +const {overview, runtime, records, total, pageNum, pageSize, status, taskType, loading, failed, + saving, runtimeSaving, refreshedAt, refresh, changeAutomation, changeRuntime, search, onPageChange, onSizeChange} = useSyncPage() const activeTab = ref('records'), creating = ref(false), retrying = ref(null) const submittedNumber = ref(null) const detailNumber = ref(typeof route.query.batch === 'string' && /^[A-Za-z0-9_-]{1,32}$/.test(route.query.batch) ? route.query.batch : null) @@ -32,6 +32,8 @@ const phases: Record = {UNKNOWN:'暂未收到心跳', IDLE:'等 const connectionLabel = computed(() => !overview.value ? '正在检查连接配置' : overview.value.is_configured ? 'T+ 读取配置就绪' : '读取配置尚未就绪') const canCreate = computed(() => user.can('tplus:collect') && overview.value?.is_configured && !failed.value) const canConfigure = computed(() => user.can('tplus:collect') && automation.value?.editable && !failed.value) +const canControl = computed(() => user.can('tplus:collect') && runtime.value?.editable && !failed.value && !saving.value) +const runtimeLabels: Record = { RUNNING: '运行中', STOPPED: '已停止', STOPPING: '正在停止', DEGRADED: '等待依赖', OFFLINE: '服务离线' } const switchReason = computed(() => !user.can('tplus:collect') ? user.permissionReason('tplus:collect') : !automation.value?.editable ? '连接配置或设置服务尚未就绪，请联系维护人员' : '开关保存后自动生效，当前读取任务不会中断') const alerts: Record = { MIGRATION_REQUIRED:'采集设置尚未就绪，请联系维护人员检查服务。', @@ -77,11 +79,18 @@ async function retry(row: api.TPlusBatch): Promise {

每周自动采集

{{ saving ? '正在保存设置…' : automation?.enabled ? '自动采集已开启' : '自动采集已关闭' }}

-
采集时间
每周{{ weekdays[automation?.weekday ?? 0] }} {{ automation?.time ?? '02:00' }}北京时间
读取服务
{{ phases[automation?.worker_phase ?? 'UNKNOWN'] }}
最近成功
{{ formatDateTime(automation?.last_success_at) }}
下次重试
{{ formatDateTime(automation.next_retry_at) }}
+
采集时间
每周{{ weekdays[automation?.weekday ?? 0] }} {{ automation?.time ?? '02:00' }}北京时间
读取阶段
{{ phases[runtime?.worker_phase ?? 'UNKNOWN'] }}
最近成功
{{ formatDateTime(automation?.last_success_at) }}
下次重试
{{ formatDateTime(automation.next_retry_at) }}

开启后按周读取；关闭只停止后续自动排队，当前任务继续完成。手动同步仍可使用。

{{ switchReason }}

+
+

读取服务

{{ runtimeLabels[runtime?.actual_state ?? 'OFFLINE'] }}
+
期望状态：{{ runtime?.desired_state === 'RUNNING' ? '运行' : '停止' }}最近心跳：{{ formatDateTime(runtime?.heartbeat_at) }}运行主机：{{ runtime.host }}（{{ runtime.platform }}）
+

{{ runtime.last_error }}

+

服务管理尚未就绪，请联系维护人员。

+
启动读取服务停止读取服务
+
查看相关任务 @@ -134,6 +143,13 @@ async function retry(row: api.TPlusBatch): Promise { .schedule-details dd { margin: 0; line-height: 1.65; font-variant-numeric: tabular-nums; } .schedule-details small { margin-left: 6px; color: var(--ys-ink-2); font-size: 13px; } .settings-help { font-size: 12px; line-height: 1.8; color: var(--ys-ink-2); margin: 18px 0 0; } +.runtime-section { margin-top: 20px; padding: 18px 0; border-bottom: 1px solid var(--ys-line); } +.runtime-heading { display: flex; align-items: center; gap: 12px; } +.runtime-heading h2 { margin: 0; font-size: 16px; } +.runtime-details { display: flex; flex-wrap: wrap; gap: 8px 22px; margin-top: 12px; font-size: 13px; color: var(--ys-ink-2); } +.runtime-error { color: var(--el-color-danger); font-size: 13px; } +.runtime-actions { display: flex; flex-wrap: wrap; gap: 8px; margin-top: 14px; } +.runtime-actions .el-button + .el-button { margin-left: 0; } .sync-boundary { display: flex; align-items: center; justify-content: space-between; flex-wrap: wrap; gap: 10px 20px; padding: 18px 0; font-size: 13px; line-height: 1.7; } .sync-boundary b { font-weight: 600; } .sync-boundary p { color: var(--ys-ink-2); margin: 4px 0 0; font-size: 12px; } diff --git a/frontend/src/views/system/tplus/useSyncPage.ts b/frontend/src/views/system/tplus/useSyncPage.ts index 390ba28..1633409 100644 --- a/frontend/src/views/system/tplus/useSyncPage.ts +++ b/frontend/src/views/system/tplus/useSyncPage.ts @@ -6,10 +6,11 @@ import { handledError } from '@/utils/error' /** 页面与筛选使用同一请求序号，后台刷新保留已显示数据和交互位置。 */ export function useSyncPage() { const overview = ref(null) + const runtime = ref(null) const records = ref([]) const total = ref(0), pageNum = ref(1), pageSize = ref(10) const status = ref(), taskType = ref() - const loading = ref(false), failed = ref(false), saving = ref(false) + const loading = ref(false), failed = ref(false), saving = ref(false), runtimeSaving = ref(false) const refreshedAt = ref(null) let shownPage = 1, shownSize = 10, sequence = 0, alive = false let timer: ReturnType | undefined @@ -18,16 +19,17 @@ export function useSyncPage() { function schedule(): void { clearTimeout(timer) if (!alive) return - const active = ['RUNNING', 'QUEUED'].includes(overview.value?.current_batch?.status ?? '') + const active = ['RUNNING', 'QUEUED'].includes(overview.value?.current_batch?.status ?? '') || + (runtime.value?.actual_state !== 'OFFLINE' && runtime.value?.desired_state !== runtime.value?.actual_state) timer = setTimeout(() => { - if (document.hidden || saving.value) schedule() + if (document.hidden || saving.value || runtimeSaving.value) schedule() else void refresh(true) }, active ? 5000 : 30000) } async function refresh(background = false): Promise { if (!alive) return - if (saving.value) { schedule(); return } + if (saving.value || runtimeSaving.value) { schedule(); return } const current = ++sequence controller?.abort() controller = new AbortController() @@ -35,9 +37,10 @@ export function useSyncPage() { const query = {pageNum: pageNum.value, pageSize: pageSize.value, status: status.value, task_type: taskType.value} const options = {silentError: background, signal: controller.signal} try { - const [summary, page] = await Promise.all([api.getOverview(options), api.listBatches(query, options)]) + const [summary, page, service] = await Promise.all([api.getOverview(options), api.listBatches(query, options), api.getRuntime(options)]) if (!alive || current !== sequence) return overview.value = summary + runtime.value = service records.value = page.list total.value = page.total shownPage = query.pageNum @@ -57,6 +60,20 @@ export function useSyncPage() { } } + async function changeRuntime(desired_state: 'RUNNING' | 'STOPPED'): Promise { + const current = runtime.value + if (!current || runtimeSaving.value || saving.value) return + runtimeSaving.value = true + ++sequence + controller?.abort() + loading.value = false + try { + runtime.value = await api.updateRuntime({desired_state, expected_version: current.version}) + ElMessage.success(desired_state === 'RUNNING' ? '已请求启动读取服务，等待状态更新' : '已请求停止读取服务，当前任务完成后停止') + } catch (error) { handledError(error) } + finally { runtimeSaving.value = false; if (alive) await refresh(true) } + } + async function changeAutomation(value: boolean | string | number): Promise { const config = overview.value?.automation if (typeof value !== 'boolean' || !config || saving.value) return @@ -80,9 +97,9 @@ export function useSyncPage() { function search(): void { pageNum.value = 1; void refresh() } function onPageChange(value: number): void { pageNum.value = value; void refresh() } function onSizeChange(value: number): void { pageSize.value = value; pageNum.value = 1; void refresh() } - function visible(): void { if (!document.hidden && !saving.value) void refresh(true) } + function visible(): void { if (!document.hidden && !saving.value && !runtimeSaving.value) void refresh(true) } onMounted(() => { alive = true; document.addEventListener('visibilitychange', visible); void refresh() }) onUnmounted(() => { alive = false; ++sequence; clearTimeout(timer); controller?.abort(); document.removeEventListener('visibilitychange', visible) }) - return {overview, records, total, pageNum, pageSize, status, taskType, loading, failed, - saving, refreshedAt, refresh, changeAutomation, search, onPageChange, onSizeChange} + return {overview, runtime, records, total, pageNum, pageSize, status, taskType, loading, failed, + saving, runtimeSaving, refreshedAt, refresh, changeAutomation, changeRuntime, search, onPageChange, onSizeChange} }