第8章: 多模态快速上手：图片/文本输入与结构化输出

# 目的：安装 DashScope Python SDK（通义千问官方/半官方调用入口）
#
# 提示：
# - 生产环境建议固定版本号（避免 API 行为变更导致线上不稳定）
# - 课堂演示可先不锁版本，跑通后再补 requirements.txt
pip install dashscope

# 目的：把 API Key 放到环境变量中，避免硬编码到源码（安全/可移植）
#
# 常见坑：
# - macOS/zsh：配置后要重新打开终端或 source 配置文件才能生效
# - IDE 运行：需要在 Run Configuration 里也配置环境变量
export DASHSCOPE_API_KEY="你的Key"

import os
import json
import re
import dashscope
from dashscope import MultiModalConversation


# =========================
# 0) 准备：API Key 与依赖
# =========================
# 设置 DashScope API Key（从环境变量读取）
# 说明：
# - 不要把 key 写进代码/仓库
# - 线上一般由容器/CI 注入环境变量
dashscope.api_key = os.getenv("DASHSCOPE_API_KEY")
if not dashscope.api_key:
    raise ValueError("请先设置环境变量 DASHSCOPE_API_KEY")


def parse_image_to_json(image_url: str) -> dict:
    """
    多模态识图并返回结构化 JSON
    Args:
        image_url: 图片地址（支持 file:// 本地路径，最稳定）
    Returns:
        dict: 包含 summary, entities, need_more_info 的结构化数据
    """
    # =========================
    # 1) 输入校验（上线必做）
    # =========================
    # 参数校验：确保传入有效图片地址
    if not image_url:
        raise ValueError("请传入图片地址（推荐使用 file:// 本地路径，最稳定）")

    # 防止使用示例占位符地址
    if "example.com" in image_url:
        raise ValueError("请传入真实图片（不要用 example.com 这种占位地址）")

    # =========================
    # 2) Prompt：把“输出合同”说清楚
    # =========================
    # 构造提示词：明确要求输出 JSON 格式，定义返回结构
    # 教学重点：
    # - 多模态上线的关键不是“让它说得好”，而是“让它输出可解析的结构化结果”
    # - prompt 必须写清楚：字段名 + 字段类型 + 输出必须是 JSON（不要解释文字）
    # - 结构越明确，解析与下游消费成本越低（也是 Guardrails/Schema 校验的前提）
    prompt = (
        "你是一个严谨的识图助手。请根据图片内容提取关键信息，并严格输出 JSON。\n"
        "输出格式：{\"summary\": string, \"entities\": [string], \"need_more_info\": [string]}"
    )

    # =========================
    # 3) 组装多模态 messages（文本 + 图片）
    # =========================
    # 构造多模态消息：包含文字指令 + 图片
    # 注意：这里的 image_url 建议优先使用 file:// 本地路径，外网 URL 可能因网络/权限失败
    messages = [
        {
            "role": "user",
            "content": [
                {"text": prompt},        # 文字指令
                {"image": image_url},    # 图片输入
            ],
        }
    ]

    # =========================
    # 4) 模型调用（外部依赖，失败概率最高）
    # =========================
    # 调用通义千问多模态模型
    # 说明：
    # - model 选择会影响价格/速度/效果
    # - 生产环境要加：超时、重试、限流、熔断、trace_id
    resp = MultiModalConversation.call(
        model="qwen-vl-plus",  # 使用支持多模态的模型
        messages=messages,
    )

    # =========================
    # 5) 错误分类（让失败“可定位、可告警”）
    # =========================
    # 错误处理：检查 API 调用是否成功
    if resp is None or getattr(resp, "output", None) is None:
        raise RuntimeError(
            f"DashScope 调用失败：status_code={getattr(resp, 'status_code', None)} "
            f"code={getattr(resp, 'code', None)} message={getattr(resp, 'message', None)} "
            f"request_id={getattr(resp, 'request_id', None)}"
        )

    # 检查返回结果是否包含有效选择
    if not getattr(resp.output, "choices", None):
        raise RuntimeError(
            f"DashScope 返回为空：status_code={getattr(resp, 'status_code', None)} "
            f"code={getattr(resp, 'code', None)} message={getattr(resp, 'message', None)} "
            f"request_id={getattr(resp, 'request_id', None)}"
        )

    # =========================
    # 6) 输出解析：把文本当成“接口响应”来处理
    # =========================
    # 提取模型返回的文本内容
    text = resp.output.choices[0].message["content"][0]["text"]

    # 清理文本：去除首尾空白
    text = (text or "").strip()
    if not text:
        raise RuntimeError("模型返回空内容（可能图片未正确读取/不支持该图片格式/图片尺寸过大）")

    # 处理可能的代码块标记（```json ... ```）
    # 常见坑：模型喜欢包 Markdown 代码块，不去掉就会 json.loads 失败
    if text.startswith("```"):
        text = re.sub(r"^```(?:json)?\s*", "", text, flags=re.IGNORECASE)  # 移除开头的 ```
        text = re.sub(r"\s*```$", "", text)  # 移除结尾的 ```

    # 尝试解析 JSON
    # 说明：
    # - 第一优先：直接 json.loads
    # - 第二优先：正则抽取第一个 {...}（容错）
    # - 生产可再加：schema 校验（字段/类型），失败走“修复回路”而不是重跑识图
    try:
        return json.loads(text)
    except json.JSONDecodeError:
        # 容错处理：尝试从文本中提取 JSON 部分
        m = re.search(r"\{.*\}", text, flags=re.S)  # 使用正则提取 JSON 对象
        if m:
            try:
                return json.loads(m.group(0))
            except json.JSONDecodeError:
                pass
        raise RuntimeError(f"模型未返回合法 JSON：{text}")


if __name__ == "__main__":
    # 使用示例：替换为你本地图像的路径（支持相对路径）
    # 教学建议：
    # - 先用本地 file:// 跑通，再考虑 URL 图片（减少网络变量）
    local_path = "eagle.png"  # 替换为你的图片路径
    abs_path = os.path.abspath(local_path)
    
    # 构造 file:// 协议的本地文件路径
    image_path = f"file://{abs_path}"
    
    # 调用识图函数并打印结果
    result = parse_image_to_json(image_path)
    print(result)

# 复制本段到 schema_repair_example.py 可直接运行验证
#
# 你将学到：
# - 为什么“合同校验”是多模态上线的关键（而不是让模型多说两句）
# - 为什么失败时优先“修 JSON”而不是“重跑识图”（更省钱、更快、更稳定）
# - 一个可落地的工程闭环：parse -> validate（合同校验）失败 -> repair（仅修 JSON，不重跑识图） -> re-validate
#
# 建议你带着问题读：
# 1) 如果模型返回了带解释文字的 JSON，怎么提取？
# 2) 如果字段缺失/类型错，怎么让系统“仍能对外提供稳定接口”？
# 3) repair 的提示词怎么写才能让模型只做修复、不重新发挥？

import json
import re
from dataclasses import dataclass
from typing import Any, Dict, List, Optional, Tuple


# -------------------------
# 1) 定义你要的“输出合同”（Schema）
# -------------------------
REQUIRED_FIELDS = {"summary", "entities", "need_more_info"}


def extract_json_object(text: str) -> str:
    """从模型输出中提取 JSON 对象字符串。

    现实情况：模型可能返回 ```json ...``` 包裹、或在 JSON 前后夹杂解释文字。
    这里做工程可用的容错：优先提取第一个 {...}。
    """
    text = (text or "").strip()

    # 去掉 ```json``` 包裹
    # 常见坑：模型会输出 Markdown 代码块，直接 json.loads 会报错
    if text.startswith("```"):
        text = re.sub(r"^```(?:json)?\s*", "", text, flags=re.IGNORECASE)
        text = re.sub(r"\s*```$", "", text)

    # 抽取 {...}
    # 说明：这里只取第一个 JSON 对象，避免“解释文字 + JSON + 解释文字”导致解析失败
    m = re.search(r"\{.*\}", text, flags=re.S)
    return m.group(0) if m else text


