XDSec Push Bot

listen.py

import requests
import feedparser
from datetime import datetime, timedelta
from zoneinfo import ZoneInfo

from flask import Flask, request

app = Flask(__name__)

@app.route("/", methods=["POST"])
def receive_event():
    event = request.json

    now = datetime.now()

    if "raw_message" in event and event["raw_message"] == "ctflist":
        push_text = ""

        feed_url = 'https://ctftime.org/event/list/upcoming/rss/'
        feed = feedparser.parse(feed_url)

        for entry in feed.entries:
            isPush = (entry.onsite == "False" and float(entry.weight) > 0.00 and (datetime.fromisoformat(str(entry.start_date)) - now <= timedelta(days=30)))

            if isPush:
                push_text += f"{entry.title}\n{entry.url}\n开始时间：{datetime.fromisoformat(str(entry.start_date)).replace(tzinfo=timezone.utc).astimezone(ZoneInfo("Asia/Shanghai")).strftime('%Y-%m-%d %H:%M:%S')}\n结束时间：{datetime.fromisoformat(str(entry.finish_date)).replace(tzinfo=timezone.utc).astimezone(ZoneInfo("Asia/Shanghai")).strftime('%Y-%m-%d %H:%M:%S')}\n\n"
        
        token = "xxx"
        url = "http://xxx/send_group_msg"

        headers = {'User-Agent': 'Mozilla/5.0', 'Authorization': f"Bearer {token}"}

        data = {"group_id": event["group_id"] if "group_id" in event else event["target_id"], "message": f"一个人给各位收集到了最近的CTF比赛信息（30天内，权重大于0）。\n{push_text}"}

        x = requests.post(url, headers=headers, data=data)

        print(x.text)
    return "OK"

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8080)

push.py

from freshrss_api import FreshRSSAPI
import requests
from openai import OpenAI
import trafilatura

# Initialize the client with direct parameters
client = FreshRSSAPI(
    host="xxx",
    username="xxx",
    password="xxx",
    verbose=False
)

def aibrief(content, rsscontent):
    client = OpenAI(
        api_key="sk-xxx",
        base_url="https://api.deepseek.com")

    response = client.chat.completions.create(
        model="deepseek-v4-flash",
        messages=[
            {"role": "system", "content": "你是一个专业的秘书，负责总结文章的内容，供网络安全协会的推送使用。请你根据给定的文章内容，生成一段不长于75字的摘要，概括文章的主要内容、思路、技术方法，供网络安全协会的成员快速判断是否对文章感兴趣。"},
            {"role": "user", "content": "trafilatura得到的文章内容，可能会因为反爬而为None或无意义字符" + str(content) + "\n 以下是订阅软件从 rss 中获取到的内容" + rsscontent}
        ],
        stream=False,
        reasoning_effort="high",
        extra_body={"thinking": {"type": "enabled"}}
    )

    print(response.choices[0].message.content)
    return response.choices[0].message.content

def getAIrecom(passage_list) -> list:
    client = OpenAI(
        api_key="sk-xxx",
        base_url="https://api.deepseek.com")
    
    toEvaluateContent = ""
    index = 0
    for i in passage_list:
        toEvaluateContent += f"第{index}篇文章：\n标题：{i[1]}\nRSS摘要：{i[3]}\n网页摘要：{i[4]}\n\n"
        index += 1

    response = client.chat.completions.create(
        model="deepseek-v4-flash",
        messages=[
            {"role": "system", "content": "你是一个专业的秘书，负责筛选有价值的文章，供网络安全协会的推送使用。请你根据给定的文章内容，回答出其中最有价值的五篇文章的序号，序号之间用空格分隔，不要有多余内容。文章的价值从重要性和影响力来评估。因为反爬的原因，有一些文章的网页内容可能为None或无意义内容，请忽视这一点，根据RSS摘要来做判断。"},
            {"role": "user", "content": toEvaluateContent}
        ],
        stream=False,
        reasoning_effort="xhigh",
        extra_body={"thinking": {"type": "enabled"}}
    )

    print(response.choices[0].message.content)
    return str(response.choices[0].message.content).split()

unread_items = client.get_unreads()

passages = []
pass_text = ""


for i in unread_items:
    passages.append([i.author, i.title, i.url, i.html, str(trafilatura.extract(trafilatura.fetch_url(i.url), output_format='markdown', include_tables=True))])
    client.set_mark(as_="read", id=i.id)

if len(passages) > 5:
    pass_text += "本次抓取文章数大于5篇，根据AI推荐，推送五篇较有价值的文章。\n"
    push_index = getAIrecom(passage_list=passages)
    for i in range(5):
        pass_text += f"Title: {passages[int(push_index[i])][1]} \nURL: {passages[int(push_index[i])][2]} \nBrief: {aibrief(passages[int(push_index[i])][4], passages[int(push_index[i])][3])}\n\n"
elif len(passages) == 0:
    exit(0)
else:
    for i in passages:
        pass_text += f"Title: {i[1]} \nURL: {i[2]} \nBrief: {aibrief(i[4], i[3])}\n\n"

print("正在执行")

token = "xxx"
url = "http://xxx/send_group_msg"

headers = {'User-Agent': 'Mozilla/5.0', 'Authorization': f"Bearer {token}"}

data = {"group_id": xxx, "message": f"最近几小时爬取到了{len(passages)}篇文章，信息如下：\n{pass_text}\n各位成员可以在 xxx 查看所有文章。"}

x = requests.post(url, headers=headers, data=data)

print(x.text)