深入解析Python中HTML/XML实体处理的完整指南

import html
import xml.sax.saxutils

# html实体处理
text = "<div>hello & world</div>"
escaped = html.escape(text)  # "&lt;div&gt;hello &amp; world&lt;/div&gt;"
unescaped = html.unescape(escaped)  # 恢复原文本

# xml实体处理
xml_text = xml.sax.saxutils.escape(text)  # xml转义
xml_original = xml.sax.saxutils.unescape(xml_text)  # xml反转义

from html import escape, unescape

# 基本转义
print(escape("10 > 5 & 3 < 8"))  # "10 &gt; 5 &amp; 3 &lt; 8"

# 自定义转义规则
def custom_escape(text):
    """只转义尖括号"""
    return text.replace("<", "&lt;").replace(">", "&gt;")

# 处理不完整实体
def safe_unescape(text):
    """安全反转义，处理无效实体"""
    try:
        return unescape(text)
    except exception:
        # 替换无效实体
        return re.sub(r"&(\w+);", "[invalid_entity]", text)

# 测试
broken_html = "&lt;div&gt;invalid &xyz; entity&lt;/div&gt;"
print(safe_unescape(broken_html))  # "<div>invalid [invalid_entity] entity</div>"

import xml.sax.saxutils as saxutils

# 基本转义
xml_safe = saxutils.escape("""<message> "hello" & 'world' </message>""")
# "&lt;message&gt; &quot;hello&quot; &amp; &apos;world&apos; &lt;/message&gt;"

# 自定义实体映射
custom_entities = {
    '"': "&quot;",
    "'": "&apos;",
    "<": "&lt;",
    ">": "&gt;",
    "&": "&amp;",
    "©": "&copyright;"  # 自定义实体
}

def custom_xml_escape(text):
    """自定义xml转义"""
    return "".join(custom_entities.get(c, c) for c in text)

# 使用示例
print(custom_xml_escape("© 2024 my company"))
# "&copyright; 2024 my company"

import re
from html.entities import html5

# 扩展html5实体字典
html5_extended = html5.copy()
html5_extended["myentity"] = "\u25a0"  # 添加自定义实体

def extended_unescape(text):
    """支持自定义实体的反转义"""
    def replace_entity(match):
        entity = match.group(1)
        if entity in html5_extended:
            return html5_extended[entity]
        elif entity.startswith("#"):
            try:
                if entity.startswith("#x"):
                    return chr(int(entity[2:], 16))
                else:
                    return chr(int(entity[1:]))
            except (valueerror, overflowerror):
                return match.group(0)
        else:
            return match.group(0)
    
    return re.sub(r"&(\w+);", replace_entity, text)

# 测试
custom_text = "&myentity; custom □"
print(extended_unescape(custom_text))  # "■ custom □"

from html.parser import htmlparser

class entityawareparser(htmlparser):
    """实体感知html解析器"""
    def __init__(self):
        super().__init__()
        self.result = []
    
    def handle_starttag(self, tag, attrs):
        self.result.append(f"<{tag}>")
    
    def handle_endtag(self, tag):
        self.result.append(f"</{tag}>")
    
    def handle_data(self, data):
        # 保留实体不解析
        self.result.append(data)
    
    def handle_entityref(self, name):
        self.result.append(f"&{name};")
    
    def handle_charref(self, name):
        self.result.append(f"&#{name};")
    
    def get_result(self):
        return "".join(self.result)

# 使用示例
parser = entityawareparser()
parser.feed("<div>hello &nbsp; world &lt;3</div>")
print(parser.get_result())  # "<div>hello &nbsp; world &lt;3</div>"

def safe_html_render(text):
    """安全html渲染"""
    # 基础转义
    safe_text = html.escape(text)
    
    # 允许安全标签白名单
    allowed_tags = {"b", "i", "u", "p", "br"}
    allowed_attrs = {"class", "style"}
    
    # 使用安全解析器
    from bs4 import beautifulsoup
    soup = beautifulsoup(safe_text, "html.parser")
    
    # 清理不安全的标签和属性
    for tag in soup.find_all(true):
        if tag.name not in allowed_tags:
            tag.unwrap()  # 移除标签保留内容
        else:
            # 清理属性
            attrs = dict(tag.attrs)
            for attr in list(attrs.keys()):
                if attr not in allowed_attrs:
                    del tag.attrs[attr]
    
    return str(soup)

# 测试
user_input = '<script>alert("xss")</script><b>safe</b> <img src=x onerror=alert(1)>'
print(safe_html_render(user_input))  # "<b>safe</b>"

from defusedxml.elementtree import parse

def safe_xml_parse(xml_data):
    """安全的xml解析，防御xxe攻击"""
    # 禁用外部实体
    parser = et.xmlparser()
    parser.entity["external"] = none
    
    try:
        # 使用defusedxml
        tree = parse(bytesio(xml_data), parser=parser)
        return tree.getroot()
    except et.parseerror as e:
        raise securityerror("invalid xml format") from e

# 替代方案：使用lxml安全配置
from lxml import etree

def safe_lxml_parse(xml_data):
    parser = etree.xmlparser(resolve_entities=false, no_network=true)
    return etree.fromstring(xml_data, parser=parser)

_escape_table = {
    ord('<'): "&lt;",
    ord('>'): "&gt;",
    ord('&'): "&amp;",
    ord('"'): "&quot;",
    ord("'"): "&apos;"
}

def fast_html_escape(text):
    """高性能html转义"""
    return text.translate(_escape_table)

# 性能对比测试
import timeit

