Geo Push Policy

管理事件推送策略，包括冷却期、观察池、推送次数限制、事件状态跟踪和频率限制保护。

Security Scan

The skill's code and instructions match its stated purpose (managing push/cooldown/watch-pool logic), and it does not request credentials or install software — but the provided source preview is truncated, so review the full file before deployment.

Purpose & Capability

The name/description, SKILL.md and the included Python module all describe push-policy logic (cooldowns, watch pool, dead-letter queue, push counts). Nothing in the manifest or shown code requests unrelated capabilities (no AWS, no broad system access).

Instruction Scope

Runtime instructions are narrowly scoped to loading/saving a state file, evaluating events, updating caches and handling a dead-letter queue. The SKILL.md examples show passing a send function (e.g., send_to_feishu) rather than embedding secret access. The skill does not instruct the agent to read arbitrary system files or exfiltrate data.

Install Mechanism

There is no install spec and no external downloads; the skill is delivered as source code + instructions, which is the lowest-risk installation model. An optional dependency 'geo_push_ops' is documented but not required by the core logic.

Credentials

The skill declares no required environment variables or credentials. The SKILL.md example references sending to Feishu (which would require credentials in a real deployment), but the skill itself does not ask for or access credentials — responsibility to supply/send is left to the integrator.

Persistence & Privilege

always:false and user-invocable:true (default). The code reads/writes its own state file via AppState.load/save; it does not attempt to modify other skills or global agent configuration. Autonomous invocation is allowed by platform default but is not excessive on its own.

Assessment

This skill appears coherent for managing push/cooldown/watch-pool logic and does not, as presented, request credentials or install external code. Before installing: (1) review the full geo_push_policy.py file (the provided preview was truncated) to confirm there are no hidden network calls or credential usages; (2) if you wire in a send function (e.g., send_to_feishu), ensure API tokens are stored securely and only provided when needed; (3) choose a safe STATE_FILE path and limit file permissions so the skill only writes to its own directory; (4) if you plan to run the skill autonomously, test it in an isolated environment first and confirm the optional dependency 'geo_push_ops' comes from a trusted source. If you want, I can scan the remainder of the file for network operations or secrets usage — provide the complete geo_push_policy.py content.

推送策略管理技能。负责事件冷却期、观察池、频率限制等推送策略管理。

功能

• 事件冷却期: 防止同一事件短时间内重复推送
• 观察池机制: 评分 50-69 的事件进入观察池，等待确认后再推送
• 推送次数限制: 单事件最多推送 3 次
• 事件状态机: 追踪事件从 new → escalating → ongoing → resolved 的演化
• 频率限制保护: 检测飞书频率限制，自动等待重试
• 死信队列: 记录推送失败的消息，支持后续补投

冷却期配置

事件级别	冷却时间	突破条件
高优先级 (≥70 分)	90 分钟	级别升级、新增权威来源、市场联动扩大
中优先级 (50-69 分)	180 分钟	同上

观察池机制

• 进入条件: 评分 50-69 分
• 池大小限制: 最多 20 条
• 升级检测: 观察池事件评分上升至≥70 分时转为推送

推送决策流程

新闻 → 评分 → 是否≥50 分？
         ↓ 否
         忽略
         ↓ 是
    检查冷却期 → 是否在冷却期？
         ↓ 是           ↓ 否
    检查是否升级      推送
         ↓ 是
        推送

使用方法

Python API

from geo_push_policy import (
    PushPolicy,
    AppState,
    should_push_event,
    update_event_cache,
    update_watch_pool,
)

# 加载状态
state = AppState.load(STATE_FILE)

# 创建策略
policy = PushPolicy(
    cooldown_high=90,
    cooldown_medium=180,
    max_push_count=3,
)

# 判断是否推送
should_push, reason = policy.should_push(event, state)
print(f"推送决策：{reason}")

# 更新事件缓存
push_events = [event1, event2]  # 本次推送的事件
update_event_cache(state, all_events, push_events)

# 更新观察池
update_watch_pool(state, all_events, push_events)

# 保存状态
state.save(STATE_FILE)

状态文件结构

{
  "last_run_time": "2026-03-27T15:00:00",
  "last_push_time": "2026-03-27T14:00:00",
  "event_cache": {
    "event_id_1": {
      "event_id": "event_id_1",
      "fingerprint": "abc123",
      "title": "事件标题",
      "severity": "high",
      "first_seen": "2026-03-27T10:00:00",
      "last_seen": "2026-03-27T15:00:00",
      "push_count": 2,
      "stage": "ongoing"
    }
  },
  "watch_pool": [...],
  "dead_letter_queue": [...]
}

死信队列

推送失败的消息会进入死信队列，最多保留 10 条。下次运行时会尝试补投。

# 处理死信队列
from geo_push_policy import process_dead_letter_queue, send_to_feishu

success = process_dead_letter_queue(state, send_to_feishu)
if success:
    print("✅ 死信补投成功")

依赖

• geo_push_ops: 推送操作模块（可选，用于死信补投）

相关文件

• 主模块：geo_push_policy.py

版本

• v1.0.0: 初始版本，从 smart-geo-push.py v2.0 拆分