#!/usr/bin/env python3

# /*
#  * Copyright (C) 2020, KylinSoft Co., Ltd.
#  *
#  * This program is free software: you can redistribute it and/or modify
#  * it under the terms of the GNU General Public License as published by
#  * the Free Software Foundation, either version 3 of the License, or
#  * (at your option) any later version.
#  *
#  * This program is distributed in the hope that it will be useful,
#  * but WITHOUT ANY WARRANTY; without even the implied warranty of
#  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
#  * GNU General Public License for more details.
#  *
#  * You should have received a copy of the GNU General Public License
#  * along with this program.  If not, see <https://www.gnu.org/licenses/>.
#  */

import argparse
import gi
import logging
import os
import stat
import hashlib
import pwd
import grp
import sys
import difflib
from datetime import datetime

gi.require_version('OSTree', '1.0')
from gi.repository import Gio, OSTree

# 全局常量定义

OVL_DIR_PRE = "/sysroot/ostree/pkgs/ovl-"

ETC_LAYER_LOWER = "/etc-ovl/etc-lower/"
ETC_LAYER_UPPER = "/etc-ovl/etc-upper/"

USR_LAYER_LOWER = "/usr-ovl/usr-lower/"
USR_LAYER_UPPER = "/usr-ovl/usr-upper/"

VAR_LAYER_LOWER = "/var-ovl/var-lower/lib/"
VAR_LAYER_UPPER = "/var-ovl/var-upper/lib/"


class MsgCollector:
    """ostree数据收集器类"""

    def __init__(self):
        """初始化sysroot和部署信息"""
        self.sysroot = None
        self.booted_deployment = None
        self.deployments = None
        self._initialize()

    def _initialize(self):
        """初始化sysroot"""
        try:
            # 初始化sysroot
            # logging.info("[MsgCollector] Starting sysroot initialization...")
            self.sysroot = OSTree.Sysroot.new(None)
            # logging.info("[MsgCollector] OSTree.Sysroot created successfully")
            self.sysroot.set_mount_namespace_in_use()
            self.sysroot.initialize()
            # logging.info("[MsgCollector] Sysroot initialized successfully")
            self.sysroot.load()
            # logging.debug("[MsgCollector] Sysroot loaded successfully")

            # 获取当前启动的部署和所有部署
            self.booted_deployment = self.sysroot.get_booted_deployment()
            self.deployments = self.sysroot.get_deployments()

        except Exception as e:
            logging.error(f"[MsgCollector] Initialization failed: {str(e)}")
            self.sysroot = None
            self.booted_deployment = None
            self.deployments = None

    def is_initialized(self):
        """检查是否初始化成功"""
        return self.sysroot is not None and self.booted_deployment is not None

    def get_current_hash(self):
        """获取当前部署的哈希值（包含部署序列号）"""
        if not self.is_initialized():
            logging.error("[MsgCollector] System not properly initialized")
            return None

        try:
            # 获取哈希值和部署序列号
            deployment_hash = self.booted_deployment.get_csum()
            deploy_serial = self.booted_deployment.get_deployserial()

            # 组合成完整的哈希值：哈希值.序列号
            full_hash = f"{deployment_hash}.{deploy_serial}"
            return full_hash
        except Exception as e:
            logging.error(f"[MsgCollector] Error: {str(e)}")
            return None

    def get_rollback_hash(self):
        """获取上一个（rollback）部署的哈希值"""
        if not self.is_initialized():
            logging.error("[MsgCollector] System not properly initialized")
            return None

        try:
            if not self.deployments:
                logging.error("[MsgCollector] No deployments found")
                return None

            # 遍历所有部署，查找第一个与当前启动部署不同的部署（rollback）
            rollback_deployment = None
            for deployment in self.deployments:
                if deployment != self.booted_deployment:
                    rollback_deployment = deployment
                    break

            if rollback_deployment:
                # 获取哈希值和部署序列号
                deployment_hash = rollback_deployment.get_csum()
                deploy_serial = rollback_deployment.get_deployserial()

                # 组合成完整的哈希值：哈希值.序列号
                full_hash = f"{deployment_hash}.{deploy_serial}"
                return full_hash
            else:
                logging.error("[MsgCollector] No rollback deployment found")
                return None

        except Exception as e:
            logging.error(f"[MsgCollector] Error: {str(e)}")
            return None

    @staticmethod
    def _build_deploy_dict(osname, hash_value):
        """根据 osname 和 hash 构建部署信息字典（deploy_dir + usr/var/etc 各层 lower/upper）"""
        return {
            "deploy_dir": f"/sysroot/ostree/deploy/{osname}/deploy/{hash_value}/",
            "ovl_usr_lower": f"{OVL_DIR_PRE}{hash_value}{USR_LAYER_LOWER}",
            "ovl_usr_upper": f"{OVL_DIR_PRE}{hash_value}{USR_LAYER_UPPER}",
            "ovl_var_lower": f"{OVL_DIR_PRE}{hash_value}{VAR_LAYER_LOWER}",
            "ovl_var_upper": f"{OVL_DIR_PRE}{hash_value}{VAR_LAYER_UPPER}",
            "ovl_etc_lower": f"{OVL_DIR_PRE}{hash_value}{ETC_LAYER_LOWER}",
            "ovl_etc_upper": f"{OVL_DIR_PRE}{hash_value}{ETC_LAYER_UPPER}",
        }

    @staticmethod
    def _empty_deploy_dict():
        """构建全 None 的空部署信息字典，每次返回新对象避免共享引用"""
        return {
            "deploy_dir": None, "ovl_usr_lower": None, "ovl_usr_upper": None,
            "ovl_var_lower": None, "ovl_var_upper": None,
            "ovl_etc_lower": None, "ovl_etc_upper": None,
        }

    def get_ostree_message(self):
        """获取当前部署和rollback部署的OS名称、哈希值及各层目录路径
        
        Returns:
            tuple: (osname, deploy_hash, rollback_hash, rollback_dict, current_dict)
                   rollback_dict/current_dict 包含:
                     deploy_dir, ovl_usr_lower, ovl_usr_upper,
                     ovl_var_lower, ovl_var_upper,
                     ovl_etc_lower, ovl_etc_upper
                   如果获取失败，返回 ("kylin", None, None, {}, {})
        """
        default_ret = ("kylin", None, None, self._empty_deploy_dict(), self._empty_deploy_dict())
        try:
            if not (self.is_initialized() and self.booted_deployment):
                logging.warning("[MsgCollector] Failed to get OS name, using default 'kylin'")
                return default_ret

            osname = self.booted_deployment.get_osname()
            deploy_hash = self.get_current_hash()
            rollback_hash = self.get_rollback_hash()

            if not deploy_hash or not rollback_hash:
                logging.warning("[MsgCollector] Failed to get deploy/rollback hash")
                return ("kylin", deploy_hash, rollback_hash, self._empty_deploy_dict(), self._empty_deploy_dict())

            rollback_dict = self._build_deploy_dict(osname, rollback_hash)
            current_dict = self._build_deploy_dict(osname, deploy_hash)

            return osname, deploy_hash, rollback_hash, rollback_dict, current_dict
        except Exception as e:
            logging.error(f"[MsgCollector] Error getting OS name: {str(e)}")
            return default_ret


