fix2

app/cleaner.py

@@ -0,0 +1,427 @@
+import glob
+import os
+import re
+import time
+from multiprocessing import Pool, cpu_count
+from pathlib import Path
+
+from langchain.agents import create_agent
+from langchain_community.chat_models import ChatLlamaCpp
+from langchain_core.messages import HumanMessage, SystemMessage
+from langchain_core.runnables import Runnable, RunnableLambda, chain
+from langchain_core.tools import tool
+from langchain_huggingface import ChatHuggingFace, HuggingFaceEndpoint
+from langchain_text_splitters import RecursiveCharacterTextSplitter, TextSplitter
+from pydantic import BaseModel, Field
+from settings import settings
+
+
+@tool
+def remove_brackets_content(text: str) -> str:
+    """
+    Remove all content inside square brackets [],
+    round brackets () and curly brackets {}.
+    Useful for removing sound descriptions, speaker labels,
+    stage directions like [applause], (laughs), {music}.
+    """
+    text = re.sub(r'\[.*?\]', '', text)   # [applause]
+    text = re.sub(r'\(.*?\)', '', text)   # (laughs)
+    text = re.sub(r'\{.*?\}', '', text)   # {music}
+    return text.strip()
+
+
+@tool
+def remove_non_alphabetic(text: str) -> str:
+    """
+    Remove all non-alphabetic characters except spaces.
+    Keeps only letters A-Z, a-z and whitespace.
+    Useful for stripping punctuation, numbers, special symbols.
+    """
+    text = re.sub(r'[^a-zA-Z\s]', '', text)
+    return text.strip()
+
+
+@tool
+def remove_newlines(text: str) -> str:
+    """
+    Remove newline characters and replace them with spaces.
+    Merges multi-line subtitle blocks into single lines.
+    """
+    text = text.replace('\n', ' ')
+    text = text.replace('\r', ' ')
+    text = re.sub(r' +', ' ', text)  # collapse multiple spaces
+    return text.strip()
+
+
+@tool
+def remove_dialog_punctuation(text: str) -> str:
+    """
+    Remove dialog-specific punctuation: dashes at line start (- text),
+    ellipsis (...), double dashes (--), quotation marks,
+    and excessive punctuation used in subtitles.
+    """
+    text = re.sub(r'^\s*-+\s*', '', text, flags=re.MULTILINE)  # leading dashes
+    text = re.sub(r'\.{2,}', '', text)    # ellipsis ...
+    text = re.sub(r'-{2,}', '', text)     # double dash --
+    text = re.sub(r'["""\'\'\']+', '', text)  # quotes
+    text = re.sub(r'[!?,;:]+', '', text)  # dialog punctuation
+    return text.strip()
+
+
+@tool
+def remove_timestamps(text: str) -> str:
+    """
+    Remove SRT/VTT subtitle timestamps.
+    Handles formats like:
+    - 00:01:23,456 --> 00:01:25,789  (SRT)
+    - 00:01:23.456 --> 00:01:25.789  (VTT)
+    Also removes bare sequence numbers (1, 2, 3...) used in SRT files.
+    """
+    # SRT timestamps
+    text = re.sub(
+        r'\d{2}:\d{2}:\d{2}[.,]\d{3}\s*-->\s*\d{2}:\d{2}:\d{2}[.,]\d{3}',
+        '', text
+    )
+    # VTT cue identifiers
+    text = re.sub(r'^\s*\d+\s*$', '', text, flags=re.MULTILINE)
+    # WEBVTT header
+    text = re.sub(r'WEBVTT.*\n?', '', text)
+    return text.strip()
+
+
+@tool
+def remove_speaker_labels(text: str) -> str:
+    """
+    Remove speaker labels commonly found in subtitles.
+    Handles formats like:
+    - JOHN: text
+    - John: text
+    - [JOHN]: text
+    - <v John> text  (VTT format)
+    """
+    text = re.sub(r'^[A-Z][A-Z\s]{1,20}:\s*', '', text, flags=re.MULTILINE)  # JOHN:
+    text = re.sub(r'^\w[\w\s]{1,20}:\s*', '', text, flags=re.MULTILINE)       # John:
+    text = re.sub(r'<v\s+[^>]+>', '', text)                                    # <v John>
+    return text.strip()
+
+
+@tool
+def remove_html_tags(text: str) -> str:
+    """
+    Remove HTML/XML tags commonly found in subtitles.
+    Handles: <i>, <b>, <u>, <font color="">, <c.colorname> etc.
+    Used in SRT and VTT files for styling.
+    """
+    text = re.sub(r'<[^>]+>', '', text)
+    return text.strip()
+
+
+@tool
+def normalize_whitespace(text: str) -> str:
+    """
+    Normalize all whitespace: collapse multiple spaces into one,
+    strip leading/trailing spaces from each line,
+    remove empty lines.
+    Final cleanup step — use after all other tools.
+    """
+    lines = text.split('\n')
+    lines = [re.sub(r' +', ' ', line).strip() for line in lines]
+    lines = [line for line in lines if line]  # remove empty
+    return ' '.join(lines)
+
+
+@tool
+def lowercase_text(text: str) -> str:
+    """
+    Convert all text to lowercase.
+    Recommended for sentiment analysis preprocessing
+    to ensure uniform token representation.
+    """
+    return text.lower()
+
+
+@tool
+def remove_filler_words(text: str) -> str:
+    """
+    Remove common spoken filler words that add noise for sentiment analysis.
+    Removes: um, uh, hmm, ah, oh, er, erm, hm, gonna, wanna, gotta etc.
+    """
+    fillers = r'\b(um+|uh+|hmm+|hm+|ah+|oh+|er+|erm+|gonna|wanna|gotta|kinda|sorta|like|okay|ok|yeah|yep|nope)\b'
+    text = re.sub(fillers, '', text, flags=re.IGNORECASE)
+    text = re.sub(r' +', ' ', text)
+    return text.strip()
+
+
+class FormattedResponse(BaseModel):
+    """Cleaned text"""
+    CLEANED_TEXT: str = Field(description="The CLEANED subtitle text")
+
+@chain
+def clean_subtitle(text: str) -> FormattedResponse:
+    """Deterministic cleaning pipeline — no LLM needed."""
+    text = remove_timestamps.invoke(text)
+    text = remove_brackets_content.invoke(text)
+    text = remove_html_tags.invoke(text)
+    text = remove_speaker_labels.invoke(text)
+    text = remove_dialog_punctuation.invoke(text)
+    text = remove_newlines.invoke(text)
+    text = remove_non_alphabetic.invoke(text)
+    text = remove_filler_words.invoke(text)
+    text = lowercase_text.invoke(text)
+    text = normalize_whitespace.invoke(text)
+    return FormattedResponse(CLEANED_TEXT=text)
+
+class PreprocessingAgent:
+    """
+    Agent for preprocessing.
+
+    Pipeline:
+        1. Dumping system prompt and subtitles into chat format
+        2. Passing this chat to agent, which can call tools
+        3. Passing agent's output to LLM to extract cleaned text
+
+    Second LLM is used, since single agent will struggle dealing with heavy prompt,
+    tools and requirements for structured output
+    """
+    def __init__(self):
+
+        self.use_hugging_face = settings.preprocessor.use_hugging_face # Flag for using local or external LLM
+        self.output_path = settings.preprocessor.output_path
+        self.num_workers = cpu_count() # If using external model several workers can be used in parallel
+
+        self.prompt ="""Your goal is to clean raw subtitle text step by step using available tools.
+            ## Input text format:
+            The user will provide the subtitle text marked as SUBTITLE_TEXT.
+
+            ## Recommended cleaning pipeline (follow this order):
+            1. remove_timestamps         — strip SRT/VTT timing info
+            2. remove_brackets_content   — remove [sound], (laughter), {{music}}
+            3. remove_html_tags          — strip <i>, <b>, <font> tags
+            4. remove_speaker_labels     — remove JOHN:, John:, <v John>
+            5. remove_dialog_punctuation — remove ---, ..., quotes, !?;:,
+            6. remove_newlines           — merge lines into single text
+            7. remove_non_alphabetic     — keep only letters and spaces
+            8. remove_filler_words       — remove um, uh, gonna, wanna...
+            9. lowercase_text            — convert to lowercase
+            10. normalize_whitespace     — final cleanup of spaces
+
+            Apply ALL steps unless the user specifies otherwise.
+            After cleaning, return the final cleaned text clearly labeled as:
+
+            CLEANED_TEXT: <result>
+            """
+
+    def clean_text(text: str) -> str:
+        return clean_subtitle.invoke(text).CLEANED_TEXT
+
+
+    def _setup_pipeline(self, tools: list) -> Runnable:
+        """
+        Initializes the pipeline
+
+        Args:
+            tools (list): list of available tools
+
+        Returns:
+            Runnable: Runnable pipeline
+        """
+        if self.use_hugging_face:
+            endpoint = HuggingFaceEndpoint(
+                repo_id=settings.preprocessor.model,
+                task='text-generation',
+                do_sample=False,
+                temperature=0.05,
+                max_new_tokens=16_000,
+                streaming=False
+            )
+
+            llm = ChatHuggingFace(llm=endpoint)
+            structured_llm = llm.with_structured_output(FormattedResponse, method='json_mode')
+        else:
+            llm = ChatLlamaCpp(
+                model_path=settings.preprocessor.local_model_path,
+                n_ctx=4096,
+                n_gpu_layers=-1,
+                n_batch=512,
+                max_tokens=16_000,
+                n_threads=cpu_count() - 1,
+                temperature=0.05,
+                verbose=True,
+                f16_kv=True
+            )
+            structured_llm = llm.with_structured_output(FormattedResponse)
+
+        agent = create_agent(llm, tools, system_prompt=SystemMessage(self.prompt))
+
+        pipeline = (
+            RunnableLambda(lambda text: {
+                'messages': [HumanMessage(f'Clean the following text:\n\nSUBTITLE_TEXT:\n{text}')]
+            })
+            | agent
+            | RunnableLambda(lambda result: f"Return this cleaned text in the required format:\n\n{result['messages'][-1].content}")
+            | structured_llm
+        )
+
+        return pipeline
+
+
+    def _init_tools(self) -> list:
+        """
+        Provides the list of available tools. This method is required due to
+        function serialization problem (when using several workers)
+
+        Returns:
+            list: tools
+        """
+        return [
+            remove_timestamps,
+            remove_brackets_content,
+            remove_html_tags,
+            remove_speaker_labels,
+            remove_dialog_punctuation,
+            remove_newlines,
+            remove_non_alphabetic,
+            remove_filler_words,
+            lowercase_text,
+            normalize_whitespace,
+        ]
+
+
+    def invoke(self, pipeline: Runnable, splitter: TextSplitter, text: str) -> str | None:
+        """
+        Preprocesses the text. Additionally, splits it to chunks to avoid
+        context overflow
+
+        Args:
+            pipeline (Runnable): pipeline for preprocessing
+            splitter (TextSplitter): text splitter. Defaults to RecursiveCharacterSplitter
+            text (str): subtitles
+
+        Returns:
+            str | None: cleaned text
+        """
+        texts = splitter.split_text(text)
+
+        ready_parts = []
+        for splitted_text in texts:
+            response = pipeline.invoke(splitted_text)
+
+            # Some HuggingFace models struggles with formatted response
+            # So we iterate over all keys and try to find ANY text
+            if self.use_hugging_face:
+                for key in response.keys():
+                    if 'text' in key.lower():
+                        ready_parts.append(response[key].strip())
+                        break
+            else:
+                # Local models have a bit different schema, so they always will have this
+                ready_parts.append(response.CLEANED_TEXT.strip())
+
+        if not len(ready_parts):
+            return None
+
+        return ' '.join(ready_parts).strip()
+
+
+    def analyze_file(self, pipeline: Runnable, splitter: TextSplitter, input_path: Path) -> None:
+        """
+        Loads file with subtitles, cleans it, and saves back
+
+        Args:
+            pipeline (Runnable): preprocessing pipeline
+            splitter (_type_): text splitter
+            input_path (Path): path to subtitles
+        """
+        with open(input_path, 'r', encoding='utf-8') as f:
+            text = f.read()
+
+        print(f'Processing {input_path.name}')
+        start_time = time.time()
+
+        content = self.invoke(pipeline, splitter, text) # cleaning logic
+
+        self.output_path.mkdir(parents=True, exist_ok=True)
+
+        with open(
+            os.path.join(self.output_path, f'{input_path.stem}.csv'), "w"
+        ) as file:
+            file.write(content)
+
+        elapsed = time.time() - start_time
+
+        print(f'Saved to {self.output_path} in {elapsed:.2f}s')
+
+
+    def _select_filenames(self) -> list[Path]:
+        """
+        Collects all the files with subtitles.
+
+        Returns:
+            list[Path]: list with path for each file
+        """
+        names = []
+
+        for filepath in glob.glob(
+            os.path.join(settings.preprocessor.input_path, '*.txt')
+        ):
+            names.append(Path(filepath))
+
+        return names
+
+
+    def _start_worker(self, files: list[Path], worker_id: int = 0) -> None:
+        """
+        Starts worker. Each worker has its own pipeline, set of tools, and
+        text splitter
+
+        Args:
+            files (list[Path]): list of files to be preprocessed
+        """
+        print(f'Worker {worker_id} is preparing')
+
+        # Note: When using multiprocessing, each process will serialize
+        # self and any other referenced object, but langchain tools are not
+        # serializable. So separate method for tools is required
+        tools = self._init_tools()
+        pipeline = self._setup_pipeline(tools)
+        splitter = RecursiveCharacterTextSplitter(
+            chunk_size=settings.preprocessor.chunk_size,
+            chunk_overlap=0,
+            length_function=len
+        )
+
+        print(f'Worker {worker_id} is starting preprocessing')
+
+        for file in files:
+            self.analyze_file(pipeline, splitter, file)
+
+        print(f'Worker {worker_id} is finishing')
+
+
+    def start_preprocessing(self) -> None:
+        """
+        Starts preprocessing.
+        """
+        offset = settings.scraper.offset
+        files = self._select_filenames()
+
+        if self.use_hugging_face:
+            # Map files to workers
+            chunk_size = (len(files) + self.num_workers - 1) // self.num_workers
+            file_mapping = [
+                files[i + offset : i + offset + chunk_size]
+                for i in range(0, len(files), chunk_size)
+            ]
+
+            # Collect worker's args
+            worker_args = [
+                (files, i + 1)
+                for i, files in enumerate(file_mapping)
+            ]
+
+            # Each worker - separate process
+            with Pool(processes=self.num_workers) as p:
+                p.starmap(self._start_worker, worker_args)
+        else:
+            files = files[offset:]
+            self._start_worker(files, 1)

