技术审计文档

def calculate_file_hash(file_path: Path, algorithm: str = 'sha256') -> str:
    if not file_path or not isinstance(file_path, Path):
        raise RightsError(f"无效的文件路径: {file_path}")
    if not file_path.exists():
        raise RightsError(f"文件不存在: {file_path}")
    if not file_path.is_file():
        raise RightsError(f"路径不是文件: {file_path}")
 
    file_size = file_path.stat().st_size
    if file_size > 10 * 1024 * 1024 * 1024:  # 10GB 限制
        raise RightsError(f"文件过大（超过10GB）: {file_path}")
 
    hash_obj = hashlib.new(algorithm)
    with open(file_path, 'rb') as f:
        chunk_size = 8192
        while True:
            chunk = f.read(chunk_size)
            if not chunk:
                break
            hash_obj.update(chunk)
    return hash_obj.hexdigest()

class MerkleTree:
    """
    Merkle Tree 实现，用于将多个文件的哈希值聚合成单个根哈希
 
    Known limitation (second-preimage resistance):
        This implementation does NOT use domain separation prefixes for leaf vs
        internal nodes (i.e. b'\\x00' for leaves, b'\\x01' for internal nodes as
        recommended by RFC 6962 §2.1).  Adding prefixes would change the root hash
        computation and break backward compatibility with all existing .nep files
        and the verification website (verify.arisfusion.com).  A future tree_version
        bump can introduce domain separation; the current version is safe for our
        threat model (user-submitted files, not adversarial tree construction).
    """
 
    def __init__(self, hash_algorithm: str = 'sha256'):
        self.hash_algorithm = hash_algorithm
        self.leaves: List[str] = []
        self.tree: List[List[str]] = []
        self.root_hash: Optional[str] = None
 
    def add_file_hash(self, file_hash: str) -> None:
        self.leaves.append(file_hash)
 
    def build(self) -> str:
        if not self.leaves:
            raise ValueError("Merkle Tree 没有叶子节点")
 
        if len(self.leaves) == 1:
            self.root_hash = self.leaves[0]
            return self.root_hash
 
        current_level = self.leaves.copy()
        self.tree = [current_level]
 
        while len(current_level) > 1:
            next_level = []
            for i in range(0, len(current_level), 2):
                if i + 1 < len(current_level):
                    combined = current_level[i] + current_level[i + 1]
                else:
                    combined = current_level[i] + current_level[i]
 
                hash_obj = hashlib.new(self.hash_algorithm)
                hash_obj.update(combined.encode('utf-8'))
                parent_hash = hash_obj.hexdigest()
                next_level.append(parent_hash)
 
            self.tree.append(next_level)
            current_level = next_level
 
        self.root_hash = current_level[0]
        return self.root_hash
 
    def get_proof(self, leaf_index: int) -> List[Dict]:
        if not self.tree:
            self.build()
 
        if leaf_index >= len(self.leaves):
            raise IndexError(f"叶子节点索引超出范围: {leaf_index}")
 
        proof = []
        current_index = leaf_index
        current_level = 0
 
        while current_level < len(self.tree) - 1:
            level = self.tree[current_level]
 
            if current_index % 2 == 0:
                sibling_index = current_index + 1
                if sibling_index < len(level):
                    proof.append({'hash': level[sibling_index], 'position': 'right'})
                else:
                    proof.append({'hash': level[current_index], 'position': 'right'})
            else:
                sibling_index = current_index - 1
                proof.append({'hash': level[sibling_index], 'position': 'left'})
 
            current_index = current_index // 2
            current_level += 1
 
        return proof
 
    def verify_proof(self, leaf_hash: str, proof: List[Dict], root_hash: str) -> bool:
        current_hash = leaf_hash
 
        for step in proof:
            sibling_hash = step['hash']
            position = step['position']
 
            if position == 'right':
                combined = current_hash + sibling_hash
            else:
                combined = sibling_hash + current_hash
 
            hash_obj = hashlib.new(self.hash_algorithm)
            hash_obj.update(combined.encode('utf-8'))
            current_hash = hash_obj.hexdigest()
 
        return current_hash == root_hash

