app.py

import base64
import io
import os
import sys
import traceback
# from http.cookiejar import domain_match
from typing import List
import gradio as gr
from PIL import Image
import time
from langchain.agents import AgentExecutor, create_react_agent
from langchain_core.prompts import ChatPromptTemplate
from model import *
from gradio import ChatMessage, on, HTML
from tools import *
from prompt import *
import sys
from gradio.components.chatbot import MessageDict
from langchain_community.llms import Tongyi
from io import BytesIO
from css import *
from concurrent.futures import ThreadPoolExecutor
from TrustedUrl import *
from hashlib import md5
import random


os.environ["OPENAI_API_KEY"] = "sb-6a683cb3bd63a9b72040aa2dd08feff8b68f08a0e1d959f5"
os.environ['OPENAI_BASE_URL'] = "https://api.openai-sb.com/v1/"
os.environ["SERPAPI_API_KEY"] = "dcc98b22d5f7d413979a175ff7d75b721c5992a3ee1e2363020b2bbdf4f82404"
os.environ['TAVILY_API_KEY'] = "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"  # you
os.environ["REPLICATE_API_TOKEN"] = "r8_IYJpjwjrxegcUfBeBbyUxErJXXsnHDM4AlSQQ"
os.environ["DASHSCOPE_API_KEY"] = "sk-8159f0ed38994c3b96b4527404ea1cda"

class ImageProcessor:
    """Handles image processing and analysis"""

    @staticmethod
    def resize_image(image: Image.Image) -> Image.Image:
        """Resize image to appropriate dimensions"""
        width, height = image.size
        while width >= 500 or height >= 400:
            width = width * 0.8
            height = height * 0.8
        return image.resize((int(width), int(height)))

    @staticmethod
    def encode_image(image: Image.Image) -> str:
        """Encode image to base64 string"""
        buffered = io.BytesIO()
        image.save(buffered, format="PNG")
        return base64.b64encode(buffered.getvalue()).decode("utf-8")

    @staticmethod
    def load_image(image_path_or_url: str) -> Optional[Image.Image]:
        """Load image from path or URL"""
        if image_path_or_url.startswith(('http://', 'https://')):
            try:
                response = requests.get(image_path_or_url, timeout=10)
                response.raise_for_status()
                return Image.open(BytesIO(response.content))
            except Exception as e:
                print(f"Failed to download image: {image_path_or_url} | Error: {e}")
                return None
        else:
            try:
                return Image.open(image_path_or_url)
            except Exception as e:
                print(f"Failed to open local image: {image_path_or_url} | Error: {e}")
                return None

    @staticmethod
    def analyze_image(image: Image.Image, prompt: str) -> str:
        base64_image = ImageProcessor.encode_image(image)
        client = OpenAI(
            api_key=os.getenv('DASHSCOPE_API_KEY'),
            base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
        )
        completion = client.chat.completions.create(
            model="qwen-vl-plus",
            messages=[
                {
                    "role": "system",
                    "content": [{"type": "text", "text": f"{prompt}"}]},
                {
                    "role": "user",
                    "content": [
                        {
                            "type": "image_url",
                            "image_url": {"url": f"data:image/png;base64,{base64_image}"},
                        },
                        {"type": "text", "text": "请分析该图片"},
                    ],
                }
            ],
        )
        return completion.choices[0].message.content


