Merge pull request 'testing' (#16) from testing into main

Reviewed-on: #16

django_code/apps/Le_journal_10h00_du_30_06_2026.txt

@@ -0,0 +1 @@
+ikgdsa

django_code/apps/betaserie/utils.py

@@ -104,3 +104,22 @@ class betaserieUtils():
             print(response.text)
 
         return response.status_code
+
+    def series_information(self,show_id,verbose=False):
+        #
+        #   Usage
+        #
+        url = "https://api.betaseries.com/shows/display?id="+str(show_id)
+        response = requests.request("GET", url, headers=self.headers, data=self.payload)
+        
+        maximum_number_seasons = response.json()['show']['seasons']
+        if verbose : print("maximum_number_seasons: "+str(maximum_number_seasons))
+        
+        seasons_details = response.json()['show']['seasons_details']
+        
+        if verbose:
+            print("Url request: "+str(url))
+            print(response.text)
+
+        return seasons_details,maximum_number_seasons
+        

django_code/apps/podcast_aggregator/admin.py

@@ -0,0 +1,9 @@
+from django.contrib import admin
+from import_export.admin import ImportExportModelAdmin
+from import_export import resources
+from .models import podcastData
+
+# Register your models here.
+@admin.register(podcastData)
+class JobAdmin(ImportExportModelAdmin):
+    list_display = ('podcast_name','podcast_url','enabled')

django_code/apps/podcast_aggregator/apps.py

@@ -0,0 +1,7 @@
+from django.apps import AppConfig
+
+
+class PodcastAggregatorConfig(AppConfig):
+    default_auto_field = 'django.db.models.BigAutoField'
+    name = 'apps.podcast_aggregator'
+    label = 'podcast'

django_code/apps/podcast_aggregator/migrations/0001_initial.py

@@ -0,0 +1,25 @@
+# Generated by Django 4.2.29 on 2026-06-30 22:05
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    initial = True
+
+    dependencies = [
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='podcastData',
+            fields=[
+                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('podcast_name', models.CharField(max_length=100, null=True)),
+                ('podcast_url', models.CharField(max_length=100, null=True)),
+            ],
+            options={
+                'db_table': 'podcasts',
+            },
+        ),
+    ]

django_code/apps/podcast_aggregator/migrations/0002_podcastdata_enabled.py

@@ -0,0 +1,18 @@
+# Generated by Django 4.2.29 on 2026-06-30 22:08
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('podcast', '0001_initial'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='podcastdata',
+            name='enabled',
+            field=models.BooleanField(default=True),
+        ),
+    ]

django_code/apps/podcast_aggregator/models.py

@@ -0,0 +1,8 @@
+from django.db import models
+
+class podcastData(models.Model):
+    podcast_name = models.CharField(max_length=100, null=True)
+    podcast_url = models.CharField(max_length=100, null=True)
+    enabled = models.BooleanField(default=True)
+    class Meta:
+        db_table = 'podcasts'

django_code/apps/podcast_aggregator/tests.py

@@ -0,0 +1,3 @@
+from django.test import TestCase
+
+# Create your tests here.

django_code/apps/podcast_aggregator/utils.py