class TSAClient:
    PROVIDERS = {
        'freetsa': {
            'name': 'FreeTSA',
            'url': 'https://freetsa.org/tsr',
            'hashname': 'sha256',
            'requires_auth': False,
            'legal_strength': 3,
            'price': 0
        },
        'digicert': {
            'name': 'DigiCert',
            'url': 'http://timestamp.digicert.com',
            'hashname': 'sha256',
            'requires_auth': False,
            'legal_strength': 4,
            'price': 0
        },
        'identrust': {
            'name': 'IdenTrust',
            'url': 'http://timestamp.identrust.com',
            'hashname': 'sha256',
            'requires_auth': False,
            'legal_strength': 4,
            'price': 0
        }
    }
 
    FAILOVER_ORDER: List[str] = ['digicert', 'freetsa', 'identrust']
 
    def __init__(self, provider: str = 'freetsa', custom_url: Optional[str] = None,
                 hashname: str = 'sha256', timeout: int = 30):
        if not RFC3161_AVAILABLE:
            raise ImportError("rfc3161ng 库未安装。请运行: pip install rfc3161ng")
 
        if custom_url:
            self.url = custom_url
            self.provider_name = "Custom TSA"
            self.provider_key = None
        elif provider in self.PROVIDERS:
            config = self.PROVIDERS[provider]
            self.url = config['url']
            self.provider_name = config['name']
            self.provider_key = provider
            hashname = config['hashname']
        else:
            raise ValueError(f"未知的 TSA 提供商: {provider}")
 
        self.hashname = hashname
        self.timeout = timeout
 
        try:
            self.stamper = rfc3161ng.RemoteTimestamper(
                url=self.url, hashname=self.hashname, timeout=self.timeout
            )
        except Exception as e:
            self.stamper = None
            self._init_error = str(e)
 
    def _call_with_retry(self, hash_bytes: bytes, max_retries: int = 3) -> bytes:
        providers_to_try = []
        if self.provider_key:
            providers_to_try.append(self.provider_key)
            for p in self.FAILOVER_ORDER:
                if p != self.provider_key:
                    providers_to_try.append(p)
        else:
            providers_to_try = [None]
 
        last_error = None
        for provider_key in providers_to_try:
            if provider_key is not None:
                config = self.PROVIDERS[provider_key]
                url = config['url']
                hashname = config['hashname']
                provider_name = config['name']
            else:
                url = self.url
                hashname = self.hashname
                provider_name = self.provider_name
 
            for attempt in range(max_retries):
                try:
                    stamper = rfc3161ng.RemoteTimestamper(
                        url=url, hashname=hashname, timeout=self.timeout
                    )
                    tsr_token = stamper(digest=hash_bytes)
                    if provider_key is not None:
                        self.provider_name = provider_name
                        self.provider_key = provider_key
                        self.url = url
                        self.stamper = stamper
                    return tsr_token
                except Exception as e:
                    last_error = e
                    if attempt < max_retries - 1:
                        time.sleep(2 ** attempt)
 
        raise Exception(
            f"所有 TSA 提供商均失败: {last_error}"
        )
 
    def timestamp_hash(self, hash_value: str, output_path: Path) -> Dict:
        hash_bytes = bytes.fromhex(hash_value)
        tsr_token = self._call_with_retry(hash_bytes)
 
        output_path = Path(output_path)
        output_path.parent.mkdir(parents=True, exist_ok=True)
        with open(output_path, 'wb') as f:
            f.write(tsr_token)
 
        tsa_timestamp = datetime.now().isoformat()
        tsa_issuer = self.provider_name
        try:
            from asn1crypto import tsp, cms
            signed_data = None
            try:
                ts_resp = tsp.TimeStampResp.load(tsr_token)
                signed_data = ts_resp['time_stamp_token']['content']
            except (ValueError, KeyError, TypeError):
                try:
                    ci = cms.ContentInfo.load(tsr_token)
                    if ci['content_type'].native == 'signed_data':
                        signed_data = ci['content']
                except (ValueError, KeyError, TypeError):
                    pass
 
            if signed_data:
                tst_info = signed_data['encap_content_info']['content'].parsed
                gen_time = tst_info['gen_time'].native
                if gen_time:
                    tsa_timestamp = gen_time.isoformat()
 
                try:
                    signer_infos = signed_data['signer_infos']
                    if signer_infos:
                        sid = signer_infos[0]['sid']
                        if sid.name == 'issuer_and_serial_number':
                            for rdn in sid.chosen['issuer'].chosen:
                                for attr in rdn:
                                    if attr['type'].dotted == '2.5.4.3':
                                        tsa_issuer = attr['value'].native
                                        break
                except (KeyError, IndexError, ValueError):
                    pass
        except ImportError:
            pass
 
        return {
            'success': True,
            'timestamp': tsa_timestamp,
            'hash': hash_value,
            'issuer': tsa_issuer,
            'tsr_path': str(output_path),
            'algorithm': self.hashname.upper(),
        }
 
    def verify_tsr(self, tsr_path: Path, data: Optional[bytes] = None,
                   digest: Optional[bytes] = None) -> Dict:
        with open(tsr_path, 'rb') as f:
            tsr_token = f.read()
 
        if digest is not None:
            verified = rfc3161ng.check_timestamp(tsr_token, digest=digest)
            if verified:
                return {
                    'valid': True,
                    'message': '时间戳验证通过（数据完整性已确认）',
                    'issuer': self._extract_issuer_from_tsr(tsr_token),
                }
            else:
                return {'valid': False, 'message': '时间戳验证失败：哈希值不匹配'}
        elif data is not None:
            verified = rfc3161ng.check_timestamp(tsr_token, data=data)
            if verified:
                return {
                    'valid': True,
                    'message': '时间戳验证通过（数据完整性已确认）',
                    'issuer': self._extract_issuer_from_tsr(tsr_token),
                }
            else:
                return {'valid': False, 'message': '时间戳验证失败：哈希值不匹配'}
        else:
            return self._verify_tsr_structure(tsr_token)

def batch_protect_works(
    file_paths: List[Path],
    author_name: str,
    inspiration: Optional[str] = None,
    output_dir: Optional[Path] = None,
    password: Optional[str] = None,
    progress_callback=None,
    tsa_provider: str = 'digicert',
    tsa_timeout: int = 30,
    cert_mode: str = 'simple',
) -> Dict:
    from .utils import build_merkle_tree_from_files
    from .rights_packer import RightsPacker
    from .pdf_generator import PDFGenerator
 
    if output_dir is None:
        output_dir = Path.cwd() / "digital_evidence"
    output_dir = Path(output_dir)
    output_dir.mkdir(parents=True, exist_ok=True)
 
    # Step 1: Merkle Tree
    tree = build_merkle_tree_from_files(file_paths, progress_callback=hash_progress)
    root_hash = tree.root_hash
    file_hashes = dict(tree.file_hashes)
 
    # Step 2: 感知哈希（非阻塞）
    fingerprints = []
    for file_path in file_paths:
        if file_path.suffix.lower() in {'.jpg', '.jpeg', '.png', '.webp', '.bmp'}:
            try:
                from .fingerprint import compute_fingerprint
                fp = compute_fingerprint(file_path, file_sha256=file_hashes.get(str(file_path)))
                fingerprints.append(fp)
            except Exception as e:
                logger.debug("Skipping perceptual hash for %s: %s", file_path.name, e)
 
    # Step 3: Manifest
    packer = RightsPacker(output_dir / "evidence.nep", password=password)
    manifest_data = packer.create_manifest(
        author_name=author_name, inspiration=inspiration,
        works=works, file_hashes=file_hashes
    )
 
    # Step 4: TSA 时间戳
    tsa_binary_path = output_dir / "proof.tsa"
    local_json_path = output_dir / "proof.json"
    timestamp_file = tsa_binary_path
 
    try:
        from .tsa_client import TSAClient
        tsa_client = TSAClient(provider=tsa_provider, timeout=tsa_timeout)
        tsa_result = tsa_client.timestamp_hash(root_hash, tsa_binary_path)
 
        if tsa_result['success']:
            timestamp_file = tsa_binary_path
            timestamp_info = {
                'timestamp': tsa_result['timestamp'],
                'hash': root_hash,
                'issuer': tsa_result['issuer'],
                'algorithm': tsa_result['algorithm'],
                'valid': True,
            }
        else:
            timestamp_file = local_json_path
            timestamp_info = {
                'timestamp': datetime.now().isoformat(),
                'hash': root_hash,
                'issuer': 'Nephele Workshop (本地)',
                'local_only': True,
                'error': tsa_result['message']
            }
            with open(local_json_path, 'w', encoding='utf-8') as f:
                json.dump(timestamp_info, f, indent=2, ensure_ascii=False)
    except ImportError:
        timestamp_file = local_json_path
        timestamp_info = {
            'timestamp': datetime.now().isoformat(),
            'hash': root_hash,
            'issuer': 'Nephele Workshop (本地)',
            'local_only': True,
            'note': '需安装 rfc3161ng 库以使用 FreeTSA 服务'
        }
        with open(local_json_path, 'w', encoding='utf-8') as f:
            json.dump(timestamp_info, f, indent=2, ensure_ascii=False)
 
    # Step 5: Manifest 加固
    manifest_data['merkle_root'] = root_hash
    manifest_data['cert_mode'] = cert_mode
 
    works_map = {}
    for idx, fp in enumerate(file_paths):
        works_map[f"works/{idx:03d}_{fp.name}"] = file_hashes.get(str(fp), "")
    manifest_data['works_map'] = works_map
 
    _manifest_for_hash = {k: v for k, v in manifest_data.items() if k != 'manifest_sha256'}
    _manifest_json = json.dumps(_manifest_for_hash, ensure_ascii=False, sort_keys=True, default=str)
    manifest_data['manifest_sha256'] = hashlib.sha256(_manifest_json.encode('utf-8')).hexdigest()
 
    # Step 6: PDF
    pdf_gen = PDFGenerator(pdf_path, cert_mode=cert_mode)
    pdf_gen.generate(
        manifest_data=manifest_data, root_hash=root_hash,
        timestamp_info=timestamp_info, verification_url=verification_url,
        locale="zh_CN", image_paths=image_paths,
    )
 
    # Step 7: Pack .nep
    nep_path = packer.pack(
        manifest_data=manifest_data, thumbnail_path=thumbnail_path,
        timestamp_file=timestamp_file, pdf_report=pdf_path,
        source_files=file_paths,
    )
 
    return {
        'success': True,
        'nep_path': str(nep_path),
        'root_hash': root_hash,
        'file_count': len(file_paths),
        'message': f"数字存证完成，共处理 {len(file_paths)} 个文件"
    }