class FileLayerAnalyzer:
    """排查工具：分析指定文件在 rollback/current 部署各层的差异"""

    ALLOWED_PREFIXES = (
        ("/usr/",     "usr"),
        ("/etc/",     "etc"),
        ("/var/lib/", "var"),
    )

    def __init__(self, collector, report_dir="/tmp"):
        self.collector = collector
        self.report_dir = report_dir

    def _validate_and_route(self, file_path):
        """校验前缀、返回 (层路由键, 层内相对路径)"""
        if not file_path or ".." in file_path.split("/"):
            raise ValueError(f"非法路径: {file_path}")
        if os.path.isdir(file_path):
            raise ValueError(f"路径是目录，不支持分析目录: {file_path}")

        for prefix, layer_key in self.ALLOWED_PREFIXES:
            if file_path.startswith(prefix):
                rel_path = file_path[len(prefix):]
                if not rel_path:
                    raise ValueError(f"路径剥离前缀后为空: {file_path}")
                return layer_key, rel_path

        raise ValueError(
            f"路径必须以 /usr/ /etc/ /var/lib/ 之一开头: {file_path}"
        )

    # 层路由键 → (deploy 路径相对前缀, lower/upper 字典键后缀)
    # deploy_dir 已以 / 结尾，故 deploy_prefix 不带前导斜杠
    _LAYER_ROUTE = {
        "usr": ("usr", "usr"),
        "etc": ("etc", "etc"),
        "var": ("var/lib", "var"),
    }

    def _resolve_layer_paths(self, deploy_dict, layer_key, rel_path):
        """对单个部署的 deploy_dir/lower/upper 三处拼出完整绝对路径"""
        deploy_prefix, dict_suffix = self._LAYER_ROUTE[layer_key]
        return {
            "deploy": f"{deploy_dict['deploy_dir']}{deploy_prefix}/{rel_path}",
            "lower":  f"{deploy_dict['ovl_' + dict_suffix + '_lower']}{rel_path}",
            "upper":  f"{deploy_dict['ovl_' + dict_suffix + '_upper']}{rel_path}",
        }

    def _compute_md5(self, abs_path):
        """分块计算文件 md5"""
        h = hashlib.md5()
        with open(abs_path, "rb") as f:
            for chunk in iter(lambda: f.read(65536), b""):
                h.update(chunk)
        return h.hexdigest()

    def _content_diff(self, path_a, path_b):
        """对两个普通文件做 unified diff，返回 diff 文本行列表"""
        try:
            with open(path_a, "r", encoding="utf-8", errors="replace") as f:
                lines_a = f.readlines()
            with open(path_b, "r", encoding="utf-8", errors="replace") as f:
                lines_b = f.readlines()
        except OSError as e:
            return [f"<读取文件失败: {e}>"]
        return list(difflib.unified_diff(
            lines_a, lines_b,
            fromfile=path_a, tofile=path_b,
            lineterm="",
        ))

    def _stat_layer(self, abs_path):
        """采集单层文件状态：存在性/类型/md5/权限/owner/whiteout/软链接/大小"""
        info = {
            "exists": False, "type": "missing", "md5": None,
            "mode": "N/A", "owner": "N/A",
            "symlink_target": None, "size": None, "path": abs_path,
        }
        if not os.path.lexists(abs_path):
            return info

        info["exists"] = True
        st = os.lstat(abs_path)
        info["mode"] = oct(stat.S_IMODE(st.st_mode))
        info["size"] = st.st_size
        try:
            info["owner"] = f"{pwd.getpwuid(st.st_uid).pw_name}:{grp.getgrgid(st.st_gid).gr_name}"
        except KeyError:
            info["owner"] = f"{st.st_uid}:{st.st_gid}"

        if stat.S_ISLNK(st.st_mode):
            info["type"] = "symlink"
            try:
                info["symlink_target"] = os.readlink(abs_path)
                if not os.path.exists(abs_path):
                    info["type"] = "broken symlink"
            except OSError:
                info["type"] = "broken symlink"
        elif stat.S_ISREG(st.st_mode):
            info["type"] = "regular"
            try:
                info["md5"] = self._compute_md5(abs_path)
            except OSError:
                info["md5"] = "<read error>"
        elif stat.S_ISDIR(st.st_mode):
            info["type"] = "directory"
        elif stat.S_ISCHR(st.st_mode) and st.st_rdev == 0:
            info["type"] = "whiteout"
        else:
            info["type"] = "special"

        return info

    @staticmethod
    def _layer_presence(stat):
        """判断单层逻辑存在情况: 存在 / 缺失 / 被删除(whiteout)"""
        if stat["type"] == "whiteout":
            return "被删除(whiteout)"
        if stat["exists"]:
            return "存在"
        return "缺失"

    DIFF_CLASSIFICATION = {
        ("rollback", "deploy", "lower"):    ("旧系统装机操作", "lower在装机时相对deploy层进行变更"),
        ("rollback", "lower", "upper"):     ("旧系统基于装机lower层的运维操作", "lower->upper修改"),
        ("rollback", "deploy", "upper"):    ("旧系统基于deploy层的运维操作", "deploy->upper修改"),
        ("current", "deploy", "lower"):     ("升级后deploy层+旧装机lower层", "新deploy层vs旧lower(拷贝)"),
        ("current", "lower", "upper"):      ("升级后lower层与upper层的对比", "lower->upper修改|或升级流程修改"),
        ("current", "deploy", "upper"):     ("升级后deploy层与upper层的对比", "该文件在新系统中的upper层变更覆盖了deploy层"),
        ("cross", "deploy", "deploy"):      ("新旧系统deploy层变更", "在新旧系统base层中的该文件发生变化（升级基线变更）"),
        ("cross", "lower", "lower"):        ("异常(lower应一致)", "在lower层中的该文件原封拷贝不应有差异"),
        ("cross", "upper", "upper"):        ("系统升级操作", "在upper层中的该文件在升级过程中被修改"),
    }

    @classmethod
    def _classify_diff(cls, deploy_ctx, layer_a, layer_b):
        """查表返回 (label, reasoning)；查不到返回 ("未知", "")"""
        return cls.DIFF_CLASSIFICATION.get(
            (deploy_ctx, layer_a, layer_b), ("未知", "")
        )

    def _summarize_diff(self, diff):
        """统计 diff 变化，返回摘要句"""
        # 情况1：完全一致
        if diff.get("identical"):
            return "文件完全一致，无变化"

        # 情况2：存在性差异
        if "existence" in diff:
            return diff["existence"]

        # 情况3：类型变化
        if "type" in diff:
            return f"类型变更: {diff['type']}"

        # 收集各维度变化
        parts = []

        # 情况4：内容修改
        if "md5" in diff:
            if "content_diff" in diff and diff["content_diff"]:
                added = sum(1 for l in diff["content_diff"] if l.startswith("+") and not l.startswith("+++"))
                deleted = sum(1 for l in diff["content_diff"] if l.startswith("-") and not l.startswith("---"))
                blocks = sum(1 for l in diff["content_diff"] if l.startswith("@@"))
                parts.append(f"内容修改: 新增 {added} 行, 删除 {deleted} 行, 共 {blocks} 处差异块")
            else:
                parts.append("内容修改但无明确行差异")

        # 情况5：权限变化
        if "mode" in diff:
            parts.append(f"权限 {diff['mode']}")

        # 属主变化
        if "owner" in diff:
            parts.append(f"属主 {diff['owner']}")

        # 软链接变化
        if "symlink" in diff:
            parts.append(f"软链接 {diff['symlink']}")

        return "; ".join(parts) if parts else "无变化"

    def _diff_pair(self, name_a, name_b, a, b):
        """两层状态对比，只返回有差异的维度；全部一致时设 identical=True"""
        diff = {}
        # 逻辑存在性：whiteout（字符设备0:0）视为删除，missing 视为不存在
        a_present = a["exists"] and a["type"] != "whiteout"
        b_present = b["exists"] and b["type"] != "whiteout"
        if not a_present and not b_present:
            if a["type"] == "whiteout" and b["type"] == "whiteout":
                diff["existence"] = f"{name_a} 和 {name_b} 都被删除(whiteout)"
            else:
                diff["existence"] = f"{name_a} 和 {name_b} 都不存在"
            return diff
        if a_present and not b_present:
            if b["type"] == "whiteout":
                diff["existence"] = f"{name_b} 用 whiteout 删除了 {name_a} 的文件"
            else:
                diff["existence"] = f"{name_b} 缺失该文件（无 whiteout）"
            return diff
        if not a_present and b_present:
            if a["type"] == "whiteout":
                diff["existence"] = f"{name_a} 用 whiteout 删除了 {name_b} 的文件"
            else:
                diff["existence"] = f"{name_a} 缺失该文件（无 whiteout）"
            return diff

        # 类型
        if a["type"] != b["type"]:
            diff["type"] = f"{a['type']} → {b['type']}"

        # md5（仅普通文件）
        if a["type"] == "regular" and b["type"] == "regular" and a["md5"] != b["md5"]:
            diff["md5"] = f"内容被修改 (md5: {a['md5']} → {b['md5']})"
            diff["content_diff"] = self._content_diff(a["path"], b["path"])

        # 权限
        if a["mode"] != b["mode"]:
            diff["mode"] = f"{a['mode']} → {b['mode']}"

        # 属主
        if a["owner"] != b["owner"]:
            diff["owner"] = f"{a['owner']} → {b['owner']}"

        # 软链接目标
        if a["type"] == "symlink" or b["type"] == "symlink":
            if a["symlink_target"] != b["symlink_target"]:
                diff["symlink"] = f"链接目标 {a['symlink_target']} → {b['symlink_target']}"

        # 全部一致
        if not diff:
            diff["identical"] = True

        return diff

    def _format_diff_section(self, name_a, name_b, diff, include_content_diff=True, cross_deploy=False):
        """格式化单组 diff 为文本段落。name_a/name_b 已含部署/层名信息。"""
        header = f"--- {name_a} vs {name_b} ---"
        lines = [header]
        # 完全一致时一句话带过（先判断，避免歧义标签）
        if diff.get("identical"):
            lines.append("  忽略xattrs外文件完全一致")
            return "\n".join(lines)
        # 分类和摘要（紧跟 header）
        if "classification" in diff:
            label, reason = diff["classification"]
            prefix = "⚠ " if "异常" in label else ""
            lines.append(f"  [分类] {prefix}{label} | 原因: {reason}")
        if "summary" in diff:
            lines.append(f"  [摘要] {diff['summary']}")
        # 存在性差异（文本已含具体层名）
        if "existence" in diff:
            lines.append(f"  {diff['existence']}")
        # 只输出有差异的维度
        if "type" in diff:
            lines.append(f"  类型: {diff['type']}")
        if "md5" in diff:
            lines.append(f"  md5: {diff['md5']}")
            if include_content_diff and "content_diff" in diff:
                lines.append("  内容差异:")
                for dl in diff["content_diff"]:
                    lines.append(f"    {dl}")
        if "mode" in diff:
            lines.append(f"  权限: {diff['mode']}")
        if "owner" in diff:
            lines.append(f"  属主: {diff['owner']}")
        if "symlink" in diff:
            lines.append(f"  软链接: {diff['symlink']}")
        return "\n".join(lines)

    def _write_report(self, content, file_basename):
        """写报告文件，返回路径；失败返回 None"""
        try:
            ts = datetime.now().strftime("%Y%m%d_%H%M%S")
            safe_name = file_basename.replace("/", "_")
            filename = f"analyze_{safe_name}_{ts}.log"
            path = os.path.join(self.report_dir, filename)
            with open(path, "w") as f:
                f.write(content)
            return path
        except OSError as e:
            logging.error(f"[FileLayerAnalyzer] 报告写入失败: {e}")
            return None

    def analyze(self, file_path):
        """主入口：校验→采集6处状态→层内diff(每部署3组)→跨部署diff(3层)→输出终端+报告"""
        # 校验与路由
        try:
            layer_key, rel_path = self._validate_and_route(file_path)
        except ValueError as e:
            logging.error(f"[FileLayerAnalyzer] {e}")
            return {}

        # 获取部署信息
        osname, deploy_hash, rollback_hash, rollback_dict, current_dict = \
            self.collector.get_ostree_message()

        if not rollback_dict.get("deploy_dir") or not current_dict.get("deploy_dir"):
            logging.error("[FileLayerAnalyzer] 部署信息获取失败，无法分析")
            return {}

        # 解析 6 处路径
        rb_paths = self._resolve_layer_paths(rollback_dict, layer_key, rel_path)
        cur_paths = self._resolve_layer_paths(current_dict, layer_key, rel_path)

        # 采集 6 处状态
        rb_stat = {k: self._stat_layer(v) for k, v in rb_paths.items()}
        cur_stat = {k: self._stat_layer(v) for k, v in cur_paths.items()}

        # 计算每层逻辑存在情况
        rb_presence = {k: self._layer_presence(rb_stat[k]) for k in ("deploy", "lower", "upper")}
        cur_presence = {k: self._layer_presence(cur_stat[k]) for k in ("deploy", "lower", "upper")}

        # 六层全部缺失：文件在系统中不存在，无需生成报告
        all_missing = all(v == "缺失" for v in rb_presence.values()) and \
                      all(v == "缺失" for v in cur_presence.values())
        if all_missing:
            logging.info(f"[FileLayerAnalyzer] 文件 {file_path} 在系统中不存在（rollback/current 的 deploy/lower/upper 六层均缺失）")
            return {"file_path": file_path, "layer": layer_key, "rel_path": rel_path, "not_exist": True}

        # 层内 diff：只对比都"存在"的层对
        def _build_intra(stat, presence, deploy_ctx):
            intra = []
            for a, b, b_label in [("deploy", "lower", "lower"),
                                  ("lower", "upper", "upper"),
                                  ("deploy", "upper", "upper (跨层)")]:
                if presence[a] == "存在" and presence[b] == "存在":
                    diff = self._diff_pair(a, b, stat[a], stat[b])
                    if "existence" not in diff:
                        diff["classification"] = self._classify_diff(deploy_ctx, a, b)
                        diff["summary"] = self._summarize_diff(diff)
                    intra.append({"a": a, "b": b_label, "diff": diff})
            return intra

        rb_intra = _build_intra(rb_stat, rb_presence, "rollback")
        cur_intra = _build_intra(cur_stat, cur_presence, "current")

        # 跨部署 diff（3 层）
        cross_layers = [("deploy", "deploy"), ("lower", "lower"), ("upper", "upper")]
        cross = []
        for la, lb in cross_layers:
            diff = self._diff_pair(f"rollback/{la}", f"current/{lb}", rb_stat[la], cur_stat[lb])
            if "existence" not in diff:
                diff["classification"] = self._classify_diff("cross", la, lb)
                diff["summary"] = self._summarize_diff(diff)
            cross.append(diff)

        result = {
            "file_path": file_path,
            "layer": layer_key,
            "rel_path": rel_path,
            "rollback": {"hash": rollback_hash, "presence": rb_presence, "intra": rb_intra},
            "current": {"hash": deploy_hash, "presence": cur_presence, "intra": cur_intra},
            "cross_deploy": cross,
        }

        # 格式化与输出
        report_text = self._format_full_report(result, include_content_diff=True)
        terminal_text = self._format_full_report(result, include_content_diff=False)
        report_path = self._write_report(report_text, os.path.basename(file_path))
        if report_path:
            logging.info(f"[FileLayerAnalyzer] 分析结果已写入文件: {report_path}")
        logging.info("\n%s", terminal_text)
        logging.info(f"[FileLayerAnalyzer] ===== 分析完成 =====")
        return result

    def _format_suggestions(self, result):
        """基于分类标签生成排查建议，异常优先"""
        suggestions = []

        # 收集跨部署 diff 中有差异的
        cross_layers = [("deploy", "deploy"), ("lower", "lower"), ("upper", "upper")]
        for i, (la, lb) in enumerate(cross_layers):
            d = result["cross_deploy"][i]
            if d.get("identical") or "classification" not in d:
                continue
            label, reason = d["classification"]
            prefix = "⚠ " if "异常" in label else ""
            suggestions.append(f"  [{prefix}{label}] rollback/{la} vs current/{lb} → {reason}")

        # 收集层内 diff 中有差异的
        for deploy_name, key in [("Rollback", "rollback"), ("Current", "current")]:
            d = result[key]
            for item in d["intra"]:
                diff = item["diff"]
                if diff.get("identical") or "classification" not in diff:
                    continue
                label, reason = diff["classification"]
                prefix = "⚠ " if "异常" in label else ""
                suggestions.append(f"  [{prefix}{label}] {deploy_name} {item['a']} vs {item['b']} → {reason}")

        # 异常优先排序
        suggestions.sort(key=lambda s: 0 if "⚠" in s else 1)

        if not suggestions:
            return "===== 排查建议 =====\n  所有层对比均一致，未发现异常"

        lines = ["===== 排查建议 ====="]
        lines.extend(suggestions)
        return "\n".join(lines)

    def _format_full_report(self, result, include_content_diff=True):
        """格式化完整报告；可选择是否包含内容 diff 明细。"""
        lines = []
        lines.append(f"===== 分析文件: {result['file_path']} =====")
        lines.append(f"路由: {result['layer']} 层 | 层内相对路径: {result['rel_path']}")

        for deploy_name, key in [("Rollback", "rollback"), ("Current", "current")]:
            d = result[key]
            lines.append("")
            lines.append(f"########## {deploy_name} 部署 ({d['hash']}) ##########\n")
            # 概览行：每层存在情况
            overview = "  ".join(f"{lk}: {d['presence'][lk]}" for lk in ("deploy", "lower", "upper"))
            lines.append(f"  层存在情况: {overview}")
            # 层内 diff（只输出实际对比的层对）
            if not d["intra"]:
                lines.append("  (仅一层或无层存在该文件，无层内对比)")
            for item in d["intra"]:
                lines.append(self._format_diff_section(
                    item["a"], item["b"], item["diff"],
                    include_content_diff=include_content_diff,
                ))

        lines.append("")
        lines.append("########## ------------------------跨部署 diff (rollback vs current)------------------------ ##########\n")
        layer_names = ["deploy_dir", "lower", "upper"]
        for i, name in enumerate(layer_names):
            lines.append(self._format_diff_section(
                f"rollback/{name}", f"current/{name}", result["cross_deploy"][i],
                include_content_diff=include_content_diff,
            ))

        # 排查建议
        lines.append("")
        lines.append(self._format_suggestions(result))

        report_text = "\n".join(lines)
        return report_text


