大模型笔记8 文本数据清洗

{"topic": 'xx', "content": ["", ""]}

{"text": "A_百度百科\n"}

{"text": "A…今后若有类似问题或者升级补丁会提前在官网告知用户。\n"}

{"text": "… 参考资料 1. iPhone陷闹钟失声门 苹果称不能保证未来没问题 2. 苹果吸金有术 用户体验点石成金\n"}

{"text": "B_百度百科\n"}

{"text": "B…今后若有类似问题或者升级补丁会提前在官网告知用户。\n"}

{"text": "B… 参考资料 1. iPhone陷闹钟失声门 苹果称不能保证未来没问题 2. 苹果吸金有术 用户体验点石成金\n"}

import json

with open('data/data.json', 'r', encoding='utf-8') as f:

    lines = [json.loads(line.strip())['text'].strip() for line in f if line.strip()]

results = []

current_topic = None

current_content = []

for line in lines:

    if line.endswith('_百度百科'):

        # 如果已经有前一个topic的内容，保存它

        if current_topic:

            results.append({

                "topic": current_topic.replace('_百度百科', ''),

                "content": current_content

            })

        # 新的 topic

        current_topic = line.replace('_百度百科', '')

        current_content = []

    else:

        current_content.append(line)

# 处理最后一个 topic

if current_topic:

    results.append({

        "topic": current_topic.replace('_百度百科', ''),

        "content": current_content

    })

# 保存到文件

# print(json.dumps(results, ensure_ascii=False, indent=2))

with open('data/processed_data.json', 'w', encoding='utf-8') as f:

    json.dump(results, f, ensure_ascii=False, indent=2)

[

  {

    "topic": "A",

    "content": [

      "A…今后若有类似问题或者升级补丁会提前在官网告知用户。",

      "A… 参考资料 1. iPhone陷闹钟失声门 苹果称不能保证未来没问题 2. 苹果吸金有术 用户体验点石成金"

    ]

  },

  {

    "topic": "A",

    "content": [

      "A…今后若有类似问题或者升级补丁会提前在官网告知用户。",

      "A… 参考资料 1. iPhone陷闹钟失声门 苹果称不能保证未来没问题 2. 苹果吸金有术 用户体验点石成金"

    ]

  }

]

# 写入 JSONL 文件

with open('processed.jsonl', 'w', encoding='utf-8') as f:

    for entry in data:

        f.write(json.dumps(entry, ensure_ascii=False) + '\n')

import json

import jsonlines

import uuid

read_file_path='data/quality_check.json'

save_repeat_path='data/jsonlines.jsonl'

# 读取 JSON 格式文件（上次保存的）

with open(read_file_path, 'r', encoding='utf-8') as f:

    data = json.load(f)

# 写入为 JSON Lines 格式

with jsonlines.open(save_repeat_path, mode='w') as writer:

    for item in data:

        writer.write(item)

data = []

with jsonlines.open(save_repeat_path, mode='r') as reader:

    for obj in reader:

        data.append(obj)

print(data[0])

import json

read_file_path='data/processed_data.json'

with open(read_file_path, 'r', encoding='utf-8') as f:

    data = json.load(f)

# 统计 topic 数量（每个 topic 对应一个对象）

topic_count = len(data)

print(f"共包含 {topic_count} 个 topic")

from collections import Counter

# 统计每个 content 的总字符串长度

lengths = [sum(len(item) for item in entry["content"]) for entry in results]

# 打印一些基本统计信息

print(f"样本总数: {len(lengths)}")

print(f"最小长度: {min(lengths)}")

print(f"最大长度: {max(lengths)}")

print(f"平均长度: {sum(lengths) / len(lengths):.2f}")

# 获取长度分布（以10为一个分组区间，例如 0-9, 10-19,...）

bin_size = 10

