Merge pull request #21 from La-Forge/ileDeFranceAppelProjet

feat: ✨ add scrapping for IDF's appel à projet page

Author: NassimBennouar

generate_feeds.py

@@ -1,20 +1,23 @@
 from scrappers.GniusScrapper import GniusScrapper, FEED_PATH as GNIUS_FEED_PATH
 from scrappers.BpifranceScrapper import BpifranceScrapper, FEED_PATH as BPI_FEED_PATH
+from scrappers.IleDeFranceScrapper import IleDeFranceScrapper, FEED_PATH as IDF_FEED_PATH
 
 import os
 import argparse
 
 
-def main(verbose, update_bpi, update_gnius):
+def main(verbose, update_bpi, update_gnius, update_idf):
     script_dir = os.path.dirname(os.path.abspath(__file__))
     feeds_dir = os.path.join(script_dir, "feeds")
     os.makedirs(feeds_dir, exist_ok=True)
 
     bpi_scrapper = BpifranceScrapper()
     gnius_scrapper = GniusScrapper()
+    idf_scrapper = IleDeFranceScrapper()
 
     bpi_feed_file = BPI_FEED_PATH  # os.path.join(feeds_dir, BPI_FEED_PATH)
     gnius_feed_file = GNIUS_FEED_PATH  # os.path.join(feeds_dir, 'gnius_feed.xml')
+    idf_feed_file = IDF_FEED_PATH  # os.path.join(feeds_dir, 'idf_feed.xml')
 
     if update_bpi:
         print(f"Updating {bpi_feed_file}...")
@@ -26,22 +29,29 @@ def main(verbose, update_bpi, update_gnius):
         gnius_scrapper.update_feed_file(gnius_feed_file, verbose=verbose)
         print(f"{gnius_feed_file} updated.")
 
+    # Updating XML's file for IDF's appel à projets
+    if update_idf:
+        print(f"Updating {idf_feed_file}...")
+        idf_scrapper.update_feed_file(idf_feed_file, verbose=verbose)
+        print(f"{idf_feed_file} updated.")
 
 if __name__ == "__main__":
     parser = argparse.ArgumentParser(
         description="Update feeds with optional verbosity."
     )
     parser.add_argument(
-        "--bpifrance", action="store_true", help="Update only bpifrance feed"
+        "--bpifrance", action="store_true", help="Update only Bpifrance feed"
     )
-    parser.add_argument("--gnius", action="store_true", help="Update only gnius feed")
+    parser.add_argument("--gnius", action="store_true", help="Update only Gnius feed")
+    parser.add_argument("--idf", action="store_true", help="Update only IDF feed")
     parser.add_argument(
         "-v", "--verbose", action="store_true", help="Enable verbose output"
     )
     args = parser.parse_args()
 
-    # Si ni bpifrance ni gnius n'est spécifié, on met à jour les deux
-    update_bpi = args.bpifrance or (not args.bpifrance and not args.gnius)
-    update_gnius = args.gnius or (not args.bpifrance and not args.gnius)
+    # Si aucune option n'est spécifiée, on met à jour tous les feeds
+    update_bpi = args.bpifrance or (not args.bpifrance and not args.gnius and not args.idf)
+    update_gnius = args.gnius or (not args.bpifrance and not args.gnius and not args.idf)
+    update_idf = args.idf or (not args.bpifrance and not args.gnius and not args.idf)
 
-    main(args.verbose, update_bpi, update_gnius)
+    main(args.verbose, update_bpi, update_gnius, update_idf)

scrappers/APIScrapper.py

