dataflow.yaml

apiVersion: 0.5.0
meta:
  name: recipe-processing
  version: 0.1.0
  namespace: examples

config:
  converter: json
  consumer:
    default_starting_offset:
      value: 0
      position: End
  producer:
    batch_size: 1000000

types:
  source:
    type: object
    properties:
      id:
        type: string
      name:
        type: string

  ingredient-list:
    type: list
    items:
      type: string

  recipe:
    type: object
    properties:
      title:
        type: string
      author:
        type: string
      ready_in_minutes:
        type: u32
      servings:
        type: u32
      ingredients:
        type: ingredient-list
      source_url:
        type: string

  recipe-list:
    type: list
    items:
      type: recipe

  recipes:
    type: object
    properties:
      chatid:
        type: string
      results:
        type: recipe-list

  ingredient-count:
    type: object
    properties:
      ingredient:
        type: string
      count:
        type: u32

  ingredient-count-list:
    type: list
    items:
      type: ingredient-count
      
  summarized-recipe:
    type: object
    properties:
      counts:
        type: ingredient-count-list

  labels:
    type: object
    properties:
      label:
        type: string
      score:
        type: f64

  categories-list:
    type: list
    items:
      type: labels

  classification:
    type: object
    properties:
      text:
        type: string
      categories:
        type: categories-list

  classification-list:
    type: list
    items:
      type: classification

  classifications:
    type: object
    properties:
      chatid:
        type: string
      results:
        type: classification-list

topics:
  userupdates:
    schema:
      value:
        type: string
        converter: raw
  recipes:
    schema:
      value:
        type: recipes
  summarized-recipes:
    schema:
      value:
        type: summarized-recipe
  classified-recipes:
    schema:
      value:
        type: classifications
  telegram-logs:
    schema:
      value:
        type: string