@@ -0,0 +1,78 @@
+import requests
+import xmltodict
+from urllib import request as urlrequest
+from apps.telegram_communication.utils import telegramUtils
+from .models import podcastData
+
+
+proxy_url = 'http://gluetun-svc.gluetun.svc:8888'
+proxy = urlrequest.ProxyHandler({
+    'http': proxy_url,
+    'https': proxy_url  # Add the handler for HTTPS
+})
+
+proxies = {
+    "http"  : proxy_url,
+    "https" : proxy_url
+}
+
+def rss_to_dict(url):
+    response = requests.get(url)
+    data = xmltodict.parse(response.content)
+    
+    print(data['rss'])
+    
+    title = data['rss']['channel']['title']
+    
+    #Get the keys if needed of debug 
+    #print(data['rss'].keys())
+    
+    #use first object as it's ordered in a chronological order
+    
+    
+    today_object = data['rss']['channel']['item']
+    if isinstance(today_object, list):
+        today_object = data['rss']['channel']['item'][0]
+        
+    today_title = today_object['title']
+    today_podcast_url = today_object['enclosure']['@url']
+        
+    return title,today_title,today_podcast_url
+
+def download_news_mp3(url_mp3, filename):
+
+    opener = urlrequest.build_opener(proxy)
+
+    urlrequest.install_opener(opener)
+    
+    print("Downloading : " + filename +" -- " + url_mp3)
+    try:
+        urlrequest.urlretrieve(url_mp3, filename)
+    except:
+        print("Can't download " + filename +" -- " + url_mp3 + "  == Cancelling")
+    return 0
+
+def pretty_filename(filename,extension=''):
+    filename = filename.replace(' ', '_')
+    pretty_filename = filename.replace('/', '_')
+    if extension != '':
+        extension = str('.'+extension)
+    return pretty_filename+extension
+
+def podcasts():
+    telegram = telegramUtils()
+    
+    #clear all expired messages before
+    telegram.clear_expired_messages()
+
+    podcasts = podcastData.objects.filter(enabled=True)
+    for podcast in podcasts:
+        url = podcast.podcast_url
+       
+        today_podcast = rss_to_dict(url)
+        print(today_podcast)
+        file= pretty_filename(today_podcast[1],'mp3')
+        download_news_mp3(today_podcast[2],file)
+        telegram.send_document(file,str(today_podcast[0])+" "+str(today_podcast[1]))
+
+#main()

django_code/apps/podcast_aggregator/views.py

@@ -0,0 +1,7 @@
+from django.http import HttpResponse
+from .utils import podcasts
+
+# Create your views here.
+def execute_podcast(request):
+    podcasts()
+    return HttpResponse('<h1>Generating the podcast!</h1>')

django_code/apps/random_simpsons/admin.py

@@ -2,7 +2,7 @@ from django.contrib import admin,messages
 from django.utils.translation import ngettext
 from prompt_toolkit.shortcuts import print_container
 from .models import SimpsonsData,simpsons_data_object
-from .utils import mark_episode_watched,unwatch_all_episodes
+from .utils import mark_episode_watched,unwatch_all_episodes,generate_new_urls_for_episode
 from import_export import resources
 from import_export.admin import ImportExportModelAdmin
 # Register your models here.
@@ -54,13 +54,18 @@ def make_unseen(modeladmin, request, queryset):
 def unseen_betaserie(modeladmin, request, queryset):
     unwatch_all_episodes()
 
+@admin.action(description="generate new urls for episode")
+def generate_new_urls(modeladmin, request, queryset):
+    generate_new_urls_for_episode()
+
+
 
 @admin.register(SimpsonsData)
 class EpisodeAdmin(ImportExportModelAdmin):
     list_display = [field.name for field in SimpsonsData._meta.fields if field.name != 'id']
     list_filter = ['seen']
     fields = ['season', 'episode', 'url', 'seen']
-    actions = [make_seen,make_unseen,unseen_betaserie]
+    actions = [make_seen,make_unseen,unseen_betaserie,generate_new_urls]
 
     def get_queryset(self, request):
         return super().get_queryset(request).using(DB)

django_code/apps/random_simpsons/utils.py

@@ -1,18 +1,25 @@
-from apps.betaserie.utils import betaserieUtils
 from django.db.models import Count
+from django.core.exceptions import MultipleObjectsReturned
 from celery.worker.state import active_requests
+from celery import shared_task
 from requests.compat import str
+from apps.betaserie.utils import betaserieUtils
 from .models import simpsons_data_object
 
 betaserie = betaserieUtils()
 
 simpsons_id = 12
 
