Upload 13 files

app.py

@@ -0,0 +1,367 @@
+import streamlit as st
+import os
+import time
+from PIL import Image
+
+# 全局缓存 Whisper 语音大模型
+@st.cache_resource
+def load_whisper_model():
+    import whisper
+    return whisper.load_model("small")
+
+# 页面配置
+st.set_page_config(
+    page_title="多模态AI生成痕迹鉴别系统",
+    page_icon="🔍",
+    layout="wide"
+)
+
+# ==========================================
+# 侧边栏：系统设置与说明
+# ==========================================
+with st.sidebar:
+    st.title("⚙️ 鉴别引擎设置")
+    st.markdown("提供轻量、可解释的多模态AI生成内容快速鉴别能力。")
+    st.divider()
+
+    st.header("🎚️ 动态判定阈值")
+    st.markdown("调整敏感度，适应不同审核场景：")
+    high_risk_threshold = st.slider(
+        "高危报警阈值",
+        min_value=0.60, max_value=0.95, value=0.80, step=0.05,
+        help="高于此值判定为“高度疑似AI生成”"
+    )
+    warning_threshold = st.slider(
+        "存疑缓冲阈值",
+        min_value=0.20, max_value=0.55, value=0.40, step=0.05,
+        help="介于警告阈值与报警阈值之间时，标记为“存疑”"
+    )
+
+    st.divider()
+    st.caption("🔹 引擎状态：轻量模型已缓存 · 本地推理")
+
+# ==========================================
+# 主界面标题
+# ==========================================
+st.title("🔍 多模态AI生成痕迹快速鉴别系统")
+st.markdown(
+    "同时支持**图像、文本、视频**单模态扫描，以及**图文/视文**联合跨模态融合鉴定。"
+)
+
+work_mode = st.radio(
+    "请选择检测模式：",
+    ["单模态独立检测（图片 / 文本 / 视频）", "多模态联合检测（图片+文案 / 视频+文案）"],
+    horizontal=True
+)
+
+st.divider()
+
+# ==========================================
+# 模式一：单模态独立检测
+# ==========================================
+if "单模态独立检测" in work_mode:
+    with st.container(border=True):
+        uploaded_file = st.file_uploader(
+            "拖拽或点击上传待检测文件",
+            type=["jpg", "png", "jpeg", "txt", "docx", "mp4", "avi", "mov"],
+            help="支持常见图片、文本、视频格式，单文件≤50MB，视频时长建议≤5分钟"
+        )
+
+    if uploaded_file is not None:
+        file_type = uploaded_file.name.split('.')[-1].lower()
+        st.success(f"✅ 已接收：**{uploaded_file.name}** | 启动鉴别引擎...")
+
+        col_content, col_result = st.columns([1, 1.2], gap="large")
+
+        # ----------------- 图像检测 -----------------
+        if file_type in ['jpg', 'png', 'jpeg']:
+            with col_content:
+                st.subheader("原始图像")
+                st.image(uploaded_file, use_container_width=True)
+
+                st.subheader("AI痕迹热力图（Grad-CAM）")
+                heatmap_placeholder = st.empty()
+
+            with col_result:
+                st.subheader("⚙️ 特征提取中...")
+                progress_bar = st.progress(0)
+                status_text = st.empty()
+
+                for percent_complete in range(100):
+                    time.sleep(0.01)
+                    progress_bar.progress(percent_complete + 1)
+                    status_text.text(f"正在提取LBP纹理、频域伪影及深度语义... {percent_complete + 1}%")
+
+                from image_module import analyze_image, load_deep_image_model, generate_image_heatmap
+
+                result = analyze_image(uploaded_file)
+
+                model, device = load_deep_image_model()
+                # 图像模块内部我们已经处理过 seek(0)，所以这里可以直接传
+                heatmap_img = generate_image_heatmap(uploaded_file, model, device)
+                heatmap_placeholder.image(
+                    heatmap_img,
+                    caption="🔴 红色区域为模型判定的高可疑伪造区域",
+                    use_container_width=True
+                )
+
+                status_text.text("特征提取完成")
+
+                st.subheader("检测报告")
+                final_prob = result['final_probability']
+
+                if final_prob >= high_risk_threshold:
+                    st.error(f"高度疑似AI生成图像（生成概率：{final_prob * 100:.1f}%）")
+                elif warning_threshold <= final_prob < high_risk_threshold:
+                    st.warning(f"可疑图像，可能经过AI修图或局部生成（概率：{final_prob * 100:.1f}%）")
+                else:
+                    st.success(f"真实图像，未检出明显生成痕迹（概率：{final_prob * 100:.1f}%）")
+
+                st.progress(float(final_prob))
+                with st.expander("展开底层特征参数", expanded=True):
+                    st.info(
+                        f"**双轨特征得分：**\n\n"
+                        f"- 传统��理特征异常度：{result['traditional_score'] * 100:.1f}%\n"
+                        f"- MobileNetV2 深度特征：{result['deep_score'] * 100:.1f}%"
+                    )
+
+        # ----------------- 文本检测 -----------------
+        elif file_type in ['txt', 'docx']:
+            text_content = ""
+            uploaded_file.seek(0) # 核心修复：倒带文件指针
+            if file_type == 'txt':
+                text_content = uploaded_file.read().decode("utf-8")
+            elif file_type == 'docx':
+                import docx
+                doc = docx.Document(uploaded_file)
+                text_content = "\n".join([para.text for para in doc.paragraphs])
+
+            with col_content:
+                st.subheader("原始文本内容")
+                text_placeholder = st.empty()
+                text_placeholder.text_area("提取的文本", text_content, height=350)
+
+            with col_result:
+                st.subheader("⚙️ 语义连贯性分析中...")
+                with st.spinner("正在计算困惑度、句法复杂度及BERT深度特征..."):
+                    from text_module import analyze_text, get_custom_text_model, generate_text_highlight_html
+
+                    result = analyze_text(text_content)
+
+                    tokenizer, model = get_custom_text_model()
+                    highlighted_html = generate_text_highlight_html(text_content, tokenizer, model)
+                    text_placeholder.markdown("**🔥 AI痕迹逐句高亮（黄色为高风险句式）**", unsafe_allow_html=True)
+                    text_placeholder.markdown(highlighted_html, unsafe_allow_html=True)
+
+                st.subheader("检测报告")
+                final_prob = result['final_probability']
+
+                if final_prob >= high_risk_threshold:
+                    st.error(f"高度疑似大语言模型生成文本（生成概率：{final_prob * 100:.1f}%）")
+                elif warning_threshold <= final_prob < high_risk_threshold:
+                    st.warning(f"可疑文本，可能经AI润色或拼接（概率：{final_prob * 100:.1f}%）")
+                else:
+                    st.success(f"真实人类写作风格，低风险（概率：{final_prob * 100:.1f}%）")
+
+                st.progress(float(final_prob))
+                with st.expander("展开底层特征参数", expanded=True):
+                    st.info(
+                        f"**多维度文本特征：**\n\n"
+                        f"- 统计学风格异常度：{result['stat_score'] * 100:.1f}%\n"
+                        f"- DistilBERT 语义鉴别得分：{result['deep_score'] * 100:.1f}%\n"
+                        f"- 补充指标：{result['details']}"
+                    )
+
+        # ----------------- 视频检测 -----------------
+        elif file_type in ['mp4', 'avi', 'mov']:
+            with col_content:
+                st.subheader("视频内容预览")
+                st.video(uploaded_file)
+
+            with col_result:
+                st.subheader("⚙️ 时空特征提取中...")
+                with st.spinner("正在进行关键帧抽帧、光流计算及音频频谱分析..."):
+                    temp_video_path = f"temp_uploaded_video.{file_type}"
+                    with open(temp_video_path, "wb") as f:
+                        uploaded_file.seek(0) # 核心修复：倒带文件指针
+                        f.write(uploaded_file.read())
+
+                    from video_module import analyze_video
+                    result = analyze_video(temp_video_path)
+
+                    if os.path.exists(temp_video_path):
+                        os.remove(temp_video_path)
+
+                if "error" in result:
+                    st.error(result["error"])
+                else:
+                    st.subheader("检测报告")
+                    final_prob = result['avg_probability']
+
+                    if final_prob >= high_risk_threshold:
+                        st.error(f"整体视频高度疑似AI生成（综合概率：{final_prob * 100:.1f}%）")
+                    elif warning_threshold <= final_prob < high_risk_threshold:
+                        st.warning(f"视频存疑，存在异常帧或拼接痕迹（概率：{final_prob * 100:.1f}%）")
+                    else:
+                        st.success(f"未发现明显时序伪造痕迹，低风险（概率：{final_prob * 100:.1f}%）")
+
+                    st.progress(float(final_prob))
+                    with st.expander("展开抽帧分析详情", expanded=True):
+                        st.info(
+                            f"**视频物理特征：**\n\n"
+                            f"- 总帧数：{result['total_frames']} 帧 (帧率 {result['fps']:.1f} fps)\n"
+                            f"- 关键帧采样数：{result['sampled_frames']} 帧\n"
+                            f"- 单帧最高风险值：{result['max_probability'] * 100:.1f}%"
+                        )
+
+# ==========================================
+# 模式二：多模态联合检测
+# ==========================================
+elif "多模态联合检测" in work_mode:
+    st.subheader("🔗 跨模态联合检测场景")
+    st.markdown(
+        "模拟真实审核场景：同时检测**图像/视频**与**配套文本**，通过决策级加权融合输出综合AI生成概率。"
+    )
+
+    # 初始化session_state记忆体（用于自动提取的文本）
+    if 'auto_text' not in st.session_state:
+        st.session_state['auto_text'] = ""
+
+    col_media, col_text = st.columns(2, gap="large")
+
+    with col_media:
+        multi_media = st.file_uploader(
+            "🖼️ / 🎞️ 第一步：上传媒体内容（图片或短视频）",
+            type=["jpg", "png", "jpeg", "mp4", "avi", "mov"],
+            key="multi_media"
+        )
+        file_type = ""
+        if multi_media:
+            file_type = multi_media.name.split('.')[-1].lower()
+            if file_type in ['mp4', 'avi', 'mov']:
+                st.video(multi_media)
+
+                # 智能语音提取按钮
+                if st.button("🎙️ 自动从视频提取语音转文字", use_container_width=True):
+                    with st.spinner("正在加载Whisper模型并转录音频（若视频较长请耐心等待数分钟）..."):
+                        try:
+                            from moviepy import VideoFileClip
+                            import whisper
+
+                            # 1. 保存临时视频
+                            temp_vid = "temp_for_audio.mp4"
+                            with open(temp_vid, "wb") as f:
+                                multi_media.seek(0)
+                                f.write(multi_media.read())
+
+                            # 2. 读取并剥离音频
+                            temp_audio = "temp_audio.wav"
+                            my_clip = VideoFileClip(temp_vid)
+
+                            # 【新增防护】检查视频到底有没有声音！
+                            if my_clip.audio is None:
+                                st.error("❌ 提取失败：检测到该视频没有声音轨道！")
+                            else:
+                                my_clip.audio.write_audiofile(temp_audio, logger=None)
+                                my_clip.close()
+
+                                # 3. 语音识别
+                                model = load_whisper_model()
+                                result = model.transcribe(
+                                    temp_audio,
+                                    language="zh",
+                                    initial_prompt="以下是一段标准的简体中文普通话录音。",
+                                    fp16=False
+                                )
+
+                                st.session_state['auto_text'] = result["text"]
+                                st.success("✅ 提取成功！")
+                                time.sleep(1)  # 停留1秒让用户看到成功提示
+                                st.rerun()
+
+                        except Exception as e:
+                            # 【新增防护】无论出什么错，直接弹在网页上！
+                            st.error(f"❌ 提取失败，底层报错信息：{str(e)}")
+                            st.info(
+                                "💡 提示：如果看到 'ffprobe' 或 'ffmpeg' 等字眼，请确保系统已通过 conda 安装了 ffmpeg。")
+
+                        finally:
+                            # 【新增防护】哪怕中途崩溃，也要把占硬盘的临时文件删掉
+                            if os.path.exists(temp_vid):
+                                try:
+                                    os.remove(temp_vid)
+                                except:
+                                    pass
+                            if os.path.exists(temp_audio):
+                                try:
+                                    os.remove(temp_audio)
+                                except:
+                                    pass
+            else:
+                st.image(multi_media, use_container_width=True)
+
+    with col_text:
+        multi_txt = st.text_area(
+            "📝 第二步：输入配套文本（或点击左侧自动提取）",
+            value=st.session_state['auto_text'],
+            height=200,
+            key="multi_txt"
+        )
+
+    # 联合检测按钮
+    if multi_media and multi_txt:
+        if st.button("🔍 启动跨模态融合鉴别", type="primary", use_container_width=True):
+            st.divider()
+            st.subheader("📊 联合检测报告")
+
+            from text_module import analyze_text
+
+            with st.spinner("并行调用视觉鉴别引擎与文本鉴别引擎..."):
+                file_type = multi_media.name.split('.')[-1].lower()
+                p_media = 0.0
+                media_label = ""
+
+                if file_type in ['mp4', 'avi', 'mov']:
+                    from video_module import analyze_video
+                    temp_path = f"temp_multi_video.{file_type}"
+                    with open(temp_path, "wb") as f:
+                        multi_media.seek(0) # 核心修复：倒带文件指针
+                        f.write(multi_media.read())
+                    res_media = analyze_video(temp_path)
+                    p_media = res_media.get('avg_probability', 0)
+                    media_label = "🎞️ 视频维度AI生成概率"
+                    if os.path.exists(temp_path):
+                        os.remove(temp_path)
+                else:
+                    from image_module import analyze_image
+                    res_media = analyze_image(multi_media)
+                    p_media = res_media['final_probability']
+                    media_label = "🖼️ 图像维度AI生成概率"
+
+                # 文本检测
+                res_txt = analyze_text(multi_txt)
+                p_txt = res_txt['final_probability']
+
+                # 融合权重（视觉60%，文本40%）
+                joint_prob = (p_media * 0.60) + (p_txt * 0.40)
+
+            # 展示三指标
+            col_d1, col_d2, col_d3 = st.columns(3)
+            col_d1.metric(label=media_label, value=f"{p_media * 100:.1f}%")
+            col_d2.metric(label="📝 文本维度AI生成概率", value=f"{p_txt * 100:.1f}%")
+            col_d3.metric(
+                label="🔗 多模态综合判定概率",
+                value=f"{joint_prob * 100:.1f}%",
+                delta="加权融合 (视觉0.6 + 文本0.4)",
+                delta_color="off"
+            )
+
+            st.progress(float(joint_prob))
+
+            # 综合判定
+            if joint_prob >= high_risk_threshold:
+                st.error("🚨 **联合研判结论：高度疑似AI生成的多模态内容**（图像/视频与文本均呈现显著生成特征）")
+            elif warning_threshold <= joint_prob < high_risk_threshold:
+                st.warning("⚠️ **联合研判结论：内容存疑**，可能存在局部AI修改或跨模态不一致，建议人工复核")
+            else:
+                st.success("✅ **联合研判结论：内容安全**，未见明显多模态伪造痕迹")

