purescript_linked_in/native/src/job_search/messages.py

import re
from datetime import date
from dataclasses import dataclass, asdict
from enum import Enum
from typing import Optional, Any
from job_search.job_storage import (
    JobOffer,
    ApplicationProcess,
    JobOfferOrigin,
    Flexibility,
)


def to_snake_case(string):
    return "".join("_" + c.lower() if c.isupper() else c for c in string)


class Message:
    ...


class BackgroundScriptMessage(Message):
    @staticmethod
    def interpret(message):
        if not isinstance(message, dict):
            raise TypeError(f"message should be a dict, got {type(message)}")

        try:
            tag = message.pop("tag")
        except KeyError:
            raise ValueError("message should contain a tag")

        message = {to_snake_case(k): v for k, v in message.items()}

        match tag:
            case "visited_linkedin_job_page":
                return VisitedLinkedInJobPageMessage(**message)
            case "initial_configuration":
                return InitialConfigurationMessage(**message)
            case _:
                raise ValueError(f"Got message with unknown tag {tag}")


class NativeMessage(Message):
    def serialize(self):
        if isinstance(self, JobOfferListMessage):
            tag = "job_offer_list"
        elif isinstance(self, JobAddedMessage):
            tag = "job_added"
        elif isinstance(self, JobAlreadyExistsMessage):
            tag = "job_already_exists"
        elif isinstance(self, LogMessage):
            tag = "log_message"
        else:
            raise TypeError(f"No tag was associated to {type(self)} for serialization")

        return asdict(self) | {"tag": tag}


@dataclass
class VisitedLinkedInJobPageMessage(BackgroundScriptMessage):
    url: str
    job_title: str
    page_title: str
    company: str
    location: str
    has_simplified_process: bool
    company_url: str
    flexibility: Optional[str] = None
    company_domain: Optional[str] = None

    def extract_job_offer(self):
        application_process = (
            ApplicationProcess.LINKED_IN_SIMPLIFIED
            if self.has_simplified_process
            else ApplicationProcess.REGULAR
        )

        if isinstance(self.flexibility, str):
            flexibility = Flexibility(self.flexibility)
        elif self.flexibility is None:
            flexibility = None

        return JobOffer(
            url=self.url,
            title=self.job_title,
            company=self.company,
            origin=JobOfferOrigin.LINKED_IN,
            application_process=application_process,
            location=self.location,
            company_domain=self.company_domain,
            company_url=self.company_url,
            flexibility=flexibility,
        )


@dataclass
class InitialConfigurationMessage(BackgroundScriptMessage):
    jobs_path: str


@dataclass
class JobOfferListMessage(NativeMessage):
    job_offers: list[JobOffer]


@dataclass
class JobAddedMessage(NativeMessage):
    job: JobOffer


@dataclass
class JobAlreadyExistsMessage(NativeMessage):
    job_id: str


class LogLevel(Enum):
    DEBUG = "debug"
    INFO = "info"
    ERROR = "error"


@dataclass
class LogMessage(NativeMessage):
    level: LogLevel
    content: Any

    @staticmethod
    def debug(**kwargs):
        return LogMessage(level=LogLevel.DEBUG, **kwargs)

    @staticmethod
    def info(**kwargs):
        return LogMessage(level=LogLevel.INFO, **kwargs)

    @staticmethod
    def error(**kwargs):
        return LogMessage(level=LogLevel.ERROR, **kwargs)