-def mark_episode_watched(season,episode,action):
-    
-    print("season_number: "+str(season))
-    print("episode_number: "+str(episode))
-    
+@shared_task()
+def mark_episode_watched(season,episode,action,verbose=False):
+    #mark episode as seen in DB
+    simpsons_data_object.filter(season=season, episode=episode).values('seen').update(seen=True)
+
+
+    if verbose : print("season_number: "+str(season))
+    if verbose : print("episode_number: "+str(episode))
+
+    #mark episode as seen in betaserie
     episode_number = "S"+str(season)+"E"+str(episode)
     
     episode_id = betaserie.identify_episode(simpsons_id,episode_number)
@@ -35,3 +42,49 @@ def unwatch_all_episodes():
         for episode in range(1,maximum_number_of_episodes+1):
             print("S"+str(season)+"E"+str(episode))
             mark_episode_watched(season,episode,"unwatched")
+
+
+def generate_new_urls_for_episode(verbose=False):
+    # kisscartoons didn't work anymore so moving to another website with the format:
+    # URL.com/simpsons/season=XX&episode=XX
+    # example
+    # https://www.1shows.org/tv/456-the-simpsons?streaming=true&season=33&episode=18
+    
+    simpsons_series_information = betaserie.series_information(simpsons_id)
+    seasons_information = simpsons_series_information[0]    
+    
+    #for the simpsons, there is a season 0 which increase the count of seasons by 1
+    # we need to remove this entry , 
+    # so we cannot trust the maximum_number_season coming from simpsons_series_information
+    
+    season_0_to_remove = seasons_information[0]
+    
+    if verbose : print(season_0_to_remove)
+    
+    seasons_information.remove(season_0_to_remove)
+    maximum_number_season = len(seasons_information)
+    
+    for season in range(1,maximum_number_season+1):
+        # the list doesn't start with season 0 anymore so we cannot use the season number as index 
+        # we need to add -1
+    
+        number_episodes_in_season = int(seasons_information[season-1]['episodes'])
+    
+        for episode in range(1,number_episodes_in_season+1):
+            if verbose : print("Season: "+str(season))
+            if verbose : print("episode: "+str(episode))
+            new_url = "https://www.1shows.org/tv/456-the-simpsons?streaming=true&season="+str(season)+"&episode="+str(episode)
+    
+            if verbose : print(new_url)
+            try:
+                obj,created = simpsons_data_object.get_or_create(
+                    season=season,
+                    episode=episode,
+                    defaults={"url":new_url,"seen":"False"}
+                )
+                obj.url=new_url
+                obj.save()
+            except MultipleObjectsReturned:
+                print("duplicated object for season: "+str(episode)+" episode: "+str(episode)+", ignoring")
+    
+    return 0

django_code/apps/random_simpsons/views.py

@@ -36,10 +36,9 @@ def episode_generator(request):
         #Set episode as seen
         # It will still update an already seem episode but it works and I'm lazy
         
-        simpsons_data_object.filter(season=random_seasons, episode=random_episode).values('seen').update(seen=True)
-
+        
         #Mark episode as seen on betaseries
-        mark_episode_watched(random_seasons,random_episode,"watched")
+        mark_episode_watched.delay(random_seasons,random_episode,"watched")
    
     return redirect(str(random_url))
 

django_code/apps/telegram_communication/admin.py

@@ -0,0 +1,8 @@
+from django.contrib import admin
+from import_export.admin import ImportExportModelAdmin
+from .models import TelegramData
+
+# Register your models here.
+@admin.register(TelegramData)
+class JobAdmin(ImportExportModelAdmin):
+    list_display = ('message_id','chat_id','description','creation_date','expiration_date')

django_code/apps/telegram_communication/apps.py

@@ -0,0 +1,7 @@
+from django.apps import AppConfig
+
+
+class TelegramCommunicationConfig(AppConfig):
+    default_auto_field = 'django.db.models.BigAutoField'
+    name = 'apps.telegram_communication'
+    label = 'telegram'

django_code/apps/telegram_communication/management/commands/new_database.py

