Implement HEIC parsing with vendor-specific handlers
- Added a new parser for HEIC files that supports recursive parsing of boxes. - Introduced vendor-specific handlers for Apple and Samsung HEIC features. - Created an abstract base handler for common functionality across vendors. - Implemented methods to find motion photo offsets and handle embedded video data. - Updated the structure to accommodate additional box types and improve parsing efficiency.
This commit is contained in:
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
+15
-37
@@ -5,17 +5,15 @@ import struct
|
||||
from io import BytesIO
|
||||
|
||||
class Box:
|
||||
"""
|
||||
Represents a generic ISOBMFF box.
|
||||
Now supports finding children and building (writing) data back.
|
||||
"""
|
||||
"""Generic ISOBMFF box supporting hierarchical parsing and rebuilds."""
|
||||
|
||||
def __init__(self, size: int, box_type: str, offset: int, raw_data: bytes):
|
||||
self.size = size
|
||||
self.type = box_type
|
||||
self.offset = offset
|
||||
self.raw_data = raw_data
|
||||
self.children: List['Box'] = []
|
||||
self.is_full_box = False # 默认不是 FullBox
|
||||
self.is_full_box = False # Flag toggled by FullBox subclasses
|
||||
|
||||
def __repr__(self) -> str:
|
||||
return f"<Box '{self.type}' size={self.size} offset={self.offset}>"
|
||||
@@ -48,17 +46,12 @@ class Box:
|
||||
return header.getvalue()
|
||||
|
||||
def build_content(self) -> bytes:
|
||||
"""
|
||||
序列化此盒的 *内容* (不包括头部)。
|
||||
对于一个通用的容器盒 (container boxes),它会递归构建所有子盒。
|
||||
"""
|
||||
"""Serialise the box payload (children included, header excluded)."""
|
||||
if not self.children:
|
||||
# 对于一个没有子盒的简单数据盒 (如 ftyp, mdat, 或未解析的盒)
|
||||
# 只需返回它持有的原始数据。
|
||||
# Leaf boxes expose their stored payload verbatim.
|
||||
return self.raw_data
|
||||
|
||||
# 对于一个容器盒 (如 'meta', 'iprp', 'ipco')
|
||||
# 递归构建每一个子盒 (完整的,包括头部) 并拼接它们
|
||||
# Container boxes rebuild each child (including headers) in order.
|
||||
content_stream = BytesIO()
|
||||
for child in self.children:
|
||||
child_data = child.build_box()
|
||||
@@ -67,22 +60,17 @@ class Box:
|
||||
|
||||
|
||||
def build_box(self) -> bytes:
|
||||
"""
|
||||
构建完整的盒 (头部 + 内容),并返回其二进制数据。
|
||||
"""
|
||||
# 1. "自底向上" 构建内容
|
||||
"""Construct the full box (header + payload) and return its bytes."""
|
||||
content_data = self.build_content()
|
||||
|
||||
# 2. 构建头部
|
||||
header_data = self.build_header(len(content_data))
|
||||
|
||||
# 3. 更新 self.size 属性 (8字节标准头 + 内容)
|
||||
self.size = len(header_data) + len(content_data)
|
||||
|
||||
return header_data + content_data
|
||||
|
||||
def find_box(self, box_type: str, recursive: bool = True) -> Optional['Box']:
|
||||
"""在子盒中查找指定类型的第一个盒子"""
|
||||
"""Return the first child matching `box_type`."""
|
||||
for child in self.children:
|
||||
if child.type == box_type:
|
||||
return child
|
||||
@@ -93,9 +81,8 @@ class Box:
|
||||
return None
|
||||
|
||||
class FullBox(Box):
|
||||
"""
|
||||
FullBox 是一种特殊的 Box,它在内容开头包含 4 字节的 version 和 flags
|
||||
"""
|
||||
"""Box variant that begins with a 4-byte version/flags header."""
|
||||
|
||||
def __init__(self, size: int, box_type: str, offset: int, raw_data: bytes):
|
||||
super().__init__(size, box_type, offset, raw_data)
|
||||
self.is_full_box = True
|
||||
@@ -104,38 +91,29 @@ class FullBox(Box):
|
||||
self._parse_full_box_header()
|
||||
|
||||
def _parse_full_box_header(self):
|
||||
"""从 raw_data 中解析 version 和 flags"""
|
||||
"""Extract version and flags from the stored payload."""
|
||||
if len(self.raw_data) >= 4:
|
||||
version_flags = struct.unpack('>I', self.raw_data[:4])[0]
|
||||
self.version = (version_flags >> 24) & 0xFF
|
||||
self.flags = version_flags & 0xFFFFFF
|
||||
|
||||
def build_full_box_header(self) -> bytes:
|
||||
"""构建 4 字节的 version/flags 头部"""
|
||||
"""Serialise the 4-byte version/flags prefix."""
|
||||
version_flags = (self.version << 24) | self.flags
|
||||
return struct.pack('>I', version_flags)
|
||||
|
||||
def build_content(self) -> bytes:
|
||||
"""
|
||||
序列化此 FullBox 的内容。
|
||||
它首先写入 4 字节的 version/flags,然后
|
||||
再写入子盒 (如果是容器) 或 version/flags 之后的
|
||||
原始数据 (如果不是容器)。
|
||||
"""
|
||||
"""Serialise the FullBox payload, ensuring the header is emitted."""
|
||||
content_stream = BytesIO()
|
||||
|
||||
# 1. 写入 FullBox 特有的 4 字节头部
|
||||
content_stream.write(self.build_full_box_header())
|
||||
|
||||
# 2. 写入剩余的内容
|
||||
if not self.children:
|
||||
# 对于一个没有子盒的简单 FullBox (如 'hdlr')
|
||||
# 写入 self.raw_data 中 *跳过* 4 字节 v/f 之后的部分
|
||||
# Emit the payload following the version/flags header.
|
||||
if len(self.raw_data) >= 4:
|
||||
content_stream.write(self.raw_data[4:])
|
||||
else:
|
||||
# 对于一个容器 FullBox (如 'meta')
|
||||
# 递归构建每一个子盒 (与 Box.build_content 相同)
|
||||
# Container-style FullBoxes rebuild each descendant.
|
||||
for child in self.children:
|
||||
child_data = child.build_box()
|
||||
content_stream.write(child_data)
|
||||
|
||||
+20
-31
@@ -1,10 +1,10 @@
|
||||
# pyheic_struct/builder.py
|
||||
|
||||
import struct
|
||||
from .heic_file import HEICFile
|
||||
from .heic_types import ItemLocationBox
|
||||
|
||||
class HEICBuilder:
|
||||
"""Rebuild a HEIC container after metadata edits."""
|
||||
|
||||
def __init__(self, heic_file: HEICFile):
|
||||
self.heic_file = heic_file
|
||||
self.mdat_box = heic_file.get_mdat_box()
|
||||
@@ -18,14 +18,14 @@ class HEICBuilder:
|
||||
if not self.iloc_box:
|
||||
raise ValueError("Cannot build: 'iloc' box not found.")
|
||||
|
||||
# 存储 'mdat' 的 *原始* 偏移量。这是至关重要的。
|
||||
# Preserve the original `mdat` offset for delta calculations.
|
||||
self.original_mdat_offset = self.mdat_box.offset
|
||||
|
||||
# 分离元数据和数据
|
||||
|
||||
# All non-`mdat` boxes form the metadata segment.
|
||||
self.top_level_meta_boxes = [b for b in self.heic_file.boxes if b.type != 'mdat']
|
||||
|
||||
def _calculate_meta_offset_delta(self) -> int:
|
||||
"""(V14) 辅助函数:计算 'meta' 盒的偏移增量"""
|
||||
"""Return the delta between original and rebuilt meta offsets."""
|
||||
ftyp_box = self.heic_file._ftyp_box
|
||||
if not ftyp_box:
|
||||
ftyp_box = next((b for b in self.top_level_meta_boxes if b.type == 'ftyp'), None)
|
||||
@@ -35,7 +35,7 @@ class HEICBuilder:
|
||||
return new_meta_offset - original_meta_offset
|
||||
|
||||
def _rebuild_iloc_with_delta(self, mdat_offset_delta: int):
|
||||
"""(V14) 辅助函数:使用给定的 delta 重建 iloc"""
|
||||
"""Rebuild the `iloc` box using the supplied `mdat` delta."""
|
||||
print(f" ... Rebuilding 'iloc' using mdat_delta: {mdat_offset_delta}")
|
||||
meta_offset_delta = self._calculate_meta_offset_delta()
|
||||
try:
|
||||
@@ -49,24 +49,20 @@ class HEICBuilder:
|
||||
)
|
||||
except Exception as e:
|
||||
print(f"CRITICAL: Failed to rebuild 'iloc' box content: {e}")
|
||||
raise # 抛出异常以停止构建
|
||||
raise
|
||||
|
||||
def _calculate_final_meta_size(self) -> int:
|
||||
"""(V14) 辅助函数:递归构建所有元数据盒以获取其最终大小"""
|
||||
"""Compute the size of the metadata stream after rebuild."""
|
||||
current_offset = 0
|
||||
for box in self.top_level_meta_boxes:
|
||||
# build_box() 会递归计算所有子盒的最终大小
|
||||
final_box_data = box.build_box()
|
||||
current_offset += len(final_box_data)
|
||||
return current_offset
|
||||
|
||||
def write(self, output_path: str):
|
||||
"""
|
||||
(V14) 使用一个多遍系统来解决 'iloc' 重建引起的循环依赖问题。
|
||||
"""
|
||||
|
||||
# --- Pass 1: 初步布局计算 ---
|
||||
# (在 'iloc' 重建 *之前* 计算一次大小)
|
||||
"""Persist the rebuilt HEIC structure to `output_path`."""
|
||||
|
||||
# Pass 1: compute a preliminary layout before touching `iloc`.
|
||||
print("--- Builder Pass 1: Preliminary Layout ---")
|
||||
preliminary_meta_size = self._calculate_final_meta_size()
|
||||
preliminary_mdat_offset = preliminary_meta_size
|
||||
@@ -76,14 +72,12 @@ class HEICBuilder:
|
||||
print(f"Preliminary mdat offset: {preliminary_mdat_offset}")
|
||||
print(f"Preliminary offset delta: {preliminary_mdat_delta}")
|
||||
|
||||
# --- Pass 2: 初步重建 'iloc' ---
|
||||
# (使用 *初步的* delta 重建 iloc,这会改变 'iloc' 的大小)
|
||||
# Pass 2: rebuild `iloc` with the preliminary delta.
|
||||
print("\n--- Builder Pass 2: Preliminary 'iloc' Rebuild ---")
|
||||
self._rebuild_iloc_with_delta(preliminary_mdat_delta)
|
||||
print(" 'iloc' preliminary rebuild complete.")
|
||||
|
||||
# --- Pass 3: 最终布局计算 ---
|
||||
# (现在 'iloc' 有了最终大小,我们 *再次* 计算 meta 的总大小)
|
||||
# Pass 3: recalculate metadata size after the first rebuild.
|
||||
print("\n--- Builder Pass 3: Final Layout Calculation ---")
|
||||
final_meta_size = self._calculate_final_meta_size()
|
||||
final_mdat_offset = final_meta_size
|
||||
@@ -93,8 +87,7 @@ class HEICBuilder:
|
||||
print(f"Final mdat offset (actual): {final_mdat_offset}")
|
||||
print(f"Final offset delta (actual): {final_mdat_delta}")
|
||||
|
||||
# --- Pass 4: 最终重建 'iloc' ---
|
||||
# (使用 *最终的、正确*的 delta 再次重建 iloc)
|
||||
# Pass 4: optionally rebuild with the corrected delta.
|
||||
if final_mdat_delta != preliminary_mdat_delta:
|
||||
print("\n--- Builder Pass 4: Final 'iloc' Rebuild (Correcting Delta) ---")
|
||||
self._rebuild_iloc_with_delta(final_mdat_delta)
|
||||
@@ -102,28 +95,24 @@ class HEICBuilder:
|
||||
else:
|
||||
print("\n--- Builder Pass 4: Skipped (Preliminary delta was correct) ---")
|
||||
|
||||
# --- Pass 5: 写入文件 ---
|
||||
# Pass 5: write metadata followed by `mdat`.
|
||||
print("\n--- Builder Pass 5: Rebuild & Write ---")
|
||||
|
||||
# 只有在 'iloc' 重建成功后,我们才打开(并创建)输出文件
|
||||
with open(output_path, 'wb') as f:
|
||||
|
||||
|
||||
current_offset = 0
|
||||
for box in self.top_level_meta_boxes:
|
||||
final_box_data = box.build_box()
|
||||
f.write(final_box_data)
|
||||
current_offset += len(final_box_data)
|
||||
print(f"Wrote '{box.type}' (final size: {len(final_box_data)})")
|
||||
|
||||
# 完整性检查
|
||||
|
||||
if current_offset != final_mdat_offset:
|
||||
print(f"WARNING: Final meta size ({current_offset}) does not match"
|
||||
f" calculated mdat offset ({final_mdat_offset})!")
|
||||
|
||||
# 写入 'mdat' 盒 (头部 + 数据)
|
||||
mdat_data = self.mdat_box.raw_data
|
||||
mdat_size = 8 + len(mdat_data) # 8-byte header
|
||||
|
||||
mdat_size = 8 + len(mdat_data) # 8-byte header
|
||||
|
||||
print(f"Writing 'mdat' (final size: {mdat_size}) at offset {current_offset}...")
|
||||
|
||||
if mdat_size > 4294967295:
|
||||
|
||||
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
+26
-62
@@ -1,8 +1,5 @@
|
||||
# pyheic_struct/heic_file.py
|
||||
|
||||
import os
|
||||
import math
|
||||
# 1. 导入新的库
|
||||
import pillow_heif
|
||||
from io import BytesIO
|
||||
from dataclasses import dataclass
|
||||
@@ -27,6 +24,8 @@ class Grid:
|
||||
output_height: int
|
||||
|
||||
class HEICFile:
|
||||
"""High-level accessor for parsed HEIC/HEIF structures."""
|
||||
|
||||
def __init__(self, filepath: str):
|
||||
self.filepath = filepath
|
||||
pillow_heif.register_heif_opener()
|
||||
@@ -38,7 +37,7 @@ class HEICFile:
|
||||
self._iprp_box: ItemPropertiesBox | None = None
|
||||
self._iref_box: ItemReferenceBox | None = None
|
||||
self.handler: VendorHandler | None = None
|
||||
self.boxes: list[Box] = [] # 顶层盒子
|
||||
self.boxes: list[Box] = [] # Top-level boxes
|
||||
|
||||
try:
|
||||
with open(self.filepath, 'rb') as f:
|
||||
@@ -51,7 +50,7 @@ class HEICFile:
|
||||
raise
|
||||
|
||||
def _find_essential_boxes(self, boxes: list[Box]):
|
||||
"""在解析树中查找关键盒子的快捷方式"""
|
||||
"""Populate shortcut pointers for commonly used boxes."""
|
||||
for box in boxes:
|
||||
if isinstance(box, ItemLocationBox): self._iloc_box = box
|
||||
if isinstance(box, ItemInfoBox): self._iinf_box = box
|
||||
@@ -61,10 +60,8 @@ class HEICFile:
|
||||
if box.type == 'ftyp': self._ftyp_box = box
|
||||
if box.children: self._find_essential_boxes(box.children)
|
||||
|
||||
# --- 辅助方法 (用于修改和构建) ---
|
||||
|
||||
def find_box(self, box_type: str, root_box_list: list[Box] | None = None) -> Box | None:
|
||||
"""递归查找第一个匹配类型的盒子"""
|
||||
"""Recursively locate the first box with the requested fourcc."""
|
||||
if root_box_list is None:
|
||||
root_box_list = self.boxes
|
||||
|
||||
@@ -78,14 +75,14 @@ class HEICFile:
|
||||
return None
|
||||
|
||||
def get_mdat_box(self) -> Box | None:
|
||||
"""获取顶层的 'mdat' 盒"""
|
||||
"""Return the top-level `mdat` box, if present."""
|
||||
for box in self.boxes:
|
||||
if box.type == 'mdat':
|
||||
return box
|
||||
return None
|
||||
|
||||
def _remove_box_recursive(self, box_type: str, box_list: list[Box]) -> bool:
|
||||
"""Helper for remove_box_by_type"""
|
||||
"""Internal helper used by `remove_box_by_type`."""
|
||||
for i, box in enumerate(box_list):
|
||||
if box.type == box_type:
|
||||
box_list.pop(i)
|
||||
@@ -96,19 +93,14 @@ class HEICFile:
|
||||
return False
|
||||
|
||||
def remove_box_by_type(self, box_type: str) -> bool:
|
||||
"""递归查找并移除第一个匹配类型的盒子"""
|
||||
"""Remove the first box matching `box_type`, searching recursively."""
|
||||
return self._remove_box_recursive(box_type, self.boxes)
|
||||
|
||||
# --- START V16 FIX ---
|
||||
def remove_item_by_id(self, item_id_to_remove: int):
|
||||
"""
|
||||
(V16 - 垃圾回收版)
|
||||
彻底从 iinf, iloc, ipma, iref 中删除一个 Item ID,
|
||||
并从 ipco 中删除孤立的属性,然后重写所有剩余的 ipma 索引。
|
||||
"""
|
||||
"""Fully remove an item from iinf/iloc/ipma/iref and clean orphaned properties."""
|
||||
print(f"Attempting to remove Item ID {item_id_to_remove} from all references (V16)...")
|
||||
|
||||
# 1. 从 iinf (Item Info) 中删除
|
||||
|
||||
# Update iinf metadata
|
||||
if self._iinf_box:
|
||||
self._iinf_box.children = [
|
||||
c for c in self._iinf_box.children
|
||||
@@ -120,7 +112,7 @@ class HEICFile:
|
||||
]
|
||||
print(f" - Removed from 'iinf' box.")
|
||||
|
||||
# 2. 从 iloc (Item Location) 中删除
|
||||
# Update iloc entries
|
||||
if self._iloc_box:
|
||||
self._iloc_box.locations = [
|
||||
loc for loc in self._iloc_box.locations
|
||||
@@ -128,7 +120,7 @@ class HEICFile:
|
||||
]
|
||||
print(f" - Removed from 'iloc' box.")
|
||||
|
||||
# 3. (新) 从 iref (Item Reference) 中删除
|
||||
# Update iref relationships
|
||||
if self._iref_box:
|
||||
for i in range(len(self._iref_box.children) - 1, -1, -1):
|
||||
ref_box = self._iref_box.children[i]
|
||||
@@ -153,7 +145,7 @@ class HEICFile:
|
||||
]
|
||||
print(f" - Cleaned 'iref.references' of to_id {item_id_to_remove}.")
|
||||
|
||||
# 4. 从 ipma 和 ipco (属性) 中删除
|
||||
# Update property associations and definitions
|
||||
if self._iprp_box and self._iprp_box.ipma and self._iprp_box.ipco:
|
||||
ipma = self._iprp_box.ipma
|
||||
ipco = self._iprp_box.ipco
|
||||
@@ -162,18 +154,17 @@ class HEICFile:
|
||||
print(f" - Item {item_id_to_remove} not in 'ipma'. No properties to clean.")
|
||||
return
|
||||
|
||||
# 4a. 找到要删除的属性索引 (1-based)
|
||||
# Identify property indices referenced exclusively by the removed item.
|
||||
props_to_remove = {
|
||||
assoc.property_index
|
||||
for assoc in ipma.entries[item_id_to_remove].associations
|
||||
}
|
||||
|
||||
# 4b. 从 ipma 中删除该 Item
|
||||
# Remove the item entry itself.
|
||||
del ipma.entries[item_id_to_remove]
|
||||
print(f" - Removed Item {item_id_to_remove} from 'ipma'.")
|
||||
|
||||
# 4c. 确定哪些属性是“孤儿”
|
||||
# (即,它们在 props_to_remove 中,但*不在*任何*剩余*的 item 关联中)
|
||||
# Determine which properties are now unused.
|
||||
all_remaining_props = set()
|
||||
for entry in ipma.entries.values():
|
||||
all_remaining_props.update(
|
||||
@@ -188,16 +179,12 @@ class HEICFile:
|
||||
|
||||
print(f" - Found orphaned properties to remove from 'ipco': {orphaned_props}")
|
||||
|
||||
# 4d. 创建一个“重映射表”
|
||||
# 我们从后往前遍历,以安全地删除
|
||||
# (索引是 1-based, 但列表是 0-based)
|
||||
# Remove orphaned properties from `ipco`, iterating from the end to keep indices valid.
|
||||
orphaned_indices_0based = sorted([p - 1 for p in orphaned_props], reverse=True)
|
||||
|
||||
# 原始 1-based 索引 -> 新 1-based 索引
|
||||
remap_table = {}
|
||||
remap_table = {}
|
||||
original_prop_count = len(ipco.children)
|
||||
|
||||
# 4e. 从 'ipco.children' 列表中删除孤儿
|
||||
for index_0 in orphaned_indices_0based:
|
||||
if 0 <= index_0 < len(ipco.children):
|
||||
removed_prop = ipco.children.pop(index_0)
|
||||
@@ -205,8 +192,6 @@ class HEICFile:
|
||||
else:
|
||||
print(f" - Warning: Orphaned index {index_0+1} out of bounds for 'ipco'.")
|
||||
|
||||
# 4f. 构建重映射表
|
||||
# (只有在 ipco 实际发生变化时才需要)
|
||||
new_prop_count = len(ipco.children)
|
||||
if new_prop_count != original_prop_count:
|
||||
print(" - Re-indexing 'ipma' associations...")
|
||||
@@ -220,7 +205,6 @@ class HEICFile:
|
||||
current_new_index += 1
|
||||
current_old_index += 1
|
||||
|
||||
# 4g. 应用重映射
|
||||
for item_id, entry in ipma.entries.items():
|
||||
new_associations = []
|
||||
for assoc in entry.associations:
|
||||
@@ -229,19 +213,15 @@ class HEICFile:
|
||||
assoc.property_index = remap_table[old_prop_index]
|
||||
new_associations.append(assoc)
|
||||
# else:
|
||||
# 该属性已被删除 (不应发生,因为我们只删除了孤儿)
|
||||
# The property was removed entirely (should not occur).
|
||||
|
||||
# print(f" - Item {item_id} associations: {entry.associations} -> {new_associations}")
|
||||
entry.associations = new_associations
|
||||
print(" - 'ipma' re-indexing complete.")
|
||||
else:
|
||||
print(" - No 'ipma' re-indexing needed.")
|
||||
# --- END V16 FIX ---
|
||||
|
||||
def set_content_identifier(self, new_content_id: str) -> bool:
|
||||
"""
|
||||
(FIXED) 在 'iinf' 盒中查找主图像的 'infe' 盒,并设置其 item_name
|
||||
"""
|
||||
"""Locate the primary `infe` entry and update its item name with a UUID."""
|
||||
primary_id = self.get_primary_item_id()
|
||||
if not primary_id:
|
||||
print("Error: Cannot find primary item ID.")
|
||||
@@ -253,9 +233,6 @@ class HEICFile:
|
||||
|
||||
print(f"Searching for 'infe' box with primary_id = {primary_id}")
|
||||
|
||||
# --- START FIX ---
|
||||
# 侦测我们日志中看到的 "shifted ID" 格式
|
||||
|
||||
target_id = primary_id
|
||||
found_ids = [box.item_id for box in self._iinf_box.children if isinstance(box, ItemInfoEntryBox)]
|
||||
|
||||
@@ -269,16 +246,14 @@ class HEICFile:
|
||||
print(f"Error: Could not find primary ID {primary_id} OR shifted ID {shifted_id}.")
|
||||
print(f"Available 'infe' item IDs found were: {found_ids}")
|
||||
return False
|
||||
# --- END FIX ---
|
||||
|
||||
# 'iinf' 盒的子盒是 'infe' 盒
|
||||
|
||||
# Update both the parsed box and the cached entry representation.
|
||||
for box in self._iinf_box.children:
|
||||
if isinstance(box, ItemInfoEntryBox):
|
||||
if box.item_id == target_id:
|
||||
print(f"Success: Found 'infe' box for target ID {target_id}. Setting item_name...")
|
||||
box.item_name = new_content_id
|
||||
|
||||
# 同时更新 self._iinf_box.entries 中的数据以保持一致
|
||||
for entry in self._iinf_box.entries:
|
||||
if entry.item_id == target_id:
|
||||
entry.name = new_content_id
|
||||
@@ -288,12 +263,8 @@ class HEICFile:
|
||||
print(f"Error: Logic failed to find 'infe' box for target ID {target_id} even after check.")
|
||||
return False
|
||||
|
||||
# --- 现有的只读方法 (无需修改) ---
|
||||
|
||||
def reconstruct_primary_image(self) -> Image.Image | None:
|
||||
"""
|
||||
使用 pillow-heif 重建主图像,它能自动处理网格图像。
|
||||
"""
|
||||
"""Reconstruct the primary image using pillow-heif (handles grid tiles)."""
|
||||
try:
|
||||
print("Reconstructing primary image using pillow-heif...")
|
||||
image = Image.open(self.filepath)
|
||||
@@ -326,11 +297,10 @@ class HEICFile:
|
||||
if not primary_id: return None
|
||||
if not self._iref_box: return None
|
||||
|
||||
# 尝试直接匹配
|
||||
if 'dimg' in self._iref_box.references and primary_id in self._iref_box.references['dimg']:
|
||||
return self._iref_box.references['dimg'].get(primary_id)
|
||||
|
||||
# 尝试匹配 "shifted" ID
|
||||
# Fall back to Samsung-style shifted IDs.
|
||||
shifted_id = primary_id << 16
|
||||
if 'dimg' in self._iref_box.references and shifted_id in self._iref_box.references['dimg']:
|
||||
print("Info: Using shifted primary ID to find grid layout.")
|
||||
@@ -343,17 +313,14 @@ class HEICFile:
|
||||
|
||||
target_id = item_id
|
||||
if item_id not in self._iprp_box.ipma.entries:
|
||||
# 尝试 "shifted" ID
|
||||
shifted_id = item_id << 16
|
||||
if shifted_id in self._iprp_box.ipma.entries:
|
||||
target_id = shifted_id
|
||||
else:
|
||||
# 尝试 "unshifted" ID (以防 item_id 本身就是 shifted 的)
|
||||
unshifted_id = item_id & 0x0000FFFF
|
||||
if unshifted_id in self._iprp_box.ipma.entries:
|
||||
target_id = unshifted_id
|
||||
else:
|
||||
# print(f"Warning: Cannot find size associations for item ID {item_id} or variants.")
|
||||
return None
|
||||
|
||||
item_associations = self._iprp_box.ipma.entries[target_id].associations
|
||||
@@ -393,15 +360,13 @@ class HEICFile:
|
||||
location = next((loc for loc in self._iloc_box.locations if loc.item_id == target_id), None)
|
||||
|
||||
if not location:
|
||||
# 尝试检查 "shifted" ID
|
||||
shifted_id = item_id << 16
|
||||
location = next((loc for loc in self._iloc_box.locations if loc.item_id == shifted_id), None)
|
||||
if location:
|
||||
print(f"Info: Located item {item_id} using shifted ID {shifted_id} in 'iloc'.")
|
||||
target_id = shifted_id
|
||||
|
||||
|
||||
if not location and (item_id & 0xFFFF0000):
|
||||
# 尝试反向检查 (如果传入的 ID 已经是 shifted 的)
|
||||
unshifted_id = item_id & 0x0000FFFF
|
||||
location = next((loc for loc in self._iloc_box.locations if loc.item_id == unshifted_id), None)
|
||||
if location:
|
||||
@@ -450,7 +415,6 @@ class HEICFile:
|
||||
|
||||
target_id = primary_id
|
||||
if primary_id not in self._iref_box.references['thmb']:
|
||||
# Lпробуйте проверить "shifted" ID
|
||||
shifted_primary_id = primary_id << 16
|
||||
if shifted_primary_id not in self._iref_box.references['thmb']:
|
||||
print(f"Info: Primary item ID {primary_id} (or shifted) has no 'thmb' reference.")
|
||||
|
||||
+34
-53
@@ -1,11 +1,9 @@
|
||||
# pyheic_struct/heic_types.py
|
||||
|
||||
import struct
|
||||
from .base import Box, FullBox
|
||||
from io import BytesIO
|
||||
from typing import List, Optional # <-- 添加 List, Optional 导入
|
||||
from typing import List, Optional
|
||||
|
||||
# --- 帮助函数 ---
|
||||
# Helper functions ---------------------------------------------------------
|
||||
|
||||
def _read_int(data: bytes, pos: int, size: int) -> int:
|
||||
"""Helper to read an integer of variable size."""
|
||||
@@ -26,24 +24,26 @@ def _write_int(value: int, size: int) -> bytes:
|
||||
if size == 8: return struct.pack('>Q', value)
|
||||
return b''
|
||||
|
||||
# --- ItemLocation ---
|
||||
# ItemLocation -------------------------------------------------------------
|
||||
class ItemLocation:
|
||||
"""Represents a single `iloc` entry with all associated extents."""
|
||||
|
||||
def __init__(self, item_id):
|
||||
self.item_id = item_id
|
||||
self.data_reference_index = 0
|
||||
self.base_offset = 0
|
||||
self.construction_method = 0
|
||||
self.extent_indices = []
|
||||
self.raw_extents = [] # (relative_offset, length)
|
||||
# extents 存储 (absolute_offset, length)
|
||||
self.extents = []
|
||||
# Stored as (relative_offset, length)
|
||||
self.raw_extents = []
|
||||
# Stored as (absolute_offset, length)
|
||||
self.extents = []
|
||||
|
||||
def __repr__(self):
|
||||
total_length = sum(ext[1] for ext in self.extents)
|
||||
return f"<ItemLocation ID={self.item_id} extents={len(self.extents)} total_size={total_length}>"
|
||||
|
||||
# --- ItemLocationBox ---
|
||||
# ('iloc')
|
||||
# ItemLocationBox (`iloc`) -------------------------------------------------
|
||||
class ItemLocationBox(FullBox):
|
||||
def __init__(self, size: int, box_type: str, offset: int, raw_data: bytes):
|
||||
self.locations: List[ItemLocation] = []
|
||||
@@ -122,13 +122,9 @@ class ItemLocationBox(FullBox):
|
||||
loc.extents.append((base_offset + extent_offset, extent_length))
|
||||
self.locations.append(loc)
|
||||
|
||||
# --- START FIX ---
|
||||
def rebuild_iloc_content(self, mdat_offset_delta: int, original_mdat_offset: int, original_mdat_size: int,
|
||||
meta_offset_delta: int, original_meta_offset: int, original_meta_size: int):
|
||||
"""
|
||||
使用新的 mdat 和 meta 偏移增量 (deltas) 更新此盒的 extents,
|
||||
并重建 self.raw_data
|
||||
"""
|
||||
"""Rebuild the `iloc` payload after `mdat` or `meta` offsets change."""
|
||||
print(f"Applying mdat delta ({mdat_offset_delta}) and meta delta ({meta_offset_delta}) to 'iloc' box...")
|
||||
content_stream = BytesIO()
|
||||
|
||||
@@ -191,9 +187,9 @@ class ItemLocationBox(FullBox):
|
||||
new_absolute_offset = _adjust_absolute_offset(original_absolute_offset)
|
||||
|
||||
if original_absolute_offset == 0:
|
||||
new_absolute_offset = 0 # 保持 0 偏移量
|
||||
|
||||
# 最后的安全检查,防止打包负数
|
||||
new_absolute_offset = 0
|
||||
|
||||
# Guard against negative offsets produced by delta adjustments.
|
||||
if new_absolute_offset < 0:
|
||||
print(f" Warning: Calculated a negative offset ({new_absolute_offset}) for item {loc.item_id}. Setting to 0.")
|
||||
new_absolute_offset = 0
|
||||
@@ -222,12 +218,11 @@ class ItemLocationBox(FullBox):
|
||||
|
||||
self.raw_data = content_stream.getvalue()
|
||||
print(" 'iloc' box content successfully rebuilt.")
|
||||
# --- END FIX ---
|
||||
|
||||
def build_content(self) -> bytes:
|
||||
return self.raw_data
|
||||
|
||||
# --- ItemInfoEntry (DataClass) ---
|
||||
# ItemInfoEntry ------------------------------------------------------------
|
||||
class ItemInfoEntry:
|
||||
def __init__(self, item_id, item_type, item_name):
|
||||
self.item_id = item_id
|
||||
@@ -236,10 +231,9 @@ class ItemInfoEntry:
|
||||
def __repr__(self):
|
||||
return f"<ItemInfoEntry ID={self.item_id} type='{self.type}' name='{self.name}'>"
|
||||
|
||||
# --- ItemInfoEntryBox ---
|
||||
# ('infe')
|
||||
# ItemInfoEntryBox (`infe`) -----------------------------------------------
|
||||
class ItemInfoEntryBox(FullBox):
|
||||
"""代表一个 'infe' 盒"""
|
||||
"""Parse and emit `infe` (ItemInfoEntryBox) structures."""
|
||||
def __init__(self, size: int, box_type: str, offset: int, raw_data: bytes):
|
||||
self.item_id: int = 0
|
||||
self.item_protection_index: int = 0
|
||||
@@ -261,7 +255,7 @@ class ItemInfoEntryBox(FullBox):
|
||||
pos += 2
|
||||
self.item_protection_index = struct.unpack('>H', stream[pos:pos+2])[0]
|
||||
pos += 2
|
||||
self.item_type = ""
|
||||
self.item_type = ""
|
||||
name_end = stream.find(b'\x00', pos)
|
||||
if name_end == -1: name_end = len(stream)
|
||||
self.item_name = stream[pos:name_end].decode('utf-8', errors='ignore')
|
||||
@@ -290,26 +284,26 @@ class ItemInfoEntryBox(FullBox):
|
||||
candidate_type = stream[pos+2:pos+6]
|
||||
|
||||
if candidate_protection <= 0x00FF and b'\x00' not in candidate_type:
|
||||
# 标准顺序: 2 字节保护索引 + 4 字节类型
|
||||
# Standard ordering: 2-byte protection index followed by the 4CC.
|
||||
self.item_protection_index = candidate_protection
|
||||
self._has_protection_field = True
|
||||
pos += 2
|
||||
type_bytes = candidate_type
|
||||
pos += 4
|
||||
else:
|
||||
# 三星文件: 直接写入类型 (如 'hvc1'),缺少保护索引
|
||||
# Samsung files often omit the protection index and write only the 4CC.
|
||||
self.item_protection_index = 0
|
||||
self._has_protection_field = False
|
||||
type_bytes = stream[pos:pos+4]
|
||||
pos += 4
|
||||
elif remaining >= 4:
|
||||
# 某些三星文件缺少 2 字节的保护索引字段, 直接紧跟 4 字节类型
|
||||
# Some vendor variants omit the 2-byte protection field entirely.
|
||||
self.item_protection_index = 0
|
||||
self._has_protection_field = False
|
||||
type_bytes = stream[pos:pos+4]
|
||||
pos += 4
|
||||
else:
|
||||
# 无法获得完整的 4 字节类型
|
||||
# Truncated data: fall back to an empty type.
|
||||
type_bytes = b''
|
||||
|
||||
self.item_type = type_bytes.decode('ascii', errors='ignore').strip('\x00')
|
||||
@@ -376,7 +370,7 @@ class ItemInfoEntryBox(FullBox):
|
||||
content.write(struct.pack('>I', self.item_id))
|
||||
if self._has_protection_field:
|
||||
content.write(struct.pack('>H', self.item_protection_index))
|
||||
# item_type 必须是 4 字节的 4CC,若不足则以 NUL 填充
|
||||
# Item types must be a 4CC; pad with NUL bytes when necessary.
|
||||
content.write(self.item_type.encode('ascii', errors='ignore')[:4].ljust(4, b'\x00'))
|
||||
content.write(item_name_bytes_to_write)
|
||||
if self.content_type is not None:
|
||||
@@ -396,8 +390,7 @@ class ItemInfoEntryBox(FullBox):
|
||||
|
||||
return content.getvalue()
|
||||
|
||||
# --- ItemInfoBox ---
|
||||
# ('iinf')
|
||||
# ItemInfoBox (`iinf`) -----------------------------------------------------
|
||||
class ItemInfoBox(FullBox):
|
||||
def __init__(self, size: int, box_type: str, offset: int, raw_data: bytes):
|
||||
self.entries: list[ItemInfoEntry] = []
|
||||
@@ -433,16 +426,12 @@ class ItemInfoBox(FullBox):
|
||||
else:
|
||||
header.write(struct.pack('>I', len(infe_children)))
|
||||
|
||||
# --- START FIX 2 ---
|
||||
# 之前是: children_data = super().build_content()
|
||||
# 我们需要调用 Box.build_content (祖父级)
|
||||
# Call the grandparent implementation so nested boxes rebuild correctly.
|
||||
children_data = super(FullBox, self).build_content()
|
||||
# --- END FIX 2 ---
|
||||
|
||||
return header.getvalue() + children_data
|
||||
|
||||
# --- PrimaryItemBox ---
|
||||
# ('pitm')
|
||||
# PrimaryItemBox (`pitm`) --------------------------------------------------
|
||||
class PrimaryItemBox(FullBox):
|
||||
def __init__(self, size: int, box_type: str, offset: int, raw_data: bytes):
|
||||
self.item_id: int = 0
|
||||
@@ -469,8 +458,7 @@ class PrimaryItemBox(FullBox):
|
||||
content.write(struct.pack('>I', self.item_id))
|
||||
return content.getvalue()
|
||||
|
||||
# --- ImageSpatialExtentsBox ---
|
||||
# ('ispe')
|
||||
# ImageSpatialExtentsBox (`ispe`) -----------------------------------------
|
||||
class ImageSpatialExtentsBox(FullBox):
|
||||
def __init__(self, size: int, box_type: str, offset: int, raw_data: bytes):
|
||||
self.image_width: int = 0
|
||||
@@ -496,7 +484,7 @@ class ImageSpatialExtentsBox(FullBox):
|
||||
content.write(struct.pack('>I', self.image_height))
|
||||
return content.getvalue()
|
||||
|
||||
# --- ItemPropertyAssociationEntry (DataClass) ---
|
||||
# ItemPropertyAssociationEntry ---------------------------------------------
|
||||
class ItemPropertyAssociation:
|
||||
def __init__(self, property_index: int, essential: bool):
|
||||
self.property_index = property_index
|
||||
@@ -517,8 +505,7 @@ class ItemPropertyAssociationEntry:
|
||||
return (f"<ItemPropertyAssociationEntry item_id={self.item_id} "
|
||||
f"associations={[repr(a) for a in self.associations]}>")
|
||||
|
||||
# --- ItemPropertyAssociationBox ---
|
||||
# ('ipma')
|
||||
# ItemPropertyAssociationBox (`ipma`) -------------------------------------
|
||||
class ItemPropertyAssociationBox(FullBox):
|
||||
def __init__(self, size: int, box_type: str, offset: int, raw_data: bytes):
|
||||
self.entries: dict[int, ItemPropertyAssociationEntry] = {}
|
||||
@@ -587,14 +574,12 @@ class ItemPropertyAssociationBox(FullBox):
|
||||
|
||||
return content.getvalue()
|
||||
|
||||
# --- ItemPropertyContainerBox ---
|
||||
# ('ipco')
|
||||
# ItemPropertyContainerBox (`ipco`) ---------------------------------------
|
||||
class ItemPropertyContainerBox(Box):
|
||||
def _post_parse_initialization(self):
|
||||
pass
|
||||
|
||||
# --- ItemPropertiesBox ---
|
||||
# ('iprp')
|
||||
# ItemPropertiesBox (`iprp`) ----------------------------------------------
|
||||
class ItemPropertiesBox(Box):
|
||||
def _post_parse_initialization(self):
|
||||
pass
|
||||
@@ -610,7 +595,7 @@ class ItemPropertiesBox(Box):
|
||||
if isinstance(child, ItemPropertyAssociationBox): return child
|
||||
return None
|
||||
|
||||
# --- ItemReferenceEntry (DataClass) ---
|
||||
# ItemReferenceEntry -------------------------------------------------------
|
||||
class ItemReferenceEntry:
|
||||
def __init__(self, from_id, to_ids):
|
||||
self.from_item_id = from_id
|
||||
@@ -618,8 +603,7 @@ class ItemReferenceEntry:
|
||||
def __repr__(self):
|
||||
return f"<ItemReferenceEntry from={self.from_item_id} to={self.to_item_ids}>"
|
||||
|
||||
# --- ItemReferenceBox ---
|
||||
# ('iref')
|
||||
# ItemReferenceBox (`iref`) ------------------------------------------------
|
||||
class ItemReferenceBox(FullBox):
|
||||
def __init__(self, size: int, box_type: str, offset: int, raw_data: bytes):
|
||||
self.references: dict[str, dict[int, list[int]]] = {}
|
||||
@@ -674,10 +658,7 @@ class ItemReferenceBox(FullBox):
|
||||
header = BytesIO()
|
||||
header.write(self.build_full_box_header())
|
||||
|
||||
# --- START FIX 2 ---
|
||||
# 之前是: children_data = super().build_content()
|
||||
# 我们需要调用 Box.build_content (祖父级)
|
||||
# Preserve nested reference payloads by delegating to Box.build_content.
|
||||
children_data = super(FullBox, self).build_content()
|
||||
# --- END FIX 2 ---
|
||||
|
||||
return header.getvalue() + children_data
|
||||
|
||||
+8
-16
@@ -1,5 +1,3 @@
|
||||
# pyheic_struct/parser.py
|
||||
|
||||
import struct
|
||||
from typing import List, BinaryIO
|
||||
from io import BytesIO
|
||||
@@ -8,7 +6,7 @@ from .base import Box, FullBox
|
||||
from .heic_types import (
|
||||
ItemLocationBox, PrimaryItemBox, ItemInfoBox, ItemPropertiesBox,
|
||||
ItemPropertyContainerBox, ItemPropertyAssociationBox, ImageSpatialExtentsBox,
|
||||
ItemReferenceBox, ItemInfoEntryBox # <-- 导入新的 'infe' 盒
|
||||
ItemReferenceBox, ItemInfoEntryBox
|
||||
)
|
||||
|
||||
BOX_TYPE_MAP = {
|
||||
@@ -20,13 +18,12 @@ BOX_TYPE_MAP = {
|
||||
'ipma': ItemPropertyAssociationBox,
|
||||
'ispe': ImageSpatialExtentsBox,
|
||||
'iref': ItemReferenceBox,
|
||||
'infe': ItemInfoEntryBox, # <-- 注册新的 'infe' 盒
|
||||
'infe': ItemInfoEntryBox,
|
||||
}
|
||||
|
||||
# 'iref' is also a container.
|
||||
# 'iinf' IS a container (它包含 'infe' 盒)
|
||||
# Boxes treated as containers (their payload should be parsed recursively).
|
||||
CONTAINER_BOXES = {'meta', 'moov', 'trak', 'iprp', 'ipco', 'dinf', 'fiinf', 'ipro', 'iinf', 'iref'}
|
||||
# 几乎所有语义盒都是 FullBox
|
||||
# Boxes that should default to the `FullBox` implementation.
|
||||
FULL_BOXES = {'meta', 'hdlr', 'pitm', 'iinf', 'iloc', 'ipma', 'ispe', 'iref', 'infe'}
|
||||
|
||||
def parse_boxes(stream: BinaryIO, max_size: int) -> List[Box]:
|
||||
@@ -75,28 +72,23 @@ def parse_boxes(stream: BinaryIO, max_size: int) -> List[Box]:
|
||||
parse_size = len(box.raw_data)
|
||||
|
||||
if box.is_full_box:
|
||||
# 跳过 4 字节的 version/flags
|
||||
# Skip the 4-byte version/flags prefix when recursing.
|
||||
if parse_size >= 4:
|
||||
child_stream.read(4)
|
||||
parse_size -= 4
|
||||
|
||||
# --- START FIX ---
|
||||
# 'iinf' 盒在 version/flags 之后还有一个 item_count 字段
|
||||
# 我们必须在解析它的子盒之前跳过它
|
||||
|
||||
if box.type == 'iinf':
|
||||
if box.version == 0:
|
||||
if parse_size >= 2:
|
||||
child_stream.read(2) # 跳过 16-bit item_count
|
||||
child_stream.read(2)
|
||||
parse_size -= 2
|
||||
else:
|
||||
if parse_size >= 4:
|
||||
child_stream.read(4) # 跳过 32-bit item_count
|
||||
child_stream.read(4)
|
||||
parse_size -= 4
|
||||
# --- END FIX ---
|
||||
|
||||
box.children = parse_boxes(child_stream, parse_size)
|
||||
|
||||
# 调用钩子
|
||||
box._post_parse_initialization()
|
||||
|
||||
boxes.append(box)
|
||||
|
||||
Reference in New Issue
Block a user