diff --git a/netx_api/importer.py b/netx_api/importer.py index 7c17905..38364df 100644 --- a/netx_api/importer.py +++ b/netx_api/importer.py @@ -1,29 +1,34 @@ from __future__ import annotations import json +import math from datetime import datetime, timezone -from io import BytesIO from typing import Any -import pandas as pd from sqlalchemy import func, select from sqlalchemy.orm import Session from .config import settings from .models import AlarmBatch, AlarmNorm, ImportErrorRow from .parser_config import ParserConfig +from .tabular import parse_datetime, read_xlsx + + +def _blank(v: Any) -> bool: + if v is None: + return True + if isinstance(v, float) and math.isnan(v): + return True + text = str(v).strip() + return not text or text.lower() == "nan" def _dt(v: Any) -> datetime | None: - if v is None: + if _blank(v): return None - text = str(v).strip() - if not text or text.lower() == "nan": + dt = parse_datetime(v) + if dt is None: return None - parsed = pd.to_datetime(text, errors="coerce") - if pd.isna(parsed): - return None - dt = parsed.to_pydatetime() # Convention: Excel source time is local timezone time. # Convert to UTC before persisting; keep DB as naive UTC for compatibility. if dt.tzinfo is None: @@ -35,20 +40,15 @@ def _dt(v: Any) -> datetime | None: def _str(v: Any) -> str: - if v is None: + if _blank(v): return "" - text = str(v).strip() - if text.lower() == "nan": - return "" - return text + return str(v).strip() def _int(v: Any) -> int: - if v is None: + if _blank(v): return 0 text = str(v).strip() - if not text or text.lower() == "nan": - return 0 try: return int(float(text)) except Exception: @@ -56,8 +56,7 @@ def _int(v: Any) -> int: def import_alarm_excel(db: Session, filename: str, content: bytes, parser: ParserConfig) -> AlarmBatch: - df = pd.read_excel(BytesIO(content)) - headers = [str(h) for h in df.columns] + headers, rows = read_xlsx(content) col_alarm_time = parser.resolve_col(headers, "alarm_time") col_clear_time = parser.resolve_col(headers, "clear_time") @@ -80,7 +79,7 @@ def import_alarm_excel(db: Session, filename: str, content: bytes, parser: Parse source_file=filename, parser_version=parser.parser_version, dict_version=parser.dict_version, - total_rows=int(len(df.index)), + total_rows=len(rows), status="processing", ) db.add(batch) @@ -89,8 +88,8 @@ def import_alarm_excel(db: Session, filename: str, content: bytes, parser: Parse success_rows = 0 failed_rows = 0 - for idx, row in df.iterrows(): - row_no = int(idx) + 2 # Excel header is row 1 + for row_no, row in rows: # row_no = sheet row (header is row 1) + raw_json = json.dumps(row, ensure_ascii=False, default=str) alarm_time = _dt(row.get(col_alarm_time)) if col_alarm_time else None raw_severity = _str(row.get(col_severity)) if col_severity else "" alarm_code = _str(row.get(col_alarm_code)) if col_alarm_code else "" @@ -111,7 +110,7 @@ def import_alarm_excel(db: Session, filename: str, content: bytes, parser: Parse batch_id=batch.batch_id, row_no=row_no, reason="missing_or_invalid_alarm_time", - raw_json=json.dumps(row.to_dict(), ensure_ascii=False, default=str), + raw_json=raw_json, ) ) continue @@ -122,7 +121,7 @@ def import_alarm_excel(db: Session, filename: str, content: bytes, parser: Parse batch_id=batch.batch_id, row_no=row_no, reason="missing_ne_name_and_ne_id", - raw_json=json.dumps(row.to_dict(), ensure_ascii=False, default=str), + raw_json=raw_json, ) ) continue @@ -151,7 +150,7 @@ def import_alarm_excel(db: Session, filename: str, content: bytes, parser: Parse vendor=settings.vendor, source_type=settings.source_type, source_file=filename, - raw_json=json.dumps(row.to_dict(), ensure_ascii=False, default=str), + raw_json=raw_json, ) ) success_rows += 1 diff --git a/netx_api/ne_service_common.py b/netx_api/ne_service_common.py index ed9ed1f..f76d6d7 100644 --- a/netx_api/ne_service_common.py +++ b/netx_api/ne_service_common.py @@ -2,10 +2,10 @@ from __future__ import annotations from datetime import datetime +import math import re from typing import Any -import pandas as pd from fastapi import HTTPException from sqlalchemy.orm import Session @@ -199,7 +199,7 @@ def _parse_import_bool(value: Any) -> bool: def _import_cell_str(value: Any) -> str: - if value is None or (isinstance(value, float) and pd.isna(value)): + if value is None or (isinstance(value, float) and math.isnan(value)): return "" text = str(value).strip() return "" if text.lower() == "nan" else text diff --git a/netx_api/ne_service_import.py b/netx_api/ne_service_import.py index 7bad1bc..ca006af 100644 --- a/netx_api/ne_service_import.py +++ b/netx_api/ne_service_import.py @@ -1,14 +1,13 @@ """Managed NE Excel import and UME inventory sync into managed_ne.""" from __future__ import annotations -from io import BytesIO from typing import Any -import pandas as pd from fastapi import HTTPException from sqlalchemy.orm import Session from .models import ManagedNE, UmeInventoryNE +from .tabular import read_table, write_csv, write_xlsx from .ne_crypto import CredentialCryptoError, encrypt_secret from .device_types import SUPPORTED_DEVICE_TYPES, SUPPORTED_VENDORS, is_placeholder_ne_source from .ne_schemas import ( @@ -151,46 +150,41 @@ def build_managed_ne_import_template(fmt: str = "xlsx") -> tuple[str, bytes, str "remark": "no direct password, use batch proxy", }, ] - df = pd.DataFrame(rows, columns=list(IMPORT_COLUMNS)) - buf = BytesIO() + headers = list(IMPORT_COLUMNS) + table = [[r.get(c, "") for c in headers] for r in rows] kind = str(fmt or "xlsx").strip().lower() if kind == "csv": - df.to_csv(buf, index=False, encoding="utf-8-sig") return ( "managed_ne_import_template.csv", - buf.getvalue(), + write_csv(headers, table), "text/csv; charset=utf-8", ) - device_types_df = pd.DataFrame({"device_type": list(SUPPORTED_DEVICE_TYPES)}) - with pd.ExcelWriter(buf, engine="openpyxl") as writer: - df.to_excel(writer, sheet_name="import", index=False) - device_types_df.to_excel(writer, sheet_name="device_type", index=False) + payload = write_xlsx([ + ("import", headers, table), + ("device_type", ["device_type"], [[t] for t in SUPPORTED_DEVICE_TYPES]), + ]) return ( "managed_ne_import_template.xlsx", - buf.getvalue(), + payload, "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", ) 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)) + headers, raw_rows = read_table(content, filename) 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 _REQUIRED_IMPORT_COLUMNS if c not in df.columns] + columns = [h.strip().lower() for h in headers] + missing = [c for c in _REQUIRED_IMPORT_COLUMNS if c not in 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 + for row_no, cells in raw_rows: + row = {k.strip().lower(): v for k, v in cells.items()} # Savepoint per row: a failing row must not leave a half-written NE behind. savepoint = db.begin_nested() try: @@ -214,16 +208,16 @@ def import_managed_ne(db: Session, content: bytes, filename: str) -> ImportResul failed.append(ImportFailure(row=row_no, reason="username_required")) savepoint.rollback() continue - port_raw = row.get("port", 22) + port_raw = row.get("port") try: - port = int(port_raw) - except (TypeError, ValueError): + port = int(float(port_raw)) if port_raw is not None else 22 + except (TypeError, ValueError, OverflowError): port = 22 if not 1 <= port <= 65535: failed.append(ImportFailure(row=row_no, reason="invalid_port")) savepoint.rollback() continue - protocol = _normalize_protocol(str(row.get("protocol", "ssh"))) + protocol = _normalize_protocol(_import_cell_str(row.get("protocol")) or "ssh") display_name = _import_cell_str(row.get("name", "")) or ip vendor_raw = _import_cell_str(row.get("vendor", "")) or "Other" vendor = "Other" diff --git a/netx_api/tabular.py b/netx_api/tabular.py new file mode 100644 index 0000000..c571456 --- /dev/null +++ b/netx_api/tabular.py @@ -0,0 +1,151 @@ +"""Small xlsx/csv reader/writer for imports and templates (replaces pandas). + +Rows come back as ``(row_no, {header: value})`` where ``row_no`` is the 1-based +sheet/file row (header = row 1), so import errors point at the row users see. +Blank rows are skipped. Excel cells keep their openpyxl types (str / int / +float / datetime / bool / None); integral floats become ints. CSV cells are +strings, empty cells are None. +""" +from __future__ import annotations + +import csv +import io +import re +from collections.abc import Iterable, Iterator +from datetime import datetime +from typing import Any + +from openpyxl import Workbook, load_workbook + +Row = tuple[int, dict[str, Any]] + + +def _headers(raw: Iterable[Any]) -> list[str]: + out: list[str] = [] + seen: dict[str, int] = {} + for i, h in enumerate(raw): + name = "" if h is None else str(h).strip() + if not name: + name = f"Unnamed: {i}" + n = seen.get(name, 0) + seen[name] = n + 1 + out.append(name if n == 0 else f"{name}.{n}") + return out + + +def _cell(v: Any) -> Any: + if isinstance(v, str): + return v if v.strip() else None + if isinstance(v, float) and v.is_integer(): + return int(v) + return v + + +def _rows(records: Iterator[tuple[int, list[Any]]]) -> tuple[list[str], list[Row]]: + headers: list[str] | None = None + rows: list[Row] = [] + for row_no, values in records: + cells = [_cell(v) for v in values] + if all(c is None for c in cells): + continue + if headers is None: + headers = _headers(values) + continue + cells += [None] * (len(headers) - len(cells)) + rows.append((row_no, dict(zip(headers, cells)))) + return headers or [], rows + + +def _decode(content: bytes) -> str: + try: + return content.decode("utf-8-sig") + except UnicodeDecodeError: + # CSVs saved by Chinese Excel default to GBK. + return content.decode("gb18030") + + +def read_csv(content: bytes) -> tuple[list[str], list[Row]]: + reader = csv.reader(io.StringIO(_decode(content), newline="")) + return _rows((i, rec) for i, rec in enumerate(reader, start=1)) + + +def read_xlsx(content: bytes) -> tuple[list[str], list[Row]]: + wb = load_workbook(io.BytesIO(content), read_only=True, data_only=True) + try: + ws = wb.worksheets[0] + return _rows((i, list(rec)) for i, rec in enumerate(ws.iter_rows(values_only=True), start=1)) + finally: + wb.close() + + +def read_table(content: bytes, filename: str = "") -> tuple[list[str], list[Row]]: + if str(filename or "").lower().endswith(".csv"): + return read_csv(content) + return read_xlsx(content) + + +def write_csv(headers: list[str], rows: Iterable[Iterable[Any]]) -> bytes: + buf = io.StringIO() + writer = csv.writer(buf, lineterminator="\n") + writer.writerow(headers) + for r in rows: + writer.writerow(["" if v is None else v for v in r]) + # BOM so Excel opens UTF-8 (Chinese) correctly. + return buf.getvalue().encode("utf-8-sig") + + +def write_xlsx(sheets: list[tuple[str, list[str], Iterable[Iterable[Any]]]]) -> bytes: + wb = Workbook() + wb.remove(wb.active) + for title, headers, rows in sheets: + ws = wb.create_sheet(title) + ws.append(headers) + for r in rows: + ws.append(list(r)) + buf = io.BytesIO() + wb.save(buf) + return buf.getvalue() + + +_DATE_FORMATS = ( + "%Y-%m-%d %H:%M:%S", + "%Y-%m-%d %H:%M", + "%Y-%m-%d", + "%Y%m%d%H%M%S", + "%Y%m%d", + "%m/%d/%Y %H:%M:%S", + "%m/%d/%Y %H:%M", + "%m/%d/%Y", + "%d-%b-%Y %H:%M:%S", + "%b %d, %Y %H:%M:%S", +) +_CN_DATE = re.compile(r"(\d{4})年(\d{1,2})月(\d{1,2})日") + + +def parse_datetime(v: Any) -> datetime | None: + """Best-effort timestamp parse for spreadsheet cells; None when unparseable.""" + if v is None: + return None + if isinstance(v, datetime): + return v + text = str(v).strip() + if not text or text.lower() in ("nan", "nat", "none", "null"): + return None + text = _CN_DATE.sub(r"\1-\2-\3", text) + # 2024/1/2 03:04:05 -> 2024-1-2 ... ; zero-pad so fromisoformat/strptime accept it. + if re.match(r"^\d{4}/\d{1,2}/\d{1,2}", text): + text = text.replace("/", "-", 2) + m = re.match(r"^(\d{4})-(\d{1,2})-(\d{1,2})(.*)$", text) + if m: + text = f"{m.group(1)}-{int(m.group(2)):02d}-{int(m.group(3)):02d}{m.group(4)}" + iso = text[:-1] + "+00:00" if text.endswith(("Z", "z")) else text + try: + return datetime.fromisoformat(iso) + except ValueError: + pass + for fmt in _DATE_FORMATS: + try: + return datetime.strptime(text, fmt) + except ValueError: + continue + return None diff --git a/pyproject.toml b/pyproject.toml index 338ad4f..a02bdb0 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -16,7 +16,6 @@ dependencies = [ "httpx>=0.27.0", "sqlalchemy>=2.0.0", "psycopg[binary]>=3.2.0", - "pandas>=2.0.0", "openpyxl>=3.1.0", "PyYAML>=6.0.0", "pydantic>=2.8.0", diff --git a/requirements.txt b/requirements.txt index d870d81..8bd50ac 100644 --- a/requirements.txt +++ b/requirements.txt @@ -4,7 +4,6 @@ websockets>=12.0 httpx>=0.27.0 sqlalchemy>=2.0.0 psycopg[binary]>=3.2.0 -pandas>=2.0.0 openpyxl>=3.1.0 PyYAML>=6.0.0 pydantic>=2.8.0 diff --git a/tests/conftest.py b/tests/conftest.py index 9014c70..c29b7da 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -15,9 +15,12 @@ os.environ["NETX_CREDENTIAL_SECRET_FILE"] = str(Path(_TEST_STATE.name) / "creden def pytest_sessionstart(session): + from sqlalchemy.engine import make_url + from netx_api.db import engine - if str(engine.url) != _TEST_DATABASE_URL: + # Compare parsed URLs: newer SQLAlchemy renders "C:" as "C%3A" in str(url). + if engine.url != make_url(_TEST_DATABASE_URL): raise pytest.UsageError("NetX tests require the isolated temporary database") diff --git a/tests/test_tabular.py b/tests/test_tabular.py new file mode 100644 index 0000000..1826dee --- /dev/null +++ b/tests/test_tabular.py @@ -0,0 +1,83 @@ +"""xlsx/csv helpers that replaced pandas in alarm + managed-NE imports.""" +import io +from datetime import datetime, timezone + +from openpyxl import Workbook, load_workbook + +from netx_api.importer import _dt, _int, _str +from netx_api.tabular import parse_datetime, read_csv, read_table, read_xlsx, write_csv, write_xlsx + + +def _xlsx(rows): + wb = Workbook() + ws = wb.active + for r in rows: + ws.append(r) + buf = io.BytesIO() + wb.save(buf) + return buf.getvalue() + + +def test_read_xlsx_types_blank_rows_and_sheet_row_numbers(): + content = _xlsx([ + ["Time", "Code", "Port", None, "Code"], + [datetime(2024, 1, 2, 3, 4, 5), 1001, 22.0, "x", "dup"], + [None, None, None, None, None], + [" ", "A", 2.5, None, None], + ]) + headers, rows = read_xlsx(content) + assert headers == ["Time", "Code", "Port", "Unnamed: 3", "Code.1"] + assert rows == [ + (2, {"Time": datetime(2024, 1, 2, 3, 4, 5), "Code": 1001, "Port": 22, "Unnamed: 3": "x", "Code.1": "dup"}), + (4, {"Time": None, "Code": "A", "Port": 2.5, "Unnamed: 3": None, "Code.1": None}), + ] + + +def test_read_csv_bom_quotes_short_rows_and_blank_lines(): + content = "a,b,c\n1,\"x, y\",\n\n2\n".encode("utf-8") + headers, rows = read_csv(content) + assert headers == ["a", "b", "c"] + assert rows == [(2, {"a": "1", "b": "x, y", "c": None}), (4, {"a": "2", "b": None, "c": None})] + + +def test_read_csv_falls_back_to_gbk(): + headers, rows = read_csv("名称,地址\n核心,10.0.0.1\n".encode("gbk")) + assert headers == ["名称", "地址"] + assert rows == [(2, {"名称": "核心", "地址": "10.0.0.1"})] + + +def test_read_table_dispatches_on_extension(): + assert read_table(b"h\nv\n", "X.CSV") == (["h"], [(2, {"h": "v"})]) + assert read_table(_xlsx([["h"], ["v"]]), "x.xlsx") == (["h"], [(2, {"h": "v"})]) + + +def test_write_csv_and_xlsx_round_trip(): + csv_bytes = write_csv(["a", "b"], [[1, None], ["中文", "x"]]) + assert csv_bytes.startswith("".encode("utf-8")) + assert read_csv(csv_bytes)[1] == [(2, {"a": "1", "b": None}), (3, {"a": "中文", "b": "x"})] + + xlsx = write_xlsx([("import", ["a"], [[1]]), ("types", ["t"], [["x"], ["y"]])]) + wb = load_workbook(io.BytesIO(xlsx)) + assert wb.sheetnames == ["import", "types"] + assert read_xlsx(xlsx) == (["a"], [(2, {"a": 1})]) + + +def test_parse_datetime_formats(): + want = datetime(2024, 1, 2, 3, 4, 5) + for text in ( + "2024-01-02 03:04:05", "2024/01/02 03:04:05", "2024/1/2 3:04:05", "2024-1-2 03:04:05", + "2024-01-02T03:04:05", "01/02/2024 03:04:05", "20240102030405", "02-Jan-2024 03:04:05", + "Jan 02, 2024 03:04:05", "2024年1月2日 03:04:05", + ): + assert parse_datetime(text) == want, text + assert parse_datetime("2024-01-02T03:04:05Z") == datetime(2024, 1, 2, 3, 4, 5, tzinfo=timezone.utc) + assert parse_datetime("2024-01-02") == datetime(2024, 1, 2) + for bad in (None, "", " ", "nan", "NaT", "abc"): + assert parse_datetime(bad) is None + + +def test_importer_cell_helpers(): + assert _dt("2024-01-02T03:04:05Z") == datetime(2024, 1, 2, 3, 4, 5) # stored as naive UTC + assert _dt(float("nan")) is None and _dt("garbage") is None + assert _str(1001) == "1001" and _str(None) == "" and _str(" nan ") == "" and _str(" a ") == "a" + assert _int("7") == 7 and _int(2.0) == 2 and _int("x") == 0 and _int(None) == 0