add dedup

jupyter/main-html-dedup/README.md

@@ -0,0 +1,22 @@
+## step1: main_html的预去重
+
+### step1-1: 计算text的hash
+
+cc_dedup_fir.ipynb
+input_path为输入的路径， ouput_path是输出的路径
+
+### step1-2: 做hash的dedup
+
+cc_dedup_dec.ipynb
+base_input_path等于step1-1的ouput_path
+output_path为去重后的id、hash对
+
+### step1-3: 将去重后的id与content等原始数据的字段join起来
+
+CC_WARC为原始的数据路径，即step1-1的input_path；
+base_unique_path为step1-2的ouput_path
+output_path 为需要输出的去重后路径
+
+## step2: span级别的去重
+
+input_path为step1-3的ouput_path

jupyter/main-html-dedup/cc_dedup_fir.ipynb

@@ -0,0 +1,258 @@
+{
+ "cells": [
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "0",
+   "metadata": {
+    "execution": {
+     "iopub.execute_input": "2025-07-11T02:24:06.149493Z",
+     "iopub.status.busy": "2025-07-11T02:24:06.149207Z",
+     "iopub.status.idle": "2025-07-11T02:24:14.378413Z",
+     "shell.execute_reply": "2025-07-11T02:24:14.377755Z",
+     "shell.execute_reply.started": "2025-07-11T02:24:06.149476Z"
+    }
+   },
+   "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",
+    "config = {\n",
+    "    \"spark_conf_name\": \"spark_4\",\n",
+    "    \"skip_success_check\": True,\n",
+    "}\n",
+    "\n",
+    "\n",
+    "MAX_OUTPUT_ROW_SIZE = 1024 * 1024 * 1024 * 1.5\n",
+    "DUMPS = [\n",
+    "    ...\n",
+    "]\n",
+    "\n",
+    "ERROR_PATH = \"xx\"\n",
+    "input_path = ['xx']\n",
+    "output_path = \"xx\"\n",
+    "spark = new_spark_session(\"cc_dumps.dedup.fir\", config)\n",
+    "sc = spark.sparkContext"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "id": "1",
+   "metadata": {},
+   "source": [
+    "# html source"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "2",
+   "metadata": {
+    "execution": {
+     "iopub.execute_input": "2025-07-11T02:24:14.379728Z",
+     "iopub.status.busy": "2025-07-11T02:24:14.379322Z",
+     "iopub.status.idle": "2025-07-11T02:24:14.382394Z",
+     "shell.execute_reply": "2025-07-11T02:24:14.381923Z",
+     "shell.execute_reply.started": "2025-07-11T02:24:14.379710Z"
+    }
+   },
+   "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 \"/warc/\" in x])"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "3",
+   "metadata": {
+    "editable": true,
+    "execution": {
+     "iopub.execute_input": "2025-07-11T02:24:14.383374Z",
+     "iopub.status.busy": "2025-07-11T02:24:14.383061Z",
+     "iopub.status.idle": "2025-07-11T02:24:14.397134Z",
+     "shell.execute_reply": "2025-07-11T02:24:14.396661Z",
+     "shell.execute_reply.started": "2025-07-11T02:24:14.383359Z"
+    },
+    "slideshow": {
+     "slide_type": ""
+    },
+    "tags": []
+   },
+   "outputs": [],
+   "source": [
+    "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",
+    "# 异常日志\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(row_iter):\n",
+    "    seen = set()\n",
+    "    \n",
+    "    # 初始化错误日志写入器\n",
+    "    s3_doc_writer = get_s3_doctor(\"dedup_fir\")\n",
+    "    error_info = None          # 错误信息初始化\n",
+    "    \n",
+    "    for zz in row_iter:\n",
+    "        try:\n",
+    "            # 读取文件并处理\n",
+    "            try:\n",
+    "                detail_datas = json_loads(zz.value)\n",
+    "                layout_id = detail_datas.get(\"layout_id\", \"\")\n",
+    "                sub_path = detail_datas.get(\"sub_path\", \"\").split('/')[-1]\n",
+    "                layout = layout_id.split(\"_\")[-1]\n",
+    "                if int(layout) < 0 :\n",
+    "                    continue\n",
+    "                # 安全地获取字段，提供默认值\n",
+    "                html_content = detail_datas.get(\"main_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\": sub_path,\n",
+    "                        \"hash_html\": hash_html,\n",
+    "                        \"track_id\": track_id,\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",
+    "                    \"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",
+    "                \"timestamp\": datetime.now().isoformat()\n",
+    "            }\n",
+    "            s3_doc_writer.write(error_info)\n",
+    "            continue\n",
+    "    \n",
+    "    if error_info:\n",
+    "        s3_doc_writer.flush()\n"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "4",
+   "metadata": {
+    "execution": {
+     "iopub.execute_input": "2025-07-11T02:24:14.397836Z",
+     "iopub.status.busy": "2025-07-11T02:24:14.397698Z",
+     "iopub.status.idle": "2025-07-11T02:24:45.430826Z",
+     "shell.execute_reply": "2025-07-11T02:24:45.429906Z",
+     "shell.execute_reply.started": "2025-07-11T02:24:14.397823Z"
+    }
+   },
+   "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",
+    "input_df = read_any_path(spark, \",\".join(input_path), config)\n",
+    "dump_html_df = input_df.rdd.mapPartitions(parse_path_to_html).toDF()"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "id": "5",
+   "metadata": {},
+   "source": [
+    "# 写出s3"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "6",
+   "metadata": {
+    "execution": {
+     "iopub.execute_input": "2025-07-11T02:24:52.435257Z",
+     "iopub.status.busy": "2025-07-11T02:24:52.434949Z",
+     "iopub.status.idle": "2025-07-11T02:26:00.233117Z",
+     "shell.execute_reply": "2025-07-11T02:26:00.232529Z",
+     "shell.execute_reply.started": "2025-07-11T02:24:52.435240Z"
+    }
+   },
+   "outputs": [],
+   "source": [
+    "config[\"skip_output_version\"] = True\n",
+    "config[\"output_compression\"] = \"gz\"\n",
+    "write_any_path(dump_html_df, output_path, config)"
+   ]
+  }
+ ],
+ "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
+}