from fastapi import APIRouter, Depends from pydantic import BaseModel from typing import List from apps.api.dependencies.auth import get_api_key from packages.core.olap import get_clickhouse_client router = APIRouter(prefix="/bi", tags=["BI Dashboard"]) class TopologyNode(BaseModel): name: str type: str # country, port, or company value: float class TopologyEdge(BaseModel): source: str target: str value: float class FlowTopologyResponse(BaseModel): nodes: List[TopologyNode] edges: List[TopologyEdge] @router.get("/supply-chain-flow", response_model=FlowTopologyResponse) async def get_supply_chain_flow( hs_code: str, year: int, api_key: str = Depends(get_api_key) ): """ 提供“全球供应链流向拓扑图”底层数据聚合接口。 查询 ClickHouse 获取特定商品在全球范围内的流向(原产国 -> 目的国)。 """ client = get_clickhouse_client() # ClickHouse 查询示例: 聚合原产国和目的国的贸易额 query = f""" SELECT origin_country, destination_country, sum(amount) as total_amount FROM customs_data.trade_records WHERE hs_code = '{hs_code}' AND toYear(trade_date) = {year} AND origin_country != '' AND destination_country != '' GROUP BY origin_country, destination_country ORDER BY total_amount DESC LIMIT 50 """ try: result = client.query(query) rows = result.result_rows nodes_dict = {} edges = [] for row in rows: orig = row[0] dest = row[1] amt = float(row[2]) # 添加节点 if orig not in nodes_dict: nodes_dict[orig] = TopologyNode(name=orig, type="country", value=0) if dest not in nodes_dict: nodes_dict[dest] = TopologyNode(name=dest, type="country", value=0) nodes_dict[orig].value += amt nodes_dict[dest].value += amt # 添加边 edges.append(TopologyEdge(source=orig, target=dest, value=amt)) return FlowTopologyResponse( nodes=list(nodes_dict.values()), edges=edges ) except Exception as e: # 如果 ClickHouse 没有运行,返回空模拟数据 print(f"ClickHouse query failed: {e}") return FlowTopologyResponse(nodes=[], edges=[]) class MarketShareResponse(BaseModel): period: str share: float growth_rate_yoy: float # 同比 growth_rate_mom: float # 环比 @router.get("/market-share") async def get_market_share( hs_code: str, target_country: str, api_key: str = Depends(get_api_key) ): """ 提供“目标市场份额环比/同比”动态分析接口。 这里仅做接口示例,返回模拟的增长率。真实场景需在 ClickHouse 内做时间窗口聚合。 """ return [ MarketShareResponse(period="2026-04", share=0.15, growth_rate_yoy=0.05, growth_rate_mom=0.02), MarketShareResponse(period="2026-05", share=0.18, growth_rate_yoy=0.08, growth_rate_mom=0.03), ]