.gitignore

@@ -137,9 +137,15 @@ dmypy.json
 # Cython debug symbols
 cython_debug/
 
+# Requests-cache
+redmine_cache.sqlite
+
 #####################################################
 # CUSTOM
 
+# requests cache
+instagram_cache.sqlite
+
 # direnv
 .envrc
 

.isort.cfg

@@ -3,4 +3,4 @@
 line_length = 88
 multi_line_output = 3
 include_trailing_comma = true
-known_third_party = PyPDF2,arrow,bleach,bs4,captcha,celery,dateutil,django,environ,faker,fastjsonschema,icalevnt,markdown,modelcluster,pirates,pytest,pytz,requests,sentry_sdk,taggit,tweepy,wagtail,wagtailmetadata,weasyprint,yaml
+known_third_party = PyPDF2,arrow,bleach,bs4,captcha,celery,dateutil,django,environ,faker,fastjsonschema,icalevents,markdown,modelcluster,pirates,pytest,pytz,requests,requests_cache,sentry_sdk,taggit,wagtail,wagtailmetadata,weasyprint,yaml

README.md

@@ -176,6 +176,7 @@ Přes CRON je třeba na pozadí spouštět Django `manage.py` commandy:
 * `update_main_timeline_articles` - aktualizuje články na `pirati.cz` z `https://piratipracuji.cz/api/`
 * `update_redmine_issues` - aktualizuje programované body MS a KS stránek napojených na Redmine (několikrát denně)
 * `update_tweets` - aktualizuje tweety podle nastavení na Homepage pirati.cz - vyžaduje mít v .env TWITTER_BEARER_TOKEN, parametr --days určuje stáří tweetů (default 1)
+* `update_instagram` - aktualizuje Instagramové posty na Homepage pirati.cz - vyžaduje mít v .env `INSTAGRAM_APP_ID` a `INSTAGRAM_APP_SECRET`.
 
 ### Fulltextové vyhledávání v češtině
 

calendar_utils/migrations/0004_auto_20220505_1228.py

@@ -3,7 +3,7 @@ from datetime import date, timedelta
 
 import arrow
 from django.db import migrations
-from icalevnt import icalevents
+from icalevents import icalevents
 
 from calendar_utils.parser import process_event_list
 

calendar_utils/models.py

@@ -4,7 +4,7 @@ from datetime import date, timedelta
 import arrow
 from django.core.serializers.json import DjangoJSONEncoder
 from django.db import models
-from icalevnt import icalevents
+from icalevents import icalevents
 
 from .parser import process_event_list
 

calendar_utils/parser.py

@@ -7,7 +7,7 @@ import bleach
 from django.conf import settings
 
 if TYPE_CHECKING:
-    from icalevnt.icalparser import Event
+    from icalevents.icalparser import Event
 
 EVENT_KEYS = ("start", "end", "all_day", "summary", "description", "location")
 

district/migrations/0106_alter_districtcrossroadpage_cards_content.py

 # Generated by Django 4.1.6 on 2023-02-28 07:51
 
-from django.db import migrations
 import wagtail.blocks
 import wagtail.fields
 import wagtail.images.blocks
+from django.db import migrations
 
 
 class Migration(migrations.Migration):
