Initial commit

.gitignore

@@ -0,0 +1,11 @@
+.DS_Store
+.idea
+*.log
+tmp/
+
+*.py[cod]
+*.egg
+build
+htmlcov
+
+*.db

README.md

@@ -0,0 +1,19 @@
+# Dump your knowledge, let AI refine it
+
+## Starting
+
+Install stuff, then, in one terminal, start the background worker:
+
+```sh
+cd src
+python worker.py
+```
+
+Start the web server:
+
+```sh
+cd src
+uvicorn webservice:app --reload --port 8080
+```
+
+For example requests, check `requests.org`.

environment.yml

@@ -0,0 +1,97 @@
+name: gistillery
+channels:
+  - pytorch
+  - nvidia
+  - defaults
+dependencies:
+  - _libgcc_mutex=0.1=main
+  - _openmp_mutex=5.1=1_gnu
+  - blas=1.0=mkl
+  - bzip2=1.0.8=h7b6447c_0
+  - ca-certificates=2023.01.10=h06a4308_0
+  - cuda-cudart=11.7.99=0
+  - cuda-cupti=11.7.101=0
+  - cuda-libraries=11.7.1=0
+  - cuda-nvrtc=11.7.99=0
+  - cuda-nvtx=11.7.91=0
+  - cuda-runtime=11.7.1=0
+  - filelock=3.9.0=py310h06a4308_0
+  - gmp=6.2.1=h295c915_3
+  - gmpy2=2.1.2=py310heeb90bb_0
+  - intel-openmp=2023.1.0=hdb19cb5_46305
+  - jinja2=3.1.2=py310h06a4308_0
+  - ld_impl_linux-64=2.38=h1181459_1
+  - libcublas=11.10.3.66=0
+  - libcufft=10.7.2.124=h4fbf590_0
+  - libcufile=1.6.1.9=0
+  - libcurand=10.3.2.106=0
+  - libcusolver=11.4.0.1=0
+  - libcusparse=11.7.4.91=0
+  - libffi=3.4.2=h6a678d5_6
+  - libgcc-ng=11.2.0=h1234567_1
+  - libgomp=11.2.0=h1234567_1
+  - libnpp=11.7.4.75=0
+  - libnvjpeg=11.8.0.2=0
+  - libstdcxx-ng=11.2.0=h1234567_1
+  - libuuid=1.41.5=h5eee18b_0
+  - markupsafe=2.1.1=py310h7f8727e_0
+  - mkl=2023.1.0=h6d00ec8_46342
+  - mpc=1.1.0=h10f8cd9_1
+  - mpfr=4.0.2=hb69a4c5_1
+  - ncurses=6.4=h6a678d5_0
+  - networkx=2.8.4=py310h06a4308_1
+  - openssl=1.1.1t=h7f8727e_0
+  - pip=23.0.1=py310h06a4308_0
+  - python=3.10.11=h7a1cb2a_2
+  - pytorch=2.0.0=py3.10_cuda11.7_cudnn8.5.0_0
+  - pytorch-cuda=11.7=h778d358_3
+  - pytorch-mutex=1.0=cuda
+  - readline=8.2=h5eee18b_0
+  - setuptools=66.0.0=py310h06a4308_0
+  - sqlite=3.41.2=h5eee18b_0
+  - sympy=1.11.1=py310h06a4308_0
+  - tbb=2021.8.0=hdb19cb5_0
+  - tk=8.6.12=h1ccaba5_0
+  - torchtriton=2.0.0=py310
+  - typing_extensions=4.5.0=py310h06a4308_0
+  - tzdata=2023c=h04d1e81_0
+  - wheel=0.38.4=py310h06a4308_0
+  - xz=5.4.2=h5eee18b_0
+  - zlib=1.2.13=h5eee18b_0
+  - pip:
+      - anyio==3.6.2
+      - black==23.3.0
+      - certifi==2022.12.7
+      - charset-normalizer==3.1.0
+      - click==8.1.3
+      - fastapi==0.95.1
+      - fsspec==2023.4.0
+      - h11==0.14.0
+      - httptools==0.5.0
+      - huggingface-hub==0.14.1
+      - idna==3.4
+      - mpmath==1.2.1
+      - mypy==1.2.0
+      - mypy-extensions==1.0.0
+      - numpy==1.24.3
+      - packaging==23.1
+      - pathspec==0.11.1
+      - platformdirs==3.5.0
+      - pydantic==1.10.7
+      - python-dotenv==1.0.0
+      - pyyaml==6.0
+      - regex==2023.5.5
+      - requests==2.29.0
+      - ruff==0.0.264
+      - sniffio==1.3.0
+      - starlette==0.26.1
+      - tokenizers==0.13.3
+      - tomli==2.0.1
+      - tqdm==4.65.0
+      - transformers==4.28.1
+      - urllib3==1.26.15
+      - uvicorn==0.22.0
+      - uvloop==0.17.0
+      - watchfiles==0.19.0
+      - websockets==11.0.2
+prefix: /home/vinh/anaconda3/envs/gistillery

