guanghulab/notion/sync-ticket.py

#!/usr/bin/env python3
"""
Notion ↔ 代码仓库双向同步器
同步周期：30分钟轮询

依赖：
  pip install requests

启动：
  export NOTION_TOKEN="secret_xxx"
  export NOTION_DATABASE_ID="xxx"
  pm2 start notion/sync-ticket.py --name notion-ticket-sync
"""

import os
import sys
import json
import time
import 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')
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'
}


def log(msg):
    """带时间戳的日志"""
    ts = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    print(f'[{ts}] {msg}', flush=True)


def get_repo_context():
    """获取代码仓库的最新状态"""
    context = {}
    try:
        # 获取最近提交
        result = subprocess.run(
            ['git', '-C', REPO_PATH, 'log', '--oneline', '-5'],
            capture_output=True, text=True, timeout=10
        )
        context['recent_commits'] = result.stdout.strip()
    except Exception as e:
        context['recent_commits'] = f'Error: {e}'
    
    try:
        # 获取最新认知链
        result = subprocess.run(
            ['ls', '-t', f'{REPO_PATH}/brain/d*-cognitive-chain.md'],
            capture_output=True, text=True, timeout=5
        )
        chains = [f for f in result.stdout.strip().split('\n') if f]
        context['latest_chain'] = chains[0] if chains else 'N/A'
    except Exception as e:
        context['latest_chain'] = f'Error: {e}'
    
    return context


def query_notion():
    """查询 Notion 工单数据库"""
    if not NOTION_TOKEN or not NOTION_DATABASE_ID:
        log('[跳过] 未配置 NOTION_TOKEN 或 NOTION_DATABASE_ID')
        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 API: {res.status_code} {res.text[:200]}')
            return []
        
        data = res.json()
        results = data.get('results', [])
        log(f'查询到 {len(results)} 条工单')
        return results
    except Exception as e:
        log(f'[异常] 查询 Notion 失败: {e}')
        return []


def find_pending_syncs(tickets):
    """找出需要同步到代码仓库的工单"""
    pending = []
    for ticket in tickets:
        props = ticket.get('properties', {})
        sync_status = props.get('同期状态', props.get('同步状态', {}))
        status = sync_status.get('select', {}).get('name', '')
        if status == '待同步':
            title = props.get('标题', {}).get('title', [{}])[0].get('plain_text', '无标题')
            ticket_type = props.get('类型', {}).get('select', {}).get('name', '任务')
            pending.append({
                'page_id': ticket['id'],
                'title': title,
                'type': ticket_type,
                'priority': props.get('优先级', {}).get('select', {}).get('name', 'P2'),
                'url': ticket.get('url', '')
            })
    return pending


def create_forgejo_issue(title, body, labels):
    """在 Forgejo 创建 Issue"""
    # Forgejo API 调用
    log(f'[创建] Issue: {title}')
    return True


def mark_synced(page_id):
    """标记 Notion 页面为已同步"""
    url = f'{NOTION_API}/pages/{page_id}'
    try:
        res = requests.patch(url, headers=HEADERS, json={
            'properties': {
                '同步状态': {'select': {'name': '已同步'}}
            }
        })
        return res.ok
    except Exception as e:
        log(f'[错误] 标记同步失败: {e}')
        return False


def save_tickets_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', ''),
                    'url': t.get('url', '')
                } for t in tickets]
            }, f, ensure_ascii=False, indent=2)
    except Exception as e:
        log(f'[警告] 保存缓存失败: {e}')


def sync_cycle():
    """一次同步周期"""
    log('=' * 50)
    log('开始同步周期')
    
    # 1. 获取仓库状态
    context = get_repo_context()
    log(f'仓库: {context.get("latest_chain", "?")}')
    
    # 2. 查询 Notion
    tickets = query_notion()
    if not tickets:
        log('无工单数据')
        return
    
    # 3. 找出待同步条目
    pending = find_pending_syncs(tickets)
    if pending:
        log(f'待同步到仓库: {len(pending)} 条')
        for item in pending:
            log(f'  → {item["title"]} ({item["priority"]})')
            # TODO: 创建 Forgejo Issue
            mark_synced(item['page_id'])
    else:
        log('无待同步条目')
    
    # 4. 保存缓存
    save_tickets_cache(tickets)
    
    log('同步周期完成')


if __name__ == '__main__':
    log(f'Notion ↔ 代码仓库同步器启动')
    log(f'轮询间隔: {INTERVAL}秒 = {INTERVAL//60}分钟')
    
    if not NOTION_TOKEN or not NOTION_DATABASE_ID:
        log('⚠️ 未配置 Notion 凭证')
        log('请设置环境变量:')
        log('  export NOTION_TOKEN="secret_xxx"')
        log('  export NOTION_DATABASE_ID="your-db-id"')
    
    # 首次同步
    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()