download_data.py

@@ -0,0 +1,46 @@
+from datasets import load_dataset
+import os
+
+print("正在连接 Hugging Face 并提取高清数据集...")
+dataset = load_dataset("Parveshiiii/AI-vs-Real", split="train", verification_mode="no_checks")
+
+label_col = 'binary_label'
+print(f"✅ 成功锁定标签列：'{label_col}'\n")
+
+os.makedirs("./data/real_batch", exist_ok=True)
+os.makedirs("./data/ai_batch", exist_ok=True)
+
+# 设置我们想要的配额
+target_count = 2000
+real_saved = 0
+ai_saved = 0
+
+print(f"开始精准抓取：目标 {target_count}张真实图 + {target_count}张AI图...")
+
+# 遍历整个数据集，直到两个配额都装满
+for item in dataset:
+    img = item['image']
+    label = item[label_col]
+
+    # 工业级防御：统一转为 RGB 格式
+    if img.mode != 'RGB':
+        img = img.convert('RGB')
+
+    # 分流并计数
+    if label == 1 and real_saved < target_count:
+        img.save(f"./data/real_batch/real_hd_{real_saved}.jpg")
+        real_saved += 1
+
+    elif label == 0 and ai_saved < target_count:
+        img.save(f"./data/ai_batch/ai_hd_{ai_saved}.jpg")
+        ai_saved += 1
+
+    # 每抓够 100 张，在终端汇报一下进度，让你心里有底
+    if (real_saved + ai_saved) % 100 == 0:
+        print(f"当前进度 -> 真实图: {real_saved}/{target_count} | AI图: {ai_saved}/{target_count}")
+
+    # 如果两边的配额都满了，就停止遍历
+    if real_saved >= target_count and ai_saved >= target_count:
+        break
+
+print("\n🎉 完美收工！这次两个文件夹绝对都塞得满满当当了！")

