Spaces:
Sleeping
Sleeping
File size: 4,827 Bytes
faa1393 162c559 faa1393 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 | """``picarones-rewrite import-corpus`` — extraction sandboxée d'un ZIP.
Sprint A14-S22.
Wrapper CLI minimal autour du ``CorpusService`` (S20) :
::
python -m picarones.interfaces.cli import-corpus mon_corpus.zip \\
--output-dir ./workspaces/sess1 \\
--corpus-name bnf_xviiie \\
--metadata language=fr \\
--metadata period=early_modern
Comportement
------------
- Lit le ZIP (path utilisateur, sans validation préalable — la CLI
fait confiance au filesystem local de l'opérateur).
- Crée un ``WorkspaceManager`` dans ``--output-dir`` (créé s'il
n'existe pas).
- Appelle ``CorpusService.import_zip``.
- Affiche un résumé lisible : n_documents, n_images sans GT, GT
orphelines, warnings.
- Code de sortie ``0`` succès, ``1`` erreur typée
(``CorpusImportError``), ``2`` erreur d'usage Click (gérée par
Click).
"""
from __future__ import annotations
import sys
from pathlib import Path
import click
from picarones.app.services import (
CorpusImportError,
CorpusService,
WorkspaceManager,
)
@click.command()
@click.argument(
"zip_path",
type=click.Path(
exists=True, dir_okay=False, file_okay=True, path_type=Path,
),
)
@click.option(
"--output-dir",
"output_dir",
type=click.Path(file_okay=False, dir_okay=True, path_type=Path),
required=True,
help=(
"Répertoire parent où créer le workspace sandboxé. Créé "
"s'il n'existe pas."
),
)
@click.option(
"--corpus-name",
default=None,
help=(
"Nom du corpus (défaut : nom du fichier ZIP sans "
"extension). Sera sanitizé automatiquement."
),
)
@click.option(
"--metadata",
"metadata_pairs",
multiple=True,
help=(
"Paires ``clé=valeur`` (option répétable). Ex : "
"``--metadata language=fr --metadata period=medieval``."
),
)
@click.option(
"--max-zip-mb",
default=100,
type=int,
show_default=True,
help="Plafond taille du blob ZIP (Mo).",
)
@click.option(
"--max-entries",
default=5000,
type=int,
show_default=True,
help="Plafond nombre d'entrées dans le ZIP (anti zip bomb).",
)
@click.option(
"--max-uncompressed-mb",
default=500,
type=int,
show_default=True,
help="Plafond taille décompressée totale (Mo).",
)
@click.option(
"--quiet",
is_flag=True,
default=False,
help="N'affiche que le chemin du dossier extrait, rien d'autre.",
)
def import_corpus_command(
zip_path: Path,
output_dir: Path,
corpus_name: str | None,
metadata_pairs: tuple[str, ...],
max_zip_mb: int,
max_entries: int,
max_uncompressed_mb: int,
quiet: bool,
) -> None:
"""Extrait un ZIP de corpus dans un workspace sandboxé."""
output_dir.mkdir(parents=True, exist_ok=True)
workspace = WorkspaceManager(output_dir)
if corpus_name is None:
corpus_name = zip_path.stem
metadata = _parse_metadata_pairs(metadata_pairs)
service = CorpusService(
workspace,
max_zip_size_bytes=max_zip_mb * 1024 * 1024,
max_entry_count=max_entries,
max_uncompressed_bytes=max_uncompressed_mb * 1024 * 1024,
)
try:
report = service.import_zip(
zip_path.read_bytes(),
corpus_name=corpus_name,
metadata=metadata,
)
except CorpusImportError as exc:
click.echo(f"erreur : {exc}", err=True)
sys.exit(1)
if quiet:
click.echo(str(report.extracted_dir))
return
click.echo(f"Corpus extrait dans : {report.extracted_dir}")
click.echo(f" documents : {report.n_documents}")
click.echo(f" sans GT : {report.n_images_without_gt}")
click.echo(f" GT orphelines : {report.n_gt_without_image}")
click.echo(f" bruit OS sauté : {report.n_skipped_noise}")
if report.warnings:
click.echo("Avertissements :")
for w in report.warnings:
click.echo(f" - {w}")
def _parse_metadata_pairs(
pairs: tuple[str, ...],
) -> dict[str, str]:
"""Parse ``("k1=v1", "k2=v2")`` → ``{"k1": "v1", "k2": "v2"}``.
Lève ``click.BadParameter`` si une paire ne contient pas ``=``.
"""
out: dict[str, str] = {}
for pair in pairs:
if "=" not in pair:
raise click.BadParameter(
f"métadonnée invalide : {pair!r} (attendu ``clé=valeur``).",
param_hint="--metadata",
)
key, _, value = pair.partition("=")
key = key.strip()
value = value.strip()
if not key:
raise click.BadParameter(
f"métadonnée à clé vide : {pair!r}.",
param_hint="--metadata",
)
out[key] = value
return out
__all__ = ["import_corpus_command"]
|