class FactChecker:
    """Main fact-checking agent class"""

    def __init__(self):
        self.tools = None
        self.selected_language = "ch"
        self.image_summaries = ""
        self.query = ""
        self.llm = Tongyi(model_name="qwen-plus", temperature=0.1)
        self.selected_contrast = "qwen"
        self.selected_model = "自研"
        self.contrast_model_prompt = ""
        self.fact_checking = False
        self.deep_search = 2
        self.trusted_url = {"TRUSTED_DOMAINS_ABOARD":TRUSTED_DOMAINS_ABOARD,"TRUSTED_DOMAINS_DOMESTIC":TRUSTED_DOMAINS_DOMESTIC,"TRUSTED_SUFFIXES":TRUSTED_SUFFIXES,"TRUSTED_PERSON":TRUSTED_PERSON}
        dashscope.api_key = "sk-8159f0ed38994c3b96b4527404ea1cda"

    def set_language(self, language: str):
        """Set the language for responses"""
        self.selected_language = "ch" if language == "中文" else "en"

    def baidu_translate(self,query):
        appid = '20250629002392938'
        appkey = 'CAk49yEcWE0MTWzFdSK0'
        from_lang = 'zh'
        to_lang = 'en'
        endpoint = 'http://api.fanyi.baidu.com'
        path = '/api/trans/vip/translate'
        url = endpoint + path

        def make_md5(s, encoding='utf-8'):
            return md5(s.encode(encoding)).hexdigest()

        salt = random.randint(32768, 65536)
        sign = make_md5(appid + query + str(salt) + appkey)

        headers = {'Content-Type': 'application/x-www-form-urlencoded'}
        payload = {'appid': appid, 'q': query, 'from': from_lang, 'to': to_lang, 'salt': salt, 'sign': sign}
        r = requests.post(url, params=payload, headers=headers)
        result = r.json()["trans_result"][0]["dst"]
        return result

    def get_prompt(self) -> ChatPromptTemplate:
        if self.selected_language == "en":
            num_marker = "(This 'Thought/Action/Action Input/Observation' loop can repeat up to 3 times)"
            title_marker = "### 6. Historical records:"
            if self.fact_checking:
                prompt_with_title = en_prompt.replace(
                    title_marker,
                    f"{title_marker}\n### 7. When encountering personal professional titles (positions), it is important to carefully identify them\n"
                )
                prompt_with_num = prompt_with_title.replace(
                    num_marker,
                    "(This 'Thought/Action/Action Input/Observation' loop can repeat up to 5 times)" if self.deep_search == 3 else "(This 'Thought/Action/Action Input/Observation' loop can repeat up to 3 times)"
                )
            else:
                prompt_with_num = en_prompt
            print(prompt_with_num)
            return ChatPromptTemplate.from_template(prompt_with_num)
        else:
            num_marker = "(这个 'Thought/Action/Action Input/Observation' 逻辑可以最多重复3次)"
            title_marker = "### 6. 以下为历史记录: "
            if self.fact_checking:
                prompt_with_title = ch_prompt.replace(
                    title_marker,
                    f"{title_marker}\n### 7. 当遇到个人职称（职位）时要仔细鉴别\n"
                )
                prompt_with_num = prompt_with_title.replace(
                    num_marker,
                    "(这个 'Thought/Action/Action Input/Observation' 逻辑可以最多重复5次)" if self.deep_search == 3 else "(这个 'Thought/Action/Action Input/Observation' 逻辑可以最多重复3次)"
                )
                
            else:
                prompt_with_num = ch_prompt
            print(prompt_with_num)
            return ChatPromptTemplate.from_template(prompt_with_num)

    def get_image_prompt(self) -> str:
        if self.selected_language == "ch":
            return image_ch_prompt
        else:
            return image_en_prompt

    def process_image(self, image_path_or_url: str) -> Optional[str]:
        """Process and analyze an image"""
        if not image_path_or_url:
            return None

        image = ImageProcessor.load_image(image_path_or_url)
        if not image:
            return None

        image = ImageProcessor.resize_image(image)
        prompt = self.get_image_prompt()
        analysis = ImageProcessor.analyze_image(image, prompt)
        return analysis.replace('*', '').replace('\n', ' ').strip()

    def process_image_load(self, image: Image.Image) -> Optional[str]:
        if not image:
            return None
        image = ImageProcessor.resize_image(image)
        prompt = self.get_image_prompt()
        analysis = ImageProcessor.analyze_image(image, prompt)
        return analysis.replace('*', '').replace('\n', ' ').strip()

    def extract_agent_thoughts(self, intermediate_steps: List) -> List[ChatMessage]:
        """Extract agent thoughts from intermediate steps"""
        agent_thought = []
        pattern1 = r"Pre Thought:(.*?)Thought: (.*?)\nAction: (.*?)\nAction Input: (.*)"
        pattern2 = r"Thought: (.*?)\nAction: (.*?)\nAction Input: (.*)"

        for index in intermediate_steps:
            match = re.search(pattern1, index[0].log, re.S)
            before_thought = match.group(1).strip() if match else ""
            thought = match.group(2).strip() if match else ""
            action = match.group(3).strip() if match else ""
            action_input = match.group(4).strip() if match else ""
            if self.image_summaries != "":
                agent_thought.append(
                    ChatMessage(
                        role="assistant",
                        content=self.image_summaries,
                        metadata={"title": "图片分析结果："}
                    )
                )
            if not match:
                match = re.search(pattern2, index[0].log, re.S)
                before_thought = match.group(1).strip() if match else ""
                thought = match.group(2).strip() if match else ""
                action_input = match.group(3).strip() if match else ""

            if len(before_thought) >= 3:
                # agent_thought.append({"title": "预思考：", "content": before_thought})
                agent_thought.append(
                    ChatMessage(
                        role="assistant",
                        content=before_thought,
                        metadata={"title": "预思考："}
                    )
                )
            if len(action_input) >= 3:
                # agent_thought.append({"title": f"工具调用:{action}", "content": action_input})
                web_search_tools = {"BoCha Webs Search","Image Search","Baidu News Search","tavily_search_results_json"}
                if action in web_search_tools or action == "":
                    action = "网页搜索"
                agent_thought.append(
                    ChatMessage(
                        role="assistant",
                        content=action_input,
                        metadata={"title": f"工具调用:{action}"}
                    )
                )

        return agent_thought

    def process_news_results(self, intermediate_steps: List) -> List[Dict]:
        news_results = []
        if not intermediate_steps:
            return news_results

        first_step = intermediate_steps[0][0]

        if first_step.tool == "tavily_search_results_json":
            for step1 in intermediate_steps[:1]:  # Only first step
                try:
                    for step2 in step1[1][:3]:  # First 3 results
                        news_results.append({
                            "title": step2['content'][:28] + "...",
                            "url": step2['url'],
                            "source": "",
                            "image": "",
                            "time": ""
                        })
                except:
                    continue

        elif first_step.tool == "BoCha Webs Search":
            for step1 in intermediate_steps:
                try:
                    for i in range(3):
                        news_results.append({
                            "title": step1[1]['data']['webPages']['value'][i]['snippet'][:28] + "...",
                            "url": step1[1]['data']['webPages']['value'][i]['url'],
                            "source": step1[1]['data']['webPages']['value'][i]['siteName'],
                            "image": step1[1]['data']['images']['value'][i]['contentUrl'],
                            "time": step1[1]['data']['webPages']['value'][i]['dateLastCrawled']
                        })
                except:
                    continue

        elif first_step.tool == "Baidu News Search":
            for step1 in intermediate_steps:
                try:
                    for i in range(3):
                        news_results.append({
                            "title": step1[1][i]['title'][:28] + "...",
                            "url": step1[1][i]['link'],
                            "source": step1[1][i]['source'],
                            "image": "",
                            "time": ""
                        })
                except:
                    continue

        return news_results

    def set_tools(self):
        self.tools = [
            BoChaSearchTool(selected_language=self.selected_language,fact_checking=self.fact_checking,trusted_url=self.trusted_url),
            ChineseTavilySearchResults(selected_language=self.selected_language,fact_checking=self.fact_checking,trusted_url=self.trusted_url),
            ImageSearchTool(selected_language=self.selected_language),
            WeatherCrossing(selected_language=self.selected_language),
            GetHoliday(selected_language=self.selected_language),
            GetLocation(selected_language=self.selected_language),
            CurrencyConversion(selected_language=self.selected_language),
            SafeCodeExecutor(selected_language=self.selected_language),
            SafeExpressionEvaluator(selected_language=self.selected_language),
            RegionInquiryTool(selected_language=self.selected_language),
            HTMLTextExtractor(selected_language=self.selected_language)
        ]

    @staticmethod
    def format_reply(reply:Dict) -> HTML:
        formatted_text = ""
        news_list = reply.get("news", [])
        container_html = '<div class="news-container">'  # 添加容器

        for news in news_list[:5]:
            card_html = f"""
                <div class="news-card">
                    <img src="{news['image']}" alt="新闻图片">
                    <div class="title"><a href="{news['url']}" target="_blank">{news['title']}</a></div>
                    <div class="source">{news['source']}</div>
                </div>
                """
            container_html += card_html
        container_html += '</div>'  # 关闭容器
        formatted_text += container_html
        formatted_text += reply.get("text", "")
        return gr.HTML(formatted_text)

    def contract(self)-> str | None:
        tool = BoChaSearchTool(self.selected_language)
        web_information = tool._run(self.query)
        information = []
        for step in web_information['data']['webPages']['value']:
            information.append(step['snippet'])
        contract_query = self.query + "搜索到的相关新闻：" + str(information)
        if self.selected_contrast == "qwen":
            result = qwen(contract_query)
        elif self.selected_contrast == "llama":
            result = llama(contract_query)
        elif self.selected_contrast == "glm":
            result = glm(contract_query)
        elif self.selected_contrast == "doubao":
            result = doubao(contract_query)
        elif self.selected_contrast == "deepseek":
            result = deepseek(contract_query)
        else:
            result = baichuan(contract_query)
        return result

    def SelectLanguage(self,option:str):
        if option == "英文":
            self.selected_language = "en"
        else:
            self.selected_language = "ch"

    def SelectModel(self,option:str):
        if option == "自研":
            self.selected_model = "gpt"
        elif option == "llama3-70b":
            self.selected_model = "llama3-70b"
        elif option == "llama3-8b":
            self.selected_model = "llama3-8b"
        else:
            self.selected_model = "mistral"

    def SelectMode(self,option:str):
        if option == "否":
            self.fact_checking = False
        else:
            self.fact_checking = True

    def SelectDeep(self,option:str):
        if option == "普通检测":
            self.deep_search = 2
        elif option == "深度检测":
            self.deep_search = 3
        else:
            self.deep_search = 1

    def SelectContrast(self,option:str):
        if option == "qwen":
            self.selected_contrast = "qwen"
        elif option == "llama":
            self.selected_contrast = "llama"
        elif option == "glm":
            self.selected_contrast = "glm"
        elif option == "doubao":
            self.selected_contrast = "doubao"
        elif option == "deepseek":
            self.selected_contrast = "deepseek"
        else:
            self.selected_contrast = "baichuan"

    def update_image_summaries(self):
        return self.image_summaries

    def ModelPrompt(self,prompt):
        self.contrast_model_prompt = prompt

    def AddTrustedSource(self,trusted_source):
        try:
            urls = trusted_source.split('\n')
            # print(urls)
            for url in urls:
                extracted = tldextract.extract(url)
                # subdomain = extracted.subdomain.lower()
                domain = extracted.domain.lower()
                suffix = extracted.suffix.lower()
                if domain not in self.trusted_url['TRUSTED_DOMAINS_DOMESTIC']:
                    # print(domain)
                    self.trusted_url['TRUSTED_DOMAINS_DOMESTIC'].add(domain)
            return "添加成功"
        except:
            return "添加失败，请重新尝试"

    def qwen_tool(self, query: str,language: str):
        
        messages = [
            {
                "role": "system",
                "content": """
                记住, 你是一个虚假信息鉴定工具, 由哈尔滨工业大学的SCIR实验室开发。
    你的任务是鉴定输入内容的真实性，可调用相关工具来搜集证据，最终给出判断标签及判断理由。

    ## 注意事项:
    ### 1. 鉴定结果的各个标签定义如下：
        (1) 虚假：内容中的所有细节或大部分细节都能被事实证明为虚假的。
        (2) 真假参半：内容中约一半是真实的，一半是虚假的。
        (3) 真实：内容中的所有细节或大部分细节都能被事实证明为真实的。
        (4) 无法鉴定：缺乏事实来验证，且完全无法推测其真实性。
                或者无法理解输入内容，输入无完整语义，歧义太大。
                或者为不可鉴定的内容（例如：未来状态，观点）。
                或者无法归纳到其它标签。

    ### 2. 按照以下专家经验判断:
        (1) 一步一步地思考，来生成解释，逻辑清晰，步骤详细，最好能体现思考及推理过程，并引用相关的证据，而不是简单的复述证据。
        (2) 如果输入内容包含多个细节，会逐一验证。
        (3) 虚假信息验证时的线索除了与事实（常识，自身知识，证据）冲突之外，还有语言学特征：
            - 词汇层面：情绪化词汇过多，模糊性词汇频繁出现
            - 语义层面: 逻辑矛盾,内容前后表述自相矛盾; 缺乏合理因果关系,随意编造因果关联，把并无实质联系的两件事强行说成因果关系; 过度夸张或违背常识.
            - 风格层面: 语言风格与来源不符, 例声称是官方发布的信息，却采用随意、口语化且不规范的语言风格；模仿痕迹明显
            - 语用层面: 意图引导性过强,不是客观呈现信息，带有强烈主观引导意图的内容; 语境不匹配
    ### 3. 执行鉴定的过程需要使用中文。

    ## 要求遵循如下格式来执行鉴定：
        Question: 需要被鉴定的内容。
        Image_information: 对输入的图片的鉴定结果。
        Pre Thought: 简要理解要验证的内容，是否包含多个要验证的细节部分，做好规划。
        Thought: 为了验证第一个细节部分, 你应该怎么做.
        Action:  工具列表({tool_names})之一的工具名，不能有其他内容
        Action Input: Action的输入参数.
        Observation: Action的执行结果.
        Thought: 为了验证第二个细节, 你应该怎么做.
        ... (这个 'Thought/Action/Action Input/Observation' 逻辑可以最多重复3次)
        Thought: 我现在知道最后的答案了。
        Final Answer: 我的最终回答. 要求包含一段鉴定过程的总结, 判断标签及判断理由。
    """,
            },
            {
                "role": "user",
                "content": query,
            },
                ] if language == "ch" else [
            {
                "role": "system",
                "content": """
                Remember, you are a misinformation identification tool developed by the SCIR Lab at Harbin Institute of Technology.
Your task is to verify the authenticity of input content by calling relevant tools to gather evidence, and ultimately provide a judgment label with reasoning.

## Important Notes:
### 1. Definition of judgment labels:
    (1) False: All details or Most details in the content can be proven false with factual evidence.
    (3) Mixed: Approximately half of the content is true and half is false.
    (5) True: All details or Most details in the content can be proven true with factual evidence.
    (8) Cannot_Determine: 
             - Lacks factual verification and cannot speculate authenticity
             - Or cannot understand input (incomplete semantics/ambiguous)
             - Or non-verifiable content (e.g., future states, opinions)
             - Or cannot be categorized into other labels

### 2. Follow these expert guidelines:
    (1) Think step-by-step to generate explanations with clear logic and detailed reasoning. Cite relevant evidence rather than simply restating it.
    (2) If input contains multiple claims, verify each one individually.
    (3) Misinformation clues include not only factual conflicts (common sense, knowledge, evidence) but also linguistic features:
        - Lexical level: Excessive emotional/ambiguous vocabulary
        - Semantic level: Logical contradictions, inconsistent statements; Lack of reasonable causation (forced correlations); Exaggeration or violation of common sense
        - Stylistic level: Language style inconsistent with claimed source (e.g., claiming official information but using informal language)
        - Pragmatic level: Strong intentional bias (non-objective presentation); Context mismatch

### 3. The verification process must be conducted in English.

## Required response format:
    Question: Content to be verified.
    Image_information: Verification results of input images.
    Pre Thought: Briefly understand the content to verify and plan verification steps for multiple claims if present.
    Thought: What should you do to verify the first claim.
    Action: Exactly one tool name from ({tool_names}), nothing else
    Action Input: Parameters for the Action.
    Observation: Result from the Action.
    Thought: What should you do to verify the second claim.
    ... (This 'Thought/Action/Action Input/Observation' loop can repeat up to 3 times)
    Thought: I now know the final answer.
    Final Answer: Must include: 
                 - Summary of verification process
                 - Judgment label 
                 - Reasoning
    """,
            },
            {
                "role": "user",
                "content": query,
            },
                ]
        client = OpenAI(
            api_key="sk-8159f0ed38994c3b96b4527404ea1cda",
            base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",  # 填写DashScope SDK的base_url
        )
        tools = [
            {
                "type": "function",
                "function": {
                    "name": "bocha_search",
                    "description": "首选搜索引擎，当需要检索信息时调用",
                    # 因为获取当前时间无需输入参数，因此parameters为空字典
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "query": {
                                "type": "string",
                                "description": "要检索的问题。",
                            }
                        },
                        "required": ["query"],
                    },
                },
            }
        ]
        try:
            completion = client.chat.completions.create(
                model="qwen-plus",
                messages=messages,
                tools=tools,
            )
            result_data = json.loads(completion.model_dump_json())
            result = result_data["choices"][0]["message"]["content"]

            separator = "Final Answer:"
            if separator in result:
                # 分割文本并取最后一部分（防止多次出现）
                parts = result.split(separator)
                answer = parts[-1].strip()
                print(answer)
                return answer
            else:
                print("none")
                return "未检索到内容，请稍后重试"
        except:
            return "请求失败"

    def check_facts_chat(self,dict, space):
        """Main fact-checking method"""
        topic = dict['text']
        image = dict['files']

        if not image:
            self.image_summaries = ""
        else:
            image_path = image[0]
            image = Image.open(image_path)
            self.image_summaries = self.process_image_load(image)

        if not topic:
            return "无输入内容"

        self.query = topic
        topic_en = self.baidu_translate(topic)

        if self.deep_search == 1:
            result_ch = self.qwen_tool(topic,"ch")
            result_en = self.qwen_tool(topic_en,"en")
            return result_ch + "\n" + result_en

        # Create agent
        prompt = self.get_prompt()
        prompt_en =ChatPromptTemplate.from_template(en_prompt)
        self.set_tools()
        agent = create_react_agent(self.llm, self.tools, prompt)
        agent_en = create_react_agent(self.llm, self.tools, prompt_en)
        # Redirect stdout to capture agent thoughts
        captured_output = io.StringIO()
        sys.stdout = captured_output

        cur_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
        # Execute agent
        agent_executor = AgentExecutor(
            agent=agent,
            tools=self.tools,
            verbose=True,
            handle_parsing_errors=True,
            return_intermediate_steps=True
        )
        agent_executor_en = AgentExecutor(
            agent=agent_en,
            tools=self.tools,
            verbose=False,
            handle_parsing_errors=True,
            return_intermediate_steps=True
        )

        def run_agent(agent_executor, input_data):
            return agent_executor.invoke(input_data)
        try:
            with ThreadPoolExecutor() as executor:
                future1 = executor.submit(run_agent, agent_executor, {
                    "current_time": cur_time,
                    "input": topic,
                    "image_information": self.image_summaries
                })

                future2 = executor.submit(run_agent, agent_executor_en, {
                    "current_time": cur_time,
                    "input": topic_en,
                    "image_information": self.image_summaries
                })

                response1 = future1.result()
                response2 = future2.result()
        # try:
        #     response = agent_executor.invoke({
        #         "current_time": cur_time,
        #         "input": topic,
        #         "image_information": self.image_summaries
        #     })
            sys.stdout = sys.__stdout__
            captured_content = str(captured_output.getvalue())
            captured_content = re.sub(r'\x1b\[[0-9;]*m', '', captured_content)

            # Extract explanation from captured output
            explain = ""
            try:
                match = re.search(r'\{.*\}(.*)', captured_content, re.DOTALL)
                if match:
                    extracted_content = match.group(1).strip()
                    match2 = re.search(r'Summary:(.*)Final Answer:', extracted_content, re.DOTALL)
                    if match2:
                        explain = match2.group(1).strip()
            except Exception as e:
                print(f"Error extracting explanation: {e}")
            # Format response
            text = re.sub(r'<.*?>','',response1['output'].replace('*','').strip()) + '<br><br>' +re.sub(r'<.*?>','',response2['output'].replace('*','').strip())
            reply_explain = {
                'text': text,
                'news': self.process_news_results(response1['intermediate_steps']) + self.process_news_results(response2['intermediate_steps'])
            }
            print(response1['intermediate_steps'])
            formatted_reply = FactChecker.format_reply(reply_explain)
            return [formatted_reply] + self.extract_agent_thoughts(response1['intermediate_steps']) 
        except ValueError as e:
            if "DataInspectionFailed" in str(e):
                return "您的输入可能包含敏感内容，请重新描述。"
            else:
                return "出现错误，请稍后再试。"