class DebPackageAnalyzer:
    """按部署层状态文件分析 deb 包版本。"""

    def __init__(self, collector, report_dir="/tmp"):
        self.collector = collector
        self.report_dir = report_dir

    @staticmethod
    def _read_package_version_from_status(status_path, package_name):
        """从 dpkg status 文件中读取指定包版本。"""
        if not os.path.exists(status_path):
            return {
                "state": "missing_status",
                "version": None,
                "message": "status 文件不存在",
            }

        try:
            with open(status_path, "r", encoding="utf-8", errors="replace") as f:
                current_pkg = None
                current_ver = None
                for raw_line in f:
                    line = raw_line.rstrip("\n")
                    if not line:
                        if current_pkg == package_name:
                            return {
                                "state": "found",
                                "version": current_ver,
                                "message": "找到包版本" if current_ver else "找到包但无版本字段",
                            }
                        current_pkg = None
                        current_ver = None
                        continue
                    if line.startswith("Package: "):
                        current_pkg = line.split(": ", 1)[1]
                    elif line.startswith("Version: "):
                        current_ver = line.split(": ", 1)[1]
                if current_pkg == package_name:
                    return {
                        "state": "found",
                        "version": current_ver,
                        "message": "找到包版本" if current_ver else "找到包但无版本字段",
                    }
        except OSError as e:
            return {
                "state": "read_error",
                "version": None,
                "message": f"读取失败: {e}",
            }

        return {
            "state": "not_found",
            "version": None,
            "message": "未找到该包",
        }

    @staticmethod
    def _build_status_result(layer_name, source_name, status_path, query_result):
        return {
            "layer": layer_name,
            "source": source_name,
            "path": status_path,
            "state": query_result["state"],
            "version": query_result["version"],
            "message": query_result["message"],
        }

    def _query_deploy_version(self, deploy_dict, package_name):
        status_path = os.path.join(deploy_dict["deploy_dir"], "var/lib/dpkg/status")
        result = self._read_package_version_from_status(status_path, package_name)
        return self._build_status_result("deploy", "deploy", status_path, result)

    def _query_overlay_version(self, deploy_dict, package_name, source_name):
        status_path = os.path.join(deploy_dict[f"ovl_var_{source_name}"], "dpkg/status")
        result = self._read_package_version_from_status(status_path, package_name)
        return self._build_status_result(source_name, source_name, status_path, result)

    @staticmethod
    def _pick_top_source(lower_result, upper_result):
        if upper_result["version"]:
            return "upper"
        if lower_result["version"]:
            return "lower"
        return None

    @staticmethod
    def _format_version(version):
        return version if version else "无"

    def _format_deploy_section(self, deploy_name, data):
        lines = [f"{deploy_name} hash: {data['hash']}"]
        for layer_name in ("deploy", "lower", "upper"):
            suffix = "(top)" if layer_name == data["top_source"] else ""
            lines.append(
                f"{deploy_name}/{layer_name}{suffix}: "
                f"{self._format_version(data[layer_name]['version'])}"
            )
        return lines

    def _format_full_report(self, result):
        lines = [f"===== 分析 deb 包: {result['package_name']} ====="]
        lines.extend(self._format_deploy_section("rollback", result["rollback"]))
        lines.append("")
        lines.extend(self._format_deploy_section("current", result["current"]))
        return "\n".join(lines)

    def analyze(self, package_name):
        """主入口：按部署层状态文件分析 deb 包版本。"""
        _, deploy_hash, rollback_hash, rollback_dict, current_dict = self.collector.get_ostree_message()
        if not rollback_dict.get("deploy_dir") or not current_dict.get("deploy_dir"):
            logging.error("[DebPackageAnalyzer] 部署信息获取失败，无法分析")
            return {}

        rollback_deploy = self._query_deploy_version(rollback_dict, package_name)
        rollback_lower = self._query_overlay_version(rollback_dict, package_name, "lower")
        rollback_upper = self._query_overlay_version(rollback_dict, package_name, "upper")
        current_deploy = self._query_deploy_version(current_dict, package_name)
        current_lower = self._query_overlay_version(current_dict, package_name, "lower")
        current_upper = self._query_overlay_version(current_dict, package_name, "upper")

        result = {
            "package_name": package_name,
            "rollback": {
                "hash": rollback_hash,
                "deploy": rollback_deploy,
                "lower": rollback_lower,
                "upper": rollback_upper,
                "top_source": self._pick_top_source(rollback_lower, rollback_upper),
            },
            "current": {
                "hash": deploy_hash,
                "deploy": current_deploy,
                "lower": current_lower,
                "upper": current_upper,
                "top_source": self._pick_top_source(current_lower, current_upper),
            },
        }

        report_text = self._format_full_report(result)
        report_path = self._write_report(report_text, f"deb_{package_name}")
        if report_path:
            logging.info(f"[DebPackageAnalyzer] 分析结果已写入文件: {report_path}")
        logging.info("\n%s", report_text)
        logging.info("[DebPackageAnalyzer] ===== 分析完成 =====")
        return result

    def _write_report(self, content, report_name):
        """写报告文件，返回路径；失败返回 None。"""
        try:
            ts = datetime.now().strftime("%Y%m%d_%H%M%S")
            safe_name = report_name.replace("/", "_")
            filename = f"analyze_{safe_name}_{ts}.log"
            path = os.path.join(self.report_dir, filename)
            with open(path, "w") as f:
                f.write(content)
            return path
        except OSError as e:
            logging.error(f"[DebPackageAnalyzer] 报告写入失败: {e}")
            return None