def validate_contract(obj: Dict[str, Any]) -> Tuple[bool, str]:
    """校验输出是否满足合同。

    返回：
    - ok: 是否通过
    - reason: 失败原因（用于日志/告警/修复提示词）
    """
    # 1) 必填字段
    missing = [k for k in REQUIRED_FIELDS if k not in obj]
    if missing:
        return False, f"missing={missing}"

    # 2) 字段类型/内容约束：summary 不能为空
    if not isinstance(obj.get("summary"), str) or not obj.get("summary"):
        return False, "summary must be non-empty string"

    # 3) entities 必须是 list[str]
    if not isinstance(obj.get("entities"), list) or not all(isinstance(x, str) for x in obj["entities"]):
        return False, "entities must be list[str]"

    # 4) need_more_info 必须是 list[str]
    if not isinstance(obj.get("need_more_info"), list) or not all(
        isinstance(x, str) for x in obj["need_more_info"]
    ):
        return False, "need_more_info must be list[str]"

    return True, "ok"


def parse_and_validate(text: str) -> Dict[str, Any]:
    """解析 + 合同校验。

    - 解析失败：抛出 JSONDecodeError
    - 合同不合规：抛出 ValueError
    """
    # 先做“提取”再解析，避免 markdown/解释文字污染
    obj = json.loads(extract_json_object(text))
    ok, reason = validate_contract(obj)
    if not ok:
        raise ValueError(f"contract_invalid: {reason}")
    return obj


# -------------------------
# 2) “自动修复回路”
# -------------------------

from langchain_community.chat_models import ChatTongyi
from langchain_core.messages import HumanMessage


class VisionLLM:
    """真实的多模态 LLM 调用：使用 DashScope 通义千问 Vision 模型。"""

    def __init__(self, model_name: str = "qwen-vl-max"):
        """初始化 Vision LLM。

        需要提前配置环境变量：export DASHSCOPE_API_KEY="your-api-key"
        """
        self.llm = ChatTongyi(model_name=model_name, temperature=0)

    def call_vision(self, image_path: str, prompt: str) -> str:
        """调用多模态模型识别图片并返回文本结果。

        Args:
            image_path: 图片路径（本地文件路径，格式如 file:///path/to/img.png）
            prompt: 提示词，要求模型输出 JSON 格式

        Returns:
            模型返回的文本字符串
        """
        # 构造多模态消息：文本 + 图片
        message = HumanMessage(
            content=[
                {"text": prompt},
                {"image": image_path},
            ]
        )
        resp = self.llm.invoke([message])
        # 适配 Tongyi Vision 返回格式
        if isinstance(resp.content, list) and len(resp.content) > 0:
            return resp.content[0].get("text", "")
        return resp.content if isinstance(resp.content, str) else str(resp.content)

    def invoke(self, prompt: str) -> dict:
        """用于 JSON 修复的纯文本调用（不携带图片）。

        Args:
            prompt: 修复提示词

        Returns:
            dict 包含 content 字段，兼容原有接口
        """
        resp = self.llm.invoke(prompt)
        text = resp.content if isinstance(resp.content, str) else resp.content[0].get("text", "")
        return {"content": text}


def repair_contract_with_llm(bad_text: str, llm) -> str:
    """让模型仅修复 JSON，使其符合合同。

    关键点：
    - 不要在这里重新做识图/推理；只做"格式/字段/类型修复"。
    - 这通常更省钱、更快、更稳定。
    """
    prompt = (
        "你是 JSON 修复器。请把下面内容修复为严格 JSON，对象必须包含 summary/entities/need_more_info 三个字段。\n"
        "要求：\n"
        "1) 只输出 JSON，不要解释；\n"
        "2) entities/need_more_info 必须是字符串数组；\n"
        "3) summary 必须是非空字符串。\n\n"
        f"待修复内容：{bad_text}"
    )
    resp = llm.invoke(prompt)
    return resp.get("content", "") if isinstance(resp, dict) else str(resp)


def parse_validate_with_repair(text: str, llm, image_path: str = None) -> dict:
    """完整闭环：先 parse+validate，失败则走一次 repair，再 parse+validate。

    Args:
        text: 待解析的文本（如果是首次调用，传 None 或空字符串）
        llm: VisionLLM 实例
        image_path: 图片路径（首次调用时需要）
    """
    # =========================
    # A) 首次调用：需要识图
    # =========================
    if not text and image_path:
        prompt = (
            "你是识图助手。请分析这张图片，并严格输出 JSON 格式：\n"
            '{"summary": "图片描述", "entities": ["实体1", "实体2"], "need_more_info": []}\n'
            "要求：summary 是图片的整体描述；entities 是图中识别到的关键对象列表；need_more_info 是需要用户补充的信息。"
        )
        # 第一次模型调用：成本高、延迟高，所以后续优先 repair 而非重跑
        text = llm.call_vision(image_path, prompt)
        print(f"🖼️  原始识图输出:\n{text}\n")

    # =========================
    # B) 第一次：parse + validate
    # =========================
    try:
        return parse_and_validate(text)
    except Exception as e:
        print(f"⚠️  首次校验失败: {e}")
        print("🔧 启动自动修复回路...")

        # 生产中建议这里打日志：trace_id + error_type + reason
        # - error_type: parse_error / contract_invalid / provider_error
        # - reason: 具体缺字段/类型错/JSON 不合法
        repaired = repair_contract_with_llm(text, llm)
        print(f"✅ 修复后内容:\n{repaired}\n")

        # =========================
        # C) 修复后：re-parse + re-validate
        # =========================
        return parse_and_validate(repaired)


# -------------------------
# 3) 可运行验证：使用真实图片调用大模型
# -------------------------

if __name__ == "__main__":
    import os

    # 检查 API Key
    # 常见坑：IDE 里运行脚本时没有继承终端环境变量
    if not os.environ.get("DASHSCOPE_API_KEY"):
        print("❌ 请先配置 DASHSCOPE_API_KEY 环境变量")
        print("   export DASHSCOPE_API_KEY='your-api-key'")
        exit(1)

    # 初始化真实的多模态 LLM
    # 教学建议：temperature=0 可显著提升 JSON 输出稳定性
    vision_llm = VisionLLM(model_name="qwen-vl-max")

    # 使用真实图片路径（请确保图片存在）
    # 这里使用相对路径示例，请根据你的实际路径修改
    image_paths = [
        "../data/img.png",      # 第一张图片
        "../data/img_1.png",    # 第二张图片（可选）
    ]

    # 查找存在的图片
    test_image = None
    for path in image_paths:
        if os.path.exists(path):
            test_image = os.path.abspath(path)
            print(f"✅ 找到测试图片: {test_image}")
            break

    if not test_image:
        print("❌ 未找到测试图片，请确保以下路径之一存在图片:")
        for p in image_paths:
            print(f"   - {p}")
        print("\n💡 提示：可以将任意图片复制到对应路径进行测试")
        exit(1)

    # 构造 file:// 协议路径
    image_path = f"file://{test_image}"

    print("\n" + "="*50)
    print("🚀 开始多模态识图 + Schema 校验 + 自动修复")
    print("="*50 + "\n")

    try:
        # 调用完整闭环：识图 → 校验 → （失败则）修复 → 再校验
        result = parse_validate_with_repair(
            text=None,           # 首次调用，需要识图
            llm=vision_llm,
            image_path=image_path
        )

        print("="*50)
        print("✅ 最终输出（通过合同校验的 JSON）:")
        print("="*50)
        print(json.dumps(result, ensure_ascii=False, indent=2))

    except Exception as e:
        print(f"\n❌ 处理失败: {e}")
        print("💡 可能原因：")
        print("   1. API Key 无效或已过期")
        print("   2. 网络连接问题")
        print("   3. 图片格式不被支持")
        exit(1)

