技术笔记

通常是因为某时刻应用程序大量请求内存导致系统内存不足造成的，这通常会触发 Linux 内核里的 Out of Memory (OOM) killer，OOM killer 会杀掉某个进程(用户态进程，不是内核线程)以腾出内存留给系统用，不致于让系统立刻崩溃。 ...

使用单行 perl 脚本将 Mailchimp 导出的数据转换为 listmonk 可用的清单。 perl -e 'print qq{email,name,attributes\n};while(<>){ my @r = split /,/; next unless $r[0] =~ /@/; map { s/"//g } @r; my $name = $r[1]; $name .= " $r[2]" if $r[2]; $name ||= "Unknown Name"; print qq{$r[0],"$name","{""mailchimp"": true}"\n}}' subscribed_segment_export_xxxxx.csv References mailchimp export convertor one-liner #688

listmonk 部署 安装 官方教程 进行即可，大致如下： # Download the compose file to the current directory. curl -LO https://github.com/knadh/listmonk/raw/master/docker-compose.yml # Run the services in the background. docker compose up -d rss to mail 脚本 主程序 main.py import feedparser import requests import json import os import logging from time import sleep from dateutil import parser from typing import List, Dict import re # 配置日志 logging.basicConfig( level=logging.DEBUG, format='%(asctime)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler('rss_checker.log'), logging.StreamHandler() ] ) logger = logging.getLogger(__name__) RSS_URL = os.getenv('RSS_URL', "https://xxx.com/feed/") LISTMONK_API_URL = os.getenv('LISTMONK_API_URL', "https://listmonk.xxx.com/api/campaigns") LISTMONK_TOKEN = os.getenv('LISTMONK_TOKEN', "bot:xxx") LISTMONK_SEND_LIST_ID = int(os.getenv('LISTMONK_SEND_LIST_ID', 4)) LISTMONK_SEND_LIST_IDS = [LISTMONK_SEND_LIST_ID] class RSSChecker: def __init__(self): self.rss_url = RSS_URL self.listmonk_url = LISTMONK_API_URL self.headers = { "Content-Type": "application/json", "Authorization": "token" + LISTMONK_TOKEN } self.max_retries = 3 self.retry_delay = 5 # seconds def clean_html_content(self, html_content: str) -> str: """清理HTML内容，移除以http://或https://开头的内容""" try: if not html_content: return "" # 移除以http://或https://开头的内容 cleaned_content = re.sub(r'https?://\S+', '', html_content) # 清理多余的空白字符 cleaned_content = re.sub(r'\s+', ' ', cleaned_content).strip() return cleaned_content except Exception as e: logger.error(f"清理HTML内容时出错: {str(e)}") return html_content # 如果处理失败，返回原始内容 def get_last_check_time(self) -> str: try: with open('last_check.txt', 'r') as f: last_time = f.read().strip() logger.debug(f"读取到上次检查时间: {last_time}") return last_time except: logger.warning("未找到上次检查时间文件") return '' def save_check_time(self, time: str) -> None: try: with open('last_check.txt', 'w') as f: f.write(time) logger.debug(f"保存本次检查时间: {time}") except Exception as e: logger.error(f"保存检查时间时出错: {str(e)}") def create_email_content(self, entries: List[Dict]) -> str: """创建美化的HTML邮件内容""" html_content = """ <style> .header { text-align: center; margin-bottom: 40px; padding: 20px; background-color: #f8f9fa; border-radius: 8px; } .main-title { font-size: 28px; color: #2c3e50; margin-bottom: 10px; } .subtitle { font-size: 20px; color: #34495e; margin-bottom: 15px; } .blog-name { font-size: 24px; color: #16a085; margin-bottom: 10px; } .blog-description { font-size: 16px; color: #7f8c8d; margin-bottom: 20px; } .article-container { font-family: Arial, sans-serif; max-width: 800px; margin: 0 auto; padding: 20px; } .article { margin-bottom: 30px; border-bottom: 1px solid #eee; padding-bottom: 20px; } .article-title { color: #333; font-size: 24px; margin-bottom: 10px; } .article-summary { color: #666; line-height: 1.6; margin-bottom: 15px; } .read-more { display: inline-block; padding: 8px 15px; background-color: #4CAF50; color: white; text-decoration: none; border-radius: 4px; } .read-more:hover { background-color: #45a049; } </style> <div class="article-container"> <div class="header"> <h1 class="main-title">烹茶室（Oskyla 晴空阁） 更新了！</h1> <h2 class="subtitle">欢迎访问 Frytea's Blog</h2> <h3 class="blog-name">Oskyla 烹茶室</h3> <p class="blog-description">价值信息藏书阁，统一门户入口。</p> </div> """ for entry in entries: # 清理文章标题和摘要中的HTML内容 clean_title = self.clean_html_content(entry.title) clean_summary = self.clean_html_content(entry.summary) html_content += f""" <div class="article"> <h2 class="article-title">{clean_title}</h2> <div class="article-summary">{clean_summary}</div> <a href="{entry.link}" class="read-more">阅读全文</a> </div> """ html_content += "</div>" return html_content def publish_campaign(self, campaign_id: int) -> bool: for attempt in range(self.max_retries): try: publish_url = f"{self.listmonk_url}/{campaign_id}/status" response = requests.put( publish_url, headers=self.headers, json={"status": "running"} ) if response.status_code == 200: logger.info(f"活动 {campaign_id} 发布成功") return True logger.warning(f"发布尝试 {attempt + 1} 失败: HTTP {response.status_code}") if attempt < self.max_retries - 1: sleep(self.retry_delay) except requests.exceptions.RequestException as e: logger.error(f"发布API请求异常: {str(e)}") if attempt < self.max_retries - 1: sleep(self.retry_delay) return False def send_newsletter(self, new_entries: List[Dict]) -> bool: try: content = self.create_email_content(new_entries) # 获取文章数量 article_count = len(new_entries) # 清理标题中的HTML内容 #titles = ", ".join(self.clean_html_content(entry.title) for entry in new_entries) data = { "name": "Frytea's Blog 更新通知", "subject": f"Frytea's Blog 更新了 {article_count} 篇新文章", "lists": LISTMONK_SEND_LIST_IDS, "content_type": "html", "body": content, "type": "regular" } logger.debug("准备发送的数据: %s", json.dumps(data, indent=2)) response = requests.post(self.listmonk_url, headers=self.headers, json=data) if response.status_code == 200: campaign_id = response.json().get('data', {}).get('id') if campaign_id: return self.publish_campaign(campaign_id) logger.error(f"创建活动失败: HTTP {response.status_code}") return False except Exception as e: logger.error(f"发送邮件时出错: {str(e)}") return False def check_and_send(self) -> None: try: logger.info(f"开始解析RSS源: {self.rss_url}") feed = feedparser.parse(self.rss_url) if feed.bozo: logger.error(f"RSS解析错误: {feed.bozo_exception}") return if not feed.entries: logger.warning("RSS源没有任何条目") return last_check = self.get_last_check_time() new_entries = [] #for entry in feed.entries: # if not last_check or entry.published > last_check: # new_entries.append(entry) for entry in feed.entries: # 将字符串解析为 datetime 对象 entry_time = parser.parse(entry.published) last_check_time = parser.parse(last_check) if last_check else None if not last_check_time or entry_time > last_check_time: new_entries.append(entry) if new_entries: logger.info(f"检测到 {len(new_entries)} 篇新文章") if self.send_newsletter(new_entries): self.save_check_time(max(entry.published for entry in new_entries)) else: logger.info("没有新文章") except Exception as e: logger.error(f"执行过程中出现未预期的错误: {str(e)}", exc_info=True) if __name__ == "__main__": checker = RSSChecker() checker.check_and_send() 依赖 requirements.txt ...

