rt0209/S4HANA_Migration_Intelligence_Agent

1"""2DocumentProcessor3=================4Reads and normalises various SAP export formats:5 6  • .abap / .txt / .prog → raw ABAP source7  • .zip                 → unpacks and concatenates multiple ABAP files8  • .xlsx / .csv         → SAP custom object lists (SE80 / SCTC exports)9  • .pdf                 → extracts text (e.g. SE01 transport logs, BD87 reports)10 11Returns a list of SourceObject dicts ready for ABAPAnalyzer.12"""13import io14import zipfile15import logging16import chardet17from pathlib import Path18from dataclasses import dataclass, field19from typing import List, Optional20 21logger = logging.getLogger(__name__)22 23# File extensions treated as ABAP source24ABAP_EXTENSIONS = {".abap", ".prog", ".fugr", ".clas", ".intf", ".dtel",25                   ".doma", ".tabl", ".ttyp", ".shlp", ".msag", ".enhs",26                   ".txt"}27 28 29@dataclass30class SourceObject:31    object_name: str32    object_type: str33    source_code: str34    additional_context: str = ""35    metadata: dict = field(default_factory=dict)36 37 38class DocumentProcessor:39 40    async def process_files(self, files: list) -> List[SourceObject]:41        """42        Accept a list of fastapi.UploadFile objects and return SourceObjects.43        """44        results: List[SourceObject] = []45 46        for upload in files:47            filename = upload.filename or "unknown"48            content = await upload.read()49            ext = Path(filename).suffix.lower()50 51            logger.info("Processing %s (%d bytes)", filename, len(content))52 53            if ext == ".zip":54                results.extend(self._handle_zip(content, filename))55            elif ext in (".xlsx", ".xls"):56                results.extend(self._handle_excel(content, filename))57            elif ext == ".csv":58                results.extend(self._handle_csv(content.decode("utf-8", errors="replace"), filename))59            elif ext == ".pdf":60                results.extend(self._handle_pdf(content, filename))61            elif ext in ABAP_EXTENSIONS:62                text = self._decode(content)63                obj_name = Path(filename).stem.upper()64                results.append(SourceObject(65                    object_name=obj_name,66                    object_type=self._infer_type(filename),67                    source_code=text,68                ))69            else:70                # Try as plain text71                try:72                    text = self._decode(content)73                    results.append(SourceObject(74                        object_name=Path(filename).stem.upper(),75                        object_type="ABAP Object",76                        source_code=text,77                    ))78                except Exception:79                    logger.warning("Skipping unsupported file: %s", filename)80 81        return results82 83    # ── Format handlers ───────────────────────────────────────────────84 85    def _handle_zip(self, data: bytes, zipname: str) -> List[SourceObject]:86        results = []87        try:88            with zipfile.ZipFile(io.BytesIO(data)) as zf:89                for name in zf.namelist():90                    ext = Path(name).suffix.lower()91                    if ext in ABAP_EXTENSIONS:92                        raw = zf.read(name)93                        text = self._decode(raw)94                        results.append(SourceObject(95                            object_name=Path(name).stem.upper(),96                            object_type=self._infer_type(name),97                            source_code=text,98                            additional_context=f"Extracted from {zipname}",99                        ))100        except zipfile.BadZipFile:101            logger.error("Bad ZIP file: %s", zipname)102        return results103 104    def _handle_excel(self, data: bytes, filename: str) -> List[SourceObject]:105        """106        Expects an SE80 or SCTC custom-object export workbook with columns:107          Object Type | Object Name | Description | Package | Developer108        Returns one SourceObject per row with metadata as context.109        Each row doesn't have source — they'll be used for the object-level110        risk catalogue (no source code available from the list alone).111        """112        try:113            import pandas as pd114            df = pd.read_excel(io.BytesIO(data))115            df.columns = [str(c).strip().upper() for c in df.columns]116 117            # Flexible column matching118            name_col = next((c for c in df.columns if "NAME" in c or "OBJECT" in c), df.columns[0])119            type_col = next((c for c in df.columns if "TYPE" in c), None)120 121            results = []122            for _, row in df.iterrows():123                obj_name = str(row.get(name_col, "UNKNOWN")).strip()124                obj_type = str(row.get(type_col, "Custom Object")).strip() if type_col else "Custom Object"125                # Build context from all columns126                ctx = " | ".join(f"{k}:{v}" for k, v in row.items() if str(v) != "nan")127                results.append(SourceObject(128                    object_name=obj_name,129                    object_type=obj_type,130                    source_code=f"-- Object metadata only (no source available from registry export)\n-- {ctx}",131                    additional_context="Sourced from custom object registry Excel export.",132                ))133            return results134        except Exception as e:135            logger.error("Excel parsing failed for %s: %s", filename, e)136            return []137 138    def _handle_csv(self, text: str, filename: str) -> List[SourceObject]:139        try:140            import pandas as pd141            df = pd.read_csv(io.StringIO(text))142            return self._handle_excel(text.encode(), filename)143        except Exception as e:144            logger.error("CSV parsing failed: %s", e)145            return []146 147    def _handle_pdf(self, data: bytes, filename: str) -> List[SourceObject]:148        try:149            import pdfplumber150            texts = []151            with pdfplumber.open(io.BytesIO(data)) as pdf:152                for page in pdf.pages:153                    t = page.extract_text()154                    if t:155                        texts.append(t)156            full_text = "\n".join(texts)157            return [SourceObject(158                object_name=Path(filename).stem.upper(),159                object_type="Document (PDF)",160                source_code=full_text,161                additional_context="Extracted from PDF — may contain transport logs or config documentation.",162            )]163        except Exception as e:164            logger.error("PDF extraction failed for %s: %s", filename, e)165            return []166 167    # ── Utilities ─────────────────────────────────────────────────────168 169    def _decode(self, data: bytes) -> str:170        detected = chardet.detect(data)171        encoding = detected.get("encoding") or "utf-8"172        return data.decode(encoding, errors="replace")173 174    def _infer_type(self, filename: str) -> str:175        ext = Path(filename).suffix.lower()176        mapping = {177            ".prog": "ABAP Report",178            ".fugr": "Function Group",179            ".clas": "ABAP Class",180            ".intf": "ABAP Interface",181            ".dtel": "Data Element",182            ".doma": "Domain",183            ".tabl": "Database Table",184            ".ttyp": "Table Type",185            ".shlp": "Search Help",186            ".msag": "Message Class",187            ".abap": "ABAP Program",188        }189        return mapping.get(ext, "ABAP Object")190