#!/usr/bin/env python3 # -*- coding: utf-8 -*- """WillNotes Bot — Telegram 消息 → Notion 任务库 用法：发送「学校 <内容>」入学校任务库；「任务 <内容>」入 Task 库 配置见同目录 .env（TELEGRAM_BOT_TOKEN / NOTION_TOKEN / SCHOOL_DB_ID / TASK_DB_ID） """ import json, os, re, time, datetime, urllib.error, urllib.parse, urllib.request BASE = os.path.dirname(os.path.abspath(__file__)) OFFSET_FILE = os.path.join(BASE, 'offset.txt') def load_env(): d = {} p = os.path.join(BASE, '.env') if os.path.exists(p): for ln in open(p, encoding='utf-8'): ln = ln.strip() if not ln or ln.startswith('#') or '=' not in ln: continue k, v = ln.split('=', 1) d[k.strip()] = v.strip().strip('"').strip("'") return d E = load_env() TG = E.get('TELEGRAM_BOT_TOKEN', '') NT = E.get('NOTION_TOKEN', '') SCHOOL_DB = E.get('SCHOOL_DB_ID', '2608f8e1-ce05-47bf-baea-aa1c429f3050') TASK_DB = E.get('TASK_DB_ID', '') STATUS_DEFAULT = E.get('STATUS_DEFAULT', '收件箱') SUBJECTS = ['语文', '数学', '英语', '科学', '道法', '美术', '音乐', '体育', '阅读', '劳动', '综合实践', '其他'] TYPES = ['作业', '交费', '准备物品', '通知', '活动', '其他'] API = 'https://api.telegram.org/bot%s/' % TG LOG = os.path.join(BASE, 'bot.log') def log(msg): line = '%s %s' % (datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'), msg) print(line, flush=True) try: with open(LOG, 'a', encoding='utf-8') as f: f.write(line + '\n') except Exception: pass def tg(method, **params): data = urllib.parse.urlencode(params).encode() req = urllib.request.Request(API + method, data=data) return json.load(urllib.request.urlopen(req, timeout=70)) def notion_create(db_id, props): body = {'parent': {'database_id': db_id}, 'properties': props} req = urllib.request.Request('https://api.notion.com/v1/pages', data=json.dumps(body).encode(), headers={'Authorization': 'Bearer ' + NT, 'Notion-Version': '2022-06-28', 'Content-Type': 'application/json'}, method='POST') return json.load(urllib.request.urlopen(req, timeout=40)) def parse_date(text): today = datetime.date.today() m = re.search(r'(?:(\d{4})[-/年])?(\d{1,2})[-/月](\d{1,2})日?', text) if m: y = int(m.group(1)) if m.group(1) else today.year try: return datetime.date(y, int(m.group(2)), int(m.group(3))).isoformat(), m.group(0) except ValueError: return None, None m = re.search(r'(今天|明天|后天)', text) if m: d = {'今天': 0, '明天': 1, '后天': 2}[m.group(1)] return (today + datetime.timedelta(days=d)).isoformat(), m.group(1) m = re.search(r'周([一二三四五六日天])', text) if m: wd = '一二三四五六日'.index(m.group(1).replace('天', '日')) + 1 delta = (wd - today.isoweekday()) % 7 or 7 return (today + datetime.timedelta(days=delta)).isoformat(), m.group(0) return None, None def build_props(kind, body): title = body props = {} for s in SUBJECTS: if s in body: title = title.replace(s, ' ', 1) if not props.get('_subj'): props['_subj'] = s for t in TYPES: if t in body: title = title.replace(t, ' ', 1) if not props.get('_type'): props['_type'] = t d, dtext = parse_date(body) if dtext: title = title.replace(dtext, ' ', 1) title = re.sub(r'\s+', ' ', title).strip(' ,，。;；') out = {'任务': {'title': [{'text': {'content': title[:1900]}}]}} if d: out['截止'] = {'date': {'start': d}} if props.get('_subj'): out['科目'] = {'select': {'name': props['_subj']}} if props.get('_type'): out['类型'] = {'select': {'name': props['_type']}} if kind == 'school': out['状态'] = {'select': {'name': STATUS_DEFAULT}} return out, title, d, props.get('_subj'), props.get('_type') def handle(text, chat_id): t = (text or '').strip() if t.startswith('学校'): kind, body, db, label = 'school', t[2:].strip(), SCHOOL_DB, '学校任务' elif t.startswith('任务'): kind, body, db, label = 'task', t[2:].strip(), TASK_DB, 'Task' else: tg('sendMessage', chat_id=chat_id, text='用法：\n「学校 <内容>」→ 学校任务库\n「任务 <内容>」→ Task 库') return if not body: tg('sendMessage', chat_id=chat_id, text='内容为空，例如：学校 数学 明天交实验报告') return if not db: tg('sendMessage', chat_id=chat_id, text='⚠️ %s 库尚未配置/未共享给集成，请先在 Notion 里授权' % label) return props, title, d, subj, typ = build_props(kind, body) try: r = notion_create(db, props) parts = ['✅ 已入「%s」' % label, '任务：%s' % title] if subj: parts.append('科目：%s' % subj) if typ: parts.append('类型：%s' % typ) if d: parts.append('截止：%s' % d) if kind == 'school': parts.append('状态：%s' % STATUS_DEFAULT) tg('sendMessage', chat_id=chat_id, text=' ｜ '.join(parts)) log('OK %s -> %s' % (label, title[:60])) except urllib.error.HTTPError as e: detail = e.read().decode('utf-8', 'ignore')[:200] tg('sendMessage', chat_id=chat_id, text='❌ 写入失败(%s)：%s' % (e.code, detail)) log('FAIL %s %s' % (e.code, detail)) except Exception as e: tg('sendMessage', chat_id=chat_id, text='❌ 异常：%s' % str(e)[:180]) log('ERR %s' % str(e)[:200]) def main(): if not TG or not NT: log('缺少 TELEGRAM_BOT_TOKEN / NOTION_TOKEN，退出') return try: me = tg('getMe') log('bot 启动: @%s' % me.get('result', {}).get('username')) except Exception as e: log('getMe 失败: %s' % e) off = 0 if os.path.exists(OFFSET_FILE): try: off = int(open(OFFSET_FILE).read().strip() or 0) except Exception: off = 0 tg('deleteWebhook', drop_pending_updates='false') while True: try: r = tg('getUpdates', offset=off, timeout=50) for u in r.get('result', []): off = u['update_id'] + 1 m = u.get('message') or {} txt = m.get('text') or '' cid = (m.get('chat') or {}).get('id') if cid and txt: handle(txt, cid) try: open(OFFSET_FILE, 'w').write(str(off)) except Exception: pass except Exception as e: log('轮询异常: %s' % str(e)[:160]) time.sleep(5) if __name__ == '__main__': main()