class RefMetadataAnalyzer:
    """读取 rollback/current 部署点的 commit metadata 简表。"""

    METADATA_KEYS = ("os_BuildID", "ostree.ref-binding")

    def __init__(self, collector, report_dir="/tmp"):
        self.collector = collector
        self.report_dir = report_dir
        self.repo = None

    def _get_repo(self):
        if self.repo is None:
            repo = OSTree.Repo.new(Gio.File.new_for_path("/sysroot/ostree/repo"))
            repo.open()
            self.repo = repo
        return self.repo

    def _load_commit_metadata(self, checksum):
        try:
            _, meta = self._get_repo().load_variant(OSTree.ObjectType.COMMIT, checksum)
            metadata = meta[0]
            return {
                "os_BuildID": metadata.get("os_BuildID"),
                "ostree.ref-binding": metadata.get("ostree.ref-binding"),
            }
        except Exception as e:
            logging.error(f"[RefMetadataAnalyzer] 读取 commit metadata 失败: {checksum}: {e}")
            return {
                "os_BuildID": None,
                "ostree.ref-binding": None,
            }

    @staticmethod
    def _format_value(value):
        return value if value not in (None, "", []) else "无"

    def _format_section(self, deploy_name, data):
        lines = [f"{deploy_name} hash: {data['hash']}"]
        for key in self.METADATA_KEYS:
            lines.append(f"{deploy_name}/{key}: {self._format_value(data[key])}")
        return lines

    def _format_full_report(self, result):
        lines = ["===== 分析部署点 metadata =====\n"]
        lines.extend(self._format_section("rollback", result["rollback"]))
        lines.append("")
        lines.extend(self._format_section("current", result["current"]))
        return "\n".join(lines)

    def analyze(self):
        _, deploy_hash, rollback_hash, rollback_dict, current_dict = self.collector.get_ostree_message()
        if not rollback_dict.get("deploy_dir") or not current_dict.get("deploy_dir"):
            logging.error("[RefMetadataAnalyzer] 部署信息获取失败，无法分析")
            return {}

        rollback_checksum = rollback_hash.split(".", 1)[0]
        current_checksum = deploy_hash.split(".", 1)[0]
        rollback_meta = self._load_commit_metadata(rollback_checksum)
        current_meta = self._load_commit_metadata(current_checksum)

        result = {
            "rollback": {"hash": rollback_hash, **rollback_meta},
            "current": {"hash": deploy_hash, **current_meta},
        }

        report_text = self._format_full_report(result)
        report_path = self._write_report(report_text, "refs")
        if report_path:
            logging.info(f"[RefMetadataAnalyzer] 分析结果已写入文件: {report_path}")
        logging.info("\n%s", report_text)
        logging.info("[RefMetadataAnalyzer] ===== 分析完成 =====")
        return result

    def _write_report(self, content, report_name):
        try:
            ts = datetime.now().strftime("%Y%m%d_%H%M%S")
            filename = f"analyze_{report_name}_{ts}.log"
            path = os.path.join(self.report_dir, filename)
            with open(path, "w") as f:
                f.write(content)
            return path
        except OSError as e:
            logging.error(f"[RefMetadataAnalyzer] 报告写入失败: {e}")
            return None