download_text_data.py

@@ -0,0 +1,42 @@
+from datasets import load_dataset
+import pandas as pd
+import os
+
+print("正在连接 Hugging Face 拉取 HC3 中文语料库...")
+# 使用 verification_mode="no_checks" 防止部分网络下元数据校验报错
+dataset = load_dataset("Hello-SimpleAI/HC3-Chinese", name="all", split="train", verification_mode="no_checks", trust_remote_code=True)
+
+human_texts = []
+ai_texts = []
+target_count = 500  # 咱们先各抓 500 条用来快速跑通流程
+
+print(f"开始清洗数据，目标：提取 {target_count}条人类文本 + {target_count}条AI文本...")
+
+for item in dataset:
+    # 提取人类回答 (剔除太短的废话)
+    if len(item['human_answers']) > 0 and len(item['human_answers'][0]) > 20:
+        if len(human_texts) < target_count:
+            human_texts.append(item['human_answers'][0])
+
+    # 提取 ChatGPT 回答
+    if len(item['chatgpt_answers']) > 0 and len(item['chatgpt_answers'][0]) > 20:
+        if len(ai_texts) < target_count:
+            ai_texts.append(item['chatgpt_answers'][0])
+
+    if len(human_texts) >= target_count and len(ai_texts) >= target_count:
+        break
+
+# 制作结构化的 DataFrame (标签规则统一：0代表AI，1代表真实)
+df_human = pd.DataFrame({'text': human_texts, 'label': 1})
+df_ai = pd.DataFrame({'text': ai_texts, 'label': 0})
+
+# 合并并打乱顺序 (Shuffle)
+df_all = pd.concat([df_human, df_ai]).sample(frac=1, random_state=42).reset_index(drop=True)
+
+# 确保 data 文件夹存在
+os.makedirs("./data", exist_ok=True)
+csv_path = "./data/text_dataset.csv"
+df_all.to_csv(csv_path, index=False, encoding='utf-8')
+
+print(f"✅ 语料库构建完毕！已保存为 {csv_path} (共 {len(df_all)} 条数据)")
+print("你可以打开这个 csv 文件看看，里面的文本非常有意思！")

