风纪委众议{'code': 25018, 'message': '不能进行此操作', 'ttl': 1}

[Evi233]

import requests
import time
from functools import reduce
from hashlib import md5
import urllib.parse
import json
import logging
import asyncio
from openai import AsyncOpenAI  # 使用异步客户端

# --- 配置日志记录 ---
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

# --- WBI签名算法实现 (根据wbi.md) ---
mixinKeyEncTab = [
    46, 47, 18, 2, 53, 8, 23, 32, 15, 50, 10, 31, 58, 3, 45, 35, 27, 43, 5, 49,
    33, 9, 42, 19, 29, 28, 14, 39, 12, 38, 41, 13, 37, 48, 7, 16, 24, 55, 40,
    61, 26, 17, 0, 1, 60, 51, 30, 4, 22, 25, 54, 21, 56, 59, 6, 63, 57, 62, 11,
    36, 20, 34, 44, 52
]

def getMixinKey(orig: str) -> str:
    '对 imgKey 和 subKey 进行字符顺序打乱编码'
    return reduce(lambda s, i: s + orig[i], mixinKeyEncTab, '')[:32]

def encWbi(params: dict, img_key: str, sub_key: str) -> dict:
    '为请求参数进行 wbi 签名'
    mixin_key = getMixinKey(img_key + sub_key)
    curr_time = round(time.time())
    params['wts'] = curr_time
    params = dict(sorted(params.items()))
    # 过滤 value 中的 "!'()*" 字符
    params = {
        k: ''.join(filter(lambda char: char not in "!'()*", str(v)))
        for k, v in params.items()
    }
    query = urllib.parse.urlencode(params)
    wbi_sign = md5((query + mixin_key).encode()).hexdigest()
    params['w_rid'] = wbi_sign
    return params

class WbiSigner:
    """WBI签名器，带缓存功能"""
    def __init__(self, session: requests.Session):
        self.session = session
        self.img_key = None
        self.sub_key = None
        self.last_refresh_time = 0
        self.cache_duration = 3600 * 12  # 缓存12小时

    def _prepare_dynamic_headers(self, url: str) -> dict:
        """为内部刷新密钥的请求动态生成headers"""
        from urllib.parse import urlparse, urlunparse
        parsed_url = urlparse(url)
        origin = f"{parsed_url.scheme}://{parsed_url.netloc}"
        referer = urlunparse((parsed_url.scheme, parsed_url.netloc, parsed_url.path, '', '', ''))
        
        headers = self.session.headers.copy()
        headers.update({
            'Origin': origin,
            'Referer': referer
        })
        return headers

    def _refresh_keys(self) -> bool:
        '获取并刷新最新的 img_key 和 sub_key'
        url = 'https://api.bilibili.com/x/web-interface/nav'
        try:
            headers = self._prepare_dynamic_headers(url)
            resp = self.session.get(url, headers=headers)
            resp.raise_for_status()
            json_content = resp.json()
            img_url: str = json_content['data']['wbi_img']['img_url']
            sub_url: str = json_content['data']['wbi_img']['sub_url']
            self.img_key = img_url.rsplit('/', 1)[1].split('.')[0]
            self.sub_key = sub_url.rsplit('/', 1)[1].split('.')[0]
            self.last_refresh_time = time.time()
            logging.info("WBI keys refreshed successfully.")
            return True
        except Exception as e:
            logging.error(f"Failed to get WBI keys: {e}")
            return False

    def sign(self, params: dict) -> dict:
        '对参数进行签名，必要时刷新wbi_key'
        if not self.img_key or time.time() - self.last_refresh_time > self.cache_duration:
            if not self._refresh_keys():
                raise Exception("Could not retrieve WBI keys, aborting signing.")
        
        return encWbi(params, self.img_key, self.sub_key)

# --- Bilibili风纪委员机器人 ---

