diff --git a/.github/Toos/色播聚合/ss.py b/.github/Toos/色播聚合/ss.py new file mode 100644 index 000000000..fd44c83eb --- /dev/null +++ b/.github/Toos/色播聚合/ss.py @@ -0,0 +1,173 @@ +# coding=utf-8 +import requests +import time +import os +import logging +from logging.handlers import TimedRotatingFileHandler + +# ===================== 配置区 ===================== +BASE_URL = "http://api.hclyz.com:81/mf" +M3U_FILE = "色播聚合.m3u" +LOG_FILE = "scraper.log" + +# 屏蔽词配置:包含以下关键词的标题将被过滤 +BLACK_LIST = ["支付宝风控解除", "依依实力带飞"] + +# Telegram 配置 +TELEGRAM_BOT_TOKEN = "YOUR_BOT_TOKEN" +TELEGRAM_CHAT_ID = "YOUR_CHAT_ID" + +HEADERS = {"User-Agent": "Mozilla/5.0"} +VALID_PREFIX = ("http://", "https://", "rtmp://") +# ================================================== + +# --- 日志配置 --- +def setup_logging(): + logger = logging.getLogger("ScraperLogger") + logger.setLevel(logging.INFO) + + # 格式化器:包含 [时间] [级别] 内容 + formatter = logging.Formatter('[%(asctime)s] %(levelname)s: %(message)s', datefmt='%Y-%m-%d %H:%M:%S') + + # 控制台输出 + console_handler = logging.StreamHandler() + console_handler.setFormatter(formatter) + logger.addHandler(console_handler) + + # 文件输出:每 7 天滚动一次,保留 1 个备份 + file_handler = TimedRotatingFileHandler( + LOG_FILE, when="D", interval=7, backupCount=1, encoding="utf-8" + ) + file_handler.setFormatter(formatter) + logger.addHandler(file_handler) + + return logger + +log = setup_logging() + +def safe_get_json(url): + """安全获取 JSON,失败返回 None""" + try: + r = requests.get(url, headers=HEADERS, timeout=10) + if r.status_code != 200: + return None + return r.json() + except Exception as e: + log.error(f"网络请求异常: {url} -> {e}") + return None + +def is_valid_stream(url): + """合法流地址判断""" + url = url.lower() + return url.startswith(VALID_PREFIX) and (".m3u8" in url or ".flv" in url or ".mp4" in url or url.startswith("rtmp://")) + +def send_to_telegram_message(bot_token, chat_id, message): + """发送文本消息到 Telegram""" + url = f"https://api.telegram.org/bot{bot_token}/sendMessage" + data = {'chat_id': chat_id, 'text': message, 'parse_mode': 'Markdown'} + try: + requests.post(url, data=data, timeout=15) + except Exception as e: + log.error(f"Telegram 消息发送失败: {e}") + +def send_to_telegram_file(file_path, bot_token, chat_id): + """发送文件到 Telegram""" + url = f"https://api.telegram.org/bot{bot_token}/sendDocument" + try: + with open(file_path, 'rb') as f: + files = {'document': f} + data = {'chat_id': chat_id} + r = requests.post(url, files=files, data=data, timeout=30) + if r.status_code == 200: + log.info(f"文件已发送到 Telegram(Chat ID: {chat_id})") + else: + log.error(f"Telegram 上传失败,状态码:{r.status_code}") + except Exception as e: + log.error(f"Telegram 上传异常:{e}") + +def main(): + total_error = 0 + total_success = 0 + total_filtered = 0 # 统计过滤数量 + + if TELEGRAM_BOT_TOKEN and TELEGRAM_CHAT_ID: + send_to_telegram_message(TELEGRAM_BOT_TOKEN, TELEGRAM_CHAT_ID, "🚀 开始采集直播源...") + + log.info("🚀 任务启动:开始抓取色播聚合数据") + + home = safe_get_json(f"{BASE_URL}/json.txt") + if not home: + log.error("首页数据获取失败,采集终止") + return + + data = home.get("pingtai", [])[1:] + data = sorted(data, key=lambda x: int(x.get("Number", 0) or 0), reverse=True) + + m3u_lines = ["#EXTM3U"] + seen_urls = set() + + for item in data: + room_title = item.get("title", "").strip() + number = item.get("Number", "") + address = item.get("address", "") + + log.info(f"📺 正在处理:{room_title}({number})") + + detail = safe_get_json(f"{BASE_URL}/{address}") + if not detail: + total_error += 1 + continue + + zhubo = detail.get("zhubo", []) + if not zhubo: + total_error += 1 + continue + + group_name = f"-{room_title}" + + for vod in zhubo: + name = vod.get("title", "").strip() + url = vod.get("address", "").strip() + + # 1. 过滤屏蔽词 + if any(keyword in name for keyword in BLACK_LIST): + log.info(f"🚫 已过滤屏蔽词频道: {name}") + total_filtered += 1 + continue + + # 2. 检查流有效性 + if not url or not is_valid_stream(url): + total_error += 1 + continue + + # 3. 去重处理 + if url in seen_urls: + continue + + seen_urls.add(url) + m3u_lines.append(f'#EXTINF:-1 group-title="{group_name}",{name}') + m3u_lines.append(url) + total_success += 1 + + time.sleep(0.3) # 防限频 + + # 保存 m3u + try: + with open(M3U_FILE, "w", encoding="utf-8") as f: + f.write("\n".join(m3u_lines)) + log.info(f"📄 播放列表已生成: {M3U_FILE}") + except Exception as e: + log.error(f"写入文件失败: {e}") + + log.info(f"✅ 完成!有效:{total_success},过滤:{total_filtered},异常:{total_error}") + + if TELEGRAM_BOT_TOKEN and TELEGRAM_CHAT_ID: + msg = (f"✅ 采集完成\n" + f"有效流:{total_success}\n" + f"已屏蔽:{total_filtered}\n" + f"异常数:{total_error}") + send_to_telegram_message(TELEGRAM_BOT_TOKEN, TELEGRAM_CHAT_ID, msg) + send_to_telegram_file(M3U_FILE, TELEGRAM_BOT_TOKEN, TELEGRAM_CHAT_ID) + +if __name__ == "__main__": + main() \ No newline at end of file