bins = [((l // bin_size) * bin_size) for l in lengths]

length_distribution = Counter(bins)

# 打印分布（按长度区间升序排序）

for length_bin in sorted(length_distribution.keys()):

    print(f"{length_bin:>3} - {length_bin + bin_size - 1:>3} : {length_distribution[length_bin]}")

共包含 2 个 topic

样本总数: 2    

最小长度: 163  

最大长度: 163  

平均长度: 163.00

160 - 169 : 2

import matplotlib.pyplot as plt

plt.hist(lengths, bins=range(0, max(lengths)+10, 10), edgecolor='black')

plt.title("Content 长度分布")

plt.xlabel("总长度")

plt.ylabel("频数")

plt.grid(True)

plt.show()

plt.savefig('length_distribution.png')

with open(save_stat_path, "w", encoding="utf-8") as f:

    f.write(f"topic总数: {len(lengths)}\n")

    f.write(f"最小长度: {min(lengths)}\n")

    f.write(f"最大长度: {max(lengths)}\n")

    f.write(f"平均长度: {sum(lengths) / len(lengths):.2f}\n")

    f.write("\n长度分布:\n")

    for length_bin in sorted(length_distribution.keys()):

        f.write(f"{length_bin:>3} - {length_bin + bin_size - 1:>3} : {length_distribution[length_bin]}\n")

import json

# import re

read_file_path='data/processed_data.json'

save_repeat_path='data/topic_deduplicate.json'

# 读取保存的文件

with open(read_file_path, 'r', encoding='utf-8') as f:

    data = json.load(f)

# 用 dict 去重 topic

unique_topics = {}

for entry in data:

    topic = entry['topic']

    contents = entry['content']

    if topic in unique_topics:

        # 合并内容并去重

        existing_contents = unique_topics[topic]

        combined_contents = list(set(existing_contents + contents))

        unique_topics[topic] = combined_contents

    else:

        unique_topics[topic] = contents

# 重新组织为原始格式

results = [{"topic": topic, "content": contents} for topic, contents in unique_topics.items()]

# 保存去重后的结果

with open(save_repeat_path, 'w', encoding='utf-8') as f:

    json.dump(results, f, ensure_ascii=False, indent=2)

print("去重完成，共保留 topic 数量：", len(results))

import json

# import re

read_file_path='data/topic_deduplicate.json'

save_repeat_path='data/content_deduplicate.json'

# 读取 deduplicated 文件

with open(read_file_path, 'r', encoding='utf-8') as f:

    data = json.load(f)

def remove_contained_contents(contents):

    unique_contents = []

    for i, s1 in enumerate(contents):

        contained = False

        for j, s2 in enumerate(contents):

            if i != j and s1 in s2:

                contained = True

                break

        if not contained:

            unique_contents.append(s1)

        else:#找到需要去除的字符串打印其前10个字符

            print("重复:",s1[0:10])

    return unique_contents

# 对每个 topic 的 content 进行处理

for entry in data:

    entry['content'] = remove_contained_contents(entry['content'])

# 保存最终结果

with open(save_repeat_path, 'w', encoding='utf-8') as f:

    json.dump(data, f, ensure_ascii=False, indent=2)

print("content中重复字符串元素去除完成")

清洗前

  {

    "topic": "A",

    "content": [

      "A",

      "A\n… 参考资料 1. iPhone陷闹钟失声门 苹果称不能保证未来没问题 2. 苹果吸金有术 用户体验点石成金",

      "B。\n…今后若有类似问题或者升级补丁会提前在官网告知用户。"

    ]

  }

清洗后

  {

    "topic": "A",

    "content": [

      "A\n… 参考资料 1. iPhone陷闹钟失声门 苹果称不能保证未来没问题 2. 苹果吸金有术 用户体验点石成金",

      "B。\n…今后若有类似问题或者升级补丁会提前在官网告知用户。"

    ]

  }

import json

import re

read_file_path='data/processed_data.json'

save_repeat_path='data/repeat_stats.json'

def deduplicate_sentences(text):

    # 按标点切句（保留标点）

    sentences = re.split(r'(?<=[。！？\?])\s*', text.strip())

    seen = set()

    deduped = []

    for s in sentences:

        s = s.strip()

        if s and s not in seen:

            seen.add(s)

            deduped.append(s)

    return ''.join(deduped)

# 读取已保存的文件

with open(read_file_path, 'r', encoding='utf-8') as f:

    data = json.load(f)

# 去重处理每条 content 中的句子

for item in data:

    item['content'] = [deduplicate_sentences(text) for text in item['content']]

# 保存新文件

with open(save_repeat_path, 'w', encoding='utf-8') as f:

    json.dump(data, f, ensure_ascii=False, indent=2)

print("句子去重完成")

import json

import re

def split_sentences(text):

    # 中文及英文常见句子结束符，分割时保留标点

    sentences = re.split(r'(?<=[。！？\?])\s*', text.strip())

    return [s.strip() for s in sentences if s.strip()]

# 读取上次保存的文件

with open('processed_deduplicated.json', 'r', encoding='utf-8') as f:

    data = json.load(f)

# 跨content去重句子

for item in data:

    seen_sentences = set()

    new_content = []

    for content_text in item['content']:

        sentences = split_sentences(content_text)

        deduped_sentences = []

        for s in sentences:

            if s not in seen_sentences:

                seen_sentences.add(s)

                deduped_sentences.append(s)

        # 合并回文本

        new_content.append(''.join(deduped_sentences))

    item['content'] = new_content

# 保存处理后的文件

with open('processed_cross_deduplicated.json', 'w', encoding='utf-8') as f:

    json.dump(data, f, ensure_ascii=False, indent=2)

print("跨content句子去重完成")

import json

import re

def split_sentences(text):

    sentences = re.split(r'(?<=[。！？\?])\s*', text.strip())

    return [s.strip() for s in sentences if s.strip()]

def ngrams(sentence, n=3):

    return set([sentence[i:i+n] for i in range(len(sentence) - n + 1)])

def jaccard_sim(s1, s2, n=3):

    ngrams1 = ngrams(s1, n)

    ngrams2 = ngrams(s2, n)

    if not ngrams1 or not ngrams2:

        return 0.0

    return len(ngrams1 & ngrams2) / len(ngrams1 | ngrams2)

def is_duplicate(s, seen_sentences, threshold=0.8):

    for prev in seen_sentences:

        if jaccard_sim(s, prev) >= threshold:

            return True

    return False

# 读取文件

with open('processed_deduplicated.json', 'r', encoding='utf-8') as f:

    data = json.load(f)

# 对每个 topic 进行去重

for item in data:

    seen_sentences = []

    new_content = []

    for content_text in item['content']:

        sentences = split_sentences(content_text)

        deduped_sentences = []

        for s in sentences:

            if not is_duplicate(s, seen_sentences):

                seen_sentences.append(s)

                deduped_sentences.append(s)

        new_content.append(''.join(deduped_sentences))

    item['content'] = new_content

# 保存结果

with open('processed_ngram_deduplicated.json', 'w', encoding='utf-8') as f:

    json.dump(data, f, ensure_ascii=False, indent=2)

print("n-gram 相似度去重完成")

import difflib

def is_duplicate(s, seen_sentences, threshold=0.85):  # 你可以微调这个阈值

    for prev in seen_sentences:

        similarity = difflib.SequenceMatcher(None, s, prev).ratio()

        if similarity >= threshold:

            return True

    return False

FutureWarning: split() requires a non-empty pattern match.

  return _compile(pattern, flags).split(string, maxsplit)

import json

import re

read_file_path='data/processed_data.json'

save_repeat_path='data/repeat_stats.json'

def get_ngrams(text, n=3):

    """

    根据给定的 n 值生成文本的 n-gram 列表

    """

    return [text[i:i+n] for i in range(len(text)-n+1)] if len(text) >= n else [text]

def jaccard_similarity(ngrams1, ngrams2):

    """

    计算两个 n-gram 集合的 Jaccard 相似度

    """

    set1 = set(ngrams1)

    set2 = set(ngrams2)

    union = set1.union(set2)

    if not union:

        return 0.0

    intersection = set1.intersection(set2)

    return len(intersection) / len(union)

def detect_duplicate_sentences(content_text, n=3, sim_threshold=0.8):

    """

    对给定的文本按照句子划分，利用 n-gram 方法检测重复句子。

    返回：

      - sentences：分割后的句子列表

      - duplicates：重复句子对的索引及相似度列表

    """

    # 简单根据中文句号、问号、感叹号切分句子

    sentences = re.split(r'[。！？]', content_text)

    # 去掉空白以及冗余的空格

    sentences = [s.strip() for s in sentences if s.strip()]

    duplicates = []

    # 分别计算每个句子的 n-gram

    ngrams_list = [get_ngrams(s, n) for s in sentences]

    num = len(sentences)

    # 两两比较句子

    for i in range(num):

        for j in range(i+1, num):

            sim = jaccard_similarity(ngrams_list[i], ngrams_list[j])

            if sim >= sim_threshold:

                duplicates.append({

                    "sentence_index_pair": (i, j),

                    "sentence_pair": (sentences[i], sentences[j]),

                    "similarity": sim

                })

    return sentences, duplicates

#文件读取

with open(read_file_path, 'r', encoding='utf-8') as f:

    data = json.load(f)

# 针对每个 topic 下的每个 content 做 n-gram 句子级重复性检测

# 检测参数：n=3, 阈值sim_threshold=0.8（可根据需要调整）

detailed_results = []

for topic_item in data:

    topic_title = topic_item["topic"]

    # processed_contents = []

    # 拼接同话题下content

    all_content=""

    for content in topic_item["content"]:

        all_content+=content

        # processed_contents.append({

        #     "original_content": content,

        #     "sentences": sentences,

        #     "duplicate_sentence_info": duplicates

        # })

    # 针对同话题下所有内容调用检测函数，对内容进行句子分割和重复性检测

    sentences, duplicates = detect_duplicate_sentences(all_content, n=3, sim_threshold=0.8)

    detailed_results.append({

        "topic": topic_title,

        "content": topic_item["content"],

        "sentences": sentences,

        "duplicate_sentence_info": duplicates

    })

#文件保存

# # 输出检测结果，格式化为 JSON

# print(json.dumps(detailed_results, ensure_ascii=False, indent=2))

# 将处理结果写入文件

with open(save_repeat_path, 'w', encoding='utf-8') as f:

    json.dump(detailed_results, f, ensure_ascii=False, indent=2)

import json

read_file_path='data/sentence_deduplicate2.json'

save_repeat_path='data/remove_empty.json'

# 读取之前处理后的文件

with open(read_file_path, 'r', encoding='utf-8') as f:

    data = json.load(f)

# 去除 content 中的空字符串

for topic in data:

    topic['content'] = [content for content in topic['content'] if content.strip()]

# 保存清洗结果

with open(save_repeat_path, 'w', encoding='utf-8') as f:

    json.dump(data, f, ensure_ascii=False, indent=2)

print("已清除 content 中的空字符串")

[

  {

    "topic": "sniffer",

    "content": [

      "…elseprintf(“protocolid:%d\\n”,iphead[9])…"

    ],

    "low_quality": [

      {

        "multilingual":0

      }

    ]

  }

]

import json

import re

# import re

read_file_path='data/topic_deduplicate.json'

save_repeat_path='data/quality_check.json'

# 中文字符判断函数

def chinese_ratio(text):

    chinese_chars = re.findall(r'[\u4e00-\u9fff]', text)

    return len(chinese_chars) / len(text) if text else 0

# 读取处理后的 JSON 文件

with open(read_file_path, 'r', encoding='utf-8') as f:

    data = json.load(f)

results = []

for item in data:

    topic = item['topic']

    content = item['content']

    low_quality = []

    for idx, text in enumerate(content):

        ratio = chinese_ratio(text)

        if ratio < 0.1:

            low_quality.append({'multilingual': idx})

    result = {

        "topic": topic,

        "content": content

    }

    # if low_quality:

    #     result["low_quality"] = low_quality

    result["low_quality"] = low_quality

    results.append(result)

# 保存结果到文件

with open(save_repeat_path, 'w', encoding='utf-8') as f:

    json.dump(results, f, ensure_ascii=False, indent=2)

print("低质量内容分析完成")

[

  {

    "topic": "iPhone闹钟门",

    "content": [

      "A… 参考资料 1. iPhone陷闹钟失声门 苹果称不能保证未来没问题 2. 苹果吸金有术 用户体验点石成金",

      "A…今后若有类似问题或者升级补丁会提前在官网告知用户。"

    ],

    "low_quality": []

  },

  {

    "topic": "sniffer",

    "content": [

      "…elseprintf(“protocolid:%d\\n”,iphead[9])…"

    ],

    "low_quality": [

      {

        "multilingual": 0

      }

    ]

  }

]

import json

import re

import jsonlines

read_file_path='data/jsonlines.jsonl'

save_repeat_path='data/quality_check.jsonl'

# 中文字符判断函数

def count_chinese_ratio(text):

    chinese_chars = re.findall(r'[\u4e00-\u9fff]', text)

    return len(chinese_chars) / len(text) if text else 0

# 未换行标题检测函数

def count_unstructured_titles(text):

    count = 0

    patterns = [

        r'\[[1-9]\d*\]',                          # [1], [2], etc.

        r'第[一二三四五六七八九十百千万零1234567890]+章',     # 第一章, 第二章

        r'\d+(?:\.\d+)+',                          # 1.1.1, 1.1.2

        r'（[一二三四五六七八九十]+）',            # （一）, （二）

        r'（\d+）',                                # （1）, （2）

        r'（[A-Z]）',                              # （A）, （B）

        r'（[a-z]）',                              # （a）, （b）

        r'（[ⅠⅡⅢⅣⅤⅥⅦⅧⅨⅩ]+）',                    # （Ⅰ）, （Ⅱ）

        r'（[ⅰⅱⅲⅳⅴⅵⅶⅷⅸⅹ]+）',                    # （i）, （ii）

        r'[一二三四五六七八九十]+、',              # 一、, 二、

        r'\d+、',                                 # 1、, 2、

        r'\d+\.',                                 # 1., 2.

        r'[A-Z]\.',                               # A., B.

        r'[a-z]\)',                               # a), b)

        r'[ⅠⅡⅢⅣⅤⅥⅦⅧⅨⅩ]+\)',                     # Ⅰ), Ⅱ)

        r'[ⅰⅱⅲⅳⅴⅵⅶⅷⅸⅹ]+\)'                      # i), ii)

    ]

    # 组合正则表达式（使用多行模式 ^ 匹配行首）

    combined_pattern = re.compile(

        r'|'.join(patterns),

        flags=re.MULTILINE  # 关键：让 ^ 匹配每行的开头

    )

    # 统计匹配次数

#     matches = combined_pattern.findall(text)

    matches = list(re.finditer(combined_pattern, text))

    for m in matches:

        if not around_has_newline(text, m.start(), m.end()):

            count += 1  

    return count

# 判断匹配内容前后是否有换行

def around_has_newline(text, start, end, window=10):

    before = text[max(0, start - 5):start]

    after = text[end:end + window]

    return '\n' in before or '\n' in after

# 读取上一步处理后的 JSON 文件

data = []

with jsonlines.open(read_file_path, mode='r') as reader:

    for obj in reader:

        data.append(obj)

# 统计

results = []

for item in data:

    topic = item['topic']

    contents = item['content']

    combined_content = "".join(item['content'])

    # print("content:",combined_content)

    # low_quality = []

    # 检测中文比例

    chinese_ratio = count_chinese_ratio(combined_content)

    # 检测未换行标题个数

    title_num=count_unstructured_titles(combined_content)

    result = {

        "topic": topic,

        "content": contents,

        'chinese_ratio': chinese_ratio,

        "title_num": title_num

    }

    results.append(result)

# 写入为 JSON Lines 格式

with jsonlines.open(save_repeat_path, mode='w') as writer:

    for item in results:

        writer.write(item)

{"topic": "A", "content": ["A\n… 参考资料 1. iPhone陷闹钟失声门 苹果称不能保证未来没问题 2. 苹果吸金有术 用户体验点石成金", "A\n…今后若有类似问题或者升级补丁会提前在官网告知用户。"], "chinese_ratio": 0.6363636363636364, "title_num": 2}

{"topic": "sniffer", "content": ["…elseprintf(“protocolid:%d\\n”,iphead[9])…"], "chinese_ratio": 0.0, "title_num": 1}

{"topic": "sniffer", "content": ["第1章…第1章…第1章…第1章…"], "chinese_ratio": 0.5, "title_num": 4}

[1] ,[2] ...,  第一章、第二章， 3.1.4

# 未换行标题检测函数

def count_unstructured_titles(text):

    count = 0

    # 模式1: [1], [2]...

    matches1 = list(re.finditer(r'\[\d+\]', text))

    for m in matches1:

        if not around_has_newline(text, m.start(), m.end()):

            count += 1

    # 模式2: 第一章、第二章等

    matches2 = list(re.finditer(r'(第[一二三四五六七八九十百千万零〇]+章)', text))

    for m in matches2:

        if not around_has_newline(text, m.start(), m.end()):

            count += 1

    # 模式3: 3.1 或 2.4.5 等小节编号

    matches3 = list(re.finditer(r'(?<!\d)(\d+\.\d+(?:\.\d+)*)(?!\d)', text))

    for m in matches3:

        if not around_has_newline(text, m.start(), m.end()):

            count += 1

    return count

# 判断匹配内容前后是否有换行

def around_has_newline(text, start, end, window=10):

    before = text[max(0, start - 5):start]

    after = text[end:end + window]

    return '\n' in before or '\n' in after

import json

import re

# import re

read_file_path='data/topic_deduplicate.json'

save_repeat_path='data/quality_check.json'

# 中文字符判断函数

def chinese_ratio(text):

    chinese_chars = re.findall(r'[\u4e00-\u9fff]', text)

    return len(chinese_chars) / len(text) if text else 0

# 未换行标题检测函数

def count_unstructured_titles(text):

    count = 0

    # 模式1: [1], [2]...

    matches1 = list(re.finditer(r'\[\d+\]', text))

    for m in matches1:

        if not around_has_newline(text, m.start(), m.end()):

            count += 1

    # 模式2: 第一章、第二章等

    matches2 = list(re.finditer(r'(第[一二三四五六七八九十百千万零〇]+章)', text))

    for m in matches2:

        if not around_has_newline(text, m.start(), m.end()):

            count += 1

    # 模式3: 3.1 或 2.4.5 等小节编号

    matches3 = list(re.finditer(r'(?<!\d)(\d+\.\d+(?:\.\d+)*)(?!\d)', text))

    for m in matches3:

        if not around_has_newline(text, m.start(), m.end()):

            count += 1

    return count

# 判断匹配内容前后是否有换行

def around_has_newline(text, start, end, window=10):

    before = text[max(0, start - 5):start]

    after = text[end:end + window]

    return '\n' in before or '\n' in after

# 读取上一步处理后的 JSON 文件

with open(read_file_path, 'r', encoding='utf-8') as f:

    data = json.load(f)

results = []

for item in data:

    topic = item['topic']

    content = item['content']

    low_quality = []

    for idx, text in enumerate(content):

        # 检测中文比例

        ratio = chinese_ratio(text)

        if ratio < 0.1:

            low_quality.append({'multilingual': idx})

        # 检测未换行标题个数

        if count_unstructured_titles(text) > 5:

            low_quality.append({"multiple_titles": idx})

    result = {

        "topic": topic,

        "content": content

    }

    # if low_quality:

    #     result["low_quality"] = low_quality

    result["low_quality"] = low_quality

    results.append(result)

# 保存结果到文件

with open(save_repeat_path, 'w', encoding='utf-8') as f:

    json.dump(results, f, ensure_ascii=False, indent=2)

print("低质量内容分析完成")

  {

    "topic": "sniffer",

    "content": [

      "第一章…第一章…第一章…第一章…第一章…第一章…第一章…"

    ],

    "low_quality": [

      {

        "multiple_titles": 0

      }

    ]

  }

import json

import matplotlib.pyplot as plt

from collections import defaultdict

read_file_path='data/quality_check.json'

# 读取之前保存的标注文件

with open(read_file_path, 'r', encoding='utf-8') as f:

    data = json.load(f)

# 初始化计数器

label_counts = defaultdict(int)

total_contents = 0

# 统计所有内容字符串的数量 + 标签出现次数

for topic in data:

    total_contents += len(topic['content'])

    if 'low_quality' in topic:

        for tag in topic['low_quality']:

            for label in tag:

                label_counts[label] += 1

# 计算比例

labels = list(label_counts.keys())

ratios = [label_counts[label] / total_contents for label in labels]

# 绘图

plt.figure(figsize=(8, 5))

bars = plt.bar(labels, ratios, color='skyblue')

# 添加具体数值标注

for bar, ratio in zip(bars, ratios):

    plt.text(bar.get_x() + bar.get_width() / 2, bar.get_height(),

             f'{ratio:.2%}', ha='center', va='bottom', fontsize=10)

plt.title('Low-quality content ratio')

plt.ylabel('ratio')

plt.ylim(0, 1)

plt.grid(axis='y', linestyle='--', alpha=0.5)

plt.tight_layout()

plt.show()

import json

import matplotlib.pyplot as plt

from collections import defaultdict

read_file_path='data/quality_check.json'

format_file_path='data/processed_data.json'

# 读取之前保存标注的文件

with open(read_file_path, 'r', encoding='utf-8') as f:

    data = json.load(f)

with open(format_file_path, 'r', encoding='utf-8') as f:

    format_data = json.load(f)

# 初始化计数器

label_counts = defaultdict(int)

total_contents = 0

total_contents_before_deduplicate = 0

# 统计所有内容字符串的数量 + 标签出现次数

for topic in data:

    total_contents += len(topic['content'])

    if 'low_quality' in topic:

        for tag in topic['low_quality']:

            for label in tag:

                label_counts[label] += 1

# 统计去重前内容字符串的数量 +

for topic in format_data:

    total_contents_before_deduplicate += len(topic['content'])

print("total_contents:",total_contents)

print("total_contents_before_deduplicate:",total_contents_before_deduplicate)

# 计算比例

labels = list(label_counts.keys())

ratios = [label_counts[label] / total_contents for label in labels]

# 去重比例

labels.append("Duplicate content ratio")

ratios.append(total_contents / total_contents_before_deduplicate)

# 绘图

plt.figure(figsize=(8, 5))

bars = plt.bar(labels, ratios, color='skyblue')

# 添加具体数值标注

for bar, ratio in zip(bars, ratios):

    plt.text(bar.get_x() + bar.get_width() / 2, bar.get_height(),

             f'{ratio:.2%}', ha='center', va='bottom', fontsize=10)

plt.title('Low-quality content ratio')

plt.ylabel('ratio')

plt.ylim(0, 1)

plt.grid(axis='y', linestyle='--', alpha=0.5)

plt.tight_layout()

plt.show()

from sklearn.model_selection import train_test_split

stratified_sample, _ = train_test_split(population, test_size=0.9, stratify=population[['label']])

print (stratified_sample)

df.groupby('Grade', group_keys=False).apply(lambda x: x.sample(2))

{"topic": "A", "content": ["A\n… 参考资料 1. iPhone陷闹钟失声门 苹果称不能保证未来没问题 2. 苹果吸金有术 用户体验点石成金", "B\n…今后若有类似问题或者升级补丁会提前在官网告知用户。"], "chinese_ratio": 0.6363636363636364, "title_num": 2}

{"topic": "sniffer", "content": ["…elseprintf(“protocolid:%d\\n”,iphead[9])…"], "chinese_ratio": 0.0, "title_num": 1}

{"topic": "sniffer", "content": ["第1章…第1章…第1章…第1章…"], "chinese_ratio": 0.5, "title_num": 4}

{"topic": "A", "content": ["A。\n… 参考资料 1. iPhone陷闹钟失声门 苹果称不能保证未来没问题 2. 苹果吸金有术 用户体验点石成金", "2011年1月1日，炙手可热的iPhone遭遇“闹钟门”，导致很多上班族迟到。\n…今后若有类似问题或者升级补丁会提前在官网告知用户。"], "chinese_ratio": 0.6363636363636364, "title_num": 2}

{"topic": "sniffer", "content": ["…elseprintf(“protocolid:%d\\n”,iphead[9])…"], "chinese_ratio": 0.0, "title_num": 1}

{"topic": "sniffer", "content": ["第1章…第1章…第1章…第1章…"], "chinese_ratio": 0.5, "title_num": 4}

total_num = 1

uniq_set = []

rule_res = []

seq_len = []

chinese_ratio = []

possible_igline_num = []

for ljson in data[:2000]:

    topic = ljson['topic']

    total_num += 1

    if topic not in uniq_set:

        uniq_set.append(topic)

        text = '\n'.join(ljson['content'])

        seq_len.append(len(text))

        crat = filter_text_based_on_chinese_ratio(text)

        num = count_qing_lines(text)

        chinese_ratio.append(crat)

        possible_igline_num.append(num)

    else:

        print("重复topic:",topic)

import jsonlines

import numpy as np

import random

# 参数设置

filename = 'processed_with_ratio.jsonl'   # 你的 jsonl 文件名

n_bins = 5                                # 分成几层（bins）

sample_ratio = 0.5                        # 每层采样的比例（0~1 之间）

random.seed(42)                           # 保证可复现

# Step 1: 读取数据

data = []

with jsonlines.open(filename, mode='r') as reader:

    for item in reader:

        data.append(item)

# Step 2: 分层采样

# 将chinese_ratio按等距分bin（如0~0.2，0.2~0.4等）

bin_edges = np.linspace(0, 1, n_bins + 1)  # eg: [0. , 0.2, 0.4, ..., 1.0]

binned_data = [[] for _ in range(n_bins)]

# 分配每个item到对应bin

for item in data:

    ratio = item.get('chinese_ratio', 0.0)

    # 确保最大值归到最后一组

    for i in range(n_bins):

        if bin_edges[i] <= ratio < bin_edges[i + 1] or (i == n_bins - 1 and ratio == 1.0):

            binned_data[i].append(item)

            break

# Step 3: 每层采样

sampled_data = []

for bin_items in binned_data:

    n = int(len(bin_items) * sample_ratio)

    sampled = random.sample(bin_items, min(n, len(bin_items)))

    sampled_data.extend(sampled)

# Step 4: 打印或保存结果

print(f"原始数量: {len(data)}, 采样后数量: {len(sampled_data)}")

# 保存采样结果（可选）

with jsonlines.open('sampled_by_chinese_ratio.jsonl', mode='w') as writer:

    writer.write_all(sampled_data)

import jsonlines

import pandas as pd

read_file_path='data/quality_check.jsonl'

save_repeat_path='data/df_data.csv'

# 读取 JSONL 格式文件（上次保存的）

data = []

with jsonlines.open(read_file_path, mode='r') as reader:

    for obj in reader:

        # 拼接 content 中的所有段落为一个字符串

        obj['content'] = ''.join(obj['content'])

        data.append(obj)

# print(data[0])

#转换为 DataFrame

df = pd.DataFrame(data)

# print(df.head())

#保存为 CSV

df.to_csv(save_repeat_path, index=False)

# 读取 CSV 文件检查

df = pd.read_csv(save_repeat_path, dtype={'chinese_ratio': 'float32'})#节省内存

# 显示前几行数据

print(df.head())

import pandas as pd

import matplotlib.pyplot as plt

import seaborn as sns

read_file_path='data/df_data.csv'

# save_repeat_path='data/df_data.csv'

# 读取 CSV 文件

df = pd.read_csv(read_file_path)

# 计算 content 字符串长度

df['content_length'] = df['content'].astype(str).apply(len)

# 横轴范围（ 0 到 25000）

x_min = 0

x_max = 200

# 绘制分布图

plt.figure(figsize=(10, 6))

# sns.histplot(df['content_length'], bins=30, kde=True, color='skyblue', edgecolor='black')

sns.histplot(

    df['content_length'],

    bins=30,

    binrange=(x_min, x_max),

    kde=True,

    color='skyblue',

    edgecolor='black'

)

plt.title('Content Length Distribution')

plt.xlabel('Content Length (characters)')

plt.ylabel('Frequency')

plt.xlim(x_min, x_max)

plt.grid(True)

plt.tight_layout()

plt.show()

df['content_length'].max()

def count_sample_num(lb,ub):

    # 统计 0-50 范围内的样本数

    sample_num = df[(df['content_length'] > lb) & (df['content_length'] <= ub)].shape[0]

    # 统计 >50 的占比

    total = df.shape[0]

    count_above_ub = df[df['content_length'] > ub].shape[0]

    ratio_above_ub = count_above_ub / total * 100

    # 输出结果

    print(f"content_length 在 {lb} 到 {ub} 之间的样本数量: {sample_num}")

    print(f"content_length 大于 {ub} 的样本占比: {ratio_above_ub:.2f}%")

count_sample_num(0,50)

输出结果

content_length 在 0 到 50 之间的样本数量: 2

content_length 大于 50 的样本占比: 33.33%

#避免文字过早截断

pd.set_option('display.max_colwidth', 1000)

def content_length_samples(lb,ub,sample_n):

    # 过滤出 content_length 在 (0, 50] 的样本

    subset = df[(df['content_length'] > lb) & (df['content_length'] <= ub)]

    # 检查子集大小，确保样本数大于等于 3

    sample_size = min(sample_n, subset.shape[0])

    # 随机抽取 3 个样本

    sampled = subset.sample(n=sample_size, random_state=42)

    # 显示结果

    print(f"---content长度 在 {lb} 到 {ub} 之间的样本---")

    print("抽取样本:")

    print(sampled[['topic', 'content']])

    # 统计 0-50 范围内的样本数

    sample_num = subset.shape[0]

    # 统计 >50 的占比

    total = df.shape[0]

    count_above_ub = df[df['content_length'] > ub].shape[0]

    ratio_above_ub = count_above_ub / total * 100

    # 输出结果

    print(f"样本数量: {sample_num}")

    print(f"样本占比: {ratio_above_ub:.2f}%")

content_length_samples(0,50,3)

x_min = 0

x_max = 1

# 绘制分布图

plt.figure(figsize=(10, 6))

sns.histplot(

    # (df['chinese_ratio'] * 100).map(lambda x: f"{x:.2f}%"),

    df['chinese_ratio'],

    bins=30,

    binrange=(x_min, x_max),

    color='skyblue',

    edgecolor='black'

)

plt.title('Chinese Ratio Distribution')

plt.xlabel('Chinese Ratio')

plt.ylabel('Frequency')

plt.xlim(x_min, x_max)

plt.grid(True)

plt.tight_layout()

plt.show()

def nonchinese_ratio_samples(lb,ub,sample_n):

    label='nonchinese_ratio'

    df[label]=1-df['chinese_ratio']

    # 过滤出 content_length 在 (0, 50] 的样本

    subset = df[(df[label] > lb) & (df[label] <= ub)]

    # 检查子集大小，确保样本数大于等于 3

    sample_size = min(sample_n, subset.shape[0])

    # 随机抽取 n 个样本

    sampled = subset.sample(n=sample_size, random_state=42)

    # 显示结果

    print(f"---content长度 在 {lb} 到 {ub} 之间的样本---")

    print("抽取样本:")

    print(sampled[['topic', 'content']])

    # 统计 范围内的样本数

    sample_num = subset.shape[0]

    # 统计上界外的占比

    total = df.shape[0]

    count_above_ub = df[df[label] > ub].shape[0]

    ratio_above_ub = count_above_ub / total * 100

    # 输出结果

    print(f"区间内样本数量: {sample_num}")

    print(f"区间内样本占比: {sample_num/total* 100:.2f}%")

    print(f"大于{ub}样本占比: {ratio_above_ub:.2f}%")

nonchinese_ratio_samples(0,1,3)

columns_to_save = ['uuid', 'topic', 'content', 'chinese_ratio']

df[columns_to_save].to_csv('selected_columns.csv', index=False)

pip install LangSegment

from LangSegment import getTexts

from collections import Counter

# 示例文本，包含中文、日文、韩文和英文

text = "你的名字叫<ja>佐々木？</ja>吗？韩语中的오빠读什么呢？あなたの体育の先生は誰ですか? 此次发布会带来了四款iPhone 15系列机型和三款Apple Watch等一系列新品，这次的iPad Air采用了LCD屏幕"

# 获取分词结果

segments = getTexts(text)

# 统计每种语言的词数

lang_counts = Counter()

for seg in segments:

    lang = seg['lang']

    words = seg['text'].split()

    lang_counts[lang] += len(words)

# 输出每种语言的词数

for lang, count in lang_counts.items():

    print(f"{lang}: {count} words")

import re

import jieba

import nltk

from nltk.tokenize import word_tokenize

nltk.download('punkt')

def count_multilingual_words(text):

    # Segment Chinese text

    chinese_words = jieba.lcut(text)

    # Tokenize English and other languages

    other_words = word_tokenize(text)

    # Combine and deduplicate words

    combined_words = set(chinese_words + other_words)

    return len(combined_words)

# Example usage

text = "我爱Python编程。Hello, world!"

print("Word count:", count_multilingual_words(text))

pip install split-lang

from split_lang import LangSplitter

lang_splitter = LangSplitter()

text = "你喜欢看アニメ吗"

substr = lang_splitter.split_by_lang(

    text=text,

)

for index, item in enumerate(substr):

    print(f"{index}|{item.lang}:{item.text}")

pip install jieba kiwipiepy

pip install sudachipy sudachidict_core

from split_lang import LangSplitter

import jieba

from kiwipiepy import Kiwi

from konoha import WordTokenizer

# 初始化分词器

lang_splitter = LangSplitter()

kiwi = Kiwi()

ja_tokenizer = WordTokenizer("nagisa")  # 使用 Nagisa 作为日语分词器

text = "你喜欢看アニメ吗？Hello, world! 안녕하세요. مرحبا بالعالم"

segments = lang_splitter.split_by_lang(text=text)

word_count = 0

for segment in segments:

    lang = segment.lang

    content = segment.text.strip()

    if not content or lang == 'pun':

        continue  # 跳过空字符串和标点

    if lang == 'zh':

        tokens = list(jieba.cut(content))

    elif lang == 'ja':

        tokens = [token.surface for token in ja_tokenizer.tokenize(content)]

    elif lang == 'ko':

        tokens = [token.form for token in kiwi.tokenize(content)]

    elif lang == 'en':

        tokens = content.split()

    else:

        # 对于其他语言，按字符数统计词数

        tokens = list(content)

    word_count += len(tokens)

print(f"总词数：{word_count}")

from split_lang import LangSplitter

import jieba

from kiwipiepy import Kiwi

from sudachipy import dictionary, tokenizer

# 初始化语言分段器和分词器

lang_splitter = LangSplitter()

kiwi = Kiwi()

sudachi_tokenizer = dictionary.Dictionary().create()

sudachi_mode = tokenizer.Tokenizer.SplitMode.C  # 使用最细粒度的分词模式

# 示例多语言文本

text = "你喜欢看アニメ吗？Hello, world! 안녕하세요. مرحبا بالعالم"

# 进行语言分段

segments = lang_splitter.split_by_lang(text=text)

# 初始化词数统计

word_count = 0

# 遍历每个语言段落

for segment in segments:

    lang = segment.lang

    content = segment.text.strip()

    if not content or lang == 'pun':

        continue  # 跳过空字符串和标点

    if lang == 'zh':

        tokens = list(jieba.cut(content))

    elif lang == 'ja':

        tokens = [m.surface() for m in sudachi_tokenizer.tokenize(content, sudachi_mode)]

    elif lang == 'ko':

        tokens = [token.form for token in kiwi.tokenize(content)]

    elif lang == 'en':

        tokens = content.split()

    else:

        # 对于其他语言，按字符数统计词数

        tokens = list(content)

    word_count += len(tokens)

print(f"总词数：{word_count}")

import re

from split_lang import LangSplitter

import jieba

from kiwipiepy import Kiwi

from sudachipy import dictionary, tokenizer

# 初始化分词器

lang_splitter = LangSplitter()

kiwi = Kiwi()

sudachi_tokenizer = dictionary.Dictionary().create()

sudachi_mode = tokenizer.Tokenizer.SplitMode.C  # 最细粒度

# 示例文本

text = "你喜欢看アニメ吗？Hello, world! 안녕하세요. مرحبا بالعالم"

# 进行语言分段

segments = lang_splitter.split_by_lang(text=text)

# 初始化词数统计

word_count = 0

# 遍历每个语言段落

for segment in segments:

    lang = segment.lang

    content = segment.text.strip()

    if not content or lang == 'pun':

        continue  # 跳过空字符串和标点

    if lang == 'zh':

        tokens = list(jieba.cut(content))

    elif lang == 'ja':

        tokens = [m.surface() for m in sudachi_tokenizer.tokenize(content, sudachi_mode)]

    elif lang == 'ko':

        tokens = [token.form for token in kiwi.tokenize(content)]

    elif lang == 'en':

        tokens = content.split()

    elif lang == 'x':

        # 判断是否为英文字符

        if re.search(r'[A-Za-z]', content):

            tokens = re.findall(r'\b\w+\b', content)

        else:

            tokens = list(content)

    else:

        tokens = list(content)

    word_count += len(tokens)

print(f"总词数：{word_count}")

# 提取英文单词

english_words = re.findall(r'\b\w+\b', text)

# 统计英文单词数

english_word_count = len(english_words)

# 移除英文单词后的剩余部分

remaining_text = text

for word in english_words:

    remaining_text = remaining_text.replace(word, '')

# 移除空格后的剩余部分

remaining_text = remaining_text.replace(' ', '')

# 统计剩余字符数

remaining_char_count = len(remaining_text)