diff --git a/.gitignore b/.gitignore
index 7457a4ea..20e6b3d1 100644
--- a/.gitignore
+++ b/.gitignore
@@ -11,7 +11,6 @@ __pycache__
 *.pyc
 .vscode
 debug/
-*.ipynb
 .idea
 .python-version
 
diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml
index c0c19fe1..e4eab679 100644
--- a/.pre-commit-config.yaml
+++ b/.pre-commit-config.yaml
@@ -51,11 +51,11 @@ repos:
           - mdformat_frontmatter
           - linkify-it-py
         exclude: '^tests/.*/assets/|llm_web_kit/model/assets/.*'
-  - repo: https://github.com/myint/docformatter
-    rev: v1.3.1
-    hooks:
-      - id: docformatter
-        args: [ "--in-place", "--wrap-descriptions", "119" ]
+  # - repo: https://github.com/myint/docformatter
+  #   rev: v1.3.1
+  #   hooks:
+  #     - id: docformatter
+  #       args: [ "--in-place", "--wrap-descriptions", "119" ]
   - repo: local
     hooks:
       - id: clear-jupyter-notebook-output
diff --git a/jupyter/pjcc-prededup/cc_dedup_fir.ipynb b/jupyter/pjcc-prededup/cc_dedup_fir.ipynb
new file mode 100644
index 00000000..97fc8a13
--- /dev/null
+++ b/jupyter/pjcc-prededup/cc_dedup_fir.ipynb
@@ -0,0 +1,274 @@
+{
+ "cells": [
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "0",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "# # 获取 cc warc path list\n",
+    "# warc_paths = []\n",
+    "# for dump in DUMPS:\n",
+    "#     dump_path = f'{CC_WARC}{dump}/'\n",
+    "#     warc_paths.extend([x for x in list(list_s3_objects(dump_path, recursive=True)) if x.endswith('.jsonl.gz')])\n",
+    "\n",
+    "index_path = \"s3://qa-huawei/chupei/cc-domain-centric-store/PJCC/FILELIST/pjcc-diff-0628.txt\"\n",
+    "index_df = read_any_path(spark, index_path, config)\n",
+    "warc_paths = index_df.rdd.map(lambda row: row.value).collect()\n"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "1",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "from pyspark.sql import Row\n",
+    "from xinghe.spark import *\n",
+    "from app.common.json_util import *\n",
+    "from xinghe.s3 import *\n",
+    "from pyspark.sql.types import StructType, StructField, StringType\n",
+    "import re\n",
+    "import hashlib\n",
+    "from lxml.etree import HTML\n",
+    "import traceback\n",
+    "from datetime import datetime\n",
+    "import uuid\n",
+    "\n",
+    "# 配置\n",
+    "config = {\n",
+    "    \"spark_conf_name\": \"spark_4\",\n",
+    "    \"skip_success_check\": True,\n",
+    "    \"spark.yarn.queue\": \"pipeline.clean\",\n",
+    "    \"spark.dynamicAllocation.maxExecutors\": 2000, # 控制1万并发\n",
+    "    # \"spark.executor.memory\": \"40g\",\n",
+    "    # \"spark.executor.memoryOverhead\": \"20g\",  # 增加到40GB\n",
+    "    # \"spark.speculation\": \"true\",     # 启用推测执行\n",
+    "    # \"maxRecordsPerFile\": 200000,      # 增加每文件记录数以减少总文件数\n",
+    "    \"output_compression\": \"gz\",\n",
+    "    \"skip_output_version\": True,\n",
+    "    \"skip_output_check\": True,\n",
+    "    \"spark.sql.shuffle.partitions\": \"20000\",\n",
+    "    \"spark.default.parallelism\": \"20000\",\n",
+    "    \"spark.network.timeout\": \"1200s\",  # 网络超时\n",
+    "    \"spark.broadcast.timeout\": \"1800s\", # 增加广播超时\n",
+    "    \"spark.broadcast.compress\": \"true\", # 确保广播压缩\n",
+    "    \"spark.task.maxFailures\": 8, \n",
+    "}\n",
+    "\n",
+    "from pyspark.sql.types import StructType, StructField, StringType\n",
+    "import re\n",
+    "import hashlib\n",
+    "from lxml.etree import HTML\n",
+    "\n",
+    "MAX_OUTPUT_ROW_SIZE = 1024 * 1024 * 1024 * 1.5\n",
+    "# DUMPS = [\n",
+    "#     \"20230801\",\n",
+    "# ]\n",
+    "\n",
+    "# DUMPS = [line.strip() for line in open('pjcc_list.txt')]\n",
+    "# print(f\"加载了 {len(DUMPS)} 个dumps\")\n",
+    "\n",
+    "ERROR_PATH = 's3://qa-huawei/chupei/cc-domain-centric-store/error_logs/'\n",
+    "CC_WARC = 's3://cn-common-crawl/jsonl/'\n",
+    "output_path = \"s3://web-parse-hw60p/PJCC-dedup/hash/v5/\"\n",
+    "spark = new_spark_session(\"cc_dumps.dedup.fir\", config)\n",
+    "sc = spark.sparkContext"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "2",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "# 查看样例和数量\n",
+    "print(f\"warc_paths 总数量: {len(warc_paths)}\")\n",
+    "if warc_paths:\n",
+    "    print(f\"第一个文件路径示例: {warc_paths[0]}\")\n",
+    "    print(f\"最后一个文件路径示例: {warc_paths[-1]}\")\n",
+    "else:\n",
+    "    print(\"warc_paths 为空\")"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "3",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "from pyspark.sql import Row\n",
+    "from xinghe.spark import *\n",
+    "from app.common.json_util import *\n",
+    "from xinghe.s3 import *\n",
+    "from pyspark.sql.types import StructType, StructField, StringType\n",
+    "import re\n",
+    "import hashlib\n",
+    "from lxml.etree import HTML\n",
+    "import traceback\n",
+    "from datetime import datetime\n",
+    "import uuid\n",
+    "\n",
+    "def html_to_content(html_str: str, url: str) -> str:\n",
+    "    if html_str.strip() and isinstance(html_str,str):\n",
+    "        html_str = re.sub(r'<\\?[^>]*\\?>', '', html_str.strip())\n",
+    "        try:\n",
+    "            html_etree = HTML(html_str)\n",
+    "        except:\n",
+    "            return None\n",
+    "        if html_etree:\n",
+    "            for element in html_etree.xpath('//*[self::script or self::style]'):\n",
+    "                element.getparent().remove(element)\n",
+    "            text = ''.join(html_etree.xpath(\"//text()\"))\n",
+    "            cleaned_text = re.sub(r'[^\\w\\s]', '', text, flags=re.UNICODE)\n",
+    "            cleaned_text = re.sub(r'\\s+', '', cleaned_text).strip()\n",
+    "            return sha256_hash(cleaned_text)\n",
+    "\n",
+    "def sha256_hash(string):\n",
+    "    return hashlib.sha256(string.encode()).hexdigest()\n",
+    "    \n",
+    "# def parse_path_to_html(iter):\n",
+    "#     seen = set()\n",
+    "#     for fpath in iter:\n",
+    "#         for zz in read_s3_rows(fpath, use_stream=True):\n",
+    "#             detail_datas = json_loads(zz.value)\n",
+    "#             hash_html = html_to_content(detail_datas.get(\"html\"), detail_datas[\"url\"]) if detail_datas.get(\"html\", \"\") else None\n",
+    "#             if hash_html and hash_html not in seen:\n",
+    "#                 seen.add(hash_html)\n",
+    "#                 line = {\n",
+    "#                     \"sub_path\": fpath.split('/')[4],\n",
+    "#                     \"hash_html\": hash_html,\n",
+    "#                     \"track_id\": detail_datas[\"track_id\"],\n",
+    "#                 }\n",
+    "#                 yield Row(**{\"value\": json_dumps(line)})\n",
+    "\n",
+    "# 异常日志\n",
+    "def get_s3_doctor(target_theme):\n",
+    "    partition_id = str(uuid.uuid4())\n",
+    "    current_time = datetime.now().strftime(\"%Y%m%d\")\n",
+    "    error_log_path = f\"{ERROR_PATH}{target_theme}/{current_time}/{partition_id}.jsonl\"\n",
+    "    s3_doc_writer = S3DocWriter(path=error_log_path)\n",
+    "    return s3_doc_writer\n",
+    "\n",
+    "def parse_path_to_html(iter):\n",
+    "    seen = set()  # 保持原有的 seen 逻辑\n",
+    "    \n",
+    "    # 初始化错误日志写入器\n",
+    "    s3_doc_writer = get_s3_doctor(\"dedup_fir\")\n",
+    "    error_info = None          # 错误信息初始化\n",
+    "    \n",
+    "    for fpath in iter:        \n",
+    "        try:\n",
+    "            # 读取文件并处理\n",
+    "            for zz in read_s3_rows(fpath, use_stream=True):\n",
+    "                try:\n",
+    "                    detail_datas = json_loads(zz.value)\n",
+    "                    # 安全地获取字段，提供默认值\n",
+    "                    html_content = detail_datas.get(\"html\", \"\")\n",
+    "                    url = detail_datas.get(\"url\", \"\")\n",
+    "                    track_id = detail_datas.get(\"track_id\", \"\")\n",
+    "                    \n",
+    "                    hash_html = html_to_content(html_content, url) if html_content else None\n",
+    "                    if hash_html and hash_html not in seen:  # 保持原有的去重逻辑\n",
+    "                        seen.add(hash_html)\n",
+    "                        line = {\n",
+    "                            \"sub_path\": fpath.split('/')[4],\n",
+    "                            \"hash_html\": hash_html,\n",
+    "                            \"track_id\": track_id,\n",
+    "                            \"file_path\": fpath,\n",
+    "                        }\n",
+    "                        yield Row(**{\"value\": json_dumps(line)})\n",
+    "                        \n",
+    "                except Exception as e:\n",
+    "                    # 记录数据解析错误\n",
+    "                    error_info = {\n",
+    "                        \"error_type\": type(e).__name__,\n",
+    "                        \"error_message\": str(e),\n",
+    "                        \"traceback\": traceback.format_exc(),\n",
+    "                        \"input_data\": zz.value if hasattr(zz, 'value') else str(zz),\n",
+    "                        \"file_path\": fpath,\n",
+    "                        \"timestamp\": datetime.now().isoformat()\n",
+    "                    }\n",
+    "                    s3_doc_writer.write(error_info)\n",
+    "                    continue\n",
+    "                    \n",
+    "        except Exception as e:\n",
+    "            # 记录文件读取错误\n",
+    "            error_info = {\n",
+    "                \"error_type\": type(e).__name__,\n",
+    "                \"error_message\": str(e),\n",
+    "                \"traceback\": traceback.format_exc(),\n",
+    "                \"input_data\": \"N/A\",\n",
+    "                \"file_path\": fpath,\n",
+    "                \"timestamp\": datetime.now().isoformat()\n",
+    "            }\n",
+    "            s3_doc_writer.write(error_info)\n",
+    "            continue\n",
+    "    \n",
+    "    if error_info:\n",
+    "        # 确保所有错误日志都被写入\n",
+    "        s3_doc_writer.flush()\n"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "4",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "# mapPartitions 对 warc path 并行解析数据\n",
+    "schema = StructType([\n",
+    "    StructField(\"value\", StringType(), True),\n",
+    "])\n",
+    "page_content = sc.parallelize(warc_paths, len(warc_paths))\n",
+    "dump_html_df = page_content.mapPartitions(parse_path_to_html).toDF(schema)"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "5",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "config[\"skip_output_version\"] = True\n",
+    "config[\"output_compression\"] = \"gz\"\n",
+    "write_any_path(dump_html_df, output_path, config)"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "6",
+   "metadata": {},
+   "outputs": [],
+   "source": []
+  }
+ ],
+ "metadata": {
+  "kernelspec": {
+   "display_name": "Python 3.10 (ipykernel)",
+   "language": "python",
+   "name": "python3.10"
+  },
+  "language_info": {
+   "codemirror_mode": {
+    "name": "ipython",
+    "version": 3
+   },
+   "file_extension": ".py",
+   "mimetype": "text/x-python",
+   "name": "python",
+   "nbconvert_exporter": "python",
+   "pygments_lexer": "ipython3",
+   "version": "3.10.9"
+  }
+ },
+ "nbformat": 4,
+ "nbformat_minor": 5
+}
diff --git a/jupyter/pjcc-prededup/cc_dedup_sec.ipynb b/jupyter/pjcc-prededup/cc_dedup_sec.ipynb
new file mode 100644
index 00000000..a2c07f2c
--- /dev/null
+++ b/jupyter/pjcc-prededup/cc_dedup_sec.ipynb
@@ -0,0 +1,421 @@
+{
+ "cells": [
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "0",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "\"\"\"\n",
+    "简单的hash_html去重脚本\n",
+    "\n",
+    "使用方法：\n",
+    "1. 直接调用函数：\n",
+    "   hash_dedup_simple(spark, input_base_path, output_base_path, config)\n",
+    "\n",
+    "2. 参数说明：\n",
+    "   - input_base_path: 输入数据路径\n",
+    "   - output_base_path: 输出路径前缀\n",
+    "   - config: xinghe配置对象\n",
+    "\n",
+    "3. 输出：\n",
+    "   - 对hash_html字段进行全量去重\n",
+    "   - 提取file_path后缀（如20241109/xxx.jsonl.gz），拼接output_base_path\n",
+    "   - 使用S3DocWriter在partition内独立写文件，避免collect操作\n",
+    "\n",
+    "4. 数据格式：\n",
+    "   输入: value包含{\"sub_path\", \"hash_html\", \"track_id\", \"file_path\"}\n",
+    "   输出: 按file_path后缀分组保存到output_base_path/后缀路径\n",
+    "\"\"\"\n",
+    "\n",
+    "from pyspark.sql import SparkSession\n",
+    "from pyspark.sql.functions import *\n",
+    "from pyspark.sql.types import *\n",
+    "from xinghe.spark import *\n",
+    "from xinghe.s3 import *\n",
+    "import json\n",
+    "import uuid\n",
+    "import traceback\n",
+    "from datetime import datetime\n",
+    "\n",
+    "# 错误日志路径\n",
+    "ERROR_PATH = \"s3://qa-huawei/chupei/cc-domain-centric-store/error_logs/\"\n",
+    "\n",
+    "# 异常日志\n",
+    "def get_s3_doctor(target_theme):\n",
+    "    partition_id = str(uuid.uuid4())\n",
+    "    current_time = datetime.now().strftime(\"%Y%m%d\")\n",
+    "    error_log_path = f\"{ERROR_PATH}{target_theme}/{current_time}/{partition_id}.jsonl\"\n",
+    "    s3_doc_writer = S3DocWriter(path=error_log_path)\n",
+    "    return s3_doc_writer\n",
+    "\n",
+    "\n",
+    "def write_partition_to_s3(iterator, output_base_path):\n",
+    "    \"\"\"\n",
+    "    在partition内使用S3DocWriter写文件\n",
+    "    \"\"\"    \n",
+    "    # 初始化错误日志写入器\n",
+    "    s3_doc_writer = get_s3_doctor(\"dedup_sec\")\n",
+    "    total_records = 0\n",
+    "    \n",
+    "    try:\n",
+    "        # 按output_path分组数据\n",
+    "        file_groups = {}\n",
+    "        for row in iterator:\n",
+    "            output_path = row.output_path\n",
+    "            if output_path not in file_groups:\n",
+    "                file_groups[output_path] = []\n",
+    "            file_groups[output_path].append(row.output_value)\n",
+    "            total_records += 1\n",
+    "        \n",
+    "        # 为每个文件写入数据\n",
+    "        for output_path, values in file_groups.items():\n",
+    "            try:\n",
+    "                writer = S3DocWriter(output_path)\n",
+    "                for value in values:\n",
+    "                    # 将JSON字符串转换回字典，S3DocWriter会自动添加换行符\n",
+    "                    data_dict = json.loads(value)\n",
+    "                    writer.write(data_dict)\n",
+    "                writer.flush()\n",
+    "                print(f\"成功写入文件: {output_path}, 记录数: {len(values)}\")\n",
+    "            except Exception as e:\n",
+    "                print(f\"写入文件失败: {output_path}, 错误: {str(e)}\")\n",
+    "                \n",
+    "    except Exception as e:\n",
+    "        # 外层异常捕获，记录到错误日志\n",
+    "        error_info = {\n",
+    "            \"error_type\": type(e).__name__,\n",
+    "            \"error_message\": str(e),\n",
+    "            \"traceback\": traceback.format_exc(),\n",
+    "            \"input_data\": \"partition_processing_error\",\n",
+    "            \"stage\": \"write_partition_to_s3\",\n",
+    "            \"timestamp\": datetime.now().isoformat()\n",
+    "        }\n",
+    "        s3_doc_writer.write(error_info)\n",
+    "        s3_doc_writer.flush()\n",
+    "        print(f\"分区处理失败: {str(e)}\")\n",
+    "    \n",
+    "    return iter([total_records])  # 返回该分区处理的记录数\n",
+    "\n",
+    "def hash_dedup_simple(spark, input_base_path, output_base_path, config):\n",
+    "    \"\"\"\n",
+    "    简单的hash_html去重处理\n",
+    "    \n",
+    "    Args:\n",
+    "        spark: SparkSession\n",
+    "        input_base_path: 输入数据路径\n",
+    "        output_base_path: 输出路径前缀\n",
+    "        config: xinghe配置对象\n",
+    "    \"\"\"\n",
+    "    print(f\"开始处理hash_html去重...\")\n",
+    "    print(f\"输入路径: {input_base_path}\")\n",
+    "    print(f\"输出路径前缀: {output_base_path}\")\n",
+    "    \n",
+    "    # 使用read_any_path读取数据\n",
+    "    input_df = read_any_path(spark, input_base_path, config)\n",
+    "    \n",
+    "    # 解析JSON数据\n",
+    "    parsed_df = input_df.select(\n",
+    "        from_json(col(\"value\"), StructType([\n",
+    "            StructField(\"sub_path\", StringType()),\n",
+    "            StructField(\"hash_html\", StringType()),\n",
+    "            StructField(\"track_id\", StringType()),\n",
+    "            StructField(\"file_path\", StringType())\n",
+    "        ])).alias(\"data\")\n",
+    "    ).select(\n",
+    "        col(\"data.*\")\n",
+    "    ).filter(col(\"hash_html\").isNotNull())\n",
+    "    \n",
+    "    print(\"开始全量去重...\")\n",
+    "    # 全量去重\n",
+    "    deduped_df = parsed_df.dropDuplicates([\"hash_html\"])\n",
+    "    \n",
+    "    # 构造输出路径和数据\n",
+    "    result_df = deduped_df.withColumn(\n",
+    "        \"output_path\",\n",
+    "        concat(\n",
+    "            lit(output_base_path + \"/\"),\n",
+    "            regexp_extract(col(\"file_path\"), r\".*/([^/]+/[^/]+\\.jsonl(?:\\.gz)?)$\", 1)\n",
+    "        )\n",
+    "    ).withColumn(\n",
+    "        \"output_value\",\n",
+    "        to_json(struct(\"sub_path\", \"hash_html\", \"track_id\", \"file_path\"))\n",
+    "    )\n",
+    "    \n",
+    "    print(\"开始写入文件...\")\n",
+    "    # 使用mapPartitions在每个partition内独立写文件，避免collect\n",
+    "    # 按output_path分区，确保同一output_path的数据在同一partition\n",
+    "    total_written_records = result_df.repartition(col(\"output_path\")) \\\n",
+    "                                    .rdd \\\n",
+    "                                    .mapPartitions(lambda iterator: write_partition_to_s3(iterator, output_base_path)) \\\n",
+    "                                    .sum()  # 对所有分区的记录数求和\n",
+    "    \n",
+    "    print(f\"处理完成! 写入的总记录数: {total_written_records}\")\n",
+    "    return deduped_df\n",
+    "\n",
+    "def hash_dedup_incremental(spark, input_base_path, output_base_path, existing_base_path, config):\n",
+    "    \"\"\"\n",
+    "    增量hash_html去重处理\n",
+    "    \n",
+    "    Args:\n",
+    "        spark: SparkSession\n",
+    "        input_base_path: 新增数据路径 (如 v6)\n",
+    "        output_base_path: 输出路径前缀 (如 v2)\n",
+    "        existing_base_path: 已存在的去重数据路径 (如 v2，用于读取已有hash_html)\n",
+    "        config: xinghe配置对象\n",
+    "    \"\"\"\n",
+    "    print(f\"开始处理增量hash_html去重...\")\n",
+    "    print(f\"新增数据路径: {input_base_path}\")\n",
+    "    print(f\"已存在数据路径: {existing_base_path}\")\n",
+    "    print(f\"输出路径前缀: {output_base_path}\")\n",
+    "    \n",
+    "    # 1. 读取新增数据\n",
+    "    print(\"读取新增数据...\")\n",
+    "    new_input_df = read_any_path(spark, input_base_path, config)\n",
+    "    \n",
+    "    # 解析新增数据\n",
+    "    new_parsed_df = new_input_df.select(\n",
+    "        from_json(col(\"value\"), StructType([\n",
+    "            StructField(\"sub_path\", StringType()),\n",
+    "            StructField(\"hash_html\", StringType()),\n",
+    "            StructField(\"track_id\", StringType()),\n",
+    "            StructField(\"file_path\", StringType())\n",
+    "        ])).alias(\"data\")\n",
+    "    ).select(\n",
+    "        col(\"data.*\")\n",
+    "    ).filter(col(\"hash_html\").isNotNull())\n",
+    "    \n",
+    "    print(\"新增数据读取完成\")\n",
+    "    \n",
+    "    # 2. 高效去重策略：避免读取全量已存在数据\n",
+    "    print(\"开始高效增量去重...\")\n",
+    "    \n",
+    "    # 策略：直接对新数据进行去重，然后使用left_anti join过滤\n",
+    "    # 先对新数据内部去重，减少需要join的数据量\n",
+    "    print(\"对新数据进行内部去重...\")\n",
+    "    new_internal_deduped = new_parsed_df.dropDuplicates([\"hash_html\"])\n",
+    "    print(\"新数据内部去重完成\")\n",
+    "    \n",
+    "    # 只读取已存在数据的hash_html字段，避免读取完整记录\n",
+    "    print(\"读取已存在数据的hash_html...\")\n",
+    "    try:\n",
+    "        existing_df = read_any_path(spark, existing_base_path, config)\n",
+    "        # 直接提取hash_html，避免解析完整JSON\n",
+    "        existing_hashes = existing_df.select(\n",
+    "            get_json_object(col(\"value\"), \"$.hash_html\").alias(\"hash_html\")\n",
+    "        ).filter(col(\"hash_html\").isNotNull())\n",
+    "        \n",
+    "        print(\"已存在hash_html读取完成\")\n",
+    "        \n",
+    "        # 使用分区优化的join，确保相同hash在同一分区\n",
+    "        print(\"执行高效去重join...\")\n",
+    "        new_unique_df = new_internal_deduped.repartition(col(\"hash_html\")) \\\n",
+    "                                           .join(existing_hashes.repartition(col(\"hash_html\")), \n",
+    "                                                [\"hash_html\"], \"left_anti\")\n",
+    "        \n",
+    "    except Exception as e:\n",
+    "        print(f\"读取已存在数据失败（可能是首次运行）: {str(e)}\")\n",
+    "        print(\"请使用 hash_dedup_simple 函数进行首次全量去重\")\n",
+    "        raise e\n",
+    "    \n",
+    "    # 4. 过滤完成\n",
+    "    print(\"高效去重完成...\")\n",
+    "    \n",
+    "    print(\"重复数据过滤完成\")\n",
+    "    \n",
+    "    # 5. 构造输出路径和数据（new_unique_df已经是去重后的数据）\n",
+    "    result_df = new_unique_df.withColumn(\n",
+    "        \"output_path\",\n",
+    "        concat(\n",
+    "            lit(output_base_path + \"/\"),\n",
+    "            regexp_extract(col(\"file_path\"), r\".*/([^/]+/[^/]+\\.jsonl(?:\\.gz)?)$\", 1)\n",
+    "        )\n",
+    "    ).withColumn(\n",
+    "        \"output_value\",\n",
+    "        to_json(struct(\"sub_path\", \"hash_html\", \"track_id\", \"file_path\"))\n",
+    "    )\n",
+    "    \n",
+    "    print(\"开始写入新增数据...\")\n",
+    "    # 使用mapPartitions在每个partition内独立写文件，和hash_dedup_simple一致\n",
+    "    total_written_records = result_df.repartition(col(\"output_path\")) \\\n",
+    "                                    .rdd \\\n",
+    "                                    .mapPartitions(lambda iterator: write_partition_to_s3(iterator, output_base_path)) \\\n",
+    "                                    .sum()  # 对所有分区的记录数求和\n",
+    "    \n",
+    "    print(f\"增量去重处理完成! 写入的总记录数: {total_written_records}\")\n",
+    "    return new_unique_df\n",
+    "\n",
+    "\n",
+    "\n",
+    "# 使用示例\n",
+    "if __name__ == \"__main__\":\n",
+    "    \n",
+    "    # 配置\n",
+    "    config = {\n",
+    "        \"spark_conf_name\": \"spark_4\",\n",
+    "        \"skip_success_check\": True,\n",
+    "        \"spark.yarn.queue\": \"pipeline.clean\",\n",
+    "        \"spark.dynamicAllocation.maxExecutors\": 1000, # 控制1万并发\n",
+    "        \"spark.executor.memory\": \"80g\",\n",
+    "        \"spark.executor.memoryOverhead\": \"40g\",  # 增加到40GB\n",
+    "        # \"spark.speculation\": \"true\",     # 启用推测执行\n",
+    "        # \"maxRecordsPerFile\": 200000,      # 增加每文件记录数以减少总文件数\n",
+    "        \"output_compression\": \"gz\",\n",
+    "        \"skip_output_version\": True,\n",
+    "        \"skip_output_check\": True,\n",
+    "        \"spark.sql.shuffle.partitions\": \"10000\",  # 减少分区数\n",
+    "        \"spark.default.parallelism\": \"10000\",\n",
+    "        \n",
+    "        # Shuffle 优化配置\n",
+    "        \"spark.shuffle.io.maxRetries\": \"10\",  # 增加shuffle重试次数\n",
+    "        \"spark.shuffle.io.retryWait\": \"30s\",  # 重试等待时间\n",
+    "        \"spark.shuffle.compress\": \"true\",  # 启用shuffle压缩\n",
+    "        \"spark.shuffle.spill.compress\": \"true\",  # 启用spill压缩\n",
+    "        \n",
+    "        # 网络和超时配置\n",
+    "        \"spark.network.timeout\": \"3600s\",  # 进一步增加网络超时\n",
+    "        \"spark.broadcast.timeout\": \"3600s\", \n",
+    "        \"spark.broadcast.compress\": \"true\",\n",
+    "        \"spark.rpc.askTimeout\": \"3600s\",   \n",
+    "        \"spark.rpc.lookupTimeout\": \"3600s\", \n",
+    "        \"spark.storage.blockManagerSlaveTimeoutMs\": \"3600000\",\n",
+    "        \n",
+    "    }\n",
+    "    spark = new_spark_session(\"cc_dumps.dedup.sec\", config)\n",
+    "    sc = spark.sparkContext\n",
+    "    sc.setLogLevel(\"ERROR\")\n",
+    "    sc\n",
+    "    \n",
+    "    # 示例用法\n",
+    "    \n",
+    "    # 方案1: 全量去重（首次运行）\n",
+    "    # input_base_path = \"s3://web-parse-hw60p/PJCC-dedup/hash/v5\"\n",
+    "    # output_base_path = \"s3://web-parse-hw60p/PJCC-dedup/hash-dedup/v2\"\n",
+    "    # result = hash_dedup_simple(spark, input_base_path, output_base_path, config)\n",
+    "    \n",
+    "    # 方案2: 增量去重（后续运行）\n",
+    "    input_base_path = \"s3://web-parse-hw60p/PJCC-dedup/hash/v6\"  # 新增数据\n",
+    "    output_base_path = \"s3://web-parse-hw60p/PJCC-dedup/hash-dedup/v2\"  # 输出路径\n",
+    "    existing_base_path = \"s3://web-parse-hw60p/PJCC-dedup/hash-dedup/v2\"  # 已存在数据路径\n",
+    "    \n",
+    "    # 执行增量去重\n",
+    "    result = hash_dedup_incremental(spark, input_base_path, output_base_path, existing_base_path, config)\n",
+    "    \n",
+    "    spark.stop() "
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "1",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "# Spark配置\n",
+    "config = {\n",
+    "    \"spark_conf_name\": \"spark_4\",\n",
+    "    \"skip_success_check\": True,\n",
+    "    \"spark.yarn.queue\": \"pipeline.clean\",\n",
+    "    \"spark.dynamicAllocation.maxExecutors\": 2000,  # 控制1万并发\n",
+    "    \"output_compression\": \"gz\",\n",
+    "    \"skip_output_version\": True,\n",
+    "    \"skip_output_check\": True,\n",
+    "    \"spark.sql.shuffle.partitions\": \"20000\",\n",
+    "    \"spark.default.parallelism\": \"20000\",\n",
+    "    \"spark.network.timeout\": \"1200s\",  # 网络超时\n",
+    "    \"spark.broadcast.timeout\": \"1800s\",  # 增加广播超时\n",
+    "    \"spark.broadcast.compress\": \"true\",  # 确保广播压缩\n",
+    "    \"spark.task.maxFailures\": 8,\n",
+    "}\n",
+    "    \n",
+    "from xinghe.spark.session_ext import new_spark_session   \n",
+    "from xinghe.spark.read_ext import read_any_path\n",
+    "spark = new_spark_session(\"extract_unique_file_paths\", config)\n",
+    "sc = spark.sparkContext\n",
+    "sc.setLogLevel(\"ERROR\")\n",
+    "input_path = \"s3://web-parse-hw60p/PJCC-dedup/hash-dedup/v2/\"\n",
+    "input_df = read_any_path(spark, input_path, config)\n",
+    "    \n",
+    "total_records = input_df.count()\n",
+    "print(f\"总记录数: {total_records:,}\")"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "2",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "from pyspark.sql import SparkSession\n",
+    "from pyspark.sql.functions import *\n",
+    "from pyspark.sql.types import *\n",
+    "from xinghe.spark import *\n",
+    "from xinghe.s3 import *\n",
+    "import json\n",
+    "import uuid\n",
+    "import traceback\n",
+    "from datetime import datetime\n",
+    "\n",
+    "input_base_path = \"s3://web-parse-hw60p/PJCC-dedup/hash/v5/20230801\"\n",
+    "input_df = read_any_path(spark, input_base_path, config)\n",
+    "input_df.count()"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "3",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "input_base_path = \"s3://web-parse-hw60p/PJCC-dedup/hash_dedup/v1/20230801\"\n",
+    "input_df = read_any_path(spark, input_base_path, config)\n",
+    "input_df.count()"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "4",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "files = list(list_s3_objects('s3://web-parse-hw60p/PJCC-dedup/hash/v5/20230801/', is_prefix=True))\n",
+    "print(len(files))\n",
+    "files = list(list_s3_objects('s3://web-parse-hw60p/PJCC-dedup/hash_dedup/v1/20230801/', is_prefix=True))\n",
+    "print(len(files))"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "5",
+   "metadata": {},
+   "outputs": [],
+   "source": []
+  }
+ ],
+ "metadata": {
+  "kernelspec": {
+   "display_name": "Python 3.10 (ipykernel)",
+   "language": "python",
+   "name": "python3.10"
+  },
+  "language_info": {
+   "codemirror_mode": {
+    "name": "ipython",
+    "version": 3
+   },
+   "file_extension": ".py",
+   "mimetype": "text/x-python",
+   "name": "python",
+   "nbconvert_exporter": "python",
+   "pygments_lexer": "ipython3",
+   "version": "3.10.9"
+  }
+ },
+ "nbformat": 4,
+ "nbformat_minor": 5
+}
diff --git a/jupyter/pjcc-prededup/cc_dedup_thr.ipynb b/jupyter/pjcc-prededup/cc_dedup_thr.ipynb
new file mode 100644
index 00000000..fcaf0076
--- /dev/null
+++ b/jupyter/pjcc-prededup/cc_dedup_thr.ipynb
@@ -0,0 +1,348 @@
+{
+ "cells": [
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "0",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "\"\"\"\n",
+    "根据hash_html去重后数据，进行html去重脚本\n",
+    "\n",
+    "使用方法：\n",
+    "1. 直接调用函数：\n",
+    "   html_dedup(spark, input_hash_base_path, input_html_base_path, output_html_base_path, output_hash_with_domain_base_path, config, index_path)\n",
+    "\n",
+    "2. 参数说明：\n",
+    "   - input_hash_base_path: 输入hash_html去重后的数据路径\n",
+    "   - input_html_base_path: 输入原始html数据路径\n",
+    "   - output_html_base_path: 输出html去重后的数据路径\n",
+    "   - output_hash_with_domain_base_path: 输出hash+domain数据路径\n",
+    "   - config: xinghe配置对象\n",
+    "   - index_path: 文件列表路径，用于获取需要处理的文件后缀\n",
+    "\n",
+    "3. 处理逻辑：\n",
+    "   - 从index_path读取文件列表，提取文件后缀进行并行处理\n",
+    "   - 对每个文件后缀，分别读取hash去重数据和原始html数据\n",
+    "   - 通过track_id字段进行inner join，保留在hash去重结果中的html数据\n",
+    "   - 从URL提取domain和domain_hash_id字段\n",
+    "   - 生成两个输出：HTML去重数据和Hash+Domain数据\n",
+    "\n",
+    "4. 文件路径示例：\n",
+    "   - Hash去重文件: s3://web-parse-hw60p/PJCC-dedup/hash-dedup/v1/20230801/xxx.jsonl.gz\n",
+    "   - 原始HTML文件: s3://cn-common-crawl/jsonl/20230801/xxx.jsonl.gz\n",
+    "   - 输出HTML文件: s3://web-parse-hw60p/PJCC-dedup/html-dedup/v1/20230801/xxx.jsonl.gz\n",
+    "   - 输出Hash+Domain文件: s3://web-parse-hw60p/PJCC-dedup/hash-dedup-with-domain/v1/20230801/xxx.jsonl.gz\n",
+    "\n",
+    "5. 数据格式：\n",
+    "   - Hash去重数据: {\"sub_path\", \"hash_html\", \"track_id\", \"file_path\"}\n",
+    "   - 原始HTML数据: {\"sub_path\", \"html\", \"track_id\", \"file_path\", \"url\", ...}\n",
+    "   - 输出HTML数据: 原始HTML数据 + domain相关字段\n",
+    "   - 输出Hash+Domain数据: Hash去重数据 + {\"url\", \"domain\", \"domain_hash_id\"}\n",
+    "\"\"\"\n",
+    "\n",
+    "from pyspark.sql import SparkSession\n",
+    "from pyspark.sql.functions import *\n",
+    "from pyspark.sql.types import *\n",
+    "from xinghe.spark import *\n",
+    "from xinghe.s3 import *\n",
+    "import json\n",
+    "import uuid\n",
+    "import traceback\n",
+    "from datetime import datetime\n",
+    "from urllib.parse import urlparse\n",
+    "import xxhash\n",
+    "\n",
+    "# 错误日志路径\n",
+    "ERROR_PATH = \"s3://qa-huawei/chupei/cc-domain-centric-store/error_logs/\"\n",
+    "\n",
+    "# 异常日志\n",
+    "def get_s3_doctor(target_theme):\n",
+    "    partition_id = str(uuid.uuid4())\n",
+    "    current_time = datetime.now().strftime(\"%Y%m%d\")\n",
+    "    error_log_path = f\"{ERROR_PATH}{target_theme}/{current_time}/{partition_id}.jsonl\"\n",
+    "    s3_doc_writer = S3DocWriter(path=error_log_path)\n",
+    "    return s3_doc_writer\n",
+    "\n",
+    "# 定义提取domain的UDF\n",
+    "def extract_domain(url):\n",
+    "    if url is None:\n",
+    "        return None\n",
+    "    try:\n",
+    "        hostname = urlparse(url).hostname\n",
+    "        return hostname.lower() if hostname else None\n",
+    "    except Exception as e:\n",
+    "        return None\n",
+    "\n",
+    "# 定义计算domain_hash_id的UDF\n",
+    "HASH_COUNT = 10000\n",
+    "def compute_domain_hash(domain):\n",
+    "    if domain is None:\n",
+    "        return None\n",
+    "    return xxhash.xxh64_intdigest(domain) % HASH_COUNT\n",
+    "\n",
+    "def process_file_pair(file_suffix, input_hash_base_path, input_html_base_path, output_html_base_path, output_hash_with_domain_base_path):\n",
+    "    \"\"\"\n",
+    "    处理一对文件的html去重 - 在partition内使用直接文件操作\n",
+    "    \"\"\"\n",
+    "    \n",
+    "    # 初始化错误日志写入器\n",
+    "    s3_doc_writer = get_s3_doctor(\"dedup_thr\")\n",
+    "    \n",
+    "    try:\n",
+    "        # 构造文件路径\n",
+    "        hash_file_path = f\"{input_hash_base_path}/{file_suffix}\"\n",
+    "        html_file_path = f\"{input_html_base_path}/{file_suffix}\"\n",
+    "        output_file_path = f\"{output_html_base_path}/{file_suffix}\"\n",
+    "        output_hash_with_domain_file_path = f\"{output_hash_with_domain_base_path}/{file_suffix}\"\n",
+    "        \n",
+    "        print(f\"开始处理文件对:\")\n",
+    "        print(f\"  Hash文件: {hash_file_path}\")\n",
+    "        print(f\"  HTML文件: {html_file_path}\")\n",
+    "        print(f\"  输出HTML文件: {output_file_path}\")\n",
+    "        print(f\"  输出Hash+Domain文件: {output_hash_with_domain_file_path}\")\n",
+    "        \n",
+    "        # 读取hash去重后的数据\n",
+    "        hash_data_dict = {}  # track_id -> hash_data\n",
+    "        for zz in read_s3_rows(hash_file_path, use_stream=True):\n",
+    "            try:\n",
+    "                hash_detail = json.loads(zz.value)\n",
+    "                track_id = hash_detail.get(\"track_id\")\n",
+    "                if track_id:\n",
+    "                    hash_data_dict[track_id] = hash_detail\n",
+    "            except Exception as e:\n",
+    "                # 记录解析hash数据的错误\n",
+    "                error_info = {\n",
+    "                    \"error_type\": type(e).__name__,\n",
+    "                    \"error_message\": str(e),\n",
+    "                    \"traceback\": traceback.format_exc(),\n",
+    "                    \"input_data\": f\"file_suffix: {file_suffix}, hash_file: {hash_file_path}, raw_data: {zz.value[:200] if hasattr(zz, 'value') else 'N/A'}\",\n",
+    "                    \"stage\": \"parse_hash_data\",\n",
+    "                    \"timestamp\": datetime.now().isoformat()\n",
+    "                }\n",
+    "                s3_doc_writer.write(error_info)\n",
+    "                print(f\"解析hash数据失败: {str(e)}\")\n",
+    "                continue\n",
+    "        \n",
+    "        print(f\"读取hash数据: {len(hash_data_dict)} 条记录\")\n",
+    "        \n",
+    "        # 初始化writer - 在循环外创建，避免重复创建\n",
+    "        html_writer = S3DocWriter(output_file_path)\n",
+    "        hash_domain_writer = S3DocWriter(output_hash_with_domain_file_path)\n",
+    "        \n",
+    "        # 读取原始html数据并进行join，同时写入结果\n",
+    "        html_count = 0\n",
+    "        hash_domain_count = 0\n",
+    "        \n",
+    "        for zz in read_s3_rows(html_file_path, use_stream=True):\n",
+    "            try:\n",
+    "                html_detail = json.loads(zz.value)\n",
+    "                track_id = html_detail.get(\"track_id\")\n",
+    "                \n",
+    "                # 通过track_id进行inner join\n",
+    "                if track_id and track_id in hash_data_dict:\n",
+    "                    # 处理URL提取domain\n",
+    "                    url = html_detail.get(\"url\")\n",
+    "                    domain = extract_domain(url)\n",
+    "                    domain_hash_id = compute_domain_hash(domain)\n",
+    "                    \n",
+    "                    # 构造HTML去重结果并立即写入\n",
+    "                    html_result = html_detail.copy()\n",
+    "                    html_result[\"domain\"] = domain\n",
+    "                    html_result[\"domain_hash_id\"] = domain_hash_id\n",
+    "                    html_writer.write(html_result)\n",
+    "                    html_count += 1\n",
+    "                    \n",
+    "                    # 构造Hash+Domain结果并立即写入\n",
+    "                    hash_result = hash_data_dict[track_id].copy()\n",
+    "                    # 去掉sub_path字段\n",
+    "                    if \"sub_path\" in hash_result:\n",
+    "                        del hash_result[\"sub_path\"]\n",
+    "                    hash_result[\"url\"] = url\n",
+    "                    hash_result[\"domain\"] = domain\n",
+    "                    hash_result[\"domain_hash_id\"] = domain_hash_id\n",
+    "                    hash_domain_writer.write(hash_result)\n",
+    "                    hash_domain_count += 1\n",
+    "                    \n",
+    "            except Exception as e:\n",
+    "                # 记录处理html数据的错误\n",
+    "                error_info = {\n",
+    "                    \"error_type\": type(e).__name__,\n",
+    "                    \"error_message\": str(e),\n",
+    "                    \"traceback\": traceback.format_exc(),\n",
+    "                    \"input_data\": f\"file_suffix: {file_suffix}, html_file: {html_file_path}, raw_data: {zz.value[:200] if hasattr(zz, 'value') else 'N/A'}\",\n",
+    "                    \"stage\": \"process_html_data\",\n",
+    "                    \"timestamp\": datetime.now().isoformat()\n",
+    "                }\n",
+    "                s3_doc_writer.write(error_info)\n",
+    "                print(f\"处理html数据失败: {str(e)}\")\n",
+    "                continue\n",
+    "        \n",
+    "        # 确保所有数据都写入\n",
+    "        html_writer.flush()\n",
+    "        hash_domain_writer.flush()\n",
+    "        \n",
+    "        print(f\"Join结果: {html_count} 条记录\")\n",
+    "        print(f\"成功写入HTML去重文件: {output_file_path}, 记录数: {html_count}\")\n",
+    "        print(f\"成功写入Hash+Domain文件: {output_hash_with_domain_file_path}, 记录数: {hash_domain_count}\")\n",
+    "            \n",
+    "        return html_count\n",
+    "        \n",
+    "    except Exception as e:\n",
+    "        # 记录错误日志\n",
+    "        error_info = {\n",
+    "            \"error_type\": type(e).__name__,\n",
+    "            \"error_message\": str(e),\n",
+    "            \"traceback\": traceback.format_exc(),\n",
+    "            \"input_data\": f\"file_suffix: {file_suffix}\",\n",
+    "            \"stage\": \"process_file_pair\",\n",
+    "            \"timestamp\": datetime.now().isoformat()\n",
+    "        }\n",
+    "        s3_doc_writer.write(error_info)\n",
+    "        s3_doc_writer.flush()\n",
+    "        print(f\"文件处理失败: {file_suffix}, 错误: {str(e)}\")\n",
+    "        return 0\n",
+    "\n",
+    "def html_dedup(spark, input_hash_base_path, input_html_base_path, output_html_base_path, output_hash_with_domain_base_path, config, index_path):\n",
+    "    \"\"\"\n",
+    "    根据hash_html去重后数据，进行html去重\n",
+    "    \n",
+    "    Args:\n",
+    "        spark: SparkSession\n",
+    "        input_hash_base_path: 输入hash_html去重后的数据路径\n",
+    "        input_html_base_path: 输入html数据路径\n",
+    "        output_html_base_path: 输出html去重后的数据路径\n",
+    "        output_hash_with_domain_base_path: 输出hash+domain数据路径\n",
+    "        config: xinghe配置对象\n",
+    "        index_path: 文件列表路径\n",
+    "    \"\"\"\n",
+    "    print(f\"开始处理html去重...\")\n",
+    "    print(f\"Hash去重数据路径: {input_hash_base_path}\")\n",
+    "    print(f\"HTML数据路径: {input_html_base_path}\")\n",
+    "    print(f\"输出HTML路径: {output_html_base_path}\")\n",
+    "    print(f\"输出Hash+Domain路径: {output_hash_with_domain_base_path}\")\n",
+    "    print(f\"文件列表路径: {index_path}\")\n",
+    "    \n",
+    "    # 读取文件列表\n",
+    "    index_df = read_any_path(spark, index_path, config)\n",
+    "    paths = index_df.rdd.map(lambda row: row.value).collect()\n",
+    "    \n",
+    "    print(f\"总文件数: {len(paths)}\")\n",
+    "    \n",
+    "    # 并行处理文件\n",
+    "    paths_rdd = spark.sparkContext.parallelize(paths)\n",
+    "    \n",
+    "    # 使用mapPartitions进行批量处理\n",
+    "    def process_partition(file_paths):\n",
+    "        results = []\n",
+    "        for file_path in file_paths:\n",
+    "            # 提取文件后缀\n",
+    "            file_suffix = file_path.split('/')[-2] + '/' + file_path.split('/')[-1]\n",
+    "            result_count = process_file_pair(\n",
+    "                file_suffix, \n",
+    "                input_hash_base_path, \n",
+    "                input_html_base_path, \n",
+    "                output_html_base_path, \n",
+    "                output_hash_with_domain_base_path\n",
+    "            )\n",
+    "            results.append((file_suffix, result_count))\n",
+    "        return results\n",
+    "    \n",
+    "    # 执行处理\n",
+    "    paths_rdd.mapPartitions(process_partition).count()  # 触发执行\n",
+    "    \n",
+    "    print(f\"处理完成!\")\n",
+    "    print(f\"总文件数: {len(paths)}\")\n",
+    "    \n",
+    "    return True\n",
+    "\n",
+    "\n",
+    "# 使用示例\n",
+    "if __name__ == \"__main__\":\n",
+    "    \n",
+    "       # 配置\n",
+    "    config = {\n",
+    "        \"spark_conf_name\": \"spark_4\",\n",
+    "        \"skip_success_check\": True,\n",
+    "        \"spark.yarn.queue\": \"pipeline.clean\",\n",
+    "        \"spark.dynamicAllocation.maxExecutors\": 2000, # 控制1万并发\n",
+    "        # \"spark.executor.memory\": \"80g\",\n",
+    "        # \"spark.executor.memoryOverhead\": \"40g\",  # 增加到40GB\n",
+    "        # \"spark.speculation\": \"true\",     # 启用推测执行\n",
+    "        # \"maxRecordsPerFile\": 200000,      # 增加每文件记录数以减少总文件数\n",
+    "        \"output_compression\": \"gz\",\n",
+    "        \"skip_output_version\": True,\n",
+    "        \"skip_output_check\": True,\n",
+    "        \"spark.sql.shuffle.partitions\": \"20000\",  # 减少分区数\n",
+    "        \"spark.default.parallelism\": \"20000\",\n",
+    "        \n",
+    "        # Shuffle 优化配置\n",
+    "        \"spark.shuffle.io.maxRetries\": \"10\",  # 增加shuffle重试次数\n",
+    "        \"spark.shuffle.io.retryWait\": \"30s\",  # 重试等待时间\n",
+    "        \"spark.shuffle.compress\": \"true\",  # 启用shuffle压缩\n",
+    "        \"spark.shuffle.spill.compress\": \"true\",  # 启用spill压缩\n",
+    "        \n",
+    "        # 网络和超时配置\n",
+    "        \"spark.network.timeout\": \"3600s\",  # 进一步增加网络超时\n",
+    "        \"spark.broadcast.timeout\": \"3600s\", \n",
+    "        \"spark.broadcast.compress\": \"true\",\n",
+    "        \"spark.rpc.askTimeout\": \"3600s\",   \n",
+    "        \"spark.rpc.lookupTimeout\": \"3600s\", \n",
+    "        \"spark.storage.blockManagerSlaveTimeoutMs\": \"3600000\",\n",
+    "        \n",
+    "    }\n",
+    "    spark = new_spark_session(\"cc_dumps.dedup.thr\", config)\n",
+    "    sc = spark.sparkContext\n",
+    "    sc.setLogLevel(\"ERROR\")\n",
+    "    sc\n",
+    "    \n",
+    "    # 示例用法\n",
+    "    input_hash_base_path = \"s3://web-parse-hw60p/PJCC-dedup/hash-dedup/v2\"\n",
+    "    input_html_base_path = \"s3://cn-common-crawl/jsonl\"\n",
+    "    output_html_base_path = \"s3://web-parse-hw60p/PJCC-dedup/html-dedup/v1\"\n",
+    "    output_hash_with_domain_base_path = \"s3://web-parse-hw60p/PJCC-dedup/hash-dedup-with-domain/v1\"\n",
+    "    index_path = \"s3://qa-huawei/chupei/cc-domain-centric-store/PJCC/FILELIST/pjcc-hash-dedup-v2.txt\"\n",
+    "    # test config\n",
+    "    # output_html_base_path = \"s3://qa-huawei/chupei/cc-domain-centric-store/PJCC-dedup/html-dedup/v1\"\n",
+    "    # output_hash_with_domain_base_path = \"s3://qa-huawei/chupei/cc-domain-centric-store/PJCC-dedup/hash-dedup-with-domain/v1\"\n",
+    "    # index_path = \"s3://qa-huawei/chupei/cc-domain-centric-store/PJCC/FILELIST/pjcc-hash-dedup-test100.txt\"\n",
+    "    \n",
+    "    \n",
+    "    # 执行去重\n",
+    "    result = html_dedup(spark, input_hash_base_path, input_html_base_path, output_html_base_path, output_hash_with_domain_base_path, config, index_path)\n",
+    "    \n",
+    "    spark.stop() "
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "1",
+   "metadata": {},
+   "outputs": [],
+   "source": []
+  }
+ ],
+ "metadata": {
+  "kernelspec": {
+   "display_name": "Python 3.10 (ipykernel)",
+   "language": "python",
+   "name": "python3.10"
+  },
+  "language_info": {
+   "codemirror_mode": {
+    "name": "ipython",
+    "version": 3
+   },
+   "file_extension": ".py",
+   "mimetype": "text/x-python",
+   "name": "python",
+   "nbconvert_exporter": "python",
+   "pygments_lexer": "ipython3",
+   "version": "3.10.9"
+  }
+ },
+ "nbformat": 4,
+ "nbformat_minor": 5
+}
diff --git a/llm_web_kit/extractor/html/recognizer/cc_math/tag_img.py b/llm_web_kit/extractor/html/recognizer/cc_math/tag_img.py
index e1f500c9..ae9a8700 100644
--- a/llm_web_kit/extractor/html/recognizer/cc_math/tag_img.py
+++ b/llm_web_kit/extractor/html/recognizer/cc_math/tag_img.py
@@ -1,3 +1,4 @@
+import re
 from urllib.parse import unquote
 
 from lxml.html import HtmlElement
