980 lines
35 KiB
Python
980 lines
35 KiB
Python
"""STEP 02 — Knowledge Plaza overview, usage, alerts."""
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import json
|
||
import math
|
||
import re
|
||
import time
|
||
from typing import Any
|
||
|
||
import h3
|
||
from falkordb import FalkorDB
|
||
from fastapi import APIRouter, Depends, HTTPException
|
||
|
||
from app.auth import CurrentUser
|
||
from app.config import settings
|
||
from app.db import get_agent_settings, get_plaza_overview, get_plaza_alerts, get_conn
|
||
from app.graph_qa_engine import answer_graph_question
|
||
from app.llm_client import LlmClient
|
||
from app.project_context import ProjectContext, get_project_context
|
||
|
||
router = APIRouter()
|
||
|
||
SPATIAL_GRAPH_NAME = "guiyang_spatial_v1"
|
||
LIBO_BUS_DATASET = "libo_bus_xlsx_v1"
|
||
|
||
FOOD_ENRICHMENT_KEYS = (
|
||
"food_fusion_status",
|
||
"food_match_confidence",
|
||
"food_match_rule",
|
||
"food_match_score",
|
||
"food_name_similarity",
|
||
"food_address_similarity",
|
||
"food_coordinate_distance_m",
|
||
"dianping_shop_id",
|
||
"dianping_name",
|
||
"dianping_address",
|
||
"dianping_url",
|
||
"dianping_rating",
|
||
"dianping_review_count",
|
||
"dianping_avg_price",
|
||
"dianping_area",
|
||
"dianping_category",
|
||
"dianping_score_details",
|
||
"dianping_ranking",
|
||
"dianping_business_status",
|
||
"dianping_business_hours",
|
||
"dianping_tags",
|
||
"dianping_transportation",
|
||
"dianping_shop_image",
|
||
"dianping_review_tags",
|
||
"recommended_dish_count",
|
||
"recommended_dish_names",
|
||
"group_buy_count",
|
||
"group_buy_min_price",
|
||
"group_buy_max_price",
|
||
"group_buy_titles",
|
||
"group_buy_items_json",
|
||
"review_sample_count",
|
||
"review_sample_avg_stars",
|
||
"review_sample_summary",
|
||
"food_review_items_json",
|
||
"food_review_tags_json",
|
||
"menu_image_count",
|
||
"candidate_dianping_name",
|
||
"food_candidate_name_similarity",
|
||
"food_candidate_address_similarity",
|
||
"enrichment_updated_at",
|
||
)
|
||
|
||
HOTEL_ENRICHMENT_KEYS = (
|
||
"hotel_fusion_status",
|
||
"hotel_match_confidence",
|
||
"hotel_match_rule",
|
||
"hotel_name_similarity",
|
||
"hotel_address_similarity",
|
||
"hotel_phone_exact_match",
|
||
"ctrip_hotel_id",
|
||
"ctrip_name_cn",
|
||
"ctrip_name_en",
|
||
"ctrip_address",
|
||
"ctrip_phone",
|
||
"ctrip_star_level",
|
||
"ctrip_diamond_level",
|
||
"ctrip_opened_year",
|
||
"ctrip_room_count",
|
||
"ctrip_hotel_type",
|
||
"ctrip_ranking",
|
||
"ctrip_rating",
|
||
"ctrip_rating_description",
|
||
"ctrip_review_count",
|
||
"ctrip_transportation",
|
||
"ctrip_tags",
|
||
"ctrip_popular_facilities",
|
||
"ctrip_score_details",
|
||
"ctrip_cleanliness_score",
|
||
"ctrip_facilities_score",
|
||
"ctrip_environment_score",
|
||
"ctrip_service_score",
|
||
"ctrip_description",
|
||
"ctrip_url",
|
||
"room_type_count",
|
||
"room_type_names",
|
||
"room_min_price",
|
||
"room_max_price",
|
||
"offer_count",
|
||
"offer_min_price",
|
||
"offer_max_price",
|
||
"room_items_json",
|
||
"facility_service_count",
|
||
"facility_summary",
|
||
"facility_category_items_json",
|
||
"policy_count",
|
||
"policy_summary",
|
||
"guest_review_sample_count",
|
||
"guest_review_sample_avg_score",
|
||
"latest_review_at",
|
||
"review_travel_types",
|
||
"review_items_json",
|
||
"hotel_image_samples",
|
||
"nearby_place_count",
|
||
"nearby_place_summary",
|
||
"nearby_category_items_json",
|
||
"candidate_ctrip_name",
|
||
"hotel_candidate_name_similarity",
|
||
"hotel_candidate_address_similarity",
|
||
"enrichment_updated_at",
|
||
)
|
||
|
||
CATEGORY_ALIASES: dict[str, list[str]] = {
|
||
"美食": ["美食", "吃", "餐厅", "饭店", "火锅", "小吃", "烧烤", "咖啡", "奶茶", "酸汤鱼"],
|
||
"景点": ["景点", "景区", "公园", "博物馆", "古镇", "夜游", "历史", "文化", "好玩"],
|
||
"酒店": ["酒店", "住宿", "住", "宾馆", "民宿"],
|
||
"商场": ["商场", "购物", "商圈", "超市", "商城"],
|
||
"医疗保健": ["医院", "诊所", "药店", "医疗", "看病", "急诊"],
|
||
"交通设施": ["地铁", "公交", "车站", "交通", "停车", "机场", "高铁"],
|
||
"生活服务": ["生活服务", "维修", "营业厅", "服务"],
|
||
"科教文化": ["学校", "大学", "图书馆", "教育", "培训"],
|
||
}
|
||
|
||
PLACE_TYPE_ALIASES = {
|
||
"美食": "eat",
|
||
"景点": "sight",
|
||
"酒店": "hotel",
|
||
"商场": "mall",
|
||
"医疗保健": "medical",
|
||
"交通设施": "transit",
|
||
"生活服务": "life",
|
||
"科教文化": "education",
|
||
}
|
||
|
||
|
||
def _haversine_m(lng1: float, lat1: float, lng2: float, lat2: float) -> float:
|
||
radius = 6_371_008.8
|
||
d_lng = math.radians(lng2 - lng1)
|
||
d_lat = math.radians(lat2 - lat1)
|
||
part = (
|
||
math.sin(d_lat / 2) ** 2
|
||
+ math.cos(math.radians(lat1))
|
||
* math.cos(math.radians(lat2))
|
||
* math.sin(d_lng / 2) ** 2
|
||
)
|
||
return 2 * radius * math.asin(math.sqrt(part))
|
||
|
||
|
||
def _h3_plan(radius_m: int) -> tuple[int, str, int]:
|
||
if radius_m <= 500:
|
||
return 9, "h3_r9", 2
|
||
if radius_m <= 1000:
|
||
return 9, "h3_r9", 4
|
||
if radius_m <= 3000:
|
||
return 8, "h3_r8", 4
|
||
res = 7
|
||
edge_m = h3.average_hexagon_edge_length(res, unit="m")
|
||
return res, "h3_r7", max(2, math.ceil(radius_m / (math.sqrt(3) * edge_m)) + 1)
|
||
|
||
|
||
def _rule_intent(question: str, radius_m: int | None) -> dict[str, Any]:
|
||
q = question.strip()
|
||
radius = radius_m or 1000
|
||
km = re.search(r"(\d+(?:\.\d+)?)\s*(?:公里|千米|km)", q, flags=re.I)
|
||
meter = re.search(r"(\d+(?:\.\d+)?)\s*(?:米|m)", q, flags=re.I)
|
||
minutes = re.search(r"(\d+(?:\.\d+)?)\s*分钟", q)
|
||
if km:
|
||
radius = int(float(km.group(1)) * 1000)
|
||
elif meter:
|
||
radius = int(float(meter.group(1)))
|
||
elif minutes:
|
||
# 步行 15 分钟约 1.1~1.3km，先用保守半径召回，后续可接路线时长。
|
||
radius = min(3000, max(500, int(float(minutes.group(1)) * 80)))
|
||
category = ""
|
||
for cat, aliases in CATEGORY_ALIASES.items():
|
||
if any(a in q for a in aliases):
|
||
category = cat
|
||
break
|
||
keywords = [
|
||
w for w in re.split(r"[，,。？?！!\s]+", q)
|
||
if w and not any(w in aliases for aliases in CATEGORY_ALIASES.values())
|
||
][:6]
|
||
return {
|
||
"radius_m": max(100, min(radius, 10000)),
|
||
"category": category,
|
||
"keywords": keywords,
|
||
"sort_preference": "综合距离、评分和语义匹配",
|
||
"user_need": q,
|
||
}
|
||
|
||
|
||
async def _deepseek_client(max_tokens: int = 900) -> LlmClient | None:
|
||
cfg = await get_agent_settings()
|
||
extract = cfg.get("extract") or {}
|
||
models = extract.get("models") or {}
|
||
deepseek_cfg = models.get("deepseek") or {}
|
||
if deepseek_cfg.get("base_url") and deepseek_cfg.get("api_key"):
|
||
return LlmClient(
|
||
deepseek_cfg["base_url"],
|
||
deepseek_cfg["api_key"],
|
||
deepseek_cfg.get("model") or "deepseek-chat",
|
||
timeout=int(extract.get("timeout") or 60),
|
||
max_tokens=max_tokens,
|
||
)
|
||
|
||
global_cfg = cfg.get("global") or {}
|
||
if global_cfg.get("base_url") and global_cfg.get("api_key"):
|
||
return LlmClient(
|
||
global_cfg["base_url"],
|
||
global_cfg["api_key"],
|
||
global_cfg.get("model") or "deepseek-chat",
|
||
timeout=int(global_cfg.get("timeout") or 45),
|
||
max_tokens=max_tokens,
|
||
)
|
||
return None
|
||
|
||
|
||
async def _llm_intent(question: str, fallback: dict[str, Any]) -> tuple[dict[str, Any], str]:
|
||
client = await _deepseek_client(max_tokens=700)
|
||
if not client:
|
||
return fallback, "DeepSeek 未配置，使用规则解析"
|
||
system = (
|
||
"你是城市知识图谱的游客问答意图解析器。只输出 JSON。"
|
||
"把用户问题解析为 nearby POI 查询意图，类别只能从："
|
||
"美食、景点、酒店、商场、医疗保健、交通设施、生活服务、科教文化、空字符串 中选择。"
|
||
"radius_m 为整数米，没说半径默认 1000。keywords 提取用户真正关心的语义词。"
|
||
)
|
||
user = json.dumps({"question": question, "rule_fallback": fallback}, ensure_ascii=False)
|
||
try:
|
||
data = await asyncio.to_thread(client.chat_json, system, user)
|
||
merged = {**fallback, **{k: v for k, v in data.items() if v not in (None, "")}}
|
||
merged["radius_m"] = max(100, min(int(merged.get("radius_m") or fallback["radius_m"]), 10000))
|
||
if merged.get("category") not in CATEGORY_ALIASES:
|
||
merged["category"] = fallback.get("category", "")
|
||
if not isinstance(merged.get("keywords"), list):
|
||
merged["keywords"] = fallback.get("keywords", [])
|
||
return merged, "DeepSeek 意图解析"
|
||
except Exception as exc: # noqa: BLE001
|
||
return fallback, f"DeepSeek 意图解析失败，使用规则解析：{str(exc)[:120]}"
|
||
|
||
|
||
def _photo_urls(value: Any) -> list[str]:
|
||
if isinstance(value, list):
|
||
return [str(v) for v in value if v]
|
||
if isinstance(value, str):
|
||
try:
|
||
parsed = json.loads(value)
|
||
if isinstance(parsed, list):
|
||
return [str(v) for v in parsed if v]
|
||
except Exception:
|
||
pass
|
||
return [v.strip() for v in re.split(r"[|,]", value) if v.strip().startswith("http")]
|
||
return []
|
||
|
||
|
||
def _rating_num(value: Any) -> float:
|
||
try:
|
||
return float(value or 0)
|
||
except Exception:
|
||
return 0.0
|
||
|
||
|
||
def _score_place(row: dict[str, Any], distance_m: float, radius_m: int, intent: dict[str, Any]) -> tuple[float, list[str]]:
|
||
score = max(0.0, 55.0 * (1 - distance_m / max(radius_m, 1)))
|
||
reasons = [f"距离约 {round(distance_m)} 米"]
|
||
rating = _rating_num(row.get("rating"))
|
||
if rating:
|
||
score += min(rating, 5) * 7
|
||
reasons.append(f"评分 {rating:g}")
|
||
category = intent.get("category") or ""
|
||
if category and row.get("type_label") == category:
|
||
score += 20
|
||
reasons.append(f"匹配类别「{category}」")
|
||
haystack = " ".join(
|
||
str(row.get(k) or "") for k in ("name", "address", "tags", "amap_type", "type_label")
|
||
)
|
||
for kw in intent.get("keywords") or []:
|
||
if kw and kw in haystack:
|
||
score += 12
|
||
reasons.append(f"命中关键词「{kw}」")
|
||
return round(score, 3), reasons[:4]
|
||
|
||
|
||
async def _llm_answer(question: str, intent: dict[str, Any], results: list[dict[str, Any]]) -> tuple[str, dict[str, str], str]:
|
||
if not results:
|
||
return "当前已采集的知识图谱中，没有在这个半径内找到匹配结果。可以扩大半径，或等高德网格续采完成后再试。", {}, "no_candidates"
|
||
client = await _deepseek_client(max_tokens=1200)
|
||
if not client:
|
||
first = results[0]
|
||
return (
|
||
f"根据当前知识图谱，优先推荐 {first['name']}，距离约 {round(first['distance_m'])} 米。"
|
||
f"下面结果已按距离、类别匹配和评分综合排序。",
|
||
{r["place_id"]: "距离近、类别匹配、来自当前空间知识图谱" for r in results[:8]},
|
||
"fallback_answer",
|
||
)
|
||
compact = [
|
||
{
|
||
"id": r["place_id"],
|
||
"name": r["name"],
|
||
"type": r["type_label"],
|
||
"distance_m": r["distance_m"],
|
||
"rating": r.get("rating"),
|
||
"address": r.get("address"),
|
||
"tags": r.get("tags"),
|
||
"score": r.get("score"),
|
||
}
|
||
for r in results[:20]
|
||
]
|
||
system = (
|
||
"你是面向游客的城市知识图谱问答助手。只输出 JSON。"
|
||
"基于候选 POI 回答用户问题，不能编造候选中没有的地点。"
|
||
"输出 answer 和 reasons，reasons 是 {候选id: 推荐理由}。"
|
||
"回答要像真实产品结果页，简洁、可解释。"
|
||
)
|
||
user = json.dumps({"question": question, "intent": intent, "candidates": compact}, ensure_ascii=False)
|
||
try:
|
||
data = await asyncio.to_thread(client.chat_json, system, user)
|
||
answer = str(data.get("answer") or "").strip()
|
||
reasons = data.get("reasons") if isinstance(data.get("reasons"), dict) else {}
|
||
return answer or "已根据当前知识图谱完成附近结果排序。", {str(k): str(v) for k, v in reasons.items()}, "DeepSeek 回答排序"
|
||
except Exception as exc: # noqa: BLE001
|
||
first = results[0]
|
||
return (
|
||
f"根据当前知识图谱，优先推荐 {first['name']}，距离约 {round(first['distance_m'])} 米。"
|
||
f"DeepSeek 回答组织暂时失败，页面仍展示规则排序结果。",
|
||
{},
|
||
f"DeepSeek 回答失败：{str(exc)[:120]}",
|
||
)
|
||
|
||
|
||
def _rule_answer(question: str, intent: dict[str, Any], results: list[dict[str, Any]]) -> tuple[str, dict[str, str], str]:
|
||
if not results:
|
||
return "当前已采集的知识图谱中，没有在这个半径内找到匹配结果。可以扩大半径，或等采集完成后再试。", {}, "rule_fast_answer"
|
||
category = intent.get("category") or "相关地点"
|
||
first = results[0]
|
||
answer = (
|
||
f"根据当前知识图谱，{category}共召回 {len(results)} 个候选；"
|
||
f"优先推荐 {first['name']}，距离约 {round(first['distance_m'])} 米。"
|
||
"下方已按距离、类别匹配和评分综合排序。"
|
||
)
|
||
reasons = {
|
||
r["place_id"]: "；".join(r.get("rank_reasons") or ["距离、类别和评分综合靠前"])
|
||
for r in results[:12]
|
||
}
|
||
return answer, reasons, "rule_fast_answer"
|
||
|
||
|
||
@router.get("/plaza/overview")
|
||
async def overview(
|
||
context: ProjectContext = Depends(get_project_context),
|
||
_user: CurrentUser = None,
|
||
):
|
||
return await get_plaza_overview(context.tenant_id, context.project_id)
|
||
|
||
|
||
@router.get("/plaza/usage")
|
||
async def usage(
|
||
context: ProjectContext = Depends(get_project_context),
|
||
_user: CurrentUser = None,
|
||
):
|
||
"""Return usage statistics — top hot and cold entities."""
|
||
s = settings.db_schema
|
||
async with get_conn() as conn:
|
||
async with conn.cursor() as cur:
|
||
await cur.execute(
|
||
f"SELECT entity_type, COUNT(*) AS cnt FROM {s}.candidate_entities "
|
||
"WHERE tenant_id=%s AND project_id=%s AND status='published' "
|
||
"GROUP BY entity_type ORDER BY cnt DESC",
|
||
(context.tenant_id, context.project_id),
|
||
)
|
||
by_type = await cur.fetchall()
|
||
|
||
await cur.execute(
|
||
f"SELECT COUNT(*) AS cnt FROM {s}.candidate_entities "
|
||
"WHERE tenant_id=%s AND project_id=%s AND status='pending_review'",
|
||
(context.tenant_id, context.project_id),
|
||
)
|
||
pending = (await cur.fetchone())["cnt"]
|
||
|
||
return {
|
||
"entities_by_type": by_type,
|
||
"pending_review": pending,
|
||
}
|
||
|
||
|
||
@router.get("/plaza/alerts")
|
||
async def alerts(
|
||
context: ProjectContext = Depends(get_project_context),
|
||
_user: CurrentUser = None,
|
||
):
|
||
return await get_plaza_alerts(context.tenant_id, context.project_id)
|
||
|
||
|
||
@router.get("/plaza/amap-config")
|
||
async def amap_config(_user: CurrentUser = None):
|
||
"""Return browser-side AMap JS API configuration for admin map canvases."""
|
||
return {
|
||
"configured": bool(settings.amap_js_key),
|
||
"js_key": settings.amap_js_key,
|
||
"security_jscode": settings.amap_security_jscode,
|
||
"security_configured": bool(settings.amap_security_jscode),
|
||
}
|
||
|
||
|
||
def _is_enterprise_travel_graph(graph_name: str) -> bool:
|
||
lower = graph_name.lower()
|
||
return "baixinghui" in lower or ("travel" in lower and graph_name != SPATIAL_GRAPH_NAME)
|
||
|
||
|
||
def _resolve_spatial_graph_name(graph_name: str) -> str:
|
||
"""Map the city project graph name to the materialized spatial POI graph."""
|
||
lower = graph_name.lower()
|
||
if graph_name == SPATIAL_GRAPH_NAME or lower.endswith("_spatial_v1"):
|
||
return graph_name
|
||
if graph_name == "guiyang_new2":
|
||
return SPATIAL_GRAPH_NAME
|
||
return graph_name
|
||
|
||
|
||
def _iso_datetime(value: Any) -> str:
|
||
return value.isoformat() if value else ""
|
||
|
||
|
||
def _build_libo_bus_route_payload(
|
||
rows: list[list[Any]],
|
||
graph_name: str,
|
||
) -> dict[str, Any]:
|
||
routes: dict[str, dict[str, Any]] = {}
|
||
stop_ids: set[str] = set()
|
||
for row in rows:
|
||
(
|
||
route_id,
|
||
line_name,
|
||
direction,
|
||
start_stop,
|
||
end_stop,
|
||
stop_count,
|
||
fare_yuan,
|
||
first_bus,
|
||
last_bus,
|
||
distance_km,
|
||
path_source,
|
||
stop_id,
|
||
stop_name,
|
||
lng,
|
||
lat,
|
||
sequence,
|
||
) = row
|
||
route_key = str(route_id or "")
|
||
if not route_key:
|
||
continue
|
||
route = routes.setdefault(
|
||
route_key,
|
||
{
|
||
"route_id": route_key,
|
||
"line_name": str(line_name or ""),
|
||
"direction": str(direction or ""),
|
||
"start_stop": str(start_stop or ""),
|
||
"end_stop": str(end_stop or ""),
|
||
"stop_count": int(stop_count or 0),
|
||
"fare_yuan": float(fare_yuan or 0),
|
||
"first_bus": str(first_bus or ""),
|
||
"last_bus": str(last_bus or ""),
|
||
"distance_km": float(distance_km or 0),
|
||
"path_source": str(path_source or "station_sequence"),
|
||
"stops": [],
|
||
},
|
||
)
|
||
if stop_id and lng is not None and lat is not None:
|
||
stop_key = str(stop_id)
|
||
stop_ids.add(stop_key)
|
||
route["stops"].append(
|
||
{
|
||
"stop_id": stop_key,
|
||
"name": str(stop_name or "未命名公交站"),
|
||
"lng": float(lng),
|
||
"lat": float(lat),
|
||
"sequence": int(sequence or 0),
|
||
}
|
||
)
|
||
|
||
items = sorted(
|
||
routes.values(),
|
||
key=lambda route: (
|
||
route["line_name"],
|
||
route["route_id"],
|
||
),
|
||
)
|
||
return {
|
||
"graph_name": graph_name,
|
||
"dataset": LIBO_BUS_DATASET,
|
||
"line_count": len({route["line_name"] for route in items}),
|
||
"direction_count": len(items),
|
||
"stop_count": len(stop_ids),
|
||
"route_stop_count": sum(len(route["stops"]) for route in items),
|
||
"items": items,
|
||
}
|
||
|
||
|
||
def _read_libo_bus_routes(graph_name: str) -> dict[str, Any]:
|
||
graph = FalkorDB(
|
||
host=settings.falkordb_host,
|
||
port=settings.falkordb_port,
|
||
).select_graph(graph_name)
|
||
rows = graph.query(
|
||
"MATCH (b:BusLine) WHERE b.dataset=$dataset "
|
||
"OPTIONAL MATCH (b)-[r:STOPS_AT]->(s:Place) "
|
||
"RETURN b.route_id,b.line_name,b.direction,b.start_stop,b.end_stop,"
|
||
"b.stop_count,b.fare_yuan,b.first_bus,b.last_bus,b.distance_km,"
|
||
"b.path_source,s.element_id,s.name,s.lng,s.lat,r.sequence "
|
||
"ORDER BY b.line_name,b.route_id,r.sequence",
|
||
{"dataset": LIBO_BUS_DATASET},
|
||
).result_set
|
||
return _build_libo_bus_route_payload(rows, graph_name)
|
||
|
||
|
||
_INTERNAL_ENRICHMENT_KEYS = {
|
||
"food_fusion_status",
|
||
"food_match_confidence",
|
||
"food_match_rule",
|
||
"food_match_score",
|
||
"food_name_similarity",
|
||
"food_address_similarity",
|
||
"food_coordinate_distance_m",
|
||
"candidate_dianping_name",
|
||
"food_candidate_name_similarity",
|
||
"food_candidate_address_similarity",
|
||
"hotel_fusion_status",
|
||
"hotel_match_confidence",
|
||
"hotel_match_rule",
|
||
"hotel_name_similarity",
|
||
"hotel_address_similarity",
|
||
"hotel_phone_exact_match",
|
||
"candidate_ctrip_name",
|
||
"hotel_candidate_name_similarity",
|
||
"hotel_candidate_address_similarity",
|
||
"ctrip_description",
|
||
}
|
||
|
||
_ZERO_MEANS_MISSING_ENRICHMENT_KEYS = {
|
||
"dianping_rating",
|
||
"dianping_review_count",
|
||
"dianping_avg_price",
|
||
"group_buy_count",
|
||
"review_sample_count",
|
||
"review_sample_avg_stars",
|
||
"ctrip_rating",
|
||
"ctrip_review_count",
|
||
"ctrip_room_count",
|
||
"room_min_price",
|
||
"room_max_price",
|
||
"offer_count",
|
||
"offer_min_price",
|
||
"offer_max_price",
|
||
"guest_review_sample_count",
|
||
"guest_review_sample_avg_score",
|
||
}
|
||
|
||
|
||
def _clean_enrichment_display_value(key: str, value: Any) -> Any:
|
||
"""Convert scraped platform values into stable, user-facing display values."""
|
||
if key in _INTERNAL_ENRICHMENT_KEYS:
|
||
return None
|
||
if key in _ZERO_MEANS_MISSING_ENRICHMENT_KEYS:
|
||
try:
|
||
if float(value) == 0:
|
||
return None
|
||
except (TypeError, ValueError):
|
||
pass
|
||
if key == "ctrip_diamond_level":
|
||
text = str(value or "").strip()
|
||
if re.fullmatch(r"\d+(?:\.\d+)?钻", text):
|
||
return text
|
||
matched = re.fullmatch(
|
||
r"(\d+(?:\.\d+)?)\s+out\s+of\s+5\s+(?:rating|diamonds?)",
|
||
text,
|
||
flags=re.IGNORECASE,
|
||
)
|
||
return f"{matched.group(1)}钻" if matched else None
|
||
return value
|
||
|
||
|
||
def _graph_poi_enrichment(
|
||
graph_name: str,
|
||
element_id: str,
|
||
) -> tuple[list[str], dict[str, Any], dict[str, Any]]:
|
||
graph = FalkorDB(
|
||
host=settings.falkordb_host,
|
||
port=settings.falkordb_port,
|
||
).select_graph(graph_name)
|
||
result = graph.query(
|
||
"MATCH (n {element_id:$element_id}) "
|
||
"RETURN labels(n), properties(n) LIMIT 1",
|
||
{"element_id": element_id},
|
||
).result_set
|
||
if not result:
|
||
return [], {}, {}
|
||
|
||
labels = [str(label) for label in (result[0][0] or [])]
|
||
properties = result[0][1] if isinstance(result[0][1], dict) else {}
|
||
|
||
def pick(keys: tuple[str, ...]) -> dict[str, Any]:
|
||
result: dict[str, Any] = {}
|
||
for key in keys:
|
||
if key not in properties:
|
||
continue
|
||
value = _clean_enrichment_display_value(key, properties[key])
|
||
if value in (None, "", "[]", "{}"):
|
||
continue
|
||
result[key] = value
|
||
return result
|
||
|
||
return labels, pick(FOOD_ENRICHMENT_KEYS), pick(HOTEL_ENRICHMENT_KEYS)
|
||
|
||
|
||
def _graph_poi_categories(graph_name: str) -> dict[str, list[str]]:
|
||
graph = FalkorDB(
|
||
host=settings.falkordb_host,
|
||
port=settings.falkordb_port,
|
||
).select_graph(graph_name)
|
||
rows = graph.query(
|
||
"MATCH (n) "
|
||
"WHERE n.element_id IS NOT NULL "
|
||
"AND any(label IN labels(n) WHERE label IN "
|
||
"['FoodPlace','Hotel','ScenicSpot','TransitFacility']) "
|
||
"RETURN n.element_id, labels(n)"
|
||
).result_set
|
||
label_names = {
|
||
"FoodPlace": "美食",
|
||
"Hotel": "酒店",
|
||
"ScenicSpot": "景点",
|
||
"TransitFacility": "交通设施",
|
||
}
|
||
result: dict[str, list[str]] = {}
|
||
for element_id, labels in rows:
|
||
categories = [
|
||
label_names[label]
|
||
for label in (labels or [])
|
||
if label in label_names
|
||
]
|
||
if element_id and categories:
|
||
result[str(element_id)] = categories
|
||
return result
|
||
|
||
|
||
@router.get("/plaza/map-pois")
|
||
async def map_pois(
|
||
context: ProjectContext = Depends(get_project_context),
|
||
_user: CurrentUser = None,
|
||
):
|
||
"""Return lightweight, project-scoped POI points for the knowledge map."""
|
||
graph_name = _resolve_spatial_graph_name(context.graph_name)
|
||
s = settings.db_schema
|
||
async with get_conn() as conn:
|
||
async with conn.cursor() as cur:
|
||
await cur.execute(
|
||
f"""SELECT element_id, gaode_poi_id, name, type_label, place_type,
|
||
lng, lat, address, district, city
|
||
FROM {s}.amap_spatial_pois
|
||
WHERE graph_name=%s
|
||
ORDER BY type_label, name
|
||
LIMIT 15000""",
|
||
(graph_name,),
|
||
)
|
||
rows = await cur.fetchall()
|
||
graph_categories = await asyncio.to_thread(
|
||
_graph_poi_categories,
|
||
graph_name,
|
||
)
|
||
|
||
items = [
|
||
{
|
||
"id": row["element_id"],
|
||
"gaode_poi_id": row["gaode_poi_id"],
|
||
"name": row["name"] or "未命名POI",
|
||
"category": row["type_label"] or "其他地点",
|
||
"categories": graph_categories.get(
|
||
str(row["element_id"]),
|
||
[row["type_label"] or "其他地点"],
|
||
),
|
||
"place_type": row["place_type"] or "poi",
|
||
"lng": float(row["lng"]),
|
||
"lat": float(row["lat"]),
|
||
"address": row["address"] or "",
|
||
"district": row["district"] or "",
|
||
"city": row["city"] or "",
|
||
}
|
||
for row in rows
|
||
if row.get("lng") is not None and row.get("lat") is not None
|
||
]
|
||
category_counts: dict[str, int] = {}
|
||
for item in items:
|
||
for category in item["categories"]:
|
||
category_counts[category] = category_counts.get(category, 0) + 1
|
||
return {
|
||
"graph_name": graph_name,
|
||
"total": len(items),
|
||
"categories": [
|
||
{"category": category, "count": count}
|
||
for category, count in sorted(
|
||
category_counts.items(),
|
||
key=lambda pair: (-pair[1], pair[0]),
|
||
)
|
||
],
|
||
"items": items,
|
||
}
|
||
|
||
|
||
@router.get("/plaza/bus-routes")
|
||
async def bus_routes(
|
||
context: ProjectContext = Depends(get_project_context),
|
||
_user: CurrentUser = None,
|
||
):
|
||
"""Return the active project's graph-backed bus routes and ordered stops."""
|
||
graph_name = _resolve_spatial_graph_name(context.graph_name)
|
||
return await asyncio.to_thread(_read_libo_bus_routes, graph_name)
|
||
|
||
|
||
@router.get("/plaza/map-pois/{place_id}")
|
||
async def map_poi_detail(
|
||
place_id: str,
|
||
context: ProjectContext = Depends(get_project_context),
|
||
_user: CurrentUser = None,
|
||
):
|
||
"""Return useful detail fields for one POI in the active project."""
|
||
graph_name = _resolve_spatial_graph_name(context.graph_name)
|
||
s = settings.db_schema
|
||
async with get_conn() as conn:
|
||
async with conn.cursor() as cur:
|
||
await cur.execute(
|
||
f"""SELECT element_id, gaode_poi_id, name, type_label, place_type,
|
||
amap_type, typecode, lng, lat, province, city, district,
|
||
adcode, business_area, address, tel, open_time, rating,
|
||
cost, level, tags, photo_urls, source, source_cell_id,
|
||
source_resolution, source_scope_adcode,
|
||
raw_jsonb,
|
||
first_fetched_at, last_fetched_at
|
||
FROM {s}.amap_spatial_pois
|
||
WHERE graph_name=%s
|
||
AND (element_id=%s OR gaode_poi_id=%s)
|
||
LIMIT 1""",
|
||
(graph_name, place_id, place_id.removeprefix("amap:")),
|
||
)
|
||
row = await cur.fetchone()
|
||
if not row:
|
||
raise HTTPException(status_code=404, detail="当前项目中未找到该POI")
|
||
|
||
raw = row.get("raw_jsonb") or {}
|
||
if not isinstance(raw, dict):
|
||
raw = {}
|
||
graph_labels, food_enrichment, hotel_enrichment = await asyncio.to_thread(
|
||
_graph_poi_enrichment,
|
||
graph_name,
|
||
row["element_id"],
|
||
)
|
||
external_photos = _photo_urls(food_enrichment.get("dianping_shop_image"))
|
||
external_photos.extend(_photo_urls(hotel_enrichment.get("hotel_image_samples")))
|
||
photo_urls = list(dict.fromkeys([
|
||
*_photo_urls(row["photo_urls"]),
|
||
*external_photos,
|
||
]))
|
||
|
||
return {
|
||
"graph_name": graph_name,
|
||
"id": row["element_id"],
|
||
"gaode_poi_id": row["gaode_poi_id"],
|
||
"name": row["name"] or "未命名POI",
|
||
"category": row["type_label"] or "其他地点",
|
||
"place_type": row["place_type"] or "poi",
|
||
"business_subcategory": raw.get("business_subcategory") or "",
|
||
"scenic_type": raw.get("scenic_type") or "",
|
||
"scenic_level": raw.get("scenic_level") or "",
|
||
"parent_scenic": raw.get("parent_scenic") or "",
|
||
"scenic_grade": raw.get("scenic_grade") or "",
|
||
"visitor_value": raw.get("visitor_value") or "",
|
||
"audit_result": raw.get("audit_result") or "",
|
||
"audit_confidence": raw.get("audit_confidence") or "",
|
||
"audit_basis": raw.get("audit_basis") or "",
|
||
"audit_date": raw.get("audit_date") or "",
|
||
"amap_type": row["amap_type"] or "",
|
||
"typecode": row["typecode"] or "",
|
||
"scan_hit_count": raw.get("scan_hit_count") or 0,
|
||
"matched_scan_types": raw.get("matched_scan_types") or [],
|
||
"lng": float(row["lng"]),
|
||
"lat": float(row["lat"]),
|
||
"province": row["province"] or "",
|
||
"city": row["city"] or "",
|
||
"district": row["district"] or "",
|
||
"adcode": row["adcode"] or "",
|
||
"business_area": row["business_area"] or "",
|
||
"address": row["address"] or "",
|
||
"tel": row["tel"] or "",
|
||
"open_time": row["open_time"] or "",
|
||
"rating": row["rating"] or "",
|
||
"cost": row["cost"] or "",
|
||
"level": row["level"] or "",
|
||
"tags": row["tags"] or "",
|
||
"photo_urls": photo_urls,
|
||
"source": row["source"] or "",
|
||
"source_cell_id": row["source_cell_id"] or "",
|
||
"source_resolution": row["source_resolution"],
|
||
"source_scope_adcode": row["source_scope_adcode"] or "",
|
||
"first_fetched_at": _iso_datetime(row["first_fetched_at"]),
|
||
"last_fetched_at": _iso_datetime(row["last_fetched_at"]),
|
||
"graph_labels": graph_labels,
|
||
"food_enrichment": food_enrichment,
|
||
"hotel_enrichment": hotel_enrichment,
|
||
}
|
||
|
||
|
||
@router.post("/plaza/user-query")
|
||
async def user_query(
|
||
body: dict,
|
||
context: ProjectContext = Depends(get_project_context),
|
||
_user: CurrentUser = None,
|
||
):
|
||
"""User-facing KG nearby query: NL intent -> H3 recall -> distance/rank -> LLM answer."""
|
||
started_at = time.perf_counter()
|
||
question = str(body.get("question") or "").strip()
|
||
if not question:
|
||
raise HTTPException(400, "question required")
|
||
graph_name = str(body.get("graph_name") or context.graph_name or SPATIAL_GRAPH_NAME)
|
||
if _is_enterprise_travel_graph(graph_name):
|
||
try:
|
||
graph_response = await answer_graph_question(
|
||
question,
|
||
graph_name,
|
||
customer_context=body.get("customer_context") if isinstance(body.get("customer_context"), dict) else {},
|
||
limit=int(body.get("limit") or 80),
|
||
)
|
||
except Exception as exc: # noqa: BLE001
|
||
raise HTTPException(status_code=502, detail=f"LLM 图查询问答失败: {str(exc)[:220]}") from exc
|
||
trace = graph_response.get("trace") or {}
|
||
trace.update({
|
||
"latency_ms": max(1, round((time.perf_counter() - started_at) * 1000)),
|
||
"rule_query_used": False,
|
||
"plaza_user_mode": "enterprise_graph_qa",
|
||
})
|
||
return {
|
||
"mode": "travel_customer_service",
|
||
"question": question,
|
||
"graph_name": graph_name,
|
||
"user_location": {
|
||
"lng": float(body.get("lng", 106.7135) or 106.7135),
|
||
"lat": float(body.get("lat", 26.5744) or 26.5744),
|
||
},
|
||
"intent": {
|
||
"radius_m": int(body.get("radius_m") or 1000),
|
||
"category": "百姓惠客服",
|
||
"keywords": [],
|
||
"user_need": question,
|
||
},
|
||
"answer": graph_response.get("copy_text") or graph_response.get("answer") or "",
|
||
"results": [],
|
||
"plans": graph_response.get("plans") or [],
|
||
"evidence": graph_response.get("evidence") or [],
|
||
"follow_up_questions": graph_response.get("follow_up_questions") or [],
|
||
"risk_notes": graph_response.get("risk_notes") or [],
|
||
"trace": trace,
|
||
}
|
||
graph_name = _resolve_spatial_graph_name(graph_name)
|
||
try:
|
||
lng = float(body.get("lng", 106.7135))
|
||
lat = float(body.get("lat", 26.5744))
|
||
except Exception as exc:
|
||
raise HTTPException(400, "lng/lat required") from exc
|
||
radius_input = body.get("radius_m")
|
||
radius_m = int(radius_input) if radius_input not in (None, "") else None
|
||
use_llm = body.get("use_llm") is True
|
||
fallback = _rule_intent(question, radius_m)
|
||
if use_llm:
|
||
intent, intent_source = await _llm_intent(question, fallback)
|
||
else:
|
||
intent, intent_source = fallback, "规则意图快路径"
|
||
radius = int(intent["radius_m"])
|
||
res, h3_col, k = _h3_plan(radius)
|
||
cells = list(h3.grid_disk(h3.latlng_to_cell(lat, lng, res), k))
|
||
category = intent.get("category") or ""
|
||
place_type = PLACE_TYPE_ALIASES.get(category, "")
|
||
|
||
s = settings.db_schema
|
||
async with get_conn() as conn:
|
||
async with conn.cursor() as cur:
|
||
await cur.execute(
|
||
f"""SELECT gaode_poi_id, element_id, name, type_label, place_type, amap_type,
|
||
typecode, lng, lat, address, district, city, adcode, business_area,
|
||
tel, rating, cost, open_time, tags, photo_urls, {h3_col} AS h3_cell,
|
||
source, first_fetched_at, last_fetched_at
|
||
FROM {s}.amap_spatial_pois
|
||
WHERE graph_name=%s AND {h3_col}=ANY(%s)
|
||
AND (%s='' OR type_label=%s OR place_type=%s)
|
||
LIMIT 8000""",
|
||
(graph_name, cells, category, category, place_type),
|
||
)
|
||
rows = await cur.fetchall()
|
||
|
||
results: list[dict[str, Any]] = []
|
||
for row in rows:
|
||
d = _haversine_m(lng, lat, float(row["lng"]), float(row["lat"]))
|
||
if d > radius:
|
||
continue
|
||
score, reasons = _score_place(dict(row), d, radius, intent)
|
||
results.append({
|
||
"place_id": row["element_id"],
|
||
"gaode_poi_id": row["gaode_poi_id"],
|
||
"name": row["name"],
|
||
"type_label": row["type_label"],
|
||
"place_type": row["place_type"],
|
||
"amap_type": row["amap_type"],
|
||
"typecode": row["typecode"],
|
||
"lng": float(row["lng"]),
|
||
"lat": float(row["lat"]),
|
||
"address": row["address"] or "",
|
||
"district": row["district"] or "",
|
||
"city": row["city"] or "",
|
||
"adcode": row["adcode"] or "",
|
||
"business_area": row["business_area"] or "",
|
||
"tel": row["tel"] or "",
|
||
"rating": row["rating"] or "",
|
||
"cost": row["cost"] or "",
|
||
"open_time": row["open_time"] or "",
|
||
"tags": row["tags"] or "",
|
||
"photo_urls": _photo_urls(row["photo_urls"]),
|
||
"h3_cell": row["h3_cell"],
|
||
"source": row["source"],
|
||
"last_fetched_at": row["last_fetched_at"].isoformat() if row.get("last_fetched_at") else "",
|
||
"distance_m": round(d, 1),
|
||
"score": score,
|
||
"rank_reasons": reasons,
|
||
})
|
||
results.sort(key=lambda r: (-r["score"], r["distance_m"]))
|
||
results = results[:60]
|
||
if use_llm:
|
||
answer, llm_reasons, answer_source = await _llm_answer(question, intent, results)
|
||
else:
|
||
answer, llm_reasons, answer_source = _rule_answer(question, intent, results)
|
||
for r in results:
|
||
if llm_reasons.get(r["place_id"]):
|
||
r["llm_reason"] = llm_reasons[r["place_id"]]
|
||
|
||
return {
|
||
"question": question,
|
||
"graph_name": graph_name,
|
||
"user_location": {"lng": lng, "lat": lat},
|
||
"intent": intent,
|
||
"answer": answer,
|
||
"results": results,
|
||
"trace": {
|
||
"intent_source": intent_source,
|
||
"answer_source": answer_source,
|
||
"latency_ms": max(1, round((time.perf_counter() - started_at) * 1000)),
|
||
"performance_target_ms": 1200,
|
||
"use_llm": use_llm,
|
||
"h3_resolution": res,
|
||
"h3_column": h3_col,
|
||
"h3_k": k,
|
||
"h3_cells": len(cells),
|
||
"h3_candidates": len(rows),
|
||
"radius_filtered": len(results),
|
||
"note": "当前结果来自已采集 guiyang_spatial_v1 空间知识图谱；采集未完成的区域会影响召回。",
|
||
},
|
||
}
|