757. PotatoChat自动提醒机器人

你想要的是把“PotatoChat 自动提醒机器人”做成什么样子？在我开始给出详细设计或代码之前，先问几个问题以便把方案对准你的需求：

目标平台/接口
- “PotatoChat”是你们内部的聊天室，有公开的 Bot/Webhook API 吗？还是类似 Discord/Slack 的平台？（如果有 API 文档/示例请贴出来）
功能范围（请选择或补充）
- 一次性提醒 / 周期性（每天、每周、每月） / 自定义 cron 表达式？
- 私聊提醒 / 群组提醒 / @某用户提醒？
- 支持时区吗？需要处理夏令时吗？
- 支持“稍后提醒（snooze）”、取消、修改提醒？
- 支持富文本、图片或附件吗？
规模与可靠性
- 预计同时有多少个活跃提醒？是否需要高可用/分布式？
技术栈偏好
- Python/Node.js/Go/Java ？有没有部署平台（Heroku、Docker、K8s、Serverless）偏好？
持久化与权限
- 是否可以用 SQLite/MySQL/Postgres 或必须用某种云数据库？
其它要求
- 日志、重试策略、限速、用户隐私合规等？

如果你只是想要一个可运行的最小可行原型（MVP），下面给出一个清晰简洁的实现思路 + Python 示例（假设 PotatoChat 支持通过 HTTP webhook 向某个 URL 发消息或我们可以用 POST 请求向一个 bot webhook URL 发消息）。

方案概览（MVP）

技术栈：Python + FastAPI（接收/管理提醒的 HTTP API） + APScheduler（调度提醒） + SQLite（持久化） + requests（发送到 PotatoChat webhook）
功能：创建一次性或周期性提醒、列出提醒、删除提醒、当时间到时 POST 到 PotatoChat 的 webhook URL（支持带 @user 的文本）。
部署：Docker 容器，适合跑在 VPS 或云实例；也可改为 Serverless（需要替换调度方式）。

简要数据库设计（SQLite）

reminders: id, user_id, channel_id, text, cron (nullable), run_at (nullable datetime), timezone, repeat (boolean), webhook_url, enabled, created_at, next_run_at

关键点

如果是基于 cron/周期：存 cron 表达式并用 APScheduler 的 cron 触发器。
一次性提醒用 datetime 存 run_at 并用 date-trigger。
持久化下一次执行时间以支持重启恢复：启动时从 DB 恢复并 re-schedule。
时区要支持用户指定时区（或默认服务器时区）。
发送失败要有重试（指数退避）并记录日志；超时、HTTP 4xx/5xx 处理。
安全：验证创建提醒的请求（认证），存储敏感数据要加密（如需要）。

示例代码（精简版）

说明：下面是最小可运行的示例，省去了很多健壮性代码（认证、输入校验、重试、详细错误处理）。你需要替换 POTATO_WEBHOOK_URL 为实际的 PotatoChat webhook。

Python 代码（文件：app.py）

import sqlite3
import uuid
from datetime import datetime, timezone, timedelta
import requests
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.date import DateTrigger
from apscheduler.triggers.cron import CronTrigger
import atexit

DB = 'reminders.db'
scheduler = BackgroundScheduler()
app = FastAPI()

# --- DB helpers ---
def init_db():
    conn = sqlite3.connect(DB)
    c = conn.cursor()
    c.execute('''
    CREATE TABLE IF NOT EXISTS reminders (
        id TEXT PRIMARY KEY,
        user_id TEXT,
        channel_id TEXT,
        text TEXT,
        run_at TEXT,
        cron TEXT,
        timezone TEXT,
        webhook_url TEXT,
        enabled INTEGER,
        created_at TEXT
    )
    ''')
    conn.commit()
    conn.close()

def save_reminder(rem):
    conn = sqlite3.connect(DB)
    c = conn.cursor()
    c.execute('''
    INSERT INTO reminders (id,user_id,channel_id,text,run_at,cron,timezone,webhook_url,enabled,created_at)
    VALUES (?,?,?,?,?,?,?,?,?,?)
    ''', (rem['id'], rem.get('user_id'), rem.get('channel_id'), rem['text'],
          rem.get('run_at'), rem.get('cron'), rem.get('timezone'),
          rem.get('webhook_url'), 1, rem.get('created_at')))
    conn.commit()
    conn.close()

def delete_reminder_db(rid):
    conn = sqlite3.connect(DB)
    c = conn.cursor()
    c.execute('DELETE FROM reminders WHERE id=?', (rid,))
    conn.commit()
    conn.close()

def load_all_reminders():
    conn = sqlite3.connect(DB)
    c = conn.cursor()
    c.execute('SELECT id,user_id,channel_id,text,run_at,cron,timezone,webhook_url,enabled,created_at FROM reminders WHERE enabled=1')
    rows = c.fetchall()
    conn.close()
    keys = ['id','user_id','channel_id','text','run_at','cron','timezone','webhook_url','enabled','created_at']
    return [dict(zip(keys, r)) for r in rows]