text = "<div>" * 10000
t1 = timeit.timeit(lambda: html.escape(text), number=100)
t2 = timeit.timeit(lambda: fast_html_escape(text), number=100)

print(f"标准库: {t1:.4f}秒, 自定义: {t2:.4f}秒")

def stream_entity_processing(input_file, output_file):
    """大文件流式实体处理"""
    with open(input_file, "r", encoding="utf-8") as fin:
        with open(output_file, "w", encoding="utf-8") as fout:
            while chunk := fin.read(4096):
                # 处理实体
                processed = html.escape(chunk)
                fout.write(processed)

# xml实体流式处理
class xmlstreamprocessor:
    def __init__(self):
        self.buffer = ""
    
    def process_chunk(self, chunk):
        self.buffer += chunk
        while "&" in self.buffer and ";" in self.buffer:
            # 查找实体边界
            start = self.buffer.index("&")
            end = self.buffer.index(";", start) + 1
            
            # 提取并处理实体
            entity = self.buffer[start:end]
            processed = self.process_entity(entity)
            
            # 更新缓冲区
            self.buffer = self.buffer[:start] + processed + self.buffer[end:]
        
        # 返回安全文本
        safe_text = self.buffer
        self.buffer = ""
        return safe_text
    
    def process_entity(self, entity):
        """处理单个实体"""
        if entity in {"<", ">", "&", """, "'"}:
            return entity  # 保留基本实体
        elif entity.startswith("&#"):
            return entity  # 保留数字实体
        else:
            return "[filtered]"  # 过滤其他实体

# 使用示例
processor = xmlstreamprocessor()
with open("large.xml") as f:
    while chunk := f.read(1024):
        safe_chunk = processor.process_chunk(chunk)
        # 写入安全输出

class entitycleaningpipeline:
    """爬虫实体清洗管道"""
    def __init__(self):
        self.entity_pattern = re.compile(r"&(\w+);")
        self.valid_entities = {"lt", "gt", "amp", "quot", "apos", "nbsp"}
    
    def process_item(self, item):
        """清洗实体"""
        if "html_content" in item:
            item["html_content"] = self.clean_html(item["html_content"])
        if "text_content" in item:
            item["text_content"] = self.clean_text(item["text_content"])
        return item
    
    def clean_html(self, html):
        """清理html中的实体"""
        # 保留基本实体，其他转为unicode
        return self.entity_pattern.sub(self.replace_entity, html)
    
    def clean_text(self, text):
        """清理纯文本中的实体"""
        # 所有实体转为实际字符
        return html.unescape(text)
    
    def replace_entity(self, match):
        """实体替换逻辑"""
        entity = match.group(1)
        if entity in self.valid_entities:
            return f"&{entity};"  # 保留有效实体
        else:
            try:
                # 尝试转换命名实体
                return html.entities.html5.get(entity, f"&{entity};")
            except keyerror:
                return "[invalid_entity]"

# 在scrapy中使用
class myspider(scrapy.spider):
    # ...
    pipeline = entitycleaningpipeline()
    
    def parse(self, response):
        item = {
            "html_content": response.body.decode("utf-8"),
            "text_content": response.text
        }
        yield self.pipeline.process_item(item)

from flask import flask, jsonify, request
import html

app = flask(__name__)

@app.route("/api/process", methods=["post"])
def process_text():
    """api文本处理端点"""
    data = request.json
    text = data.get("text", "")
    
    # 安全处理选项
    mode = data.get("mode", "escape")
    
    if mode == "escape":
        result = html.escape(text)
    elif mode == "unescape":
        result = html.unescape(text)
    elif mode == "clean":
        # 自定义清理：只保留字母数字和基本标点
        cleaned = re.sub(r"[^\w\s.,!?;:]", "", html.unescape(text))
        result = cleaned
    else:
        return jsonify({"error": "invalid mode"}), 400
    
    return jsonify({"result": result})

# 测试
# curl -x post -h "content-type: application/json" -d '{"text":"hello <world>", "mode":"unescape"}' http://localhost:5000/api/process
# {"result": "hello <world>"}

# 所有用户输入必须转义
user_input = request.form["comment"]
safe_comment = html.escape(user_input)

def escape_for_context(text, context):
    if context == "html":
        return html.escape(text)
    elif context == "xml":
        return saxutils.escape(text)
    elif context == "js":
        return json.dumps(text)[1:-1]  # js字符串转义
    else:
        return text

# 只允许白名单实体
allowed_entities = {"lt", "gt", "amp", "quot", "apos"}
cleaned_text = re.sub(
    r"&(?!(" + "|".join(allowed_entities) + r");)\w+;", 
    "", 
    text
)

# 禁用外部实体
parser = et.xmlparser()
parser.entity["external"] = none
tree = et.parse("data.xml", parser=parser)

# 预编译实体映射
_escape_map = str.maketrans({
    "<": "&lt;",
    ">": "&gt;",
    "&": "&amp;",
    '"': "&quot;",
    "'": "&apos;"
})

def fast_escape(text):
    return text.translate(_escape_map)

import unittest

class testentityhandling(unittest.testcase):
    def test_html_escape(self):
        self.assertequal(html.escape("<div>"), "&lt;div&gt;")
    
    def test_xss_protection(self):
        input = "<script>alert('xss')</script>"
        safe = safe_html_render(input)
        self.assertnotin("<script>", safe)
    
    def test_xxe_protection(self):
        malicious_xml = """
        <!doctype root [
            <!entity xxe system "file:///etc/passwd">
        ]>
        <root>&xxe;</root>
        """
        with self.assertraises(securityerror):
            safe_xml_parse(malicious_xml)