Docker 部署方法 参考：https://github.com/mautic/docker-mautic/tree/mautic5/examples 增加插件 使用如下 Dockerfile FROM mautic/mautic:5-apache COPY ./plugins/ /var/www/html/docroot/plugins/ 结合以下 Makefile all: docker build -t mautic/mautic:5-apache-my . 整个目录架构是这样： root@tencent-gz1:/data/docker/mautic/add-something# tree -L 2 . . ├── Dockerfile ├── Makefile ├── plugins │ └── MauticRssToEmailBundle └── translations └── zh_CN.zip 执行 增加语言包 在 Dockerfile 增加一个目录： ...

rbd du -p ssd | awk ' NR>1 { size=$4 unit=$5 if (unit=="MiB") size=size else if (unit=="GiB") size=size*1024 else if (unit=="TiB") size=size*1024*1024 print size " " unit " " $0 }' | sort -nr | head -n 50 | cut -d" " -f3- by claude 3.5 效果： root@pve1:~# rbd du -p ssd | awk ' NR>1 { size=$4 unit=$5 if (unit=="MiB") size=size else if (unit=="GiB") size=size*1024 else if (unit=="TiB") size=size*1024*1024 print size " " unit " " $0 }' | sort -nr | head -n 50 | cut -d" " -f3- <TOTAL> 111 TiB 47 TiB vm-1033-disk-0 20 TiB 20 TiB vm-1054-disk-1@bak20240731 1000 GiB 677 GiB vm-503-disk-3 500 GiB 500 GiB vm-502-disk-3 500 GiB 500 GiB vm-501-disk-3 500 GiB 500 GiB vm-1054-disk-1@bak20231106 1000 GiB 475 GiB vm-497-disk-0@backup0530 500 GiB 340 GiB vm-1279-disk-1 320 GiB 320 GiB vm-1090-disk-0 300 GiB 300 GiB

