┌─────────────┐    gRPC    ┌──────────────────┐
│ Action      │ ────────── │                  │
│ Pipeline    │ ────────── │  Function Runtime │
│ Rule Engine │ ────────── │  (Reasoning & Decision Layer)        │
│ Dashboard   │ ────────── │                  │
└─────────────┘            └──────────────────┘

my_functions/
├── pyproject.toml
├── py.typed
├── __init__.py
├── functions/
│   ├── __init__.py
│   ├── risk_scoring.py
│   ├── notification.py
│   └── data_validation.py
├── models/
│   ├── __init__.py
│   └── schemas.py
└── tests/
    ├── __init__.py
    ├── test_risk_scoring.py
    └── test_notification.py

# pyproject.toml
[project]
name = "my-onto-functions"
version = "1.0.0"
requires-python = ">=3.11"

dependencies = [
    "ontology-sdk>=1.0.0",
    "pydantic>=2.0",
]

[project.optional-dependencies]
dev = [
    "pytest>=7.0",
    "pytest-asyncio>=0.21",
    "black",
    "ruff",
    "mypy",
]

# functions/risk_scoring.py
from ontology_sdk.functions import function, FunctionContext
from pydantic import BaseModel, Field
from typing import Optional
from enum import Enum


class RiskLevel(str, Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    CRITICAL = "critical"


class RiskInput(BaseModel):
    """风险评分输入参数"""
    customer_rid: str = Field(description="客户对象 RID")
    assessment_type: str = Field(default="standard", description="评估类型")


class RiskOutput(BaseModel):
    """风险评分输出结果"""
    risk_level: RiskLevel
    risk_score: float = Field(ge=0, le=100)
    risk_factors: list[str]
    recommendations: list[str]
    confidence: float = Field(ge=0, le=1)


@function(
    name="calculate_risk_score",
    display_name="客户风险评分",
    description="基于客户交易历史和行为数据计算风险评分",
    version="1.0.0",
    input_type=RiskInput,
    output_type=RiskOutput,
    tags=["risk", "scoring", "customer"],
)
def calculate_risk_score(ctx: FunctionContext, input: RiskInput) -> RiskOutput:
    """计算客户风险评分"""

    # 获取客户数据
    customer = ctx.platform.objects.get_by_rid(input.customer_rid)
    if not customer:
        raise ValueError(f"Customer not found: {input.customer_rid}")

    # 获取交易历史
    transactions = ctx.platform.oql.execute(f"""
        FIND Transaction
        TRAVERSE belongs_to -> Customer
        WHERE Customer.rid = '{input.customer_rid}'
        SELECT amount, transaction_date, type, status
        ORDER BY transaction_date DESC
        LIMIT 100
    """)

    # 计算风险因子
    risk_factors = []
    score = 50.0  # 基准分

    # 因子 1：交易频率异常
    recent_count = sum(1 for t in transactions
                       if (ctx.now() - t.transaction_date).days < 7)
    if recent_count > 20:
        score += 15
        risk_factors.append("近7天交易频率异常偏高")

    # 因子 2：大额交易占比
    large_txns = [t for t in transactions if t.amount > 100000]
    if len(large_txns) / max(len(transactions), 1) > 0.3:
        score += 20
        risk_factors.append("大额交易占比超过30%")

    # 因子 3：失败交易率
    failed = [t for t in transactions if t.status == "failed"]
    fail_rate = len(failed) / max(len(transactions), 1)
    if fail_rate > 0.1:
        score += 10
        risk_factors.append(f"交易失败率 {fail_rate:.1%}")

    # 因子 4：客户等级
    if customer.tier == "bronze":
        score += 5
        risk_factors.append("低等级客户")

    # 确定风险等级
    score = min(100, max(0, score))
    if score >= 80:
        level = RiskLevel.CRITICAL
    elif score >= 60:
        level = RiskLevel.HIGH
    elif score >= 40:
        level = RiskLevel.MEDIUM
    else:
        level = RiskLevel.LOW

    # 生成建议
    recommendations = []
    if level in (RiskLevel.CRITICAL, RiskLevel.HIGH):
        recommendations.append("建议人工审核该客户近期交易")
        recommendations.append("启用增强身份验证")
    if "大额交易" in str(risk_factors):
        recommendations.append("对大额交易启用双重审批流程")

    return RiskOutput(
        risk_level=level,
        risk_score=score,
        risk_factors=risk_factors,
        recommendations=recommendations,
        confidence=0.85,
    )

# functions/notification.py
from ontology_sdk.functions import async_function, FunctionContext
from pydantic import BaseModel, Field
from typing import Optional
import httpx


class NotificationInput(BaseModel):
    channel: str = Field(description="通知渠道: email/dingtalk/webhook")
    recipients: list[str] = Field(description="接收人列表")
    subject: str
    body: str
    template: Optional[str] = None
    priority: str = Field(default="normal")


class NotificationOutput(BaseModel):
    sent_count: int
    failed_count: int
    failures: list[dict] = Field(default_factory=list)


@async_function(
    name="send_notification",
    display_name="发送通知",
    description="通过指定渠道发送通知消息",
    version="1.0.0",
    input_type=NotificationInput,
    output_type=NotificationOutput,
    timeout_seconds=30,
    retry_on_failure=True,
    max_retries=3,
)
async def send_notification(
    ctx: FunctionContext, input: NotificationInput
) -> NotificationOutput:
    """发送通知到指定渠道"""

    sent = 0
    failed = 0
    failures = []

    # 渲染模板
    body = input.body
    if input.template:
        template = await ctx.platform.templates.get(input.template)
        body = template.render(body=input.body, subject=input.subject)

    async with httpx.AsyncClient() as client:
        for recipient in input.recipients:
            try:
                if input.channel == "email":
                    await _send_email(client, recipient, input.subject, body)
                elif input.channel == "dingtalk":
                    await _send_dingtalk(client, recipient, input.subject, body)
                elif input.channel == "webhook":
                    await _send_webhook(client, recipient, input.subject, body)
                sent += 1
            except Exception as e:
                failed += 1
                failures.append({
                    "recipient": recipient,
                    "error": str(e)
                })
                ctx.logger.warning(f"Failed to send to {recipient}: {e}")

    return NotificationOutput(
        sent_count=sent,
        failed_count=failed,
        failures=failures,
    )


async def _send_email(client, recipient, subject, body):
    await client.post(
        "http://mail-service:8080/api/send",
        json={"to": recipient, "subject": subject, "body": body}
    )


async def _send_dingtalk(client, recipient, subject, body):
    await client.post(
        f"https://oapi.dingtalk.com/robot/send?access_token={recipient}",
        json={"msgtype": "markdown", "markdown": {"title": subject, "text": body}}
    )


async def _send_webhook(client, recipient, subject, body):
    await client.post(recipient, json={"subject": subject, "body": body})

from ontology_sdk import OntoPlatform

platform = OntoPlatform(
    control_plane_url="localhost:50051",
    intelligence_plane_url="localhost:50053"
)

# 从模块注册
platform.functions.register_module("functions.risk_scoring")
platform.functions.register_module("functions.notification")

# 从目录批量注册
platform.functions.register_directory("functions/")

# 查看已注册函数
for fn in platform.functions.list():
    print(f"{fn.name} v{fn.version} - {fn.display_name}")
    print(f"  输入: {fn.input_schema}")
    print(f"  输出: {fn.output_schema}")

# 同步调用
result = platform.functions.invoke(
    "calculate_risk_score",
    input={
        "customer_rid": "ri.ontology.object.customer.c-001",
        "assessment_type": "enhanced"
    }
)
print(f"风险等级: {result.risk_level}")
print(f"风险分数: {result.risk_score}")

# 异步调用
import asyncio

async def main():
    result = await platform.functions.invoke_async(
        "send_notification",
        input={
            "channel": "dingtalk",
            "recipients": ["webhook-token-123"],
            "subject": "风险预警",
            "body": "客户 XX 风险评分异常升高"
        }
    )
    print(f"发送: {result.sent_count}, 失败: {result.failed_count}")

asyncio.run(main())

# 批量调用
batch_results = platform.functions.invoke_batch(
    "calculate_risk_score",
    inputs=[
        {"customer_rid": f"ri.ontology.object.customer.c-{i:03d}"}
        for i in range(1, 101)
    ],
    concurrency=10
)

# tests/test_risk_scoring.py
import pytest
from unittest.mock import MagicMock, AsyncMock
from functions.risk_scoring import calculate_risk_score, RiskInput, RiskLevel
from ontology_sdk.functions import FunctionContext


@pytest.fixture
def mock_context():
    ctx = MagicMock(spec=FunctionContext)

    # 模拟客户数据
    customer = MagicMock()
    customer.tier = "gold"
    customer.annual_revenue = 5000000
    ctx.platform.objects.get_by_rid.return_value = customer

    # 模拟交易数据
    ctx.platform.oql.execute.return_value = [
        MagicMock(amount=50000, transaction_date=ctx.now() - timedelta(days=1),
                  type="purchase", status="success"),
        MagicMock(amount=200000, transaction_date=ctx.now() - timedelta(days=3),
                  type="purchase", status="success"),
    ]

    return ctx


class TestRiskScoring:
    def test_low_risk_customer(self, mock_context):
        """低风险客户应返回 LOW 等级"""
        input_data = RiskInput(
            customer_rid="ri.ontology.object.customer.c-001"
        )

        result = calculate_risk_score(mock_context, input_data)

        assert result.risk_level == RiskLevel.LOW
        assert result.risk_score < 40
        assert result.confidence > 0

    def test_high_risk_customer(self, mock_context):
        """大量大额交易的客户应返回高风险"""
        # 模拟大量大额交易
        from datetime import timedelta
        now = mock_context.now()
        mock_context.platform.oql.execute.return_value = [
            MagicMock(amount=500000, transaction_date=now - timedelta(days=i),
                      type="purchase", status="success")
            for i in range(25)
        ] + [
            MagicMock(amount=100, transaction_date=now - timedelta(days=i),
                      type="purchase", status="failed")
            for i in range(5)
        ]

        input_data = RiskInput(
            customer_rid="ri.ontology.object.customer.c-002"
        )

        result = calculate_risk_score(mock_context, input_data)

        assert result.risk_level in (RiskLevel.HIGH, RiskLevel.CRITICAL)
        assert result.risk_score >= 60
        assert len(result.risk_factors) > 0
        assert len(result.recommendations) > 0

    def test_customer_not_found(self, mock_context):
        """客户不存在应抛出异常"""
        mock_context.platform.objects.get_by_rid.return_value = None

        input_data = RiskInput(
            customer_rid="ri.ontology.object.customer.nonexistent"
        )

        with pytest.raises(ValueError, match="Customer not found"):
            calculate_risk_score(mock_context, input_data)

# tests/test_integration.py
import pytest
from ontology_sdk import OntoPlatform
from ontology_sdk.testing import TestPlatform


@pytest.fixture
def test_platform():
    """使用测试平台（内存模式）"""
    return TestPlatform(
        seed_data={
            "Customer": [
                {"rid": "c-001", "name": "测试客户A", "tier": "gold"},
                {"rid": "c-002", "name": "测试客户B", "tier": "bronze"},
            ],
            "Transaction": [
                {"customer_rid": "c-001", "amount": 50000, "status": "success"},
                {"customer_rid": "c-001", "amount": 200000, "status": "success"},
                {"customer_rid": "c-002", "amount": 500000, "status": "failed"},
            ]
        }
    )


class TestRiskScoringIntegration:
    def test_end_to_end(self, test_platform):
        """端到端集成测试"""
        result = test_platform.functions.invoke(
            "calculate_risk_score",
            input={"customer_rid": "c-001"}
        )

        assert result.risk_level is not None
        assert 0 <= result.risk_score <= 100

@function(name="enrich_project_data", ...)
def enrich_project_data(ctx: FunctionContext, input: dict) -> dict:
    # 通过 RID 获取对象
    project = ctx.platform.objects.get_by_rid(input["project_rid"])

    # 通过过滤条件查找
    dept = ctx.platform.objects.get(
        "Department",
        filters={"code": project.department_code}
    )

    # OQL 查询
    team_members = ctx.platform.oql.execute(f"""
        FIND Employee
        TRAVERSE participates_in -> Project
        WHERE Project.rid = '{project.rid}'
        SELECT Employee.name, Employee.level, participates_in.role
    """)

    return {
        "project_name": project.name,
        "department": dept.name if dept else "Unknown",
        "team_size": len(team_members),
        "team": [{"name": m.name, "role": m.role} for m in team_members],
    }

@function(name="create_project_report", ...)
def create_project_report(ctx: FunctionContext, input: dict) -> dict:
    # 创建新对象
    report = ctx.platform.objects.create(
        object_type="Report",
        properties={
            "title": f"项目月报 - {input['month']}",
            "project_rid": input["project_rid"],
            "generated_at": ctx.now().isoformat(),
            "content": generate_report_content(ctx, input),
        }
    )

    # 创建关系
    ctx.platform.relations.create(
        relation_type="has_report",
        source_rid=input["project_rid"],
        target_rid=report.rid,
    )

    return {"report_rid": report.rid, "status": "created"}

# 注册新版本
@function(name="calculate_risk_score", version="2.0.0", ...)
def calculate_risk_score_v2(ctx, input):
    # 新版本逻辑
    pass

# 查看版本列表
versions = platform.functions.list_versions("calculate_risk_score")
for v in versions:
    print(f"v{v.version}: {v.status} (部署于 {v.deployed_at})")

# 灰度发布
platform.functions.set_traffic_split("calculate_risk_score", {
    "1.0.0": 80,  # 80% 流量走旧版本
    "2.0.0": 20,  # 20% 流量走新版本
})

# 全量切换
platform.functions.promote("calculate_risk_score", version="2.0.0")

# 回滚
platform.functions.rollback("calculate_risk_score", target_version="1.0.0")

# actions/assess_customer_risk.yaml
name: assess_customer_risk
type: function_call
function: calculate_risk_score
input_mapping:
  customer_rid: $trigger.object_rid
output_handling:
  - condition: output.risk_level == 'critical'
    action: send_notification
    params:
      channel: dingtalk
      subject: "高危客户预警"
      body: "客户 ${trigger.object.name} 风险评分: ${output.risk_score}"

transform:
  stages:
    - name: risk_enrichment
      type: function_call
      function: calculate_risk_score
      input_mapping:
        customer_rid: $record.customer_rid
      output_mapping:
        risk_level: $output.risk_level
        risk_score: $output.risk_score

# Dashboard widget 数据源
@function(name="dashboard_project_health", ...)
def dashboard_project_health(ctx: FunctionContext, input: dict) -> dict:
    projects = ctx.platform.oql.execute("""
        FIND Project WHERE status = 'in_progress'
        SELECT name, budget, end_date, priority
    """)

    return {
        "total_projects": len(projects),
        "by_priority": {
            "critical": len([p for p in projects if p.priority == "critical"]),
            "high": len([p for p in projects if p.priority == "high"]),
            "medium": len([p for p in projects if p.priority == "medium"]),
            "low": len([p for p in projects if p.priority == "low"]),
        },
        "total_budget": sum(p.budget for p in projects),
    }

from ontology_sdk.functions import function, FunctionContext
from functools import lru_cache


@function(name="get_department_info", cache_ttl=300, ...)
def get_department_info(ctx: FunctionContext, input: dict) -> dict:
    """带缓存的部门信息查询（TTL 5分钟）"""
    dept = ctx.platform.objects.get("Department", filters={"code": input["dept_code"]})
    return {"name": dept.name, "location": dept.location, "head_count": dept.head_count}

@function(
    name="batch_risk_assessment",
    concurrency_limit=20,
    timeout_seconds=60,
    ...
)
def batch_risk_assessment(ctx: FunctionContext, input: dict) -> dict:
    """批量风险评估，限制并发为 20"""
    results = []
    for customer_rid in input["customer_rids"]:
        result = ctx.invoke("calculate_risk_score", {"customer_rid": customer_rid})
        results.append(result)
    return {"assessments": results}

from ontology_sdk.functions import function, FunctionContext
from pydantic import BaseModel, Field
from typing import Optional
from datetime import datetime


class ApprovalInput(BaseModel):
    request_rid: str
    approver_rid: str
    decision: str  # approve / reject
    comment: Optional[str] = None


class ApprovalOutput(BaseModel):
    status: str
    next_approver: Optional[str] = None
    is_final: bool
    audit_trail_rid: str


@function(
    name="process_approval",
    display_name="处理审批",
    description="处理审批请求，支持多级审批流程",
    version="1.0.0",
    input_type=ApprovalInput,
    output_type=ApprovalOutput,
)
def process_approval(ctx: FunctionContext, input: ApprovalInput) -> ApprovalOutput:
    # 获取审批请求
    request = ctx.platform.objects.get_by_rid(input.request_rid)
    if not request:
        raise ValueError(f"Approval request not found: {input.request_rid}")

    # 验证审批人权限
    approver = ctx.platform.objects.get_by_rid(input.approver_rid)
    if request.current_approver_rid != input.approver_rid:
        raise PermissionError("当前用户不是指定的审批人")

    # 记录审批意见
    audit = ctx.platform.objects.create(
        object_type="AuditTrail",
        properties={
            "request_rid": input.request_rid,
            "approver_rid": input.approver_rid,
            "decision": input.decision,
            "comment": input.comment,
            "timestamp": ctx.now().isoformat(),
        }
    )

    # 处理审批决定
    if input.decision == "reject":
        ctx.platform.objects.update(
            rid=input.request_rid,
            properties={"status": "rejected", "completed_at": ctx.now().isoformat()}
        )
        return ApprovalOutput(
            status="rejected",
            is_final=True,
            audit_trail_rid=audit.rid,
        )

    # 查找下一级审批人
    approval_chain = ctx.platform.oql.execute(f"""
        FIND ApprovalStep
        TRAVERSE belongs_to -> ApprovalFlow
        WHERE ApprovalFlow.type = '{request.flow_type}'
        SELECT step_order, approver_role, min_amount
        ORDER BY step_order ASC
    """)

    current_step = next(
        (s for s in approval_chain if s.step_order == request.current_step), None
    )
    next_step = next(
        (s for s in approval_chain if s.step_order == request.current_step + 1), None
    )

    if next_step and request.amount >= next_step.min_amount:
        # 需要下一级审批
        next_approver = ctx.platform.oql.execute(f"""
            FIND Employee WHERE role = '{next_step.approver_role}' LIMIT 1
        """)

        ctx.platform.objects.update(
            rid=input.request_rid,
            properties={
                "current_step": request.current_step + 1,
                "current_approver_rid": next_approver[0].rid if next_approver else None,
                "status": "pending_approval",
            }
        )

        return ApprovalOutput(
            status="pending_next_approval",
            next_approver=next_approver[0].rid if next_approver else None,
            is_final=False,
            audit_trail_rid=audit.rid,
        )
    else:
        # 审批完成
        ctx.platform.objects.update(
            rid=input.request_rid,
            properties={"status": "approved", "completed_at": ctx.now().isoformat()}
        )
        return ApprovalOutput(
            status="approved",
            is_final=True,
            audit_trail_rid=audit.rid,
        )