@@ -0,0 +1,41 @@
+from scrappers.BaseScrapper import BaseScrapper
+from html import unescape
+import requests
+from bs4 import BeautifulSoup
+import dateparser
+from feedgen.feed import FeedGenerator
+from sentry_sdk import capture_exception
+
+class APIScrapper(BaseScrapper):
+
+    def __init__(self, base_url, host, feed_title, feed_author, feed_link):
+        self.base_url = base_url
+        self.host = host
+        self.feed_title = feed_title
+        self.feed_author = feed_author
+        self.feed_link = feed_link
+
+    def generate_feed(self, verbose=True):
+        fg = FeedGenerator()
+        fg.title(self.feed_title)
+        fg.id(self.feed_link)
+        fg.author({"name": self.feed_author})
+        fg.link(href=self.feed_link, rel="alternate")
+        fg.subtitle("Powered by www.la-forge.ai")
+        fg.language("fr")
+
+        try:
+            articles = self.scrapPages(verbose=verbose)
+            for article in articles:
+                fe = fg.add_entry()
+                fe.id(article["link"])
+                fe.title(article["title"])
+                fe.link(href=article["link"])
+                fe.description(article["description"])
+                fe.pubDate(article["date"])
+        except Exception as e:
+            print(e)
+            capture_exception(e)
+
+        atomfeed = fg.atom_str(pretty=True)
+        return atomfeed

scrappers/BaseScrapper.py

@@ -21,17 +21,6 @@ def __init__(self, base_url, host, feed_title, feed_author, feed_link):
         self.feed_link = feed_link
 
     def scrapPages(self, verbose=False):
-        posts = []
-        page = 0
-        count_for_current_page = -1
-        while count_for_current_page != 0:
-            posts_on_current_page = self.scrapPage(pageNumber=page, verbose=verbose)
-            posts.extend(posts_on_current_page)
-            count_for_current_page = len(posts_on_current_page)
-            page = page + 1
-        return posts
-
-    def scrapPage(self, pageNumber, verbose=False):
         raise NotImplementedError("This method should be overridden by subclasses")
 
     def print_data(self, verbose=False):
@@ -48,47 +37,10 @@ def write_feed_to_file(self, feed, filename):
         with open(filename, "wb") as file:
             file.write(feed)
 
-    def get_full_article_content(self, article_url, content_class):
-        response = requests.get(article_url)
-        if response.status_code == 200:
-            soup = BeautifulSoup(response.content, "html.parser")
-            article_content = soup.find(class_=content_class).get_text()
-            return article_content
-        else:
-            print(f"Failed to fetch article content from URL: {article_url}")
-            return ""
-
     def generate_feed(self, verbose=True):
-        fg = FeedGenerator()
-        fg.title(self.feed_title)
-        fg.id(self.feed_link)
-        fg.author({"name": self.feed_author})
-        fg.link(href=self.feed_link, rel="alternate")
-        fg.subtitle("Powered by www.la-forge.ai")
-        fg.language("fr")
-
-        try:
-            articles = self.scrapPages(verbose=verbose)
-            for article in articles:
-                fe = fg.add_entry()
-                fe.id(article["link"])
-                fe.title(article["title"])
-                fe.link(href=article["link"])
-                fe.description(article["description"])
-                fe.pubDate(article["date"])
-                full_content = self.get_full_article_content(
-                    article["link"], article.get("content_class")
-                )
-                if full_content:
-                    fe.content(full_content, type="CDATA")
-        except Exception as e:
-            print(e)
-            capture_exception(e)
-
-        atomfeed = fg.atom_str(pretty=True)
-        return atomfeed
+        raise NotImplementedError("This method should be overridden by subclasses")
 
     def update_feed_file(self, filename="feed.xml", verbose=False):
         feed = self.generate_feed(verbose=verbose)
         self.write_feed_to_file(feed, filename)
-        return feed
+        return feed

scrappers/BpifranceScrapper.py

@@ -1,12 +1,18 @@
-from scrappers.BaseScrapper import BaseScrapper
+from scrappers.WebScrapper import WebScrapper
 from html import unescape
 import requests
 from bs4 import BeautifulSoup
 import dateparser