image_module.py

@@ -0,0 +1,172 @@
+import cv2
+import numpy as np
+import torch
+import torch.nn as nn
+from torchvision import models, transforms
+from skimage import feature
+from scipy.fftpack import dct
+from PIL import Image, ImageOps
+import streamlit as st
+
+# 引入热力图相关库
+from pytorch_grad_cam import GradCAM
+from pytorch_grad_cam.utils.model_targets import ClassifierOutputTarget
+
+
+# ==========================================
+# 辅助函数：智能读取图像（修正 EXIF 方向）
+# ==========================================
+def get_pil_image(image_input):
+    if isinstance(image_input, str):
+        img = Image.open(image_input)
+    else:
+        image_input.seek(0)
+        img = Image.open(image_input)
+
+    # 强制应用隐藏的 EXIF 旋转信息，还原真实方向
+    img = ImageOps.exif_transpose(img)
+    return img.convert('RGB')
+
+
+def get_cv_image(image_input):
+    pil_img = get_pil_image(image_input)
+    # 保证 OpenCV 提取物理特征时，方向与 PIL 绝对一致
+    return cv2.cvtColor(np.array(pil_img), cv2.COLOR_RGB2BGR)
+
+
+# ==========================================
+# 第一部分：传统物理特征检测 (权重 15%)
+# ==========================================
+def extract_traditional_features(image_input):
+    img = get_cv_image(image_input)
+    if img is None:
+        return 0.0
+
+    gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)
+
+    # 1. LBP 局部二值模式
+    lbp = feature.local_binary_pattern(gray, P=8, R=1, method="uniform")
+    hist, _ = np.histogram(lbp.ravel(), bins=np.arange(0, 11), density=True)
+    lbp_entropy = -np.sum(hist * np.log2(hist + 1e-7))
+
+    # 2. DCT 离散余弦变换
+    dct_data = dct(dct(gray.T, norm='ortho').T, norm='ortho')
+    high_freq = np.sum(np.abs(dct_data[10:, 10:]))
+    total_energy = np.sum(np.abs(dct_data))
+    dct_ratio = high_freq / (total_energy + 1e-7)
+
+    # 3. FFT 快速傅里叶变换
+    f = np.fft.fft2(gray)
+    fshift = np.fft.fftshift(f)
+    magnitude = 20 * np.log(np.abs(fshift) + 1)
+    h, w = magnitude.shape
+    center_h, center_w = h // 2, w // 2
+    top_left = magnitude[0:center_h, 0:center_w]
+    bottom_right = np.flip(magnitude[center_h + (h % 2):, center_w + (w % 2):])
+
+    if top_left.shape != bottom_right.shape:
+        bottom_right = cv2.resize(bottom_right, (top_left.shape[1], top_left.shape[0]))
+
+    fft_sym_error = np.mean(np.abs(top_left - bottom_right))
+
+    score = 0.0
+    if lbp_entropy > 3.6: score += 0.3
+    if dct_ratio < 0.985: score += 0.1
+    if fft_sym_error > 13.8: score += 0.1
+
+    return min(score, 1.0)
+
+
+# ==========================================
+# 第二部分：深度模型缓存与提取
+# ==========================================
+@st.cache_resource
+def load_deep_image_model():
+    device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
+    model = models.mobilenet_v2(weights=None)
+    num_ftrs = model.classifier[1].in_features
+    model.classifier[1] = nn.Linear(num_ftrs, 2)
+
+    # 加载自己炼丹生成的权重文件
+    model.load_state_dict(torch.load('mobilenet_finetuned.pth', map_location=device))
+    model = model.to(device)
+    model.eval()
+
+    return model, device
+
+
+def extract_deep_features(image_input, model, device):
+    pil_img = get_pil_image(image_input)
+    transform = transforms.Compose([
+        transforms.Resize((224, 224)),
+        transforms.ToTensor(),
+        transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225])
+    ])
+    input_tensor = transform(pil_img).unsqueeze(0).to(device)
+
+    with torch.no_grad():
+        outputs = model(input_tensor)
+        probs = torch.softmax(outputs, dim=1)
+        fake_prob = probs[0][0].item()
+
+    return fake_prob
+
+
+# ==========================================
+# 第三部分：多模态加权融合引擎
+# ==========================================
+def analyze_image(image_input):
+    trad_score = extract_traditional_features(image_input)
+    model, device = load_deep_image_model()
+    deep_score = extract_deep_features(image_input, model, device)
+
+    final_prob = (trad_score * 0.15) + (deep_score * 0.85)
+
+    return {
+        'traditional_score': trad_score,
+        'deep_score': deep_score,
+        'final_probability': final_prob
+    }
+
+
+# ==========================================
+# 第四部分：XAI 热力图渲染引擎
+# ==========================================
+def generate_image_heatmap(image_input, model, device):
+    """使用 Grad-CAM 生成热力图，并无损拉伸回原图尺寸，带动态透明度"""
+    # ⚠️ 获取原图，绝对没有任何 resize 操作干扰！
+    original_pil_img = get_pil_image(image_input)
+    orig_width, orig_height = original_pil_img.size
+
+    transform = transforms.Compose([
+        transforms.Resize((224, 224)),
+        transforms.ToTensor(),
+        transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225])
+    ])
+    input_tensor = transform(original_pil_img).unsqueeze(0).to(device)
+
+    target_layers = [model.features[-1]]
+
+    try:
+        with GradCAM(model=model, target_layers=target_layers) as cam:
+            targets = [ClassifierOutputTarget(0)]
+            # 跑出 224x224 的 0~1 原始热力图矩阵
+            grayscale_cam_224 = cam(input_tensor=input_tensor, targets=targets)[0, :]
+
+        # 将 224x224 的热力图拉伸回真实的高清原图宽高！
+        grayscale_cam_resized = cv2.resize(grayscale_cam_224, (orig_width, orig_height))
+
+        heatmap_color = cv2.applyColorMap(np.uint8(255 * grayscale_cam_resized), cv2.COLORMAP_JET)
+        heatmap_color = cv2.cvtColor(heatmap_color, cv2.COLOR_BGR2RGB)
+        heatmap_color = np.float32(heatmap_color) / 255.0
+
+        orig_img_array = np.array(original_pil_img, dtype=np.float32) / 255.0
+
+        alpha = grayscale_cam_resized[..., np.newaxis]
+        alpha = np.power(alpha, 1.5) * 0.65
+
+        visualization = orig_img_array * (1 - alpha) + heatmap_color * alpha
+        return np.uint8(255 * visualization)
+
+    except Exception as e:
+        return np.array(original_pil_img)

