Menü
Standardmäßig ist der Prompt ausgewählt, der zuerst die Quelle prüft. Sie können zu einem direkten Befehl wechseln oder eine lokale Kopie herunterladen.
Lesen Sie SKILL.md und alle von SkillsMP angezeigten Begleitdateien, bevor Sie sich für eine Installation entscheiden.
Mit Codex oder Claude installieren Kopieren Sie diesen Prompt, fügen Sie ihn in Codex, Claude oder einen anderen Assistant ein und lassen Sie die Skill-Seite prüfen und installieren.
Ein direkter Befehl überspringt den Prüf-Prompt. Prüfen Sie die Quelle, bevor Sie ihn ausführen.
npx skills add https://github.com/clowlove/Hermes-House --skill report-distribution-agentDer Befehl bleibt in einer Zeile. Scrollen Sie horizontal, um ihn vor dem Kopieren vollständig zu prüfen.
Sie bevorzugen eine lokale Kopie? Laden Sie die Dateien herunter, die SkillsMP derzeit vorliegen.
SKILL.md wird angezeigt
Basierend auf der SOC-Berufsklassifikation
| 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,
}