@staticmethod
def _write_zip_contents(zipf, manifest_data, thumbnail_path, timestamp_file,
                        pdf_report, additional_files, source_files):
    # 1. manifest.json
    manifest_json = json.dumps(manifest_data, indent=2, ensure_ascii=False)
    zipf.writestr('manifest.json', manifest_json.encode('utf-8'))
 
    # 2. 缩略图
    if thumbnail_path and thumbnail_path.exists():
        zipf.write(thumbnail_path, 'thumbnail.jpg')
 
    # 3. 时间戳（保持原始扩展名区分等级）
    if timestamp_file and timestamp_file.exists():
        archive_name = 'proof.tsa' if timestamp_file.suffix == '.tsa' else 'proof.json'
        zipf.write(timestamp_file, archive_name)
 
    # 4. PDF 报告
    if pdf_report and pdf_report.exists():
        zipf.write(pdf_report, 'VerificationReport.pdf')
 
    # 5. 原始作品 → works/ 目录，索引命名防撞名
    if source_files:
        for idx, file_path in enumerate(source_files):
            if isinstance(file_path, str):
                file_path = Path(file_path)
            if file_path.exists() and file_path.is_file():
                zipf.write(file_path, f'works/{idx:03d}_{file_path.name}')
 
    # 6. 额外文件
    if additional_files:
        for file_path in additional_files:
            if file_path.exists():
                zipf.write(file_path, f'additional/{file_path.name}')

def verify_evidence_package(tsa_path: Path, file_paths: List[Path]) -> Tuple[bool, Dict]:
    from .utils import build_merkle_tree_from_files
 
    if not tsa_path.exists():
        return False, {'valid': False, 'message': f"时间戳文件不存在: {tsa_path}"}
 
    missing = [str(p) for p in file_paths if not p.exists()]
    if missing:
        return False, {'valid': False, 'message': f"原始文件缺失: {', '.join(missing)}"}
 
    # Step 1: 重算 Merkle Tree
    tree = build_merkle_tree_from_files(file_paths)
    computed_root = tree.root_hash
 
    # Step 2: 本地时间戳检查
    suffix = tsa_path.suffix.lower()
    if suffix == '.json':
        tsr_result = parse_tsr(tsa_path)
        tsr_hash = tsr_result.get('hash', tsr_result.get('work_identity', ''))
        if computed_root.lower() == tsr_hash.lower():
            return False, {
                'valid': False,
                'message': "本地时间戳无第三方签名，不具备密码学证明力。文件哈希一致但无法证明时间。",
                'root_hash': computed_root,
                'local_only': True,
            }
        else:
            return False, {
                'valid': False,
                'message': "验证失败：Merkle Root 与本地时间戳记录不匹配",
                'computed_root': computed_root,
            }
 
    # Step 3: RFC 3161 TSR — 密码学签名验证
    computed_digest = bytes.fromhex(computed_root)
    try:
        from .tsa_client import TSAClient
        tsa_client = TSAClient()
        verify_result = tsa_client.verify_tsr(tsa_path, digest=computed_digest)
 
        if verify_result.get('valid'):
            tsr_result = parse_tsr(tsa_path)
            return True, {
                'valid': True,
                'message': f"深度验证通过：{len(file_paths)} 个文件的 Merkle Root 与 TSA 签名匹配",
                'timestamp': tsr_result.get('timestamp'),
                'issuer': tsr_result.get('issuer'),
                'root_hash': computed_root,
            }
        else:
            return False, {
                'valid': False,
                'message': f"TSA 签名验证失败：{verify_result.get('message', 'unknown')}",
                'root_hash': computed_root,
            }
    except ImportError:
        # rfc3161ng 未安装，降级为结构性比对
        tsr_result = parse_tsr(tsa_path)
        tsr_hash_raw = tsr_result.get('hash', '')
        tsr_hash = tsr_hash_raw.split(':', 1)[1] if ':' in tsr_hash_raw else tsr_hash_raw
 
        if computed_root.lower() == tsr_hash.lower():
            return True, {
                'valid': True,
                'message': "结构验证通过（安装 rfc3161ng 可启用密码学签名验证）",
                'partial_verification': True,
                'root_hash': computed_root,
            }
        else:
            return False, {
                'valid': False,
                'message': "验证失败：Merkle Root 与 TSR 记录的哈希不匹配",
                'computed_root': computed_root,
                'tsr_hash': tsr_hash,
            }

async def _async_screenshot(self, path: Optional[str]) -> dict:
    import tempfile
    if not path:
        path = os.path.join(tempfile.gettempdir(), "nephele_screenshot.png")
    await self._page.screenshot(path=path, full_page=False)
    return {"success": True, "message": f"截图已保存: {path}", "output_path": path}

def capture(self, url: str, progress_callback=None) -> Dict:
    total_phases = 5
    self._record(f"Starting evidence capture for: {url}")
 
    try:
        # Phase 1: Environment + DNS + TLS
        environment = self.collect_environment()
        dns_info = self.resolve_dns(url)
        tls_info = self.capture_tls_certificate(url)
 
        # Phase 2: Browser capture
        artifacts = self.capture_page(url, progress_callback=None)
 
        # Phase 3: Hash all artifacts
        file_hashes = self.hash_artifacts(artifacts)
 
        # Phase 4: Manifest
        log_path = Path(self.save_log())
        self._log_committed = True
        file_hashes[log_path.name] = {
            "sha256": self._sha256_file(log_path),
            "size": log_path.stat().st_size,
            "type": "operation_log",
        }
 
        manifest = self.generate_manifest(
            target_url=url, environment=environment,
            dns_info=dns_info, artifacts=artifacts,
            file_hashes=file_hashes, tls_info=tls_info,
        )
 
        # Phase 5: Timestamp
        ts_info = self.timestamp_manifest(manifest)
 
        # Phase 6: Package .nep
        nep_path = self._package_nep()
 
        return {
            "success": True,
            "evidence_id": self._evidence_id,
            "output_dir": str(self._output_dir),
            "nep_path": str(nep_path),
            "manifest": manifest,
            "timestamp_info": ts_info,
            "message": f"URL evidence captured"
        }

