Productivity Automation Kit

┌─────────────────────────────────────────────────┐
│ WORKFLOW: [工作流名称]                           │
│ 版本: v1.0    创建日期: [日期]                   │
├─────────────────────────────────────────────────┤
│ 触发器 TRIGGER                                  │
│   类型: [schedule|webhook|event|manual]         │
│   条件: [触发条件描述]                           │
├─────────────────────────────────────────────────┤
│ 输入 INPUTS                                    │
│   - [输入项1]: 来源 [来源系统]                   │
│   - [输入项2]: 来源 [来源系统]                   │
├─────────────────────────────────────────────────┤
│ 步骤 STEPS                                      │
│   Step 1: [操作名称]                            │
│     执行: [具体动作]                            │
│     成功→: Step 2                              │
│     失败→: 错误处理器                           │
│   Step 2: [操作名称]                            │
│     ...                                         │
├─────────────────────────────────────────────────┤
│ 错误处理 ERROR HANDLING                         │
│   重试: 最多3次，指数退避                       │
│   告警: [失败时通知渠道]                        │
├─────────────────────────────────────────────────┤
│ 输出 OUTPUTS                                   │
│   - [输出项1]: 目的地 [目标系统/文件]            │
│   - [输出项2]: 目的地 [目标系统/文件]            │
└─────────────────────────────────────────────────┘

触发器: 每天 09:00 (cron: 0 9 * * *)
步骤:
  1. 生成当日励志语录 (AI生成)
  2. 生成配套图片提示词
  3. 发布到社交媒体队列
  4. 记录发布日志
错误处理: 失败发送通知，重新执行最多3次
输出: content_queue/daily_YYYY-MM-DD.json

触发器: 每周一 08:00 (cron: 0 8 * * 1)
步骤:
  1. 汇总本周任务完成数据
  2. 计算KPI达成率
  3. 生成结构化周报
  4. 发送至指定邮箱/群组
输出: reports/weekly_YYYY-WXX.json

触发器: 新表单提交 / 新邮件到达
步骤:
  1. 验证数据完整性，去重
  2. 补充企业信息（自动化查询）
  3. 评分 (0-100 ICP匹配度)
  4. 路由:
     - 80分+: 即时通知 + 日历链接
     - 40-79分: 加入培育序列
     - <40分: 自动回复资料
  5. 记录至CRM
输出: leads/processed_YYYY-MM-DD.json

触发器: 发票收据到达（邮件附件 / 上传）
步骤:
  1. 提取关键信息（供应商、金额、到期日）
  2. 匹配至预算类别 / 采购订单
  3. 审批路由:
     - 金额在核准范围内 → 自动审批
     - 超阈值 → 转交经理
     - 无匹配PO → 标记待审
  4. 更新财务系统状态
  5. 发送付款确认通知
输出: invoices/processed_YYYY-MM-DD.json

月节省时间(小时) = (单次分钟数 / 60) × 月执行次数
自动化投入 = (搭建时间 × 时薪) + 工具月费
回收周期(月) = 投入 / 月节省价值

示例:
  任务: 手动填写表单 (15分钟/次, 20次/月)
  节省: 15/60×20 = 5小时/月
  投入: 1小时搭建 + $20/月工具费
  回收周期: 0.2个月 → 立即值得！

┌──────────┬────────────────────────────────┐
│ 时段     │ 安排                           │
├──────────┼────────────────────────────────┤
│ 06:00-08:00 │ 晨间准备 + 深度工作(黄金时段) │
│ 08:00-10:00 │ 高价值任务 (创意/决策)        │
│ 10:00-12:00 │ 会议 + 协作                   │
│ 12:00-13:30 │ 午餐 + 休息                   │
│ 13:30-15:30 │ 下午工作 (适合邮件/琐事)       │
│ 15:30-17:30 │ 收尾工作 + 明日计划           │
│ 17:30-19:00 │ 个人时间                      │
│ 19:00-22:00 │ 弹性时间 / 副业               │
│ 22:00-06:00 │ 睡眠                          │
└──────────┴────────────────────────────────┘

高能量时段 (认知高峰) → 深度工作、创意决策、学习新技能
中能量时段 → 会议、邮件、协作沟通
低能量时段 → 机械性任务、数据整理、归档

每周日 (20分钟)：
  1. 回顾上周完成与未完成
  2. 确定本周3大核心目标
  3. 分配至本周时间块
  4. 预判潜在阻碍，准备备选方案

每天晚间 (5分钟)：
  1. 回顾今日完成
  2. 确认明日Top 3任务
  3. 清理收件箱至0

📌 今日必做 (MIT - Most Important Tasks)
   - 最多3项，必须今天完成
   - 通常是最高价值的工作

📋 本周承诺
   - 本周内需要完成的任务
   - 来源：目标分解、会议决策、承诺

🔄 等待中
   - 委托给他人的任务
   - 等待外部条件的任务

✅ 已完成
   - 记录已完成的重要任务
   - 用于回顾和数据统计

任务提醒模板:
  任务: [任务名称]
  来源: [来自哪里：会议/邮件/自己安排]
  截止时间: [YYYY-MM-DD HH:MM]
  优先级: [P0/P1/P2/P3]
  预计耗时: [X小时/分钟]
  关联目标: [对应的目标或项目]
  阻碍因素: [当前卡点]
  需要资源: [完成所需资源]
  提醒时间: [提前多久提醒：1小时/1天/1周]
  提醒方式: [通知/邮件/消息]

