feat: ajout d'une fonctionnalité d'importation de l'historique shell avec prévisualisation et importation des nouvelles commandes

Author: serge-eric-kalaga

backend/api/routers/imports.py

@@ -1,5 +1,7 @@
 from __future__ import annotations
 
+import re
+
 from fastapi import APIRouter, Depends, HTTPException, status
 from pydantic import BaseModel, Field
 from sqlalchemy import select
@@ -39,6 +41,36 @@ class ImportResponse(BaseModel):
     commands_created: int
 
 
+class HistoryImportRequest(BaseModel):
+    group_id: int
+    history: str = Field(min_length=1)
+    tags: list[str] = Field(default_factory=list)
+
+
+class HistoryImportPreviewItem(BaseModel):
+    command: str
+    status: str  # new | duplicate | noise
+    reason: str | None = None
+
+
+class HistoryImportPreviewResponse(BaseModel):
+    total_lines: int
+    parsed: int
+    created_candidates: int
+    duplicate_candidates: int
+    noise_candidates: int
+    truncated: bool
+    items: list[HistoryImportPreviewItem]
+
+
+class HistoryImportResponse(BaseModel):
+    created: int
+    skipped_duplicates: int
+    skipped_noise: int
+    total_lines: int
+    parsed: int
+
+
 def _normalize_tags(raw: list[str]) -> list[str]:
     names: list[str] = []
     seen: set[str] = set()
@@ -81,6 +113,148 @@ def _get_or_create_tags(db: Session, raw_names: list[str]) -> list[Tag]:
 )
 
 