# 复制本段到 image_governance_example.py 可直接运行验证
#
# 这段代码在讲什么？（教学导读）
# - 多模态上线第一道防线：输入治理（Input Guardrails）
# - 核心目标：把“任何图片都能进来”的不确定性，变成“可控成本 + 可控风险”的确定性
#
# 你将学到：
# 1) precheck：格式白名单/大小限制（不合规直接拒绝，避免成本失控）
# 2) 分辨率治理：大图建议缩放（否则延迟和 token 成本会飙升）
# 3) hash 缓存：重复图片复用结果（省钱、降低 P95）
# 4) 脱敏占位：生产里常见的人脸/证件/车牌打码通常发生在这层
#
# 目标：precheck（大小/格式/分辨率） → 压缩/缩放 → hash 缓存 → 调用识图 → 返回结果

import hashlib
import os
import json
from dataclasses import dataclass
from typing import Any, Dict, Tuple, Callable

from langchain_community.chat_models import ChatTongyi
from langchain_core.messages import HumanMessage


# -------------------------
# 1) 治理策略配置（可按业务调整）
# -------------------------
# 教学点：
# - 这些阈值应该“配置化”（env/config/feature flag），而不是写死
# - 不同业务的容忍度不同：电商图 vs 身份证图，策略差异巨大
# -------------------------
MAX_BYTES = 5 * 1024 * 1024          # 5MB
ALLOWED_EXT = {".jpg", ".jpeg", ".png", ".webp"}
MAX_LONG_EDGE = 1600                 # 长边 1600px
CACHE: Dict[str, Dict[str, Any]] = {}  # 内存缓存（生产建议用 Redis）


def precheck_image(path: str) -> None:
    """校验大小、格式、分辨率。

    - 格式/大小不合规：直接拒绝（避免成本失控）
    - 分辨率过大：后续会自动缩放
    """
    # 1.1 格式白名单
    ext = os.path.splitext(path)[1].lower()
    if ext not in ALLOWED_EXT:
        raise ValueError(f"不支持的格式: {ext}")

    # 1.2 大小限制：超过直接拒绝（先别急着“帮用户压缩”）
    # 教学解释：
    # - 压缩本身也要 CPU/时间；线上通常要先拒绝，再引导用户上传合规图片
    size = os.path.getsize(path)
    if size > MAX_BYTES:
        raise ValueError(f"图片过大: {size} > {MAX_BYTES}")

    # 1.3 可选：分辨率检查（需要 Pillow）
    # 常见策略：长边超过阈值则缩放；这里只演示“检查”，不直接修改原图
    # try:
    #     from PIL import Image
    #     w, h = Image.open(path).size
    #     if max(w, h) > MAX_LONG_EDGE:
    #         raise ValueError(f"分辨率过大: {w}x{h}")
    # except Exception:
    #     pass  # Pillow 未装就跳过


def sha256_file(path: str) -> str:
    """计算文件 SHA-256，用于缓存键。"""
    # 教学点：
    # - 用文件内容 hash 做 key，可以天然“去重”
    # - 生产环境可以加：TTL + LRU + 最大缓存条数（防止内存爆）
    h = hashlib.sha256()
    with open(path, "rb") as f:
        for chunk in iter(lambda: f.read(1024 * 1024), b""):
            h.update(chunk)
    return h.hexdigest()


# -------------------------
# 2) 图片治理演示（仅读取，绝不修改原图）
# -------------------------
def analyze_image_info(path: str) -> Dict[str, Any]:
    """分析图片信息，仅读取元数据，绝不修改原图。
    
    这个函数展示真正的"治理"：在不破坏原图的前提下获取信息。
    
    返回：
    - size_bytes: 文件大小
    - dimensions: 图片尺寸
    - format: 图片格式
    - needs_compression: 是否需要压缩
    - compression_ratio: 如果压缩，建议的压缩比例
    """
    try:
        from PIL import Image
        
        # 仅读取图片信息，不做任何修改
        img = Image.open(path)
        w, h = img.size
        size_bytes = os.path.getsize(path)
        
        # 计算是否需要压缩
        needs_compression = max(w, h) > MAX_LONG_EDGE
        compression_ratio = MAX_LONG_EDGE / max(w, h) if needs_compression else 1.0
        
        return {
            "size_bytes": size_bytes,
            "size_mb": round(size_bytes / (1024 * 1024), 2),
            "dimensions": (w, h),
            "format": img.format,
            "needs_compression": needs_compression,
            "compression_ratio": round(compression_ratio, 2),
            "target_dimensions": (int(w * compression_ratio), int(h * compression_ratio)) if needs_compression else (w, h),
            "estimated_compressed_size": round(size_bytes * compression_ratio * compression_ratio / (1024 * 1024), 2) if needs_compression else round(size_bytes / (1024 * 1024), 2)
        }
    except Exception as e:
        return {"error": f"无法分析图片: {e}"}


def simulate_compression_plan(path: str) -> Dict[str, Any]:
    """模拟压缩计划，展示压缩后的效果，但不实际执行压缩。
    
    这个函数展示"规划"的重要性：先评估，再执行。
    """
    info = analyze_image_info(path)
    
    if "error" in info:
        return info
    
    if not info["needs_compression"]:
        return {
            "action": "无需压缩",
            "reason": f"图片尺寸 {info['dimensions']} 在限制范围内",
            "original_info": info
        }
    
    return {
        "action": "建议压缩",
        "original_info": info,
        "compression_plan": {
            "current_size_mb": info["size_mb"],
            "estimated_size_mb": info["estimated_compressed_size"],
            "space_saved_mb": round(info["size_mb"] - info["estimated_compressed_size"], 2),
            "current_dimensions": info["dimensions"],
            "target_dimensions": info["target_dimensions"],
            "compression_ratio": info["compression_ratio"]
        },
        "safety_note": "这只是模拟分析，原图不会被修改。实际压缩时会创建临时文件。"
    }


# -------------------------
# 3) 缓存 + 调用闭环
# -------------------------

@dataclass
class VisionResult:
    """识图返回的结构化结果（示例）"""
    summary: str
    entities: list[str]
    need_more_info: list[str]

    def to_dict(self) -> Dict[str, Any]:
        """转为普通字典，便于 json.dumps 序列化。"""
        return {
            "summary": self.summary,
            "entities": self.entities,
            "need_more_info": self.need_more_info,
        }


class RealVisionCall:
    """真实的多模态识图调用：使用 DashScope 通义千问 Vision 模型。

    需要提前配置环境变量：export DASHSCOPE_API_KEY="your-api-key"
    """

    def __init__(self, model_name: str = "qwen-vl-max"):
        """初始化 Vision LLM。"""
        self.llm = ChatTongyi(model_name=model_name, temperature=0)

    def __call__(self, image_path: str) -> VisionResult:
        """调用多模态模型识别图片并返回结构化结果。

        Args:
            image_path: 图片路径（本地文件路径）

        Returns:
            VisionResult 结构化结果
        """
        # 构造 file:// 协议路径
        file_path = f"file://{os.path.abspath(image_path)}"

        # 构造多模态消息：文本 + 图片
        prompt = (
            "你是识图助手。请分析这张图片，并严格输出 JSON 格式：\n"
            '{"summary": "图片描述", "entities": ["实体1", "实体2"], "need_more_info": []}\n'
            "要求：summary 是图片的整体描述；entities 是图中识别到的关键对象列表；need_more_info 是需要用户补充的信息。"
        )
        message = HumanMessage(
            content=[
                {"text": prompt},
                {"image": file_path},
            ]
        )

        try:
            resp = self.llm.invoke([message])
            # 适配 Tongyi Vision 返回格式
            if isinstance(resp.content, list) and len(resp.content) > 0:
                text = resp.content[0].get("text", "")
            else:
                text = resp.content if isinstance(resp.content, str) else str(resp.content)

            # 解析 JSON 输出
            try:
                obj = json.loads(text)
                return VisionResult(
                    summary=obj.get("summary", "解析失败"),
                    entities=obj.get("entities", []),
                    need_more_info=obj.get("need_more_info", [])
                )
            except json.JSONDecodeError:
                # JSON 解析失败，返回原始文本
                return VisionResult(
                    summary=f"JSON解析失败，原始输出：{text[:100]}",
                    entities=[],
                    need_more_info=["模型输出格式不正确"]
                )
        except Exception as e:
            return VisionResult(
                summary=f"识图调用失败: {str(e)}",
                entities=[],
                need_more_info=["请检查 API Key 和网络连接"]
            )


