惯性聚合 高效追踪和阅读你感兴趣的博客、新闻、科技资讯
阅读原文 在惯性聚合中打开

推荐订阅源

博客园_首页
IT之家
IT之家
博客园 - Franky
Stack Overflow Blog
Stack Overflow Blog
宝玉的分享
宝玉的分享
Recent Announcements
Recent Announcements
Engineering at Meta
Engineering at Meta
S
SegmentFault 最新的问题
V
Visual Studio Blog
freeCodeCamp Programming Tutorials: Python, JavaScript, Git & More
OSCHINA 社区最新新闻
OSCHINA 社区最新新闻
Last Week in AI
Last Week in AI
H
Help Net Security
V
V2EX
H
Hackread – Cybersecurity News, Data Breaches, AI and More
量子位
博客园 - 叶小钗
J
Java Code Geeks
博客园 - 【当耐特】
月光博客
月光博客
爱范儿
爱范儿
人人都是产品经理
人人都是产品经理
酷 壳 – CoolShell
酷 壳 – CoolShell
小众软件
小众软件

Jiajun的技术笔记

你好,2026! TiDB 源码阅读(六):TiDB Coprocessor 源码解析 性能优化的核心思想 TiDB 源码阅读(五):索引 TiDB 源码阅读(四):AST、逻辑计划、物理计划 CockroachDB Serverless Architecture podman 无故退出 Cursor Control-L (CTRL-L) Keyboard Shortcuts in Terminal Replace docker with podman Using xmonad with xfce4 A RC script for freebsd frpc 自己动手写一个k8s controller AI 会取代你的(编程)岗位吗? 自建DERP服务器提升Tailscale连接速度(使用Nginx转发) 自动升级Docker容器 再读《程序员修炼之道-从小工到专家》 让浏览器下载文件 再读《软件随想录》/《黑客与画家》/《软技能》 HTTP 压力测试中的 Coordinated Omission 2的补码 编程语言中的 context 是什么? flutter macOS 构建出错 Flatpak 使用小记 Golang CAS 操作是怎么实现的 PostgreSQL 当MQ来使用 Clash 结合 工作VPN 的网络设计 使用 PostgreSQL 搭建 JuiceFS PostgreSQL 配置优化和日志分析 有GitHub Copilot?那就可以搭建你的ChatGPT4服务 窗口函数的使用(以PG为例)
写了一个Telegram Bot:自动化分享高质量内容
Jiajun Huang · 2018-09-02 · via Jiajun的技术笔记

平时我们读到优秀的资源例如文章,视频或者电子书等等,总是会忍不住收藏起来,但是如果我们能分享出来给所有人看到,那会不会更好呢?

所以我做了一个Telegram Bot,平时我只要把阅读到的不错的链接分享给我的bot,然后选择性的添加一些我自己的评语(或者叫推荐 理由)。然后会有另外一个页面来动态的渲染所有我分享的链接和评语。

欢迎直接访问:https://jiajunhuang.com/sharing

有Bot需求欢迎来找我定制开发哟

Bot代码

首先你要去Telegram上和BotFather聊天,其实就是输入一些命令,来创建一个你自己的Bot,它会下发给你一个Token。然后参考 https://core.telegram.org/bots/api 开始开发。当然,为了方便省事儿,我是直接用的一个Python封装好的包来做的。Bot的核心代码 如下:

import gevent.monkey
gevent.monkey.patch_all()  # noqa

import logging
logging.basicConfig(level=logging.INFO)  # noqa

from telegram.ext import Updater, CommandHandler, MessageHandler, Filters
from telegram import MessageEntity

from models import get_session, URLShare
from config import config


AUTHORS_FILTER = Filters.user(username="@jiajunhuang")


def report_error(func):
    def wrapper(bot, update, *args, **kwargs):
        try:
            return func(bot, update, *args, **kwargs)
        except Exception as e:
            logging.exception("failed to handle message from telegram")
            bot.send_message(chat_id=update.message.chat_id, text="出错啦:" + str(e))
    return wrapper


def save_url(url):
    with get_session() as s:
        url_share = URLShare(url=url)
        s.add(url_share)
        s.flush()
        return url_share.id


