前言

以Linux.do的平均水平，应该不需要介绍啥是RSS，为啥RSS了。

信息熵越来越高，尤其是最近几个月的AI信息大爆炸，不能通过RSS订阅的信息源越来越多，所以春节期间断断续续研究了几天，主要解决几个问题：

1. 如何订阅不支持RSS的网站
2. 如何把微信公众号纳入进RSS里
3. 如何解决信息重复，浪费注意力的问题。
   1. 上了一部新电影/电视剧，PT上充斥大量下载源，我只需要看到一次就可以了。
   2. 同样或者相似的内容在多个信息源，例如同时在网站和公众号都有发，但彼此可能又有独立信息。

RSS聚合订阅服务

开源和在线服务都花了些时间研究：
在线服务：Feedly、Inoreader
开源项目：Tiny Tiny RSS、FreshRSS

直接说结论，我选FreshRSS，相对TTRSS来说它更现代化，它们都支持Fever API可以整合到RSS客户端，FreshRSS还支持Google Reader API，它更强大、效率更高。简单来说，你在手机上看过的文章，通过FreshRSS可以同步到你的电脑上，这也是与直接使用RSS阅读器直接订阅源的重要区别。

FreshRSS可以当做是一个服务端和管理后台来使用，因此我对它的前端交互和美观并没有太多要求，真正的阅读在客户端。

FreshRSS自带一些简单的数据统计功能，例如长期无更新的订阅源，我觉得是比较实用的功能，该取关的取关。

FreshRSS自带了HTML+XPath抓取的功能和RSS-Bridge插件，这往往是在线服务的付费功能，不过，我选择使用RSSHub，下面有介绍。

FreshRSS的官方docker镜像不能用在群晖NAS上，似乎是因为群晖的kernel版本太低导致的，所以我使用了Linuxserver的版本，有一说一，Linuxserver的版本真贴心，把该映射到宿主机的文件都集合在一起映射了出来，很方便。以后我应该会多用Linuxserver发布的镜像。

FreshRSS支持多种数据库格式，我选择用PostgreSQL，最初是因为要使用pg_trgm扩展，不过后来废弃了。PG版本选择了 15.5，原因是参考了AWS RDS的默认最佳实践的选择。总之我不建议用SQLite，因为还有一些其他需求要操作数据库，下面会有具体写。

介绍下几个插件

顺便一提，某些插件（比如FreshRss FlareSolverr）在启用时，会把插件文件复制到其他地方，所以在“为啥我改了文件没生效啊”、“有缓存吗？重启下试试”、“怎么还不行”这种问题上也浪费了我两根烟。具体见插件的extension.php

目前插件这块，还有一些问题没有去解决，有空时我会再去研究和更新。

TODO：

TranslateTitlesCN插件选择谷歌翻译时会存在网络不通的问题。我没有选择给docker全局梯子，因为很多订阅源可以直接访问，无需梯子，但是它又无法满足Clash的 - RULE-SET,cncidr, 漏网之鱼 的条件，为每个订阅源维护代理规则真的太蠢了，所以最好的方式就是对插件的CURL请求做一些修改，有空时再处理。

[17-Feb-2024 07:50:43 Asia/Shanghai] Error in translation: Failed to get content from Google Translate API.
[17-Feb-2024 17:42:42 Asia/Shanghai] PHP Deprecated:  Creation of dynamic property FreshRSS_Feed::$TranslateTitles is deprecated in /config/www/freshrss/extensions/TranslateTitlesCN/extension.php on line 67

News Assistant插件代码运行有报错，有空时再去看看。

[18-Feb-2024 00:28:28 Asia/Shanghai] PHP Warning:  Attempt to read property "message" on true in /config/www/freshrss/extensions/xExtension-NewsAssistant/helper.php on line 62

有一些订阅源虽然支持RSS，但FresshRSS不能正确工作，等待进一步研究。

[15-Feb-2024 23:25:10 Asia/Shanghai] PHP Notice: A feed could not be found at `https://www.jiqizhixin.com/rss`; the status code is `200` and content-type is `application/rss+xml,application/xml; charset=utf-8` in /app/www/lib/SimplePie/SimplePie.php on line 1829

RSS阅读器

