from fastapi import APIRouter, Depends from sqlalchemy.orm import Session from sqlalchemy import func, distinct from typing import List, Optional from pydantic import BaseModel from app.models import get_db from app.models.dit import DitRecord from app.models.sample import SampleRecord from app.models.order import OrderRecord from app.models.match import MatchResult, TargetType, MatchStatus router = APIRouter(prefix="/dashboard", tags=["Dashboard"]) # --- Pydantic Models --- class DashboardKPI(BaseModel): total_dit: int sample_rate: float hit_rate: float fulfillment_rate: float no_order_sample_rate: float total_revenue: float class FunnelData(BaseModel): name: str value: int fill: str class DitSchema(BaseModel): id: int op_id: Optional[str] = None customer: str pn: str eau: float = 0 stage: Optional[str] = None date: Optional[str] = None class Config: from_attributes = True class SampleSchema(BaseModel): id: int order_no: Optional[str] = None customer: str pn: str qty: int = 0 date: Optional[str] = None class Config: from_attributes = True class OrderSchema(BaseModel): id: int order_no: Optional[str] = None customer: str pn: str qty: int = 0 amount: float = 0 status: Optional[str] = None class Config: from_attributes = True class AttributionRow(BaseModel): dit: DitSchema sample: Optional[SampleSchema] = None order: Optional[OrderSchema] = None match_source: Optional[str] = None attributed_qty: int fulfillment_rate: float # --- Routes --- @router.get("/kpi", response_model=DashboardKPI) def get_kpi(db: Session = Depends(get_db)): total_dit = db.query(DitRecord).count() if total_dit == 0: return DashboardKPI( total_dit=0, sample_rate=0, hit_rate=0, fulfillment_rate=0, no_order_sample_rate=0, total_revenue=0 ) # Get valid matches valid_statuses = [MatchStatus.auto_matched, MatchStatus.accepted] # 1. Matches for Samples sample_matches = db.query(MatchResult.dit_id).filter( MatchResult.target_type == TargetType.SAMPLE, MatchResult.status.in_(valid_statuses) ).distinct().count() # 2. Matches for Orders order_matches = db.query(MatchResult.dit_id).filter( MatchResult.target_type == TargetType.ORDER, MatchResult.status.in_(valid_statuses) ).distinct().count() # 3. Revenue # Join MatchResult -> OrderRecord to sum amount revenue = db.query(func.sum(OrderRecord.amount)).join( MatchResult, MatchResult.target_id == OrderRecord.id ).filter( MatchResult.target_type == TargetType.ORDER, MatchResult.status.in_(valid_statuses) ).scalar() or 0.0 # 4. Fulfillment (Total Matched Order Qty / Total Matched DIT EAU) total_order_qty = db.query(func.sum(OrderRecord.qty)).join( MatchResult, MatchResult.target_id == OrderRecord.id ).filter( MatchResult.target_type == TargetType.ORDER, MatchResult.status.in_(valid_statuses) ).scalar() or 0 total_eau = db.query(func.sum(DitRecord.eau)).scalar() or 1 # Avoid div/0 sample_rate = round((sample_matches / total_dit) * 100, 1) hit_rate = round((order_matches / total_dit) * 100, 1) fulfillment_rate = round((total_order_qty / total_eau) * 100, 1) if total_eau > 0 else 0 # No Order Sample Rate dit_with_samples = set(x[0] for x in db.query(MatchResult.dit_id).filter( MatchResult.target_type == TargetType.SAMPLE, MatchResult.status.in_(valid_statuses) ).distinct().all()) dit_with_orders = set(x[0] for x in db.query(MatchResult.dit_id).filter( MatchResult.target_type == TargetType.ORDER, MatchResult.status.in_(valid_statuses) ).distinct().all()) dit_sample_no_order = len(dit_with_samples - dit_with_orders) no_order_sample_rate = round((dit_sample_no_order / len(dit_with_samples) * 100), 1) if dit_with_samples else 0.0 return DashboardKPI( total_dit=total_dit, sample_rate=sample_rate, hit_rate=hit_rate, fulfillment_rate=fulfillment_rate, no_order_sample_rate=no_order_sample_rate, total_revenue=revenue ) @router.get("/funnel", response_model=List[FunnelData]) def get_funnel(db: Session = Depends(get_db)): valid_statuses = [MatchStatus.auto_matched, MatchStatus.accepted] # Stage 1: DIT total_dit = db.query(DitRecord).count() # Stage 2: Sample dit_with_samples = db.query(MatchResult.dit_id).filter( MatchResult.target_type == TargetType.SAMPLE, MatchResult.status.in_(valid_statuses) ).distinct().count() # Stage 3: Order dit_with_orders = db.query(MatchResult.dit_id).filter( MatchResult.target_type == TargetType.ORDER, MatchResult.status.in_(valid_statuses) ).distinct().count() return [ FunnelData(name="DIT 總案", value=total_dit, fill="#6366f1"), FunnelData(name="成功送樣", value=dit_with_samples, fill="#a855f7"), FunnelData(name="取得訂單", value=dit_with_orders, fill="#10b981"), ] @router.get("/attribution", response_model=List[AttributionRow]) def get_attribution(db: Session = Depends(get_db)): valid_statuses = [MatchStatus.auto_matched, MatchStatus.accepted] matches = db.query(MatchResult).filter(MatchResult.status.in_(valid_statuses)).all() if not matches: return [] dit_ids = set(m.dit_id for m in matches) dits = db.query(DitRecord).filter(DitRecord.id.in_(dit_ids)).all() dit_map = {d.id: d for d in dits} sample_match_rows = [m for m in matches if m.target_type == TargetType.SAMPLE] order_match_rows = [m for m in matches if m.target_type == TargetType.ORDER] sample_ids = [m.target_id for m in sample_match_rows] order_ids = [m.target_id for m in order_match_rows] samples = db.query(SampleRecord).filter(SampleRecord.id.in_(sample_ids)).all() orders = db.query(OrderRecord).filter(OrderRecord.id.in_(order_ids)).all() sample_lookup = {s.id: s for s in samples} order_lookup = {o.id: o for o in orders} results = [] for dit_id, dit in dit_map.items(): s_matches = [m for m in matches if m.dit_id == dit_id and m.target_type == TargetType.SAMPLE] best_sample_match = max(s_matches, key=lambda x: x.score) if s_matches else None sample_obj = sample_lookup.get(best_sample_match.target_id) if best_sample_match else None o_matches = [m for m in matches if m.dit_id == dit_id and m.target_type == TargetType.ORDER] best_order_match = max(o_matches, key=lambda x: x.score) if o_matches else None order_obj = order_lookup.get(best_order_match.target_id) if best_order_match else None attributed_qty = 0 for om in o_matches: o = order_lookup.get(om.target_id) if o: attributed_qty += o.qty fulfillment_rate = round((attributed_qty / dit.eau * 100), 1) if dit.eau > 0 else 0 dit_schema = DitSchema.model_validate(dit) # Handle date to string conversion if needed, Pydantic often handles date -> string automatically in JSON response # checking earlier 'test_server_login' response showed JSON string for 'created_at'. # But here I set it manually to safe string just in case dit_schema.date = str(dit.date) if dit.date else None sample_schema = None if sample_obj: sample_schema = SampleSchema.model_validate(sample_obj) sample_schema.date = str(sample_obj.date) if sample_obj.date else None order_schema = None if order_obj: order_schema = OrderSchema.model_validate(order_obj) results.append(AttributionRow( dit=dit_schema, sample=sample_schema, order=order_schema, match_source=best_order_match.match_source if best_order_match else (best_order_match.reason if best_order_match else None), attributed_qty=attributed_qty, fulfillment_rate=fulfillment_rate )) return results