def save_comment(comment):
    with get_session() as s:
        share = s.query(URLShare).order_by(URLShare.id.desc()).first()
        if share:
            share.comment = comment
            s.add(share)
            return "mapped with url: " + share.url

        return "not found"


def update_comment(share_id, comment):
    with get_session() as s:
        share = s.query(URLShare).filter(URLShare.id == share_id).first()
        if share:
            share.comment = comment
            s.add(share)
            return "mapped with url: " + share.url

        return "not found"


@report_error
def comment_handler(bot, update, args):
    if len(args) == 0:
        text = "Usage: /comment <your comments>"
    else:
        text = save_comment("".join(args))

    bot.send_message(chat_id=update.message.chat_id, text=text)


@report_error
def update_comment_handler(bot, update, args):
    if len(args) == 0:
        text = "Usage: /update <id> <new comments>"
    else:
        text = update_comment(int(args[0]), "".join(args[1:]))

    bot.send_message(chat_id=update.message.chat_id, text=text)


@report_error
def url_share_handler(bot, update):
    bot.send_message(chat_id=update.message.chat_id, text="save with id: {}".format(save_url(update.message.text)))


if __name__ == "__main__":
    updater = Updater(token=config.TGBOTTOKEN)
    dispatcher = updater.dispatcher
    dispatcher.add_handler(
        CommandHandler(
            'comment', comment_handler, pass_args=True, filters=AUTHORS_FILTER,
        )
    )
    dispatcher.add_handler(
        CommandHandler(
            'update', update_comment_handler, pass_args=True, filters=AUTHORS_FILTER,
        ),
    )
    dispatcher.add_handler(MessageHandler(
        Filters.text & (
            Filters.entity(MessageEntity.URL) | Filters.entity(MessageEntity.TEXT_LINK)
        ) & AUTHORS_FILTER,
        url_share_handler,
    ))
    updater.start_polling()

models定义如下:

import datetime
import contextlib

from sqlalchemy import create_engine, Column, DateTime, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

from config import config

engine = create_engine(config.SQLALCHEMY_DB_URI, echo=config.SQLALCHEMY_ECHO)
Session = sessionmaker(bind=engine)

Base = declarative_base()


class BaseMixin:
    id = Column(Integer, primary_key=True, autoincrement=True)

    created_at = Column(DateTime, nullable=False, default=datetime.datetime.now)
    updated_at = Column(DateTime, nullable=False, default=datetime.datetime.now, onupdate=datetime.datetime.now)
    deleted_at = Column(DateTime, nullable=True, index=True)


@contextlib.contextmanager
def get_session():
    s = Session()
    try:
        yield s
        s.commit()
    except Exception:
        s.rollback()
        raise
    finally:
        s.close()


class URLShare(Base, BaseMixin):
    __tablename__ = "url_share"

    url = Column(String(1024), nullable=False)
    comment = Column(String(512))

当然上面的代码里 config.py 里的内容我就不贴出来了,毕竟为了简单方便,我直接吧token和数据库URL写到了代码里,在实际 工作上,这是 不好 的习惯,请不要学,谢谢。 有网友提醒说其实把 config.py 的涉密内容删掉就可以了,也对:

class Config:
    def __init__(self):
        self.SQLALCHEMY_DB_URI = "sqlite:////tmp/tgbot.db"  # 举个例子,这样。还是要修改成具体你的SQLite文件的路径
        self.SQLALCHEMY_ECHO = False
        self.TGBOTTOKEN = "你从BotFather那里申请来的token"


config = Config()

在第一次使用之前,需要初始化数据库schema,如果不想生成migration的话,就配置好 config.py 之后保存一个 gen.py

from models import Base, engine

Base.metadata.create_all(engine)

执行一下就可以了。

数据库用的是SQLite。为啥不用MySQL或者PG?答:为啥要用大炮打蚊子?而且,SQLite没有你想象中的那么弱。

虽然是一个简单到不行的网页,但是为了自动化方便,那也是要支持RSS的!

访问 https://jiajunhuang.com/sharing/rss 获取feed。有了RSS,你可以选择在IFTTT上增加有新的订阅时就给你发送一条消息,这样 当我有新的分享时,你就可以自动收到推送啦 :)