本章介绍 Bash 脚本的条件判断语法。 if 结构 if是最常用的条件判断结构，只有符合给定条件时，才会执行指定的命令。它的语法如下。 if commands; then commands [elif commands; then commands...] [else commands] fi 这个命令分成三个部分：if、elif和else。其中，后两个部分是可选的。 ...

linuxserver 版 docker --- services: smokeping: image: lscr.io/linuxserver/smokeping:latest container_name: smokeping environment: - PUID=1000 - PGID=1000 - TZ=Asia/Shanghai - MASTER_URL=http://<master-host-ip>:80/smokeping/ #optional - SHARED_SECRET=password #optional - CACHE_DIR=/tmp #optional volumes: - ./data/config:/config - ./data/data:/data ports: - 80:80 restart: unless-stopped 其中的 data/config/Targets 为监控目标配置。 配置说明 General 自行配置。 *** General *** owner = frytea.com contact = songtianlun@frytea.com mailhost = my.mail.host # NOTE: do not put the Image Cache below cgi-bin # since all files under cgi-bin will be executed ... this is not # good for images. cgiurl = http://localhost/smokeping/smokeping.cgi # specify this to get syslog logging syslogfacility = local0 # each probe is now run in its own process # disable this to revert to the old behaviour # concurrentprobes = no display_name = Frytea's SmokePing @include /config/pathnames Probes 配置侦测频率。 ...

常用网络测试工具的对比矩阵: 工具网站 PING TCPing MTR DNS HTTP(S) 端口扫描 路由追踪 IP 查询 多节点 SSL证书 IPv6 API 特色功能 ping.pe ✓ ✓ ✓ ✓ - ✓ ✓ - ✓ ✓ 全球节点,命令行风格 ping.chinaz.com ✓ - ✓ ✓ - ✓ ✓ ✓ - - 国内节点覆盖全,历史记录 itdog.cn ✓ - ✓ ✓ ✓ ✓ ✓ ✓ - - 界面友好,检测报告详细 17ce.com ✓ - ✓ ✓ - ✓ ✓ ✓ - ✓ 性能监控,故障分析 ipip.net ✓ ✓ ✓ - - ✓ ✓ ✓ - ✓ ✓ IP地理位置,AS信息 tool.lu ✓ - ✓ ✓ ✓ ✓ - ✓ - - 工具集合,便捷使用 boce.com ✓ - ✓ ✓ - ✓ ✓ ✓ - ✓ 网站监控,性能分析 ping.eu ✓ - ✓ ✓ ✓ ✓ ✓ - ✓ - 欧洲节点为主 zhale.me ✓ ✓ ✓ ✓ ✓ ✓ ✓ 全球1000+网络拨测节点，模拟用户访问域名/IP，实用小工具，运维必备 ping.sx ✓ ✓ ✓ ✓ ✓ tance.cc ✓ ✓ ✓ ✓ ✓ cesu.net ✓ ✓ ✓ ✓ ✓ 功能说明: ...

Docker Engine 可以在 Rocky Linux 服务器上运行原生 Docker 风格的容器工作负载。在运行完整的 Docker Desktop 环境时，有时会首选这种方式。 添加 Docker 仓库 使用 dnf 工具将 Docker 仓库添加到你的 Rocky Linux 服务器。输入： ...

汇总表格 名称 IPv4 IPv6 DNSPod 119.29.29.29 2402:4e00:: AliDNS 223.5.5.5 223.6.6.6 2400:3200::1 2400:3200:baba::1 BaiduDNS 180.76.76.76 2400:da00::6666 360DNS 101.226.4.6 218.30.118.6 123.125.81.6 140.207.198.6 114DNS 114.114.114.114 114.114.115.115 ChinaDNS 1.2.4.8 210.2.4.8 2001:dc7:1000::1 OneDNS 117.50.11.11 117.50.22.22 Hi!XNS DNS 40.73.101.101 TWNIC DNS 101.101.101.101 101.102.103.104 2001:de4::101 2001:de4::102 国内知名公共 DNS 服务器 腾讯 DNS (DNSPod) 由 DNSPod 提供的公共免费 DNS，后来 DNSPod 被腾讯(Tencent)收购，现在属于腾讯公司所有，稳定性和连通性也是不错的，经测试海外也可以使用。 DNSPod 除了 IPv4，现在同时支持 IPv6 DNS 和 DoT/DoH 服务。 ...