class BiliJuryBot:
    def __init__(self, cookie: str, deepseek_api_key: str):
        if not cookie: raise ValueError("Cookie 不能为空。")
        if not deepseek_api_key: raise ValueError("DeepSeek API Key 不能为空。")
            
        self.session = requests.Session()
        # 初始化基础请求头，不包含动态的Origin和Referer
        self.session.headers.update({
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/138.0.0.0 Safari/537.36 Edg/138.0.0.0',
            'Accept': 'application/json, text/plain, */*',
            'Accept-Language': 'zh-CN,zh;q=0.9',
        })
        self.session.headers['Cookie'] = cookie
        try:
            self.csrf = [c.strip() for c in cookie.split(';') if 'bili_jct' in c][0].split('=')[1]
        except IndexError:
            raise ValueError("Cookie中未找到'bili_jct' (CSRF token)。")

        self.wbi_signer = WbiSigner(self.session)
        self.llm_client = AsyncOpenAI(api_key=deepseek_api_key, base_url="https://yunwu.ai/v1")

    def _prepare_dynamic_headers(self, url: str) -> dict:
        """根据URL动态生成请求头中的Origin和Referer"""
        from urllib.parse import urlparse, urlunparse
        parsed_url = urlparse(url)
        origin = f"{parsed_url.scheme}://{parsed_url.netloc}"
        referer = urlunparse((parsed_url.scheme, parsed_url.netloc, parsed_url.path, '', '', ''))
        
        headers = self.session.headers.copy()
        headers.update({
            'Origin': origin,
            'Referer': referer
        })
        return headers

    def get_next_case(self) -> str or None:
        """获取下一个众议ID"""
        url = 'https://api.bilibili.com/x/credit/v2/jury/case/next'
        try:
            headers = self._prepare_dynamic_headers(url)
            response = self.session.get(url, headers=headers)
            response.raise_for_status()
            data = response.json()
            if data['code'] == 0:
                case_id = data['data']['case_id']
                logging.info(f"获取到新案件ID: {case_id}")
                return case_id
            elif data['code'] in [25005, 25008]:
                logging.warning(f"无法获取新案件: {data['message']}")
                return None
            else:
                logging.error(f"获取新案件失败: {data}")
                return None
        except Exception as e:
            logging.error(f"请求 get_next_case 异常: {e}")
            return None

    def get_case_info(self, case_id: str) -> dict or None:
        """获取众议详情"""
        url = f'https://api.bilibili.com/x/credit/v2/jury/case/info?case_id={case_id}'
        try:
            headers = self._prepare_dynamic_headers(url)
            response = self.session.get(url, headers=headers)
            response.raise_for_status()
            data = response.json()
            if data['code'] == 0:
                logging.info(f"获取案件 {case_id} 详情成功。")
                return data['data']
            else:
                logging.error(f"获取案件 {case_id} 详情失败: {data}")
                return None
        except Exception as e:
            logging.error(f"请求 get_case_info 异常: {e}")
            return None

    def get_video_details(self, avid: int) -> dict or None:
        """获取视频详情，主要是为了up_mid"""
        url = 'https://api.bilibili.com/x/web-interface/view'
        params = {'aid': avid}
        try:
            full_url = url + '?' + urllib.parse.urlencode(params)
            headers = self._prepare_dynamic_headers(full_url)
            response = self.session.get(url, params=params, headers=headers)
            response.raise_for_status()
            data = response.json()
            return data['data'] if data['code'] == 0 else None
        except Exception as e:
            logging.error(f"请求 get_video_details (aid: {avid}) 异常: {e}")
            return None

    def get_video_summary(self, aid: int, cid: int, up_mid: int) -> dict or None:
        """获取视频AI总结"""
        url = 'https://api.bilibili.com/x/web-interface/view/conclusion/get'
        base_params = {'aid': aid, 'cid': cid, 'up_mid': up_mid}
        try:
            signed_params = self.wbi_signer.sign(base_params.copy())
            full_url = url + '?' + urllib.parse.urlencode(signed_params)
            headers = self._prepare_dynamic_headers(full_url)
            response = self.session.get(url, params=signed_params, headers=headers)
            response.raise_for_status()
            data = response.json()
            if data['code'] == 0 and data['data']['code'] == 0:
                logging.info(f"获取视频(aid: {aid})AI总结成功。")
                return data['data']['model_result']
            else:
                logging.warning(f"视频(aid: {aid})无AI总结或获取失败: {data}")
                return None
        except Exception as e:
            logging.error(f"请求 get_video_summary (aid: {aid}) 异常: {e}")
            return None

    def submit_vote(self, case_id: str, vote: int, content: str) -> bool:
        """提交投票"""
        url = 'https://api.bilibili.com/x/credit/v2/jury/vote'
        payload = {
            'case_id': case_id, 'vote': vote, 'content': content,
            'anonymous': 1, 'insiders': 0, 'csrf': self.csrf
        }
        print(payload)
        try:
            headers = self._prepare_dynamic_headers(url)
            response = self.session.post(url, data=payload, headers=headers)
            response.raise_for_status()
            data = response.json()
            if data['code'] == 0:
                logging.info(f"案件 {case_id} 投票成功 (选项: {vote}, 理由: '{content}')。")
                return True
            else:
                logging.error(f"案件 {case_id} 投票失败: {data}")
                return False
        except Exception as e:
            logging.error(f"请求 submit_vote (case_id: {case_id}) 异常: {e}")
            return False

    def translate_case_type(self, case_type_id: int) -> str:
        """将案件类型ID转换为可读字符串"""
        type_map = {1: "单条评论", 2: "评论氛围", 3: "单条弹幕", 4: "弹幕氛围"}
        return type_map.get(case_type_id, f"未知类型 ({case_type_id})")

    def has_link_or_image(self, data) -> bool:
        """递归检查数据中是否包含链接或图片标签"""
        if isinstance(data, dict):
            return any(self.has_link_or_image(v) for v in data.values())
        if isinstance(data, list):
            return any(self.has_link_or_image(elem) for elem in data)
        if isinstance(data, str):
            return 'http://' in data or 'https://' in data or '[img]' in data
        return False

    async def analyze_with_llm(self, case_type: str, title: str, summary: dict or None, case_info: dict) -> dict or None:
        """使用两阶段LLM进行分析，并返回投票决策"""
        logging.info("--- [LLM模块] 开始分析 ---")

        logging.info("第一阶段: 使用 deepseek-reasoner 进行初步分析...")
        prompt_stage1 = f"""你是一个bilibili风纪委员会成员，你需要通过给出信息判断是否违规。
版本信息：V2
任务类型：{case_type}
视频标题：{title}
视频总结的内容：{json.dumps(summary, ensure_ascii=False) if summary else "无"}
案件信息：{json.dumps(case_info, ensure_ascii=False)}
请你的输出为无格式的json输出,包括success,vote和content部分。
'success'是是否解析成功,如果案件信息含有链接或者图片（或者不含有视频总结）并且因为信息的缺失影响你的判决，请设置为false。如果不影响判决则为true
'vote' 的值：
如果任务类型为单条弹幕或者单条评论应为 1(合适), 2(一般), 3(不合适), 4(无法判断)。
如果任务类型为弹幕氛围或者评论氛围应为11(好),12(一般),13(差),14(无法判断)
'content' 是投票理由,不大于20字,自然,不包括多余的符号如句末句号,表情符号等。"""

        try:
            response_stage1 = await self.llm_client.chat.completions.create(
                model="deepseek-reasoner",
                messages=[{"role": "user", "content": prompt_stage1}],
                stream=False
            )
            final_decision_content = response_stage1.choices[0].message.content
            logging.info("第一阶段完成，获得初步结果。")
        except Exception as e:
            logging.error(f"LLM第一阶段调用失败: {e}")
            return {"vote": 4, "content": "模型分析失败"}

        logging.info("第二阶段: 使用 gpt-4o-mini 进行格式化与验证...")
        try:
            response_stage2 = await self.llm_client.chat.completions.create(
                model="gpt-4o-mini",
                messages=[
                    {"role": "system", "content": "用户会给你输入一段文字，包含json（有时不包含，可能需要自行查找），把它输出为纯粹json格式"},
                    {"role": "user", "content": final_decision_content}
                ],
                response_format={
                    "type": "json_schema",
                    "json_schema": {
                        "name": "vote_result", "strict": True,
                        "schema": {
                            "type": "object",
                            "properties": {
                                "success": {"type": "boolean", "description": "操作是否成功"},
                                "vote": {"type": "number", "description": "投票数"},
                                "content": {"type": "string", "description": "与本次操作或投票相关的内容"}
                            },
                            "required": ["success", "vote", "content"],
                            "additionalProperties": False
                        }
                    }
                }
            )
            
            result_json = json.loads(response_stage2.choices[0].message.content)
            logging.info(f"第二阶段完成，最终决策: {result_json}")
            result_json.pop('success', None)
            return result_json

        except Exception as e:
            logging.error(f"LLM第二阶段调用或解析失败: {e}")
            return {"vote": 4, "content": "模型格式化失败"}

    async def run(self, max_cases: int = 10, delay_seconds: int = 15):
        """主运行循环 (异步)"""
        cases_processed = 0
        while cases_processed < max_cases:
            case_id = self.get_next_case()
            if not case_id:
                logging.info("无更多案件或已达今日上限，程序退出。")
                break

            case_details = self.get_case_info(case_id)
            if not case_details:
                await asyncio.sleep(delay_seconds)
                continue
            
            case_type_id = case_details.get('case_type', 0)
            case_type_str = self.translate_case_type(case_type_id)

            video_summary = None
            if case_details.get('avid') and case_details.get('cid'):
                video_details = self.get_video_details(case_details['avid'])
                if video_details and video_details.get('owner'):
                    up_mid = video_details['owner']['mid']
                    video_summary = self.get_video_summary(case_details['avid'], case_details['cid'], up_mid)
            
            llm_decision = await self.analyze_with_llm(
                case_type_str, case_details.get('title', '无标题'), 
                video_summary, case_details.get('case_info', {})
            )

            if llm_decision is None:
                logging.info(f"已跳过案件 {case_id}，获取下一个。")
                continue

            if self.submit_vote(case_id, llm_decision['vote'], llm_decision['content']):
                cases_processed += 1

            logging.info(f"已处理 {cases_processed}/{max_cases} 案件。等待 {delay_seconds} 秒...")
            await asyncio.sleep(delay_seconds)
            
        logging.info(f"运行结束。共处理案件: {cases_processed}")


