From 042c01504587e281c92f4cb5b4710d515f2f48a4 Mon Sep 17 00:00:00 2001 From: oliver Date: Wed, 27 May 2026 23:34:29 +0800 Subject: [PATCH] feat(ops): add managed NE management and batch CLI collection Introduce workbench operations for multi-vendor NE CRUD/connect-test and Netmiko batch collection with job lifecycle controls, log downloads, and paginated run filters. Co-authored-by: Cursor --- .env.example | 4 + .gitignore | 1 + netx_api/collection_job_state.py | 78 +++++ netx_api/collection_recovery.py | 52 +++ netx_api/collection_router.py | 100 ++++++ netx_api/collection_schemas.py | 41 +++ netx_api/collection_service.py | 385 +++++++++++++++++++++ netx_api/config.py | 10 + netx_api/db.py | 8 + netx_api/device_types.py | 45 +++ netx_api/main.py | 35 +- netx_api/managed_ne_router.py | 92 +++++ netx_api/models.py | 61 ++++ netx_api/ne_collect_runner.py | 174 ++++++++++ netx_api/ne_collection_paths.py | 26 ++ netx_api/ne_connect.py | 193 +++++++++++ netx_api/ne_crypto.py | 40 +++ netx_api/ne_netmiko.py | 17 + netx_api/ne_schemas.py | 93 +++++ netx_api/ne_service.py | 280 +++++++++++++++ requirements.txt | 2 + tests/test_collection_job_state.py | 35 ++ tests/test_managed_ne.py | 175 ++++++++++ tests/test_ne_collection.py | 15 + web/WEB.md | 15 + web/src/App.tsx | 4 + web/src/config/modules.ts | 20 +- web/src/constants/queryKeys.ts | 12 + web/src/i18n/en.ts | 113 ++++++ web/src/i18n/zh.ts | 112 ++++++ web/src/index.css | 233 ++++++++++++- web/src/pages/CollectPage.tsx | 531 +++++++++++++++++++++++++++++ web/src/pages/NePage.tsx | 493 ++++++++++++++++++++++++++ web/src/pages/WorkbenchPage.tsx | 2 +- web/src/services/api.ts | 126 +++++++ web/src/types.ts | 89 +++++ 36 files changed, 3695 insertions(+), 17 deletions(-) create mode 100644 netx_api/collection_job_state.py create mode 100644 netx_api/collection_recovery.py create mode 100644 netx_api/collection_router.py create mode 100644 netx_api/collection_schemas.py create mode 100644 netx_api/collection_service.py create mode 100644 netx_api/device_types.py create mode 100644 netx_api/managed_ne_router.py create mode 100644 netx_api/ne_collect_runner.py create mode 100644 netx_api/ne_collection_paths.py create mode 100644 netx_api/ne_connect.py create mode 100644 netx_api/ne_crypto.py create mode 100644 netx_api/ne_netmiko.py create mode 100644 netx_api/ne_schemas.py create mode 100644 netx_api/ne_service.py create mode 100644 tests/test_collection_job_state.py create mode 100644 tests/test_managed_ne.py create mode 100644 tests/test_ne_collection.py create mode 100644 web/src/pages/CollectPage.tsx create mode 100644 web/src/pages/NePage.tsx diff --git a/.env.example b/.env.example index 116a311..3184c89 100644 --- a/.env.example +++ b/.env.example @@ -1,4 +1,8 @@ NETX_DATABASE_URL=postgresql+psycopg://netx:netx@127.0.0.1:5432/netx +# Fernet key for managed NE passwords (python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())") +NETX_CREDENTIAL_SECRET_KEY= +NETX_NE_CONNECT_MAX_WORKERS=5 +NETX_NE_CONNECT_TIMEOUT_SEC=30 NETX_HOST=127.0.0.1 NETX_PORT=8890 NETX_VENDOR=ZTE diff --git a/.gitignore b/.gitignore index 2fe42e2..539a603 100644 --- a/.gitignore +++ b/.gitignore @@ -7,3 +7,4 @@ build/ dist/ scripts/.run/ ume/ +data/ne_collections/ diff --git a/netx_api/collection_job_state.py b/netx_api/collection_job_state.py new file mode 100644 index 0000000..ec24223 --- /dev/null +++ b/netx_api/collection_job_state.py @@ -0,0 +1,78 @@ +from __future__ import annotations + +from datetime import datetime + +from sqlalchemy.orm import Session + +from .config import settings +from .models import NeCollectionJob, NeCollectionRun + +_TERMINAL = frozenset({"success", "fail", "cancelled"}) + + +def _sync_job_counts(job: NeCollectionJob, runs: list[NeCollectionRun]) -> None: + job.success_count = sum(1 for r in runs if str(r.status) == "success") + job.fail_count = sum(1 for r in runs if str(r.status) in ("fail", "cancelled")) + + +def finalize_collection_job(db: Session, job_id: str) -> None: + runs = db.query(NeCollectionRun).filter(NeCollectionRun.job_id == job_id).all() + if not runs: + return + if any(str(r.status or "") not in _TERMINAL for r in runs): + return + job = db.get(NeCollectionJob, job_id) + if not job: + return + _sync_job_counts(job, runs) + job_status = str(job.status or "") + finish_at = datetime.now() + if job_status == "paused": + if not job.ended_at: + job.ended_at = finish_at + else: + success = int(job.success_count or 0) + job.status = "done" if success > 0 else "failed" + if not job.ended_at: + job.ended_at = finish_at + job.last_run_at = job.ended_at or finish_at + db.commit() + + +def reconcile_stale_collection_job(db: Session, job_id: str) -> bool: + """Mark long-running pending/running rows as failed and finalize job if possible.""" + job = db.get(NeCollectionJob, job_id) + if not job: + return False + if str(job.status or "") in ("done", "failed", "paused"): + return False + run_stale_sec = max(60, int(settings.ne_collect_stale_run_sec or 900)) + pending_stale_sec = max(30, int(settings.ne_collect_pending_stale_sec or 180)) + now = datetime.now() + runs = db.query(NeCollectionRun).filter(NeCollectionRun.job_id == job_id).all() + changed = False + for run in runs: + st = str(run.status or "") + if st in _TERMINAL: + continue + if st == "pending": + anchor = job.started_at or job.created_at + limit = pending_stale_sec + reason = "collection_pending_stale" + else: + anchor = run.started_at or job.started_at or job.created_at + limit = run_stale_sec + reason = "collection_timeout_stale" + if not anchor: + continue + age = (now - anchor).total_seconds() + if age < limit: + continue + run.status = "fail" + run.message = f"{reason} ({int(age)}s)" + run.ended_at = now + changed = True + if changed: + db.commit() + finalize_collection_job(db, job_id) + return changed diff --git a/netx_api/collection_recovery.py b/netx_api/collection_recovery.py new file mode 100644 index 0000000..c6468f4 --- /dev/null +++ b/netx_api/collection_recovery.py @@ -0,0 +1,52 @@ +from __future__ import annotations + +import logging + +from sqlalchemy.orm import Session + +from .collection_job_state import finalize_collection_job, reconcile_stale_collection_job +from .models import NeCollectionJob, NeCollectionRun +from .ne_collect_runner import schedule_collection_runs + +_log = logging.getLogger("netx.collection.recovery") + + +def _parse_commands(text: str) -> list[str]: + lines: list[str] = [] + for raw in str(text or "").splitlines(): + line = raw.strip() + if not line or line.startswith("#"): + continue + lines.append(line) + return lines + + +def recover_collection_jobs_on_startup(db: Session) -> int: + """Re-queue pending runs for running jobs after API restart; reconcile stale rows.""" + jobs = db.query(NeCollectionJob).filter(NeCollectionJob.status == "running").all() + resumed = 0 + for job in jobs: + job_id = str(job.id) + reconcile_stale_collection_job(db, job_id) + db.refresh(job) + if str(job.status or "") in ("done", "failed"): + continue + commands = _parse_commands(str(job.commands or "")) + if not commands: + job.status = "failed" + job.error_message = "commands_empty_on_recovery" + db.commit() + continue + pending = ( + db.query(NeCollectionRun) + .filter(NeCollectionRun.job_id == job_id, NeCollectionRun.status == "pending") + .all() + ) + if not pending: + finalize_collection_job(db, job_id) + continue + run_ids = [str(r.id) for r in pending] + schedule_collection_runs(job_id, run_ids, commands) + resumed += len(run_ids) + _log.info("resumed collection job=%s pending_runs=%s", job_id, len(run_ids)) + return resumed diff --git a/netx_api/collection_router.py b/netx_api/collection_router.py new file mode 100644 index 0000000..3aa1a9d --- /dev/null +++ b/netx_api/collection_router.py @@ -0,0 +1,100 @@ +from __future__ import annotations + +from fastapi import APIRouter, Depends, Query +from fastapi.responses import FileResponse, Response +from sqlalchemy.orm import Session + +from .collection_service import ( + build_collection_job_zip, + create_and_start_collection, + delete_collection_job, + get_collection_job, + list_collection_jobs, + list_collection_runs, + list_eligible_ne, + pause_collection_job, + resolve_run_output_path, + restart_collection_job, +) +from .collection_schemas import CollectionJobCreate +from .db import get_db +from .models import NeCollectionRun + +router = APIRouter(prefix="/v1/ne-collections", tags=["ne-collections"]) + + +@router.get("/eligible-ne") +def api_eligible_ne( + page: int = Query(default=1, ge=1), + page_size: int = Query(default=200, ge=1, le=500), + db: Session = Depends(get_db), +): + return list_eligible_ne(db, page=page, page_size=page_size) + + +@router.post("") +def api_create_collection(body: CollectionJobCreate, db: Session = Depends(get_db)): + return create_and_start_collection(db, body).model_dump() + + +@router.get("") +def api_list_collections( + page: int = Query(default=1, ge=1), + page_size: int = Query(default=20, ge=1, le=100), + db: Session = Depends(get_db), +): + return list_collection_jobs(db, page=page, page_size=page_size) + + +@router.get("/runs/{run_id}/download") +def api_download_run_output(run_id: str, db: Session = Depends(get_db)): + run = db.get(NeCollectionRun, run_id) + if not run: + from fastapi import HTTPException + + raise HTTPException(status_code=404, detail="collection_run_not_found") + path = resolve_run_output_path(str(run.output_rel_path or "")) + filename = path.name + return FileResponse(path, media_type="text/plain; charset=utf-8", filename=filename) + + +@router.get("/{job_id}/download") +def api_download_collection_job(job_id: str, db: Session = Depends(get_db)): + filename, payload = build_collection_job_zip(db, job_id) + return Response( + content=payload, + media_type="application/zip", + headers={"content-disposition": f'attachment; filename="{filename}"'}, + ) + + +@router.get("/{job_id}/runs") +def api_list_collection_runs( + job_id: str, + page: int = Query(default=1, ge=1), + page_size: int = Query(default=20, ge=1, le=200), + status: str = Query(default=""), + keyword: str = Query(default=""), + db: Session = Depends(get_db), +): + return list_collection_runs(db, job_id, page=page, page_size=page_size, status=status, keyword=keyword) + + +@router.get("/{job_id}") +def api_get_collection(job_id: str, db: Session = Depends(get_db)): + return get_collection_job(db, job_id) + + +@router.post("/{job_id}/pause") +def api_pause_collection(job_id: str, db: Session = Depends(get_db)): + return pause_collection_job(db, job_id).model_dump() + + +@router.post("/{job_id}/restart") +def api_restart_collection(job_id: str, db: Session = Depends(get_db)): + return restart_collection_job(db, job_id).model_dump() + + +@router.delete("/{job_id}") +def api_delete_collection(job_id: str, db: Session = Depends(get_db)): + return delete_collection_job(db, job_id) diff --git a/netx_api/collection_schemas.py b/netx_api/collection_schemas.py new file mode 100644 index 0000000..4635b87 --- /dev/null +++ b/netx_api/collection_schemas.py @@ -0,0 +1,41 @@ +from __future__ import annotations + +from datetime import datetime + +from pydantic import BaseModel, Field + + +class CollectionJobCreate(BaseModel): + title: str = "" + commands: str = Field(min_length=1) + ne_ids: list[str] = Field(min_length=1) + + +class CollectionRunOut(BaseModel): + id: str + job_id: str + ne_id: str + ne_name: str + ne_ip: str + status: str + message: str + output_rel_path: str + has_output: bool + started_at: datetime | None + ended_at: datetime | None + + +class CollectionJobOut(BaseModel): + id: str + title: str + commands: str + status: str + ne_count: int + success_count: int + fail_count: int + output_count: int = 0 + error_message: str + created_at: datetime + started_at: datetime | None + ended_at: datetime | None + last_run_at: datetime | None = None diff --git a/netx_api/collection_service.py b/netx_api/collection_service.py new file mode 100644 index 0000000..419f0b1 --- /dev/null +++ b/netx_api/collection_service.py @@ -0,0 +1,385 @@ +from __future__ import annotations + +import io +import logging +import re +import shutil +import zipfile +from datetime import datetime +from pathlib import Path +from typing import Any + +from fastapi import HTTPException +from sqlalchemy import func, or_ +from sqlalchemy.orm import Session + +from .models import ManagedNE, NeCollectionJob, NeCollectionRun +from .collection_job_state import finalize_collection_job, reconcile_stale_collection_job, _sync_job_counts +from .collection_schemas import CollectionJobCreate, CollectionJobOut, CollectionRunOut +from .ne_collect_runner import schedule_collection_runs +from .ne_collection_paths import clear_run_output_files, collection_data_root + +_log = logging.getLogger("netx.collection") + + +def _now() -> datetime: + return datetime.utcnow() + + +def _parse_commands(text: str) -> list[str]: + lines = [] + for raw in str(text or "").splitlines(): + line = raw.strip() + if not line or line.startswith("#"): + continue + lines.append(line) + return lines + + +def _output_counts_for_jobs(db: Session, job_ids: list[str]) -> dict[str, int]: + if not job_ids: + return {} + rows = ( + db.query(NeCollectionRun.job_id, func.count()) + .filter( + NeCollectionRun.job_id.in_(job_ids), + NeCollectionRun.output_rel_path != "", + NeCollectionRun.output_rel_path.isnot(None), + ) + .group_by(NeCollectionRun.job_id) + .all() + ) + return {str(job_id): int(count) for job_id, count in rows} + + +def _output_count_for_job(db: Session, job_id: str) -> int: + return int( + db.query(func.count()) + .select_from(NeCollectionRun) + .filter( + NeCollectionRun.job_id == job_id, + NeCollectionRun.output_rel_path != "", + NeCollectionRun.output_rel_path.isnot(None), + ) + .scalar() + or 0 + ) + + +def job_to_out(row: NeCollectionJob, *, output_count: int | None = None) -> CollectionJobOut: + return CollectionJobOut( + id=str(row.id), + title=str(row.title or ""), + commands=str(row.commands or ""), + status=str(row.status or "pending"), + ne_count=int(row.ne_count or 0), + success_count=int(row.success_count or 0), + fail_count=int(row.fail_count or 0), + output_count=int(output_count if output_count is not None else 0), + error_message=str(row.error_message or "")[:1000], + created_at=row.created_at, + started_at=row.started_at, + ended_at=row.ended_at, + last_run_at=row.last_run_at, + ) + + +def run_to_out(row: NeCollectionRun) -> CollectionRunOut: + rel = str(row.output_rel_path or "").strip() + return CollectionRunOut( + id=str(row.id), + job_id=str(row.job_id), + ne_id=str(row.ne_id), + ne_name=str(row.ne_name or ""), + ne_ip=str(row.ne_ip or ""), + status=str(row.status or "pending"), + message=str(row.message or "")[:1000], + output_rel_path=rel, + has_output=bool(rel), + started_at=row.started_at, + ended_at=row.ended_at, + ) + + +def list_eligible_ne(db: Session, *, page: int = 1, page_size: int = 200) -> dict[str, Any]: + stmt = db.query(ManagedNE).filter(ManagedNE.connect_status == "pass") + total = int(stmt.count()) + rows = ( + stmt.order_by(ManagedNE.name.asc()) + .offset((page - 1) * page_size) + .limit(page_size) + .all() + ) + items = [ + { + "id": str(x.id), + "name": str(x.name or ""), + "vendor": str(x.vendor or ""), + "device_type": str(x.device_type or ""), + "ip_address": str(x.ip_address or ""), + "connect_status": str(x.connect_status or ""), + "connect_tested_at": x.connect_tested_at.isoformat() if x.connect_tested_at else None, + } + for x in rows + ] + return {"total": total, "page": page, "page_size": page_size, "items": items} + + +def create_and_start_collection(db: Session, body: CollectionJobCreate) -> CollectionJobOut: + commands = _parse_commands(body.commands) + if not commands: + raise HTTPException(status_code=400, detail="commands_empty") + ne_ids = [str(x).strip() for x in body.ne_ids if str(x).strip()] + if not ne_ids: + raise HTTPException(status_code=400, detail="ne_ids_required") + + ne_rows: list[ManagedNE] = [] + missing: list[str] = [] + not_pass: list[str] = [] + for ne_id in ne_ids: + row = db.get(ManagedNE, ne_id) + if not row: + missing.append(ne_id) + continue + if str(row.connect_status or "") != "pass": + not_pass.append(ne_id) + continue + ne_rows.append(row) + if missing: + raise HTTPException(status_code=404, detail=f"managed_ne_not_found: {','.join(missing[:5])}") + if not_pass: + raise HTTPException(status_code=400, detail=f"ne_connect_not_pass: {','.join(not_pass[:5])}") + if not ne_rows: + raise HTTPException(status_code=400, detail="no_eligible_ne") + + now = _now() + job = NeCollectionJob( + title=str(body.title or "").strip() or f"collect-{now.strftime('%Y%m%d-%H%M%S')}", + commands="\n".join(commands), + status="running", + ne_count=len(ne_rows), + created_at=now, + started_at=now, + last_run_at=now, + ) + db.add(job) + db.flush() + + run_ids: list[str] = [] + for ne in ne_rows: + run = NeCollectionRun( + job_id=str(job.id), + ne_id=str(ne.id), + ne_name=str(ne.name or ne.ip_address or ""), + ne_ip=str(ne.ip_address or ""), + status="pending", + ) + db.add(run) + run_ids.append(str(run.id)) + db.commit() + db.refresh(job) + + schedule_collection_runs(str(job.id), run_ids, commands) + return job_to_out(job, output_count=0) + + +def list_collection_jobs(db: Session, *, page: int = 1, page_size: int = 20) -> dict[str, Any]: + stmt = db.query(NeCollectionJob) + total = int(stmt.count()) + rows = stmt.order_by(NeCollectionJob.created_at.desc()).offset((page - 1) * page_size).limit(page_size).all() + for row in rows: + if str(row.status or "") not in ("done", "failed"): + reconcile_stale_collection_job(db, str(row.id)) + db.refresh(row) + job_ids = [str(x.id) for x in rows] + output_counts = _output_counts_for_jobs(db, job_ids) + return { + "total": total, + "page": page, + "page_size": page_size, + "items": [job_to_out(x, output_count=output_counts.get(str(x.id), 0)).model_dump() for x in rows], + } + + +def get_collection_job(db: Session, job_id: str) -> dict[str, Any]: + job = db.get(NeCollectionJob, job_id) + if not job: + raise HTTPException(status_code=404, detail="collection_job_not_found") + if str(job.status or "") not in ("done", "failed"): + reconcile_stale_collection_job(db, job_id) + db.refresh(job) + return { + "job": job_to_out(job, output_count=_output_count_for_job(db, job_id)).model_dump(), + } + + +def list_collection_runs( + db: Session, + job_id: str, + *, + page: int = 1, + page_size: int = 20, + status: str = "", + keyword: str = "", +) -> dict[str, Any]: + job = db.get(NeCollectionJob, job_id) + if not job: + raise HTTPException(status_code=404, detail="collection_job_not_found") + stmt = db.query(NeCollectionRun).filter(NeCollectionRun.job_id == job_id) + st = str(status or "").strip() + if st: + stmt = stmt.filter(NeCollectionRun.status == st) + kw = str(keyword or "").strip() + if kw: + like = f"%{kw}%" + stmt = stmt.filter(or_(NeCollectionRun.ne_name.ilike(like), NeCollectionRun.ne_ip.ilike(like))) + total = int(stmt.count()) + rows = ( + stmt.order_by(NeCollectionRun.ne_name.asc(), NeCollectionRun.ne_ip.asc()) + .offset((page - 1) * page_size) + .limit(page_size) + .all() + ) + return { + "total": total, + "page": page, + "page_size": page_size, + "items": [run_to_out(x).model_dump() for x in rows], + } + + +def _active_runs(runs: list[NeCollectionRun]) -> bool: + return any(str(r.status or "") in ("pending", "running") for r in runs) + + +def pause_collection_job(db: Session, job_id: str) -> CollectionJobOut: + job = db.get(NeCollectionJob, job_id) + if not job: + raise HTTPException(status_code=404, detail="collection_job_not_found") + if str(job.status or "") != "running": + raise HTTPException(status_code=400, detail="collection_job_not_running") + now = _now() + runs = db.query(NeCollectionRun).filter(NeCollectionRun.job_id == job_id).all() + for run in runs: + if str(run.status or "") != "pending": + continue + run.status = "cancelled" + run.message = "paused" + run.ended_at = now + job.status = "paused" + _sync_job_counts(job, runs) + if not any(str(r.status or "") not in {"success", "fail", "cancelled"} for r in runs): + job.ended_at = now + db.commit() + db.refresh(job) + finalize_collection_job(db, job_id) + db.refresh(job) + return job_to_out(job, output_count=_output_count_for_job(db, job_id)) + + +def restart_collection_job(db: Session, job_id: str) -> CollectionJobOut: + job = db.get(NeCollectionJob, job_id) + if not job: + raise HTTPException(status_code=404, detail="collection_job_not_found") + runs = db.query(NeCollectionRun).filter(NeCollectionRun.job_id == job_id).all() + if not runs: + raise HTTPException(status_code=400, detail="collection_no_runs") + if str(job.status or "") == "running" or _active_runs(runs): + raise HTTPException(status_code=400, detail="collection_job_running") + commands = _parse_commands(str(job.commands or "")) + if not commands: + raise HTTPException(status_code=400, detail="commands_empty") + retry_ids: list[str] = [] + for run in runs: + st = str(run.status or "") + if st in ("pending", "running"): + continue + clear_run_output_files(job_id, str(run.id)) + run.status = "pending" + run.message = "" + run.output_rel_path = "" + run.started_at = None + run.ended_at = None + retry_ids.append(str(run.id)) + if not retry_ids: + raise HTTPException(status_code=400, detail="collection_nothing_to_retry") + now = _now() + job.status = "running" + job.ended_at = None + job.error_message = "" + job.success_count = 0 + job.fail_count = 0 + job.started_at = now + job.last_run_at = now + db.commit() + db.refresh(job) + schedule_collection_runs(job_id, retry_ids, commands) + return job_to_out(job, output_count=0) + + +def delete_collection_job(db: Session, job_id: str) -> dict[str, bool]: + job = db.get(NeCollectionJob, job_id) + if not job: + raise HTTPException(status_code=404, detail="collection_job_not_found") + runs = db.query(NeCollectionRun).filter(NeCollectionRun.job_id == job_id).all() + if str(job.status or "") == "running" or _active_runs(runs): + raise HTTPException(status_code=400, detail="collection_job_running") + db.query(NeCollectionRun).filter(NeCollectionRun.job_id == job_id).delete() + db.delete(job) + db.commit() + job_dir = (collection_data_root() / job_id).resolve() + root = collection_data_root().resolve() + if str(job_dir).startswith(str(root)) and job_dir.is_dir(): + shutil.rmtree(job_dir, ignore_errors=True) + return {"ok": True} + + +def _safe_archive_part(text: str, fallback: str = "device") -> str: + s = re.sub(r'[<>:"/\\|?*]', "_", str(text or "").strip()) + return (s[:80] or fallback).strip("._") or fallback + + +def build_collection_job_zip(db: Session, job_id: str) -> tuple[str, bytes]: + job = db.get(NeCollectionJob, job_id) + if not job: + raise HTTPException(status_code=404, detail="collection_job_not_found") + runs = ( + db.query(NeCollectionRun) + .filter(NeCollectionRun.job_id == job_id) + .order_by(NeCollectionRun.ne_name.asc()) + .all() + ) + files: list[tuple[str, Path]] = [] + used_names: set[str] = set() + for run in runs: + rel = str(run.output_rel_path or "").strip() + if not rel: + continue + path = resolve_run_output_path(rel) + arcname = path.name + if arcname in used_names: + arcname = f"{str(run.id)[:8]}_{path.name}" + used_names.add(arcname) + files.append((arcname, path)) + if not files: + raise HTTPException(status_code=404, detail="collection_outputs_not_found") + buf = io.BytesIO() + with zipfile.ZipFile(buf, mode="w", compression=zipfile.ZIP_DEFLATED) as zf: + for arcname, path in files: + zf.write(path, arcname=arcname) + title = _safe_archive_part(str(job.title or "collect"), "collect") + zip_name = f"{title}_{job_id[:8]}.zip" + return zip_name, buf.getvalue() + + +def resolve_run_output_path(rel_path: str) -> Path: + rel = str(rel_path or "").strip().replace("\\", "/") + if not rel or ".." in rel.split("/"): + raise HTTPException(status_code=400, detail="invalid_output_path") + full = (collection_data_root() / rel).resolve() + root = collection_data_root() + if not str(full).startswith(str(root)): + raise HTTPException(status_code=400, detail="invalid_output_path") + if not full.is_file(): + raise HTTPException(status_code=404, detail="output_file_not_found") + return full diff --git a/netx_api/config.py b/netx_api/config.py index 010f94a..fc1baaf 100644 --- a/netx_api/config.py +++ b/netx_api/config.py @@ -57,6 +57,16 @@ class Settings(BaseSettings): ume_ne_path: str = "/restconf/data/zte-resources-module:network-elements" ume_alarms_path: str = "/restconf/data/zte-alarms:alarms/alarm-list" ume_sync_alarms_history_every_hours: int = 24 + # Managed NE credentials (Fernet key; generate with cryptography.fernet.Fernet.generate_key()) + credential_secret_key: str = "" + ne_connect_max_workers: int = 5 + ne_connect_timeout_sec: int = 30 + ne_collect_max_workers: int = 5 + ne_collect_read_timeout_sec: int = 120 + ne_collect_stale_run_sec: int = 900 + ne_collect_pending_stale_sec: int = 180 + ne_collect_run_timeout_cap_sec: int = 600 + ne_collection_data_dir: str = "data/ne_collections" settings = Settings() diff --git a/netx_api/db.py b/netx_api/db.py index d00422a..4e4f544 100644 --- a/netx_api/db.py +++ b/netx_api/db.py @@ -12,3 +12,11 @@ class Base(DeclarativeBase): engine = create_engine(settings.database_url, future=True, pool_pre_ping=True) SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False, expire_on_commit=False) + + +def get_db(): + db = SessionLocal() + try: + yield db + finally: + db.close() diff --git a/netx_api/device_types.py b/netx_api/device_types.py new file mode 100644 index 0000000..f9e2baa --- /dev/null +++ b/netx_api/device_types.py @@ -0,0 +1,45 @@ +"""Netmiko device_type values supported for managed network elements.""" + +from __future__ import annotations + +SUPPORTED_DEVICE_TYPES: tuple[str, ...] = ( + "zte_zxros", + "alcatel_aos", + "alcatel_sros", + "cisco_asa", + "cisco_ftd", + "cisco_ios", + "cisco_nxos", + "cisco_s200", + "cisco_s300", + "cisco_tp", + "cisco_viptela", + "cisco_wlc", + "cisco_xe", + "cisco_xr", + "ericsson_ipos", + "ericsson_mltn63", + "ericsson_mltn66", + "huawei", + "huawei_smartax", + "huawei_olt", + "huawei_vrp", + "huawei_vrpv8", + "juniper", + "juniper_junos", + "juniper_screenos", + "mikrotik_routeros", + "mikrotik_switchos", + "nokia_sros", + "nokia_srl", + "ruijie_os", +) + +SUPPORTED_VENDORS: tuple[str, ...] = ( + "ZTE", + "Huawei", + "Cisco", + "Juniper", + "Nokia", + "Other", +) diff --git a/netx_api/main.py b/netx_api/main.py index 58fde86..21a11e9 100644 --- a/netx_api/main.py +++ b/netx_api/main.py @@ -19,13 +19,18 @@ import uvicorn from .ap_client import analyze_with_oclaw, health_with_oclaw from .config import settings -from .db import Base, SessionLocal, engine +from .db import Base, SessionLocal, engine, get_db +from .collection_router import router as collection_router +from .managed_ne_router import router as managed_ne_router from .importer import aggregate_alarms, import_alarm_excel, query_alarms from .models import ( AiAnalyzeHistory, AlarmBatch, AlarmNorm, ImportErrorRow, + ManagedNE, + NeCollectionJob, + NeCollectionRun, UmeAlarmCurrent, UmeAlarmHistory, UmeInventoryNE, @@ -72,6 +77,8 @@ from .schemas import ( ) app = FastAPI(title="netx ops tool", version="0.1.0") +app.include_router(managed_ne_router) +app.include_router(collection_router) parser_cfg = load_parser_config() _UME_CLIENT_SINGLETON = UMEClient( token_loader=lambda: load_shared_token(), @@ -418,14 +425,6 @@ def _maybe_wait_for_sync_interval( _sleep_or_until_paused(task_id, wait_s) -def get_db(): - db = SessionLocal() - try: - yield db - finally: - db.close() - - def _parse_time(text: str | None) -> datetime | None: s = str(text or "").strip() if not s: @@ -658,6 +657,17 @@ def on_startup() -> None: Base.metadata.create_all(bind=engine) _reset_runtime_pause_flags() _fail_stale_running_sync_jobs_on_startup() + db = SessionLocal() + try: + from .collection_recovery import recover_collection_jobs_on_startup + + resumed = recover_collection_jobs_on_startup(db) + if resumed: + _schedule_log.info("startup: resumed %s pending ne collection runs", resumed) + except Exception: + _schedule_log.exception("startup: ne collection job recovery failed") + finally: + db.close() # Best-effort schema evolution for new columns (no migrations framework). # Safe for Postgres (IF NOT EXISTS); ignored on failure. try: @@ -730,6 +740,13 @@ def on_startup() -> None: conn.exec_driver_sql("ALTER TABLE ume_alarms_current DROP COLUMN IF EXISTS user_label") conn.exec_driver_sql("ALTER TABLE ume_alarms_history DROP COLUMN IF EXISTS ne_name") conn.exec_driver_sql("ALTER TABLE ume_alarms_history DROP COLUMN IF EXISTS user_label") + conn.exec_driver_sql( + "ALTER TABLE ne_collection_job ADD COLUMN IF NOT EXISTS last_run_at TIMESTAMP" + ) + conn.exec_driver_sql( + "UPDATE ne_collection_job SET last_run_at = COALESCE(ended_at, started_at, created_at) " + "WHERE last_run_at IS NULL" + ) conn.exec_driver_sql("COMMENT ON TABLE ume_inventory_ne IS '网元对象详细信息'") conn.exec_driver_sql("COMMENT ON COLUMN ume_inventory_ne.ne_id IS '网元uuid'") conn.exec_driver_sql("COMMENT ON COLUMN ume_inventory_ne.ne_name IS '资源名称'") diff --git a/netx_api/managed_ne_router.py b/netx_api/managed_ne_router.py new file mode 100644 index 0000000..f691641 --- /dev/null +++ b/netx_api/managed_ne_router.py @@ -0,0 +1,92 @@ +from __future__ import annotations + +from fastapi import APIRouter, Depends, File, HTTPException, Query, UploadFile +from sqlalchemy.orm import Session + +from .db import get_db +from .device_types import SUPPORTED_DEVICE_TYPES, SUPPORTED_VENDORS +from .ne_connect import schedule_connect_tests +from .ne_crypto import credentials_configured +from .ne_schemas import ConnectTestRequest, ManagedNeCreate, ManagedNeUpdate +from .ne_service import ( + create_managed_ne, + delete_managed_ne, + get_managed_ne, + import_managed_ne, + list_managed_ne, + update_managed_ne, +) +from .models import ManagedNE + +router = APIRouter(prefix="/v1/managed-ne", tags=["managed-ne"]) + + +@router.get("") +def api_list_managed_ne( + keyword: str | None = Query(default=None), + vendor: str | None = Query(default=None), + connect_status: str | None = Query(default=None), + page: int = Query(default=1, ge=1), + page_size: int = Query(default=50, ge=1, le=500), + db: Session = Depends(get_db), +): + return list_managed_ne( + db, + keyword=keyword, + vendor=vendor, + connect_status=connect_status, + page=page, + page_size=page_size, + ) + + +@router.get("/meta/device-types") +def api_device_types(): + return {"device_types": list(SUPPORTED_DEVICE_TYPES), "vendors": list(SUPPORTED_VENDORS)} + + +@router.get("/meta/credentials-configured") +def api_credentials_configured(): + return {"configured": credentials_configured()} + + +@router.post("") +def api_create_managed_ne(body: ManagedNeCreate, db: Session = Depends(get_db)): + return create_managed_ne(db, body).model_dump() + + +@router.get("/{ne_id}") +def api_get_managed_ne(ne_id: str, db: Session = Depends(get_db)): + return get_managed_ne(db, ne_id).model_dump() + + +@router.patch("/{ne_id}") +def api_update_managed_ne(ne_id: str, body: ManagedNeUpdate, db: Session = Depends(get_db)): + return update_managed_ne(db, ne_id, body).model_dump() + + +@router.delete("/{ne_id}") +def api_delete_managed_ne(ne_id: str, db: Session = Depends(get_db)): + return delete_managed_ne(db, ne_id) + + +@router.post("/import") +async def api_import_managed_ne(file: UploadFile = File(...), db: Session = Depends(get_db)): + content = await file.read() + if not content: + raise HTTPException(status_code=400, detail="empty_file") + return import_managed_ne(db, content, file.filename or "import.xlsx").model_dump() + + +@router.post("/connect-test") +def api_connect_test(body: ConnectTestRequest, db: Session = Depends(get_db)): + ids = [str(x).strip() for x in body.ids if str(x).strip()] + if not ids: + raise HTTPException(status_code=400, detail="ids_required") + rows = db.query(ManagedNE).filter(ManagedNE.id.in_(ids)).all() + found_ids = {str(r.id) for r in rows} + missing = [x for x in ids if x not in found_ids] + if missing: + raise HTTPException(status_code=404, detail=f"managed_ne_not_found: {','.join(missing[:5])}") + submitted = schedule_connect_tests(ids) + return {"ok": True, "submitted": submitted} diff --git a/netx_api/models.py b/netx_api/models.py index b6a8e70..43397b6 100644 --- a/netx_api/models.py +++ b/netx_api/models.py @@ -221,3 +221,64 @@ class UmeTokenCache(Base): lock_owner: Mapped[str] = mapped_column(String(128), default="", index=True) lock_expires_at_epoch_s: Mapped[int] = mapped_column(Integer, default=0, index=True) updated_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow, index=True) + + +class ManagedNE(Base): + """Locally managed network element (SSH/Telnet), independent of UME inventory.""" + + __tablename__ = "managed_ne" + + id: Mapped[str] = mapped_column(String(64), primary_key=True, default=lambda: uuid4().hex) + name: Mapped[str] = mapped_column(String(256), default="", index=True) + vendor: Mapped[str] = mapped_column(String(64), default="Other", index=True) + device_type: Mapped[str] = mapped_column(String(128), default="") + ip_address: Mapped[str] = mapped_column(String(128), unique=True, index=True) + port: Mapped[int] = mapped_column(Integer, default=22) + protocol: Mapped[str] = mapped_column(String(16), default="ssh") + username: Mapped[str] = mapped_column(String(128), default="") + password_enc: Mapped[str] = mapped_column(Text, default="") + enable_secret_enc: Mapped[str] = mapped_column(Text, default="") + connect_status: Mapped[str] = mapped_column(String(32), default="unknown", index=True) + connect_message: Mapped[str] = mapped_column(String(512), default="") + connect_tested_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True) + site: Mapped[str] = mapped_column(String(256), default="") + tags: Mapped[str] = mapped_column(String(512), default="") + remark: Mapped[str] = mapped_column(String(1024), default="") + created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow) + updated_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow, index=True) + + +class NeCollectionJob(Base): + """Batch CLI collection job over managed NEs.""" + + __tablename__ = "ne_collection_job" + + id: Mapped[str] = mapped_column(String(64), primary_key=True, default=lambda: uuid4().hex) + title: Mapped[str] = mapped_column(String(256), default="") + commands: Mapped[str] = mapped_column(Text, default="") + status: Mapped[str] = mapped_column(String(32), default="pending", index=True) + ne_count: Mapped[int] = mapped_column(Integer, default=0) + success_count: Mapped[int] = mapped_column(Integer, default=0) + fail_count: Mapped[int] = mapped_column(Integer, default=0) + error_message: Mapped[str] = mapped_column(String(1024), default="") + created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow, index=True) + started_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True) + ended_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True) + last_run_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True, index=True) + + +class NeCollectionRun(Base): + """Per-NE execution within a collection job.""" + + __tablename__ = "ne_collection_run" + + id: Mapped[str] = mapped_column(String(64), primary_key=True, default=lambda: uuid4().hex) + job_id: Mapped[str] = mapped_column(String(64), index=True) + ne_id: Mapped[str] = mapped_column(String(64), index=True) + ne_name: Mapped[str] = mapped_column(String(256), default="") + ne_ip: Mapped[str] = mapped_column(String(128), default="") + status: Mapped[str] = mapped_column(String(32), default="pending", index=True) + message: Mapped[str] = mapped_column(String(1024), default="") + output_rel_path: Mapped[str] = mapped_column(String(1024), default="") + started_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True) + ended_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True) diff --git a/netx_api/ne_collect_runner.py b/netx_api/ne_collect_runner.py new file mode 100644 index 0000000..3809b2e --- /dev/null +++ b/netx_api/ne_collect_runner.py @@ -0,0 +1,174 @@ +from __future__ import annotations + +import logging +import re +from concurrent.futures import ThreadPoolExecutor, TimeoutError as FuturesTimeout +from datetime import datetime +from pathlib import Path +from typing import Any + +from netmiko import ConnectHandler + +from .collection_job_state import finalize_collection_job +from .config import settings +from .db import SessionLocal +from .models import ManagedNE, NeCollectionJob, NeCollectionRun +from .ne_collection_paths import clear_run_output_files, run_output_dir +from .ne_crypto import CredentialCryptoError +from .ne_netmiko import normalize_netmiko_device_type +from .ne_service import get_device_credentials + +_log = logging.getLogger("netx.ne.collect") +_executor: ThreadPoolExecutor | None = None + + +def _executor_pool() -> ThreadPoolExecutor: + global _executor + if _executor is None: + workers = max(1, int(settings.ne_collect_max_workers or 5)) + _executor = ThreadPoolExecutor(max_workers=workers, thread_name_prefix="ne-collect") + return _executor + + +def _safe_filename_part(text: str) -> str: + s = re.sub(r'[<>:"/\\|?*]', "_", str(text or "").strip()) + return s[:80] or "device" + + +def _collect_on_device(creds: dict[str, Any], commands: list[str]) -> str: + device_type = normalize_netmiko_device_type(creds["device_type"], creds["protocol"]) + per_cmd = int(settings.ne_collect_read_timeout_sec or 120) + dev: dict[str, Any] = { + "device_type": device_type, + "host": creds["ip_address"], + "username": creds["username"], + "password": creds["password"], + "port": int(creds["port"] or 22), + "conn_timeout": int(settings.ne_connect_timeout_sec or 30), + "auth_timeout": int(settings.ne_connect_timeout_sec or 30), + "banner_timeout": int(settings.ne_connect_timeout_sec or 30), + "session_timeout": per_cmd * max(1, len(commands)) + 60, + } + secret = str(creds.get("enable_secret") or "").strip() + if secret: + dev["secret"] = secret + chunks: list[str] = [] + with ConnectHandler(**dev) as conn: + prompt = str(conn.find_prompt() or "") + for command in commands: + ts = datetime.now().isoformat(timespec="seconds") + chunks.append(f'>>> [{ts}] {{"String":"{command}", "Match":"{prompt}", "Timeout":0}}\n') + out = conn.send_command(command_string=command, read_timeout=per_cmd) + chunks.append(str(out or "")) + chunks.append("\n") + return "".join(chunks) + + +def _collect_with_timeout(creds: dict[str, Any], commands: list[str]) -> str: + per_cmd = int(settings.ne_collect_read_timeout_sec or 120) + cap = int(settings.ne_collect_run_timeout_cap_sec or 600) + budget = min(cap, per_cmd * max(1, len(commands)) + 90) + with ThreadPoolExecutor(max_workers=1) as pool: + fut = pool.submit(_collect_on_device, creds, commands) + try: + return fut.result(timeout=budget) + except FuturesTimeout as exc: + raise TimeoutError(f"collection_timeout ({budget}s)") from exc + + +def _update_run(run_id: str, **fields: Any) -> None: + db = SessionLocal() + try: + row = db.get(NeCollectionRun, run_id) + if not row: + return + for key, val in fields.items(): + setattr(row, key, val) + db.commit() + finally: + db.close() + + +def _collection_aborted(job_id: str, run_id: str) -> bool: + db = SessionLocal() + try: + job = db.get(NeCollectionJob, job_id) + run = db.get(NeCollectionRun, run_id) + if not job or not run: + return True + if str(job.status or "") == "paused": + if str(run.status or "") == "pending": + run.status = "cancelled" + run.message = "paused" + run.ended_at = datetime.now() + db.commit() + return True + if str(run.status or "") in ("cancelled", "success", "fail"): + return True + return False + finally: + db.close() + + +def _run_single(job_id: str, run_id: str, commands: list[str]) -> None: + if _collection_aborted(job_id, run_id): + db = SessionLocal() + try: + finalize_collection_job(db, job_id) + finally: + db.close() + return + db = SessionLocal() + try: + run = db.get(NeCollectionRun, run_id) + if not run: + return + ne = db.get(ManagedNE, str(run.ne_id)) + if not ne: + _update_run(run_id, status="fail", message="managed_ne_not_found", ended_at=datetime.now()) + return + started_at = datetime.now() + _update_run(run_id, status="running", message="collecting", started_at=started_at) + try: + creds = get_device_credentials(ne) + output = _collect_with_timeout(creds, commands) + finished_at = datetime.now() + name_part = _safe_filename_part(str(run.ne_name or creds.get("name") or "ne")) + ip_part = _safe_filename_part(str(run.ne_ip or creds.get("ip_address") or "ip")) + ts = finished_at.strftime("%Y%m%d%H%M%S") + rel_dir = Path(job_id) / run_id + out_dir = run_output_dir(job_id, run_id) + clear_run_output_files(job_id, run_id) + out_dir.mkdir(parents=True, exist_ok=True) + filename = f"{name_part}-{ip_part}-{ts}.txt" + full_path = out_dir / filename + full_path.write_text(output, encoding="utf-8", errors="replace") + rel_path = str(rel_dir / filename).replace("\\", "/") + _update_run( + run_id, + status="success", + message="collected", + output_rel_path=rel_path, + ended_at=finished_at, + ) + except CredentialCryptoError as exc: + _update_run(run_id, status="fail", message=str(exc)[:1000], ended_at=datetime.now()) + except Exception as exc: + _log.exception("collection failed run=%s", run_id) + _update_run(run_id, status="fail", message=str(exc).split("\n")[0][:1000], ended_at=datetime.now()) + finally: + db.close() + db2 = SessionLocal() + try: + finalize_collection_job(db2, job_id) + finally: + db2.close() + + +def schedule_collection_runs(job_id: str, run_ids: list[str], commands: list[str]) -> int: + pool = _executor_pool() + submitted = 0 + for run_id in run_ids: + pool.submit(_run_single, job_id, run_id, list(commands)) + submitted += 1 + return submitted diff --git a/netx_api/ne_collection_paths.py b/netx_api/ne_collection_paths.py new file mode 100644 index 0000000..7bb7e3b --- /dev/null +++ b/netx_api/ne_collection_paths.py @@ -0,0 +1,26 @@ +from __future__ import annotations + +from pathlib import Path + +from .config import settings + + +def collection_data_root() -> Path: + root = Path(str(settings.ne_collection_data_dir or "data/ne_collections")) + root.mkdir(parents=True, exist_ok=True) + return root.resolve() + + +def run_output_dir(job_id: str, run_id: str) -> Path: + return collection_data_root() / job_id / run_id + + +def clear_run_output_files(job_id: str, run_id: str) -> None: + """Remove prior log files for this run so re-collection overwrites in place.""" + out_dir = run_output_dir(job_id, run_id).resolve() + root = collection_data_root() + if not str(out_dir).startswith(str(root)) or not out_dir.is_dir(): + return + for path in out_dir.iterdir(): + if path.is_file(): + path.unlink(missing_ok=True) diff --git a/netx_api/ne_connect.py b/netx_api/ne_connect.py new file mode 100644 index 0000000..91e8b9d --- /dev/null +++ b/netx_api/ne_connect.py @@ -0,0 +1,193 @@ +from __future__ import annotations + +import logging +import re +from concurrent.futures import ThreadPoolExecutor +from datetime import datetime +from typing import Any + +from netmiko import ConnectHandler + +from .config import settings +from .db import SessionLocal +from .models import ManagedNE +from .ne_crypto import CredentialCryptoError, decrypt_secret +from .ne_netmiko import normalize_netmiko_device_type +from .ne_service import get_device_credentials + +_log = logging.getLogger("netx.ne.connect") +_executor: ThreadPoolExecutor | None = None + + +def _executor_pool() -> ThreadPoolExecutor: + global _executor + if _executor is None: + workers = max(1, int(settings.ne_connect_max_workers or 5)) + _executor = ThreadPoolExecutor(max_workers=workers, thread_name_prefix="ne-connect") + return _executor + + +def hostname_probe_command(device_type: str, vendor: str) -> str | None: + """ + Per-vendor CLI to read system name (ported from legacy connect.extract_dev_command). + ZTE: rely on login prompt / empty command path. + """ + dt = str(device_type or "").lower() + v = str(vendor or "").lower() + if "huawei" in dt or v == "huawei": + return "display current-configuration | include sysname" + if "juniper" in dt or v == "juniper": + return "show system host-name" + if "cisco" in dt or v == "cisco": + return "show hostname" + return None + + +def parse_hostname_from_output( + device_type: str, + vendor: str, + output: str, + prompt: str = "", +) -> str | None: + """ + Parse device name from command output or prompt (legacy connect.extract_hostname). + """ + dt = str(device_type or "").lower() + v = str(vendor or "").lower() + text = str(output or "") + + if "huawei" in dt or v == "huawei": + m = re.search(r"sysname\s+(\S+)", text, re.IGNORECASE) + if m: + return m.group(1).strip() + + if "juniper" in dt or v == "juniper": + m = re.search(r"host-name\s+(\S+)", text, re.IGNORECASE) + if m: + return m.group(1).strip().rstrip(";") + m = re.search(r"^\s*name\s+(\S+)", text, re.IGNORECASE | re.MULTILINE) + if m: + return m.group(1).strip().rstrip(";") + + if "cisco" in dt or v == "cisco": + lines = [ln.strip() for ln in text.splitlines() if ln.strip()] + for ln in reversed(lines): + if ln.startswith("%") or "invalid" in ln.lower(): + continue + token = ln.split()[0].strip("<>[]") + if token: + return token + + if "zte" in dt or v == "zte": + lines = [ln.strip() for ln in text.splitlines() if ln.strip()] + if lines: + last = lines[-1].strip() + if last and len(last) <= 128 and not last.startswith("%"): + return last + + cleaned = _clean_prompt_hostname(prompt) + if cleaned: + return cleaned + return None + + +def _clean_prompt_hostname(prompt: str) -> str | None: + p = str(prompt or "").strip() + if not p: + return None + p = re.sub(r"[\s#>$]+\s*$", "", p).strip() + p = re.sub(r"^[<\[]|[>\]]$", "", p).strip() + if not p or p.lower() in (">", "#"): + return None + return p[:256] + + +def _probe_device(creds: dict[str, Any]) -> tuple[str, str, str | None]: + """Login via Netmiko, probe hostname, return (status, message, discovered_name).""" + device_type = normalize_netmiko_device_type(creds["device_type"], creds["protocol"]) + vendor = str(creds.get("vendor") or "") + dev: dict[str, Any] = { + "device_type": device_type, + "host": creds["ip_address"], + "username": creds["username"], + "password": creds["password"], + "port": int(creds["port"] or 22), + "conn_timeout": int(settings.ne_connect_timeout_sec or 30), + "auth_timeout": int(settings.ne_connect_timeout_sec or 30), + "banner_timeout": int(settings.ne_connect_timeout_sec or 30), + } + secret = str(creds.get("enable_secret") or "").strip() + if secret: + dev["secret"] = secret + try: + with ConnectHandler(**dev) as conn: + prompt = str(conn.find_prompt() or "") + command = hostname_probe_command(creds["device_type"], vendor) + output = "" + if command: + output = conn.send_command(command_string=command, read_timeout=30) + hostname = parse_hostname_from_output(creds["device_type"], vendor, output, prompt) + if hostname: + return "pass", f"connected: {hostname}", hostname + if command: + return "pass", "connected (hostname not parsed)", None + fallback = _clean_prompt_hostname(prompt) + if fallback: + return "pass", f"connected: {fallback}", fallback + return "pass", "connected", None + except Exception as exc: + msg = str(exc).split("\n")[0][:480] + return "fail", msg, None + + +def _update_row(ne_id: str, status: str, message: str, discovered_name: str | None = None) -> None: + db = SessionLocal() + try: + row = db.get(ManagedNE, ne_id) + if not row: + return + row.connect_status = status + row.connect_message = str(message or "")[:500] + row.connect_tested_at = datetime.utcnow() + if discovered_name: + row.name = discovered_name[:256] + row.updated_at = datetime.utcnow() + db.commit() + finally: + db.close() + + +def _run_single(ne_id: str) -> None: + db = SessionLocal() + try: + row = db.get(ManagedNE, ne_id) + if not row: + return + row.connect_status = "testing" + row.connect_message = "" + row.updated_at = datetime.utcnow() + db.commit() + try: + creds = get_device_credentials(row) + except CredentialCryptoError as exc: + _update_row(ne_id, "fail", str(exc)) + return + status, message, discovered = _probe_device(creds) + _update_row(ne_id, status, message, discovered) + except Exception as exc: + _log.exception("connect test failed for %s", ne_id) + _update_row(ne_id, "fail", str(exc)[:480]) + finally: + db.close() + + +def schedule_connect_tests(ne_ids: list[str]) -> int: + pool = _executor_pool() + submitted = 0 + for ne_id in ne_ids: + ne_id = str(ne_id or "").strip() + if not ne_id: + continue + pool.submit(_run_single, ne_id) + submitted += 1 + return submitted diff --git a/netx_api/ne_crypto.py b/netx_api/ne_crypto.py new file mode 100644 index 0000000..4ba7617 --- /dev/null +++ b/netx_api/ne_crypto.py @@ -0,0 +1,40 @@ +from __future__ import annotations + +from cryptography.fernet import Fernet, InvalidToken + +from .config import settings + + +class CredentialCryptoError(RuntimeError): + pass + + +def _fernet() -> Fernet: + key = str(settings.credential_secret_key or "").strip() + if not key: + raise CredentialCryptoError("credential_secret_key_not_configured") + try: + return Fernet(key.encode("ascii")) + except Exception as exc: + raise CredentialCryptoError("credential_secret_key_invalid") from exc + + +def encrypt_secret(value: str) -> str: + plain = str(value or "") + if not plain: + return "" + return _fernet().encrypt(plain.encode("utf-8")).decode("ascii") + + +def decrypt_secret(value: str) -> str: + enc = str(value or "").strip() + if not enc: + return "" + try: + return _fernet().decrypt(enc.encode("ascii")).decode("utf-8") + except InvalidToken as exc: + raise CredentialCryptoError("credential_decrypt_failed") from exc + + +def credentials_configured() -> bool: + return bool(str(settings.credential_secret_key or "").strip()) diff --git a/netx_api/ne_netmiko.py b/netx_api/ne_netmiko.py new file mode 100644 index 0000000..ec6b97b --- /dev/null +++ b/netx_api/ne_netmiko.py @@ -0,0 +1,17 @@ +"""Shared Netmiko helpers for managed NE connect/collect.""" + +from __future__ import annotations + + +def normalize_netmiko_device_type(device_type: str, protocol: str) -> str: + dt = str(device_type or "").strip() + proto = str(protocol or "ssh").strip().lower() + if "zte" in dt.lower(): + if dt == "zte": + return f"zte_zxros_{proto}" + if "telnet" not in dt and "ssh" not in dt: + return f"{dt}_{proto}" + return dt + if "telnet" not in dt and "ssh" not in dt: + return f"{dt}_{proto}" + return dt diff --git a/netx_api/ne_schemas.py b/netx_api/ne_schemas.py new file mode 100644 index 0000000..e3a6789 --- /dev/null +++ b/netx_api/ne_schemas.py @@ -0,0 +1,93 @@ +from __future__ import annotations + +from datetime import datetime +from typing import Literal + +from pydantic import BaseModel, Field, field_validator + +from .device_types import SUPPORTED_VENDORS + +ConnectStatus = Literal["unknown", "testing", "pass", "fail"] + + +class ManagedNeCreate(BaseModel): + name: str = "" + vendor: str + device_type: str + ip_address: str + port: int = 22 + protocol: str = "ssh" + username: str + password: str + tags: str = "" + remark: str = "" + + @field_validator("vendor") + @classmethod + def normalize_vendor(cls, v: str) -> str: + raw = str(v or "").strip() + if not raw: + raise ValueError("vendor_required") + for item in SUPPORTED_VENDORS: + if item.lower() == raw.lower(): + return item + return "Other" + + +class ManagedNeUpdate(BaseModel): + name: str | None = None + vendor: str | None = None + device_type: str | None = None + ip_address: str | None = None + port: int | None = None + protocol: str | None = None + username: str | None = None + password: str | None = None + tags: str | None = None + remark: str | None = None + + @field_validator("vendor") + @classmethod + def normalize_vendor_update(cls, v: str | None) -> str | None: + if v is None: + return None + raw = str(v).strip() + if not raw: + raise ValueError("vendor_required") + for item in SUPPORTED_VENDORS: + if item.lower() == raw.lower(): + return item + return "Other" + + +class ManagedNeOut(BaseModel): + id: str + name: str + vendor: str + device_type: str + ip_address: str + port: int + protocol: str + username: str + connect_status: ConnectStatus + connect_message: str + connect_tested_at: datetime | None + tags: str + remark: str + created_at: datetime + updated_at: datetime + + +class ConnectTestRequest(BaseModel): + ids: list[str] = Field(min_length=1) + + +class ImportFailure(BaseModel): + row: int + reason: str + + +class ImportResult(BaseModel): + inserted: int + updated: int + failed: list[ImportFailure] diff --git a/netx_api/ne_service.py b/netx_api/ne_service.py new file mode 100644 index 0000000..65d1c18 --- /dev/null +++ b/netx_api/ne_service.py @@ -0,0 +1,280 @@ +from __future__ import annotations + +from datetime import datetime +from io import BytesIO +from typing import Any + +import pandas as pd +from fastapi import HTTPException +from sqlalchemy.orm import Session + +from .device_types import SUPPORTED_DEVICE_TYPES, SUPPORTED_VENDORS +from .models import ManagedNE +from .ne_crypto import CredentialCryptoError, credentials_configured, decrypt_secret, encrypt_secret +from .ne_schemas import ImportFailure, ImportResult, ManagedNeCreate, ManagedNeOut, ManagedNeUpdate + +IMPORT_COLUMNS = ( + "device_type", + "ip", + "username", + "password", + "port", + "protocol", + "name", + "vendor", +) + + +def _now() -> datetime: + return datetime.utcnow() + + +def _require_crypto() -> None: + if not credentials_configured(): + raise HTTPException(status_code=503, detail="credential_secret_key_not_configured") + + +def _normalize_ip(ip: str) -> str: + return str(ip or "").strip() + + +def _normalize_protocol(protocol: str) -> str: + p = str(protocol or "ssh").strip().lower() + return p if p in ("ssh", "telnet") else "ssh" + + +def row_to_out(row: ManagedNE) -> ManagedNeOut: + status = str(row.connect_status or "unknown") + if status not in ("unknown", "testing", "pass", "fail"): + status = "unknown" + return ManagedNeOut( + id=str(row.id), + name=str(row.name or ""), + vendor=str(row.vendor or "Other"), + device_type=str(row.device_type or ""), + ip_address=str(row.ip_address or ""), + port=int(row.port or 22), + protocol=str(row.protocol or "ssh"), + username=str(row.username or ""), + connect_status=status, # type: ignore[arg-type] + connect_message=str(row.connect_message or "")[:500], + connect_tested_at=row.connect_tested_at, + tags=str(row.tags or ""), + remark=str(row.remark or ""), + created_at=row.created_at, + updated_at=row.updated_at, + ) + + +def list_managed_ne( + db: Session, + *, + keyword: str | None = None, + vendor: str | None = None, + connect_status: str | None = None, + page: int = 1, + page_size: int = 50, +) -> dict[str, Any]: + stmt = db.query(ManagedNE) + kw = str(keyword or "").strip() + if kw: + stmt = stmt.filter( + ManagedNE.name.contains(kw) + | ManagedNE.ip_address.contains(kw) + | ManagedNE.username.contains(kw) + | ManagedNE.tags.contains(kw) + ) + v = str(vendor or "").strip() + if v: + stmt = stmt.filter(ManagedNE.vendor == v) + cs = str(connect_status or "").strip() + if cs: + stmt = stmt.filter(ManagedNE.connect_status == cs) + total = int(stmt.count()) + rows = ( + stmt.order_by(ManagedNE.updated_at.desc()) + .offset((page - 1) * page_size) + .limit(page_size) + .all() + ) + return { + "total": total, + "page": page, + "page_size": page_size, + "items": [row_to_out(x).model_dump() for x in rows], + } + + +def get_managed_ne(db: Session, ne_id: str) -> ManagedNeOut: + row = db.get(ManagedNE, ne_id) + if not row: + raise HTTPException(status_code=404, detail="managed_ne_not_found") + return row_to_out(row) + + +def create_managed_ne(db: Session, body: ManagedNeCreate) -> ManagedNeOut: + _require_crypto() + ip = _normalize_ip(body.ip_address) + if not ip: + raise HTTPException(status_code=400, detail="ip_address_required") + if body.device_type not in SUPPORTED_DEVICE_TYPES: + raise HTTPException(status_code=400, detail="unsupported_device_type") + existing = db.query(ManagedNE).filter(ManagedNE.ip_address == ip).first() + if existing: + raise HTTPException(status_code=400, detail="ip_address_exists") + now = _now() + row = ManagedNE( + name=str(body.name or "").strip() or ip, + vendor=body.vendor, + device_type=body.device_type, + ip_address=ip, + port=int(body.port or 22), + protocol=_normalize_protocol(body.protocol), + username=str(body.username or "").strip(), + password_enc=encrypt_secret(body.password), + enable_secret_enc="", + connect_status="unknown", + tags=str(body.tags or "").strip(), + remark=str(body.remark or "").strip(), + created_at=now, + updated_at=now, + ) + db.add(row) + db.commit() + db.refresh(row) + return row_to_out(row) + + +def update_managed_ne(db: Session, ne_id: str, body: ManagedNeUpdate) -> ManagedNeOut: + row = db.get(ManagedNE, ne_id) + if not row: + raise HTTPException(status_code=404, detail="managed_ne_not_found") + data = body.model_dump(exclude_unset=True) + if "ip_address" in data: + ip = _normalize_ip(data["ip_address"]) + if not ip: + raise HTTPException(status_code=400, detail="ip_address_required") + other = db.query(ManagedNE).filter(ManagedNE.ip_address == ip, ManagedNE.id != ne_id).first() + if other: + raise HTTPException(status_code=400, detail="ip_address_exists") + row.ip_address = ip + if "device_type" in data: + if data["device_type"] not in SUPPORTED_DEVICE_TYPES: + raise HTTPException(status_code=400, detail="unsupported_device_type") + row.device_type = data["device_type"] + if "vendor" in data: + v = str(data["vendor"] or "").strip() + row.vendor = v if v in SUPPORTED_VENDORS else "Other" + if "name" in data: + row.name = str(data["name"] or "").strip() + if "port" in data and data["port"] is not None: + row.port = int(data["port"]) + if "protocol" in data and data["protocol"] is not None: + row.protocol = _normalize_protocol(data["protocol"]) + if "username" in data and data["username"] is not None: + row.username = str(data["username"]).strip() + if "tags" in data and data["tags"] is not None: + row.tags = str(data["tags"]).strip() + if "remark" in data and data["remark"] is not None: + row.remark = str(data["remark"]).strip() + if "password" in data and data["password"]: + _require_crypto() + row.password_enc = encrypt_secret(str(data["password"])) + row.updated_at = _now() + db.commit() + db.refresh(row) + return row_to_out(row) + + +def delete_managed_ne(db: Session, ne_id: str) -> dict[str, bool]: + row = db.get(ManagedNE, ne_id) + if not row: + raise HTTPException(status_code=404, detail="managed_ne_not_found") + db.delete(row) + db.commit() + return {"ok": True} + + +def import_managed_ne(db: Session, content: bytes, filename: str) -> ImportResult: + _require_crypto() + name = str(filename or "").lower() + try: + if name.endswith(".csv"): + df = pd.read_csv(BytesIO(content)) + else: + df = pd.read_excel(BytesIO(content)) + except Exception as exc: + raise HTTPException(status_code=400, detail=f"import_parse_failed: {exc}") from exc + df.columns = [str(c).strip().lower() for c in df.columns] + missing = [c for c in IMPORT_COLUMNS if c not in df.columns] + if missing: + raise HTTPException(status_code=400, detail=f"import_missing_columns: {','.join(missing)}") + inserted = 0 + updated = 0 + failed: list[ImportFailure] = [] + for idx, row in df.iterrows(): + row_no = int(idx) + 2 + try: + ip = _normalize_ip(str(row.get("ip", ""))) + if not ip: + failed.append(ImportFailure(row=row_no, reason="ip_required")) + continue + device_type = str(row.get("device_type", "")).strip() + if device_type not in SUPPORTED_DEVICE_TYPES: + failed.append(ImportFailure(row=row_no, reason="unsupported_device_type")) + continue + username = str(row.get("username", "")).strip() + password = str(row.get("password", "")).strip() + if not username or not password: + failed.append(ImportFailure(row=row_no, reason="username_password_required")) + continue + port_raw = row.get("port", 22) + try: + port = int(port_raw) + except (TypeError, ValueError): + port = 22 + protocol = _normalize_protocol(str(row.get("protocol", "ssh"))) + display_name = str(row.get("name", "") or "").strip() or ip + vendor_raw = str(row.get("vendor", "") or "Other").strip() + vendor = "Other" + for v in SUPPORTED_VENDORS: + if v.lower() == vendor_raw.lower(): + vendor = v + break + existing = db.query(ManagedNE).filter(ManagedNE.ip_address == ip).first() + now = _now() + if existing is None: + existing = ManagedNE(ip_address=ip, created_at=now) + db.add(existing) + inserted += 1 + else: + updated += 1 + existing.name = display_name + existing.vendor = vendor + existing.device_type = device_type + existing.port = port + existing.protocol = protocol + existing.username = username + existing.password_enc = encrypt_secret(password) + existing.updated_at = now + except CredentialCryptoError as exc: + failed.append(ImportFailure(row=row_no, reason=str(exc))) + except Exception as exc: + failed.append(ImportFailure(row=row_no, reason=str(exc)[:200])) + db.commit() + return ImportResult(inserted=inserted, updated=updated, failed=failed) + + +def get_device_credentials(row: ManagedNE) -> dict[str, Any]: + return { + "id": str(row.id), + "vendor": str(row.vendor or ""), + "device_type": str(row.device_type or ""), + "ip_address": str(row.ip_address or ""), + "port": int(row.port or 22), + "protocol": str(row.protocol or "ssh"), + "username": str(row.username or ""), + "password": decrypt_secret(row.password_enc), + "enable_secret": decrypt_secret(row.enable_secret_enc), + "name": str(row.name or ""), + } diff --git a/requirements.txt b/requirements.txt index a0917ba..0b254b0 100644 --- a/requirements.txt +++ b/requirements.txt @@ -10,3 +10,5 @@ pydantic>=2.8.0 pydantic-settings>=2.3.0 python-multipart>=0.0.9 websocket-client>=1.8.0 +cryptography>=42.0.0 +netmiko>=4.3.0 diff --git a/tests/test_collection_job_state.py b/tests/test_collection_job_state.py new file mode 100644 index 0000000..6a2d1d6 --- /dev/null +++ b/tests/test_collection_job_state.py @@ -0,0 +1,35 @@ +from __future__ import annotations + +import unittest +from datetime import datetime +from unittest.mock import MagicMock + +from netx_api.collection_job_state import finalize_collection_job + + +class CollectionJobFinalizeTests(unittest.TestCase): + def test_paused_job_stays_paused_when_all_runs_terminal(self): + job = MagicMock() + job.status = "paused" + job.ended_at = None + job.success_count = 0 + job.fail_count = 0 + + run_ok = MagicMock(status="success") + run_cancel = MagicMock(status="cancelled") + + db = MagicMock() + db.query.return_value.filter.return_value.all.return_value = [run_ok, run_cancel] + db.get.return_value = job + + finalize_collection_job(db, "job-1") + + self.assertEqual(job.status, "paused") + self.assertEqual(job.success_count, 1) + self.assertEqual(job.fail_count, 1) + self.assertIsNotNone(job.ended_at) + db.commit.assert_called_once() + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_managed_ne.py b/tests/test_managed_ne.py new file mode 100644 index 0000000..8962db3 --- /dev/null +++ b/tests/test_managed_ne.py @@ -0,0 +1,175 @@ +from __future__ import annotations + +import unittest +from unittest.mock import patch + +from cryptography.fernet import Fernet +from fastapi.testclient import TestClient +from sqlalchemy import create_engine +from sqlalchemy.orm import sessionmaker +from sqlalchemy.pool import StaticPool + +from netx_api.config import settings +from netx_api.db import Base, get_db +from netx_api.main import app +from netx_api.models import ManagedNE # noqa: F401 — register table on Base +from netx_api.ne_connect import hostname_probe_command, parse_hostname_from_output +from netx_api.ne_crypto import decrypt_secret, encrypt_secret +from netx_api.ne_service import create_managed_ne, import_managed_ne +from netx_api.ne_schemas import ManagedNeCreate + + +class ManagedNeHostnameParseTests(unittest.TestCase): + def test_huawei_sysname(self): + out = " sysname PE-CORE-01\n" + self.assertEqual(parse_hostname_from_output("huawei", "Huawei", out), "PE-CORE-01") + + def test_juniper_hostname(self): + out = "host-name ROUTER-A;\nname ROUTER-A\n" + self.assertEqual(parse_hostname_from_output("juniper", "Juniper", out), "ROUTER-A") + + def test_zte_last_line(self): + out = "line1\nZXR10-PE1#" + self.assertEqual(parse_hostname_from_output("zte_zxros", "ZTE", out), "ZXR10-PE1#") + + def test_probe_commands(self): + self.assertIn("sysname", hostname_probe_command("huawei", "Huawei") or "") + self.assertEqual(hostname_probe_command("zte_zxros", "ZTE"), None) + + +class ManagedNeCryptoTests(unittest.TestCase): + def setUp(self): + self._orig = settings.credential_secret_key + settings.credential_secret_key = Fernet.generate_key().decode() + + def tearDown(self): + settings.credential_secret_key = self._orig + + def test_encrypt_roundtrip(self): + enc = encrypt_secret("secret-pass") + self.assertEqual(decrypt_secret(enc), "secret-pass") + + +class ManagedNeApiTests(unittest.TestCase): + def setUp(self): + self._orig_key = settings.credential_secret_key + settings.credential_secret_key = Fernet.generate_key().decode() + self.engine = create_engine( + "sqlite://", + connect_args={"check_same_thread": False}, + poolclass=StaticPool, + ) + ManagedNE.__table__.create(bind=self.engine, checkfirst=True) + self.Session = sessionmaker(bind=self.engine, autoflush=False, autocommit=False) + + def override_get_db(): + db = self.Session() + try: + yield db + finally: + db.close() + + app.dependency_overrides[get_db] = override_get_db + self._session_patch = patch("netx_api.ne_connect.SessionLocal", self.Session) + self._session_patch.start() + self.client = TestClient(app) + + def tearDown(self): + app.dependency_overrides.clear() + self._session_patch.stop() + settings.credential_secret_key = self._orig_key + + def test_crud_flow(self): + r = self.client.post( + "/v1/managed-ne", + json={ + "name": "PE-01", + "vendor": "ZTE", + "device_type": "zte_zxros", + "ip_address": "10.0.0.1", + "port": 22, + "protocol": "ssh", + "username": "admin", + "password": "pass123", + }, + ) + self.assertEqual(r.status_code, 200, r.text) + ne_id = r.json()["id"] + self.assertNotIn("password", r.json()) + + r2 = self.client.get(f"/v1/managed-ne/{ne_id}") + self.assertEqual(r2.status_code, 200) + self.assertEqual(r2.json()["ip_address"], "10.0.0.1") + + r3 = self.client.patch(f"/v1/managed-ne/{ne_id}", json={"name": "PE-01-upd"}) + self.assertEqual(r3.status_code, 200) + self.assertEqual(r3.json()["name"], "PE-01-upd") + + r4 = self.client.get("/v1/managed-ne", params={"keyword": "10.0.0"}) + self.assertEqual(r4.status_code, 200) + self.assertEqual(r4.json()["total"], 1) + + r5 = self.client.delete(f"/v1/managed-ne/{ne_id}") + self.assertEqual(r5.status_code, 200) + + def test_create_without_crypto_key(self): + settings.credential_secret_key = "" + r = self.client.post( + "/v1/managed-ne", + json={ + "vendor": "ZTE", + "device_type": "zte_zxros", + "ip_address": "10.0.0.2", + "username": "admin", + "password": "x", + }, + ) + self.assertEqual(r.status_code, 503, r.text) + + @patch("netx_api.ne_connect._probe_device", return_value=("pass", "ok", None)) + def test_connect_test(self, _mock_probe): + r = self.client.post( + "/v1/managed-ne", + json={ + "vendor": "Huawei", + "device_type": "huawei", + "ip_address": "10.0.0.3", + "username": "admin", + "password": "pass", + }, + ) + ne_id = r.json()["id"] + r2 = self.client.post("/v1/managed-ne/connect-test", json={"ids": [ne_id]}) + self.assertEqual(r2.status_code, 200) + self.assertEqual(r2.json()["submitted"], 1) + + +class ManagedNeServiceImportTests(unittest.TestCase): + def setUp(self): + self._orig = settings.credential_secret_key + settings.credential_secret_key = Fernet.generate_key().decode() + self.engine = create_engine( + "sqlite://", + connect_args={"check_same_thread": False}, + poolclass=StaticPool, + ) + ManagedNE.__table__.create(bind=self.engine, checkfirst=True) + self.db = sessionmaker(bind=self.engine)() + + def tearDown(self): + self.db.close() + settings.credential_secret_key = self._orig + + def test_csv_import(self): + csv = ( + "device_type,ip,username,password,port,protocol,name,vendor\n" + "zte_zxros,10.1.1.1,u1,p1,22,ssh,NE-A,ZTE\n" + ).encode("utf-8") + result = import_managed_ne(self.db, csv, "devices.csv") + self.assertEqual(result.inserted, 1) + self.assertEqual(result.updated, 0) + self.assertEqual(len(result.failed), 0) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_ne_collection.py b/tests/test_ne_collection.py new file mode 100644 index 0000000..df80e64 --- /dev/null +++ b/tests/test_ne_collection.py @@ -0,0 +1,15 @@ +from __future__ import annotations + +import unittest + +from netx_api.collection_service import _parse_commands + + +class NeCollectionParseTests(unittest.TestCase): + def test_parse_commands_skips_comments(self): + cmds = _parse_commands("display version\n# comment\ndisplay ip int brief\n") + self.assertEqual(cmds, ["display version", "display ip int brief"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/web/WEB.md b/web/WEB.md index 5029605..d0eb4f7 100644 --- a/web/WEB.md +++ b/web/WEB.md @@ -33,6 +33,8 @@ src/ |------|------|----------| | `/` | 工作台 | — | | `/ume` | UME 同步 | `ume` | +| `/ne` | 网元管理 | `managed-ne` | +| `/collect` | 批量采集 | `collect` | **新增模块只需改 `config/modules.ts`：** @@ -60,6 +62,19 @@ src/ - 失效缓存用 prefix key（如 `queryKeys.umeSyncStatusAll`） - 顶栏连接状态：`App` 轮询 `GET /v1/integrations/status`（5s），展示 **netx api** 与 **oclaw bridge**（含延迟 / 错误类型） +## 网元管理（独立于 UME） + +- API：`/v1/managed-ne/*`（CRUD、导入、连通性测试） +- 环境变量：`NETX_CREDENTIAL_SECRET_KEY`（Fernet，用于加密存储 SSH 密码） +- 导入列：`device_type,ip,username,password,port,protocol,name,vendor` +- 连通性测试成功后会**始终**用探测到的设备名覆盖「名称」 + +## 批量采集 + +- API：`/v1/ne-collections/*`（仅 `connect_status=pass` 的网元可参与） +- 采集日志目录：`NETX_NE_COLLECTION_DATA_DIR`（默认 `data/ne_collections`） +- 命令每行一条，`#` 为注释；输出格式与旧版 NetX 采集 `.txt` 一致 + ## Toast - 使用 `ToastProvider`（`main.tsx`）+ `useToast()` diff --git a/web/src/App.tsx b/web/src/App.tsx index 64fe603..a102fd0 100644 --- a/web/src/App.tsx +++ b/web/src/App.tsx @@ -3,6 +3,8 @@ import { useQuery } from "@tanstack/react-query"; import { AppLayout } from "./layout/AppLayout"; import { queryKeys } from "./constants/queryKeys"; import { WorkbenchPage } from "./pages/WorkbenchPage"; +import { CollectPage } from "./pages/CollectPage"; +import { NePage } from "./pages/NePage"; import { UmePage } from "./pages/UmePage"; import { fetchIntegrationStatus } from "./services/api"; @@ -40,6 +42,8 @@ function App() { } /> } /> } /> + } /> + } /> } /> diff --git a/web/src/config/modules.ts b/web/src/config/modules.ts index 174399b..2e8ff16 100644 --- a/web/src/config/modules.ts +++ b/web/src/config/modules.ts @@ -3,7 +3,7 @@ */ export type ModuleIconTone = "blue" | "green" | "amber"; -export type WorkbenchSection = "monitoring"; +export type WorkbenchSection = "monitoring" | "operations"; export type ModuleDefinition = { moduleId: string; @@ -25,6 +25,24 @@ export const MODULES: readonly ModuleDefinition[] = [ iconTone: "blue", titleKey: "layout.titleUme", }, + { + moduleId: "managed-ne", + path: "/ne", + section: "operations", + labelKey: "workbench.cards.managedNe", + descKey: "workbench.cards.managedNeDesc", + iconTone: "green", + titleKey: "layout.titleManagedNe", + }, + { + moduleId: "collect", + path: "/collect", + section: "operations", + labelKey: "workbench.cards.collect", + descKey: "workbench.cards.collectDesc", + iconTone: "amber", + titleKey: "layout.titleCollect", + }, ] as const; export function getModuleById(moduleId: string): ModuleDefinition | undefined { diff --git a/web/src/constants/queryKeys.ts b/web/src/constants/queryKeys.ts index f1b0732..9318727 100644 --- a/web/src/constants/queryKeys.ts +++ b/web/src/constants/queryKeys.ts @@ -10,6 +10,18 @@ export const queryKeys = { umeCurrentAlarmsAll: ["umeCurrentAlarms"] as const, umeSyncStatus: (page: number, pageSize: number) => ["umeSyncStatus", page, pageSize] as const, umeNE: (keyword: string, page: number, pageSize: number) => ["umeNE", keyword, page, pageSize] as const, + managedNeMeta: ["managedNeMeta"] as const, + managedNeAll: ["managedNe"] as const, + managedNe: (keyword: string, vendor: string, connectStatus: string, page: number, pageSize: number) => + ["managedNe", keyword, vendor, connectStatus, page, pageSize] as const, + collectionEligibleNeAll: ["collectionEligibleNe"] as const, + collectionEligibleNe: (page: number) => ["collectionEligibleNe", page] as const, + neCollectionsAll: ["neCollections"] as const, + neCollections: (page: number) => ["neCollections", page] as const, + neCollectionDetail: (jobId: string) => ["neCollection", jobId] as const, + neCollectionRunsAll: ["neCollectionRuns"] as const, + neCollectionRuns: (jobId: string, page: number, status: string, keyword: string) => + ["neCollectionRuns", jobId, page, status, keyword] as const, umeCurrentAlarms: ( severity: string, cleared: string, diff --git a/web/src/i18n/en.ts b/web/src/i18n/en.ts index 6fb5f86..bfa133b 100644 --- a/web/src/i18n/en.ts +++ b/web/src/i18n/en.ts @@ -18,14 +18,21 @@ const en = { openInNewTab: "Open in new tab", openModule: "Open or switch to this module tab", monitoring: "Monitoring", + operations: "Operations", cards: { umeSync: "UME Sync", umeSyncDesc: "UME alarms, subscription & inventory", + managedNe: "NE Management", + managedNeDesc: "Multi-vendor SSH/Telnet NEs & connectivity test", + collect: "Batch Collect", + collectDesc: "Run custom CLI commands on connected NEs and save logs", }, }, layout: { title: "UME Integration", titleUme: "UME Sync", + titleManagedNe: "NE Management", + titleCollect: "Batch Collect", navUme: "UME", netxApi: "netx api", oclawBridge: "oclaw bridge", @@ -34,6 +41,112 @@ const en = { langZh: "中文", langEn: "English", }, + collect: { + eligible: { + title: "Eligible NEs (connectivity passed)", + hint: "Only NEs with connect_status=pass. Run connectivity test in NE Management first.", + empty: "No NEs with passed connectivity test.", + }, + form: { + title: "Collection setup", + jobTitle: "Job title", + jobTitlePh: "Auto-generated if empty", + commands: "Commands (one per line, # for comments)", + commandsPh: "display version\ndisplay current-configuration", + commandsHint: "Commands run in order on each device; output is saved as a text log (legacy NetX format).", + meta: "{{ne}} NE(s) selected · {{cmd}} command(s)", + start: "Start collection", + starting: "Submitting…", + }, + started: "Collection job started: {{id}}", + paused: "Job paused", + restarted: "Collection restarted", + deleted: "Job deleted", + nothingToRetry: "No failed devices to retry", + confirmDelete: "Delete this collection job and its log files?", + jobs: { + title: "Collection jobs", + col: { + title: "Job", + status: "Status", + progress: "Progress", + created: "Created", + lastRun: "Last run", + actions: "Actions", + }, + ok: "ok", + fail: "fail", + expand: "Details", + collapse: "Hide", + pause: "Pause", + restart: "Restart", + downloadResults: "Download results", + delete: "Delete", + autoRefresh: "Auto-refresh every 2s while jobs are running", + runsInProgress: "Some devices are still being collected…", + }, + runs: { + status: "Status", + message: "Message", + download: "Log", + downloadFile: "Download", + filterStatus: "Status filter", + filterDevice: "Device filter", + deviceKeywordPh: "Name / IP", + allStatus: "All statuses", + }, + }, + managedNe: { + title: "Network Elements", + credsNotConfigured: + "NETX_CREDENTIAL_SECRET_KEY is not set. Configure a Fernet key in .env and restart the API to store passwords.", + add: "Add NE", + importBtn: "Import", + importing: "Importing…", + edit: "Edit", + delete: "Delete", + confirmDelete: "Delete this network element?", + keywordPh: "name / IP / user / tags", + allVendors: "All vendors", + allConnectStatus: "All connect status", + col: { + name: "Name", + vendor: "Vendor", + deviceType: "Device type", + ip: "Management IP", + port: "Port", + protocol: "Protocol", + user: "Username", + password: "Password", + tags: "Tags", + remark: "Remark", + connect: "Connect", + testedAt: "Tested at", + actions: "Actions", + }, + connect: { + run: "Connectivity test", + running: "Testing…", + submitted: "Submitted tests for {{n}} device(s)", + }, + importResult: { + done: "Import done: {{inserted}} inserted, {{updated}} updated, {{failed}} failed row(s)", + }, + form: { + requiredHint: "Fields marked with * are required", + nameConnectHint: "After a successful connectivity test, name is replaced with the device hostname", + createTitle: "Add network element", + editTitle: "Edit network element", + save: "Save", + saving: "Saving…", + cancel: "Cancel", + created: "Network element created", + updated: "Network element updated", + deleted: "Network element deleted", + passwordRequired: "Password is required", + passwordOptional: "leave blank to keep unchanged", + }, + }, ume: { token: { title: "UME Token", diff --git a/web/src/i18n/zh.ts b/web/src/i18n/zh.ts index a83950b..edb4437 100644 --- a/web/src/i18n/zh.ts +++ b/web/src/i18n/zh.ts @@ -18,14 +18,21 @@ const zh = { openInNewTab: "在新标签页中打开", openModule: "打开或切换到该模块页签", monitoring: "监控", + operations: "运维", cards: { umeSync: "UME同步", umeSyncDesc: "UME 告警同步、订阅与清单", + managedNe: "网元管理", + managedNeDesc: "多厂商 SSH/Telnet 网元与连通性测试", + collect: "批量采集", + collectDesc: "对已连通网元执行定制 CLI 命令并保存日志", }, }, layout: { title: "UME 对接", titleUme: "UME同步", + titleManagedNe: "网元管理", + titleCollect: "批量采集", navUme: "UME 对接", netxApi: "netx api", oclawBridge: "oclaw bridge", @@ -34,6 +41,111 @@ const zh = { langZh: "中文", langEn: "English", }, + collect: { + eligible: { + title: "可选网元（连通性已通过）", + hint: "仅展示 connect_status=pass 的网元。请先在「网元管理」完成连通性测试。", + empty: "暂无已通过连通性测试的网元。", + }, + form: { + title: "采集配置", + jobTitle: "任务名称", + jobTitlePh: "留空则自动生成", + commands: "采集命令（每行一条，# 开头为注释）", + commandsPh: "display version\ndisplay current-configuration", + commandsHint: "命令将按顺序在每台设备上执行，输出合并为一份文本日志（格式与旧版 NetX 采集一致）。", + meta: "已选 {{ne}} 台网元 · {{cmd}} 条命令", + start: "开始采集", + starting: "提交中…", + }, + started: "采集任务已启动：{{id}}", + paused: "任务已暂停", + restarted: "已重新开始采集", + deleted: "任务已删除", + nothingToRetry: "没有可重试的失败网元", + confirmDelete: "确定删除该采集任务？相关日志文件将一并删除。", + jobs: { + title: "采集任务", + col: { + title: "任务", + status: "状态", + progress: "进度", + created: "创建时间", + lastRun: "最近执行", + actions: "操作", + }, + ok: "成功", + fail: "失败", + expand: "详情", + collapse: "收起", + pause: "暂停", + restart: "重新开始", + downloadResults: "下载结果", + delete: "删除", + autoRefresh: "任务进行中，每 2 秒自动刷新", + runsInProgress: "部分网元仍在采集中，请稍候…", + }, + runs: { + status: "状态", + message: "说明", + download: "日志", + downloadFile: "下载", + filterStatus: "状态筛选", + filterDevice: "设备筛选", + deviceKeywordPh: "名称 / IP", + allStatus: "全部状态", + }, + }, + managedNe: { + title: "网元管理", + credsNotConfigured: "未配置 NETX_CREDENTIAL_SECRET_KEY，无法保存密码。请在 .env 中设置 Fernet 密钥后重启 API。", + add: "新增网元", + importBtn: "批量导入", + importing: "导入中…", + edit: "编辑", + delete: "删除", + confirmDelete: "确定删除该网元？", + keywordPh: "名称 / IP / 用户名 / 标签", + allVendors: "全部厂商", + allConnectStatus: "全部连通状态", + col: { + name: "名称", + vendor: "厂商", + deviceType: "设备类型", + ip: "管理地址", + port: "端口", + protocol: "协议", + user: "用户名", + password: "密码", + tags: "标签", + remark: "备注", + connect: "连通性", + testedAt: "测试时间", + actions: "操作", + }, + connect: { + run: "连通性测试", + running: "测试中…", + submitted: "已提交 {{n}} 台设备测试", + }, + importResult: { + done: "导入完成：新增 {{inserted}}，更新 {{updated}}，失败 {{failed}} 行", + }, + form: { + requiredHint: "带 * 为必填项", + nameConnectHint: "连通性测试成功后将自动更新为设备名称", + createTitle: "新增网元", + editTitle: "编辑网元", + save: "保存", + saving: "保存中…", + cancel: "取消", + created: "网元已创建", + updated: "网元已更新", + deleted: "网元已删除", + passwordRequired: "请填写密码", + passwordOptional: "留空则不修改", + }, + }, ume: { token: { title: "UME Token 状态", diff --git a/web/src/index.css b/web/src/index.css index c31cadf..3a98027 100644 --- a/web/src/index.css +++ b/web/src/index.css @@ -511,6 +511,22 @@ pre { cursor: pointer; } +.link-btn--danger { + color: #c62828; +} + +.link-btn:disabled { + opacity: 0.5; + cursor: not-allowed; +} + +.table-actions { + display: flex; + flex-wrap: wrap; + gap: 8px; + align-items: center; +} + .status { margin-top: 12px; color: #9db2cf; @@ -555,21 +571,28 @@ pre { position: fixed; right: 20px; bottom: 20px; - padding: 10px 12px; + padding: 12px 16px; border-radius: 8px; - border: 1px solid #2c405f; - background: #0f1e34; + border: 1px solid #cbd5e1; + background: #fff; + color: #1e293b; + font-size: 14px; + line-height: 1.45; z-index: 60; max-width: 420px; - box-shadow: 0 8px 24px rgba(0, 0, 0, 0.3); + box-shadow: 0 10px 28px rgba(15, 23, 42, 0.14); } .toast--ok { - border-color: #1f5a3a; + border-color: #81c784; + background: #e8f5e9; + color: #1b5e20; } .toast--error { - border-color: #5f1f2a; + border-color: #ef9a9a; + background: #ffebee; + color: #b71c1c; } .inline-spinner { @@ -739,6 +762,201 @@ pre { } } +.page-stack { + display: flex; + flex-direction: column; + gap: 16px; +} + +.panel--warn { + border-color: #ffe082; + background: #fff8e1; + color: #5d4037; +} + +.panel__toolbar { + display: flex; + justify-content: space-between; + align-items: center; + gap: 12px; + flex-wrap: wrap; + margin-bottom: 12px; +} + +.panel__toolbar h2 { + margin: 0; +} + +.panel__actions { + display: flex; + flex-wrap: wrap; + gap: 8px; +} + +.table-actions { + display: flex; + gap: 8px; + flex-wrap: wrap; +} + +.modal-backdrop { + position: fixed; + inset: 0; + z-index: 50; + background: rgba(15, 23, 42, 0.45); + display: flex; + align-items: center; + justify-content: center; + padding: 16px; +} + +.modal { + width: min(720px, 100%); + max-height: 90vh; + overflow: auto; + background: #fff; + border-radius: 10px; + padding: 20px; + box-shadow: 0 16px 48px rgba(15, 23, 42, 0.2); +} + +.modal h3 { + margin: 0 0 16px; +} + +.modal__actions { + display: flex; + justify-content: flex-end; + gap: 8px; + margin-top: 16px; +} + +.form-grid { + display: grid; + grid-template-columns: repeat(2, minmax(0, 1fr)); + gap: 12px; +} + +.form-hint { + margin: 0 0 12px; + font-size: 12px; + color: #64748b; +} + +.form-label__required { + color: #c62828; + font-weight: 600; +} + +.form-label__optional { + color: #94a3b8; + font-weight: 400; +} + +.form-field-hint { + font-size: 11px; + color: #94a3b8; + line-height: 1.35; +} + +.form-grid label { + display: flex; + flex-direction: column; + gap: 4px; + font-size: 12px; + color: #64748b; +} + +.form-grid input, +.form-grid select { + font-size: 14px; +} + +.form-grid__full { + grid-column: 1 / -1; +} + +.form-grid--single { + grid-template-columns: 1fr; +} + +.panel__hint { + margin: 0 0 12px; + font-size: 13px; + color: #64748b; + line-height: 1.45; +} + +.panel__hint--live { + color: #1565c0; +} + +.collect-commands { + width: 100%; + min-height: 160px; + font-family: ui-monospace, Consolas, monospace; + font-size: 13px; + line-height: 1.4; + padding: 10px; + border: 1px solid #cbd5e1; + border-radius: 6px; + resize: vertical; +} + +.collect-meta { + font-size: 13px; + color: #64748b; + margin-right: auto; +} + +.collect-cmd-preview { + font-size: 12px; + background: #f8fafc; + border: 1px solid #e2e8f0; + border-radius: 6px; + padding: 10px; + max-height: 120px; + overflow: auto; + white-space: pre-wrap; +} + +.collect-runs-panel { + display: flex; + flex-direction: column; + gap: 10px; +} + +.collect-runs-toolbar { + display: flex; + flex-wrap: wrap; + align-items: center; + gap: 12px; +} + +.collect-runs-filter { + display: flex; + align-items: center; + gap: 8px; + font-size: 13px; + color: #475569; +} + +.collect-runs-filter select { + min-width: 140px; + padding: 6px 8px; + border: 1px solid #cbd5e1; + border-radius: 6px; + background: #fff; +} + +.collect-runs-filter input { + min-width: 180px; + padding: 6px 8px; + border: 1px solid #cbd5e1; + border-radius: 6px; + background: #fff; +} + @media (max-width: 1200px) { .cards { grid-template-columns: 1fr; @@ -752,4 +970,7 @@ pre { .split { grid-template-columns: 1fr; } + .form-grid { + grid-template-columns: 1fr; + } } diff --git a/web/src/pages/CollectPage.tsx b/web/src/pages/CollectPage.tsx new file mode 100644 index 0000000..8b75d41 --- /dev/null +++ b/web/src/pages/CollectPage.tsx @@ -0,0 +1,531 @@ +import { useMemo, useState } from "react"; +import { useMutation, useQuery, useQueryClient } from "@tanstack/react-query"; +import { + createNeCollection, + deleteCollectionJob, + fetchCollectionJob, + fetchCollectionRuns, + fetchEligibleNe, + fetchNeCollections, + pauseCollectionJob, + restartCollectionJob, + collectionJobDownloadUrl, + collectionRunDownloadUrl, +} from "../services/api"; +import { queryKeys } from "../constants/queryKeys"; +import { useI18n } from "../i18n"; +import { useToast } from "../hooks/useToast"; +import type { CollectionJobDetail, CollectionJobItem, EligibleNeItem } from "../types"; +import { pageCount } from "../utils/display"; +import { formatSystemTime } from "../utils/time"; + +export function CollectPage() { + const { t } = useI18n(); + const { showOk, showError } = useToast(); + const queryClient = useQueryClient(); + + const [commands, setCommands] = useState(""); + const [title, setTitle] = useState(""); + const [selected, setSelected] = useState([]); + const [nePage, setNePage] = useState(1); + const [jobPage, setJobPage] = useState(1); + const [expandedJobId, setExpandedJobId] = useState(""); + + const POLL_MS = 2000; + const ELIGIBLE_PAGE_SIZE = 20; + + const eligibleQuery = useQuery({ + queryKey: queryKeys.collectionEligibleNe(nePage), + queryFn: () => fetchEligibleNe({ page: nePage, pageSize: ELIGIBLE_PAGE_SIZE }), + staleTime: 5000, + }); + + const jobsQuery = useQuery({ + queryKey: queryKeys.neCollections(jobPage), + queryFn: () => fetchNeCollections({ page: jobPage, pageSize: 20 }), + staleTime: 1000, + refetchInterval: (q) => { + const items = q.state.data?.items ?? []; + return items.some((j) => j.status === "running") ? POLL_MS : false; + }, + }); + + const detailQuery = useQuery({ + queryKey: queryKeys.neCollectionDetail(expandedJobId), + queryFn: () => fetchCollectionJob(expandedJobId), + enabled: Boolean(expandedJobId), + staleTime: 500, + refetchInterval: (q) => (q.state.data?.job.status === "running" ? POLL_MS : false), + }); + + const jobs = jobsQuery.data?.items ?? []; + const jobActive = jobs.some((j) => j.status === "running"); + const expandedJobRunning = + Boolean(expandedJobId) && + (jobs.find((j) => j.id === expandedJobId)?.status === "running" || detailQuery.data?.job.status === "running"); + const autoPolling = jobActive || expandedJobRunning; + + const refreshAll = async () => { + await Promise.all([ + queryClient.invalidateQueries({ queryKey: queryKeys.neCollectionsAll }), + expandedJobId + ? queryClient.invalidateQueries({ queryKey: queryKeys.neCollectionDetail(expandedJobId) }) + : Promise.resolve(), + expandedJobId + ? queryClient.invalidateQueries({ queryKey: queryKeys.neCollectionRunsAll }) + : Promise.resolve(), + ]); + }; + + const invalidateJobs = async (jobId?: string) => { + await queryClient.invalidateQueries({ queryKey: queryKeys.neCollectionsAll }); + if (jobId) { + await queryClient.invalidateQueries({ queryKey: queryKeys.neCollectionDetail(jobId) }); + await queryClient.invalidateQueries({ queryKey: queryKeys.neCollectionRunsAll }); + } + }; + + const pauseMutation = useMutation({ + mutationFn: pauseCollectionJob, + onSuccess: async (job) => { + showOk(t("collect.paused")); + await invalidateJobs(job.id); + }, + onError: (err) => showError(String(err)), + }); + + const restartMutation = useMutation({ + mutationFn: restartCollectionJob, + onSuccess: async (job) => { + showOk(t("collect.restarted")); + setExpandedJobId(job.id); + await invalidateJobs(job.id); + }, + onError: (err) => showError(String(err)), + }); + + const deleteMutation = useMutation({ + mutationFn: deleteCollectionJob, + onSuccess: async (_, jobId) => { + showOk(t("collect.deleted")); + if (expandedJobId === jobId) setExpandedJobId(""); + await invalidateJobs(); + }, + onError: (err) => showError(String(err)), + }); + + const startMutation = useMutation({ + mutationFn: () => + createNeCollection({ + title: title.trim(), + commands, + ne_ids: selected, + }), + onSuccess: async (job) => { + showOk(t("collect.started", { id: job.id })); + setExpandedJobId(job.id); + await queryClient.invalidateQueries({ queryKey: queryKeys.neCollectionsAll }); + await queryClient.invalidateQueries({ queryKey: queryKeys.neCollectionDetail(job.id) }); + }, + onError: (err) => showError(String(err)), + }); + + const items = eligibleQuery.data?.items ?? []; + const allSelected = items.length > 0 && items.every((x) => selected.includes(x.id)); + + const toggleAll = () => { + if (allSelected) { + const ids = new Set(items.map((x) => x.id)); + setSelected((prev) => prev.filter((id) => !ids.has(id))); + } else { + setSelected((prev) => [...new Set([...prev, ...items.map((x) => x.id)])]); + } + }; + + const neTotal = eligibleQuery.data?.total ?? 0; + const nePages = pageCount(neTotal, ELIGIBLE_PAGE_SIZE); + + const jobTotal = jobsQuery.data?.total ?? 0; + const jobPages = pageCount(jobTotal, 20); + + const commandLines = useMemo( + () => + commands + .split("\n") + .map((l) => l.trim()) + .filter((l) => l && !l.startsWith("#")).length, + [commands], + ); + + return ( +
+
+
+
+

{t("collect.eligible.title")}

+

{t("collect.eligible.hint")}

+
+ +
+ {eligibleQuery.isLoading ?

{t("common.refreshing")}

: null} + {!eligibleQuery.isLoading && items.length === 0 ? ( +

{t("collect.eligible.empty")}

+ ) : ( + + + + + + + + + + + + {items.map((row: EligibleNeItem) => ( + + + + + + + + ))} + +
+ + {t("managedNe.col.name")}{t("managedNe.col.vendor")}{t("managedNe.col.ip")}{t("managedNe.col.connect")}
+ + setSelected((prev) => + prev.includes(row.id) ? prev.filter((x) => x !== row.id) : [...prev, row.id], + ) + } + /> + {row.name || row.ip_address}{row.vendor}{row.ip_address} + {row.connect_status} +
+ )} + {neTotal > 0 ? ( +
+
{t("common.pagerMeta", { total: neTotal, page: nePage, pages: nePages })}
+
+ + +
+
+ ) : null} +
+ +
+

{t("collect.form.title")}

+

{t("collect.form.commandsHint")}

+
+ +