def cached_vision_call(image_path: str, vision_fn: Callable[[str], VisionResult]) -> Dict[str, Any]:
    """带缓存、治理的识图调用闭环。

    返回：
    - _cache: bool 是否命中缓存
    - result: VisionResult 结构化结果
    - _original_path: 原始路径（用于日志）
    - _governance_info: 图片治理信息（尺寸、格式等）
    """
    # 1) 治理校验
    precheck_image(image_path)

    # 2) 分析图片信息（仅读取，不修改）
    governance_info = analyze_image_info(image_path)
    
    # 3) 缓存键（基于原图）
    cache_key = sha256_file(image_path)
    if cache_key in CACHE:
        return {"_cache": True, "result": CACHE[cache_key].to_dict(), "_original_path": image_path, "_governance_info": governance_info}

    # 4) 调用识图（使用原图，确保安全）
    result = vision_fn(image_path)

    # 5) 存入缓存（生产建议加 TTL）
    CACHE[cache_key] = result

    return {"_cache": False, "result": result.to_dict(), "_original_path": image_path, "_governance_info": governance_info}


# -------------------------
# 4) 可运行验证：图片治理演示（绝对安全）
# -------------------------

if __name__ == "__main__":
    # 检查 API Key
    if not os.environ.get("DASHSCOPE_API_KEY"):
        print("❌ 请先配置 DASHSCOPE_API_KEY 环境变量")
        print("   export DASHSCOPE_API_KEY='your-api-key'")
        exit(1)

    # 初始化真实的多模态识图调用
    vision = RealVisionCall(model_name="qwen-vl-max")
    print("✅ 已初始化 DashScope Vision 模型 (qwen-vl-max)")

    # 使用真实图片路径（请确保这些文件存在）
    real_paths = ["../data/img.png", "../data/img_1.png"]
    
    # 检查文件是否存在
    existing_paths = []
    for p in real_paths:
        if os.path.exists(p):
            existing_paths.append(p)
            print(f"✅ 找到图片: {p}")
        else:
            print(f"⚠️  图片不存在: {p}")
    
    if not existing_paths:
        print("❌ 没有找到可用的图片文件，请确保 ../data/img.png 或 ../data/img_1.png 存在")
        print("💡 提示：可以将真实图片复制到 ../data/ 目录下进行测试")
        exit(1)

    print("\n=== 图片治理演示（仅分析，不修改原图）===")
    
    # 演示图片分析功能
    for p in existing_paths:
        print(f"\n==== 分析 {p} ====")
        analysis = analyze_image_info(p)
        print(json.dumps(analysis, ensure_ascii=False, indent=2))
        
        # 演示压缩计划
        plan = simulate_compression_plan(p)
        print(f"\n压缩计划:")
        print(json.dumps(plan, ensure_ascii=False, indent=2))

    print("\n=== 识图调用演示 ===")
    
    # 第一次调用第一个图片（未命中缓存）
    first_image = existing_paths[0]
    out1 = cached_vision_call(first_image, vision)
    print(f"\n==== {first_image}（首次）====")
    print(json.dumps(out1, ensure_ascii=False, indent=2))

    # 第二次调用同一图片（命中缓存）
    out1_cached = cached_vision_call(first_image, vision)
    print(f"\n==== {first_image}（缓存命中）====")
    print(json.dumps(out1_cached, ensure_ascii=False, indent=2))

    # 如果有第二张图片，调用它（新图，未命中缓存）
    if len(existing_paths) > 1:
        second_image = existing_paths[1]
        out2 = cached_vision_call(second_image, vision)
        print(f"\n==== {second_image}（新图）====")
        print(json.dumps(out2, ensure_ascii=False, indent=2))

    print("\n=== 安全验证 ===")
    print("✅ 原图文件未被修改")
    print("✅ 仅读取图片元数据进行分析")
    print("✅ 缓存基于原图内容，无需临时文件")
    print("✅ 治理信息包含尺寸、格式等关键数据")
    print("✅ 演示代码不会创建或覆盖任何文件")

# 复制本段到 observability_example.py 可直接运行验证
#
# 这段代码在讲什么？（教学导读）
# - 多模态系统最大的痛点是“黑盒”：慢/错/不稳定时你不知道发生了什么
# - 可观测的目标：每一次请求都有 trace_id，每一步都有耗时与状态，错误能分类统计
#
# 你将学到：
# 1) Trace：把一次调用拆成 precheck/vision_call/post_parse 三段并记录事件
# 2) 错误分类：input/provider/parse/system 四类，便于告警与优化
# 3) 日志脱敏：不要把原图路径/原始输出全文/敏感标识写进日志
#
# 目标：trace_id + 分段计时（precheck/vision_call/post_parse） + 错误分类 + 日志脱敏 + 完整闭环

import time
import uuid
import json
import hashlib
import os
from dataclasses import dataclass
from typing import Any, Dict, Optional, Callable

from langchain_community.chat_models import ChatTongyi
from langchain_core.messages import HumanMessage


# -------------------------
# 1) Trace 与错误分类（可观测核心）
# -------------------------
class Trace:
    """一次多模态请求的完整追踪器。

    用法：
    - trace = Trace()
    - trace.mark("precheck", True, {"size": 1024})
    - trace.mark("vision_call", False, {"error_type": "provider_error"})
    - print(trace.dump())
    """
    def __init__(self):
        # trace_id 用于串起“一次请求”的全链路（日志/监控/排障）
        self.trace_id = f"t-{uuid.uuid4().hex[:8]}"
        self.t0 = time.time()
        self.events = []

    def mark(self, name: str, ok: bool, meta: Optional[Dict[str, Any]] = None):
        """记录一个事件节点。

        Args:
            name: 阶段名，建议 precheck/vision_call/post_parse
            ok: 是否成功
            meta: 附加信息（不要放敏感内容，如原图/完整识别文本）
        """
        self.events.append({
            "name": name,
            "ok": ok,
            "ms": round((time.time() - self.t0) * 1000),
            "meta": meta or {},
        })

    def dump(self) -> Dict[str, Any]:
        """导出为可日志/上报的 JSON 结构。"""
        return {"trace_id": self.trace_id, "events": self.events}


def classify_error(e: Exception) -> str:
    """把异常映射为可统计的错误类型。

    分类依据：错误信息关键字 + 常见工程场景。
    """
    # 教学点：分类是为了“可统计”，不是为了写得花
    # - 线上看 dashboard：哪个错误类型占比最高？就优先优化它
    s = str(e).lower()
    if "图片" in s or "format" in s or "size" in s or "path" in s:
        return "input_error"
    if "status_code" in s or "request_id" in s or "dashscope" in s or "timeout" in s:
        return "provider_error"
    if "json" in s or "contract" in s or "schema" in s:
        return "parse_error"
    return "system_error"


