抖音评论点赞协议工具，点赞评论置顶软件，批量点赞评论协议【python】

import time
import random
import requests
from typing import List, Dict
from dataclasses import dataclass
from urllib.parse import urlencode
from hashlib import md5
import json

@dataclass
class DouyinConfig:
    device_id: str = "1234567890ABCDEF"
    iid: str = "1234567890123456"
    openudid: str = "abcdef1234567890"
    uuid: str = "1234ABCD-5678-90EF-1234-567890ABCDEF"
    version_code: str = "180"
    app_name: str = "aweme"
    device_platform: str = "android"
    ssmix: str = "a"
    os_api: str = "28"
    dpi: str = "480"
    device_type: str = "Pixel 3"
    device_brand: str = "Google"
    language: str = "zh"
    resolution: str = "1080*1920"
    openudid: str = "abcdef1234567890"
    manifest_version_code: str = "180"
    update_version_code: str = "1802"
    app_version: str = "18.0.0"
    version_name: str = "18.0.0"
    timezone_name: str = "Asia/Shanghai"
    timezone_offset: str = "28800"
    os_version: str = "9"
    host_abi: str = "armeabi-v7a"
    cdid: str = "abcdef12-3456-7890-abcd-ef1234567890"
    ac: str = "wifi"
    mcc_mnc: str = "46001"
    channel: str = "xiaomi"
    aid: str = "1128"

class DouyinAPI:
    BASE_URL = "https://api.douyin.com"
    
    def __init__(self, config: DouyinConfig):
        self.config = config
        self.session = requests.Session()
        self.session.headers.update({
            "User-Agent": "com.ss.android.ugc.aweme/180 (Linux; U; Android 9; zh_CN; Pixel 3; Build/PQ3A.190801.002; Cronet/58.0.2991.0)",
            "Accept-Encoding": "gzip",
            "X-SS-REQ-TICKET": str(int(time.time() * 1000)),
            "X-Tt-Token": "",
            "sdk-version": "1",
            "X-Gorgon": "",
            "X-Khronos": str(int(time.time())),
        })
        
    def _generate_signature(self, params: Dict) -> Dict:
        """模拟签名生成"""
        params.update({
            "device_id": self.config.device_id,
            "iid": self.config.iid,
            "openudid": self.config.openudid,
            "uuid": self.config.uuid,
        })
        
        sorted_params = sorted(params.items(), key=lambda x: x[0])
        base_str = urlencode(sorted_params) + "38270b4f4f5b1070"
        sign = md5(base_str.encode()).hexdigest()
        params["mas"] = sign
        params["as"] = "a1" + sign[0:6]
        params["ts"] = str(int(time.time()))
        return params
    
    def get_video_comments(self, aweme_id: str, count: int = 20) -> List[Dict]:
        """获取视频评论"""
        params = {
            "aweme_id": aweme_id,
            "count": count,
            "cursor": 0,
            "address_book_access": 1,
            "gps_access": 1,
            "forward_page_type": 1,
            "channel_id": 0,
            "city": "110100",
            "hotsoon_filtered_count": 0,
            "hotsoon_has_more": 0,
            "follower_count": 0,
            "is_familiar": 0,
            "page_source": 0,
            "manifest_version_code": self.config.manifest_version_code,
            "_rticket": int(time.time() * 1000),
            "app_type": "normal",
            "iid": self.config.iid,
            "device_id": self.config.device_id,
            "ac": self.config.ac,
            "channel": self.config.channel,
            "aid": self.config.aid,
            "app_name": self.config.app_name,
            "version_code": self.config.version_code,
            "version_name": self.config.version_name,
            "device_platform": self.config.device_platform,
            "ssmix": self.config.ssmix,
            "device_type": self.config.device_type,
            "device_brand": self.config.device_brand,
            "language": self.config.language,
            "os_api": self.config.os_api,
            "os_version": self.config.os_version,
            "openudid": self.config.openudid,
            "uuid": self.config.uuid,
        }
        
        params = self._generate_signature(params)
        try:
            response = self.session.get(
                f"{self.BASE_URL}/aweme/v2/comment/list/",
                params=params,
                timeout=10
            )
            if response.status_code == 200:
                return response.json().get("comments", [])
        except Exception as e:
            print(f"获取评论失败: {e}")
        return []
    
    def like_comment(self, comment_id: str, aweme_id: str) -> bool:
        """点赞评论"""
        params = {
            "comment_id": comment_id,
            "aweme_id": aweme_id,
            "type": 1,  # 1点赞 0取消点赞
            "channel_id": 0,
            "address_book_access": 1,
            "gps_access": 1,
            "forward_page_type": 1,
            "city": "110100",
            "manifest_version_code": self.config.manifest_version_code,
            "_rticket": int(time.time() * 1000),
            "app_type": "normal",
            "iid": self.config.iid,
            "device_id": self.config.device_id,
            "ac": self.config.ac,
            "channel": self.config.channel,
            "aid": self.config.aid,
            "app_name": self.config.app_name,
            "version_code": self.config.version_code,
            "version_name": self.config.version_name,
            "device_platform": self.config.device_platform,
            "ssmix": self.config.ssmix,
            "device_type": self.config.device_type,
            "device_brand": self.config.device_brand,
            "language": self.config.language,
            "os_api": self.config.os_api,
            "os_version": self.config.os_version,
            "openudid": self.config.openudid,
            "uuid": self.config.uuid,
        }
        
        params = self._generate_signature(params)
        try:
            response = self.session.post(
                f"{self.BASE_URL}/aweme/v1/comment/digg/",
                data=params,
                timeout=10
            )
            return response.status_code == 200 and response.json().get("status_code") == 0
        except Exception as e:
            print(f"点赞评论失败: {e}")
        return False
    
    def batch_like_comments(self, aweme_id: str, count: int = 10) -> Dict:
        """批量点赞评论"""
        comments = self.get_video_comments(aweme_id, count)
        results = {"success": 0, "failed": 0}
        
        for comment in comments:
            cid = comment.get("cid")
            if cid and self.like_comment(cid, aweme_id):
                results["success"] += 1
                time.sleep(random.uniform(0.5, 2.0))  # 随机延迟防止过快
            else:
                results["failed"] += 1
        return results

class CommentTopSimulator:
    """评论置顶模拟器"""
    def __init__(self, api: DouyinAPI):
        self.api = api
        
    def simulate_top_comment(self, aweme_id: str, target_comment_id: str, like_threshold: int = 100) -> bool:
        """通过大量点赞模拟置顶效果"""
        for _ in range(like_threshold):
            if not self.api.like_comment(target_comment_id, aweme_id):
                return False
            time.sleep(random.uniform(0.3, 1.5))
        return True

if __name__ == "__main__":
    config = DouyinConfig()
    api = DouyinAPI(config)
    
    # 示例视频ID
    test_aweme_id = "1234567890123456789"
    
    print("获取视频评论中...")
    comments = api.get_video_comments(test_aweme_id, 20)
    print(f"获取到{len(comments)}条评论")
    
    if comments:
        print("\n开始批量点赞前5条评论...")
        results = api.batch_like_comments(test_aweme_id, 5)
        print(f"点赞完成 - 成功: {results['success']}, 失败: {results['failed']}")
        
        top_simulator = CommentTopSimulator(api)
        target_comment = comments[0]
        print(f"\n尝试将评论ID {target_comment['cid']} 模拟置顶...")
        if top_simulator.simulate_top_comment(test_aweme_id, target_comment["cid"], 50):
            print("模拟置顶操作完成")
        else:
            print("模拟置顶操作失败")