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

推荐订阅源

云风的 BLOG
云风的 BLOG
M
MIT News - Artificial intelligence
Recent Announcements
Recent Announcements
Cyber Security Advisories - MS-ISAC
Cyber Security Advisories - MS-ISAC
Stack Overflow Blog
Stack Overflow Blog
J
Java Code Geeks
Microsoft Azure Blog
Microsoft Azure Blog
罗磊的独立博客
博客园 - 【当耐特】
H
Help Net Security
腾讯CDC
大猫的无限游戏
大猫的无限游戏
GbyAI
GbyAI
Last Week in AI
Last Week in AI
Jina AI
Jina AI
博客园 - 聂微东
Blog — PlanetScale
Blog — PlanetScale
A
About on SuperTechFans
Apple Machine Learning Research
Apple Machine Learning Research
P
Proofpoint News Feed
Y
Y Combinator Blog
C
Check Point Blog
博客园 - 司徒正美
钛媒体:引领未来商业与生活新知
钛媒体:引领未来商业与生活新知

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上增加有新的订阅时就给你发送一条消息,这样 当我有新的分享时,你就可以自动收到推送啦 :)