419 lines
19 KiB
Python
419 lines
19 KiB
Python
"""Intégration IA locale (llama-cpp-python) pour assister le scraping.
|
||
|
||
Quatre capacités, combinables :
|
||
- pick_next_link : désigne, parmi les liens candidats d'une page, celui
|
||
qui mène le plus probablement à la suite de l'histoire/du chapitre —
|
||
en complément (ou remplacement, sur les pages où il échoue) de
|
||
l'heuristique regex de link_detector.py, qui reste rapide/gratuite
|
||
mais fragile face aux structures de page inhabituelles. Indique aussi
|
||
si la suite trouvée est la pagination interne du même chapitre (pages
|
||
d'un même chapitre réparties sur plusieurs URLs) ou un nouveau
|
||
chapitre — voir NextLinkResult. Sait aussi viser spécifiquement le
|
||
premier chapitre depuis une table des matières (voir from_toc).
|
||
- extract_main_content : extrait le texte narratif principal d'une page
|
||
(mode lecture), en écartant menus/pubs/commentaires — une alternative
|
||
au mode "capture de bloc au survol" quand le contenu est trop
|
||
fragmenté pour être capturé en un clic.
|
||
- estimate_chapter_count : tente de déterminer le nombre total de
|
||
chapitres d'une histoire à partir de sa page de départ, pour le mode
|
||
"0 chapitre = limite détectée automatiquement" du scraping
|
||
automatique (voir MainWindow._toggle_auto_scrape). Best-effort : ne
|
||
lève jamais d'exception, retourne None si indéterminable.
|
||
- is_table_of_contents : détecte si la page actuelle est une table des
|
||
matières/liste de chapitres plutôt que le contenu narratif d'un
|
||
chapitre — sans quoi le scraping automatique tenterait d'en "extraire
|
||
le contenu narratif" (résultat vide/aberrant capturé comme un faux
|
||
chapitre) et choisirait un lien "suivant" arbitraire parmi tous les
|
||
chapitres listés plutôt que viser le premier. Best-effort, comme
|
||
estimate_chapter_count.
|
||
|
||
Le modèle (fichier .gguf) n'est pas embarqué dans ce dépôt — l'utilisateur
|
||
fournit son propre fichier via la fenêtre de configuration IA (voir
|
||
ai_settings_dialog.py). Le chargement est paresseux (au premier appel
|
||
réel) et mis en cache : un modèle de quelques Go peut prendre plusieurs
|
||
secondes à charger, d'où l'exécution systématique en arrière-plan côté
|
||
appelant (voir ai_worker.py) pour ne jamais geler l'interface.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import re
|
||
from dataclasses import dataclass
|
||
from typing import Callable
|
||
|
||
from .ai_config import AiConfig
|
||
|
||
|
||
class LocalAiError(RuntimeError):
|
||
"""Erreur lisible, destinée à être affichée telle quelle à l'utilisateur."""
|
||
|
||
|
||
@dataclass
|
||
class LinkCandidate:
|
||
text: str
|
||
href: str
|
||
# Texte du conteneur parent immédiat du lien, quand il diffère du
|
||
# texte du lien lui-même — capte les cas où l'indice "suivant" est une
|
||
# légende/icône voisine plutôt que le texte du lien (ex. pagination
|
||
# avec une simple flèche sans texte propre). Voir ALL_LINKS_FOR_AI_JS
|
||
# dans main_window.py, qui applique la même logique que
|
||
# link_detector.py pour l'heuristique regex.
|
||
context: str = ""
|
||
|
||
|
||
@dataclass
|
||
class NextLinkResult:
|
||
"""Résultat de pick_next_link() : le lien choisi (ou None si aucun ne
|
||
convient), et — quand un lien est choisi — si l'IA l'estime être une
|
||
simple suite du MÊME chapitre (pagination interne, ex. page 2/3 d'un
|
||
chapitre réparti sur plusieurs pages web) ou un NOUVEAU chapitre.
|
||
Utilisé par le scraping automatique pour relier automatiquement les
|
||
extraits d'un même chapitre (voir MainWindow._auto_scrape_on_next_link_found
|
||
et models.group_into_chapters)."""
|
||
|
||
candidate: LinkCandidate | None
|
||
same_chapter: bool = False
|
||
|
||
|
||
# Prompt par défaut, utilisé sauf si l'utilisateur en a défini un autre
|
||
# dans AiConfig.next_link_prompt (voir ai_settings_dialog.py — un champ
|
||
# libre y permet d'affiner ce texte, ex. pour un site particulièrement
|
||
# atypique). Le format de réponse JSON attendu (index + same_chapter) est
|
||
# décrit explicitement : un prompt personnalisé doit le conserver pour que
|
||
# _parse_next_link_response() continue à fonctionner.
|
||
DEFAULT_NEXT_LINK_PROMPT = (
|
||
"Tu aides à identifier, parmi une liste de liens présents sur une page web, "
|
||
"celui qui mène le plus probablement à la SUITE de ce qui est affiché. Cela "
|
||
"inclut aussi bien un lien vers le chapitre/la partie suivante qu'un simple "
|
||
"lien de pagination vers la page suivante D'UN MÊME CHAPITRE réparti sur "
|
||
"plusieurs pages web (ex. texte '2', '›', 'Page suivante', 'Suite', ou une "
|
||
"flèche sans texte propre — dans ce cas le \"contexte_proche\" fourni pour "
|
||
"ce lien peut donner l'indice). Dans tous ces cas, choisis ce lien : c'est "
|
||
"la suite logique de la lecture.\n\n"
|
||
"Indique aussi si ce lien mène à la suite du MÊME chapitre (pagination "
|
||
"interne, pas de nouveau titre/chapitre) ou à un NOUVEAU chapitre distinct, "
|
||
"via le champ same_chapter (true/false) — utile pour savoir si les deux "
|
||
"pages doivent être fusionnées en une seule section dans le document final. "
|
||
"Si on te fournit un extrait de la fin du contenu de la page actuelle, "
|
||
"utilise-le comme indice supplémentaire pour ce choix : un texte qui "
|
||
"s'interrompt en plein milieu d'une scène/d'une phrase suggère une simple "
|
||
"pagination interne (same_chapter=true) ; un texte qui se termine "
|
||
"proprement (dénouement, dialogue conclu, \"Fin du chapitre\"...) suggère "
|
||
"plutôt un nouveau chapitre (same_chapter=false).\n\n"
|
||
"Si aucun lien ne semble mener à une suite (menu, pub, autres histoires "
|
||
"sans rapport), réponds qu'aucun ne convient.\n\n"
|
||
"Réponds STRICTEMENT avec un objet JSON sur une seule ligne, de la forme "
|
||
'{"index": <numéro du lien choisi>, "same_chapter": true|false} ou '
|
||
'{"index": null}. Aucun autre texte.'
|
||
)
|
||
|
||
# Nombre de caractères de fin de page transmis à pick_next_link — la FIN
|
||
# du contenu (pas le début) est ce qui renseigne le mieux sur une coupure
|
||
# probable en milieu de chapitre. Volontairement court : ce n'est qu'un
|
||
# indice complémentaire, pas le texte à analyser en profondeur (ça, c'est
|
||
# le rôle d'extract_main_content).
|
||
_NEXT_LINK_CONTENT_EXCERPT_CHARS = 800
|
||
|
||
_EXTRACT_SYSTEM_PROMPT = (
|
||
"Tu extrais le texte narratif principal (l'histoire/l'article) d'un texte brut "
|
||
"issu d'une page web, qui contient probablement aussi du texte hors-sujet "
|
||
"(menus, publicités, commentaires, liens de navigation, avertissements). "
|
||
"Renvoie UNIQUEMENT ce texte narratif, tel quel, sans le résumer, le traduire "
|
||
"ni le commenter, en conservant sa mise en forme en paragraphes. Ne renvoie "
|
||
"rien d'autre : pas de préambule, pas d'explication."
|
||
)
|
||
|
||
_ESTIMATE_CHAPTER_COUNT_PROMPT = (
|
||
"Tu aides à estimer le nombre TOTAL de chapitres d'une histoire, à partir du "
|
||
"titre et du texte brut de sa page de départ (page d'accueil du récit, table "
|
||
"des matières, description...). Cherche un indice EXPLICITE et fiable (ex. "
|
||
"\"Chapitre 3 sur 12\", une liste/table des matières numérotée, \"Histoire "
|
||
"terminée — 15 chapitres\", \"Chapters: 20/20\"). Ne devine JAMAIS un nombre "
|
||
"en l'absence d'un tel indice explicite dans le texte fourni — réponds alors "
|
||
"que c'est indéterminable.\n\n"
|
||
"Réponds STRICTEMENT avec un objet JSON sur une seule ligne, de la forme "
|
||
'{"count": <nombre entier total de chapitres>} ou {"count": null}. Aucun '
|
||
"autre texte."
|
||
)
|
||
|
||
_IS_TABLE_OF_CONTENTS_PROMPT = (
|
||
"Tu détermines si une page web est une TABLE DES MATIÈRES / liste de "
|
||
"chapitres d'une histoire — essentiellement une liste de titres/liens vers "
|
||
"des chapitres, avec peu ou pas de texte narratif propre — plutôt que le "
|
||
"contenu narratif d'un chapitre lui-même. Une page d'accueil de série avec "
|
||
"un résumé suivi d'une liste de chapitres compte aussi comme une table des "
|
||
"matières. Si le texte fourni raconte une histoire (dialogues, description "
|
||
"d'actions, narration suivie), ce n'est PAS une table des matières, même "
|
||
"s'il contient par ailleurs quelques liens de navigation.\n\n"
|
||
"Réponds STRICTEMENT avec un objet JSON sur une seule ligne, de la forme "
|
||
'{"is_toc": true} ou {"is_toc": false}. Aucun autre texte.'
|
||
)
|
||
|
||
|
||
class LocalAi:
|
||
"""Enveloppe autour de llama-cpp-python, avec chargement paresseux du modèle."""
|
||
|
||
def __init__(self, config: AiConfig) -> None:
|
||
self.config = config
|
||
self._llm = None # instancié à la demande, voir _ensure_loaded()
|
||
|
||
def reset(self) -> None:
|
||
"""Invalide le modèle chargé (ex. après changement de configuration)."""
|
||
self._llm = None
|
||
|
||
def is_configured(self) -> bool:
|
||
return self.config.is_configured
|
||
|
||
# ------------------------------------------------------------------ #
|
||
# Chargement du modèle
|
||
# ------------------------------------------------------------------ #
|
||
def _ensure_loaded(self):
|
||
if self._llm is not None:
|
||
return self._llm
|
||
|
||
if not self.config.model_path:
|
||
raise LocalAiError(
|
||
"Aucun modèle IA configuré. Ouvrez « ⚙️ Configurer l'IA locale » "
|
||
"et indiquez le chemin d'un fichier .gguf."
|
||
)
|
||
|
||
try:
|
||
from llama_cpp import Llama
|
||
except ImportError as exc:
|
||
raise LocalAiError(
|
||
"Le paquet llama-cpp-python n'est pas installé.\n"
|
||
"Installez-le avec : pip install llama-cpp-python"
|
||
) from exc
|
||
|
||
try:
|
||
self._llm = Llama(
|
||
model_path=self.config.model_path,
|
||
n_ctx=self.config.n_ctx,
|
||
n_threads=self.config.n_threads or None,
|
||
verbose=False,
|
||
)
|
||
except Exception as exc: # noqa: BLE001 — remonté proprement à l'UI
|
||
raise LocalAiError(f"Échec du chargement du modèle IA :\n{exc}") from exc
|
||
|
||
return self._llm
|
||
|
||
# ------------------------------------------------------------------ #
|
||
# Capacités
|
||
# ------------------------------------------------------------------ #
|
||
def pick_next_link(
|
||
self,
|
||
candidates: list[LinkCandidate],
|
||
page_title: str,
|
||
current_content: str = "",
|
||
from_toc: bool = False,
|
||
on_activity: Callable[[], None] | None = None,
|
||
) -> NextLinkResult:
|
||
"""Interroge le modèle pour désigner le lien menant à la suite,
|
||
parmi `candidates`, et s'il s'agit d'une suite du même chapitre ou
|
||
d'un nouveau chapitre. `candidate` vaut None si aucun ne convient.
|
||
|
||
`current_content` : texte de la page actuellement affichée (idéalement
|
||
déjà extrait/nettoyé, voir extract_main_content) — sans lui, le
|
||
modèle ne voit que les titres/textes des liens, ce qui est souvent
|
||
insuffisant pour juger si la page s'interrompt en plein chapitre ou
|
||
se termine proprement (voir MainWindow._auto_scrape_on_links, qui
|
||
réutilise le texte déjà extrait du dernier extrait capturé). Seule
|
||
la FIN du texte est transmise (voir _NEXT_LINK_CONTENT_EXCERPT_CHARS)
|
||
: c'est elle qui renseigne sur une coupure probable, pas le début.
|
||
|
||
`from_toc` : la page actuelle est une table des matières (voir
|
||
is_table_of_contents) — le choix ne se fait alors plus par rapport
|
||
à une "suite" d'un contenu affiché (il n'y en a pas), mais en visant
|
||
explicitement le TOUT PREMIER chapitre de la liste. `current_content`
|
||
est ignoré dans ce cas (pas de contenu narratif à juger), et le
|
||
résultat force same_chapter=False (quitter une table des matières
|
||
pour un chapitre est toujours un nouveau chapitre).
|
||
"""
|
||
if not candidates:
|
||
return NextLinkResult(candidate=None)
|
||
|
||
llm = self._ensure_loaded()
|
||
lines = []
|
||
for i, c in enumerate(candidates):
|
||
line = f"{i}. texte={c.text!r}"
|
||
if c.context and c.context != c.text:
|
||
line += f" contexte_proche={c.context!r}"
|
||
lines.append(line)
|
||
numbered = "\n".join(lines)
|
||
|
||
user_prompt = f"Titre de la page actuelle : {page_title!r}\n"
|
||
if from_toc:
|
||
user_prompt += (
|
||
"\nCette page est une TABLE DES MATIÈRES / liste de chapitres, pas le "
|
||
"contenu narratif d'un chapitre : choisis le lien qui mène au TOUT "
|
||
"PREMIER chapitre de l'histoire (le premier dans l'ordre de lecture, "
|
||
"généralement le plus petit numéro), pas une suite quelconque.\n"
|
||
)
|
||
else:
|
||
excerpt = current_content.strip()[-_NEXT_LINK_CONTENT_EXCERPT_CHARS:]
|
||
if excerpt:
|
||
user_prompt += f"\nFin du contenu de cette page :\n…{excerpt}\n"
|
||
user_prompt += f"\nLiens candidats :\n{numbered}"
|
||
|
||
system_prompt = self.config.next_link_prompt.strip() or DEFAULT_NEXT_LINK_PROMPT
|
||
raw = self._complete(llm, system_prompt, user_prompt, max_tokens=80, on_activity=on_activity)
|
||
index, same_chapter = self._parse_next_link_response(raw, len(candidates))
|
||
if index is None:
|
||
return NextLinkResult(candidate=None)
|
||
if from_toc:
|
||
same_chapter = False
|
||
return NextLinkResult(candidate=candidates[index], same_chapter=same_chapter)
|
||
|
||
def extract_main_content(
|
||
self,
|
||
page_title: str,
|
||
raw_text: str,
|
||
max_input_chars: int = 6000,
|
||
on_activity: Callable[[], None] | None = None,
|
||
) -> str:
|
||
"""Demande au modèle d'extraire le texte narratif principal d'un
|
||
texte brut de page (déjà tronqué à `max_input_chars` caractères —
|
||
les petits modèles locaux ont une fenêtre de contexte limitée)."""
|
||
llm = self._ensure_loaded()
|
||
trimmed = raw_text.strip()[:max_input_chars]
|
||
if not trimmed:
|
||
raise LocalAiError("Aucun texte à analyser sur cette page.")
|
||
|
||
user_prompt = f"Titre de la page : {page_title!r}\n\n--- TEXTE BRUT ---\n{trimmed}"
|
||
result = self._complete(
|
||
llm,
|
||
_EXTRACT_SYSTEM_PROMPT,
|
||
user_prompt,
|
||
max_tokens=max(self.config.max_tokens, len(trimmed) // 3),
|
||
on_activity=on_activity,
|
||
)
|
||
result = result.strip()
|
||
if not result:
|
||
raise LocalAiError("L'IA n'a renvoyé aucun contenu exploitable pour cette page.")
|
||
return result
|
||
|
||
def estimate_chapter_count(
|
||
self,
|
||
page_title: str,
|
||
raw_text: str,
|
||
max_input_chars: int = 4000,
|
||
on_activity: Callable[[], None] | None = None,
|
||
) -> int | None:
|
||
"""Tente de déterminer le nombre total de chapitres d'une histoire
|
||
à partir du titre/texte de sa page de départ (voir
|
||
MainWindow._toggle_auto_scrape, quand l'utilisateur choisit "0
|
||
chapitre" = limite à détecter automatiquement). Ne lève jamais
|
||
d'exception : retourne None si indéterminable OU en cas d'échec de
|
||
l'inférence — cette détection est un simple bonus, son échec ne
|
||
doit jamais empêcher le scraping automatique de démarrer."""
|
||
try:
|
||
llm = self._ensure_loaded()
|
||
trimmed = raw_text.strip()[:max_input_chars]
|
||
if not trimmed:
|
||
return None
|
||
|
||
user_prompt = f"Titre de la page : {page_title!r}\n\n--- TEXTE BRUT ---\n{trimmed}"
|
||
raw = self._complete(
|
||
llm, _ESTIMATE_CHAPTER_COUNT_PROMPT, user_prompt, max_tokens=40, on_activity=on_activity
|
||
)
|
||
except LocalAiError:
|
||
return None
|
||
|
||
match = re.search(r'"count"\s*:\s*(null|\d+)', raw)
|
||
if not match or match.group(1) == "null":
|
||
return None
|
||
try:
|
||
count = int(match.group(1))
|
||
except ValueError:
|
||
return None
|
||
return count if count > 0 else None
|
||
|
||
def is_table_of_contents(
|
||
self,
|
||
page_title: str,
|
||
raw_text: str,
|
||
max_input_chars: int = 3000,
|
||
on_activity: Callable[[], None] | None = None,
|
||
) -> bool:
|
||
"""Détecte si la page actuelle est une table des matières / liste de
|
||
chapitres plutôt que le contenu narratif d'un chapitre — sans ce
|
||
garde-fou, le scraping automatique tente d'"extraire le contenu
|
||
narratif" d'une simple liste de liens (résultat vide ou aberrant,
|
||
capturé comme un faux chapitre) et choisit un "lien suivant" au
|
||
hasard parmi tous les chapitres listés plutôt que viser le premier
|
||
(voir MainWindow._auto_scrape_on_page_text). Best-effort, comme
|
||
estimate_chapter_count : ne lève jamais, retourne False (page
|
||
normale, comportement inchangé) en cas d'échec ou d'ambiguïté."""
|
||
try:
|
||
llm = self._ensure_loaded()
|
||
trimmed = raw_text.strip()[:max_input_chars]
|
||
if not trimmed:
|
||
return False
|
||
|
||
user_prompt = f"Titre de la page : {page_title!r}\n\n--- TEXTE BRUT ---\n{trimmed}"
|
||
raw = self._complete(
|
||
llm, _IS_TABLE_OF_CONTENTS_PROMPT, user_prompt, max_tokens=20, on_activity=on_activity
|
||
)
|
||
except LocalAiError:
|
||
return False
|
||
|
||
match = re.search(r'"is_toc"\s*:\s*(true|false)', raw)
|
||
return bool(match and match.group(1) == "true")
|
||
|
||
# ------------------------------------------------------------------ #
|
||
def _complete(
|
||
self,
|
||
llm,
|
||
system_prompt: str,
|
||
user_prompt: str,
|
||
max_tokens: int,
|
||
on_activity: Callable[[], None] | None = None,
|
||
) -> str:
|
||
"""Exécute l'inférence en streaming plutôt qu'en un seul appel
|
||
bloquant : `on_activity()` est rappelé à chaque fragment généré,
|
||
et sert de "signe de vie" à l'appelant (voir MainWindow._run_ai_task
|
||
et son délai d'INACTIVITÉ, pas de durée totale — un modèle lent
|
||
mais qui progresse encore n'est ainsi jamais interrompu à tort,
|
||
contrairement à un simple timeout sur la durée totale de l'appel)."""
|
||
try:
|
||
chunks = llm.create_chat_completion(
|
||
messages=[
|
||
{"role": "system", "content": system_prompt},
|
||
{"role": "user", "content": user_prompt},
|
||
],
|
||
max_tokens=max_tokens,
|
||
temperature=0.1,
|
||
stream=True,
|
||
)
|
||
parts: list[str] = []
|
||
for chunk in chunks:
|
||
delta = (chunk.get("choices") or [{}])[0].get("delta") or {}
|
||
piece = delta.get("content")
|
||
if piece:
|
||
parts.append(piece)
|
||
if on_activity is not None:
|
||
on_activity()
|
||
return "".join(parts)
|
||
except LocalAiError:
|
||
raise
|
||
except Exception as exc: # noqa: BLE001 — remonté proprement à l'UI
|
||
raise LocalAiError(f"Échec de l'inférence IA :\n{exc}") from exc
|
||
|
||
@staticmethod
|
||
def _parse_next_link_response(raw: str, count: int) -> tuple[int | None, bool]:
|
||
index_match = re.search(r'"index"\s*:\s*(null|\d+)', raw)
|
||
if not index_match or index_match.group(1) == "null":
|
||
return None, False
|
||
try:
|
||
index = int(index_match.group(1))
|
||
except ValueError:
|
||
return None, False
|
||
if not (0 <= index < count):
|
||
return None, False
|
||
|
||
same_chapter = False
|
||
chapter_match = re.search(r'"same_chapter"\s*:\s*(true|false)', raw)
|
||
if chapter_match:
|
||
same_chapter = chapter_match.group(1) == "true"
|
||
return index, same_chapter
|