class History:
    @staticmethod
    def generate_chat_title(conversation: list[MessageDict]) -> str:
        title = ""
        for message in conversation:
            if message["role"] == "user":
                if isinstance(message["content"], str):
                    title += message["content"]
                    break
                else:
                    title += "📎 "
        if len(title) > 40:
            title = title[:40] + "..."

        # print(title)
        return title or "Conversation"

    @staticmethod
    def load_chat_history(conversations):
        return gr.Dataset(
            samples=[
                [History.generate_chat_title(conv)]
                for conv in conversations or []
                if conv
            ]
        )
    @staticmethod
    def save_conversation(
            index: int | None,
            conversation: list[MessageDict],
            saved_conversations: list[list[MessageDict]],
    ):
        if index is not None:
            saved_conversations[index] = conversation
        else:
            saved_conversations.append(conversation)
            index = len(saved_conversations) - 1
        return index, saved_conversations

    @staticmethod
    def load_conversation(
            index: int,
            conversations: list[list[MessageDict]],
    ):
        return (
            index,
            gr.Chatbot(
                value=conversations[index],  # type: ignore
                feedback_value=[],
                type="messages"
            ),
        )


def build_interface() -> gr.Blocks:
    with gr.Blocks(css=css, theme='soft') as demo:
        gr.HTML(
            "<h1 style='font-size: 36px; text-align: center; color: #333333; margin-bottom: 20px;'>惊堂木-虚假信息检测系统</h1>")
        with gr.Tab(label='Chat'):
            with gr.Row():
                with gr.Sidebar():
                    with gr.Column(scale=1):
                        gr.Textbox(visible=False)
                    with gr.Column(scale=1):
                        gr.Textbox(visible=False)
                    with gr.Column(scale=1):
                        new_chat_button = gr.Button(
                            "New chat",
                            variant="primary",
                            size="md",
                            icon="plus.svg",
                        )
                        chat_history_dataset = gr.Dataset(
                            components=[gr.Textbox(visible=False)],
                            show_label=False,
                            layout="table",
                            type="index",
                        )

                        language_select = gr.Dropdown(["中文", "英文"], label="请选择要使用的语言", scale=1, value="中文")
                        # model_select = gr.Dropdown(["自研", "llama3-70b", "llama3-8b", "mistral"],
                        #                            label="请选择要使用的大模型",
                        #                            scale=1, value="自研")
                        mode_select = gr.Dropdown(["否","是"],label= "是否过滤可信信源",scale= 1, value= "否")
                        deep_select = gr.Dropdown(["轻度检测","普通检测","深度检测"],label= "选择检测模式",scale= 1, value= "普通检测")

                with gr.Column(scale=3):
                    bot = gr.ChatInterface(
                        fn=factChecker.check_facts_chat,
                        examples=[
                            {
                                "text": "9月19日，马来西亚最高元首 Ibrahim 应邀对中国进行为期8天国事访问，亦是2024年1月上任以来首次访问东盟外国家。"},
                            {
                                "text": "据最新天文研究，火星的轨道将逐渐接近地球，最终成为地球的“第二月亮”。天文学家预测这一变化将在2025年发生，届时火星将在夜空中与月亮一样明亮，影响全球潮汐和生态平衡。"}
                        ],
                        chatbot=gr.Chatbot(label='惊堂木',
                                           avatar_images=("./image/user.png", "./image/logo.png"),
                                           type="messages",
                                           height=600,
                                           layout="bubble",
                                           show_copy_button=True,
                                           show_copy_all_button=True,
                                           resizable=True
                                           ),
                        multimodal=True,
                        show_progress='full',
                        type="messages",
                        flagging_mode='manual',
                        cache_examples=False,
                        example_icons=["./image/search.png", "./image/search.png"],

                    )

                with gr.Column(scale=1):
                    with gr.Accordion("添加可信信源", open=False):
                        # img_info = gr.Textbox(label="分析结果", lines=5)
                        trusted_source = gr.Textbox(label="请输入需要添加的可信信源",interactive=True,show_copy_button=True,placeholder="请用回车隔开，输入示例：url1\nurl2",lines=5)
                    # contrast_model = gr.Textbox(label="对比模型", lines=5)
                    # with gr.Accordion("展示对比模型prompt", open=False):
                    #     model_prompt = gr.Textbox(label="对比模型prompt", lines=5, placeholder=constract_model_prompt, interactive=True)
                    # contrast_select = gr.Dropdown(["qwen", "llama", "glm", "doubao", "deepseek", "baichuan"],
                    #                                label="请选择使用的对比模型", scale=1, value="qwen")
                    # contrast_bn = gr.Button("展示对比模型")
                    # prompt_bn = gr.Button("确定更改提示词")
                    trusted_source_bn = gr.Button("确定添加可信信源")
            # img_bn.click(factChecker.update_image_summaries, [], img_info)
            language_select.change(factChecker.SelectLanguage, language_select, [])
            # model_select.change(factChecker.SelectModel, model_select, [])
            mode_select.change(factChecker.SelectMode, mode_select, [])
            trusted_source_bn.click(factChecker.AddTrustedSource,trusted_source,trusted_source)
            deep_select.change(factChecker.SelectDeep, deep_select, [])
            # contrast_select.change(factChecker.SelectContrast, contrast_select, [])
            # contrast_bn.click(factChecker.contract, [], contrast_model)
            # prompt_bn.click(factChecker.ModelPrompt,model_prompt, [])
            new_chat_button.click(
                lambda x: x,
                [bot.chatbot],
                [bot.chatbot_state],
                show_api=False,
                queue=False,
            ).then(
                History.save_conversation,
                [bot.conversation_id, bot.chatbot_state, bot.saved_conversations],
                [bot.conversation_id, bot.saved_conversations]
            ).then(
                lambda: (None, []),
                None,
                [bot.conversation_id, bot.chatbot],
                show_api=False,
                queue=False,
            ).then(
                lambda x: x,
                [bot.chatbot],
                [bot.chatbot_state],
                show_api=False,
                queue=False,
            )
            on(
                triggers=[demo.load, bot.saved_conversations.change],
                fn=History.load_chat_history,
                inputs=bot.saved_conversations,
                outputs=chat_history_dataset,
                show_api=False,
                queue=False,
            )
            chat_history_dataset.click(
                lambda: [],
                None,
                [bot.chatbot],
                show_api=False,
                queue=False,
                show_progress="hidden",
            ).then(
                History.load_conversation,
                [chat_history_dataset, bot.saved_conversations],
                [bot.conversation_id, bot.chatbot],
                show_api=False,
                queue=False,
                show_progress="hidden",
            )

        return demo

if __name__ == "__main__":
    factChecker = FactChecker()
    demo = build_interface()
    demo.launch()