clowlove
clowlove/Hermes-House
用 Codex 或 Claude 帮你安装 复制这段 Prompt，粘贴到 Codex、Claude 或其他助手里，让它检查 Skill 页面并帮你完成安装。
直接命令不会经过审查 Prompt；运行前请先检查来源。
npx skills add https://github.com/clowlove/Hermes-House --skill report-distribution-agent命令会保持在同一行。复制前请横向滚动并检查完整内容。
想先保存到本地？可下载 SkillsMP 当前能够提供的文件。
Migrate Hermes Agent to a new server while keeping both instances running in parallel. Covers backup, SSH troubleshooting, skill/memory sync, and GitHub remote setup.
Backup, restore, and migrate Hermes Agent data across machines. Covers the local backup scripts, cron scheduling, retention policies, git remote management, shallow-clone migration, and cross-machine parallel deployment.
Modify PDF appearance without changing content/structure: change font colors, remove highlights, adjust styling. Preserves all text, layout, fonts, and embedded resources.
基于 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,
}