eomsangdonandClaude Fable 5
6e195afb69
횡단설계(B06) 다음을 상세설계 → 수량산출 → 설계도서 순으로 재배열하고, 폴더 번호가 흐름과 일치하도록 이름을 맞바꾼다. - B08_DesignDetail → B07_DesignDetail, B07_Quantity → B08_Quantity (파일 접두어·식별자·라우트·locale 키 전량 스왑) - STAGE_KEYS 4=DESIGN_DETAIL, 5=QUANTITY 스왑 + 라우터 stage 리터럴 교체 - CAD 마운트 /b08-cad → /b07-cad (main.py·vite proxy·iframe URL), openwebcad Toolbar 라벨 B07로 수정 후 재빌드 - 유지: openwebcad postMessage 프로토콜 aislo:b08:*·패키지명(내부 식별자) - 기존 프로젝트 storage 폴더 rename + project_manifest 갱신, DB project_workflow_stages stage_no 4↔5 행 스왑 완료 Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
665 lines
23 KiB
Python
665 lines
23 KiB
Python
"""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)를 반환한다.
|
|
|
|
B08이 도면 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 get_cross_section_chainages(
|
|
connection: aiomysql.Connection, route_id: int
|
|
) -> list[float]:
|
|
"""경로에 **행이 존재하는** 측점의 chainage(m) 목록. 구조물(비정규) 측점처럼
|
|
행 자체가 없는 자리를 가려내 확정 때 정본으로 채우기 위해 쓴다(2026-08-24)."""
|
|
async with connection.cursor() as cursor:
|
|
await cursor.execute(
|
|
"SELECT chainage_m FROM cross_sections WHERE route_id = %s", (route_id,)
|
|
)
|
|
rows = await cursor.fetchall()
|
|
return [float(row[0]) for row in rows]
|
|
|
|
|
|
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,),
|
|
)
|