+from feedgen.feed import FeedGenerator
+from sentry_sdk import capture_exception
+
 
 FEED_PATH = 'feeds/bpi_feed.xml'
 
-class BpifranceScrapper(BaseScrapper):
+class BpifranceScrapper(WebScrapper):
+    """
+    Classe pour scrapper les données de BpiFrance - appel à projets. 
+    """
     def __init__(self):
         super().__init__(
             base_url = "https://www.bpifrance.fr/views/ajax?_wrapper_format=drupal_ajax&labels=All&view_name=events_before_end_date&view_display_id=events_finishing_more_week&view_args=496&view_path=%2Fnode%2F7620&view_base_path=&view_dom_id=de2b6579af442525efdb3720e2433d578ae6af46c8d2cb9812d17facde4592ff&pager_element=0&_drupal_ajax=1&ajax_page_state%5Btheme%5D=bpi_main&ajax_page_state%5Btheme_token%5D=vUo2YdcgaSQx1XGJHIa_CX496Ili2qa2-fmRJpfpgV8&ajax_page_state%5Blibraries%5D=eJxtztsOwjAIBuAXqusjNXTFDkcPFqrOp3fuYotxN-TnCxA8qmJz-KpFMLgr8dqKha7FSfeJ1PjzkYgZG7DxlRzDe3GYlXSxCShv-A02cvHAFxkbVZV_14UpR1OhQWxQJ7Gh9Qo8HDL0XLtnkgmDuXca53Vltns6M0d5_VwUlERp3K8eYmQRxWQ9CJoH4VPsVge4wesHUgmd8QPX0HW2",
@@ -109,4 +115,4 @@ def get_article_content(self, article: BeautifulSoup) -> str:
         content = ""
         if p and len(p):
             content = p[0].text.strip()
-        return content
+        return content

scrappers/GniusScrapper.py

@@ -1,13 +1,18 @@
-from scrappers.BaseScrapper import BaseScrapper
+from scrappers.WebScrapper import WebScrapper
 from html import unescape
 import requests
 from bs4 import BeautifulSoup
 import dateparser
+from feedgen.feed import FeedGenerator
+from sentry_sdk import capture_exception
 
 
 FEED_PATH = 'feeds/gnius_feed.xml'
 
-class GniusScrapper(BaseScrapper):
+class GniusScrapper(WebScrapper):
+    """
+    Classe pour scrapper les données de GNius - actualités. 
+    """
     def __init__(self):
         super().__init__(
             base_url="https://gnius.esante.gouv.fr/fr/a-la-une/actualites?page=<page-number>",

scrappers/IleDeFranceScrapper.py

@@ -0,0 +1,84 @@
+import requests
+from scrappers.APIScrapper import APIScrapper
+import datetime 
+
+FEED_PATH = 'feeds/idf_feed.xml'
+
+class IleDeFranceScrapper(APIScrapper):
+    """
+    Classe pour scrapper les données de l'Ile-de-France - appel à projets. 
+    """
+    def __init__(self):
+        super().__init__(
+            base_url="https://data.iledefrance.fr/api/explore/v2.1/catalog/datasets/aides-appels-a-projets/records",
+            host="data.iledefrance.fr",
+            feed_title="Aides et Appels à Projets - Île-de-France",
+            feed_author="Île-de-France",
+            feed_link="https://data.iledefrance.fr/explore/dataset/aides-appels-a-projets/",
+        )
+        self.limit_per_request = 100  
+
+    def scrapPages(self, verbose=False):
+        """
+        Récupère tous les enregistrements en gérant la limite de taille par requête
+        """
+        all_records = []
+        offset = 0
+
+        while True:
+            params = {
+                "limit": self.limit_per_request,
+                "offset": offset,
+            }
+
+            response = requests.get(self.base_url, params=params)
+            response.raise_for_status()  
+            data = response.json()
+
+            current_records = data.get("results", [])
+            all_records.extend(current_records)
+
+            if verbose:
+                print(f"Page avec offset {offset} : {len(current_records)} enregistrements récupérés.")
+
+            if len(current_records) < self.limit_per_request:
+                break
+
+            offset += self.limit_per_request
+
+        if verbose:
+            print(f"Nombre total d'enregistrements récupérés : {len(all_records)}")
+
+        return self.format_articles(all_records)
+
+    def format_articles(self, data):
+        """
+        Formate les données API dans le format attendu par BaseScrapper.
+        """
+        articles = []
+        for record in data:
+            fields = record
+            description_parts = [
+            f"Description : {fields.get('chapo_txt', 'Pas de description disponible')}",
+            f"Pour quel type de projet : {fields.get('objectif_txt', 'Non spécifié')}",
+            f"Qui peut en bénéficier : {', '.join(fields.get('qui_peut_en_beneficier', [])) or 'Non spécifié'}"
+            ]
+            description = "\n".join(description_parts)
+            articles.append({
+                "title": fields.get("nom_de_l_aide_de_la_demarche", "Titre inconnu"),
+                "link": fields.get("url_descriptif", ""),
+                "description": description,
+                "date": self.parse_date(fields.get("date")),
+                "content_class": None, 
+            })
+        return articles
+
+    @staticmethod
+    def parse_date(date_str):
+        """
+        Transforme une date au format ISO 8601 en format RSS (RFC 822).
+        """
+        try:
+            return datetime.strptime(date_str, "%Y-%m-%dT%H:%M:%S%z").strftime("%a, %d %b %Y %H:%M:%S %z")
+        except Exception:
+            return None

scrappers/WebScrapper.py

@@ -0,0 +1,69 @@
+from scrappers.BaseScrapper import BaseScrapper
+from html import unescape
+import requests
+from bs4 import BeautifulSoup
+import dateparser
+from feedgen.feed import FeedGenerator
+from sentry_sdk import capture_exception
+
+class WebScrapper(BaseScrapper):
+
+    def __init__(self, base_url, host, feed_title, feed_author, feed_link):
+        self.base_url = base_url
+        self.host = host
+        self.feed_title = feed_title
+        self.feed_author = feed_author
+        self.feed_link = feed_link
+
+    def scrapPages(self, verbose=False):
+        posts = []
+        page = 0
+        count_for_current_page = -1
+
+        while count_for_current_page != 0:
+            posts_on_current_page = self.scrapPage(pageNumber=page, verbose=verbose)
+            count_for_current_page = len(posts_on_current_page)
+            posts.extend(posts_on_current_page)
+            page += 1
+
+        return posts
+    
+    def get_full_article_content(self, article_url, content_class):
+        response = requests.get(article_url)
+        if response.status_code == 200:
+            soup = BeautifulSoup(response.content, "html.parser")
+            article_content = soup.find(class_=content_class).get_text()
+            return article_content
+        else:
+            print(f"Failed to fetch article content from URL: {article_url}")
+            return ""
+        
+    def generate_feed(self, verbose=True):
+        fg = FeedGenerator()
+        fg.title(self.feed_title)
+        fg.id(self.feed_link)
+        fg.author({"name": self.feed_author})
+        fg.link(href=self.feed_link, rel="alternate")
+        fg.subtitle("Powered by www.la-forge.ai")
+        fg.language("fr")
+
+        try:
+            articles = self.scrapPages(verbose=verbose)
+            for article in articles:
+                fe = fg.add_entry()
+                fe.id(article["link"])
+                fe.title(article["title"])
+                fe.link(href=article["link"])
+                fe.description(article["description"])
+                fe.pubDate(article["date"])
+                full_content = self.get_full_article_content(
+                    article["link"], article.get("content_class")
+                )
+                if full_content:
+                    fe.content(full_content, type="CDATA")
+        except Exception as e:
+            print(e)
+            capture_exception(e)
+
+        atomfeed = fg.atom_str(pretty=True)
+        return atomfeed