@@ -0,0 +1,47 @@
+from django.core.management.base import BaseCommand
+from django.conf import settings
+import os
+#import jinja2
+from jinja2 import Environment, FileSystemLoader
+
+router_path = os.path.join(settings.BASE_DIR,'core','routers.py')
+
+class Command(BaseCommand):
+    args = '<team_id>'
+    help = 'Affiche la liste des backlogs'
+
+    def extract_settings_database(self):
+        settings_databases = []
+        for database in settings.DATABASES:
+            if database == 'default':
+                continue
+            print(database)
+
+            database_label = database.split('_db')[0]
+            
+            settings_databases.append([database,database_label])
+        return settings_databases
+
+    def coucou(self):
+        self.stdout.write('Coucou !')        
+
+    def templating(self,databases_list):
+        template_dir = os.path.join(settings.BASE_DIR, 'templates','routers')
+        env = Environment(loader=FileSystemLoader(template_dir))
+        template = env.get_template('routers.jinja.py')
+        output = template.render(databases_list = databases_list)
+    
+        with open(router_path, 'w') as f:
+            print(output, file = f)
+        
+        print(output)
+
+    def main(self):
+        databases_list = self.extract_settings_database()
+        print(databases_list)
+        self.templating(databases_list)
+        self.coucou()
+
+
+    def handle(self, *args, **options):
+        self.main()

django_code/apps/telegram_communication/management/commands/test_message.py

@@ -0,0 +1,19 @@
+from apps.telegram_communication.utils import telegramUtils
+from django.core.management.base import BaseCommand
+import datetime
+
+
+class Command(BaseCommand):
+    telegram = telegramUtils()
+
+    date_now = datetime.datetime.now()
+    
+    message = "Test message sent at "+str(date_now)
+
+    
+    telegram.send_message(message)
+    args = '<team_id>'
+    help = 'Affiche la liste des backlogs'
+
+    def handle(self, *args, **options):
+        self.stdout.write('Message sent !')

django_code/apps/telegram_communication/migrations/0001_initial.py

@@ -0,0 +1,28 @@
+# Generated by Django 4.2.30 on 2026-06-30 12:13
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    initial = True
+
+    dependencies = [
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='TelegramData',
+            fields=[
+                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('message_id', models.IntegerField(max_length=100, null=True)),
+                ('chat_id', models.IntegerField(max_length=100, null=True)),
+                ('description', models.CharField(max_length=100, null=True)),
+                ('creation_date', models.DateField(blank=True, null=True, verbose_name='Published')),
+                ('expiration_date', models.DateField(blank=True, null=True, verbose_name='Published')),
+            ],
+            options={
+                'db_table': 'messages',
+            },
+        ),
+    ]

django_code/apps/telegram_communication/migrations/0002_alter_telegramdata_chat_id_and_more.py

@@ -0,0 +1,33 @@
+# Generated by Django 4.2.30 on 2026-06-30 12:14
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('telegram', '0001_initial'),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name='telegramdata',
+            name='chat_id',
+            field=models.IntegerField(),
+        ),
+        migrations.AlterField(
+            model_name='telegramdata',
+            name='creation_date',
+            field=models.DateField(blank=True, verbose_name='Published'),
+        ),
+        migrations.AlterField(
+            model_name='telegramdata',
+            name='expiration_date',
+            field=models.DateField(blank=True, verbose_name='Published'),
+        ),
+        migrations.AlterField(
+            model_name='telegramdata',
+            name='message_id',
+            field=models.IntegerField(),
+        ),
+    ]

django_code/apps/telegram_communication/models.py

@@ -0,0 +1,12 @@
+from django.db import models
+
+
+class TelegramData(models.Model):
+    message_id = models.IntegerField(null=False)
+    chat_id = models.IntegerField(null=False)
+    description = models.CharField(max_length=100, null=True)
+    creation_date = models.DateField(blank=True, null=False, verbose_name='Published')
+    expiration_date = models.DateField(blank=True, null=False, verbose_name='Published')
+    class Meta:
+        db_table = 'messages'
+

django_code/apps/telegram_communication/tests.py

@@ -0,0 +1,3 @@
+from django.test import TestCase
+
+# Create your tests here.

django_code/apps/telegram_communication/utils.py