-
     dependencies = [
-        ('district', '0105_alter_districtarticlepage_content'),
+        ("district", "0105_alter_districtarticlepage_content"),
     ]
 
     operations = [
         migrations.AlterField(
-            model_name='districtcrossroadpage',
-            name='cards_content',
-            field=wagtail.fields.StreamField([('cards', wagtail.blocks.StructBlock([('headline', wagtail.blocks.CharBlock(label='Titulek bloku', required=False)), ('card_items', wagtail.blocks.ListBlock(wagtail.blocks.StructBlock([('image', wagtail.images.blocks.ImageChooserBlock(label='Obrázek')), ('title', wagtail.blocks.CharBlock(label='Titulek', required=True)), ('text', wagtail.blocks.RichTextBlock(label='Krátký text pod nadpisem', required=False)), ('page', wagtail.blocks.PageChooserBlock(label='Stránka', page_type=['district.DistrictArticlePage', 'district.DistrictArticlesPage', 'district.DistrictCenterPage', 'district.DistrictContactPage', 'district.DistrictCrossroadPage', 'district.DistrictCustomPage', 'district.DistrictElectionCampaignPage', 'district.DistrictElectionProgramPage', 'district.DistrictElectionRootPage', 'district.DistrictPeoplePage', 'district.DistrictPersonPage', 'district.DistrictPostElectionStrategyPage', 'district.DistrictProgramPage'], required=False)), ('link', wagtail.blocks.URLBlock(label='Odkaz', required=False))]), label='Karty s odkazy'))]))], blank=True, use_json_field=True, verbose_name='Karty rozcestníku'),
+            model_name="districtcrossroadpage",
+            name="cards_content",
+            field=wagtail.fields.StreamField(
+                [
+                    (
+                        "cards",
+                        wagtail.blocks.StructBlock(
+                            [
+                                (
+                                    "headline",
+                                    wagtail.blocks.CharBlock(
+                                        label="Titulek bloku", required=False
+                                    ),
+                                ),
+                                (
+                                    "card_items",
+                                    wagtail.blocks.ListBlock(
+                                        wagtail.blocks.StructBlock(
+                                            [
+                                                (
+                                                    "image",
+                                                    wagtail.images.blocks.ImageChooserBlock(
+                                                        label="Obrázek"
+                                                    ),
+                                                ),
+                                                (
+                                                    "title",
+                                                    wagtail.blocks.CharBlock(
+                                                        label="Titulek", required=True
+                                                    ),
+                                                ),
+                                                (
+                                                    "text",
+                                                    wagtail.blocks.RichTextBlock(
+                                                        label="Krátký text pod nadpisem",
+                                                        required=False,
+                                                    ),
+                                                ),
+                                                (
+                                                    "page",
+                                                    wagtail.blocks.PageChooserBlock(
+                                                        label="Stránka",
+                                                        page_type=[
+                                                            "district.DistrictArticlePage",
+                                                            "district.DistrictArticlesPage",
+                                                            "district.DistrictCenterPage",
+                                                            "district.DistrictContactPage",
+                                                            "district.DistrictCrossroadPage",
+                                                            "district.DistrictCustomPage",
+                                                            "district.DistrictElectionCampaignPage",
+                                                            "district.DistrictElectionProgramPage",
+                                                            "district.DistrictElectionRootPage",
+                                                            "district.DistrictPeoplePage",
+                                                            "district.DistrictPersonPage",
+                                                            "district.DistrictPostElectionStrategyPage",
+                                                            "district.DistrictProgramPage",
+                                                        ],
+                                                        required=False,
+                                                    ),
+                                                ),
+                                                (
+                                                    "link",
+                                                    wagtail.blocks.URLBlock(
+                                                        label="Odkaz", required=False
+                                                    ),
+                                                ),
+                                            ]
+                                        ),
+                                        label="Karty s odkazy",
+                                    ),
+                                ),
+                            ]
+                        ),
+                    )
+                ],
+                blank=True,
+                use_json_field=True,
+                verbose_name="Karty rozcestníku",
+            ),
         ),
     ]

twitter_utils/apps.py → instagram_utils/apps.py

 from django.apps import AppConfig
 
 
-class TwitterUtilsConfig(AppConfig):
-    name = "twitter_utils"
+class InstagramUtilsConfig(AppConfig):
+    name = "instagram_utils"

instagram_utils/management/commands/update_instagram.py

+from django.conf import settings
+from django.core.management.base import BaseCommand
+
+from ...services import InstagramDownloadService
+
+
+class Command(BaseCommand):
+    def handle(self, *args, **options):
+        service = InstagramDownloadService(
+            app_id=settings.INSTAGRAM_APP_ID,
+            app_secret=settings.INSTAGRAM_APP_SECRET,
+        )
+        service.perform_update()
+
+        self.stdout.write("\nInstagram post update finished.")

instagram_utils/migrations/0001_initial.py

+# Generated by Django 4.1.6 on 2023-04-05 17:16
+
+from django.db import migrations, models
+import instagram_utils.models
+
+
+class Migration(migrations.Migration):
+
+    initial = True
+
+    dependencies = [
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='InstagramPost',
+            fields=[
+                ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('author_name', models.CharField(max_length=64, verbose_name='Jméno autora')),
+                ('author_username', models.CharField(max_length=64, verbose_name='Username autora')),
+                ('timestamp', models.DateTimeField(default=instagram_utils.models.get_current_datetime, verbose_name='Datum a čas vytvoření')),
+                ('caption', models.TextField(blank=True, null=True, verbose_name='Popis')),
+                ('image', models.ImageField(upload_to='instagram', verbose_name='Obrázek')),
+                ('url', models.URLField(blank=True, null=True, verbose_name='Odkaz')),
+            ],
+            options={
+                'ordering': ('timestamp',),
+            },
+        ),
+    ]

instagram_utils/migrations/0002_instagrampost_remote_id.py

+# Generated by Django 4.1.6 on 2023-04-05 17:18
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('instagram_utils', '0001_initial'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='instagrampost',
+            name='remote_id',
+            field=models.CharField(default='', max_length=64, verbose_name='ID Postu'),
+            preserve_default=False,
+        ),
+    ]

instagram_utils/migrations/0003_alter_instagrampost_remote_id.py

+# Generated by Django 4.1.6 on 2023-04-05 17:19
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('instagram_utils', '0002_instagrampost_remote_id'),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name='instagrampost',
+            name='remote_id',
+            field=models.CharField(max_length=64, unique=True, verbose_name='ID Postu'),
+        ),
+    ]

instagram_utils/models.py

