AI 实践｜Dify 实现埋点巡检方案

def calculate_fluctuation_rate(self, data: Dict, point_names: List[str]) -> Dict:        """        计算埋点波动率并识别异常        """        results = {}
        try:            if "data" not in data or not data["data"]:                return {}
            # 按埋点名称和日期组织数据            point_data = {}            for item in data["data"]:                point_name = item.get("point_name")                if point_name not in point_names:                    continue
                if point_name not in point_data:                    point_data[point_name] = {}
                date_str = item.get("date")                point_data[point_name][date_str] = {                    "pv": item.get("pv", 0),                    "uv": item.get("uv", 0)                }
            # 计算每个埋点的波动率            for point_name, dates in point_data.items():                if len(dates) < 2:                    logger.warning(f"埋点 {point_name} 数据不足，无法计算波动率")                    continue
                # 按日期排序                sorted_dates = sorted(dates.keys())                latest_date = sorted_dates[-1]                previous_date = sorted_dates[-2]
                latest_data = dates[latest_date]                previous_data = dates[previous_date]
                # 计算PV和UV的波动率                pv_change = self._calculate_change_rate(                    previous_data["pv"], latest_data["pv"]                )                uv_change = self._calculate_change_rate(                    previous_data["uv"], latest_data["uv"]                )
                # 判断是否异常                threshold = float(os.getenv("FLUCTUATION_THRESHOLD", 30.0))                is_abnormal = (                    abs(pv_change) > threshold or                     abs(uv_change) > threshold or                    latest_data["pv"] == 0 or  # 零值异常                    latest_data["uv"] == 0                )
                results[point_name] = {                    "latest_date": latest_date,                    "previous_date": previous_date,                    "pv_fluctuation": round(pv_change, 2),                    "uv_fluctuation": round(uv_change, 2),                    "latest_pv": latest_data["pv"],                    "latest_uv": latest_data["uv"],                    "previous_pv": previous_data["pv"],                    "previous_uv": previous_data["uv"],                    "is_abnormal": is_abnormal,                    "abnormal_reason": self._get_abnormal_reason(                        pv_change, uv_change,                         latest_data["pv"], latest_data["uv"],                        threshold                    ) if is_abnormal else None                }
        except Exception as e:            logger.error(f"计算波动率时发生错误: {str(e)}")
        return results
    def _calculate_change_rate(self, previous: int, current: int) -> float:        """计算变化率"""        if previous == 0:            return 100.0 if current > 0 else 0.0        return ((current - previous) / previous) * 100
    def _get_abnormal_reason(self, pv_change: float, uv_change: float,                             current_pv: int, current_uv: int, threshold: float) -> str:        """获取异常原因"""        reasons = []
        if current_pv == 0:            reasons.append("PV为零")        elif abs(pv_change) > threshold:            reasons.append(f"PV波动({pv_change:.2f}%)超过阈值({threshold}%)")
        if current_uv == 0:            reasons.append("UV为零")        elif abs(uv_change) > threshold:            reasons.append(f"UV波动({uv_change:.2f}%)超过阈值({threshold}%)")
        return "; ".join(reasons)

@app.get("/api/data/test_api")def test_api():    return {"msg": "success"}

gunicorn -c gunicornconf.py api_server.main:app -k uvicorn.workers.UvicornWorker -n qa_tracking