543 lines
22 KiB
Python
543 lines
22 KiB
Python
import os
|
|
from typing import Optional, List, Dict
|
|
import httpx
|
|
|
|
from flask import Flask, g, render_template, request, abort
|
|
from urllib.parse import urlencode
|
|
from dotenv import load_dotenv
|
|
|
|
|
|
|
|
APP_NAME = "1호선 고장코드"
|
|
|
|
|
|
def create_app() -> Flask:
|
|
app = Flask(__name__)
|
|
app.config.update(TEMPLATES_AUTO_RELOAD=True)
|
|
|
|
# 환경변수 로드 및 Supabase 기본값 설정
|
|
load_dotenv()
|
|
# 기본: Kong 프록시(8000) 또는 사용자가 지정한 URL
|
|
app.config.setdefault("SUPABASE_URL", os.environ.get("SUPABASE_URL", "http://192.168.0.180:8000"))
|
|
app.config.setdefault("SUPABASE_ANON_KEY", os.environ.get("SUPABASE_ANON_KEY", "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJyb2xlIjoiYW5vbiIsImlzcyI6InN1cGFiYXNlIiwiaWF0IjoxNzU4NTUxNjY2LCJleHAiOjQxMDI0NDQ4MDB9.jMCGL3Q-N2o_l7JQE_HrO7Uoct86CMgLsVxpabisG4I"))
|
|
# Kong Basic Auth(선택)
|
|
app.config.setdefault("SUPABASE_BASIC_USER", os.environ.get("SUPABASE_BASIC_USER", ""))
|
|
app.config.setdefault("SUPABASE_BASIC_PASSWORD", os.environ.get("SUPABASE_BASIC_PASSWORD", ""))
|
|
|
|
# 더 이상 SQLite 초기화/연결을 사용하지 않음 (Supabase만 사용)
|
|
|
|
# PostgREST(REST) 클라이언트 빌더
|
|
def build_pg_client() -> httpx.Client:
|
|
base = app.config["SUPABASE_URL"].rstrip("/") + "/rest/v1"
|
|
basic_user = app.config.get("SUPABASE_BASIC_USER") or ""
|
|
basic_pass = app.config.get("SUPABASE_BASIC_PASSWORD") or ""
|
|
headers = {
|
|
"apikey": app.config.get("SUPABASE_ANON_KEY", ""),
|
|
"Accept-Profile": "public",
|
|
"Content-Profile": "public",
|
|
}
|
|
auth = httpx.BasicAuth(basic_user, basic_pass) if basic_user else None
|
|
if not auth and app.config.get("SUPABASE_ANON_KEY"):
|
|
headers["Authorization"] = f"Bearer {app.config['SUPABASE_ANON_KEY']}"
|
|
return httpx.Client(base_url=base, headers=headers, auth=auth, timeout=10)
|
|
|
|
def pg_unique(col: str) -> List[str]:
|
|
with build_pg_client() as c:
|
|
r = c.get("/Fault_Code_Table", params={"select": col})
|
|
r.raise_for_status()
|
|
vals = [row.get(col) for row in (r.json() or []) if row.get(col)]
|
|
seen: Dict[str, bool] = {}
|
|
out: List[str] = []
|
|
for v in vals:
|
|
if v not in seen:
|
|
seen[v] = True
|
|
out.append(v)
|
|
return out
|
|
|
|
def pg_unique_from(table: str, col: str) -> List[str]:
|
|
"""지정 테이블에서 고유값 리스트를 반환한다."""
|
|
with build_pg_client() as c:
|
|
r = c.get(f"/{table}", params={"select": col})
|
|
r.raise_for_status()
|
|
vals = [row.get(col) for row in (r.json() or []) if row.get(col)]
|
|
seen: Dict[str, bool] = {}
|
|
out: List[str] = []
|
|
for v in vals:
|
|
if v not in seen:
|
|
seen[v] = True
|
|
out.append(v)
|
|
return out
|
|
|
|
@app.route("/")
|
|
def index():
|
|
return render_template(
|
|
"index.html",
|
|
app_name=APP_NAME,
|
|
)
|
|
|
|
# 기존 SQLite 기반 라우트 제거됨
|
|
|
|
@app.route("/modal/close")
|
|
def modal_close():
|
|
return ""
|
|
|
|
@app.route("/health")
|
|
def health():
|
|
return {"status": "ok"}
|
|
|
|
# 빈 파비콘 응답으로 404 제거
|
|
@app.route("/favicon.ico")
|
|
def favicon():
|
|
return ("", 204, {"Content-Type": "image/x-icon"})
|
|
|
|
# 브라우저/툴 호환을 위해 루트 경로에도 매니페스트 노출 (정적 파일 경로 사용 권장)
|
|
# 더 이상 /api/* 동기화 엔드포인트 제공하지 않음
|
|
|
|
# ----------------- Supabase 기반 라우트 -----------------
|
|
@app.route("/sb/tabs")
|
|
def sb_tabs():
|
|
try:
|
|
with build_pg_client() as c:
|
|
r1 = c.get("/Fault_Code_Table", params={"select": "manufacturer"})
|
|
r1.raise_for_status()
|
|
vals1 = [row.get("manufacturer") for row in (r1.json() or []) if row.get("manufacturer")]
|
|
vals2: List[str] = []
|
|
for path in ("/Signals", "/signals"):
|
|
try:
|
|
r2 = c.get(path, params={"select": "manufacturer"})
|
|
r2.raise_for_status()
|
|
vals2 = [row.get("manufacturer") for row in (r2.json() or []) if row.get("manufacturer")]
|
|
break
|
|
except httpx.HTTPError:
|
|
continue
|
|
vals = vals1 + vals2
|
|
seen: Dict[str, bool] = {}
|
|
manufacturers: List[str] = []
|
|
for v in vals:
|
|
if v not in seen:
|
|
seen[v] = True
|
|
manufacturers.append(v)
|
|
return render_template("partials/sb_tabs.html", manufacturers=manufacturers)
|
|
except Exception as e:
|
|
return render_template(
|
|
"partials/sb_error.html",
|
|
error_message=str(e),
|
|
supabase_url=app.config.get("SUPABASE_URL"),
|
|
)
|
|
@app.route("/sb")
|
|
def sb_home():
|
|
try:
|
|
# ---- MMI 코드 분기 ----
|
|
if request.args.get("section", "fault").strip() == "mmicode":
|
|
PAGE_SIZE = 50
|
|
page = int(request.args.get("page", "0"))
|
|
offset = page * PAGE_SIZE
|
|
|
|
# 셀렉트박스/검색 필드
|
|
with build_pg_client() as c:
|
|
# 제조사
|
|
res = c.get("/MMI_Code", params={"select": "manufacturer"})
|
|
res.raise_for_status()
|
|
manufacturers = sorted({row.get("manufacturer") for row in (res.json() or []) if row.get("manufacturer")})
|
|
# 차량분류(alias_name)
|
|
res2 = c.get("/MMI_Code", params={"select": "alias_name"})
|
|
res2.raise_for_status()
|
|
alias_names = sorted({row.get("alias_name") for row in (res2.json() or []) if row.get("alias_name")})
|
|
|
|
selected_manufacturer = request.args.get("manufacturer", "").strip()
|
|
selected_alias_name = request.args.get("alias_name", "").strip()
|
|
q = request.args.get("q", "").strip()
|
|
group_code = request.args.get("group_code", "").strip().lower() == "on"
|
|
# MMI 쿼리
|
|
params = {
|
|
"select": "id,code_name,code_description,data_type,car_id,alias_name,manufacturer",
|
|
"order": "code_name.asc",
|
|
"limit": str(PAGE_SIZE),
|
|
"offset": str(offset),
|
|
}
|
|
if selected_manufacturer:
|
|
params["manufacturer"] = f"eq.{selected_manufacturer}"
|
|
if selected_alias_name:
|
|
params["alias_name"] = f"eq.{selected_alias_name}"
|
|
if q:
|
|
# 여러 컬럼에 대해 ilike filter
|
|
params["or"] = f"(code_name.ilike.*{q}*,code_description.ilike.*{q}*,alias_name.ilike.*{q}*)"
|
|
|
|
with build_pg_client() as c:
|
|
res = c.get("/MMI_Code", params=params)
|
|
res.raise_for_status()
|
|
rows = res.json() or []
|
|
# dedup if group_code on
|
|
if group_code:
|
|
seen = set()
|
|
dedup = []
|
|
for r in rows:
|
|
code = r.get("code_name")
|
|
if code and code not in seen:
|
|
seen.add(code)
|
|
dedup.append(r)
|
|
rows = dedup
|
|
|
|
query_params = request.args.to_dict()
|
|
if "page" in query_params:
|
|
del query_params["page"]
|
|
query_params_string = urlencode(query_params)
|
|
|
|
# 템플릿 렌더 및 선택 파라미터 전달
|
|
return render_template(
|
|
"partials/sb_mmi_list.html",
|
|
rows=rows,
|
|
manufacturers=manufacturers,
|
|
alias_names=alias_names,
|
|
selected_manufacturer=selected_manufacturer,
|
|
selected_alias_name=selected_alias_name,
|
|
q=q,
|
|
group_code="on" if group_code else "off",
|
|
section="mmicode",
|
|
page=page,
|
|
page_size=PAGE_SIZE,
|
|
query_params_string=query_params_string,
|
|
)
|
|
|
|
# ---- 기존(고장코드/TCMS) ----
|
|
mf_fault = pg_unique("manufacturer")
|
|
mf_signal = pg_unique_from("Signals", "manufacturer")
|
|
seen_mf: Dict[str, bool] = {}
|
|
manufacturers: List[str] = []
|
|
for v in (mf_fault + mf_signal):
|
|
if v and v not in seen_mf:
|
|
seen_mf[v] = True
|
|
manufacturers.append(v)
|
|
|
|
devices = pg_unique("device")
|
|
car_types = pg_unique("car_type")
|
|
car_ids = pg_unique("car_id")
|
|
# alias_name 필터는 제작사에 따라 값 제한: 우진/로템/다대 포함
|
|
all_alias = pg_unique("alias_name")
|
|
selected_manufacturer = request.args.get("manufacturer", "").strip()
|
|
def alias_allowed(name: str) -> bool:
|
|
if not name:
|
|
return False
|
|
low = name.lower()
|
|
if selected_manufacturer.lower() == "woojin":
|
|
return ("우진" in name) or ("woojin" in low)
|
|
if selected_manufacturer.lower() == "rotem":
|
|
return ("로템" in name) or ("다대" in name) or ("rotem" in low)
|
|
return True
|
|
alias_names = [a for a in all_alias if alias_allowed(a)]
|
|
signal_classifications = pg_unique_from("Signals", "classification")
|
|
|
|
# 선택값 처리 (쿼리스트링에 manufacturer 키가 있으면 빈 문자열이라도 그대로 유지)
|
|
if "manufacturer" in request.args:
|
|
selected_manufacturer = (request.args.get("manufacturer") or "").strip()
|
|
else:
|
|
selected_manufacturer = manufacturers[0] if manufacturers else ""
|
|
|
|
# 다른 필터의 현재 선택값 유지
|
|
selected_device = request.args.get("device", "").strip()
|
|
selected_car_type = request.args.get("car_type", "").strip()
|
|
selected_alias_name = request.args.get("alias_name", "").strip()
|
|
selected_classification = request.args.get("classification", "").strip()
|
|
q = request.args.get("q", "").strip()
|
|
section = request.args.get("section", "fault").strip() or "fault"
|
|
|
|
return render_template(
|
|
"partials/sb_manufacturer.html",
|
|
manufacturers=manufacturers,
|
|
devices=devices,
|
|
car_types=car_types,
|
|
car_ids=car_ids,
|
|
alias_names=alias_names,
|
|
signal_classifications=signal_classifications,
|
|
selected_manufacturer=selected_manufacturer,
|
|
selected_device=selected_device,
|
|
selected_car_type=selected_car_type,
|
|
selected_alias_name=selected_alias_name,
|
|
selected_classification=selected_classification,
|
|
q=q,
|
|
section=section,
|
|
)
|
|
except Exception as e:
|
|
return render_template(
|
|
"partials/sb_error.html",
|
|
error_message=str(e),
|
|
supabase_url=app.config.get("SUPABASE_URL"),
|
|
)
|
|
|
|
@app.route("/sb/faults/list")
|
|
def sb_faults_list():
|
|
try:
|
|
PAGE_SIZE = 50
|
|
page = int(request.args.get("page", "0"))
|
|
offset = page * PAGE_SIZE
|
|
|
|
manufacturer = request.args.get("manufacturer", "").strip()
|
|
device = request.args.get("device", "").strip()
|
|
car_type = request.args.get("car_type", "").strip()
|
|
car_id = request.args.get("car_id", "").strip()
|
|
alias_name = request.args.get("alias_name", "").strip()
|
|
q = request.args.get("q", "").strip()
|
|
section = request.args.get("section", "fault").strip()
|
|
params: Dict[str, str] = {
|
|
"select": "f_code,f_code_num,f_name,manufacturer,device,car_type,car_id,fault_detail,alias_name",
|
|
"order": "f_code.asc",
|
|
"limit": str(PAGE_SIZE),
|
|
"offset": str(offset),
|
|
}
|
|
if manufacturer:
|
|
params["manufacturer"] = f"eq.{manufacturer}"
|
|
if device:
|
|
params["device"] = f"eq.{device}"
|
|
if car_type:
|
|
params["car_type"] = f"eq.{car_type}"
|
|
if car_id:
|
|
params["car_id"] = f"eq.{car_id}"
|
|
if alias_name:
|
|
params["alias_name"] = f"eq.{alias_name}"
|
|
if q:
|
|
params["or"] = f"(f_code.ilike.*{q}*,f_name.ilike.*{q}*)"
|
|
|
|
with build_pg_client() as c:
|
|
r = c.get("/Fault_Code_Table", params=params)
|
|
r.raise_for_status()
|
|
rows = r.json() or []
|
|
# 그룹핑 옵션: 같은 f_code를 첫 항목만 남김
|
|
group_code = request.args.get("group_code", "").strip().lower()
|
|
if group_code == "on":
|
|
seen: Dict[str, bool] = {}
|
|
dedup: List[Dict] = []
|
|
for r in rows:
|
|
code = r.get("f_code")
|
|
if code and code not in seen:
|
|
seen[code] = True
|
|
dedup.append(r)
|
|
rows = dedup
|
|
|
|
query_params = request.args.to_dict()
|
|
if "page" in query_params:
|
|
del query_params["page"]
|
|
query_params_string = urlencode(query_params)
|
|
|
|
return render_template(
|
|
"partials/sb_fault_list.html",
|
|
rows=rows,
|
|
page=page,
|
|
page_size=PAGE_SIZE,
|
|
query_params_string=query_params_string,
|
|
)
|
|
except Exception as e:
|
|
return render_template(
|
|
"partials/sb_error.html",
|
|
error_message=str(e),
|
|
supabase_url=app.config.get("SUPABASE_URL"),
|
|
)
|
|
|
|
@app.route("/sb/faults/<string:f_code>")
|
|
def sb_fault_detail(f_code: str):
|
|
try:
|
|
params = {
|
|
"select": "f_code,f_code_num,f_name,car_type,f_class,grade,device,fault_detail,fault_reaction,fault_detection,fault_clear,fault_action,fault_schematics,car_id,manufacturer",
|
|
"f_code": f"eq.{f_code}",
|
|
"limit": "1",
|
|
}
|
|
with build_pg_client() as c:
|
|
r = c.get("/Fault_Code_Table", params=params)
|
|
r.raise_for_status()
|
|
rows = r.json() or []
|
|
if not rows:
|
|
abort(404)
|
|
row = rows[0]
|
|
return render_template("partials/sb_fault_detail.html", row=row)
|
|
except Exception as e:
|
|
return render_template(
|
|
"partials/sb_error.html",
|
|
error_message=str(e),
|
|
supabase_url=app.config.get("SUPABASE_URL"),
|
|
)
|
|
|
|
# ----------------- Signals (TCMS) -----------------
|
|
@app.route("/sb/signals/list")
|
|
def sb_signals_list():
|
|
try:
|
|
PAGE_SIZE = 50
|
|
page = int(request.args.get("page", "0"))
|
|
offset = page * PAGE_SIZE
|
|
|
|
manufacturer = request.args.get("manufacturer", "").strip()
|
|
classification = request.args.get("classification", "").strip()
|
|
q = request.args.get("q", "").strip()
|
|
alias_name = request.args.get("alias_name", "").strip()
|
|
|
|
params: Dict[str, str] = {
|
|
"select": "id,sig_num,signal_abbreviation,signal_description,status_value,manufacturer,classification,alias_name",
|
|
"order": "sig_num.asc",
|
|
"limit": str(PAGE_SIZE),
|
|
"offset": str(offset),
|
|
}
|
|
if manufacturer:
|
|
params["manufacturer"] = f"eq.{manufacturer}"
|
|
if classification:
|
|
params["classification"] = f"eq.{classification}"
|
|
if alias_name:
|
|
params["alias_name"] = f"eq.{alias_name}"
|
|
if q:
|
|
params["or"] = f"(signal_abbreviation.ilike.*{q}*,signal_description.ilike.*{q}*)"
|
|
|
|
with build_pg_client() as c:
|
|
rows = []
|
|
last_error: Exception | None = None
|
|
for path in ("/Signals", "/signals"):
|
|
try:
|
|
r = c.get(path, params=params)
|
|
r.raise_for_status()
|
|
rows = r.json() or []
|
|
last_error = None
|
|
break
|
|
except Exception as e:
|
|
last_error = e
|
|
rows = []
|
|
continue
|
|
if last_error and not rows:
|
|
raise last_error
|
|
# TCMS도 group_code=on이면 sig_num/dedup(또는 id?)
|
|
group_code = request.args.get("group_code", "").strip().lower()
|
|
if group_code == "on":
|
|
seen: Dict[str, bool] = {}
|
|
dedup: List[Dict] = []
|
|
for r in rows:
|
|
code = r.get("id")
|
|
if code and code not in seen:
|
|
seen[code] = True
|
|
dedup.append(r)
|
|
rows = dedup
|
|
|
|
query_params = request.args.to_dict()
|
|
if "page" in query_params:
|
|
del query_params["page"]
|
|
query_params_string = urlencode(query_params)
|
|
|
|
return render_template(
|
|
"partials/sb_signal_list.html",
|
|
rows=rows,
|
|
page=page,
|
|
page_size=PAGE_SIZE,
|
|
query_params_string=query_params_string,
|
|
)
|
|
except Exception as e:
|
|
return render_template(
|
|
"partials/sb_error.html",
|
|
error_message=str(e),
|
|
supabase_url=app.config.get("SUPABASE_URL"),
|
|
)
|
|
|
|
@app.route("/sb/signals/<string:item_id>")
|
|
def sb_signal_detail(item_id: str):
|
|
try:
|
|
params = {
|
|
"select": "id,sig_num,signal_abbreviation,signal_description,status_value,manufacturer,classification,alias_name,original_data,created_at,updated_at",
|
|
"id": f"eq.{item_id}",
|
|
"limit": "1",
|
|
}
|
|
with build_pg_client() as c:
|
|
rows = []
|
|
last_error: Exception | None = None
|
|
for path in ("/Signals", "/signals"):
|
|
try:
|
|
r = c.get(path, params=params)
|
|
r.raise_for_status()
|
|
rows = r.json() or []
|
|
last_error = None
|
|
break
|
|
except Exception as e:
|
|
last_error = e
|
|
rows = []
|
|
continue
|
|
if last_error and not rows:
|
|
raise last_error
|
|
if not rows:
|
|
abort(404)
|
|
row = rows[0]
|
|
return render_template("partials/sb_signal_detail.html", row=row)
|
|
except Exception as e:
|
|
return render_template(
|
|
"partials/sb_error.html",
|
|
error_message=str(e),
|
|
supabase_url=app.config.get("SUPABASE_URL"),
|
|
)
|
|
|
|
@app.route("/sb/health")
|
|
def sb_health():
|
|
try:
|
|
with build_pg_client() as c:
|
|
r = c.get("/Fault_Code_Table", params={"select": "f_code", "limit": "1"})
|
|
r.raise_for_status()
|
|
return {"sb": "ok", "url": app.config.get("SUPABASE_URL")}
|
|
except Exception as e:
|
|
return {"sb": "error", "url": app.config.get("SUPABASE_URL"), "error": str(e)}
|
|
|
|
# 간단한 Signals 테스트 엔드포인트: 상위 5개 레코드를 JSON으로 반환
|
|
@app.route("/sb/signals/test")
|
|
def sb_signals_test():
|
|
try:
|
|
params = {
|
|
"select": "uuid,sig_num,signal_abbreviation,manufacturer,classification,alias_name",
|
|
"order": "sig_num.asc",
|
|
"limit": "5",
|
|
}
|
|
with build_pg_client() as c:
|
|
rows = []
|
|
used_path = None
|
|
last_error = None
|
|
for path in ("/Signals", "/signals"):
|
|
try:
|
|
r = c.get(path, params=params)
|
|
r.raise_for_status()
|
|
rows = r.json() or []
|
|
used_path = path
|
|
last_error = None
|
|
break
|
|
except Exception as e:
|
|
last_error = str(e)
|
|
rows = []
|
|
continue
|
|
if last_error and not rows:
|
|
return {"ok": False, "url": app.config.get("SUPABASE_URL"), "tried": ["/Signals", "/signals"], "error": last_error}, 500
|
|
return {"ok": True, "url": app.config.get("SUPABASE_URL"), "path": used_path, "rows": rows}
|
|
except Exception as e:
|
|
return {"ok": False, "url": app.config.get("SUPABASE_URL"), "error": str(e)}, 500
|
|
|
|
# 상세 디버그: 두 경로 각각의 상태/본문을 그대로 반환
|
|
@app.route("/sb/signals/debug")
|
|
def sb_signals_debug():
|
|
try:
|
|
out = []
|
|
with build_pg_client() as c:
|
|
for path in ("/Signals", "/signals"):
|
|
try:
|
|
r = c.get(path, params={"select": "*", "limit": "5"})
|
|
ct = r.headers.get("content-type", "")
|
|
body = None
|
|
try:
|
|
body = r.json()
|
|
except Exception:
|
|
body = r.text
|
|
out.append({
|
|
"path": path,
|
|
"status": r.status_code,
|
|
"content_type": ct,
|
|
"body": body,
|
|
})
|
|
except httpx.HTTPError as e:
|
|
out.append({"path": path, "error": str(e)})
|
|
return {"ok": True, "url": app.config.get("SUPABASE_URL"), "results": out}
|
|
except Exception as e:
|
|
return {"ok": False, "url": app.config.get("SUPABASE_URL"), "error": str(e)}, 500
|
|
|
|
return app
|
|
|
|
|
|
app = create_app()
|
|
if __name__ == "__main__":
|
|
port = int(os.environ.get("PORT", "5000"))
|
|
app.run(host="0.0.0.0", port=port, debug=True)
|