services:
  fetch-recipes:
    sources:
      - type: topic
        id: userupdates
    transforms:
      - operator: map
        dependencies:
          - name: sdf-http
            git: "https://github.com/infinyon/sdf-http-guest"
            tag: "v0.4.0"
          - name: serde_json
            version: "1.0.117"
        run: |
          fn fetch_recipes(chatid: String) -> Result<Recipes> {
              use sdf_http::blocking::send;
              use sdf_http::http::Request;
              use serde_json::{Value};

              let url = "https://api.spoonacular.com/recipes/random?number=5";
          
              let api_key = std::env::var("SPOONACULAR_KEY")?;

              let request = Request::builder()
                  .method("GET")
                  .header("x-api-key", api_key)
                  .header("User-Agent","Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/127.0.0.0 Safari/537.36")
                  .uri(url)
                  .body("")?;

              let response = send(request)?;
              let body = response.into_body();
              let body_str = String::from_utf8(body.clone())?;

              let json: Value = serde_json::from_str(&body_str)?;

              let recipes = json["recipes"]
                  .as_array()
                  .map(|recipes| {
                      recipes.iter().map(|recipe| {
                          let title = recipe["title"].as_str().unwrap_or("").to_string();
                          let ready_in_minutes = recipe["readyInMinutes"].as_u64().unwrap_or(0) as u32;
                          let servings = recipe["servings"].as_u64().unwrap_or(0) as u32;
                          let source_url = recipe["sourceUrl"].as_str().unwrap_or("").to_string();
                          let ingredients = recipe["extendedIngredients"]
                              .as_array()
                              .unwrap_or(&Vec::new())
                              .iter()
                              .map(|ingredient| ingredient["name"].as_str().unwrap_or("").to_string())
                              .collect::<Vec<String>>();

                          Recipe {
                              title,
                              author: "Spoonacular".to_string(),
                              ready_in_minutes,
                              servings,
                              ingredients,
                              source_url,
                          }
                      }).collect::<Vec<Recipe>>()
                  })
                  .unwrap_or_else(Vec::new);

              let result = Recipes {
                  chatid,
                  results: recipes,
              };

              Ok(result)
            }

    sinks:
      - type: topic
        id: recipes
  summarize-recipes:
    sources:
      - type: topic
        id: recipes

    transforms:
      - operator: map
        run: |
          fn ingredient_count_of_recipes(recipes: Recipes) -> Result<SummarizedRecipe> {

            let mut ingredient_count = std::collections::HashMap::new();
            for recipe in &recipes.results {
              for ingredient in &recipe.ingredients {
                *ingredient_count.entry(ingredient.clone()).or_insert(0) += 1;
              }
            }
            let ingredient_counts: Vec<IngredientCount> = ingredient_count.into_iter()
              .map(|(ingredient, count)| IngredientCount { ingredient, count })
              .collect();

            Ok(SummarizedRecipe {
                counts: ingredient_counts
              })
          }

    sinks:
      - type: topic
        id: summarized-recipes
  classify-recipes:
    sources:
      - type: topic
        id: recipes

    transforms:
      - operator: map
        dependencies:
          - name: sdf-http
            git: "https://github.com/infinyon/sdf-http-guest"
            tag: "v0.4.0"
          - name: serde_json
            version: "1.0.117"
        run: |
          fn classify_recipes(recipes: Recipes) -> Result<Classifications> {
              use sdf_http::blocking::send;
              use sdf_http::http::Request;
              use serde_json::{Value};

              let url = "https://api.spoonacular.com/recipes/categorize";
              let mut classification_list: Vec<Classification> = Vec::new();

              for recipe in recipes.results.iter().take(5) {
                  let text = recipe.title.clone();
                  let payload = format!("text={}", text);
          
                  let api_key = std::env::var("SPOONACULAR_KEY")?;                  

                  let request = Request::builder()
                      .method("POST")
                      .uri(url)
                      .header("x-api-key", api_key)
                      .header("Content-Type", "application/x-www-form-urlencoded")
                      .body(payload)?;

                  let response = send(request)?;
                  let body = response.into_body();
                  let body_str = String::from_utf8(body.clone())?;

                  let mut json: Value = serde_json::from_str(&body_str)?;

                  if let Some(categories) = json.get_mut("categories").and_then(Value::as_array_mut) {
                      let owned_categories: Vec<Labels> = categories.iter()
                          .map(|category| Labels {
                              label: category.get("label").unwrap_or(&Value::Null).as_str().unwrap_or("").to_string(),
                              score: category.get("score").unwrap_or(&Value::Null).as_f64().unwrap_or(0.0),
                          })
                          .collect();

                      classification_list.push(Classification {
                          text: text.clone(),
                          categories: owned_categories,
                      });
                  }
              }

              Ok(Classifications {
                  chatid: recipes.chatid.clone(),
                  results: classification_list,
              })
          }

    sinks:
      - type: topic
        id: classified-recipes
        
  send-to-telegram:
    sources:
      - type: topic
        id: classified-recipes
    transforms:
      - operator: map
        dependencies:
          - name: sdf-http
            git: "https://github.com/infinyon/sdf-http-guest"
            tag: "v0.4.0"
          - name: serde_json
            version: 1.0.117
        run: |
          fn format_for_telegram(classifications: Classifications) -> Result<String> {
              use sdf_http::http::Request;
              use sdf_http::blocking::send;
              use std::time::{SystemTime, UNIX_EPOCH};

            
            let bot_token = std::env::var("BOT_TOKEN")?;
            let chat_id = classifications.chatid;
            let mut text = String::new();
            text.push_str(&format!("*Classifying Recipes:*\n\n"));

            for classification in &classifications.results {
                text.push_str(&format!("*Recipe:* {}\n", classification.text));
                text.push_str(&format!("*Classified Labels:*\n"));
                for category in &classification.categories {
                    text.push_str(&format!(" - *{}*: {:.2}\n", category.label, category.score));
                }
                text.push('\n');
            }

            let url = format!("https://api.telegram.org/bot{}/sendMessage", bot_token);
            let body = serde_json::json!({
                "chat_id": chat_id,
                "text": text,
                "parse_mode": "Markdown"
            });

            let request = Request::builder()
                .method("POST")
                .uri(url)
                .header("Content-Type", "application/json")
                .body(body.to_string())?;

            let response = send(request)?;
            let body = response.into_body();
            let body_str = String::from_utf8(body)?;

            let json: serde_json::Value = serde_json::from_str(&body_str)?;
            let duration_since_epoch = SystemTime::now().duration_since(UNIX_EPOCH)?;
            let timestamp = duration_since_epoch.as_secs();
                if json.get("ok").and_then(serde_json::Value ::as_bool).unwrap_or(false) {
                    Ok(format!(
                    "timestamp: {} chatid: {} message sent successfully text: {}",
                        timestamp,
                        chat_id,
                        text
                    ))
                } else {
                    
                    let error_message = json.get("description")
                    .and_then(serde_json::Value::as_str)
                    .unwrap_or("Unknown error");
            
                    Ok(format!(
                    "timestamp: {} chatid: {} error: {}",
                    timestamp,
                    chat_id,
                    error_message
                    ))
                }
          }
    sinks:
      - type: topic
        id: telegram-logs