up repo

app/controllers/valuation.py

@@ -24,6 +24,80 @@ class ValuationController:
     model = ValuationAssessment
     step_model = ValuationCalculationStep
     
+    # 参数说明映射表：将参数名（英文）映射到中文说明
+    PARAM_DESCRIPTIONS = {
+        # 财务价值相关
+        "three_year_income": "近三年收益（万元）",
+        "annual_revenue_3_years": "近三年收益（万元）",
+        "financial_value_f": "财务价值F",
+        
+        # 法律强度相关
+        "patent_score": "专利分",
+        "popularity_score": "普及分",
+        "infringement_score": "侵权分",
+        "legal_strength_l": "法律强度L",
+        
+        # 发展潜力相关
+        "patent_count": "专利数量",
+        "esg_score": "ESG分",
+        "innovation_ratio": "创新投入比",
+        "development_potential_d": "发展潜力D",
+        
+        # 行业系数
+        "industry_coefficient": "行业系数I",
+        "target_industry_roe": "目标行业ROE",
+        "benchmark_industry_roe": "基准行业ROE",
+        
+        # 流量因子相关
+        "search_index_s1": "搜索指数S1",
+        "industry_average_s2": "行业均值S2",
+        "social_media_spread_s3": "社交媒体传播度S3",
+        "likes": "点赞数",
+        "comments": "评论数",
+        "shares": "转发数",
+        "sales_volume": "销售量",
+        "link_views": "链接浏览量",
+        
+        # 政策乘数相关
+        "implementation_stage": "实施阶段评分",
+        "funding_support": "资金支持度",
+        "policy_match_score": "政策匹配度",
+        
+        # 文化价值相关
+        "inheritor_level_coefficient": "传承人等级系数",
+        "offline_sessions": "线下传习次数",
+        "douyin_views": "抖音浏览量",
+        "bilibili_views": "B站浏览量",
+        "kuaishou_views": "快手浏览量",
+        "cross_border_depth": "跨界合作深度",
+        "historical_inheritance": "历史传承度HI",
+        "structure_complexity": "结构复杂度SC",
+        "normalized_entropy": "归一化信息熵H",
+        
+        # 风险调整相关
+        "highest_price": "最高价格",
+        "lowest_price": "最低价格",
+        "inheritor_ages": "传承人年龄列表",
+        "lawsuit_status": "诉讼状态",
+        
+        # 市场估值相关
+        "manual_bids": "手动竞价列表",
+        "expert_valuations": "专家估值列表",
+        "weighted_average_price": "加权平均价格",
+        "daily_browse_volume": "日均浏览量",
+        "collection_count": "收藏数",
+        "issuance_level": "发行量",
+        "recent_market_activity": "最近市场活动时间",
+        
+        # 动态质押率相关
+        "monthly_transaction_amount": "月交易额（万元）",
+        "monthly_amount": "月交易额（万元）",
+        "heritage_asset_level": "非遗等级",
+        "dynamic_pledge_rate": "动态质押率",
+        "base_pledge_rate": "基础质押率",
+        "flow_correction": "流量修正系数",
+    }
+
     async def create_calculation_step(self, data: ValuationCalculationStepCreate) -> ValuationCalculationStepOut:
         """
         创建估值计算步骤
@@ -223,13 +297,12 @@ class ValuationController:
         """
         根据估值ID生成计算过程的 Markdown 报告。
 
-        此方法会查询所有相关的计算步骤，按照公式的层级关系组织，
+        此方法会查询所有相关的计算步骤，按照公式顺序组织，
         并生成格式化的 Markdown 文档，包含：
-        - 公式名称和说明
+        - 公式名称
         - 输入参数
+        - 公式文本
         - 输出结果
-        - 计算状态
-        - 错误信息（如果有）
 
         Args:
             valuation_id (int): 估值的唯一标识符。
@@ -245,10 +318,10 @@ class ValuationController:
         if not valuation:
             raise ValueError(f"估值记录不存在: {valuation_id}")
         
-        # 获取所有计算步骤
+        # 获取所有计算步骤，按 step_order 排序
         steps = await self.step_model.filter(valuation_id=valuation_id).order_by('step_order')
         if not steps:
-            return f"# 估值计算报告\n\n**估值ID**: {valuation_id}\n\n**资产名称**: {valuation.asset_name}\n\n> 暂无计算步骤记录。\n"
+            return f"# 计算摘要\n\n**估值ID**: {valuation_id}\n\n**资产名称**: {valuation.asset_name}\n\n> 暂无计算步骤记录。\n"
         
         # 转换为字典列表，便于处理
         steps_data = []
@@ -256,11 +329,8 @@ class ValuationController:
             step_dict = ValuationCalculationStepOut.model_validate(step).model_dump()
             steps_data.append(step_dict)
         
-        # 构建公式树形结构
-        formula_tree = self._build_formula_tree(steps_data)
-        
         # 生成 Markdown
-        markdown = self._generate_markdown(valuation, formula_tree)
+        markdown = self._generate_markdown(valuation, steps_data)
         
         logger.info("calcstep.report_markdown generated valuation_id={} steps_count={}", valuation_id, len(steps_data))
         return markdown
@@ -327,139 +397,202 @@ class ValuationController:
         
         return {'roots': root_nodes, 'all': tree}
     
-    def _generate_markdown(self, valuation, formula_tree: Dict) -> str:
+    def _generate_markdown(self, valuation, steps_data: List[Dict]) -> str:
         """
         生成 Markdown 格式的报告。
 
         Args:
             valuation: 估值评估对象。
-            formula_tree: 公式树形结构。
+            steps_data: 计算步骤列表（已按 step_order 排序）。
 
         Returns:
             str: Markdown 格式的字符串。
         """
         lines = []
         
-        # 标题和基本信息
-        lines.append("# 估值计算报告")
+        # 标题
+        lines.append("# 计算摘要")
         lines.append("")
-        lines.append("## 基本信息")
-        lines.append("")
-        lines.append("| 字段 | 值 |")
-        lines.append("|------|----|")
-        lines.append(f"| 估值ID | {valuation.id} |")
-        lines.append(f"| 资产名称 | {valuation.asset_name} |")
-        lines.append(f"| 所属机构 | {valuation.institution} |")
-        lines.append(f"| 所属行业 | {valuation.industry} |")
-        heritage_value = valuation.heritage_level if valuation.heritage_level else "-"
-        lines.append(f"| 非遗等级 | {heritage_value} |")
-        created_at_str = valuation.created_at.strftime("%Y-%m-%d %H:%M:%S") if valuation.created_at else "N/A"
-        lines.append(f"| 创建时间 | {created_at_str} |")
         lines.append("")
         
-        # 计算结果摘要
-        if valuation.final_value_ab is not None:
-            lines.append("## 计算结果摘要")
-            lines.append("")
-            lines.append("| 项目 | 数值（万元） |")
-            lines.append("|------|-------------|")
-            if valuation.model_value_b is not None:
-                lines.append(f"| 模型估值B | {valuation.model_value_b:.2f} |")
-            if valuation.market_value_c is not None:
-                lines.append(f"| 市场估值C | {valuation.market_value_c:.2f} |")
-            lines.append(f"| **最终估值AB** | **{valuation.final_value_ab:.2f}** |")
-            if valuation.dynamic_pledge_rate is not None:
-                lines.append(f"| 动态质押率 | {valuation.dynamic_pledge_rate:.4f} |")
-            lines.append("")
-        
-        # 详细计算过程
-        lines.append("## 详细计算过程")
-        lines.append("")
-        
-        def _format_json_block(value: Any, indent_prefix: str = "") -> List[str]:
-            """格式化 JSON 代码块"""
-            json_text = json.dumps(value, ensure_ascii=False, indent=2)
-            block_lines = [f"{indent_prefix}```json"]
-            for line in json_text.splitlines():
-                block_lines.append(f"{indent_prefix}{line}")
-            block_lines.append(f"{indent_prefix}```")
-            return block_lines
-
-        # 递归生成公式树
-        heading_levels = ["####", "#####", "######", "######", "######"]
-
-        def render_node(node: Dict, level: int = 0, prefix: str = ""):
-            step = node['step']
-            heading = heading_levels[min(level, len(heading_levels) - 1)]
+        # 遍历所有步骤，按顺序生成
+        for step in steps_data:
             name = step.get('formula_name', step.get('step_name', '未知'))
             formula_text = step.get('formula_text', step.get('step_description', ''))
