Closed Sendu-sanxing closed 2 months ago
今天发现new插件里的https://news.ycombinator.com/被墙了,国内不能再访问了,导致插件报错,于是写了一个新的news插件:
# news.yml keywords: [ "新闻" ] plugin_name: "news" news_api_key: "8ee2b469d70fdbbd99ea110ca2452602" #申请地址:https://www.tianapi.com/console/ news_num: 50 # 新闻数量设置,最高50条 request_interval: 3600 # 请求间隔时间,单位为秒,默认1小时 # news.py import http.client import urllib import json import yaml from loguru import logger from utils.plugin_interface import PluginInterface import pywxdll from datetime import datetime, timedelta class news(PluginInterface): def __init__(self): # 读取插件配置 config_path = 'plugins/news.yml' with open(config_path, 'r', encoding='utf-8') as f: config = yaml.safe_load(f.read()) self.api_key = config['news_api_key'] self.news_num = config.get('news_num', 10) # 默认新闻数量为10 self.request_interval = config.get('request_interval', 3600) # 默认请求间隔为1小时 # 读取主设置中的机器人IP和端口 main_config_path = 'main_config.yml' with open(main_config_path, 'r', encoding='utf-8') as f: main_config = yaml.safe_load(f.read()) self.ip = main_config['ip'] self.port = main_config['port'] self.bot = pywxdll.Pywxdll(self.ip, self.port) self.last_request_times = {} # 用于记录每个群的上次请求时间 async def run(self, recv): wxid = recv['wxid'] current_time = datetime.now() last_request_time = self.last_request_times.get(wxid) if last_request_time and (current_time - last_request_time).total_seconds() < self.request_interval: wait_time = self.request_interval - (current_time - last_request_time).total_seconds() self.bot.send_txt_msg(wxid, f"️⚠️请等待 {int(wait_time // 60)} 分钟 {int(wait_time % 60)} 秒后再试。") return news_data = self.get_news() if news_data: messages = self.format_news(news_data) for message in messages: self.bot.send_txt_msg(wxid, message) self.last_request_times[wxid] = current_time else: self.bot.send_txt_msg(wxid, "⚠️无法获取新闻信息,请稍后再试。") def get_news(self): try: conn = http.client.HTTPSConnection('apis.tianapi.com') params = urllib.parse.urlencode({'key': self.api_key, 'num': self.news_num}) headers = {'Content-type': 'application/x-www-form-urlencoded'} conn.request('POST', '/world/index', params, headers) response = conn.getresponse() result = response.read() data = result.decode('utf-8') dict_data = json.loads(data) logger.info(f"API返回的数据: {dict_data}") # 添加日志记录返回的数据 return dict_data except Exception as e: logger.error(f"⚠️获取新闻时发生错误: {e}") return None def format_news(self, news_data): try: news_list = news_data.get('result', {}).get('newslist', []) if not news_list: return ["⚠️未获取到任何新闻信息。"] today = datetime.now().date() yesterday = today - timedelta(days=1) filtered_news = [news for news in news_list if datetime.strptime(news.get('ctime', ''), '%Y-%m-%d %H:%M').date() in [today, yesterday]] formatted_news = [] news_chunk = [] for idx, news in enumerate(filtered_news, 1): title = news.get('title', '无标题') description = news.get('description', '无描述') url = news.get('url', '') ctime = news.get('ctime', '无时间信息') source = news.get('source', '无发行单位') news_chunk.append(f"{idx}. 📰标题: {title}\nℹ时间: {ctime}\n📺发行单位: {source}\n🔗链接: {url}\n\n") if idx % 10 == 0 or idx == len(filtered_news): formatted_news.append("".join(news_chunk)) news_chunk = [] return formatted_news except Exception as e: logger.error(f"⚠️格式化新闻时发生错误: {e}") return ["⚠️新闻数据格式化错误。"]
pr一下吧
news_api_key复制时忘删掉了吧,赶紧删掉吧 。
news_api_key
哈哈,谢谢提醒,不过这个API差不多等于是免费的,我也又重置了一下
今天发现new插件里的https://news.ycombinator.com/被墙了,国内不能再访问了,导致插件报错,于是写了一个新的news插件: