1f5b393675
Both tables are ReplacingMergeTree; dedup happens only during background part merges, not at read time. The ongoing etl_rw feed keeps inserting into pf_board, so right after an insert that updates an existing row there is a short window (before the forced merge, min_age_to_force_merge_seconds=15) where the same task_id / board_key exists in two parts. A plain SELECT reads both parts and returns the row twice. The /api/boards query already collapses duplicates via argMax + GROUP BY, but /api/bookings and the two /api/map queries read raw rows, so they could intermittently double a booking (timeline/balloon) or a surface marker. Add FINAL to those three reads (pf_board in bookings + map current, board_info in map surfaces). At 463/515 rows the cost is negligible and it removes the class of transient duplicates regardless of merge timing. Output verified unchanged against the single-part baseline. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
297 lines
13 KiB
Python
297 lines
13 KiB
Python
import os
|
|
import httpx
|
|
from fastapi import FastAPI, Query
|
|
from fastapi.responses import FileResponse
|
|
from fastapi.staticfiles import StaticFiles
|
|
|
|
CH_HOST = os.environ.get("CLICKHOUSE_HOST", "ClickHouse")
|
|
CH_PORT = os.environ.get("CLICKHOUSE_PORT", "8123")
|
|
CH_USER = os.environ.get("CLICKHOUSE_USER", "default")
|
|
# Credentials come from the environment (see .env / docker-compose env_file).
|
|
# No secret is kept in source; empty default fails fast if the env is missing.
|
|
CH_PASSWORD = os.environ.get("CLICKHOUSE_PASSWORD", "")
|
|
CH_URL = f"http://{CH_HOST}:{CH_PORT}/"
|
|
|
|
app = FastAPI()
|
|
|
|
# --- Manager name normalization -------------------------------------------
|
|
# Planfix has emitted manager names in two word orders over time ("Имя Фамилия"
|
|
# from the 26 Jul 2026 backfill, "Фамилия Имя" from later re-syncs), so a single
|
|
# person shows up as two distinct strings (e.g. "Александра Столбова" and
|
|
# "Столбова Александра"). We collapse them with a word-order- and case-independent
|
|
# key: lowercase, split into words, sort the words, rejoin. The SQL key
|
|
# expression (MANAGER_KEY_ARR) and the Python key (_mkey) must stay in sync so a
|
|
# selected manager matches rows stored under either order. For display we map the
|
|
# key back to the canonical "Фамилия Имя" form in CANONICAL_MANAGERS.
|
|
MANAGER_KEY_ARR = (
|
|
"arrayMap(x -> arrayStringConcat("
|
|
"arraySort(splitByChar(' ', lowerUTF8(trimBoth(x)))), ' '), manager)"
|
|
)
|
|
|
|
# Canonical display names, in the chosen "Фамилия Имя" order. Unknown names fall
|
|
# through untouched (see canon_manager), so new managers still appear as-is.
|
|
CANONICAL_MANAGERS = [
|
|
"Вербицкий Александр",
|
|
"Иванова Диана",
|
|
"Кабатов Евгений",
|
|
"Колбина Ирина",
|
|
"Столбова Александра",
|
|
]
|
|
|
|
|
|
def _mkey(name: str) -> str:
|
|
return " ".join(sorted(name.lower().split()))
|
|
|
|
|
|
_MANAGER_CANON = {_mkey(n): n for n in CANONICAL_MANAGERS}
|
|
|
|
|
|
def canon_manager(name: str) -> str:
|
|
# Map any word order / casing of a known manager to the canonical
|
|
# "Фамилия Имя" string; leave unknown names untouched.
|
|
return _MANAGER_CANON.get(_mkey(name), name)
|
|
# --------------------------------------------------------------------------
|
|
|
|
CITY_COND = "(length({cities:Array(String)}) = 0 OR city IN {cities:Array(String)})"
|
|
DIM_COND = "(length({dimensions:Array(String)}) = 0 OR dimension IN {dimensions:Array(String)})"
|
|
BRAND_COND = "(length({brands:Array(String)}) = 0 OR brand IN {brands:Array(String)})"
|
|
# manager is Array(String) in pf_board (a task may have several managers), so we
|
|
# match with hasAny: keep the row if any selected manager is among its managers.
|
|
MANAGER_COND = f"(length({{managers:Array(String)}}) = 0 OR hasAny({MANAGER_KEY_ARR}, {{managers:Array(String)}}))"
|
|
STATUS_COND = "(length({statuses:Array(String)}) = 0 OR task_status IN {statuses:Array(String)})"
|
|
SEARCH_COND = (
|
|
"({search:String} = ''"
|
|
" OR positionCaseInsensitive(address, {search:String}) > 0"
|
|
" OR positionCaseInsensitive(board_id, {search:String}) > 0)"
|
|
)
|
|
# Independent date bounds: start_date >= X ("Дата начала"), end_date <= Y
|
|
# ("Дата окончания"). Each included only when its value is provided.
|
|
# NB: the bookings SELECT aliases toString(start_date) AS start_date, which would
|
|
# shadow the Date column in WHERE and cause a String-vs-Date type error. Qualify
|
|
# with the table name so the comparison always binds to the real Date column.
|
|
START_COND = "pf_board.start_date >= {date_start:Date}"
|
|
END_COND = "pf_board.end_date <= {date_end:Date}"
|
|
|
|
# Map endpoint filters board_info (the surface inventory that carries coordinates).
|
|
# board_info has no `brand` column, so the brand facet is expressed as "this
|
|
# surface has at least one booking of the selected brand" via a subquery on pf_board.
|
|
MAP_BRAND_COND = (
|
|
"(length({brands:Array(String)}) = 0"
|
|
" OR board_key IN (SELECT board_key FROM default.pf_board WHERE brand IN {brands:Array(String)}))"
|
|
)
|
|
# Same idea for the manager facet on the map: keep surfaces that have at least one
|
|
# booking managed by one of the selected managers.
|
|
MAP_MANAGER_COND = (
|
|
"(length({managers:Array(String)}) = 0"
|
|
f" OR board_key IN (SELECT board_key FROM default.pf_board WHERE hasAny({MANAGER_KEY_ARR}, {{managers:Array(String)}})))"
|
|
)
|
|
# Same idea for the status facet on the map: keep surfaces that have at least one
|
|
# booking whose task_status is among the selected values.
|
|
MAP_STATUS_COND = (
|
|
"(length({statuses:Array(String)}) = 0"
|
|
" OR board_key IN (SELECT board_key FROM default.pf_board WHERE task_status IN {statuses:Array(String)}))"
|
|
)
|
|
|
|
|
|
def build_where(city=False, dimension=False, brand=False, manager=False, status=False, search=False, date_start=False, date_end=False):
|
|
# Faceted WHERE: include only the requested facet conditions. For the
|
|
# dependent-filter option lists we exclude a facet's own condition so its
|
|
# available values reflect the OTHER filters (all-but-self).
|
|
parts = ["char_length(city) > 0"]
|
|
if city:
|
|
parts.append(CITY_COND)
|
|
if dimension:
|
|
parts.append(DIM_COND)
|
|
if brand:
|
|
parts.append(BRAND_COND)
|
|
if manager:
|
|
parts.append(MANAGER_COND)
|
|
if status:
|
|
parts.append(STATUS_COND)
|
|
if search:
|
|
parts.append(SEARCH_COND)
|
|
if date_start:
|
|
parts.append(START_COND)
|
|
if date_end:
|
|
parts.append(END_COND)
|
|
return "\n AND ".join(parts)
|
|
|
|
|
|
async def ch_query(sql: str, params: dict):
|
|
form = {"query": sql, "default_format": "JSONEachRow"}
|
|
for k, v in params.items():
|
|
if isinstance(v, list):
|
|
form[f"param_{k}"] = "[" + ",".join("'" + str(x).replace("'", "\\'") + "'" for x in v) + "]"
|
|
else:
|
|
form[f"param_{k}"] = str(v)
|
|
async with httpx.AsyncClient(timeout=15) as client:
|
|
r = await client.post(CH_URL, params=form, auth=(CH_USER, CH_PASSWORD))
|
|
r.raise_for_status()
|
|
text = r.text.strip()
|
|
if not text:
|
|
return []
|
|
import json
|
|
return [json.loads(line) for line in text.splitlines()]
|
|
|
|
|
|
def parse_filters(city: str, dimension: str, brand: str, manager: str, status: str, search: str, date_start: str, date_end: str):
|
|
cities = [c for c in city.split(",") if c] if city else []
|
|
dimensions = [d for d in dimension.split(",") if d] if dimension else []
|
|
brands = [b for b in brand.split(",") if b] if brand else []
|
|
# Normalize each selected manager to its word-order-independent key so it
|
|
# matches rows stored under either name order (see MANAGER_KEY_ARR).
|
|
managers = [_mkey(m) for m in manager.split(",") if m] if manager else []
|
|
statuses = [s for s in status.split(",") if s] if status else []
|
|
return {
|
|
"cities": cities,
|
|
"dimensions": dimensions,
|
|
"brands": brands,
|
|
"managers": managers,
|
|
"statuses": statuses,
|
|
"search": search or "",
|
|
"date_start": date_start or "",
|
|
"date_end": date_end or "",
|
|
}
|
|
|
|
|
|
@app.get("/api/meta")
|
|
async def meta(city: str = "", dimension: str = "", brand: str = "", manager: str = "", status: str = "", search: str = "", date_start: str = "", date_end: str = ""):
|
|
params = parse_filters(city, dimension, brand, manager, status, search, date_start, date_end)
|
|
hds = bool(params["date_start"])
|
|
hde = bool(params["date_end"])
|
|
cities = await ch_query(
|
|
f"SELECT DISTINCT city FROM default.pf_board WHERE {build_where(dimension=True, brand=True, manager=True, status=True, search=True, date_start=hds, date_end=hde)} AND char_length(city) > 0 ORDER BY city",
|
|
params,
|
|
)
|
|
brands = await ch_query(
|
|
f"SELECT DISTINCT brand FROM default.pf_board WHERE {build_where(city=True, dimension=True, manager=True, status=True, search=True, date_start=hds, date_end=hde)} AND char_length(brand) > 0 ORDER BY brand",
|
|
params,
|
|
)
|
|
dimensions = await ch_query(
|
|
f"SELECT DISTINCT dimension FROM default.pf_board WHERE {build_where(city=True, brand=True, manager=True, status=True, search=True, date_start=hds, date_end=hde)} AND char_length(dimension) > 0 ORDER BY dimension",
|
|
params,
|
|
)
|
|
# manager is an Array(String) column — ARRAY JOIN to enumerate distinct names.
|
|
managers = await ch_query(
|
|
f"SELECT DISTINCT m AS manager FROM default.pf_board ARRAY JOIN manager AS m WHERE {build_where(city=True, dimension=True, brand=True, status=True, search=True, date_start=hds, date_end=hde)} AND char_length(m) > 0 ORDER BY manager",
|
|
params,
|
|
)
|
|
statuses = await ch_query(
|
|
f"SELECT DISTINCT task_status FROM default.pf_board WHERE {build_where(city=True, dimension=True, brand=True, manager=True, search=True, date_start=hds, date_end=hde)} AND char_length(task_status) > 0 ORDER BY task_status",
|
|
params,
|
|
)
|
|
return {
|
|
"cities": [r["city"] for r in cities],
|
|
"dimensions": [r["dimension"] for r in dimensions],
|
|
"brands": [r["brand"] for r in brands],
|
|
"managers": sorted({canon_manager(r["manager"]) for r in managers}),
|
|
"statuses": [r["task_status"] for r in statuses],
|
|
}
|
|
|
|
|
|
@app.get("/api/boards")
|
|
async def boards(city: str = "", dimension: str = "", brand: str = "", manager: str = "", status: str = "", search: str = "", date_start: str = "", date_end: str = ""):
|
|
params = parse_filters(city, dimension, brand, manager, status, search, date_start, date_end)
|
|
where = build_where(city=True, dimension=True, brand=True, manager=True, status=True, search=True, date_start=bool(params["date_start"]), date_end=bool(params["date_end"]))
|
|
sql = f"""
|
|
SELECT
|
|
board_id,
|
|
argMax(address, last_activity) AS board_address,
|
|
argMax(city, last_activity) AS board_city,
|
|
argMax(dimension, last_activity) AS board_dimension
|
|
FROM default.pf_board
|
|
WHERE {where}
|
|
GROUP BY board_id
|
|
ORDER BY board_address
|
|
"""
|
|
rows = await ch_query(sql, params)
|
|
return rows
|
|
|
|
|
|
@app.get("/api/bookings")
|
|
async def bookings(city: str = "", dimension: str = "", brand: str = "", manager: str = "", status: str = "", search: str = "", date_start: str = "", date_end: str = ""):
|
|
params = parse_filters(city, dimension, brand, manager, status, search, date_start, date_end)
|
|
where = build_where(city=True, dimension=True, brand=True, manager=True, status=True, search=True, date_start=bool(params["date_start"]), date_end=bool(params["date_end"]))
|
|
sql = f"""
|
|
SELECT
|
|
board_id,
|
|
toString(task_id) AS task_id,
|
|
task_status,
|
|
toString(start_date) AS start_date,
|
|
toString(end_date) AS end_date,
|
|
brand,
|
|
company_name
|
|
FROM default.pf_board FINAL
|
|
WHERE {where}
|
|
ORDER BY board_id, start_date
|
|
"""
|
|
rows = await ch_query(sql, params)
|
|
return rows
|
|
|
|
|
|
@app.get("/api/map")
|
|
async def map_data(city: str = "", dimension: str = "", brand: str = "", manager: str = "", status: str = "", search: str = ""):
|
|
# Surfaces come from board_info (has coordinates). Occupancy is computed "as of
|
|
# today" and is intentionally NOT affected by the date-range filters — only the
|
|
# Город/Бренд/Размер/Менеджер/Поиск facets narrow which surfaces are shown.
|
|
params = parse_filters(city, dimension, brand, manager, status, search, "", "")
|
|
where = " AND ".join([
|
|
"char_length(city) > 0",
|
|
CITY_COND,
|
|
DIM_COND,
|
|
SEARCH_COND,
|
|
MAP_BRAND_COND,
|
|
MAP_MANAGER_COND,
|
|
MAP_STATUS_COND,
|
|
])
|
|
surfaces_sql = f"""
|
|
SELECT
|
|
board_key,
|
|
board_id,
|
|
city,
|
|
address,
|
|
dimension,
|
|
board_type,
|
|
side,
|
|
latitude AS lat,
|
|
longitude AS lon
|
|
FROM default.board_info FINAL
|
|
WHERE {where}
|
|
ORDER BY city, address
|
|
"""
|
|
surfaces = await ch_query(surfaces_sql, params)
|
|
|
|
# All bookings active today, keyed by surface, for marker color + balloon.
|
|
current_sql = """
|
|
SELECT
|
|
board_key,
|
|
toString(task_id) AS task_id,
|
|
task_status,
|
|
toString(start_date) AS start_date,
|
|
toString(end_date) AS end_date,
|
|
brand,
|
|
company_name
|
|
FROM default.pf_board FINAL
|
|
WHERE board_key != 0
|
|
AND pf_board.start_date <= today()
|
|
AND pf_board.end_date >= today()
|
|
"""
|
|
current = await ch_query(current_sql, params)
|
|
|
|
by_key: dict = {}
|
|
for b in current:
|
|
by_key.setdefault(b["board_key"], []).append(b)
|
|
|
|
out = []
|
|
for s in surfaces:
|
|
active = by_key.get(s["board_key"], [])
|
|
out.append({
|
|
**s,
|
|
"occupied_now": len(active) > 0,
|
|
"bookings": active,
|
|
})
|
|
return out
|
|
|
|
|
|
app.mount("/", StaticFiles(directory="static", html=True), name="static")
|