def safe_meta(meta: Dict[str, Any]) -> Dict[str, Any]:
    """脱敏：去除敏感字段，避免日志泄露隐私。"""
    # 教学点：
    # - 你很容易“为了调试”把原始输出打印出来，结果把隐私/密钥写进日志
    # - 线上日志一般会汇聚到第三方系统，泄露风险更大
    sensitive_keys = {"image_path", "raw_text", "full_output", "request_id"}
    return {k: v for k, v in meta.items() if k not in sensitive_keys}


# -------------------------
# 2) 带可观测的识图闭环（真实 API 调用）
# -------------------------
@dataclass
class VisionResult:
    """识图返回的结构化结果（示例）"""
    summary: str
    entities: list[str]
    need_more_info: list[str]

class RealVisionWithTrace:
    """真实的多模态识图调用：使用 DashScope 通义千问 Vision 模型。

    需要提前配置环境变量：export DASHSCOPE_API_KEY="your-api-key"
    """

    def __init__(self, model_name: str = "qwen-vl-max"):
        """初始化 Vision LLM。"""
        self.llm = ChatTongyi(model_name=model_name, temperature=0)

    def __call__(self, image_path: str, trace: Trace) -> VisionResult:
        """调用多模态模型识别图片并返回结构化结果。

        Args:
            image_path: 图片路径（本地文件路径）
            trace: Trace 追踪器

        Returns:
            VisionResult 结构化结果
        """
        try:
            # 构造 file:// 协议路径
            file_path = f"file://{os.path.abspath(image_path)}"

            # 构造多模态消息
            prompt = (
                "你是识图助手。请分析这张图片，并严格输出 JSON 格式：\n"
                '{"summary": "图片描述", "entities": ["实体1", "实体2"], "need_more_info": []}\n'
                "要求：summary 是图片的整体描述；entities 是图中识别到的关键对象列表；need_more_info 是需要用户补充的信息。"
            )
            message = HumanMessage(
                content=[
                    {"text": prompt},
                    {"image": file_path},
                ]
            )

            resp = self.llm.invoke([message])

            # 适配 Tongyi Vision 返回格式
            if isinstance(resp.content, list) and len(resp.content) > 0:
                text = resp.content[0].get("text", "")
            else:
                text = resp.content if isinstance(resp.content, str) else str(resp.content)

            # 记录成功
            trace.mark("vision_call", True, {"model": "qwen-vl-max"})

            # 解析 JSON 输出
            try:
                obj = json.loads(text)
                return VisionResult(
                    summary=obj.get("summary", "解析失败"),
                    entities=obj.get("entities", []),
                    need_more_info=obj.get("need_more_info", [])
                )
            except json.JSONDecodeError:
                return VisionResult(
                    summary=f"JSON解析失败，原始输出：{text[:100]}",
                    entities=[],
                    need_more_info=["模型输出格式不正确"]
                )

        except Exception as e:
            trace.mark("vision_call", False, {"error": str(e)})
            raise


