zhuyuan-ice-gl-zy001/CODEX-DEV-20260823/mother-persona-neuron-extraction-20260823/tools/index_chatgpt_export.py

216 lines
9.5 KiB
Python

#!/usr/bin/env python3
"""Stream a ChatGPT export and build provenance-first indexes without loading it all."""
from __future__ import annotations
from collections import Counter
from datetime import datetime, timezone
import hashlib
import json
from pathlib import Path
SOURCE = Path("/Volumes/JZAO/铸渊-ICE-GL-ZY001/CODEX-DEV-20260823/mother-persona-neuron-extraction-20260823/extracted-readonly/CORPUS-BS-GPT-0001/GPTconversations.json")
OUT_ROOT = Path("/Volumes/JZAO/铸渊-ICE-GL-ZY001/CODEX-DEV-20260823/mother-persona-neuron-extraction-20260823/corpus-manifests")
SOURCE_SHA256 = "ecfb4c8870751ecf402c08e9950e3452519678ef24bc7565394d6cf48e1d44f7"
CORPUS_ID = "CORPUS-BS-GPT-0001"
def stable_json(value) -> str:
return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
def digest_text(value: str) -> str:
return hashlib.sha256(value.encode("utf-8")).hexdigest()
def iso_time(value):
if not isinstance(value, (int, float)):
return None
return datetime.fromtimestamp(value, tz=timezone.utc).isoformat().replace("+00:00", "Z")
def iter_array(path: Path, chunk_size: int = 1024 * 1024):
decoder = json.JSONDecoder()
with path.open("r", encoding="utf-8") as stream:
buffer = ""
position = 0
started = False
ended = False
while True:
chunk = stream.read(chunk_size)
eof = chunk == ""
buffer += chunk
while True:
while position < len(buffer) and buffer[position].isspace():
position += 1
if not started:
if position >= len(buffer):
break
if buffer[position] != "[":
raise ValueError("top-level ChatGPT export is not an array")
position += 1
started = True
continue
while position < len(buffer) and (buffer[position].isspace() or buffer[position] == ","):
position += 1
if position < len(buffer) and buffer[position] == "]":
position += 1
ended = True
break
if position >= len(buffer):
break
try:
value, next_position = decoder.raw_decode(buffer, position)
except json.JSONDecodeError:
if eof:
raise
break
yield value
position = next_position
if ended:
if buffer[position:].strip():
raise ValueError("non-whitespace data follows top-level array")
return
if eof:
raise ValueError("unterminated top-level array")
if position:
buffer = buffer[position:]
position = 0
def main() -> None:
if SOURCE.stat().st_size != 697293638:
raise ValueError("source size changed")
source_hash = hashlib.sha256()
with SOURCE.open("rb") as stream:
for chunk in iter(lambda: stream.read(1024 * 1024), b""):
source_hash.update(chunk)
if source_hash.hexdigest() != SOURCE_SHA256:
raise ValueError(f"source hash changed: {source_hash.hexdigest()}")
segment_path = OUT_ROOT / f"{CORPUS_ID}.direct-user-role-segments.jsonl"
conversation_path = OUT_ROOT / f"{CORPUS_ID}.conversation-index.jsonl"
stats_path = OUT_ROOT / f"{CORPUS_ID}.stream-audit.json"
for path in (segment_path, conversation_path, stats_path):
if path.exists():
raise FileExistsError(path)
conversations = 0
messages = 0
user_segments = 0
roles = Counter()
content_types = Counter()
user_min_time = None
user_max_time = None
total_user_projection_chars = 0
non_string_user_parts = 0
with segment_path.open("x", encoding="utf-8") as segment_out, conversation_path.open("x", encoding="utf-8") as conversation_out:
for ordinal, conversation in enumerate(iter_array(SOURCE), start=1):
conversations += 1
conversation_id = conversation.get("conversation_id") or conversation.get("id")
title = conversation.get("title")
mapping = conversation.get("mapping") or {}
local_counts = Counter()
local_user = 0
for node_id, node in mapping.items():
message = (node or {}).get("message")
if not isinstance(message, dict):
continue
messages += 1
role = ((message.get("author") or {}).get("role") or "unknown")
roles[role] += 1
local_counts[role] += 1
content = message.get("content") or {}
content_type = content.get("content_type") or "unknown"
content_types[content_type] += 1
if role != "user":
continue
parts = content.get("parts")
if not isinstance(parts, list):
parts = []
string_parts = [part for part in parts if isinstance(part, str)]
non_string_count = len(parts) - len(string_parts)
non_string_user_parts += non_string_count
text_projection = "\n".join(string_parts)
part_canon = stable_json(parts)
message_id = message.get("id") or node_id
segment_seed = f"{SOURCE_SHA256}:{conversation_id}:{message_id}"
segment_id = f"BSGPT-SEG-{digest_text(segment_seed)[:24]}"
created = message.get("create_time")
if isinstance(created, (int, float)):
user_min_time = created if user_min_time is None else min(user_min_time, created)
user_max_time = created if user_max_time is None else max(user_max_time, created)
record = {
"schema": "guanghu.corpus-source-segment/v1",
"segment_id": segment_id,
"corpus_id": CORPUS_ID,
"source_file": str(SOURCE),
"source_file_sha256": SOURCE_SHA256,
"conversation_ordinal": ordinal,
"conversation_id": conversation_id,
"conversation_title_metadata": title,
"node_id": node_id,
"message_id": message_id,
"speaker_export_role": "user",
"speaker_identity_state": "BINGSHUO_CANDIDATE_REQUIRES_CONTENT_BOUNDARY_AUDIT",
"source_class_state": "S01_CANDIDATE_NOT_PROMOTED",
"content_type": content_type,
"occurred_at_utc": iso_time(created),
"time_precision": "EXPORT_MESSAGE_TIMESTAMP" if created is not None else "UNKNOWN",
"content_parts": parts,
"content_parts_canonical_sha256": digest_text(part_canon),
"text_projection_sha256": digest_text(text_projection),
"text_projection_chars": len(text_projection),
"non_string_part_count": non_string_count,
"exact_locator": f"conversation[{ordinal}].mapping[{json.dumps(node_id, ensure_ascii=False)}].message.content.parts",
"promotion_state": "OBSERVED_SEGMENT_ONLY",
}
segment_out.write(stable_json(record) + "\n")
user_segments += 1
local_user += 1
total_user_projection_chars += len(text_projection)
conversation_record = {
"schema": "guanghu.chatgpt-conversation-index/v1",
"corpus_id": CORPUS_ID,
"ordinal": ordinal,
"conversation_id": conversation_id,
"title_metadata": title,
"create_time_utc": iso_time(conversation.get("create_time")),
"update_time_utc": iso_time(conversation.get("update_time")),
"mapping_node_count": len(mapping),
"message_role_counts": dict(sorted(local_counts.items())),
"user_segment_count": local_user,
}
conversation_out.write(stable_json(conversation_record) + "\n")
stats = {
"schema": "guanghu.chatgpt-corpus-stream-audit/v1",
"corpus_id": CORPUS_ID,
"source_file": str(SOURCE),
"source_file_sha256": SOURCE_SHA256,
"json_syntax": "PASS_100",
"top_level": "ARRAY",
"conversation_count": conversations,
"message_count": messages,
"message_role_counts": dict(sorted(roles.items())),
"content_type_counts": dict(sorted(content_types.items())),
"direct_user_role_segment_count": user_segments,
"direct_user_text_projection_chars": total_user_projection_chars,
"non_string_user_part_count": non_string_user_parts,
"user_time_min_utc": iso_time(user_min_time),
"user_time_max_utc": iso_time(user_max_time),
"segment_index": str(segment_path),
"conversation_index": str(conversation_path),
"source_authority_boundary": "USER_EXPORT_ROLE_IS_ONLY_A_BINGSHUO_CANDIDATE_UNTIL_CONTENT_AND_SPEAKER_BOUNDARY_AUDIT",
"neuron_promotion_state": "NONE",
}
with stats_path.open("x", encoding="utf-8") as output:
json.dump(stats, output, ensure_ascii=False, indent=2)
output.write("\n")
print(json.dumps(stats, ensure_ascii=False, indent=2))
if __name__ == "__main__":
main()