最近在帮市场团队解决跨平台营销活动排期的问题，发现手动协调不仅效率低，还容易出错。于是用Python折腾了一个自动化排期系统，过程中踩了不少坑，也总结了些经验，分享给有类似需求的同学。

一、为什么要做自动化排期？

市场部同事的日常是这样的：

• Excel里有2026全年营销计划
• 微信群确认各平台负责人时间
• 人工录入到企业微信/钉钉日历
• 发现冲突再挨个打电话调整

用mermaid画的SWOT分析看得更清楚：

graph TD
    A[传统排期方式] --> B[优势: 操作简单]
    A --> C[劣势: 错误率32%]
    A --> D[机会: 已有历史数据]
    A --> E[威胁: 跨时区协作难]

二、技术方案选型

对比了三种主流方案：

| 方案 | 时区支持 | 富文本 | 免费配额 | |-----------------|----------|--------|--------------| | iCalendar | 一般 | ❌ | 无限制 | | Google Calendar | 完善 | ✅ | 100次/分钟 | | Outlook API | 完善 | ✅ | 需企业订阅 |

最终选择Google Calendar API + iCalendar解析的组合。

三、核心代码实现

1. 解析营销日历数据

from icalendar import Calendar

with open('marketing_2026.ics', 'rb') as f:
    cal = Calendar.from_ical(f.read())
    for event in cal.walk('VEVENT'):
        print(f"{event.get('summary')} 开始于 {event.get('dtstart').dt}")

2. 时间冲突检测（优化版）

def check_conflicts(events):
    # O(nlogn) 区间排序算法
    sorted_events = sorted(events, key=lambda x: x['start'])

    for i in range(1, len(sorted_events)):
        prev_end = sorted_events[i-1]['end']
        curr_start = sorted_events[i]['start']
        if curr_start < prev_end:
            raise ValueError(f"冲突事件: {sorted_events[i-1]['name']} 与 {sorted_events[i]['name']}")

3. Google API授权实战

关键是要处理token刷新：

from google.oauth2.credentials import Credentials
from google.auth.transport.requests import Request

creds = None
if os.path.exists('token.json'):
    creds = Credentials.from_authorized_user_file('token.json')

if not creds or not creds.valid:
    if creds and creds.expired and creds.refresh_token:
        creds.refresh(Request())  # 自动刷新token

四、生产环境必做的优化

处理夏令时的正确姿势

from pytz import timezone
import datetime

# 关键：始终使用带时区的datetime对象
start_time = datetime.datetime(2026, 3, 8, 9, 0, tzinfo=timezone('US/Eastern'))

异步任务用Celery

避免同步阻塞：

@app.task(bind=True)
def sync_to_calendar(self, event_data):
    try:
        service.events().insert(calendarId='primary', body=event_data).execute()
    except Exception as e:
        self.retry(exc=e, countdown=60)

五、踩过的三个大坑

1. API配额超限：

   # 解决方案：指数退避重试
   from time import sleep
   
   def call_api():
       for n in range(5):
           try:
               return api_call()
           except QuotaExceededError:
               sleep(2 ** n + random.random())

2. 时区转换丢失： 总是先转换为UTC再处理：

   dt = dt.astimezone(pytz.utc)

3. 重复事件同步： 用eventId做幂等性校验：

   if service.events().get(calendarId='primary', eventId=event_id).execute():
       logger.warning(f"事件 {event_id} 已存在")

六、未来优化方向

可以考虑用Airflow搭建跨年调度系统：

with DAG('marketing_calendar', schedule_interval='@yearly') as dag:
    parse_task = PythonOperator(task_id='parse_ical', python_callable=parse_ical)
    sync_task = PythonOperator(task_id='sync_gcal', python_callable=sync_to_calendar)

    parse_task >> sync_task

完整代码Colab笔记本

实际运行后，市场团队反馈排期效率提升了3倍多。建议从简单原型开始，逐步添加企业微信/钉钉对接，这个方案特别适合有海外业务的公司。