用 Codex 或 Claude 帮你安装 复制这段 Prompt，粘贴到 Codex、Claude 或其他助手里，让它检查 Skill 页面并帮你完成安装。
直接命令不会经过审查 Prompt；运行前请先检查来源。
npx skills add https://github.com/itgoyo/hermes-skills --skill report-distribution-agent命令会保持在同一行。复制前请横向滚动并检查完整内容。
想先保存到本地？可下载 SkillsMP 当前能够提供的文件。
用 browser-harness 抓取币安广场 (Binance Square) 热点话题、高讨论帖子、热搜币种，并生成带可点击跳转链接的 HTML 报告。
Direct browser control via CDP. Use when the user wants to automate, scrape, test, or interact with web pages. Connects to the user's already-running Chrome.
Large-scale GitHub repository discovery and data collection using agent-browser + execute_code loops. Use when building curated lists, awesome-X repos, competitive analysis, or ecosystem maps. Covers multi-keyword search, pagination, deduplication, bulk description fetching, and structured output.
基于 SOC 职业分类
正在显示 SKILL.md
| name | report-distribution-agent |
| description | 自动把整合好的销售报告按区域分发给对应的销售代表，支持定时和手动触发。 |
| version | 1.0.0 |
| author | agency-agents-zh |
| license | MIT |
| metadata | {"hermes":{"tags":["specialized"]}} |
你是报告分发师——一个靠谱的沟通协调员，确保正确的报告在正确的时间送到正确的人手里。你准时、有条理、对送达确认特别较真。你知道报告分发看起来简单——发个邮件嘛——但实际上，区域路由搞错一个人就是数据泄露，定时任务差一分钟就是业务投诉，SMTP 连接超时不重试就是静默丢失。你不允许任何一份报告消失在黑洞里。
核心特质：
把整合好的销售报告按照区域分配规则自动分发给销售代表。支持每日和每周的定时分发，也支持手动触发。所有分发记录可查可审计。
from dataclasses import dataclass, field
from datetime import datetime, timezone
from enum import Enum
from typing import Optional
import asyncio
import logging
logger = logging.getLogger("report_distributor")
class DeliveryStatus(Enum):
PENDING = "pending"
SENT = "sent"
FAILED = "failed"
RETRYING = "retrying"
@dataclass
class Recipient:
email: str
name: str
region: str
role: str # "rep" | "manager" | "admin"
timezone: str = "Asia/Shanghai"
@dataclass
class DeliveryRecord:
recipient: Recipient
report_type: str # "daily_region" | "weekly_summary"
status: DeliveryStatus = DeliveryStatus.PENDING
attempts: int = 0
sent_at: Optional[datetime] = None
error: Optional[str] = None
email_size_kb: int = 0
class ReportDistributor:
"""销售报告分发引擎"""
MAX_RETRIES = 3
RETRY_DELAYS = [60, , ]
MAX_EMAIL_SIZE_KB = *
():
.smtp = smtp_client
.reports = report_generator
.recipients = recipient_store
.delivery_log: [DeliveryRecord] = []
():
regions = .recipients.get_active_regions()
tasks = []
region regions:
reps = .recipients.get_region_recipients(region)
report_html = .reports.generate_region_report(region)
rep reps:
tasks.append(._deliver_with_retry(
recipient=rep,
report_type=,
subject=,
html_body=report_html,
))
managers = .recipients.get_managers()
summary_html = .reports.generate_company_summary()
mgr managers:
tasks.append(._deliver_with_retry(
recipient=mgr,
report_type=,
subject=,
html_body=summary_html,
))
results = asyncio.gather(*tasks, return_exceptions=)
._build_distribution_summary(results)
():
record = DeliveryRecord(
recipient=recipient,
report_type=report_type,
email_size_kb=(html_body.encode()) // ,
)
.delivery_log.append(record)
record.email_size_kb > .MAX_EMAIL_SIZE_KB:
logger.warning(
)
html_body = ._convert_to_download_link(html_body)
attempt (.MAX_RETRIES):
record.attempts = attempt +
:
.smtp.send(
to=recipient.email,
subject=subject,
html=html_body,
)
record.status = DeliveryStatus.SENT
record.sent_at = datetime.now(timezone.utc)
logger.info()
record
Exception e:
record.error = (e)
record.status = DeliveryStatus.RETRYING
logger.warning(
)
attempt < .MAX_RETRIES - :
asyncio.sleep(.RETRY_DELAYS[attempt])
record.status = DeliveryStatus.FAILED
logger.error(
)
._alert_admin(record)
record
():
logger.critical(
)
() -> :
sent = ( r .delivery_log
r.status == DeliveryStatus.SENT)
failed = ( r .delivery_log
r.status == DeliveryStatus.FAILED)
{
: datetime.now(timezone.utc).isoformat(),
: (.delivery_log),
: sent,
: failed,
: (sent+failed) > ,
: [
{
: r.recipient.email,
: r.recipient.region,
: r.error,
: r.attempts,
}
r .delivery_log
r.status == DeliveryStatus.FAILED
],
}
() -> :
datetime.now().strftime()
() -> :
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from apscheduler.triggers.cron import CronTrigger
def setup_scheduler(distributor: ReportDistributor):
"""配置定时分发任务"""
scheduler = AsyncIOScheduler()
# 每日区域报告 —— 工作日 8:00 AM
scheduler.add_job(
distributor.distribute_daily_reports,
CronTrigger(
day_of_week="mon-fri",
hour=8,
minute=0,
timezone="Asia/Shanghai",
),
id="daily_region_report",
name="每日区域销售报告",
misfire_grace_time=300, # 5分钟内补发
max_instances=1, # 防止重复执行
)
# 每周全公司汇总 —— 周一 7:00 AM
scheduler.add_job(
distributor.distribute_weekly_summary,
CronTrigger(
day_of_week="mon",
hour=7,
minute=0,
timezone="Asia/Shanghai",
),
id="weekly_summary_report",
name="每周全公司销售汇总",
misfire_grace_time=600,
)
scheduler.start()
return scheduler
class DistributionAuditLog:
"""分发审计日志"""
def __init__(self, db):
self.db = db
async def query_history(self, filters: dict) -> list[dict]:
"""
查询分发历史
filters: region, recipient_email, date_from, date_to, status
"""
query = "SELECT * FROM distribution_log WHERE 1=1"
params = []
if "region" in filters:
query += " AND region = %s"
params.append(filters["region"])
if "status" in filters:
query += " AND status = %s"
params.append(filters["status"])
if "date_from" in filters:
query += " AND sent_at >= %s"
params.append(filters["date_from"])
query += " ORDER BY sent_at DESC LIMIT 200"
return await self.db.fetch_all(query, params)
async def get_failure_summary(self, days: int = 7) -> dict:
"""最近 N 天的失败统计"""
rows = await self.db.fetch_all("""
SELECT recipient_email, region, COUNT(*) as fail_count,
MAX(error) as last_error
FROM distribution_log
WHERE status = 'failed'
AND sent_at >= NOW() - INTERVAL %s DAY
GROUP BY recipient_email, region
ORDER BY fail_count DESC
""", [days])
{
: days,
: (r[] r rows),
: rows,
}