spb/wp14_uqo

1#!/usr/bin/env python32# -*- coding: utf-8 -*-3"""4Extraction des specs de propriété via GPT-4o-mini, à partir des descriptions.5Parallèle, résumable (cache specs.jsonl), avec retry/backoff. Aucune dépendance.6"""7import json, os, time, threading, urllib.request, urllib.error8from concurrent.futures import ThreadPoolExecutor, as_completed910SRC = "data/processed/maisons_avec_prix.json"11OUT = "data/processed/specs.jsonl"12WORKERS = int(os.environ.get("WORKERS", "24"))13NSHARDS = int(os.environ.get("NSHARDS", "1"))14SHARD = int(os.environ.get("SHARD", "0"))15MODEL = "gpt-4o-mini"1617# clé18for line in open(".env", encoding="utf-8"):19    if line.startswith("OPENAI_API_KEY="):20        os.environ["OPENAI_API_KEY"] = line.split("=", 1)[1].strip()21KEY = os.environ["OPENAI_API_KEY"]2223SCHEMA = {24    "type": "object", "additionalProperties": False,25    "properties": {26        "condition": {"type": "string", "enum": ["a_renover","original","bon","renove","refait_neuf","neuf","inconnu"]},27        "turnkey": {"type": "boolean"},28        "recently_renovated": {"type": "boolean"},29        "flooring": {"type": "array", "items": {"type": "string", "enum": ["bois_franc","ceramique","plancher_flottant","tapis","vinyle","beton","autre"]}},30        "countertop": {"type": "string", "enum": ["granit","quartz","marbre","stratifie","bois","autre","inconnu"]},31        "heating": {"type": "array", "items": {"type": "string", "enum": ["electrique","plinthes","thermopompe","bois","gaz","mazout","radiant","geothermie","autre"]}},32        "garage": {"type": "boolean"},33        "pool": {"type": "boolean"},34        "waterfront": {"type": "boolean"},35        "view": {"type": "string", "enum": ["eau","montagne","ville","aucune","inconnu"]},36        "finished_basement": {"type": "boolean"},37        "fireplace": {"type": "boolean"},38        "central_ac": {"type": "boolean"},39        "new_construction": {"type": "boolean"},40        "income_property": {"type": "boolean"},41        "renovated_kitchen": {"type": "boolean"},42        "renovated_bathroom": {"type": "boolean"},43        "large_lot": {"type": "boolean"},44        "quiet_area": {"type": "boolean"},45        "highlights": {"type": "array", "items": {"type": "string"}}46    },47    "required": ["condition","turnkey","recently_renovated","flooring","countertop",48                 "heating","garage","pool","waterfront","view","finished_basement",49                 "fireplace","central_ac","new_construction","income_property",50                 "renovated_kitchen","renovated_bathroom","large_lot","quiet_area",51                 "highlights"]52}53SYS = ("Tu es un extracteur d'attributs immobiliers. À partir de la description "54       "d'une propriété (français ou anglais), renvoie les specs au format JSON "55       "strict. Base-toi UNIQUEMENT sur le texte ; si une info est absente, mets "56       "false / 'inconnu' / [] selon le champ. highlights = 3 à 6 mots-clés de vente.")5758def call(desc, tries=5):59    body = {"model": MODEL, "temperature": 0,60            "messages": [{"role": "system", "content": SYS},61                         {"role": "user", "content": desc[:4000]}],62            "response_format": {"type": "json_schema", "json_schema":63                {"name": "specs", "strict": True, "schema": SCHEMA}}}64    data = json.dumps(body).encode()65    for t in range(tries):66        try:67            req = urllib.request.Request(68                "https://api.openai.com/v1/chat/completions", data=data,69                headers={"Authorization": f"Bearer {KEY}",70                         "Content-Type": "application/json"})71            with urllib.request.urlopen(req, timeout=90) as r:72                out = json.load(r)73            return json.loads(out["choices"][0]["message"]["content"])74        except urllib.error.HTTPError as e:75            if e.code in (429, 500, 502, 503, 504) and t < tries - 1:76                time.sleep(2 ** t + 0.5); continue77            raise78        except (urllib.error.URLError, TimeoutError):79            if t < tries - 1:80                time.sleep(2 ** t + 0.5); continue81            raise82    return None8384# charge données + cache85data = json.load(open(SRC, encoding="utf-8"))86done = set()87if os.path.exists(OUT):88    for line in open(OUT, encoding="utf-8"):89        try:90            done.add(json.loads(line)["uid"])91        except Exception:92            pass9394todo = []95for d in data:96    uid = f"{d.get('source')}:{d.get('listingId')}"97    desc = d.get("description")98    if uid in done or not desc:99        continue100    todo.append((uid, desc))101102# sharding : partition disjointe et complète par index (déterministe)103if NSHARDS > 1:104    todo = [t for i, t in enumerate(todo) if i % NSHARDS == SHARD]105106print(f"[shard {SHARD}/{NSHARDS}, workers={WORKERS}] Total={len(data)}  "107      f"déjà fait={len(done)}  à traiter (ce shard)={len(todo)}", flush=True)108109lock = threading.Lock()110counter = {"n": 0, "err": 0}111fh = open(OUT, "a", encoding="utf-8")112113def work(item):114    uid, desc = item115    try:116        specs = call(desc)117        rec = {"uid": uid, "specs": specs}118    except Exception as e:119        rec = {"uid": uid, "error": str(e)[:200]}120    with lock:121        fh.write(json.dumps(rec, ensure_ascii=False) + "\n")122        fh.flush()123        counter["n"] += 1124        if "error" in rec:125            counter["err"] += 1126        if counter["n"] % 250 == 0:127            print(f"  {counter['n']}/{len(todo)}  (err={counter['err']})",128                  flush=True)129130t0 = time.time()131with ThreadPoolExecutor(max_workers=WORKERS) as ex:132    list(ex.map(work, todo))133fh.close()134print(f"Terminé : {counter['n']} traités, {counter['err']} erreurs, "135      f"{time.time()-t0:.0f}s", flush=True)136