"""评价模板管理 + 评分计算 API。 提供模板 CRUD、权重计算、评分历史查询等接口。 """ from typing import Any from fastapi import APIRouter, Depends, Query from pydantic import BaseModel, Field from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.core.database import get_db from app.core.dependencies import get_current_user from app.models.evaluation_template import EvaluationTemplate from app.models.user import User from app.schemas.common import ApiResponse, error, success from app.services.evaluation_engine import ( calculate_weighted_score, compute_weights, get_dimension_score_key, ) router = APIRouter(prefix="/evaluation", tags=["evaluation"]) # --- 请求/响应模型 --- class WeightComputeRequest(BaseModel): """权重计算请求。""" fund_type: str = Field(..., description="基金类型") fund_lifecycle: str = Field("investment", description="存续期阶段") company_stage: str = Field("a", description="企业阶段") industry: str = Field("ai", description="产业赛道") strategy: str = Field("growth", description="投资策略") investor_type: str = Field("investor", description="投资人类型") class TemplateCreateRequest(BaseModel): """创建模板请求。""" name: str = Field(..., description="模板名称") fund_type: str = Field(..., description="基金类型") fund_lifecycle: str = Field("investment", description="存续期阶段") company_stage: str = Field("a", description="企业阶段") industry: str = Field("ai", description="产业赛道") strategy: str = Field("growth", description="投资策略") investor_type: str = Field("investor", description="投资人类型") weights_json: dict[str, float] | None = Field(None, description="自定义权重（不传则自动计算）") is_default: bool = Field(False, description="是否为默认模板") class ScoreCalculateRequest(BaseModel): """评分计算请求。""" company_id: str = Field(..., description="企业 ID") template_id: str | None = Field(None, description="模板 ID（不传则自动匹配）") fund_type: str | None = Field(None, description="基金类型（无模板时用于自动匹配）") fund_lifecycle: str | None = Field(None, description="存续期阶段") company_stage: str | None = Field(None, description="企业阶段") industry: str | None = Field(None, description="产业赛道") strategy: str | None = Field(None, description="投资策略") structured_data: dict[str, Any] = Field(default_factory=dict, description="月报结构化数据") # --- API 端点 --- @router.post("/weights/compute", response_model=ApiResponse[dict]) async def compute_evaluation_weights( req: WeightComputeRequest, user: User = Depends(get_current_user), ): """根据 6 轴参数实时计算权重（不持久化）。""" result = compute_weights( fund_type=req.fund_type, fund_lifecycle=req.fund_lifecycle, company_stage=req.company_stage, industry=req.industry, strategy=req.strategy, investor_type=req.investor_type, ) return success(data=result) @router.get("/templates", response_model=ApiResponse[list]) async def list_templates( fund_type: str | None = Query(default=None, description="基金类型筛选"), fund_lifecycle: str | None = Query(default=None, description="存续期阶段筛选"), company_stage: str | None = Query(default=None, description="企业阶段筛选"), industry: str | None = Query(default=None, description="产业赛道筛选"), strategy: str | None = Query(default=None, description="投资策略筛选"), is_default: bool | None = Query(default=None, description="仅默认模板"), db: AsyncSession = Depends(get_db), user: User = Depends(get_current_user), ): """获取评价模板列表。""" query = select(EvaluationTemplate).where( EvaluationTemplate.tenant_id == user.tenant_id, EvaluationTemplate.is_active == True, # noqa: E712 ) if fund_type: query = query.where(EvaluationTemplate.fund_type == fund_type) if fund_lifecycle: query = query.where(EvaluationTemplate.fund_lifecycle == fund_lifecycle) if company_stage: query = query.where(EvaluationTemplate.company_stage == company_stage) if industry: query = query.where(EvaluationTemplate.industry == industry) if strategy: query = query.where(EvaluationTemplate.strategy == strategy) if is_default is not None: query = query.where(EvaluationTemplate.is_default == is_default) query = query.order_by(EvaluationTemplate.fund_type, EvaluationTemplate.company_stage) result = await db.execute(query) templates = result.scalars().all() return success(data=[ { "id": str(t.id), "name": t.name, "fund_type": t.fund_type, "fund_lifecycle": t.fund_lifecycle, "company_stage": t.company_stage, "industry": t.industry, "strategy": t.strategy, "investor_type": t.investor_type, "weights": t.weights_json, "enabled_dimensions": t.enabled_dimensions, "disabled_dimensions": t.disabled_dimensions, "custom_metrics": t.custom_metrics_json, "is_default": t.is_default, "version": t.version, } for t in templates ]) @router.get("/templates/{template_id}", response_model=ApiResponse[dict]) async def get_template( template_id: str, db: AsyncSession = Depends(get_db), user: User = Depends(get_current_user), ): """获取单个评价模板详情。""" result = await db.execute( select(EvaluationTemplate).where(EvaluationTemplate.id == template_id) ) tmpl = result.scalar_one_or_none() if not tmpl: return error(code=404, message="模板不存在") return success(data={ "id": str(tmpl.id), "name": tmpl.name, "fund_type": tmpl.fund_type, "fund_lifecycle": tmpl.fund_lifecycle, "company_stage": tmpl.company_stage, "industry": tmpl.industry, "strategy": tmpl.strategy, "investor_type": tmpl.investor_type, "weights": tmpl.weights_json, "enabled_dimensions": tmpl.enabled_dimensions, "disabled_dimensions": tmpl.disabled_dimensions, "custom_metrics": tmpl.custom_metrics_json, "lp_focus_metrics": tmpl.lp_focus_metrics, "regional_benchmark": tmpl.regional_benchmark, "is_default": tmpl.is_default, "is_active": tmpl.is_active, "version": tmpl.version, "created_at": tmpl.created_at.isoformat() if tmpl.created_at else None, "updated_at": tmpl.updated_at.isoformat() if tmpl.updated_at else None, }) @router.post("/templates", response_model=ApiResponse[dict]) async def create_template( req: TemplateCreateRequest, db: AsyncSession = Depends(get_db), user: User = Depends(get_current_user), ): """创建自定义评价模板。""" # 如果未提供权重，自动计算 if req.weights_json is None: result = compute_weights( fund_type=req.fund_type, fund_lifecycle=req.fund_lifecycle, company_stage=req.company_stage, industry=req.industry, strategy=req.strategy, investor_type=req.investor_type, ) weights_json = result["weights"] enabled_dims = result["enabled_dimensions"] disabled_dims = result["disabled_dimensions"] custom_metrics = {"metrics": result["custom_metrics"]} else: weights_json = req.weights_json # 从权重 key 推导启用/禁用维度 all_dims = list(get_dimension_score_key(k).replace("_score", "") for k in weights_json) enabled_dims = [k for k, v in weights_json.items() if v > 0] disabled_dims = [k for k in all_dims if k not in enabled_dims] custom_metrics = None tmpl = EvaluationTemplate( tenant_id=user.tenant_id, name=req.name, fund_type=req.fund_type, fund_lifecycle=req.fund_lifecycle, company_stage=req.company_stage, industry=req.industry, strategy=req.strategy, investor_type=req.investor_type, weights_json=weights_json, enabled_dimensions=enabled_dims, disabled_dimensions=disabled_dims, custom_metrics_json=custom_metrics, is_default=req.is_default, is_active=True, version=1, created_by=user.id, ) db.add(tmpl) await db.flush() return success(data={ "id": str(tmpl.id), "name": tmpl.name, "weights": tmpl.weights_json, }) @router.post("/score", response_model=ApiResponse[dict]) async def calculate_score( req: ScoreCalculateRequest, db: AsyncSession = Depends(get_db), user: User = Depends(get_current_user), ): """根据模板和月报数据计算评价评分。""" from app.models.health_score import HealthScore from app.services.health_calculator import calculate_health_score # 获取模板 tmpl: EvaluationTemplate | None = None if req.template_id: result = await db.execute( select(EvaluationTemplate).where(EvaluationTemplate.id == req.template_id) ) tmpl = result.scalar_one_or_none() if not tmpl: # 自动匹配模板 query = select(EvaluationTemplate).where( EvaluationTemplate.tenant_id == user.tenant_id, EvaluationTemplate.is_active == True, # noqa: E712 EvaluationTemplate.is_default == True, ) if req.fund_type: query = query.where(EvaluationTemplate.fund_type == req.fund_type) if req.fund_lifecycle: query = query.where(EvaluationTemplate.fund_lifecycle == req.fund_lifecycle) if req.company_stage: query = query.where(EvaluationTemplate.company_stage == req.company_stage) if req.industry: query = query.where(EvaluationTemplate.industry == req.industry) result = await db.execute(query.limit(1)) tmpl = result.scalar_one_or_none() # 计算各维度评分（复用现有 health_calculator） dimension_scores = calculate_health_score(req.structured_data) # 如果有模板，使用模板权重计算加权总分 if tmpl: weights = tmpl.weights_json # 将维度评分转换为权重 key 格式 dim_scores_by_weight_key: dict[str, float] = {} for weight_key in weights: score_key = get_dimension_score_key(weight_key) dim_scores_by_weight_key[weight_key] = dimension_scores.get(score_key, 0.0) total_score = calculate_weighted_score(dim_scores_by_weight_key, weights) # 保存评分记录 score_record = HealthScore( company_id=req.company_id, total_score=total_score, financial_score=dimension_scores.get("financial_score"), operational_score=dimension_scores.get("operational_score"), ai_commercial_score=dimension_scores.get("ai_commercial_score"), ai_cost_score=dimension_scores.get("ai_cost_score"), org_talent_score=dimension_scores.get("org_talent_score"), product_tech_score=dimension_scores.get("product_tech_score"), market_compete_score=dimension_scores.get("market_compete_score"), governance_score=dimension_scores.get("governance_score"), financing_score=dimension_scores.get("financing_score"), synergy_score=dimension_scores.get("synergy_score"), ai_model_product_score=dimension_scores.get("ai_model_product_score"), data_compliance_score=dimension_scores.get("data_compliance_score"), team_tech_score=dimension_scores.get("team_tech_score"), customer_success_score=dimension_scores.get("customer_success_score"), template_id=tmpl.id, fund_type=tmpl.fund_type, fund_lifecycle=tmpl.fund_lifecycle, company_stage=tmpl.company_stage, industry=tmpl.industry, strategy=tmpl.strategy, evidence_json={"template_name": tmpl.name, "weights": weights}, ) db.add(score_record) await db.flush() return success(data={ "score_id": str(score_record.id), "total_score": total_score, "dimension_scores": {k: v for k, v in dimension_scores.items() if k != "total_score"}, "template": { "id": str(tmpl.id), "name": tmpl.name, "weights": weights, }, }) else: # 无模板，使用默认计算 total_score = dimension_scores.get("total_score", 0.0) return success(data={ "total_score": total_score, "dimension_scores": {k: v for k, v in dimension_scores.items() if k != "total_score"}, "template": None, }) @router.get("/scores", response_model=ApiResponse[list]) async def list_scores( company_id: str | None = Query(default=None, description="企业 ID 筛选"), template_id: str | None = Query(default=None, description="模板 ID 筛选"), limit: int = Query(default=20, ge=1, le=100), db: AsyncSession = Depends(get_db), user: User = Depends(get_current_user), ): """获取评价评分历史。""" from app.models.health_score import HealthScore from app.models.company import Company query = ( select(HealthScore) .join(Company, HealthScore.company_id == Company.id) .where(Company.tenant_id == user.tenant_id) ) if company_id: query = query.where(HealthScore.company_id == company_id) if template_id: query = query.where(HealthScore.template_id == template_id) query = query.order_by(HealthScore.calculated_at.desc()).limit(limit) result = await db.execute(query) scores = result.scalars().all() return success(data=[ { "id": str(s.id), "company_id": s.company_id, "total_score": s.total_score, "financial_score": s.financial_score, "operational_score": s.operational_score, "ai_commercial_score": s.ai_commercial_score, "ai_cost_score": s.ai_cost_score, "org_talent_score": s.org_talent_score, "product_tech_score": s.product_tech_score, "market_compete_score": s.market_compete_score, "governance_score": s.governance_score, "financing_score": s.financing_score, "synergy_score": s.synergy_score, "ai_model_product_score": s.ai_model_product_score, "data_compliance_score": s.data_compliance_score, "team_tech_score": s.team_tech_score, "customer_success_score": s.customer_success_score, "trend": s.trend, "template_id": s.template_id, "fund_type": s.fund_type, "fund_lifecycle": s.fund_lifecycle, "company_stage": s.company_stage, "industry": s.industry, "strategy": s.strategy, "calculated_at": s.calculated_at.isoformat() if s.calculated_at else None, } for s in scores ]) # --- 基金管理端点 --- @router.get("/funds", response_model=ApiResponse[list]) async def list_funds( db: AsyncSession = Depends(get_db), user: User = Depends(get_current_user), ): """获取当前租户的基金列表。""" from app.models.fund import Fund result = await db.execute( select(Fund) .where(Fund.tenant_id == user.tenant_id, Fund.is_active == True) # noqa: E712 .order_by(Fund.established_date.desc()) ) funds = result.scalars().all() return success(data=[ { "id": str(f.id), "name": f.name, "fund_type": f.fund_type, "strategy": f.strategy, "established_date": f.established_date.isoformat() if f.established_date else None, "total_lifespan_months": f.total_lifespan_months, "investment_period_months": f.investment_period_months, "current_lifecycle": f.current_lifecycle, "lp_composition": f.lp_composition_json, "primary_market": f.primary_market, } for f in funds ]) @router.post("/funds", response_model=ApiResponse[dict]) async def create_fund( req: dict[str, Any], db: AsyncSession = Depends(get_db), user: User = Depends(get_current_user), ): """创建基金。""" from app.models.fund import Fund from datetime import date fund = Fund( tenant_id=user.tenant_id, name=req.get("name", ""), fund_type=req.get("fund_type", "early_vc"), strategy=req.get("strategy", "growth"), established_date=date.fromisoformat(req["established_date"]) if req.get("established_date") else None, total_lifespan_months=req.get("total_lifespan_months", 84), investment_period_months=req.get("investment_period_months", 48), lp_composition_json=req.get("lp_composition"), primary_market=req.get("primary_market"), ) db.add(fund) await db.flush() return success(data={ "id": str(fund.id), "name": fund.name, "current_lifecycle": fund.current_lifecycle, })