def main():
    """主函数：文件层差异分析工具"""
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s [%(levelname)s] %(message)s",
        datefmt="%Y-%m-%d %H:%M:%S",
    )

    parser = argparse.ArgumentParser(
        description="分析 ostree 部署层中的文件差异或 deb 包版本"
    )
    parser.add_argument("file_path", nargs="?", help="要分析的绝对文件路径")
    parser.add_argument("--deb", dest="deb_package", help="要分析的已安装 deb 包名")
    parser.add_argument("--refs", action="store_true", help="打印 rollback/current 部署点的 metadata")
    args = parser.parse_args()
    selected_modes = sum(bool(x) for x in (args.file_path, args.deb_package, args.refs))
    if selected_modes != 1:
        parser.error("必须且只能选择一种模式：文件路径、--deb <包名> 或 --refs")

    msg_collector = MsgCollector()
    if args.deb_package:
        analyzer = DebPackageAnalyzer(msg_collector, report_dir="/tmp")
        analyzer.analyze(args.deb_package)
    elif args.refs:
        analyzer = RefMetadataAnalyzer(msg_collector, report_dir="/tmp")
        analyzer.analyze()
    else:
        analyzer = FileLayerAnalyzer(msg_collector, report_dir="/tmp")
        analyzer.analyze(args.file_path)


if __name__ == "__main__":
    main()