说结论，macOS选择Reeder5、iOS也选择Reeder5，Android选择FeedMe，使用体验都很好。Win上我看了几个都不符合预期，就网页版随便看看。

更多阅读器，FreshRSS给出了一份列表，可自行挑选。

我推荐开启 自动标为已读（Mark as read on scroll） 功能，挺方便的，如果你要用FreshRSS网页版浏览，记得开打开文章时将其置顶选项，经验教训。

获取订阅源

原生支持RSS的信息源

没啥好说的，直接用。
现在很多个人站点是基于WordPress构建的，虽然页面上没有提供RSS，但是试试在他的网址后面增加一点东西，说不定有惊喜，例如

https://www.xxx.com/rss
https://www.xxx.com/rss2
https://www.xxx.com/feed
https://www.xxx.com/atom
https://www.xxx.com/rss.xml
https://www.xxx.com/atom.xml
https://www.xxx.com/feed.xml

不支持原生RSS的信息源

宝藏级开源项目RSSHub，真的香。现在已经有大量的路由可以直接使用，可以慢慢看慢慢玩。

对于没有支持的网站和路由，上手难度也是相当的低，我从没写过Node.JS，对着别人写好的路由照葫芦画瓢，加上cocopilot的辅助和coze的答疑，竟也能一小时写出3个来。其实主要就是2类，一类是从HTML中获取数据，一类是从API接口中(JSON)中获取数据，所以只要写出第一个来，后面就是复制粘贴，轻轻松松。

举个例子，一个没写过Node.JS的非职业程序员写的路由（还没PR到项目中）。

module.exports = {
    'game520.com': {
        _name: 'Game520',
        '.': [
            {
                title: '最近更新',
                docs: 'https://docs.rsshub.app/routes/program-update#game520',
                source: ['/'],
                target: '/gamer520/latest',
            }
        ],
    },
};

module.exports = function (router) {
    router.get('/latest', require('./latest'));
};

module.exports = {
    '/latest': ['ShiFangJuMie'],
};

const got = require('@/utils/got');
const cheerio = require('cheerio');

const rootUrl = 'https://www.gamer520.com';

module.exports = async (ctx) => {
    const response = await got(rootUrl, {
        headers: {
            "Accept": '*/*',
            "Cookie": 'cao_notice_cookie=1',
            "Host": 'www.gamer520.com',
            "User-Agent": 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/121.0.0.0 Safari/537.36',
        }
    });

    const $ = cheerio.load(response.data);
    const list = $('body > div.site > div.site-content > div > main > div > div > div > div > div > main > div.row.posts-wrapper > div')
        .map((_, item) => ({
            title: $(item).find('div.entry-wrapper > header > h2 > a').text(),
            link: new URL($(item).find('div.entry-wrapper > header > h2 > a').attr('href'), rootUrl).href,
            pubDate: $(item).find('div.entry-wrapper > div > ul > li.meta-date > time').attr('datetime'),
            description: '',
            category: $(item).find('div.entry-wrapper > header > div > span > a').text(),
        }))
        .get();
/*
        // 站长留言说：“不要扒我了”，所以这里注释掉了
        const items = await Promise.all(
            list.map((item) =>
                ctx.cache.tryGet(item.link, async () => {
                    const detailResponse = await got(item.link);
                    const content = cheerio.load(detailResponse.data);
                    item.description = content('div.entry-content.u-text-format.u-clearfix').text();
                    // 休息0.5秒，防止被封
                    await new Promise((resolve) => setTimeout(resolve, 500));

                    return item;
                })
            )
        );
*/
    ctx.state.data = {
        title: 'Gamer520 - 最近更新',
        link: rootUrl,
        item: list,
    };
};

RSSHub本质上是爬虫，所以你不得不面对反爬虫的问题，对于要在RSS里要输出全文来说，短时间大量请求还是比较容易触发封禁的，这个各位各显神通吧。我个人来说对阅读器上看全文不是刚需，我还是比较喜欢点到原站点中观看。

如何解决重复/高相似度内容的噪音问题

最初想使用pg_trgm来做相似度检查，但如果我有1000篇文章，想要查重就要进行1000x1000次SQL查询遍历，我认为这是不合理的做法。也许还有其他高级玩法，但对我这个非专业程序员和资源本不富裕的NAS来说，我希望选择轻量级的解决方案。