mobilenet_finetuned.pth

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:22343e127cd0d6a43a5cb78176171b808a3956cd8dfac8db799b77f753e9620c
+size 9149899

packages.txt

@@ -0,0 +1 @@
+ffmpeg

requirements.txt

@@ -1,3 +1,12 @@
-altair
-pandas
-streamlit
+streamlit==1.32.0
+torch
+torchvision
+transformers
+opencv-python-headless
+moviepy
+openai-whisper
+pytorch-grad-cam
+scikit-image
+scipy
+Pillow
+numpy

test.py

@@ -0,0 +1,2 @@
+import streamlit as st
+st.title("如果能看到这行字，说明环境没问题！")

text_module.py

@@ -0,0 +1,116 @@
+import jieba
+import torch
+import re
+import numpy as np
+import streamlit as st
+from transformers import AutoTokenizer, AutoModelForSequenceClassification
+
+
+# ==========================================
+# 第一部分：统计风格特征提取 (占权重 20%)
+# ==========================================
+def extract_text_statistics(text):
+    """
+    提取文本的统计学与风格特征。
+    """
+    words = list(jieba.cut(text))
+    words = [w for w in words if w.strip() and len(w) > 0]
+    if len(words) == 0: return 0.0, 0, 0
+
+    unique_words = set(words)
+    richness = len(unique_words) / len(words)
+
+    sentences = re.split(r'[。！？\n]', text)
+    sentences = [s for s in sentences if len(s.strip()) > 0]
+
+    sentence_lengths = [len(s) for s in sentences]
+    avg_sentence_len = np.mean(sentence_lengths) if sentence_lengths else 0
+    sentence_len_std = np.std(sentence_lengths) if sentence_lengths else 0
+
+    score = 0.0
+    if richness < 0.55:
+        score += 0.5
+    if sentence_len_std < 10.0 and avg_sentence_len > 15:
+        score += 0.5
+
+    return min(score, 1.0), richness, sentence_len_std
+
+
+# ==========================================
+# 第二部分：深度语义特征提取 (换上我们自己微调的专属模型)
+# ==========================================
+@st.cache_resource
+def get_custom_text_model():
+    """加载我们刚刚炼丹微调出来的专属 BERT 模型"""
+    # 直接指向你刚刚生成的那个文件夹
+    model_path = "./finetuned_text_model"
+    tokenizer = AutoTokenizer.from_pretrained(model_path)
+    model = AutoModelForSequenceClassification.from_pretrained(model_path)
+    model.eval()
+    return tokenizer, model
+
+
+def extract_deep_text_features(text, tokenizer, model):
+    """使用专属模型进行推理打分"""
+    inputs = tokenizer(text, return_tensors="pt", truncation=True, max_length=128)
+    with torch.no_grad():
+        outputs = model(**inputs)
+
+    # 提取输出并转换为概率分布 (0: AI, 1: 真实)
+    logits = outputs.logits
+    probs = torch.softmax(logits, dim=1)
+
+    # 获取它被判定为 AI (标签0) 的概率百分比
+    fake_prob = probs[0][0].item()
+    return fake_prob
+
+
+# ==========================================
+# 第三部分：多模态加权融合
+# ==========================================
+def analyze_text(text_content):
+    # 1. 统计特征
+    stat_score, richness, std_len = extract_text_statistics(text_content)
+
+    # 2. 深度特征 (加载你的专属模型)
+    tokenizer, model = get_custom_text_model()
+    deep_score = extract_deep_text_features(text_content, tokenizer, model)
+
+    # 3. 加权融合 (既然我们自己炼了丹，深度模型变得极强，权重拉高到 80%)
+    final_prob = (stat_score * 0.2) + (deep_score * 0.8)
+
+    return {
+        "stat_score": stat_score,
+        "deep_score": deep_score,
+        "final_probability": final_prob,
+        "details": f"词汇丰富度: {richness:.2f} | 句长波动率: {std_len:.1f}"
+    }
+
+
+def generate_text_highlight_html(text, tokenizer, model):
+    """逐句扫描文本，生成带有高亮背景色的 HTML 代码"""
+    # 按照标点符号将文章拆分为句子
+    sentences = re.split(r'([。！？\n])', text)
+    # 重新把句子和标点拼装起来
+    sentences = ["".join(i) for i in zip(sentences[0::2], sentences[1::2] + [""])]
+
+    html_content = '<div style="line-height: 1.6; font-size: 16px; padding: 10px; border: 1px solid #ddd; border-radius: 5px;">'
+
+    for s in sentences:
+        if len(s.strip()) < 2:
+            html_content += s.replace('\n', '<br>')
+            continue
+
+        # 让专属模型给这一句话单独打分
+        prob = extract_deep_text_features(s, tokenizer, model)
+
+        # 只有当 AI 概率大于 50% 时才开始变红，概率越高颜色越深
+        if prob > 0.5:
+            alpha = (prob - 0.5) * 2  # 映射到 0~1 透明度
+            color = f"rgba(255, 75, 75, {alpha:.2f})"
+            html_content += f'<span style="background-color: {color}; border-radius: 3px;">{s}</span>'
+        else:
+            html_content += s
+
+    html_content += "</div>"
+    return html_content

