舆情追踪修复

aidso_geo/models/fix_encoding.py

+# cp1252 特殊字符码点集合（0x80-0x9F 区间映射到的 Unicode 字符）
+from asyncio import as_completed
+from concurrent.futures import ThreadPoolExecutor
+from loguru import logger
+from aidso_geo.models.process import platform_process
+from aidso_geo.models.yuanbao_data_process import yuanbao_process_original_data
+from aidso_geo.utils import bh_utils
+import requests
+import json
+from aidso_geo.utils.tos_utils import get_string_from_tos, put_string_to_tos, check_file_in_tos
+
+_CP1252_SPECIAL = {
+    0x20AC, 0x201A, 0x0192, 0x201E, 0x2026, 0x2020, 0x2021, 0x02C6,
+    0x2030, 0x0160, 0x2039, 0x0152, 0x017D, 0x2018, 0x2019, 0x201C,
+    0x201D, 0x2022, 0x2013, 0x2014, 0x02DC, 0x2122, 0x0161, 0x203A,
+    0x0153, 0x017E, 0x0178,
+    0x81, 0x8D, 0x8F, 0x90, 0x9D,
+}
+
+_LATIN_EXT_RANGE = range(0x00C0, 0x0100)
+
+
+_CP1252_REVERSE = {
+    0x20AC: 0x80, 0x201A: 0x82, 0x0192: 0x83, 0x201E: 0x84,
+    0x2026: 0x85, 0x2020: 0x86, 0x2021: 0x87, 0x02C6: 0x88,
+    0x2030: 0x89, 0x0160: 0x8A, 0x2039: 0x8B, 0x0152: 0x8C,
+    0x017D: 0x8E, 0x2018: 0x91, 0x2019: 0x92, 0x201C: 0x93,
+    0x201D: 0x94, 0x2022: 0x95, 0x2013: 0x96, 0x2014: 0x97,
+    0x02DC: 0x98, 0x2122: 0x99, 0x0161: 0x9A, 0x203A: 0x9B,
+    0x0153: 0x9C, 0x017E: 0x9E, 0x0178: 0x9F,
+}
+
+for _p in (0x81, 0x8D, 0x8F, 0x90, 0x9D):
+    _CP1252_REVERSE[_p] = _p
+
+
+def is_mojibake(text, sample_size=200000, threshold=0.08):
+    """
+    检测文本是否为 UTF-8 双重编码乱码。
+    """
+    if not text:
+        return False
+
+    sample = text[:sample_size]
+    total = len(sample)
+    if total == 0:
+        return False
+
+    latin_ext_count = 0
+    cp1252_special_count = 0
+    chinese_count = 0
+
+    for c in sample:
+        code = ord(c)
+
+        if code in _LATIN_EXT_RANGE:
+            latin_ext_count += 1
+        elif code in _CP1252_SPECIAL:
+            cp1252_special_count += 1
+        elif 0x4E00 <= code <= 0x9FFF:
+            chinese_count += 1
+
+    # 如果已经有比较多正常中文，通常不认为是乱码
+    if chinese_count / total > 0.99:
+        return False
+
+    utf8_lead_chars = set("äåæçèé")
+    trigram_hits = 0
+
+    for i in range(len(sample) - 2):
+        if sample[i] in utf8_lead_chars:
+            b1 = ord(sample[i + 1])
+            b2 = ord(sample[i + 2])
+
+            b1_is_cont = (0x80 <= b1 <= 0xBF) or (b1 in _CP1252_SPECIAL)
+            b2_is_cont = (0x80 <= b2 <= 0xBF) or (b2 in _CP1252_SPECIAL)
+
+            if b1_is_cont and b2_is_cont:
+                trigram_hits += 1
+
+    if trigram_hits >= 2:
+        return True
+
+    latin_ratio = latin_ext_count / total
+    cp1252_ratio = cp1252_special_count / total
+    combined = latin_ratio + cp1252_ratio
+
+    if combined > 0.25:
+        return True
+
+    return False
+
+
+def fix_mojibake_text(text):
+    """
+    修复 UTF-8 被 cp1252/Latin-1 误读后产生的 mojibake 文本。
+
+    例如：
+        正在æœç´¢ -> 正在搜索
+        中文测试 -> 中文测试
+    """
+    if not text:
+        return text
+
+    raw_bytes = bytearray()
+
+    for c in text:
+        code = ord(c)
+
+        if code < 0x100:
+            raw_bytes.append(code)
+        elif code in _CP1252_REVERSE:
+            raw_bytes.append(_CP1252_REVERSE[code])
+        else:
+            # 普通字符保持原样写回 UTF-8
+            raw_bytes.extend(c.encode("utf-8"))
+
+    return raw_bytes.decode("utf-8", errors="replace")
+
+
+def get_mojibake_sample(text):
+    """
+    提取一段疑似乱码样例。
+    """
+    import re
+
+    match = re.search(
+        r'[äåæçèé][\x80-\xbf\u0080-\u00bf\u0152\u0153\u0160\u0161]{2,20}',
+        text
+    )
+    return match.group(0) if match else text[:100]
+
+
+def is_mojibake_tos_file(file_path):
+    """
+    从 TOS 读取 original.text 内容，并检测/修复 UTF-8 双重编码乱码。
+
+    Args:
+        file_path: TOS 文件路径
+
+    Returns:
+        dict: {
+            "is_mojibake": bool,
+            "encoding": str,
+            "sample": str,
+            "sample_fixed": str,
+            "fixed_content": str,
+            "original_content": str,
+        }
+    """
+
+    try:
+        original_content = get_string_from_tos(file_path)
+    except Exception as e:
+        return {
+            "is_mojibake": False,
+            "encoding": f"TOS读取失败: {e}",
+            "sample": "",
+            "sample_fixed": "",
+            "fixed_content": "",
+            "original_content": "",
+        }
+
+    if not original_content:
+        return {
+            "is_mojibake": False,
+            "encoding": "TOS内容为空",
+            "sample": "",
+            "sample_fixed": "",
+            "fixed_content": "",
+            "original_content": "",
+        }
+
+    result = is_mojibake(original_content)
+
+    if not result:
+        return {
+            "is_mojibake": False,
+            "encoding": "UTF-8（正常）",
+            "sample": "",
+            "sample_fixed": "",
+            "fixed_content": original_content,
+            "original_content": original_content,
+        }
+
+    sample = get_mojibake_sample(original_content)
+    sample_fixed = fix_mojibake_text(sample)
+    fixed_content = fix_mojibake_text(original_content)
+
+    return {
+        "is_mojibake": True,
+        "encoding": "UTF-8 双重编码（经 cp1252/Latin-1 误读）",
+        "sample": sample,
+        "sample_fixed": sample_fixed,
+        "fixed_content": fixed_content,
+        "original_content": original_content,
+    }
+
+
+def fix_req(req_ids):
+
+
+    url = "https://api.aidso.com/openapi/ywt/reReq"
+
+    # payload = json.dumps([
+    #     "35506100-213c-4264-bcc4-67d3449431ac"
+    # ])
+    headers = {
+        'Content-Type': 'application/json'
+    }
+
+    response = requests.request("POST", url, headers=headers,json=req_ids )
+
+    print(response.text)
+
+
+if __name__ == "__main__":
+    query =bh_utils.query_data("select taskId,reqId from geo_commit_task where platform = 'TXYB' and insertime > 1780502400")
+    # query =bh_utils.query_data("select taskId,reqId from geo_commit_task where platform = 'TXYB' and taskId = 'd7984dca-3c17-40ac-89d5-ece0ba94a29c'")
+    # query =[1]
+    req_id_list = []
+    for q in query:
+        taskId = q.get('taskId')
+        reqId = q.get('reqId')
+        file_path = f"geo/{taskId}/TXYB/original.text"
+        result = is_mojibake_tos_file(file_path)
+        if result.get('is_mojibake'):
+            print(reqId)
+            req_id_list.append(reqId)
+            fixed_content = result["fixed_content"]
+            put_string_to_tos(f"geo/{taskId}/TXYB/original.text", fixed_content)
+            yuanbao_process_original_data(f"geo/{taskId}/TXYB/original.text")
+    print(req_id_list)
+    fix_req(req_id_list)
+
+
+    req_id_sql = ",".join([f"'{req_id}'" for req_id in req_id_list])
+    # # # print(len(req_ids))
+    query_sql = f"select * from geo_commit_task where reqId in ({req_id_sql})"
+    #
+    data_list = bh_utils.query_data(query_sql)
+    print(data_list)
+    def handle_item(i):
+        if i.get('comWordsMap'):
+            i['comWordsMap'] = json.loads(i.get('comWordsMap'))
+        if i.get('brandWords'):
+            i['brandWords'] = json.loads(i.get('brandWords'))
+        if i.get('comWords'):
+            i['comWords'] = json.loads(i.get('comWords'))
+        if i.get('keywords'):
+            i['keywords'] = json.loads(i.get('keywords'))
+        if i.get('productWordsMap'):
+            i['productWordsMap'] = json.loads(i.get('productWordsMap'))
+        type_t = i.get('type')
+
+        # return task_send_queue(i,type_t)
+        return platform_process(i)
+
+    if data_list:
+        with ThreadPoolExecutor(max_workers=30) as executor:
+            futures = [executor.submit(handle_item, i) for i in data_list]
+
+            for future in as_completed(futures):
+                try:
+                    future.result()
+                except Exception as e:
+                    logger.exception(f"platform_process 执行异常: {e}")
\ No newline at end of file