[fix/feat]:[FL-77][修复杆塔高程回填任务 actor_user_id 传递]

Co-authored-by: multica-agent <github@multica.ai>

api/app/services/elevation_service.py

@@ -1192,7 +1192,7 @@ def create_apply_job(
     if not saved:
         raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="创建任务失败")
 
-    task = _dispatch_elevation_apply_task(job_id=saved.id)
+    task = _dispatch_elevation_apply_task(job_id=saved.id, actor_user_id=actor.id)
     saved.task_id = task.id
     saved.update_user = actor.id
     saved.update_date = utcnow()
@@ -1211,10 +1211,10 @@ def create_apply_job(
     return ElevationApplyJobCreateResponse(job=serialize_job(latest), queued=True)
 
 
-def _dispatch_elevation_apply_task(*, job_id: str):
+def _dispatch_elevation_apply_task(*, job_id: str, actor_user_id: str | None):
     from ..tasks.elevation_tasks import apply_elevation_for_line_job
 
-    return apply_elevation_for_line_job.delay(job_id)
+    return apply_elevation_for_line_job.delay(job_id, actor_user_id)
 
 
 def _dispatch_elevation_dataset_analysis_task(*, dataset_id: str, actor_user_id: str | None):
@@ -1229,7 +1229,7 @@ def _dispatch_elevation_dataset_terrain_task(*, dataset_id: str, actor_user_id:
     return build_elevation_dataset_terrain_job.delay(dataset_id, actor_user_id)
 
 
-def execute_apply_job(job_id: str) -> None:
+def execute_apply_job(job_id: str, actor_user_id: str | None = None) -> None:
     db = SessionLocal()
     try:
         job = get_job_by_id(db, job_id)
@@ -1237,6 +1237,7 @@ def execute_apply_job(job_id: str) -> None:
             return
         if job.status in {"success", "failed"}:
             return
+        resolved_actor_user_id = actor_user_id or job.create_user or job.update_user
 
         job.status = "running"
         job.started_at = utcnow()
@@ -1298,13 +1299,13 @@ def execute_apply_job(job_id: str) -> None:
         job.finished_at = utcnow()
         job.update_date = utcnow()
         line.update_date = utcnow()
-        line.update_user = actor_user_id
+        line.update_user = resolved_actor_user_id
         record_line_preparation_source(
             line,
             component="ground_slope",
             payload={
                 "prepared_at": utcnow().isoformat(),
-                "prepared_by_user_id": actor_user_id,
+                "prepared_by_user_id": resolved_actor_user_id,
                 "dataset_id": dataset.id,
                 "dataset_code": dataset.code,
                 "job_id": job.id,

api/app/tasks/elevation_tasks.py

@@ -5,8 +5,8 @@ from ..services.elevation_service import execute_apply_job, execute_dataset_anal
 
 
 @celery_app.task(name="app.tasks.elevation_tasks.apply_elevation_for_line_job")
-def apply_elevation_for_line_job(job_id: str) -> dict[str, str]:
-    execute_apply_job(job_id)
+def apply_elevation_for_line_job(job_id: str, actor_user_id: str | None = None) -> dict[str, str]:
+    execute_apply_job(job_id=job_id, actor_user_id=actor_user_id)
     return {"job_id": job_id, "status": "done"}
 
 

api/tests/test_elevation_apply_service.py

@@ -0,0 +1,162 @@
+from __future__ import annotations
+
+from types import SimpleNamespace
+
+from sqlalchemy import create_engine, select
+from sqlalchemy.orm import Session, sessionmaker
+from sqlalchemy.pool import StaticPool
+
+from app.core.database import Base
+from app.models.elevation import ElevationApplyJob, ElevationDataset
+from app.models.line import Line
+from app.models.line_tower import LineTower
+from app.models.tower_profile import TowerProfile
+from app.schemas.elevation import ElevationApplyJobCreateRequest
+from app.services import elevation_service
+
+
+def _build_session_factory() -> sessionmaker[Session]:
+    engine = create_engine(
+        "sqlite+pysqlite://",
+        connect_args={"check_same_thread": False},
+        poolclass=StaticPool,
+    )
+    Base.metadata.create_all(
+        bind=engine,
+        tables=[
+            Line.__table__,
+            LineTower.__table__,
+            TowerProfile.__table__,
+            ElevationDataset.__table__,
+            ElevationApplyJob.__table__,
+        ],
+    )
+    return sessionmaker(bind=engine, autocommit=False, autoflush=False, expire_on_commit=False)
+
+
+def test_create_apply_job_dispatches_actor_user_id(monkeypatch) -> None:
+    testing_session = _build_session_factory()
+    session = testing_session()
+    try:
+        line = Line(code="L-APPLY-001", name="回填线路", voltage_kv=220, lightning_param_json={})
+        dataset = ElevationDataset(
+            code="DEM-APPLY-001",
+            name="高程数据集",
+            file_format="csv",
+            mount_code="default",
+            dataset_dir="/elevation/datasets/DEM-APPLY-001",
+            file_path="/elevation/datasets/DEM-APPLY-001/dataset.csv",
+            status="active",
+            usage_status="idle",
+        )
+        session.add_all([line, dataset])
+        session.flush()
+        session.add(LineTower(line_id=line.id, seq_no=1, tower_no="T1", longitude=120.0, latitude=30.0))
+        session.commit()
+
+        dispatched: dict[str, str | None] = {}
+
+        def _fake_dispatch(*, job_id: str, actor_user_id: str | None) -> SimpleNamespace:
+            dispatched["job_id"] = job_id
+            dispatched["actor_user_id"] = actor_user_id
+            return SimpleNamespace(id="celery-task-1")
+
+        monkeypatch.setattr(elevation_service, "_dispatch_elevation_apply_task", _fake_dispatch)
+        monkeypatch.setattr(elevation_service, "_publish_elevation_change", lambda *args, **kwargs: None)
+
+        response = elevation_service.create_apply_job(
+            session,
+            ElevationApplyJobCreateRequest(line_id=line.id, dataset_id=dataset.id, mode="overwrite_all"),
+            actor=SimpleNamespace(id="tester"),
+        )
+
+        saved_job = session.get(ElevationApplyJob, response.job.id)
+        assert response.queued is True
+        assert dispatched == {"job_id": response.job.id, "actor_user_id": "tester"}
+        assert saved_job is not None
+        assert saved_job.task_id == "celery-task-1"
+        assert saved_job.create_user == "tester"
+        assert saved_job.update_user == "tester"
+    finally:
+        session.close()
+
+
+def test_execute_apply_job_uses_saved_actor_for_preparation_source(monkeypatch) -> None:
+    testing_session = _build_session_factory()
+    session = testing_session()
+    try:
+        line = Line(code="L-APPLY-002", name="高程回填线路", voltage_kv=110, lightning_param_json={})
+        dataset = ElevationDataset(
+            code="DEM-APPLY-002",
+            name="高程数据集",
+            file_format="csv",
+            mount_code="default",
+            dataset_dir="/elevation/datasets/DEM-APPLY-002",
+            file_path="/elevation/datasets/DEM-APPLY-002/dataset.csv",
+            status="active",
+            usage_status="idle",
+        )
+        session.add_all([line, dataset])
+        session.flush()
+
+        meter_to_lat = 1 / 111_320.0
+        session.add_all(
+            [
+                LineTower(line_id=line.id, seq_no=1, tower_no="P1", longitude=120.0, latitude=30.0 + 300 * meter_to_lat),
+                LineTower(line_id=line.id, seq_no=2, tower_no="P2", longitude=120.0, latitude=30.0 + 600 * meter_to_lat),
+                LineTower(line_id=line.id, seq_no=3, tower_no="P3", longitude=120.0, latitude=30.0 + 900 * meter_to_lat),
+            ]
+        )
+        session.flush()
+
+        job = ElevationApplyJob(
+            line_id=line.id,
+            dataset_id=dataset.id,
+            mode="overwrite_all",
+            status="pending",
+            total_tower_count=3,
+            create_user="tester",
+            update_user="tester",
+        )
+        session.add(job)
+        session.commit()
+
+        points = [
+            elevation_service.ElevationSamplePoint(
+                lon=120.0,
+                lat=30.0 + distance_m * meter_to_lat,
+                altitude_m=100.0 + distance_m * 0.12,
+            )
+            for distance_m in range(0, 1251, 50)
+        ]
+
+        monkeypatch.setattr(elevation_service, "SessionLocal", testing_session)
+        monkeypatch.setattr(elevation_service, "_load_dataset_points", lambda *_args, **_kwargs: (points, []))
+        monkeypatch.setattr(elevation_service, "_publish_elevation_change", lambda *args, **kwargs: None)
+        monkeypatch.setattr(elevation_service, "_publish_line_change", lambda *args, **kwargs: None)
+        monkeypatch.setattr(elevation_service, "_refresh_dataset_usage_status", lambda *args, **kwargs: None)
+
+        elevation_service.execute_apply_job(job.id)
+
+        verification_session = testing_session()
+        try:
+            saved_job = verification_session.get(ElevationApplyJob, job.id)
+            saved_line = verification_session.get(Line, line.id)
+            towers = verification_session.execute(
+                select(LineTower).where(LineTower.line_id == line.id).order_by(LineTower.seq_no.asc())
+            ).scalars().all()
+
+            assert saved_job is not None
+            assert saved_job.status == "success"
+            assert saved_line is not None
+            assert saved_line.update_user == "tester"
+            assert all(tower.altitude_m is not None for tower in towers)
+
+            source = saved_line.lightning_param_json["preparation_sources"]["ground_slope"]
+            assert source["prepared_by_user_id"] == "tester"
+            assert source["dataset_id"] == dataset.id
+            assert source["job_id"] == job.id
+        finally:
+            verification_session.close()
+    finally:
+        session.close()