MonetTag专用刷量机器人版本

-  新增MonetTag专用刷量机器人 (monetag_bot.py)
-  添加MonetTag专用配置文件 (monetag_config.json)
-  添加一键启动脚本 (run_monetag_bot.bat)
-  添加详细使用说明 (MonetTag刷量机器人使用说明.md)
-  删除通用广告机器人文件(改为专用版本)

核心特性:
-  Zone ID: 157708
-  点击率控制: 10%
-  美国IP代理支持
-  手机端优先 (90%)
-  支持5种广告类型
-  展示量+点击量双重刷量
This commit is contained in:
huangzhenpc
2025-07-18 14:43:23 +08:00
parent 39ce9404db
commit cba3421b2c
7 changed files with 1067 additions and 1590 deletions

649
monetag_bot.py Normal file
View File

@@ -0,0 +1,649 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
MonetTag 专用刷量机器人
Zone: 157708
点击率控制: 10%
地区: 美国
设备: 主要手机端
"""
import requests
import time
import random
import json
import logging
from urllib.parse import urlparse, urljoin, parse_qs
import re
import base64
import hashlib
import uuid
# 配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('monetag_bot.log', encoding='utf-8'),
logging.StreamHandler()
]
)
logger = logging.getLogger(__name__)
class MonetTagBot:
def __init__(self, config_file='monetag_config.json'):
"""
初始化MonetTag刷量机器人
"""
self.config = self.load_config(config_file)
self.session = None
# 🎯 MonetTag 核心配置
self.zone_id = "157708"
self.base_domain = "fpyf8.com"
self.tag_url = f"https://{self.base_domain}/88/tag.min.js"
self.sw_domain = "aiharsoreersu.net"
# 🎯 广告类型配置
self.ad_types = {
"push_notifications": {
"name": "Push Notifications",
"zone": "9583728",
"weight": 0.2,
"click_rate": 0.12
},
"vignette_banner": {
"name": "Vignette Banner",
"zone": "9583727",
"weight": 0.25,
"click_rate": 0.08
},
"native_banner": {
"name": "Native Banner (Interstitial)",
"zone": "9583726",
"weight": 0.2,
"click_rate": 0.15
},
"in_page_push": {
"name": "In-Page Push",
"zone": "9583725",
"weight": 0.2,
"click_rate": 0.10
},
"onclick_popunder": {
"name": "OnClick (Popunder)",
"zone": "9583724",
"weight": 0.15,
"click_rate": 0.05
}
}
# 🇺🇸 美国手机端User-Agent
self.us_mobile_agents = [
"Mozilla/5.0 (iPhone; CPU iPhone OS 17_1_2 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.1 Mobile/15E148 Safari/604.1",
"Mozilla/5.0 (iPhone; CPU iPhone OS 16_6_1 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/16.6 Mobile/15E148 Safari/604.1",
"Mozilla/5.0 (iPhone; CPU iPhone OS 17_0_3 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.0.3 Mobile/15E148 Safari/604.1",
"Mozilla/5.0 (Linux; Android 14; SM-G991U) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Mobile Safari/537.36",
"Mozilla/5.0 (Linux; Android 13; SM-A515U) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/119.0.0.0 Mobile Safari/537.36",
"Mozilla/5.0 (Linux; Android 14; Pixel 7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Mobile Safari/537.36",
"Mozilla/5.0 (Linux; Android 13; Pixel 6) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/119.0.0.0 Mobile Safari/537.36",
"Mozilla/5.0 (iPad; CPU OS 17_1_2 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.1 Mobile/15E148 Safari/604.1",
]
# 📊 统计数据
self.stats = {
"total_visits": 0,
"impressions": 0,
"clicks": 0,
"click_rate": 0,
"ad_types_stats": {},
"errors": 0
}
# 🔧 请求配置
self.request_config = {
"timeout": 10,
"max_retries": 3,
"retry_delay": 2
}
def load_config(self, config_file):
"""加载配置文件"""
try:
with open(config_file, 'r', encoding='utf-8') as f:
config = json.load(f)
logger.info(f"✅ 配置文件加载成功: {config_file}")
return config
except FileNotFoundError:
logger.warning(f"⚠️ 配置文件未找到: {config_file},使用默认配置")
return self.get_default_config()
except json.JSONDecodeError as e:
logger.error(f"❌ 配置文件格式错误: {e}")
return self.get_default_config()
def get_default_config(self):
"""获取默认配置"""
return {
"target_website": "https://your-website.com",
"proxy": {
"enabled": True,
"host": "us-proxy.com",
"port": "8080",
"username": "your-username",
"password": "your-password"
},
"settings": {
"click_rate": 0.10,
"mobile_rate": 0.90,
"us_traffic_rate": 1.0,
"visits_per_hour": 100,
"min_delay": 30,
"max_delay": 90
}
}
def setup_session(self):
"""设置请求会话"""
self.session = requests.Session()
self.session.cookies.clear()
# 🇺🇸 设置美国代理
if self.config.get('proxy', {}).get('enabled', False):
proxy_config = self.config['proxy']
proxy_url = f"http://{proxy_config['username']}:{proxy_config['password']}@{proxy_config['host']}:{proxy_config['port']}"
self.session.proxies = {
'http': proxy_url,
'https': proxy_url
}
logger.info(f"🇺🇸 美国代理配置: {proxy_config['host']}:{proxy_config['port']}")
# 📱 设置手机端User-Agent (90%概率)
if random.random() < self.config['settings']['mobile_rate']:
user_agent = random.choice(self.us_mobile_agents)
is_mobile = True
else:
# 10%概率使用美国桌面端
desktop_agents = [
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.1 Safari/605.1.15"
]
user_agent = random.choice(desktop_agents)
is_mobile = False
# 🇺🇸 美国特征请求头
headers = {
"User-Agent": user_agent,
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8",
"Accept-Language": "en-US,en;q=0.9",
"Accept-Encoding": "gzip, deflate, br",
"Connection": "keep-alive",
"Upgrade-Insecure-Requests": "1",
"Sec-Fetch-Dest": "document",
"Sec-Fetch-Mode": "navigate",
"Sec-Fetch-Site": "none",
"Sec-Fetch-User": "?1",
"Cache-Control": "max-age=0",
"DNT": "0",
"Sec-GPC": "1" if random.random() < 0.3 else None
}
if is_mobile:
headers.update({
"Sec-CH-UA-Mobile": "?1",
"Viewport-Width": str(random.choice([375, 414, 390, 393, 412])),
"Sec-CH-UA-Platform": '"iOS"' if 'iPhone' in user_agent else '"Android"'
})
# 移除None值
headers = {k: v for k, v in headers.items() if v is not None}
self.session.headers.update(headers)
logger.info(f"🎭 会话配置完成:")
logger.info(f" 📱 移动端: {is_mobile}")
logger.info(f" 🇺🇸 美国IP: 已配置")
logger.info(f" 🎯 MonetTag Zone: {self.zone_id}")
logger.info(f" 📊 目标点击率: {self.config['settings']['click_rate']*100}%")
def generate_device_fingerprint(self):
"""生成设备指纹"""
device_id = str(uuid.uuid4())
session_id = hashlib.md5(str(time.time()).encode()).hexdigest()
fingerprint = {
"device_id": device_id,
"session_id": session_id,
"screen_width": random.choice([375, 414, 390, 393, 412, 360, 1920, 1440]),
"screen_height": random.choice([667, 896, 844, 851, 915, 640, 1080, 900]),
"timezone": "America/New_York",
"language": "en-US",
"platform": "iPhone" if "iPhone" in self.session.headers.get("User-Agent", "") else "Android"
}
return fingerprint
def load_tag_script(self):
"""加载MonetTag标签脚本"""
try:
logger.info(f"📥 加载MonetTag标签脚本...")
# 构造tag.min.js请求
tag_params = {
"data-zone": self.zone_id,
"async": "true",
"data-cfasync": "false"
}
response = self.session.get(
self.tag_url,
params=tag_params,
timeout=self.request_config["timeout"]
)
if response.status_code == 200:
logger.info(f"✅ 标签脚本加载成功 ({len(response.content)} 字节)")
# 解析脚本中的配置
self._parse_tag_script(response.text)
return True
else:
logger.error(f"❌ 标签脚本加载失败: {response.status_code}")
return False
except Exception as e:
logger.error(f"❌ 加载标签脚本异常: {e}")
return False
def _parse_tag_script(self, script_content):
"""解析标签脚本内容"""
# 提取可能的端点和配置
endpoints = re.findall(r'https?://[^\s"\']+', script_content)
zones = re.findall(r'zone["\']?\s*[:=]\s*["\']?(\d+)', script_content)
logger.info(f"🔍 脚本分析结果:")
logger.info(f" 发现端点: {len(endpoints)}")
logger.info(f" 发现区域: {zones}")
def simulate_page_load(self):
"""模拟页面加载"""
target_url = self.config["target_website"]
try:
logger.info(f"🌐 模拟页面加载: {target_url}")
# 页面加载前延迟
pre_load_delay = random.uniform(1, 3)
time.sleep(pre_load_delay)
# 设置referrer
referrers = [
"https://www.google.com/search?q=free+games",
"https://www.google.com/search?q=online+games",
"https://www.bing.com/search?q=mobile+games",
"https://duckduckgo.com/?q=html5+games",
"https://www.reddit.com/r/WebGames/",
"direct"
]
referrer = random.choice(referrers)
if referrer != "direct":
self.session.headers.update({"Referer": referrer})
# 发起页面请求
response = self.session.get(target_url, timeout=15)
if response.status_code == 200:
logger.info(f"✅ 页面加载成功 ({response.status_code})")
# 模拟页面解析时间
parse_time = random.uniform(0.5, 1.5)
time.sleep(parse_time)
return True
else:
logger.error(f"❌ 页面加载失败: {response.status_code}")
return False
except Exception as e:
logger.error(f"❌ 页面加载异常: {e}")
return False
def generate_ad_requests(self):
"""生成广告请求"""
fingerprint = self.generate_device_fingerprint()
# 为每种广告类型生成请求
ad_requests = []
for ad_type, config in self.ad_types.items():
if random.random() < config["weight"]:
request_data = {
"zone": config["zone"],
"type": ad_type,
"device_id": fingerprint["device_id"],
"session_id": fingerprint["session_id"],
"screen_width": fingerprint["screen_width"],
"screen_height": fingerprint["screen_height"],
"timezone": fingerprint["timezone"],
"language": fingerprint["language"],
"platform": fingerprint["platform"],
"timestamp": int(time.time()),
"referrer": self.session.headers.get("Referer", "direct"),
"user_agent": self.session.headers.get("User-Agent", ""),
"ip_country": "US"
}
ad_requests.append(request_data)
return ad_requests
def send_impression_requests(self, ad_requests):
"""发送展示请求"""
impression_count = 0
for request_data in ad_requests:
try:
# 构造展示请求
impression_url = f"https://{self.sw_domain}/impression"
# 添加MonetTag特有的参数
params = {
"zone": request_data["zone"],
"type": request_data["type"],
"device": request_data["device_id"],
"session": request_data["session_id"],
"sw": request_data["screen_width"],
"sh": request_data["screen_height"],
"tz": request_data["timezone"],
"lang": request_data["language"],
"platform": request_data["platform"],
"ts": request_data["timestamp"],
"ref": base64.b64encode(request_data["referrer"].encode()).decode(),
"ua": base64.b64encode(request_data["user_agent"].encode()).decode(),
"country": "US",
"r": random.random()
}
# 发送请求
response = self.session.get(
impression_url,
params=params,
timeout=self.request_config["timeout"]
)
if response.status_code in [200, 204]:
impression_count += 1
logger.info(f"📊 展示请求成功: {request_data['type']} (Zone: {request_data['zone']})")
else:
logger.warning(f"⚠️ 展示请求失败: {response.status_code}")
# 请求间隔
time.sleep(random.uniform(0.1, 0.5))
except Exception as e:
logger.error(f"❌ 展示请求异常: {e}")
self.stats["errors"] += 1
continue
self.stats["impressions"] += impression_count
logger.info(f"📊 本次展示量: {impression_count}")
return impression_count
def send_click_requests(self, ad_requests):
"""发送点击请求"""
click_count = 0
target_click_rate = self.config["settings"]["click_rate"]
for request_data in ad_requests:
# 根据广告类型和设置的点击率决定是否点击
ad_config = self.ad_types.get(request_data["type"], {})
base_click_rate = ad_config.get("click_rate", 0.1)
# 综合考虑全局点击率和广告类型点击率
final_click_rate = (target_click_rate + base_click_rate) / 2
if random.random() < final_click_rate:
try:
# 模拟用户思考时间
think_time = random.uniform(1, 4)
time.sleep(think_time)
# 构造点击请求
click_url = f"https://{self.sw_domain}/click"
# 点击请求参数
click_params = {
"zone": request_data["zone"],
"type": request_data["type"],
"device": request_data["device_id"],
"session": request_data["session_id"],
"click_id": hashlib.md5(str(time.time()).encode()).hexdigest(),
"timestamp": int(time.time()),
"x": random.randint(10, 200),
"y": random.randint(10, 300),
"country": "US",
"r": random.random()
}
# 发送点击请求
response = self.session.post(
click_url,
data=click_params,
timeout=self.request_config["timeout"]
)
if response.status_code in [200, 201, 204]:
click_count += 1
logger.info(f"🖱️ 点击请求成功: {request_data['type']} (Zone: {request_data['zone']})")
# 模拟点击后行为
self._simulate_post_click_behavior(request_data)
else:
logger.warning(f"⚠️ 点击请求失败: {response.status_code}")
except Exception as e:
logger.error(f"❌ 点击请求异常: {e}")
self.stats["errors"] += 1
continue
self.stats["clicks"] += click_count
logger.info(f"🖱️ 本次点击量: {click_count}")
return click_count
def _simulate_post_click_behavior(self, request_data):
"""模拟点击后行为"""
# 模拟广告页面停留时间
stay_time = random.uniform(2, 8)
# 根据广告类型调整行为
if request_data["type"] == "onclick_popunder":
# 弹窗广告,用户可能快速关闭
if random.random() < 0.7:
stay_time = random.uniform(0.5, 2)
logger.info(f" 🚫 快速关闭弹窗: {stay_time:.1f}")
elif request_data["type"] == "native_banner":
# 原生广告,用户可能更感兴趣
if random.random() < 0.4:
stay_time = random.uniform(5, 15)
logger.info(f" 📖 阅读原生广告: {stay_time:.1f}")
time.sleep(stay_time)
def run_single_session(self):
"""运行单次会话"""
logger.info("🚀 开始单次MonetTag刷量会话")
# 设置会话
self.setup_session()
try:
# 1. 加载标签脚本
if not self.load_tag_script():
logger.error("❌ 标签脚本加载失败")
return False
# 2. 模拟页面加载
if not self.simulate_page_load():
logger.error("❌ 页面加载失败")
return False
# 3. 生成广告请求
ad_requests = self.generate_ad_requests()
if not ad_requests:
logger.warning("⚠️ 未生成广告请求")
return False
logger.info(f"🎯 生成了 {len(ad_requests)} 个广告请求")
# 4. 发送展示请求
impression_count = self.send_impression_requests(ad_requests)
# 5. 发送点击请求
click_count = self.send_click_requests(ad_requests)
# 6. 更新统计
self.stats["total_visits"] += 1
if impression_count > 0:
self.stats["click_rate"] = self.stats["clicks"] / self.stats["impressions"]
logger.info(f"✅ 会话完成: 展示={impression_count}, 点击={click_count}")
return True
except Exception as e:
logger.error(f"❌ 会话执行异常: {e}")
self.stats["errors"] += 1
return False
finally:
if self.session:
self.session.close()
def run_continuous_sessions(self, total_sessions=None, delay_range=None):
"""运行连续会话"""
if total_sessions is None:
total_sessions = self.config['settings'].get('visits_per_hour', 100)
if delay_range is None:
delay_range = (
self.config['settings']['min_delay'],
self.config['settings']['max_delay']
)
success_count = 0
logger.info(f"🎯 开始连续MonetTag刷量目标: {total_sessions}")
logger.info(f"📊 目标点击率: {self.config['settings']['click_rate']*100}%")
logger.info(f"🇺🇸 主要地区: 美国")
logger.info(f"📱 主要设备: 手机端 ({self.config['settings']['mobile_rate']*100}%)")
for i in range(total_sessions):
logger.info(f"{'='*60}")
logger.info(f"🔄 第 {i+1}/{total_sessions} 次会话")
logger.info(f"{'='*60}")
session_start = time.time()
if self.run_single_session():
success_count += 1
session_time = time.time() - session_start
# 实时统计
current_click_rate = (self.stats["clicks"] / max(self.stats["impressions"], 1)) * 100
logger.info(f"✅ 第 {i+1} 次会话成功!耗时: {session_time:.1f}")
logger.info(f"📊 累计统计: 访问={self.stats['total_visits']}, 展示={self.stats['impressions']}, 点击={self.stats['clicks']}, 点击率={current_click_rate:.1f}%")
else:
logger.error(f"❌ 第 {i+1} 次会话失败!")
# 会话间隔
if i < total_sessions - 1:
delay = random.uniform(delay_range[0], delay_range[1])
logger.info(f"⏳ 等待 {delay:.1f}")
time.sleep(delay)
# 最终统计
self._print_final_stats(success_count, total_sessions)
return success_count
def _print_final_stats(self, success_count, total_sessions):
"""打印最终统计"""
success_rate = (success_count / total_sessions) * 100
click_rate = (self.stats["clicks"] / max(self.stats["impressions"], 1)) * 100
logger.info(f"🎉 MonetTag刷量任务完成")
logger.info(f"📊 最终统计:")
logger.info(f" 成功率: {success_count}/{total_sessions} ({success_rate:.1f}%)")
logger.info(f" 总访问: {self.stats['total_visits']}")
logger.info(f" 总展示: {self.stats['impressions']}")
logger.info(f" 总点击: {self.stats['clicks']}")
logger.info(f" 点击率: {click_rate:.1f}%")
logger.info(f" 错误数: {self.stats['errors']}")
logger.info(f" 🎯 Zone ID: {self.zone_id}")
logger.info(f" 🇺🇸 主要地区: 美国")
logger.info(f" 📱 主要设备: 手机端")
def main():
"""主函数"""
print("=" * 80)
print("🎯 MonetTag 专用刷量机器人")
print("=" * 80)
print("💡 核心特性:")
print(" 🎯 MonetTag Zone: 157708")
print(" 📊 点击率控制: 10%")
print(" 🇺🇸 主要地区: 美国")
print(" 📱 主要设备: 手机端 (90%)")
print(" 🔄 支持广告类型: Push, Banner, Native, Popunder")
print()
try:
bot = MonetTagBot()
print("请选择运行模式:")
print("1. 🧪 单次会话测试")
print("2. 🚀 连续会话 (使用配置参数)")
print("3. ⚙️ 自定义连续会话")
choice = input("请输入选择 (1/2/3): ").strip()
if choice == "1":
logger.info("🧪 开始单次会话测试")
success = bot.run_single_session()
if success:
print("🎉 单次会话测试成功!")
else:
print("😞 单次会话测试失败!")
elif choice == "2":
logger.info("🚀 开始连续会话")
success_count = bot.run_continuous_sessions()
print(f"🎉 连续会话完成!成功: {success_count}")
elif choice == "3":
try:
session_count = int(input("请输入会话次数: ").strip())
min_delay = int(input("请输入最小延迟秒数: ").strip())
max_delay = int(input("请输入最大延迟秒数: ").strip())
logger.info(f"🚀 开始自定义会话,次数: {session_count}")
success_count = bot.run_continuous_sessions(
total_sessions=session_count,
delay_range=(min_delay, max_delay)
)
print(f"🎉 自定义会话完成!成功: {success_count}/{session_count}")
except ValueError:
print("❌ 输入参数错误!")
else:
print("❌ 无效选择!")
except KeyboardInterrupt:
print("\n⚠️ 用户中断执行")
except Exception as e:
logger.error(f"程序执行出错: {e}")
print("❌ 程序执行出错,请查看日志文件 monetag_bot.log")
if __name__ == "__main__":
main()