def capture_tls_certificate(self, url: str) -> Dict:
    import ssl
    from urllib.parse import urlparse
 
    parsed = urlparse(url if "://" in url else f"https://{url}")
    hostname = parsed.hostname or ""
    port = parsed.port or 443
 
    if not hostname:
        return {"error": "Invalid hostname"}
 
    try:
        ctx = ssl.create_default_context()
        with ctx.wrap_socket(
            socket.socket(socket.AF_INET, socket.SOCK_STREAM),
            server_hostname=hostname,
        ) as sock:
            sock.settimeout(10)
            sock.connect((hostname, port))
            cert = sock.getpeercert()
            cert_der = sock.getpeercert(binary_form=True)
 
        cert_path = self._output_dir / "server_certificate.der"
        cert_path.write_bytes(cert_der)
 
        # ... PEM 转换和字段提取 ...
        return cert_info
    except Exception as e:
        return {"error": str(e)}

_CAPTCHA_KEYWORDS = (
    "验证码", "验证", "captcha", "verify", "challenge",
    "human verification", "robot", "机器人",
)
 
def _is_captcha_page(self, title: str, url: str) -> bool:
    text = (title or "").lower()
    url_lower = (url or "").lower()
    for kw in self._CAPTCHA_KEYWORDS:
        if kw in text or kw in url_lower:
            return True
    return False

class WaterMark:
    def __init__(self, password_wm=1, password_img=1, block_shape=(4, 4), mode='common', processes=None):
        self.bwm_core = WaterMarkCore(password_img=password_img, mode=mode, processes=processes)
        self.password_wm = password_wm
        self.wm_bit = None
        self.wm_size = 0
 
    def read_img(self, filename=None, img=None):
        if img is None:
            img = cv2.imread(filename, flags=cv2.IMREAD_UNCHANGED)
        self.bwm_core.read_img_arr(img=img)
        return img
 
    def read_wm(self, wm_content, mode='img'):
        if mode == 'bit':
            self.wm_bit = np.array(wm_content)
        # ... img / str 模式省略 ...
 
        self.wm_size = self.wm_bit.size
        # 水印加密：用 password_wm 作为种子对 bit 序列做伪随机置乱
        np.random.RandomState(self.password_wm).shuffle(self.wm_bit)
        self.bwm_core.read_wm(self.wm_bit)
 
    def embed(self, filename=None, compression_ratio=None):
        embed_img = self.bwm_core.embed()
        if filename is not None:
            cv2.imwrite(filename=filename, img=embed_img)
        return embed_img
 
    def extract_decrypt(self, wm_avg):
        # 逆置乱：根据相同的 seed 生成相同的 shuffle index，再逆序还原
        wm_index = np.arange(self.wm_size)
        np.random.RandomState(self.password_wm).shuffle(wm_index)
        wm_avg[wm_index] = wm_avg.copy()
        return wm_avg
 
    def extract(self, filename=None, embed_img=None, wm_shape=None, mode='img'):
        if filename is not None:
            embed_img = cv2.imread(filename, flags=cv2.IMREAD_COLOR)
        self.wm_size = np.array(wm_shape).prod()
 
        if mode in ('str', 'bit'):
            wm_avg = self.bwm_core.extract_with_kmeans(img=embed_img, wm_shape=wm_shape)
        else:
            wm_avg = self.bwm_core.extract(img=embed_img, wm_shape=wm_shape)
 
        wm = self.extract_decrypt(wm_avg=wm_avg)
        return wm

