refactor(B05,B06): B05_wf2_Route -> B05_Profile, B06_wf3_ProfileCross -> B06_Section 동시 개명
- 한몸으로 동작하는 두 페이지라 한 커밋으로 처리 (상호 참조 다수) - B05 37파일 + B06 20파일 접두사 개명 (git mv, 이력 보존) - 참조 치환 91파일: import 경로, 라우트 슬러그(b05-profile/b06-section), 라우트 키(B05_PROFILE/B06_SECTION), B03 자동 체인, storage 상수, pyproject 제외 경로 - 로직 변경 없음. typecheck·백엔드 import 검증 통과 Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Claude Fable 5
@@ -0,0 +1,651 @@
|
||||
"""B06 종횡단 결과의 aiomysql Raw SQL 접근.
|
||||
|
||||
longitudinal_sections(종단면 1건), cross_sections(측점별 다건) 테이블에
|
||||
메타데이터와 상대 경로를 기록한다. 상세 샘플 데이터는 파일에 저장하고 DB에는
|
||||
요약 data(JSON)와 경로만 기록한다.
|
||||
"""
|
||||
|
||||
import json
|
||||
from collections.abc import Callable
|
||||
from pathlib import PurePosixPath
|
||||
from typing import Any
|
||||
from uuid import UUID
|
||||
|
||||
import aiomysql
|
||||
|
||||
_STAGE_ROOT = "B06_Section"
|
||||
|
||||
|
||||
def _validate_stage_path(relative_path: str) -> str:
|
||||
normalized = PurePosixPath(relative_path.replace("\\", "/"))
|
||||
if normalized.is_absolute() or ".." in normalized.parts:
|
||||
raise ValueError("DB에는 프로젝트 루트 기준 상대 경로만 저장할 수 있습니다.")
|
||||
if not normalized.parts or normalized.parts[0] != _STAGE_ROOT:
|
||||
raise ValueError(f"B06 산출물 경로는 {_STAGE_ROOT} 아래여야 합니다.")
|
||||
return normalized.as_posix()
|
||||
|
||||
|
||||
async def get_confirmed_route_context(
|
||||
connection: aiomysql.Connection, project_id: UUID
|
||||
) -> dict[str, Any] | None:
|
||||
"""프로젝트의 최신 확정 경로와 연결된 지표면 좌표계를 조회한다.
|
||||
|
||||
surface_models.crs_epsg가 NULL이면(분석에 사용한 입력 파일에 좌표계가
|
||||
없던 경우) 같은 프로젝트 input_files의 감지된 좌표계로 폴백한다.
|
||||
"""
|
||||
async with connection.cursor(aiomysql.DictCursor) as cursor:
|
||||
await cursor.execute(
|
||||
"""
|
||||
SELECT r.id AS route_id,
|
||||
COALESCE(
|
||||
sm.crs_epsg,
|
||||
(SELECT f.crs_epsg
|
||||
FROM input_files f
|
||||
WHERE f.project_id = r.project_id AND f.crs_epsg IS NOT NULL
|
||||
ORDER BY f.id DESC
|
||||
LIMIT 1)
|
||||
) AS crs_epsg
|
||||
FROM routes r
|
||||
LEFT JOIN surface_models sm ON sm.id = r.surface_model_id
|
||||
WHERE r.project_id = %s AND r.status = 'CONFIRMED'
|
||||
ORDER BY r.computed_at DESC, r.id DESC
|
||||
LIMIT 1
|
||||
""",
|
||||
(str(project_id),),
|
||||
)
|
||||
row = await cursor.fetchone()
|
||||
if not row:
|
||||
return None
|
||||
return {
|
||||
"route_id": int(row["route_id"]),
|
||||
"crs_epsg": int(row["crs_epsg"]) if row["crs_epsg"] is not None else None,
|
||||
}
|
||||
|
||||
|
||||
async def get_latest_section_options(
|
||||
connection: aiomysql.Connection, project_id: UUID
|
||||
) -> dict[str, Any] | None:
|
||||
"""프로젝트 최신 종단면 data에 저장된 생성 옵션 스냅샷을 반환한다 (없으면 None)."""
|
||||
async with connection.cursor() as cursor:
|
||||
await cursor.execute(
|
||||
"""
|
||||
SELECT data
|
||||
FROM longitudinal_sections
|
||||
WHERE project_id = %s
|
||||
ORDER BY id DESC
|
||||
LIMIT 1
|
||||
""",
|
||||
(str(project_id),),
|
||||
)
|
||||
row = await cursor.fetchone()
|
||||
if not row or not row[0]:
|
||||
return None
|
||||
data = row[0]
|
||||
if isinstance(data, str):
|
||||
data = json.loads(data)
|
||||
options = data.get("options") if isinstance(data, dict) else None
|
||||
return options if isinstance(options, dict) else None
|
||||
|
||||
|
||||
async def list_recent_company_projects(
|
||||
connection: aiomysql.Connection, company_id: int, exclude_project_id: UUID
|
||||
) -> list[dict[str, Any]]:
|
||||
"""같은 회사의 최근 프로젝트 5개(최근 갱신순, 현재·삭제 프로젝트 제외).
|
||||
|
||||
설계값 보유 여부와 무관하게 보여준다 — 선택 시 설계값 조회에서 없으면 안내한다.
|
||||
"""
|
||||
async with connection.cursor(aiomysql.DictCursor) as cursor:
|
||||
await cursor.execute(
|
||||
"""
|
||||
SELECT p.id AS project_id, p.name AS name
|
||||
FROM projects p
|
||||
WHERE p.company_id = %s
|
||||
AND p.id <> %s
|
||||
AND p.deleted_at IS NULL
|
||||
ORDER BY p.updated_at DESC
|
||||
LIMIT 5
|
||||
""",
|
||||
(company_id, str(exclude_project_id)),
|
||||
)
|
||||
rows = await cursor.fetchall()
|
||||
return [{"project_id": row["project_id"], "name": row["name"]} for row in rows]
|
||||
|
||||
|
||||
async def get_project_standard_cross_section(
|
||||
connection: aiomysql.Connection, company_id: int, project_id: UUID
|
||||
) -> dict[str, Any] | None:
|
||||
"""회사 스코프로 특정 프로젝트의 표준횡단 설정값을 반환한다(타 회사 접근 차단).
|
||||
|
||||
company_id 조건으로 남의 회사 프로젝트 값은 조회되지 않는다(권한 강제).
|
||||
"""
|
||||
async with connection.cursor() as cursor:
|
||||
await cursor.execute(
|
||||
"""
|
||||
SELECT l.data
|
||||
FROM longitudinal_sections l
|
||||
JOIN projects p ON p.id = l.project_id
|
||||
WHERE l.project_id = %s
|
||||
AND p.company_id = %s
|
||||
AND p.deleted_at IS NULL
|
||||
AND JSON_EXTRACT(l.data, '$.options.standard_cross_section') IS NOT NULL
|
||||
ORDER BY l.id DESC
|
||||
LIMIT 1
|
||||
""",
|
||||
(str(project_id), company_id),
|
||||
)
|
||||
row = await cursor.fetchone()
|
||||
if not row or not row[0]:
|
||||
return None
|
||||
data = row[0]
|
||||
if isinstance(data, str):
|
||||
data = json.loads(data)
|
||||
options = data.get("options") if isinstance(data, dict) else None
|
||||
standard = options.get("standard_cross_section") if isinstance(options, dict) else None
|
||||
return standard if isinstance(standard, dict) else None
|
||||
|
||||
|
||||
async def get_latest_grade_options(
|
||||
connection: aiomysql.Connection, project_id: UUID
|
||||
) -> dict[str, Any] | None:
|
||||
"""최신 종단면 data에 저장된 계획선 **사용자 입력값**을 반환한다 (없으면 None).
|
||||
|
||||
해석이 끝난 기준값(`grade_options`)이 아니라 사용자가 명시 입력한 값
|
||||
(`grade_overrides`)만 돌려준다. 전자를 폴백으로 되쓰면 등급·지형 구분을 바꿔도
|
||||
옛 기본값이 법정값을 이겨 갱신되지 않는다.
|
||||
"""
|
||||
async with connection.cursor() as cursor:
|
||||
await cursor.execute(
|
||||
"""
|
||||
SELECT data
|
||||
FROM longitudinal_sections
|
||||
WHERE project_id = %s
|
||||
ORDER BY id DESC
|
||||
LIMIT 1
|
||||
""",
|
||||
(str(project_id),),
|
||||
)
|
||||
row = await cursor.fetchone()
|
||||
if not row or not row[0]:
|
||||
return None
|
||||
data = row[0]
|
||||
if isinstance(data, str):
|
||||
data = json.loads(data)
|
||||
overrides = data.get("grade_overrides") if isinstance(data, dict) else None
|
||||
return overrides if isinstance(overrides, dict) else None
|
||||
|
||||
|
||||
async def get_route_generation_source(
|
||||
connection: aiomysql.Connection, project_id: UUID, route_id: int
|
||||
) -> dict[str, Any] | None:
|
||||
"""종횡단 재생성에 필요한 경로 GeoJSON 경로와 좌표계를 route_id로 조회한다."""
|
||||
async with connection.cursor(aiomysql.DictCursor) as cursor:
|
||||
await cursor.execute(
|
||||
"""
|
||||
SELECT r.route_data_path,
|
||||
COALESCE(
|
||||
sm.crs_epsg,
|
||||
(SELECT f.crs_epsg
|
||||
FROM input_files f
|
||||
WHERE f.project_id = r.project_id AND f.crs_epsg IS NOT NULL
|
||||
ORDER BY f.id DESC
|
||||
LIMIT 1)
|
||||
) AS crs_epsg
|
||||
FROM routes r
|
||||
LEFT JOIN surface_models sm ON sm.id = r.surface_model_id
|
||||
WHERE r.id = %s AND r.project_id = %s
|
||||
LIMIT 1
|
||||
""",
|
||||
(route_id, str(project_id)),
|
||||
)
|
||||
row = await cursor.fetchone()
|
||||
if not row or not row["route_data_path"]:
|
||||
return None
|
||||
return {
|
||||
"route_data_path": str(row["route_data_path"]),
|
||||
"crs_epsg": int(row["crs_epsg"]) if row["crs_epsg"] is not None else None,
|
||||
}
|
||||
|
||||
|
||||
async def delete_sections_for_route(connection: aiomysql.Connection, route_id: int) -> None:
|
||||
"""경로 재생성 전에 기존 종횡단 레코드를 삭제한다 (멱등 재실행)."""
|
||||
async with connection.cursor() as cursor:
|
||||
await cursor.execute("DELETE FROM cross_sections WHERE route_id = %s", (route_id,))
|
||||
await cursor.execute("DELETE FROM longitudinal_sections WHERE route_id = %s", (route_id,))
|
||||
|
||||
|
||||
async def create_longitudinal_section(
|
||||
connection: aiomysql.Connection,
|
||||
*,
|
||||
project_id: UUID,
|
||||
route_id: int,
|
||||
data: dict[str, Any] | None,
|
||||
longitudinal_file_path: str,
|
||||
status: str = "DRAFT",
|
||||
) -> int:
|
||||
"""종단면 메타데이터를 저장하고 생성된 ID를 반환한다."""
|
||||
file_rel = _validate_stage_path(longitudinal_file_path)
|
||||
async with connection.cursor() as cursor:
|
||||
await cursor.execute(
|
||||
"""
|
||||
INSERT INTO longitudinal_sections (
|
||||
project_id, route_id, computed_at, data, longitudinal_file_path, status
|
||||
)
|
||||
VALUES (%s, %s, CURRENT_TIMESTAMP, %s, %s, %s)
|
||||
""",
|
||||
(
|
||||
str(project_id),
|
||||
route_id,
|
||||
json.dumps(data, ensure_ascii=False) if data is not None else None,
|
||||
file_rel,
|
||||
status,
|
||||
),
|
||||
)
|
||||
new_id = cursor.lastrowid
|
||||
if not new_id:
|
||||
raise RuntimeError("longitudinal_sections 레코드 생성 결과에 ID가 없습니다.")
|
||||
return int(new_id)
|
||||
|
||||
|
||||
async def insert_cross_sections(
|
||||
connection: aiomysql.Connection,
|
||||
*,
|
||||
project_id: UUID,
|
||||
route_id: int,
|
||||
sections: list[dict[str, Any]],
|
||||
) -> int:
|
||||
"""측점별 횡단면 레코드를 일괄 저장하고 저장 건수를 반환한다.
|
||||
|
||||
각 section dict: {chainage_m, sequence_num, data, cross_section_file_path, status?}
|
||||
"""
|
||||
if not sections:
|
||||
return 0
|
||||
rows = []
|
||||
for section in sections:
|
||||
file_rel = _validate_stage_path(section["cross_section_file_path"])
|
||||
data = section.get("data")
|
||||
rows.append(
|
||||
(
|
||||
str(project_id),
|
||||
route_id,
|
||||
section.get("chainage_m"),
|
||||
section.get("sequence_num"),
|
||||
json.dumps(data, ensure_ascii=False) if data is not None else None,
|
||||
file_rel,
|
||||
section.get("status", "DRAFT"),
|
||||
)
|
||||
)
|
||||
async with connection.cursor() as cursor:
|
||||
await cursor.executemany(
|
||||
"""
|
||||
INSERT INTO cross_sections (
|
||||
project_id, route_id, chainage_m, sequence_num,
|
||||
data, cross_section_file_path, status
|
||||
)
|
||||
VALUES (%s, %s, %s, %s, %s, %s, %s)
|
||||
""",
|
||||
rows,
|
||||
)
|
||||
return len(rows)
|
||||
|
||||
|
||||
async def get_longitudinal_section(
|
||||
connection: aiomysql.Connection, project_id: UUID, route_id: int
|
||||
) -> dict[str, Any] | None:
|
||||
"""경로의 종단면 메타데이터를 조회한다 (없으면 None)."""
|
||||
async with connection.cursor() as cursor:
|
||||
await cursor.execute(
|
||||
"""
|
||||
SELECT id, data, longitudinal_file_path, status, computed_at
|
||||
FROM longitudinal_sections
|
||||
WHERE project_id = %s AND route_id = %s
|
||||
ORDER BY id DESC
|
||||
LIMIT 1
|
||||
""",
|
||||
(str(project_id), route_id),
|
||||
)
|
||||
row = await cursor.fetchone()
|
||||
if not row:
|
||||
return None
|
||||
data = row[1]
|
||||
if isinstance(data, str):
|
||||
data = json.loads(data)
|
||||
return {
|
||||
"id": int(row[0]),
|
||||
"data": data if isinstance(data, dict) else None,
|
||||
"longitudinal_file_path": row[2],
|
||||
"status": row[3],
|
||||
"computed_at": row[4].isoformat() if row[4] else None,
|
||||
}
|
||||
|
||||
|
||||
async def count_cross_sections(connection: aiomysql.Connection, route_id: int) -> int:
|
||||
"""경로에 저장된 횡단면 개수를 반환한다."""
|
||||
async with connection.cursor() as cursor:
|
||||
await cursor.execute("SELECT COUNT(*) FROM cross_sections WHERE route_id = %s", (route_id,))
|
||||
row = await cursor.fetchone()
|
||||
return int(row[0]) if row else 0
|
||||
|
||||
|
||||
async def update_cross_section_design(
|
||||
connection: aiomysql.Connection,
|
||||
*,
|
||||
route_id: int,
|
||||
chainage_m: float,
|
||||
design: dict[str, Any],
|
||||
project_id: UUID | None = None,
|
||||
) -> bool:
|
||||
"""측점 하나의 data.design(잠정 설계 지정·단면적)을 병합 저장한다.
|
||||
|
||||
기존 data 요약을 보존하고 design 키만 갱신한다. 대상 측점을 chainage 근사로
|
||||
찾으며(부동소수 오차 허용), 갱신 여부를 반환한다.
|
||||
구조물(비정규) 측점은 B05 확정이 파일만 쓰고 DB 행을 만들지 않으므로, 행이 없고
|
||||
project_id가 오면 새 행을 삽입한다(upsert — 구조물 측점 설계 저장 보장).
|
||||
"""
|
||||
async with connection.cursor() as cursor:
|
||||
await cursor.execute(
|
||||
"""
|
||||
SELECT id, data FROM cross_sections
|
||||
WHERE route_id = %s AND ABS(chainage_m - %s) < 0.01
|
||||
ORDER BY id DESC
|
||||
LIMIT 1
|
||||
""",
|
||||
(route_id, chainage_m),
|
||||
)
|
||||
row = await cursor.fetchone()
|
||||
if not row:
|
||||
if project_id is None:
|
||||
return False
|
||||
await cursor.execute(
|
||||
"""
|
||||
INSERT INTO cross_sections (project_id, route_id, chainage_m, data, status)
|
||||
VALUES (%s, %s, %s, %s, 'DRAFT')
|
||||
""",
|
||||
(
|
||||
str(project_id),
|
||||
route_id,
|
||||
chainage_m,
|
||||
json.dumps({"design": design}, ensure_ascii=False),
|
||||
),
|
||||
)
|
||||
return True
|
||||
data = row[1]
|
||||
if isinstance(data, str):
|
||||
data = json.loads(data)
|
||||
if not isinstance(data, dict):
|
||||
data = {}
|
||||
data["design"] = design
|
||||
await cursor.execute(
|
||||
"UPDATE cross_sections SET data = %s WHERE id = %s",
|
||||
(json.dumps(data, ensure_ascii=False), int(row[0])),
|
||||
)
|
||||
return True
|
||||
|
||||
|
||||
async def merge_cross_section_design_by_round(
|
||||
connection: aiomysql.Connection,
|
||||
*,
|
||||
route_id: int,
|
||||
chainage_int: int,
|
||||
patch: dict[str, Any],
|
||||
) -> bool:
|
||||
"""정수 chainage(m) 측점의 data.design에 patch를 병합 저장한다.
|
||||
|
||||
B07 도면 확정(전체 설계 재계산 결과) 및 확정 해제(status 되돌림)에서 사용한다.
|
||||
도면 ID가 정수 m라 ROUND로 매칭한다. 갱신 여부를 반환한다.
|
||||
"""
|
||||
async with connection.cursor() as cursor:
|
||||
await cursor.execute(
|
||||
"""
|
||||
SELECT id, data FROM cross_sections
|
||||
WHERE route_id = %s AND ROUND(chainage_m) = %s
|
||||
ORDER BY id DESC
|
||||
LIMIT 1
|
||||
""",
|
||||
(route_id, chainage_int),
|
||||
)
|
||||
row = await cursor.fetchone()
|
||||
if not row:
|
||||
return False
|
||||
data = row[1]
|
||||
if isinstance(data, str):
|
||||
data = json.loads(data)
|
||||
if not isinstance(data, dict):
|
||||
data = {}
|
||||
design = data.get("design")
|
||||
if not isinstance(design, dict):
|
||||
design = {}
|
||||
design.update(patch)
|
||||
data["design"] = design
|
||||
await cursor.execute(
|
||||
"UPDATE cross_sections SET data = %s WHERE id = %s",
|
||||
(json.dumps(data, ensure_ascii=False), int(row[0])),
|
||||
)
|
||||
return True
|
||||
|
||||
|
||||
async def get_cross_section_design(
|
||||
connection: aiomysql.Connection, route_id: int, chainage_int: int
|
||||
) -> dict[str, Any] | None:
|
||||
"""정수 chainage(m)에 해당하는 측점의 잠정 설계(data.design)를 반환한다.
|
||||
|
||||
B07이 도면 ID(cross_{정수m}m)로 조회하므로 ROUND로 근사 매칭한다.
|
||||
"""
|
||||
async with connection.cursor() as cursor:
|
||||
await cursor.execute(
|
||||
"""
|
||||
SELECT data FROM cross_sections
|
||||
WHERE route_id = %s AND ROUND(chainage_m) = %s
|
||||
ORDER BY id DESC
|
||||
LIMIT 1
|
||||
""",
|
||||
(route_id, chainage_int),
|
||||
)
|
||||
row = await cursor.fetchone()
|
||||
if not row or not row[0]:
|
||||
return None
|
||||
data = row[0]
|
||||
if isinstance(data, str):
|
||||
data = json.loads(data)
|
||||
design = data.get("design") if isinstance(data, dict) else None
|
||||
return design if isinstance(design, dict) and design.get("ground_type") else None
|
||||
|
||||
|
||||
async def get_cross_section_designs(
|
||||
connection: aiomysql.Connection, route_id: int
|
||||
) -> list[dict[str, Any]]:
|
||||
"""경로 측점별 저장된 설계 지정(data.design)을 chainage와 함께 반환한다.
|
||||
|
||||
상세 조회가 파일 기반이라 DB에만 있는 잠정 설계 지정을 화면 복원용으로 얹기 위함.
|
||||
"""
|
||||
async with connection.cursor() as cursor:
|
||||
await cursor.execute(
|
||||
"SELECT chainage_m, data FROM cross_sections WHERE route_id = %s", (route_id,)
|
||||
)
|
||||
rows = await cursor.fetchall()
|
||||
designs: list[dict[str, Any]] = []
|
||||
for row in rows:
|
||||
data = row[1]
|
||||
if isinstance(data, str):
|
||||
data = json.loads(data)
|
||||
design = data.get("design") if isinstance(data, dict) else None
|
||||
if isinstance(design, dict) and design.get("ground_type"):
|
||||
designs.append({"chainage_m": float(row[0]), "design": design})
|
||||
return designs
|
||||
|
||||
|
||||
async def count_cross_sections_without_design(
|
||||
connection: aiomysql.Connection, route_id: int
|
||||
) -> int:
|
||||
"""지반유형(data.design.ground_type)이 아직 지정되지 않은 측점 수를 반환한다.
|
||||
|
||||
확정 게이팅에 사용한다. JSON 함수 대신 애플리케이션에서 판정해
|
||||
MariaDB JSON 함수 가용성에 의존하지 않는다.
|
||||
"""
|
||||
async with connection.cursor() as cursor:
|
||||
await cursor.execute("SELECT data FROM cross_sections WHERE route_id = %s", (route_id,))
|
||||
rows = await cursor.fetchall()
|
||||
missing = 0
|
||||
for row in rows:
|
||||
data = row[0]
|
||||
if isinstance(data, str):
|
||||
data = json.loads(data)
|
||||
design = data.get("design") if isinstance(data, dict) else None
|
||||
if not isinstance(design, dict) or not design.get("ground_type"):
|
||||
missing += 1
|
||||
return missing
|
||||
|
||||
|
||||
async def get_cross_sections_missing_design_chainages(
|
||||
connection: aiomysql.Connection, route_id: int
|
||||
) -> list[float]:
|
||||
"""지반유형이 아직 지정되지 않은 측점의 chainage(m) 목록을 반환한다.
|
||||
|
||||
확정 시 기본값(토사/좌절토)으로 일괄 채우기 위해 사용한다.
|
||||
"""
|
||||
async with connection.cursor() as cursor:
|
||||
await cursor.execute(
|
||||
"SELECT chainage_m, data FROM cross_sections WHERE route_id = %s", (route_id,)
|
||||
)
|
||||
rows = await cursor.fetchall()
|
||||
chainages: list[float] = []
|
||||
for row in rows:
|
||||
data = row[1]
|
||||
if isinstance(data, str):
|
||||
data = json.loads(data)
|
||||
design = data.get("design") if isinstance(data, dict) else None
|
||||
if not isinstance(design, dict) or not design.get("ground_type"):
|
||||
chainages.append(float(row[0]))
|
||||
return chainages
|
||||
|
||||
|
||||
async def merge_cross_section_design_patch(
|
||||
connection: aiomysql.Connection,
|
||||
*,
|
||||
route_id: int,
|
||||
chainage_m: float,
|
||||
patch: dict[str, Any],
|
||||
) -> bool:
|
||||
"""측점 하나(부동소수 chainage 근사 매칭)의 data.design에 patch를 병합한다.
|
||||
|
||||
확정 시 프론트 세션 보관값(암 경계선 오프셋 등)을 기존 설계를 보존한 채 얹기
|
||||
위해 사용한다. design이 없던 측점이면 patch만으로 design을 만든다.
|
||||
"""
|
||||
if not patch:
|
||||
return False
|
||||
async with connection.cursor() as cursor:
|
||||
await cursor.execute(
|
||||
"""
|
||||
SELECT id, data FROM cross_sections
|
||||
WHERE route_id = %s AND ABS(chainage_m - %s) < 0.01
|
||||
ORDER BY id DESC
|
||||
LIMIT 1
|
||||
""",
|
||||
(route_id, chainage_m),
|
||||
)
|
||||
row = await cursor.fetchone()
|
||||
if not row:
|
||||
return False
|
||||
data = row[1]
|
||||
if isinstance(data, str):
|
||||
data = json.loads(data)
|
||||
if not isinstance(data, dict):
|
||||
data = {}
|
||||
design = data.get("design")
|
||||
if not isinstance(design, dict):
|
||||
design = {}
|
||||
design.update(patch)
|
||||
data["design"] = design
|
||||
await cursor.execute(
|
||||
"UPDATE cross_sections SET data = %s WHERE id = %s",
|
||||
(json.dumps(data, ensure_ascii=False), int(row[0])),
|
||||
)
|
||||
return True
|
||||
|
||||
|
||||
async def _patch_latest_longitudinal_data(
|
||||
connection: aiomysql.Connection,
|
||||
route_id: int,
|
||||
apply_patch: Callable[[dict[str, Any]], None],
|
||||
) -> bool:
|
||||
"""경로 최신 종단면 행의 data(JSON)를 읽어 apply_patch로 고친 뒤 되쓴다.
|
||||
|
||||
행이 없으면 아무것도 하지 않고 False를 반환한다.
|
||||
"""
|
||||
async with connection.cursor() as cursor:
|
||||
await cursor.execute(
|
||||
"""
|
||||
SELECT id, data FROM longitudinal_sections
|
||||
WHERE route_id = %s
|
||||
ORDER BY id DESC
|
||||
LIMIT 1
|
||||
""",
|
||||
(route_id,),
|
||||
)
|
||||
row = await cursor.fetchone()
|
||||
if not row:
|
||||
return False
|
||||
data = row[1]
|
||||
if isinstance(data, str):
|
||||
data = json.loads(data)
|
||||
if not isinstance(data, dict):
|
||||
data = {}
|
||||
apply_patch(data)
|
||||
await cursor.execute(
|
||||
"UPDATE longitudinal_sections SET data = %s WHERE id = %s",
|
||||
(json.dumps(data, ensure_ascii=False), int(row[0])),
|
||||
)
|
||||
return True
|
||||
|
||||
|
||||
async def merge_longitudinal_section_options(
|
||||
connection: aiomysql.Connection,
|
||||
*,
|
||||
route_id: int,
|
||||
options_patch: dict[str, Any],
|
||||
) -> bool:
|
||||
"""경로 최신 종단면 data.options에 patch를 병합 저장한다.
|
||||
|
||||
표준 횡단면 설정 등 확정 시점 옵션을 기존 생성 옵션 스냅샷을 보존한 채 갱신한다.
|
||||
갱신 여부를 반환한다.
|
||||
"""
|
||||
if not options_patch:
|
||||
return False
|
||||
|
||||
def apply_patch(data: dict[str, Any]) -> None:
|
||||
options = data.get("options")
|
||||
if not isinstance(options, dict):
|
||||
options = {}
|
||||
options.update(options_patch)
|
||||
data["options"] = options
|
||||
|
||||
return await _patch_latest_longitudinal_data(connection, route_id, apply_patch)
|
||||
|
||||
|
||||
async def merge_longitudinal_section_data(
|
||||
connection: aiomysql.Connection,
|
||||
*,
|
||||
route_id: int,
|
||||
data_patch: dict[str, Any],
|
||||
) -> bool:
|
||||
"""경로 최신 종단면 data의 **최상위 키**에 patch를 병합 저장한다.
|
||||
|
||||
생성 옵션이 아닌 산출 결과(유토곡선 등)를 보관하는 데 쓴다. 갱신 여부를 반환한다.
|
||||
"""
|
||||
if not data_patch:
|
||||
return False
|
||||
return await _patch_latest_longitudinal_data(
|
||||
connection, route_id, lambda data: data.update(data_patch)
|
||||
)
|
||||
|
||||
|
||||
async def confirm_sections_for_route(connection: aiomysql.Connection, route_id: int) -> None:
|
||||
"""경로의 종횡단면 상태를 CONFIRMED로 변경한다."""
|
||||
async with connection.cursor() as cursor:
|
||||
await cursor.execute(
|
||||
"UPDATE longitudinal_sections SET status = 'CONFIRMED' WHERE route_id = %s",
|
||||
(route_id,),
|
||||
)
|
||||
await cursor.execute(
|
||||
"UPDATE cross_sections SET status = 'CONFIRMED' WHERE route_id = %s",
|
||||
(route_id,),
|
||||
)
|
||||
Reference in New Issue
Block a user