以下是我的解决方案，也是我不选择SQLite的原因：

import psycopg2
from thefuzz import fuzz
try:
conn = psycopg2.connect(
host=“192.168.31.23”,
port=“5433”,
database=“freshrss”,
user=“freshrss”,
password=“freshrss”
)
cur = conn.cursor()
except Exception as e:
print(’连接数据库失败！')
查询未读文章
def Query_feeds():
cur.execute(“SELECT id, title, id_feed FROM admin_entry WHERE is_read = 0 ORDER BY date;”)
rows = cur.fetchall()
thefuzz(rows)
文章比较, 相似度大于80%的文章自动标记为已读
def thefuzz(arrays):
for i in range(len(arrays)):
# 跳过已经被置空的数组
if not arrays[i]:
continue
seed_id = arrays[i][0]
seed_title = arrays[i][1]
seed_feed = arrays[i][2]
# 遍历数组，比较相似度
for j in range(i + 1, len(arrays)):
# 跳过已经被置空的数组
if not arrays[j]:
continue
entry_id = arrays[j][0]
entry_title = arrays[j][1]
entry_feed = arrays[j][2]
# 同一RSS源的文章不进行比较
if seed_feed == entry_feed:
continue
else:
# 文章标题相似度比较
similarity = fuzz.ratio(seed_title, entry_title)
if similarity > 80:
# 相似度大于80%的文章自动标记为已读
Auto_mark_as_read(seed_id, seed_title, entry_id, entry_title, similarity)
# 将已经比较过的数组置空，避免重复比较
arrays[j] = None
print(f’{seed_id}: {seed_title} [{seed_feed}]’)
print(f’{entry_id}: {entry_title} [{entry_feed}] are similar by {similarity}%')
自动标记为已读
def Auto_mark_as_read(seed_id, seed_title, entry_id, entry_title, Similarity):
cur.execute(f"UPDATE admin_entry SET is_read = 1 WHERE id = {entry_id};")
cur.execute(f"INSERT INTO auto_marked_logs (seed_id, seed_title, entry_id, entry_title, Similarity) VALUES ({seed_id}, ‘{seed_title}’, {entry_id}, ‘{entry_title}’, ‘{Similarity}’);")
conn.commit()
print(f’————-’)
关闭数据库连接
def Close_connection():
cur.close()
conn.close()
if name == ‘main’:
Query_feeds()
if conn:
Close_connection()

-- ----------------------------
-- Table structure for auto_marked_logs
-- ----------------------------
DROP TABLE IF EXISTS "public"."auto_marked_logs";
CREATE TABLE "public"."auto_marked_logs" (
  "id" int4 NOT NULL DEFAULT nextval('automarkedlogs_id_seq'::regclass),
  "seed_id" varchar(255) COLLATE "pg_catalog"."default" NOT NULL,
  "seed_title" text COLLATE "pg_catalog"."default",
  "entry_id" varchar(255) COLLATE "pg_catalog"."default" NOT NULL,
  "entry_title" text COLLATE "pg_catalog"."default",
  "timestamp" timestamptz(6) DEFAULT CURRENT_TIMESTAMP,
  "similarity" float4
)
;

-- ----------------------------
-- Primary Key structure for table auto_marked_logs
-- ----------------------------
ALTER TABLE "public"."auto_marked_logs" ADD CONSTRAINT "automarkedlogs_pkey" PRIMARY KEY ("id");

非职业程序员出品，能用就行。期待有大佬出手AI+一下~~

理论上一定会有误伤，所以增加了同一个源之间不互相比较的判断。不同源之间的误伤，我能接受。

有时间时想抄一下琉璃的广告识别，先TODO一下。

如何将微信公众号的推文加入RSS

我花了几天时间研究了各种方式，尝试了RSSHub中的所有方案，坦白说目前没有完美方案。

先说结论，我目前的方案是 wechatbot-webhook + Wechat2RSS + 二十次幂 ，多渠道必然面临的问题就是信息重复，上一节已经解决了这个问题。

为什么一定要费劲折腾公众号转RSS订阅呢？a) 因为我厌倦了公众号页面的广告；b) 公众号的消息不是按照时间线排序，我不想让算法决定信息茧房；c) 我不想看重复的信息，例如这两天被SORA刷屏。

