fix table identifier length

Author: felixevers

main.py

@@ -1,29 +1,44 @@
+import hashlib
 import logging
-import pandas as pd
-from sqlalchemy import create_engine
 
-from typing import List
+import pandas as pd
+from preprocessing.core import Preprocessor
+from preprocessing.loader import BibLoader, TxtLoader
+from preprocessing.models import DocumentRecord, PreprocessedDocument
 from scystream.sdk.core import entrypoint
 from scystream.sdk.env.settings import (
     EnvSettings,
+    FileSettings,
     InputSettings,
     OutputSettings,
-    FileSettings,
-    PostgresSettings
+    PostgresSettings,
 )
 from scystream.sdk.file_handling.s3_manager import S3Operations
-
-from preprocessing.core import Preprocessor
-from preprocessing.loader import TxtLoader, BibLoader
-from preprocessing.models import DocumentRecord, PreprocessedDocument
+from sqlalchemy import create_engine
+from sqlalchemy.sql import quoted_name
 
 logging.basicConfig(
     level=logging.INFO,
-    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
+    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
 )
 logger = logging.getLogger(__name__)
 
 
+def _normalize_table_name(table_name: str) -> str:
+    max_length = 63
+    if len(table_name) <= max_length:
+        return table_name
+    digest = hashlib.sha1(table_name.encode("utf-8")).hexdigest()[:10]
+    prefix_length = max_length - len(digest) - 1
+    return f"{table_name[:prefix_length]}_{digest}"
+
+
+def _resolve_db_table(settings: PostgresSettings) -> str:
+    normalized_name = _normalize_table_name(settings.DB_TABLE)
+    settings.DB_TABLE = normalized_name
+    return normalized_name
+
+
 class NormalizedDocsOutput(PostgresSettings, OutputSettings):
     __identifier__ = "normalized_docs"
 
@@ -69,31 +84,40 @@ class PreprocessBIB(EnvSettings):
 
 
 def _write_preprocessed_docs_to_postgres(
-        preprocessed_ouput: List[PreprocessedDocument],
-        settings: PostgresSettings
+    preprocessed_ouput: list[PreprocessedDocument],
+    settings: PostgresSettings,
 ):
-    df = pd.DataFrame([
-        {
-            "doc_id": d.doc_id,
-            "tokens": d.tokens
-        }
-        for d in preprocessed_ouput
-    ])
-
-    logger.info(f"Writing {len(df)} processed documents to DB table '{
-                settings.DB_TABLE}'…")
+    resolved_table_name = _resolve_db_table(settings)
+    df = pd.DataFrame(
+        [
+            {
+                "doc_id": d.doc_id,
+                "tokens": d.tokens,
+            }
+            for d in preprocessed_ouput
+        ],
+    )
+
+    logger.info(
+        "Writing %s processed documents to DB table '%s'…",
+        len(df),
+        resolved_table_name,
+    )
     engine = create_engine(
         f"postgresql+psycopg2://{settings.PG_USER}:{settings.PG_PASS}"
-        f"@{settings.PG_HOST}:{int(settings.PG_PORT)}/"
+        f"@{settings.PG_HOST}:{int(settings.PG_PORT)}/",
     )
 
-    df.to_sql(settings.DB_TABLE, engine, if_exists="replace", index=False)
+    table_name = quoted_name(resolved_table_name, quote=True)
+    df.to_sql(table_name, engine, if_exists="replace", index=False)
 
-    logger.info(f"Successfully stored normalized documents into '{
-                settings.DB_TABLE}'.")
+    logger.info(
+        "Successfully stored normalized documents into '%s'.",
+        resolved_table_name,
+    )
 
 
-def _preprocess_and_store(documents: List[DocumentRecord], settings):
+def _preprocess_and_store(documents: list[DocumentRecord], settings):
     """Shared preprocessing logic for TXT and BIB."""
     logger.info(f"Starting preprocessing with {len(documents)} documents")
 
@@ -110,7 +134,9 @@ def _preprocess_and_store(documents: List[DocumentRecord], settings):
     result = pre.generate_normalized_output()
 
     _write_preprocessed_docs_to_postgres(
-        result, settings.normalized_docs_output)
+        result,
+        settings.normalized_docs_output,
+    )
 
     logger.info("Preprocessing completed successfully.")