train_image_model.py

@@ -0,0 +1,81 @@
+import os
+import torch
+import torch.nn as nn
+import torch.optim as optim
+from torchvision import datasets, models, transforms
+from torch.utils.data import DataLoader
+from tqdm import tqdm  # 引入了实时进度条神器
+
+
+def train_model():
+    device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
+    print(f"当前使用的计算设备: {device}")
+
+    if device.type == 'cpu':
+        print("⚠️ 警告：当前正在使用 CPU 训练，4000张图片预计每轮需要 15-30 分钟，请保持耐心！")
+
+    data_transforms = transforms.Compose([
+        transforms.Resize((224, 224)),
+        transforms.RandomHorizontalFlip(),
+        transforms.ToTensor(),
+        transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225])
+    ])
+
+    data_dir = './data'
+    image_dataset = datasets.ImageFolder(data_dir, data_transforms)
+    # CPU 训练比较慢，我们把 batch_size 稍微调大一点点到 16
+    dataloader = DataLoader(image_dataset, batch_size=16, shuffle=True)
+
+    print(f"总计训练图片数量: {len(image_dataset)} 张\n")
+
+    print("正在加载 MobileNetV2 模型...")
+    model = models.mobilenet_v2(weights=models.MobileNet_V2_Weights.IMAGENET1K_V1)
+
+    num_ftrs = model.classifier[1].in_features
+    model.classifier[1] = nn.Linear(num_ftrs, 2)
+    model = model.to(device)
+
+    criterion = nn.CrossEntropyLoss()
+    optimizer = optim.Adam(model.parameters(), lr=1e-4)
+
+    # 为了用 CPU 能快点看到结果，我们先设为 3 轮
+    num_epochs = 3
+    print("\n--- 开始模型微调 ---")
+
+    for epoch in range(num_epochs):
+        model.train()
+        running_loss = 0.0
+        corrects = 0
+
+        # 【核心修改】：用 tqdm 包装 dataloader，生成实时进度条
+        progress_bar = tqdm(dataloader, desc=f"第 {epoch + 1}/{num_epochs} 轮", leave=False, colour='green')
+
+        for inputs, labels in progress_bar:
+            inputs = inputs.to(device)
+            labels = labels.to(device)
+
+            optimizer.zero_grad()
+            outputs = model(inputs)
+            _, preds = torch.max(outputs, 1)
+            loss = criterion(outputs, labels)
+
+            loss.backward()
+            optimizer.step()
+
+            running_loss += loss.item() * inputs.size(0)
+            corrects += torch.sum(preds == labels.data)
+
+            # 让进度条实时显示当前的误差值
+            progress_bar.set_postfix({'loss': f"{loss.item():.4f}"})
+
+        epoch_loss = running_loss / len(image_dataset)
+        epoch_acc = corrects.double() / len(image_dataset)
+        print(f"✅ 第 {epoch + 1}/{num_epochs} 轮完成 | 平均损失: {epoch_loss:.4f} | 准确率: {epoch_acc:.4f}")
+
+    save_path = 'mobilenet_finetuned.pth'
+    torch.save(model.state_dict(), save_path)
+    print(f"\n🎉 训练完成！模型权重已保存至: {save_path}")
+
+
+if __name__ == '__main__':
+    train_model()