-            status = step.get('status', 'unknown')
             input_params = step.get('input_params')
             output_result = step.get('output_result')
-            error_message = step.get('error_message')
-            duration_ms = None
-            if output_result and isinstance(output_result, dict):
-                duration_ms = output_result.get('duration_ms')
-            lines.append(f"{heading} {prefix}{name}")
+            
+            # 公式标题（二级标题）
+            lines.append(f"## {name}")
             lines.append("")
             
-            # 公式说明
-            if formula_text:
-                if level == 0:
-                    lines.append(f"> {formula_text}")
-                    lines.append("")
-                else:
-                    lines.append("计算公式：")
-                    lines.append(f"`{formula_text}`")
-                    lines.append("")
-            
-            # 状态和耗时
-            status_label = {
-                'processing': '计算中',
-                'completed': '已完成',
-                'failed': '计算失败'
-            }.get(status, status)
-            lines.append(f"**状态**: {status_label}")
-            if duration_ms is not None:
-                lines.append(f"**耗时**: {duration_ms}ms")
-            lines.append("")
-            
-            # 输入参数
+            # 参数部分
             if input_params:
-                lines.append("**输入参数**:")
-                lines.extend(_format_json_block(input_params, ""))
+                lines.append("**参数：**")
+                lines.append("")
+                # 格式化参数显示
+                param_lines = self._format_params(input_params)
+                lines.extend(param_lines)
                 lines.append("")
             
-            # 输出结果
+            # 公式部分
+            if formula_text:
+                lines.append("**公式：**")
+                lines.append("")
+                lines.append("```")
+                lines.append(formula_text)
+                lines.append("```")
+                lines.append("")
+            
+            # 结果部分
             if output_result:
-                # 移除 duration_ms，因为已经在状态中显示了
-                result_display = {k: v for k, v in output_result.items() if k != 'duration_ms'}
-                if result_display:
-                    lines.append("**输出结果**:")
-                    lines.extend(_format_json_block(result_display, ""))
+                # 提取主要结果值
+                result_value = self._extract_main_result(output_result, name)
+                if result_value is not None:
+                    lines.append("**结果：**")
+                    lines.append("")
+                    lines.append(f"`{result_value}`")
                     lines.append("")
             
-            # 错误信息
-            if error_message:
-                lines.append(f"> ⚠️ **错误**: {error_message}")
-                lines.append("")
-            
-            # 子节点
-            children = node.get('children', [])
-            if children:
-                for idx, child in enumerate(children, start=1):
-                    lines.append("")
-                    child_prefix = f"{prefix}{idx}."
-                    render_node(child, level + 1, child_prefix)
             lines.append("")
         
-        # 渲染所有根节点
-        for idx, root in enumerate(formula_tree['roots'], start=1):
-            prefix = f"{idx}."
-            render_node(root, 0, prefix)
-
         return "\n".join(lines)
     
