القائمة
يُحدَّد Prompt الذي يراجع المصدر أولًا بشكل افتراضي. يمكنك التبديل إلى أمر مباشر أو تنزيل نسخة محلية.
اقرأ SKILL.md وأي ملفات مرافقة يعرضها SkillsMP قبل أن تقرر التثبيت.
التثبيت باستخدام 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,
}