🌅 晨间 (5分钟)
  → 查看今日MIT (最多3项)
  → 确认时间块安排
  → 清空昨日遗留（决定做/删/推迟）

☀️ 日间
  → 执行时间块任务
  → 新任务立即捕获至收集箱
  → 2分钟法则处理琐事

🌙 晚间 (5分钟)
  → 标记完成/未完成
  → 明日Top 3确认
  → 收件箱归零

触发器: [定时/文件变化/手动]
输入: [原始数据来源：CSV/JSON/API/表单]
处理步骤:
  1. 数据验证 (格式、完整性)
  2. 数据清洗 (去重、格式化、缺失值处理)
  3. 数据分类 (按规则打标签/分组)
  4. 数据统计 (汇总指标、计算KPI)
  5. 输出格式化 (生成报告/导出)
输出: [整理后数据/报告]
错误处理: [异常记录 + 告警]

✅ 格式验证：字段类型、长度、格式符合预期
✅ 完整性检查：无关键字段缺失
✅ 去重检查：无重复记录（按唯一ID判断）
✅ 一致性检查：同一实体数据在不同来源一致
✅ 时效性检查：数据是否为最新版本
✅ 权限检查：读取/写入权限正确

数据报告自动化:
  名称: [报告名称]
  频率: [每日/每周/每月]
  数据源:
    - [源系统1]: 连接方式 [API/文件/数据库]
    - [源系统2]: 连接方式 [...]
  指标计算:
    - 指标1: [计算公式]
    - 指标2: [计算公式]
  告警规则:
    - 触发条件: [指标 > 阈值]
    - 告警方式: [通知渠道]
  输出格式:
    - 摘要: [简短总结，用于消息推送]
    - 完整报告: [详细报告，存档或发送邮件]

#!/bin/bash
# 数据整理自动化脚本
# 用途: [描述]
# 频率: [执行频率]

set -euo pipefail
LOG_FILE="logs/data_process_$(date +%Y%m%d).log"
TIMESTAMP=$(date -u +"%Y-%m-%dT%H:%M:%SZ")

log() { echo "[$TIMESTAMP] $1" | tee -a "$LOG_FILE"; }

# Step 1: 数据采集
log "采集数据..."
DATA=$(curl -s -H "Authorization: Bearer $API_TOKEN" \
  "https://api.example.com/endpoint" || echo "")

# Step 2: 数据验证
if [ -z "$DATA" ]; then
  log "ERROR: 数据采集失败"
  exit 1
fi

# Step 3: 数据清洗
log "数据清洗..."
CLEANED=$(echo "$DATA" | jq '[.items[] | select(.status == "active")]')

# Step 4: 数据统计
COUNT=$(echo "$CLEANED" | jq 'length')
log "处理完成: $COUNT 条记录"

# Step 5: 输出
echo "$CLEANED" > "data/processed_$(date +%Y%m%d).json"
log "数据已保存"

#!/usr/bin/env python3
"""数据整理自动化脚本"""
import json
import csv
from datetime import datetime, timedelta
from pathlib import Path

def load_data(filepath: str) -> list:
    """加载原始数据"""
    with open(filepath, 'r', encoding='utf-8') as f:
        return json.load(f)

def clean_data(raw_data: list) -> list:
    """数据清洗"""
    cleaned = []
    seen = set()
    for item in raw_data:
        # 去重
        item_id = item.get('id')
        if item_id and item_id not in seen:
            seen.add(item_id)
            # 格式化字段
            cleaned.append({
                'id': item_id,
                'name': item.get('name', '').strip(),
                'value': float(item.get('value', 0)),
                'timestamp': item.get('created_at', '')
            })
    return cleaned

def calculate_metrics(data: list) -> dict:
    """计算统计指标"""
    if not data:
        return {'count': 0, 'total': 0, 'average': 0}
    total = sum(d['value'] for d in data)
    return {
        'count': len(data),
        'total': total,
        'average': total / len(data)
    }

def generate_report(data: list, metrics: dict) -> str:
    """生成报告摘要"""
    date_str = datetime.now().strftime('%Y-%m-%d')
    return f"""# 数据报告 - {date_str}

## 统计摘要
- 记录总数: {metrics['count']}
- 总值: {metrics['total']:.2f}
- 平均值: {metrics['average']:.2f}

## 最近更新
{chr(10).join(f"- {d['name']}: {d['value']}" for d in data[-5:])}
"""

if __name__ == '__main__':
    raw = load_data('data/raw/input.json')
    cleaned = clean_data(raw)
    metrics = calculate_metrics(cleaned)
    report = generate_report(cleaned, metrics)
    
    output_dir = Path('data/processed')
    output_dir.mkdir(parents=True, exist_ok=True)
    
    with open(f'data/processed/report_{datetime.now().strftime("%Y%m%d")}.md', 'w') as f:
        f.write(report)
    
    with open(f'data/processed/data_{datetime.now().strftime("%Y%m%d")}.json', 'w') as f:
        json.dump(cleaned, f, ensure_ascii=False, indent=2)
    
    print(f"报告生成完成: {metrics['count']} 条记录")