class WaterMarkCore:
    def __init__(self, password_img=1, mode='common', processes=None):
        self.block_shape = np.array([4, 4])
        self.password_img = password_img
        self.d1, self.d2 = 36, 20  # 量化步长：越大鲁棒性越强，但失真越大
        self.pool = AutoPool(mode=mode, processes=processes)

    def read_img_arr(self, img):
        # 处理透明图
        self.alpha = None
        if img.shape[2] == 4:
            if img[:, :, 3].min() < 255:
                self.alpha = img[:, :, 3]
                img = img[:, :, :3]
 
        # BGR -> YUV，补白边使像素变偶数（DWT 要求）
        self.img = img.astype(np.float32)
        self.img_shape = self.img.shape[:2]
        self.img_YUV = cv2.copyMakeBorder(
            cv2.cvtColor(self.img, cv2.COLOR_BGR2YUV),
            0, self.img.shape[0] % 2, 0, self.img.shape[1] % 2,
            cv2.BORDER_CONSTANT, value=(0, 0, 0)
        )
 
        # 对 Y/U/V 三个通道分别做 1 级 Haar DWT
        self.ca_shape = [(i + 1) // 2 for i in self.img_shape]
        self.ca_block_shape = (
            self.ca_shape[0] // self.block_shape[0],
            self.ca_shape[1] // self.block_shape[1],
            self.block_shape[0], self.block_shape[1]
        )
 
        for channel in range(3):
            self.ca[channel], self.hvd[channel] = dwt2(
                self.img_YUV[:, :, channel], 'haar'
            )
            # 将 CA（近似系数）转为 4D 分块数组
            self.ca_block[channel] = np.lib.stride_tricks.as_strided(
                self.ca[channel].astype(np.float32),
                self.ca_block_shape,
                strides=4 * np.array([
                    self.ca_shape[1] * self.block_shape[0],
                    self.block_shape[1], self.ca_shape[1], 1
                ])
            )

    def block_add_wm_slow(self, arg):
        block, shuffler, i = arg
        wm_1 = self.wm_bit[i % self.wm_size]
 
        # Step 1: 对 4x4 块做 DCT
        block_dct = dct(block)
 
        # Step 2: flatten 后按 shuffler 打乱顺序（块内置乱）
        block_dct_shuffled = block_dct.flatten()[shuffler].reshape(self.block_shape)
 
        # Step 3: SVD 分解
        u, s, v = svd(block_dct_shuffled)
 
        # Step 4: 在奇异值上嵌入 1 bit 水印
        # 量化公式：把 s[0] 量化到 d1 的整数倍，再根据 wm_1 偏移 1/4 个步长
        s[0] = (s[0] // self.d1 + 1/4 + 1/2 * wm_1) * self.d1
        if self.d2:
            s[1] = (s[1] // self.d2 + 1/4 + 1/2 * wm_1) * self.d2
 
        # Step 5: 逆 SVD
        block_dct_flatten = np.dot(u, np.dot(np.diag(s), v)).flatten()
 
        # Step 6: 逆置乱
        block_dct_flatten[shuffler] = block_dct_flatten.copy()
 
        # Step 7: 逆 DCT
        return idct(block_dct_flatten.reshape(self.block_shape))

s'[0] = (floor(s[0] / d1) + 1/4 + 1/2 * w) * d1

w = 1  if (s[0] mod d1) > (d1 / 2)  else 0

    def block_get_wm_slow(self, args):
        block, shuffler = args
        block_dct_shuffled = dct(block).flatten()[shuffler].reshape(self.block_shape)
        u, s, v = svd(block_dct_shuffled)
 
        # 从 s[0] 提取 bit
        wm = (s[0] % self.d1 > self.d1 / 2) * 1
        if self.d2:
            # s[1] 作为辅助，加权平均
            tmp = (s[1] % self.d2 > self.d2 / 2) * 1
            wm = (wm * 3 + tmp * 1) / 4
        return wm

    def embed(self):
        self.init_block_index()
        embed_ca = copy.deepcopy(self.ca)
 
        # 生成块选择置乱序列（跨块置乱）
        self.idx_shuffle = random_strategy1(
            self.password_img, self.block_num,
            self.block_shape[0] * self.block_shape[1]
        )
 
        for channel in range(3):
            # 对每个块并行执行 block_add_wm
            tmp = self.pool.map(self.block_add_wm, [
                (self.ca_block[channel][self.block_index[i]], self.idx_shuffle[i], i)
                for i in range(self.block_num)
            ])
 
            # 写回 4D 数组
            for i in range(self.block_num):
                self.ca_block[channel][self.block_index[i]] = tmp[i]
 
            # 4D -> 2D，拼接回 CA 子带
            self.ca_part[channel] = np.concatenate(np.concatenate(self.ca_block[channel], 1), 1)
            embed_ca[channel][:self.part_shape[0], :self.part_shape[1]] = self.ca_part[channel]
 
            # 逆 DWT
            embed_YUV[channel] = idwt2((embed_ca[channel], self.hvd[channel]), "haar")
 
        # 合并三通道，YUV -> BGR，裁剪回原始尺寸
        embed_img_YUV = np.stack(embed_YUV, axis=2)
        embed_img_YUV = embed_img_YUV[:self.img_shape[0], :self.img_shape[1]]
        embed_img = cv2.cvtColor(embed_img_YUV, cv2.COLOR_YUV2BGR)
        embed_img = np.clip(embed_img, a_min=0, a_max=255)
 
        if self.alpha is not None:
            embed_img = cv2.merge([embed_img.astype(np.uint8), self.alpha])
        return embed_img

def one_dim_kmeans(inputs):
    threshold = 0
    e_tol = 10 ** (-6)
    center = [inputs.min(), inputs.max()]
    for i in range(300):
        threshold = (center[0] + center[1]) / 2
        is_class01 = inputs > threshold
        center = [inputs[~is_class01].mean(), inputs[is_class01].mean()]
        if np.abs((center[0] + center[1]) / 2 - threshold) < e_tol:
            threshold = (center[0] + center[1]) / 2
            break
    is_class01 = inputs > threshold
    return is_class01

def random_strategy1(seed, size, block_shape):
    return np.random.RandomState(seed) \
        .random(size=(size, block_shape)) \
        .argsort(axis=1)

WATERMARK_BYTES = 32
WATERMARK_BITS = WATERMARK_BYTES * 8
_WM_PASSWORD_IMG = <REDACTED>   # 图片置乱种子
_WM_PASSWORD_WM = <REDACTED>    # 水印加密密钥
 
 
def _text_to_bits(text: str) -> list[int]:
    raw = text.encode("utf-8")[:WATERMARK_BYTES]
    padded = raw.ljust(WATERMARK_BYTES, b"\x00")
    bits = []
    for byte in padded:
        for i in range(7, -1, -1):
            bits.append((byte >> i) & 1)
    return bits
 
 
def _bits_to_text(bits: list) -> str:
    raw = bytearray()
    for i in range(0, len(bits), 8):
        chunk = bits[i:i+8]
        if len(chunk) < 8:
            break
        val = 0
        for b in chunk:
            val = (val << 1) | (1 if b > 0.5 else 0)
        raw.append(val)
    return raw.rstrip(b"\x00").decode("utf-8", errors="replace")

class WatermarkEngine:
    _instance = None
 
    def __new__(cls):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
        return cls._instance
 
    def embed(self, image: Image.Image, text: str) -> Image.Image:
        try:
            from blind_watermark import WaterMark
 
            rgb = image.convert("RGB")
            alpha = image.split()[3] if image.mode == "RGBA" else None
 
            with tempfile.TemporaryDirectory() as tmpdir:
                orig_path = str(Path(tmpdir) / "orig.png")
                wm_path = str(Path(tmpdir) / "watermarked.png")
                rgb.save(orig_path, format="PNG")
 
                bits = _text_to_bits(text)
 
                bwm = WaterMark(
                    password_img=_WM_PASSWORD_IMG, password_wm=_WM_PASSWORD_WM
                )
                bwm.read_img(orig_path)
                bwm.read_wm(np.array(bits), mode="bit")
                bwm.embed(wm_path)
 
                result_img = Image.open(wm_path).convert("RGB")
 
                # Round-trip verify
                extracted_bits = WaterMark(
                    password_img=_WM_PASSWORD_IMG, password_wm=_WM_PASSWORD_WM
                ).extract(wm_path, wm_shape=WATERMARK_BITS, mode="bit")
                extracted_text = _bits_to_text(extracted_bits)
 
                if extracted_text == text[:WATERMARK_BYTES]:
                    logger.info("[Watermark] Verified: '%s'", extracted_text)
                else:
                    logger.warning(
                        "[Watermark] Verify mismatch: '%s' -> '%s'",
                        text[:WATERMARK_BYTES], extracted_text,
                    )
 
            if alpha:
                result_img = result_img.convert("RGBA")
                result_img.putalpha(alpha)
            return result_img
 
        except Exception as e:
            logger.error("Embed failed: %s", e, exc_info=True)
            return image

    def extract(self, image: Image.Image) -> Optional[str]:
        try:
            from blind_watermark import WaterMark
            rgb = image.convert("RGB")
            bit_len = WATERMARK_BITS
 
            with tempfile.TemporaryDirectory() as tmpdir:
                img_path = str(Path(tmpdir) / "check.png")
                rgb.save(img_path, format="PNG")
 
                extracted_bits = WaterMark(
                    password_img=_WM_PASSWORD_IMG, password_wm=_WM_PASSWORD_WM
                ).extract(img_path, wm_shape=bit_len, mode="bit")
 
            text = _bits_to_text(extracted_bits)
            return text.strip() if text.strip() else None
        except Exception as e:
            logger.warning("Extract failed: %s", e)
            return None

def protect_image(
    image: Image.Image,
    level: str = "none",
    copyright_info: str = "ARIS",
) -> Image.Image:
    level = LEVEL_ALIASES.get(level, level)
    if level == "invisible":
        return WatermarkEngine().embed(image, copyright_info)
    return image
 
 
def extract_watermark(image: Image.Image) -> Optional[str]:
    return WatermarkEngine().extract(image)

def pack_image(
    input_path: str,
    watermark_path: Optional[str] = None,
    output_dir: Optional[str] = None,
    watermark_mode: str = "center",
    watermark_opacity: float = 0.3,
    preview_max_size: int = 1920,
    thumbnail_max_size: int = 500,
    protection_level: str = "none",
    copyright_info: str = "© ArisFusion Studio",
    output_folder_name: str = "Delivery_Pack",
) -> dict:
    ...
    packer = DeliveryPacker(
        ...,
        protection_level=protection_level,
        copyright_info=copyright_info,
    )
    result_dir, results = packer.process_image(...)
    ...

class WatermarkExtractWorker(QThread):
    finished = Signal(str)  # watermark result or empty string
    logMessage = Signal(str, str)
 
    def run(self):
        try:
            ensure_src_path()
            from PIL import Image
            from tools.packer.watermark_protection import extract_watermark
 
            img_path = Path(self.image_path)
            if not img_path.exists():
                self.finished.emit("")
                return
 
            image = Image.open(img_path)
            watermark = extract_watermark(image)
 
            if watermark:
                self.finished.emit(watermark)
            else:
                self.finished.emit("")
        except Exception as e:
            self.finished.emit("")

用户选择图片
  -> core/workers/ai_detector_worker.py
  -> tools/validator/logic.py:MetaDataDetector.detect()
       ├── Pillow 读取 PNG/JPEG/WebP/TIFF 元数据（PNG info / EXIF）
       ├── tools/validator/c2pa_verifier.py:verify_c2pa_file()（官方 SDK）
       └── 原始字节扫描（JUMBF / APP11 fallback）
  -> 返回 {status, reason, tool, evidence}
  -> UI 映射证据标签

{
    "status": "ai" | "unknown" | "human" | "error",
    "reason": str,
    "tool": str,
    "evidence": str,
}

class MetaDataDetector:
    """
    Detects AI generation metadata from image files using heuristic analysis
    of EXIF, PNG info chunks, and generation parameters.
    """
 
    # 1. 明确的软件签名 (强特征，优先匹配专有字符串)
    # 顺序：Midjourney 先于 Gemini，以避免误判
    AI_SOFTWARE_SIGNATURES = {
        "Midjourney": [
            re.compile(r"job id:\s*[a-f0-9\-]+", re.IGNORECASE),  # 专有 Job ID
            re.compile(r"--ar\s*\d+:\d+", re.IGNORECASE),         # --ar 参数
            re.compile(r"--v\s*\d+", re.IGNORECASE),              # --v 参数
            re.compile(r"--stylize\s*\d+", re.IGNORECASE),        # --stylize
            re.compile(r"midjourney", re.IGNORECASE),             # 显式名称
            re.compile(r"mj v", re.IGNORECASE),
            re.compile(r"mj_", re.IGNORECASE),
        ],
        "ComfyUI": [
            re.compile(r"comfyui", re.IGNORECASE),
            re.compile(r"comfyland", re.IGNORECASE),
        ],
        "Gemini (Google)": [
            re.compile(r"gemini", re.IGNORECASE),
            re.compile(r"google deepmind", re.IGNORECASE),
            re.compile(r"generated by google", re.IGNORECASE),
            re.compile(r"google imagen", re.IGNORECASE),
            re.compile(r"imagen by google", re.IGNORECASE),
            re.compile(r"synthid", re.IGNORECASE),
            re.compile(r"nano banana", re.IGNORECASE),
            re.compile(r"nanobanana", re.IGNORECASE),
        ],
        "DALL-E": [re.compile(r"dall-e", re.IGNORECASE), re.compile(r"dalle", re.IGNORECASE)],
        "NovelAI": [re.compile(r"novelai", re.IGNORECASE), re.compile(r"nai-diffusion", re.IGNORECASE)],
        "InvokeAI": [re.compile(r"invokeai", re.IGNORECASE), re.compile(r"invoke ai", re.IGNORECASE)],
        "Fooocus": [re.compile(r"fooocus", re.IGNORECASE)],
        "Stable Diffusion": [re.compile(r"stable diffusion", re.IGNORECASE), re.compile(r"sd\.?next", re.IGNORECASE), re.compile(r"forge", re.IGNORECASE)],
        "Leonardo.ai": [re.compile(r"leonardo.ai", re.IGNORECASE)],
        "Adobe Firefly": [re.compile(r"adobe firefly", re.IGNORECASE)],
        "Bing Image Creator": [re.compile(r"bing image creator", re.IGNORECASE)],
    }
 
    # 2. 生成参数指纹 (次优先，当没有软件名时)
    GENERATION_PARAM_FINGERPRINTS = [
        (r"Steps:\s*\d+", "Stable Diffusion (Parameters)"),
        (r"CFG scale:\s*[\d\.]+", "Stable Diffusion (Parameters)"),
        (r"Sampler:\s*\w+", "Stable Diffusion (Parameters)"),
        (r"Seed:\s*\d+", "Stable Diffusion (Parameters)"),
        (r"Model hash:\s*[a-f0-9]+", "Stable Diffusion (Parameters)"),
        (r"Model:\s*[^,\n]+", "Stable Diffusion (Parameters)"),
        (r"Negative prompt:", "Stable Diffusion (Parameters)"),
        (r"Size:\s*\d+x\d+", "Stable Diffusion (Parameters)"),
        (r"Clip skip:\s*\d+", "Stable Diffusion (Parameters)"),
        (r"Schedule type:\s*[^,\n]+", "Stable Diffusion (Parameters)"),
        (r"Denoising strength:\s*[\d\.]+", "Stable Diffusion (Parameters)"),
        (r"Hires upscale:\s*[\d\.]+", "Stable Diffusion (Parameters)"),
        (r"DigitalSourceType\s*[:=]\s*(?:http://cv\.iptc\.org/newscodes/digitalsourcetype/)?trainedAlgorithmicMedia", "Generative AI (IPTC/XMP Standard)"),
    ]
 
    RAW_METADATA_SCAN_LIMIT = 128 * 1024 * 1024
 
    C2PA_CONTAINER_MARKERS = [
        b"c2pa", b"jumbf", b"content credentials",
        b"contentcredentials", b"contentauth",
    ]
 
    C2PA_AI_MARKERS = [
        b"trainedalgorithmicmedia",
        b"compositewithtrainedalgorithmicmedia",
        b"algorithmicmedia",
        b"generated by ai", b"ai generated",
        b"adobe firefly", b"google imagen", b"synthid",
        b"dall-e", b"dalle", b"midjourney", b"stable diffusion",
    ]

def detect(self, image_path: str) -> Dict[str, str]:
    img_path = Path(image_path)
    if not img_path.exists():
        return {"status": "error", "reason": "文件不存在", "tool": "", "evidence": ""}
 
    evidence_found = []
    weak_evidence_found = []
    context_evidence_found = []
    detected_tool = None
    extension_mismatch = False
    minimal_web_jpeg = False
 
    try:
        with Image.open(img_path) as img:
            img.load()  # Ensure header is loaded
            actual_format = (img.format or "").upper()
            suffix = img_path.suffix.lower()
            if actual_format == "JPEG" and suffix not in {".jpg", ".jpeg", ".jpe"}:
                extension_mismatch = True
                context_evidence_found.append(f"File extension mismatch: {suffix} file contains JPEG data")
            elif actual_format == "PNG" and suffix != ".png":
                extension_mismatch = True
                context_evidence_found.append(f"File extension mismatch: {suffix} file contains PNG data")
 
            minimal_jpeg_keys = {"jfif", "jfif_density", "jfif_unit", "jfif_version", "progression", "progressive"}
            if actual_format == "JPEG" and set(img.info.keys()).issubset(minimal_jpeg_keys):
                minimal_web_jpeg = True
                context_evidence_found.append("Minimal JPEG metadata only")
 
            # --- 1. Check PNG Info / tEXt chunks ---
            if hasattr(img, 'info') and img.info:
                software_value = img.info.get("Software") or img.info.get("software")
                if isinstance(software_value, bytes):
                    software_text = software_value.decode("utf-8", errors="ignore")
                else:
                    software_text = str(software_value or "")
                software_lower = software_text.lower()
                if "celsys" in software_lower or "clip studio" in software_lower:
                    context_evidence_found.append("Edited/exported by CELSYS/Clip Studio Paint")
 
                if not detected_tool:
                    for key, val in img.info.items():
                        if not isinstance(val, (str, bytes)):
                            continue
                        structured_res = self._detect_structured_generator_metadata(key, val)
                        if structured_res:
                            detected_tool, marker = structured_res
                            evidence_found.append(marker)
                            break
 
                # Case A: A1111 / SD
                if 'parameters' in img.info:
                    val = img.info['parameters']
                    if isinstance(val, str) and ("Steps:" in val or "Prompt" in val):
                        detected_tool = "Stable Diffusion (A1111)"
                        evidence_found.append("Stable Diffusion parameters chunk")
                        res = self._analyze_text(val)
                        if res:
                            detected_tool, marker = res
                            evidence_found.append(f"Parameters: {marker}")
 
                # Case B: ComfyUI (专有检查)
                if not detected_tool and ('workflow' in img.info or 'prompt' in img.info):
                    try:
                        if 'workflow' in img.info:
                            json.loads(img.info['workflow'])
                            evidence_found.append("Valid 'workflow' JSON")
                        if 'prompt' in img.info:
                            json.loads(img.info['prompt'])
                            evidence_found.append("Valid 'prompt' JSON")
                        detected_tool = "ComfyUI"
                    except json.JSONDecodeError:
                        pass
 
                # Case C: Generic Scan (其他 info)
                if not detected_tool:
                    for key, val in img.info.items():
                        if isinstance(val, (str, bytes)):
                            val_str = self._metadata_value_to_text(val)
                            res = self._analyze_text(val_str)
                            if res:
                                detected_tool, marker = res
                                evidence_found.append(f"PNG Info '{key}': {marker}")
                                break
 
                for key, val in img.info.items():
                    if not isinstance(val, (str, bytes)):
                        continue
                    val_str = val.decode("utf-8", errors="ignore") if isinstance(val, bytes) else str(val)
                    if "DigitalSourceType" in val_str and "trainedAlgorithmicMedia" in val_str:
                        marker = f"PNG Info '{key}': IPTC/XMP trainedAlgorithmicMedia"
                        if marker not in evidence_found:
                            evidence_found.append(marker)
 
            # --- 2. Check EXIF / XMP Data ---
            if not detected_tool:
                exif = img.getexif()
                if exif:
                    for tag_id, value in exif.items():
                        tag_name = ExifTags.TAGS.get(tag_id, str(tag_id))
                        if isinstance(value, bytes):
                            try:
                                value_str = value.decode('utf-8', errors='ignore')
                            except:
                                continue
                        else:
                            value_str = str(value)
                        res = self._analyze_text(value_str)
                        if res:
                            detected_tool, marker = res
                            evidence_found.append(f"EXIF {tag_name}: {marker}")
                            break
 
    except Exception as e:
        return {"status": "error", "reason": f"读取错误: {str(e)}", "tool": "", "evidence": ""}
 
    # --- 3. Official C2PA manifest/signature verification ---
    c2pa_available = False
    c2pa_has_manifest = False
    c2pa_claim_generator = ""
    try:
        from tools.validator.c2pa_verifier import verify_c2pa_file
 
        c2pa_result = verify_c2pa_file(img_path)
        c2pa_available = c2pa_result.available
        if c2pa_result.has_manifest:
            c2pa_has_manifest = True
            c2pa_claim_generator = c2pa_result.claim_generator or ""
            evidence_found.append(c2pa_result.evidence_summary())
            if c2pa_result.ai_generated:
                detected_tool = "Generative AI (C2PA Content Credentials)"
        elif c2pa_result.available and c2pa_result.error:
            evidence_found.append(f"C2PA verification error: {c2pa_result.error}")
        elif c2pa_result.available:
            evidence_found.append("No C2PA manifest found")
    except Exception as e:
        evidence_found.append(f"C2PA verification error: {e}")
 
    # --- 4. Raw metadata fallback: C2PA/JUMBF/XMP payloads ---
    if not detected_tool and not c2pa_has_manifest:
        res, has_c2pa_container = self._scan_raw_metadata(img_path)
        if res:
            detected_tool, marker = res
            prefix = "Raw metadata"
            if not c2pa_available and has_c2pa_container:
                prefix = "Raw metadata (official C2PA verifier unavailable)"
            evidence_found.append(f"{prefix}: {marker}")
        elif has_c2pa_container:
            if c2pa_available:
                evidence_found.append("C2PA Content Credentials found, no AI generation marker")
            else:
                evidence_found.append("C2PA Content Credentials found, official verifier unavailable")
 
    # --- 5. Visible Google/Gemini watermark ---
    google_context = "google" in c2pa_claim_generator.lower()
    filename_lower = img_path.name.lower()
    filename_gemini_hint = "gemini" in filename_lower or "google" in filename_lower
    if not detected_tool and (google_context or filename_gemini_hint):
        if self._detect_google_visible_watermark(img_path):
            detected_tool = "Gemini (Google Visible Watermark)"
            evidence_found.append("Visible watermark: Google/Gemini sparkle mark")
 
    # --- 6. Final Fallback: Filename Check ---
    if not detected_tool:
        res = self._analyze_text(filename_lower, include_weak_markers=False)
        if res:
            weak_tool, marker = res
            weak_evidence_found.append(f"Filename suggests {weak_tool}: {marker}")
 
    if extension_mismatch and minimal_web_jpeg:
        context_evidence_found.append("Downloaded file appears re-encoded or metadata-stripped")
 
    # --- Result Construction ---
    if detected_tool:
        return {
            "status": "ai",
            "reason": f"检测到 {detected_tool} 元数据",
            "tool": detected_tool,
            "evidence": "; ".join(evidence_found + context_evidence_found)
        }
    else:
        if weak_evidence_found:
            return {
                "status": "unknown",
                "reason": "AI indicators found, but C2PA credentials do not declare AI generation",
                "tool": "",
                "evidence": "; ".join(evidence_found + context_evidence_found + weak_evidence_found)
            }
        reason = "C2PA Content Credentials do not declare AI generation" if c2pa_has_manifest else "No known AI generation metadata detected"
        return {
            "status": "human",
            "reason": reason,
            "tool": "",
            "evidence": "; ".join(evidence_found + context_evidence_found) if evidence_found or context_evidence_found else "No metadata signatures found"
        }

def _decode_metadata_bytes(self, data: bytes) -> str:
    """Best-effort decoding for embedded XMP/C2PA text inside binary assets."""
    if not data:
        return ""
    text = data.decode("utf-8", errors="ignore")
    if len(text.strip()) < 8:
        text = data.decode("latin-1", errors="ignore")
    return text
 
def _metadata_value_to_text(self, value: Any) -> str:
    if isinstance(value, bytes):
        return value.decode("utf-8", errors="ignore")
    return str(value)
 
def _json_loads(self, value: str) -> Any:
    try:
        return json.loads(value)
    except Exception:
        return None
 
def _json_has_keys(self, value: Any, keys: set[str]) -> bool:
    if isinstance(value, dict):
        lowered = {str(key).lower() for key in value.keys()}
        if lowered.intersection(keys):
            return True
        return any(self._json_has_keys(item, keys) for item in value.values())
    if isinstance(value, list):
        return any(self._json_has_keys(item, keys) for item in value)
    return False
 
def _detect_structured_generator_metadata(self, key: str, value: Any) -> Optional[Tuple[str, str]]:
    text = self._metadata_value_to_text(value)
    lowered_key = key.lower()
    lowered_text = text.lower()
    parsed = self._json_loads(text)
 
    if "invoke" in lowered_key or "invokeai" in lowered_text or "invoke ai" in lowered_text:
        return "InvokeAI", f"PNG Info '{key}': InvokeAI metadata"
 
    if "fooocus" in lowered_key or "fooocus" in lowered_text:
        return "Fooocus", f"PNG Info '{key}': Fooocus metadata"
 
    if "novelai" in lowered_text or "nai-diffusion" in lowered_text:
        return "NovelAI", f"PNG Info '{key}': NovelAI metadata"
 
    if isinstance(parsed, dict):
        has_generation_keys = self._json_has_keys(
            parsed,
            {"sampler", "sampler_name", "steps", "scale", "cfg_scale", "seed", "model", "model_hash", "uc"},
        )
        has_novelai_shape = self._json_has_keys(parsed, {"uc"}) and self._json_has_keys(parsed, {"sampler", "steps", "scale"})
        if has_novelai_shape:
            return "NovelAI", f"PNG Info '{key}': NovelAI generation JSON"
 
        if "invoke" in lowered_key and has_generation_keys:
            return "InvokeAI", f"PNG Info '{key}': InvokeAI generation JSON"
 
        if lowered_key in {"sd-metadata", "sd_metadata", "generation_data", "generation_data_formatted"} and has_generation_keys:
            return "Stable Diffusion", f"PNG Info '{key}': Stable Diffusion generation JSON"
 
    return None

def _analyze_c2pa_bytes(self, data: bytes) -> Optional[Tuple[str, str]]:
    """Detect AI signals embedded in C2PA/JUMBF Content Credentials payloads."""
    if not data:
        return None
 
    lowered = data.lower()
    has_c2pa_container = any(marker in lowered for marker in self.C2PA_CONTAINER_MARKERS)
    if not has_c2pa_container:
        return None
 
    for marker in self.C2PA_AI_MARKERS:
        if marker in lowered:
            return "Generative AI (C2PA Content Credentials)", marker.decode("ascii", errors="ignore")
 
    text = self._decode_metadata_bytes(data)
    res = self._analyze_text(text, include_weak_markers=False)
    if res:
        tool, marker = res
        return tool, f"C2PA payload: {marker}"
 
    return None
 
def _scan_raw_metadata(self, image_path: Path) -> Tuple[Optional[Tuple[str, str]], bool]:
    """
    Scan raw image bytes for metadata payloads Pillow does not expose.
 
    C2PA manifests are commonly stored as JUMBF boxes or JPEG APP11 segments.
    Pillow can open those files without surfacing the manifest through
    Image.info or EXIF, so a raw marker scan is needed as a fallback.
    """
    try:
        size = image_path.stat().st_size
        if size > self.RAW_METADATA_SCAN_LIMIT:
            return None, False
        data = image_path.read_bytes()
    except OSError:
        return None, False
 
    lowered = data.lower()
    has_c2pa_container = any(marker in lowered for marker in self.C2PA_CONTAINER_MARKERS)
    c2pa_res = self._analyze_c2pa_bytes(data)
    if c2pa_res:
        return c2pa_res, has_c2pa_container
 
    text = self._decode_metadata_bytes(data)
    return self._analyze_text(text, include_weak_markers=False), has_c2pa_container

def _analyze_text(self, text: str, include_weak_markers: bool = True) -> Optional[Tuple[str, str]]:
    """
    Analyze a string for AI markers.
    Returns: (tool_name, found_marker) or None
 
    检测优先级：
    1. 先检查明确的软件签名（使用 regex 更严格匹配）
    2. 再检查生成参数指纹（如果 IPTC 标记，检查是否有 Google 证据，否则通用）
    """
    if not text:
        return None
 
    text_lower = text.lower()
 
    # 1. Check Explicit Software Names (按字典顺序，Midjourney 先)
    weak_patterns = {r"mj_"}
    for tool, patterns in self.AI_SOFTWARE_SIGNATURES.items():
        for pattern in patterns:
            if not include_weak_markers and pattern.pattern in weak_patterns:
                continue
            if pattern.search(text_lower):
                return tool, pattern.pattern
 
    # 2. Check Generation Parameter Fingerprints
    match_count = 0
    evidence = []
    detected_tool = None
    for pattern, tool_name in self.GENERATION_PARAM_FINGERPRINTS:
        match = re.search(pattern, text, re.IGNORECASE)
        if match:
            if "IPTC/XMP Standard" in tool_name:
                # 通用 IPTC，优先检查是否有 Google 证据
                if any(re.search(p, text_lower) for p in self.AI_SOFTWARE_SIGNATURES["Gemini (Google)"]):
                    return "Gemini (Google)", "IPTC/XMP with Google Evidence"
                else:
                    detected_tool = "Generative AI (Unknown)"
                    evidence.append("IPTC/XMP Signature")
            else:
                match_count += 1
                evidence.append(pattern)
 
    if match_count >= 1:
        return "Stable Diffusion WebUI", "Generation Parameters Detected"
 
    if detected_tool:
        return detected_tool, "; ".join(evidence)
 
    return None