@@ -0,0 +1,86 @@
+import os
+import requests
+from apps.bitwarden.utils import bitwardenUtils
+from .models import TelegramData
+from datetime import datetime, timedelta
+
+class telegramUtils():
+
+    token_uid = "a88dbe52-b862-422a-a116-b479009733ab"
+    chat_id_uid = "13e948ab-494a-4b09-bb3a-b47900974cef"
+    
+    __bitwarden_access_token = os.environ.get('bitwarden_access_token')
+    __bitwarden_organization_id = os.environ.get('bitwarden_organization_id')
+    
+    bw = bitwardenUtils(__bitwarden_access_token,__bitwarden_organization_id)
+    
+    __TOKEN = bw.read_value_secret(token_uid)
+    __CHAT_ID = bw.read_value_secret(chat_id_uid)
+    
+    
+    
+    def get_chat_id(self):
+        # Source - https://stackoverflow.com/a/75118239
+        # Posted by Yasantha Biyuranga, modified by community. See post 'Timeline' for change history
+        # Retrieved 2026-06-30, License - CC BY-SA 4.0
+    
+        url = f"https://api.telegram.org/bot{self.__TOKEN}/getUpdates"
+        print(requests.get(url).json())
+        return 0
+    
+    def send_message(self,message):
+        # Source - https://stackoverflow.com/a/75118239
+        # Posted by Yasantha Biyuranga, modified by community. See post 'Timeline' for change history
+        # Retrieved 2026-06-30, License - CC BY-SA 4.0
+    
+        url = f"https://api.telegram.org/bot{self.__TOKEN}/sendMessage?chat_id={self.__CHAT_ID}&text={message}"
+       
+        result = requests.get(url).json() # this sends the message
+        send_result = result['ok']
+        TelegramData.objects.create(message_id=result['result']['message_id'],chat_id=self.__CHAT_ID,description=message[0:99],creation_date=datetime.now(),expiration_date=datetime.now()+timedelta(days=1))
+        return send_result
+    
+    def send_document(self,file,message=''):
+        url = 'https://api.telegram.org/bot{}/sendDocument'.format(self.__TOKEN)
+        data = {'chat_id': self.__CHAT_ID ,'caption': message}
+        document = open(file, 'rb')
+        files = {'document': document}
+        response = requests.post(url, data=data, files=files)
+        result = response.json()
+        
+        #Create data , for the dercsription use a message of set , or use the filename , cut it to grab the end if the filename is more then 99 characters  
+        TelegramData.objects.create(message_id=result['result']['message_id'],chat_id=self.__CHAT_ID,description=message[0:99] or str(file)[-99:],creation_date=datetime.now(),expiration_date=datetime.now()+timedelta(days=1))
+        return result['ok'],result['result']['message_id'],result
+        
+        #example send_document('apps/Le_journal_10h00_du_30_06_2026.mp3')
+    
+    def delete_message(self,message_id_list):
+        send_result = []
+        for message_id in message_id_list:
+            url = f"https://api.telegram.org/bot{self.__TOKEN}/deleteMessage?chat_id={self.__CHAT_ID}&message_id={message_id}"
+            
+            result = requests.get(url).json()
+            send_result.append(["Message id: "+str(message_id),result['ok']])
+            if result['ok']:
+                message = TelegramData.objects.filter(message_id=message_id)
+                message.delete()
+        return send_result
+
+    def clear_expired_messages(self):
+        today = datetime.now()
+        #filter all the messages that expired yesterday or before
+        expired_messages_list = [] 
+        expired_messages = TelegramData.objects.filter(expiration_date__lte=today)
+        
+        #leave if the list is empty 
+        if len(expired_messages) == 0:
+            return 1
+        
+        #create a list with only the message_id
+        for expired_message in expired_messages:
+            expired_messages_list.append(expired_message.message_id)
+            
+        #run the delete query with the list created before    
+        result = self.delete_message(expired_messages_list)
+        return result[0]
+

django_code/apps/telegram_communication/views.py

@@ -0,0 +1,3 @@
+from django.shortcuts import render
+
+# Create your views here.