def precheck_with_trace(image_path: str, trace: Trace) -> None:
    """带可观测的治理校验。"""
    # 校验文件是否存在
    if not os.path.exists(image_path):
        raise ValueError(f"图片不存在: {image_path}")

    # 校验格式
    ext = os.path.splitext(image_path)[1].lower()
    if ext not in {".jpg", ".jpeg", ".png", ".webp"}:
        raise ValueError(f"不支持的格式: {ext}")

    # 校验大小 (5MB)
    size = os.path.getsize(image_path)
    if size > 5 * 1024 * 1024:
        raise ValueError(f"图片过大: {size} > 5MB")

    trace.mark("precheck", True, {"size_kb": size // 1024})


def post_parse_with_trace(result: VisionResult, trace: Trace) -> Dict[str, Any]:
    """带可观测的后处理（JSON 合同校验等）。"""
    # 校验字段
    if not result.summary:
        raise ValueError("contract_invalid: summary is empty")
    if not isinstance(result.entities, list):
        raise ValueError("contract_invalid: entities must be list")

    trace.mark("post_parse", True, {"fields": len(result.entities)})
    return {"summary": result.summary, "entities": result.entities, "need_more_info": result.need_more_info}


def run_vision_with_observability(image_path: str, vision_fn: Callable[[str, Trace], VisionResult]) -> Dict[str, Any]:
    """完整的多模态调用闭环，带可观测。

    返回：
    - trace: 完整追踪信息
    - result: 结构化结果（失败时为 None）
    - error: 异常信息（成功时为 None）
    """
    trace = Trace()
    try:
        precheck_with_trace(image_path, trace)
        result = vision_fn(image_path, trace)
        parsed = post_parse_with_trace(result, trace)
        return {"trace": trace.dump(), "result": parsed, "error": None}
    except Exception as e:
        trace.mark("error", False, {"type": classify_error(e), "msg": str(e)})
        return {"trace": trace.dump(), "result": None, "error": {"type": classify_error(e), "msg": str(e)}}


# -------------------------
# 3) 可运行验证：使用真实图片调用大模型
# -------------------------
if __name__ == "__main__":
    # 检查 API Key
    if not os.environ.get("DASHSCOPE_API_KEY"):
        print("❌ 请先配置 DASHSCOPE_API_KEY 环境变量")
        print("   export DASHSCOPE_API_KEY='your-api-key'")
        exit(1)

    # 初始化真实的多模态识图调用
    vision = RealVisionWithTrace(model_name="qwen-vl-max")
    print("✅ 已初始化 DashScope Vision 模型 (qwen-vl-max)")

    # 使用真实图片路径（请确保这些文件存在）
    image_paths = ["../data/img.png", "../data/img_1.png"]

    # 查找存在的图片
    test_image = None
    for path in image_paths:
        if os.path.exists(path):
            test_image = os.path.abspath(path)
            print(f"✅ 找到测试图片: {test_image}")
            break

    if not test_image:
        print("❌ 未找到测试图片，请确保以下路径之一存在图片:")
        for p in image_paths:
            print(f"   - {p}")
        print("\n💡 提示：可以将任意图片复制到对应路径进行测试")
        exit(1)

    print("\n" + "=" * 50)
    print("🚀 开始多模态识图 + 可观测性追踪")
    print("=" * 50 + "\n")

    # 调用带可观测性的识图闭环
    out = run_vision_with_observability(test_image, vision)

    print("=" * 50)
    print("📊 完整 Trace 与结果:")
    print("=" * 50)
    print(json.dumps(out, ensure_ascii=False, indent=2))

import json
import os
import re
from typing import Any, Dict, List, Tuple

from langchain_community.chat_models import ChatTongyi
from langchain_core.messages import HumanMessage
from langchain_core.runnables import RunnableBranch, RunnableLambda, RunnablePassthrough


# =========================
# 这段代码在讲什么？（教学导读）
# =========================
# 目标：把“多模态识图”做成一条可复用、可观测、可降级的工程链路。
#
# 关键工程点：
# 1) LCEL（Runnable）是 LangChain 的“编排语言”：每一步都是可替换节点。
# 2) RunnableBranch 是“降级开关”：Vision 失败就走 OCR/Text 分支。
# 3) with_retry 是“链路级重试”：解析/校验失败统一重试，不把 try/except 写满业务。
# 4) 合同优先：不管 Vision 还是 OCR，最终都输出同一个 JSON 合同，Writer 无感。
#
# 运行前准备：
# - 需要提前配置：export DASHSCOPE_API_KEY="..."
# - 可选启用 LangSmith Trace（课堂讲可观测非常有用）：
#     export LANGCHAIN_TRACING_V2=true
#     export LANGCHAIN_API_KEY="你的 LangSmith Key"
#     export LANGCHAIN_PROJECT="vision-writer-demo"


# =========================
# 0) 模型选择与稳定性
# =========================
# - Vision 模型：负责“看图”
# - Text 模型：负责“读 OCR 文本并抽合同”（fallback 路径）
# - Writer 模型：负责“把合同变交付物”（文案/报告/工单都可以换）
#
# 温度建议：temperature=0 有助于结构化输出稳定（更利于 JSON 解析/校验）
vision_llm = ChatTongyi(model_name="qwen-vl-max", temperature=0)
text_llm = ChatTongyi(model_name="qwen-max", temperature=0)
writer_llm = ChatTongyi(model_name="qwen-max", temperature=0)


# =========================
# 1) 输出合同（Schema）
# =========================
# 说明：合同字段越稳定，下游越好写（前端渲染/落库/检索/下游 Writer 都依赖它）
REQUIRED = {"summary", "entities", "need_more_info"}


def _extract_json(text: str) -> str:
    text = (text or "").strip()
    # 常见坑：模型会用 ```json ...``` 包起来，不去掉就无法 json.loads
    if text.startswith("```"):
        text = re.sub(r"^```(?:json)?\s*", "", text, flags=re.IGNORECASE)
        text = re.sub(r"\s*```$", "", text)
    m = re.search(r"\{.*\}", text, flags=re.S)
    return m.group(0) if m else text


def _vision_call(inp: Dict[str, Any]) -> Dict[str, Any]:
    """节点1：尝试 Vision。

    返回统一结构，便于 RunnableBranch 决策：
    - ok: 是否成功
    - raw: Vision 原始输出文本
    - error: 失败原因（字符串）
    """
    # 允许手动强制走 OCR 分支（课堂演示时很好用）
    # 说明：线上一般用“错误类型/超时/置信度”触发，这里用 force_ocr 简化讲解
    if inp.get("force_ocr"):
        return {"ok": False, "raw": "", "error": "force_ocr"}

    image_path = inp.get("image_path")
    if not image_path:
        return {"ok": False, "raw": "", "error": "missing_image_path"}

    prompt = (
        "你是识图助手。请严格输出 JSON（不要输出多余文本/代码块）："
        '{"summary": string, "entities": [string], "need_more_info": [string]}'
    )

    try:
        msg = HumanMessage(content=[{"text": prompt}, {"image": image_path}])
        resp = vision_llm.invoke([msg])
        raw = resp.content[0]["text"]
        if not (isinstance(raw, str) and raw.strip()):
            return {"ok": False, "raw": "", "error": "empty_vision_output"}
        return {"ok": True, "raw": raw, "error": ""}
    except Exception as e:
        # 注意：这里不直接抛异常，而是把失败“结构化”出来交给 Branch 决策
        return {"ok": False, "raw": "", "error": f"vision_error:{type(e).__name__}"}


def _parse_contract(v: Dict[str, Any]) -> Dict[str, Any]:
    """节点2：把 Vision 原始文本解析为合同 JSON。

    说明：
    - 只有在 ok=True 的分支才会走到这里
    - 解析失败抛异常，可触发链路重试
    """
    # 解析失败抛异常：交给 with_retry 做统一重试
    return json.loads(_extract_json(v.get("raw", "")))


def _ocr_text_to_contract(inp: Dict[str, Any]) -> Dict[str, Any]:
    """降级分支：OCR/Text → 合同 JSON。

    实战里通常是：
    - 先用 OCR 引擎把图片转成文字（比如 PaddleOCR/Tesseract/厂商 OCR）
    - 再把 ocr_text 喂给文本 LLM 生成同一个 JSON 合同（保持“合同口径一致”）

    这里为了示例可运行性：假设你已经拿到了 ocr_text。
    """
    # 输入约定：ocr_text 是 OCR 输出的纯文本（可由 PaddleOCR/Tesseract/厂商 OCR 产出）
    ocr_text = (inp.get("ocr_text") or "").strip()
    if not ocr_text:
        raise ValueError("missing_ocr_text")

    prompt = (
        "你是信息抽取助手。请根据以下 OCR 文本，严格输出 JSON（不要输出多余文本/代码块）："
        '{"summary": string, "entities": [string], "need_more_info": [string]}\n\n'
        f"OCR_TEXT=\n{ocr_text}"
    )
    resp = text_llm.invoke(prompt)
    raw = resp.content if isinstance(resp.content, str) else resp.content[0].get("text", "")
    return json.loads(_extract_json(raw))


def _guard_contract(obj: Dict[str, Any]) -> Dict[str, Any]:
    """节点3：Guardrails（合同层）— schema 校验 + 轻量脱敏（示例）。"""
    # 2.1 合同校验：字段缺失/类型错就当失败（不要让脏数据进入下游）
    miss = [k for k in REQUIRED if k not in obj]
    if miss:
        raise ValueError(f"contract_missing={miss}")
    if not isinstance(obj.get("entities"), list):
        raise ValueError("entities must be list")
    if not isinstance(obj.get("need_more_info"), list):
        raise ValueError("need_more_info must be list")

    # 2.2 合规：轻量脱敏（示例）
    # 生产环境建议：脱敏策略配置化 + 记录命中原因（用于审计）
    def clean(s: Any) -> str:
        s = "" if s is None else str(s)
        s = re.sub(r"\b\d{17}[0-9Xx]\b", "[REDACTED_ID]", s)
        s = re.sub(r"\b1\d{10}\b", "[REDACTED_PHONE]", s)
        s = re.sub(r"[\w.+-]+@[\w-]+\.[\w.-]+", "[REDACTED_EMAIL]", s)
        return s

    out = dict(obj)
    out["summary"] = clean(out.get("summary"))
    out["entities"] = [clean(x) for x in out.get("entities", [])]
    out["need_more_info"] = [clean(x) for x in out.get("need_more_info", [])]
    return out


def _writer_call(contract: Dict[str, Any]) -> str:
    """节点4：Writer 产出交付物（纯文本）。"""
    prompt = (
        "你是运营文案专家。\n"
        "请根据以下 JSON 生成一段 80 字以内的商品介绍，强调卖点。\n"
        "输出必须是纯文本，不要 JSON，不要 Markdown，不要列表符号，不要链接，不要联系方式。\n\n"
        f"JSON={json.dumps(contract, ensure_ascii=False)}"
    )
    resp = writer_llm.invoke(prompt)
    out = resp.content if isinstance(resp.content, str) else resp.content[0].get("text", "")
    return (out or "").strip()


def _guard_deliverable(text: str) -> str:
    """节点5：Guardrails（交付物层）— 长度/敏感/危险指令拦截（示例）。"""
    # 最小门禁：空/太长/危险指令/PII
    if not text:
        raise ValueError("deliverable_empty")
    if len(text) > 80:
        raise ValueError(f"deliverable_too_long={len(text)}")
    if re.search(r"\b(import\s+os|os\.system|subprocess|rm\s+-rf)\b", text, flags=re.I):
        raise ValueError("deliverable_unsafe")
    if re.search(r"\b1\d{10}\b", text) or re.search(r"[\w.+-]+@[\w-]+\.[\w.-]+", text):
        raise ValueError("deliverable_pii")
    return text


# =========================
# 3) 把节点串成链路：RunnableBranch 做降级
# =========================
# 关键：Runnable 链路（LCEL） + RunnableBranch（失败降级）
# - 主分支：Vision → parse → guard
# - 降级分支：OCR/Text → guard
# - 分支合流后：writer → deliverable_guard
# 说明：contract_chain 专注“生成合规合同”，后续 Writer 只依赖合同
contract_chain = (
    RunnablePassthrough()
    | RunnableLambda(_vision_call)
    | RunnableBranch(
        # 条件：Vision 成功就走主分支
        (lambda v: bool(v.get("ok")), RunnableLambda(_parse_contract)),
        # 默认分支：Vision 失败 → 回到原始输入，用 OCR/Text 生成合同
        # 教学点：这就是“降级”。让链路有结果，而不是报错退出。
        RunnablePassthrough() | RunnableLambda(_ocr_text_to_contract),
    )
    | RunnableLambda(_guard_contract)
)

chain = (
    RunnablePassthrough()
    | contract_chain
    | RunnableLambda(_writer_call)
    | RunnableLambda(_guard_deliverable)
).with_retry(stop_after_attempt=3)


if __name__ == "__main__":
    image_path = f"file://{os.path.abspath('eagle.png')}"

    # 正常情况：直接走 Vision
    print(chain.invoke({"image_path": image_path}))

    # 课堂演示：强制走 OCR 降级分支（只要提供 ocr_text 即可）
    # 说明：在课堂上你可以先演示 Vision 版本，再演示“Vision 挂了也不影响交付”的 fallback。
    # print(chain.invoke({"force_ocr": True, "ocr_text": "Apple iPhone 15 Pro 128GB\n价格 7999\n钛金属"}))

import json
import os
import re
from typing import Any, Dict, List, Tuple

from langchain_community.chat_models import ChatTongyi
from langchain_core.messages import HumanMessage


# =========================
# 这段端到端示例在讲什么？（教学导读）
# =========================
# 目标：把多模态系统拆成三个独立模块：
# - Vision：只负责“看图 → 输出合同(JSON)”
# - Writer：只负责“合同(JSON) → 交付物(文本/报告/工单)”
# - Guardrails：夹在中间做质量门禁（schema 校验、敏感过滤、失败重试/降级）
#
# 为什么要拆？
# - Vision 模型输出再不稳定，只要合同稳定，下游就稳定。
# - Writer 可以频繁迭代（换风格/换模板/换输出类型），不影响 Vision。
# - Guardrails 统一收口风险：把“不确定性”挡在业务前。

vision_llm = ChatTongyi(model_name="qwen-vl-max", temperature=0)
writer_llm = ChatTongyi(model_name="qwen-max", temperature=0)


REQUIRED = {"summary", "entities", "need_more_info"}


def _extract_json(text: str) -> str:
    text = (text or "").strip()
    # 常见坑：模型包 ```json``` / 或前后夹解释文字
    if text.startswith("```"):
        text = re.sub(r"^```(?:json)?\s*", "", text, flags=re.IGNORECASE)
        text = re.sub(r"\s*```$", "", text)
    m = re.search(r"\{.*\}", text, flags=re.S)
    return m.group(0) if m else text


def validate_contract(obj: Dict[str, Any]) -> Tuple[bool, str]:
    # 合同校验：字段缺失/类型错误都视为失败
    miss = [k for k in REQUIRED if k not in obj]
    if miss:
        return False, f"missing={miss}"
    if not isinstance(obj.get("entities"), list):
        return False, "entities must be list"
    if not isinstance(obj.get("need_more_info"), list):
        return False, "need_more_info must be list"
    return True, "ok"


def guard_contract(obj: Dict[str, Any]) -> Tuple[Dict[str, Any], List[str]]:
    """Guardrails（合同层）：schema 校验 + 敏感字段过滤（最小示例）。"""
    issues: List[str] = []

    ok, reason = validate_contract(obj)
    if not ok:
        issues.append(f"schema:{reason}")
        return obj, issues

    def _clean_text(s: Any) -> str:
        s = "" if s is None else str(s)
        s = re.sub(r"\b\d{17}[0-9Xx]\b", "[REDACTED_ID]", s)
        s = re.sub(r"\b1\d{10}\b", "[REDACTED_PHONE]", s)
        s = re.sub(r"[\w.+-]+@[\w-]+\.[\w.-]+", "[REDACTED_EMAIL]", s)
        return s

    obj = dict(obj)
    obj["summary"] = _clean_text(obj.get("summary"))
    obj["entities"] = [_clean_text(x) for x in obj.get("entities", [])]
    obj["need_more_info"] = [_clean_text(x) for x in obj.get("need_more_info", [])]
    return obj, issues


def guard_deliverable(text: str, max_chars: int = 80) -> Tuple[bool, str]:
    """Guardrails（交付物层）：长度/敏感/注入等检查（最小示例）。"""
    t = (text or "").strip()
    if not t:
        return False, "empty"
    if len(t) > max_chars:
        return False, f"too_long:{len(t)}"
    if re.search(r"\b(ssh-rsa|BEGIN\s+PRIVATE\s+KEY|AKIA[0-9A-Z]{16})\b", t, flags=re.I):
        return False, "secret_like"
    if re.search(r"\b(import\s+os|os\.system|subprocess|rm\s+-rf)\b", t, flags=re.I):
        return False, "unsafe_instruction"
    if re.search(r"\b1\d{10}\b", t) or re.search(r"[\w.+-]+@[\w-]+\.[\w.-]+", t):
        return False, "pii"
    return True, "ok"


def vision_to_contract(image_path: str, max_retries: int = 2) -> Dict[str, Any]:
    """Vision → JSON 合同，并在 Guardrails 层做校验/重试/降级。"""
    # 教学点：
    # - max_retries 控制成本与可用性权衡
    # - 生产建议把失败原因写入 need_more_info 或日志（便于告警/复盘）
    last_err = ""
    for attempt in range(max_retries + 1):
        prompt = (
            "你是识图助手。请严格输出 JSON（不要输出多余文本/代码块）："
            '{"summary": string, "entities": [string], "need_more_info": [string]}'
        )
        msg = HumanMessage(content=[{"text": prompt}, {"image": image_path}])
        resp = vision_llm.invoke([msg])
        text = resp.content[0]["text"]

        try:
            obj = json.loads(_extract_json(text))
        except Exception as e:
            last_err = f"json_parse:{e}"
            continue

        obj, issues = guard_contract(obj)
        if not issues:
            return obj
        last_err = ";".join(issues)

    return {
        "summary": "",
        "entities": [],
        "need_more_info": ["contract_invalid", last_err],
        "_degraded": True,
    }


def contract_to_deliverable(contract: Dict[str, Any], max_retries: int = 1) -> str:
    """JSON 合同 → 交付物，并在 Guardrails 层做校验/重试/降级。"""
    # 教学点：
    # - 交付物也要做门禁（长度、PII、危险指令），避免“模型胡写”直接对外发布
    last_err = ""
    for attempt in range(max_retries + 1):
        prompt = (
            "你是运营文案专家。\n"
            "请根据以下 JSON 生成一段 80 字以内的商品介绍，强调卖点。\n"
            "输出必须是纯文本，不要 JSON，不要 Markdown，不要列表符号，不要链接，不要联系方式。\n\n"
            f"JSON={json.dumps(contract, ensure_ascii=False)}"
        )
        resp = writer_llm.invoke(prompt)
        out = resp.content if isinstance(resp.content, str) else resp.content[0].get("text", "")
        ok, reason = guard_deliverable(out, max_chars=80)
        if ok:
            return out.strip()
        last_err = reason

    return "（生成失败：输出未通过 Guardrails 校验）"


if __name__ == "__main__":
    image_path = f"file://{os.path.abspath('eagle.png')}"
    contract = vision_to_contract(image_path)
    deliverable = contract_to_deliverable(contract)
    print("contract=", json.dumps(contract, ensure_ascii=False, indent=2))
    print("deliverable=", deliverable)

# 依赖：pip install opencv-python
#
# 教学导读：这段在做“关键帧路线”的最小闭环
# 1) 用 OpenCV 从视频抽取 N 张关键帧（默认 6 张，成本/延迟可控）
# 2) 把这些帧作为多张图片一次性喂给 Vision 模型（覆盖场景变化）
# 3) 输出结构化 JSON（summary/scenes/actions）供下游消费
#
# 生产建议：
# - 帧图存储到临时目录，并确保 finally 清理
# - 对超大视频做时长/分辨率限制，避免 OOM/耗时过长

import os
import json
import cv2
from langchain_community.chat_models import ChatTongyi
from langchain_core.messages import HumanMessage


def extract_frames(video_path: str, frame_count: int = 6) -> list[str]:
    # 1) 打开视频文件
    cap = cv2.VideoCapture(video_path)
    if not cap.isOpened():
        raise ValueError(f"无法打开视频：{video_path}")

    total = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
    if total <= 0:
        raise ValueError("视频帧数为 0")

    # 2) 均匀采样：用 step 控制抽帧间隔
    # 更高级做法：检测“场景变化”再抽帧（减少冗余，提高有效信息密度）
    step = max(1, total // frame_count)
    paths: list[str] = []

    for idx in range(0, total, step):
        # 3) 跳到指定帧位置读取
        cap.set(cv2.CAP_PROP_POS_FRAMES, idx)
        ok, frame = cap.read()
        if not ok:
            continue
        # 4) 写出临时图片文件（生产建议写到安全的 tmp 目录）
        out = f"_tmp_frame_{len(paths)}.jpg"
        cv2.imwrite(out, frame)
        paths.append(out)
        if len(paths) >= frame_count:
            break

    cap.release()
    return paths


def analyze_video_by_frames(video_path: str) -> dict:
    # 抽帧
    frames = extract_frames(video_path, frame_count=6)
    llm = ChatTongyi(model_name="qwen-vl-max", temperature=0)

    prompt = (
        "你将看到一个视频的关键帧序列。请综合这些帧推断视频内容，并严格输出 JSON：\n"
        '{"summary": string, "scenes": [string], "actions": [string]}'
    )

    # 把多张图片拼到同一次 HumanMessage 里
    content = [{"text": prompt}]
    for p in frames:
        content.append({"image": f"file://{os.path.abspath(p)}"})

    resp = llm.invoke([HumanMessage(content=content)])

    # 清理临时帧
    # 常见坑：异常中断会残留临时文件；生产建议用 try/finally
    for p in frames:
        try:
            os.remove(p)
        except OSError:
            pass

    text = resp.content[0]["text"]
    try:
        return json.loads(text)
    except json.JSONDecodeError:
        return {"_raw": text}


if __name__ == "__main__":
    video_path = "sample.mp4"  # 替换为你的视频路径
    print(analyze_video_by_frames(video_path))

# -*- coding: utf-8 -*-
# 依赖：pip install -U openai-whisper
# 说明：Whisper 依赖本机 ffmpeg，请先安装 ffmpeg
#
# 教学导读：这段在做“音频路线”的最小闭环
# 1) 用 ffmpeg 从视频提取音频（wav, 16kHz, 单声道）
# 2) Whisper 做 ASR（语音转文字）
# 3) 把转录文本交给文本 LLM 输出结构化 JSON（topic/key_points/summary）
#
# 为什么这条路线是 fallback？
# - 画面信息不重要/或抽帧成本过高时，用语音就能覆盖主要信息
# - 当 Vision 输出不稳定时，ASR+Text 往往更稳

import os
import json
import subprocess
import whisper
from langchain_community.chat_models import ChatTongyi
from langchain_core.messages import HumanMessage


def get_ffmpeg_path():
    """获取 ffmpeg 路径"""
    # 教学点：很多同学“代码没问题但就是跑不起来”，原因 99% 是 ffmpeg 不在 PATH
    # 检查系统路径中的 ffmpeg
    try:
        subprocess.run(["ffmpeg", "-version"], capture_output=True, check=True)
        return "ffmpeg"
    except (subprocess.CalledProcessError, FileNotFoundError):
        pass
    
    # 检查用户目录中的 ffmpeg
    home_dir = os.path.expanduser("~")
    ffmpeg_path = os.path.join(home_dir, "ffmpeg")
    try:
        subprocess.run([ffmpeg_path, "-version"], capture_output=True, check=True)
        return ffmpeg_path
    except (subprocess.CalledProcessError, FileNotFoundError):
        pass
    
    return None


def extract_audio(video_path: str, audio_path: str = "_tmp_audio.wav") -> str:
    """从视频中提取音频"""
    ffmpeg_path = get_ffmpeg_path()
    if not ffmpeg_path:
        raise RuntimeError(
            "ffmpeg 未安装！请安装 ffmpeg：\n"
            "macOS: brew install ffmpeg\n"
            "Ubuntu: sudo apt install ffmpeg\n"
            "Windows: 下载 ffmpeg 并添加到 PATH\n\n"
            "快速安装方案：\n"
            "  curl -L https://evermeet.cx/ffmpeg/getrelease/ffmpeg/zip -o ffmpeg.zip\n"
            "  unzip ffmpeg.zip && mv ffmpeg ~/ffmpeg"
        )
    
    # 使用 subprocess 直接调用 ffmpeg
    # 参数说明：
    # -ac 1：单声道（更省资源）
    # -ar 16000：16k 采样率（ASR 常用）
    # -y：覆盖输出文件（避免交互确认卡住脚本）
    try:
        subprocess.run([
            ffmpeg_path, '-i', video_path, 
            '-ac', '1', '-ar', '16000', 
            '-y', audio_path
        ], capture_output=True, check=True)
    except subprocess.CalledProcessError as e:
        raise RuntimeError(f"音频提取失败: {e.stderr.decode() if e.stderr else str(e)}")
    
    return audio_path


def transcribe(audio_path: str) -> str:
    """转录音频为文字"""
    # 教学点：Whisper 默认模型越大越慢；课堂演示可先用 tiny/base
    # 设置 ffmpeg 路径环境变量，让 whisper 能找到 ffmpeg
    ffmpeg_path = get_ffmpeg_path()
    if ffmpeg_path and ffmpeg_path != "ffmpeg":
        # 如果是自定义路径，添加到 PATH
        current_path = os.environ.get('PATH', '')
        os.environ['PATH'] = f"{os.path.dirname(ffmpeg_path)}:{current_path}"
    
    model = whisper.load_model("base")
    r = model.transcribe(audio_path)
    return (r.get("text") or "").strip()


def analyze_video_by_audio(video_path: str) -> dict:
    """分析视频内容（通过音频转录）"""
    # 1) 抽音频
    audio_path = extract_audio(video_path)
    # 2) ASR 转写
    text = transcribe(audio_path)
    
    # 清理临时文件（生产建议 finally）
    try:
        os.remove(audio_path)
    except OSError:
        pass

    # 使用 LLM 分析转录内容
    llm = ChatTongyi(model_name="qwen-max", temperature=0)
    prompt = (
        "请根据以下视频语音转录内容做结构化总结，并严格输出 JSON：\n"
        f"转录内容：{text}\n"
        '{"summary": string, "key_points": [string], "topic": string}'
    )

    resp = llm.invoke([HumanMessage(content=prompt)])
    out = resp.content if isinstance(resp.content, str) else resp.content[0].get("text", "")
    
    try:
        return json.loads(out)
    except json.JSONDecodeError:
        return {"transcript": text, "_raw": out}


if __name__ == "__main__":
    video_path = "lecture.mp4"  # 替换为你的视频路径
    
    # 检查 ffmpeg
    if not get_ffmpeg_path():
        print("❌ ffmpeg 未安装！")
        print("快速安装：")
        print("  curl -L https://evermeet.cx/ffmpeg/getrelease/ffmpeg/zip -o ffmpeg.zip")
        print("  unzip ffmpeg.zip && mv ffmpeg ~/ffmpeg")
        exit(1)
    
    try:
        result = analyze_video_by_audio(video_path)
        print("✅ 视频分析完成：")
        print(json.dumps(result, ensure_ascii=False, indent=2))
    except Exception as e:
        print(f"❌ 分析失败：{e}")
        exit(1)