pyproject.toml

@@ -0,0 +1,14 @@
+[tool.black]
+line-length = 88
+target_version = ['py310', 'py311']
+preview = true
+
+[tool.isort]
+profile = "black"
+
+[tool.mypy]
+no_implicit_optional = true
+strict = true
+
+[[tool.mypy-transformers]]
+ignore_missing_imports = true

requests.org

@@ -0,0 +1,57 @@
+#+title: Requests
+
+#+begin_src bash
+curl -X 'GET' \
+  'http://localhost:8080/clear/' \
+  -H 'accept: application/json'
+#+end_src
+
+#+RESULTS:
+: OK
+
+#+begin_src bash
+# curl command to localhost and post the message "hi there"
+curl -X 'POST' \
+  'http://localhost:8080/submit/' \
+  -H 'accept: application/json' \
+  -H 'Content-Type: application/json' \
+  -d '{
+  "author": "ben",
+  "content": "SAN FRANCISCO, May 2, 2023 PRNewswire -- GitLab Inc., the most comprehensive, scalable enterprise DevSecOps platform for software innovation, and Google Cloud today announced an extension of its strategic partnership to deliver secure AI offerings to the enterprise. GitLab is trusted by more than 50% of the Fortune 100 to secure and protect their most valuable assets, and leads with a privacy-first approach to AI. By leveraging Google Cloud'\''s customizable foundation models and open generative AI infrastructure, GitLab will provide customers with AI-assisted features directly within the enterprise DevSecOps platform."
+}'
+#+end_src
+
+#+RESULTS:
+: Submitted job 04d44970ced1473999dfab77b02202b8
+
+#+begin_src bash
+curl -X 'POST' \
+  'http://localhost:8080/submit/' \
+  -H 'accept: application/json' \
+  -H 'Content-Type: application/json' \
+  -d '{
+  "author": "ben",
+  "content": "In literature discussing why ChatGPT is able to capture so much of our imagination, I often come across two narratives: Scale: throwing more data and compute at it. UX: moving from a prompt interface to a more natural chat interface. A narrative that is often glossed over in the demo frenzy is the incredible technical creativity that went into making models like ChatGPT work. One such cool idea is RLHF (Reinforcement Learning from Human Feedback): incorporating reinforcement learning and human feedback into NLP. RL has been notoriously difficult to work with, and therefore, mostly confined to gaming and simulated environments like Atari or MuJoCo. Just five years ago, both RL and NLP were progressing pretty much orthogonally – different stacks, different techniques, and different experimentation setups. It’s impressive to see it work in a new domain at a massive scale. So, how exactly does RLHF work? Why does it work? This post will discuss the answers to those questions."
+}'
+#+end_src
+
+#+RESULTS:
+: Submitted job 3cc2104aec0748b1bd5743c321b169ac
+
+#+begin_src bash
+curl -X 'GET' \
+  'http://localhost:8080/check_status/22b158499b744f42918912cd387fd657' \
+  -H 'accept: application/json'
+#+end_src
+
+#+RESULTS:
+| {"id":"22b158499b744f42918912cd387fd657" | status:"done" | last_updated:"2023-05-05T14:54:11"} |
+
+#+begin_src bash
+curl -X 'GET' \
+  'http://localhost:8080/recent/' \
+  -H 'accept: application/json'
+#+end_src
+
+#+RESULTS:
+| [{"id":"3cc2104aec0748b1bd5743c321b169ac" | author:"ben" | summary:"A new approach to NLP that incorporates reinforcement learning and human feedback. How does it work? Why does it work? In this post | I’ll explain how it works. RLHF is a new approach to NLP that incorporates reinforcement learning and human feedback. It’s a new approach to NLP that incorporates reinforcement learning and human feedback. It’s a new approach to NLP that incorporates reinforcement learning and human feedback. It’s a new approach to NLP that incorporates reinforcement learning and human feedback. It’s a new approach to NLP that incorporates reinforcement learning and human feedback." | tags:["#general" | #rlhf] | date:"2023-05-05T14:56:32"} | {"id":"04d44970ced1473999dfab77b02202b8" | author:"ben" | summary:"GitLab | the most comprehensive | scalable enterprise DevSecOps platform for software innovation | and Google Cloud today announced an extension of their strategic partnership to deliver secure AI offerings to the enterprise. By leveraging Google Cloud's customizable foundation models and open generative AI infrastructure | GitLab will provide customers with AI-assisted features directly within the enterprise DevSecOps platform. The company's AI capabilities are designed to help enterprises improve productivity and reduce costs." | tags:["#general"] | date:"2023-05-05T14:56:31"}] |

requirements-dev.txt

@@ -0,0 +1,4 @@
+black
+isort
+mypy
+ruff

requirements.txt

@@ -0,0 +1,5 @@
+fastapi
+httpx
+uvicorn[standard]
+torch
+transformers

src/base.py

@@ -0,0 +1,37 @@
+import datetime as dt
+import enum
+
+from pydantic import BaseModel
+
+
+class RequestInput(BaseModel):
+    author: str
+    content: str
+
+
+class EntriesResult(BaseModel):
+    id: str
+    author: str
+    summary: str
+    tags: list[str]
+    date: dt.datetime
+
+
+class JobInput(BaseModel):
+    id: str
+    author: str
+    content: str
+
+
+class JobStatus(str, enum.Enum):
+    pending = "pending"
+    done = "done"
+    failed = "failed"
+    cancelled = "cancelled"
+    not_found = "not found"
+
+
+class JobStatusResult(BaseModel):
+    id: str
+    status: JobStatus
+    last_updated: dt.datetime | None

src/db.py

@@ -0,0 +1,102 @@
+import logging
+import sqlite3
+from contextlib import contextmanager
+from typing import Generator
+
+logger = logging.getLogger(__name__)
+logger.setLevel(logging.DEBUG)
+
+
+schema_entries = """
+CREATE TABLE entries
+(
+    id TEXT PRIMARY KEY,
+    author TEXT NOT NULL,
+    source TEXT NOT NULL,
+    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+)
+"""
+
+# create schema for 'summary' table, id is a uuid4
+schema_summary = """
+CREATE TABLE summaries
+(
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    entry_id TEXT NOT NULL,
+    summary TEXT NOT NULL,
+    summarizer_name TEXT NOT NULL,
+    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+    FOREIGN KEY(entry_id) REFERENCES entries(id)
+)
+"""
+
+schema_tag = """
+CREATE TABLE tags
+(
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    entry_id TEXT NOT NULL,
+    tag TEXT NOT NULL,
+    tagger_name TEXT NOT NULL,
+    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+    FOREIGN KEY(entry_id) REFERENCES entries(id)
+)
+"""
+
+schema_job = """
+CREATE TABLE jobs
+(
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    entry_id TEXT NOT NULL,
+    status TEXT NOT NULL DEFAULT 'pending',
+    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+    last_updated TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+    FOREIGN KEY(entry_id) REFERENCES entries(id)
+)
+"""
+
+TABLES = {
+    "entries": schema_entries,
+    "summaries": schema_summary,
+    "tags": schema_tag,
+    "jobs": schema_job,
+}
+TABLES_CREATED = False
+
+
+def _get_db_connection() -> sqlite3.Connection:
+    global TABLES_CREATED
+
+    # sqlite cannot deal with concurrent access, so we set a big timeout
+    conn = sqlite3.connect("sqlite-data.db", timeout=30)
+    if TABLES_CREATED:
+        return conn
+
+    cursor = conn.cursor()
+
+    # create tables if needed
+    for table_name, schema in TABLES.items():
+        cursor.execute(
+            "SELECT name FROM sqlite_master WHERE type='table' AND name=?",
+            (table_name,),
+        )
+        table_exists = cursor.fetchone() is not None
+        if not table_exists:
+            logger.info(f"'{table_name}' table does not exist, creating it now...")
+            cursor.execute(schema)
+            conn.commit()
+            logger.info("done")
+
+    TABLES_CREATED = True
+    return conn
+
+
+@contextmanager
+def get_db_cursor() -> Generator[sqlite3.Cursor, None, None]:
+    conn = _get_db_connection()
+    cursor = conn.cursor()
+    try:
+        yield cursor
+    finally:
+        conn.commit()
+        cursor.close()
+        conn.close()

src/ml.py

@@ -0,0 +1,143 @@
+import abc
+import logging
+import re
+
+import httpx
+from transformers import AutoModelForSeq2SeqLM, AutoTokenizer, GenerationConfig
+
+from base import JobInput
+
+logger = logging.getLogger(__name__)
+logger.setLevel(logging.DEBUG)
+
+MODEL_NAME = "google/flan-t5-large"
+model = AutoModelForSeq2SeqLM.from_pretrained(MODEL_NAME)
+tokenizer = AutoTokenizer.from_pretrained(MODEL_NAME)
+
+
+class Summarizer:
+    def __init__(self) -> None:
+        self.template = "Summarize the text below in two sentences:\n\n{}"
+        self.generation_config = GenerationConfig.from_pretrained(MODEL_NAME)
+        self.generation_config.max_new_tokens = 200
+        self.generation_config.min_new_tokens = 100
+        self.generation_config.top_k = 5
+        self.generation_config.repetition_penalty = 1.5
+
+    def __call__(self, x: str) -> str:
+        text = self.template.format(x)
+        inputs = tokenizer(text, return_tensors="pt")
+        outputs = model.generate(**inputs, generation_config=self.generation_config)
+        output = tokenizer.batch_decode(outputs, skip_special_tokens=True)[0]
+        assert isinstance(output, str)
+        return output
+
+    def get_name(self) -> str:
+        return f"Summarizer({MODEL_NAME})"
+
+
+class Tagger:
+    def __init__(self) -> None:
+        self.template = (
+            "Create a list of tags for the text below. The tags should be high level "
+            "and specific. Prefix each tag with a hashtag.\n\n{}\n\nTags: #general"
+        )
+        self.generation_config = GenerationConfig.from_pretrained(MODEL_NAME)
+        self.generation_config.max_new_tokens = 50
+        self.generation_config.min_new_tokens = 25
+        # increase the temperature to make the model more creative
+        self.generation_config.temperature = 1.5
+
+    def _extract_tags(self, text: str) -> list[str]:
+        tags = set()
+        for tag in text.split():
+            if tag.startswith("#"):
+                tags.add(tag.lower())
+        return sorted(tags)
+
+    def __call__(self, x: str) -> list[str]:
+        text = self.template.format(x)
+        inputs = tokenizer(text, return_tensors="pt")
+        outputs = model.generate(**inputs, generation_config=self.generation_config)
+        output = tokenizer.batch_decode(outputs, skip_special_tokens=True)[0]
+        tags = self._extract_tags(output)
+        return tags
+
+    def get_name(self) -> str:
+        return f"Tagger({MODEL_NAME})"
+
+
+class Processor(abc.ABC):
+    def __call__(self, job: JobInput) -> str:
+        _id = job.id
+        logger.info(f"Processing {input} with {self.__class__.__name__} (id={_id[:8]})")
+        result = self.process(job)
+        logger.info(f"Finished processing input (id={_id[:8]})")
+        return result
+
+    def process(self, input: JobInput) -> str:
+        raise NotImplementedError
+
+    def match(self, input: JobInput) -> bool:
+        raise NotImplementedError
+
+    def get_name(self) -> str:
+        raise NotImplementedError
+
+
+class RawProcessor(Processor):
+    def match(self, input: JobInput) -> bool:
+        return True
+
+    def process(self, input: JobInput) -> str:
+        return input.content
+
+    def get_name(self) -> str:
+        return self.__class__.__name__
+
+
+class PlainUrlProcessor(Processor):
+    def __init__(self) -> None:
+        self.client = httpx.Client()
+        self.regex = re.compile(r"(https?://[^\s]+)")
+        self.url = None
+        self.template = "{url}\n\n{content}"
+
+    def match(self, input: JobInput) -> bool:
+        urls = list(self.regex.findall(input.content))
+        if len(urls) == 1:
+            self.url = urls[0]
+            return True
+        return False
+
+    def process(self, input: JobInput) -> str:
+        """Get content of website and return it as string"""
+        assert isinstance(self.url, str)
+        text = self.client.get(self.url).text
+        assert isinstance(text, str)
+        text = self.template.format(url=self.url, content=text)
+        return text
+
+    def get_name(self) -> str:
+        return self.__class__.__name__
+
+
+class ProcessorRegistry:
+    def __init__(self) -> None:
+        self.registry: list[Processor] = []
+        self.default_registry: list[Processor] = []
+        self.set_default_processors()
+
+    def set_default_processors(self) -> None:
+        self.default_registry.extend([PlainUrlProcessor(), RawProcessor()])
+
+    def register(self, processor: Processor) -> None:
+        self.registry.append(processor)
+
+    def dispatch(self, input: JobInput) -> Processor:
+        for processor in self.registry + self.default_registry:
+            if processor.match(input):
+                return processor
+
+        # should never be requires, but eh
+        return RawProcessor()

src/webservice.py

@@ -0,0 +1,112 @@
+import logging
+import uuid
+
+from fastapi import FastAPI
+
+from base import EntriesResult, JobStatus, JobStatusResult, RequestInput
+from db import TABLES, get_db_cursor
+
+
+logger = logging.getLogger(__name__)
+logger.setLevel(logging.DEBUG)
+
+
+app = FastAPI()
+
+
+@app.post("/submit/")
+def submit_job(input: RequestInput) -> str:
+    # submit a new job, poor man's job queue
+    _id = uuid.uuid4().hex
+    logger.info(f"Submitting job for (_id={_id[:8]})")
+
+    with get_db_cursor() as cursor:
+        # create a job
+        query = "INSERT INTO jobs (entry_id, status) VALUES (?, ?)"
+        cursor.execute(query, (_id, "pending"))
+        # create an entry
+        query = "INSERT INTO entries (id, author, source) VALUES (?, ?, ?)"
+        cursor.execute(query, (_id, input.author, input.content))
+
+    return f"Submitted job {_id}"
+
+
+@app.get("/check_status/{_id}")
+def check_status(_id: str) -> JobStatusResult:
+    with get_db_cursor() as cursor:
+        cursor.execute(
+            "SELECT status, last_updated FROM jobs WHERE entry_id = ?", (_id,)
+        )
+        result = cursor.fetchone()
+
+    if result is None:
+        return JobStatusResult(id=_id, status=JobStatus.not_found, last_updated=None)
+
+    status, last_updated = result
+    return JobStatusResult(id=_id, status=status, last_updated=last_updated)
+
+
+@app.get("/recent/")
+def recent() -> list[EntriesResult]:
+    with get_db_cursor() as cursor:
+        # get the last 10 entries, join summary and tag, where each tag is
+        # joined to a comma separated str
+        cursor.execute("""
+            SELECT e.id, e.author, s.summary, GROUP_CONCAT(t.tag, ","), e.created_at
+            FROM entries e
+            JOIN summaries s ON e.id = s.entry_id
+            JOIN tags t ON e.id = t.entry_id
+            GROUP BY e.id
+            ORDER BY e.created_at DESC
+            LIMIT 10
+            """)
+        results = cursor.fetchall()
+
+    entries = []
+    for _id, author, summary, tags, date in results:
+        entry = EntriesResult(
+            id=_id, author=author, summary=summary, tags=tags.split(","), date=date
+        )
+        entries.append(entry)
+    return entries
+
+
+@app.get("/recent/{tag}")
+def recent_tag(tag: str) -> list[EntriesResult]:
+    if not tag.startswith("#"):
+        tag = "#" + tag
+
+    # same as recent, but filter by tag
+    with get_db_cursor() as cursor:
+        cursor.execute(
+            """
+            SELECT e.id, e.author, s.summary, GROUP_CONCAT(t.tag, ","), e.created_at
+            FROM entries e
+            JOIN summaries s ON e.id = s.entry_id
+            JOIN tags t ON e.id = t.entry_id
+            WHERE t.tag = ?
+            GROUP BY e.id
+            ORDER BY e.created_at DESC
+            LIMIT 10
+            """,
+            (tag,),
+        )
+        results = cursor.fetchall()
+
+    entries = []
+    for _id, author, summary, tags, date in results:
+        entry = EntriesResult(
+            id=_id, author=author, summary=summary, tags=tags.split(","), date=date
+        )
+        entries.append(entry)
+    return entries
+
+
+@app.get("/clear/")
+def clear() -> str:
+    # clear all tables
+    logger.warning("Clearing all tables")
+    with get_db_cursor() as cursor:
+        for table_name in TABLES:
+            cursor.execute(f"DELETE FROM {table_name}")
+    return "OK"

src/worker.py

@@ -0,0 +1,117 @@
+import time
+
+from base import JobInput
+from db import get_db_cursor
+from ml import ProcessorRegistry, Summarizer, Tagger
+
+SLEEP_INTERVAL = 5
+
+
+processor_registry = ProcessorRegistry()
+summarizer = Summarizer()
+tagger = Tagger()
+print("loaded ML models")
+
+
+def check_pending_jobs() -> list[JobInput]:
+    """Check DB for pending jobs"""
+    with get_db_cursor() as cursor:
+        # fetch pending jobs, join authro and content from entries table
+        query = """
+        SELECT j.entry_id, e.author, e.source
+        FROM jobs j
+        JOIN entries e
+        ON j.entry_id = e.id
+        WHERE j.status = 'pending'
+        """
+        res = list(cursor.execute(query))
+    return [
+        JobInput(id=_id, author=author, content=content) for _id, author, content in res
+    ]
+
+
+def store(
+    job: JobInput,
+    *,
+    summary: str,
+    tags: list[str],
+    processor_name: str,
+    summarizer_name: str,
+    tagger_name: str,
+) -> None:
+    with get_db_cursor() as cursor:
+        # write to entries, summary, tags tables
+        cursor.execute(
+            (
+                "INSERT INTO summaries (entry_id, summary, summarizer_name)"
+                " VALUES (?, ?, ?)"
+            ),
+            (job.id, summary, summarizer_name),
+        )
+        cursor.executemany(
+            "INSERT INTO tags (entry_id, tag, tagger_name) VALUES (?, ?, ?)",
+            [(job.id, tag, tagger_name) for tag in tags],
+        )
+
+
+def process_job(job: JobInput) -> None:
+    tic = time.perf_counter()
+    print(f"Processing job for (id={job.id[:8]})")
+
+    # care: acquire cursor (which leads to locking) as late as possible, since
+    # the processing and we don't want to block other workers during that time
+    try:
+        processor = processor_registry.dispatch(job)
+        processor_name = processor.get_name()
+        processed = processor(job)
+
+        tagger_name = tagger.get_name()
+        tags = tagger(processed)
+
+        summarizer_name = summarizer.get_name()
+        summary = summarizer(processed)
+
+        store(
+            job,
+            summary=summary,
+            tags=tags,
+            processor_name=processor_name,
+            summarizer_name=summarizer_name,
+            tagger_name=tagger_name,
+        )
+        # update job status to done
+        with get_db_cursor() as cursor:
+            cursor.execute(
+                "UPDATE jobs SET status = 'done' WHERE entry_id = ?", (job.id,)
+            )
+    except Exception as e:
+        # update job status to failed
+        with get_db_cursor() as cursor:
+            cursor.execute(
+                "UPDATE jobs SET status = 'failed' WHERE entry_id = ?", (job.id,)
+            )
+        print(f"Failed to process job for (id={job.id[:8]}): {e}")
+
+    toc = time.perf_counter()
+    print(f"Finished processing job (id={job.id[:8]}) in {toc - tic:0.3f} seconds")
+
+
+def main() -> None:
+    while True:
+        jobs = check_pending_jobs()
+        if not jobs:
+            print("No pending jobs found, sleeping...")
+            time.sleep(SLEEP_INTERVAL)
+            continue
+
+        print(f"Found {len(jobs)} pending job(s), processing...")
+        for job in jobs:
+            process_job(job)
+
+
+if __name__ == "__main__":
+    try:
+        main()
+    except KeyboardInterrupt:
+        print("Shutting down...")
+        exit(0)