好的，你写吧。
好的，下面是完整的“标准件萃取器”代码，它将每篇产出按段落拆分为可独立使用的“标准件”。
```python
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
标准件萃取器 V1.0
从 output_放大器 中提取段落级标准件
每个段落独立存储，标注情绪、场景、关键词等元数据
不调用API，纯规则驱动
"""
import os
import re
import json
import hashlib
from typing import List, Dict, Any, Optional
from datetime import datetime
from collections import defaultdict
# ==================== 路径配置 ====================
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
os.chdir(BASE_DIR)
INPUT_DIR = "output_放大器"
OUTPUT_DIR = "excerpt_memory"
PARAGRAPH_DIR = os.path.join(OUTPUT_DIR, "paragraphs")
INDEX_PATH = os.path.join(OUTPUT_DIR, "index", "paragraph_index.json")
os.makedirs(PARAGRAPH_DIR, exist_ok=True)
os.makedirs(os.path.dirname(INDEX_PATH), exist_ok=True)
# ==================== 词库 ====================
# 七情（同前）
QING_WORDS = {
"xi": ["喜", "悦", "欣", "怡", "庆", "贺", "得", "成", "畅", "欢", "乐", "安"],
"nu": ["怒", "愤", "恨", "仇", "怨", "暴", "烈", "恼", "憎", "恶", "忿"],
"you": ["忧", "虑", "愁", "烦", "苦", "闷", "郁", "戚", "哀", "伤", "恸"],
"si": ["思", "念", "想", "恋", "忆", "盼", "望", "怀", "牵", "挂", "慕"],
"bei": ["悲", "哀", "痛", "哭", "丧", "绝", "殇", "凄", "凉", "惨", "涕"],
"kong": ["恐", "惧", "怕", "骇", "颤", "栗", "惶", "畏", "怯", "悚", "悸"],
"jing": ["惊", "骇", "震", "突", "变", "危", "乱", "愕", "诧", "异", "骤"]
}
QING_LABELS = {
"xi": "喜", "nu": "怒", "you": "忧", "si": "思",
"bei": "悲", "kong": "恐", "jing": "惊"
}
# 人文地貌（同前）
HUMAN_GEO = {
"建筑": ["庐", "舍", "屋", "宅", "院", "庭", "堂", "楼", "阁", "亭", "台", "榭",
"庙", "寺", "观", "庵", "祠", "殿", "宫", "苑", "城", "垣", "墙", "门",
"桥", "梁", "栈", "驿", "铺", "肆", "坊", "巷", "街", "市", "墟"],
"器物": ["犁", "耙", "锄", "镰", "刀", "剑", "矛", "盾", "弓", "箭", "鼎", "钟",
"釜", "甑", "碗", "盏", "杯", "壶", "缶", "甕", "筐", "篓", "笥", "箱",
"梳", "镜", "簪", "钗", "环", "佩", "带", "韘", "笄", "璎", "珞"],
"服饰": ["衣", "裳", "袍", "襦", "裙", "裾", "袂", "袖", "带", "绦", "绶", "冠",
"冕", "巾", "帻", "笠", "屐", "履", "靴", "袜", "佩", "缨", "绂", "韐"],
"饮食": ["酒", "茶", "羹", "汤", "粥", "饭", "米", "粟", "麦", "黍", "稻", "粱",
"肉", "鱼", "禽", "蔬", "果", "药", "酿", "醅", "醴", "醪", "饴", "饵"],
"民俗": ["祭", "祀", "祷", "祈", "卜", "占", "巫", "傩", "戏", "乐", "舞", "歌",
"丧", "葬", "殡", "奠", "婚", "嫁", "聘", "宴", "会", "社", "赛"],
"地理": ["山", "岭", "峰", "峦", "谷", "壑", "涧", "溪", "河", "江", "湖", "海",
"泉", "瀑", "潭", "泽", "渚", "洲", "岛", "原", "野", "田", "埂", "垄",
"城", "关", "塞", "驿", "津", "渡", "浦", "汊"],
"气象": ["风", "雨", "雪", "霜", "冰", "雹", "露", "雾", "云", "霞", "虹", "雷",
"电", "月", "日", "星", "辰", "光", "影", "霭", "霰", "霪"]
}
# 写作手法（同前）
WRITING_TECHNIQUES = {
"比喻": ["如", "似", "若", "犹", "仿佛", "宛如", "好似", "像", "同"],
"拟人": ["哭", "笑", "叹", "诉", "说", "思", "念", "忆", "忧", "惧"],
"倒叙": ["忆", "昔", "往", "前", "曾经", "当时", "那年", "记得"],
"插叙": ["忽", "然", "间", "却", "又", "这时", "此时"],
"悬念": ["疑", "惑", "莫", "非", "难道", "竟然", "究竟", "到底"],
"伏笔": ["早", "便", "已", "先", "暗中", "悄然", "不知不觉"],
"白描": ["是", "有", "见", "看", "望", "闻", "听", "觉", "感"],
"细描": ["纹", "理", "色", "香", "味", "触", "声", "光", "影"],
"烘托": ["愈加", "更加", "越发", "倍", "更"],
"衬托": ["反衬", "映", "衬", "托"],
"虚实相生": ["犹", "似", "若", "仿佛", "恍", "惚"],
"直抒": ["吾", "我", "余", "予", "窃", "愚"],
"借景抒情": ["望", "见", "观", "登", "临", "凭", "倚"],
"寓情于景": ["含", "带", "藏", "隐", "寓", "寄"],
"用典": ["古人", "先贤", "圣", "经", "典", "史", "诗"]
}
# 小说元素（同前）
NOVEL_ELEMENTS = {
"人物身份": ["翁", "妪", "叟", "媪", "童", "稚", "女", "子", "郎", "君",
"士", "农", "工", "商", "僧", "道", "医", "卜", "巫", "匠"],
"人物外貌": ["面", "容", "貌", "额", "眉", "目", "眼", "鼻", "口", "唇",
"齿", "须", "发", "鬓", "肤", "肌", "骨", "形", "体", "身"],
"人物动作": ["行", "走", "步", "趋", "奔", "驰", "跃", "跳", "坐", "立",
"卧", "倚", "靠", "握", "持", "执", "举", "挥", "刺", "砍",
"斩", "劈", "击", "打", "敲", "磨", "刻", "雕", "画", "写"],
"环境空间": ["间", "中", "内", "外", "上", "下", "前", "后", "左", "右",
"东", "西", "南", "北", "处", "所", "地", "方", "旁", "侧"],
"时间线索": ["朝", "暮", "晨", "昏", "日", "夜", "春", "夏", "秋", "冬",
"今", "昔", "往", "来", "岁", "年", "月", "时", "刻"],
"自然意象": ["花", "草", "木", "叶", "竹", "松", "柏", "柳", "梅", "菊",
"山", "水", "月", "风", "云", "雪", "霜", "雨", "露", "星"],
"人文意象": ["书", "画", "琴", "棋", "酒", "茶", "剑", "弓", "旗", "鼓"]
}
# ===== 新增：场景类型词库 =====
SCENE_TYPES = {
"田园日常": ["田", "埂", "稻", "麦", "农", "耕", "犁", "锄", "牛", "羊", "庄", "舍", "炊烟"],
"市井街巷": ["市", "街", "巷", "铺", "肆", "摊", "商", "卖", "买", "叫卖", "行人"],
"山水旅途": ["山", "岭", "谷", "溪", "河", "江", "湖", "海", "泉", "瀑", "行", "旅", "舟", "渡"],
"室内家居": ["屋", "庭", "堂", "楼", "阁", "窗", "门", "案", "床", "席", "灯", "炉"],
"祭祀庙宇": ["庙", "寺", "观", "庵", "祠", "祭", "祀", "祷", "祈", "巫", "卜"],
"夜色星辰": ["夜", "月", "星", "辰", "晚", "暮", "昏", "宵", "漏", "更"],
"风雪寒凉": ["雪", "霜", "冰", "寒", "冷", "冻", "凛", "冽"],
"雨雾朦胧": ["雨", "雾", "霭", "湿", "濛", "润", "霖"],
"宴饮欢聚": ["宴", "酒", "饮", "杯", "盏", "欢", "乐", "舞", "歌"],
"离别思念": ["别", "离", "送", "远", "思", "念", "望", "怀"],
"劳作耕耘": ["耕", "种", "锄", "犁", "刈", "获", "舂", "织"],
"战斗冲突": ["战", "斗", "杀", "伐", "攻", "击", "破", "斩", "刃"],
"静谧沉思": ["静", "寂", "闲", "思", "虑", "默", "独", "清"],
"古意苍茫": ["古", "旧", "残", "墟", "荒", "碑", "苔", "沧桑"],
}
# 辅助：场景类型的优先级（如果同时匹配多个，可排序）
SCENE_PRIORITY = ["古意苍茫", "战斗冲突", "离别思念", "夜色星辰", "雨雾朦胧", "风雪寒凉", "祭祀庙宇", "山水旅途", "田园日常", "市井街巷", "室内家居", "宴饮欢聚", "劳作耕耘", "静谧沉思"]
# ==================== 萃取器 ====================
class ParagraphExtractor:
def __init__(self):
self.index = self._load_index()
self.log_entries = []
def _load_index(self) -> Dict[str, Any]:
if os.path.exists(INDEX_PATH):
try:
with open(INDEX_PATH, 'r', encoding='utf-8') as f:
return json.load(f)
except:
pass
return {"processed_files": [], "total_paragraphs": 0}
def _save_index(self):
with open(INDEX_PATH, 'w', encoding='utf-8') as f:
json.dump(self.index, f, ensure_ascii=False, indent=2)
def _log(self, msg: str):
ts = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
entry = f"[{ts}] {msg}"
print(entry)
self.log_entries.append(entry)
def _is_processed(self, filename: str) -> bool:
for p in self.index.get("processed_files", []):
if p.get("file") == filename:
return True
return False
def _mark_processed(self, filename: str, count: int):
self.index["processed_files"].append({
"file": filename,
"path": os.path.join(INPUT_DIR, filename),
"extracted_at": datetime.now().isoformat(),
"paragraph_count": count
})
self.index["total_paragraphs"] += count
self._save_index()
def _read_file(self, filepath: str) -> Optional[str]:
try:
with open(filepath, 'r', encoding='utf-8', errors='ignore') as f:
return f.read()
except Exception as e:
self._log(f"⚠️ 读取失败: {filepath} - {e}")
return None
def _split_paragraphs(self, text: str) -> List[str]:
# 按空行分割
raw_paras = re.split(r'\n\s*\n', text)
result = []
for p in raw_paras:
p = p.strip()
if len(p) > 30: # 至少30字才保留
result.append(p)
return result
# ----- 分析函数（同前，略作调整） -----
def _extract_qing(self, text: str) -> Dict[str, float]:
scores = {}
total = 0
for qing, words in QING_WORDS.items():
score = 0
for w in words:
score += text.count(w)
scores[qing] = score
total += score
if total == 0:
return {}
for qing in scores:
scores[qing] = round(scores[qing] / total * 100, 1)
return scores
def _get_dominant_qing(self, scores: Dict[str, float]) -> str:
if not scores:
return "未知"
max_qing = max(scores, key=scores.get)
return QING_LABELS.get(max_qing, "未知")
def _extract_human_geo(self, text: str) -> Dict[str, List[str]]:
result = defaultdict(list)
for category, words in HUMAN_GEO.items():
for w in words:
if w in text:
result[category].append(w)
return {k: list(set(v)) for k, v in result.items()}
def _extract_techniques(self, text: str) -> List[str]:
techniques = []
for name, indicators in WRITING_TECHNIQUES.items():
for ind in indicators:
if ind in text:
techniques.append(name)
break
return list(set(techniques))
def _extract_elements(self, text: str) -> Dict[str, List[str]]:
result = defaultdict(list)
for category, words in NOVEL_ELEMENTS.items():
for w in words:
if w in text:
result[category].append(w)
return {k: list(set(v)) for k, v in result.items()}
def _extract_keywords(self, text: str, limit: int = 6) -> List[str]:
words = re.findall(r'[\u4e00-\u9fff]{2,4}', text)
freq = {}
for w in words:
freq[w] = freq.get(w, 0) + 1
sorted_words = sorted(freq.items(), key=lambda x: x[1], reverse=True)
return [w for w, _ in sorted_words[:limit] if len(w) > 1]
def _detect_scene_type(self, text: str) -> List[str]:
detected = []
for scene, keywords in SCENE_TYPES.items():
for kw in keywords:
if kw in text:
detected.append(scene)
break
# 去重并按优先级排序
if detected:
detected = list(set(detected))
# 按优先级排序（优先级高的排前面）
detected.sort(key=lambda x: SCENE_PRIORITY.index(x) if x in SCENE_PRIORITY else 999)
return detected
def process_file(self, filename: str) -> Dict[str, Any]:
filepath = os.path.join(INPUT_DIR, filename)
if not os.path.exists(filepath):
return {"status": "error", "message": f"文件不存在: {filepath}"}
if self._is_processed(filename):
return {"status": "skipped", "message": f"已萃取: {filename}"}
content = self._read_file(filepath)
if not content:
return {"status": "error", "message": f"读取失败: {filename}"}
paragraphs = self._split_paragraphs(content)
if not paragraphs:
return {"status": "warning", "message": "无有效段落"}
para_objects = []
for idx, para in enumerate(paragraphs):
if len(para) < 20:
continue
# 生成ID
para_id = hashlib.md5(f"{filename}_{idx}_{para[:30]}".encode()).hexdigest()[:12]
qing_scores = self._extract_qing(para)
dominant_qing = self._get_dominant_qing(qing_scores)
human_geo = self._extract_human_geo(para)
techniques = self._extract_techniques(para)
elements = self._extract_elements(para)
keywords = self._extract_keywords(para)
scene_types = self._detect_scene_type(para)
obj = {
"id": para_id,
"paragraph": para,
"word_count": len(para),
"dominant_qing": dominant_qing,
"qing_distribution": qing_scores,
"scene_type": scene_types,
"human_geo": human_geo,
"writing_techniques": techniques,
"novel_elements": elements,
"keywords": keywords,
"source_file": filename,
"source_path": filepath,
"extracted_at": datetime.now().isoformat()
}
para_objects.append(obj)
if para_objects:
# 保存为一个JSON文件
output_filename = filename.replace('.txt', '') + '.json'
output_path = os.path.join(PARAGRAPH_DIR, output_filename)
with open(output_path, 'w', encoding='utf-8') as f:
json.dump(para_objects, f, ensure_ascii=False, indent=2)
self._mark_processed(filename, len(para_objects))
self._log(f" ✅ 萃取 {len(para_objects)} 个段落标准件")
return {
"status": "success",
"file": filename,
"paragraph_count": len(para_objects),
"first_qing": para_objects[0].get("dominant_qing") if para_objects else "未知"
}
return {"status": "warning", "message": "处理后无有效段落"}
def scan_and_process(self):
if not os.path.exists(INPUT_DIR):
self._log(f"⚠️ 输入目录不存在: {INPUT_DIR}")
return
files = [f for f in os.listdir(INPUT_DIR) if f.endswith('.txt')]
files.sort()
if not files:
self._log(f"⚠️ 输入目录为空: {INPUT_DIR}")
return
total = len(files)
processed = 0
skipped = 0
errors = 0
self._log(f"📂 发现 {total} 个文件")
self._log(f"📊 开始萃取段落标准件...")
for i, filename in enumerate(files, 1):
self._log(f" 处理 [{i}/{total}]: {filename}")
result = self.process_file(filename)
if result.get("status") == "success":
processed += 1
self._log(f" ✅ {result['paragraph_count']} 段 | 基调: {result.get('first_qing', '未知')}")
elif result.get("status") == "skipped":
skipped += 1
else:
errors += 1
self._log(f" ❌ {result.get('message', '未知错误')}")
self._log(f"\n{'='*50}")
self._log(f"📊 萃取完成")
self._log(f" 总文件: {total}")
self._log(f" 已处理: {processed}")
self._log(f" 已跳过: {skipped}")
self._log(f" 错误: {errors}")
self._log(f" 总段落数: {self.index.get('total_paragraphs', 0)}")
self._log(f" 标准件目录: {PARAGRAPH_DIR}")
self._log(f" 索引: {INDEX_PATH}")
self._log(f"{'='*50}")
# ==================== 主入口 ====================
def main():
print("\n" + "="*70)
print("🧩 标准件萃取器 V1.0")
print(" 从 output_放大器 提取段落级标准件")
print(" 每个段落独立存储，带标签")
print(" 不调用API，纯规则驱动")
print("="*70 + "\n")
extractor = ParagraphExtractor()
extractor.scan_and_process()
print("\n✅ 完成！")
print(f" 标准件目录: {PARAGRAPH_DIR}")
print(f" 索引: {INDEX_PATH}")
if __name__ == "__main__":
main()
```
---
## 📋 使用说明
### 运行
```bash
python excerpt_paragraph_v1.py
```
### 输出
- 每个源文件生成一个 JSON，放在 `excerpt_memory/paragraphs/` 下
- 每个 JSON 包含该文件的所有段落对象（数组）
- 每个段落对象包含原文 + 所有标签
### 段落标准件示例
```json
{
"id": "para_001",
"paragraph": "沇水缓缓地绕着村庄流淌，水面上泛着粼粼的波光。孩子们在溪边嬉戏玩耍，光着脚丫踩在清凉的水里，溅起一串串水花。老人拄着拐杖，宿醉未醒，脚步有些踉跄，春潮一次次地涨满了旧日的河床，水声哗哗，像是时光在低语。",
"word_count": 95,
"dominant_qing": "喜",
"qing_distribution": {"xi": 40, "you": 20, "si": 20, "bei": 10, "jing": 10},
"scene_type": ["田园日常", "静谧沉思"],
"human_geo": {"地理": ["溪", "河"], "气象": ["光"]},
"writing_techniques": ["白描", "拟人"],
"novel_elements": {"人物身份": ["翁"], "人物动作": ["行", "走"], "自然意象": ["水", "光"]},
"keywords": ["沇水", "村庄", "溪边", "春潮"],
"source_file": "round_388.txt",
"source_path": "output_放大器/round_388.txt",
"extracted_at": "2026-07-08T10:00:00"
}
```
---
## ✅ 特点
| 特性 | 说明 |
|:---|:---|
| **段落级** | 保留原文自然段落，不拆句子 |
| **标准件** | 每个段落独立，可直接使用 |
| **无API** | 纯规则，无成本 |
| **可检索** | 带有七情、场景、关键词等标签 |
| **增量处理** | 已处理的文件不会重复萃取 |
你可以直接运行，然后用这些标准件来组合新的文章。🎯 |
鄂公网安备42060002000282号 )