+_WS_RE = re.compile(r"\s+")
+
+
+def _normalize_history_line(raw: str) -> str:
+    line = (raw or "").strip()
+    if not line:
+        return ""
+
+    # zsh format: ": 1712345678:0;the command"
+    if line.startswith(":") and ";" in line:
+        line = line.split(";", 1)[1].strip()
+
+    return line
+
+
+def _normalize_command_text(cmd: str) -> str:
+    cmd = (cmd or "").strip()
+    cmd = _WS_RE.sub(" ", cmd)
+    return cmd
+
+
+def _is_noise_command(cmd: str) -> tuple[bool, str | None]:
+    if not cmd:
+        return True, "empty"
+
+    if cmd.startswith("#"):
+        return True, "comment"
+
+    if len(cmd) < 3:
+        return True, "too short"
+
+    first = cmd.split(" ", 1)[0].lower()
+
+    # common noise commands in history
+    if first in {"cd", "ls", "pwd", "clear", "exit", "history", "alias", "unalias"}:
+        return True, f"noise: {first}"
+
+    return False, None
+
+
+def _make_title_from_command(cmd: str) -> str:
+    tokens = cmd.split(" ")
+    if not tokens:
+        return "Imported command"
+
+    if len(tokens) == 1:
+        base = tokens[0]
+    else:
+        base = f"{tokens[0]} {tokens[1]}".strip()
+
+    if len(base) < 3:
+        base = tokens[0]
+
+    # Keep within schema max_length=160
+    return base[:160]
+
+
+def _preview_history_import(
+    db: Session,
+    history: str,
+    max_items: int = 300,
+) -> tuple[list[HistoryImportPreviewItem], int, int, int, int, int, bool]:
+    lines = history.splitlines()
+    total_lines = len(lines)
+
+    existing = {
+        _normalize_command_text(cmd)
+        for cmd in db.scalars(select(Command.command)).all()
+        if _normalize_command_text(cmd)
+    }
+
+    seen_in_payload: set[str] = set()
+    items: list[HistoryImportPreviewItem] = []
+
+    parsed = 0
+    created_candidates = 0
+    duplicate_candidates = 0
+    noise_candidates = 0
+    truncated = False
+
+    for raw in lines:
+        normalized_line = _normalize_history_line(raw)
+        normalized_cmd = _normalize_command_text(normalized_line)
+        if not normalized_cmd:
+            continue
+
+        parsed += 1
+
+        is_noise, reason = _is_noise_command(normalized_cmd)
+        if is_noise:
+            noise_candidates += 1
+            if len(items) < max_items:
+                items.append(
+                    HistoryImportPreviewItem(
+                        command=normalized_cmd,
+                        status="noise",
+                        reason=reason,
+                    )
+                )
+            else:
+                truncated = True
+            continue
+
+        if normalized_cmd in seen_in_payload:
+            duplicate_candidates += 1
+            if len(items) < max_items:
+                items.append(
+                    HistoryImportPreviewItem(
+                        command=normalized_cmd,
+                        status="duplicate",
+                        reason="duplicate in history",
+                    )
+                )
+            else:
+                truncated = True
+            continue
+
+        seen_in_payload.add(normalized_cmd)
+
+        if normalized_cmd in existing:
+            duplicate_candidates += 1
+            if len(items) < max_items:
+                items.append(
+                    HistoryImportPreviewItem(
+                        command=normalized_cmd,
+                        status="duplicate",
+                        reason="already exists",
+                    )
+                )
+            else:
+                truncated = True
+            continue
+
+        created_candidates += 1
+        if len(items) < max_items:
+            items.append(HistoryImportPreviewItem(command=normalized_cmd, status="new"))
+        else:
+            truncated = True
+
+    return items, total_lines, parsed, created_candidates, duplicate_candidates, noise_candidates, truncated
+
+
 @router.post("", response_model=ImportResponse, status_code=status.HTTP_200_OK)
 def import_vault(
     payload: ImportRequest, db: Session = Depends(get_db)
@@ -137,3 +311,70 @@ def import_vault(
         groups_created=len(payload.groups),
         commands_created=len(payload.commands),
     )
+
+
+@router.post("/history/preview", response_model=HistoryImportPreviewResponse, status_code=status.HTTP_200_OK)
+def preview_history_import(payload: HistoryImportRequest, db: Session = Depends(get_db)) -> HistoryImportPreviewResponse:
+    group = db.get(Group, payload.group_id)
+    if group is None:
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Group does not exist")
+
+    items, total_lines, parsed, created, dupes, noise, truncated = _preview_history_import(
+        db=db,
+        history=payload.history,
+    )
+
+    return HistoryImportPreviewResponse(
+        total_lines=total_lines,
+        parsed=parsed,
+        created_candidates=created,
+        duplicate_candidates=dupes,
+        noise_candidates=noise,
+        truncated=truncated,
+        items=items,
+    )
+
+
+@router.post("/history", response_model=HistoryImportResponse, status_code=status.HTTP_200_OK)
+def import_history(payload: HistoryImportRequest, db: Session = Depends(get_db)) -> HistoryImportResponse:
+    group = db.get(Group, payload.group_id)
+    if group is None:
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Group does not exist")
+
+    items, total_lines, parsed, created, dupes, noise, _truncated = _preview_history_import(
+        db=db,
+        history=payload.history,
+        max_items=10_000_000,  # no truncation for actual import
+    )
+
+    tags = _normalize_tags(payload.tags)
+
+    try:
+        for item in items:
+            if item.status != "new":
+                continue
+
+            cmd = Command(
+                group_id=payload.group_id,
+                title=_make_title_from_command(item.command),
+                command=item.command,
+                description=None,
+                default_variables={},
+                is_favorite=False,
+                copy_count=0,
+            )
+            cmd.tag_entities = _get_or_create_tags(db, tags)
+            db.add(cmd)
+
+        db.commit()
+    except Exception:
+        db.rollback()
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Import failed")
+
+    return HistoryImportResponse(
+        created=created,
+        skipped_duplicates=dupes,
+        skipped_noise=noise,
+        total_lines=total_lines,
+        parsed=parsed,
+    )

backend/tests/test_commands_groups_search.py

@@ -379,3 +379,82 @@ def test_top_copied_commands_returns_sorted_by_copy_count(client):
     # Ensure sorted by copy_count desc for these three
     top_three = [item for item in items if item["id"] in {c1["id"], c2["id"], c3["id"]}]
     assert [item["id"] for item in top_three][:3] == [c2["id"], c1["id"], c3["id"]]
+
+
+def test_import_history_preview_and_import_creates_only_new_commands(client):
+    login = _login(client, "admin", "admin")
+    token = login["access_token"]
+
+    res = client.post(
+        "/auth/change-password",
+        json={"old_password": "admin", "new_password": "admin123"},
+        headers=_auth_headers(token),
+    )
+    assert res.status_code == 200
+
+    login2 = _login(client, "admin", "admin123")
+    token2 = login2["access_token"]
+
+    res = client.post(
+        "/groups", json={"name": "Projet A"}, headers=_auth_headers(token2)
+    )
+    assert res.status_code == 201
+    group = res.json()
+
+    # Existing command in DB (should be treated as duplicate)
+    res = client.post(
+        "/commands",
+        json={
+            "group_id": group["id"],
+            "title": "Docker list",
+            "command": "docker ps -a",
+            "tags": [],
+        },
+        headers=_auth_headers(token2),
+    )
+    assert res.status_code == 201
+
+    history = "\n".join(
+        [
+            "cd /tmp",
+            "ls -la",
+            ": 1712345678:0;docker ps -a",
+            "docker ps -a",
+            "docker system prune -af",
+            "echo hello",
+        ]
+    )
+
+    res = client.post(
+        "/import/history/preview",
+        json={"group_id": group["id"], "history": history, "tags": ["devops"]},
+        headers=_auth_headers(token2),
+    )
+    assert res.status_code == 200
+    body = res.json()
+    assert body["total_lines"] == 6
+    assert body["parsed"] == 6
+    assert body["created_candidates"] == 2
+    assert body["duplicate_candidates"] == 2
+    assert body["noise_candidates"] == 2
+
+    res = client.post(
+        "/import/history",
+        json={"group_id": group["id"], "history": history, "tags": ["devops"]},
+        headers=_auth_headers(token2),
+    )
+    assert res.status_code == 200
+    body2 = res.json()
+    assert body2["created"] == 2
+    assert body2["skipped_duplicates"] == 2
+    assert body2["skipped_noise"] == 2
+
+    res = client.get(
+        "/commands", params={"group_id": group["id"]}, headers=_auth_headers(token2)
+    )
+    assert res.status_code == 200
+    cmds = res.json()
+    assert len(cmds) == 3
+    created_commands = {c["command"] for c in cmds}
+    assert "docker system prune -af" in created_commands
+    assert "echo hello" in created_commands

frontend/src/components/AppSidebar.tsx

@@ -21,6 +21,7 @@ interface SidebarProps {
   onNewGroup: () => void;
   onExport: () => void;
   onImport: (file: File) => void;
+  onImportHistory: () => void;
   onLogout: () => void;
   dark: boolean;
   onToggleTheme: () => void;
@@ -32,7 +33,7 @@ export default function AppSidebar({
   groups, stats, activeView, onViewChange, onNewGroup,
   loading = false,
   importing = false,
-  onExport, onImport, onLogout, dark, onToggleTheme, onEditGroup, onDeleteGroup,
+  onExport, onImport, onImportHistory, onLogout, dark, onToggleTheme, onEditGroup, onDeleteGroup,
 }: SidebarProps) {
   const [mobileOpen, setMobileOpen] = useState(false);
   const fileRef = useRef<HTMLInputElement>(null);
@@ -162,6 +163,14 @@ export default function AppSidebar({
               <Upload className="w-4 h-4" />
               Importer
             </DropdownMenuItem>
+            <DropdownMenuItem
+              onSelect={onImportHistory}
+              disabled={loading || importing}
+              className="gap-2"
+            >
+              <Upload className="w-4 h-4" />
+              Importer historique
+            </DropdownMenuItem>
             <DropdownMenuItem onSelect={onExport} className="gap-2">
               <Download className="w-4 h-4" />
               Exporter

frontend/src/components/HelpModal.tsx

@@ -121,6 +121,9 @@ export default function HelpModal({ open, onClose }: Props) {
                                 Export (JSON/CSV/PDF) et import JSON sont accessibles depuis la sidebar.
                                 Tu peux aussi exporter une sélection via la barre d’actions.
                             </p>
+                            <p>
+                                Tu peux aussi importer ton <strong className="text-foreground">historique shell</strong> (bash/zsh) via “Importer historique”.
+                            </p>
                         </div>
                     </section>