async def main():
    # ==================== 配置区域 ====================
    # 1. 您的 Bilibili Cookie 已填入。
    MY_COOKIE = "bili_jct=647***; SESSDATA=259963***"
    
    # 2. 请在此处填入您的 DeepSeek API Key。
    DEEPSEEK_API_KEY = "sk-EQgn***" # <--- 在这里替换你的 KEY

    # 3. 设置本次运行要处理的最大案件数量。
    MAX_CASES_TO_PROCESS = 5

    # 4. 设置每次处理案件后的延迟时间（秒），以避免请求过于频繁。
    DELAY_BETWEEN_CASES = 20
    # ===============================================
    
    if "sk-xxxx" in DEEPSEEK_API_KEY:
        logging.error("错误: 请在脚本的配置区域填入您的 DEEPSEEK_API_KEY。")
        return

    try:
        bot = BiliJuryBot(cookie=MY_COOKIE, deepseek_api_key=DEEPSEEK_API_KEY)
        await bot.run(max_cases=MAX_CASES_TO_PROCESS, delay_seconds=DELAY_BETWEEN_CASES)
    except Exception as e:
        logging.critical(f"程序发生未处理的严重错误: {e}")

if __name__ == '__main__':
    asyncio.run(main())

原来是一个用AI自动判断风纪的一个东西，但是提交的时候总是{'code': 25018, 'message': '不能进行此操作', 'ttl': 1}。之前遇到这个是通过 #1019 解决的，但是没好两次又出现这个问题。求助！