train_text_model.py

@@ -0,0 +1,96 @@
+import pandas as pd
+import torch
+from torch.utils.data import Dataset, DataLoader
+from transformers import AutoTokenizer, AutoModelForSequenceClassification
+from torch.optim import AdamW
+from tqdm import tqdm
+
+
+# 1. 定义专属的 PyTorch 文本数据集
+class AITextDataset(Dataset):
+    def __init__(self, csv_file, tokenizer, max_len=128):
+        self.data = pd.read_csv(csv_file)
+        self.tokenizer = tokenizer
+        self.max_len = max_len
+
+    def __len__(self):
+        return len(self.data)
+
+    def __getitem__(self, index):
+        text = str(self.data.iloc[index, 0])
+        label = int(self.data.iloc[index, 1])
+
+        # 将汉字切成 token 序列
+        encoding = self.tokenizer(
+            text,
+            add_special_tokens=True,
+            max_length=self.max_len,
+            padding='max_length',
+            truncation=True,
+            return_attention_mask=True,
+            return_tensors='pt',
+        )
+        return {
+            'input_ids': encoding['input_ids'].flatten(),
+            'attention_mask': encoding['attention_mask'].flatten(),
+            'labels': torch.tensor(label, dtype=torch.long)
+        }
+
+
+def train_text():
+    device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
+    print(f"💻 当前计算设备: {device}")
+    if device.type == 'cpu':
+        print("⚠️ 警告：当前使用 CPU 炼丹。NLP模型参数量巨大，这可能需要一些时间，请耐心等待...")
+
+    print("正在加载预训练的中文 BERT 分词器与模型权重...")
+
+    # 【核心修复】：换成了官方真实存在、最经典的 bert-base-chinese
+    model_name = "bert-base-chinese"
+    tokenizer = AutoTokenizer.from_pretrained(model_name)
+    model = AutoModelForSequenceClassification.from_pretrained(model_name, num_labels=2)
+    model = model.to(device)
+
+    print("正在封装数据...")
+    dataset = AITextDataset('./data/text_dataset.csv', tokenizer, max_len=128)
+    # 批量大小设为 8，防止 CPU 内存吃紧
+    dataloader = DataLoader(dataset, batch_size=8, shuffle=True)
+
+    optimizer = AdamW(model.parameters(), lr=2e-5)
+
+    epochs = 1
+
+    print("\n🚀 --- 开始文本模型微调 ---")
+    model.train()
+
+    for epoch in range(epochs):
+        progress_bar = tqdm(dataloader, desc=f"第 {epoch + 1}/{epochs} 轮", leave=True, colour='blue')
+        running_loss = 0.0
+
+        for batch in progress_bar:
+            optimizer.zero_grad()
+
+            input_ids = batch['input_ids'].to(device)
+            attention_mask = batch['attention_mask'].to(device)
+            labels = batch['labels'].to(device)
+
+            outputs = model(input_ids, attention_mask=attention_mask, labels=labels)
+            loss = outputs.loss
+
+            loss.backward()
+            optimizer.step()
+
+            running_loss += loss.item()
+            progress_bar.set_postfix({'loss': f"{loss.item():.4f}"})
+
+        print(f"✅ 第 {epoch + 1} 轮完成 | 平均 Loss: {running_loss / len(dataloader):.4f}")
+
+    # 保存咱们微调后的专属大模型权重
+    save_dir = "./finetuned_text_model"
+    model.save_pretrained(save_dir)
+    tokenizer.save_pretrained(save_dir)
+    print(f"\n🎉 炼丹成功！专属的文本鉴别模型已保存在: {save_dir} 文件夹中。")
+
+
+if __name__ == "__main__":
+    train_text()