对于各类微信转RSS的方案，我的评测如下：

1. 已经失效、更新不完整或不可控、必须满足特定条件订阅的方案，全部放弃。
2. 搜狗来源，可用，但只有一篇文章。
3. Telegram 频道来源，详见这里，可用，但我放弃了该方案，它比较重，也比较繁琐，最重要的是，我真的没有那么多手机、那么多微信号可以分配给他使用。一旦你从手机上退出微信/切换账号，它也会一同掉线，所以他会占用一台你的手机。
4. Wechat2RSS 来源，它很稳，因为作者将xml结果文件提供给订阅，但只能在作者支持的公众号中选择，或向作者推荐，略感遗憾。作者有支持self-hosted的想法，重点关注，我愿称之为最期待的方案之一。
5. 二十次幂来源，好用，对于Wechat2RSS不支持的公众号，它是我的首选。2个小技巧，其一，搜索公众号次数受限，但创建榜单时按下你的F12，有奇效；其二，如果你想订阅的公众号没有被收录，你可以在这里提交，然后过一会再来。请低调使用，优先使用Wechat2RSS，减少对二十次幂的服务器压力。
6. 琉璃，也很吊的项目，亮点在备份和去广告。最后没有使用的原因是 a) 它现在并不是个活跃项目；b) 它的微信订阅源是使用的搜狗来源，所以和直接使用搜狗来源没差别；c) 使用企业微信需要域名在国内做ICP备案。所以无法发挥它的全部能力。
7. wechatbot-webhook，它不是一个与RSS有关的方案，它是一个微信机器人，我原本使用它给自己发一些通知（代替PushPlus一类的服务），下面开个小节专门说。

如何使用wechatbot-webhook订阅微信公众号

wechatbot-webhook是一个基于wechaty的新项目，它不是用来做RSS的，只是我把他用来做RSS，在他的众多能力中包含了支持公众号推文链接，这很符合我的期望。

它转RSS存在一些缺点，

1. 它和TG、搜狗一样，只能得到一篇推文（问题已经反馈给作者，不知道是否可解）
2. 对历史数据不可回溯，所以掉线期间的推文就是拿不到，好在还有其他方案并行兜底。
3. 它拥有其他微信个人号的通病，例如无法登录/掉线，详见作者的FAQ。

对一个非职业程序员来说，没脸发到Github/Gitlab，自己玩玩，也没有用环境变量。
没有cocopilot估计要写很久，现在只要一个晚上。始皇牛逼，感谢 @Llxyyds666 大佬带我上车。

main.py

