import html
import xml.sax.saxutils

# HTML實(shí)體處理
text = "<div>Hello & World</div>"
escaped = html.escape(text)  # "&lt;div&gt;Hello &amp; World&lt;/div&gt;"
unescaped = html.unescape(escaped)  # 恢復(fù)原文本

# XML實(shí)體處理
xml_text = xml.sax.saxutils.escape(text)  # XML轉(zhuǎn)義
xml_original = xml.sax.saxutils.unescape(xml_text)  # XML反轉(zhuǎn)義

from html import escape, unescape

# 基本轉(zhuǎn)義
print(escape("10 > 5 & 3 < 8"))  # "10 &gt; 5 &amp; 3 &lt; 8"

# 自定義轉(zhuǎn)義規(guī)則
def custom_escape(text):
    """只轉(zhuǎn)義尖括號(hào)"""
    return text.replace("<", "&lt;").replace(">", "&gt;")

# 處理不完整實(shí)體
def safe_unescape(text):
    """安全反轉(zhuǎn)義，處理無(wú)效實(shí)體"""
    try:
        return unescape(text)
    except Exception:
        # 替換無(wú)效實(shí)體
        return re.sub(r"&(\w+);", "[INVALID_ENTITY]", text)

# 測(cè)試
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

# 基本轉(zhuǎn)義
xml_safe = saxutils.escape("""<message> "Hello" & 'World' </message>""")
# "&lt;message&gt; &quot;Hello&quot; &amp; &apos;World&apos; &lt;/message&gt;"

# 自定義實(shí)體映射
custom_entities = {
    '"': "&quot;",
    "'": "&apos;",
    "<": "&lt;",
    ">": "&gt;",
    "&": "&amp;",
    "?": "&copyright;"  # 自定義實(shí)體
}

def custom_xml_escape(text):
    """自定義XML轉(zhuǎn)義"""
    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

# 擴(kuò)展HTML5實(shí)體字典
html5_extended = html5.copy()
html5_extended["myentity"] = "\u25A0"  # 添加自定義實(shí)體

def extended_unescape(text):
    """支持自定義實(shí)體的反轉(zhuǎn)義"""
    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)

# 測(cè)試
custom_text = "&myentity; Custom □"
print(extended_unescape(custom_text))  # "■ Custom □"

from html.parser import HTMLParser

class EntityAwareParser(HTMLParser):
    """實(shí)體感知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):
        # 保留實(shí)體不解析
        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渲染"""
    # 基礎(chǔ)轉(zhuǎn)義
    safe_text = html.escape(text)
    
    # 允許安全標(biāo)簽白名單
    allowed_tags = {"b", "i", "u", "p", "br"}
    allowed_attrs = {"class", "style"}
    
    # 使用安全解析器
    from bs4 import BeautifulSoup
    soup = BeautifulSoup(safe_text, "html.parser")
    
    # 清理不安全的標(biāo)簽和屬性
    for tag in soup.find_all(True):
        if tag.name not in allowed_tags:
            tag.unwrap()  # 移除標(biāo)簽保留內(nèi)容
        else:
            # 清理屬性
            attrs = dict(tag.attrs)
            for attr in list(attrs.keys()):
                if attr not in allowed_attrs:
                    del tag.attrs[attr]
    
    return str(soup)

# 測(cè)試
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攻擊"""
    # 禁用外部實(shí)體
    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轉(zhuǎn)義"""
    return text.translate(_escape_table)

# 性能對(duì)比測(cè)試
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"標(biāo)準(zhǔn)庫(kù): {t1:.4f}秒, 自定義: {t2:.4f}秒")

def stream_entity_processing(input_file, output_file):
    """大文件流式實(shí)體處理"""
    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):
                # 處理實(shí)體
                processed = html.escape(chunk)
                fout.write(processed)

# XML實(shí)體流式處理
class XMLStreamProcessor:
    def __init__(self):
        self.buffer = ""
    
    def process_chunk(self, chunk):
        self.buffer += chunk
        while "&" in self.buffer and ";" in self.buffer:
            # 查找實(shí)體邊界
            start = self.buffer.index("&")
            end = self.buffer.index(";", start) + 1
            
            # 提取并處理實(shí)體
            entity = self.buffer[start:end]
            processed = self.process_entity(entity)
            
            # 更新緩沖區(qū)
            self.buffer = self.buffer[:start] + processed + self.buffer[end:]
        
        # 返回安全文本
        safe_text = self.buffer
        self.buffer = ""
        return safe_text
    
    def process_entity(self, entity):
        """處理單個(gè)實(shí)體"""
        if entity in {"<", ">", "&", """, "'"}:
            return entity  # 保留基本實(shí)體
        elif entity.startswith("&#"):
            return entity  # 保留數(shù)字實(shí)體
        else:
            return "[FILTERED]"  # 過(guò)濾其他實(shí)體

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

class EntityCleaningPipeline:
    """爬蟲實(shí)體清洗管道"""
    def __init__(self):
        self.entity_pattern = re.compile(r"&(\w+);")
        self.valid_entities = {"lt", "gt", "amp", "quot", "apos", "nbsp"}
    
    def process_item(self, item):
        """清洗實(shí)體"""
        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中的實(shí)體"""
        # 保留基本實(shí)體，其他轉(zhuǎn)為Unicode
        return self.entity_pattern.sub(self.replace_entity, html)
    
    def clean_text(self, text):
        """清理純文本中的實(shí)體"""
        # 所有實(shí)體轉(zhuǎn)為實(shí)際字符
        return html.unescape(text)
    
    def replace_entity(self, match):
        """實(shí)體替換邏輯"""
        entity = match.group(1)
        if entity in self.valid_entities:
            return f"&{entity};"  # 保留有效實(shí)體
        else:
            try:
                # 嘗試轉(zhuǎn)換命名實(shí)體
                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文本處理端點(diǎn)"""
    data = request.json
    text = data.get("text", "")
    
    # 安全處理選項(xiàng)
    mode = data.get("mode", "escape")
    
    if mode == "escape":
        result = html.escape(text)
    elif mode == "unescape":
        result = html.unescape(text)
    elif mode == "clean":
        # 自定義清理：只保留字母數(shù)字和基本標(biāo)點(diǎn)
        cleaned = re.sub(r"[^\w\s.,!?;:]", "", html.unescape(text))
        result = cleaned
    else:
        return jsonify({"error": "Invalid mode"}), 400
    
    return jsonify({"result": result})

# 測(cè)試
# curl -X POST -H "Content-Type: application/json" -d '{"text":"Hello <World>", "mode":"unescape"}' http://localhost:5000/api/process
# {"result": "Hello <World>"}

# 所有用戶輸入必須轉(zhuǎn)義
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字符串轉(zhuǎn)義
    else:
        return text

# 只允許白名單實(shí)體
ALLOWED_ENTITIES = {"lt", "gt", "amp", "quot", "apos"}
cleaned_text = re.sub(
    r"&(?!(" + "|".join(ALLOWED_ENTITIES) + r");)\w+;", 
    "", 
    text
)

# 禁用外部實(shí)體
parser = ET.XMLParser()
parser.entity["external"] = None
tree = ET.parse("data.xml", parser=parser)

# 預(yù)編譯實(shí)體映射
_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)