# --- Reminder sending ---
def send_to_potato(webhook_url, text):
    # Simple POST; adjust payload to PotatoChat's API format
    payload = {"content": text}
    try:
        r = requests.post(webhook_url, json=payload, timeout=10)
        r.raise_for_status()
        return True
    except Exception as e:
        print("Send failed:", e)
        return False

def job_runner(rem):
    text = rem['text']
    webhook = rem['webhook_url']
    print(f"Firing reminder {rem['id']} -> {webhook}: {text}")
    ok = send_to_potato(webhook, text)
    if not ok:
        print("Failed to deliver reminder", rem['id'])
    # If it's a one-time reminder, disable it in DB
    if not rem.get('cron'):
        conn = sqlite3.connect(DB)
        c = conn.cursor()
        c.execute('UPDATE reminders SET enabled=0 WHERE id=?', (rem['id'],))
        conn.commit()
        conn.close()

def schedule_reminder(rem):
    # rem: dict with id, run_at (iso) or cron
    if rem.get('cron'):
        # naive: assume cron in standard five-field "min hour day month dow"
        trigger = CronTrigger.from_crontab(rem['cron'])
        scheduler.add_job(job_runner, trigger, args=[rem], id=rem['id'], replace_existing=True)
    elif rem.get('run_at'):
        dt = datetime.fromisoformat(rem['run_at'])
        if dt.tzinfo is None:
            dt = dt.replace(tzinfo=timezone.utc)
        trigger = DateTrigger(run_date=dt)
        scheduler.add_job(job_runner, trigger, args=[rem], id=rem['id'], replace_existing=True)
    else:
        print("No schedule for", rem['id'])

# --- API models ---
class CreateReminder(BaseModel):
    text: str
    run_at: str = None   # ISO format datetime, optional
    cron: str = None     # cron expression, optional
    webhook_url: str
    user_id: str = None
    channel_id: str = None
    timezone: str = None

@app.on_event("startup")
def startup():
    init_db()
    scheduler.start()
    # restore jobs from DB
    for rem in load_all_reminders():
        schedule_reminder(rem)
    atexit.register(lambda: scheduler.shutdown())

@app.post("/reminders")
def create_reminder(r: CreateReminder):
    if not (r.run_at or r.cron):
        raise HTTPException(status_code=400, detail="Either run_at or cron required")
    rid = str(uuid.uuid4())
    rem = {
        'id': rid,
        'user_id': r.user_id,
        'channel_id': r.channel_id,
        'text': r.text,
        'run_at': r.run_at,
        'cron': r.cron,
        'timezone': r.timezone,
        'webhook_url': r.webhook_url,
        'created_at': datetime.utcnow().isoformat()
    }
    save_reminder(rem)
    schedule_reminder(rem)
    return {"id": rid}

@app.get("/reminders")
def list_reminders():
    return load_all_reminders()

@app.delete("/reminders/{rid}")
def delete_reminder(rid: str):
    try:
        scheduler.remove_job(rid)
    except Exception:
        pass
    delete_reminder_db(rid)
    return {"deleted": rid}

如何使用

启动：pip install fastapi uvicorn apscheduler requests pydantic
uvicorn app:app –reload
创建提醒 POST /reminders, JSON 例如：
- 一次性： {"text":"开会了","run_at":"2026-03-10T09:00:00+00:00","webhook_url":"https://potato.example/webhook/xxx"}
- 每天 9 点： {"text":"每日总结","cron":"0 9 * * *","webhook_url":"…"}
PotatoChat webhook 的 payload 可能需要按平台格式修改 send_to_potato()

后续增强建议（可选）

鉴权（API token / OAuth）
更健壮的重试策略和失败告警
支持用户时区、解析自然语言时间（例如使用 parsedatetime）
支持 snooze（返回提醒 id 后可调用 /reminders/{id}/snooze）
UI（网页/聊天内 slash command）集成
分布式环境：用 Redis + RQ / Celery 或使用数据库驱动的调度器；或使用云调度服务
单元测试与监控（Prometheus、Sentry）

如果你确认了 PotatoChat 的 webhook/消息格式、希望的功能（例如只需一次性提醒 vs 需要周期/cron，是否需要支持 @用户），我可以：

基于具体 API 把 send_to_potato() 改成正确的请求体/鉴权；
提供完整 Dockerfile、部署脚本和示例 curl 请求；
或者给出 Node.js/Go 版本。

你先告诉我上面第 1-6 点的关键信息，或直接说“给我一个最小可用原型（可以部署）”，我就把代码、Dockerfile 和部署说明整理成可运行包。

757. PotatoChat自动提醒机器人

更多文章

767. PotatoChat社群活动组织

764. PotatoChat社群活跃度提升

758. PotatoChat与第三方办公工具集成

761. PotatoChat社区怎么运营