tune_thresholds.py

@@ -0,0 +1,47 @@
+import os
+import cv2
+import numpy as np
+from image_module import get_lbp_entropy, get_dct_high_freq_energy, get_fft_symmetry_error
+
+
+def process_folder(folder_path, label):
+    results = []
+    print(f"正在分析 [{label}] 样本集: {folder_path}")
+
+    for filename in os.listdir(folder_path):
+        if filename.lower().endswith(('.png', '.jpg', '.jpeg')):
+            filepath = os.path.join(folder_path, filename)
+            # 读取图片并转换为灰度图
+            img = cv2.imdecode(np.fromfile(filepath, dtype=np.uint8), cv2.IMREAD_COLOR)
+            if img is None: continue
+            gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)
+
+            # 提取特征
+            lbp = get_lbp_entropy(gray)
+            dct = get_dct_high_freq_energy(gray)
+            fft = get_fft_symmetry_error(gray)
+
+            results.append((lbp, dct, fft))
+
+    if not results:
+        return None
+
+    results_arr = np.array(results)
+    print(f"--- {label} 统计结果 ({len(results)}张) ---")
+    print(
+        f"LBP 熵  : 平均 {np.mean(results_arr[:, 0]):.3f} | 范围 [{np.min(results_arr[:, 0]):.3f} - {np.max(results_arr[:, 0]):.3f}]")
+    print(
+        f"DCT 占比: 平均 {np.mean(results_arr[:, 1]):.3f} | 范围 [{np.min(results_arr[:, 1]):.3f} - {np.max(results_arr[:, 1]):.3f}]")
+    print(
+        f"FFT 误差: 平均 {np.mean(results_arr[:, 2]):.3f} | 范围 [{np.min(results_arr[:, 2]):.3f} - {np.max(results_arr[:, 2]):.3f}]\n")
+    return results_arr
+
+
+if __name__ == "__main__":
+    # 请在当前目录下新建这两个文件夹，分别放几十张真实的和AI生成的图进去
+    real_folder = "./data/real"
+    ai_folder = "./data/ai"
+
+    print("开始执行特征分布寻优...\n")
+    process_folder(real_folder, "真实图像")
+    process_folder(ai_folder, "AI生成图像")

video_module.py

@@ -0,0 +1,68 @@
+import cv2
+import numpy as np
+import os
+from PIL import Image
+from image_module import analyze_image
+
+
+def analyze_video(video_path, num_samples=10):
+    """
+    使用 OpenCV 对视频进行均匀抽帧，并复用图像引擎进行鉴别
+    :param video_path: 视频文件的路径
+    :param num_samples: 准备抽取的代表性帧数（默认 10 帧）
+    """
+    # 1. 打开视频文件
+    cap = cv2.VideoCapture(video_path)
+    if not cap.isOpened():
+        return {"error": "无法打开视频文件"}
+
+    # 2. 获取视频的基础信息
+    total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
+    fps = cap.get(cv2.CAP_PROP_FPS)
+
+    # 3. 计算均匀抽帧的索引 (比如从 300 帧里均匀选 10 个时间点)
+    if total_frames < num_samples:
+        num_samples = total_frames  # 如果视频太短，有几帧抽几帧
+    intervals = np.linspace(0, total_frames - 1, num_samples, dtype=int)
+
+    frame_scores = []
+
+    # 4. 开始逐帧提取
+    for frame_idx in intervals:
+        cap.set(cv2.CAP_PROP_POS_FRAMES, frame_idx)  # 跳转到指定帧
+        ret, frame = cap.read()
+
+        if ret:
+            # OpenCV 默认读取的是 BGR 格式，我们需要转成正常的 RGB 格式
+            frame_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
+            pil_img = Image.fromarray(frame_rgb)
+
+            # 临时保存为图片文件，喂给咱们之前写好的图像模块
+            temp_path = f"temp_frame_{frame_idx}.jpg"
+            pil_img.save(temp_path)
+
+            try:
+                # 🌟 核心：直接调用咱们炼好的图像鉴别引擎！
+                result = analyze_image(temp_path)
+                frame_scores.append(result['final_probability'])
+            finally:
+                # 阅后即焚，清理临时文件
+                if os.path.exists(temp_path):
+                    os.remove(temp_path)
+
+    cap.release()
+
+    if not frame_scores:
+        return {"error": "未能成功提取任何视频帧"}
+
+    # 5. 综合计算这 10 张图的得分
+    avg_score = np.mean(frame_scores)
+    max_score = np.max(frame_scores)  # 记录最可疑的一帧
+
+    return {
+        "avg_probability": avg_score,
+        "max_probability": max_score,
+        "sampled_frames": num_samples,
+        "total_frames": total_frames,
+        "fps": fps
+    }