+    def _format_params(self, params: Dict[str, Any]) -> List[str]:
+        """
+        格式化参数显示，优先使用列表格式（如果是数组），否则显示为列表项。
+        参数名会附带中文说明（如果存在）。
+        
+        Args:
+            params: 参数字典
+            
+        Returns:
+            List[str]: 格式化后的参数行列表
+        """
+        lines = []
+        
+        def _get_param_label(key: str) -> str:
+            """获取参数标签，包含中文说明"""
+            description = self.PARAM_DESCRIPTIONS.get(key)
+            if description:
+                return f"{key}（{description}）"
+            return key
+        
+        # 如果参数只有一个键，且值是数组，直接显示数组（不带参数名，符合示例格式）
+        if len(params) == 1:
+            key, value = next(iter(params.items()))
+            if isinstance(value, (list, tuple)):
+                # 格式化为列表：- [12.2, 13.2, 14.2]
+                value_str = json.dumps(list(value), ensure_ascii=False)
+                lines.append(f"- {value_str}")
+                return lines
+        
+        # 多个参数或非数组，显示为列表项（带说明）
+        for key, value in params.items():
+            param_label = _get_param_label(key)
+            if isinstance(value, (list, tuple)):
+                value_str = json.dumps(list(value), ensure_ascii=False)
+                lines.append(f"- **{param_label}**: {value_str}")
+            elif isinstance(value, dict):
+                value_str = json.dumps(value, ensure_ascii=False)
+                lines.append(f"- **{param_label}**: {value_str}")
+            else:
+                lines.append(f"- **{param_label}**: {value}")
+        
+        return lines
+    
+    def _extract_main_result(self, output_result: Dict[str, Any], formula_name: str) -> Optional[str]:
+        """
+        从输出结果中提取主要结果值。
+        
+        优先顺序：
+        1. 如果结果中只有一个数值类型的值，返回该值
+        2. 如果结果中包含与公式名称相关的字段（如 "财务价值 F" -> "financial_value_f"），返回该值
+        3. 如果结果中包含常见的计算结果字段（如 "result", "value", "output"），返回该值
+        4. 返回第一个数值类型的值
+        
+        Args:
+            output_result: 输出结果字典
+            formula_name: 公式名称
+            
+        Returns:
+            Optional[str]: 主要结果值的字符串表示，如果找不到则返回 None
+        """
+        if not output_result or not isinstance(output_result, dict):
+            return None
+        
+        # 移除 duration_ms 等元数据字段
+        filtered_result = {k: v for k, v in output_result.items() 
+                          if k not in ['duration_ms', 'duration', 'timestamp', 'status']}
+        
+        if not filtered_result:
+            return None
+        
+        # 如果只有一个值，直接返回
+        if len(filtered_result) == 1:
+            value = next(iter(filtered_result.values()))
+            if isinstance(value, (int, float)):
+                return str(value)
+            elif isinstance(value, (list, tuple)) and len(value) == 1:
+                return str(value[0])
+            else:
+                return json.dumps(value, ensure_ascii=False)
+        
+        # 尝试根据公式名称匹配字段
+        # 例如："财务价值 F" -> 查找 "financial_value_f", "财务价值F" 等
+        # 提取公式名称中的关键部分（通常是最后一个字母或单词）
+        name_parts = formula_name.split()
+        if name_parts:
+            # 获取最后一个部分（通常是字母，如 "F", "L", "D"）
+            last_part = name_parts[-1].lower()
+            # 构建可能的字段名：如 "financial_value_f", "legal_strength_l" 等
+            # 将中文名称转换为可能的英文字段名模式
+            possible_keys = []
+            
+            # 1. 直接匹配包含最后部分的字段（如包含 "f", "l", "d"）
+            for key in filtered_result.keys():
+                if last_part in key.lower() or key.lower().endswith(f"_{last_part}"):
+                    possible_keys.append(key)
+            
+            # 2. 尝试匹配常见的命名模式
+            # 例如："财务价值 F" -> "financial_value_f"
+            # 这里我们尝试匹配以最后部分结尾的字段
+            suffix_patterns = [
+                f"_{last_part}",
+                f"_{last_part}_",
+                last_part,
+            ]
+            
+            for key in filtered_result.keys():
+                key_lower = key.lower()
+                for pattern in suffix_patterns:
+                    if key_lower.endswith(pattern) or pattern in key_lower:
+                        if key not in possible_keys:
+                            possible_keys.append(key)
+            
+            # 按优先级匹配
+            for key in possible_keys:
+                if key in filtered_result:
+                    value = filtered_result[key]
+                    if isinstance(value, (int, float)):
+                        return str(value)
+        
+        # 查找常见的结果字段
+        common_result_keys = ['result', 'value', 'output', 'final_value', 'calculated_value']
+        for key in common_result_keys:
+            if key in filtered_result:
+                value = filtered_result[key]
+                if isinstance(value, (int, float)):
+                    return str(value)
+        
+        # 返回第一个数值类型的值
+        for key, value in filtered_result.items():
+            if isinstance(value, (int, float)):
+                return str(value)
+        
+        # 如果都没有，返回整个结果的 JSON（但简化显示）
+        return json.dumps(filtered_result, ensure_ascii=False)
+    
     async def create(self, data: ValuationAssessmentCreate, user_id: int) -> ValuationAssessmentOut:
         """创建估值评估"""
         # 将用户ID添加到数据中