from flask import Flask, request, Response
from feedgen.feed import FeedGenerator
from db_manager import DatabaseManager
from request_handler import parse_post_request
from logger import logger
实例化Flask应用
app = Flask(name)
实例化数据库管理器
db_manager = DatabaseManager({
‘dbname’: ‘wxmp2rss’,
‘user’: ‘wxmp2rss’,
‘password’: ‘wxmp2rss’,
‘host’: ‘192.168.31.23’,
‘port’: ‘5433’
})
设置路由，接收微信消息
@app.route("/receive", methods=[‘POST’])
def receive_data():
# 解析POST请求
data = parse_post_request(request)
if data:
try:
# 保存数据
db_manager.save_data(data)
return “Data received and stored successfully.”, 200
# 如果发生异常，记录日志
except Exception as e:
logger.error(f"An error occurred while processing the data.: {e}")
return None, 500
# 如果数据解析失败，记录日志
else:
logger.error(f"Invalid data received.")
return None, 500
设置路由，生成RSS
@app.route(’/rss2’, methods=[‘GET’])
def generate_rss():
# 实例化FeedGenerator，设置基本信息
fg = FeedGenerator()
fg.title(’微信公众号订阅小能手’)
fg.generator(‘Shifang by WechatBot-webhook’)
fg.link(href=‘https://192.168.31.23’, rel=‘alternate’)
fg.description(’由WechatBot-Webhook提供公众号消息服务。’)
fg.language(‘zh-cn’)
# 从数据库中获取数据
entries = db_manager.get_data()

# 生成RSS的条目
for entry in reversed(entries):
    fe = fg.add_entry()
    fe.category({'term': entry[2]})
    fe.title(entry[3])
    fe.id(entry[4])
    fe.link(href=entry[4])
    fe.description(f"<img src=\"{entry[5]}\" />")
    fe.pubDate(entry[6])
# 返回RSS，状态码200， MIME类型为application/rss+xml，浏览器会自动识别并显示RSS
response = Response(response=fg.rss_str(pretty=True), status=200, mimetype='application/rss+xml')
return response

if name == “main”:
app.run(host=‘0.0.0.0’, port=8080)
# 开启debug模式，方便调试，无需重启服务
# app.run(host=‘0.0.0.0’, port=8080, debug=True)

request_handler.py

from logger import logger
解析POST请求
def parse_post_request(request_data):
try:
parsed_data = {
’type’: request_data.form[’type’],
‘content’: request_data.form[‘content’],
‘source’: request_data.form[‘source’],
‘isMentioned’: request_data.form[‘isMentioned’],
‘isSystemEvent’: request_data.form[‘isSystemEvent’]
}
return parsed_data
except Exception as e:
logger.error(f"Error parsing request: {e}")
return None, 500

db_manager.py

from psycopg2 import pool
import json
from logger import logger
class DatabaseManager:
# 初始化数据库连接池
def init(self, db_config):
self.db_pool = pool.SimpleConnectionPool(minconn=1, maxconn=10, **db_config)
def fetch_query(self, query, vars=None):
    # 从连接池中获取一个连接
    connection = self.db_pool.getconn()
    result = None
    try:
        # 使用with语句，确保连接使用完毕后自动关闭
        with connection.cursor() as cursor:
            cursor.execute(query, vars)
            result = cursor.fetchall()
            connection.commit()
    # 如果发生异常，记录日志
    except Exception as e:
        logger.error(f"Error fetching query: {e}")
        connection.rollback()
    # 无论是否发生异常，都将连接放回连接池
    finally:
        self.db_pool.putconn(connection)
    # 返回查询结果
    return result

# 执行写入操作
def execute_write_query(self, query, vars=None):
    connection = self.db_pool.getconn()
    try:
        with connection.cursor() as cursor:
            cursor.execute(query, vars)
            connection.commit()
    except Exception as e:
        logger.error(f"Error executing write query: {e}")
        connection.rollback()
    finally:
        self.db_pool.putconn(connection)

# 保存数据
def save_data(self, data):
    # 如果消息类型是urlLink（公众号推文），则解析消息内容
    if data["type"] == "urlLink":
        # 解析消息内容
        source_data = json.loads(data["source"])
        content_data = json.loads(data["content"])
        # 执行写入操作
        query = """
        INSERT INTO msg (from_id, from_name, title, url, image) 
        VALUES (%s, %s, %s, %s, %s)
        """
        self.execute_write_query(query,
                                 (source_data["from"]['id'], source_data["from"]["payload"]["name"],
                                  content_data["title"], content_data["url"], content_data["thumbnailUrl"]))
    # 写个日志
    query = """
    INSERT INTO logs (type, content, source, is_mentioned, is_system_event) 
    VALUES (%s, %s, %s, %s, %s)
    """
    self.execute_write_query(query,
                             (data['type'], data['content'], data['source'], data['isMentioned'],
                              data['isSystemEvent']))

# 获取数据
def get_data(self):
    query = "SELECT * FROM msg ORDER BY id DESC LIMIT 200"
    return self.fetch_query(query)

logger.py

import logging
import sys
设置日志的配置
logging.basicConfig(
stream=sys.stdout,
level=logging.INFO,
format=’%(levelname)s - %(message)s’
)
使用日志
logger = logging.getLogger(name)

requirements.txt

Flask
psycopg2-binary
feedgen

Dockerfile‘

FROM python:3.9-slim
设置工作目录
WORKDIR /app
设置环境变量
RUN python -m venv venv
ENV PATH="/app/venv/bin:$PATH"
复制当前目录的内容到容器的/app目录
COPY . /app
更新pip并安装依赖
RUN pip install –upgrade pip && pip install –no-cache-dir -r requirements.txt
更新并安装依赖
RUN apt-get update && apt-get install -y apt-utils && apt-get install -y curl wget
安装依赖
RUN pip install –no-cache-dir -r requirements.txt
声明容器监听的端口
EXPOSE 8080
启动Python应用
CMD [“python”, “./main.py”]

public.sql

-- ----------------------------
-- Sequence structure for logs_id_seq
-- ----------------------------
DROP SEQUENCE IF EXISTS "public"."logs_id_seq";
CREATE SEQUENCE "public"."logs_id_seq" 
INCREMENT 1
MINVALUE  1
MAXVALUE 2147483647
START 1
CACHE 1;

– Sequence structure for msg_id_seq

DROP SEQUENCE IF EXISTS “public”.“msg_id_seq”;
CREATE SEQUENCE “public”.“msg_id_seq”
INCREMENT 1
MINVALUE  1
MAXVALUE 2147483647
START 1
CACHE 1;

– Table structure for logs

DROP TABLE IF EXISTS “public”.“logs”;
CREATE TABLE “public”.“logs” (
“id” int4 NOT NULL GENERATED ALWAYS AS IDENTITY (
INCREMENT 1
MINVALUE  1
MAXVALUE 2147483647
START 1
CACHE 1
),
“type” varchar(255) COLLATE “pg_catalog”.“default” NOT NULL,
“content” text COLLATE “pg_catalog”.“default” NOT NULL,
“source” text COLLATE “pg_catalog”.“default” NOT NULL,
“is_mentioned” int2 NOT NULL,
“is_system_event” int2 NOT NULL,
“timestamp” timestamptz(6) DEFAULT CURRENT_TIMESTAMP
)
;

– Table structure for msg

DROP TABLE IF EXISTS “public”.“msg”;
CREATE TABLE “public”.“msg” (
“id” int4 NOT NULL GENERATED ALWAYS AS IDENTITY (
INCREMENT 1
MINVALUE  1
MAXVALUE 2147483647
START 1
CACHE 1
),
“from_id” varchar(255) COLLATE “pg_catalog”.“default”,
“from_name” varchar(255) COLLATE “pg_catalog”.“default”,
“title” varchar(255) COLLATE “pg_catalog”.“default”,
“url” varchar(255) COLLATE “pg_catalog”.“default”,
“image” varchar(255) COLLATE “pg_catalog”.“default”,
“timestamp” timestamptz(6) DEFAULT CURRENT_TIMESTAMP
)
;

– Alter sequences owned by

ALTER SEQUENCE “public”.“logs_id_seq”
OWNED BY “public”.“logs”.“id”;
SELECT setval(’“public”.“logs_id_seq”’, 665, true);

– Alter sequences owned by

ALTER SEQUENCE “public”.“msg_id_seq”
OWNED BY “public”.“msg”.“id”;
SELECT setval(’“public”.“msg_id_seq”’, 29, true);

– Primary Key structure for table logs

ALTER TABLE “public”.“logs” ADD CONSTRAINT “logs_pkey” PRIMARY KEY (“id”);

– Primary Key structure for table msg

ALTER TABLE “public”.“msg” ADD CONSTRAINT “msg_pkey” PRIMARY KEY (“id”);

如何让微信小号快速上岗

wechatbot-webhook既然是一个个人号的机器人，就意味着这个微信号将失去登录PC/MAC的能力，所以也许你需要一个小号。

小号需要实名认证，这不是wechatbot-webhook的要求，而是微信的风控要求，具体还是看wechatbot-webhook的FAQ。

为了快速的将微信中的公众号转移到小号中，我尝试了一些办法，最后我选择 —— 按键精灵

找一台安卓手机，装一个自动点击软件，我使用的是贝利自动点击器，没什么门槛，大多数情况下都可以使用按钮点击功能精准的找到。

这个过程分2个阶段，第一阶段是将大号的公众号自动推荐给小号（大约每100个公众号需要30分钟）。第二阶段是在小号上关注这些公众号（不知道时间多久，我手动关注的，因为我做错了一件事）。

做错的这件事就是，我没有提前把小号下线！！所以等我登录小号的时候，大号推荐的聊天消息在另一台iPhone手机上=、=

遗憾的是我没办法把录好的脚本分享出来，以下是第一阶段的脚本：

1 个帖子 - 1 位参与者

阅读完整话题