guanghulab/notion/sync-ticket.py

#!/usr/bin/env python3
"""
Notion ↔ 代码仓库双向同步器
对接数据库：⚡ 半体工单 · Half-Body Orders (05719bae-9e53-40f4-8233-6bf54551ac18)
同步周期：30分钟轮询

依赖：
  pip install requests

启动：
  export NOTION_TOKEN="xxx"
  export NOTION_DATABASE_ID="05719bae-9e53-40f4-8233-6bf54551ac18"
  pm2 start notion/sync-ticket.py --name notion-ticket-sync
"""

import os, sys, json, time, subprocess
from datetime import datetime

try:
    import requests
except ImportError:
    print('[Sync] 请先安装 requests: pip install requests')
    sys.exit(1)

# ── 配置 ──
NOTION_TOKEN = os.environ.get('NOTION_TOKEN')
NOTION_DATABASE_ID = os.environ.get('NOTION_DATABASE_ID') or '05719bae-9e53-40f4-8233-6bf54551ac18'
REPO_PATH = '/opt/guanghulab-repo'
CACHE_FILE = os.path.join(REPO_PATH, 'notion/tickets-cache.json')
INTERVAL = int(os.environ.get('SYNC_INTERVAL', '1800'))  # 30分钟

NOTION_API = 'https://api.notion.com/v1'
HEADERS = {
    'Authorization': f'Bearer {NOTION_TOKEN}',
    'Notion-Version': '2022-06-28',
    'Content-Type': 'application/json'
}

# 负责Agent选项（含铸渊）
AGENTS = ['译典A05', '培园A04', '录册A02', '霜砚Web', '铸渊']


def log(msg):
    ts = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    print(f'[{ts}] {msg}', flush=True)


def query_notion_tickets():
    """查询Notion半体工单数据库"""
    if not NOTION_TOKEN:
        log('[跳过] 未配置 NOTION_TOKEN')
        return []
    url = f'{NOTION_API}/databases/{NOTION_DATABASE_ID}/query'
    body = {
        'page_size': 50,
        'sorts': [{'timestamp': 'last_edited_time', 'direction': 'descending'}]
    }
    try:
        res = requests.post(url, headers=HEADERS, json=body, timeout=15)
        if res.status_code == 401:
            log('[错误] Notion token 无效或已过期')
            return []
        if not res.ok:
            log(f'[错误] Notion {res.status_code}')
            return []
        data = res.json()
        results = data.get('results', [])
        log(f'查询到 {len(results)} 条工单')
        return results
    except Exception as e:
        log(f'[异常] {e}')
        return []


def get_my_tickets(tickets):
    """找出分配给铸渊或未分配的工单"""
    mine = []
    for t in tickets:
        props = t.get('properties', {})
        agent = props.get('负责Agent', {}).get('select', {}).get('name', '')
        status = props.get('状态', {}).get('select', {}).get('name', '')
        title = props.get('任务标题', {}).get('title', [{}])[0].get('plain_text', '无标题')
        if agent in ('铸渊', '') and status not in ('已完成', '暂缓'):
            mine.append({
                'page_id': t['id'],
                'title': title,
                'status': status,
                'agent': agent,
                'url': t.get('url', ''),
                'code': props.get('编号', {}).get('rich_text', [{}])[0].get('plain_text', '') if not props.get('编号', {}).get('type') else ''
            })
    return mine


def save_cache(tickets):
    try:
        with open(CACHE_FILE, 'w') as f:
            json.dump({
                'last_sync': datetime.now().isoformat(),
                'count': len(tickets),
                'tickets': [{
                    'id': t['id'],
                    'title': t.get('properties', {}).get('任务标题', {}).get('title', [{}])[0].get('plain_text', ''),
                    'status': t.get('properties', {}).get('状态', {}).get('select', {}).get('name', '')
                } for t in tickets]
            }, f, ensure_ascii=False, indent=2)
    except Exception as e:
        log(f'[警告] 缓存失败: {e}')


def sync_cycle():
    log('=' * 50)
    log('开始同步周期')
    
    tickets = query_notion_tickets()
    if not tickets:
        return
    
    mine = get_my_tickets(tickets)
    if mine:
        log(f'铸渊相关工单: {len(mine)} 条')
        for item in mine:
            log(f'  → {item["title"]} [{item["status"]}]')
    else:
        log('无铸渊相关工单')
    
    save_cache(tickets)
    log('同步周期完成')


if __name__ == '__main__':
    log(f'Notion ↔ 代码仓库同步器启动')
    log(f'轮询间隔: {INTERVAL}秒')
    log(f'数据库: {NOTION_DATABASE_ID}')
    
    if not NOTION_TOKEN:
        log('⚠️ 未配置 NOTION_TOKEN')
        log('请设置: export NOTION_TOKEN="secret_xxx"')
    
    sync_cycle()
    
    while True:
        time.sleep(INTERVAL)
        try:
            sync_cycle()
        except KeyboardInterrupt:
            log('同步器停止')
            break
        except Exception as e:
            log(f'[严重] {e}')
            import traceback; traceback.print_exc()