chainyo

from __future__ import annotations

from dataclasses import dataclass, field
from typing import TYPE_CHECKING, Any, Literal, Union

from lightrag.core import Component, DataClass, Generator, GeneratorOutput
from lightrag.components.model_client import GroqAPIClient
from lightrag.components.output_parsers import JsonOutputParser

chainyo

from typing import List, Optional, Union

import patito as pt
import polars as pl


def normalize_list_ids(v: Union[str, List[str], None]) -> Union[List[str], None]:
    """Normalize the ids to a list."""
    if v is None or isinstance(v, list):
        return v

chainyo

    def download_dir(self, s3_path, local_path, bucket: str = None):
        """
        Fonction permettant le téléchargement les fichiers d'un répertoire entier depuis un bucket S3.

        Parameters
        ----------
        s3_path: str
            path du fichier ou dossier sur S3.
        local_path: str
            path du fichier ou dossier en local.

chainyo

#!/bin/sh

set -e

# activate our virtual environment here
. /opt/pysetup/.venv/bin/activate

# You can put other setup logic here

# Evaluating passed command:

chainyo

FROM python:3.10-slim as python-base

ENV PYTHONUNBUFFERED=1 \
    PYTHONDONTWRITEBYTECODE=1 \
    PIP_NO_CACHE_DIR=off \
    PIP_DISABLE_PIP_VERSION_CHECK=on \
    PIP_DEFAULT_TIMEOUT=100 \
    POETRY_HOME="/opt/poetry" \
    POETRY_VIRTUALENVS_IN_PROJECT=true \
    POETRY_NO_INTERACTION=1 \

chainyo

async def send_summary_as_dm(
        self,
        guild: discord.Guild,
        user: discord.User,
        summary_size: str,
        timeframe: str,
        language: str,
        job_name: str,
        token: str,
        summarized_chat: Optional[List[str]] = None,

chainyo

import logging
from datetime import datetime, timedelta
from pytimeparse import parse
from typing import Dict, List, Optional

import discord
from discord import app_commands

from wordcab import start_summary
from wordcab.core_objects import InMemorySource

chainyo

async def add_a_guild(self, discord_guild_id: int, guild_owner_id: int):
    """Add a guild."""
    async with AsyncSession(self.engine) as session:
        guild = await session.exec(select(Guilds).where(Guilds.discord_guild_id == discord_guild_id))
        try:
            guild = guild.one()
            print("Guild already exists.")
        except NoResultFound:
            guild = Guilds(discord_guild_id=discord_guild_id, guild_owner_id=guild_owner_id)
            session.add(guild)

chainyo

import os
from dotenv import load_dotenv

from sqlalchemy.ext.asyncio import create_async_engine

from sqlmodel import SQLModel
from sqlmodel.ext.asyncio.session import AsyncSession


load_dotenv()

chainyo

from typing import Optional
from sqlmodel import Field, SQLModel


class Guilds(SQLModel, table=True):
    """Guilds table."""
    id: Optional[int] = Field(default=None, primary_key=True)
    logged_in: bool = Field(default=False)
    discord_guild_id: int = Field(unique=True, index=True)
    guild_owner_id: int