app/main.py

@@ -1,7 +1,9 @@
+
 from fastapi import FastAPI
 
+from .cleaner import PreprocessingAgent
 from .model import get_model
-from .schemas import EmotionResponse, SubtitleRequest
+from .schemas import CleanResponse, EmotionResponse, SubtitleRequest
 
 app = FastAPI(title="Emotion Analyzer API")
 
@@ -19,4 +21,10 @@ async def analyze(req: SubtitleRequest):
 
     return EmotionResponse(
         data=df.to_dict(orient="records")
+    )
+
+@app.post('/clean', response_model=CleanResponse)
+async def clean(req: SubtitleRequest):
+    return CleanResponse(
+        text=PreprocessingAgent.clean_text(req.text)
     )

app/schemas.py

@@ -6,4 +6,7 @@ class SubtitleRequest(BaseModel):
 
 
 class EmotionResponse(BaseModel):
-    data: list[dict]
+    data: list[dict]
+
+class CleanResponse(BaseModel):
+    text: str

app/settings.py

@@ -0,0 +1,189 @@
+import os
+from pathlib import Path
+
+from dotenv import load_dotenv
+from pydantic import BaseModel, Field
+
+load_dotenv() # for manual use (please, ensure that there is .env file in the same folder)
+
+BASE_DIR = Path(__file__).resolve().parent
+
+
+class ScraperSettings(BaseModel):
+    url: str = Field(
+        default='https://subslikescript.com',
+        description="Base URL for scraping"
+    )
+
+    output_folder: str = Field(
+        default=os.path.join(BASE_DIR, 'data'),
+        description="Directory where scraped files will be saved"
+    )
+
+    # Do not set False if using Docker
+    headless: bool = Field(
+        default=False,
+        description="Run browser in headless mode (disable UI)"
+    )
+
+    delay: int = Field(
+        default=3,
+        description="Delay between requests in seconds"
+    )
+
+    num_workers: int = Field(
+        default=4,
+        description="Number of concurrent workers"
+    )
+
+    offset: int = Field(
+        default=386,
+        description="End page for the first worker at the last run"
+    )
+
+
+class EmotionAnalyzerSettings(BaseModel):
+    input_path: str = Field(
+        default='./data',
+        description="Path to preprocessed data"
+    )
+
+    output_path: str = Field(
+        default='./emotion_analysis/embeddings',
+        description="Path to files with embeddings"
+    )
+
+
+class PreprocessorSettings(BaseModel):
+    use_hugging_face: bool = Field(
+        default=False,
+        description="Flag to use propitiate HuggingFace model"
+    )
+
+    local_model_path: str = Field(
+        default='./emotion_analysis/weights/mistral_prep/model.gguf',
+        description="Path to file with local model"
+    )
+
+    model: str = Field(
+        default='Qwen/Qwen2.5-7B-Instruct',
+        description="Model name"
+    )
+
+    input_path: str = Field(
+        default='./data',
+        description="Path to raw subtitles"
+    )
+
+    output_path: str = Field(
+        default=Path('./preprocessing/ready_data'),
+        description="Path to folder with preprocessed data"
+    )
+
+    chunk_size: int = Field(
+        default=4096,
+        description="Each text is divided into chunks since " \
+        "entire sequence can not fit into context"
+    )
+
+    offset: int = Field(
+        default=0,
+        description="Determines how much first files will be skipped "
+        "(in case some have already been preprocessed)"
+    )
+
+
+class DBSettings(BaseModel):
+    db_url: str = Field(
+        default=os.environ['DB_URL'],
+        description="DB connection string"
+    )
+
+
+class GraphSettings(BaseModel):
+    max_depth: int = Field(
+        default=5,
+        description="Maximum graph depth (excluding the highest node (root))"
+    )
+
+    min_samples_leaf: int = Field(
+        default=6,
+        description="Minimum number of movies in each leaf"
+    )
+
+    max_nodes: int = Field(
+        default=800,
+        description="Maximum number of nodes in the graph (important for KMeans)"
+    )
+
+    target_leaf_size: int = Field(
+        default=50,
+        description="Desired number of movies in each leaf"
+    )
+
+    # Important - with current algorithm min/max_fanout can be violated for very small
+    # number of nodes. To understand the reason check the algorithm description
+    # in ./clustering/graph_creator.py
+    min_fanout: int = Field(
+        default=3,
+        description="Minimal number of successors for each node"
+    )
+
+    max_fanout: int = Field(
+        default=8,
+        description="Maximum number of successors for each node"
+    )
+
+
+class NameCreatorSettings(BaseModel):
+    model_path: str = Field(
+        default='./emotion_analysis/weights/qwen/model.gguf',
+        description="Path to model"
+    )
+
+
+class APISettings(BaseModel):
+    app_path: str = Field(
+        default='api.api:app',
+        description='Path to file with FastAPI app'
+    )
+
+    host: str = Field(
+        default='0.0.0.0',
+    )
+
+    port: int = Field(
+        default=int(os.environ['API_PORT']),
+    )
+
+
+class MoviesValidatorSettings(BaseModel):
+    path_to_files: str = Field(
+        default='./movies_for_validation'
+    )
+
+
+class AdminPanelSettings(BaseModel):
+    login: str = Field(
+        default=os.environ['ADMIN_LOGIN']
+    )
+
+    password: str = Field(
+        default=os.environ['ADMIN_PASSWORD']
+    )
+
+
+class Settings(BaseModel):
+    scraper: ScraperSettings = ScraperSettings()
+    emotion_analyzer: EmotionAnalyzerSettings = EmotionAnalyzerSettings()
+    preprocessor: PreprocessorSettings = PreprocessorSettings()
+    name_creator: NameCreatorSettings = NameCreatorSettings()
+    movie_validator: MoviesValidatorSettings = MoviesValidatorSettings()
+    admin: AdminPanelSettings = AdminPanelSettings()
+    db: DBSettings = DBSettings()
+    graph: GraphSettings = GraphSettings()
+    api: APISettings = APISettings()
+    base_dir: str = BASE_DIR
+
+
+settings = Settings()

requirements.txt

@@ -6,4 +6,16 @@ pandas
 numpy
 peft
 accelerate
-bitsandbytes
+bitsandbytes
+langchain==1.2.10
+langchain-classic==1.0.1
+langchain-community==0.4.1
+langchain-core==1.2.13
+langchain-huggingface==1.2.0
+langchain-text-splitters==1.1.0
+langgraph==1.0.8
+langgraph-checkpoint==4.0.0
+langgraph-prebuilt==1.0.7
+langgraph-sdk==0.3.6
+langsmith==0.7.3
+dotenv