@@ -45,7 +46,10 @@ def is_display_mode(node, src_name):
                 return True
 
             # 4. 检查图片尺寸
-            if node.get('width') and int(node.get('width', '0')) > 100:
+            width_str = node.get('width', '')
+            # 提取数字部分，处理带单位的情况（如 "100px"）
+            width_match = re.match(r'^(\d+)', width_str)
+            if width_match and int(width_match.group(1)) > 100:
                 return True
 
             # 5. 检查是否后面紧跟<br>标签
diff --git a/tests/llm_web_kit/simple/test_simple.py b/tests/llm_web_kit/simple/test_simple.py
index 442af4a2..d5305894 100644
--- a/tests/llm_web_kit/simple/test_simple.py
+++ b/tests/llm_web_kit/simple/test_simple.py
@@ -718,6 +718,56 @@ def test_extract_main_html_with_table_with_math(self):
         self.assertIn('| $n$ | $785$ | $885$ | $1667$ |', md)
         self.assertIn('| $\\chi(n)$ | $e\\left(\\frac{3}{4}\\right)$ | $e\\left(\\frac{2}{3}\\right)$ | $-1$ |', md)
 
+    def test_extract_main_html_with_math_img_width_various_formats(self):
+        """测试img标签width属性各种格式的情况，验证不会抛出异常."""
+        main_html = r'''
+        <p>Some text with inline formula:</p>
+        <!-- 带单位 px -->
+        <img src="https://latex.codecogs.com/gif.latex?E=mc^2" alt="$E=mc^2$" width="50px" height="20px" />
+        <p>And a larger image:</p>
+        <!-- 带单位 px，超过100 -->
+        <img src="https://example.com/large.png" alt="large image" width="200px" />
+        <p>Image with percent width:</p>
+        <!-- 百分比 -->
+        <img src="https://latex.codecogs.com/gif.latex?a^2+b^2=c^2" alt="$a^2+b^2=c^2$" width="80%" />
+        <p>Image with float width:</p>
+        <!-- 浮点数 -->
+        <img src="https://latex.codecogs.com/gif.latex?x^n" alt="$x^n$" width="512.123" />
+        <p>Image with float width and unit:</p>
+        <!-- 浮点数带单位 -->
+        <img src="https://latex.codecogs.com/gif.latex?y^m" alt="$y^m$" width="123.456px" />
+        <p>Image with auto width:</p>
+        <!-- auto 值 -->
+        <img src="https://latex.codecogs.com/gif.latex?z^k" alt="$z^k$" width="auto" />
+        <p>Image with em unit:</p>
+        <!-- em 单位 -->
+        <img src="https://latex.codecogs.com/gif.latex?w^j" alt="$w^j$" width="10em" />
+        <p>Image with empty width:</p>
+        <!-- 空值 -->
+        <img src="https://latex.codecogs.com/gif.latex?v^i" alt="$v^i$" width="" />
+        '''
+
+        # 这个测试主要验证不会因为各种 width 值而抛出异常
+        md = extract_content_from_main_html(self.url, main_html)
+        print(md)
+
+        # 验证文本内容存在
+        self.assertIn('Some text with inline formula', md)
+        self.assertIn('Image with float width', md)
+        self.assertIn('Image with auto width', md)
+
+        # 验证 img 中的数学公式被正确提取
+        # width <= 100 的是行内公式 $...$
+        self.assertIn('$E=mc^2$', md)  # width="50px", 50 <= 100
+        self.assertIn('$a^2+b^2=c^2$', md)  # width="80%", 80 <= 100
+        self.assertIn('$z^k$', md)  # width="auto", 无数字
+        self.assertIn('$w^j$', md)  # width="10em", 10 <= 100
+        self.assertIn('$v^i$', md)  # width="", 空值
+
+        # width > 100 的是行间公式 $$...$$ (多行格式)
+        self.assertIn('$$\nx^n\n$$', md)  # width="512.123", 512 > 100
+        self.assertIn('$$\ny^m\n$$', md)  # width="123.456px", 123 > 100
+
     def test_extract_magic_html_with_mathjax(self):
         """测试包含MathJax数学公式的HTML内容提取."""
         raw_html = r'''
@@ -752,3 +802,4 @@ def test_extract_magic_html_with_mathjax(self):
 
 if __name__ == '__main__':
     unittest.main(verbosity=2)
+    TestSimple().test_extract_main_html_with_math_img_width_various_formats()