spb/trouve-ka

1# Trouve-KA — réparation du mojibake dans l'index2# Author: Simon-Pierre Boucher3# Contact: contact@spboucher.ai45"""Répare les documents indexés avec du double encodage UTF-8→cp12526(« é » → « é ») produit avant le fix de decode_html.78Pour chaque doc contenant des signatures : répare title/description/body/headings9via fix_mojibake (réparation strictement non destructive), réécrit le doc et10efface son embedding (recalculé par le backfill sur le texte propre).11Synchronise aussi title/description dans Postgres (table documents).1213Usage : python scripts/fix-mojibake/repair.py14"""1516import asyncio1718from opensearchpy import AsyncOpenSearch19from opensearchpy.helpers import async_bulk2021from trouveka.config import get_settings22from trouveka.database import Database23from trouveka.parser import fix_mojibake2425FIELDS = ["title", "description", "body", "headings"]26QUERY = {27    "bool": {28        "should": [29            {"match_phrase": {"body": "Ã"}},30            {"match_phrase": {"title": "Ã"}},31            {"match_phrase": {"description": "Ã"}},32            {"query_string": {"query": 'description:"é" OR body:"é" OR title:"é" OR body:"â€" OR body:"è"'}},33        ],34        "minimum_should_match": 1,35    }36}373839async def main() -> None:40    s = get_settings()41    client = AsyncOpenSearch(hosts=[s.search_url], timeout=120)42    db = Database(s.database_url, pool_min=1, pool_max=3)43    await db.connect()44    repaired = scanned = 045    search_after = None46    try:47        while True:48            body = {49                "size": 200,50                "query": QUERY,51                "_source": FIELDS + ["url"],52                # Curseur stable : on balaie TOUT le stock une fois, même si53                # certains documents matchent sans être réparables.54                "sort": [{"_id": "asc"}],55            }56            if search_after is not None:57                body["search_after"] = search_after58            res = await client.search(index=s.search_index, body=body)59            hits = res["hits"]["hits"]60            if not hits:61                break62            search_after = hits[-1]["sort"]63            actions = []64            for hit in hits:65                scanned += 166                src = hit["_source"]67                changes = {}68                for f in FIELDS:69                    val = src.get(f)70                    if isinstance(val, str):71                        fixed = fix_mojibake(val)72                        if fixed != val:73                            changes[f] = fixed74                    elif isinstance(val, list):75                        fixed_list = [fix_mojibake(v) if isinstance(v, str) else v for v in val]76                        if fixed_list != val:77                            changes[f] = fixed_list78                if not changes:79                    continue  # signatures détectées mais irréparables : recrawl s'en chargera80                repaired += 181                actions.append({82                    "_op_type": "update",83                    "_index": s.search_index,84                    "_id": hit["_id"],85                    # embedding effacé : le backfill le recalcule sur le texte propre86                    "script": {"source": "ctx._source.remove('embedding')"},87                })88                actions.append({89                    "_op_type": "update",90                    "_index": s.search_index,91                    "_id": hit["_id"],92                    "doc": changes,93                })94                if "title" in changes or "description" in changes:95                    await db.pool.execute(96                        """97                        UPDATE documents SET98                            title = COALESCE($2, title),99                            description = COALESCE($3, description)100                        WHERE url_id = (SELECT id FROM urls WHERE url = $1 OR canonical_url = $1 LIMIT 1)101                        """,102                        src.get("url"),103                        changes.get("title"),104                        changes.get("description"),105                    )106            if actions:107                await async_bulk(client, actions, raise_on_error=False)108            print(f"→ scannés {scanned}, réparés {repaired}")109    finally:110        print(f"✓ terminé : {repaired} documents réparés sur {scanned} scannés")111        await db.close()112        await client.close()113114115if __name__ == "__main__":116    asyncio.run(main())117