你的理解能力超强,为你点赞。你写代码吧。
**收到!我现在就写一个完整的、可以直接跑的批量抽取器 + 记忆库统计代码。**
---
### 🧩 完整代码:`structure_memory_builder.py`
```python
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
结构记忆库构建器(丢蛋的鸡)
功能:扫描文件夹内所有 UTF-8 .txt 文档,只提取结构骨架,
丢弃所有具体文字,统计每种结构模式的出现频次。
"""
import os
import re
import json
import hashlib
from pathlib import Path
from collections import defaultdict
from datetime import datetime
# ============================================================
# 第一部分:结构骨架抽取器(只抽鸡,不要蛋)
# ============================================================
class StructureExtractor:
"""从纯文本中提取结构骨架,丢弃所有具体文字内容"""
def __init__(self):
# 模式匹配规则(只识别结构,不捕获具体文字)
self.patterns = {
"work_number": re.compile(r'^第\d+轮作品$'), # 作品编号
"metadata_label": re.compile(r'^[^::]+[::]'), # 标签(冒号前部分)
"numbered_heading": re.compile(r'^\d+\.\s'), # 数字编号开头
"chinese_numbered": re.compile(r'^[一二三四五六七八九十]+[、.]'), # 中文编号
"list_item": re.compile(r'^[-*•]\s'), # 列表项
"indent": re.compile(r'^(\s{2,}|\t)'), # 缩进(2个以上空格或Tab)
}
def extract(self, text):
"""
输入:文本内容(字符串)
输出:结构骨架序列(只含类型和节奏符号,不含任何具体文字)
"""
lines = text.split('\n')
structure = []
prev_blank = False
line_count = 0
for line in lines:
raw = line
stripped = raw.strip()
line_count += 1
# 检测空行(== 顿)
if not stripped:
if not prev_blank: # 避免连续空行重复记录
structure.append({"type": "break", "symbol": "‖"})
prev_blank = True
continue
prev_blank = False
# 检测行特征(按优先级从高到低)
feature = self._classify_line(raw, stripped)
# 检测内部节奏(标点密度)
rhythm = self._detect_rhythm(stripped)
# 检测行长(气息长度)
length_mark = self._detect_length(stripped)
structure.append({
"type": feature["type"],
"level": feature.get("level", 0),
"rhythm": rhythm,
"length": length_mark,
"symbol": feature["symbol"]
})
# 标记结束
structure.append({"type": "end", "symbol": "‖‖"})
return structure
def _classify_line(self, raw, stripped):
"""识别行的结构类型(只取类型,丢掉具体文字)"""
# 1. 作品编号
if self.patterns["work_number"].match(stripped):
return {"type": "work_number", "symbol": "G4𝄐", "level": 0}
# 2. 元数据标签(冒号前部分,但只取类型,不取标签名)
if self.patterns["metadata_label"].match(stripped):
return {"type": "metadata_label", "symbol": "D4>", "level": 0}
# 3. 数字编号标题(1. 2. 3.)
if self.patterns["numbered_heading"].match(stripped):
return {"type": "numbered_heading", "symbol": "C5", "level": 1}
# 4. 中文编号标题(一、 二、)
if self.patterns["chinese_numbered"].match(stripped):
return {"type": "chinese_heading", "symbol": "C5", "level": 1}
# 5. 列表项(- * •)
if self.patterns["list_item"].match(stripped):
return {"type": "list_item", "symbol": "•", "level": 2}
# 6. 缩进(从属内容)
if self.patterns["indent"].match(raw):
return {"type": "indented_body", "symbol": "B3", "level": 1}
# 7. 普通正文(默认)
return {"type": "body", "symbol": "F4", "level": 0}
def _detect_rhythm(self, text):
"""检测文本的内部节奏(基于标点密度)"""
# 句号类(强断点)
full_stops = text.count('。') + text.count('.') + text.count('.')
# 逗号类(弱断点)
commas = text.count(',') + text.count(',') + text.count('、')
# 分号类(中断点)
semicolons = text.count(';') + text.count(';')
total = full_stops + commas + semicolons
if total == 0:
return "legato" # 无标点,连奏
if full_stops > commas + semicolons:
return "staccato" # 句号多,短促
if commas > full_stops:
return "legato" # 逗号多,连绵
return "normal" # 中等
def _detect_length(self, text):
"""检测行长(气息长度)"""
length = len(text)
if length > 50:
return "𝄐" # 长句,延长
if length > 30:
return "—" # 中句,保持
return "·" # 短句,短促
# ============================================================
# 第二部分:结构签名生成器(把骨架序列压缩成唯一签名)
# ============================================================
class StructureSigner:
"""将结构骨架序列压缩为一个可去重的签名(丢弃所有具体文字)"""
@staticmethod
def to_signature(structure_sequence):
"""
输入:结构骨架序列(list of dict)
输出:一个字符串签名(只含类型和符号,不含任何文字)
"""
# 只提取类型和符号,忽略具体位置
tokens = []
for item in structure_sequence:
# 核心:只取 type 和 symbol
t = item.get("type", "unknown")
s = item.get("symbol", "")
tokens.append(f"{t}:{s}")
# 用 | 连接成一个字符串
raw_signature = "|".join(tokens)
# 用 MD5 生成短签名(便于索引和去重)
hash_obj = hashlib.md5(raw_signature.encode('utf-8'))
short_id = hash_obj.hexdigest()[:12]
return {
"signature_id": short_id,
"signature_string": raw_signature,
"token_count": len(tokens)
}
# ============================================================
# 第三部分:记忆库(只存模式频次,不存原文)
# ============================================================
class StructureMemory:
"""结构记忆库:只记录每种结构模式出现的频次"""
def __init__(self, storage_path="memory.json"):
self.storage_path = storage_path
self.memory = {
"total_documents": 0,
"unique_patterns": {},
"pattern_frequency": defaultdict(int)
}
self._load()
def _load(self):
if os.path.exists(self.storage_path):
try:
with open(self.storage_path, 'r', encoding='utf-8') as f:
data = json.load(f)
self.memory["total_documents"] = data.get("total_documents", 0)
self.memory["unique_patterns"] = data.get("unique_patterns", {})
self.memory["pattern_frequency"] = defaultdict(
int, data.get("pattern_frequency", {})
)
except:
pass
def save(self):
"""保存记忆库到文件"""
data = {
"total_documents": self.memory["total_documents"],
"unique_patterns": self.memory["unique_patterns"],
"pattern_frequency": dict(self.memory["pattern_frequency"]),
"updated_at": datetime.now().isoformat()
}
with open(self.storage_path, 'w', encoding='utf-8') as f:
json.dump(data, f, ensure_ascii=False, indent=2)
def add_document(self, signature_id, signature_string, token_count):
"""向记忆库添加一个文档的结构模式"""
self.memory["total_documents"] += 1
self.memory["pattern_frequency"][signature_id] += 1
# 记录模式详情(只存第一次出现的完整签名)
if signature_id not in self.memory["unique_patterns"]:
self.memory["unique_patterns"][signature_id] = {
"signature": signature_string,
"token_count": token_count,
"first_seen": datetime.now().isoformat()
}
def get_report(self):
"""生成统计报告"""
freq = self.memory["pattern_frequency"]
total = self.memory["total_documents"]
unique = len(self.memory["unique_patterns"])
# 找出最高频模式
top_pattern = None
top_count = 0
for pid, count in freq.items():
if count > top_count:
top_count = count
top_pattern = pid
report = {
"total_documents_scanned": total,
"unique_structure_patterns": unique,
"most_frequent_pattern": {
"id": top_pattern,
"count": top_count,
"coverage": f"{top_count/total*100:.2f}%" if total > 0 else "0%"
},
"top_10_patterns": sorted(freq.items(), key=lambda x: x[1], reverse=True)[:10]
}
return report
# ============================================================
# 第四部分:批量处理器(丢蛋的鸡)
# ============================================================
class BatchProcessor:
"""批量扫描文件夹,只抽取结构,丢掉所有文字"""
def __init__(self, input_dir, memory_path="memory.json"):
self.input_dir = Path(input_dir)
self.extractor = StructureExtractor()
self.signer = StructureSigner()
self.memory = StructureMemory(memory_path)
self.errors = []
def scan_single(self, file_path):
"""处理单个文件:抽取结构 → 生成签名 → 丢入记忆库"""
try:
with open(file_path, 'r', encoding='utf-8') as f:
text = f.read()
# 1. 抽取结构骨架(只抽鸡,不要蛋)
structure = self.extractor.extract(text)
# 2. 生成唯一签名(丢掉所有具体文字)
signature = self.signer.to_signature(structure)
# 3. 投入记忆库(只记结构频次,不存原文)
self.memory.add_document(
signature_id=signature["signature_id"],
signature_string=signature["signature_string"],
token_count=signature["token_count"]
)
return {"file": file_path.name, "status": "success", "sig": signature["signature_id"]}
except Exception as e:
self.errors.append({"file": file_path.name, "error": str(e)})
return {"file": file_path.name, "status": "fail", "error": str(e)}
def run(self):
"""批量扫描所有 .txt 文件"""
files = list(self.input_dir.glob("*.txt"))
if not files:
print("❌ 未找到任何 .txt 文件!")
return
print(f"📁 找到 {len(files)} 个 .txt 文件,开始处理...")
for idx, f in enumerate(files, 1):
result = self.scan_single(f)
if result["status"] == "success":
print(f" ✅ [{idx}/{len(files)}] {result['file']} → {result['sig']}")
else:
print(f" ❌ [{idx}/{len(files)}] {result['file']} → 失败: {result['error']}")
# 保存记忆库
self.memory.save()
print("\n💾 记忆库已保存。")
# 输出统计报告
report = self.memory.get_report()
print("\n" + "="*50)
print("📊 结构记忆库统计报告")
print("="*50)
print(f"总扫描文档数:{report['total_documents_scanned']}")
print(f"唯一结构模式数:{report['unique_structure_patterns']}")
if report['most_frequent_pattern']['id']:
print(f"最高频模式 ID:{report['most_frequent_pattern']['id']}")
print(f"出现次数:{report['most_frequent_pattern']['count']}")
print(f"覆盖率:{report['most_frequent_pattern']['coverage']}")
print("="*50)
# 显示 Top 5 模式
print("\n🏆 Top 5 高频结构模式:")
for pid, count in report['top_10_patterns'][:5]:
pattern_info = self.memory.memory["unique_patterns"].get(pid, {})
sig_preview = pattern_info.get("signature", "")[:80] + "..."
print(f" {pid}: {count} 次 → {sig_preview}")
# 如果有错误,显示错误汇总
if self.errors:
print(f"\n⚠️ 有 {len(self.errors)} 个文件处理失败:")
for err in self.errors[:5]:
print(f" ❌ {err['file']}: {err['error']}")
# 保存报告到文件
report_path = self.input_dir / "structure_report.json"
with open(report_path, 'w', encoding='utf-8') as f:
json.dump(report, f, ensure_ascii=False, indent=2)
print(f"\n📄 详细报告已保存至:{report_path}")
# ============================================================
# 第五部分:入口(直接运行)
# ============================================================
if __name__ == "__main__":
import sys
# 用法:python structure_memory_builder.py ./my_documents
if len(sys.argv) < 2:
print("用法:python structure_memory_builder.py <文档文件夹路径>")
print("示例:python structure_memory_builder.py ./docs")
sys.exit(1)
input_dir = sys.argv[1]
if not os.path.isdir(input_dir):
print(f"❌ 错误:'{input_dir}' 不是一个有效的文件夹")
sys.exit(1)
print("🐔 丢蛋的鸡 —— 结构记忆库构建器")
print(f"📂 扫描目标:{input_dir}")
print("="*50)
processor = BatchProcessor(input_dir)
processor.run()
print("\n✅ 全部完成!只保留了结构骨架,所有文字已被丢弃。")
```
---
### 🚀 如何使用
1. **保存代码**:将上面的代码保存为 `structure_memory_builder.py`
2. **准备文件夹**:把所有 UTF-8 编码的 `.txt` 文件放在一个文件夹里,比如 `./my_docs`
3. **运行**:
```bash
python structure_memory_builder.py ./my_docs
```
4. **输出**:
- **`memory.json`**:持久化的记忆库,记录每种结构模式的出现频次
- **`structure_report.json`**:统计报告(总文档数、唯一模式数、最高频模式、Top 10)
---
### 📦 示例输出
```
🐔 丢蛋的鸡 —— 结构记忆库构建器
📂 扫描目标:./my_docs
==================================================
📁 找到 47 个 .txt 文件,开始处理...
✅ [1/47] round_1011548.txt → 3a7f1d9e2b4c
✅ [2/47] round_1014638.txt → 3a7f1d9e2b4c ← 注意:结构和上一个相同!
✅ [3/47] report_2024.txt → 8f2a1d7e3c9b
...
💾 记忆库已保存。
==================================================
📊 结构记忆库统计报告
==================================================
总扫描文档数:47
唯一结构模式数:23
最高频模式 ID:3a7f1d9e2b4c
出现次数:12
覆盖率:25.53%
==================================================
🏆 Top 5 高频结构模式:
3a7f1d9e2b4c: 12 次 → work_number:G4𝄐|metadata_label 4>|break:‖|numbered_heading:C5|body:F4|end:‖‖...
8f2a1d7e3c9b: 8 次 → work_number:G4𝄐|break:‖|numbered_heading:C5|list_item:•|list_item:•|end:‖‖...
...
📄 详细报告已保存至:./my_docs/structure_report.json
✅ 全部完成!只保留了结构骨架,所有文字已被丢弃。
```
---
### 🔑 核心逻辑总结
1. **只抽鸡**:每个文档只提取 `类型 + 节奏符号 + 层级`,具体文字全部丢弃
2. **唯一签名**:相同结构的文档会生成相同的 MD5 签名
3. **记忆库统计**:只记录每种模式出现的**频次**,不存任何原文
4. **批量处理**:支持一次扫描上千个 `.txt` 文件
---
你现在就可以拿自己的文档文件夹试一下。如果跑通了,你就能看到你的文档库里**到底有多少种“结构形状”,以及哪种形状最常出现**。🎯 |