+import datetime
+
+from django.db import models
+
+
+def get_current_datetime() -> datetime.datetime:
+    return datetime.datetime.now(tz=datetime.timezone.utc)
+
+
+class InstagramPost(models.Model):
+    """
+    Model representing an Instgram post obtained from its API through the
+    update_instagram management command.
+    """
+
+    remote_id = models.CharField(
+        verbose_name="ID Postu",
+        max_length=64,
+        unique=True,
+    )
+    timestamp = models.DateTimeField(
+        verbose_name="Datum a čas vytvoření",
+        default=get_current_datetime
+    )
+
+    author_name = models.CharField(
+        verbose_name="Jméno autora",
+        max_length=64,
+    )
+    author_username = models.CharField(
+        verbose_name="Username autora",
+        max_length=64,
+    )
+
+    caption = models.TextField(
+        verbose_name="Popis",
+        blank=True,
+        null=True,
+    )
+    image = models.ImageField(
+        verbose_name="Obrázek",
+        upload_to="instagram",
+    )
+    url = models.URLField(
+        verbose_name="Odkaz",
+        blank=True,
+        null=True,
+    )
+
+    def __str__(self) -> str:
+        return f"@{self.author_username} - {self.caption}"
+
+    class Meta:
+        ordering = ("timestamp",)

instagram_utils/services.py

+import datetime
+import logging
+import io
+import os
+import requests_cache
+
+from django.core.files import File
+from main.models import MainHomePage, MainPersonPage
+
+from .models import InstagramPost
+
+
+logger = logging.getLogger()
+
+
+class InstagramDownloadService:
+    """
+    TODO
+    """
+
+    def __init__(self, app_id: int, app_secret: str):
+        self.session = requests_cache.CachedSession("instagram_cache")
+        
+        self.app_id = app_id
+        self.app_secret = app_secret
+
+    def get_user_info_list(self) -> list[str]:
+        access_block = MainHomePage.objects.first().instagram_access
+
+        homepage_access_list = [
+            (
+                block["value"]["name"],
+                block["value"]["access_token"]
+            )
+            for block in access_block.raw_data
+        ]
+
+        people_access_list = []
+
+        for people_page in MainPersonPage.objects.all():
+            people_access_list += [
+                (
+                    block["value"]["name"],
+                    block["value"]["access_token"]
+                )
+                for block in people_page.instagram_access.raw_data
+            ]
+
+        # Remove duplicates
+        return list({*people_access_list, *homepage_access_list})
+
+    def download_remote_image(self, image_url) -> (str, File):
+        try:
+            response = self.session.get(image_url)
+            response.raise_for_status()
+        except Exception as exc:
+            logger.warning(
+                "Error getting Instagram image at %s: %s",
+                image_url, exc
+            )
+            return "", None
+
+        return os.path.basename(image_url), File(io.BytesIO(response.content))
+
+    def get_user_data(self, access_token: str) -> dict:
+        user_data = self.session.get(
+            f"https://graph.instagram.com/v16.0/me?access_token={access_token}"
+            "&fields=id,username"
+        )
+        user_data.raise_for_status()
+
+        return user_data.json()
+
+    def get_recent_media(self, user_data: dict, access_token: str) -> list[dict]:
+        with self.session.cache_disabled():
+            recent_media = self.session.get(
+                f"https://graph.instagram.com/v16.0/{user_data['id']}/media?access_token="
+                f"{access_token}&fields=id,timestamp,caption,media_type,permalink,"
+                "media_url,thumbnail_url"
+            )
+
+        if not recent_media.ok:
+            logger.warning(
+                "Error getting media for user %s: %s",
+                user_data["id"],
+                recent_media.status_code
+            )
+
+            return []
+
+        logger.debug("Parsing Instagram feed: %s", recent_media)
+
+        return recent_media.json()["data"]
+
+    def parse_media_for_user(self, name: str, access_token: str) -> None:
+        user_data = self.get_user_data(access_token)
+        recent_media_json = self.get_recent_media(user_data, access_token)
+
+        if len(recent_media_json) == 0:
+            return
+
+        posts = []
+
+        for media_data in recent_media_json:
+            # Don't recreate existing posts'
+            if InstagramPost.objects.filter(remote_id=media_data["id"]).exists():
+                logging.info(
+                    "Skipping Instagram post ID %s, already exists",
+                    media_data["id"]
+                )
+
+                continue
+
+            post = InstagramPost(
+                remote_id=media_data["id"],
+                author_name=name,
+                author_username=user_data["username"],
+                timestamp=datetime.datetime.strptime(
+                    media_data["timestamp"],
+                    "%Y-%m-%dT%H:%M:%S%z",
+                ),
+                caption=media_data["caption"],
+                url=media_data["permalink"],
+            )
+
+            post.image.save(
+                *self.download_remote_image(media_data["media_url"]),
+                False,  # Don't save yet
+            )
+
+            post.save()
+
+            logger.info(
+                "Saved Instagram post ID %s",
+                post.remote_id,
+            )
+
+    def perform_update(self) -> None:
+        user_info_list = self.get_user_info_list()
+
+        media_list = []
+
+        for user_info in user_info_list:
+            self.parse_media_for_user(*user_info)